diff --git a/parakeet/src/cache.rs b/parakeet/src/cache.rs deleted file mode 100644 index 2f70fff9..00000000 --- a/parakeet/src/cache.rs +++ /dev/null @@ -1,121 +0,0 @@ -use dataloader::async_cached::AsyncCache; -use moka::future::Cache; -use std::time::Duration; - -/// In-memory Loader Cache using moka -/// -/// Provides TinyLFU eviction with optional TTL and size limits. -/// No serialization needed for in-memory caching. -pub struct LoaderCache -where - V: Clone + Send + Sync + 'static, -{ - cache: Cache, -} - -impl LoaderCache -where - V: Clone + Send + Sync + 'static, -{ - pub fn new(ttl_seconds: Option, max_capacity: u64) -> Self { - let mut builder = Cache::builder().max_capacity(max_capacity); - - if let Some(ttl) = ttl_seconds { - builder = builder.time_to_live(Duration::from_secs(ttl)); - } - - Self { - cache: builder.build(), - } - } -} - -impl AsyncCache for LoaderCache -where - V: Clone + Send + Sync + 'static, -{ - type Key = String; - type Val = V; - - async fn get(&mut self, key: &Self::Key) -> Option { - self.cache.get(key).await - } - - async fn insert(&mut self, key: Self::Key, val: Self::Val) { - self.cache.insert(key, val).await; - } - - async fn remove(&mut self, key: &Self::Key) -> Option { - self.cache.remove(key).await - } - - async fn clear(&mut self) { - self.cache.invalidate_all(); - // Run pending tasks to complete the invalidation - self.cache.run_pending_tasks().await; - } -} - -/// A Loader Cache with a key prefix -/// -/// Uses moka for in-memory caching with automatic prefix handling. -/// Supports both String and i32 keys (anything that implements Display). -pub struct PrefixedLoaderCache -where - K: std::fmt::Display + std::hash::Hash + Eq + Send + Sync + 'static, - V: Clone + Send + Sync + 'static, -{ - cache: Cache, - prefix: String, - _phantom: std::marker::PhantomData, -} - -impl PrefixedLoaderCache -where - K: std::fmt::Display + std::hash::Hash + Eq + Send + Sync + 'static, - V: Clone + Send + Sync + 'static, -{ - pub fn new(prefix: String, ttl_seconds: Option, max_capacity: u64) -> Self { - let mut builder = Cache::builder().max_capacity(max_capacity); - - if let Some(ttl) = ttl_seconds { - builder = builder.time_to_live(Duration::from_secs(ttl)); - } - - Self { - cache: builder.build(), - prefix, - _phantom: std::marker::PhantomData, - } - } -} - -impl AsyncCache for PrefixedLoaderCache -where - K: std::fmt::Display + std::hash::Hash + Eq + Send + Sync + 'static, - V: Clone + Send + Sync + 'static, -{ - type Key = K; - type Val = V; - - async fn get(&mut self, key: &Self::Key) -> Option { - let cache_key = format!("{}{}", self.prefix, key); - self.cache.get(&cache_key).await - } - - async fn insert(&mut self, key: Self::Key, val: Self::Val) { - let cache_key = format!("{}{}", self.prefix, key); - self.cache.insert(cache_key, val).await; - } - - async fn remove(&mut self, key: &Self::Key) -> Option { - let cache_key = format!("{}{}", self.prefix, key); - self.cache.remove(&cache_key).await - } - - async fn clear(&mut self) { - self.cache.invalidate_all(); - // Run pending tasks to complete the invalidation - self.cache.run_pending_tasks().await; - } -} diff --git a/parakeet/src/common/auth.rs b/parakeet/src/common/auth.rs new file mode 100644 index 00000000..3d89ea6a --- /dev/null +++ b/parakeet/src/common/auth.rs @@ -0,0 +1,214 @@ +use did_resolver::Resolver; +use jsonwebtoken::{Algorithm, DecodingKey, Validation}; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::sync::{Arc, LazyLock}; +use tokio::sync::RwLock; + +static DUMMY_KEY: LazyLock = LazyLock::new(|| DecodingKey::from_secret(&[])); +static NO_VERIFY: LazyLock = LazyLock::new(|| { + let mut val_no_verify = Validation::default(); + val_no_verify.insecure_disable_signature_validation(); + val_no_verify.validate_aud = false; + val_no_verify +}); + +#[derive(Debug, Deserialize, Serialize)] +pub struct Claims { + pub aud: String, + pub exp: usize, + pub iat: usize, + pub iss: String, + pub lxm: Option, + pub jti: String, +} + +pub struct JwtVerifier { + aud: String, + resolver: Arc, + key_cache: RwLock>, +} + +impl JwtVerifier { + pub fn new(aud: String, resolver: Arc) -> Self { + Self { + aud, + resolver, + key_cache: RwLock::new(HashMap::new()), + } + } + + pub async fn resolve_and_verify_jwt(&self, token: &str, aud: Option<&str>) -> Option { + // first we need to decode without verifying, to get iss. + let unsafe_data = jsonwebtoken::decode::(token, &DUMMY_KEY, &NO_VERIFY).ok()?; + let unsafe_iss = unsafe_data.claims.iss; + + let maybe_cached_key = { + let l = self.key_cache.read().await; + l.get(&unsafe_iss).cloned() + }; + let multibase_key = match maybe_cached_key { + Some(key) => key, + None => self.resolve_key(&unsafe_iss).await?, + }; + + let aud = aud.unwrap_or(&self.aud); + self.verify_jwt_multibase_with_alg(token, &multibase_key, unsafe_data.header.alg, aud) + } + + async fn resolve_key(&self, did: &str) -> Option { + tracing::trace!("resolving multikey for {did}"); + let did_doc = self.resolver.resolve_did(did).await.ok()??; + + // try find the multibase key + let multikey = did_doc.find_verif_method_by_type("Multikey")?; + + { + let mut l = self.key_cache.write().await; + drop(l.insert(did.to_owned(), multikey.public_key_multibase.clone())); + } + + Some(multikey.public_key_multibase.clone()) + } + + pub fn verify_jwt_multibase(&self, token: &str, multibase_key: &str) -> Option { + let alg = jsonwebtoken::decode_header(token).ok()?.alg; + + self.verify_jwt_multibase_with_alg(token, multibase_key, alg, &self.aud) + } + + pub fn verify_jwt_multibase_with_alg( + &self, + token: &str, + multibase_key: &str, + alg: Algorithm, + aud: &str, + ) -> Option { + // decode the multibase key + let (_, key) = multibase::decode(multibase_key).ok()?; + + let key = DecodingKey::from_ec_der(&key[2..]); + + let mut validation = Validation::new(alg); + validation.validate_aud = false; + validation.set_audience(&[&aud]); + + let decoded = jsonwebtoken::decode::(token, &key, &validation).ok()?; + + Some(decoded.claims) + } +} + +// ============================================================================ +// Axum Extractors (from extract.rs) +// ============================================================================ + +#[derive(Debug)] +pub struct LabelConfigItem { + pub labeler: String, + pub redact: bool, +} + +impl std::str::FromStr for LabelConfigItem { + type Err = std::convert::Infallible; + + fn from_str(val: &str) -> Result { + let v = val.trim(); + + let Some((did, rem)) = v.split_once(';') else { + return Ok(Self { + labeler: v.to_owned(), + redact: false, + }); + }; + + Ok(Self { + labeler: did.to_owned(), + redact: rem == "redact", + }) + } +} + +#[derive(Debug)] +pub struct AtpAcceptLabelers(pub Vec); + +impl FromRequestParts for AtpAcceptLabelers +where + S: Send + Sync, +{ + type Rejection = (StatusCode, &'static str); + + async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result { + let labelers = match parts.headers.get("Atproto-Accept-Labelers") { + Some(labelers) => { + let labelers = labelers.to_str().map_err(|_err| { + ( + StatusCode::BAD_REQUEST, + "Atproto-Accept-Labelers was invalid", + ) + })?; + + labelers + .trim() + .split(",") + .map(|s| s.parse().unwrap()) + .collect() + } + None => vec![], + }; + + Ok(Self(labelers)) + } +} + +#[derive(Clone, Debug)] +pub struct AtpAuth(pub String); + +type BearerHeader = TypedHeader>; + +impl FromRequestParts for AtpAuth { + type Rejection = (StatusCode, String); + + async fn from_request_parts( + parts: &mut Parts, + state: &GlobalState, + ) -> Result { + let hdr = >::from_request_parts( + parts, state, + ) + .await + .map_err(|err| (StatusCode::BAD_REQUEST, err.to_string()))? + .ok_or_else(|| (StatusCode::UNAUTHORIZED, "missing JWT".to_owned()))?; + + let jwt_result = state.jwt.resolve_and_verify_jwt(hdr.token(), None).await; + match jwt_result { + Some(claims) => Ok(Self(claims.iss)), + None => Err((StatusCode::INTERNAL_SERVER_ERROR, "JWT error".to_owned())), + } + } +} + +impl OptionalFromRequestParts for AtpAuth { + type Rejection = (StatusCode, String); + + async fn from_request_parts( + parts: &mut Parts, + state: &GlobalState, + ) -> Result, Self::Rejection> { + let Some(hdr) = + >::from_request_parts( + parts, state, + ) + .await + .map_err(|err| (StatusCode::BAD_REQUEST, err.to_string()))? + else { + return Ok(None); + }; + + let jwt_result = state.jwt.resolve_and_verify_jwt(hdr.token(), None).await; + match jwt_result { + Some(claims) => Ok(Some(Self(claims.iss))), + None => Err((StatusCode::INTERNAL_SERVER_ERROR, "JWT error".to_owned())), + } + } +} diff --git a/parakeet/src/id_cache_helpers.rs b/parakeet/src/common/cache/id_helpers.rs similarity index 100% rename from parakeet/src/id_cache_helpers.rs rename to parakeet/src/common/cache/id_helpers.rs diff --git a/parakeet/src/cache_listener.rs b/parakeet/src/common/cache/listener.rs similarity index 100% rename from parakeet/src/cache_listener.rs rename to parakeet/src/common/cache/listener.rs diff --git a/parakeet/src/common/cache/mod.rs b/parakeet/src/common/cache/mod.rs new file mode 100644 index 00000000..aca8306c --- /dev/null +++ b/parakeet/src/common/cache/mod.rs @@ -0,0 +1,10 @@ +//! Cache infrastructure modules + +pub mod id_helpers; +pub mod listener; +pub mod timeline; + +// Re-export commonly used items +pub use id_helpers::{get_actor_id_or_fetch, get_actor_dids_or_fetch, get_actor_ids_or_fetch}; +pub use listener::spawn_cache_listener; +pub use timeline::{AuthorFeedCache, TimelineCache}; \ No newline at end of file diff --git a/parakeet/src/timeline_cache.rs b/parakeet/src/common/cache/timeline.rs similarity index 100% rename from parakeet/src/timeline_cache.rs rename to parakeet/src/common/cache/timeline.rs diff --git a/parakeet/src/xrpc/error.rs b/parakeet/src/common/errors.rs similarity index 100% rename from parakeet/src/xrpc/error.rs rename to parakeet/src/common/errors.rs diff --git a/parakeet/src/xrpc/helpers_entity.rs b/parakeet/src/common/helpers.rs similarity index 100% rename from parakeet/src/xrpc/helpers_entity.rs rename to parakeet/src/common/helpers.rs diff --git a/parakeet/src/common/mod.rs b/parakeet/src/common/mod.rs new file mode 100644 index 00000000..15e13efc --- /dev/null +++ b/parakeet/src/common/mod.rs @@ -0,0 +1,12 @@ +//! Common infrastructure and utilities shared across the application + +pub mod auth; +pub mod cache; +pub mod errors; +pub mod helpers; +pub mod rate_limiting; + +// Re-export commonly used items +pub use auth::{AtpAcceptLabelers, AtpAuth, JwtVerifier}; +pub use errors::{Error, XrpcResult}; +pub use rate_limiting::{rate_limit_middleware, RateLimiter}; \ No newline at end of file diff --git a/parakeet/src/rate_limit.rs b/parakeet/src/common/rate_limiting.rs similarity index 100% rename from parakeet/src/rate_limit.rs rename to parakeet/src/common/rate_limiting.rs diff --git a/parakeet/src/lib.rs b/parakeet/src/lib.rs index d25bb3d9..e64bd68b 100644 --- a/parakeet/src/lib.rs +++ b/parakeet/src/lib.rs @@ -5,14 +5,9 @@ use diesel_async::pooled_connection::deadpool::Pool; use diesel_async::AsyncPgConnection; use std::sync::Arc; -pub mod cache_listener; +pub mod common; pub mod config; pub mod entities; -pub mod id_cache_helpers; -pub mod middleware; -pub mod rate_limit; -pub mod search; -pub mod timeline_cache; pub mod xrpc; // Re-export entities for use in main @@ -22,13 +17,13 @@ pub use entities::{ProfileEntity, PostEntity, FeedGeneratorEntity, ListEntity, S pub struct GlobalState { pub pool: Pool, pub resolver: Arc, - pub jwt: Arc, - pub cdn: Arc, + pub jwt: Arc, + pub cdn: Arc, pub id_cache: Arc, - pub rate_limiter: Arc, + pub rate_limiter: Arc, pub rate_limit_config: config::ConfigRateLimit, - pub timeline_cache: Arc, - pub author_feed_cache: Arc, + pub timeline_cache: Arc, + pub author_feed_cache: Arc, pub http_client: reqwest::Client, // Entity-based system (replaces old hydration/loaders/caches) pub profile_entity: Arc, diff --git a/parakeet/src/main.rs b/parakeet/src/main.rs index f313bc6f..1cd1d326 100644 --- a/parakeet/src/main.rs +++ b/parakeet/src/main.rs @@ -63,12 +63,12 @@ async fn main() -> eyre::Result<()> { plc_directory: conf.plc_directory, ..Default::default() })?); - let jwt = Arc::new(xrpc::jwt::JwtVerifier::new( + let jwt = Arc::new(common::auth::JwtVerifier::new( conf.service.did.clone(), resolver.clone(), )); - let cdn = Arc::new(xrpc::cdn::BskyCdn::new(conf.cdn.base.clone(), conf.cdn.video_base)); + let cdn = Arc::new(common::helpers::BskyCdn::new(conf.cdn.base.clone(), conf.cdn.video_base)); // Initialize shared HTTP client with connection pooling @@ -85,13 +85,13 @@ async fn main() -> eyre::Result<()> { let state = { // Initialize rate limiter (in-memory with DashMap) - let rate_limiter = Arc::new(rate_limit::RateLimiter::new()); + let rate_limiter = Arc::new(common::rate_limiting::RateLimiter::new()); // Initialize timeline cache (60 second TTL, 10k max items) - let timeline_cache = Arc::new(timeline_cache::TimelineCache::new(60, 10_000)); + let timeline_cache = Arc::new(common::cache::TimelineCache::new(60, 10_000)); // Initialize author feed cache (60 second TTL, 10k max items) - let author_feed_cache = Arc::new(timeline_cache::AuthorFeedCache::new(60, 10_000)); + let author_feed_cache = Arc::new(common::cache::AuthorFeedCache::new(60, 10_000)); // Initialize new entity-centric implementations (replacing old caches) @@ -151,7 +151,7 @@ async fn main() -> eyre::Result<()> { // Spawn cache invalidation listener (PostgreSQL LISTEN/NOTIFY) // This listens for cache_invalidate notifications from the consumer - if let Err(e) = cache_listener::spawn_cache_listener(Arc::new(state.clone())).await { + if let Err(e) = common::cache::spawn_cache_listener(Arc::new(state.clone())).await { tracing::warn!("Failed to spawn cache listener: {}", e); // Continue anyway - cache will rely on TTL for freshness } @@ -164,7 +164,7 @@ async fn main() -> eyre::Result<()> { ) .layer(axum::middleware::from_fn_with_state( state.clone(), - middleware::rate_limit::rate_limit_middleware, + common::rate_limiting::rate_limit_middleware, )) .layer(TraceLayer::new_for_http()) .layer(cors) diff --git a/parakeet/src/middleware/mod.rs b/parakeet/src/middleware/mod.rs deleted file mode 100644 index 382585de..00000000 --- a/parakeet/src/middleware/mod.rs +++ /dev/null @@ -1 +0,0 @@ -pub mod rate_limit; diff --git a/parakeet/src/middleware/rate_limit.rs b/parakeet/src/middleware/rate_limit.rs deleted file mode 100644 index bd74f3d9..00000000 --- a/parakeet/src/middleware/rate_limit.rs +++ /dev/null @@ -1,186 +0,0 @@ -//! Rate limiting middleware for Axum -//! -//! Implements request-level rate limiting with DashMap (in-memory). -//! Adds rate limit headers to responses and returns 429 when limits exceeded. - -use axum::{ - body::Body, - extract::State, - http::{Request, StatusCode}, - middleware::Next, - response::{IntoResponse, Response}, -}; - -use crate::xrpc::extract::AtpAuth; -use crate::GlobalState; - -/// Rate limit middleware that checks quotas before allowing requests -/// -/// Rate limits are applied per (endpoint, identifier) where identifier is: -/// - Authenticated: user DID -/// - Unauthenticated: IP address from X-Forwarded-For or connection -/// -/// Adds rate limit headers to all responses: -/// - X-RateLimit-Limit: Maximum requests allowed -/// - X-RateLimit-Remaining: Requests remaining in current window -/// - X-RateLimit-Reset: Unix timestamp when limit resets -pub async fn rate_limit_middleware( - State(state): State, - req: Request, - next: Next, -) -> Response { - // Extract identifier (authenticated DID or IP address) - let identifier = extract_identifier(&req); - - // Get endpoint path for rate limit key - let endpoint = req.uri().path(); - - // Determine rate limit based on endpoint and config - let (max_requests, window_secs) = - get_rate_limit_for_endpoint(endpoint, &state.rate_limit_config); - - // Build rate limit key - let key = format!("{}:{}", endpoint, identifier); - - // Check rate limit (no locking needed - DashMap is already thread-safe) - let result = state.rate_limiter.check_rate_limit(&key, max_requests, window_secs); - - if !result.allowed { - // Rate limit exceeded - tracing::warn!( - endpoint = endpoint, - identifier = identifier, - "Rate limit exceeded" - ); - - let mut response = ( - StatusCode::TOO_MANY_REQUESTS, - "Rate limit exceeded. Please try again later.", - ) - .into_response(); - - // Add rate limit headers - response.headers_mut().insert( - "X-RateLimit-Limit", - max_requests.to_string().parse().unwrap(), - ); - response.headers_mut().insert( - "X-RateLimit-Remaining", - "0".parse().unwrap(), - ); - response.headers_mut().insert( - "X-RateLimit-Reset", - result.reset_at.to_string().parse().unwrap(), - ); - - return response; - } - - // Process the request - let mut response = next.run(req).await; - - // Add rate limit headers to successful response - response.headers_mut().insert( - "X-RateLimit-Limit", - max_requests.to_string().parse().unwrap(), - ); - response.headers_mut().insert( - "X-RateLimit-Remaining", - result.remaining.to_string().parse().unwrap(), - ); - response.headers_mut().insert( - "X-RateLimit-Reset", - result.reset_at.to_string().parse().unwrap(), - ); - - response -} - -/// Extract identifier for rate limiting from request -fn extract_identifier(req: &Request) -> String { - // Try to get authenticated DID first - if let Some(auth) = req.extensions().get::() { - return format!("user:{}", auth.0); - } - - // Fall back to IP address - req.headers() - .get("x-forwarded-for") - .and_then(|h| h.to_str().ok()) - .and_then(|s| s.split(',').next()) - .map(|s| s.trim().to_string()) - .unwrap_or_else(|| "unknown".to_string()) -} - -/// Get rate limit (max_requests, window_secs) for an endpoint -/// -/// Different endpoints have different limits based on their expense. -/// Expensive queries (timeline, threads) have lower limits. -/// Uses configuration values with sensible defaults. -fn get_rate_limit_for_endpoint( - endpoint: &str, - config: &crate::config::ConfigRateLimit, -) -> (u32, u64) { - let limit = match endpoint { - // Timeline queries are expensive - lower limit - _ if endpoint.contains("getTimeline") => config.timeline, - - // Thread queries can be expensive - moderate limit - _ if endpoint.contains("getPostThread") => config.thread, - - // Feed queries - moderate limit - _ if endpoint.contains("getAuthorFeed") - || endpoint.contains("getFeed") - || endpoint.contains("getListFeed") => - { - config.feed - } - - // Default for all other endpoints - _ => config.default, - }; - - (limit, config.window_secs) -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::config::ConfigRateLimit; - - fn test_config() -> ConfigRateLimit { - ConfigRateLimit::default() - } - - #[test] - fn test_get_rate_limit_for_timeline() { - let config = test_config(); - let (limit, window) = get_rate_limit_for_endpoint("/xrpc/app.bsky.feed.getTimeline", &config); - assert_eq!(limit, 900); // 15 req/sec - assert_eq!(window, 60); - } - - #[test] - fn test_get_rate_limit_for_thread() { - let config = test_config(); - let (limit, window) = get_rate_limit_for_endpoint("/xrpc/app.bsky.feed.getPostThread", &config); - assert_eq!(limit, 900); // 15 req/sec - assert_eq!(window, 60); - } - - #[test] - fn test_get_rate_limit_for_feed() { - let config = test_config(); - let (limit, window) = get_rate_limit_for_endpoint("/xrpc/app.bsky.feed.getAuthorFeed", &config); - assert_eq!(limit, 900); // 15 req/sec - assert_eq!(window, 60); - } - - #[test] - fn test_get_rate_limit_default() { - let config = test_config(); - let (limit, window) = get_rate_limit_for_endpoint("/xrpc/app.bsky.actor.getProfile", &config); - assert_eq!(limit, 6000); // 100 req/sec - assert_eq!(window, 60); - } -} diff --git a/parakeet/src/search/mod.rs b/parakeet/src/search/mod.rs deleted file mode 100644 index 6eb25129..00000000 --- a/parakeet/src/search/mod.rs +++ /dev/null @@ -1,38 +0,0 @@ -pub mod query_parser; -pub mod search_service; - -pub use query_parser::QueryParser; -pub use search_service::SearchService; - -/// Legacy search query result (stub for compatibility with existing search endpoint) -/// -/// The existing search.rs expects this structure. This is a stub that returns -/// the query text as-is without advanced operator parsing. Once we fully migrate -/// to token-based search, this can be removed or enhanced. -pub struct LegacySearchQuery { - pub text: String, - pub from: Option, - pub mentions: Option, - pub lang: Option, - pub domain: Option, - pub tags: Vec, - pub since: Option, - pub until: Option, -} - -/// Parse search query (stub for compatibility) -/// -/// Currently returns the query text as-is without operator parsing. -/// TODO: Implement full operator parsing (from:, mentions:, lang:, domain:, etc.) -pub fn parse_search_query(query: &str, _viewer_did: Option<&str>) -> LegacySearchQuery { - LegacySearchQuery { - text: query.to_string(), - from: None, - mentions: None, - lang: None, - domain: None, - tags: Vec::new(), - since: None, - until: None, - } -} diff --git a/parakeet/src/search/query_parser.rs b/parakeet/src/search/query_parser.rs deleted file mode 100644 index 26a2697c..00000000 --- a/parakeet/src/search/query_parser.rs +++ /dev/null @@ -1,169 +0,0 @@ -use regex::Regex; -use std::collections::HashSet; -use std::sync::LazyLock; -use unicode_segmentation::UnicodeSegmentation; - -static HASHTAG_RE: LazyLock = LazyLock::new(|| Regex::new(r"#([\p{L}\p{N}_]+)").unwrap()); -static MENTION_RE: LazyLock = LazyLock::new(|| Regex::new(r"@([\w-]+)\.[\w.-]+").unwrap()); - -/// Query parser for token-based search -/// -/// Tokenizes search query text using the same logic as post indexing. -/// Uses same rules: hashtags, mentions, words (with stopword filtering). -/// Does NOT parse operators - those come from explicit API parameters. -pub struct QueryParser { - stopwords: HashSet, -} - -impl QueryParser { - pub fn new() -> Self { - Self { - stopwords: Self::load_stopwords(), - } - } - - /// Tokenize query text into search tokens - /// - /// Returns tokens that will be matched against post_search_tokens.tokens - /// using PostgreSQL array containment (@> operator for AND semantics). - pub fn tokenize(&self, query: &str) -> Vec { - let mut tokens = Vec::new(); - let mut seen = HashSet::new(); - - // Extract hashtags (preserve as single tokens) - for cap in HASHTAG_RE.captures_iter(query) { - let tag = cap[1].to_lowercase(); - if tag.len() >= 2 && seen.insert(tag.clone()) { - tokens.push(tag); - } - } - - // Extract mentions (username only) - for cap in MENTION_RE.captures_iter(query) { - let username = cap[1].to_lowercase(); - if username.len() >= 2 && seen.insert(username.clone()) { - tokens.push(username); - } - } - - // Extract regular words - for word in UnicodeSegmentation::unicode_words(query) { - // Skip URLs - if word.starts_with("http") || word.contains(".com") || word.contains(".org") { - continue; - } - - let normalized = word.to_lowercase(); - - // Filter: too short, stopwords, already seen - if normalized.len() < 2 { - continue; - } - if self.stopwords.contains(&normalized) { - continue; - } - if !seen.insert(normalized.clone()) { - continue; - } - - tokens.push(normalized); - } - - tokens - } - - fn load_stopwords() -> HashSet { - // Same stopword list as tokenizer - let words = vec![ - "a", "an", "and", "are", "as", "at", "be", "by", "for", - "from", "has", "he", "in", "is", "it", "its", "of", "on", - "that", "the", "to", "was", "will", "with", "this", "but", - "they", "have", "had", "what", "when", "where", "who", "which", - "not", "or", "if", "so", "can", "been", "would", "could", - "should", "may", "do", "does", "did", "me", "we", "us", - "our", "your", "their", "his", "her", "she", "him", "them", - "my", "you", "i", "am", "up", "out", "about", "into", - "than", "then", "now", "all", "some", "any", "no", "more", - "most", "just", "only", "very", "too", "also", "such", - "much", "many", "even", "still", "how", "why", "here", - "there", "said", "each", "which", "these", "those", "other", - "another", "both", "few", "own", "same", "get", "got", - ]; - words.into_iter().map(String::from).collect() - } -} - -impl Default for QueryParser { - fn default() -> Self { - Self::new() - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_simple_query() { - let parser = QueryParser::new(); - let tokens = parser.tokenize("rust programming"); - - assert_eq!(tokens.len(), 2); - assert!(tokens.contains(&"rust".to_string())); - assert!(tokens.contains(&"programming".to_string())); - } - - #[test] - fn test_stopword_filtering() { - let parser = QueryParser::new(); - let tokens = parser.tokenize("the quick brown fox"); - - // "the" is a stopword, should be filtered out - assert!(!tokens.contains(&"the".to_string())); - assert!(tokens.contains(&"quick".to_string())); - assert!(tokens.contains(&"brown".to_string())); - assert!(tokens.contains(&"fox".to_string())); - } - - #[test] - fn test_hashtag_extraction() { - let parser = QueryParser::new(); - let tokens = parser.tokenize("#Rust2024 #Bluesky programming"); - - // Hashtags extracted as lowercase without # - assert!(tokens.contains(&"rust2024".to_string())); - assert!(tokens.contains(&"bluesky".to_string())); - assert!(tokens.contains(&"programming".to_string())); - } - - #[test] - fn test_mention_extraction() { - let parser = QueryParser::new(); - let tokens = parser.tokenize("@alice.bsky.social check this out"); - - // Mentions extracted as username only - assert!(tokens.contains(&"alice".to_string())); - assert!(tokens.contains(&"check".to_string())); - } - - #[test] - fn test_url_skipping() { - let parser = QueryParser::new(); - let tokens = parser.tokenize("Check https://example.com amazing"); - - // URLs should be skipped - assert!(!tokens.iter().any(|t| t.contains("http") || t.contains("example"))); - assert!(tokens.contains(&"check".to_string())); - assert!(tokens.contains(&"amazing".to_string())); - } - - #[test] - fn test_deduplication() { - let parser = QueryParser::new(); - let tokens = parser.tokenize("rust rust rust programming"); - - // "rust" appears 3 times but should only be one token - assert_eq!(tokens.iter().filter(|t| *t == "rust").count(), 1); - assert!(tokens.contains(&"programming".to_string())); - } -} diff --git a/parakeet/src/search/search_service.rs b/parakeet/src/search/search_service.rs deleted file mode 100644 index b6831906..00000000 --- a/parakeet/src/search/search_service.rs +++ /dev/null @@ -1,182 +0,0 @@ -use crate::search::query_parser::QueryParser; -use chrono::NaiveDateTime; -use tokio_postgres::GenericClient; - -pub struct SearchService { - parser: QueryParser, -} - -#[derive(Debug, Clone)] -pub struct SearchResult { - pub uri: String, - pub rank: f64, -} - -impl SearchService { - pub fn new() -> Self { - Self { - parser: QueryParser::new(), - } - } - - /// Search posts using token-based search - /// - /// Returns post URIs matching the search criteria - /// Aligned with AT Protocol app.bsky.feed.searchPosts parameters - /// All filtering done via JOINs (no denormalized data in search table) - #[allow(clippy::too_many_arguments)] - pub async fn search( - &self, - conn: &C, - query_str: &str, - limit: i64, - author_did: Option<&str>, - mentions_did: Option<&str>, - lang: Option<&str>, - url: Option<&str>, - tags: &[String], - has_media: bool, - has_links: bool, - since: Option, - until: Option, - ) -> Result, tokio_postgres::Error> { - // Tokenize query text using same logic as indexing - let tokens = self.parser.tokenize(query_str); - - // Build query dynamically based on parameters - // Base query from posts table directly - let mut query = String::from( - "SELECT DISTINCT p.rkey, a.did - FROM posts p - JOIN actors a ON a.id = p.actor_id", - ); - - query.push_str(" WHERE 1=1"); - - // Add WHERE clauses - let mut param_index = 1; - let mut params: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> = Vec::new(); - - // Text search tokens - if !tokens.is_empty() { - query.push_str(&format!(" AND p.tokens && ${}", param_index)); - param_index += 1; - } - - // Author filter (filter by post's actor_id via JOIN, no denormalization) - if author_did.is_some() { - query.push_str(&format!(" AND p.actor_id = (SELECT id FROM actors WHERE did = ${})", param_index)); - param_index += 1; - } - - // Mentions filter (use mentions array column) - if mentions_did.is_some() { - query.push_str(&format!(" AND EXISTS (SELECT 1 FROM actors ma WHERE ma.id = ANY(p.mentions) AND ma.did = ${})", param_index)); - param_index += 1; - } - - // Language filter (check posts.langs array) - if lang.is_some() { - query.push_str(&format!(" AND ${} = ANY(p.langs)", param_index)); - param_index += 1; - } - - // URL filter (use ext_embed composite column) - if url.is_some() { - query.push_str(&format!(" AND (p.ext_embed).uri = ${}", param_index)); - param_index += 1; - } - - // Tags filter (lowercase matching using existing tokens) - if !tags.is_empty() { - query.push_str(&format!(" AND p.tokens && ${}", param_index)); - param_index += 1; - } - - // Media filter (check image_1 or video_embed columns) - if has_media { - query.push_str(" AND (p.image_1 IS NOT NULL OR p.video_embed IS NOT NULL)"); - } - - // Links filter (check ext_embed column) - if has_links { - query.push_str(" AND p.ext_embed IS NOT NULL"); - } - - // Time range filters (use TID timestamp) - if since.is_some() { - query.push_str(&format!(" AND tid_timestamp(p.rkey) >= to_timestamp(${})", param_index)); - param_index += 1; - } - if until.is_some() { - query.push_str(&format!(" AND tid_timestamp(p.rkey) <= to_timestamp(${})", param_index)); - param_index += 1; - } - - // Order by post creation time (descending) and limit - query.push_str(&format!(" ORDER BY p.rkey DESC LIMIT ${}", param_index)); - - // Build parameter vector - let tokens_vec: Vec; - let tags_vec: Vec; - - if !tokens.is_empty() { - tokens_vec = tokens; - params.push(&tokens_vec); - } - if author_did.is_some() { - params.push(&author_did); - } - if mentions_did.is_some() { - params.push(&mentions_did); - } - if lang.is_some() { - params.push(&lang); - } - if url.is_some() { - params.push(&url); - } - if !tags.is_empty() { - tags_vec = tags.iter().map(|t| t.to_lowercase()).collect(); - params.push(&tags_vec); - } - // Convert NaiveDateTime to timestamp for postgres compatibility - let since_ts = since.map(|dt| dt.and_utc().timestamp()); - let until_ts = until.map(|dt| dt.and_utc().timestamp()); - - if since_ts.is_some() { - params.push(&since_ts); - } - if until_ts.is_some() { - params.push(&until_ts); - } - params.push(&limit); - - let rows = conn.query(&query, ¶ms).await?; - - let results: Vec = rows - .iter() - .map(|row| { - let rkey: i64 = row.get(0); - let author_did: String = row.get(1); - - // Convert TID rkey to timestamp for ranking - let created_at = parakeet_db::tid_util::tid_to_datetime(rkey); - let uri = format!("at://{}/app.bsky.feed.post/{}", author_did, parakeet_db::tid_util::encode_tid(rkey)); - - SearchResult { - uri, - rank: created_at.timestamp() as f64, - } - }) - .collect(); - - Ok(results) - } -} - -impl Default for SearchService { - fn default() -> Self { - Self::new() - } -} diff --git a/parakeet/src/xrpc/app_bsky/actor.rs b/parakeet/src/xrpc/app_bsky/actor.rs index d995cce4..2d4a1f0d 100644 --- a/parakeet/src/xrpc/app_bsky/actor.rs +++ b/parakeet/src/xrpc/app_bsky/actor.rs @@ -1,5 +1,5 @@ -use crate::xrpc::error::{Error, XrpcResult}; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::{Error, XrpcResult}; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::GlobalState; use axum::extract::{Query, State}; use axum::response::{IntoResponse as _, Response}; @@ -259,7 +259,7 @@ pub async fn get_suggestions( } // Convert DIDs to actor_ids for efficient profile loading - let did_to_actor_id = crate::id_cache_helpers::get_actor_ids_or_fetch( + let did_to_actor_id = crate::common::cache::get_actor_ids_or_fetch( &state.pool, &state.id_cache, &all_dids, diff --git a/parakeet/src/xrpc/app_bsky/bookmark.rs b/parakeet/src/xrpc/app_bsky/bookmark.rs index 742c4cf4..5716c81d 100644 --- a/parakeet/src/xrpc/app_bsky/bookmark.rs +++ b/parakeet/src/xrpc/app_bsky/bookmark.rs @@ -1,5 +1,5 @@ -use crate::xrpc::error::XrpcResult; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::XrpcResult; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::xrpc::{datetime_cursor, CursorQuery}; use crate::GlobalState; use axum::extract::{Query, State}; @@ -21,7 +21,7 @@ pub async fn create_bookmark( auth: AtpAuth, Json(form): Json, ) -> XrpcResult<()> { - use crate::xrpc::error::Error; + use crate::common::errors::Error; let mut conn = state.pool.get().await?; // Resolve auth DID to actor_id using ProfileEntity @@ -99,7 +99,7 @@ pub async fn delete_bookmark( auth: AtpAuth, Json(form): Json, ) -> XrpcResult<()> { - use crate::xrpc::error::Error; + use crate::common::errors::Error; let mut conn = state.pool.get().await?; // Resolve auth DID to actor_id using ProfileEntity @@ -166,7 +166,7 @@ pub async fn get_bookmarks( auth: AtpAuth, Query(query): Query, ) -> XrpcResult> { - use crate::xrpc::error::Error; + use crate::common::errors::Error; let mut conn = state.pool.get().await?; // Resolve auth DID to actor_id using ProfileEntity @@ -261,7 +261,7 @@ pub async fn get_bookmarks_count( State(state): State, auth: AtpAuth, ) -> XrpcResult> { - use crate::xrpc::error::Error; + use crate::common::errors::Error; let mut conn = state.pool.get().await?; // Resolve auth DID to actor_id using ProfileEntity diff --git a/parakeet/src/xrpc/app_bsky/feed/feedgen.rs b/parakeet/src/xrpc/app_bsky/feed/feedgen.rs index b62f59ea..3808d88c 100644 --- a/parakeet/src/xrpc/app_bsky/feed/feedgen.rs +++ b/parakeet/src/xrpc/app_bsky/feed/feedgen.rs @@ -1,5 +1,5 @@ -use crate::xrpc::error::{Error, XrpcResult}; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::{Error, XrpcResult}; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::xrpc::{datetime_cursor, ActorWithCursorQuery}; use crate::GlobalState; use axum::extract::{Query, State}; diff --git a/parakeet/src/xrpc/app_bsky/feed/get_timeline.rs b/parakeet/src/xrpc/app_bsky/feed/get_timeline.rs index 6089ffbe..e25cb697 100644 --- a/parakeet/src/xrpc/app_bsky/feed/get_timeline.rs +++ b/parakeet/src/xrpc/app_bsky/feed/get_timeline.rs @@ -1,6 +1,6 @@ use crate::xrpc::datetime_cursor; -use crate::xrpc::error::{Error, XrpcResult}; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::{Error, XrpcResult}; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::GlobalState; use axum::extract::{Query, State}; use axum::Json; @@ -41,7 +41,7 @@ pub async fn get_timeline( // Resolve actor_id using ProfileEntity let user_actor_id = state.profile_entity.resolve_identifier(&user_did).await - .map_err(|_| crate::xrpc::error::Error::actor_not_found(&user_did))?; + .map_err(|_| crate::common::errors::Error::actor_not_found(&user_did))?; // Try cache first let mut step_timer = std::time::Instant::now(); @@ -100,7 +100,7 @@ pub async fn get_timeline( // Get timeline posts using PostEntity let posts_result = state.post_entity.get_timeline_posts(&followed_ids, cursor_value.as_ref(), limit + 1).await - .map_err(|e| crate::xrpc::error::Error::server_error(Some(&e.to_string())))?; + .map_err(|e| crate::common::errors::Error::server_error(Some(&e.to_string())))?; // Check for next page let has_next = posts_result.len() > limit as usize; @@ -221,7 +221,7 @@ pub async fn get_author_feed( // Resolve actor to actor_id using ProfileEntity let actor_id = state.profile_entity.resolve_identifier(&query.actor).await - .map_err(|_| crate::xrpc::error::Error::actor_not_found(&query.actor))?; + .map_err(|_| crate::common::errors::Error::actor_not_found(&query.actor))?; // Parse cursor let cursor_value = datetime_cursor(query.cursor.as_ref()); diff --git a/parakeet/src/xrpc/app_bsky/feed/likes.rs b/parakeet/src/xrpc/app_bsky/feed/likes.rs index 481fff28..193fccda 100644 --- a/parakeet/src/xrpc/app_bsky/feed/likes.rs +++ b/parakeet/src/xrpc/app_bsky/feed/likes.rs @@ -1,5 +1,5 @@ -use crate::xrpc::error::{Error, XrpcResult}; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::{Error, XrpcResult}; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::xrpc::{datetime_cursor, ActorWithCursorQuery}; use crate::GlobalState; use axum::extract::{Query, State}; diff --git a/parakeet/src/xrpc/app_bsky/feed/posts/feeds.rs b/parakeet/src/xrpc/app_bsky/feed/posts/feeds.rs index a44726e6..60f27554 100644 --- a/parakeet/src/xrpc/app_bsky/feed/posts/feeds.rs +++ b/parakeet/src/xrpc/app_bsky/feed/posts/feeds.rs @@ -4,8 +4,8 @@ use lexica::app_bsky::feed::{FeedViewPost, GeneratorView}; use serde::{Deserialize, Serialize}; use crate::xrpc::datetime_cursor; -use crate::xrpc::error::XrpcResult; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::XrpcResult; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::GlobalState; #[derive(Debug, Deserialize)] @@ -84,7 +84,7 @@ pub async fn get_feed_generator( let view = state.feedgen_entity .get_by_uri(&query.feed, viewer_did.as_deref()) .await? - .ok_or_else(|| crate::xrpc::error::Error::not_found())?; + .ok_or_else(|| crate::common::errors::Error::not_found())?; Ok(Json(GetFeedGeneratorRes { view, diff --git a/parakeet/src/xrpc/app_bsky/feed/posts/helpers.rs b/parakeet/src/xrpc/app_bsky/feed/posts/helpers.rs index 47e14406..6f4f6aaa 100644 --- a/parakeet/src/xrpc/app_bsky/feed/posts/helpers.rs +++ b/parakeet/src/xrpc/app_bsky/feed/posts/helpers.rs @@ -9,7 +9,7 @@ use lexica::app_bsky::feed::{ use reqwest::Url; use std::collections::HashMap; -use crate::xrpc::error::{Error, XrpcResult}; +use crate::common::errors::{Error, XrpcResult}; #[expect(dead_code)] const FEEDGEN_SERVICE_ID: &str = "#bsky_fg"; diff --git a/parakeet/src/xrpc/app_bsky/feed/posts/queries.rs b/parakeet/src/xrpc/app_bsky/feed/posts/queries.rs index e86679eb..42119e6e 100644 --- a/parakeet/src/xrpc/app_bsky/feed/posts/queries.rs +++ b/parakeet/src/xrpc/app_bsky/feed/posts/queries.rs @@ -4,8 +4,8 @@ use axum_extra::extract::Query as ExtraQuery; use lexica::app_bsky::feed::PostView; use serde::{Deserialize, Serialize}; -use crate::xrpc::error::XrpcResult; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::XrpcResult; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::GlobalState; #[derive(Debug, Deserialize)] @@ -67,7 +67,7 @@ pub async fn get_post( let post_view = state.post_entity .get_by_uri(&query.uri, viewer_did.as_deref()) .await? - .ok_or_else(|| crate::xrpc::error::Error::not_found())?; + .ok_or_else(|| crate::common::errors::Error::not_found())?; // Extract the record value from the PostView // This is a simplified response - the actual endpoint returns the raw record @@ -131,11 +131,11 @@ pub async fn get_quotes( let embed_actor_id = state.profile_entity .resolve_identifier(embed_did) .await - .map_err(|_| crate::xrpc::error::Error::not_found())?; + .map_err(|_| crate::common::errors::Error::not_found())?; // Decode rkey let embed_rkey = parakeet_db::tid_util::decode_tid(embed_rkey_str) - .map_err(|_| crate::xrpc::error::Error::invalid_request(Some("Invalid rkey".to_string())))?; + .map_err(|_| crate::common::errors::Error::invalid_request(Some("Invalid rkey".to_string())))?; // Parse cursor let (cursor_actor_id, cursor_rkey) = if let Some(ref cursor) = query.cursor { @@ -155,7 +155,7 @@ pub async fn get_quotes( // Get quotes from database using PostEntity let cursor = cursor_actor_id.zip(cursor_rkey); let results = state.post_entity.get_quotes(embed_actor_id, embed_rkey, cursor, limit).await - .map_err(|e| crate::xrpc::error::Error::server_error(Some(&e.to_string())))?; + .map_err(|e| crate::common::errors::Error::server_error(Some(&e.to_string())))?; // Convert to PostViews let mut posts = Vec::new(); @@ -235,11 +235,11 @@ pub async fn get_reposted_by( let post_actor_id = state.profile_entity .resolve_identifier(post_did) .await - .map_err(|_| crate::xrpc::error::Error::not_found())?; + .map_err(|_| crate::common::errors::Error::not_found())?; // Decode rkey let post_rkey = parakeet_db::tid_util::decode_tid(post_rkey_str) - .map_err(|_| crate::xrpc::error::Error::invalid_request(Some("Invalid rkey".to_string())))?; + .map_err(|_| crate::common::errors::Error::invalid_request(Some("Invalid rkey".to_string())))?; // Parse cursor let cursor_rkey = query.cursor.as_ref() @@ -247,7 +247,7 @@ pub async fn get_reposted_by( // Get reposted by from database using PostEntity let results = state.post_entity.get_reposted_by(post_actor_id, post_rkey, cursor_rkey, limit).await - .map_err(|e| crate::xrpc::error::Error::server_error(Some(&e.to_string())))?; + .map_err(|e| crate::common::errors::Error::server_error(Some(&e.to_string())))?; // Convert to ProfileViews let actor_ids: Vec = results.iter().map(|(actor_id, _)| *actor_id).collect(); diff --git a/parakeet/src/xrpc/app_bsky/feed/posts/threads.rs b/parakeet/src/xrpc/app_bsky/feed/posts/threads.rs index 4cb3c6ed..50a92682 100644 --- a/parakeet/src/xrpc/app_bsky/feed/posts/threads.rs +++ b/parakeet/src/xrpc/app_bsky/feed/posts/threads.rs @@ -5,8 +5,8 @@ use serde::{Deserialize, Serialize}; use std::collections::HashMap; use crate::entities::post::ThreadItem; -use crate::xrpc::error::{Error, XrpcResult}; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::{Error, XrpcResult}; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::GlobalState; use super::helpers::postview_to_tvpt; diff --git a/parakeet/src/xrpc/app_bsky/feed/search.rs b/parakeet/src/xrpc/app_bsky/feed/search.rs index 75cd1ae3..7cea6c51 100644 --- a/parakeet/src/xrpc/app_bsky/feed/search.rs +++ b/parakeet/src/xrpc/app_bsky/feed/search.rs @@ -1,5 +1,5 @@ -use crate::xrpc::error::{Error, XrpcResult}; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::{Error, XrpcResult}; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::GlobalState; use axum::extract::{Query, State}; use axum::Json; diff --git a/parakeet/src/xrpc/app_bsky/graph/lists.rs b/parakeet/src/xrpc/app_bsky/graph/lists.rs index a2bd77c6..24a67e1f 100644 --- a/parakeet/src/xrpc/app_bsky/graph/lists.rs +++ b/parakeet/src/xrpc/app_bsky/graph/lists.rs @@ -1,5 +1,5 @@ -use crate::xrpc::error::{Error, XrpcResult}; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::{Error, XrpcResult}; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::xrpc::{datetime_cursor, ActorWithCursorQuery, CursorQuery}; use crate::GlobalState; use axum::extract::{Query, State}; diff --git a/parakeet/src/xrpc/app_bsky/graph/mutes.rs b/parakeet/src/xrpc/app_bsky/graph/mutes.rs index e463aa47..ec612a5f 100644 --- a/parakeet/src/xrpc/app_bsky/graph/mutes.rs +++ b/parakeet/src/xrpc/app_bsky/graph/mutes.rs @@ -1,6 +1,6 @@ use crate::xrpc::datetime_cursor; -use crate::xrpc::error::{Error, XrpcResult}; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::{Error, XrpcResult}; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::xrpc::CursorQuery; use crate::GlobalState; use axum::extract::{Query, State}; diff --git a/parakeet/src/xrpc/app_bsky/graph/relations.rs b/parakeet/src/xrpc/app_bsky/graph/relations.rs index dacc3ddb..f19ba333 100644 --- a/parakeet/src/xrpc/app_bsky/graph/relations.rs +++ b/parakeet/src/xrpc/app_bsky/graph/relations.rs @@ -1,6 +1,6 @@ use crate::xrpc::datetime_cursor; -use crate::xrpc::error::{Error, XrpcResult}; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::{Error, XrpcResult}; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::xrpc::{ActorWithCursorQuery, CursorQuery}; use crate::GlobalState; diff --git a/parakeet/src/xrpc/app_bsky/graph/search.rs b/parakeet/src/xrpc/app_bsky/graph/search.rs index ede18311..3c0e3944 100644 --- a/parakeet/src/xrpc/app_bsky/graph/search.rs +++ b/parakeet/src/xrpc/app_bsky/graph/search.rs @@ -1,5 +1,5 @@ -use crate::xrpc::error::XrpcResult; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::XrpcResult; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::GlobalState; use axum::extract::{Query, State}; use axum::Json; @@ -37,7 +37,7 @@ pub async fn search_actors( // Search actors using ProfileEntity let results = state.profile_entity.search_actors(&query.q, limit as i64, cursor_value) .await - .map_err(|e| crate::xrpc::error::Error::server_error(Some(&e.to_string())))?; + .map_err(|e| crate::common::errors::Error::server_error(Some(&e.to_string())))?; // Check for pagination let has_more = results.len() > limit as usize; @@ -82,7 +82,7 @@ pub async fn search_actors_skeleton( // Search actors using ProfileEntity let results = state.profile_entity.search_actors(&query.q, limit as i64, cursor_value) .await - .map_err(|e| crate::xrpc::error::Error::server_error(Some(&e.to_string())))?; + .map_err(|e| crate::common::errors::Error::server_error(Some(&e.to_string())))?; // Check for pagination let has_more = results.len() > limit as usize; diff --git a/parakeet/src/xrpc/app_bsky/graph/starter_packs.rs b/parakeet/src/xrpc/app_bsky/graph/starter_packs.rs index 0306a386..5727aa6f 100644 --- a/parakeet/src/xrpc/app_bsky/graph/starter_packs.rs +++ b/parakeet/src/xrpc/app_bsky/graph/starter_packs.rs @@ -1,5 +1,5 @@ -use crate::xrpc::error::{Error, XrpcResult}; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::{Error, XrpcResult}; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::xrpc::{datetime_cursor, ActorWithCursorQuery}; use crate::GlobalState; use axum::extract::{Query, State}; diff --git a/parakeet/src/xrpc/app_bsky/graph/suggestions.rs b/parakeet/src/xrpc/app_bsky/graph/suggestions.rs index 6ac55601..8dfd8450 100644 --- a/parakeet/src/xrpc/app_bsky/graph/suggestions.rs +++ b/parakeet/src/xrpc/app_bsky/graph/suggestions.rs @@ -1,5 +1,5 @@ -use crate::xrpc::error::XrpcResult; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::XrpcResult; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::GlobalState; use axum::extract::{Query, State}; use axum::Json; @@ -32,7 +32,7 @@ pub async fn get_suggestions( // Resolve viewer to actor_id using ProfileEntity let viewer_id = state.profile_entity.resolve_identifier(&viewer_did).await - .map_err(|_| crate::xrpc::error::Error::actor_not_found(&viewer_did))?; + .map_err(|_| crate::common::errors::Error::actor_not_found(&viewer_did))?; // Get suggested actors - for now just return top followed actors // TODO: Implement proper suggestion algorithm (friends of friends, similar interests, etc) @@ -73,7 +73,7 @@ pub async fn get_suggestions_skeleton( // Resolve viewer to actor_id using ProfileEntity let viewer_id = state.profile_entity.resolve_identifier(&viewer_did).await - .map_err(|_| crate::xrpc::error::Error::actor_not_found(&viewer_did))?; + .map_err(|_| crate::common::errors::Error::actor_not_found(&viewer_did))?; // Get suggested actors from database // Get suggested actors - for now just return top followed actors diff --git a/parakeet/src/xrpc/app_bsky/graph/thread_mutes.rs b/parakeet/src/xrpc/app_bsky/graph/thread_mutes.rs index 68ac43a0..a25087c9 100644 --- a/parakeet/src/xrpc/app_bsky/graph/thread_mutes.rs +++ b/parakeet/src/xrpc/app_bsky/graph/thread_mutes.rs @@ -1,5 +1,5 @@ -use crate::xrpc::error::XrpcResult; -use crate::xrpc::extract::AtpAuth; +use crate::common::errors::XrpcResult; +use crate::common::auth::AtpAuth; use crate::GlobalState; use axum::extract::State; use axum::Json; @@ -15,11 +15,11 @@ pub async fn mute_thread( auth: AtpAuth, Json(form): Json, ) -> XrpcResult> { - use crate::xrpc::error::Error; + use crate::common::errors::Error; let mut conn = state.pool.get().await?; // Resolve authenticated user's actor_id via IdCache - let actor_id = crate::id_cache_helpers::get_actor_id_or_fetch( + let actor_id = crate::common::cache::get_actor_id_or_fetch( &state.pool, &state.id_cache, &auth.0, @@ -39,7 +39,7 @@ pub async fn mute_thread( let rkey_bigint = parakeet_db::tid_util::decode_tid(rkey_str) .map_err(|_| Error::invalid_request(Some("Invalid TID in root URI".into())))?; - let root_post_actor_id = crate::id_cache_helpers::get_actor_id_or_fetch( + let root_post_actor_id = crate::common::cache::get_actor_id_or_fetch( &state.pool, &state.id_cache, root_did, @@ -75,11 +75,11 @@ pub async fn unmute_thread( auth: AtpAuth, Json(form): Json, ) -> XrpcResult> { - use crate::xrpc::error::Error; + use crate::common::errors::Error; let mut conn = state.pool.get().await?; // Resolve authenticated user's actor_id via IdCache - let actor_id = crate::id_cache_helpers::get_actor_id_or_fetch( + let actor_id = crate::common::cache::get_actor_id_or_fetch( &state.pool, &state.id_cache, &auth.0, @@ -99,7 +99,7 @@ pub async fn unmute_thread( let rkey_bigint = parakeet_db::tid_util::decode_tid(rkey_str) .map_err(|_| Error::invalid_request(Some("Invalid TID in root URI".into())))?; - let root_post_actor_id = crate::id_cache_helpers::get_actor_id_or_fetch( + let root_post_actor_id = crate::common::cache::get_actor_id_or_fetch( &state.pool, &state.id_cache, root_did, diff --git a/parakeet/src/xrpc/app_bsky/labeler.rs b/parakeet/src/xrpc/app_bsky/labeler.rs index 3386744e..d9bb09a2 100644 --- a/parakeet/src/xrpc/app_bsky/labeler.rs +++ b/parakeet/src/xrpc/app_bsky/labeler.rs @@ -1,5 +1,5 @@ -use crate::xrpc::error::XrpcResult; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::XrpcResult; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::GlobalState; use axum::extract::State; use axum::Json; diff --git a/parakeet/src/xrpc/app_bsky/notification/mod.rs b/parakeet/src/xrpc/app_bsky/notification/mod.rs index 52c02aa5..a382ba56 100644 --- a/parakeet/src/xrpc/app_bsky/notification/mod.rs +++ b/parakeet/src/xrpc/app_bsky/notification/mod.rs @@ -1,5 +1,5 @@ -use crate::xrpc::error::{Error, XrpcResult}; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::{Error, XrpcResult}; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::GlobalState; use axum::extract::{Query, State}; use axum::Json; @@ -194,7 +194,7 @@ pub async fn list_notifications( // Resolve actor_ids to DIDs using IdCache helper (with automatic database fallback) let actor_id_vec: Vec = actor_ids_to_resolve.into_iter().collect(); - let actor_did_map = crate::id_cache_helpers::get_actor_dids_or_fetch( + let actor_did_map = crate::common::cache::get_actor_dids_or_fetch( &state.pool, &state.id_cache, &actor_id_vec, @@ -276,7 +276,7 @@ pub async fn list_notifications( let actor_id_vec: Vec = additional_actor_ids.into_iter().collect(); // Use IdCache helper for additional actors - let additional_dids = crate::id_cache_helpers::get_actor_dids_or_fetch( + let additional_dids = crate::common::cache::get_actor_dids_or_fetch( &state.pool, &state.id_cache, &actor_id_vec, diff --git a/parakeet/src/xrpc/app_bsky/unspecced/mod.rs b/parakeet/src/xrpc/app_bsky/unspecced/mod.rs index b6db6a76..cfb15e36 100644 --- a/parakeet/src/xrpc/app_bsky/unspecced/mod.rs +++ b/parakeet/src/xrpc/app_bsky/unspecced/mod.rs @@ -1,6 +1,6 @@ use std::collections::HashMap; -use crate::xrpc::error::{Error, XrpcResult}; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::{Error, XrpcResult}; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::GlobalState; use axum::extract::{Query, State}; use axum::Json; @@ -231,7 +231,7 @@ pub async fn get_suggested_users( } // Convert DIDs to actor_ids for profile loading - let actor_id_map = match crate::id_cache_helpers::get_actor_ids_or_fetch( + let actor_id_map = match crate::common::cache::get_actor_ids_or_fetch( &state.pool, &state.id_cache, &all_dids, @@ -563,7 +563,7 @@ pub async fn get_suggested_starter_packs( // Hydrate starter packs maintaining order let (maybe_did, maybe_actor_id) = if let Some(auth) = maybe_auth { let did = auth.0.clone(); - let actor_id = crate::id_cache_helpers::get_actor_id_or_fetch(&state.pool, &state.id_cache, &did).await.ok(); + let actor_id = crate::common::cache::get_actor_id_or_fetch(&state.pool, &state.id_cache, &did).await.ok(); (Some(did), actor_id) } else { (None, None) diff --git a/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/other_replies.rs b/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/other_replies.rs index 5f912aac..a92b2d4d 100644 --- a/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/other_replies.rs +++ b/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/other_replies.rs @@ -4,8 +4,8 @@ use axum::Json; use std::collections::HashMap; // use crate::hydration::StatefulHydrator; // Removed - using entities now -use crate::xrpc::error::{Error, XrpcResult}; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::{Error, XrpcResult}; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::xrpc::normalise_at_uri; use crate::GlobalState; @@ -59,7 +59,7 @@ pub async fn get_post_thread_other_v2( let is_authenticated = maybe_auth.is_some(); let (maybe_did, maybe_actor_id) = if let Some(auth) = maybe_auth { let did = auth.0.clone(); - let actor_id = crate::id_cache_helpers::get_actor_id_or_fetch(&state.pool, &state.id_cache, &did).await.ok(); + let actor_id = crate::common::cache::get_actor_id_or_fetch(&state.pool, &state.id_cache, &did).await.ok(); (Some(did), actor_id) } else { (None, None) diff --git a/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/post_thread.rs b/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/post_thread.rs index dfd740a0..79fe9da1 100644 --- a/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/post_thread.rs +++ b/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/post_thread.rs @@ -3,8 +3,8 @@ use axum::response::{IntoResponse as _, Response}; use axum::Json; // use crate::hydration::StatefulHydrator; // Removed - using entities now -use crate::xrpc::error::{Error, XrpcResult}; -use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; +use crate::common::errors::{Error, XrpcResult}; +use crate::common::auth::{AtpAcceptLabelers, AtpAuth}; use crate::xrpc::normalise_at_uri; use crate::GlobalState; @@ -28,7 +28,7 @@ pub async fn get_post_thread_v2( let is_authenticated = maybe_auth.is_some(); let (maybe_did, maybe_actor_id) = if let Some(auth) = maybe_auth { let did = auth.0.clone(); - let actor_id = crate::id_cache_helpers::get_actor_id_or_fetch(&state.pool, &state.id_cache, &did).await.ok(); + let actor_id = crate::common::cache::get_actor_id_or_fetch(&state.pool, &state.id_cache, &did).await.ok(); (Some(did), actor_id) } else { (None, None) diff --git a/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/thread_builder.rs b/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/thread_builder.rs index 5a64974d..8756f840 100644 --- a/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/thread_builder.rs +++ b/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/thread_builder.rs @@ -63,7 +63,7 @@ impl ThreadBuilder<'_> { pub async fn build_thread( &self, conn: &mut diesel_async::AsyncPgConnection, - ) -> crate::xrpc::error::XrpcResult<(Vec, bool)> { + ) -> crate::common::errors::XrpcResult<(Vec, bool)> { let start = std::time::Instant::now(); let mut thread_items = Vec::new(); @@ -103,7 +103,7 @@ impl ThreadBuilder<'_> { &self, conn: &mut diesel_async::AsyncPgConnection, thread_items: &mut Vec, - ) -> crate::xrpc::error::XrpcResult<()> { + ) -> crate::common::errors::XrpcResult<()> { // Use actor_id-based query with IdCache // NO FALLBACK - we require IdCache to be populated (should always be from hydrate_post) let db_start = std::time::Instant::now(); @@ -133,10 +133,10 @@ impl ThreadBuilder<'_> { // Get actor_id from cache (should always be cached after hydrate_post) let cached_actor = self.id_cache.get_actor_id(anchor_did).await - .ok_or_else(|| crate::xrpc::error::Error::server_error(Some("Actor not in cache")))?; + .ok_or_else(|| crate::common::errors::Error::server_error(Some("Actor not in cache")))?; let anchor_rkey = parakeet_db::tid_util::decode_tid(anchor_rkey_base32) - .map_err(|_| crate::xrpc::error::Error::invalid_request(Some("Invalid rkey".to_string())))?; + .map_err(|_| crate::common::errors::Error::invalid_request(Some("Invalid rkey".to_string())))?; // Get root actor_id if we have a root URI let root_info = if let Some(ref root_uri_str) = root_uri { @@ -145,12 +145,12 @@ impl ThreadBuilder<'_> { let root_did = root_parts[0]; let root_rkey_base32 = root_parts[2]; let root_cached = self.id_cache.get_actor_id(root_did).await - .ok_or_else(|| crate::xrpc::error::Error::server_error(Some("Root actor not in cache")))?; + .ok_or_else(|| crate::common::errors::Error::server_error(Some("Root actor not in cache")))?; let root_rkey = parakeet_db::tid_util::decode_tid(root_rkey_base32) - .map_err(|_| crate::xrpc::error::Error::invalid_request(Some("Invalid root rkey".to_string())))?; + .map_err(|_| crate::common::errors::Error::invalid_request(Some("Invalid root rkey".to_string())))?; (root_cached.actor_id, root_rkey) } else { - return Err(crate::xrpc::error::Error::invalid_request(Some("Invalid root URI".to_string())).into()); + return Err(crate::common::errors::Error::invalid_request(Some("Invalid root URI".to_string())).into()); } } else { // No root means this IS the root (top-level post) @@ -166,7 +166,7 @@ impl ThreadBuilder<'_> { ) .await? } else { - return Err(crate::xrpc::error::Error::invalid_request(Some("Invalid URI format".to_string())).into()); + return Err(crate::common::errors::Error::invalid_request(Some("Invalid URI format".to_string())).into()); }; let db_elapsed = db_start.elapsed().as_secs_f64() * 1000.0; @@ -248,7 +248,7 @@ impl ThreadBuilder<'_> { &self, conn: &mut diesel_async::AsyncPgConnection, thread_items: &mut Vec, - ) -> crate::xrpc::error::XrpcResult<()> { + ) -> crate::common::errors::XrpcResult<()> { // Get all replies from database with branching factor // Query with branching_factor + 1 to know if there are more replies // Use actor_id-based query with IdCache @@ -263,10 +263,10 @@ impl ThreadBuilder<'_> { // Get actor_id from cache (should always be cached after hydrate_post) let cached_actor = self.id_cache.get_actor_id(anchor_did).await - .ok_or_else(|| crate::xrpc::error::Error::server_error(Some("Actor not in cache")))?; + .ok_or_else(|| crate::common::errors::Error::server_error(Some("Actor not in cache")))?; let anchor_rkey = parakeet_db::tid_util::decode_tid(anchor_rkey_base32) - .map_err(|_| crate::xrpc::error::Error::invalid_request(Some("Invalid rkey".to_string())))?; + .map_err(|_| crate::common::errors::Error::invalid_request(Some("Invalid rkey".to_string())))?; self.post_entity.get_thread_children_by_arrays( cached_actor.actor_id, @@ -276,7 +276,7 @@ impl ThreadBuilder<'_> { ) .await? } else { - return Err(crate::xrpc::error::Error::invalid_request(Some("Invalid URI format".to_string())).into()); + return Err(crate::common::errors::Error::invalid_request(Some("Invalid URI format".to_string())).into()); }; let db_elapsed = db_start.elapsed().as_secs_f64() * 1000.0; diff --git a/parakeet/src/xrpc/cdn.rs b/parakeet/src/xrpc/cdn.rs deleted file mode 100644 index ff63bc17..00000000 --- a/parakeet/src/xrpc/cdn.rs +++ /dev/null @@ -1,41 +0,0 @@ -/// For a CDN that uses paths identically to Bluesky -pub struct BskyCdn { - cdn_base: String, - video_base: String, -} - -impl BskyCdn { - pub fn new(cdn_base: String, video_base: String) -> Self { - Self { - cdn_base, - video_base, - } - } - - pub fn avatar(&self, did: &str, cid: &str) -> String { - format!("{}/img/avatar/plain/{did}/{cid}@jpeg", self.cdn_base) - } - - pub fn banner(&self, did: &str, cid: &str) -> String { - format!("{}/img/banner/plain/{did}/{cid}@jpeg", self.cdn_base) - } - - pub fn embed_thumb(&self, did: &str, cid: &str) -> String { - format!( - "{}/img/feed_thumbnail/plain/{did}/{cid}@jpeg", - self.cdn_base - ) - } - - pub fn embed_fullsize(&self, did: &str, cid: &str) -> String { - format!("{}/img/feed_fullsize/plain/{did}/{cid}@jpeg", self.cdn_base) - } - - pub fn video_thumb(&self, did: &str, cid: &str) -> String { - format!("{}/watch/{did}/{cid}/thumbnail.jpg", self.video_base) - } - - pub fn video_playlist(&self, did: &str, cid: &str) -> String { - format!("{}/watch/{did}/{cid}/playlist.m3u8", self.video_base) - } -} diff --git a/parakeet/src/xrpc/com_atproto/identity.rs b/parakeet/src/xrpc/com_atproto/identity.rs index 074b08ea..059544cc 100644 --- a/parakeet/src/xrpc/com_atproto/identity.rs +++ b/parakeet/src/xrpc/com_atproto/identity.rs @@ -1,5 +1,5 @@ -use crate::xrpc::error::{Error, XrpcResult}; -use crate::xrpc::extract::AtpAuth; +use crate::common::errors::{Error, XrpcResult}; +use crate::common::auth::AtpAuth; use crate::GlobalState; use axum::extract::{Query, State}; use axum::Json; diff --git a/parakeet/src/xrpc/com_atproto/repo.rs b/parakeet/src/xrpc/com_atproto/repo.rs index b92ef672..3cf405d0 100644 --- a/parakeet/src/xrpc/com_atproto/repo.rs +++ b/parakeet/src/xrpc/com_atproto/repo.rs @@ -1,6 +1,6 @@ use crate::xrpc::check_actor_status; -use crate::xrpc::error::{Error, XrpcResult}; -use crate::xrpc::extract::AtpAuth; +use crate::common::errors::{Error, XrpcResult}; +use crate::common::auth::AtpAuth; use crate::GlobalState; use axum::extract::{Query, State}; use axum::Json; diff --git a/parakeet/src/xrpc/community_lexicon/bookmarks.rs b/parakeet/src/xrpc/community_lexicon/bookmarks.rs index f0a496fa..aa3d81cc 100644 --- a/parakeet/src/xrpc/community_lexicon/bookmarks.rs +++ b/parakeet/src/xrpc/community_lexicon/bookmarks.rs @@ -1,6 +1,6 @@ use crate::xrpc::datetime_cursor; -use crate::xrpc::error::XrpcResult; -use crate::xrpc::extract::AtpAuth; +use crate::common::errors::XrpcResult; +use crate::common::auth::AtpAuth; use crate::GlobalState; use axum::extract::{Query, State}; use axum::Json; @@ -31,7 +31,7 @@ pub async fn get_actor_bookmarks( let limit = query.limit.unwrap_or(50).clamp(1, 100); // Resolve DID to actor_id via IdCache - let actor_id = crate::id_cache_helpers::get_actor_id_or_fetch( + let actor_id = crate::common::cache::get_actor_id_or_fetch( &state.pool, &state.id_cache, &auth.0, diff --git a/parakeet/src/xrpc/cursor.rs b/parakeet/src/xrpc/cursor.rs deleted file mode 100644 index 06cb903b..00000000 --- a/parakeet/src/xrpc/cursor.rs +++ /dev/null @@ -1,44 +0,0 @@ -use serde::Deserialize; - -/// Parses a TID cursor string (base32-encoded) into an i64 rkey -/// -/// This matches the official Bluesky API which uses base32-encoded TIDs -/// like "3m47vaoniuy2d" for pagination cursors. -pub fn tid_cursor(cursor: Option<&String>) -> Option { - cursor.and_then(|v| parakeet_db::tid_util::decode_tid(v).ok()) -} - -/// Parses a datetime cursor string (ISO 8601 format) into a DateTime -/// -/// This matches the official Bluesky API which uses ISO 8601 timestamps -/// like "2025-10-27T01:36:17.072Z" instead of millisecond integers. -/// -/// For backward compatibility, also accepts millisecond timestamps. -pub fn datetime_cursor(cursor: Option<&String>) -> Option> { - cursor.and_then(|v| { - // Try ISO 8601 first (official format) - if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(v) { - return Some(dt.with_timezone(&chrono::Utc)); - } - - // Fall back to millisecond timestamp for backward compatibility - v.parse::() - .ok() - .and_then(chrono::DateTime::from_timestamp_millis) - }) -} - -/// Common query structure for endpoints with cursor pagination -#[derive(Debug, Deserialize)] -pub struct CursorQuery { - pub limit: Option, - pub cursor: Option, -} - -/// Common query structure for endpoints with actor and cursor pagination -#[derive(Debug, Deserialize)] -pub struct ActorWithCursorQuery { - pub actor: String, - pub limit: Option, - pub cursor: Option, -} diff --git a/parakeet/src/xrpc/extract.rs b/parakeet/src/xrpc/extract.rs deleted file mode 100644 index 42c2e6ac..00000000 --- a/parakeet/src/xrpc/extract.rs +++ /dev/null @@ -1,117 +0,0 @@ -use crate::GlobalState; -use axum::extract::{FromRequestParts, OptionalFromRequestParts}; -use axum::http::request::Parts; -use axum::http::StatusCode; -use axum_extra::headers::authorization::Bearer; -use axum_extra::headers::Authorization; -use axum_extra::TypedHeader; - -#[derive(Debug)] -pub struct LabelConfigItem { - pub labeler: String, - pub redact: bool, -} - -impl std::str::FromStr for LabelConfigItem { - type Err = std::convert::Infallible; - - fn from_str(val: &str) -> Result { - let v = val.trim(); - - let Some((did, rem)) = v.split_once(';') else { - return Ok(Self { - labeler: v.to_owned(), - redact: false, - }); - }; - - Ok(Self { - labeler: did.to_owned(), - redact: rem == "redact", - }) - } -} - -#[derive(Debug)] -pub struct AtpAcceptLabelers(pub Vec); - -impl FromRequestParts for AtpAcceptLabelers -where - S: Send + Sync, -{ - type Rejection = (StatusCode, &'static str); - - async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result { - let labelers = match parts.headers.get("Atproto-Accept-Labelers") { - Some(labelers) => { - let labelers = labelers.to_str().map_err(|_err| { - ( - StatusCode::BAD_REQUEST, - "Atproto-Accept-Labelers was invalid", - ) - })?; - - labelers - .trim() - .split(",") - .map(|s| s.parse().unwrap()) - .collect() - } - None => vec![], - }; - - Ok(Self(labelers)) - } -} - -#[derive(Clone, Debug)] -pub struct AtpAuth(pub String); - -type BearerHeader = TypedHeader>; - -impl FromRequestParts for AtpAuth { - type Rejection = (StatusCode, String); - - async fn from_request_parts( - parts: &mut Parts, - state: &GlobalState, - ) -> Result { - let hdr = >::from_request_parts( - parts, state, - ) - .await - .map_err(|err| (StatusCode::BAD_REQUEST, err.to_string()))? - .ok_or_else(|| (StatusCode::UNAUTHORIZED, "missing JWT".to_owned()))?; - - let jwt_result = state.jwt.resolve_and_verify_jwt(hdr.token(), None).await; - match jwt_result { - Some(claims) => Ok(Self(claims.iss)), - None => Err((StatusCode::INTERNAL_SERVER_ERROR, "JWT error".to_owned())), - } - } -} - -impl OptionalFromRequestParts for AtpAuth { - type Rejection = (StatusCode, String); - - async fn from_request_parts( - parts: &mut Parts, - state: &GlobalState, - ) -> Result, Self::Rejection> { - let Some(hdr) = - >::from_request_parts( - parts, state, - ) - .await - .map_err(|err| (StatusCode::BAD_REQUEST, err.to_string()))? - else { - return Ok(None); - }; - - let jwt_result = state.jwt.resolve_and_verify_jwt(hdr.token(), None).await; - match jwt_result { - Some(claims) => Ok(Some(Self(claims.iss))), - None => Err((StatusCode::INTERNAL_SERVER_ERROR, "JWT error".to_owned())), - } - } -} diff --git a/parakeet/src/xrpc/jwt.rs b/parakeet/src/xrpc/jwt.rs deleted file mode 100644 index ba5e29ad..00000000 --- a/parakeet/src/xrpc/jwt.rs +++ /dev/null @@ -1,100 +0,0 @@ -use did_resolver::Resolver; -use jsonwebtoken::{Algorithm, DecodingKey, Validation}; -use serde::{Deserialize, Serialize}; -use std::collections::HashMap; -use std::sync::{Arc, LazyLock}; -use tokio::sync::RwLock; - -static DUMMY_KEY: LazyLock = LazyLock::new(|| DecodingKey::from_secret(&[])); -static NO_VERIFY: LazyLock = LazyLock::new(|| { - let mut val_no_verify = Validation::default(); - val_no_verify.insecure_disable_signature_validation(); - val_no_verify.validate_aud = false; - val_no_verify -}); - -#[derive(Debug, Deserialize, Serialize)] -pub struct Claims { - pub aud: String, - pub exp: usize, - pub iat: usize, - pub iss: String, - pub lxm: Option, - pub jti: String, -} - -pub struct JwtVerifier { - aud: String, - resolver: Arc, - key_cache: RwLock>, -} - -impl JwtVerifier { - pub fn new(aud: String, resolver: Arc) -> Self { - Self { - aud, - resolver, - key_cache: RwLock::new(HashMap::new()), - } - } - - pub async fn resolve_and_verify_jwt(&self, token: &str, aud: Option<&str>) -> Option { - // first we need to decode without verifying, to get iss. - let unsafe_data = jsonwebtoken::decode::(token, &DUMMY_KEY, &NO_VERIFY).ok()?; - let unsafe_iss = unsafe_data.claims.iss; - - let maybe_cached_key = { - let l = self.key_cache.read().await; - l.get(&unsafe_iss).cloned() - }; - let multibase_key = match maybe_cached_key { - Some(key) => key, - None => self.resolve_key(&unsafe_iss).await?, - }; - - let aud = aud.unwrap_or(&self.aud); - self.verify_jwt_multibase_with_alg(token, &multibase_key, unsafe_data.header.alg, aud) - } - - async fn resolve_key(&self, did: &str) -> Option { - tracing::trace!("resolving multikey for {did}"); - let did_doc = self.resolver.resolve_did(did).await.ok()??; - - // try find the multibase key - let multikey = did_doc.find_verif_method_by_type("Multikey")?; - - { - let mut l = self.key_cache.write().await; - drop(l.insert(did.to_owned(), multikey.public_key_multibase.clone())); - } - - Some(multikey.public_key_multibase.clone()) - } - - pub fn verify_jwt_multibase(&self, token: &str, multibase_key: &str) -> Option { - let alg = jsonwebtoken::decode_header(token).ok()?.alg; - - self.verify_jwt_multibase_with_alg(token, multibase_key, alg, &self.aud) - } - - pub fn verify_jwt_multibase_with_alg( - &self, - token: &str, - multibase_key: &str, - alg: Algorithm, - aud: &str, - ) -> Option { - // decode the multibase key - let (_, key) = multibase::decode(multibase_key).ok()?; - - let key = DecodingKey::from_ec_der(&key[2..]); - - let mut validation = Validation::new(alg); - validation.validate_aud = false; - validation.set_audience(&[&aud]); - - let decoded = jsonwebtoken::decode::(token, &key, &validation).ok()?; - - Some(decoded.claims) - } -} diff --git a/parakeet/src/xrpc/mod.rs b/parakeet/src/xrpc/mod.rs index b49f1474..5b5953cd 100644 --- a/parakeet/src/xrpc/mod.rs +++ b/parakeet/src/xrpc/mod.rs @@ -1,20 +1,16 @@ mod app_bsky; -pub mod cdn; mod com_atproto; mod community_lexicon; -pub mod cursor; -pub mod error; -pub mod extract; -mod helpers_entity; -pub mod jwt; use axum::routing::get; use axum::{Json, Router}; use serde::Serialize; -// Re-export commonly used items -pub use cursor::{datetime_cursor, tid_cursor, ActorWithCursorQuery, CursorQuery}; -pub use helpers_entity::{check_actor_status, normalise_at_uri}; +// Re-export commonly used items from common module +pub use crate::common::helpers::{ + check_actor_status, datetime_cursor, normalise_at_uri, tid_cursor, ActorWithCursorQuery, + CursorQuery, +}; #[derive(Serialize)] struct HealthResponse {