From 3c46e5fc73d180bd6eb64b230358512dcc279e95 Mon Sep 17 00:00:00 2001 From: Lewis Date: Sun, 12 Jul 2026 00:21:32 +0300 Subject: [PATCH] lexicon: Nsid instead of strs for collections Lewis: May this revision serve well! --- crates/tranquil-api/src/identity/account.rs | 1 + crates/tranquil-api/src/moderation/mod.rs | 2 + crates/tranquil-api/src/repo/blob.rs | 2 +- crates/tranquil-api/src/repo/record/batch.rs | 6 +- .../src/repo/record/validation.rs | 6 +- .../src/server/passkey_account.rs | 1 + .../tranquil-api/src/server/service_auth.rs | 20 +- crates/tranquil-auth/src/token.rs | 4 +- crates/tranquil-auth/src/types.rs | 2 +- crates/tranquil-lexicon/src/dynamic.rs | 174 ++++++++++-------- crates/tranquil-lexicon/src/registry.rs | 51 +++-- crates/tranquil-lexicon/src/resolve.rs | 58 +++--- crates/tranquil-lexicon/src/schema.rs | 3 +- crates/tranquil-lexicon/src/validate.rs | 69 +++---- .../tests/resolve_integration.rs | 65 +++++-- crates/tranquil-pds/src/api/proxy.rs | 12 +- crates/tranquil-pds/src/auth/extractor.rs | 10 +- crates/tranquil-pds/src/auth/mod.rs | 10 +- crates/tranquil-pds/src/auth/scope_check.rs | 5 +- .../tranquil-pds/src/auth/scope_verified.rs | 20 +- crates/tranquil-pds/src/auth/service.rs | 2 +- crates/tranquil-pds/src/sync/import.rs | 4 +- crates/tranquil-pds/src/validation/mod.rs | 11 +- crates/tranquil-pds/tests/jwt_security.rs | 1 + .../tranquil-pds/tests/record_validation.rs | 5 + crates/tranquil-pds/tests/repo_batch.rs | 2 +- crates/tranquil-pds/tests/scope_edge_cases.rs | 11 +- crates/tranquil-scopes/src/permission_set.rs | 53 +++--- crates/tranquil-scopes/src/permissions.rs | 43 +++-- .../src/metastore/record_ops.rs | 18 +- .../tranquil-store/src/metastore/records.rs | 29 ++- .../tranquil-store/src/metastore/recovery.rs | 13 +- 32 files changed, 427 insertions(+), 286 deletions(-) diff --git a/crates/tranquil-api/src/identity/account.rs b/crates/tranquil-api/src/identity/account.rs index 9c520c4..3e1447f 100644 --- a/crates/tranquil-api/src/identity/account.rs +++ b/crates/tranquil-api/src/identity/account.rs @@ -210,6 +210,7 @@ pub async fn create_account( let token = extracted.token; if is_service_token(&token) { let verifier = ServiceTokenVerifier::new(); + let create_account_lxm = Nsid::from("com.atproto.server.createAccount".to_string()); match verifier .verify_service_token(&token, Some("com.atproto.server.createAccount")) .await diff --git a/crates/tranquil-api/src/moderation/mod.rs b/crates/tranquil-api/src/moderation/mod.rs index de8a715..6b7e84d 100644 --- a/crates/tranquil-api/src/moderation/mod.rs +++ b/crates/tranquil-api/src/moderation/mod.rs @@ -11,6 +11,7 @@ use tranquil_pds::api::ApiError; use tranquil_pds::api::proxy_client::{is_ssrf_safe, proxy_client}; use tranquil_pds::auth::{AnyUser, Auth}; use tranquil_pds::state::AppState; +use tranquil_pds::types::{Did, Nsid}; #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub enum ReportReasonType { @@ -135,6 +136,7 @@ async fn proxy_to_report_service( }, }; + let report_lxm = Nsid::from("com.atproto.moderation.createReport".to_string()); let service_token = match tranquil_pds::auth::create_service_token( &auth_user.did, service_did, diff --git a/crates/tranquil-api/src/repo/blob.rs b/crates/tranquil-api/src/repo/blob.rs index 8087d64..d491aa0 100644 --- a/crates/tranquil-api/src/repo/blob.rs +++ b/crates/tranquil-api/src/repo/blob.rs @@ -48,7 +48,7 @@ pub async fn upload_blob( ) -> Result { let (did, controller_did): (Did, Option) = match &auth { AuthAny::Service(service) => { - service.require_lxm("com.atproto.repo.uploadBlob")?; + service.require_lxm(&Nsid::from("com.atproto.repo.uploadBlob".to_string()))?; (service.did.clone(), None) } AuthAny::User(user) => { diff --git a/crates/tranquil-api/src/repo/record/batch.rs b/crates/tranquil-api/src/repo/record/batch.rs index 1af2dee..77482fb 100644 --- a/crates/tranquil-api/src/repo/record/batch.rs +++ b/crates/tranquil-api/src/repo/record/batch.rs @@ -353,9 +353,9 @@ pub async fn apply_writes( &auth, &input.writes, |w| match w { - WriteOp::Create { collection, .. } => collection.as_str(), - WriteOp::Update { collection, .. } => collection.as_str(), - WriteOp::Delete { collection, .. } => collection.as_str(), + WriteOp::Create { collection, .. } => collection, + WriteOp::Update { collection, .. } => collection, + WriteOp::Delete { collection, .. } => collection, }, |w| match w { WriteOp::Create { .. } => WriteOpKind::Create, diff --git a/crates/tranquil-api/src/repo/record/validation.rs b/crates/tranquil-api/src/repo/record/validation.rs index 7c2150c..ecabeb0 100644 --- a/crates/tranquil-api/src/repo/record/validation.rs +++ b/crates/tranquil-api/src/repo/record/validation.rs @@ -9,13 +9,13 @@ pub async fn validate_record_with_status( require_lexicon: bool, ) -> Result { let registry = tranquil_lexicon::LexiconRegistry::global(); - if !registry.has_schema(collection.as_str()) { - let _ = registry.resolve_dynamic(collection.as_str()).await; + if !registry.has_schema(collection) { + let _ = registry.resolve_dynamic(collection).await; } let validator = RecordValidator::new().require_lexicon(require_lexicon); validator - .validate_with_rkey(record, collection.as_str(), rkey.map(|v| v.as_str())) + .validate_with_rkey(record, collection, rkey) .map_err(validation_error_to_api_error) } diff --git a/crates/tranquil-api/src/server/passkey_account.rs b/crates/tranquil-api/src/server/passkey_account.rs index 7002ae7..b9a958c 100644 --- a/crates/tranquil-api/src/server/passkey_account.rs +++ b/crates/tranquil-api/src/server/passkey_account.rs @@ -69,6 +69,7 @@ pub async fn create_passkey_account( let token = extracted.token; if is_service_token(&token) { let verifier = ServiceTokenVerifier::new(); + let create_account_lxm = Nsid::from("com.atproto.server.createAccount".to_string()); match verifier .verify_service_token(&token, Some("com.atproto.server.createAccount")) .await diff --git a/crates/tranquil-api/src/server/service_auth.rs b/crates/tranquil-api/src/server/service_auth.rs index 07ff609..d9f48ce 100644 --- a/crates/tranquil-api/src/server/service_auth.rs +++ b/crates/tranquil-api/src/server/service_auth.rs @@ -169,18 +169,14 @@ pub async fn get_service_auth( } } - let service_token = match tranquil_pds::auth::create_service_token( - &auth.did, - params.aud.as_str(), - lxm.map(|v| v.as_str()), - &key_bytes, - ) { - Ok(t) => t, - Err(e) => { - error!("Failed to create service token: {:?}", e); - return ApiError::InternalError(None).into_response(); - } - }; + let service_token = + match tranquil_pds::auth::create_service_token(&auth.did, ¶ms.aud, lxm, &key_bytes) { + Ok(t) => t, + Err(e) => { + error!("Failed to create service token: {:?}", e); + return ApiError::InternalError(None).into_response(); + } + }; ( StatusCode::OK, Json(GetServiceAuthOutput { diff --git a/crates/tranquil-auth/src/token.rs b/crates/tranquil-auth/src/token.rs index bb39846..dbd50aa 100644 --- a/crates/tranquil-auth/src/token.rs +++ b/crates/tranquil-auth/src/token.rs @@ -125,7 +125,7 @@ pub fn create_refresh_token_with_jti( pub fn create_service_token( did: &str, aud: &str, - lxm: Option<&str>, + lxm: Option<&Nsid>, key_bytes: &[u8], ) -> Result { let signing_key = SigningKey::from_slice(key_bytes)?; @@ -279,7 +279,7 @@ pub fn create_refresh_token_hs256_with_metadata( pub fn create_service_token_hs256( did: &str, aud: &str, - lxm: &str, + lxm: &Nsid, secret: &[u8], ) -> Result { let expiration = Utc::now() diff --git a/crates/tranquil-auth/src/types.rs b/crates/tranquil-auth/src/types.rs index 6872b7d..0fb4e1b 100644 --- a/crates/tranquil-auth/src/types.rs +++ b/crates/tranquil-auth/src/types.rs @@ -216,7 +216,7 @@ pub struct Claims { #[serde(skip_serializing_if = "Option::is_none")] pub scope: Option, #[serde(skip_serializing_if = "Option::is_none")] - pub lxm: Option, + pub lxm: Option, pub jti: String, #[serde(skip_serializing_if = "Option::is_none")] pub act: Option, diff --git a/crates/tranquil-lexicon/src/dynamic.rs b/crates/tranquil-lexicon/src/dynamic.rs index 48691e4..7130ae4 100644 --- a/crates/tranquil-lexicon/src/dynamic.rs +++ b/crates/tranquil-lexicon/src/dynamic.rs @@ -6,6 +6,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::{Duration, Instant}; use tokio::sync::Notify; +use tranquil_types::Nsid; const NEGATIVE_CACHE_TTL: Duration = Duration::from_secs(24 * 60 * 60); const POSITIVE_CACHE_TTL: Duration = Duration::from_secs(24 * 60 * 60); @@ -34,20 +35,20 @@ impl CacheEntry { } struct SchemaStore { - schemas: HashMap, - insertion_order: VecDeque, + schemas: HashMap, + insertion_order: VecDeque, } pub struct DynamicRegistry { store: RwLock, - negative_cache: RwLock>, - in_flight: RwLock>>, + negative_cache: RwLock>, + in_flight: RwLock>>, network_disabled: AtomicBool, } struct InFlightGuard<'a> { registry: &'a DynamicRegistry, - nsid: String, + nsid: Nsid, } impl Drop for InFlightGuard<'_> { @@ -84,7 +85,7 @@ impl DynamicRegistry { self.network_disabled.store(disabled, Ordering::Relaxed); } - pub fn get_cached(&self, nsid: &str) -> Option> { + pub fn get_cached(&self, nsid: &Nsid) -> Option> { self.store .read() .schemas @@ -92,7 +93,7 @@ impl DynamicRegistry { .map(|e| Arc::clone(&e.doc)) } - pub(crate) fn get_entry(&self, nsid: &str) -> Option { + pub(crate) fn get_entry(&self, nsid: &Nsid) -> Option { let now = Instant::now(); self.store.read().schemas.get(nsid).map(|e| { if e.expires_at > now { @@ -103,21 +104,21 @@ impl DynamicRegistry { }) } - pub fn is_negative_cached(&self, nsid: &str) -> bool { + pub fn is_negative_cached(&self, nsid: &Nsid) -> bool { let cache = self.negative_cache.read(); cache .get(nsid) .is_some_and(|entry| entry.expires_at > Instant::now()) } - fn insert_negative(&self, nsid: &str) { + fn insert_negative(&self, nsid: &Nsid) { let mut cache = self.negative_cache.write(); if cache.len() >= MAX_DYNAMIC_SCHEMAS { let now = Instant::now(); cache.retain(|_, entry| entry.expires_at > now); } cache.insert( - nsid.to_string(), + nsid.clone(), NegativeEntry { expires_at: Instant::now() + NEGATIVE_CACHE_TTL, }, @@ -158,25 +159,25 @@ impl DynamicRegistry { arc } - fn bump_expiry(&self, nsid: &str, duration: Duration) { + fn bump_expiry(&self, nsid: &Nsid, duration: Duration) { let mut store = self.store.write(); if let Some(entry) = store.schemas.get_mut(nsid) { entry.expires_at = Instant::now() + duration; } } - pub async fn resolve_and_cache(&self, nsid: &str) -> Result, ResolveError> { + pub async fn resolve_and_cache(&self, nsid: &Nsid) -> Result, ResolveError> { self.resolve_and_cache_with(nsid, |n| async move { resolve_lexicon(&n).await }) .await } async fn resolve_and_cache_with( &self, - nsid: &str, + nsid: &Nsid, resolver: F, ) -> Result, ResolveError> where - F: FnOnce(String) -> Fut, + F: FnOnce(Nsid) -> Fut, Fut: std::future::Future>, { match self.get_entry(nsid) { @@ -188,12 +189,12 @@ impl DynamicRegistry { async fn refresh_stale( &self, - nsid: &str, + nsid: &Nsid, stale: Arc, resolver: F, ) -> Result, ResolveError> where - F: FnOnce(String) -> Fut, + F: FnOnce(Nsid) -> Fut, Fut: std::future::Future>, { if self.network_disabled.load(Ordering::Relaxed) { @@ -201,12 +202,12 @@ impl DynamicRegistry { } match self.acquire_leadership(nsid) { - Some(_guard) => match resolver(nsid.to_string()).await { + Some(_guard) => match resolver(nsid.clone()).await { Ok(doc) => Ok(self.insert_schema(doc)), Err(e) => { self.bump_expiry(nsid, REFRESH_FAILURE_BACKOFF); tracing::warn!( - nsid = nsid, + nsid = %nsid, error = %e, "lexicon refresh failed, serving stale cached entry" ); @@ -222,11 +223,11 @@ impl DynamicRegistry { async fn resolve_fresh( &self, - nsid: &str, + nsid: &Nsid, resolver: F, ) -> Result, ResolveError> where - F: FnOnce(String) -> Fut, + F: FnOnce(Nsid) -> Fut, Fut: std::future::Future>, { if self.network_disabled.load(Ordering::Relaxed) { @@ -234,17 +235,17 @@ impl DynamicRegistry { } if self.is_negative_cached(nsid) { return Err(ResolveError::NegativelyCached { - nsid: nsid.to_string(), + nsid: nsid.clone(), ttl_secs: NEGATIVE_CACHE_TTL.as_secs(), }); } match self.acquire_leadership(nsid) { - Some(_guard) => match resolver(nsid.to_string()).await { + Some(_guard) => match resolver(nsid.clone()).await { Ok(doc) => Ok(self.insert_schema(doc)), Err(e) => { self.insert_negative(nsid); - tracing::debug!(nsid = nsid, error = %e, "caching negative resolution result"); + tracing::debug!(nsid = %nsid, error = %e, "caching negative resolution result"); Err(e) } }, @@ -253,31 +254,29 @@ impl DynamicRegistry { match self.get_cached(nsid) { Some(doc) => Ok(doc), None if self.is_negative_cached(nsid) => Err(ResolveError::NegativelyCached { - nsid: nsid.to_string(), + nsid: nsid.clone(), ttl_secs: NEGATIVE_CACHE_TTL.as_secs(), }), - None => Err(ResolveError::LeaderAborted { - nsid: nsid.to_string(), - }), + None => Err(ResolveError::LeaderAborted { nsid: nsid.clone() }), } } } } - fn acquire_leadership(&self, nsid: &str) -> Option> { + fn acquire_leadership(&self, nsid: &Nsid) -> Option> { let mut map = self.in_flight.write(); if map.contains_key(nsid) { None } else { - map.insert(nsid.to_string(), Arc::new(Notify::new())); + map.insert(nsid.clone(), Arc::new(Notify::new())); Some(InFlightGuard { registry: self, - nsid: nsid.to_string(), + nsid: nsid.clone(), }) } } - async fn wait_for_leader(&self, nsid: &str) { + async fn wait_for_leader(&self, nsid: &Nsid) { let notify = { let map = self.in_flight.read(); match map.get(nsid) { @@ -300,7 +299,7 @@ impl DynamicRegistry { } #[cfg(test)] - fn expire_now(&self, nsid: &str) { + fn expire_now(&self, nsid: &Nsid) { let mut store = self.store.write(); if let Some(entry) = store.schemas.get_mut(nsid) { entry.expires_at = Instant::now(); @@ -318,22 +317,26 @@ impl Default for DynamicRegistry { mod tests { use super::*; + fn nsid(s: &str) -> Nsid { + s.parse().unwrap() + } + #[test] fn test_negative_cache() { let registry = DynamicRegistry::new(); - assert!(!registry.is_negative_cached("com.example.test")); + assert!(!registry.is_negative_cached(&nsid("com.example.test"))); - registry.insert_negative("com.example.test"); - assert!(registry.is_negative_cached("com.example.test")); + registry.insert_negative(&nsid("com.example.test")); + assert!(registry.is_negative_cached(&nsid("com.example.test"))); } #[tokio::test] async fn test_negative_cache_returns_appropriate_error_variant() { let registry = DynamicRegistry::new(); - registry.insert_negative("com.example.cached"); + registry.insert_negative(&nsid("com.example.cached")); let err = registry - .resolve_and_cache("com.example.cached") + .resolve_and_cache(&nsid("com.example.cached")) .await .unwrap_err(); @@ -347,7 +350,11 @@ mod tests { #[test] fn test_empty_lookup() { let registry = DynamicRegistry::new(); - assert!(registry.get_cached("com.example.nonexistent").is_none()); + assert!( + registry + .get_cached(&nsid("com.example.nonexistent")) + .is_none() + ); assert_eq!(registry.schema_count(), 0); } @@ -356,7 +363,7 @@ mod tests { let registry = DynamicRegistry::new(); let doc = LexiconDoc { lexicon: 1, - id: "com.example.test".to_string(), + id: nsid("com.example.test"), defs: HashMap::new(), }; @@ -364,11 +371,11 @@ mod tests { assert_eq!(arc.id, "com.example.test"); assert_eq!(registry.schema_count(), 1); - let retrieved = registry.get_cached("com.example.test"); + let retrieved = registry.get_cached(&nsid("com.example.test")); assert!(retrieved.is_some()); assert_eq!(retrieved.unwrap().id, "com.example.test"); - let entry = registry.get_entry("com.example.test").unwrap(); + let entry = registry.get_entry(&nsid("com.example.test")).unwrap(); assert!(entry.is_fresh(), "freshly inserted entry must be fresh"); } @@ -376,17 +383,17 @@ mod tests { fn test_negative_cache_cleared_on_insert() { let registry = DynamicRegistry::new(); - registry.insert_negative("com.example.test"); - assert!(registry.is_negative_cached("com.example.test")); + registry.insert_negative(&nsid("com.example.test")); + assert!(registry.is_negative_cached(&nsid("com.example.test"))); let doc = LexiconDoc { lexicon: 1, - id: "com.example.test".to_string(), + id: nsid("com.example.test"), defs: HashMap::new(), }; registry.insert_schema(doc); - assert!(!registry.is_negative_cached("com.example.test")); + assert!(!registry.is_negative_cached(&nsid("com.example.test"))); } #[test] @@ -394,17 +401,25 @@ mod tests { let registry = DynamicRegistry::new(); let doc = LexiconDoc { lexicon: 1, - id: "pet.nel.stale".to_string(), + id: nsid("pet.nel.stale"), defs: HashMap::new(), }; registry.insert_schema(doc); - assert!(registry.get_entry("pet.nel.stale").unwrap().is_fresh()); + assert!( + registry + .get_entry(&nsid("pet.nel.stale")) + .unwrap() + .is_fresh() + ); - registry.expire_now("pet.nel.stale"); + registry.expire_now(&nsid("pet.nel.stale")); assert!( - !registry.get_entry("pet.nel.stale").unwrap().is_fresh(), + !registry + .get_entry(&nsid("pet.nel.stale")) + .unwrap() + .is_fresh(), "entry past expiry must be reported stale" ); } @@ -414,14 +429,14 @@ mod tests { let registry = DynamicRegistry::new(); let doc = LexiconDoc { lexicon: 1, - id: "pet.nel.flaky".to_string(), + id: nsid("pet.nel.flaky"), defs: HashMap::new(), }; registry.insert_schema(doc); - registry.expire_now("pet.nel.flaky"); + registry.expire_now(&nsid("pet.nel.flaky")); let result = registry - .resolve_and_cache_with("pet.nel.flaky", |n| async move { + .resolve_and_cache_with(&nsid("pet.nel.flaky"), |n| async move { Err::(ResolveError::DnsLookup { domain: n, reason: "simulated failure".to_string(), @@ -432,11 +447,14 @@ mod tests { let served = result.expect("stale entry must be served when refresh fails"); assert_eq!(served.id, "pet.nel.flaky"); assert!( - registry.get_entry("pet.nel.flaky").unwrap().is_fresh(), + registry + .get_entry(&nsid("pet.nel.flaky")) + .unwrap() + .is_fresh(), "failed refresh must bump expiry so subsequent lookups skip the resolver" ); assert!( - !registry.is_negative_cached("pet.nel.flaky"), + !registry.is_negative_cached(&nsid("pet.nel.flaky")), "stale refresh failure must not poison negative cache" ); } @@ -446,13 +464,13 @@ mod tests { let registry = DynamicRegistry::new(); let doc = LexiconDoc { lexicon: 1, - id: "pet.nel.fresh".to_string(), + id: nsid("pet.nel.fresh"), defs: HashMap::new(), }; registry.insert_schema(doc); let result = registry - .resolve_and_cache_with("pet.nel.fresh", |_| async move { + .resolve_and_cache_with(&nsid("pet.nel.fresh"), |_| async move { panic!("resolver must not run on fresh hit") }) .await; @@ -465,15 +483,15 @@ mod tests { let registry = DynamicRegistry::new(); let doc = LexiconDoc { lexicon: 1, - id: "pet.nel.offline".to_string(), + id: nsid("pet.nel.offline"), defs: HashMap::new(), }; registry.insert_schema(doc); - registry.expire_now("pet.nel.offline"); + registry.expire_now(&nsid("pet.nel.offline")); registry.set_network_disabled(true); let result = registry - .resolve_and_cache_with("pet.nel.offline", |_| async move { + .resolve_and_cache_with(&nsid("pet.nel.offline"), |_| async move { panic!("resolver must not run when network disabled") }) .await; @@ -486,16 +504,21 @@ mod tests { let registry = DynamicRegistry::new(); let doc = LexiconDoc { lexicon: 1, - id: "pet.nel.refresh".to_string(), + id: nsid("pet.nel.refresh"), defs: HashMap::new(), }; registry.insert_schema(doc); - registry.expire_now("pet.nel.refresh"); + registry.expire_now(&nsid("pet.nel.refresh")); - assert!(!registry.get_entry("pet.nel.refresh").unwrap().is_fresh()); + assert!( + !registry + .get_entry(&nsid("pet.nel.refresh")) + .unwrap() + .is_fresh() + ); let refreshed = registry - .resolve_and_cache_with("pet.nel.refresh", |n| async move { + .resolve_and_cache_with(&nsid("pet.nel.refresh"), |n| async move { Ok(LexiconDoc { lexicon: 1, id: n, @@ -507,7 +530,10 @@ mod tests { assert_eq!(refreshed.id, "pet.nel.refresh"); assert!( - registry.get_entry("pet.nel.refresh").unwrap().is_fresh(), + registry + .get_entry(&nsid("pet.nel.refresh")) + .unwrap() + .is_fresh(), "refresh must restore freshness" ); } @@ -524,7 +550,7 @@ mod tests { let calls = Arc::clone(&calls); tokio::spawn(async move { registry - .resolve_and_cache_with("pet.nel.herd", |n| { + .resolve_and_cache_with(&nsid("pet.nel.herd"), |n| { let calls = Arc::clone(&calls); async move { calls.fetch_add(1, Ordering::SeqCst); @@ -565,7 +591,7 @@ mod tests { let calls = Arc::clone(&calls); tokio::spawn(async move { registry - .resolve_and_cache_with("pet.nel.failHerd", |n| { + .resolve_and_cache_with(&nsid("pet.nel.failHerd"), |n| { let calls = Arc::clone(&calls); async move { calls.fetch_add(1, Ordering::SeqCst); @@ -590,7 +616,7 @@ mod tests { 1, "single-flight must coalesce failing resolves too" ); - assert!(registry.is_negative_cached("pet.nel.failHerd")); + assert!(registry.is_negative_cached(&nsid("pet.nel.failHerd"))); } async fn futures_collect(handles: Vec>) -> Vec { @@ -608,7 +634,7 @@ mod tests { (0..MAX_DYNAMIC_SCHEMAS).for_each(|i| { let doc = LexiconDoc { lexicon: 1, - id: format!("pet.nel.schema{}", i), + id: nsid(&format!("pet.nel.schema{}", i)), defs: HashMap::new(), }; registry.insert_schema(doc); @@ -617,23 +643,23 @@ mod tests { let trigger = LexiconDoc { lexicon: 1, - id: "pet.nel.trigger".to_string(), + id: nsid("pet.nel.trigger"), defs: HashMap::new(), }; registry.insert_schema(trigger); assert!( - registry.get_cached("pet.nel.schema0").is_none(), + registry.get_cached(&nsid("pet.nel.schema0")).is_none(), "oldest entry should be evicted" ); assert!( - registry.get_cached("pet.nel.trigger").is_some(), + registry.get_cached(&nsid("pet.nel.trigger")).is_some(), "newly inserted entry should exist" ); let evict_count = MAX_DYNAMIC_SCHEMAS / 4; assert!( registry - .get_cached(&format!("pet.nel.schema{}", evict_count)) + .get_cached(&nsid(&format!("pet.nel.schema{}", evict_count))) .is_some(), "entry after eviction window should survive" ); @@ -644,7 +670,7 @@ mod tests { let registry = DynamicRegistry::new(); let doc = LexiconDoc { lexicon: 1, - id: "pet.nel.tracked".to_string(), + id: nsid("pet.nel.tracked"), defs: HashMap::new(), }; let arc = registry.insert_schema(doc); @@ -656,7 +682,7 @@ mod tests { (0..MAX_DYNAMIC_SCHEMAS).for_each(|i| { registry.insert_schema(LexiconDoc { lexicon: 1, - id: format!("pet.nel.filler{}", i), + id: nsid(&format!("pet.nel.filler{}", i)), defs: HashMap::new(), }); }); diff --git a/crates/tranquil-lexicon/src/registry.rs b/crates/tranquil-lexicon/src/registry.rs index 2aba6ea..ff6d68a 100644 --- a/crates/tranquil-lexicon/src/registry.rs +++ b/crates/tranquil-lexicon/src/registry.rs @@ -1,11 +1,12 @@ use crate::schema::{LexDef, LexObject, LexiconDoc, ParsedRef, parse_ref}; use std::collections::HashMap; use std::sync::{Arc, OnceLock}; +use tranquil_types::Nsid; static REGISTRY: OnceLock = OnceLock::new(); pub struct LexiconRegistry { - schemas: HashMap>, + schemas: HashMap>, #[cfg(feature = "resolve")] dynamic: crate::dynamic::DynamicRegistry, } @@ -39,11 +40,17 @@ impl LexiconRegistry { self.dynamic.insert_schema(doc); } - pub fn get_doc(&self, nsid: &str) -> Option> { - self.schemas.get(nsid).cloned().or_else(|| { + pub fn get_doc(&self, nsid: &Nsid) -> Option> { + self.get_doc_by_key(nsid.as_str()) + } + + fn get_doc_by_key(&self, key: &str) -> Option> { + self.schemas.get(key).cloned().or_else(|| { #[cfg(feature = "resolve")] { - self.dynamic.get_cached(nsid) + Nsid::new(key) + .ok() + .and_then(|nsid| self.dynamic.get_cached(&nsid)) } #[cfg(not(feature = "resolve"))] { @@ -52,7 +59,7 @@ impl LexiconRegistry { }) } - pub fn get_record_def(&self, nsid: &str) -> Option> { + pub fn get_record_def(&self, nsid: &Nsid) -> Option> { let doc = self.get_doc(nsid)?; match doc.defs.get("main")? { LexDef::Record(_) => Some(doc), @@ -67,11 +74,11 @@ impl LexiconRegistry { Self::def_to_resolved(&doc, local) } ParsedRef::Qualified { nsid, fragment } => { - let doc = self.get_doc(nsid)?; + let doc = self.get_doc_by_key(nsid)?; Self::def_to_resolved(&doc, fragment) } ParsedRef::Bare(nsid) => { - let doc = self.get_doc(nsid)?; + let doc = self.get_doc_by_key(nsid)?; Self::def_to_resolved(&doc, "main") } } @@ -90,7 +97,7 @@ impl LexiconRegistry { } } - pub fn has_schema(&self, nsid: &str) -> bool { + pub fn has_schema(&self, nsid: &Nsid) -> bool { self.get_doc(nsid).is_some() } @@ -109,13 +116,13 @@ impl LexiconRegistry { #[cfg(feature = "resolve")] pub async fn resolve_dynamic( &self, - nsid: &str, + nsid: &Nsid, ) -> Result, crate::resolve::ResolveError> { self.dynamic.resolve_and_cache(nsid).await } #[cfg(feature = "resolve")] - pub fn is_negative_cached(&self, nsid: &str) -> bool { + pub fn is_negative_cached(&self, nsid: &Nsid) -> bool { self.dynamic.is_negative_cached(nsid) } } @@ -146,11 +153,15 @@ impl ResolvedRef { mod tests { use super::*; + fn nsid(s: &str) -> Nsid { + s.parse().unwrap() + } + #[test] fn test_empty_registry() { let registry = LexiconRegistry::new(); assert_eq!(registry.schema_count(), 0); - assert!(!registry.has_schema("app.bsky.feed.post")); + assert!(!registry.has_schema(&nsid("app.bsky.feed.post"))); } #[test] @@ -158,19 +169,19 @@ mod tests { let mut registry = LexiconRegistry::new(); let doc = LexiconDoc { lexicon: 1, - id: "com.example.test".to_string(), + id: nsid("com.example.test"), defs: HashMap::new(), }; registry.register(doc); assert_eq!(registry.schema_count(), 1); - assert!(registry.has_schema("com.example.test")); - assert!(!registry.has_schema("com.example.other")); + assert!(registry.has_schema(&nsid("com.example.test"))); + assert!(!registry.has_schema(&nsid("com.example.other"))); } #[test] fn test_get_record_def() { let registry = crate::test_schemas::test_registry(); - let doc = registry.get_record_def("com.test.basic"); + let doc = registry.get_record_def(&nsid("com.test.basic")); assert!(doc.is_some()); let doc = doc.unwrap(); match doc.defs.get("main").unwrap() { @@ -185,7 +196,11 @@ mod tests { #[test] fn test_get_record_def_unknown() { let registry = LexiconRegistry::new(); - assert!(registry.get_record_def("com.example.nonexistent").is_none()); + assert!( + registry + .get_record_def(&nsid("com.example.nonexistent")) + .is_none() + ); } #[test] @@ -205,7 +220,7 @@ mod tests { #[test] fn test_has_schema() { let registry = crate::test_schemas::test_registry(); - assert!(registry.has_schema("com.test.basic")); - assert!(!registry.has_schema("com.example.nonexistent")); + assert!(registry.has_schema(&nsid("com.test.basic"))); + assert!(!registry.has_schema(&nsid("com.example.nonexistent"))); } } diff --git a/crates/tranquil-lexicon/src/resolve.rs b/crates/tranquil-lexicon/src/resolve.rs index ec50952..e1c41b6 100644 --- a/crates/tranquil-lexicon/src/resolve.rs +++ b/crates/tranquil-lexicon/src/resolve.rs @@ -4,6 +4,7 @@ use hickory_resolver::config::{ResolverConfig, ResolverOpts}; use reqwest::Client; use std::sync::OnceLock; use std::time::Duration; +use tranquil_types::{Did, Nsid}; static RESOLVER_CLIENT: OnceLock = OnceLock::new(); @@ -67,18 +68,15 @@ pub enum ResolveError { #[error("schema deserialization failed: {0}")] InvalidSchema(String), #[error("schema resolution recently failed for {nsid}, cached for {ttl_secs}s")] - NegativelyCached { nsid: String, ttl_secs: u64 }, + NegativelyCached { nsid: Nsid, ttl_secs: u64 }, #[error("network resolution disabled")] NetworkDisabled, #[error("leader task for {nsid} aborted before completion")] - LeaderAborted { nsid: String }, + LeaderAborted { nsid: Nsid }, } -pub fn nsid_to_authority(nsid: &str) -> Result { +pub fn nsid_to_authority(nsid: &Nsid) -> String { let mut segments: Vec<&str> = nsid.split('.').collect(); - if segments.len() < 3 { - return Err(ResolveError::InvalidNsid(nsid.to_string())); - } segments.pop(); segments.reverse(); Ok(segments.join(".")) @@ -191,13 +189,13 @@ fn extract_pds_endpoint(doc: &serde_json::Value) -> Option { pub async fn fetch_schema_from_pds( pds_endpoint: &str, did: &str, - nsid: &str, + nsid: &Nsid, ) -> Result { let url = format!( "{}/xrpc/com.atproto.repo.getRecord?repo={}&collection=com.atproto.lexicon.schema&rkey={}", pds_endpoint.trim_end_matches('/'), - urlencoding::encode(did), - urlencoding::encode(nsid) + urlencoding::encode(did.as_str()), + urlencoding::encode(nsid.as_str()) ); let resp = client() @@ -241,8 +239,8 @@ pub async fn fetch_schema_from_pds( .map_err(|e| ResolveError::InvalidSchema(e.to_string())) } -fn validate_fetched_schema(doc: &LexiconDoc, nsid: &str) -> Result<(), ResolveError> { - if doc.id != nsid { +fn validate_fetched_schema(doc: &LexiconDoc, nsid: &Nsid) -> Result<(), ResolveError> { + if doc.id != *nsid { return Err(ResolveError::InvalidSchema(format!( "schema id '{}' does not match requested NSID '{}'", doc.id, nsid @@ -257,22 +255,22 @@ fn validate_fetched_schema(doc: &LexiconDoc, nsid: &str) -> Result<(), ResolveEr Ok(()) } -pub async fn resolve_lexicon(nsid: &str) -> Result { +pub async fn resolve_lexicon(nsid: &Nsid) -> Result { resolve_lexicon_with_config(nsid, None).await } pub async fn resolve_lexicon_with_config( - nsid: &str, + nsid: &Nsid, plc_directory_url: Option<&str>, ) -> Result { - let authority = nsid_to_authority(nsid)?; - tracing::debug!(nsid = nsid, authority = %authority, "resolving lexicon schema"); + let authority = nsid_to_authority(nsid); + tracing::debug!(nsid = %nsid, authority = %authority, "resolving lexicon schema"); let did = resolve_did_from_dns(&authority).await?; - tracing::debug!(nsid = nsid, did = %did, "resolved authority DID"); + tracing::debug!(nsid = %nsid, did = %did, "resolved authority DID"); let pds_endpoint = resolve_pds_endpoint(&did, plc_directory_url).await?; - tracing::debug!(nsid = nsid, pds = %pds_endpoint, "resolved PDS endpoint"); + tracing::debug!(nsid = %nsid, pds = %pds_endpoint, "resolved PDS endpoint"); let doc = fetch_schema_from_pds(&pds_endpoint, &did, nsid).await?; validate_fetched_schema(&doc, nsid)?; @@ -281,7 +279,7 @@ pub async fn resolve_lexicon_with_config( } pub async fn resolve_lexicon_from_did( - nsid: &str, + nsid: &Nsid, did: &str, plc_directory_url: Option<&str>, ) -> Result { @@ -295,18 +293,22 @@ pub async fn resolve_lexicon_from_did( mod tests { use super::*; + fn nsid(s: &str) -> Nsid { + s.parse().unwrap() + } + #[test] fn test_nsid_to_authority() { assert_eq!( - nsid_to_authority("app.bsky.feed.post").unwrap(), + nsid_to_authority(&nsid("app.bsky.feed.post")), "feed.bsky.app" ); assert_eq!( - nsid_to_authority("com.atproto.repo.strongRef").unwrap(), + nsid_to_authority(&nsid("com.atproto.repo.strongRef")), "repo.atproto.com" ); assert_eq!( - nsid_to_authority("com.germnetwork.social.post").unwrap(), + nsid_to_authority(&nsid("com.germnetwork.social.post")), "social.germnetwork.com" ); assert!(nsid_to_authority("tooShort").is_err()); @@ -315,7 +317,7 @@ mod tests { #[test] fn test_nsid_to_authority_three_segments() { assert_eq!( - nsid_to_authority("org.example.record").unwrap(), + nsid_to_authority(&nsid("org.example.record")), "example.org" ); } @@ -375,20 +377,20 @@ mod tests { fn test_validate_fetched_schema_ok() { let doc = LexiconDoc { lexicon: 1, - id: "com.example.thing".to_string(), + id: nsid("com.example.thing"), defs: Default::default(), }; - assert!(validate_fetched_schema(&doc, "com.example.thing").is_ok()); + assert!(validate_fetched_schema(&doc, &nsid("com.example.thing")).is_ok()); } #[test] fn test_validate_fetched_schema_id_mismatch() { let doc = LexiconDoc { lexicon: 1, - id: "com.example.other".to_string(), + id: nsid("com.example.other"), defs: Default::default(), }; - let err = validate_fetched_schema(&doc, "com.example.thing").unwrap_err(); + let err = validate_fetched_schema(&doc, &nsid("com.example.thing")).unwrap_err(); assert!(matches!(err, ResolveError::InvalidSchema(_))); } @@ -396,10 +398,10 @@ mod tests { fn test_validate_fetched_schema_bad_version() { let doc = LexiconDoc { lexicon: 99, - id: "com.example.thing".to_string(), + id: nsid("com.example.thing"), defs: Default::default(), }; - let err = validate_fetched_schema(&doc, "com.example.thing").unwrap_err(); + let err = validate_fetched_schema(&doc, &nsid("com.example.thing")).unwrap_err(); assert!(matches!(err, ResolveError::InvalidSchema(_))); } } diff --git a/crates/tranquil-lexicon/src/schema.rs b/crates/tranquil-lexicon/src/schema.rs index f1a3d2b..d71b12f 100644 --- a/crates/tranquil-lexicon/src/schema.rs +++ b/crates/tranquil-lexicon/src/schema.rs @@ -1,10 +1,11 @@ use serde::Deserialize; use std::collections::HashMap; +use tranquil_types::Nsid; #[derive(Debug, Deserialize)] pub struct LexiconDoc { pub lexicon: u32, - pub id: String, + pub id: Nsid, #[serde(default)] pub defs: HashMap, } diff --git a/crates/tranquil-lexicon/src/validate.rs b/crates/tranquil-lexicon/src/validate.rs index 6876a5e..9261b34 100644 --- a/crates/tranquil-lexicon/src/validate.rs +++ b/crates/tranquil-lexicon/src/validate.rs @@ -47,7 +47,7 @@ fn ref_to_context_nsid<'a>(reference: &'a str, current_context: &'a str) -> &'a pub fn validate_record( registry: &LexiconRegistry, - nsid: &str, + nsid: &tranquil_types::Nsid, value: &serde_json::Value, ) -> Result<(), LexValidationError> { let doc = registry @@ -527,6 +527,11 @@ mod tests { use super::*; use crate::test_schemas::test_registry; use serde_json::json; + use tranquil_types::Nsid; + + fn nsid(s: &str) -> Nsid { + s.parse().unwrap() + } #[test] fn test_validate_valid_record() { @@ -536,7 +541,7 @@ mod tests { "text": "Hello, world!", "createdAt": "2024-01-01T00:00:00.000Z" }); - assert!(validate_record(®istry, "com.test.basic", &record).is_ok()); + assert!(validate_record(®istry, &nsid("com.test.basic"), &record).is_ok()); } #[test] @@ -546,7 +551,7 @@ mod tests { "$type": "com.test.basic", "createdAt": "2024-01-01T00:00:00.000Z" }); - let err = validate_record(®istry, "com.test.basic", &record).unwrap_err(); + let err = validate_record(®istry, &nsid("com.test.basic"), &record).unwrap_err(); assert!(matches!(err, LexValidationError::MissingRequired { .. })); } @@ -558,7 +563,7 @@ mod tests { "text": "a".repeat(101), "createdAt": "2024-01-01T00:00:00.000Z" }); - let err = validate_record(®istry, "com.test.basic", &record).unwrap_err(); + let err = validate_record(®istry, &nsid("com.test.basic"), &record).unwrap_err(); assert!(matches!(err, LexValidationError::InvalidField { .. })); } @@ -570,7 +575,7 @@ mod tests { "text": "a".repeat(51), "createdAt": "2024-01-01T00:00:00.000Z" }); - let err = validate_record(®istry, "com.test.basic", &record).unwrap_err(); + let err = validate_record(®istry, &nsid("com.test.basic"), &record).unwrap_err(); assert!(matches!(err, LexValidationError::InvalidField { .. })); } @@ -582,7 +587,7 @@ mod tests { "$type": "com.test.profile", "displayName": emoji_text }); - let err = validate_record(®istry, "com.test.profile", &record).unwrap_err(); + let err = validate_record(®istry, &nsid("com.test.profile"), &record).unwrap_err(); assert!(matches!(err, LexValidationError::InvalidField { .. })); } @@ -595,7 +600,7 @@ mod tests { "createdAt": "2024-01-01T00:00:00.000Z", "count": 101 }); - let err = validate_record(®istry, "com.test.basic", &record).unwrap_err(); + let err = validate_record(®istry, &nsid("com.test.basic"), &record).unwrap_err(); assert!(matches!(err, LexValidationError::InvalidField { .. })); let record_neg = json!({ @@ -604,7 +609,7 @@ mod tests { "createdAt": "2024-01-01T00:00:00.000Z", "count": -1 }); - let err = validate_record(®istry, "com.test.basic", &record_neg).unwrap_err(); + let err = validate_record(®istry, &nsid("com.test.basic"), &record_neg).unwrap_err(); assert!(matches!(err, LexValidationError::InvalidField { .. })); } @@ -617,7 +622,7 @@ mod tests { "createdAt": "2024-01-01T00:00:00.000Z", "count": 5.0 }); - assert!(validate_record(®istry, "com.test.basic", &record).is_ok()); + assert!(validate_record(®istry, &nsid("com.test.basic"), &record).is_ok()); let record_frac = json!({ "$type": "com.test.basic", @@ -625,7 +630,7 @@ mod tests { "createdAt": "2024-01-01T00:00:00.000Z", "count": 5.5 }); - let err = validate_record(®istry, "com.test.basic", &record_frac).unwrap_err(); + let err = validate_record(®istry, &nsid("com.test.basic"), &record_frac).unwrap_err(); assert!(matches!(err, LexValidationError::InvalidField { .. })); } @@ -638,7 +643,7 @@ mod tests { "createdAt": "2024-01-01T00:00:00.000Z", "active": "not-a-bool" }); - let err = validate_record(®istry, "com.test.basic", &record).unwrap_err(); + let err = validate_record(®istry, &nsid("com.test.basic"), &record).unwrap_err(); assert!(matches!(err, LexValidationError::InvalidField { .. })); } @@ -651,7 +656,7 @@ mod tests { "createdAt": "2024-01-01T00:00:00.000Z", "tags": ["a", "b", "c", "d"] }); - let err = validate_record(®istry, "com.test.basic", &record).unwrap_err(); + let err = validate_record(®istry, &nsid("com.test.basic"), &record).unwrap_err(); assert!(matches!(err, LexValidationError::InvalidField { .. })); } @@ -664,7 +669,7 @@ mod tests { "createdAt": "2024-01-01T00:00:00.000Z", "tags": ["a", "b", "c"] }); - assert!(validate_record(®istry, "com.test.basic", &record).is_ok()); + assert!(validate_record(®istry, &nsid("com.test.basic"), &record).is_ok()); } #[test] @@ -678,7 +683,7 @@ mod tests { }, "createdAt": "2024-01-01T00:00:00.000Z" }); - assert!(validate_record(®istry, "com.test.withref", &record).is_ok()); + assert!(validate_record(®istry, &nsid("com.test.withref"), &record).is_ok()); } #[test] @@ -691,7 +696,7 @@ mod tests { }, "createdAt": "2024-01-01T00:00:00.000Z" }); - let err = validate_record(®istry, "com.test.withref", &record).unwrap_err(); + let err = validate_record(®istry, &nsid("com.test.withref"), &record).unwrap_err(); assert!(matches!(err, LexValidationError::MissingRequired { .. })); } @@ -713,7 +718,7 @@ mod tests { } } }); - assert!(validate_record(®istry, "com.test.withreply", &record).is_ok()); + assert!(validate_record(®istry, &nsid("com.test.withreply"), &record).is_ok()); } #[test] @@ -738,7 +743,7 @@ mod tests { ] } }); - assert!(validate_record(®istry, "com.test.withreply", &record).is_ok()); + assert!(validate_record(®istry, &nsid("com.test.withreply"), &record).is_ok()); let bad_embed = json!({ "$type": "com.test.withreply", @@ -750,7 +755,7 @@ mod tests { } }); assert!( - validate_record(®istry, "com.test.withreply", &bad_embed).is_err(), + validate_record(®istry, &nsid("com.test.withreply"), &bad_embed).is_err(), "union with bare NSID ref must validate the matched schema" ); } @@ -852,7 +857,7 @@ mod tests { } } }); - assert!(validate_record(®istry, "com.test.withreply", &record).is_ok()); + assert!(validate_record(®istry, &nsid("com.test.withreply"), &record).is_ok()); let bad_external = json!({ "$type": "com.test.withreply", @@ -866,7 +871,7 @@ mod tests { } }); assert!( - validate_record(®istry, "com.test.withreply", &bad_external).is_err(), + validate_record(®istry, &nsid("com.test.withreply"), &bad_external).is_err(), "local #ref in cross-schema union must resolve against the correct schema" ); } @@ -882,7 +887,7 @@ mod tests { { "$type": "com.test.withgate#disableRule" } ] }); - assert!(validate_record(®istry, "com.test.withgate", &record).is_ok()); + assert!(validate_record(®istry, &nsid("com.test.withgate"), &record).is_ok()); } #[test] @@ -893,14 +898,14 @@ mod tests { "subject": "did:plc:abc123", "createdAt": "2024-01-01T00:00:00.000Z" }); - assert!(validate_record(®istry, "com.test.withdid", &record).is_ok()); + assert!(validate_record(®istry, &nsid("com.test.withdid"), &record).is_ok()); let bad_did = json!({ "$type": "com.test.withdid", "subject": "not-a-did", "createdAt": "2024-01-01T00:00:00.000Z" }); - assert!(validate_record(®istry, "com.test.withdid", &bad_did).is_err()); + assert!(validate_record(®istry, &nsid("com.test.withdid"), &bad_did).is_err()); } #[test] @@ -911,14 +916,15 @@ mod tests { "name": "test", "value": null }); - assert!(validate_record(®istry, "com.test.nullable", &record).is_ok()); + assert!(validate_record(®istry, &nsid("com.test.nullable"), &record).is_ok()); } #[test] fn test_validate_unknown_lexicon() { let registry = test_registry(); let record = json!({"$type": "com.example.nonexistent"}); - let err = validate_record(®istry, "com.example.nonexistent", &record).unwrap_err(); + let err = + validate_record(®istry, &nsid("com.example.nonexistent"), &record).unwrap_err(); assert!(matches!(err, LexValidationError::LexiconNotFound(_))); } @@ -931,14 +937,14 @@ mod tests { "createdAt": "2024-01-01T00:00:00.000Z", "unknownField": "this is fine" }); - assert!(validate_record(®istry, "com.test.basic", &record).is_ok()); + assert!(validate_record(®istry, &nsid("com.test.basic"), &record).is_ok()); } #[test] fn test_validate_no_required_fields() { let registry = test_registry(); let record = json!({"$type": "com.test.profile"}); - assert!(validate_record(®istry, "com.test.profile", &record).is_ok()); + assert!(validate_record(®istry, &nsid("com.test.profile"), &record).is_ok()); } #[test] @@ -948,7 +954,7 @@ mod tests { "$type": "com.test.profile", "displayName": "a".repeat(11) }); - let err = validate_record(®istry, "com.test.profile", &record).unwrap_err(); + let err = validate_record(®istry, &nsid("com.test.profile"), &record).unwrap_err(); assert!(matches!(err, LexValidationError::InvalidField { .. })); } @@ -961,7 +967,7 @@ mod tests { "value": null }); assert!( - validate_record(®istry, "com.test.requirednullable", &record).is_ok(), + validate_record(®istry, &nsid("com.test.requirednullable"), &record).is_ok(), "a field that is both required and nullable must accept null values" ); } @@ -975,7 +981,8 @@ mod tests { }); assert!( matches!( - validate_record(®istry, "com.test.requirednullable", &record).unwrap_err(), + validate_record(®istry, &nsid("com.test.requirednullable"), &record) + .unwrap_err(), LexValidationError::MissingRequired { .. } ), "a field that is required+nullable must still be present (even if null)" @@ -991,7 +998,7 @@ mod tests { "value": "hello" }); assert!( - validate_record(®istry, "com.test.requirednullable", &record).is_ok(), + validate_record(®istry, &nsid("com.test.requirednullable"), &record).is_ok(), "a field that is required+nullable must accept non-null values" ); } diff --git a/crates/tranquil-lexicon/tests/resolve_integration.rs b/crates/tranquil-lexicon/tests/resolve_integration.rs index f9e0518..eb5bb3e 100644 --- a/crates/tranquil-lexicon/tests/resolve_integration.rs +++ b/crates/tranquil-lexicon/tests/resolve_integration.rs @@ -167,9 +167,13 @@ async fn test_fetch_schema_from_pds_success() { .mount(&pds_server) .await; - let doc = fetch_schema_from_pds(&pds_server.uri(), did, nsid) - .await - .unwrap(); + let doc = fetch_schema_from_pds( + &pds_server.uri(), + &did.parse().unwrap(), + &nsid.parse().unwrap(), + ) + .await + .unwrap(); assert_eq!(doc.id, nsid); assert_eq!(doc.lexicon, 1); assert!(doc.defs.contains_key("main")); @@ -190,7 +194,12 @@ async fn test_fetch_schema_missing_value_field() { .mount(&pds_server) .await; - let result = fetch_schema_from_pds(&pds_server.uri(), did, nsid).await; + let result = fetch_schema_from_pds( + &pds_server.uri(), + &did.parse().unwrap(), + &nsid.parse().unwrap(), + ) + .await; assert!(matches!(result, Err(ResolveError::SchemaFetch { .. }))); } @@ -212,7 +221,12 @@ async fn test_fetch_schema_invalid_lexicon_json() { .mount(&pds_server) .await; - let result = fetch_schema_from_pds(&pds_server.uri(), did, nsid).await; + let result = fetch_schema_from_pds( + &pds_server.uri(), + &did.parse().unwrap(), + &nsid.parse().unwrap(), + ) + .await; assert!(matches!(result, Err(ResolveError::InvalidSchema(_)))); } @@ -240,9 +254,13 @@ async fn test_full_chain_plc_to_schema() { .mount(&pds_server) .await; - let doc = resolve_lexicon_from_did(nsid, did, Some(&plc_server.uri())) - .await - .unwrap(); + let doc = resolve_lexicon_from_did( + &nsid.parse().unwrap(), + &did.parse().unwrap(), + Some(&plc_server.uri()), + ) + .await + .unwrap(); assert_eq!(doc.id, nsid); assert_eq!(doc.lexicon, 1); } @@ -271,7 +289,12 @@ async fn test_full_chain_schema_id_mismatch_rejected() { .mount(&pds_server) .await; - let result = resolve_lexicon_from_did(nsid, did, Some(&plc_server.uri())).await; + let result = resolve_lexicon_from_did( + &nsid.parse().unwrap(), + &did.parse().unwrap(), + Some(&plc_server.uri()), + ) + .await; assert!(matches!(result, Err(ResolveError::InvalidSchema(_)))); } @@ -304,7 +327,12 @@ async fn test_full_chain_bad_lexicon_version_rejected() { .mount(&pds_server) .await; - let result = resolve_lexicon_from_did(nsid, did, Some(&plc_server.uri())).await; + let result = resolve_lexicon_from_did( + &nsid.parse().unwrap(), + &did.parse().unwrap(), + Some(&plc_server.uri()), + ) + .await; assert!(matches!(result, Err(ResolveError::InvalidSchema(_)))); } @@ -323,9 +351,13 @@ async fn test_pds_trailing_slash_handled() { .await; let pds_url_with_slash = format!("{}/", pds_server.uri()); - let doc = fetch_schema_from_pds(&pds_url_with_slash, did, nsid) - .await - .unwrap(); + let doc = fetch_schema_from_pds( + &pds_url_with_slash, + &did.parse().unwrap(), + &nsid.parse().unwrap(), + ) + .await + .unwrap(); assert_eq!(doc.id, nsid); } @@ -344,7 +376,12 @@ async fn test_fetch_schema_error_status_gives_meaningful_error() { .mount(&pds_server) .await; - let result = fetch_schema_from_pds(&pds_server.uri(), did, nsid).await; + let result = fetch_schema_from_pds( + &pds_server.uri(), + &did.parse().unwrap(), + &nsid.parse().unwrap(), + ) + .await; let err = result.unwrap_err(); let err_msg = err.to_string(); assert!( diff --git a/crates/tranquil-pds/src/api/proxy.rs b/crates/tranquil-pds/src/api/proxy.rs index efdf723..7ee8b68 100644 --- a/crates/tranquil-pds/src/api/proxy.rs +++ b/crates/tranquil-pds/src/api/proxy.rs @@ -5,6 +5,7 @@ use std::sync::LazyLock; use crate::api::error::ApiError; use crate::api::proxy_client::proxy_client; use crate::state::AppState; +use crate::types::{Did, Nsid}; use crate::util::get_header_str; use axum::{ body::Bytes, @@ -275,6 +276,10 @@ async fn proxy_handler( .await { Ok(auth_user) => { + let Ok(method_nsid) = method.parse::() else { + return ApiError::InvalidRequest(format!("Invalid XRPC method: {}", method)) + .into_response(); + }; if let Err(e) = crate::auth::scope_check::check_rpc_scope( &auth_user.auth_source, auth_user.scope.as_deref(), @@ -319,7 +324,12 @@ async fn proxy_handler( // getFeed must be audienced to the feed generator, not the AppView. let (token_aud, token_lxm) = if method == "app.bsky.feed.getFeed" { match resolve_feed_generator_did(&resolved.url, query.as_deref()).await { - Some(feed_did) => (feed_did, "app.bsky.feed.getFeedSkeleton"), + Some(feed_did) => ( + feed_did, + "app.bsky.feed.getFeedSkeleton" + .parse::() + .expect("getFeedSkeleton is a valid NSID"), + ), None => { warn!( "getFeed proxy: could not resolve feed generator DID; refusing \ diff --git a/crates/tranquil-pds/src/auth/extractor.rs b/crates/tranquil-pds/src/auth/extractor.rs index f999f9b..4beefb9 100644 --- a/crates/tranquil-pds/src/auth/extractor.rs +++ b/crates/tranquil-pds/src/auth/extractor.rs @@ -340,7 +340,11 @@ impl Auth

{ self.0.permissions() } - pub fn check_repo_scope(&self, action: RepoAction, collection: &str) -> Result<(), ApiError> { + pub fn check_repo_scope( + &self, + action: RepoAction, + collection: &crate::types::Nsid, + ) -> Result<(), ApiError> { if !self.needs_scope_check() { return Ok(()); } @@ -424,7 +428,7 @@ pub struct ServiceAuth { } impl ServiceAuth { - pub fn require_lxm(&self, expected_lxm: &str) -> Result<(), ApiError> { + pub fn require_lxm(&self, expected_lxm: &crate::types::Nsid) -> Result<(), ApiError> { match &self.claims.lxm { Some(lxm) if crate::auth::lxm_permits(lxm, expected_lxm) => Ok(()), Some(lxm) => Err(ApiError::AuthorizationError(format!( @@ -505,7 +509,7 @@ impl AuthAny

{ matches!(self, Self::Service(_)) } - pub fn require_lxm(&self, expected_lxm: &str) -> Result<(), ApiError> { + pub fn require_lxm(&self, expected_lxm: &crate::types::Nsid) -> Result<(), ApiError> { match self { Self::User(_) => Ok(()), Self::Service(auth) => auth.require_lxm(expected_lxm), diff --git a/crates/tranquil-pds/src/auth/mod.rs b/crates/tranquil-pds/src/auth/mod.rs index 1e8ac56..48cf852 100644 --- a/crates/tranquil-pds/src/auth/mod.rs +++ b/crates/tranquil-pds/src/auth/mod.rs @@ -57,8 +57,8 @@ pub use tranquil_auth::{ verify_refresh_token_hs256, verify_token, verify_token_es256k, verify_totp_code, }; -pub fn lxm_permits(lxm: &str, expected: &str) -> bool { - lxm == "*" || lxm == expected +pub fn lxm_permits(lxm: &str, expected: &crate::types::Nsid) -> bool { + lxm == "*" || lxm == expected.as_str() } pub fn try_decrypt_user_key( @@ -182,7 +182,7 @@ impl AuthenticatedUser { self.auth_source.service_claims() } - pub fn require_lxm(&self, expected_lxm: &str) -> Result<(), ApiError> { + pub fn require_lxm(&self, expected_lxm: &crate::types::Nsid) -> Result<(), ApiError> { match self.auth_source.service_claims() { Some(claims) => match &claims.lxm { Some(lxm) if lxm_permits(lxm, expected_lxm) => Ok(()), @@ -632,6 +632,10 @@ pub async fn validate_token_with_dpop( mod tests { use super::*; + fn n(s: &str) -> crate::types::Nsid { + crate::types::Nsid::from(s.to_string()) + } + #[test] fn test_lxm_permits_exact_match() { assert!(lxm_permits( diff --git a/crates/tranquil-pds/src/auth/scope_check.rs b/crates/tranquil-pds/src/auth/scope_check.rs index f9a14c0..fb8f1c2 100644 --- a/crates/tranquil-pds/src/auth/scope_check.rs +++ b/crates/tranquil-pds/src/auth/scope_check.rs @@ -2,6 +2,7 @@ use crate::api::error::ApiError; use crate::oauth::scopes::{ AccountAction, AccountAttr, IdentityAttr, RepoAction, ScopePermissions, }; +use crate::types::Nsid; use super::{AuthSource, TokenScope}; @@ -19,7 +20,7 @@ pub fn check_repo_scope( auth_source: &AuthSource, scope: Option<&str>, action: RepoAction, - collection: &str, + collection: &Nsid, ) -> Result<(), ApiError> { if !requires_scope_check(auth_source, scope) { return Ok(()); @@ -50,7 +51,7 @@ pub fn check_rpc_scope( auth_source: &AuthSource, scope: Option<&str>, aud: &str, - lxm: &str, + lxm: &Nsid, ) -> Result<(), ApiError> { if !requires_scope_check(auth_source, scope) { return Ok(()); diff --git a/crates/tranquil-pds/src/auth/scope_verified.rs b/crates/tranquil-pds/src/auth/scope_verified.rs index 436710c..5b757c0 100644 --- a/crates/tranquil-pds/src/auth/scope_verified.rs +++ b/crates/tranquil-pds/src/auth/scope_verified.rs @@ -7,7 +7,7 @@ use crate::api::error::ApiError; use crate::oauth::scopes::{ AccountAction, AccountAttr, IdentityAttr, RepoAction, ScopePermissions, }; -use crate::types::Did; +use crate::types::{Did, Nsid}; use super::AuthenticatedUser; @@ -236,24 +236,24 @@ pub fn verify_batch_write_scopes<'a, T, C, F>( classify: C, ) -> Result, ScopeVerificationError> where - F: Fn(&T) -> &str, + F: Fn(&T) -> &Nsid, C: Fn(&T) -> WriteOpKind, { use std::collections::HashSet; - let create_collections: HashSet<&str> = writes + let create_collections: HashSet<&Nsid> = writes .iter() .filter(|w| matches!(classify(w), WriteOpKind::Create)) .map(&get_collection) .collect(); - let update_collections: HashSet<&str> = writes + let update_collections: HashSet<&Nsid> = writes .iter() .filter(|w| matches!(classify(w), WriteOpKind::Update)) .map(&get_collection) .collect(); - let delete_collections: HashSet<&str> = writes + let delete_collections: HashSet<&Nsid> = writes .iter() .filter(|w| matches!(classify(w), WriteOpKind::Delete)) .map(&get_collection) @@ -300,7 +300,7 @@ pub trait VerifyScope { fn verify_repo_create<'a>( &'a self, - collection: &str, + collection: &Nsid, ) -> Result, ScopeVerificationError> where Self: AsRef, @@ -322,7 +322,7 @@ pub trait VerifyScope { fn verify_repo_update<'a>( &'a self, - collection: &str, + collection: &Nsid, ) -> Result, ScopeVerificationError> where Self: AsRef, @@ -344,7 +344,7 @@ pub trait VerifyScope { fn verify_repo_delete<'a>( &'a self, - collection: &str, + collection: &Nsid, ) -> Result, ScopeVerificationError> where Self: AsRef, @@ -366,7 +366,7 @@ pub trait VerifyScope { fn verify_repo_upsert<'a>( &'a self, - collection: &str, + collection: &Nsid, ) -> Result, ScopeVerificationError> where Self: AsRef, @@ -414,7 +414,7 @@ pub trait VerifyScope { fn verify_rpc<'a>( &'a self, aud: &str, - lxm: &str, + lxm: &Nsid, ) -> Result, ScopeVerificationError> where Self: AsRef, diff --git a/crates/tranquil-pds/src/auth/service.rs b/crates/tranquil-pds/src/auth/service.rs index 1cafb69..2771099 100644 --- a/crates/tranquil-pds/src/auth/service.rs +++ b/crates/tranquil-pds/src/auth/service.rs @@ -170,7 +170,7 @@ impl ServiceTokenVerifier { pub async fn verify_service_token( &self, token: &str, - required_lxm: Option<&str>, + required_lxm: Option<&Nsid>, ) -> Result { let jwt = JwtParts::parse(token)?; diff --git a/crates/tranquil-pds/src/sync/import.rs b/crates/tranquil-pds/src/sync/import.rs index 070f3bc..a20d11b 100644 --- a/crates/tranquil-pds/src/sync/import.rs +++ b/crates/tranquil-pds/src/sync/import.rs @@ -171,7 +171,7 @@ pub fn extract_links(value: &Ipld, links: &mut Vec) { #[derive(Debug)] pub struct ImportedRecord { - pub collection: String, + pub collection: Nsid, pub rkey: String, pub cid: Cid, pub blob_refs: Vec, @@ -227,7 +227,7 @@ fn walk_mst_node( let blob_refs = find_blob_refs_ipld(&record_value, 0); let parts: Vec<&str> = full_key.split('/').collect(); if parts.len() >= 2 { - let collection = parts[..parts.len() - 1].join("/"); + let collection = Nsid::from(parts[..parts.len() - 1].join("/")); let rkey = parts[parts.len() - 1].to_string(); records.push(ImportedRecord { collection, diff --git a/crates/tranquil-pds/src/validation/mod.rs b/crates/tranquil-pds/src/validation/mod.rs index 2da0cb6..e626b44 100644 --- a/crates/tranquil-pds/src/validation/mod.rs +++ b/crates/tranquil-pds/src/validation/mod.rs @@ -1,3 +1,4 @@ +use crate::types::{Nsid, Rkey}; use serde_json::Value; use thiserror::Error; use tranquil_lexicon::LexValidationError; @@ -65,7 +66,7 @@ impl RecordValidator { pub fn validate( &self, record: &Value, - collection: &str, + collection: &Nsid, ) -> Result { self.validate_with_rkey(record, collection, None) } @@ -73,13 +74,13 @@ impl RecordValidator { pub fn validate_with_rkey( &self, record: &Value, - collection: &str, + collection: &Nsid, rkey: Option<&str>, ) -> Result { let (record_type, obj) = validate_preamble(record, collection)?; let registry = tranquil_lexicon::LexiconRegistry::global(); - match tranquil_lexicon::validate_record(registry, record_type, record) { + match tranquil_lexicon::validate_record(registry, collection, record) { Ok(()) => { check_banned_content(record_type, obj, rkey)?; Ok(ValidationStatus::Valid) @@ -110,7 +111,7 @@ impl RecordValidator { fn validate_preamble<'a>( record: &'a Value, - collection: &str, + collection: &Nsid, ) -> Result<(&'a str, &'a serde_json::Map), ValidationError> { let obj = record .as_object() @@ -119,7 +120,7 @@ fn validate_preamble<'a>( .get("$type") .and_then(|v| v.as_str()) .ok_or(ValidationError::MissingType)?; - if record_type != collection { + if record_type != *collection { return Err(ValidationError::TypeMismatch { expected: collection.to_string(), actual: record_type.to_string(), diff --git a/crates/tranquil-pds/tests/jwt_security.rs b/crates/tranquil-pds/tests/jwt_security.rs index aaebb16..aa8bbad 100644 --- a/crates/tranquil-pds/tests/jwt_security.rs +++ b/crates/tranquil-pds/tests/jwt_security.rs @@ -14,6 +14,7 @@ use tranquil_pds::auth::{ get_did_from_token, get_jti_from_token, verify_access_token, verify_refresh_token, verify_token, }; +use tranquil_types::{Did, Nsid}; fn generate_user_key() -> Vec { let secret_key = SecretKey::random(&mut OsRng); diff --git a/crates/tranquil-pds/tests/record_validation.rs b/crates/tranquil-pds/tests/record_validation.rs index f812783..11ba1e6 100644 --- a/crates/tranquil-pds/tests/record_validation.rs +++ b/crates/tranquil-pds/tests/record_validation.rs @@ -3,11 +3,16 @@ use tranquil_pds::validation::{ RecordValidator, ValidationError, ValidationStatus, validate_collection_nsid, validate_record_key, }; +use tranquil_types::Nsid; fn now() -> String { chrono::Utc::now().to_rfc3339() } +fn c(s: &str) -> Nsid { + Nsid::from(s.to_string()) +} + #[test] fn test_type_mismatch() { let validator = RecordValidator::new(); diff --git a/crates/tranquil-pds/tests/repo_batch.rs b/crates/tranquil-pds/tests/repo_batch.rs index 9a4e97d..9821134 100644 --- a/crates/tranquil-pds/tests/repo_batch.rs +++ b/crates/tranquil-pds/tests/repo_batch.rs @@ -4,7 +4,7 @@ use common::*; use reqwest::StatusCode; use serde_json::{Value, json}; use tranquil_db_traits::{Backlink, BacklinkPath}; -use tranquil_pds::types::{AtUri, Did, Nsid}; +use tranquil_pds::types::{AtUri, Did, Nsid, Rkey}; #[tokio::test] async fn test_apply_writes_create() { diff --git a/crates/tranquil-pds/tests/scope_edge_cases.rs b/crates/tranquil-pds/tests/scope_edge_cases.rs index 59f3d3c..c9d7ed3 100644 --- a/crates/tranquil-pds/tests/scope_edge_cases.rs +++ b/crates/tranquil-pds/tests/scope_edge_cases.rs @@ -3,6 +3,11 @@ use tranquil_pds::oauth::scopes::{ AccountAction, IdentityAttr, ParsedScope, RepoAction, ScopePermissions, parse_scope, parse_scope_string, }; +use tranquil_types::Nsid; + +fn c(s: &str) -> Nsid { + Nsid::from(s.to_string()) +} #[test] fn test_repo_star_defaults_to_all_actions() { @@ -141,9 +146,9 @@ fn test_multiple_scopes_parsing() { fn test_permissions_null_scope_defaults_atproto() { let perms = ScopePermissions::from_scope_string(None); assert!(!perms.has_full_access()); - assert!(!perms.allows_repo(RepoAction::Create, "any.collection")); - assert!(!perms.allows_repo(RepoAction::Update, "any.collection")); - assert!(!perms.allows_repo(RepoAction::Delete, "any.collection")); + assert!(!perms.allows_repo(RepoAction::Create, &c("cafe.oyster.record"))); + assert!(!perms.allows_repo(RepoAction::Update, &c("cafe.oyster.record"))); + assert!(!perms.allows_repo(RepoAction::Delete, &c("cafe.oyster.record"))); } #[test] diff --git a/crates/tranquil-scopes/src/permission_set.rs b/crates/tranquil-scopes/src/permission_set.rs index 96cf4f1..0f77d32 100644 --- a/crates/tranquil-scopes/src/permission_set.rs +++ b/crates/tranquil-scopes/src/permission_set.rs @@ -6,6 +6,7 @@ use std::collections::HashMap; use std::sync::LazyLock; use tokio::sync::RwLock; use tracing::debug; +use tranquil_types::{Did, Nsid}; #[derive(Debug, thiserror::Error)] pub enum ScopeExpansionError { @@ -81,7 +82,9 @@ pub async fn expand_include_scopes(scope_string: &str) -> Result { let (nsid_base, aud) = parse_include_scope(rest); - expand_permission_set(nsid_base, aud).await + let nsid = Nsid::new(nsid_base) + .map_err(|_| ScopeExpansionError::InvalidNsid(nsid_base.to_string()))?; + expand_permission_set(&nsid, aud).await } None => Ok(scope.to_string()), } @@ -105,7 +108,7 @@ fn parse_include_scope(rest: &str) -> (&str, Option<&str>) { } async fn expand_permission_set( - nsid: &str, + nsid: &Nsid, aud: Option<&str>, ) -> Result { let cache_key = match aud { @@ -118,7 +121,7 @@ async fn expand_permission_set( if let Some(cached) = cache.get(&cache_key) && cached.cached_at.elapsed().as_secs() < CACHE_TTL_SECS { - debug!(nsid, "Using cached permission set expansion"); + debug!(nsid = %nsid, "Using cached permission set expansion"); return Ok(cached.expanded_scope.clone()); } } @@ -162,29 +165,25 @@ async fn expand_permission_set( ); } - debug!(nsid, expanded = %expanded, "Successfully expanded permission set"); + debug!(nsid = %nsid, expanded = %expanded, "Successfully expanded permission set"); Ok(expanded) } -async fn fetch_lexicon_via_atproto(nsid: &str) -> Result { +async fn fetch_lexicon_via_atproto(nsid: &Nsid) -> Result { let parts: Vec<&str> = nsid.split('.').collect(); - if parts.len() < 3 { - return Err(ScopeExpansionError::InvalidNsid(nsid.to_string())); - } - let authority = parts[..parts.len() - 1] .iter() .rev() .cloned() .collect::>() .join("."); - debug!(nsid, authority = %authority, "Resolving lexicon DID authority via DNS"); + debug!(nsid = %nsid, authority = %authority, "Resolving lexicon DID authority via DNS"); let did = resolve_lexicon_did_authority(&authority).await?; - debug!(nsid, did = %did, "Resolved lexicon DID authority"); + debug!(nsid = %nsid, did = %did, "Resolved lexicon DID authority"); let pds_endpoint = resolve_did_to_pds(&did).await?; - debug!(nsid, pds = %pds_endpoint, "Resolved DID to PDS endpoint"); + debug!(nsid = %nsid, pds = %pds_endpoint, "Resolved DID to PDS endpoint"); let client = Client::builder() .timeout(std::time::Duration::from_secs(10)) @@ -194,10 +193,10 @@ async fn fetch_lexicon_via_atproto(nsid: &str) -> Result Result { )) } -fn extract_namespace_authority(nsid: &str) -> String { +fn extract_namespace_authority(nsid: &Nsid) -> String { let parts: Vec<&str> = nsid.split('.').collect(); - if parts.len() >= 2 { - parts[..parts.len() - 1].join(".") - } else { - nsid.to_string() - } + parts[..parts.len() - 1].join(".") } fn is_under_authority(target_nsid: &str, authority: &str) -> bool { @@ -385,14 +380,18 @@ mod tests { assert_eq!(aud, Some("did:web:example.com")); } + fn nsid(s: &str) -> Nsid { + s.parse().unwrap() + } + #[test] fn test_extract_namespace_authority() { assert_eq!( - extract_namespace_authority("io.atcr.authFullApp"), + extract_namespace_authority(&nsid("io.atcr.authFullApp")), "io.atcr" ); assert_eq!( - extract_namespace_authority("app.bsky.authFullApp"), + extract_namespace_authority(&nsid("app.bsky.authFullApp")), "app.bsky" ); } @@ -400,7 +399,7 @@ mod tests { #[test] fn test_extract_namespace_authority_deep_nesting() { assert_eq!( - extract_namespace_authority("io.atcr.sailor.star.collection"), + extract_namespace_authority(&nsid("io.atcr.sailor.star.collection")), "io.atcr.sailor.star" ); } @@ -601,7 +600,7 @@ mod tests { ); } - let result = expand_permission_set(cache_key, None).await; + let result = expand_permission_set(&nsid(cache_key), None).await; assert!(result.is_ok()); assert_eq!(result.unwrap(), cached_value); @@ -629,7 +628,7 @@ mod tests { ); } - let result = expand_permission_set(nsid, Some(aud)).await; + let result = expand_permission_set(&nsid.parse().unwrap(), Some(aud)).await; assert!(result.is_ok()); assert_eq!(result.unwrap(), cached_value); @@ -655,7 +654,7 @@ mod tests { ); } - let result = expand_permission_set(cache_key, None).await; + let result = expand_permission_set(&nsid(cache_key), None).await; assert!(result.is_err()); { diff --git a/crates/tranquil-scopes/src/permissions.rs b/crates/tranquil-scopes/src/permissions.rs index 7835e2d..ce8a03c 100644 --- a/crates/tranquil-scopes/src/permissions.rs +++ b/crates/tranquil-scopes/src/permissions.rs @@ -4,6 +4,7 @@ use super::parser::{ RepoScope, RpcScope, parse_scope_string, }; use std::collections::HashSet; +use tranquil_types::Nsid; #[derive(Debug, Clone)] pub struct ScopePermissions { @@ -105,7 +106,7 @@ impl ScopePermissions { }) } - pub fn assert_repo(&self, action: RepoAction, collection: &str) -> Result<(), ScopeError> { + pub fn assert_repo(&self, action: RepoAction, collection: &Nsid) -> Result<(), ScopeError> { if self.has_transition_generic { return Ok(()); } @@ -156,7 +157,7 @@ impl ScopePermissions { } } - pub fn assert_rpc(&self, aud: &str, lxm: &str) -> Result<(), ScopeError> { + pub fn assert_rpc(&self, aud: &str, lxm: &Nsid) -> Result<(), ScopeError> { if lxm.starts_with("chat.bsky.") { if self.has_transition_chat { return Ok(()); @@ -256,7 +257,7 @@ impl ScopePermissions { .any(|a| a.attr == AccountAttr::Email || a.attr == AccountAttr::Wildcard) } - pub fn allows_repo(&self, action: RepoAction, collection: &str) -> bool { + pub fn allows_repo(&self, action: RepoAction, collection: &Nsid) -> bool { self.assert_repo(action, collection).is_ok() } @@ -264,7 +265,7 @@ impl ScopePermissions { self.assert_blob(mime).is_ok() } - pub fn allows_rpc(&self, aud: &str, lxm: &str) -> bool { + pub fn allows_rpc(&self, aud: &str, lxm: &Nsid) -> bool { self.assert_rpc(aud, lxm).is_ok() } @@ -341,6 +342,10 @@ impl Default for ScopePermissions { mod tests { use super::*; + fn c(s: &str) -> Nsid { + s.parse().unwrap() + } + #[test] fn test_atproto_scope_is_identity_only() { let perms = ScopePermissions::from_scope_string(Some("atproto")); @@ -388,7 +393,7 @@ mod tests { let perms = ScopePermissions::from_scope_string(None); assert!(perms.has_scope("atproto")); assert!(!perms.has_full_access()); - assert!(!perms.allows_repo(RepoAction::Create, "any.collection")); + assert!(!perms.allows_repo(RepoAction::Create, &c("cafe.oyster.record"))); } #[test] @@ -412,8 +417,8 @@ mod tests { fn test_granular_repo_wildcard() { let perms = ScopePermissions::from_scope_string(Some("atproto repo:*?action=create blob:*/*")); - assert!(perms.allows_repo(RepoAction::Create, "app.bsky.feed.post")); - assert!(perms.allows_repo(RepoAction::Create, "any.collection")); + assert!(perms.allows_repo(RepoAction::Create, &c("app.bsky.feed.post"))); + assert!(perms.allows_repo(RepoAction::Create, &c("cafe.oyster.record"))); assert!(perms.allows_blob("image/png")); } @@ -473,9 +478,9 @@ mod tests { fn test_granular_scopes_without_atproto() { let perms = ScopePermissions::from_scope_string(Some("repo:*?action=create")); assert!(!perms.has_full_access()); - assert!(perms.allows_repo(RepoAction::Create, "any.collection")); - assert!(!perms.allows_repo(RepoAction::Update, "any.collection")); - assert!(!perms.allows_repo(RepoAction::Delete, "any.collection")); + assert!(perms.allows_repo(RepoAction::Create, &c("cafe.oyster.record"))); + assert!(!perms.allows_repo(RepoAction::Update, &c("cafe.oyster.record"))); + assert!(!perms.allows_repo(RepoAction::Delete, &c("cafe.oyster.record"))); } #[test] @@ -483,9 +488,9 @@ mod tests { let perms = ScopePermissions::from_scope_string(Some( "atproto repo:*?action=create repo:*?action=update repo:*?action=delete blob:*/*", )); - assert!(perms.allows_repo(RepoAction::Create, "any.collection")); - assert!(perms.allows_repo(RepoAction::Update, "any.collection")); - assert!(perms.allows_repo(RepoAction::Delete, "any.collection")); + assert!(perms.allows_repo(RepoAction::Create, &c("cafe.oyster.record"))); + assert!(perms.allows_repo(RepoAction::Update, &c("cafe.oyster.record"))); + assert!(perms.allows_repo(RepoAction::Delete, &c("cafe.oyster.record"))); assert!(perms.allows_blob("image/png")); assert!(perms.allows_blob("video/mp4")); } @@ -530,9 +535,9 @@ mod tests { let perms = ScopePermissions::from_scope_string(Some("atproto repo:*?action=create blob:*/*")); assert!(!perms.has_full_access()); - assert!(perms.allows_repo(RepoAction::Create, "any.collection")); - assert!(!perms.allows_repo(RepoAction::Delete, "any.collection")); - assert!(!perms.allows_repo(RepoAction::Update, "any.collection")); + assert!(perms.allows_repo(RepoAction::Create, &c("cafe.oyster.record"))); + assert!(!perms.allows_repo(RepoAction::Delete, &c("cafe.oyster.record"))); + assert!(!perms.allows_repo(RepoAction::Update, &c("cafe.oyster.record"))); assert!(perms.allows_blob("image/png")); assert!(!perms.allows_rpc("did:web:api.bsky.app", "app.bsky.feed.getTimeline")); } @@ -541,9 +546,9 @@ mod tests { fn test_atproto_alone_grants_nothing() { let perms = ScopePermissions::from_scope_string(Some("atproto")); assert!(!perms.has_full_access()); - assert!(!perms.allows_repo(RepoAction::Create, "any.collection")); - assert!(!perms.allows_repo(RepoAction::Delete, "any.collection")); - assert!(!perms.allows_repo(RepoAction::Update, "any.collection")); + assert!(!perms.allows_repo(RepoAction::Create, &c("cafe.oyster.record"))); + assert!(!perms.allows_repo(RepoAction::Delete, &c("cafe.oyster.record"))); + assert!(!perms.allows_repo(RepoAction::Update, &c("cafe.oyster.record"))); assert!(!perms.allows_blob("image/png")); assert!(!perms.allows_rpc("did:web:api.bsky.app", "app.bsky.feed.getTimeline")); } diff --git a/crates/tranquil-store/src/metastore/record_ops.rs b/crates/tranquil-store/src/metastore/record_ops.rs index 0fd8a68..34cb8aa 100644 --- a/crates/tranquil-store/src/metastore/record_ops.rs +++ b/crates/tranquil-store/src/metastore/record_ops.rs @@ -78,7 +78,7 @@ impl RecordOps { records: &[RecordWrite<'_>], ) -> Result<(), MetastoreError> { records.iter().try_for_each(|rec| { - let key = record_key(user_hash, rec.collection.as_str(), rec.rkey.as_str()); + let key = record_key(user_hash, rec.collection, rec.rkey); let cid_bytes = cid_link_to_bytes(rec.cid)?; let existing_takedown = self .repo_data @@ -102,7 +102,7 @@ impl RecordOps { records: &[RecordDelete<'_>], ) { records.iter().for_each(|rec| { - let key = record_key(user_hash, rec.collection.as_str(), rec.rkey.as_str()); + let key = record_key(user_hash, rec.collection, rec.rkey); batch.remove(&self.repo_data, key.as_slice()); }); } @@ -126,7 +126,7 @@ impl RecordOps { Some(h) => h, None => return Ok(None), }; - let key = record_key(user_hash, collection.as_str(), rkey.as_str()); + let key = record_key(user_hash, collection, rkey); point_lookup( &self.repo_data, @@ -147,24 +147,23 @@ impl RecordOps { None => return Ok(Vec::new()), }; - let coll_str = query.collection.as_str(); - let coll_prefix = record_collection_prefix(user_hash, coll_str); + let coll_prefix = record_collection_prefix(user_hash, query.collection); let coll_upper = exclusive_upper_bound(coll_prefix.as_slice()) .expect("collection prefix always contains non-0xFF bytes"); let start_key = query .rkey_start - .map(|rs| record_key(user_hash, coll_str, rs.as_str())); + .map(|rs| record_key(user_hash, query.collection, rs)); let end_key_upper = query .rkey_end - .map(|re| record_key(user_hash, coll_str, re.as_str())) + .map(|re| record_key(user_hash, query.collection, re)) .map(|ek| { exclusive_upper_bound(ek.as_slice()) .expect("record key always contains non-0xFF bytes") }); let cursor_key = query .cursor - .map(|c| record_key(user_hash, coll_str, c.as_str())); + .map(|c| record_key(user_hash, query.collection, c)); let mut range_lo: &[u8] = coll_prefix.as_slice(); let mut range_hi: &[u8] = coll_upper.as_slice(); @@ -287,11 +286,12 @@ impl RecordOps { }; let (key_bytes, _) = guard.into_inner().map_err(MetastoreError::Fjall)?; let collection = parse_record_key_collection(&key_bytes) + .map(Nsid::from) .ok_or(MetastoreError::CorruptData("invalid record key"))?; let coll_prefix = record_collection_prefix(user_hash, &collection); seek_from = exclusive_upper_bound(coll_prefix.as_slice()) .expect("collection prefix always contains non-0xFF bytes"); - collections.push(Nsid::from(collection)); + collections.push(collection); } Ok(collections) diff --git a/crates/tranquil-store/src/metastore/records.rs b/crates/tranquil-store/src/metastore/records.rs index 4973f88..8f5c824 100644 --- a/crates/tranquil-store/src/metastore/records.rs +++ b/crates/tranquil-store/src/metastore/records.rs @@ -1,5 +1,6 @@ use serde::{Deserialize, Serialize}; use smallvec::SmallVec; +use tranquil_types::{Nsid, Rkey}; use super::encoding::KeyBuilder; use super::keys::{KeyTag, UserHash}; @@ -30,7 +31,7 @@ impl RecordValue { } } -pub fn record_key(user_hash: UserHash, collection: &str, rkey: &str) -> SmallVec<[u8; 128]> { +pub fn record_key(user_hash: UserHash, collection: &Nsid, rkey: &Rkey) -> SmallVec<[u8; 128]> { KeyBuilder::new() .tag(KeyTag::RECORDS) .u64(user_hash.raw()) @@ -39,7 +40,7 @@ pub fn record_key(user_hash: UserHash, collection: &str, rkey: &str) -> SmallVec .build() } -pub fn record_collection_prefix(user_hash: UserHash, collection: &str) -> SmallVec<[u8; 128]> { +pub fn record_collection_prefix(user_hash: UserHash, collection: &Nsid) -> SmallVec<[u8; 128]> { KeyBuilder::new() .tag(KeyTag::RECORDS) .u64(user_hash.raw()) @@ -63,6 +64,14 @@ mod tests { use super::*; use crate::metastore::encoding::KeyReader; + fn nsid(s: &str) -> Nsid { + s.parse().unwrap() + } + + fn rkey(s: &str) -> Rkey { + s.parse().unwrap() + } + #[test] fn record_value_roundtrip() { let value = RecordValue { @@ -114,7 +123,7 @@ mod tests { #[test] fn record_key_roundtrip() { let hash = UserHash::from_raw(0xDEAD_BEEF_CAFE_BABE); - let key = record_key(hash, "app.bsky.feed.post", "3k2abcd"); + let key = record_key(hash, &nsid("app.bsky.feed.post"), &rkey("3k2abcd")); let mut reader = KeyReader::new(&key); assert_eq!(reader.tag(), Some(KeyTag::RECORDS.raw())); assert_eq!(reader.u64(), Some(0xDEAD_BEEF_CAFE_BABE)); @@ -128,10 +137,10 @@ mod tests { let h1 = UserHash::from_raw(1); let h2 = UserHash::from_raw(2); - let k1 = record_key(h1, "app.bsky.feed.like", "aaa"); - let k2 = record_key(h1, "app.bsky.feed.post", "aaa"); - let k3 = record_key(h1, "app.bsky.feed.post", "bbb"); - let k4 = record_key(h2, "app.bsky.feed.like", "aaa"); + let k1 = record_key(h1, &nsid("app.bsky.feed.like"), &rkey("aaa")); + let k2 = record_key(h1, &nsid("app.bsky.feed.post"), &rkey("aaa")); + let k3 = record_key(h1, &nsid("app.bsky.feed.post"), &rkey("bbb")); + let k4 = record_key(h2, &nsid("app.bsky.feed.like"), &rkey("aaa")); assert!(k1.as_slice() < k2.as_slice()); assert!(k2.as_slice() < k3.as_slice()); @@ -141,8 +150,8 @@ mod tests { #[test] fn collection_prefix_is_prefix_of_full_key() { let hash = UserHash::from_raw(42); - let prefix = record_collection_prefix(hash, "app.bsky.feed.post"); - let full = record_key(hash, "app.bsky.feed.post", "some_rkey"); + let prefix = record_collection_prefix(hash, &nsid("app.bsky.feed.post")); + let full = record_key(hash, &nsid("app.bsky.feed.post"), &rkey("some_rkey")); assert!(full.as_slice().starts_with(prefix.as_slice())); } @@ -150,7 +159,7 @@ mod tests { fn user_prefix_is_prefix_of_collection_prefix() { let hash = UserHash::from_raw(42); let user_pfx = record_user_prefix(hash); - let coll_pfx = record_collection_prefix(hash, "app.bsky.feed.post"); + let coll_pfx = record_collection_prefix(hash, &nsid("app.bsky.feed.post")); assert!(coll_pfx.as_slice().starts_with(user_pfx.as_slice())); } diff --git a/crates/tranquil-store/src/metastore/recovery.rs b/crates/tranquil-store/src/metastore/recovery.rs index f716672..7bc6028 100644 --- a/crates/tranquil-store/src/metastore/recovery.rs +++ b/crates/tranquil-store/src/metastore/recovery.rs @@ -1,6 +1,7 @@ use std::collections::HashSet; use serde::{Deserialize, Serialize}; +use tranquil_types::{Nsid, Rkey}; use super::backlink_ops::remove_backlinks_for_record; use super::backlinks::{BacklinkValue, backlink_by_user_key, backlink_key, discriminant_to_path}; @@ -117,7 +118,11 @@ pub fn replay_mutation_set( batch.insert(repo_data, meta_key.as_slice(), updated_meta.serialize()); mutation_set.record_upserts.iter().for_each(|u| { - let key = record_key(user_hash, &u.collection, &u.rkey); + let key = record_key( + user_hash, + &Nsid::from(u.collection.clone()), + &Rkey::from(u.rkey.clone()), + ); let value = RecordValue { record_cid: u.cid_bytes.clone(), takedown_ref: None, @@ -126,7 +131,11 @@ pub fn replay_mutation_set( }); mutation_set.record_deletes.iter().for_each(|d| { - let key = record_key(user_hash, &d.collection, &d.rkey); + let key = record_key( + user_hash, + &Nsid::from(d.collection.clone()), + &Rkey::from(d.rkey.clone()), + ); batch.remove(repo_data, key.as_slice()); });