From cd0232a75e7159b8a2b24664c0396408c6112d27 Mon Sep 17 00:00:00 2001 From: Seongmin Lee Date: Fri, 2 Oct 2026 22:34:08 +0900 Subject: [PATCH] bobbin-edge-index: ClaimIndex Signed-off-by: Seongmin Lee --- bobbin/crates/edge-index/src/claims.rs | 334 +++++++++++++++++++++++++ bobbin/crates/edge-index/src/lib.rs | 2 + 2 files changed, 336 insertions(+) create mode 100644 bobbin/crates/edge-index/src/claims.rs diff --git a/bobbin/crates/edge-index/src/claims.rs b/bobbin/crates/edge-index/src/claims.rs new file mode 100644 index 000000000..e496dd76a --- /dev/null +++ b/bobbin/crates/edge-index/src/claims.rs @@ -0,0 +1,334 @@ +//! Claim state for repo manifests, which only become visible once the declared owner confirms them. +//! +//! - manifest `at://R/org.tangled.repo.manifest/self` declaring `(owner, slug)` is claimed while +//! `at://owner/org.tangled.repo.declaration/slug` points back at R. +//! +//! Records are stored regardless of claim state. Every write returns the edge rewrites the caller must apply to the +//! `EdgeStore`: a claimed source gets its candidate edges, an unclaimed one gets none. + +use std::collections::HashMap; + +use bobbin_types::edges::Edge; +use jacquard_common::DefaultStr; +use jacquard_common::types::string::AtUri; +use parking_lot::Mutex; + +const MANIFEST: &str = "org.tangled.repo.manifest"; +const DECLARATION: &str = "org.tangled.repo.declaration"; + +type Ident = (String, String); + +#[derive(Debug, Default)] +pub struct ClaimDelta { + pub rewrites: Vec, + pub transitions: Vec, +} + +/// Replace all edges of `source` with `edges` (empty when unclaimed). +#[derive(Debug, Clone, PartialEq)] +pub struct Rewrite { + pub source: AtUri, + pub edges: Vec, +} + +/// A repo became claimed under an identity it was never claimed under before. +/// `previous` is `None` the first time, else the last claimed `(owner, slug)`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RepoTransition { + pub repo: String, + pub owner: String, + pub slug: String, + pub previous: Option, +} + +struct Manifest { + source: AtUri, + ident: Ident, + edges: Vec, +} + +#[derive(Default)] +struct Inner { + manifests: HashMap, + declarations: HashMap, + last_claimed: HashMap, +} + +#[derive(Default)] +pub struct ClaimIndex { + inner: Mutex, +} + +impl ClaimIndex { + pub fn new() -> Self { + Self::default() + } + + pub fn upsert_manifest( + &self, + source: &AtUri, + owner: &str, + slug: &str, + edges: Vec, + ) -> ClaimDelta { + let Some((repo, _, _)) = split(source.as_ref()) else { + return ClaimDelta::default(); + }; + let mut inner = self.inner.lock(); + let ident = (owner.to_owned(), slug.to_owned()); + inner.manifests.insert( + repo.to_owned(), + Manifest { + source: source.clone(), + ident, + edges, + }, + ); + let mut delta = ClaimDelta::default(); + inner.recheck_repo(repo, &mut delta); + delta + } + + pub fn upsert_declaration(&self, source: &AtUri, repo: &str) -> ClaimDelta { + let Some((owner, _, slug)) = split(source.as_ref()) else { + return ClaimDelta::default(); + }; + let mut inner = self.inner.lock(); + let previous = inner + .declarations + .insert((owner.to_owned(), slug.to_owned()), repo.to_owned()); + let mut delta = ClaimDelta::default(); + if let Some(prev) = previous.filter(|p| p != repo) { + inner.recheck_repo(&prev, &mut delta); + } + inner.recheck_repo(repo, &mut delta); + delta + } + + pub fn delete(&self, source: &AtUri) -> ClaimDelta { + let Some((authority, collection, rkey)) = split(source.as_ref()) else { + return ClaimDelta::default(); + }; + let mut inner = self.inner.lock(); + let mut delta = ClaimDelta::default(); + match collection { + MANIFEST => { + if let Some(m) = inner.manifests.remove(authority) { + delta.rewrites.push(Rewrite { + source: m.source, + edges: Vec::new(), + }); + } + } + DECLARATION => { + let key = (authority.to_owned(), rkey.to_owned()); + if let Some(repo) = inner.declarations.remove(&key) { + inner.recheck_repo(&repo, &mut delta); + } + } + _ => {} + } + delta + } + + /// Claimed `(owner, slug)` of repo R. + pub fn repo(&self, repo: &str) -> Option { + let inner = self.inner.lock(); + inner + .manifest_claimed(repo) + .then(|| inner.manifests[repo].ident.clone()) + } + + pub fn repo_by_name(&self, owner: &str, slug: &str) -> Option { + let inner = self.inner.lock(); + let repo = inner + .declarations + .get(&(owner.to_owned(), slug.to_owned()))?; + inner.manifest_claimed(repo).then(|| repo.clone()) + } + + pub fn is_claimed(&self, source: &AtUri) -> bool { + let inner = self.inner.lock(); + match split(source.as_ref()) { + Some((repo, MANIFEST, _)) => inner.manifest_claimed(repo), + _ => false, + } + } +} + +impl Inner { + fn manifest_claimed(&self, repo: &str) -> bool { + self.manifests + .get(repo) + .is_some_and(|m| self.declarations.get(&m.ident).is_some_and(|r| r == repo)) + } + + fn recheck_repo(&mut self, repo: &str, delta: &mut ClaimDelta) { + let claimed = self.manifest_claimed(repo); + let Some(m) = self.manifests.get(repo) else { + return; + }; + delta.rewrites.push(Rewrite { + source: m.source.clone(), + edges: if claimed { m.edges.clone() } else { Vec::new() }, + }); + if !claimed { + return; + } + let ident = m.ident.clone(); + let previous = self.last_claimed.insert(repo.to_owned(), ident.clone()); + if previous.as_ref() != Some(&ident) { + delta.transitions.push(RepoTransition { + repo: repo.to_owned(), + owner: ident.0, + slug: ident.1, + previous, + }); + } + } +} + +/// `at://authority/collection/rkey` → parts +fn split(uri: &str) -> Option<(&str, &str, &str)> { + let mut parts = uri.strip_prefix("at://")?.splitn(3, '/'); + Some((parts.next()?, parts.next()?, parts.next()?)) +} + +#[cfg(test)] +mod tests { + use super::*; + use bobbin_types::ids::SubjectRef; + use jacquard_common::types::did::Did; + use jacquard_common::types::string::Nsid; + + const R: &str = "did:plc:repo"; + const R2: &str = "did:plc:repo2"; + const ALICE: &str = "did:plc:alice"; + const BOB: &str = "did:plc:bob"; + + fn uri(s: &str) -> AtUri { + AtUri::new_owned(s).unwrap() + } + + fn manifest_uri(repo: &str) -> AtUri { + uri(&format!("at://{repo}/{MANIFEST}/self")) + } + + fn decl_uri(owner: &str, slug: &str) -> AtUri { + uri(&format!("at://{owner}/{DECLARATION}/{slug}")) + } + + fn edge(source: &AtUri, subject: &str) -> Vec { + vec![Edge { + kind: Nsid::new_owned("org.tangled.test").unwrap(), + subject: SubjectRef::Did(Did::new_owned(subject).unwrap()), + source: source.clone(), + sort_micros: 0, + }] + } + + fn rewrite(source: &AtUri, edges: Vec) -> Rewrite { + Rewrite { + source: source.clone(), + edges, + } + } + + #[test] + fn manifest_then_declaration() { + let idx = ClaimIndex::new(); + let m = manifest_uri(R); + let d = idx.upsert_manifest(&m, ALICE, "foo", edge(&m, ALICE)); + assert_eq!(d.rewrites, vec![rewrite(&m, vec![])]); + assert!(idx.repo(R).is_none()); + + let d = idx.upsert_declaration(&decl_uri(ALICE, "foo"), R); + assert_eq!(d.rewrites, vec![rewrite(&m, edge(&m, ALICE))]); + assert_eq!(d.transitions[0].previous, None); + assert_eq!(idx.repo(R), Some((ALICE.into(), "foo".into()))); + assert_eq!(idx.repo_by_name(ALICE, "foo").as_deref(), Some(R)); + assert!(idx.is_claimed(&m)); + } + + #[test] + fn declaration_then_manifest() { + let idx = ClaimIndex::new(); + let m = manifest_uri(R); + let d = idx.upsert_declaration(&decl_uri(ALICE, "foo"), R); + assert!(d.rewrites.is_empty()); + let d = idx.upsert_manifest(&m, ALICE, "foo", edge(&m, ALICE)); + assert_eq!(d.rewrites, vec![rewrite(&m, edge(&m, ALICE))]); + assert_eq!(d.transitions.len(), 1); + } + + #[test] + fn repointed_declaration_unclaims_old_repo() { + let idx = ClaimIndex::new(); + let (m1, m2) = (manifest_uri(R), manifest_uri(R2)); + idx.upsert_manifest(&m1, ALICE, "foo", edge(&m1, ALICE)); + idx.upsert_manifest(&m2, ALICE, "foo", edge(&m2, ALICE)); + idx.upsert_declaration(&decl_uri(ALICE, "foo"), R); + let d = idx.upsert_declaration(&decl_uri(ALICE, "foo"), R2); + assert_eq!( + d.rewrites, + vec![rewrite(&m1, vec![]), rewrite(&m2, edge(&m2, ALICE))] + ); + assert!(idx.repo(R).is_none()); + assert_eq!(idx.repo_by_name(ALICE, "foo").as_deref(), Some(R2)); + } + + #[test] + fn rename_unclaims_until_new_declaration() { + let idx = ClaimIndex::new(); + let m = manifest_uri(R); + idx.upsert_manifest(&m, ALICE, "foo", edge(&m, ALICE)); + idx.upsert_declaration(&decl_uri(ALICE, "foo"), R); + + let d = idx.upsert_manifest(&m, BOB, "bar", edge(&m, BOB)); + assert_eq!(d.rewrites, vec![rewrite(&m, vec![])]); + assert!(d.transitions.is_empty()); + assert!(idx.repo_by_name(ALICE, "foo").is_none()); + + let d = idx.upsert_declaration(&decl_uri(BOB, "bar"), R); + assert_eq!(d.rewrites, vec![rewrite(&m, edge(&m, BOB))]); + assert_eq!( + d.transitions, + vec![RepoTransition { + repo: R.into(), + owner: BOB.into(), + slug: "bar".into(), + previous: Some((ALICE.into(), "foo".into())), + }] + ); + + // stale declaration going away only re-asserts the current edges + let d = idx.delete(&decl_uri(ALICE, "foo")); + assert_eq!(d.rewrites, vec![rewrite(&m, edge(&m, BOB))]); + assert!(d.transitions.is_empty()); + assert!(idx.is_claimed(&m)); + } + + #[test] + fn reclaim_under_same_identity_is_not_a_transition() { + let idx = ClaimIndex::new(); + let m = manifest_uri(R); + idx.upsert_manifest(&m, ALICE, "foo", edge(&m, ALICE)); + idx.upsert_declaration(&decl_uri(ALICE, "foo"), R); + idx.delete(&decl_uri(ALICE, "foo")); + let d = idx.upsert_declaration(&decl_uri(ALICE, "foo"), R); + assert!(d.transitions.is_empty()); + assert!(idx.is_claimed(&m)); + } + + #[test] + fn manifest_delete_clears_edges() { + let idx = ClaimIndex::new(); + let m = manifest_uri(R); + idx.upsert_manifest(&m, ALICE, "foo", edge(&m, ALICE)); + idx.upsert_declaration(&decl_uri(ALICE, "foo"), R); + let d = idx.delete(&m); + assert_eq!(d.rewrites, vec![rewrite(&m, vec![])]); + assert!(idx.repo(R).is_none()); + assert!(idx.repo_by_name(ALICE, "foo").is_none()); + } +} diff --git a/bobbin/crates/edge-index/src/lib.rs b/bobbin/crates/edge-index/src/lib.rs index 694d37084..79df95f47 100644 --- a/bobbin/crates/edge-index/src/lib.rs +++ b/bobbin/crates/edge-index/src/lib.rs @@ -64,6 +64,7 @@ impl BucketKey { } } +pub mod claims; pub mod coverage; pub mod invites; pub mod settle; @@ -71,6 +72,7 @@ mod sorted_blocks; pub mod state_index; mod state_view; pub mod stream; +pub use claims::{ClaimDelta, ClaimIndex, RepoTransition, Rewrite}; pub use coverage::{Coverage, CoverageWatch, HydrantCursor, PromotionSignal}; pub use invites::{InviteIndex, InviteScope, Offer}; pub use settle::{ParsedCid, RecordOutcome, Rejection, Settlement, Settlements, Watch}; -- 2.51.2