tangled.org / core
Monorepo for Tangled tangled.org
2
Fork 30

Configure Feed

Select the types of activity you want to include in your feed.

bobbin: add sh.tangled.repo.countForks

Signed-off-by: dawn <dawn@tangled.org>

author
dawn date (Jul 25, 2026, 9:13 PM +0300) commit 1a86edf3 parent 69a0a7d2 change-id xzlyyplk
+354 -2
+128
bobbin/crates/ingest/src/lib.rs
··· 3039 3039 } 3040 3040 3041 3041 #[tokio::test] 3042 + async fn fork_record_edges_against_its_source() { 3043 + let (store, issue_states, pull_statuses, cov, resolver) = fresh(); 3044 + let fork: HydrantFrame = parse_frame(json!({ 3045 + "id": 1, 3046 + "type": "record", 3047 + "record": { 3048 + "live": false, 3049 + "did": "did:plc:olaren", 3050 + "rev": fresh_tid().as_str(), 3051 + "collection": "sh.tangled.repo", 3052 + "rkey": "abcabcabcabcz", 3053 + "action": "create", 3054 + "record": { 3055 + "$type": "sh.tangled.repo", 3056 + "createdAt": "2026-05-01T00:00:00Z", 3057 + "knot": "oyster.cafe", 3058 + "name": "abalone", 3059 + "source": "did:plc:abalone" 3060 + } 3061 + } 3062 + })); 3063 + handle_frame( 3064 + fork, 3065 + &store, 3066 + &issue_states, 3067 + &pull_statuses, 3068 + &cov, 3069 + &NoopSearchSink, 3070 + &NoopRecordStore, 3071 + &resolver, 3072 + &sys_clock(), 3073 + now(), 3074 + ) 3075 + .await; 3076 + let key = bobbin_types::ids::EdgeKey::new( 3077 + Nsid::new_static(bobbin_types::edges::REPO_SOURCE_EDGE_KIND).unwrap(), 3078 + did_subj("did:plc:abalone"), 3079 + ); 3080 + assert_eq!(store.count(&key), 1); 3081 + } 3082 + 3083 + fn fork_frame(source: &str) -> HydrantFrame { 3084 + parse_frame(json!({ 3085 + "id": 1, 3086 + "type": "record", 3087 + "record": { 3088 + "live": false, 3089 + "did": "did:plc:olaren", 3090 + "rev": fresh_tid().as_str(), 3091 + "collection": "sh.tangled.repo", 3092 + "rkey": "abcabcabcabcz", 3093 + "action": "create", 3094 + "record": { 3095 + "$type": "sh.tangled.repo", 3096 + "createdAt": "2026-05-01T00:00:00Z", 3097 + "knot": "oyster.cafe", 3098 + "name": "abalone", 3099 + "source": source 3100 + } 3101 + } 3102 + })) 3103 + } 3104 + 3105 + fn fork_edge_key(subject: SubjectRef) -> bobbin_types::ids::EdgeKey { 3106 + bobbin_types::ids::EdgeKey::new( 3107 + Nsid::new_static(bobbin_types::edges::REPO_SOURCE_EDGE_KIND).unwrap(), 3108 + subject, 3109 + ) 3110 + } 3111 + 3112 + #[tokio::test] 3113 + async fn fork_by_at_uri_edges_against_the_source_did() { 3114 + let (store, issue_states, pull_statuses, cov, resolver) = fresh(); 3115 + resolver 3116 + .observe( 3117 + Did::new_owned("did:plc:nel").unwrap(), 3118 + Rkey::new_owned("core").unwrap(), 3119 + Some(Did::new_owned("did:plc:abalone").unwrap()), 3120 + ) 3121 + .await; 3122 + handle_frame( 3123 + fork_frame("at://did:plc:nel/sh.tangled.repo/core"), 3124 + &store, 3125 + &issue_states, 3126 + &pull_statuses, 3127 + &cov, 3128 + &NoopSearchSink, 3129 + &NoopRecordStore, 3130 + &resolver, 3131 + &sys_clock(), 3132 + now(), 3133 + ) 3134 + .await; 3135 + assert_eq!(store.count(&fork_edge_key(did_subj("did:plc:abalone"))), 1); 3136 + assert_eq!( 3137 + store.count(&fork_edge_key(uri_subj( 3138 + "at://did:plc:nel/sh.tangled.repo/core" 3139 + ))), 3140 + 0, 3141 + "the rkey form would never match a bare-DID query", 3142 + ); 3143 + } 3144 + 3145 + #[tokio::test] 3146 + async fn fork_of_a_repo_without_a_did_has_nothing_to_count_against() { 3147 + let (store, issue_states, pull_statuses, cov, resolver) = fresh(); 3148 + handle_frame( 3149 + fork_frame("at://did:plc:nel/sh.tangled.repo/core"), 3150 + &store, 3151 + &issue_states, 3152 + &pull_statuses, 3153 + &cov, 3154 + &NoopSearchSink, 3155 + &NoopRecordStore, 3156 + &resolver, 3157 + &sys_clock(), 3158 + now(), 3159 + ) 3160 + .await; 3161 + assert_eq!( 3162 + store.count(&fork_edge_key(uri_subj( 3163 + "at://did:plc:nel/sh.tangled.repo/core" 3164 + ))), 3165 + 0, 3166 + ); 3167 + } 3168 + 3169 + #[tokio::test] 3042 3170 async fn repo_record_without_a_name_indexes_its_rkey() { 3043 3171 let (store, issue_states, pull_statuses, cov, resolver) = fresh(); 3044 3172 let repo: HydrantFrame = parse_frame(json!({
+98 -2
bobbin/crates/types/src/edges.rs
··· 3 3 4 4 use jacquard_common::types::did::Did; 5 5 use jacquard_common::types::nsid::Nsid; 6 - use jacquard_common::types::string::{AtStrError, AtUri, Datetime}; 6 + use jacquard_common::types::string::{AtStrError, AtUri, Datetime, UriValue}; 7 7 use jacquard_common::types::tid::Tid; 8 8 use jacquard_common::{BosStr, DefaultStr}; 9 9 ··· 228 228 Self::Knot(_) => Ok(owner_self_edges("sh.tangled.knot", source)), 229 229 Self::LabelDefinition(_) => Ok(owner_self_edges("sh.tangled.label.definition", source)), 230 230 Self::PublicKey(_) => Ok(owner_self_edges("sh.tangled.publicKey", source)), 231 - Self::Repo(_) => Ok(owner_self_edges("sh.tangled.repo", source)), 231 + Self::Repo(r) => Ok(repo_edges(source, r)), 232 232 Self::Spindle(_) => Ok(owner_self_edges("sh.tangled.spindle", source)), 233 233 Self::TangledString(_) => Ok(owner_self_edges("sh.tangled.string", source)), 234 234 Self::Vouch(r) => vouch_edges(source, r), 235 235 Self::Profile(_) => Ok(Vec::new()), 236 236 } 237 + } 238 + } 239 + 240 + pub const REPO_SOURCE_EDGE_KIND: &str = "sh.tangled.repo.source"; 241 + 242 + fn repo_edges(source: &AtUri<DefaultStr>, record: &RepoRecord<DefaultStr>) -> Vec<Edge> { 243 + let mut edges = owner_self_edges("sh.tangled.repo", source); 244 + if let Some(subject) = fork_source_subject(record) { 245 + edges.push(Edge { 246 + kind: nsid_static(REPO_SOURCE_EDGE_KIND), 247 + subject, 248 + source: source.clone(), 249 + sort_micros: 0, 250 + }); 251 + } 252 + edges 253 + } 254 + 255 + fn fork_source_subject(record: &RepoRecord<DefaultStr>) -> Option<SubjectRef> { 256 + match record.source.as_ref()? { 257 + UriValue::Did(did) => Some(SubjectRef::Did(did.clone())), 258 + UriValue::At(uri) => uri_subject_for_record(uri), 259 + _ => None, 237 260 } 238 261 } 239 262 ··· 578 601 fn extract(collection: &'static str, source: &str, body: serde_json::Value) -> Vec<Edge> { 579 602 let parsed = Record::from_json_value(&nsid(collection), body).expect("parse"); 580 603 parsed.primary_edges(&at(source)).expect("extract") 604 + } 605 + 606 + fn repo_body(source: Option<&str>) -> serde_json::Value { 607 + let mut body = json!({ 608 + "$type": "sh.tangled.repo", 609 + "createdAt": "2026-05-01T00:00:00Z", 610 + "knot": "oyster.cafe", 611 + "name": "abalone" 612 + }); 613 + if let Some(source) = source { 614 + body["source"] = json!(source); 615 + } 616 + body 617 + } 618 + 619 + #[test] 620 + fn plain_repo_only_makes_its_owner_edge() { 621 + let edges = extract( 622 + "sh.tangled.repo", 623 + "at://did:plc:nel/sh.tangled.repo/abcabcabcabcz", 624 + repo_body(None), 625 + ); 626 + assert_eq!(edges.len(), 1); 627 + assert_eq!(edges[0].kind, nsid("sh.tangled.repo")); 628 + assert_eq!(edges[0].subject, did_subj("did:plc:nel")); 629 + } 630 + 631 + #[test] 632 + fn fork_of_a_repo_did_keys_on_the_did() { 633 + let edges = extract( 634 + "sh.tangled.repo", 635 + "at://did:plc:nel/sh.tangled.repo/abcabcabcabcz", 636 + repo_body(Some("did:plc:abalone")), 637 + ); 638 + let fork: Vec<&Edge> = edges 639 + .iter() 640 + .filter(|e| e.kind.as_ref() == REPO_SOURCE_EDGE_KIND) 641 + .collect(); 642 + assert_eq!(fork.len(), 1); 643 + assert_eq!(fork[0].subject, did_subj("did:plc:abalone")); 644 + } 645 + 646 + #[test] 647 + fn fork_of_a_repo_without_a_did_keys_on_the_record() { 648 + let edges = extract( 649 + "sh.tangled.repo", 650 + "at://did:plc:nel/sh.tangled.repo/abcabcabcabcz", 651 + repo_body(Some("at://did:plc:olaren/sh.tangled.repo/core")), 652 + ); 653 + let fork: Vec<&Edge> = edges 654 + .iter() 655 + .filter(|e| e.kind.as_ref() == REPO_SOURCE_EDGE_KIND) 656 + .collect(); 657 + assert_eq!(fork.len(), 1); 658 + assert_eq!( 659 + fork[0].subject, 660 + uri_subj("at://did:plc:olaren/sh.tangled.repo/core") 661 + ); 662 + } 663 + 664 + #[test] 665 + fn fork_of_something_off_network_makes_no_fork_edge() { 666 + let edges = extract( 667 + "sh.tangled.repo", 668 + "at://did:plc:nel/sh.tangled.repo/abcabcabcabcz", 669 + repo_body(Some("https://github.com/90-008/awawa")), 670 + ); 671 + assert!( 672 + edges 673 + .iter() 674 + .all(|e| e.kind.as_ref() != REPO_SOURCE_EDGE_KIND), 675 + "a clone url is not a repo we can count against", 676 + ); 581 677 } 582 678 583 679 #[test]
+19
bobbin/crates/xrpc/src/lib.rs
··· 33 33 SearchCursor, SearchError, SearchFilters, SearchHit, SearchOffset, SearchReader, 34 34 }; 35 35 use bobbin_slingshot_client::{SlingshotClient, SlingshotError}; 36 + use bobbin_types::edges::REPO_SOURCE_EDGE_KIND; 36 37 use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static}; 37 38 use bobbin_types::knot_acl::{KnotOwnedSource, decode_knot_owned_source, knot_did_host}; 38 39 use bobbin_types::record::RecordBody; ··· 239 240 ) 240 241 .route("/xrpc/sh.tangled.repo.listRepos", get(list_repos)) 241 242 .route("/xrpc/sh.tangled.repo.countRepos", get(count_repos)) 243 + .route("/xrpc/sh.tangled.repo.countForks", get(count_forks)) 242 244 .route("/xrpc/sh.tangled.knot.listKnots", get(list_knots)) 243 245 .route("/xrpc/sh.tangled.knot.countKnots", get(count_knots)) 244 246 .route("/xrpc/sh.tangled.spindle.listSpindles", get(list_spindles)) ··· 1017 1019 "sh.tangled.spindle.member" => SpindleMemberRecord, SubjectShape::BareDid, mirror SpindleMemberBy; 1018 1020 "sh.tangled.string" => TangledStringRecord, SubjectShape::BareDid; 1019 1021 } 1022 + 1023 + // the fork edge is not a collection so it is not in the table above 1024 + const FORK_SUBJECT_SHAPE: SubjectShape = SubjectShape::BareDid; 1020 1025 1021 1026 fn parse_subject(raw: &SubjectQuery, shape: SubjectShape) -> Result<SubjectRef, XrpcError> { 1022 1027 let uri = match raw { ··· 2103 2108 XrpcQuery(q): XrpcQuery<CountQuery>, 2104 2109 ) -> Result<Json<CountResponse>, XrpcError> { 2105 2110 count_for::<RepoRecord>(&state, q).map(Json) 2111 + } 2112 + 2113 + // forks are repo records pointing back at a repo, so they get their own edge 2114 + // kind instead of a collection 2115 + async fn count_forks( 2116 + State(state): State<AppState>, 2117 + XrpcQuery(q): XrpcQuery<CountQuery>, 2118 + ) -> Result<Json<CountResponse>, XrpcError> { 2119 + let subject = parse_subject(&q.subject, FORK_SUBJECT_SHAPE)?; 2120 + let key = EdgeKey::new(nsid_static(REPO_SOURCE_EDGE_KIND), subject); 2121 + Ok(Json(CountResponse { 2122 + count: state.edges.count(&key), 2123 + distinct_authors: state.edges.count_distinct_authors(&key), 2124 + })) 2106 2125 } 2107 2126 2108 2127 async fn list_knots(
+70
bobbin/crates/xrpc/tests/aggregation.rs
··· 351 351 } 352 352 353 353 #[tokio::test] 354 + async fn count_forks_counts_repos_pointing_at_the_source() { 355 + let h = Harness::new().await; 356 + let subject = at("at://did:plc:abalone"); 357 + h.add_edge( 358 + &nsid("sh.tangled.repo.source"), 359 + &subject, 360 + &at("at://did:plc:nel/sh.tangled.repo/f1"), 361 + ); 362 + h.add_edge( 363 + &nsid("sh.tangled.repo.source"), 364 + &subject, 365 + &at("at://did:plc:olaren/sh.tangled.repo/f2"), 366 + ); 367 + // the source owner's other repos are not forks of it 368 + h.add_edge( 369 + &nsid("sh.tangled.repo"), 370 + &subject, 371 + &at("at://did:plc:nel/sh.tangled.repo/r1"), 372 + ); 373 + 374 + let app = router(h.state.clone()); 375 + let resp = app 376 + .oneshot(list_request( 377 + "sh.tangled.repo.countForks", 378 + subject.as_ref(), 379 + &[], 380 + )) 381 + .await 382 + .unwrap(); 383 + let (status, body) = json_response(resp).await; 384 + assert_eq!(status, StatusCode::OK); 385 + assert_eq!(body["count"], json!(2)); 386 + assert_eq!(body["distinctAuthors"], json!(2)); 387 + } 388 + 389 + #[tokio::test] 390 + async fn count_forks_of_an_unforked_repo_is_zero() { 391 + let h = Harness::new().await; 392 + let app = router(h.state.clone()); 393 + let resp = app 394 + .oneshot(list_request( 395 + "sh.tangled.repo.countForks", 396 + "did:plc:abalone", 397 + &[], 398 + )) 399 + .await 400 + .unwrap(); 401 + let (status, body) = json_response(resp).await; 402 + assert_eq!(status, StatusCode::OK); 403 + assert_eq!(body["count"], json!(0)); 404 + } 405 + 406 + #[tokio::test] 407 + async fn count_forks_rejects_an_at_uri_subject() { 408 + let h = Harness::new().await; 409 + let app = router(h.state.clone()); 410 + let resp = app 411 + .oneshot(list_request( 412 + "sh.tangled.repo.countForks", 413 + "at://did:plc:nel/sh.tangled.repo/core", 414 + &[], 415 + )) 416 + .await 417 + .unwrap(); 418 + let (status, body) = json_response(resp).await; 419 + assert_eq!(status, StatusCode::BAD_REQUEST); 420 + assert_eq!(body["error"], "InvalidRequest"); 421 + } 422 + 423 + #[tokio::test] 354 424 async fn list_items_stable_across_coverage_promotion() { 355 425 let h = Harness::new().await; 356 426 let subject = at("at://did:plc:abalone");
+39
lexicons/repo/countForks.json
··· 1 + { 2 + "lexicon": 1, 3 + "id": "sh.tangled.repo.countForks", 4 + "defs": { 5 + "main": { 6 + "type": "query", 7 + "parameters": { 8 + "type": "params", 9 + "required": ["subject"], 10 + "properties": { 11 + "subject": { 12 + "type": "string", 13 + "format": "did", 14 + "description": "Repo DID to count forks of." 15 + } 16 + } 17 + }, 18 + "output": { 19 + "encoding": "application/json", 20 + "schema": { 21 + "type": "object", 22 + "required": ["count", "distinctAuthors"], 23 + "properties": { 24 + "count": { 25 + "type": "integer", 26 + "minimum": 0, 27 + "description": "Total number of matching records." 28 + }, 29 + "distinctAuthors": { 30 + "type": "integer", 31 + "minimum": 0, 32 + "description": "Number of distinct authors among the matching records." 33 + } 34 + } 35 + } 36 + } 37 + } 38 + } 39 + }