From d07b62a9ca6ef576ffb7064a471c05e3520c334b Mon Sep 17 00:00:00 2001 From: oppiliappan Date: Wed, 7 Oct 2026 18:06:11 +0100 Subject: [PATCH] bobbin: serve !hide/!warn labels through enrich Signed-off-by: oppiliappan --- bobbin/crates/bobbin/src/main.rs | 16 ++++-- bobbin/crates/xrpc/src/enrich.rs | 46 ++++++++++++++- bobbin/crates/xrpc/src/labels.rs | 67 ++++++++++++++++++++++ bobbin/crates/xrpc/src/lib.rs | 9 +++ bobbin/crates/xrpc/tests/enrich.rs | 91 +++++++++++++++++++++++++++++- lexicons/query/enrichResponse.json | 14 ++++- 6 files changed, 233 insertions(+), 10 deletions(-) create mode 100644 bobbin/crates/xrpc/src/labels.rs diff --git a/bobbin/crates/bobbin/src/main.rs b/bobbin/crates/bobbin/src/main.rs index ea3ea1b1e..fe23b9ce0 100644 --- a/bobbin/crates/bobbin/src/main.rs +++ b/bobbin/crates/bobbin/src/main.rs @@ -377,7 +377,7 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { let ingest_coverage = coverage.clone(); let labels = Arc::new(LabelIndex::new(edges.clone(), hasher.clone())); - let _moderation = match cfg.moderation.labeler()? { + let moderation = match cfg.moderation.labeler()? { config::Moderation::Labeler { did: labeler, url: service, @@ -387,9 +387,9 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { %service, "tailing labeler for moderation decisions", ); - Some(tokio::spawn(bobbin_ingest::moderation::run( + tokio::spawn(bobbin_ingest::moderation::run( bobbin_ingest::moderation::ModerationConfig { - labeler, + labeler: labeler.clone(), service, }, bobbin_ingest::moderation::ModerationRuntime { @@ -399,7 +399,12 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { entropy: entropy.clone(), cancel: cancel.clone(), }, - ))) + )); + Some(bobbin_xrpc::Moderation::new( + labels.clone(), + labeler, + clock.clone(), + )) } config::Moderation::Off => None, }; @@ -480,7 +485,8 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { .with_settlements(settlements) .with_invites(invites) .with_claims(claims) - .with_stream_health(stream_health); + .with_stream_health(stream_health) + .with_moderation(moderation); let app = router(state).layer(axum::middleware::from_fn(metrics::xrpc)); let _debug_server = match (debug_bind, mem_probe) { diff --git a/bobbin/crates/xrpc/src/enrich.rs b/bobbin/crates/xrpc/src/enrich.rs index 6d254fda9..38526fc2a 100644 --- a/bobbin/crates/xrpc/src/enrich.rs +++ b/bobbin/crates/xrpc/src/enrich.rs @@ -7,6 +7,7 @@ use axum::{ http::{Request, StatusCode}, response::Response, }; +use bobbin_types::sh_tangled::query::enrich_response::Labels; use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static, owner_did_from_aturi}; use futures::stream::{self, StreamExt as _}; use jacquard_common::DefaultStr; @@ -29,13 +30,15 @@ pub const TYPE_VIEWER: &str = "sh.tangled.query.enrichResponse#viewer"; /// output of upstream blue.microcosm.identity.resolveMiniDoc pub const TYPE_MINIDOC: &str = "blue.microcosm.identity.resolveMiniDoc"; pub const TYPE_REPO_VIEW_BASIC: &str = "sh.tangled.repo.defs#repoViewBasic"; +pub const TYPE_LABELS: &str = "sh.tangled.query.enrichResponse#labels"; -const KNOWN_TYPES: [&str; 5] = [ +const KNOWN_TYPES: [&str; 6] = [ TYPE_COUNT, TYPE_DISTINCT_AUTHORS, TYPE_VIEWER, TYPE_MINIDOC, TYPE_REPO_VIEW_BASIC, + TYPE_LABELS, ]; /// a payload type nsid, with an optional #fragment for lexicon defs. the raw @@ -305,6 +308,27 @@ pub async fn enrich( } } } + TYPE_LABELS => { + let Some(moderation) = state.moderation.as_ref() else { + continue; + }; + let labels = moderation.labels(&subject); + if labels.is_empty() { + continue; + } + let payload = serde_json::to_value(Labels { + labels, + extra_data: None, + }) + .map_err(|e| XrpcError::Internal(format!("labels encode: {e}")))?; + put( + &mut data, + subject.as_str(), + descriptor.source.as_str(), + descriptor.ty.as_str(), + payload, + ); + } TYPE_REPO_VIEW_BASIC => { let sources = view_targets.entry(subject).or_default(); if !sources.contains(&descriptor.source) { @@ -421,6 +445,15 @@ fn descriptor_error(source: &LinkSource, msg: &str) -> XrpcError { } fn validate_source(source: &LinkSource, ty: &PayloadType) -> Result<(), XrpcError> { + if ty.as_str() == TYPE_LABELS { + return match source.path { + SourcePath::Itself | SourcePath::Repo => Ok(()), + SourcePath::Subject => Err(descriptor_error( + source, + "labels are on the record itself or its repo; try collection:. or collection:.repo", + )), + }; + } match source.path { SourcePath::Itself => match view_type(source.collection.as_ref()) { Some(view) if view == ty.as_str() => Ok(()), @@ -476,6 +509,7 @@ fn applicable_subject( SourcePath::Repo => Ok(repo_did(reference).map(SubjectRef::Did)), // repos are mostly referenced by their repo did, not their record uri SourcePath::Itself => Ok(match reference { + SubjectRef::Did(_) if descriptor.ty.as_str() == TYPE_LABELS => None, SubjectRef::Did(_) => Some(reference.clone()), SubjectRef::Uri(uri) => uri .collection() @@ -710,4 +744,14 @@ mod tests { assert!(PayloadType::parse("sh.tangled.query.enrichResponse#").is_err()); assert!(PayloadType::parse("nope#count").is_err()); } + + #[test] + fn labels_read_the_record_or_its_repo() { + let labels = PayloadType::parse(TYPE_LABELS).unwrap(); + for ok in ["org.tangled.actor.profile:.", "org.tangled.feed.comment:.repo"] { + assert!(validate_source(&LinkSource::parse(ok).unwrap(), &labels).is_ok()); + } + let subject = LinkSource::parse("org.tangled.feed.star:subject").unwrap(); + assert!(validate_source(&subject, &labels).is_err()); + } } diff --git a/bobbin/crates/xrpc/src/labels.rs b/bobbin/crates/xrpc/src/labels.rs new file mode 100644 index 000000000..87bf14374 --- /dev/null +++ b/bobbin/crates/xrpc/src/labels.rs @@ -0,0 +1,67 @@ +use std::sync::Arc; + +use bobbin_edge_index::{LabelEntry, LabelIndex, LabelTarget, LabelVal, ModerationLabel}; +use bobbin_runtime::Clock; +use bobbin_types::com_atproto::label::Label; +use bobbin_types::ids::SubjectRef; +use jacquard_common::DefaultStr; +use jacquard_common::types::did::Did; +use jacquard_common::types::uri::UriValue; + +use crate::datetime_of; + +const LABEL_VERSION: i64 = 1; + +#[derive(Clone)] +pub struct Moderation { + index: Arc, + labeler: Did, + clock: Arc, +} + +impl Moderation { + pub fn new(index: Arc, labeler: Did, clock: Arc) -> Self { + Self { + index, + labeler, + clock, + } + } + + pub(crate) fn labels(&self, subject: &SubjectRef) -> Vec> { + let (target, subject) = match subject { + SubjectRef::Did(did) => (LabelTarget::Account(did), UriValue::Did(did.clone())), + SubjectRef::Uri(uri) => (LabelTarget::Record(uri), UriValue::At(uri.clone())), + SubjectRef::Global => return Vec::new(), + }; + self.index + .labels(target, self.clock.now_unix_micros()) + .into_iter() + .filter(|entry| client_facing(entry.val)) + .filter_map(|entry| self.view(&subject, entry)) + .collect() + } + + fn view(&self, subject: &UriValue, entry: LabelEntry) -> Option> { + let cts = datetime_of(entry.cts)?; + Some(Label { + src: self.labeler.clone(), + uri: subject.clone(), + cid: entry.cid.map(|cid| cid.to_string().into()), + val: self.index.val_str(entry.val).into(), + cts, + exp: entry.exp.and_then(datetime_of), + neg: None, + sig: None, + ver: Some(LABEL_VERSION), + extra_data: None, + }) + } +} + +fn client_facing(val: LabelVal) -> bool { + matches!( + val, + LabelVal::Moderation(ModerationLabel::Hide | ModerationLabel::Warn) + ) +} diff --git a/bobbin/crates/xrpc/src/lib.rs b/bobbin/crates/xrpc/src/lib.rs index a149b8895..ac8a1f1a0 100644 --- a/bobbin/crates/xrpc/src/lib.rs +++ b/bobbin/crates/xrpc/src/lib.rs @@ -98,6 +98,7 @@ mod commit_stats; mod enrich; mod feed; mod filter; +mod labels; mod pocket; mod profile_activity; mod recordpath; @@ -113,6 +114,7 @@ pub use client_address::{ClientAddress, SocketPeer}; use filter::{ CountFilter, IssueFilter, ListFilter, NoFilter, PullFilter, StateViewQuery, VouchNetworkFilter, }; +pub use labels::Moderation; pub use pocket::PocketProxy; use trusted_proxies::TrustedProxies; @@ -217,6 +219,7 @@ pub struct AppState { pub claims: Arc, /// the ingest stream's liveness, absent when nothing publishes it pub stream_health: Option>, + pub moderation: Option, cors: Cors, service_auth_config: service_auth::ServiceAuthConfig>, enrich_router: Arc>, @@ -261,6 +264,7 @@ impl AppState { invites: Arc::new(InviteIndex::new()), claims: Arc::new(ClaimIndex::new()), stream_health: None, + moderation: None, cors: Cors::Off, service_auth_config: ServiceAuthConfig::new( Did::new_static("did:web:localhost").unwrap(), @@ -327,6 +331,11 @@ impl AppState { self } + pub fn with_moderation(mut self, moderation: Option) -> Self { + self.moderation = moderation; + self + } + pub fn with_stream_health(mut self, stream_health: Arc) -> Self { self.stream_health = Some(stream_health); self diff --git a/bobbin/crates/xrpc/tests/enrich.rs b/bobbin/crates/xrpc/tests/enrich.rs index 52b258119..e84de0e52 100644 --- a/bobbin/crates/xrpc/tests/enrich.rs +++ b/bobbin/crates/xrpc/tests/enrich.rs @@ -1,15 +1,17 @@ use std::sync::Arc; use axum::body::{Body, to_bytes}; -use bobbin_edge_index::{CoverageWatch, EdgeStore, TicketStateKind}; +use bobbin_edge_index::{ + CoverageWatch, EdgeStore, IncomingLabel, LabelIndex, LabelTarget, TicketStateKind, +}; use bobbin_record_lru::{CacheCapacity, LruRecordStore}; use bobbin_resolver::RepoIdResolver; -use bobbin_runtime::{RuntimeHasher, SystemClock}; +use bobbin_runtime::{RuntimeHasher, SystemClock, UnixMicros}; 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 bobbin_xrpc::{AppState, Moderation, router}; use http::{Request, StatusCode}; use jacquard_common::DefaultStr; use jacquard_common::types::did::Did; @@ -40,6 +42,7 @@ const DISTINCT_AUTHORS: &str = "sh.tangled.query.enrichResponse#distinctAuthors" const VIEWER: &str = "sh.tangled.query.enrichResponse#viewer"; const MINIDOC: &str = "blue.microcosm.identity.resolveMiniDoc"; const REPO_VIEW: &str = "sh.tangled.repo.defs#repoViewBasic"; +const LABELS: &str = "sh.tangled.query.enrichResponse#labels"; fn at(s: &str) -> AtUri { AtUri::new_owned(s).unwrap() @@ -753,3 +756,85 @@ async fn hydrates_repo_views_by_record_uri() { // the owner did is not targeted, so it gets no view assert!(body["data"][owner.as_str()].is_null(), "{body}"); } + +#[tokio::test] +async fn labels_mark_records_and_their_authors() { + let h = Harness::new().await; + let owner = did("did:plc:nel"); + for (i, fan) in ["did:plc:a", "did:plc:b"].iter().enumerate() { + h.add_edge( + "org.tangled.graph.follow", + SubjectRef::Did(owner.clone()), + &at(&format!("at://{fan}/org.tangled.graph.follow/f{i}")), + ); + h.mount( + &did(fan), + "org.tangled.graph.follow", + &format!("f{i}"), + follow_body(), + ) + .await; + } + let labels = Arc::new(LabelIndex::new(h.edges.clone(), RuntimeHasher::default())); + let followed = at("at://did:plc:b/org.tangled.graph.follow/f1"); + for (seq, (target, val)) in [ + (LabelTarget::Account(&did("did:plc:a")), "!hide"), + (LabelTarget::Record(&followed), "!warn"), + (LabelTarget::Account(&did("did:plc:b")), "!takedown"), + ] + .into_iter() + .enumerate() + { + labels.apply(IncomingLabel { + target, + cid: None, + val, + neg: false, + cts: UnixMicros::new(1), + exp: None, + seq: seq as u64, + }); + } + let labeler = did("did:plc:labeler"); + let state = h.state.clone().with_moderation(Some(Moderation::new( + labels, + labeler.clone(), + Arc::new(SystemClock::new()), + ))); + + let app = router(state); + let (status, body) = json_response( + app.oneshot(enrich_request(json!({ + "xrpc": "sh.tangled.graph.listFollows", + "params": { "subject": owner.as_ref() }, + "enrich": [ + { "source": "org.tangled.graph.follow:.", "type": LABELS }, + { "source": "org.tangled.graph.follow:.repo", "type": LABELS } + ] + }))) + .await + .unwrap(), + ) + .await; + + assert_eq!(status, StatusCode::OK, "{body}"); + let data = &body["data"]; + let hidden = &data["did:plc:a"]["org.tangled.graph.follow:.repo"][LABELS]["labels"]; + assert_eq!(hidden[0]["val"], json!("!hide"), "{body}"); + assert_eq!(hidden[0]["src"], json!(labeler.as_ref()), "{body}"); + assert_eq!(hidden[0]["uri"], json!("did:plc:a"), "{body}"); + let warned = &data[followed.as_str()]["org.tangled.graph.follow:."][LABELS]["labels"]; + assert_eq!(warned[0]["val"], json!("!warn"), "{body}"); + assert!( + data["did:plc:b"]["org.tangled.graph.follow:.repo"] + .get(LABELS) + .is_none(), + "{body}" + ); + assert!( + data["at://did:plc:a/org.tangled.graph.follow/f0"]["org.tangled.graph.follow:."] + .get(LABELS) + .is_none(), + "{body}" + ); +} diff --git a/lexicons/query/enrichResponse.json b/lexicons/query/enrichResponse.json index 2b287d577..1cd9a8cd2 100644 --- a/lexicons/query/enrichResponse.json +++ b/lexicons/query/enrichResponse.json @@ -84,7 +84,8 @@ "sh.tangled.query.enrichResponse#distinctAuthors", "sh.tangled.query.enrichResponse#viewer", "blue.microcosm.identity.resolveMiniDoc", - "sh.tangled.repo.defs#repoViewBasic" + "sh.tangled.repo.defs#repoViewBasic", + "sh.tangled.query.enrichResponse#labels" ], "description": "NSID of the payload type to produce, which also defines the payload's shape. Results land at data[ref][source][type]." }, @@ -96,6 +97,17 @@ "description": "Record paths into the inner response selecting references for this payload. All references are considered when omitted." } } + }, + "labels": { + "type": "object", + "description": "Moderation labels the configured labeler applied. Only !hide and !warn are served, takedowns leave subjects out instead. Use collection:. for labels on the record, collection:.repo for labels on its author's account. Absent when there are none.", + "required": ["labels"], + "properties": { + "labels": { + "type": "array", + "items": { "type": "ref", "ref": "com.atproto.label.defs#label" } + } + } } } } -- 2.54.0