mirrors/tranquil-pds
1
0
Fork 0
mirror of https://tangled.org/tranquil.farm/tranquil-pds synced 2026-07-20 23:12:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

lexicon: Nsid instead of strs for collections

Browse Source 
Lewis: May this revision serve well! <lu5a@proton.me>
This commit is contained in:
Lewis
2026-07-12 00:21:32 +03:00
parent f330dcd366
commit 3c46e5fc73
32 changed files with 427 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/tranquil-api/src/identity/account.rs
+1
View File
@@ -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
crates/tranquil-api/src/moderation/mod.rs
+2
View File
@@ -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,
crates/tranquil-api/src/repo/blob.rs
+1 -1
View File
@@ -48,7 +48,7 @@ pub async fn upload_blob(
) -> Result<Response, ApiError> {
let (did, controller_did): (Did, Option<Did>) = 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) => {
crates/tranquil-api/src/repo/record/batch.rs
+3 -3
View File
@@ -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,
crates/tranquil-api/src/repo/record/validation.rs
+3 -3
View File
@@ -9,13 +9,13 @@ pub async fn validate_record_with_status(
require_lexicon: bool,
) -> Result<ValidationStatus, ApiError> {
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)
}
crates/tranquil-api/src/server/passkey_account.rs
+1
View File
@@ -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
crates/tranquil-api/src/server/service_auth.rs
+8 -12
View File
@@ -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, &params.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 {
crates/tranquil-auth/src/token.rs
+2 -2
View File
@@ -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<String> {
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<String> {
let expiration = Utc::now()
crates/tranquil-auth/src/types.rs
+1 -1
View File
@@ -216,7 +216,7 @@ pub struct Claims {
#[serde(skip_serializing_if = "Option::is_none")]
pub scope: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub lxm: Option<String>,
pub lxm: Option<Nsid>,
pub jti: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub act: Option<ActClaim>,
crates/tranquil-lexicon/src/dynamic.rs
+100 -74
View File
@@ -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<String, PositiveEntry>,
insertion_order: VecDeque<String>,
schemas: HashMap<Nsid, PositiveEntry>,
insertion_order: VecDeque<Nsid>,
}
pub struct DynamicRegistry {
store: RwLock<SchemaStore>,
negative_cache: RwLock<HashMap<String, NegativeEntry>>,
in_flight: RwLock<HashMap<String, Arc<Notify>>>,
negative_cache: RwLock<HashMap<Nsid, NegativeEntry>>,
in_flight: RwLock<HashMap<Nsid, Arc<Notify>>>,
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<Arc<LexiconDoc>> {
pub fn get_cached(&self, nsid: &Nsid) -> Option<Arc<LexiconDoc>> {
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<CacheEntry> {
pub(crate) fn get_entry(&self, nsid: &Nsid) -> Option<CacheEntry> {
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<Arc<LexiconDoc>, ResolveError> {
pub async fn resolve_and_cache(&self, nsid: &Nsid) -> Result<Arc<LexiconDoc>, ResolveError> {
self.resolve_and_cache_with(nsid, |n| async move { resolve_lexicon(&n).await })
.await
}
async fn resolve_and_cache_with<F, Fut>(
&self,
nsid: &str,
nsid: &Nsid,
resolver: F,
) -> Result<Arc<LexiconDoc>, ResolveError>
where
F: FnOnce(String) -> Fut,
F: FnOnce(Nsid) -> Fut,
Fut: std::future::Future<Output = Result<LexiconDoc, ResolveError>>,
{
match self.get_entry(nsid) {
@@ -188,12 +189,12 @@ impl DynamicRegistry {
async fn refresh_stale<F, Fut>(
&self,
nsid: &str,
nsid: &Nsid,
stale: Arc<LexiconDoc>,
resolver: F,
) -> Result<Arc<LexiconDoc>, ResolveError>
where
F: FnOnce(String) -> Fut,
F: FnOnce(Nsid) -> Fut,
Fut: std::future::Future<Output = Result<LexiconDoc, ResolveError>>,
{
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<F, Fut>(
&self,
nsid: &str,
nsid: &Nsid,
resolver: F,
) -> Result<Arc<LexiconDoc>, ResolveError>
where
F: FnOnce(String) -> Fut,
F: FnOnce(Nsid) -> Fut,
Fut: std::future::Future<Output = Result<LexiconDoc, ResolveError>>,
{
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<InFlightGuard<'_>> {
fn acquire_leadership(&self, nsid: &Nsid) -> Option<InFlightGuard<'_>> {
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::<LexiconDoc, _>(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<T>(handles: Vec<tokio::task::JoinHandle<T>>) -> Vec<T> {
@@ -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(),
});
});
crates/tranquil-lexicon/src/registry.rs
+33 -18
View File
@@ -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<LexiconRegistry> = OnceLock::new();
pub struct LexiconRegistry {
schemas: HashMap<String, Arc<LexiconDoc>>,
schemas: HashMap<Nsid, Arc<LexiconDoc>>,
#[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<Arc<LexiconDoc>> {
self.schemas.get(nsid).cloned().or_else(|| {
pub fn get_doc(&self, nsid: &Nsid) -> Option<Arc<LexiconDoc>> {
self.get_doc_by_key(nsid.as_str())
}
fn get_doc_by_key(&self, key: &str) -> Option<Arc<LexiconDoc>> {
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<Arc<LexiconDoc>> {
pub fn get_record_def(&self, nsid: &Nsid) -> Option<Arc<LexiconDoc>> {
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<Arc<LexiconDoc>, 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")));
}
}
crates/tranquil-lexicon/src/resolve.rs
+30 -28
View File
@@ -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<Client> = 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<String, ResolveError> {
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<String> {
pub async fn fetch_schema_from_pds(
pds_endpoint: &str,
did: &str,
nsid: &str,
nsid: &Nsid,
) -> Result<LexiconDoc, ResolveError> {
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<LexiconDoc, ResolveError> {
pub async fn resolve_lexicon(nsid: &Nsid) -> Result<LexiconDoc, ResolveError> {
resolve_lexicon_with_config(nsid, None).await
}
pub async fn resolve_lexicon_with_config(
nsid: &str,
nsid: &Nsid,
plc_directory_url: Option<&str>,
) -> Result<LexiconDoc, ResolveError> {
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<LexiconDoc, ResolveError> {
@@ -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(_)));
}
}
crates/tranquil-lexicon/src/schema.rs
+2 -1
View File
@@ -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<String, LexDef>,
}
crates/tranquil-lexicon/src/validate.rs
+38 -31
View File
@@ -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(&registry, "com.test.basic", &record).is_ok());
assert!(validate_record(&registry, &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(&registry, "com.test.basic", &record).unwrap_err();
let err = validate_record(&registry, &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(&registry, "com.test.basic", &record).unwrap_err();
let err = validate_record(&registry, &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(&registry, "com.test.basic", &record).unwrap_err();
let err = validate_record(&registry, &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(&registry, "com.test.profile", &record).unwrap_err();
let err = validate_record(&registry, &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(&registry, "com.test.basic", &record).unwrap_err();
let err = validate_record(&registry, &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(&registry, "com.test.basic", &record_neg).unwrap_err();
let err = validate_record(&registry, &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(&registry, "com.test.basic", &record).is_ok());
assert!(validate_record(&registry, &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(&registry, "com.test.basic", &record_frac).unwrap_err();
let err = validate_record(&registry, &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(&registry, "com.test.basic", &record).unwrap_err();
let err = validate_record(&registry, &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(&registry, "com.test.basic", &record).unwrap_err();
let err = validate_record(&registry, &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(&registry, "com.test.basic", &record).is_ok());
assert!(validate_record(&registry, &nsid("com.test.basic"), &record).is_ok());
}
#[test]
@@ -678,7 +683,7 @@ mod tests {
},
"createdAt": "2024-01-01T00:00:00.000Z"
});
assert!(validate_record(&registry, "com.test.withref", &record).is_ok());
assert!(validate_record(&registry, &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(&registry, "com.test.withref", &record).unwrap_err();
let err = validate_record(&registry, &nsid("com.test.withref"), &record).unwrap_err();
assert!(matches!(err, LexValidationError::MissingRequired { .. }));
}
@@ -713,7 +718,7 @@ mod tests {
}
}
});
assert!(validate_record(&registry, "com.test.withreply", &record).is_ok());
assert!(validate_record(&registry, &nsid("com.test.withreply"), &record).is_ok());
}
#[test]
@@ -738,7 +743,7 @@ mod tests {
]
}
});
assert!(validate_record(&registry, "com.test.withreply", &record).is_ok());
assert!(validate_record(&registry, &nsid("com.test.withreply"), &record).is_ok());
let bad_embed = json!({
"$type": "com.test.withreply",
@@ -750,7 +755,7 @@ mod tests {
}
});
assert!(
validate_record(&registry, "com.test.withreply", &bad_embed).is_err(),
validate_record(&registry, &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(&registry, "com.test.withreply", &record).is_ok());
assert!(validate_record(&registry, &nsid("com.test.withreply"), &record).is_ok());
let bad_external = json!({
"$type": "com.test.withreply",
@@ -866,7 +871,7 @@ mod tests {
}
});
assert!(
validate_record(&registry, "com.test.withreply", &bad_external).is_err(),
validate_record(&registry, &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(&registry, "com.test.withgate", &record).is_ok());
assert!(validate_record(&registry, &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(&registry, "com.test.withdid", &record).is_ok());
assert!(validate_record(&registry, &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(&registry, "com.test.withdid", &bad_did).is_err());
assert!(validate_record(&registry, &nsid("com.test.withdid"), &bad_did).is_err());
}
#[test]
@@ -911,14 +916,15 @@ mod tests {
"name": "test",
"value": null
});
assert!(validate_record(&registry, "com.test.nullable", &record).is_ok());
assert!(validate_record(&registry, &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(&registry, "com.example.nonexistent", &record).unwrap_err();
let err =
validate_record(&registry, &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(&registry, "com.test.basic", &record).is_ok());
assert!(validate_record(&registry, &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(&registry, "com.test.profile", &record).is_ok());
assert!(validate_record(&registry, &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(&registry, "com.test.profile", &record).unwrap_err();
let err = validate_record(&registry, &nsid("com.test.profile"), &record).unwrap_err();
assert!(matches!(err, LexValidationError::InvalidField { .. }));
}
@@ -961,7 +967,7 @@ mod tests {
"value": null
});
assert!(
validate_record(&registry, "com.test.requirednullable", &record).is_ok(),
validate_record(&registry, &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(&registry, "com.test.requirednullable", &record).unwrap_err(),
validate_record(&registry, &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(&registry, "com.test.requirednullable", &record).is_ok(),
validate_record(&registry, &nsid("com.test.requirednullable"), &record).is_ok(),
"a field that is required+nullable must accept non-null values"
);
}
crates/tranquil-lexicon/tests/resolve_integration.rs
+51 -14
View File
@@ -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!(
crates/tranquil-pds/src/api/proxy.rs
+11 -1
View File
@@ -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::<Nsid>() 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::<Nsid>()
.expect("getFeedSkeleton is a valid NSID"),
),
None => {
warn!(
"getFeed proxy: could not resolve feed generator DID; refusing \
crates/tranquil-pds/src/auth/extractor.rs
+7 -3
View File
@@ -340,7 +340,11 @@ impl<P: AuthPolicy> Auth<P> {
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<P: AuthPolicy> AuthAny<P> {
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),
crates/tranquil-pds/src/auth/mod.rs
+7 -3
View File
@@ -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(
crates/tranquil-pds/src/auth/scope_check.rs
+3 -2
View File
@@ -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(());
crates/tranquil-pds/src/auth/scope_verified.rs
+10 -10
View File
@@ -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<BatchWriteScopes<'a>, 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<ScopeVerified<'a, RepoCreate>, ScopeVerificationError>
where
Self: AsRef<AuthenticatedUser>,
@@ -322,7 +322,7 @@ pub trait VerifyScope {
fn verify_repo_update<'a>(
&'a self,
collection: &str,
collection: &Nsid,
) -> Result<ScopeVerified<'a, RepoUpdate>, ScopeVerificationError>
where
Self: AsRef<AuthenticatedUser>,
@@ -344,7 +344,7 @@ pub trait VerifyScope {
fn verify_repo_delete<'a>(
&'a self,
collection: &str,
collection: &Nsid,
) -> Result<ScopeVerified<'a, RepoDelete>, ScopeVerificationError>
where
Self: AsRef<AuthenticatedUser>,
@@ -366,7 +366,7 @@ pub trait VerifyScope {
fn verify_repo_upsert<'a>(
&'a self,
collection: &str,
collection: &Nsid,
) -> Result<ScopeVerified<'a, RepoUpsert>, ScopeVerificationError>
where
Self: AsRef<AuthenticatedUser>,
@@ -414,7 +414,7 @@ pub trait VerifyScope {
fn verify_rpc<'a>(
&'a self,
aud: &str,
lxm: &str,
lxm: &Nsid,
) -> Result<ScopeVerified<'a, RpcCall>, ScopeVerificationError>
where
Self: AsRef<AuthenticatedUser>,
crates/tranquil-pds/src/auth/service.rs
+1 -1
View File
@@ -170,7 +170,7 @@ impl ServiceTokenVerifier {
pub async fn verify_service_token(
&self,
token: &str,
required_lxm: Option<&str>,
required_lxm: Option<&Nsid>,
) -> Result<ServiceTokenClaims, ServiceTokenError> {
let jwt = JwtParts::parse(token)?;
crates/tranquil-pds/src/sync/import.rs
+2 -2
View File
@@ -171,7 +171,7 @@ pub fn extract_links(value: &Ipld, links: &mut Vec<Cid>) {
#[derive(Debug)]
pub struct ImportedRecord {
pub collection: String,
pub collection: Nsid,
pub rkey: String,
pub cid: Cid,
pub blob_refs: Vec<BlobRef>,
@@ -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,
crates/tranquil-pds/src/validation/mod.rs
+6 -5
View File
@@ -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<ValidationStatus, ValidationError> {
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<ValidationStatus, ValidationError> {
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<String, Value>), 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(),
crates/tranquil-pds/tests/jwt_security.rs
+1
View File
@@ -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<u8> {
let secret_key = SecretKey::random(&mut OsRng);
crates/tranquil-pds/tests/record_validation.rs
+5
View File
@@ -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();
crates/tranquil-pds/tests/repo_batch.rs
+1 -1
View File
@@ -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() {
crates/tranquil-pds/tests/scope_edge_cases.rs
+8 -3
View File
@@ -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]
crates/tranquil-scopes/src/permission_set.rs
+26 -27
View File
@@ -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<String, ScopeEx
match scope.strip_prefix("include:") {
Some(rest) => {
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<String, ScopeExpansionError> {
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<LexiconDoc, ScopeExpansionError> {
async fn fetch_lexicon_via_atproto(nsid: &Nsid) -> Result<LexiconDoc, ScopeExpansionError> {
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::<Vec<_>>()
.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<LexiconDoc, ScopeExpans
let url = format!(
"{}/xrpc/com.atproto.repo.getRecord?repo={}&collection=com.atproto.lexicon.schema&rkey={}",
pds_endpoint,
urlencoding::encode(&did),
urlencoding::encode(nsid)
urlencoding::encode(did.as_str()),
urlencoding::encode(nsid.as_str())
);
debug!(nsid, url = %url, "Fetching lexicon from PDS");
debug!(nsid = %nsid, url = %url, "Fetching lexicon from PDS");
let response = client
.get(&url)
@@ -296,13 +295,9 @@ async fn resolve_did_to_pds(did: &str) -> Result<String, ScopeExpansionError> {
))
}
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());
{
crates/tranquil-scopes/src/permissions.rs
+24 -19
View File
@@ -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"));
}
crates/tranquil-store/src/metastore/record_ops.rs
+9 -9
View File
@@ -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)
crates/tranquil-store/src/metastore/records.rs
+19 -10
View File
@@ -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()));
}
crates/tranquil-store/src/metastore/recovery.rs
+11 -2
View File
@@ -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());
});