diff --git a/Cargo.lock b/Cargo.lock index 21ae47ba..673430ef 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -932,10 +932,14 @@ dependencies = [ name = "bobbin-search" version = "0.0.1" dependencies = [ + "base32", "bobbin-runtime", "bobbin-types", "jacquard-common", + "levenshtein_automata", + "parking_lot", "tantivy", + "tantivy-fst", "thiserror 2.0.18", "tokio", "tracing", diff --git a/Cargo.toml b/Cargo.toml index d555b419..e9796510 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -162,6 +162,9 @@ proptest = "1" divan = "0.1.21" tantivy = "0.26" +tantivy-fst = "0.5" +levenshtein_automata = "0.2" +parking_lot = "0.12" gengo-language = "0.14" regex = "1" globset = "0.4" diff --git a/bobbin/crates/bobbin/src/main.rs b/bobbin/crates/bobbin/src/main.rs index 4258a723..f1510f78 100644 --- a/bobbin/crates/bobbin/src/main.rs +++ b/bobbin/crates/bobbin/src/main.rs @@ -18,7 +18,7 @@ use bobbin_runtime::{ Clock, GuardedWs, MemoryBudget, NetworkError, OsEntropy, ReqwestHttp, RuntimeHasher, SystemClock, TungsteniteWs, WsTransport, }; -use bobbin_search::{SearchIndex, SearchReader}; +use bobbin_search::{ActorIndex, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; use bobbin_slingshot_client::default_http_client; use bobbin_xrpc::{ @@ -187,12 +187,14 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { let records: Arc = Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(lru_cap))); let slingshot = SlingshotClient::with_default_http(cfg.slingshot.url.clone())?; + let actors = Arc::new(ActorIndex::new()); let identity = Arc::new( IdentityResolver::with_slingshot(slingshot.clone(), clock.clone(), hasher.clone()) .with_hydrant(HydrantClient::new( cfg.hydrant.url.clone(), ReqwestHttp::shared(default_http_client()?), - )?), + )?) + .with_sink(actors.clone()), ); let mut resolver_opts = ResolverOptions::default(); // NOTE: see https://tangled.org/nonbinary.computer/jacquard/issues/39. @@ -293,6 +295,26 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { parallelism, }; let cancel = CancellationToken::new(); + let actor_rebuilder = actors.clone(); + let actor_rebuilder_cancel = cancel.clone(); + let _actor_rebuilder = tokio::spawn(async move { + loop { + tokio::select! { + _ = actor_rebuilder_cancel.cancelled() => break, + _ = tokio::time::sleep(Duration::from_secs(30)) => {} + } + let actors = actor_rebuilder.clone(); + match tokio::task::spawn_blocking(move || actors.rebuild()).await { + Ok(Ok(stats)) => tracing::debug!( + actors = stats.num_actors, + bytes = stats.total_bytes, + "rebuilt actor handle list", + ), + Ok(Err(error)) => tracing::warn!(%error, "could not rebuild actor handle list"), + Err(error) => tracing::warn!(%error, "actor handle rebuild stopped unexpectedly"), + } + } + }); let ingest_coverage = coverage.clone(); let knot_acl_dev = !cfg.knot.require_https; @@ -401,6 +423,7 @@ async fn run(cfg: BobbinConfig) -> anyhow::Result<()> { ) .with_codesearch(codesearch) .with_identity(identity) + .with_actors(actors) .with_limiter(limiter) .with_mirror(mirror) .with_mirror_v2(mirror_v2) diff --git a/bobbin/crates/resolver/src/identity.rs b/bobbin/crates/resolver/src/identity.rs index 21837c01..c1876889 100644 --- a/bobbin/crates/resolver/src/identity.rs +++ b/bobbin/crates/resolver/src/identity.rs @@ -4,6 +4,7 @@ use std::time::Duration; use bobbin_runtime::{Clock, RuntimeHasher}; use bobbin_slingshot_client::{SlingshotClient, SlingshotError}; +use bobbin_types::identity::IdentitySink; use jacquard_common::DefaultStr; use jacquard_common::types::crypto::PublicKey; use jacquard_common::types::did::Did; @@ -152,6 +153,8 @@ pub struct IdentityResolver { hydrant: Option, probe: Option, clock: Option>, + sink: Option>, + mutations: Mutex<()>, stats: IdentityResolverStats, } @@ -173,6 +176,11 @@ impl IdentityResolver { self } + pub fn with_sink(mut self, sink: Arc) -> Self { + self.sink = Some(sink); + self + } + fn new(probe: Option, hasher: RuntimeHasher) -> Self { let clock = probe.as_ref().map(|probe| probe.clock.clone()); let (warm_tx, warm_rx) = mpsc::unbounded_channel(); @@ -187,6 +195,8 @@ impl IdentityResolver { hydrant: None, probe, clock, + sink: None, + mutations: Mutex::new(()), stats: IdentityResolverStats::default(), } } @@ -328,6 +338,10 @@ impl IdentityResolver { } fn observe_doc(&self, doc: MiniDoc) { + let _mutation = self + .mutations + .lock() + .expect("identity mutation mutex poisoned"); let (did, handle) = (doc.did.clone(), doc.handle.clone()); let mut previous_handle = None; match self.by_did.entry_sync(did.clone()) { @@ -350,10 +364,17 @@ impl IdentityResolver { } } self.remove_by_handle_if_owned(&did, previous_handle.as_ref()); - self.insert_by_handle(did, handle); + self.insert_by_handle(did.clone(), handle.clone()); + if let Some(sink) = self.sink.as_ref() { + sink.identity_changed(&did, previous_handle.as_ref(), &handle); + } } pub fn deactivate(&self, did: Did) { + let _mutation = self + .mutations + .lock() + .expect("identity mutation mutex poisoned"); let mut previous_handle = None; match self.by_did.entry_sync(did.clone()) { MapEntry::Occupied(mut occupied) => { @@ -365,6 +386,9 @@ impl IdentityResolver { } } self.remove_by_handle_if_owned(&did, previous_handle.as_ref()); + if let Some(sink) = self.sink.as_ref() { + sink.identity_removed(&did, previous_handle.as_ref()); + } } fn remove_by_handle_if_owned( @@ -514,18 +538,26 @@ impl IdentityResolver { } fn insert_fetched_by_did(&self, doc: MiniDoc) -> Result { - match self.by_did.entry_sync(doc.did.clone()) { + let _mutation = self + .mutations + .lock() + .expect("identity mutation mutex poisoned"); + let (did, handle) = match self.by_did.entry_sync(doc.did.clone()) { MapEntry::Occupied(occupied) => match occupied.get() { - IdentityState::Inactive => Err(IdentityResolveError::NotFound), - IdentityState::Cached(current) => Ok(current.clone()), + IdentityState::Inactive => return Err(IdentityResolveError::NotFound), + IdentityState::Cached(current) => return Ok(current.clone()), }, MapEntry::Vacant(vacant) => { let (did, handle) = (doc.did.clone(), doc.handle.clone()); vacant.insert_entry(IdentityState::Cached(doc.clone())); - self.insert_by_handle(did, handle); - Ok(doc) + (did, handle) } + }; + self.insert_by_handle(did.clone(), handle.clone()); + if let Some(sink) = self.sink.as_ref() { + sink.identity_changed(&did, None, &handle); } + Ok(doc) } } @@ -778,10 +810,11 @@ mod tests { } async fn drain_warming(resolver: &Arc, did: &Did) { + let settled_before = resolver.stats().warm_resolved + resolver.stats().warm_failed; let cancel = CancellationToken::new(); let warmer = tokio::spawn(resolver.clone().run_warming(cancel.clone())); tokio::time::timeout(Duration::from_secs(5), async { - while resolver.stats().warm_resolved + resolver.stats().warm_failed == 0 { + while resolver.stats().warm_resolved + resolver.stats().warm_failed == settled_before { tokio::task::yield_now().await; } }) @@ -1028,4 +1061,220 @@ mod tests { resolver.warm(&identity); assert_eq!(resolver.stats().warm_queued, 0); } + + #[derive(Clone, Debug, Eq, PartialEq)] + enum SinkEvent { + Changed { + did: Did, + previous_handle: Option>, + handle: Handle, + }, + Removed { + did: Did, + previous_handle: Option>, + }, + } + + #[derive(Default)] + struct RecordingSink { + events: Mutex>, + } + + impl IdentitySink for RecordingSink { + fn identity_changed( + &self, + did: &Did, + previous_handle: Option<&Handle>, + handle: &Handle, + ) { + self.events.lock().unwrap().push(SinkEvent::Changed { + did: did.clone(), + previous_handle: previous_handle.cloned(), + handle: handle.clone(), + }); + } + + fn identity_removed( + &self, + did: &Did, + previous_handle: Option<&Handle>, + ) { + self.events.lock().unwrap().push(SinkEvent::Removed { + did: did.clone(), + previous_handle: previous_handle.cloned(), + }); + } + } + + struct IdentityCheckingSink { + resolver: Arc>>, + events: Mutex>, + } + + impl IdentitySink for IdentityCheckingSink { + fn identity_changed( + &self, + did: &Did, + previous_handle: Option<&Handle>, + handle: &Handle, + ) { + let resolver = self.resolver.get().expect("resolver initialized"); + let doc = resolver.get_by_did(did).expect("did must resolve"); + assert_eq!(&doc.handle, handle); + let by_handle = resolver + .cached(&AtIdentifier::Handle(handle.clone())) + .expect("cache lookup must succeed") + .expect("handle must resolve"); + assert_eq!(&by_handle.did, did); + if let Some(prev) = previous_handle + && prev != handle + { + assert!( + resolver + .cached(&AtIdentifier::Handle(prev.clone())) + .expect("cache lookup must succeed") + .is_none() + ); + } + self.events.lock().unwrap().push(SinkEvent::Changed { + did: did.clone(), + previous_handle: previous_handle.cloned(), + handle: handle.clone(), + }); + } + + fn identity_removed( + &self, + did: &Did, + previous_handle: Option<&Handle>, + ) { + let resolver = self.resolver.get().expect("resolver initialized"); + assert_eq!( + resolver.get_by_did(did), + Err(IdentityResolveError::NotFound) + ); + if let Some(prev) = previous_handle { + assert!( + resolver + .cached(&AtIdentifier::Handle(prev.clone())) + .expect("cache lookup must succeed") + .is_none() + ); + } + self.events.lock().unwrap().push(SinkEvent::Removed { + did: did.clone(), + previous_handle: previous_handle.cloned(), + }); + } + } + + #[test] + fn callback_runs_after_identity_maps_are_updated() { + let cell = Arc::new(OnceCell::new()); + let sink = Arc::new(IdentityCheckingSink { + resolver: cell.clone(), + events: Mutex::new(Vec::new()), + }); + let resolver = Arc::new(resolver().with_sink(sink.clone())); + cell.set(resolver.clone()).ok().expect("cell set"); + + let identity = did("did:plc:dawn"); + let initial_handle = handle("old.ptr.pet"); + let updated_handle = handle("new.ptr.pet"); + + // 1. Initial observation + resolver.observe(identity.clone(), initial_handle.clone()); + // 2. Rename + resolver.observe(identity.clone(), updated_handle.clone()); + // 3. Deactivate + resolver.deactivate(identity.clone()); + + let checked = sink.events.lock().unwrap().clone(); + assert_eq!(checked.len(), 3); + assert_eq!( + checked[0], + SinkEvent::Changed { + did: identity.clone(), + previous_handle: None, + handle: initial_handle.clone(), + } + ); + assert_eq!( + checked[1], + SinkEvent::Changed { + did: identity.clone(), + previous_handle: Some(initial_handle), + handle: updated_handle.clone(), + } + ); + assert_eq!( + checked[2], + SinkEvent::Removed { + did: identity, + previous_handle: Some(updated_handle), + } + ); + } + + #[tokio::test] + async fn hydrant_results_are_sent_to_identity_sink() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/repos/did:plc:live")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "did": "did:plc:live", + "status": "active", + "tracked": true, + "handle": "live.example.com", + }))) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/repos/did:plc:dead")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "did": "did:plc:dead", + "status": "deleted", + "tracked": true, + "handle": "dead.example.com", + }))) + .mount(&server) + .await; + + let sink = Arc::new(RecordingSink::default()); + let base = Url::parse(&server.uri()).unwrap(); + let resolver = Arc::new( + IdentityResolver::with_slingshot( + SlingshotClient::with_default_http(base.clone()).unwrap(), + Arc::new(SystemClock::new()), + hasher(), + ) + .with_hydrant( + HydrantClient::new(base, ReqwestHttp::shared(default_http_client().unwrap())) + .unwrap(), + ) + .with_sink(sink.clone()), + ); + + let live_did = did("did:plc:live"); + resolver.fill_one(&live_did).await.unwrap(); + + let dead_did = did("did:plc:dead"); + resolver.fill_one(&dead_did).await.unwrap(); + + let events = sink.events.lock().unwrap().clone(); + assert_eq!( + events, + vec![ + SinkEvent::Changed { + did: live_did, + previous_handle: None, + handle: handle("live.example.com"), + }, + SinkEvent::Removed { + did: dead_did, + previous_handle: None, + }, + ] + ); + } } diff --git a/bobbin/crates/search/Cargo.toml b/bobbin/crates/search/Cargo.toml index 51ba79a5..334297ae 100644 --- a/bobbin/crates/search/Cargo.toml +++ b/bobbin/crates/search/Cargo.toml @@ -10,7 +10,11 @@ bobbin-runtime = { workspace = true } bobbin-types = { workspace = true } jacquard-common = { workspace = true } +base32 = { workspace = true } tantivy = { workspace = true } +tantivy-fst = { workspace = true } +levenshtein_automata = { workspace = true } +parking_lot = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true, features = ["rt", "sync", "time"] } tracing = { workspace = true } diff --git a/bobbin/crates/search/src/actor.rs b/bobbin/crates/search/src/actor.rs new file mode 100644 index 00000000..74e5a044 --- /dev/null +++ b/bobbin/crates/search/src/actor.rs @@ -0,0 +1,975 @@ +use std::collections::{BTreeMap, HashSet}; +use std::mem; +use std::sync::Arc; + +use base32::Alphabet; +use bobbin_types::identity::IdentitySink; +use jacquard_common::DefaultStr; +use jacquard_common::types::did::Did; +use jacquard_common::types::string::Handle; +use levenshtein_automata::{DFA, Distance, LevenshteinAutomatonBuilder, SINK_STATE}; +use parking_lot::{Mutex, RwLock}; +use tantivy_fst::{Automaton, IntoStreamer, Map, MapBuilder, Streamer}; +use thiserror::Error; + +const TYPO_PREFIX_MIN_CHARS: usize = 3; +const SHARD_PREFIX_LEN: usize = 2; +const INVALID_HANDLE: &str = "handle.invalid"; +const PLC_PREFIX: &str = "did:plc:"; + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ActorSuggestion { + pub did: Did, + pub handle: String, +} + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct ActorIndexStats { + pub num_actors: u64, + /// Encoded snapshots plus logical overlay payloads; excludes container overhead. + pub total_bytes: u64, +} + +#[derive(Debug, Error)] +pub enum ActorIndexError { + #[error("could not build actor handle map: {0}")] + Build(#[from] tantivy_fst::Error), + #[error("stored actor DID is invalid: {0}")] + InvalidDid(String), + #[error("actor search worker stopped: {0}")] + WorkerFailed(String), + #[error("actor DID table is full")] + CapacityReached, +} + +#[derive(Clone)] +pub struct ActorIndex { + inner: Arc, +} + +struct ActorIndexInner { + state: RwLock, + rebuilding: Mutex<()>, + typo_matcher: LevenshteinAutomatonBuilder, +} + +struct ActorIndexState { + // Fixed-width prefixes preserve global lexical order between immutable shards. + shards: BTreeMap>, + // These changes stay visible while their shards are rebuilt. + changes_being_rebuilt: Option>, + // Identity changes that arrived after the current rebuild started. + recent_changes: ShardChanges, +} + +type ShardPrefix = [u8; SHARD_PREFIX_LEN]; +type ShardChanges = BTreeMap; +type HandleChanges = BTreeMap, HandleChange>; + +enum HandleChange { + Present(Box), + Removed, +} + +struct IndexedActors { + handles: Map>, + dids: DidTable, + byte_size: usize, +} + +// A PLC DID fits directly in one entry. Other DID forms point into `text`. +#[derive(Default)] +struct DidTable { + entries: Vec<[u8; 16]>, + text: String, +} + +struct TypoMatcher<'a>(&'a DFA); + +impl Automaton for TypoMatcher<'_> { + type State = u32; + + fn start(&self) -> Self::State { + self.0.initial_state() + } + + fn is_match(&self, state: &Self::State) -> bool { + matches!(self.0.distance(*state), Distance::Exact(_)) + } + + fn can_match(&self, state: &Self::State) -> bool { + *state != SINK_STATE + } + + fn accept(&self, state: &Self::State, byte: u8) -> Self::State { + self.0.transition(*state, byte) + } +} + +impl ActorIndex { + pub fn new() -> Self { + Self { + inner: Arc::new(ActorIndexInner { + typo_matcher: LevenshteinAutomatonBuilder::new(1, true), + state: RwLock::new(ActorIndexState { + shards: BTreeMap::new(), + changes_being_rebuilt: None, + recent_changes: BTreeMap::new(), + }), + rebuilding: Mutex::new(()), + }), + } + } + + pub async fn suggest( + &self, + query: &str, + limit: u32, + ) -> Result, ActorIndexError> { + if limit == 0 { + return Ok(Vec::new()); + } + let normalized = query.trim().trim_start_matches('@').to_ascii_lowercase(); + if normalized.is_empty() { + return Ok(Vec::new()); + } + let index = self.clone(); + match tokio::task::spawn_blocking(move || { + index.suggest_blocking(&normalized, limit as usize) + }) + .await + { + Ok(result) => result, + Err(error) if error.is_panic() => std::panic::resume_unwind(error.into_panic()), + Err(error) => Err(ActorIndexError::WorkerFailed(error.to_string())), + } + } + + pub fn rebuild(&self) -> Result { + self.rebuild_after_freeze(|| {}) + } + + fn rebuild_after_freeze( + &self, + after_freeze: impl FnOnce(), + ) -> Result { + let _rebuild_guard = self.inner.rebuilding.lock(); + let (snapshots, changes_being_rebuilt) = { + let mut state = self.inner.state.write(); + if state.recent_changes.is_empty() { + return Ok(stats_locked(&state)); + } + let changes_being_rebuilt = Arc::new(mem::take(&mut state.recent_changes)); + let snapshots = changes_being_rebuilt + .keys() + .map(|prefix| { + ( + *prefix, + state + .shards + .get(prefix) + .expect("changed actor shard must exist") + .clone(), + ) + }) + .collect::>(); + state.changes_being_rebuilt = Some(changes_being_rebuilt.clone()); + (snapshots, changes_being_rebuilt) + }; + after_freeze(); + + // Building can take time, so identity updates continue in `recent_changes`. + let built = snapshots + .into_iter() + .map(|(prefix, snapshot)| { + IndexedActors::apply_changes( + &snapshot, + changes_being_rebuilt + .get(&prefix) + .expect("changed actor shard must have changes"), + ) + .map(|snapshot| (prefix, Arc::new(snapshot))) + }) + .collect::, _>>(); + match built { + Ok(shards) => { + let mut state = self.inner.state.write(); + state.shards.extend(shards); + state.changes_being_rebuilt = None; + Ok(stats_locked(&state)) + } + Err(error) => { + let mut state = self.inner.state.write(); + state.changes_being_rebuilt = None; + let frozen = Arc::try_unwrap(changes_being_rebuilt) + .unwrap_or_else(|_| unreachable!("rebuild batch must be uniquely owned")); + for (prefix, changes) in frozen { + let recent = state.recent_changes.entry(prefix).or_default(); + for (handle, value) in changes { + recent.entry(handle).or_insert(value); + } + } + Err(error) + } + } + } + + fn suggest_blocking( + &self, + normalized: &str, + limit: usize, + ) -> Result, ActorIndexError> { + let state = self.inner.state.read(); + let mut output = Vec::with_capacity(limit); + let mut seen_dids = HashSet::with_capacity(limit); + collect_exact_matches(&state, normalized, limit, &mut output, &mut seen_dids)?; + if output.len() == limit || normalized.chars().count() < TYPO_PREFIX_MIN_CHARS { + return Ok(output); + } + collect_typo_matches( + &state, + &self.inner.typo_matcher, + normalized, + limit, + &mut output, + &mut seen_dids, + )?; + Ok(output) + } +} + +impl Default for ActorIndex { + fn default() -> Self { + Self::new() + } +} +impl ActorIndexState { + fn record_change(&mut self, handle: String, change: HandleChange) { + let prefix = shard_prefix(&handle); + if matches!(&change, HandleChange::Removed) && !self.shards.contains_key(&prefix) { + return; + } + self.shards.entry(prefix).or_insert_with(|| { + Arc::new(IndexedActors::empty().expect("an empty actor shard must be valid")) + }); + self.recent_changes + .entry(prefix) + .or_default() + .insert(handle.into_boxed_str(), change); + } +} + +impl IdentitySink for ActorIndex { + fn identity_changed( + &self, + did: &Did, + previous_handle: Option<&Handle>, + handle: &Handle, + ) { + let normalized = handle.as_ref().to_ascii_lowercase(); + let previous = previous_handle.map(|handle| handle.as_ref().to_ascii_lowercase()); + let present = + (normalized != INVALID_HANDLE).then(|| HandleChange::Present(did.as_ref().into())); + let mut state = self.inner.state.write(); + if let Some(previous) = previous { + state.record_change(previous, HandleChange::Removed); + } + if let Some(present) = present { + state.record_change(normalized, present); + } + } + + fn identity_removed( + &self, + _did: &Did, + previous_handle: Option<&Handle>, + ) { + let Some(previous) = previous_handle else { + return; + }; + let previous = previous.as_ref().to_ascii_lowercase(); + self.inner + .state + .write() + .record_change(previous, HandleChange::Removed); + } +} +fn shard_prefix(handle: &str) -> ShardPrefix { + handle.as_bytes()[..SHARD_PREFIX_LEN] + .try_into() + .expect("validated handle must contain a shard prefix") +} + +impl IndexedActors { + fn empty() -> Result { + let handles = Map::from_iter(std::iter::empty::<(String, u64)>())?; + Ok(Self::from_parts(handles, DidTable::default())) + } + + fn from_parts(handles: Map>, dids: DidTable) -> Self { + // TODO: load the finished handle map from a memory-mapped file. + let byte_size = handles.as_fst().size() + + dids.entries.len() * mem::size_of::<[u8; 16]>() + + dids.text.len(); + Self { + handles, + dids, + byte_size, + } + } + + fn apply_changes( + snapshot: &IndexedActors, + changes: &HandleChanges, + ) -> Result { + let mut builder = MapBuilder::memory(); + let mut dids = DidTable::default(); + let mut indexed = snapshot.handles.stream(); + let mut changes = changes.iter().peekable(); + + 'indexed: while let Some((handle, did_index)) = indexed.next() { + while let Some((changed_handle, changed_did)) = changes.peek() { + match changed_handle.as_bytes().cmp(handle) { + std::cmp::Ordering::Less => { + if let HandleChange::Present(did) = changed_did { + add_indexed_actor( + &mut builder, + &mut dids, + changed_handle.as_bytes(), + did, + )?; + } + changes.next(); + } + std::cmp::Ordering::Equal => { + if let HandleChange::Present(did) = changed_did { + add_indexed_actor(&mut builder, &mut dids, handle, did)?; + } + changes.next(); + continue 'indexed; + } + std::cmp::Ordering::Greater => break, + } + } + add_existing_actor(&mut builder, &mut dids, &snapshot.dids, handle, did_index)?; + } + for (handle, change) in changes { + if let HandleChange::Present(did) = change { + add_indexed_actor(&mut builder, &mut dids, handle.as_bytes(), did)?; + } + } + + let handles = Map::from_bytes(builder.into_inner()?)?; + Ok(Self::from_parts(handles, dids)) + } +} +fn add_indexed_actor( + builder: &mut MapBuilder>, + dids: &mut DidTable, + handle: &[u8], + did: &str, +) -> Result<(), ActorIndexError> { + let did_index = dids.insert(did)?; + builder.insert(handle, did_index)?; + Ok(()) +} +fn add_existing_actor( + builder: &mut MapBuilder>, + dids: &mut DidTable, + source_dids: &DidTable, + handle: &[u8], + source_did_index: u64, +) -> Result<(), ActorIndexError> { + let did_index = dids.copy_from(source_dids, source_did_index)?; + builder.insert(handle, did_index)?; + Ok(()) +} + +impl DidTable { + fn insert(&mut self, did: &str) -> Result { + let mut slot = [0u8; 16]; + if let Some(plc) = did.strip_prefix(PLC_PREFIX).and_then(decode_plc) { + slot[0] = 0; + slot[1..].copy_from_slice(&plc); + } else { + let offset = + u32::try_from(self.text.len()).map_err(|_| ActorIndexError::CapacityReached)?; + let len = u32::try_from(did.len()).map_err(|_| ActorIndexError::CapacityReached)?; + slot[0] = 1; + slot[1..5].copy_from_slice(&offset.to_le_bytes()); + slot[5..9].copy_from_slice(&len.to_le_bytes()); + self.text.push_str(did); + } + let did_index = self.entries.len() as u64; + self.entries.push(slot); + Ok(did_index) + } + + fn copy_from(&mut self, source: &Self, did_index: u64) -> Result { + let mut entry = *source.entries.get(did_index as usize).ok_or_else(|| { + ActorIndexError::InvalidDid(format!("missing DID at index {did_index}")) + })?; + match entry[0] { + 0 => {} + 1 => { + let offset = u32::from_le_bytes(entry[1..5].try_into().unwrap()) as usize; + let len = u32::from_le_bytes(entry[5..9].try_into().unwrap()) as usize; + let did = source + .text + .get(offset..offset.saturating_add(len)) + .ok_or_else(|| ActorIndexError::InvalidDid("DID text range".into()))?; + let offset = + u32::try_from(self.text.len()).map_err(|_| ActorIndexError::CapacityReached)?; + entry[1..5].copy_from_slice(&offset.to_le_bytes()); + self.text.push_str(did); + } + tag => { + return Err(ActorIndexError::InvalidDid(format!( + "unknown DID tag {tag}" + ))); + } + } + let did_index = self.entries.len() as u64; + self.entries.push(entry); + Ok(did_index) + } + + fn get(&self, did_index: u64) -> Result { + let entry = self.entries.get(did_index as usize).ok_or_else(|| { + ActorIndexError::InvalidDid(format!("missing DID at index {did_index}")) + })?; + match entry[0] { + 0 => Ok(format!( + "{PLC_PREFIX}{}", + encode_plc((&entry[1..]).try_into().unwrap()) + )), + 1 => { + let offset = u32::from_le_bytes(entry[1..5].try_into().unwrap()) as usize; + let len = u32::from_le_bytes(entry[5..9].try_into().unwrap()) as usize; + let raw = self + .text + .get(offset..offset.saturating_add(len)) + .ok_or_else(|| ActorIndexError::InvalidDid("DID text range".into()))?; + Ok(raw.to_owned()) + } + tag => Err(ActorIndexError::InvalidDid(format!( + "unknown DID tag {tag}" + ))), + } + } +} + +type ChangeEntry<'a> = (&'a Box, &'a HandleChange); + +struct ChangeCursor<'a, I> +where + I: Iterator>, +{ + changes_being_rebuilt: std::iter::Peekable, + recent_changes: std::iter::Peekable, +} + +impl<'a, I> ChangeCursor<'a, I> +where + I: Iterator>, +{ + // A newer change for the same handle always wins. + fn next(&mut self) -> Option<(&'a str, &'a HandleChange)> { + let ordering = match ( + self.changes_being_rebuilt.peek(), + self.recent_changes.peek(), + ) { + (Some((changes_being_rebuilt, _)), Some((recent_changes, _))) => { + Some(changes_being_rebuilt.cmp(recent_changes)) + } + (Some(_), None) => Some(std::cmp::Ordering::Less), + (None, Some(_)) => Some(std::cmp::Ordering::Greater), + (None, None) => None, + }?; + match ordering { + std::cmp::Ordering::Less => self + .changes_being_rebuilt + .next() + .map(|(handle, value)| (handle.as_ref(), value)), + std::cmp::Ordering::Equal => { + self.changes_being_rebuilt.next(); + self.recent_changes + .next() + .map(|(handle, value)| (handle.as_ref(), value)) + } + std::cmp::Ordering::Greater => self + .recent_changes + .next() + .map(|(handle, value)| (handle.as_ref(), value)), + } + } +} + +fn exact_changes<'a>( + changes_being_rebuilt: Option<&'a HandleChanges>, + recent_changes: Option<&'a HandleChanges>, + prefix: &'a str, +) -> ChangeCursor<'a, impl Iterator> + 'a> { + fn matching<'a>( + changes: Option<&'a HandleChanges>, + prefix: &'a str, + ) -> impl Iterator> + 'a { + changes.into_iter().flat_map(move |changes| { + changes + .range::(( + std::ops::Bound::Included(prefix), + std::ops::Bound::Unbounded, + )) + .take_while(move |(handle, _)| handle.starts_with(prefix)) + }) + } + ChangeCursor { + changes_being_rebuilt: matching(changes_being_rebuilt, prefix).peekable(), + recent_changes: matching(recent_changes, prefix).peekable(), + } +} + +fn typo_changes<'a>( + changes_being_rebuilt: Option<&'a HandleChanges>, + recent_changes: Option<&'a HandleChanges>, + matcher: &'a DFA, +) -> ChangeCursor<'a, impl Iterator> + 'a> { + fn matching<'a>( + changes: Option<&'a HandleChanges>, + matcher: &'a DFA, + ) -> impl Iterator> + 'a { + changes.into_iter().flat_map(move |changes| { + changes.iter().filter(move |(handle, _)| { + matches!(matcher.eval(handle.as_bytes()), Distance::Exact(_)) + }) + }) + } + ChangeCursor { + changes_being_rebuilt: matching(changes_being_rebuilt, matcher).peekable(), + recent_changes: matching(recent_changes, matcher).peekable(), + } +} + +fn matcher_can_match_shard(matcher: &DFA, shard: &ShardPrefix) -> bool { + shard.iter().fold(matcher.initial_state(), |state, byte| { + matcher.transition(state, *byte) + }) != SINK_STATE +} + +fn collect_exact_matches( + state: &ActorIndexState, + prefix: &str, + limit: usize, + output: &mut Vec, + seen_dids: &mut HashSet, +) -> Result<(), ActorIndexError> { + let bytes = prefix.as_bytes(); + let start = [bytes[0], bytes.get(1).copied().unwrap_or(0)]; + let end = [bytes[0], bytes.get(1).copied().unwrap_or(u8::MAX)]; + for (shard_prefix, snapshot) in state.shards.range(start..=end) { + collect_exact_shard( + snapshot, + state + .changes_being_rebuilt + .as_deref() + .and_then(|changes| changes.get(shard_prefix)), + state.recent_changes.get(shard_prefix), + prefix, + limit, + output, + seen_dids, + )?; + if output.len() == limit { + break; + } + } + Ok(()) +} + +fn collect_exact_shard( + snapshot: &IndexedActors, + changes_being_rebuilt: Option<&HandleChanges>, + recent_changes: Option<&HandleChanges>, + prefix: &str, + limit: usize, + output: &mut Vec, + seen_dids: &mut HashSet, +) -> Result<(), ActorIndexError> { + let mut range = snapshot.handles.range().ge(prefix.as_bytes()); + if let Some(end) = prefix_end(prefix.as_bytes()) { + range = range.lt(end); + } + let mut stream = range.into_stream(); + collect_matches( + snapshot, + limit, + output, + seen_dids, + || { + stream.next().map(|(handle, did_index)| { + (String::from_utf8_lossy(handle).into_owned(), did_index) + }) + }, + exact_changes(changes_being_rebuilt, recent_changes, prefix), + ) +} + +fn collect_typo_matches( + state: &ActorIndexState, + builder: &LevenshteinAutomatonBuilder, + query: &str, + limit: usize, + output: &mut Vec, + seen_dids: &mut HashSet, +) -> Result<(), ActorIndexError> { + let matcher = builder.build_prefix_dfa(query); + for (shard_prefix, snapshot) in &state.shards { + if !matcher_can_match_shard(&matcher, shard_prefix) { + continue; + } + collect_typo_shard( + snapshot, + state + .changes_being_rebuilt + .as_deref() + .and_then(|changes| changes.get(shard_prefix)), + state.recent_changes.get(shard_prefix), + &matcher, + limit, + output, + seen_dids, + )?; + if output.len() == limit { + break; + } + } + Ok(()) +} + +fn collect_typo_shard( + snapshot: &IndexedActors, + changes_being_rebuilt: Option<&HandleChanges>, + recent_changes: Option<&HandleChanges>, + matcher: &DFA, + limit: usize, + output: &mut Vec, + seen_dids: &mut HashSet, +) -> Result<(), ActorIndexError> { + let mut stream = snapshot.handles.search(TypoMatcher(matcher)).into_stream(); + collect_matches( + snapshot, + limit, + output, + seen_dids, + || { + stream.next().map(|(handle, did_index)| { + (String::from_utf8_lossy(handle).into_owned(), did_index) + }) + }, + typo_changes(changes_being_rebuilt, recent_changes, matcher), + ) +} + +fn collect_matches<'a, I>( + snapshot: &IndexedActors, + limit: usize, + output: &mut Vec, + seen_dids: &mut HashSet, + mut next_indexed: impl FnMut() -> Option<(String, u64)>, + mut changes: ChangeCursor<'a, I>, +) -> Result<(), ActorIndexError> +where + I: Iterator>, +{ + let mut indexed = next_indexed(); + let mut changed = changes.next(); + while output.len() < limit && (indexed.is_some() || changed.is_some()) { + let (handle, did) = match (indexed.as_ref(), changed.as_ref()) { + (Some((indexed_handle, did_index)), Some((changed_handle, value))) => { + match indexed_handle.as_str().cmp(changed_handle) { + std::cmp::Ordering::Less => { + let result = (indexed_handle.clone(), Some(snapshot.dids.get(*did_index)?)); + indexed = next_indexed(); + result + } + std::cmp::Ordering::Equal => { + let result = (changed_handle.to_string(), changed_did(value)); + indexed = next_indexed(); + changed = changes.next(); + result + } + std::cmp::Ordering::Greater => { + let result = (changed_handle.to_string(), changed_did(value)); + changed = changes.next(); + result + } + } + } + (Some((indexed_handle, did_index)), None) => { + let result = (indexed_handle.clone(), Some(snapshot.dids.get(*did_index)?)); + indexed = next_indexed(); + result + } + (None, Some((changed_handle, value))) => { + let result = (changed_handle.to_string(), changed_did(value)); + changed = changes.next(); + result + } + (None, None) => break, + }; + add_result(handle, did, output, seen_dids)?; + } + Ok(()) +} + +fn changed_did(value: &HandleChange) -> Option { + match value { + HandleChange::Present(did) => Some(did.to_string()), + HandleChange::Removed => None, + } +} + +fn add_result( + handle: String, + did: Option, + output: &mut Vec, + seen_dids: &mut HashSet, +) -> Result<(), ActorIndexError> { + let Some(did) = did else { + return Ok(()); + }; + if seen_dids.insert(did.clone()) { + output.push(ActorSuggestion { + did: parse_did(did)?, + handle, + }); + } + Ok(()) +} + +fn parse_did(raw: String) -> Result, ActorIndexError> { + Did::new_owned(raw).map_err(|error| ActorIndexError::InvalidDid(error.to_string())) +} + +fn prefix_end(prefix: &[u8]) -> Option> { + let mut end = prefix.to_vec(); + while let Some(last) = end.pop() { + if last != u8::MAX { + end.push(last + 1); + return Some(end); + } + } + None +} + +fn decode_plc(raw: &str) -> Option<[u8; 15]> { + if raw.len() != 24 { + return None; + } + base32::decode(Alphabet::Rfc4648Lower { padding: false }, raw)? + .try_into() + .ok() +} + +fn encode_plc(raw: &[u8; 15]) -> String { + base32::encode(Alphabet::Rfc4648Lower { padding: false }, raw) +} + +fn stats_locked(state: &ActorIndexState) -> ActorIndexStats { + let change_bytes = state + .changes_being_rebuilt + .iter() + .flat_map(|changes| changes.values()) + .chain(state.recent_changes.values()) + .flat_map(|changes| changes.iter()) + .map(|(handle, value)| { + handle.len() + + mem::size_of::() + + match value { + HandleChange::Present(did) => did.len(), + HandleChange::Removed => 0, + } + }) + .sum::(); + ActorIndexStats { + num_actors: state + .shards + .values() + .map(|shard| shard.handles.len() as u64) + .sum(), + total_bytes: (state + .shards + .values() + .map(|shard| shard.byte_size) + .sum::() + + change_bytes) as u64, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn did(raw: &str) -> Did { + Did::new_owned(raw).unwrap() + } + + fn handle(raw: &str) -> Handle { + Handle::new_owned(raw).unwrap() + } + + #[tokio::test] + async fn exact_results_come_before_typo_matches_and_survive_a_rebuild() { + let index = ActorIndex::new(); + index.identity_changed( + &did("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa"), + None, + &handle("dawn.example.com"), + ); + index.identity_changed( + &did("did:plc:bbbbbbbbbbbbbbbbbbbbbbbb"), + None, + &handle("fawn.example.com"), + ); + let before = index.suggest("dawn", 10).await.unwrap(); + assert_eq!( + before + .iter() + .map(|actor| actor.handle.as_str()) + .collect::>(), + vec!["dawn.example.com", "fawn.example.com"] + ); + index.rebuild().unwrap(); + assert_eq!(index.suggest("dawn", 10).await.unwrap(), before); + } + + #[tokio::test] + async fn rebuild_keeps_changes_that_arrive_while_it_runs() { + let index = ActorIndex::new(); + index.identity_changed( + &did("did:web:actor.example.com"), + None, + &handle("actor.example.com"), + ); + let (frozen_tx, frozen_rx) = std::sync::mpsc::channel(); + let (resume_tx, resume_rx) = std::sync::mpsc::channel(); + let index_being_rebuilt = index.clone(); + let rebuilding = std::thread::spawn(move || { + index_being_rebuilt + .rebuild_after_freeze(move || { + frozen_tx.send(()).unwrap(); + resume_rx.recv().unwrap(); + }) + .unwrap() + }); + frozen_rx.recv().unwrap(); + index.identity_changed( + &did("did:web:late.example.com"), + None, + &handle("late.example.com"), + ); + resume_tx.send(()).unwrap(); + rebuilding.join().unwrap(); + + let suggestions = index.suggest("late", 10).await.unwrap(); + assert_eq!(suggestions.len(), 1); + assert_eq!(suggestions[0].handle, "late.example.com"); + } + #[tokio::test] + async fn earlier_removals_do_not_hide_later_changes_at_small_limits() { + let index = ActorIndex::new(); + let actor = did("did:web:a1.example.com"); + let indexed = handle("a1.example.com"); + index.identity_changed(&actor, None, &indexed); + index.rebuild().unwrap(); + + index.identity_removed( + &did("did:web:a0.example.com"), + Some(&handle("a0.example.com")), + ); + index.identity_removed(&actor, Some(&indexed)); + index.identity_changed( + &did("did:web:a2.example.com"), + None, + &handle("a2.example.com"), + ); + + assert_eq!( + index.suggest("a", 1).await.unwrap()[0].handle, + "a2.example.com" + ); + } + + #[tokio::test] + async fn rename_deactivation_and_invalid_handle_override_indexed_values() { + let index = ActorIndex::new(); + let actor = did("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa"); + let old = handle("old.example.com"); + let new = handle("new.example.com"); + index.identity_changed(&actor, None, &old); + index.rebuild().unwrap(); + index.identity_changed(&actor, Some(&old), &new); + assert!(index.suggest("old", 10).await.unwrap().is_empty()); + assert_eq!(index.suggest("new", 10).await.unwrap()[0].did, actor); + index.identity_removed(&actor, Some(&new)); + assert!(index.suggest("new", 10).await.unwrap().is_empty()); + index.identity_changed(&actor, None, &handle(INVALID_HANDLE)); + assert!(index.suggest("handle", 10).await.unwrap().is_empty()); + } + + #[tokio::test] + async fn rebuild_replaces_only_dirty_shards() { + let index = ActorIndex::new(); + let first = did("did:web:first.example.com"); + let second = did("did:web:second.example.com"); + let untouched = did("did:web:untouched.example.com"); + let old = handle("aa.example.com"); + let new = handle("ac.example.com"); + index.identity_changed(&first, None, &old); + index.identity_changed(&second, None, &handle("ab.example.com")); + index.identity_changed(&untouched, None, &handle("ba.example.com")); + index.rebuild().unwrap(); + + let (ab_before, ba_before) = { + let state = index.inner.state.read(); + (state.shards[b"ab"].clone(), state.shards[b"ba"].clone()) + }; + index.identity_changed(&first, Some(&old), &new); + index.rebuild().unwrap(); + + { + let state = index.inner.state.read(); + assert!(Arc::ptr_eq(&ab_before, &state.shards[b"ab"])); + assert!(Arc::ptr_eq(&ba_before, &state.shards[b"ba"])); + } + assert_eq!( + index + .suggest("a", 10) + .await + .unwrap() + .into_iter() + .map(|actor| actor.handle) + .collect::>(), + ["ab.example.com", "ac.example.com"] + ); + } + + #[test] + fn did_table_round_trips_plc_and_web_dids() { + let plc = "did:plc:abcdefghijklmnopqrstuvwx"; + let web = "did:web:example.com"; + let mut dids = DidTable::default(); + let plc_index = dids.insert(plc).unwrap(); + assert!(dids.text.is_empty(), "did:plc stays in one fixed entry"); + let web_index = dids.insert(web).unwrap(); + assert_eq!(dids.text, web, "did:web uses the text side table"); + assert_eq!(dids.get(plc_index).unwrap(), plc); + assert_eq!(dids.get(web_index).unwrap(), web); + + let mut copied = DidTable::default(); + assert_eq!(copied.copy_from(&dids, plc_index).unwrap(), 0); + assert_eq!(copied.copy_from(&dids, web_index).unwrap(), 1); + assert_eq!(copied.get(0).unwrap(), plc); + assert_eq!(copied.get(1).unwrap(), web); + } +} diff --git a/bobbin/crates/search/src/lib.rs b/bobbin/crates/search/src/lib.rs index 58858595..bca8ca97 100644 --- a/bobbin/crates/search/src/lib.rs +++ b/bobbin/crates/search/src/lib.rs @@ -1,3 +1,7 @@ +mod actor; + +pub use actor::{ActorIndex, ActorIndexError, ActorIndexStats, ActorSuggestion}; + use std::future::Future; use std::ops::Bound; use std::pin::Pin; @@ -303,43 +307,7 @@ impl Inner { limit: usize, ) -> Result { let parsed = self.parser.parse_query(q)?; - let final_query: Box = if filters.is_empty() { - parsed - } else { - let mut clauses: Vec<(Occur, Box)> = Vec::with_capacity(5); - clauses.push((Occur::Must, parsed)); - if let Some(n) = &filters.nsid { - let term = Term::from_field_text(self.fields.nsid, n.as_ref()); - clauses.push(( - Occur::Must, - Box::new(TermQuery::new(term, IndexRecordOption::Basic)), - )); - } - if let Some(a) = &filters.author { - let term = Term::from_field_text(self.fields.author, a.as_ref()); - clauses.push(( - Occur::Must, - Box::new(TermQuery::new(term, IndexRecordOption::Basic)), - )); - } - if let Some(r) = &filters.repo { - let term = Term::from_field_text(self.fields.repo, r.as_ref()); - clauses.push(( - Occur::Must, - Box::new(TermQuery::new(term, IndexRecordOption::Basic)), - )); - } - if filters.since.is_some() || filters.until.is_some() { - let lower = filters.since.map_or(Bound::Unbounded, |s| { - Bound::Included(Term::from_field_i64(self.fields.created_at, s)) - }); - let upper = filters.until.map_or(Bound::Unbounded, |u| { - Bound::Excluded(Term::from_field_i64(self.fields.created_at, u)) - }); - clauses.push((Occur::Must, Box::new(RangeQuery::new(lower, upper)))); - } - Box::new(BooleanQuery::new(clauses)) - }; + let final_query = self.filtered_query(parsed, filters); let collector = TopDocs::with_limit(limit + 1) .and_offset(offset) .order_by_score(); @@ -354,9 +322,9 @@ impl Inner { let doc: TantivyDocument = searcher.doc(*addr)?; let uri_str = stored_text(&doc, self.fields.uri, "uri")?; let nsid_str = stored_text(&doc, self.fields.nsid, "nsid")?; - let uri = AtUri::::new_owned(uri_str.as_str()) + let uri = AtUri::::new_owned(uri_str) .map_err(|e| SearchError::InvalidUri(format!("{e}: {uri_str}")))?; - let nsid = Nsid::::new_owned(nsid_str.as_str()) + let nsid = Nsid::::new_owned(nsid_str) .map_err(|e| SearchError::InvalidNsid(format!("{e}: {nsid_str}")))?; Ok::<_, SearchError>(SearchHit { uri, @@ -370,6 +338,46 @@ impl Inner { next: next_cursor(offset, limit, has_more), }) } + + fn filtered_query(&self, query: Box, filters: &SearchFilters) -> Box { + if filters.is_empty() { + return query; + } + let mut clauses: Vec<(Occur, Box)> = Vec::with_capacity(5); + clauses.push((Occur::Must, query)); + for (field, value) in [ + ( + self.fields.nsid, + filters.nsid.as_ref().map(|value| value.as_ref()), + ), + ( + self.fields.author, + filters.author.as_ref().map(|value| value.as_ref()), + ), + ( + self.fields.repo, + filters.repo.as_ref().map(|value| value.as_ref()), + ), + ] { + if let Some(value) = value { + let term = Term::from_field_text(field, value); + clauses.push(( + Occur::Must, + Box::new(TermQuery::new(term, IndexRecordOption::Basic)), + )); + } + } + if filters.since.is_some() || filters.until.is_some() { + let lower = filters.since.map_or(Bound::Unbounded, |s| { + Bound::Included(Term::from_field_i64(self.fields.created_at, s)) + }); + let upper = filters.until.map_or(Bound::Unbounded, |u| { + Bound::Excluded(Term::from_field_i64(self.fields.created_at, u)) + }); + clauses.push((Occur::Must, Box::new(RangeQuery::new(lower, upper)))); + } + Box::new(BooleanQuery::new(clauses)) + } } fn next_cursor(offset: usize, limit: usize, has_more: bool) -> Option { @@ -475,13 +483,13 @@ fn commit_and_reload(writer: &mut IndexWriter, reader: &IndexReader) { } } -fn stored_text( - doc: &TantivyDocument, +fn stored_text<'a>( + doc: &'a TantivyDocument, field: Field, name: &'static str, -) -> Result { +) -> Result<&'a str, SearchError> { doc.get_first(field) - .and_then(|v| v.as_str().map(|s| s.to_owned())) + .and_then(|value| value.as_str()) .ok_or(SearchError::MissingField(name)) } diff --git a/bobbin/crates/types/src/identity.rs b/bobbin/crates/types/src/identity.rs new file mode 100644 index 00000000..58bd7234 --- /dev/null +++ b/bobbin/crates/types/src/identity.rs @@ -0,0 +1,34 @@ +use jacquard_common::DefaultStr; +use jacquard_common::types::did::Did; +use jacquard_common::types::string::Handle; + +pub trait IdentitySink: Send + Sync { + fn identity_changed( + &self, + did: &Did, + previous_handle: Option<&Handle>, + handle: &Handle, + ); + + fn identity_removed(&self, did: &Did, previous_handle: Option<&Handle>); +} + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct NoopIdentitySink; + +impl IdentitySink for NoopIdentitySink { + fn identity_changed( + &self, + _did: &Did, + _previous_handle: Option<&Handle>, + _handle: &Handle, + ) { + } + + fn identity_removed( + &self, + _did: &Did, + _previous_handle: Option<&Handle>, + ) { + } +} diff --git a/bobbin/crates/types/src/lib.rs b/bobbin/crates/types/src/lib.rs index 11b564b2..a010510e 100644 --- a/bobbin/crates/types/src/lib.rs +++ b/bobbin/crates/types/src/lib.rs @@ -4,6 +4,7 @@ extern crate alloc; pub use lexicons::*; pub mod edges; +pub mod identity; pub mod ids; pub mod knot_acl; pub mod legacy; diff --git a/bobbin/crates/xrpc/src/lib.rs b/bobbin/crates/xrpc/src/lib.rs index 9cb71966..5e6b6308 100644 --- a/bobbin/crates/xrpc/src/lib.rs +++ b/bobbin/crates/xrpc/src/lib.rs @@ -35,12 +35,14 @@ use bobbin_record_lru::RecordStore; use bobbin_resolver::{IdentityResolveError, IdentityResolver, RepoIdResolver}; use bobbin_runtime::ReqwestHttp; use bobbin_search::{ - SearchCursor, SearchError, SearchFilters, SearchHit, SearchOffset, SearchReader, + ActorIndex, ActorIndexError, SearchCursor, SearchError, SearchFilters, SearchHit, SearchOffset, + SearchReader, }; use bobbin_slingshot_client::{SlingshotClient, SlingshotError}; use bobbin_types::edges::REPO_SOURCE_EDGE_KIND; use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static, owner_did_from_aturi}; use bobbin_types::knot_acl::{KnotOwnedSource, decode_knot_owned_source, knot_did_host}; +use bobbin_types::org_tangled::feed::subscription::SubscriptionRecord; use bobbin_types::record::RecordBody; use bobbin_types::search::SearchableRecord; use bobbin_types::sh_tangled::actor::profile::{Profile, ProfileGetRecordOutput, ProfileRecord}; @@ -84,7 +86,6 @@ use bobbin_types::sh_tangled::spindle::{Spindle, SpindleRecord}; use bobbin_types::sh_tangled::string::{ TangledString, TangledStringGetRecordOutput, TangledStringRecord, }; -use bobbin_types::org_tangled::feed::subscription::SubscriptionRecord; use futures::Stream; use futures::stream::{self, StreamExt, TryStreamExt}; @@ -131,6 +132,9 @@ use filter::{ use trusted_proxies::TrustedProxies; const DEFAULT_LIMIT: u32 = 50; +const ACTOR_TYPEAHEAD_DEFAULT_LIMIT: u32 = 10; +const ACTOR_TYPEAHEAD_MAX_LIMIT: u32 = 100; +const ACTOR_TYPEAHEAD_MAX_QUERY_BYTES: usize = 253; const FETCH_CONCURRENCY: usize = 8; const TANGLED_NSID_PREFIX: &str = "sh.tangled."; @@ -154,6 +158,7 @@ pub struct AppState { pub search: Arc, /// Zoekt-backed code search. `None` when unconfigured, and the endpoint answers 501. pub codesearch: Option>, + pub actors: Arc, pub resolver: Arc, pub identity: Arc, pub directory: Arc, @@ -197,6 +202,7 @@ impl AppState { mirror_v2: None, search, codesearch: None, + actors: Arc::new(ActorIndex::new()), resolver, identity, directory: directory.clone(), @@ -213,6 +219,10 @@ impl AppState { } } + pub fn with_actors(mut self, actors: Arc) -> Self { + self.actors = actors; + self + } pub fn with_limiter(mut self, limiter: Option>) -> Self { self.limiter = limiter; self @@ -548,6 +558,10 @@ pub fn router(state: AppState) -> Router { "/xrpc/sh.tangled.query.enrichResponse", axum::routing::post(enrich::enrich), ) + .route( + "/xrpc/sh.tangled.actor.searchActorsTypeahead", + get(search_actors_typeahead), + ) .route("/xrpc/sh.tangled.bobbin.getCoverage", get(get_coverage)) .route("/xrpc/sh.tangled.bobbin.awaitRecord", get(await_record)) .route( @@ -862,6 +876,12 @@ struct SearchQueryParams { limit: Option, } +#[derive(Debug, Deserialize)] +struct ActorTypeaheadParams { + q: String, + limit: Option, +} + pub struct XrpcQuery(pub T); impl FromRequestParts for XrpcQuery @@ -1099,6 +1119,18 @@ struct SearchHitView { value: SearchableRecord, } +#[derive(Serialize)] +struct ActorTypeaheadResponse { + actors: Vec, +} + +#[derive(Serialize)] +struct ActorTypeaheadView { + uri: AtUri, + did: Did, + handle: String, +} + fn map_slingshot(err: SlingshotError) -> XrpcError { use SlingshotError as E; match err { @@ -1205,9 +1237,9 @@ fn parse_subject(raw: &SubjectQuery, shape: SubjectShape) -> Result { return match shape { - SubjectShape::BareDid | SubjectShape::BareDidOrOneOfCollections(_) | SubjectShape::BareDidOrAnyAtUri => { - Ok(SubjectRef::Did(did.clone())) - } + SubjectShape::BareDid + | SubjectShape::BareDidOrOneOfCollections(_) + | SubjectShape::BareDidOrAnyAtUri => Ok(SubjectRef::Did(did.clone())), SubjectShape::Collection(expected) => Err(XrpcError::InvalidParams(format!( "subject must be at:///{expected}/, got bare did" ))), @@ -1260,7 +1292,9 @@ fn parse_subject(raw: &SubjectQuery, shape: SubjectShape) -> Result// with nsid in [{}], got collection {c}", allowed.join(", "), ))), - SubjectShape::AnyAtUri | SubjectShape::BareDidOrAnyAtUri => Ok(SubjectRef::Uri(uri.clone())), + SubjectShape::AnyAtUri | SubjectShape::BareDidOrAnyAtUri => { + Ok(SubjectRef::Uri(uri.clone())) + } } } @@ -2190,7 +2224,6 @@ async fn get_subscription_for_actor( get_for::(&state, q).map(Json) } - async fn list_follows( State(state): State, XrpcQuery(q): XrpcQuery>, @@ -3017,6 +3050,51 @@ async fn list_recipients( Ok(Json(ListRecipientsResponse { dids })) } +async fn search_actors_typeahead( + State(state): State, + XrpcQuery(q): XrpcQuery, +) -> Result, XrpcError> { + let query = q.q.trim().trim_start_matches('@'); + if query.is_empty() { + return Err(XrpcError::InvalidParams("q must not be empty".into())); + } + if query.len() > ACTOR_TYPEAHEAD_MAX_QUERY_BYTES { + return Err(XrpcError::InvalidParams(format!( + "q must be at most {ACTOR_TYPEAHEAD_MAX_QUERY_BYTES} bytes", + ))); + } + let limit = q.limit.unwrap_or(ACTOR_TYPEAHEAD_DEFAULT_LIMIT); + if !(1..=ACTOR_TYPEAHEAD_MAX_LIMIT).contains(&limit) { + return Err(XrpcError::InvalidParams(format!( + "limit must be between 1 and {ACTOR_TYPEAHEAD_MAX_LIMIT}", + ))); + } + let _permit = state.heavy_permit()?; + let suggestions = state + .actors + .suggest(query, limit) + .await + .map_err(map_actor_search_err)?; + let actors = suggestions + .into_iter() + .map(|suggestion| { + let uri = AtUri::new_owned(format!( + "at://{}/sh.tangled.actor.profile/self", + suggestion.did + )) + .map_err(|error| { + XrpcError::Internal(format!("could not build profile URI: {error}")) + })?; + Ok(ActorTypeaheadView { + uri, + did: suggestion.did, + handle: suggestion.handle, + }) + }) + .collect::, XrpcError>>()?; + Ok(Json(ActorTypeaheadResponse { actors })) +} + async fn search_query( State(state): State, XrpcQuery(q): XrpcQuery, @@ -3130,6 +3208,9 @@ fn map_search_err(err: SearchError) -> XrpcError { | E::Cancelled(_)) => XrpcError::Internal(format!("search: {e}")), } } +fn map_actor_search_err(err: ActorIndexError) -> XrpcError { + XrpcError::Internal(format!("actor search: {err}")) +} fn map_proxy_error(err: KnotProxyError) -> XrpcError { match err { diff --git a/bobbin/crates/xrpc/tests/search.rs b/bobbin/crates/xrpc/tests/search.rs index a50cc8d9..0d08c2f0 100644 --- a/bobbin/crates/xrpc/tests/search.rs +++ b/bobbin/crates/xrpc/tests/search.rs @@ -6,15 +6,17 @@ 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::ActorIndex; use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; use bobbin_slingshot_client::SlingshotClient; +use bobbin_types::identity::IdentitySink; use bobbin_types::search::{SearchDoc, SearchSink}; use bobbin_xrpc::{AppState, router}; use http::{Request, StatusCode}; use jacquard_common::DefaultStr; use jacquard_common::types::nsid::Nsid; use jacquard_common::types::recordkey::Rkey; -use jacquard_common::types::string::{AtUri, Did}; +use jacquard_common::types::string::{AtUri, Did, Handle}; use serde_json::{Value, json}; use tower::ServiceExt; use url::Url; @@ -31,6 +33,9 @@ fn at(s: &str) -> AtUri { fn did(s: &str) -> Did { Did::new_owned(s).unwrap() } +fn handle(s: &str) -> Handle { + Handle::new_owned(s).unwrap() +} fn rkey(s: &str) -> Rkey { Rkey::new_owned(s).unwrap() @@ -48,6 +53,7 @@ struct Harness { server: MockServer, coverage: Arc, search: Arc, + actors: Arc, state: AppState, } @@ -78,10 +84,13 @@ impl Harness { Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), Arc::new(bobbin_xrpc::default_directory()), ); + let actors = Arc::new(ActorIndex::new()); + let state = state.with_actors(actors.clone()); Self { server, coverage, search, + actors, state, } } @@ -264,6 +273,18 @@ fn search_request(extras: &[(&str, &str)]) -> Request { .unwrap() } +fn typeahead_request(extras: &[(&str, &str)]) -> Request { + let qs = extras + .iter() + .map(|(k, v)| format!("{k}={}", enc(v))) + .collect::>() + .join("&"); + Request::builder() + .uri(format!("/xrpc/sh.tangled.actor.searchActorsTypeahead?{qs}",)) + .body(Body::empty()) + .unwrap() +} + 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(); @@ -900,3 +921,132 @@ async fn search_drops_undecodable_hit_and_returns_others() { ), ); } + +#[tokio::test] +async fn actor_typeahead_matches_handles_without_hydration() { + let h = Harness::new().await; + for (owner, hdl) in [ + ("did:plc:dawn", "dawn.example.com"), + ("did:plc:fawn", "fawn.example.com"), + ] { + h.actors.identity_changed(&did(owner), None, &handle(hdl)); + } + h.search + .upsert(SearchDoc { + uri: at("at://did:plc:issue/sh.tangled.repo.issue/i1"), + nsid: nsid("sh.tangled.repo.issue"), + title: "dawn breaks search".to_owned(), + body: String::new(), + author: Some(did("did:plc:issue")), + created_at: None, + repo: None, + }) + .await; + h.search.flush().await; + + let app = router(h.state.clone()); + let response = app + .oneshot(typeahead_request(&[("q", "@DAWN")])) + .await + .unwrap(); + let (status, body) = json_response(response).await; + assert_eq!(status, StatusCode::OK); + let actors = body["actors"].as_array().expect("actors array"); + assert_eq!(actors.len(), 2, "profile nsid filter excludes issue"); + assert_eq!( + actors[0], + json!({ + "uri": "at://did:plc:dawn/sh.tangled.actor.profile/self", + "did": "did:plc:dawn", + "handle": "dawn.example.com", + }), + "exact prefix ranks before typo match", + ); + assert_eq!(actors[1]["did"], json!("did:plc:fawn")); + assert_eq!( + actors[1]["uri"], + json!("at://did:plc:fawn/sh.tangled.actor.profile/self") + ); + assert_eq!(actors[1]["handle"], json!("fawn.example.com")); +} + +#[tokio::test] +async fn actor_typeahead_handles_typos_and_rejects_invalid_bounds() { + let h = Harness::new().await; + h.actors + .identity_changed(&did("did:plc:dawn"), None, &handle("dawn.example.com")); + let app = router(h.state.clone()); + + let response = app + .clone() + .oneshot(typeahead_request(&[("q", "dwn")])) + .await + .unwrap(); + let (status, body) = json_response(response).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(body["actors"].as_array().unwrap().len(), 1); + assert_eq!(body["actors"][0]["handle"], json!("dawn.example.com")); + + for params in [ + [("q", " "), ("limit", "10")], + [("q", "dawn"), ("limit", "0")], + [("q", "dawn"), ("limit", "101")], + ] { + let response = app + .clone() + .oneshot(typeahead_request(¶ms)) + .await + .unwrap(); + let (status, body) = json_response(response).await; + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(body["error"], json!("InvalidRequest")); + } +} + +#[tokio::test] +async fn actor_typeahead_handles_rename_and_removal() { + let h = Harness::new().await; + let dawn = did("did:plc:dawn"); + let old_hdl = handle("dawn.example.com"); + let new_hdl = handle("dawn-renamed.example.com"); + let fawn = did("did:plc:fawn"); + let fawn_hdl = handle("fawn.example.com"); + + h.actors.identity_changed(&dawn, None, &old_hdl); + h.actors.identity_changed(&fawn, None, &fawn_hdl); + + h.actors.identity_changed(&dawn, Some(&old_hdl), &new_hdl); + h.actors.identity_removed(&fawn, Some(&fawn_hdl)); + + let app = router(h.state.clone()); + + let response = app + .clone() + .oneshot(typeahead_request(&[("q", "dawn-renamed")])) + .await + .unwrap(); + let (status, body) = json_response(response).await; + assert_eq!(status, StatusCode::OK); + let actors = body["actors"].as_array().unwrap(); + assert_eq!(actors.len(), 1); + assert_eq!(actors[0]["handle"], json!("dawn-renamed.example.com")); + assert_eq!(actors[0]["did"], json!("did:plc:dawn")); + assert_eq!( + actors[0]["uri"], + json!("at://did:plc:dawn/sh.tangled.actor.profile/self") + ); + + let response = app + .oneshot(typeahead_request(&[("q", "fawn")])) + .await + .unwrap(); + let (status, body) = json_response(response).await; + assert_eq!(status, StatusCode::OK); + assert!( + body["actors"] + .as_array() + .unwrap() + .iter() + .all(|actor| actor["did"] != json!("did:plc:fawn")), + ); +} diff --git a/lexicons/actor/searchActorsTypeahead.json b/lexicons/actor/searchActorsTypeahead.json new file mode 100644 index 00000000..677b1aff --- /dev/null +++ b/lexicons/actor/searchActorsTypeahead.json @@ -0,0 +1,63 @@ +{ + "lexicon": 1, + "id": "sh.tangled.actor.searchActorsTypeahead", + "defs": { + "main": { + "type": "query", + "description": "Find actors by their verified handle.", + "parameters": { + "type": "params", + "required": ["q"], + "properties": { + "q": { + "type": "string", + "minLength": 1, + "maxLength": 254, + "description": "The beginning of a handle. A leading @ is optional." + }, + "limit": { + "type": "integer", + "minimum": 1, + "maximum": 100, + "default": 10 + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["actors"], + "properties": { + "actors": { + "type": "array", + "items": { + "type": "ref", + "ref": "#actor" + } + } + } + } + } + }, + "actor": { + "type": "object", + "required": ["uri", "did", "handle"], + "properties": { + "uri": { + "type": "string", + "format": "at-uri", + "description": "AT-URI for the actor's Tangled profile." + }, + "did": { + "type": "string", + "format": "did" + }, + "handle": { + "type": "string", + "format": "handle" + } + } + } + } +}