Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205use std::sync::Arc;
use axum::body::{Body, to_bytes};use bobbin_edge_index::{ Coverage, CoverageWatch, EdgeStore, HydrantCursor, IssueStateKind, PageToken, PullStatusKind, StateIndex,};use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig};use bobbin_record_lru::{CacheCapacity, LruRecordStore};use bobbin_resolver::RepoIdResolver;use bobbin_runtime::{RuntimeHasher, SystemClock};use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader};use bobbin_slingshot_client::SlingshotClient;use bobbin_types::edges::Edge;use bobbin_types::ids::SubjectRef;use bobbin_xrpc::{AppState, router};use futures::stream::{self, StreamExt};use http::{Request, StatusCode};use jacquard_common::DefaultStr;use jacquard_common::types::did::Did;use jacquard_common::types::nsid::Nsid;use jacquard_common::types::recordkey::Rkey;use jacquard_common::types::string::AtUri;use serde_json::{Value, json};use tower::ServiceExt;use url::Url;use url::form_urlencoded::byte_serialize;use wiremock::matchers::{method, path, query_param};use wiremock::{Mock, MockServer, ResponseTemplate};
const CID: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i";
fn at(s: &str) -> AtUri<DefaultStr> { AtUri::new_owned(s).unwrap()}
fn did(s: &str) -> Did<DefaultStr> { Did::new_owned(s).unwrap()}
fn rkey(s: &str) -> Rkey<DefaultStr> { Rkey::new_owned(s).unwrap()}
fn nsid(s: &'static str) -> Nsid<DefaultStr> { Nsid::new_static(s).unwrap()}
fn subj(s: &str) -> SubjectRef { Did::<DefaultStr>::new_owned(s) .map(SubjectRef::Did) .unwrap_or_else(|_| SubjectRef::Uri(AtUri::new_owned(s).unwrap()))}
struct Harness { server: MockServer, edges: Arc<EdgeStore>, coverage: Arc<CoverageWatch>, state: AppState,}
static EDGE_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
fn next_sort_micros() -> u64 { EDGE_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed)}
impl Harness { async fn new() -> Self { let server = MockServer::start().await; let edges = Arc::new(EdgeStore::new(RuntimeHasher::default())); let issue_states = Arc::new(StateIndex::new(RuntimeHasher::default())); let pull_statuses = Arc::new(StateIndex::new(RuntimeHasher::default())); let coverage = Arc::new(CoverageWatch::new()); let state = AppState::new( Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap(), edges.clone(), issue_states.clone(), pull_statuses.clone(), coverage.clone(), Arc::new( KnotProxy::new( KnotProxyConfig::default(), KnotHttpConfig::default(), Arc::new(SystemClock::new()), RuntimeHasher::default(), ) .unwrap(), ), Arc::new( SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap(), ) as Arc<dyn SearchReader>, Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), ); Self { server, edges, coverage, state, } }
fn add_edge( &self, kind: &Nsid<DefaultStr>, subject: &AtUri<DefaultStr>, source: &AtUri<DefaultStr>, ) { self.edges.add(Edge { kind: kind.clone(), subject: subj(subject.as_ref()), source: source.clone(), sort_micros: next_sort_micros(), }); }
async fn mount( &self, did: &Did<DefaultStr>, collection: &Nsid<DefaultStr>, rkey: &Rkey<DefaultStr>, value: Value, ) { let uri = format!( "at://{}/{}/{}", did.as_ref(), collection.as_ref(), rkey.as_ref() ); let body = json!({ "uri": uri, "cid": CID, "value": value }); Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .and(query_param("repo", did.as_ref())) .and(query_param("collection", collection.as_ref())) .and(query_param("rkey", rkey.as_ref())) .respond_with(ResponseTemplate::new(200).set_body_json(body)) .mount(&self.server) .await; }
fn promote_ready(&self, events: u64, cursor: u64) { self.coverage.update(|_| Coverage::Ready { events_processed: events, last_cursor: HydrantCursor::new(cursor), }); }
fn warming(&self, events: u64, cursor: u64) { self.coverage.update(|_| Coverage::Warming { events_processed: events, last_cursor: HydrantCursor::new(cursor), }); }}
fn list_request(endpoint: &str, subject: &str, extras: &[(&str, &str)]) -> Request<Body> { let mut qs = format!("subject={subject}"); extras.iter().for_each(|(k, v)| { qs.push('&'); qs.push_str(k); qs.push('='); qs.push_str(&encode(v)); }); Request::builder() .uri(format!("/xrpc/{endpoint}?{qs}")) .body(Body::empty()) .unwrap()}
fn encode(s: &str) -> String { byte_serialize(s.as_bytes()).collect()}
async fn json_response(resp: axum::response::Response) -> (StatusCode, Value) { let status = resp.status(); let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap(); let parsed: Value = serde_json::from_slice(&bytes).expect("JSON body"); (status, parsed)}
fn issue_body(repo_did: &Did<DefaultStr>, title: &str) -> Value { json!({ "$type": "sh.tangled.repo.issue", "repo": repo_did.as_ref(), "title": title, "createdAt": "2026-05-01T00:00:00Z" })}
fn pull_body(repo_did: &Did<DefaultStr>, title: &str) -> Value { json!({ "$type": "sh.tangled.repo.pull", "title": title, "createdAt": "2026-05-01T00:00:00Z", "rounds": [], "target": { "repo": repo_did.as_ref(), "branch": "main" } })}
fn star_body(subject_did: &Did<DefaultStr>) -> Value { json!({ "$type": "sh.tangled.feed.star", "createdAt": "2026-05-01T00:00:00Z", "subject": { "$type": "sh.tangled.feed.star#repo", "did": subject_did.as_ref() } })}
fn follow_body(subject_did: &Did<DefaultStr>) -> Value { json!({ "$type": "sh.tangled.graph.follow", "createdAt": "2026-05-01T00:00:00Z", "subject": subject_did.as_ref() })}
#[tokio::test]async fn list_issues_with_no_edges_returns_empty_items() { let h = Harness::new().await; let app = router(h.state.clone()); let resp = app .oneshot(list_request( "sh.tangled.repo.listIssues", "at://did:plc:abalone", &[], )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::OK); assert_eq!(body["items"], json!([])); assert!(body["cursor"].is_null());}
#[tokio::test]async fn count_issues_with_no_edges_returns_zero() { let h = Harness::new().await; let app = router(h.state.clone()); let resp = app .oneshot(list_request( "sh.tangled.repo.countIssues", "at://did:plc:abalone", &[], )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::OK); assert_eq!(body["count"], json!(0)); assert_eq!(body["distinctAuthors"], json!(0));}
#[tokio::test]async fn list_issues_hydrates_via_slingshot_when_edges_present() { let h = Harness::new().await; let repo = did("did:plc:abalone"); let subject = at(&format!("at://{}", repo.as_ref())); let owners = [ ("did:plc:nel", "i1", "first"), ("did:plc:olaren", "i2", "second"), ]; stream::iter(owners) .for_each(|(d, r, title)| { let h = &h; let subject = subject.clone(); let repo = repo.clone(); async move { let d_did = did(d); let rk = rkey(r); h.add_edge( &nsid("sh.tangled.repo.issue"), &subject, &at(&format!( "at://{}/sh.tangled.repo.issue/{}", d_did.as_ref(), rk.as_ref() )), ); h.mount( &d_did, &nsid("sh.tangled.repo.issue"), &rk, issue_body(&repo, title), ) .await; } }) .await;
let app = router(h.state.clone()); let resp = app .oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[], )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().expect("items array"); assert_eq!(items.len(), 2); let titles: Vec<&str> = items .iter() .map(|v| v["value"]["title"].as_str().unwrap()) .collect(); assert!(titles.contains(&"first")); assert!(titles.contains(&"second")); assert_eq!(items[0]["cid"], CID); assert!(items[0]["uri"].as_str().unwrap().starts_with("at://"));}
#[tokio::test]async fn count_distinct_authors_dedupes_per_author() { let h = Harness::new().await; let subject = at("at://did:plc:abalone"); h.add_edge( &nsid("sh.tangled.feed.star"), &subject, &at("at://did:plc:nel/sh.tangled.feed.star/s1"), ); h.add_edge( &nsid("sh.tangled.feed.star"), &subject, &at("at://did:plc:nel/sh.tangled.feed.star/s2"), ); h.add_edge( &nsid("sh.tangled.feed.star"), &subject, &at("at://did:plc:olaren/sh.tangled.feed.star/s3"), );
let app = router(h.state.clone()); let resp = app .oneshot(list_request( "sh.tangled.feed.countStars", subject.as_ref(), &[], )) .await .unwrap(); let (_, body) = json_response(resp).await; assert_eq!(body["count"], json!(3)); assert_eq!(body["distinctAuthors"], json!(2));}
#[tokio::test]async fn list_items_stable_across_coverage_promotion() { let h = Harness::new().await; let subject = at("at://did:plc:abalone"); let nel = did("did:plc:nel"); h.add_edge( &nsid("sh.tangled.feed.star"), &subject, &at(&format!("at://{}/sh.tangled.feed.star/s1", nel.as_ref())), ); h.mount( &nel, &nsid("sh.tangled.feed.star"), &rkey("s1"), star_body(&did("did:plc:abalone")), ) .await;
let app = router(h.state.clone()); h.warming(1, 5); let (_, before) = json_response( app.clone() .oneshot(list_request( "sh.tangled.feed.listStars", subject.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!(before["items"].as_array().unwrap().len(), 1);
h.promote_ready(2, 9); let (_, after) = json_response( app.oneshot(list_request( "sh.tangled.feed.listStars", subject.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!( after["items"].as_array().unwrap().len(), before["items"].as_array().unwrap().len(), ); assert_eq!(after["items"], before["items"]);}
#[tokio::test]async fn list_paginates_via_cursor() { let h = Harness::new().await; let subject = at("at://did:plc:abalone"); let repo = did("did:plc:abalone"); let owners = [ ("did:plc:nel", "i1"), ("did:plc:olaren", "i2"), ("did:plc:teq", "i3"), ("did:plc:lyna", "i4"), ("did:plc:bailey", "i5"), ]; stream::iter(owners) .for_each(|(d, r)| { let h = &h; let subject = subject.clone(); let repo = repo.clone(); async move { let d_did = did(d); let rk = rkey(r); h.add_edge( &nsid("sh.tangled.repo.issue"), &subject, &at(&format!( "at://{}/sh.tangled.repo.issue/{}", d_did.as_ref(), rk.as_ref() )), ); h.mount( &d_did, &nsid("sh.tangled.repo.issue"), &rk, issue_body(&repo, &format!("issue-{}", rk.as_ref())), ) .await; } }) .await;
let app = router(h.state.clone()); let (_, page1) = json_response( app.clone() .oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[("limit", "2")], )) .await .unwrap(), ) .await; let page1_items = page1["items"].as_array().unwrap().clone(); assert_eq!(page1_items.len(), 2); let cursor = page1["cursor"] .as_str() .expect("first page must yield a cursor") .to_owned(); assert!( PageToken::decode_token(&cursor).is_ok(), "cursor must be a TID-shaped token" );
let (_, page2) = json_response( app.oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[("limit", "10"), ("cursor", &cursor)], )) .await .unwrap(), ) .await; let page2_items = page2["items"].as_array().unwrap().clone(); assert_eq!(page2_items.len(), 3); assert!(page2["cursor"].is_null(), "tail page must not promise more");
let union: Vec<&str> = page1_items .iter() .chain(page2_items.iter()) .map(|item| item["uri"].as_str().unwrap()) .collect(); assert_eq!(union.len(), owners.len(), "union covers every owner"); let mut sorted = union.clone(); sorted.sort(); sorted.dedup(); assert_eq!(sorted.len(), owners.len(), "no duplicates across pages");}
#[tokio::test]async fn pagination_unaffected_by_coverage_promotion() { let h = Harness::new().await; let subject = at("at://did:plc:abalone"); let repo = did("did:plc:abalone"); let owners = [("did:plc:nel", "i1"), ("did:plc:olaren", "i2")]; stream::iter(owners) .for_each(|(d, r)| { let h = &h; let subject = subject.clone(); let repo = repo.clone(); async move { let d_did = did(d); let rk = rkey(r); h.add_edge( &nsid("sh.tangled.repo.issue"), &subject, &at(&format!( "at://{}/sh.tangled.repo.issue/{}", d_did.as_ref(), rk.as_ref() )), ); h.mount( &d_did, &nsid("sh.tangled.repo.issue"), &rk, issue_body(&repo, &format!("issue-{}", rk.as_ref())), ) .await; } }) .await;
h.warming(1, 5); let app = router(h.state.clone()); let (_, page1) = json_response( app.clone() .oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[("limit", "1")], )) .await .unwrap(), ) .await; let cursor = page1["cursor"].as_str().unwrap().to_owned();
h.promote_ready(2, 9); let (_, page2) = json_response( app.oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[("limit", "10"), ("cursor", &cursor)], )) .await .unwrap(), ) .await; assert_eq!(page2["items"].as_array().unwrap().len(), 1);}
#[tokio::test]async fn invalid_cursor_returns_400() { let h = Harness::new().await; let app = router(h.state.clone()); let resp = app .oneshot(list_request( "sh.tangled.repo.listIssues", "at://did:plc:abalone", &[("cursor", "not-a-number")], )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_REQUEST); assert_eq!(body["error"], "InvalidRequest");}
#[tokio::test]async fn list_follows_subject_is_followee_did() { let h = Harness::new().await; let followee = did("did:plc:bailey"); let subject = at(&format!("at://{}", followee.as_ref())); h.add_edge( &nsid("sh.tangled.graph.follow"), &subject, &at("at://did:plc:nel/sh.tangled.graph.follow/f1"), ); h.mount( &did("did:plc:nel"), &nsid("sh.tangled.graph.follow"), &rkey("f1"), follow_body(&followee), ) .await;
let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.graph.listFollows", subject.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().unwrap(); assert_eq!(items.len(), 1); assert_eq!(items[0]["value"]["subject"], followee.as_ref());}
#[tokio::test]async fn upstream_failure_during_hydration_drops_only_that_item() { let h = Harness::new().await; let subject = at("at://did:plc:squid"); let kind = nsid("sh.tangled.repo.issue"); let repo = did("did:plc:squid"); h.add_edge( &kind, &subject, &at("at://did:plc:nel/sh.tangled.repo.issue/ok"), ); h.add_edge( &kind, &subject, &at("at://did:plc:teq/sh.tangled.repo.issue/flaky"), ); h.mount( &did("did:plc:nel"), &kind, &rkey("ok"), issue_body(&repo, "kelp survey"), ) .await; Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .and(query_param("repo", "did:plc:teq")) .and(query_param("collection", "sh.tangled.repo.issue")) .and(query_param("rkey", "flaky")) .respond_with(ResponseTemplate::new(503)) .mount(&h.server) .await;
let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().expect("items array"); assert_eq!(items.len(), 1, "flaky item dropped, healthy sibling kept"); assert_eq!( items[0]["uri"].as_str().unwrap(), "at://did:plc:nel/sh.tangled.repo.issue/ok", );}
#[tokio::test]async fn transient_failure_keeps_edge_so_count_stays_whole() { let h = Harness::new().await; let subject = at("at://did:plc:squid"); let kind = nsid("sh.tangled.repo.issue"); let repo = did("did:plc:squid"); h.add_edge( &kind, &subject, &at("at://did:plc:nel/sh.tangled.repo.issue/ok"), ); h.add_edge( &kind, &subject, &at("at://did:plc:teq/sh.tangled.repo.issue/flaky"), ); h.mount( &did("did:plc:nel"), &kind, &rkey("ok"), issue_body(&repo, "kelp survey"), ) .await; Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .and(query_param("repo", "did:plc:teq")) .and(query_param("collection", "sh.tangled.repo.issue")) .and(query_param("rkey", "flaky")) .respond_with(ResponseTemplate::new(503)) .mount(&h.server) .await;
let app = router(h.state.clone()); let (status, body) = json_response( app.clone() .oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); assert_eq!(body["items"].as_array().unwrap().len(), 1);
let (cstatus, cbody) = json_response( app.oneshot(list_request( "sh.tangled.repo.countIssues", subject.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!(cstatus, StatusCode::OK); assert_eq!( cbody["count"], json!(2), "a transient 503 must not evict the edge, count stays whole", );}
#[tokio::test]async fn gone_item_is_evicted_so_count_converges_to_list() { let h = Harness::new().await; let subject = at("at://did:plc:squid"); let kind = nsid("sh.tangled.repo.issue"); let repo = did("did:plc:squid"); h.add_edge( &kind, &subject, &at("at://did:plc:nel/sh.tangled.repo.issue/ok"), ); h.add_edge( &kind, &subject, &at("at://did:plc:teq/sh.tangled.repo.issue/gone"), ); h.mount( &did("did:plc:nel"), &kind, &rkey("ok"), issue_body(&repo, "kelp survey"), ) .await; Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .and(query_param("repo", "did:plc:teq")) .and(query_param("collection", "sh.tangled.repo.issue")) .and(query_param("rkey", "gone")) .respond_with(ResponseTemplate::new(404)) .mount(&h.server) .await;
let app = router(h.state.clone()); let (status, body) = json_response( app.clone() .oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); assert_eq!( body["items"].as_array().unwrap().len(), 1, "gone item dropped from the page" );
let (cstatus, cbody) = json_response( app.oneshot(list_request( "sh.tangled.repo.countIssues", subject.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!(cstatus, StatusCode::OK); assert_eq!( cbody["count"], json!(1), "a definitive 404 must evict the dead edge so count matches the list", );}
#[tokio::test]async fn handle_authority_subject_is_400() { let h = Harness::new().await; let app = router(h.state.clone()); let cases = [ "sh.tangled.feed.listStars", "sh.tangled.feed.countStars", "sh.tangled.graph.listFollows", "sh.tangled.graph.countFollows", "sh.tangled.repo.listIssues", "sh.tangled.repo.countIssues", "sh.tangled.repo.listPulls", "sh.tangled.repo.countPulls", "sh.tangled.feed.listComments", "sh.tangled.feed.countComments", ]; stream::iter(cases) .for_each(|endpoint| { let app = app.clone(); async move { let resp = app .oneshot(list_request(endpoint, "at://oyster.cafe", &[])) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint}"); assert_eq!(body["error"], "InvalidRequest", "{endpoint}"); assert!( body["message"] .as_str() .unwrap_or_default() .contains("did, not a handle"), "{endpoint}: {}", body["message"] ); } }) .await;}
#[tokio::test]async fn empty_subject_is_400() { let h = Harness::new().await; let app = router(h.state.clone()); let resp = app .oneshot(list_request("sh.tangled.repo.listIssues", "", &[])) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_REQUEST); assert_eq!(body["error"], "InvalidRequest");}
#[tokio::test]async fn limit_below_min_or_above_max_is_400() { let h = Harness::new().await; let app = router(h.state.clone()); let cases = [("0", "below"), ("1001", "above")]; stream::iter(cases) .for_each(|(limit, label)| { let app = app.clone(); async move { let resp = app .oneshot(list_request( "sh.tangled.repo.listIssues", "at://did:plc:abalone", &[("limit", limit)], )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_REQUEST, "limit {label}"); assert_eq!(body["error"], "InvalidRequest", "limit {label}"); } }) .await;}
#[tokio::test]async fn count_after_remove_source_returns_zero() { let h = Harness::new().await; let subject = at("at://did:plc:abalone"); let source = at("at://did:plc:nel/sh.tangled.feed.star/s1"); h.add_edge(&nsid("sh.tangled.feed.star"), &subject, &source); h.edges.remove_source(&source);
let app = router(h.state.clone()); let (_, body) = json_response( app.oneshot(list_request( "sh.tangled.feed.countStars", subject.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!(body["count"], json!(0)); assert_eq!(body["distinctAuthors"], json!(0));}
#[tokio::test]async fn list_feed_comments_hydrates_end_to_end() { let h = Harness::new().await; let issue_uri = at("at://did:plc:abalone/sh.tangled.repo.issue/i1"); let nel = did("did:plc:nel"); let rk = rkey("c1"); h.add_edge( &nsid("sh.tangled.feed.comment"), &issue_uri, &at(&format!( "at://{}/sh.tangled.feed.comment/{}", nel.as_ref(), rk.as_ref() )), ); h.mount( &nel, &nsid("sh.tangled.feed.comment"), &rk, json!({ "$type": "sh.tangled.feed.comment", "subject": { "uri": issue_uri.as_ref(), "cid": "bafkqaaa" }, "body": { "$type": "sh.tangled.markup.markdown", "text": "thoughts" }, "createdAt": "2026-05-01T00:00:00Z" }), ) .await;
let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.feed.listComments", issue_uri.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().unwrap(); assert_eq!(items.len(), 1); assert_eq!(items[0]["value"]["body"]["text"], json!("thoughts")); assert_eq!( items[0]["value"]["subject"]["uri"], json!(issue_uri.as_ref()) );}
#[tokio::test]async fn list_item_cid_is_present() { let h = Harness::new().await; let subject = at("at://did:plc:abalone"); let nel = did("did:plc:nel"); h.add_edge( &nsid("sh.tangled.feed.star"), &subject, &at(&format!("at://{}/sh.tangled.feed.star/s1", nel.as_ref())), ); h.mount( &nel, &nsid("sh.tangled.feed.star"), &rkey("s1"), star_body(&did("did:plc:abalone")), ) .await;
let app = router(h.state.clone()); let (_, body) = json_response( app.oneshot(list_request( "sh.tangled.feed.listStars", subject.as_ref(), &[], )) .await .unwrap(), ) .await; let item = &body["items"][0]; assert!( item.as_object().unwrap().contains_key("cid"), "list items must mirror getRecord output shape and include cid" ); assert_eq!(item["cid"], json!(CID));}
#[tokio::test]async fn count_feed_comments_subjects_on_issue_uri() { let h = Harness::new().await; let issue_uri = at("at://did:plc:abalone/sh.tangled.repo.issue/i1"); h.add_edge( &nsid("sh.tangled.feed.comment"), &issue_uri, &at("at://did:plc:nel/sh.tangled.feed.comment/c1"), ); h.add_edge( &nsid("sh.tangled.feed.comment"), &issue_uri, &at("at://did:plc:olaren/sh.tangled.feed.comment/c2"), );
let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.feed.countComments", issue_uri.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); assert_eq!(body["count"], json!(2)); assert_eq!(body["distinctAuthors"], json!(2));}
#[tokio::test]async fn list_item_404_dropped_not_404_for_subject() { let h = Harness::new().await; let subject = at("at://did:plc:squid"); let kind = nsid("sh.tangled.repo.issue"); let repo = did("did:plc:squid"); h.add_edge( &kind, &subject, &at("at://did:plc:nel/sh.tangled.repo.issue/live"), ); h.add_edge( &kind, &subject, &at("at://did:plc:teq/sh.tangled.repo.issue/missing"), ); h.mount( &did("did:plc:nel"), &kind, &rkey("live"), issue_body(&repo, "kelp survives"), ) .await; Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .and(query_param("repo", "did:plc:teq")) .and(query_param("collection", "sh.tangled.repo.issue")) .and(query_param("rkey", "missing")) .respond_with(ResponseTemplate::new(404).set_body_json(json!({ "error": "RecordNotFound", "message": "could not find record" }))) .mount(&h.server) .await;
let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!( status, StatusCode::OK, "a stale-index 404 drops that item, it must not 404 or 502 the subject's list", ); let items = body["items"].as_array().expect("items array"); assert_eq!(items.len(), 1, "stale 404 item dropped, live sibling kept"); assert_eq!( items[0]["uri"].as_str().unwrap(), "at://did:plc:nel/sh.tangled.repo.issue/live", );}
#[tokio::test]async fn list_item_with_wrong_type_tag_dropped() { let h = Harness::new().await; let subject = at("at://did:plc:squid"); let kind = nsid("sh.tangled.feed.star"); h.add_edge( &kind, &subject, &at("at://did:plc:nel/sh.tangled.feed.star/good"), ); h.add_edge( &kind, &subject, &at("at://did:plc:teq/sh.tangled.feed.star/wrong"), ); h.mount( &did("did:plc:nel"), &kind, &rkey("good"), star_body(&did("did:plc:squid")), ) .await; Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) .and(query_param("repo", "did:plc:teq")) .and(query_param("collection", "sh.tangled.feed.star")) .and(query_param("rkey", "wrong")) .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "uri": "at://did:plc:teq/sh.tangled.feed.star/wrong", "cid": CID, "value": { "$type": "sh.tangled.feed.reaction", "createdAt": "2026-05-01T00:00:00Z", "subject": "at://did:plc:squid" } }))) .mount(&h.server) .await;
let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.feed.listStars", subject.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().expect("items array"); assert_eq!(items.len(), 1, "wrong-type item dropped, valid star kept"); assert_eq!( items[0]["uri"].as_str().unwrap(), "at://did:plc:nel/sh.tangled.feed.star/good", );}
#[tokio::test]async fn list_item_with_mismatched_collection_dropped() { let h = Harness::new().await; let subject = at("at://did:plc:squid"); let kind = nsid("sh.tangled.repo.issue"); let repo = did("did:plc:squid"); h.add_edge( &kind, &subject, &at("at://did:plc:nel/sh.tangled.repo.issue/live"), ); h.add_edge( &kind, &subject, &at("at://did:plc:teq/sh.tangled.feed.star/whelk"), ); h.mount( &did("did:plc:nel"), &kind, &rkey("live"), issue_body(&repo, "kelp survives"), ) .await;
let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!( status, StatusCode::OK, "a mismatched-collection index edge must not 400 the subject's list", ); let items = body["items"].as_array().expect("items array"); assert_eq!( items.len(), 1, "mismatched-collection edge dropped, live sibling kept" ); assert_eq!( items[0]["uri"].as_str().unwrap(), "at://did:plc:nel/sh.tangled.repo.issue/live", );}
#[tokio::test]async fn bare_did_endpoints_reject_at_uri_subject() { let h = Harness::new().await; let app = router(h.state.clone()); let cases = [ "sh.tangled.graph.listFollows", "sh.tangled.graph.countFollows", ]; stream::iter(cases) .for_each(|endpoint| { let app = app.clone(); async move { let resp = app .oneshot(list_request( endpoint, "at://did:plc:abalone/sh.tangled.repo/r1", &[], )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint}"); assert_eq!(body["error"], "InvalidRequest", "{endpoint}"); assert!( body["message"] .as_str() .unwrap_or_default() .contains("bare did"), "{endpoint}: {}", body["message"], ); } }) .await;}
#[tokio::test]async fn repo_pointing_endpoints_reject_at_uri_subject() { let h = Harness::new().await; let app = router(h.state.clone()); let cases = [ "sh.tangled.repo.listIssues", "sh.tangled.repo.countIssues", "sh.tangled.repo.listPulls", "sh.tangled.repo.countPulls", "sh.tangled.repo.listArtifacts", "sh.tangled.repo.countArtifacts", ]; stream::iter(cases) .for_each(|endpoint| { let app = app.clone(); async move { let resp = app .oneshot(list_request( endpoint, "at://did:plc:abalone/sh.tangled.repo/r1", &[], )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!( status, StatusCode::BAD_REQUEST, "{endpoint} must reject rkey-form subjects since rkeys are unstable; clients must send the repoDID", ); assert!( body["message"] .as_str() .unwrap_or_default() .contains("bare did"), "{endpoint}: {}", body["message"], ); } }) .await;}
#[tokio::test]async fn repo_pointing_endpoints_accept_bare_did() { let h = Harness::new().await; let app = router(h.state.clone()); let cases = [ "sh.tangled.repo.listIssues", "sh.tangled.repo.countIssues", "sh.tangled.repo.listPulls", "sh.tangled.repo.countPulls", "sh.tangled.repo.listArtifacts", "sh.tangled.repo.countArtifacts", ]; stream::iter(cases) .for_each(|endpoint| { let app = app.clone(); async move { let resp = app .oneshot(list_request(endpoint, "did:plc:abalone", &[])) .await .unwrap(); let (status, _body) = json_response(resp).await; assert_eq!(status, StatusCode::OK, "{endpoint} must accept bare did"); } }) .await;}
#[tokio::test]async fn feed_comment_endpoints_reject_bare_did_or_wrong_collection() { let h = Harness::new().await; let app = router(h.state.clone()); let endpoints = [ "sh.tangled.feed.listComments", "sh.tangled.feed.countComments", ]; let inputs = [ "at://did:plc:abalone", "at://did:plc:abalone/sh.tangled.repo/r1", ]; let cases = endpoints .iter() .copied() .flat_map(|endpoint| inputs.iter().copied().map(move |input| (endpoint, input))); stream::iter(cases) .for_each(|(endpoint, input)| { let app = app.clone(); async move { let resp = app .oneshot(list_request(endpoint, input, &[])) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint} input={input}"); let msg = body["message"].as_str().unwrap_or_default(); assert!( msg.contains("sh.tangled.repo.issue") && msg.contains("sh.tangled.repo.pull") && msg.contains("sh.tangled.string"), "{endpoint} input={input}: {msg}", ); } }) .await;}
#[tokio::test]async fn star_endpoints_reject_unrelated_collection() { let h = Harness::new().await; let app = router(h.state.clone()); let endpoints = ["sh.tangled.feed.listStars", "sh.tangled.feed.countStars"]; stream::iter(endpoints) .for_each(|endpoint| { let app = app.clone(); async move { let resp = app .oneshot(list_request( endpoint, "at://did:plc:abalone/sh.tangled.knot/k1", &[], )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::BAD_REQUEST, "{endpoint}"); let msg = body["message"].as_str().unwrap_or_default(); assert!(msg.contains("sh.tangled.string"), "{endpoint}: {msg}",); } }) .await;}
#[tokio::test]async fn star_endpoints_reject_repo_uri_subject() { let h = Harness::new().await; let app = router(h.state.clone()); let resp = app .oneshot(list_request( "sh.tangled.feed.countStars", "at://did:plc:abalone/sh.tangled.repo/r1", &[], )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!( status, StatusCode::BAD_REQUEST, "rkey-form repo URI must be rejected; clients must send the repoDID directly", ); let msg = body["message"].as_str().unwrap_or_default(); assert!(msg.contains("sh.tangled.string"), "{msg}");}
#[tokio::test]async fn star_endpoints_accept_string_subject_form() { let h = Harness::new().await; let app = router(h.state.clone()); let resp = app .oneshot(list_request( "sh.tangled.feed.countStars", "at://did:plc:abalone/sh.tangled.string/k1", &[], )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::OK); assert_eq!(body["count"], json!(0));}
#[tokio::test]async fn list_after_remove_source_returns_empty_items() { let h = Harness::new().await; let subject = at("at://did:plc:abalone"); let source = at("at://did:plc:nel/sh.tangled.feed.star/s1"); h.add_edge(&nsid("sh.tangled.feed.star"), &subject, &source); h.edges.remove_source(&source);
let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.feed.listStars", subject.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); assert_eq!(body["items"], json!([])); assert!(body["cursor"].is_null());}
#[tokio::test]async fn list_pulls_hydrates_via_slingshot_when_edges_present() { let h = Harness::new().await; let target_did = did("did:plc:abalone"); let subject = at(&format!("at://{}", target_did.as_ref())); let source_did = did("did:plc:nel"); let rk = rkey("p1"); h.add_edge( &nsid("sh.tangled.repo.pull"), &subject, &at(&format!( "at://{}/sh.tangled.repo.pull/{}", source_did.as_ref(), rk.as_ref() )), ); h.mount( &source_did, &nsid("sh.tangled.repo.pull"), &rk, json!({ "$type": "sh.tangled.repo.pull", "title": "ship it", "createdAt": "2026-05-01T00:00:00Z", "rounds": [], "target": {"repo": target_did.as_ref(), "branch": "main"}, }), ) .await; let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.repo.listPulls", subject.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().unwrap(); assert_eq!(items.len(), 1); assert_eq!(items[0]["value"]["title"], json!("ship it")); assert_eq!( items[0]["value"]["target"]["repo"], json!(target_did.as_ref()) );}
#[tokio::test]async fn count_pulls_returns_distinct_authors() { let h = Harness::new().await; let subject = at("at://did:plc:abalone"); h.add_edge( &nsid("sh.tangled.repo.pull"), &subject, &at("at://did:plc:nel/sh.tangled.repo.pull/p1"), ); h.add_edge( &nsid("sh.tangled.repo.pull"), &subject, &at("at://did:plc:olaren/sh.tangled.repo.pull/p2"), ); h.add_edge( &nsid("sh.tangled.repo.pull"), &subject, &at("at://did:plc:nel/sh.tangled.repo.pull/p3"), ); let app = router(h.state.clone()); let (_, body) = json_response( app.oneshot(list_request( "sh.tangled.repo.countPulls", subject.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!(body["count"], json!(3)); assert_eq!(body["distinctAuthors"], json!(2));}
#[tokio::test]async fn extractor_to_xrpc_round_trip_for_star() { let h = Harness::new().await; let subject_did = did("did:plc:abalone"); let source_did = did("did:plc:nel"); let rk = rkey("s1"); let source = at(&format!( "at://{}/sh.tangled.feed.star/{}", source_did.as_ref(), rk.as_ref() )); let body = star_body(&subject_did); let parsed = bobbin_types::edges::Record::from_json_value(&nsid("sh.tangled.feed.star"), body.clone()) .expect("parse star record"); parsed .extract_edges(&source) .expect("extract") .into_iter() .for_each(|e| h.edges.add(e)); h.mount(&source_did, &nsid("sh.tangled.feed.star"), &rk, body) .await;
let app = router(h.state.clone()); let (status, json) = json_response( app.oneshot(list_request( "sh.tangled.feed.listStars", &format!("at://{}", subject_did.as_ref()), &[], )) .await .unwrap(), ) .await; assert_eq!( status, StatusCode::OK, "extractor key must match handler subject, body was {json}", ); let items = json["items"].as_array().unwrap(); assert_eq!(items.len(), 1, "expected exactly one star edge"); assert_eq!( items[0]["value"]["subject"]["did"], json!(subject_did.as_ref()) );}
#[tokio::test]async fn list_issues_includes_state_comment_count_and_state_updated_at() { let h = Harness::new().await; let repo = did("did:plc:limpet"); let subject = at(&format!("at://{}", repo.as_ref())); let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri); h.mount( &did("did:plc:nel"), &nsid("sh.tangled.repo.issue"), &rkey("i1"), issue_body(&repo, "hi"), ) .await; h.add_edge( &nsid("sh.tangled.feed.comment"), &issue_uri, &at("at://did:plc:olaren/sh.tangled.feed.comment/c1"), ); h.add_edge( &nsid("sh.tangled.feed.comment"), &issue_uri, &at("at://did:plc:teq/sh.tangled.feed.comment/c2"), );
h.state.issue_states.upsert( at("at://did:plc:nel/sh.tangled.repo.issue.state/s1"), issue_uri.clone(), 1_777_593_600_000_000, IssueStateKind::Open, ); h.state.issue_states.upsert( at("at://did:plc:nel/sh.tangled.repo.issue.state/s2"), issue_uri.clone(), 1_777_593_700_000_000, IssueStateKind::Closed, );
let app = router(h.state.clone()); let resp = app .oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[], )) .await .unwrap(); let (status, body) = json_response(resp).await; assert_eq!(status, StatusCode::OK); let item = &body["items"][0]; assert_eq!(item["state"], json!("closed")); assert_eq!(item["commentCount"], json!(2)); let updated = item["stateUpdatedAt"] .as_str() .expect("stateUpdatedAt must serialize as RFC3339 string"); assert!( updated.starts_with("2026-"), "expected 2026 timestamp, got {updated}" );}
#[tokio::test]async fn list_issues_defaults_to_open_when_no_state_record() { let h = Harness::new().await; let repo = did("did:plc:limpet"); let subject = at(&format!("at://{}", repo.as_ref())); let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri); h.mount( &did("did:plc:nel"), &nsid("sh.tangled.repo.issue"), &rkey("i1"), issue_body(&repo, "no state yet"), ) .await;
let app = router(h.state.clone()); let (_status, body) = json_response( app.oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[], )) .await .unwrap(), ) .await; let item = &body["items"][0]; assert_eq!( item["state"], json!("open"), "absent state record defaults to open" ); assert!( item.get("stateUpdatedAt").is_none(), "stateUpdatedAt must be absent without a state record", ); assert_eq!(item["commentCount"], json!(0));}
#[tokio::test]async fn list_issues_author_filter_restricts_to_matching_did() { let h = Harness::new().await; let repo = did("did:plc:limpet"); let subject = at(&format!("at://{}", repo.as_ref())); let owners = [ ("did:plc:nel", "n1"), ("did:plc:nel", "n2"), ("did:plc:olaren", "o1"), ("did:plc:olaren", "o2"), ]; stream::iter(owners) .for_each(|(d, r)| { let h = &h; let subject = subject.clone(); let repo = repo.clone(); async move { let d_did = did(d); let rk = rkey(r); h.add_edge( &nsid("sh.tangled.repo.issue"), &subject, &at(&format!( "at://{}/sh.tangled.repo.issue/{}", d_did.as_ref(), rk.as_ref() )), ); h.mount( &d_did, &nsid("sh.tangled.repo.issue"), &rk, issue_body(&repo, &format!("issue-{}", rk.as_ref())), ) .await; } }) .await;
let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[("author", "did:plc:nel")], )) .await .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().expect("items array"); assert_eq!(items.len(), 2, "two issues authored by nel"); let all_nel = items .iter() .all(|i| i["uri"].as_str().unwrap().starts_with("at://did:plc:nel/")); assert!(all_nel, "every returned uri must be authored by nel");}
#[tokio::test]async fn list_issues_invalid_author_returns_400() { let h = Harness::new().await; let subject = "at://did:plc:limpet".to_owned(); let app = router(h.state.clone()); let resp = app .oneshot(list_request( "sh.tangled.repo.listIssues", &subject, &[("author", "not-a-did")], )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::BAD_REQUEST);}
#[tokio::test]async fn list_pulls_includes_merged_state_and_comment_count() { let h = Harness::new().await; let repo = did("did:plc:limpet"); let subject = at(&format!("at://{}", repo.as_ref())); let pull_uri = at("at://did:plc:nel/sh.tangled.repo.pull/p1"); h.add_edge(&nsid("sh.tangled.repo.pull"), &subject, &pull_uri); h.mount( &did("did:plc:nel"), &nsid("sh.tangled.repo.pull"), &rkey("p1"), pull_body(&repo, "fix bug"), ) .await; h.add_edge( &nsid("sh.tangled.feed.comment"), &pull_uri, &at("at://did:plc:teq/sh.tangled.feed.comment/c1"), ); h.state.pull_statuses.upsert( at("at://did:plc:nel/sh.tangled.repo.pull.status/s1"), pull_uri.clone(), 1_777_593_600_000_000, PullStatusKind::Open, ); h.state.pull_statuses.upsert( at("at://did:plc:nel/sh.tangled.repo.pull.status/s2"), pull_uri.clone(), 1_777_593_800_000_000, PullStatusKind::Merged, );
let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.repo.listPulls", subject.as_ref(), &[], )) .await .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); let item = &body["items"][0]; assert_eq!(item["state"], json!("merged")); assert_eq!(item["commentCount"], json!(1));}
#[tokio::test]async fn list_issues_state_filter_open_includes_records_without_state() { let h = Harness::new().await; let repo = did("did:plc:limpet"); let subject = at(&format!("at://{}", repo.as_ref())); let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri); h.mount( &did("did:plc:nel"), &nsid("sh.tangled.repo.issue"), &rkey("i1"), issue_body(&repo, "fresh"), ) .await;
let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[("state", "open")], )) .await .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().expect("items array"); assert_eq!( items.len(), 1, "absent state record still matches state=open" );}
#[tokio::test]async fn list_issues_state_filter_ignores_third_party_state_source() { let h = Harness::new().await; let repo = did("did:plc:limpet"); let subject = at(&format!("at://{}", repo.as_ref())); let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri); h.mount( &did("did:plc:nel"), &nsid("sh.tangled.repo.issue"), &rkey("i1"), issue_body(&repo, "open issue"), ) .await; h.state.issue_states.upsert( at("at://did:plc:nautilus/sh.tangled.repo.issue.state/spoof"), issue_uri.clone(), 1_777_593_800_000_000, IssueStateKind::Closed, );
let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[("state", "open")], )) .await .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().expect("items array"); assert_eq!( items.len(), 1, "third-party Closed record must not flip filter result for state=open", ); assert_eq!(items[0]["state"], json!("open")); assert!( items[0].get("stateUpdatedAt").is_none(), "third-party state source must not surface stateUpdatedAt", );}
#[tokio::test]async fn list_pulls_status_filter_ignores_third_party_status_source() { let h = Harness::new().await; let repo = did("did:plc:limpet"); let subject = at(&format!("at://{}", repo.as_ref())); let pull_uri = at("at://did:plc:nel/sh.tangled.repo.pull/p1"); h.add_edge(&nsid("sh.tangled.repo.pull"), &subject, &pull_uri); h.mount( &did("did:plc:nel"), &nsid("sh.tangled.repo.pull"), &rkey("p1"), pull_body(&repo, "wip"), ) .await; h.state.pull_statuses.upsert( at("at://did:plc:nautilus/sh.tangled.repo.pull.status/spoof"), pull_uri.clone(), 1_777_593_800_000_000, PullStatusKind::Merged, );
let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.repo.listPulls", subject.as_ref(), &[("status", "merged")], )) .await .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().expect("items array"); assert_eq!( items.len(), 0, "third-party Merged record must not satisfy status=merged" );}
#[tokio::test]async fn list_issues_state_filter_accepts_repo_owner_state_source() { let h = Harness::new().await; let repo_owner = did("did:plc:limpet"); let subject = at(&format!("at://{}", repo_owner.as_ref())); let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri); h.mount( &did("did:plc:nel"), &nsid("sh.tangled.repo.issue"), &rkey("i1"), issue_body(&repo_owner, "owner closed"), ) .await; h.state.issue_states.upsert( at("at://did:plc:limpet/sh.tangled.repo.issue.state/legit"), issue_uri.clone(), 1_777_593_800_000_000, IssueStateKind::Closed, );
let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[("state", "closed")], )) .await .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().expect("items array"); assert_eq!( items.len(), 1, "repo-owner state record must satisfy state=closed" ); assert_eq!(items[0]["state"], json!("closed"));}
#[tokio::test]async fn list_issues_order_asc_returns_oldest_first() { let h = Harness::new().await; let repo = did("did:plc:limpet"); let subject = at(&format!("at://{}", repo.as_ref())); let rkeys = ["a", "b", "c"]; stream::iter(rkeys) .for_each(|r| { let h = &h; let subject = subject.clone(); let repo = repo.clone(); async move { let rk = rkey(r); let issue_uri = at(&format!( "at://did:plc:nel/sh.tangled.repo.issue/{}", rk.as_ref() )); h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri); h.mount( &did("did:plc:nel"), &nsid("sh.tangled.repo.issue"), &rk, issue_body(&repo, &format!("issue-{}", rk.as_ref())), ) .await; } }) .await;
let app = router(h.state.clone()); let (_, asc) = json_response( app.clone() .oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[("order", "asc")], )) .await .unwrap(), ) .await; let (_, desc) = json_response( app.oneshot(list_request( "sh.tangled.repo.listIssues", subject.as_ref(), &[("order", "desc")], )) .await .unwrap(), ) .await; let asc_uris: Vec<_> = asc["items"] .as_array() .unwrap() .iter() .map(|i| i["uri"].as_str().unwrap().to_owned()) .collect(); let desc_uris: Vec<_> = desc["items"] .as_array() .unwrap() .iter() .map(|i| i["uri"].as_str().unwrap().to_owned()) .collect(); let mut reversed = asc_uris.clone(); reversed.reverse(); assert_eq!(asc_uris.len(), 3); assert_eq!(desc_uris, reversed, "desc must be exact reverse of asc");}
#[tokio::test]async fn list_issues_by_state_filter_narrows_results() { let h = Harness::new().await; let author = did("did:plc:nel"); let repo = did("did:plc:limpet"); let open_uri = at("at://did:plc:nel/sh.tangled.repo.issue/open1"); let closed_uri = at("at://did:plc:nel/sh.tangled.repo.issue/closed1"); let author_subject = at(&format!("at://{}", author.as_ref())); h.edges.add(Edge { kind: nsid("sh.tangled.repo.issue.by"), subject: SubjectRef::Did(author.clone()), source: open_uri.clone(), sort_micros: next_sort_micros(), }); h.edges.add(Edge { kind: nsid("sh.tangled.repo.issue.by"), subject: SubjectRef::Did(author.clone()), source: closed_uri.clone(), sort_micros: next_sort_micros(), }); h.mount( &author, &nsid("sh.tangled.repo.issue"), &rkey("open1"), issue_body(&repo, "still open"), ) .await; h.mount( &author, &nsid("sh.tangled.repo.issue"), &rkey("closed1"), issue_body(&repo, "shut"), ) .await; h.state.issue_states.upsert( at("at://did:plc:nel/sh.tangled.repo.issue.state/s1"), closed_uri.clone(), 1_777_593_800_000_000, IssueStateKind::Closed, );
let app = router(h.state.clone()); let (status, body) = json_response( app.oneshot(list_request( "sh.tangled.repo.listIssuesBy", author_subject.as_ref(), &[("state", "closed")], )) .await .unwrap(), ) .await; assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().expect("items array"); assert_eq!( items.len(), 1, "only the closed issue survives state=closed" ); assert_eq!(items[0]["uri"], json!(closed_uri.as_ref()));}
#[tokio::test]async fn knot_owned_member_is_synthesized_without_slingshot() { let harness = Harness::new().await; let knot = bobbin_types::knot_acl::host_to_knot_did("kt.oyster.cafe").unwrap(); let subject = did("did:plc:boltless"); let created = chrono::DateTime::parse_from_rfc3339("2026-06-01T00:00:00Z").unwrap(); let micros = created.timestamp_micros() as u64; let (source, edges) = bobbin_types::knot_acl::member_upsert(&knot, &subject, micros).unwrap(); harness.edges.upsert_source(&source, edges); harness.promote_ready(1, 1);
let (status, body) = json_response( router(harness.state.clone()) .oneshot(list_request( "sh.tangled.knot.listMembers", subject.as_ref(), &[], )) .await .unwrap(), ) .await;
assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().expect("items array"); assert_eq!( items.len(), 1, "synthesized member must hydrate with no slingshot mock mounted" ); assert_eq!(items[0]["uri"], json!(source.as_ref())); assert!(items[0]["cid"].is_null()); assert_eq!(items[0]["value"]["domain"], json!("kt.oyster.cafe")); assert_eq!(items[0]["value"]["subject"], json!("did:plc:boltless")); let got = chrono::DateTime::parse_from_rfc3339( items[0]["value"]["createdAt"] .as_str() .expect("createdAt string"), ) .unwrap(); assert_eq!(got.timestamp_micros(), micros as i64);}
#[tokio::test]async fn knot_owned_member_lists_by_knot_did() { let harness = Harness::new().await; let knot = bobbin_types::knot_acl::host_to_knot_did("kt.oyster.cafe").unwrap(); let subject = did("did:plc:boltless"); let created = chrono::DateTime::parse_from_rfc3339("2026-06-01T00:00:00Z").unwrap(); let micros = created.timestamp_micros() as u64; let (source, edges) = bobbin_types::knot_acl::member_upsert(&knot, &subject, micros).unwrap(); harness.edges.upsert_source(&source, edges); harness.promote_ready(1, 1);
let (status, body) = json_response( router(harness.state.clone()) .oneshot(list_request( "sh.tangled.knot.listMembersBy", knot.as_ref(), &[], )) .await .unwrap(), ) .await;
assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().expect("items array"); assert_eq!(items.len(), 1); assert_eq!(items[0]["uri"], json!(source.as_ref())); assert!(items[0]["cid"].is_null()); assert_eq!(items[0]["value"]["domain"], json!("kt.oyster.cafe")); assert_eq!(items[0]["value"]["subject"], json!("did:plc:boltless"));}
#[tokio::test]async fn knot_owned_collaborator_is_synthesized_without_slingshot() { let harness = Harness::new().await; let repo = did("did:plc:scallop"); let subject = did("did:plc:olaren"); let created = chrono::DateTime::parse_from_rfc3339("2026-06-03T12:00:00Z").unwrap(); let micros = created.timestamp_micros() as u64; let (source, edges) = bobbin_types::knot_acl::collaborator_upsert(&repo, &subject, micros).unwrap(); harness.edges.upsert_source(&source, edges); harness.promote_ready(1, 1);
let (status, body) = json_response( router(harness.state.clone()) .oneshot(list_request( "sh.tangled.repo.listCollaborators", repo.as_ref(), &[], )) .await .unwrap(), ) .await;
assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().expect("items array"); assert_eq!(items.len(), 1); assert_eq!(items[0]["uri"], json!(source.as_ref())); assert!(items[0]["cid"].is_null()); assert_eq!(items[0]["value"]["repo"], json!("did:plc:scallop")); assert_eq!(items[0]["value"]["subject"], json!("did:plc:olaren"));}
#[tokio::test]async fn knot_owned_collaborator_lists_by_subject_did() { let harness = Harness::new().await; let repo = did("did:plc:scallop"); let subject = did("did:plc:olaren"); let created = chrono::DateTime::parse_from_rfc3339("2026-06-03T12:00:00Z").unwrap(); let micros = created.timestamp_micros() as u64; let (source, edges) = bobbin_types::knot_acl::collaborator_upsert(&repo, &subject, micros).unwrap(); harness.edges.upsert_source(&source, edges); harness.promote_ready(1, 1);
let (status, body) = json_response( router(harness.state.clone()) .oneshot(list_request( "sh.tangled.repo.listCollaboratorsBy", subject.as_ref(), &[], )) .await .unwrap(), ) .await;
assert_eq!(status, StatusCode::OK); let items = body["items"].as_array().expect("items array"); assert_eq!(items.len(), 1); assert_eq!(items[0]["uri"], json!(source.as_ref())); assert!(items[0]["cid"].is_null()); assert_eq!(items[0]["value"]["repo"], json!("did:plc:scallop")); assert_eq!(items[0]["value"]["subject"], json!("did:plc:olaren"));}