From cfac0b2a09a333adcbbbcd3f78d172469a056843 Mon Sep 17 00:00:00 2001 From: Timothy Quilling Date: Wed, 17 Dec 2025 14:12:30 -0500 Subject: [PATCH] reorganize --- parakeet/src/common/auth.rs | 7 + parakeet/src/common/cache/id_helpers.rs | 2 +- parakeet/src/common/helpers.rs | 116 +++++++++++++-- parakeet/src/common/rate_limiting.rs | 187 ++++++++++++++++++++++++ 4 files changed, 300 insertions(+), 12 deletions(-) diff --git a/parakeet/src/common/auth.rs b/parakeet/src/common/auth.rs index 3d89ea6a..e8ab2e34 100644 --- a/parakeet/src/common/auth.rs +++ b/parakeet/src/common/auth.rs @@ -1,3 +1,10 @@ +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; use did_resolver::Resolver; use jsonwebtoken::{Algorithm, DecodingKey, Validation}; use serde::{Deserialize, Serialize}; diff --git a/parakeet/src/common/cache/id_helpers.rs b/parakeet/src/common/cache/id_helpers.rs index 3f651206..e5883f2b 100644 --- a/parakeet/src/common/cache/id_helpers.rs +++ b/parakeet/src/common/cache/id_helpers.rs @@ -3,7 +3,7 @@ //! These functions combine IdCache lookups with database queries, //! automatically fetching and caching missing entries. -use crate::xrpc::error::{Error, XrpcResult}; +use crate::common::errors::{Error, XrpcResult}; use diesel::sql_types::{Array, Integer, Text}; use diesel_async::pooled_connection::deadpool::Pool; use diesel_async::{AsyncPgConnection, RunQueryDsl}; diff --git a/parakeet/src/common/helpers.rs b/parakeet/src/common/helpers.rs index 8e991328..62fa544f 100644 --- a/parakeet/src/common/helpers.rs +++ b/parakeet/src/common/helpers.rs @@ -2,7 +2,7 @@ /// /// These replace the old hydration/dataloader-based helpers with direct entity access -use crate::xrpc::error; +use crate::common::errors; use crate::entities::ProfileEntity; use diesel::prelude::*; use diesel_async::{AsyncPgConnection, RunQueryDsl}; @@ -12,7 +12,7 @@ use diesel_async::pooled_connection::deadpool::Pool; pub async fn get_actor_did( profile_entity: &ProfileEntity, actor: String, -) -> error::XrpcResult { +) -> errors::XrpcResult { if actor.starts_with("did:") { Ok(actor) } else { @@ -20,12 +20,12 @@ pub async fn get_actor_did( let actor_id = profile_entity .resolve_identifier(&actor) .await - .map_err(|_| error::Error::actor_not_found(&actor))?; + .map_err(|_| errors::Error::actor_not_found(&actor))?; profile_entity .get_did_by_id(actor_id) .await - .map_err(|_| error::Error::actor_not_found(&actor)) + .map_err(|_| errors::Error::actor_not_found(&actor)) } } @@ -46,7 +46,7 @@ pub async fn get_actor_dids( } /// Normalize an AT URI (no-op for now, could add validation) -pub async fn normalise_at_uri(uri: &str) -> error::XrpcResult { +pub async fn normalise_at_uri(uri: &str) -> errors::XrpcResult { // Could add validation here Ok(uri.to_string()) } @@ -56,11 +56,11 @@ pub async fn check_actor_status( pool: &Pool, profile_entity: &ProfileEntity, did: &str, -) -> error::XrpcResult<()> { +) -> errors::XrpcResult<()> { let actor_id = profile_entity .resolve_identifier(did) .await - .map_err(|_| error::Error::actor_not_found(did))?; + .map_err(|_| errors::Error::actor_not_found(did))?; let mut conn = pool.get().await?; @@ -75,7 +75,7 @@ pub async fn check_actor_status( .await?; if !is_active { - return Err(error::Error::actor_not_found(did)); + return Err(errors::Error::actor_not_found(did)); } Ok(()) @@ -85,7 +85,7 @@ pub async fn check_actor_status( pub async fn resolve_did_no_cache( handle: &str, pool: &Pool, -) -> error::XrpcResult { +) -> errors::XrpcResult { let mut conn = pool.get().await?; use parakeet_db::schema::actors; @@ -97,5 +97,99 @@ pub async fn resolve_did_no_cache( .await .ok(); - did.ok_or_else(|| error::Error::actor_not_found(handle)) -} \ No newline at end of file + did.ok_or_else(|| errors::Error::actor_not_found(handle)) +} +// ============================================================================ +// CDN URL Helpers (from cdn.rs) +// ============================================================================ + +/// 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) + } +} + +// ============================================================================ +// Cursor Helpers (from cursor.rs) +// ============================================================================ + +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/common/rate_limiting.rs b/parakeet/src/common/rate_limiting.rs index 0d69ae86..c602ff09 100644 --- a/parakeet/src/common/rate_limiting.rs +++ b/parakeet/src/common/rate_limiting.rs @@ -246,3 +246,190 @@ mod tests { assert_eq!(limiter.entry_count(), 0); } } + + +// ============================================================================ +// Rate Limiting Middleware (from middleware/rate_limit.rs) +// ============================================================================ + +use axum::{ + body::Body, + extract::State, + http::{Request, StatusCode}, + middleware::Next, + response::{IntoResponse, Response}, +}; + +use crate::common::auth::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 middleware_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); + } +} -- 2.51.2