alice.pds.demo.boltless.dev
This repository has no description
1use std::sync::Arc;
2use std::sync::atomic::{AtomicU64, Ordering};
3
4use bobbin_runtime::RuntimeHasher;
5use bobbin_slingshot_client::{SlingshotClient, SlingshotError};
6use jacquard_common::DefaultStr;
7use jacquard_common::types::did::Did;
8use jacquard_common::types::ident::AtIdentifier;
9use jacquard_common::types::string::Handle;
10use scc::hash_cache::Entry as CacheEntry;
11use scc::{HashCache as SccCache, HashMap as SccMap};
12use serde::{Deserialize, Serialize};
13use thiserror::Error;
14use tokio::sync::OnceCell;
15
16pub const DEFAULT_IDENTITY_CACHE_ENTRIES: usize = 100_000;
17
18#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
19#[serde(rename_all = "camelCase")]
20pub struct MiniDoc {
21 pub did: Did<DefaultStr>,
22 pub handle: Handle<DefaultStr>,
23 #[serde(skip_serializing_if = "Option::is_none")]
24 pub pds: Option<String>,
25}
26
27#[derive(Clone, Debug, Error, Eq, PartialEq)]
28pub enum IdentityResolveError {
29 #[error("identity not found")]
30 NotFound,
31 #[error("identity upstream: {0}")]
32 Upstream(String),
33 #[error("invalid identity response: {0}")]
34 Decode(String),
35}
36
37impl From<SlingshotError> for IdentityResolveError {
38 fn from(error: SlingshotError) -> Self {
39 match error {
40 SlingshotError::NotFound => Self::NotFound,
41 other => Self::Upstream(other.to_string()),
42 }
43 }
44}
45
46// hydrant doesnt send pds so we have a separate states to have
47// public resolve call upgrade from a partial-cached to a full-cached
48#[derive(Clone)]
49enum IdentityState {
50 // seen by bobbin from hydrant
51 Observed(MiniDoc),
52 // fetched by bobbin from slingshot
53 Fetched(MiniDoc),
54 Inactive,
55}
56
57impl IdentityState {
58 fn doc(&self) -> Option<&MiniDoc> {
59 match self {
60 Self::Observed(doc) | Self::Fetched(doc) => Some(doc),
61 Self::Inactive => None,
62 }
63 }
64}
65
66#[derive(Clone, Copy, Debug, Eq, PartialEq)]
67pub struct IdentityResolverStatsSnapshot {
68 pub entries: usize,
69 pub capacity: usize,
70 pub hits: u64,
71 pub misses: u64,
72 pub upstream_requests: u64,
73}
74
75#[derive(Default)]
76struct IdentityResolverStats {
77 hits: AtomicU64,
78 misses: AtomicU64,
79 upstream_requests: AtomicU64,
80}
81
82pub struct IdentityResolver {
83 by_did: SccCache<Did<DefaultStr>, IdentityState, RuntimeHasher>,
84 by_handle: SccMap<Handle<DefaultStr>, Did<DefaultStr>, RuntimeHasher>,
85 in_flight: SccMap<String, Arc<OnceCell<Result<MiniDoc, IdentityResolveError>>>, RuntimeHasher>,
86 slingshot: Option<SlingshotClient>,
87 stats: IdentityResolverStats,
88}
89
90impl IdentityResolver {
91 pub fn with_slingshot(
92 slingshot: SlingshotClient,
93 hasher: RuntimeHasher,
94 capacity: usize,
95 ) -> Self {
96 Self::new(Some(slingshot), hasher, capacity)
97 }
98
99 pub fn detached(hasher: RuntimeHasher, capacity: usize) -> Self {
100 Self::new(None, hasher, capacity)
101 }
102
103 fn new(slingshot: Option<SlingshotClient>, hasher: RuntimeHasher, capacity: usize) -> Self {
104 Self {
105 by_did: SccCache::with_capacity_and_hasher(0, capacity, hasher.clone()),
106 by_handle: SccMap::with_hasher(hasher.clone()),
107 in_flight: SccMap::with_hasher(hasher),
108 slingshot,
109 stats: IdentityResolverStats::default(),
110 }
111 }
112
113 pub fn stats(&self) -> IdentityResolverStatsSnapshot {
114 IdentityResolverStatsSnapshot {
115 entries: self.by_did.len(),
116 capacity: *self.by_did.capacity_range().end(),
117 hits: self.stats.hits.load(Ordering::Relaxed),
118 misses: self.stats.misses.load(Ordering::Relaxed),
119 upstream_requests: self.stats.upstream_requests.load(Ordering::Relaxed),
120 }
121 }
122
123 pub fn observe(&self, did: Did<DefaultStr>, handle: Handle<DefaultStr>) {
124 let mut evicted = None;
125 let mut previous_handle = None;
126 match self.by_did.entry_sync(did.clone()) {
127 CacheEntry::Occupied(mut occupied) => {
128 let (pds, fetched) = match occupied.get() {
129 IdentityState::Observed(previous) => {
130 previous_handle = Some(previous.handle.clone());
131 (previous.pds.clone(), false)
132 }
133 IdentityState::Fetched(previous) => {
134 previous_handle = Some(previous.handle.clone());
135 (previous.pds.clone(), previous.handle == handle)
136 }
137 IdentityState::Inactive => (None, false),
138 };
139 let doc = MiniDoc {
140 did: did.clone(),
141 handle: handle.clone(),
142 pds,
143 };
144 occupied.put(if fetched {
145 IdentityState::Fetched(doc)
146 } else {
147 IdentityState::Observed(doc)
148 });
149 }
150 CacheEntry::Vacant(vacant) => {
151 let (removed, occupied) = vacant.put_entry(IdentityState::Observed(MiniDoc {
152 did: did.clone(),
153 handle: handle.clone(),
154 pds: None,
155 }));
156 evicted = removed;
157 drop(occupied);
158 }
159 }
160 self.remove_by_handle_if_owned(&did, previous_handle.as_ref());
161 self.remove_by_handle_for_removed(evicted);
162 self.insert_by_handle(did, handle);
163 }
164
165 pub fn deactivate(&self, did: Did<DefaultStr>) {
166 let mut evicted = None;
167 let mut previous_handle = None;
168 match self.by_did.entry_sync(did.clone()) {
169 CacheEntry::Occupied(mut occupied) => {
170 previous_handle = occupied.get().doc().map(|doc| doc.handle.clone());
171 occupied.put(IdentityState::Inactive);
172 }
173 CacheEntry::Vacant(vacant) => {
174 let (removed, occupied) = vacant.put_entry(IdentityState::Inactive);
175 evicted = removed;
176 drop(occupied);
177 }
178 }
179 self.remove_by_handle_if_owned(&did, previous_handle.as_ref());
180 self.remove_by_handle_for_removed(evicted);
181 }
182
183 fn remove_by_handle_if_owned(
184 &self,
185 did: &Did<DefaultStr>,
186 handle: Option<&Handle<DefaultStr>>,
187 ) {
188 let Some(handle) = handle else {
189 return;
190 };
191 self.by_handle.remove_if_sync(handle, |owner| owner == did);
192 }
193
194 fn remove_by_handle_for_removed(&self, removed: Option<(Did<DefaultStr>, IdentityState)>) {
195 let Some((did, state)) = removed else {
196 return;
197 };
198 let Some(doc) = state.doc() else {
199 return;
200 };
201 self.remove_by_handle_if_owned(&did, Some(&doc.handle));
202 }
203
204 fn insert_by_handle(&self, did: Did<DefaultStr>, handle: Handle<DefaultStr>) {
205 if let Some(displaced_did) = self
206 .by_handle
207 .upsert_sync(handle.clone(), did.clone())
208 .filter(|displaced_did| displaced_did != &did)
209 {
210 let removed = self.by_did.remove_if_sync(&displaced_did, |state| {
211 state.doc().is_some_and(|doc| doc.handle == handle)
212 });
213 self.remove_by_handle_for_removed(removed);
214 }
215
216 if self.by_did_matches_handle(&did, &handle) {
217 return;
218 }
219 self.remove_by_handle_if_owned(&did, Some(&handle));
220 }
221
222 fn by_did_matches_handle(&self, did: &Did<DefaultStr>, handle: &Handle<DefaultStr>) -> bool {
223 self.by_did
224 .get_sync(did)
225 .is_some_and(|state| state.get().doc().is_some_and(|doc| doc.handle == *handle))
226 }
227
228 fn cached(
229 &self,
230 identifier: &AtIdentifier<DefaultStr>,
231 require_fetched: bool,
232 ) -> Result<Option<MiniDoc>, IdentityResolveError> {
233 match identifier {
234 AtIdentifier::Did(did) => match self.by_did.get_sync(did).as_deref() {
235 Some(IdentityState::Observed(doc)) => Ok((!require_fetched).then(|| doc.clone())),
236 Some(IdentityState::Fetched(doc)) => Ok(Some(doc.clone())),
237 Some(IdentityState::Inactive) => Err(IdentityResolveError::NotFound),
238 None => Ok(None),
239 },
240 AtIdentifier::Handle(handle) => {
241 let Some(did) = self
242 .by_handle
243 .get_sync(handle)
244 .map(|entry| entry.get().clone())
245 else {
246 return Ok(None);
247 };
248 let state = self.by_did.get_sync(&did).map(|state| state.get().clone());
249 match state {
250 Some(IdentityState::Observed(doc)) if doc.handle == *handle => {
251 return Ok((!require_fetched).then_some(doc));
252 }
253 Some(IdentityState::Fetched(doc)) if doc.handle == *handle => {
254 return Ok(Some(doc));
255 }
256 Some(IdentityState::Observed(_))
257 | Some(IdentityState::Fetched(_))
258 | Some(IdentityState::Inactive)
259 | None => {}
260 }
261 self.remove_by_handle_if_owned(&did, Some(handle));
262 Ok(None)
263 }
264 }
265 }
266
267 /// Resolve a DID using Hydrant's partial identity data when available.
268 pub async fn resolve_by_did(
269 &self,
270 did: &Did<DefaultStr>,
271 ) -> Result<MiniDoc, IdentityResolveError> {
272 self.resolve_with_cache(&AtIdentifier::Did(did.clone()), false)
273 .await
274 }
275
276 /// Resolve a minidoc, fetching Hydrant-only observations upstream first.
277 pub async fn resolve_minidoc(
278 &self,
279 identifier: &AtIdentifier<DefaultStr>,
280 ) -> Result<MiniDoc, IdentityResolveError> {
281 self.resolve_with_cache(identifier, true).await
282 }
283
284 async fn resolve_with_cache(
285 &self,
286 identifier: &AtIdentifier<DefaultStr>,
287 require_fetched: bool,
288 ) -> Result<MiniDoc, IdentityResolveError> {
289 match self.cached(identifier, require_fetched) {
290 Ok(Some(doc)) => {
291 self.stats.hits.fetch_add(1, Ordering::Relaxed);
292 return Ok(doc);
293 }
294 Err(error) => {
295 self.stats.hits.fetch_add(1, Ordering::Relaxed);
296 return Err(error);
297 }
298 Ok(None) => self.stats.misses.fetch_add(1, Ordering::Relaxed),
299 };
300
301 let key = identifier.as_str().to_owned();
302 let cell = self
303 .in_flight
304 .entry_async(key.clone())
305 .await
306 .or_insert_with(|| Arc::new(OnceCell::new()))
307 .get()
308 .clone();
309 let result = cell
310 .get_or_init(|| async { self.fetch_minidoc(identifier).await })
311 .await
312 .clone();
313 self.in_flight.remove_async(&key).await;
314 result
315 }
316
317 async fn fetch_minidoc(
318 &self,
319 identifier: &AtIdentifier<DefaultStr>,
320 ) -> Result<MiniDoc, IdentityResolveError> {
321 let client = self
322 .slingshot
323 .as_ref()
324 .ok_or(IdentityResolveError::NotFound)?;
325 self.stats.upstream_requests.fetch_add(1, Ordering::Relaxed);
326 let bytes = client
327 .resolve_mini_doc(identifier)
328 .await
329 .map_err(IdentityResolveError::from)?;
330 let doc = serde_json::from_slice::<MiniDoc>(&bytes)
331 .map_err(|error| IdentityResolveError::Decode(error.to_string()))?;
332 self.insert_fetched_by_did(doc)
333 }
334
335 fn insert_fetched_by_did(&self, doc: MiniDoc) -> Result<MiniDoc, IdentityResolveError> {
336 let did = doc.did.clone();
337 let handle = doc.handle.clone();
338 let mut evicted = None;
339 let mut previous_handle = None;
340 let stored = match self.by_did.entry_sync(did.clone()) {
341 CacheEntry::Occupied(mut occupied) => match occupied.get_mut() {
342 IdentityState::Inactive => return Err(IdentityResolveError::NotFound),
343 IdentityState::Observed(observed) if observed.handle != handle => {
344 observed.pds = doc.pds;
345 return Ok(observed.clone());
346 }
347 IdentityState::Observed(previous) | IdentityState::Fetched(previous) => {
348 previous_handle = Some(previous.handle.clone());
349 occupied.put(IdentityState::Fetched(doc.clone()));
350 doc
351 }
352 },
353 CacheEntry::Vacant(vacant) => {
354 let (removed, occupied) = vacant.put_entry(IdentityState::Fetched(doc.clone()));
355 evicted = removed;
356 drop(occupied);
357 doc
358 }
359 };
360 self.remove_by_handle_if_owned(&did, previous_handle.as_ref());
361 self.remove_by_handle_for_removed(evicted);
362 self.insert_by_handle(did, handle);
363 Ok(stored)
364 }
365}
366
367#[cfg(test)]
368mod tests {
369 use super::*;
370 use url::Url;
371 use wiremock::matchers::{method, path, query_param};
372 use wiremock::{Mock, MockServer, ResponseTemplate};
373
374 const TEST_CAPACITY: usize = 64;
375
376 fn hasher() -> RuntimeHasher {
377 RuntimeHasher::from_seeds(1, 2, 3, 4)
378 }
379
380 fn resolver() -> IdentityResolver {
381 IdentityResolver::detached(hasher(), TEST_CAPACITY)
382 }
383
384 fn did(value: &str) -> Did<DefaultStr> {
385 Did::new_owned(value).unwrap()
386 }
387
388 fn handle(value: &str) -> Handle<DefaultStr> {
389 Handle::new_owned(value).unwrap()
390 }
391
392 #[tokio::test]
393 async fn observed_identity_resolves_by_did_without_upstream() {
394 let resolver = resolver();
395 resolver.observe(did("did:plc:dawn"), handle("ptr.pet"));
396
397 let doc = resolver.resolve_by_did(&did("did:plc:dawn")).await.unwrap();
398 assert_eq!(doc.handle, handle("ptr.pet"));
399 assert_eq!(doc.pds, None);
400 assert_eq!(resolver.stats().hits, 1);
401 }
402
403 #[tokio::test]
404 async fn observed_identity_updates_existing_did_and_removes_old_handle() {
405 let resolver = resolver();
406 let identity = did("did:plc:dawn");
407 resolver.observe(identity.clone(), handle("ptr.pet"));
408 resolver.observe(identity.clone(), handle("new.ptr.pet"));
409
410 assert!(
411 resolver
412 .cached(&AtIdentifier::Handle(handle("ptr.pet")), false)
413 .unwrap()
414 .is_none()
415 );
416 let updated = resolver.resolve_by_did(&identity).await.unwrap();
417 assert_eq!(updated.handle, handle("new.ptr.pet"));
418 }
419
420 #[tokio::test]
421 async fn handle_reassignment_keeps_the_new_owner() {
422 let resolver = resolver();
423 let first = did("did:plc:first");
424 let second = did("did:plc:second");
425 let shared = handle("shared.example.com");
426 resolver.observe(first.clone(), shared.clone());
427 resolver.observe(second.clone(), shared.clone());
428 resolver.observe(first, handle("first.example.com"));
429
430 let cached = resolver
431 .cached(&AtIdentifier::Handle(shared), false)
432 .unwrap()
433 .unwrap();
434 assert_eq!(cached.did, second);
435 }
436
437 #[test]
438 fn handle_lookup_discards_an_unvalidated_reverse_hint() {
439 let resolver = resolver();
440 let identity = did("did:plc:dawn");
441 let stale = handle("stale.example.com");
442 resolver.observe(identity.clone(), handle("current.example.com"));
443 resolver
444 .by_handle
445 .upsert_sync(stale.clone(), identity.clone());
446
447 assert!(
448 resolver
449 .cached(&AtIdentifier::Handle(stale.clone()), false)
450 .unwrap()
451 .is_none()
452 );
453 assert!(resolver.by_handle.get_sync(&stale).is_none());
454 }
455
456 #[test]
457 fn minidoc_lookup_keeps_a_valid_observed_handle_hint() {
458 let resolver = resolver();
459 let identity = did("did:plc:dawn");
460 let handle = handle("ptr.pet");
461 resolver.observe(identity.clone(), handle.clone());
462
463 assert!(
464 resolver
465 .cached(&AtIdentifier::Handle(handle.clone()), true)
466 .unwrap()
467 .is_none()
468 );
469 assert_eq!(
470 resolver
471 .by_handle
472 .get_sync(&handle)
473 .map(|entry| entry.get().clone()),
474 Some(identity)
475 );
476 }
477
478 #[tokio::test]
479 async fn partial_observation_fetches_and_preserves_pds() {
480 let server = MockServer::start().await;
481 Mock::given(method("GET"))
482 .and(path("/xrpc/com.bad-example.identity.resolveMiniDoc"))
483 .and(query_param("identifier", "did:plc:dawn"))
484 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
485 "did": "did:plc:dawn",
486 "handle": "ptr.pet",
487 "pds": "https://pds.example.com"
488 })))
489 .expect(1)
490 .mount(&server)
491 .await;
492 let client =
493 SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap();
494 let resolver = IdentityResolver::with_slingshot(client, hasher(), TEST_CAPACITY);
495 let identity = did("did:plc:dawn");
496 resolver.observe(identity.clone(), handle("ptr.pet"));
497
498 let doc = resolver
499 .resolve_minidoc(&AtIdentifier::Did(identity.clone()))
500 .await
501 .unwrap();
502 assert_eq!(doc.pds.as_deref(), Some("https://pds.example.com"));
503
504 resolver.observe(identity.clone(), handle("ptr.pet"));
505 let cached = resolver
506 .resolve_minidoc(&AtIdentifier::Did(identity))
507 .await
508 .unwrap();
509 assert_eq!(cached.pds.as_deref(), Some("https://pds.example.com"));
510 assert_eq!(resolver.stats().upstream_requests, 1);
511 }
512
513 #[tokio::test]
514 async fn inactive_identity_rejects_an_in_flight_result() {
515 let resolver = resolver();
516 let identity = did("did:plc:dawn");
517 resolver.observe(identity.clone(), handle("ptr.pet"));
518 resolver.deactivate(identity.clone());
519
520 assert_eq!(
521 resolver.resolve_by_did(&identity).await,
522 Err(IdentityResolveError::NotFound)
523 );
524 assert_eq!(
525 resolver.insert_fetched_by_did(MiniDoc {
526 did: identity,
527 handle: handle("ptr.pet"),
528 pds: Some("https://pds.example.com".to_owned()),
529 }),
530 Err(IdentityResolveError::NotFound)
531 );
532 assert!(
533 resolver
534 .cached(&AtIdentifier::Handle(handle("ptr.pet")), false)
535 .unwrap()
536 .is_none()
537 );
538 }
539
540 #[test]
541 fn cache_capacity_bounds_forward_and_reverse_indexes() {
542 let resolver = resolver();
543 for n in 0..512 {
544 resolver.observe(
545 did(&format!("did:plc:user{n}")),
546 handle(&format!("user{n}.example.com")),
547 );
548 }
549
550 let stats = resolver.stats();
551 assert!(stats.entries <= stats.capacity);
552 assert!(resolver.by_handle.len() <= stats.capacity);
553 }
554}