diff --git a/crates/tranquil-lexicon/Cargo.toml b/crates/tranquil-lexicon/Cargo.toml index 99e7eb0..2974aac 100644 --- a/crates/tranquil-lexicon/Cargo.toml +++ b/crates/tranquil-lexicon/Cargo.toml @@ -5,10 +5,11 @@ edition.workspace = true license.workspace = true [features] -resolve = ["dep:reqwest", "dep:hickory-resolver", "dep:tokio", "dep:parking_lot", "dep:tracing", "dep:urlencoding"] +resolve = ["dep:reqwest", "dep:hickory-resolver", "dep:tokio", "dep:parking_lot", "dep:tracing", "dep:tranquil-infra"] [dependencies] -tranquil-types = { path = "../tranquil-types", default-features = false } +tranquil-types = { workspace = true } +tranquil-infra = { workspace = true, optional = true, features = ["cache-keys"] } serde = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } @@ -19,9 +20,9 @@ hickory-resolver = { workspace = true, optional = true } tokio = { workspace = true, optional = true } parking_lot = { workspace = true, optional = true } tracing = { workspace = true, optional = true } -urlencoding = { workspace = true, optional = true } [dev-dependencies] wiremock = { workspace = true } tokio = { workspace = true } futures = { workspace = true } +tranquil-infra = { workspace = true, features = ["testing", "cache-keys"] } diff --git a/crates/tranquil-lexicon/src/dynamic.rs b/crates/tranquil-lexicon/src/dynamic.rs index 96581e2..5d5db99 100644 --- a/crates/tranquil-lexicon/src/dynamic.rs +++ b/crates/tranquil-lexicon/src/dynamic.rs @@ -6,9 +6,11 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::{Duration, Instant}; use tokio::sync::Notify; +use tranquil_infra::cache_keys::{lexicon_doc_key, lexicon_negative_key}; +use tranquil_infra::{Cache, read_json, write_json}; use tranquil_types::Nsid; -const NEGATIVE_CACHE_TTL: Duration = Duration::from_secs(24 * 60 * 60); +const NEGATIVE_CACHE_TTL: Duration = Duration::from_secs(60 * 60); const POSITIVE_CACHE_TTL: Duration = Duration::from_secs(24 * 60 * 60); const REFRESH_FAILURE_BACKOFF: Duration = Duration::from_secs(60); const MAX_DYNAMIC_SCHEMAS: usize = 1024; @@ -17,6 +19,13 @@ struct NegativeEntry { expires_at: Instant, } +fn negative_ttl_for(error: &ResolveError) -> Duration { + match error.is_definitive() { + true => NEGATIVE_CACHE_TTL, + false => REFRESH_FAILURE_BACKOFF, + } +} + struct PositiveEntry { doc: Arc, expires_at: Instant, @@ -44,6 +53,7 @@ pub struct DynamicRegistry { negative_cache: RwLock>, in_flight: RwLock>>, network_disabled: AtomicBool, + shared: RwLock>>, } struct InFlightGuard<'a> { @@ -70,9 +80,18 @@ impl DynamicRegistry { negative_cache: RwLock::new(HashMap::new()), in_flight: RwLock::new(HashMap::new()), network_disabled: AtomicBool::new(false), + shared: RwLock::new(None), } } + pub fn set_shared_cache(&self, cache: Arc) { + *self.shared.write() = Some(cache); + } + + fn shared_cache(&self) -> Option> { + self.shared.read().clone() + } + pub fn from_env() -> Self { let registry = Self::new(); let disabled = @@ -105,13 +124,17 @@ impl DynamicRegistry { } 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()) + self.negative_remaining(nsid).is_some() } - fn insert_negative(&self, nsid: &Nsid) { + fn negative_remaining(&self, nsid: &Nsid) -> Option { + self.negative_cache + .read() + .get(nsid) + .and_then(|entry| entry.expires_at.checked_duration_since(Instant::now())) + } + + fn insert_negative(&self, nsid: &Nsid, ttl: Duration) { let mut cache = self.negative_cache.write(); if cache.len() >= MAX_DYNAMIC_SCHEMAS { let now = Instant::now(); @@ -120,7 +143,7 @@ impl DynamicRegistry { cache.insert( nsid.clone(), NegativeEntry { - expires_at: Instant::now() + NEGATIVE_CACHE_TTL, + expires_at: Instant::now() + ttl, }, ); } @@ -159,6 +182,44 @@ impl DynamicRegistry { arc } + async fn shared_get(&self, nsid: &Nsid) -> Option> { + let cache = self.shared_cache()?; + let doc = read_json::(cache.as_ref(), &lexicon_doc_key(nsid)).await?; + Some(self.insert_schema(doc)) + } + + async fn shared_put(&self, doc: &LexiconDoc) { + let Some(cache) = self.shared_cache() else { + return; + }; + write_json( + cache.as_ref(), + &lexicon_doc_key(&doc.id), + doc, + POSITIVE_CACHE_TTL, + ) + .await; + let _ = cache.delete(&lexicon_negative_key(&doc.id)).await; + } + + async fn shared_is_negative(&self, nsid: &Nsid) -> bool { + match self.shared_cache() { + Some(cache) => cache.get(&lexicon_negative_key(nsid)).await.is_some(), + None => false, + } + } + + async fn shared_put_negative(&self, nsid: &Nsid, error: &ResolveError) { + if !error.is_definitive() { + return; + } + if let Some(cache) = self.shared_cache() { + let _ = cache + .set(&lexicon_negative_key(nsid), "1", NEGATIVE_CACHE_TTL) + .await; + } + } + fn bump_expiry(&self, nsid: &Nsid, duration: Duration) { let mut store = self.store.write(); if let Some(entry) = store.schemas.get_mut(nsid) { @@ -203,15 +264,23 @@ impl DynamicRegistry { match self.acquire_leadership(nsid) { Some(_guard) => match resolver(nsid.clone()).await { - Ok(doc) => Ok(self.insert_schema(doc)), + Ok(doc) => { + self.shared_put(&doc).await; + Ok(self.insert_schema(doc)) + } Err(e) => { + let (doc, source) = match self.shared_get(nsid).await { + Some(doc) => (doc, "shared"), + None => (stale, "local"), + }; self.bump_expiry(nsid, REFRESH_FAILURE_BACKOFF); tracing::warn!( nsid = %nsid, error = %e, - "lexicon refresh failed, serving stale cached entry" + source, + "lexicon refresh failed, serving cached entry" ); - Ok(stale) + Ok(doc) } }, None => { @@ -230,34 +299,59 @@ impl DynamicRegistry { F: FnOnce(Nsid) -> Fut, Fut: std::future::Future>, { - if self.network_disabled.load(Ordering::Relaxed) { - return Err(ResolveError::NetworkDisabled); + if let Some(doc) = self.shared_get(nsid).await { + return Ok(doc); } - if self.is_negative_cached(nsid) { + + if let Some(remaining) = self.negative_remaining(nsid) { return Err(ResolveError::NegativelyCached { nsid: nsid.clone(), - ttl_secs: NEGATIVE_CACHE_TTL.as_secs(), + ttl_secs: remaining.as_secs(), }); } + if self.shared_is_negative(nsid).await { + // Cache reports 0 remaining TTL for shared negative hit, + // so we mirror for the backoff rather than a full `NEGATIVE_CACHE_TTL`. + self.insert_negative(nsid, REFRESH_FAILURE_BACKOFF); + return Err(ResolveError::NegativelyCached { + nsid: nsid.clone(), + ttl_secs: REFRESH_FAILURE_BACKOFF.as_secs(), + }); + } + + if self.network_disabled.load(Ordering::Relaxed) { + return Err(ResolveError::NetworkDisabled); + } + match self.acquire_leadership(nsid) { Some(_guard) => match resolver(nsid.clone()).await { - Ok(doc) => Ok(self.insert_schema(doc)), + Ok(doc) => { + self.shared_put(&doc).await; + Ok(self.insert_schema(doc)) + } Err(e) => { - self.insert_negative(nsid); - tracing::debug!(nsid = %nsid, error = %e, "caching negative resolution result"); + let ttl = negative_ttl_for(&e); + self.insert_negative(nsid, ttl); + self.shared_put_negative(nsid, &e).await; + tracing::debug!( + nsid = %nsid, + error = %e, + ttl_secs = ttl.as_secs(), + "caching negative resolution result" + ); Err(e) } }, None => { self.wait_for_leader(nsid).await; - match self.get_cached(nsid) { - Some(doc) => Ok(doc), - None if self.is_negative_cached(nsid) => Err(ResolveError::NegativelyCached { + match (self.get_cached(nsid), self.negative_remaining(nsid)) { + (Some(doc), _) => Ok(doc), + (None, Some(remaining)) => Err(ResolveError::NegativelyCached { nsid: nsid.clone(), - ttl_secs: NEGATIVE_CACHE_TTL.as_secs(), + ttl_secs: remaining.as_secs(), }), - None => Err(ResolveError::LeaderAborted { nsid: nsid.clone() }), + (None, None) => Err(ResolveError::LeaderAborted { nsid: nsid.clone() }), } } } @@ -316,6 +410,7 @@ impl Default for DynamicRegistry { #[cfg(test)] mod tests { use super::*; + use tranquil_infra::MemoryCache; fn nsid(s: &str) -> Nsid { s.parse().unwrap() @@ -324,19 +419,19 @@ mod tests { #[test] fn test_negative_cache() { let registry = DynamicRegistry::new(); - assert!(!registry.is_negative_cached(&nsid("com.example.test"))); + assert!(!registry.is_negative_cached(&nsid("pet.nel.negative"))); - registry.insert_negative(&nsid("com.example.test")); - assert!(registry.is_negative_cached(&nsid("com.example.test"))); + registry.insert_negative(&nsid("pet.nel.negative"), NEGATIVE_CACHE_TTL); + assert!(registry.is_negative_cached(&nsid("pet.nel.negative"))); } #[tokio::test] async fn test_negative_cache_returns_appropriate_error_variant() { let registry = DynamicRegistry::new(); - registry.insert_negative(&nsid("com.example.cached")); + registry.insert_negative(&nsid("pet.nel.cached"), NEGATIVE_CACHE_TTL); let err = registry - .resolve_and_cache(&nsid("com.example.cached")) + .resolve_and_cache(&nsid("pet.nel.cached")) .await .unwrap_err(); @@ -383,17 +478,17 @@ mod tests { fn test_negative_cache_cleared_on_insert() { let registry = DynamicRegistry::new(); - registry.insert_negative(&nsid("com.example.test")); - assert!(registry.is_negative_cached(&nsid("com.example.test"))); + registry.insert_negative(&nsid("pet.nel.cleared"), NEGATIVE_CACHE_TTL); + assert!(registry.is_negative_cached(&nsid("pet.nel.cleared"))); let doc = LexiconDoc { lexicon: 1, - id: nsid("com.example.test"), + id: nsid("pet.nel.cleared"), defs: HashMap::new(), }; registry.insert_schema(doc); - assert!(!registry.is_negative_cached(&nsid("com.example.test"))); + assert!(!registry.is_negative_cached(&nsid("pet.nel.cleared"))); } #[test] @@ -692,4 +787,95 @@ mod tests { "evicted Arc should be freed when no external references remain" ); } + + #[tokio::test] + async fn test_shared_positive_hit_skips_resolver() { + let registry = DynamicRegistry::new(); + let cache = Arc::new(MemoryCache::new()); + registry.set_shared_cache(cache.clone()); + let doc = LexiconDoc { + lexicon: 1, + id: nsid("pet.nel.sharedDoc"), + defs: HashMap::new(), + }; + cache + .set( + &lexicon_doc_key(&nsid("pet.nel.sharedDoc")), + &serde_json::to_string(&doc).unwrap(), + POSITIVE_CACHE_TTL, + ) + .await + .unwrap(); + + let resolved = registry + .resolve_and_cache_with(&nsid("pet.nel.sharedDoc"), |_| async move { + panic!("resolver mustn't run on a shared positive hit") + }) + .await + .unwrap(); + + assert_eq!(resolved.id, "pet.nel.sharedDoc"); + assert!(registry.get_cached(&nsid("pet.nel.sharedDoc")).is_some()); + } + + #[tokio::test] + async fn test_definitive_failure_writes_shared_negative_and_peers_mirror_it() { + let cache = Arc::new(MemoryCache::new()); + let registry = DynamicRegistry::new(); + registry.set_shared_cache(cache.clone()); + + let _ = registry + .resolve_and_cache_with(&nsid("pet.nel.gone"), |n| async move { + Err::(ResolveError::SchemaNotFound { + nsid: n, + url: "https://oyster.cafe".to_string(), + }) + }) + .await; + assert!( + cache + .get(&lexicon_negative_key(&nsid("pet.nel.gone"))) + .await + .is_some(), + "definitive failure must write the shared negative key" + ); + + let _ = registry + .resolve_and_cache_with(&nsid("pet.nel.transient"), |n| async move { + Err::(ResolveError::DnsLookup { + domain: n.into_inner(), + reason: "simulated".to_string(), + }) + }) + .await; + assert!( + cache + .get(&lexicon_negative_key(&nsid("pet.nel.transient"))) + .await + .is_none(), + "transient failure must stay out of the shared negative key" + ); + + let peer = DynamicRegistry::new(); + peer.set_shared_cache(cache); + let err = peer + .resolve_and_cache_with(&nsid("pet.nel.gone"), |_| async move { + panic!("resolver mustn't run on a shared negative hit") + }) + .await + .unwrap_err(); + match err { + ResolveError::NegativelyCached { ttl_secs, .. } => assert!( + ttl_secs <= REFRESH_FAILURE_BACKOFF.as_secs(), + "local mirror must use the backoff TTL, got {}s", + ttl_secs + ), + other => panic!("expected NegativelyCached, got: {}", other), + } + assert!( + peer.negative_remaining(&nsid("pet.nel.gone")) + .expect("local mirror exists") + <= REFRESH_FAILURE_BACKOFF + ); + } } diff --git a/crates/tranquil-lexicon/src/registry.rs b/crates/tranquil-lexicon/src/registry.rs index 2fa209d..65a14f8 100644 --- a/crates/tranquil-lexicon/src/registry.rs +++ b/crates/tranquil-lexicon/src/registry.rs @@ -125,6 +125,11 @@ impl LexiconRegistry { pub fn is_negative_cached(&self, nsid: &Nsid) -> bool { self.dynamic.is_negative_cached(nsid) } + + #[cfg(feature = "resolve")] + pub fn set_shared_cache(&self, cache: Arc) { + self.dynamic.set_shared_cache(cache); + } } pub struct ResolvedRef { diff --git a/crates/tranquil-lexicon/src/resolve.rs b/crates/tranquil-lexicon/src/resolve.rs index 4662223..c66f69b 100644 --- a/crates/tranquil-lexicon/src/resolve.rs +++ b/crates/tranquil-lexicon/src/resolve.rs @@ -4,7 +4,10 @@ use hickory_resolver::config::{ResolverConfig, ResolverOpts}; use reqwest::Client; use std::sync::OnceLock; use std::time::Duration; -use tranquil_types::{Did, Nsid}; +use tranquil_types::did_doc::extract_pds_endpoint; +use tranquil_types::{ + Did, Nsid, SchemaHostUrl, UrlKind, dns_guard, redirect_policy, url_kind, url_reach_permits, +}; static RESOLVER_CLIENT: OnceLock = OnceLock::new(); @@ -17,7 +20,8 @@ fn client() -> &'static Client { .connect_timeout(Duration::from_secs(5)) .pool_max_idle_per_host(4) .pool_idle_timeout(Duration::from_secs(60)) - .redirect(reqwest::redirect::Policy::limited(3)) + .redirect(redirect_policy(url_kind::SchemaHost::REACH_POLICY)) + .dns_resolver(dns_guard(url_kind::SchemaHost::REACH_POLICY)) .build() .expect("failed to build lexicon resolver HTTP client") }) @@ -63,6 +67,8 @@ pub enum ResolveError { NoPdsEndpoint { did: Did }, #[error("schema fetch failed from {url}: {reason}")] SchemaFetch { url: String, reason: String }, + #[error("no schema record for {nsid} at {url}")] + SchemaNotFound { nsid: Nsid, url: String }, #[error("schema deserialization failed: {0}")] InvalidSchema(String), #[error("schema resolution recently failed for {nsid}, cached for {ttl_secs}s")] @@ -73,6 +79,23 @@ pub enum ResolveError { LeaderAborted { nsid: Nsid }, } +impl ResolveError { + pub fn is_definitive(&self) -> bool { + match self { + Self::NoDid { .. } + | Self::NoPdsEndpoint { .. } + | Self::InvalidSchema(_) + | Self::SchemaNotFound { .. } => true, + Self::DnsLookup { .. } + | Self::DidResolution { .. } + | Self::SchemaFetch { .. } + | Self::NegativelyCached { .. } + | Self::NetworkDisabled + | Self::LeaderAborted { .. } => false, + } + } +} + pub fn nsid_to_authority(nsid: &Nsid) -> String { let mut segments: Vec<&str> = nsid.split('.').collect(); segments.pop(); @@ -123,7 +146,7 @@ pub async fn resolve_did_from_dns(authority: &str) -> Result pub async fn resolve_pds_endpoint( did: &Did, plc_directory_url: Option<&str>, -) -> Result { +) -> Result { let plc_base = plc_directory_url.unwrap_or(DEFAULT_PLC_DIRECTORY); let url = match did @@ -131,7 +154,20 @@ pub async fn resolve_pds_endpoint( .and_then(|(_, rest)| rest.split_once(':')) { Some(("plc", _)) => format!("{}/{}", plc_base.trim_end_matches('/'), did), - Some(("web", domain)) => format!("https://{}/.well-known/did.json", domain), + Some(("web", domain)) => { + let url = format!("https://{}/.well-known/did.json", domain); + let permitted = reqwest::Url::parse(&url) + .is_ok_and(|u| url_reach_permits(&u, url_kind::SchemaHost::REACH_POLICY)); + match permitted { + true => url, + false => { + return Err(ResolveError::DidResolution { + did: did.clone(), + reason: "did:web host is outside the allowed host reach".to_string(), + }); + } + } + } _ => { return Err(ResolveError::DidResolution { did: did.clone(), @@ -162,39 +198,29 @@ pub async fn resolve_pds_endpoint( reason: e.to_string(), })?; - extract_pds_endpoint(&doc).ok_or_else(|| ResolveError::NoPdsEndpoint { did: did.clone() }) + extract_pds_endpoint(&doc).map_err(|_| ResolveError::NoPdsEndpoint { did: did.clone() }) } -fn extract_pds_endpoint(doc: &serde_json::Value) -> Option { - doc.get("service") - .and_then(|s| s.as_array()) - .and_then(|services| { - services.iter().find_map(|svc| { - let is_pds = svc - .get("type") - .and_then(|t| t.as_str()) - .is_some_and(|t| t == "AtprotoPersonalDataServer"); - is_pds - .then(|| svc.get("serviceEndpoint").and_then(|ep| ep.as_str()))? - .map(|s| s.to_string()) - }) - }) +fn is_record_absent(xrpc_error: &str, xrpc_message: &str) -> bool { + xrpc_error == "RecordNotFound" + || xrpc_error == "InvalidRequest" && xrpc_message.starts_with("Could not locate record") } pub async fn fetch_schema_from_pds( - pds_endpoint: &str, + pds_endpoint: &SchemaHostUrl, did: &Did, 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.as_str()), - urlencoding::encode(nsid.as_str()) - ); + let mut request_url = pds_endpoint.endpoint("xrpc/com.atproto.repo.getRecord"); + request_url + .query_pairs_mut() + .append_pair("repo", did.as_str()) + .append_pair("collection", "com.atproto.lexicon.schema") + .append_pair("rkey", nsid.as_str()); + let url = request_url.to_string(); let resp = client() - .get(&url) + .get(request_url) .send() .await .map_err(|e| ResolveError::SchemaFetch { @@ -204,10 +230,27 @@ pub async fn fetch_schema_from_pds( let status = resp.status(); if !status.is_success() { - return Err(ResolveError::SchemaFetch { - url, - reason: format!("HTTP {}", status), - }); + let body = read_body_limited(resp, MAX_RESPONSE_BYTES) + .await + .ok() + .and_then(|bytes| serde_json::from_slice::(&bytes).ok()) + .unwrap_or(serde_json::Value::Null); + let field = |name: &str| { + body.get(name) + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string() + }; + return match is_record_absent(&field("error"), &field("message")) { + true => Err(ResolveError::SchemaNotFound { + nsid: nsid.clone(), + url, + }), + false => Err(ResolveError::SchemaFetch { + url, + reason: format!("HTTP {}", status), + }), + }; } let body = read_body_limited(resp, MAX_RESPONSE_BYTES) @@ -292,6 +335,27 @@ mod tests { s.parse().unwrap() } + #[test] + fn is_record_absent_recognizes_only_the_reference_pds_absence_shapes() { + assert!(is_record_absent( + "RecordNotFound", + "Could not locate record: at://did:plc:nel/com.atproto.lexicon.schema/x" + )); + assert!(is_record_absent("RecordNotFound", "")); + assert!(is_record_absent( + "InvalidRequest", + "Could not locate record" + )); + assert!(!is_record_absent( + "InvalidRequest", + "Error: rkey must be a valid record key" + )); + assert!(!is_record_absent("InvalidRequest", "")); + assert!(!is_record_absent("InternalServerError", "")); + assert!(!is_record_absent("RateLimitExceeded", "")); + assert!(!is_record_absent("", "")); + } + #[test] fn test_nsid_to_authority() { assert_eq!( @@ -316,57 +380,6 @@ mod tests { ); } - #[test] - fn test_extract_pds_endpoint_valid() { - let doc = serde_json::json!({ - "service": [{ - "type": "AtprotoPersonalDataServer", - "serviceEndpoint": "https://pds.example.com" - }] - }); - assert_eq!( - extract_pds_endpoint(&doc), - Some("https://pds.example.com".to_string()) - ); - } - - #[test] - fn test_extract_pds_endpoint_multiple_services() { - let doc = serde_json::json!({ - "service": [ - { - "type": "AtprotoLabeler", - "serviceEndpoint": "https://labeler.example.com" - }, - { - "type": "AtprotoPersonalDataServer", - "serviceEndpoint": "https://pds.example.com" - } - ] - }); - assert_eq!( - extract_pds_endpoint(&doc), - Some("https://pds.example.com".to_string()) - ); - } - - #[test] - fn test_extract_pds_endpoint_missing() { - let doc = serde_json::json!({ - "service": [{ - "type": "AtprotoLabeler", - "serviceEndpoint": "https://labeler.example.com" - }] - }); - assert_eq!(extract_pds_endpoint(&doc), None); - } - - #[test] - fn test_extract_pds_endpoint_no_services() { - let doc = serde_json::json!({}); - assert_eq!(extract_pds_endpoint(&doc), None); - } - #[test] fn test_validate_fetched_schema_ok() { let doc = LexiconDoc { diff --git a/crates/tranquil-lexicon/src/schema.rs b/crates/tranquil-lexicon/src/schema.rs index d71b12f..261e766 100644 --- a/crates/tranquil-lexicon/src/schema.rs +++ b/crates/tranquil-lexicon/src/schema.rs @@ -1,8 +1,8 @@ -use serde::Deserialize; +use serde::{Deserialize, Serialize}; use std::collections::HashMap; use tranquil_types::Nsid; -#[derive(Debug, Deserialize)] +#[derive(Debug, Serialize, Deserialize)] pub struct LexiconDoc { pub lexicon: u32, pub id: Nsid, @@ -10,7 +10,7 @@ pub struct LexiconDoc { pub defs: HashMap, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Serialize, Deserialize)] #[serde(tag = "type")] pub enum LexDef { #[serde(rename = "record")] @@ -35,14 +35,14 @@ pub enum LexDef { PermissionSet {}, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Serialize, Deserialize)] pub struct LexRecord { #[serde(default)] pub key: Option, pub record: LexObject, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Serialize, Deserialize)] pub struct LexObject { #[serde(default)] pub required: Vec, @@ -52,7 +52,7 @@ pub struct LexObject { pub properties: HashMap, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Serialize, Deserialize)] #[serde(tag = "type")] pub enum LexProperty { #[serde(rename = "string")] @@ -79,7 +79,7 @@ pub enum LexProperty { Object(LexObject), } -#[derive(Debug, Deserialize)] +#[derive(Debug, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LexString { #[serde(default)] @@ -102,7 +102,7 @@ pub struct LexString { pub default: Option, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Serialize, Deserialize)] pub struct LexInteger { #[serde(default)] pub minimum: Option, @@ -116,7 +116,7 @@ pub struct LexInteger { pub const_value: Option, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LexBytes { #[serde(default)] @@ -125,7 +125,7 @@ pub struct LexBytes { pub min_length: Option, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LexBlob { #[serde(default)] @@ -134,7 +134,7 @@ pub struct LexBlob { pub max_size: Option, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LexArray { pub items: Box, @@ -144,7 +144,7 @@ pub struct LexArray { pub max_length: Option, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Serialize, Deserialize)] pub struct LexUnion { #[serde(default)] pub refs: Vec, @@ -152,14 +152,14 @@ pub struct LexUnion { pub closed: bool, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LexRef { #[serde(rename = "ref")] pub reference: String, } -#[derive(Debug, Clone, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub enum StringFormat { #[serde(rename = "did")] Did, @@ -204,6 +204,6 @@ pub fn parse_ref(reference: &str) -> ParsedRef<'_> { } } -#[derive(Debug, Deserialize)] +#[derive(Debug, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LexStringDef {} diff --git a/crates/tranquil-lexicon/tests/resolve_integration.rs b/crates/tranquil-lexicon/tests/resolve_integration.rs index 68158ee..c0f2c62 100644 --- a/crates/tranquil-lexicon/tests/resolve_integration.rs +++ b/crates/tranquil-lexicon/tests/resolve_integration.rs @@ -74,7 +74,7 @@ async fn test_resolve_pds_endpoint_from_plc() { let endpoint = resolve_pds_endpoint(&did.parse().unwrap(), Some(&plc_server.uri())) .await .unwrap(); - assert_eq!(endpoint, "https://pds.example.com"); + assert_eq!(endpoint.as_str(), "https://pds.example.com"); } #[tokio::test] @@ -130,14 +130,17 @@ async fn test_resolve_pds_endpoint_multiple_services_picks_pds() { "id": did, "service": [ { + "id": "#atproto_labeler", "type": "AtprotoLabeler", "serviceEndpoint": "https://labeler.example.com" }, { + "id": "#bsky_notif", "type": "BskyNotificationService", "serviceEndpoint": "https://notify.example.com" }, { + "id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": "https://pds.example.com" } @@ -149,7 +152,7 @@ async fn test_resolve_pds_endpoint_multiple_services_picks_pds() { let endpoint = resolve_pds_endpoint(&did.parse().unwrap(), Some(&plc_server.uri())) .await .unwrap(); - assert_eq!(endpoint, "https://pds.example.com"); + assert_eq!(endpoint.as_str(), "https://pds.example.com"); } #[tokio::test] @@ -168,7 +171,7 @@ async fn test_fetch_schema_from_pds_success() { .await; let doc = fetch_schema_from_pds( - &pds_server.uri(), + &pds_server.uri().parse().unwrap(), &did.parse().unwrap(), &nsid.parse().unwrap(), ) @@ -195,7 +198,7 @@ async fn test_fetch_schema_missing_value_field() { .await; let result = fetch_schema_from_pds( - &pds_server.uri(), + &pds_server.uri().parse().unwrap(), &did.parse().unwrap(), &nsid.parse().unwrap(), ) @@ -222,7 +225,7 @@ async fn test_fetch_schema_invalid_lexicon_json() { .await; let result = fetch_schema_from_pds( - &pds_server.uri(), + &pds_server.uri().parse().unwrap(), &did.parse().unwrap(), &nsid.parse().unwrap(), ) @@ -352,7 +355,7 @@ async fn test_pds_trailing_slash_handled() { let pds_url_with_slash = format!("{}/", pds_server.uri()); let doc = fetch_schema_from_pds( - &pds_url_with_slash, + &pds_url_with_slash.parse().unwrap(), &did.parse().unwrap(), &nsid.parse().unwrap(), ) @@ -377,7 +380,7 @@ async fn test_fetch_schema_error_status_gives_meaningful_error() { .await; let result = fetch_schema_from_pds( - &pds_server.uri(), + &pds_server.uri().parse().unwrap(), &did.parse().unwrap(), &nsid.parse().unwrap(), )