init: parakeet-appview

parakeet-appview-macros/Cargo.toml

···
+[package]
+name = "parakeet-appview-macros"
+version = "0.1.0"
+edition = "2021"
+
+[lib]
+proc-macro = true
+
+[dependencies]
+syn = { version = "2.0", features = ["full", "extra-traits"] }
+quote = "1.0"
+proc-macro2 = "1.0"
+
+[lints.rust]
+warnings = { level = "warn", priority = -1 }
+deprecated-safe = { level = "warn", priority = -1 }

parakeet-appview-macros/src/lib.rs

···
+//! Procedural macros for parakeet-appview.
+//!
+//! This crate provides the `#[xrpc_handler]` attribute macro that generates
+//! XRPC handler boilerplate, working with Jacquard types.
+
+use proc_macro::TokenStream;
+use quote::quote;
+use syn::{parse_macro_input, parse::{Parse, ParseStream}, ItemFn, Token, LitStr};
+
+/// Procedural macro for XRPC handlers that generates boilerplate code.
+///
+/// This macro:
+/// - Generates proper Axum handler signatures
+/// - Handles authentication requirements
+/// - Manages identifier resolution
+/// - Provides pagination support
+/// - Maps errors appropriately
+///
+/// # Attributes
+///
+/// - `auth`: Authentication requirement (required/optional/none)
+/// - `resolve`: Identifier resolution mapping
+/// - `paginate`: Enable pagination with default settings
+/// - `rate_limit`: Rate limiting configuration
+///
+/// # Example
+///
+/// ```rust
+/// #[xrpc_handler(
+///     auth = "optional",
+///     paginate = true
+/// )]
+/// async fn get_profile(req: GetProfileRequest) -> Result<GetProfileOutput, XrpcError> {
+///     // Handler implementation
+/// }
+/// ```
+///
+/// This generates an Axum handler with:
+/// - Optional authentication extraction
+/// - Pagination parameters
+/// - Proper error handling
+/// - Context building
+#[proc_macro_attribute]
+pub fn xrpc_handler(args: TokenStream, input: TokenStream) -> TokenStream {
+    let config = parse_macro_input!(args as HandlerConfig);
+    let input_fn = parse_macro_input!(input as ItemFn);
+
+    // Generate the expanded handler
+    let expanded = generate_handler(&input_fn, &config);
+
+    TokenStream::from(expanded)
+}
+
+/// Configuration parsed from the macro attributes
+struct HandlerConfig {
+    auth: AuthRequirement,
+    paginate: bool,
+    resolve: Vec<ResolveConfig>,
+    rate_limit: Option<String>,
+}
+
+#[derive(Clone)]
+enum AuthRequirement {
+    Required,
+    Optional,
+    None,
+}
+
+struct ResolveConfig {
+    field: syn::Ident,
+    target: syn::Ident,
+}
+
+impl Parse for HandlerConfig {
+    fn parse(input: ParseStream) -> syn::Result<Self> {
+        let mut auth = AuthRequirement::None;
+        let mut paginate = false;
+        let mut resolve = Vec::new();
+        let mut rate_limit = None;
+
+        while !input.is_empty() {
+            let ident: syn::Ident = input.parse()?;
+            input.parse::<Token![=]>()?;
+
+            match ident.to_string().as_str() {
+                "auth" => {
+                    let value: LitStr = input.parse()?;
+                    auth = match value.value().as_str() {
+                        "required" => AuthRequirement::Required,
+                        "optional" => AuthRequirement::Optional,
+                        "none" => AuthRequirement::None,
+                        _ => {
+                            return Err(syn::Error::new_spanned(
+                                value,
+                                "auth must be 'required', 'optional', or 'none'",
+                            ))
+                        }
+                    };
+                }
+                "paginate" => {
+                    let value: syn::LitBool = input.parse()?;
+                    paginate = value.value();
+                }
+                "resolve" => {
+                    // Parse resolve mapping like: req.actor -> actor_id
+                    let content;
+                    syn::braced!(content in input);
+
+                    let field_path: syn::ExprField = content.parse()?;
+                    content.parse::<Token![->]>()?;
+                    let target: syn::Ident = content.parse()?;
+
+                    if let syn::Expr::Field(field) = field_path.base.as_ref() {
+                        if let syn::Member::Named(field_name) = &field.member {
+                            resolve.push(ResolveConfig {
+                                field: field_name.clone(),
+                                target,
+                            });
+                        }
+                    }
+                }
+                "rate_limit" => {
+                    let value: LitStr = input.parse()?;
+                    rate_limit = Some(value.value());
+                }
+                _ => {
+                    return Err(syn::Error::new_spanned(
+                        &ident,
+                        format!("Unknown attribute: {}", ident),
+                    ))
+                }
+            }
+
+            if !input.is_empty() {
+                input.parse::<Token![,]>()?;
+            }
+        }
+
+        Ok(HandlerConfig {
+            auth,
+            paginate,
+            resolve,
+            rate_limit,
+        })
+    }
+}
+
+fn generate_handler(input_fn: &ItemFn, config: &HandlerConfig) -> proc_macro2::TokenStream {
+    let fn_name = &input_fn.sig.ident;
+    let fn_body = &input_fn.block;
+    let vis = &input_fn.vis;
+    let _asyncness = &input_fn.sig.asyncness;
+
+    // Extract request and response types from the function signature
+    let (req_type, resp_type) = extract_types(&input_fn.sig);
+
+    // Generate auth parameter based on requirement
+    let auth_param = match config.auth {
+        AuthRequirement::Required => quote! {
+            auth: crate::xrpc::AtpAuth,
+        },
+        AuthRequirement::Optional => quote! {
+            maybe_auth: ::core::option::Option<crate::xrpc::AtpAuth>,
+        },
+        AuthRequirement::None => quote! {},
+    };
+
+    // Generate pagination parameter if enabled
+    let pagination_param = if config.paginate {
+        quote! {
+            pagination: ::axum::extract::Query<crate::xrpc::PaginationParams>,
+        }
+    } else {
+        quote! {}
+    };
+
+    // Generate context building based on auth
+    let build_context = match config.auth {
+        AuthRequirement::Required => quote! {
+            let ctx = crate::xrpc::XrpcContextBuilder::new(state.clone())
+                .auth(auth.0.clone(), None)
+                .build();
+        },
+        AuthRequirement::Optional => quote! {
+            let ctx = match maybe_auth {
+                Some(auth) => crate::xrpc::XrpcContextBuilder::new(state.clone())
+                    .auth(auth.0.clone(), None)
+                    .build(),
+                None => crate::xrpc::XrpcContext {
+                    state: state.clone(),
+                    auth_did: None,
+                    viewer_id: None,
+                    labelers: labelers.0.clone(),
+                },
+            };
+        },
+        AuthRequirement::None => quote! {
+            let ctx = crate::xrpc::XrpcContext {
+                state: state.clone(),
+                auth_did: None,
+                viewer_id: None,
+                labelers: labelers.0.clone(),
+            };
+        },
+    };
+
+    // Generate resolver calls for identifier resolution
+    let resolver_calls = config.resolve.iter().map(|r| {
+        let field = &r.field;
+        let target = &r.target;
+        quote! {
+            let #target = if let Some(identifier) = req.#field.as_ref() {
+                ctx.state.profile_entity
+                    .resolve_identifier(identifier)
+                    .await
+                    .map_err(|e| crate::xrpc::XrpcError::BadRequest(e.to_string()))?
+                    .ok_or_else(|| crate::xrpc::XrpcError::ActorNotFound(
+                        format!("Actor not found: {}", identifier)
+                    ))?
+            } else {
+                return Err(crate::xrpc::XrpcError::BadRequest(
+                    format!("Missing required field: {}", stringify!(#field))
+                ));
+            };
+        }
+    });
+
+    // Handle pagination extraction
+    let pagination_extract = if config.paginate {
+        quote! {
+            let pagination_params = pagination.0;
+        }
+    } else {
+        quote! {}
+    };
+
+
+    // Generate rate limit check if configured
+    let rate_limit_check = if let Some(limit_config) = &config.rate_limit {
+        match config.auth {
+            AuthRequirement::Required => quote! {
+                // Rate limit by authenticated DID
+                let rate_limit_key = auth.0.clone();
+                if !state.rate_limiter.check(&rate_limit_key, Some(#limit_config)) {
+                    return Err(crate::xrpc::XrpcError::RateLimitExceeded);
+                }
+            },
+            AuthRequirement::Optional => quote! {
+                // Rate limit by DID if authenticated, otherwise by a default key
+                let rate_limit_key = maybe_auth
+                    .as_ref()
+                    .map(|a| a.0.clone())
+                    .unwrap_or_else(|| "anonymous".to_string());
+                if !state.rate_limiter.check(&rate_limit_key, Some(#limit_config)) {
+                    return Err(crate::xrpc::XrpcError::RateLimitExceeded);
+                }
+            },
+            AuthRequirement::None => quote! {
+                // Rate limit by a default key for unauthenticated endpoints
+                if !state.rate_limiter.check("anonymous", Some(#limit_config)) {
+                    return Err(crate::xrpc::XrpcError::RateLimitExceeded);
+                }
+            },
+        }
+    } else {
+        quote! {}
+    };
+
+    // Generate the expanded handler
+    quote! {
+        // Public wrapper function with Axum signature
+        #vis async fn #fn_name(
+            ::axum::extract::State(state): ::axum::extract::State<::std::sync::Arc<crate::xrpc::GlobalState>>,
+            #pagination_param
+            labelers: crate::xrpc::AtpAcceptLabelers,
+            #auth_param
+            crate::xrpc::ExtractXrpc(req): crate::xrpc::ExtractXrpc<#req_type>,
+        ) -> crate::xrpc::XrpcResult<::axum::Json<#resp_type>> {
+            #rate_limit_check
+            #build_context
+            #pagination_extract
+            #(#resolver_calls)*
+
+            // Original handler body with injected context
+            // The body is expected to return Result<Response, XrpcError>
+            let result_res: Result<#resp_type, crate::xrpc::XrpcError> = {
+                #fn_body
+            };
+            let result = result_res?;
+
+            Ok(::axum::Json(result))
+        }
+    }
+}
+
+fn extract_types(sig: &syn::Signature) -> (proc_macro2::TokenStream, proc_macro2::TokenStream) {
+    // Extract request type from first parameter
+    let req_type = if let Some(arg) = sig.inputs.first() {
+        if let syn::FnArg::Typed(pat_type) = arg {
+            let ty = &pat_type.ty;
+            quote! { #ty }
+        } else {
+            // Default to unit type if no request parameter
+            quote! { () }
+        }
+    } else {
+        quote! { () }
+    };
+
+    // Extract response type from return type
+    let resp_type = match &sig.output {
+        syn::ReturnType::Type(_, ty) => {
+            // Handle Result<T, E> by extracting T
+            if let syn::Type::Path(type_path) = ty.as_ref() {
+                if let Some(segment) = type_path.path.segments.last() {
+                    if segment.ident == "Result" {
+                        if let syn::PathArguments::AngleBracketed(args) = &segment.arguments {
+                            if let Some(syn::GenericArgument::Type(inner_ty)) = args.args.first() {
+                                return (req_type, quote! { #inner_ty });
+                            }
+                        }
+                    }
+                }
+            }
+            quote! { #ty }
+        }
+        syn::ReturnType::Default => quote! { () },
+    };
+
+    (req_type, resp_type)
+}

parakeet-appview/Cargo.toml

···
+[package]
+name = "parakeet-appview"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+# Jacquard dependencies for AT Protocol types
+jacquard = { workspace = true }
+jacquard-api = { workspace = true }
+jacquard-axum = { workspace = true }
+jacquard-common = { workspace = true }
+
+# Web framework
+axum = { version = "0.8", features = ["json", "macros"] }
+axum-extra = { version = "0.10", features = ["query", "typed-header"] }
+
+# Database
+diesel = { version = "2.2", features = ["chrono", "serde_json"] }
+diesel-async = { version = "0.5", features = ["deadpool", "postgres"] }
+deadpool = "0.12"
+
+# Caching
+moka = { version = "0.12", features = ["future"] }
+dashmap = "6.0"
+
+# Async runtime
+tokio = { version = "1.42", features = ["full"] }
+futures = "0.3"
+async-trait = "0.1"
+
+# Serialization
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+serde_urlencoded = "0.7"
+
+# Error handling
+eyre = "0.6"
+color-eyre = "0.6"
+thiserror = "2.0"
+
+# Utilities
+chrono = { version = "0.4", features = ["serde"] }
+tracing = "0.1"
+itertools = "0.14"
+reqwest = { version = "0.12", features = ["json"] }
+
+# Internal dependencies
+parakeet-db = { path = "../parakeet-db" }
+parakeet-appview-macros = { path = "../parakeet-appview-macros", optional = true }
+
+[features]
+default = ["macros"]
+macros = ["dep:parakeet-appview-macros"]
+
+[dev-dependencies]
+tokio-test = "0.4"
+
+[lints.rust]
+warnings = { level = "warn", priority = -1 }
+deprecated-safe = { level = "warn", priority = -1 }

parakeet-appview/src/entity/implementations.rs

···
+//! Entity implementations for parakeet-db models.
+
+use std::time::Duration;
+
+use async_trait::async_trait;
+use diesel_async::AsyncPgConnection;
+use eyre::Result;
+
+use parakeet_db::models::{Actor, Post};
+
+use super::traits::{CachedEntity, ResolvableEntity, InvalidatableEntity};
+
+/// Profile entity for actor caching.
+pub struct ProfileEntity;
+
+#[async_trait]
+impl CachedEntity for ProfileEntity {
+    type Id = i32;
+    type Model = Actor;
+
+    async fn fetch_one(
+        &self,
+        _conn: &mut AsyncPgConnection,
+        _id: &Self::Id,
+    ) -> Result<Option<Self::Model>> {
+        // TODO: Implement
+        Ok(None)
+    }
+
+    async fn fetch_batch(
+        &self,
+        _conn: &mut AsyncPgConnection,
+        _ids: &[Self::Id],
+    ) -> Result<Vec<(Self::Id, Self::Model)>> {
+        // TODO: Implement
+        Ok(Vec::new())
+    }
+
+    fn cache_ttl(&self) -> Duration {
+        Duration::from_secs(3600) // 1 hour
+    }
+
+    fn cache_key(&self, id: &Self::Id) -> String {
+        format!("profile:{}", id)
+    }
+}
+
+#[async_trait]
+impl ResolvableEntity for ProfileEntity {
+    async fn resolve_identifier(
+        &self,
+        _conn: &mut AsyncPgConnection,
+        _identifier: &str,
+    ) -> Result<Option<Self::Id>> {
+        // TODO: Implement DID/handle resolution
+        Ok(None)
+    }
+}
+
+#[async_trait]
+impl InvalidatableEntity for ProfileEntity {
+    async fn invalidate(&self, _payload: &str) -> Result<()> {
+        // Payload format: "profile:123"
+        Ok(())
+    }
+
+    fn parse_invalidation(&self, payload: &str) -> Option<Self::Id> {
+        payload
+            .strip_prefix("profile:")
+            .and_then(|id_str| id_str.parse::<i32>().ok())
+    }
+}
+
+/// Post entity for caching posts.
+pub struct PostEntity;
+
+#[async_trait]
+impl CachedEntity for PostEntity {
+    type Id = (i32, i64); // (actor_id, rkey)
+    type Model = Post;
+
+    async fn fetch_one(
+        &self,
+        _conn: &mut AsyncPgConnection,
+        _id: &Self::Id,
+    ) -> Result<Option<Self::Model>> {
+        // TODO: Implement
+        Ok(None)
+    }
+
+    async fn fetch_batch(
+        &self,
+        _conn: &mut AsyncPgConnection,
+        _ids: &[Self::Id],
+    ) -> Result<Vec<(Self::Id, Self::Model)>> {
+        // TODO: Implement
+        Ok(Vec::new())
+    }
+
+    fn cache_ttl(&self) -> Duration {
+        Duration::from_secs(1800) // 30 minutes
+    }
+
+    fn cache_key(&self, id: &Self::Id) -> String {
+        format!("post:{}:{}", id.0, id.1)
+    }
+}
+
+#[async_trait]
+impl InvalidatableEntity for PostEntity {
+    async fn invalidate(&self, _payload: &str) -> Result<()> {
+        // Payload format: "post:123:456789"
+        Ok(())
+    }
+
+    fn parse_invalidation(&self, payload: &str) -> Option<Self::Id> {
+        let parts: Vec<&str> = payload.strip_prefix("post:")?.split(':').collect();
+        if parts.len() == 2 {
+            let actor_id = parts[0].parse::<i32>().ok()?;
+            let rkey = parts[1].parse::<i64>().ok()?;
+            Some((actor_id, rkey))
+        } else {
+            None
+        }
+    }
+}

parakeet-appview/src/entity/store.rs

···
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+use std::time::Duration;
+
+use deadpool::managed::Pool;
+use diesel_async::AsyncPgConnection;
+use diesel_async::pooled_connection::AsyncDieselConnectionManager;
+use eyre::Result;
+use moka::future::Cache;
+use tracing::{debug, trace, warn};
+
+use super::traits::{BatchResult, CachedEntity, EntityConfig, InvalidatableEntity, ResolvableEntity};
+
+/// Generic entity store that provides caching and batching for any entity type.
+///
+/// This store handles:
+/// - Multi-level caching (entity cache + optional identifier resolution cache)
+/// - Batch fetching with deduplication
+/// - Cache invalidation
+/// - Automatic TTL management via Moka
+pub struct EntityStore<E: CachedEntity> {
+    /// The entity implementation
+    entity: Arc<E>,
+
+    /// Main cache for entities (ID -> Model)
+    cache: Cache<E::Id, E::Model>,
+
+    /// Optional cache for identifier resolution (e.g., DID -> ID)
+    identifier_cache: Option<Cache<String, E::Id>>,
+
+    /// Database connection pool
+    db_pool: Arc<Pool<AsyncDieselConnectionManager<AsyncPgConnection>>>,
+
+    /// Configuration
+    config: EntityConfig,
+}
+
+impl<E: CachedEntity> EntityStore<E> {
+    /// Create a new entity store
+    pub fn new(
+        entity: E,
+        db_pool: Arc<Pool<AsyncDieselConnectionManager<AsyncPgConnection>>>,
+        config: EntityConfig,
+    ) -> Self {
+        // Build the main entity cache with Moka
+        let mut cache_builder = Cache::builder()
+            .max_capacity(config.max_capacity)
+            .time_to_live(config.cache_ttl);
+
+        if let Some(idle) = config.idle_timeout {
+            cache_builder = cache_builder.time_to_idle(idle);
+        }
+
+        let cache = cache_builder.build();
+
+        Self {
+            entity: Arc::new(entity),
+            cache,
+            identifier_cache: None,
+            db_pool,
+            config,
+        }
+    }
+
+    /// Enable identifier resolution caching
+    pub fn with_identifier_cache(mut self, ttl: Duration) -> Self
+    where
+        E: ResolvableEntity,
+    {
+        self.identifier_cache = Some(
+            Cache::builder()
+                .max_capacity(self.config.max_capacity * 2) // More identifiers than entities
+                .time_to_live(ttl)
+                .build(),
+        );
+        self
+    }
+
+    /// Get a single entity by ID
+    pub async fn get(&self, id: &E::Id) -> Result<Option<E::Model>> {
+        // Check cache first
+        if let Some(cached) = self.cache.get(id).await {
+            trace!("Cache hit for entity {:?}", self.entity.cache_key(id));
+            return Ok(Some(cached));
+        }
+
+        // Cache miss - fetch from database
+        trace!("Cache miss for entity {:?}", self.entity.cache_key(id));
+
+        let mut conn = self.db_pool.get().await?;
+        let result = self.entity.fetch_one(&mut conn, id).await?;
+
+        // Cache the result if found
+        if let Some(ref model) = result {
+            self.cache.insert(id.clone(), model.clone()).await;
+            debug!("Cached entity {:?}", self.entity.cache_key(id));
+        }
+
+        Ok(result)
+    }
+
+    /// Get multiple entities by IDs with batching and caching
+    pub async fn get_many(&self, ids: &[E::Id]) -> Result<BatchResult<E::Id, E::Model>> {
+        if ids.is_empty() {
+            return Ok(BatchResult::new(vec![], vec![]));
+        }
+
+        let mut results = Vec::with_capacity(ids.len());
+        let mut missing_ids = Vec::new();
+        let mut ids_to_fetch = HashSet::new();
+
+        // First pass: check cache
+        for id in ids {
+            if let Some(cached) = self.cache.get(id).await {
+                results.push((id.clone(), cached));
+            } else {
+                ids_to_fetch.insert(id.clone());
+            }
+        }
+
+        // Fetch missing items from database
+        if !ids_to_fetch.is_empty() {
+            debug!("Fetching {} entities from database", ids_to_fetch.len());
+
+            let fetch_vec: Vec<E::Id> = ids_to_fetch.iter().cloned().collect();
+            let mut conn = self.db_pool.get().await?;
+            let fetched = self.entity.fetch_batch(&mut conn, &fetch_vec).await?;
+
+            // Cache fetched items and add to results
+            let mut fetched_map = HashMap::new();
+            for (id, model) in fetched {
+                self.cache.insert(id.clone(), model.clone()).await;
+                fetched_map.insert(id.clone(), model);
+            }
+
+            // Add fetched items to results, track missing
+            for id in ids_to_fetch {
+                if let Some(model) = fetched_map.remove(&id) {
+                    results.push((id, model));
+                } else {
+                    missing_ids.push(id);
+                }
+            }
+        }
+
+        // Preserve original order
+        let result_map: HashMap<E::Id, E::Model> = results.into_iter().collect();
+        let mut ordered_results = Vec::new();
+        let mut ordered_missing = Vec::new();
+
+        for id in ids {
+            if let Some(model) = result_map.get(id) {
+                ordered_results.push((id.clone(), model.clone()));
+            } else if !result_map.contains_key(id) {
+                ordered_missing.push(id.clone());
+            }
+        }
+
+        Ok(BatchResult::new(ordered_results, ordered_missing))
+    }
+
+    // Note: Jacquard view conversion is now handled directly by parakeet-db models
+    // Models have methods like to_profile_view(), to_post_view(), etc.
+
+    /// Clear all cached entries
+    pub async fn clear_cache(&self) {
+        self.cache.invalidate_all();
+        self.cache.run_pending_tasks().await;
+
+        if let Some(ref id_cache) = self.identifier_cache {
+            id_cache.invalidate_all();
+            id_cache.run_pending_tasks().await;
+        }
+    }
+
+    /// Get cache statistics
+    pub fn cache_stats(&self) -> CacheStats {
+        CacheStats {
+            entry_count: self.cache.entry_count(),
+            weighted_size: self.cache.weighted_size(),
+        }
+    }
+}
+
+/// Extended methods for entities that support identifier resolution
+impl<E> EntityStore<E>
+where
+    E: CachedEntity + ResolvableEntity,
+{
+    /// Resolve an identifier to an ID
+    pub async fn resolve_identifier(&self, identifier: &str) -> Result<Option<E::Id>> {
+        // Check identifier cache if enabled
+        if let Some(ref id_cache) = self.identifier_cache {
+            if let Some(cached_id) = id_cache.get(identifier).await {
+                trace!("Identifier cache hit for {}", identifier);
+                return Ok(Some(cached_id));
+            }
+        }
+
+        // Resolve from database
+        let mut conn = self.db_pool.get().await?;
+        let result = self.entity.resolve_identifier(&mut conn, identifier).await?;
+
+        // Cache the result if found
+        if let (Some(ref id_cache), Some(ref id)) = (&self.identifier_cache, &result) {
+            id_cache.insert(identifier.to_string(), id.clone()).await;
+            debug!("Cached identifier mapping {} -> {:?}", identifier, self.entity.cache_key(id));
+        }
+
+        Ok(result)
+    }
+
+    /// Resolve and get an entity by identifier
+    pub async fn get_by_identifier(&self, identifier: &str) -> Result<Option<E::Model>> {
+        let id = self.resolve_identifier(identifier).await?;
+        match id {
+            Some(id) => self.get(&id).await,
+            None => Ok(None),
+        }
+    }
+}
+
+/// Extended methods for entities that support invalidation
+impl<E> EntityStore<E>
+where
+    E: CachedEntity + InvalidatableEntity,
+{
+    /// Invalidate cache based on a notification payload
+    pub async fn invalidate(&self, payload: &str) -> Result<()> {
+        if let Some(id) = self.entity.parse_invalidation(payload) {
+            self.cache.invalidate(&id).await;
+            debug!("Invalidated cache for {:?}", self.entity.cache_key(&id));
+        } else {
+            warn!("Could not parse invalidation payload: {}", payload);
+        }
+        Ok(())
+    }
+
+    /// Invalidate a specific entity by ID
+    pub async fn invalidate_by_id(&self, id: &E::Id) -> Result<()> {
+        self.cache.invalidate(id).await;
+        debug!("Invalidated cache for {:?}", self.entity.cache_key(id));
+        Ok(())
+    }
+}
+
+/// Cache statistics
+#[derive(Debug, Clone)]
+pub struct CacheStats {
+    /// Number of entries currently in the cache
+    pub entry_count: u64,
+
+    /// Total weighted size of entries
+    pub weighted_size: u64,
+}
+
+#[cfg(test)]
+mod tests {
+    // TODO: Add comprehensive tests for EntityStore
+}

parakeet-appview/src/entity/traits.rs

···
+use std::hash::Hash;
+use std::time::Duration;
+
+use async_trait::async_trait;
+use diesel_async::AsyncPgConnection;
+use eyre::Result;
+
+/// Core trait for entities that can be cached.
+///
+/// This trait provides the foundation for all entity operations, including:
+/// - Database fetching (single and batch)
+/// - Cache management with TTL
+#[async_trait]
+pub trait CachedEntity: Send + Sync + 'static {
+    /// The type used as the unique identifier for this entity
+    type Id: Hash + Eq + Clone + Send + Sync + 'static;
+
+    /// The database model type (from parakeet-db)
+    /// Models should have methods like to_profile_view(), to_post_view(), etc.
+    type Model: Clone + Send + Sync + 'static;
+
+    /// Fetch a single entity from the database by ID
+    async fn fetch_one(
+        &self,
+        conn: &mut AsyncPgConnection,
+        id: &Self::Id,
+    ) -> Result<Option<Self::Model>>;
+
+    /// Fetch multiple entities from the database in a single query
+    async fn fetch_batch(
+        &self,
+        conn: &mut AsyncPgConnection,
+        ids: &[Self::Id],
+    ) -> Result<Vec<(Self::Id, Self::Model)>>;
+
+    /// Get the cache TTL for this entity type
+    fn cache_ttl(&self) -> Duration;
+
+    /// Generate a cache key for the given ID
+    fn cache_key(&self, id: &Self::Id) -> String;
+
+    /// Maximum number of items to keep in cache
+    fn cache_max_capacity(&self) -> u64 {
+        10_000 // Default, can be overridden
+    }
+}
+
+/// Extended trait for entities that support identifier resolution
+/// (e.g., resolving DIDs or handles to internal IDs)
+#[async_trait]
+pub trait ResolvableEntity: CachedEntity {
+    /// Resolve an external identifier (DID, handle, URI) to an internal ID
+    async fn resolve_identifier(
+        &self,
+        conn: &mut AsyncPgConnection,
+        identifier: &str,
+    ) -> Result<Option<Self::Id>>;
+
+    /// Get the cache TTL for identifier mappings (usually longer than entity TTL)
+    fn identifier_cache_ttl(&self) -> Duration {
+        Duration::from_secs(86400) // 24 hours default
+    }
+}
+
+/// Trait for entities that support invalidation via PostgreSQL NOTIFY
+#[async_trait]
+pub trait InvalidatableEntity: CachedEntity {
+    /// Invalidate cache entries based on a notification payload
+    async fn invalidate(&self, payload: &str) -> Result<()>;
+
+    /// Parse invalidation message to extract the ID
+    fn parse_invalidation(&self, payload: &str) -> Option<Self::Id>;
+}
+
+/// Configuration for entity caching behavior
+#[derive(Debug, Clone)]
+pub struct EntityConfig {
+    /// Time-to-live for cached entities
+    pub cache_ttl: Duration,
+
+    /// Maximum number of items in cache
+    pub max_capacity: u64,
+
+    /// Time-to-idle for cache entries
+    pub idle_timeout: Option<Duration>,
+
+    /// Whether to use write-through caching
+    pub write_through: bool,
+}
+
+impl Default for EntityConfig {
+    fn default() -> Self {
+        Self {
+            cache_ttl: Duration::from_secs(3600), // 1 hour
+            max_capacity: 10_000,
+            idle_timeout: Some(Duration::from_secs(1800)), // 30 minutes
+            write_through: true,
+        }
+    }
+}
+
+/// Batch fetch result with ordering preserved
+pub struct BatchResult<Id, Model> {
+    /// The fetched items with their IDs
+    pub items: Vec<(Id, Model)>,
+
+    /// IDs that were requested but not found
+    pub missing: Vec<Id>,
+}
+
+impl<Id: Clone, Model> BatchResult<Id, Model> {
+    /// Create a new batch result
+    pub fn new(items: Vec<(Id, Model)>, missing: Vec<Id>) -> Self {
+        Self { items, missing }
+    }
+
+    /// Convert to a map for easy lookups
+    pub fn into_map(self) -> std::collections::HashMap<Id, Model>
+    where
+        Id: Hash + Eq,
+    {
+        self.items.into_iter().collect()
+    }
+}

parakeet-appview/src/handlers/actor.rs

···
+//! Actor/Profile XRPC handlers.
+
+use std::sync::Arc;
+use axum::{extract::State, Json};
+use crate::xrpc::{AtpAcceptLabelers, AtpAuth, GlobalState, XrpcResult};
+
+/// Example handler showing the pattern.
+pub async fn example_get_profile(
+    State(_state): State<Arc<GlobalState>>,
+    _labelers: AtpAcceptLabelers,
+    _maybe_auth: Option<AtpAuth>,
+) -> XrpcResult<Json<String>> {
+    // TODO: Implement actual profile fetching
+    Ok(Json("Profile endpoint placeholder".to_string()))
+}

parakeet-appview/src/lib.rs

···
+//! AppView implementation for AT Protocol.
+
+pub mod entity {
+    //! Entity abstraction layer for caching and data access.
+    //!
+    //! This module provides a generic entity system that works with Jacquard AT Protocol types,
+    //! handling caching, batch fetching, and invalidation for all entity types.
+
+    mod implementations;
+    mod store;
+    mod traits;
+
+    pub use implementations::{PostEntity, ProfileEntity};
+    pub use moka::future::Cache;
+    pub use store::{CacheStats, EntityStore};
+    pub use traits::{
+        BatchResult, CachedEntity, EntityConfig, InvalidatableEntity, ResolvableEntity,
+    };
+
+    /// Prelude for entity implementations
+    pub mod prelude {
+        pub use super::{
+            BatchResult, CachedEntity, EntityConfig, EntityStore, InvalidatableEntity,
+            ResolvableEntity,
+        };
+    }
+}
+
+pub mod macros;
+
+pub mod xrpc {
+    //! XRPC handler utilities.
+    mod auth;
+    mod context;
+    mod error;
+    mod helpers;
+    mod rate_limiter;
+
+    pub use auth::{AtpAcceptLabelers, AtpAuth, AuthError, ExtractXrpc};
+    pub use context::{GlobalState, XrpcContext, XrpcContextBuilder};
+    pub use error::{IntoXrpcError, XrpcError, XrpcResult};
+    pub use helpers::{
+        build_at_uri, decode_compound_cursor, encode_compound_cursor, encode_int_cursor,
+        encode_timestamp_cursor, is_did, is_handle, normalize_handle, paginate, parse_at_uri,
+        preserve_order, preserve_order_optional, PaginatedResponse, PaginationParams,
+    };
+    pub use rate_limiter::{RateLimiter, RateLimitConfig};
+
+    /// Prelude for handler implementations.
+    pub mod prelude {
+        pub use super::{
+            AtpAcceptLabelers, AtpAuth, ExtractXrpc, GlobalState, PaginatedResponse,
+            PaginationParams, XrpcContext, XrpcError, XrpcResult,
+        };
+
+        pub use jacquard_api::app_bsky;
+        pub use jacquard_axum::IntoRouter;
+        pub use jacquard_common::types::{
+            did::Did,
+            handle::Handle,
+            string::{AtUri, Datetime},
+        };
+    }
+}
+
+pub mod handlers {
+    //! Handler implementations.
+    pub mod actor;
+}
+
+#[cfg(feature = "macros")]
+pub use parakeet_appview_macros::xrpc_handler;
+
+pub use entity::{CachedEntity, EntityStore, InvalidatableEntity, ResolvableEntity};
+pub use xrpc::{GlobalState, XrpcContext, XrpcError, XrpcResult};
+
+/// Prelude module for convenient imports.
+pub mod prelude {
+    pub use crate::batch_fetch;
+    pub use crate::define_entity;
+    pub use crate::entity::prelude::*;
+    pub use crate::xrpc::prelude::*;
+    #[cfg(feature = "macros")]
+    pub use crate::xrpc_handler;
+
+    pub use axum::{extract::State, Json};
+    pub use eyre::Result;
+}

parakeet-appview/src/macros.rs

···
+//! Declarative macros for entity definitions and handlers.
+
+/// Define a cached entity with automatic implementation of common patterns.
+///
+/// This macro generates:
+/// - Entity struct implementation
+/// - CachedEntity trait implementation
+/// - Optional ResolvableEntity implementation for identifier resolution
+/// - Helper methods for common operations
+///
+/// # Example
+///
+/// ```rust
+/// define_entity! {
+///     ProfileEntity {
+///         // Basic entity configuration
+///         id: i32,
+///         model: parakeet_db::models::Actor,
+///         table: parakeet_db::schema::actors,
+///         ttl: 3600,  // 1 hour in seconds
+///
+///         // Optional: Define indexes for resolution
+///         indexes: {
+///             did: String => did_to_id,
+///             handle: String => handle_to_id,
+///         },
+///
+///         // Optional: Custom fetch query
+///         fetch_one: |conn, id| {
+///             use diesel::prelude::*;
+///             actors::table
+///                 .filter(actors::id.eq(id))
+///                 .filter(actors::status.eq(ActorStatus::Active))
+///                 .first(conn)
+///         },
+///     }
+/// }
+/// ```
+#[macro_export]
+macro_rules! define_entity {
+    (
+        $name:ident {
+            id: $id_type:ty,
+            model: $model:ty,
+            table: $table:path,
+            ttl: $ttl:expr,
+
+            $(
+                // Optional indexes for resolution
+                indexes: {
+                    $($index_name:ident : $index_type:ty => $index_field:ident),* $(,)?
+                }$(,)?
+            )?
+
+            $(
+                // Optional custom fetch implementation
+                fetch_one: |$conn_var:ident, $id_var:ident| $fetch_expr:expr
+            )?$(,)?
+        }
+    ) => {
+        pub struct $name;
+
+        #[async_trait::async_trait]
+        impl $crate::entity::CachedEntity for $name {
+            type Id = $id_type;
+            type Model = $model;
+
+            async fn fetch_one(
+                &self,
+                conn: &mut diesel_async::AsyncPgConnection,
+                id: &Self::Id,
+            ) -> ::eyre::Result<Option<Self::Model>> {
+                use diesel::prelude::*;
+                use diesel_async::RunQueryDsl;
+
+                $(
+                    // Use custom fetch if provided
+                    let $conn_var = conn;
+                    let $id_var = id;
+                    return Ok($fetch_expr.await.optional()?);
+                )?
+
+                // Default fetch implementation
+                #[allow(unreachable_code)]
+                {
+                    Ok($table::table
+                        .find(id.clone())
+                        .first::<Self::Model>(conn)
+                        .await
+                        .optional()?)
+                }
+            }
+
+            async fn fetch_batch(
+                &self,
+                conn: &mut diesel_async::AsyncPgConnection,
+                ids: &[Self::Id],
+            ) -> ::eyre::Result<Vec<(Self::Id, Self::Model)>> {
+                use diesel::prelude::*;
+                use diesel_async::RunQueryDsl;
+
+                // Default batch fetch implementation
+                let models: Vec<Self::Model> = $table::table
+                    .filter($table::id.eq_any(ids))
+                    .load(conn)
+                    .await?;
+
+                // Map models back to their IDs (assuming model has id field)
+                Ok(models.into_iter()
+                    .map(|model| {
+                        let id = model.id.clone();
+                        (id, model)
+                    })
+                    .collect())
+            }
+
+
+            fn cache_ttl(&self) -> ::std::time::Duration {
+                ::std::time::Duration::from_secs($ttl)
+            }
+
+            fn cache_key(&self, id: &Self::Id) -> String {
+                format!("{}:{:?}", stringify!($name), id)
+            }
+        }
+
+        // Generate index methods if indexes are defined
+        $(
+            impl $name {
+                $(
+                    pub async fn $index_name(
+                        &self,
+                        conn: &mut diesel_async::AsyncPgConnection,
+                        value: &$index_type,
+                    ) -> ::eyre::Result<Option<$id_type>> {
+                        use diesel::prelude::*;
+                        use diesel_async::RunQueryDsl;
+
+                        Ok($table::table
+                            .filter($table::$index_field.eq(value))
+                            .select($table::id)
+                            .first::<$id_type>(conn)
+                            .await
+                            .optional()?)
+                    }
+                )*
+            }
+
+            // Implement ResolvableEntity if indexes are defined
+            #[async_trait::async_trait]
+            impl $crate::entity::ResolvableEntity for $name {
+                async fn resolve_identifier(
+                    &self,
+                    conn: &mut diesel_async::AsyncPgConnection,
+                    identifier: &str,
+                ) -> ::eyre::Result<Option<Self::Id>> {
+                    // Try each index method to resolve the identifier
+                    $(
+                        // Check if this index type can parse the identifier
+                        if let Ok(value) = identifier.parse::<$index_type>() {
+                            if let Some(id) = self.$index_name(conn, &value).await? {
+                                return Ok(Some(id));
+                            }
+                        }
+                    )*
+
+                    Ok(None)
+                }
+            }
+        )?
+    };
+}
+
+/// Helper macro for batch operations on entities.
+/// Macro for batch operations on entities.
+///
+/// Simplifies the pattern of fetching multiple entities and converting them to views.
+#[macro_export]
+macro_rules! batch_fetch {
+    (
+        $entity:expr,
+        ids: $ids:expr,
+        viewer: $viewer:expr
+        $(, preserve_order: $preserve:literal)?
+    ) => {{
+        let batch_result = $entity.get_many($ids).await?;
+        let views = $entity.to_jacquard_views(&batch_result.items, $viewer).await?;
+
+        $(
+            if $preserve {
+                let view_map: ::std::collections::HashMap<_, _> = batch_result.items
+                    .into_iter()
+                    .zip(views)
+                    .collect();
+
+                let ordered = $ids.iter()
+                    .filter_map(|id| view_map.get(id).cloned())
+                    .collect();
+
+                ordered
+            } else {
+                views
+            }
+        )?
+
+        #[allow(unreachable_code)]
+        views
+    }};
+}

parakeet-appview/src/xrpc/auth.rs

···
+//! Authentication types for XRPC handlers.
+
+use axum::{
+    extract::FromRequestParts,
+    http::{header, request::Parts, StatusCode},
+    response::{IntoResponse, Response},
+};
+use serde::Deserialize;
+
+/// Authentication token containing the authenticated user's DID.
+#[derive(Debug, Clone)]
+pub struct AtpAuth(pub String);
+
+impl<S> FromRequestParts<S> for AtpAuth
+where
+    S: Send + Sync,
+{
+    type Rejection = AuthError;
+
+    async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
+        let auth_header = parts
+            .headers
+            .get(header::AUTHORIZATION)
+            .and_then(|h| h.to_str().ok())
+            .ok_or(AuthError::MissingToken)?;
+
+        let token = auth_header
+            .strip_prefix("Bearer ")
+            .ok_or(AuthError::InvalidToken)?;
+
+        // TODO: Validate JWT and extract DID
+        Ok(AtpAuth(token.to_string()))
+    }
+}
+
+/// List of labeler DIDs accepted by the client.
+#[derive(Debug, Clone, Default)]
+pub struct AtpAcceptLabelers(pub Vec<String>);
+
+impl<S> FromRequestParts<S> for AtpAcceptLabelers
+where
+    S: Send + Sync,
+{
+    type Rejection = std::convert::Infallible;
+
+    async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
+        let labelers = parts
+            .headers
+            .get("atproto-accept-labelers")
+            .and_then(|h| h.to_str().ok())
+            .map(|s| s.split(',').map(|s| s.trim().to_string()).collect())
+            .unwrap_or_default();
+
+        Ok(AtpAcceptLabelers(labelers))
+    }
+}
+
+/// Authentication error.
+#[derive(Debug)]
+pub enum AuthError {
+    /// No Authorization header was provided
+    MissingToken,
+    /// The Authorization header was malformed
+    InvalidToken,
+    /// The token failed validation
+    ValidationFailed(String),
+}
+
+impl IntoResponse for AuthError {
+    fn into_response(self) -> Response {
+        let (status, message) = match self {
+            AuthError::MissingToken => (StatusCode::UNAUTHORIZED, "Missing authentication token"),
+            AuthError::InvalidToken => (StatusCode::UNAUTHORIZED, "Invalid authentication token"),
+            AuthError::ValidationFailed(_msg) => {
+                (StatusCode::UNAUTHORIZED, "Token validation failed")
+            }
+        };
+
+        (status, message).into_response()
+    }
+}
+
+/// Wrapper for extracting XRPC request types.
+///
+/// Placeholder until jacquard_axum integration is complete.
+#[derive(Debug)]
+pub struct ExtractXrpc<T>(pub T);
+
+impl<S, T> FromRequestParts<S> for ExtractXrpc<T>
+where
+    S: Send + Sync,
+    T: for<'de> Deserialize<'de> + Send,
+{
+    type Rejection = StatusCode;
+
+    async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
+        let query = parts.uri.query().unwrap_or("");
+        let value = serde_urlencoded::from_str(query).map_err(|_| StatusCode::BAD_REQUEST)?;
+
+        Ok(ExtractXrpc(value))
+    }
+}

parakeet-appview/src/xrpc/context.rs

···
+use std::sync::Arc;
+
+use deadpool::managed::Pool;
+use diesel_async::pooled_connection::AsyncDieselConnectionManager;
+use diesel_async::AsyncPgConnection;
+
+use crate::entity::{EntityStore, PostEntity, ProfileEntity};
+use super::rate_limiter::RateLimiter;
+
+/// Global application state.
+pub struct GlobalState {
+    pub pool: Arc<Pool<AsyncDieselConnectionManager<AsyncPgConnection>>>,
+    pub profile_entity: Arc<EntityStore<ProfileEntity>>,
+    pub post_entity: Arc<EntityStore<PostEntity>>,
+    pub http_client: reqwest::Client,
+    pub rate_limiter: RateLimiter,
+}
+
+/// Context for XRPC handlers.
+#[derive(Clone)]
+pub struct XrpcContext {
+    pub state: Arc<GlobalState>,
+    pub auth_did: Option<String>,
+    pub viewer_id: Option<i32>,
+    pub labelers: Vec<String>,
+}
+
+impl XrpcContext {
+    pub fn new(state: Arc<GlobalState>) -> Self {
+        Self {
+            state,
+            auth_did: None,
+            viewer_id: None,
+            labelers: Vec::new(),
+        }
+    }
+
+    pub fn with_auth(mut self, did: String, viewer_id: Option<i32>) -> Self {
+        self.auth_did = Some(did);
+        self.viewer_id = viewer_id;
+        self
+    }
+
+    pub fn with_labelers(mut self, labelers: Vec<String>) -> Self {
+        self.labelers = labelers;
+        self
+    }
+
+    pub fn is_authenticated(&self) -> bool {
+        self.auth_did.is_some()
+    }
+
+    pub fn viewer_did(&self) -> Option<&str> {
+        self.auth_did.as_deref()
+    }
+
+    pub async fn db_conn(
+        &self,
+    ) -> Result<
+        deadpool::managed::Object<AsyncDieselConnectionManager<AsyncPgConnection>>,
+        deadpool::managed::PoolError<diesel_async::pooled_connection::PoolError>,
+    > {
+        self.state.pool.get().await
+    }
+}
+
+/// Builder for XrpcContext.
+pub struct XrpcContextBuilder {
+    state: Arc<GlobalState>,
+    auth_did: Option<String>,
+    viewer_id: Option<i32>,
+    labelers: Vec<String>,
+}
+
+impl XrpcContextBuilder {
+    pub fn new(state: Arc<GlobalState>) -> Self {
+        Self {
+            state,
+            auth_did: None,
+            viewer_id: None,
+            labelers: Vec::new(),
+        }
+    }
+
+    pub fn auth(mut self, did: String, viewer_id: Option<i32>) -> Self {
+        self.auth_did = Some(did);
+        self.viewer_id = viewer_id;
+        self
+    }
+
+    pub fn labelers(mut self, labelers: Vec<String>) -> Self {
+        self.labelers = labelers;
+        self
+    }
+
+    pub fn build(self) -> XrpcContext {
+        XrpcContext {
+            state: self.state,
+            auth_did: self.auth_did,
+            viewer_id: self.viewer_id,
+            labelers: self.labelers,
+        }
+    }
+}

parakeet-appview/src/xrpc/error.rs

···
+use axum::http::StatusCode;
+use axum::response::{IntoResponse, Response};
+use serde_json::json;
+use thiserror::Error;
+
+/// XRPC error type for the appview
+///
+/// This is a simplified error type that can be extended to integrate
+/// with Jacquard error types when needed.
+#[derive(Debug, Error)]
+pub enum XrpcError {
+    /// Resource not found error (404)
+    #[error("Not found: {0}")]
+    NotFound(String),
+
+    /// Malformed or invalid request syntax (400)
+    #[error("Bad request: {0}")]
+    BadRequest(String),
+
+    /// Request validation failed (400)
+    #[error("Invalid request: {0}")]
+    InvalidRequest(String),
+
+    /// Authentication required or failed (401)
+    #[error("Unauthorized: {0}")]
+    Unauthorized(String),
+
+    /// User lacks permission for this operation (403)
+    #[error("Forbidden: {0}")]
+    Forbidden(String),
+
+    /// Requested actor/DID does not exist (404)
+    #[error("Actor not found: {0}")]
+    ActorNotFound(String),
+
+    /// Requested post does not exist (404)
+    #[error("Post not found: {0}")]
+    PostNotFound(String),
+
+    /// Requested feed does not exist (404)
+    #[error("Feed not found: {0}")]
+    FeedNotFound(String),
+
+    /// Request exceeded rate limits (429)
+    #[error("Rate limit exceeded")]
+    RateLimitExceeded,
+
+    /// Unexpected server error (500)
+    #[error("Internal server error: {0}")]
+    InternalServerError(String),
+
+    /// Database operation failed (500)
+    #[error("Database error: {0}")]
+    DatabaseError(String),
+
+    /// Wrapper for other error types
+    #[error(transparent)]
+    Other(#[from] eyre::Report),
+}
+
+/// Result type for XRPC handlers
+pub type XrpcResult<T> = Result<T, XrpcError>;
+
+impl XrpcError {
+    /// Create an actor not found error
+    pub fn actor_not_found(identifier: &str) -> Self {
+        Self::ActorNotFound(format!("Actor not found: {}", identifier))
+    }
+
+    /// Create a post not found error
+    pub fn post_not_found(uri: &str) -> Self {
+        Self::PostNotFound(format!("Post not found: {}", uri))
+    }
+
+    /// Create a feed not found error
+    pub fn feed_not_found(uri: &str) -> Self {
+        Self::FeedNotFound(format!("Feed not found: {}", uri))
+    }
+
+    /// Create a bad request error
+    pub fn bad_request(message: &str) -> Self {
+        Self::BadRequest(message.to_string())
+    }
+
+    /// Get the HTTP status code for this error
+    pub fn status_code(&self) -> StatusCode {
+        match self {
+            Self::NotFound(_) => StatusCode::NOT_FOUND,
+            Self::BadRequest(_) | Self::InvalidRequest(_) => StatusCode::BAD_REQUEST,
+            Self::Unauthorized(_) => StatusCode::UNAUTHORIZED,
+            Self::Forbidden(_) => StatusCode::FORBIDDEN,
+            Self::ActorNotFound(_) | Self::PostNotFound(_) | Self::FeedNotFound(_) => {
+                StatusCode::NOT_FOUND
+            }
+            Self::RateLimitExceeded => StatusCode::TOO_MANY_REQUESTS,
+            Self::InternalServerError(_) | Self::DatabaseError(_) | Self::Other(_) => {
+                StatusCode::INTERNAL_SERVER_ERROR
+            }
+        }
+    }
+}
+
+impl IntoResponse for XrpcError {
+    fn into_response(self) -> Response {
+        let status = self.status_code();
+        let body = json!({
+            "error": self.to_string(),
+        });
+
+        (status, axum::Json(body)).into_response()
+    }
+}
+
+/// Extension trait for converting errors into XrpcError
+pub trait IntoXrpcError {
+    /// Convert this error type into an XrpcError
+    fn into_xrpc_error(self) -> XrpcError;
+}
+
+impl IntoXrpcError for eyre::Report {
+    fn into_xrpc_error(self) -> XrpcError {
+        XrpcError::Other(self)
+    }
+}
+
+impl IntoXrpcError for diesel::result::Error {
+    fn into_xrpc_error(self) -> XrpcError {
+        XrpcError::DatabaseError(self.to_string())
+    }
+}

parakeet-appview/src/xrpc/helpers.rs

···
+use chrono::{DateTime, Utc};
+use eyre::{eyre, Result};
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+
+/// Pagination helpers for XRPC endpoints
+/// Standard pagination parameters used across XRPC endpoints
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PaginationParams {
+    /// Maximum number of items to return in a single page
+    pub limit: Option<i32>,
+    /// Opaque cursor for fetching the next page of results
+    pub cursor: Option<String>,
+}
+
+impl PaginationParams {
+    /// Get the limit value, clamped to a range with a default
+    pub fn get_limit(&self, default: usize, min: usize, max: usize) -> usize {
+        self.limit
+            .map(|l| l as usize)
+            .unwrap_or(default)
+            .clamp(min, max)
+    }
+
+    /// Parse a datetime cursor
+    pub fn parse_datetime_cursor(&self) -> Option<DateTime<Utc>> {
+        self.cursor
+            .as_ref()
+            .and_then(|c| c.parse::<i64>().ok())
+            .and_then(DateTime::from_timestamp_millis)
+    }
+
+    /// Parse an integer cursor
+    pub fn parse_int_cursor(&self) -> Option<i64> {
+        self.cursor.as_ref().and_then(|c| c.parse().ok())
+    }
+}
+
+/// Response with pagination support
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PaginatedResponse<T> {
+    /// The list of items for the current page
+    pub items: Vec<T>,
+    /// Cursor for fetching the next page, if more items exist
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub cursor: Option<String>,
+}
+
+impl<T> PaginatedResponse<T> {
+    /// Create a paginated response from items
+    pub fn new(mut items: Vec<T>, limit: usize) -> Self {
+        let has_more = items.len() > limit;
+        if has_more {
+            items.truncate(limit);
+        }
+
+        Self {
+            items,
+            cursor: None, // Caller should set if has_more
+        }
+    }
+
+    /// Set the cursor if there are more items
+    pub fn with_cursor<F>(mut self, has_more: bool, cursor_fn: F) -> Self
+    where
+        F: FnOnce(&T) -> String,
+    {
+        if has_more && !self.items.is_empty() {
+            self.cursor = Some(cursor_fn(self.items.last().unwrap()));
+        }
+        self
+    }
+}
+
+/// Helper to paginate a vector of items
+pub fn paginate<T>(
+    items: Vec<T>,
+    limit: usize,
+    cursor_fn: impl Fn(&T) -> String,
+) -> PaginatedResponse<T> {
+    let has_more = items.len() > limit;
+    let mut items = items;
+
+    if has_more {
+        items.truncate(limit);
+    }
+
+    let cursor = if has_more && !items.is_empty() {
+        Some(cursor_fn(items.last().unwrap()))
+    } else {
+        None
+    };
+
+    PaginatedResponse { items, cursor }
+}
+
+/// AT URI parsing helpers
+/// Parse an AT URI into its components
+///
+/// Note: This is a simplified implementation.
+/// In production, use the proper AtUri parser from jacquard_common.
+pub fn parse_at_uri(uri: &str) -> Result<(String, String, String)> {
+    // Simple parsing: at://did/collection/rkey
+    if !uri.starts_with("at://") {
+        return Err(eyre!("Not an AT URI"));
+    }
+
+    let without_scheme = &uri[5..]; // Remove "at://"
+    let parts: Vec<&str> = without_scheme.split('/').collect();
+
+    if parts.len() < 3 {
+        return Err(eyre!("AT URI path incomplete"));
+    }
+
+    let did = parts[0].to_string();
+    let collection = parts[1].to_string();
+    let rkey = parts[2].to_string();
+
+    Ok((did, collection, rkey))
+}
+
+/// Build an AT URI from components
+pub fn build_at_uri(did: &str, collection: &str, rkey: &str) -> String {
+    format!("at://{}/{}/{}", did, collection, rkey)
+}
+
+/// Identifier resolution helpers
+/// Check if a string is a valid DID
+pub fn is_did(identifier: &str) -> bool {
+    identifier.starts_with("did:")
+}
+
+/// Check if a string is a valid handle
+pub fn is_handle(identifier: &str) -> bool {
+    // Basic handle validation - contains a dot and doesn't start with did:
+    !is_did(identifier) && identifier.contains('.')
+}
+
+/// Normalize a handle (remove @ prefix if present)
+pub fn normalize_handle(handle: &str) -> &str {
+    handle.strip_prefix('@').unwrap_or(handle)
+}
+
+/// Cursor encoding/decoding
+/// Encode a timestamp as a cursor
+pub fn encode_timestamp_cursor(ts: DateTime<Utc>) -> String {
+    ts.timestamp_millis().to_string()
+}
+
+/// Encode an integer as a cursor
+pub fn encode_int_cursor(value: i64) -> String {
+    value.to_string()
+}
+
+/// Encode a compound cursor with multiple values
+pub fn encode_compound_cursor(values: &[(&str, &str)]) -> String {
+    let params: Vec<String> = values
+        .iter()
+        .map(|(k, v)| format!("{}={}", k, v))
+        .collect();
+    params.join("&")
+}
+
+/// Decode a compound cursor
+pub fn decode_compound_cursor(cursor: &str) -> HashMap<String, String> {
+    cursor
+        .split('&')
+        .filter_map(|part| {
+            let mut split = part.split('=');
+            match (split.next(), split.next()) {
+                (Some(k), Some(v)) => Some((k.to_string(), v.to_string())),
+                _ => None,
+            }
+        })
+        .collect()
+}
+
+/// Order preservation helpers
+/// Preserve the order of items based on the order of IDs
+pub fn preserve_order<Id, Item>(
+    ids: &[Id],
+    mut items: Vec<(Id, Item)>,
+) -> Vec<Item>
+where
+    Id: Eq + std::hash::Hash + Clone,
+    Item: Clone,
+{
+    let item_map: HashMap<Id, Item> = items.drain(..).collect();
+
+    ids.iter()
+        .filter_map(|id| item_map.get(id).cloned())
+        .collect()
+}
+
+/// Preserve order with optional items
+pub fn preserve_order_optional<Id, Item>(
+    ids: &[Id],
+    mut items: Vec<(Id, Item)>,
+) -> Vec<Option<Item>>
+where
+    Id: Eq + std::hash::Hash + Clone,
+    Item: Clone,
+{
+    let item_map: HashMap<Id, Item> = items.drain(..).collect();
+
+    ids.iter()
+        .map(|id| item_map.get(id).cloned())
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_parse_at_uri() {
+        let uri = "at://did:plc:example/app.bsky.feed.post/3k2yv5";
+        let (did, collection, rkey) = parse_at_uri(uri).unwrap();
+
+        assert_eq!(did, "did:plc:example");
+        assert_eq!(collection, "app.bsky.feed.post");
+        assert_eq!(rkey, "3k2yv5");
+    }
+
+    #[test]
+    fn test_pagination() {
+        let items = vec![1, 2, 3, 4, 5];
+        let response = paginate(items, 3, |i| i.to_string());
+
+        assert_eq!(response.items, vec![1, 2, 3]);
+        assert_eq!(response.cursor, Some("3".to_string()));
+    }
+
+    #[test]
+    fn test_compound_cursor() {
+        let cursor = encode_compound_cursor(&[("id", "123"), ("ts", "456")]);
+        assert_eq!(cursor, "id=123&ts=456");
+
+        let decoded = decode_compound_cursor(&cursor);
+        assert_eq!(decoded.get("id"), Some(&"123".to_string()));
+        assert_eq!(decoded.get("ts"), Some(&"456".to_string()));
+    }
+}

parakeet-appview/src/xrpc/rate_limiter.rs

···
+//! Rate limiting for XRPC endpoints.
+
+use dashmap::DashMap;
+use std::str::FromStr;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+/// Rate limiter for tracking request rates per client.
+#[derive(Clone)]
+pub struct RateLimiter {
+    /// Stores rate limit state per key (e.g., DID, IP address)
+    buckets: Arc<DashMap<String, TokenBucket>>,
+    /// Default configuration for all rate limits
+    default_config: RateLimitConfig,
+}
+
+/// Configuration for a rate limit.
+#[derive(Clone, Debug)]
+pub struct RateLimitConfig {
+    /// Maximum number of requests
+    pub max_requests: u32,
+    /// Time window for the requests
+    pub window: Duration,
+}
+
+impl FromStr for RateLimitConfig {
+    type Err = String;
+
+    /// Parse from a string like "100/60s" or "1000/1h"
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let parts: Vec<&str> = s.split('/').collect();
+        if parts.len() != 2 {
+            return Err(format!("Invalid rate limit format: {}", s));
+        }
+
+        let max_requests = parts[0]
+            .parse::<u32>()
+            .map_err(|_| format!("Invalid request count: {}", parts[0]))?;
+
+        let window =
+            parse_duration(parts[1]).ok_or_else(|| format!("Invalid duration: {}", parts[1]))?;
+
+        Ok(RateLimitConfig {
+            max_requests,
+            window,
+        })
+    }
+}
+
+/// Parse duration strings like "60s", "5m", "1h"
+fn parse_duration(s: &str) -> Option<Duration> {
+    let (num_str, unit) = s.split_at(s.len() - 1);
+    let num = num_str.parse::<u64>().ok()?;
+
+    match unit {
+        "s" => Some(Duration::from_secs(num)),
+        "m" => Some(Duration::from_secs(num * 60)),
+        "h" => Some(Duration::from_secs(num * 3600)),
+        _ => None,
+    }
+}
+
+/// Token bucket for rate limiting.
+struct TokenBucket {
+    tokens: u32,
+    last_refill: Instant,
+}
+
+impl RateLimiter {
+    /// Create a new rate limiter with default configuration.
+    pub fn new() -> Self {
+        Self {
+            buckets: Arc::new(DashMap::new()),
+            default_config: RateLimitConfig {
+                max_requests: 100,
+                window: Duration::from_secs(60), // 100 requests per minute default
+            },
+        }
+    }
+
+    /// Check if a request is allowed for the given key and configuration.
+    pub fn check(&self, key: &str, config_str: Option<&str>) -> bool {
+        let config = if let Some(s) = config_str {
+            match s.parse::<RateLimitConfig>() {
+                Ok(c) => c,
+                Err(e) => {
+                    tracing::warn!("Invalid rate limit config '{}': {}, using default", s, e);
+                    self.default_config.clone()
+                }
+            }
+        } else {
+            self.default_config.clone()
+        };
+
+        let mut entry = self
+            .buckets
+            .entry(key.to_string())
+            .or_insert_with(|| TokenBucket {
+                tokens: config.max_requests,
+                last_refill: Instant::now(),
+            });
+
+        let now = Instant::now();
+        let time_passed = now.duration_since(entry.last_refill);
+
+        // Refill tokens based on time passed
+        if time_passed >= config.window {
+            // Full refill
+            entry.tokens = config.max_requests;
+            entry.last_refill = now;
+        } else {
+            // Partial refill
+            let refill_rate = config.max_requests as f64 / config.window.as_secs_f64();
+            let tokens_to_add = (time_passed.as_secs_f64() * refill_rate) as u32;
+            if tokens_to_add > 0 {
+                entry.tokens = (entry.tokens + tokens_to_add).min(config.max_requests);
+                entry.last_refill = now;
+            }
+        }
+
+        // Check if we have tokens available
+        if entry.tokens > 0 {
+            entry.tokens -= 1;
+            true
+        } else {
+            false
+        }
+    }
+
+    /// Clear rate limit data for a specific key.
+    pub fn clear(&self, key: &str) {
+        self.buckets.remove(key);
+    }
+
+    /// Clear all rate limit data.
+    pub fn clear_all(&self) {
+        self.buckets.clear();
+    }
+}
+
+impl Default for RateLimiter {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::thread;
+
+    #[test]
+    fn test_rate_limit_config_parsing() {
+        let config = "100/60s".parse::<RateLimitConfig>().unwrap();
+        assert_eq!(config.max_requests, 100);
+        assert_eq!(config.window, Duration::from_secs(60));
+
+        let config = "1000/5m".parse::<RateLimitConfig>().unwrap();
+        assert_eq!(config.max_requests, 1000);
+        assert_eq!(config.window, Duration::from_secs(300));
+
+        let config = "5000/1h".parse::<RateLimitConfig>().unwrap();
+        assert_eq!(config.max_requests, 5000);
+        assert_eq!(config.window, Duration::from_secs(3600));
+
+        assert!("invalid".parse::<RateLimitConfig>().is_err());
+        assert!("100".parse::<RateLimitConfig>().is_err());
+        assert!("100/60x".parse::<RateLimitConfig>().is_err());
+    }
+
+    #[test]
+    fn test_rate_limiting() {
+        let limiter = RateLimiter::new();
+
+        // Test basic limiting - 2 requests per second
+        let key = "test_user";
+        let config = "2/1s";
+
+        // First two requests should succeed
+        assert!(limiter.check(key, Some(config)));
+        assert!(limiter.check(key, Some(config)));
+
+        // Third request should fail
+        assert!(!limiter.check(key, Some(config)));
+
+        // After waiting, should refill
+        thread::sleep(Duration::from_millis(1100));
+        assert!(limiter.check(key, Some(config)));
+        assert!(limiter.check(key, Some(config)));
+        assert!(!limiter.check(key, Some(config)));
+    }
+
+    #[test]
+    fn test_different_keys() {
+        let limiter = RateLimiter::new();
+        let config = "1/1s";
+
+        // Different keys should have independent limits
+        assert!(limiter.check("user1", Some(config)));
+        assert!(limiter.check("user2", Some(config)));
+        assert!(limiter.check("user3", Some(config)));
+
+        // But second request for same key should fail
+        assert!(!limiter.check("user1", Some(config)));
+        assert!(!limiter.check("user2", Some(config)));
+        assert!(!limiter.check("user3", Some(config)));
+    }
+}