diff --git a/parakeet-db/src/actor_cache.rs b/parakeet-db/src/actor_cache.rs index 4e683b42..1b0ad212 100644 --- a/parakeet-db/src/actor_cache.rs +++ b/parakeet-db/src/actor_cache.rs @@ -11,7 +11,7 @@ use dashmap::DashMap; use std::sync::Arc; -use crate::types::{ActorStatus, ActorSyncState, AllowlistStatus}; +use crate::types::{ActorStatus, ActorSyncState}; /// Basic actor metadata stored in the cache #[derive(Debug, Clone)] @@ -68,41 +68,6 @@ impl CachedActorStore { dids.iter().map(|did| self.get_actor(did)).collect() } - /// Get allowlist status for a DID (cache-only, no DB fallback) - /// - /// Returns: - /// - `FullyAllowed` if actor is Synced/Dirty/Processing - /// - `PartiallyAllowed` if actor is Partial - /// - `NotAllowed` if actor is not in cache - pub fn get_allowlist_status(&self, did: &str) -> AllowlistStatus { - self.actors - .get(did) - .map(|entry| AllowlistStatus::from(entry.sync_state)) - .unwrap_or(AllowlistStatus::NotAllowed) - } - - /// Get allowlist status for multiple DIDs at once (cache-only, no DB fallback) - pub fn get_allowlist_status_batch(&self, dids: &[&str]) -> Vec { - dids.iter() - .map(|did| self.get_allowlist_status(did)) - .collect() - } - - /// Check if any DID in a list passes the allowlist (faster than batch for early exit) - /// - /// This is optimized for the routing filter use case where we want to know - /// "is at least one DID fully allowed?" without checking all of them. - /// - /// Returns true as soon as a fully allowed DID is found. - pub fn has_fully_allowed_did(&self, dids: &[&str]) -> bool { - dids.iter().any(|&did| { - self.actors - .get(did) - .map(|entry| AllowlistStatus::from(entry.sync_state).is_fully_allowed()) - .unwrap_or(false) - }) - } - /// Add or update an actor in the cache pub fn upsert_actor(&self, actor: ActorMetadata) { self.actors.insert(actor.did.clone(), actor); @@ -297,71 +262,4 @@ mod tests { assert!(!cache.contains_actor("did:plc:anyone")); } - #[test] - fn test_batch_allowlist_status() { - // Question: Does get_allowlist_status_batch correctly batch check multiple DIDs? - let cache = CachedActorStore::new(); - - // Create test actors with different sync states - let mut synced = make_test_actor("did:plc:synced"); - synced.sync_state = ActorSyncState::Synced; - - let mut partial = make_test_actor("did:plc:partial"); - partial.sync_state = ActorSyncState::Partial; - - cache.update_cache(vec![synced, partial]); - - // Batch check mixed DIDs - let dids = vec!["did:plc:synced", "did:plc:partial", "did:plc:unknown"]; - let results = cache.get_allowlist_status_batch(&dids); - - assert_eq!(results.len(), 3); - assert!(results[0].is_fully_allowed(), "Synced should be fully allowed"); - assert_eq!(results[1], AllowlistStatus::PartiallyAllowed); - assert_eq!(results[2], AllowlistStatus::NotAllowed); - } - - #[test] - fn test_has_fully_allowed_did_finds_allowed() { - // Question: Does has_fully_allowed_did correctly find a fully allowed DID? - let cache = CachedActorStore::new(); - - let mut synced = make_test_actor("did:plc:synced"); - synced.sync_state = ActorSyncState::Synced; - - let mut partial = make_test_actor("did:plc:partial"); - partial.sync_state = ActorSyncState::Partial; - - cache.update_cache(vec![synced, partial]); - - // Mixed list should return true (has synced actor) - let dids = vec!["did:plc:partial", "did:plc:synced", "did:plc:unknown"]; - assert!(cache.has_fully_allowed_did(&dids), "Should find fully allowed DID"); - - // Only partial and unknown should return false - let dids = vec!["did:plc:partial", "did:plc:unknown"]; - assert!(!cache.has_fully_allowed_did(&dids), "Should not find fully allowed DID"); - - // Empty list should return false - assert!(!cache.has_fully_allowed_did(&[]), "Empty list should return false"); - } - - #[test] - fn test_has_fully_allowed_did_early_exit() { - // Question: Does has_fully_allowed_did exit early when it finds a match? - let cache = CachedActorStore::new(); - - let mut synced1 = make_test_actor("did:plc:synced1"); - synced1.sync_state = ActorSyncState::Synced; - - let mut synced2 = make_test_actor("did:plc:synced2"); - synced2.sync_state = ActorSyncState::Synced; - - cache.update_cache(vec![synced1, synced2]); - - // Should return true as soon as first fully allowed is found - // (This test validates correctness; performance would need benchmarking) - let dids = vec!["did:plc:synced1", "did:plc:synced2"]; - assert!(cache.has_fully_allowed_did(&dids), "Should find at least one fully allowed"); - } } diff --git a/parakeet-db/src/allowlist.rs b/parakeet-db/src/allowlist.rs deleted file mode 100644 index bbb6f858..00000000 --- a/parakeet-db/src/allowlist.rs +++ /dev/null @@ -1,298 +0,0 @@ -//! Shared definitions for the allowlist functionality -//! -//! This module defines common types and traits for allowlist functionality -//! across different database backends (Diesel and tokio-postgres). -//! -//! Uses DashMap for efficient concurrent access without complex sharding. - -use dashmap::DashSet; -use std::fmt::Debug; -use std::sync::Arc; - -/// Represents an entry in the allowlist -#[derive(Debug, Clone)] -pub struct AllowlistEntry { - /// The unique ID of the entry - pub id: Option, - /// The DID in the allowlist - pub did: String, - /// Optional description of why the DID is allowed - pub description: Option, - /// When the entry was created - pub created_at: Option, -} - -/// Clean and normalize a DID -/// -/// Handles the following cases: -/// - Removes "at://" prefix if present -/// - Extracts DID part before any slashes -/// - Adds "did:plc:" prefix for raw PLC keys -/// - Adds "did:web:" prefix for domain-like inputs -/// -/// Returns None if the input doesn't appear to be a valid DID format -pub fn clean_did(did: &str) -> Option { - let did = did.trim(); - - // Remove at:// prefix if present - let did = if let Some(stripped) = did.strip_prefix("at://") { - stripped - } else { - did - }; - - // Extract the DID part before any slashes - let did_part = did.split('/').next().unwrap_or(did); - - // Ensure it starts with "did:" - if !did_part.starts_with("did:") { - // If it looks like a raw PLC key, prepend "did:plc:" - if did_part.chars().all(|c| c.is_ascii_alphanumeric()) - && (did_part.len() >= 16 && did_part.len() <= 64) - { - return Some(format!("did:plc:{}", did_part)); - } else { - // Try to interpret it as a web DID (e.g. example.com) - return Some(format!("did:web:{}", did_part)); - } - } - - Some(did_part.to_string()) -} - -/// A thread-safe cached allowlist that periodically refreshes from the database -/// -/// Uses DashSet for efficient concurrent access. DashSet is internally sharded -/// (64 shards by default) which provides excellent parallel performance without -/// the complexity of manual sharding and bloom filters. -/// -/// DashSet provides: -/// - Lock-free reads (using internal RwLock per shard) -/// - Automatic hash-based sharding (64 shards) -/// - O(1) average-case lookups -/// - Simple API without bloom filter complexity -#[derive(Clone, Debug)] -pub struct CachedAllowlist { - /// Set of allowed DIDs (internally sharded by DashSet) - dids: Arc>, -} - -impl CachedAllowlist { - /// Create a new empty cached allowlist - pub fn new() -> Self { - Self { - dids: Arc::new(DashSet::new()), - } - } - - /// Update the cache with new DIDs (replaces existing set) - /// - /// Returns true if the set of DIDs changed (additions or removals) - pub fn update_cache(&self, dids: Vec) -> bool { - // Check if the set changed by comparing sizes and contents - let old_len = self.dids.len(); - let new_len = dids.len(); - - // Fast path: different sizes means definitely changed - if old_len != new_len { - self.dids.clear(); - for did in dids { - self.dids.insert(did); - } - return true; - } - - // Same size: check if all DIDs are the same - let changed = dids.iter().any(|did| !self.dids.contains(did)); - - if changed { - self.dids.clear(); - for did in dids { - self.dids.insert(did); - } - } - - changed - } - - /// Check if a DID is in the cache - pub fn contains_did(&self, did: &str) -> bool { - self.dids.contains(did) - } - - /// Add a DID to the cache - pub fn add_did(&self, did: &str) { - self.dids.insert(did.to_owned()); - } - - /// Remove a DID from the cache - pub fn remove_did(&self, did: &str) { - self.dids.remove(did); - } - - /// Get the number of entries in the cache - pub fn len(&self) -> usize { - self.dids.len() - } - - /// Check if the cache is empty - pub fn is_empty(&self) -> bool { - self.dids.is_empty() - } - - /// Get all DIDs in the cache - /// - /// Returns a snapshot of all DIDs currently in the allowlist. - pub fn get_all_dids(&self) -> Vec { - self.dids.iter().map(|entry| entry.clone()).collect() - } -} - -impl Default for CachedAllowlist { - fn default() -> Self { - Self::new() - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_contains_did_finds_allowlisted() { - let cache = CachedAllowlist::new(); - - let test_dids = vec![ - "did:plc:user1".to_string(), - "did:plc:user2".to_string(), - "did:plc:user3".to_string(), - ]; - - cache.update_cache(test_dids.clone()); - - // All allowlisted DIDs should be found - for did in &test_dids { - assert!( - cache.contains_did(did), - "Should find allowlisted DID: {}", - did - ); - } - } - - #[test] - fn test_contains_did_rejects_non_members() { - let cache = CachedAllowlist::new(); - - let allowed_dids = vec![ - "did:plc:allowed1".to_string(), - "did:plc:allowed2".to_string(), - ]; - - cache.update_cache(allowed_dids); - - // Non-members should be rejected - let non_member = "did:plc:notinlist"; - assert!(!cache.contains_did(non_member), "Non-member should not be found"); - } - - #[test] - fn test_add_did() { - let cache = CachedAllowlist::new(); - - // Start with empty cache - assert!(!cache.contains_did("did:plc:newuser")); - - // Add a DID - cache.add_did("did:plc:newuser"); - - // Should now be found - assert!( - cache.contains_did("did:plc:newuser"), - "Newly added DID should be found immediately" - ); - } - - #[test] - fn test_remove_did() { - let cache = CachedAllowlist::new(); - - let dids = vec![ - "did:plc:user1".to_string(), - "did:plc:user2".to_string(), - "did:plc:user3".to_string(), - ]; - - cache.update_cache(dids); - - // Verify user2 exists - assert!(cache.contains_did("did:plc:user2")); - - // Remove user2 - cache.remove_did("did:plc:user2"); - - // user2 should no longer be found - assert!( - !cache.contains_did("did:plc:user2"), - "Removed DID should not be found" - ); - - // Other DIDs should still be found - assert!(cache.contains_did("did:plc:user1")); - assert!(cache.contains_did("did:plc:user3")); - } - - #[test] - fn test_update_cache_replaces_entries() { - let cache = CachedAllowlist::new(); - - // Initial set - cache.update_cache(vec!["did:plc:old1".to_string(), "did:plc:old2".to_string()]); - - assert!(cache.contains_did("did:plc:old1")); - assert!(!cache.contains_did("did:plc:new1")); - - // Update with new set - cache.update_cache(vec!["did:plc:new1".to_string(), "did:plc:new2".to_string()]); - - // Old DIDs should be gone - assert!(!cache.contains_did("did:plc:old1")); - assert!(!cache.contains_did("did:plc:old2")); - - // New DIDs should be present - assert!(cache.contains_did("did:plc:new1")); - assert!(cache.contains_did("did:plc:new2")); - } - - #[test] - fn test_large_dataset() { - let cache = CachedAllowlist::new(); - - // Generate 1000 DIDs - let dids: Vec = (0..1000).map(|i| format!("did:plc:user{:04}", i)).collect(); - - cache.update_cache(dids.clone()); - - // All 1000 DIDs should be found - for did in &dids { - assert!( - cache.contains_did(did), - "Should find DID in large dataset: {}", - did - ); - } - - // Non-existent DID should not be found - let non_existent = "did:plc:notinlist9999"; - assert!(!cache.contains_did(non_existent)); - } - - #[test] - fn test_empty_cache() { - let cache = CachedAllowlist::new(); - - assert_eq!(cache.len(), 0); - assert!(cache.is_empty()); - assert!(!cache.contains_did("did:plc:anyone")); - } -} diff --git a/parakeet-db/src/lib.rs b/parakeet-db/src/lib.rs index 1e9cd97c..549c47cb 100644 --- a/parakeet-db/src/lib.rs +++ b/parakeet-db/src/lib.rs @@ -1,5 +1,4 @@ pub mod actor_cache; -pub mod allowlist; pub mod at_uri_util; pub mod cid_util; pub mod composite_types; diff --git a/parakeet-db/src/types.rs b/parakeet-db/src/types.rs index 5e018cf7..0e75c9a7 100644 --- a/parakeet-db/src/types.rs +++ b/parakeet-db/src/types.rs @@ -496,64 +496,6 @@ diesel_enum!(RepostStatus, RepostStatus, { Stub = "stub", }); -// ============================================================================= -// HELPER TRAITS FOR ALLOWLIST -// ============================================================================= - -/// Allowlist status for an actor -/// -/// Represents whether an actor is on the allowlist and how they should be indexed. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub enum AllowlistStatus { - /// Actor is fully allowed (Synced/Dirty/Processing) - FullyAllowed, - /// Actor is partially allowed (Partial - interacts with allowlisted users) - PartiallyAllowed, - /// Actor is not on the allowlist - NotAllowed, -} - -impl AllowlistStatus { - /// Check if this status allows processing the actor's content - pub fn should_process(self) -> bool { - matches!(self, Self::FullyAllowed | Self::PartiallyAllowed) - } - - /// Check if this is a fully allowed actor - pub fn is_fully_allowed(self) -> bool { - matches!(self, Self::FullyAllowed) - } - - /// Check if this is a partially allowed actor - pub fn is_partially_allowed(self) -> bool { - matches!(self, Self::PartiallyAllowed) - } - - /// Check if this actor is not allowed - pub fn is_not_allowed(self) -> bool { - matches!(self, Self::NotAllowed) - } -} - -impl From for AllowlistStatus { - fn from(state: ActorSyncState) -> Self { - match state { - ActorSyncState::Synced | ActorSyncState::Dirty | ActorSyncState::Processing => { - Self::FullyAllowed - } - ActorSyncState::Partial => Self::PartiallyAllowed, - } - } -} - -impl From> for AllowlistStatus { - fn from(state: Option) -> Self { - match state { - Some(state) => state.into(), - None => Self::NotAllowed, - } - } -} // ============================================================================= // NOTES ON ENUM EXTENSIBILITY diff --git a/parakeet/src/admin/mod.rs b/parakeet/src/admin/mod.rs deleted file mode 100644 index 5a6ae8ec..00000000 --- a/parakeet/src/admin/mod.rs +++ /dev/null @@ -1,424 +0,0 @@ -//! Admin API endpoints for managing Parakeet - -use axum::{ - extract::{Path, State}, - http::StatusCode, - response::{IntoResponse as _, Response}, - routing::{delete, get, post}, - Json, Router, -}; -use chrono::prelude::*; - -use serde::{Deserialize, Serialize}; - -use crate::allowlist; -use crate::GlobalState; -use parakeet_db::allowlist::AllowlistEntry as SharedAllowlistEntry; - -/// Register admin routes -pub fn routes() -> Router { - Router::new() - .route("/admin/allowlist", get(get_allowlist)) - .route("/admin/allowlist", post(add_to_allowlist)) - .route("/admin/allowlist/{did}", delete(remove_from_allowlist)) - .route("/admin/allowlist/{did}", get(get_allowlist_entry)) - .route("/admin/backfill", post(queue_backfill)) - .route("/admin/allowlist-and-backfill", post(allowlist_and_backfill)) -} - -/// Response for allowlist entries -#[derive(Serialize)] -struct AllowlistEntry { - id: i32, - did: String, - created_at: DateTime, - description: Option, -} - -impl From<&SharedAllowlistEntry> for AllowlistEntry { - fn from(entry: &SharedAllowlistEntry) -> Self { - Self { - id: entry.id.unwrap_or(0), - did: entry.did.clone(), - created_at: entry.created_at.map_or_else(Utc::now, |naive| { - DateTime::from_naive_utc_and_offset(naive, Utc) - }), - description: entry.description.clone(), - } - } -} - -/// Request to add a DID to the allowlist -#[derive(Deserialize)] -struct AddAllowlistRequest { - did: String, - description: Option, -} - -/// Response for the full allowlist -#[derive(Serialize)] -struct AllowlistResponse { - entries: Vec, -} - -/// Get all DIDs in the allowlist -async fn get_allowlist( - State(state): State, -) -> Result, Response> { - let mut conn = state.pool.get().await.map_err(internal_error)?; - - let rows = allowlist::get_all(&mut conn) - .await - .map_err(internal_error)?; - - let entries = rows - .into_iter() - .map(|row| AllowlistEntry { - id: row.id, - did: row.did.clone(), - created_at: row.created_at.and_utc(), - description: row.description, - }) - .collect(); - - Ok(Json(AllowlistResponse { entries })) -} - -/// Add a DID to the allowlist -async fn add_to_allowlist( - State(state): State, - Json(request): Json, -) -> Result { - let mut conn = state.pool.get().await.map_err(internal_error)?; - let description_ref = request.description.as_deref(); - - let add_result = allowlist::add(&mut conn, &request.did, description_ref).await; - match add_result { - Ok(rows_affected) => { - if rows_affected > 0 { - Ok(StatusCode::CREATED) - } else { - // DID was already in the allowlist (affected 0 rows due to conflict) - Err(( - StatusCode::CONFLICT, - format!("DID {did} is already in the allowlist", did = request.did), - ) - .into_response()) - } - } - Err(e) => { - // For any other database errors - Err(internal_error(e)) - } - } -} - -/// Remove a DID from the allowlist -async fn remove_from_allowlist( - State(state): State, - Path(did): Path, -) -> Result { - let mut conn = state.pool.get().await.map_err(internal_error)?; - - let result = allowlist::remove(&mut conn, &did) - .await - .map_err(internal_error)?; - - if result == 0 { - return Err((StatusCode::NOT_FOUND, format!("DID {did} not found")).into_response()); - } - - Ok(StatusCode::NO_CONTENT) -} - -/// Get a specific DID's entry from the allowlist -async fn get_allowlist_entry( - State(state): State, - Path(did): Path, -) -> Result, Response> { - let mut conn = state.pool.get().await.map_err(internal_error)?; - - let row = allowlist::get_entry(&mut conn, &did) - .await - .map_err(internal_error)?; - - if let Some(row) = row { - Ok(Json(AllowlistEntry { - id: row.id.unwrap_or(0), - did: row.did.clone(), - created_at: row.created_at.map_or_else(Utc::now, |naive| { - DateTime::from_naive_utc_and_offset(naive, Utc) - }), - description: row.description, - })) - } else { - Err((StatusCode::NOT_FOUND, format!("DID {did} not found")).into_response()) - } -} - -/// Request to add DID(s) to allowlist and queue for backfill -#[derive(Deserialize)] -struct AllowlistAndBackfillRequest { - /// DID(s) to add and backfill (single DID or array of DIDs) - #[serde(deserialize_with = "deserialize_did_or_dids")] - dids: Vec, - /// Optional description to apply to all DIDs - description: Option, - /// Whether to reset the repo_rev (forces full re-download) - /// Defaults to true for new allowlist additions - #[serde(default = "default_reset_repo_rev")] - #[expect(dead_code, reason = "reset_repo_rev accepted for API compatibility but backfill reset logic not yet implemented")] - reset_repo_rev: bool, -} - -fn default_reset_repo_rev() -> bool { - true -} - -/// Response for combined allowlist and backfill operation -#[derive(Serialize)] -struct AllowlistAndBackfillResponse { - /// DIDs newly added to allowlist - added: Vec, - /// DIDs already in allowlist - already_listed: Vec, - /// DIDs successfully queued for backfill - queued: Vec, - /// Errors encountered during processing - errors: Vec, -} - -/// Request to queue a backfill -#[derive(Deserialize)] -struct BackfillRequest { - /// DID(s) to backfill (single DID or array of DIDs) - #[serde(deserialize_with = "deserialize_did_or_dids")] - dids: Vec, - /// Whether to reset the repo_rev (forces full re-download) - #[serde(default)] - #[expect(dead_code, reason = "reset_repo_rev accepted for API compatibility but backfill reset logic not yet implemented")] - reset_repo_rev: bool, -} - -/// Custom deserializer to handle either a single DID string or an array of DIDs -fn deserialize_did_or_dids<'de, D>(deserializer: D) -> Result, D::Error> -where - D: serde::Deserializer<'de>, -{ - use serde::de::Error; - use serde_json::Value; - - let value = Value::deserialize(deserializer)?; - match value { - Value::String(s) => Ok(vec![s]), - Value::Array(arr) => arr - .into_iter() - .map(|v| { - v.as_str() - .map(String::from) - .ok_or_else(|| D::Error::custom("DID array must contain strings")) - }) - .collect(), - _ => Err(D::Error::custom( - "dids must be a string or array of strings", - )), - } -} - -/// Response for backfill operation -#[derive(Serialize)] -struct BackfillResponse { - queued: Vec, - errors: Vec, -} - -#[derive(Serialize)] -struct BackfillError { - did: String, - error: String, -} - -/// Add DID(s) to allowlist and queue for immediate backfill -async fn allowlist_and_backfill( - State(state): State, - Json(request): Json, -) -> Result, Response> { - let mut conn = state.pool.get().await.map_err(internal_error)?; - - let mut added = Vec::new(); - let mut already_listed = Vec::new(); - let mut queued = Vec::new(); - let mut errors = Vec::new(); - - for did in request.dids { - // Clean the DID - let Some(clean_did) = parakeet_db::allowlist::clean_did(&did) else { - errors.push(BackfillError { - did: did.clone(), - error: "Invalid DID format".to_string(), - }); - continue; - }; - - // Add to allowlist (this also updates the cache) - let description_ref = request.description.as_deref(); - match state - .allowlist - .add_did(&mut conn, &clean_did, description_ref) - .await - { - Ok(was_added) => { - if was_added { - added.push(clean_did.clone()); - } else { - already_listed.push(clean_did.clone()); - } - } - Err(e) => { - errors.push(BackfillError { - did: clean_did.clone(), - error: format!("Failed to add to allowlist: {}", e), - }); - continue; - } - } - - // Enqueue backfill job - let enqueue_result = enqueue_backfill_job(&mut conn, &clean_did).await; - match enqueue_result { - Ok(()) => { - queued.push(clean_did.clone()); - } - Err(e) => { - errors.push(BackfillError { - did: clean_did.clone(), - error: format!("Failed to enqueue backfill: {}", e), - }); - } - } - } - - Ok(Json(AllowlistAndBackfillResponse { - added, - already_listed, - queued, - errors, - })) -} - -/// Enqueue a backfill job for a DID -async fn enqueue_backfill_job( - conn: &mut diesel_async::AsyncPgConnection, - did: &str, -) -> Result<(), diesel::result::Error> { - use diesel::sql_types::Text; - - diesel_async::RunQueryDsl::execute( - diesel::sql_query( - "INSERT INTO backfill_jobs (did, status, scheduled_at) - VALUES ($1, 'pending', NOW()) - ON CONFLICT (did) - DO UPDATE SET - status = 'pending', - scheduled_at = NOW() - WHERE backfill_jobs.status != 'successful'" - ) - .bind::(did), - conn, - ) - .await?; - - Ok(()) -} - -/// Queue one or more DIDs for backfill -async fn queue_backfill( - State(state): State, - Json(request): Json, -) -> Result, Response> { - let mut conn = state.pool.get().await.map_err(internal_error)?; - - let mut queued = Vec::new(); - let mut errors = Vec::new(); - - for did in request.dids { - // Clean the DID - let Some(clean_did) = parakeet_db::allowlist::clean_did(&did) else { - errors.push(BackfillError { - did: did.clone(), - error: "Invalid DID format".to_string(), - }); - continue; - }; - - // Check if actor exists in database - let actor_exists = check_actor_exists(&mut conn, &clean_did).await; - match actor_exists { - Ok(false) => { - errors.push(BackfillError { - did: clean_did.clone(), - error: "Actor not found in database".to_string(), - }); - continue; - } - Err(e) => { - errors.push(BackfillError { - did: clean_did.clone(), - error: format!("Database error: {}", e), - }); - continue; - } - Ok(true) => {} // Actor exists, continue - } - - // Enqueue backfill job - match enqueue_backfill_job(&mut conn, &clean_did).await { - Ok(()) => { - queued.push(clean_did.clone()); - } - Err(e) => { - errors.push(BackfillError { - did: clean_did.clone(), - error: format!("Failed to enqueue backfill: {}", e), - }); - } - } - } - - Ok(Json(BackfillResponse { queued, errors })) -} - -/// Check if an actor exists in the database -async fn check_actor_exists( - conn: &mut diesel_async::AsyncPgConnection, - did: &str, -) -> Result { - use diesel::sql_types::{Bool, Text}; - - #[derive(diesel::QueryableByName)] - #[allow(unused_qualifications, reason = "Diesel QueryableByName macro generates unnecessary qualifications")] - struct ExistsResult { - #[diesel(sql_type = Bool)] - exists: bool, - } - - let result: ExistsResult = diesel_async::RunQueryDsl::get_result( - diesel::sql_query("SELECT EXISTS(SELECT 1 FROM actors WHERE did = $1)").bind::(did), - conn, - ) - .await?; - - Ok(result.exists) -} - -/// Helper function to convert any error into a 500 response -fn internal_error(err: E) -> Response -where - E: std::error::Error, -{ - ( - StatusCode::INTERNAL_SERVER_ERROR, - format!("Internal server error: {err}"), - ) - .into_response() -} diff --git a/parakeet/src/allowlist.rs b/parakeet/src/allowlist.rs deleted file mode 100644 index 95d97c69..00000000 --- a/parakeet/src/allowlist.rs +++ /dev/null @@ -1,237 +0,0 @@ -//! Allowlist functionality for DIDs with caching support -//! -//! This module provides functionality for checking, adding, and removing DIDs -//! from the allowlist, with an efficient caching layer to reduce database load. - -use chrono::NaiveDateTime; -use diesel::sql_types::{Bool, Integer, Nullable, Text, Timestamp}; -use diesel::{prelude::QueryableByName, result::QueryResult}; -use diesel_async::AsyncPgConnection; -use parakeet_db::allowlist::{clean_did, AllowlistEntry, CachedAllowlist}; -use tracing::debug; - -/// Raw database model for allowlist entries from SQL queries -#[derive(Debug, QueryableByName)] -#[diesel(check_for_backend(diesel::pg::Pg))] -#[allow(unused_qualifications, reason = "Diesel QueryableByName macro generates unnecessary qualifications")] -pub struct AllowlistEntryRaw { - #[diesel(sql_type = Integer)] - pub id: i32, - #[diesel(sql_type = Text)] - pub did: String, - #[diesel(sql_type = Timestamp)] - pub created_at: NaiveDateTime, - #[diesel(sql_type = Nullable)] - pub description: Option, -} - -impl From for AllowlistEntry { - fn from(raw: AllowlistEntryRaw) -> Self { - Self { - id: Some(raw.id), - did: raw.did, - description: raw.description, - created_at: Some(raw.created_at), - } - } -} - -/// A thread-safe cached allowlist that periodically refreshes from the database -#[derive(Clone)] -pub struct Allowlist { - /// The cached allowlist state - cache: CachedAllowlist, -} - -impl Allowlist { - /// Create a new empty cached allowlist - pub fn new() -> Self { - Self { - cache: CachedAllowlist::new(), - } - } - - /// Initialize the cache by loading the allowlist from the database - pub async fn initialize(&self, conn: &mut AsyncPgConnection) -> QueryResult<()> { - // Get all DIDs from the database - let entries = get_all(conn).await?; - - // Update our cache - extract just the DIDs - self.cache - .update_cache(entries.into_iter().map(|entry| entry.did).collect()); - - debug!( - "Allowlist cache initialized with {} entries", - self.cache.len() - ); - - Ok(()) - } - - /// Check if a DID is in the allowlist - pub async fn is_allowed(&self, _conn: &mut AsyncPgConnection, did: &str) -> QueryResult { - let Some(did) = clean_did(did) else { - return Ok(false); // Invalid DID format - }; - - // Check if the DID is in our cache (updated in real-time via LISTEN/NOTIFY) - Ok(self.cache.contains_did(&did)) - } - - /// Add a DID to the allowlist (both in the database and the cache) - pub async fn add_did( - &self, - conn: &mut AsyncPgConnection, - did: &str, - description: Option<&str>, - ) -> QueryResult { - let Some(did) = clean_did(did) else { - return Err(diesel::result::Error::DeserializationError( - "Invalid DID format".into(), - )); - }; - - // First, add to database - let result = add(conn, &did, description).await?; - - // If successful or already exists, add to our cache - self.cache.add_did(&did); - - debug!("Added DID {} to allowlist", did); - - // Return true if it was newly added, false if it already existed - Ok(result > 0) - } - - /// Remove a DID from the allowlist (both from the database and the cache) - pub async fn remove_did(&self, conn: &mut AsyncPgConnection, did: &str) -> QueryResult { - let Some(did) = clean_did(did) else { - return Err(diesel::result::Error::DeserializationError( - "Invalid DID format".into(), - )); - }; - - // First, remove from database - let result = remove(conn, &did).await?; - - // If successfully removed, update the cache - if result > 0 { - self.cache.remove_did(&did); - debug!("Removed DID {} from allowlist", did); - } - - Ok(result > 0) - } -} - -impl Default for Allowlist { - fn default() -> Self { - Self::new() - } -} - -/// Check if a DID is in the allowlist (direct database query) -/// -/// An actor is considered "allowlisted" if their sync_state is Synced, Dirty, or Processing. -pub async fn is_allowed(conn: &mut AsyncPgConnection, did: &str) -> QueryResult { - let Some(did) = clean_did(did) else { - return Ok(false); // Invalid DID format - }; - - #[derive(QueryableByName)] - #[allow(unused_qualifications, reason = "Diesel QueryableByName macro generates unnecessary qualifications")] - struct ExistsResult { - #[diesel(sql_type = Bool)] - exists: bool, - } - - diesel_async::RunQueryDsl::get_result( - diesel::sql_query( - "SELECT EXISTS( - SELECT 1 FROM actors - WHERE did = $1 AND sync_state IN ('synced', 'dirty', 'processing') - )" - ) - .bind::(&did), - conn, - ) - .await - .map(|result: ExistsResult| result.exists) -} - -/// Get all DIDs in the allowlist -/// -/// An actor is considered "allowlisted" if their sync_state is Synced, Dirty, or Processing. -/// Partial actors are not allowlisted (they only interact with allowlisted users). -pub async fn get_all(conn: &mut AsyncPgConnection) -> QueryResult> { - diesel_async::RunQueryDsl::load( - diesel::sql_query( - "SELECT id, did, last_indexed as created_at, NULL::text as description - FROM actors - WHERE sync_state IN ('synced', 'dirty', 'processing') - ORDER BY last_indexed DESC NULLS LAST", - ), - conn, - ) - .await -} - -/// Get a specific DID entry from the allowlist -pub async fn get_entry( - conn: &mut AsyncPgConnection, - did: &str, -) -> QueryResult> { - let Some(did) = clean_did(did) else { - return Err(diesel::result::Error::DeserializationError( - "Invalid DID format".into(), - )); - }; - - let results: Vec = diesel_async::RunQueryDsl::load( - diesel::sql_query( - "SELECT id, did, last_indexed as created_at, NULL::text as description - FROM actors - WHERE did = $1 AND sync_state IN ('synced', 'dirty', 'processing')" - ) - .bind::(&did), - conn, - ) - .await?; - - Ok(results.into_iter().next().map(Into::into)) -} - -/// Add a DID to the allowlist -/// -/// Sets the actor's sync_state to 'dirty' to mark them as allowlisted and needing backfill. -/// Creates the actor if they don't exist. -/// Returns the number of rows affected (1 if newly added/updated, 0 if already allowlisted) -pub async fn add( - conn: &mut AsyncPgConnection, - did: &str, - _description: Option<&str>, -) -> QueryResult { - // Note: description parameter kept for API compatibility but is no longer stored - // Admin notes should be kept in external documentation - diesel_async::RunQueryDsl::execute( - diesel::sql_query( - "INSERT INTO actors (did, status, sync_state) - VALUES ($1, 'active'::actor_status, 'dirty'::actor_sync_state) - ON CONFLICT (did) DO UPDATE - SET sync_state = 'dirty'::actor_sync_state - WHERE actors.sync_state = 'partial'::actor_sync_state" - ) - .bind::(did), - conn, - ) - .await -} - -/// Remove a DID from the allowlist -/// -/// Sets the actor's sync_state to 'partial' to mark them as no longer allowlisted. -/// This preserves the actor record (and any social graph data) while removing allowlist status. -/// Returns the number of rows affected (1 if removed, 0 if not allowlisted) -pub async fn remove(_conn: &mut AsyncPgConnection, _did: &str) -> QueryResult { - todo!("Use PostgreSQL queue to communicate back to consumer worker (see consumer/src/db/backfill_jobs.rs)") -} diff --git a/parakeet/src/id_cache_helpers.rs b/parakeet/src/id_cache_helpers.rs index 5634770c..3f651206 100644 --- a/parakeet/src/id_cache_helpers.rs +++ b/parakeet/src/id_cache_helpers.rs @@ -46,12 +46,10 @@ pub async fn get_actor_id_or_fetch( id: i32, #[diesel(sql_type = diesel::sql_types::Nullable)] handle: Option, - #[diesel(sql_type = diesel::sql_types::Text)] - sync_state: String, } let actor: ActorRow = diesel::sql_query( - "SELECT id, handle, sync_state::text + "SELECT id, handle FROM actors WHERE did = $1" ) @@ -74,13 +72,11 @@ pub async fn get_actor_id_or_fetch( })?; // Cache both forward and reverse lookups - let is_allowlisted = matches!(actor.sync_state.as_str(), "synced" | "dirty" | "processing"); - id_cache.set_actor_id( did.to_string(), CachedActor { actor_id: actor.id, - is_allowlisted, + is_allowlisted: false, // Allowlist concept removed - default to false }, ).await; @@ -237,12 +233,10 @@ pub async fn get_actor_ids_or_fetch( did: String, #[diesel(sql_type = diesel::sql_types::Nullable)] handle: Option, - #[diesel(sql_type = diesel::sql_types::Text)] - sync_state: String, } let db_results: Vec = diesel::sql_query( - "SELECT id, did, handle, sync_state::text + "SELECT id, did, handle FROM actors WHERE did = ANY($1)" ) @@ -259,13 +253,11 @@ pub async fn get_actor_ids_or_fetch( // Update cache and result for row in db_results { - let is_allowlisted = matches!(row.sync_state.as_str(), "synced" | "dirty" | "processing"); - id_cache.set_actor_id( row.did.clone(), CachedActor { actor_id: row.id, - is_allowlisted, + is_allowlisted: false, // Allowlist concept removed - default to false }, ).await; diff --git a/parakeet/src/lib.rs b/parakeet/src/lib.rs index 7ba149fb..d25bb3d9 100644 --- a/parakeet/src/lib.rs +++ b/parakeet/src/lib.rs @@ -5,8 +5,6 @@ use diesel_async::pooled_connection::deadpool::Pool; use diesel_async::AsyncPgConnection; use std::sync::Arc; -pub mod admin; -pub mod allowlist; pub mod cache_listener; pub mod config; pub mod entities; @@ -26,7 +24,6 @@ pub struct GlobalState { pub resolver: Arc, pub jwt: Arc, pub cdn: Arc, - pub allowlist: Arc, pub id_cache: Arc, pub rate_limiter: Arc, pub rate_limit_config: config::ConfigRateLimit, diff --git a/parakeet/src/main.rs b/parakeet/src/main.rs index ca2e0aa7..f313bc6f 100644 --- a/parakeet/src/main.rs +++ b/parakeet/src/main.rs @@ -84,16 +84,6 @@ async fn main() -> eyre::Result<()> { let did_doc = did_web_doc(&conf.service); let state = { - // Initialize the allowlist - let allowlist = Arc::new(allowlist::Allowlist::new()); - - // Populate the allowlist initially - let mut conn = pool.get().await?; - if let Err(e) = allowlist.initialize(&mut conn).await { - tracing::warn!("Failed to initialize allowlist: {}", e); - // Continue anyway - the cache will refresh on first use - } - // Initialize rate limiter (in-memory with DashMap) let rate_limiter = Arc::new(rate_limit::RateLimiter::new()); @@ -144,7 +134,6 @@ async fn main() -> eyre::Result<()> { resolver, jwt, cdn, - allowlist, id_cache, rate_limiter, rate_limit_config: conf.rate_limit.clone(), @@ -173,7 +162,6 @@ async fn main() -> eyre::Result<()> { "/.well-known/did.json", axum::routing::get(async || axum::Json(did_doc)), ) - .merge(admin::routes()) .layer(axum::middleware::from_fn_with_state( state.clone(), middleware::rate_limit::rate_limit_middleware, diff --git a/parakeet/src/xrpc/app_bsky/unspecced/mod.rs b/parakeet/src/xrpc/app_bsky/unspecced/mod.rs index 19439abb..b6db6a76 100644 --- a/parakeet/src/xrpc/app_bsky/unspecced/mod.rs +++ b/parakeet/src/xrpc/app_bsky/unspecced/mod.rs @@ -96,28 +96,9 @@ pub struct GetConfigResponse { /// Returns a static configuration response /// If authenticated user is not allowlisted, queues them for backfill pub async fn get_config( - State(state): State, - maybe_auth: Option, + State(_state): State, + _maybe_auth: Option, ) -> Json { - // If user is authenticated, check if they need backfill - if let Some(AtpAuth(viewer_did)) = maybe_auth { - // Check allowlist cache (in-memory bloom filter + hashset, very fast) - // Note: is_allowed doesn't actually use the conn, just checks the cache - let is_allowlisted = match state.pool.get().await { - Ok(mut conn) => state.allowlist.is_allowed(&mut conn, &viewer_did).await.unwrap_or(false), - Err(_) => false, // If can't get connection, assume not allowlisted - }; - - if !is_allowlisted { - // TODO: Implement PostgreSQL-based backfill queueing (see consumer/src/db/backfill_jobs.rs) - // For now, just log that user is not allowlisted - tracing::debug!( - did = %viewer_did, - "Authenticated non-allowlisted user detected (backfill queueing not yet implemented)" - ); - } - } - let live_now = vec![ // LiveNowEntry { // did: "did:plc:".to_string(), diff --git a/parakeet/src/xrpc/extract.rs b/parakeet/src/xrpc/extract.rs index 08b56ccb..42c2e6ac 100644 --- a/parakeet/src/xrpc/extract.rs +++ b/parakeet/src/xrpc/extract.rs @@ -85,36 +85,7 @@ impl FromRequestParts for AtpAuth { let jwt_result = state.jwt.resolve_and_verify_jwt(hdr.token(), None).await; match jwt_result { - Some(claims) => { - let allowlist_enabled = false; - if !allowlist_enabled { - return Ok(Self(claims.iss)); - } - // Check if the DID is in the allowlist from DB - let mut conn = match state.pool.get().await { - Ok(conn) => conn, - Err(e) => { - tracing::error!("Failed to get DB connection for allowlist check: {}", e); - return Err(( - StatusCode::INTERNAL_SERVER_ERROR, - "database error".to_owned(), - )); - } - }; - - let allowlist_result = state.allowlist.is_allowed(&mut conn, &claims.iss).await; - match allowlist_result { - Ok(true) => Ok(Self(claims.iss)), - Ok(false) => Err((StatusCode::FORBIDDEN, "forbidden".to_owned())), - Err(e) => { - tracing::error!("Allowlist check failed: {}", e); - Err(( - StatusCode::INTERNAL_SERVER_ERROR, - "allowlist check failed".to_owned(), - )) - } - } - } + Some(claims) => Ok(Self(claims.iss)), None => Err((StatusCode::INTERNAL_SERVER_ERROR, "JWT error".to_owned())), } } @@ -139,39 +110,7 @@ impl OptionalFromRequestParts for AtpAuth { let jwt_result = state.jwt.resolve_and_verify_jwt(hdr.token(), None).await; match jwt_result { - Some(claims) => { - // For optional auth, allowlist checking is disabled - // Non-allowlisted users just get None instead of 403 - let allowlist_enabled = false; - if !allowlist_enabled { - return Ok(Some(Self(claims.iss))); - } - - // Check if the DID is in the allowlist from DB - let mut conn = match state.pool.get().await { - Ok(conn) => conn, - Err(e) => { - tracing::error!("Failed to get DB connection for allowlist check: {}", e); - return Err(( - StatusCode::INTERNAL_SERVER_ERROR, - "database error".to_owned(), - )); - } - }; - - let allowlist_result = state.allowlist.is_allowed(&mut conn, &claims.iss).await; - match allowlist_result { - Ok(true) => Ok(Some(Self(claims.iss))), - Ok(false) => Ok(None), // Return None instead of 403 for optional auth - Err(e) => { - tracing::error!("Allowlist check failed: {}", e); - Err(( - StatusCode::INTERNAL_SERVER_ERROR, - "allowlist check failed".to_owned(), - )) - } - } - } + Some(claims) => Ok(Some(Self(claims.iss))), None => Err((StatusCode::INTERNAL_SERVER_ERROR, "JWT error".to_owned())), } }