diff --git a/bobbin/crates/edge-index/src/lib.rs b/bobbin/crates/edge-index/src/lib.rs index a5f125086..4f929e84f 100644 --- a/bobbin/crates/edge-index/src/lib.rs +++ b/bobbin/crates/edge-index/src/lib.rs @@ -3222,6 +3222,41 @@ mod tests { assert_eq!(store.count_distinct_authors(&key), 4, "nel fully removed"); } + #[test] + fn upsert_source_moves_a_subscription_to_its_mute_kind_and_back() { + let store = store(); + let source = at("at://did:plc:bob/org.tangled.feed.subscription/r1"); + let subject = did_subj("did:plc:repo"); + let sub_key = EdgeKey::new(nsid("org.tangled.feed.subscription"), subject.clone()); + let mute_key = EdgeKey::new( + nsid("org.tangled.feed.subscription.ignore"), + subject.clone(), + ); + let edge_of = |kind: &'static str| Edge { + kind: nsid(kind), + subject: subject.clone(), + source: source.clone(), + sort_micros: 1, + }; + + store.upsert_source(&source, vec![edge_of("org.tangled.feed.subscription")]); + assert_eq!((store.count(&sub_key), store.count(&mute_key)), (1, 0)); + + store.upsert_source( + &source, + vec![edge_of("org.tangled.feed.subscription.ignore")], + ); + assert_eq!((store.count(&sub_key), store.count(&mute_key)), (0, 1)); + assert_eq!( + store.viewer_source(&mute_key, "did:plc:bob"), + Some(source.clone()) + ); + assert_eq!(store.viewer_source(&sub_key, "did:plc:bob"), None); + + store.upsert_source(&source, vec![edge_of("org.tangled.feed.subscription")]); + assert_eq!((store.count(&sub_key), store.count(&mute_key)), (1, 0)); + } + #[test] fn source_collections_set_and_clear() { let store = store(); diff --git a/bobbin/crates/types/src/edges.rs b/bobbin/crates/types/src/edges.rs index 02c7c157a..188fff60d 100644 --- a/bobbin/crates/types/src/edges.rs +++ b/bobbin/crates/types/src/edges.rs @@ -355,6 +355,8 @@ fn star_edges( Ok(one_edge("org.tangled.feed.star", subject, source)) } +pub const SUBSCRIPTION_IGNORE_KIND: &str = "org.tangled.feed.subscription.ignore"; + fn subscription_edges( source: &AtUri, record: &Subscription, @@ -369,7 +371,13 @@ fn subscription_edges( } SubscriptionSubject::Repo(v) => SubjectRef::Did(v.did.clone()), }; - Ok(one_edge("org.tangled.feed.subscription", subject, source)) + // a mute gets its own kind, so counts never pick it up + let kind = if record.ignore == Some(true) { + SUBSCRIPTION_IGNORE_KIND + } else { + "org.tangled.feed.subscription" + }; + Ok(one_edge(kind, subject, source)) } fn reaction_edges( @@ -602,6 +610,34 @@ mod tests { ); } + #[test] + fn subscription_mute_moves_to_the_ignore_kind() { + let source = "at://did:plc:nel/org.tangled.feed.subscription/3l"; + let body = |ignore: Option| { + let mut body = json!({ + "$type": "org.tangled.feed.subscription", + "subject": { + "$type": "org.tangled.feed.subscription#record", + "uri": "at://did:plc:abalone/org.tangled.track.ticket/3k" + }, + "createdAt": "2026-05-01T00:00:00Z" + }); + if let Some(ignore) = ignore { + body["ignore"] = json!(ignore); + } + body + }; + for (ignore, kind) in [ + (None, "org.tangled.feed.subscription"), + (Some(false), "org.tangled.feed.subscription"), + (Some(true), SUBSCRIPTION_IGNORE_KIND), + ] { + let edges = extract_full("org.tangled.feed.subscription", source, body(ignore)); + let kinds: Vec<&str> = edges.iter().map(|e| e.kind.as_ref()).collect(); + assert_eq!(kinds, [kind], "ignore {ignore:?}"); + } + } + #[test] fn declaration_makes_no_edges() { let edges = extract_full( diff --git a/bobbin/crates/xrpc/src/lib.rs b/bobbin/crates/xrpc/src/lib.rs index 27d7b40cb..5e44e5fbc 100644 --- a/bobbin/crates/xrpc/src/lib.rs +++ b/bobbin/crates/xrpc/src/lib.rs @@ -33,7 +33,7 @@ use bobbin_resolver::{IdentityResolveError, IdentityResolver, RepoIdResolver}; use bobbin_runtime::{ReqwestHttp, UnixMicros}; use bobbin_search::{ActorIndex, ActorIndexError, SearchReader}; use bobbin_slingshot_client::{SlingshotClient, SlingshotError}; -use bobbin_types::edges::REPO_SOURCE_EDGE_KIND; +use bobbin_types::edges::{REPO_SOURCE_EDGE_KIND, SUBSCRIPTION_IGNORE_KIND}; use bobbin_types::edges::{TICKET_ISSUE_KIND, TICKET_PULL_KIND, ticket_editor}; use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static, owner_did_from_aturi}; use bobbin_types::org_tangled::actor::profile::{Profile, ProfileGetRecordOutput, ProfileRecord}; @@ -2216,11 +2216,33 @@ async fn get_star( ) -> Result, XrpcError> { get_for::(&state, q).map(Json) } + +#[derive(Serialize)] +struct SubscriptionForActorResponse { + uri: AtUri, + ignore: bool, +} + async fn get_subscription_for_actor( State(state): State, XrpcQuery(q): XrpcQuery, -) -> Result, XrpcError> { - get_for::(&state, q).map(Json) +) -> Result, XrpcError> { + let subject = parse_subject(&q.subject, SubscriptionRecord::SHAPE)?; + let find = |kind: &'static str| { + let key = EdgeKey::new(nsid_static(kind), subject.clone()); + state.edges.viewer_source(&key, q.actor.as_str()) + }; + // a mute is a subscription record too, so comparing uris compares tid rkeys and the newer one wins + let (uri, ignore) = match ( + find(SubscriptionRecord::NSID), + find(SUBSCRIPTION_IGNORE_KIND), + ) { + (Some(sub), Some(mute)) if mute.as_str() > sub.as_str() => (mute, true), + (Some(sub), _) => (sub, false), + (None, Some(mute)) => (mute, true), + (None, None) => return Err(XrpcError::NotFound), + }; + Ok(Json(SubscriptionForActorResponse { uri, ignore })) } async fn list_follows( @@ -3417,7 +3439,7 @@ async fn list_recipients( )); }; - let key = EdgeKey::new(nsid_static("org.tangled.feed.subscription"), subject_ref); + let key = EdgeKey::new(nsid_static(SubscriptionRecord::NSID), subject_ref); let sources = state.edges.sources_for(&key); // NOTE: collection filtering intentionally disabled. diff --git a/bobbin/crates/xrpc/tests/aggregation.rs b/bobbin/crates/xrpc/tests/aggregation.rs index 9d84a42c3..4705b6a5d 100644 --- a/bobbin/crates/xrpc/tests/aggregation.rs +++ b/bobbin/crates/xrpc/tests/aggregation.rs @@ -2531,6 +2531,123 @@ async fn list_recipients_repo_subject_returns_repo_subscribers() { assert_eq!(body["dids"], json!(["did:plc:watcher"])); } +fn subscription_body(ignore: Option) -> Value { + let mut body = json!({ + "$type": "org.tangled.feed.subscription", + "subject": { + "$type": "org.tangled.feed.subscription#record", + "uri": "at://did:plc:repo/org.tangled.track.ticket/abc" + }, + "createdAt": "2026-06-03T12:00:00Z" + }); + if let Some(ignore) = ignore { + body["ignore"] = json!(ignore); + } + body +} + +fn index_subscription(h: &Harness, source: &AtUri, ignore: Option) { + let edges = bobbin_types::edges::Record::from_json_value( + &nsid("org.tangled.feed.subscription"), + subscription_body(ignore), + ) + .unwrap() + .extract_edges(source) + .unwrap(); + h.edges.upsert_source(source, edges); +} + +async fn recipients_and_viewer_subscription(h: &Harness) -> (Value, StatusCode, Value) { + let app = router(h.state.clone()); + let (_, recipients) = json_response( + app.clone() + .oneshot( + Request::builder() + .uri("/xrpc/org.tangled.temp.notification.listRecipients?subject=at://did:plc:repo/org.tangled.track.ticket/abc") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(), + ) + .await; + let (status, viewer) = json_response( + app.oneshot(list_request( + "org.tangled.feed.subscription.getForActor", + "at://did:plc:repo/org.tangled.track.ticket/abc", + &[("actor", "did:plc:bob")], + )) + .await + .unwrap(), + ) + .await; + (recipients["dids"].clone(), status, viewer) +} + +#[tokio::test] +async fn muted_subscription_is_no_recipient_but_found_for_actor() { + let h = Harness::new().await; + let source = at("at://did:plc:bob/org.tangled.feed.subscription/rkey1"); + + index_subscription(&h, &source, Some(true)); + let (dids, status, viewer) = recipients_and_viewer_subscription(&h).await; + assert_eq!(dids, json!([])); + assert_eq!(status, StatusCode::OK); + assert_eq!(viewer, json!({ "uri": source.as_ref(), "ignore": true })); +} + +#[tokio::test] +async fn flipping_ignore_moves_subscription_between_recipients_and_mutes() { + let h = Harness::new().await; + let source = at("at://did:plc:bob/org.tangled.feed.subscription/rkey1"); + + for (ignore, recipient) in [ + (None, true), + (Some(true), false), + (Some(false), true), + (Some(true), false), + ] { + index_subscription(&h, &source, ignore); + let (dids, status, viewer) = recipients_and_viewer_subscription(&h).await; + assert_eq!( + dids, + if recipient { + json!(["did:plc:bob"]) + } else { + json!([]) + } + ); + assert_eq!(status, StatusCode::OK); + assert_eq!( + viewer, + json!({ "uri": source.as_ref(), "ignore": !recipient }), + "ignore {ignore:?}" + ); + } + + h.edges.remove_source(&source); + let (dids, status, _) = recipients_and_viewer_subscription(&h).await; + assert_eq!(dids, json!([])); + assert_eq!(status, StatusCode::NOT_FOUND); +} + +#[tokio::test] +async fn the_newer_of_a_subscription_and_a_mute_wins() { + let h = Harness::new().await; + let older = at("at://did:plc:bob/org.tangled.feed.subscription/3kaaaaaaaaaa2"); + let newer = at("at://did:plc:bob/org.tangled.feed.subscription/3kaaaaaaaaab2"); + + index_subscription(&h, &older, None); + index_subscription(&h, &newer, Some(true)); + let (_, _, viewer) = recipients_and_viewer_subscription(&h).await; + assert_eq!(viewer, json!({ "uri": newer.as_ref(), "ignore": true })); + + index_subscription(&h, &older, Some(true)); + index_subscription(&h, &newer, None); + let (_, _, viewer) = recipients_and_viewer_subscription(&h).await; + assert_eq!(viewer, json!({ "uri": newer.as_ref(), "ignore": false })); +} + #[tokio::test] async fn list_recipients_empty_for_unknown_subject() { let h = Harness::new().await;