diff --git a/parakeet/src/common/cache_id_helpers.rs b/parakeet/src/common/cache_id_helpers.rs deleted file mode 100644 index e5883f2b..00000000 --- a/parakeet/src/common/cache_id_helpers.rs +++ /dev/null @@ -1,276 +0,0 @@ -//! IdCache helper functions with automatic database fallback -//! -//! These functions combine IdCache lookups with database queries, -//! automatically fetching and caching missing entries. - -use crate::common::errors::{Error, XrpcResult}; -use diesel::sql_types::{Array, Integer, Text}; -use diesel_async::pooled_connection::deadpool::Pool; -use diesel_async::{AsyncPgConnection, RunQueryDsl}; -use parakeet_db::id_cache::{CachedActor, CachedActorData, IdCache}; -use std::collections::HashMap; -use std::sync::Arc; - -/// Get actor_id for a DID, fetching from database if not cached -/// -/// This automatically: -/// 1. Checks IdCache for the DID -/// 2. If miss, queries database -/// 3. Updates cache with result -/// 4. Returns the actor_id -/// -/// # Errors -/// Returns error if DID not found in database -pub async fn get_actor_id_or_fetch( - pool: &Pool, - id_cache: &Arc, - did: &str, -) -> XrpcResult { - // Try cache first - if let Some(actor_id) = id_cache.get_actor_id_only(did).await { - return Ok(actor_id); - } - - // Cache miss - query database - let mut conn = pool.get().await.map_err(|e| { - Error::new( - axum::http::StatusCode::INTERNAL_SERVER_ERROR, - "DatabaseError", - Some(format!("Failed to get database connection: {}", e)), - ) - })?; - - #[derive(diesel::QueryableByName)] - struct ActorRow { - #[diesel(sql_type = Integer)] - id: i32, - #[diesel(sql_type = diesel::sql_types::Nullable)] - handle: Option, - } - - let actor: ActorRow = diesel::sql_query( - "SELECT id, handle - FROM actors - WHERE did = $1" - ) - .bind::(did) - .get_result(&mut conn) - .await - .map_err(|e| match e { - diesel::result::Error::NotFound => { - Error::new( - axum::http::StatusCode::NOT_FOUND, - "ActorNotFound", - Some(format!("Actor not found: {}", did)), - ) - } - _ => Error::new( - axum::http::StatusCode::INTERNAL_SERVER_ERROR, - "DatabaseError", - Some(format!("Failed to query actor: {}", e)), - ), - })?; - - // Cache both forward and reverse lookups - id_cache.set_actor_id( - did.to_string(), - CachedActor { - actor_id: actor.id, - is_allowlisted: false, // Allowlist concept removed - default to false - }, - ).await; - - id_cache.set_actor_data( - actor.id, - CachedActorData { - did: did.to_string(), - handle: actor.handle, - }, - ).await; - - Ok(actor.id) -} - -/// Get DIDs for multiple actor_ids, fetching from database for cache misses -/// -/// This automatically: -/// 1. Checks IdCache for each actor_id -/// 2. For cache misses, queries database in batch -/// 3. Updates cache with results -/// 4. Returns HashMap of actor_id → DID -/// -/// # Returns -/// HashMap with all requested actor_ids (omits IDs not found in database) -pub async fn get_actor_dids_or_fetch( - pool: &Pool, - id_cache: &Arc, - actor_ids: &[i32], -) -> XrpcResult> { - if actor_ids.is_empty() { - return Ok(HashMap::new()); - } - - // Try cache first - let cached = id_cache.get_actor_data_many(actor_ids).await; - let mut result: HashMap = cached - .into_iter() - .map(|(id, data)| (id, data.did)) - .collect(); - - // Find cache misses - let missing: Vec = actor_ids - .iter() - .filter(|id| !result.contains_key(id)) - .copied() - .collect(); - - if missing.is_empty() { - return Ok(result); - } - - // Query database for misses - let mut conn = pool.get().await.map_err(|e| { - Error::new( - axum::http::StatusCode::INTERNAL_SERVER_ERROR, - "DatabaseError", - Some(format!("Failed to get database connection: {}", e)), - ) - })?; - - #[derive(diesel::QueryableByName)] - struct ActorRow { - #[diesel(sql_type = Integer)] - id: i32, - #[diesel(sql_type = Text)] - did: String, - #[diesel(sql_type = diesel::sql_types::Nullable)] - handle: Option, - } - - let db_results: Vec = diesel::sql_query( - "SELECT id, did, handle - FROM actors - WHERE id = ANY($1)" - ) - .bind::, _>(&missing) - .load(&mut conn) - .await - .map_err(|e| { - Error::new( - axum::http::StatusCode::INTERNAL_SERVER_ERROR, - "DatabaseError", - Some(format!("Failed to resolve actor DIDs: {}", e)), - ) - })?; - - // Update cache and result - for row in db_results { - id_cache.set_actor_data( - row.id, - CachedActorData { - did: row.did.clone(), - handle: row.handle, - }, - ).await; - - result.insert(row.id, row.did); - } - - Ok(result) -} - -/// Get actor_ids for multiple DIDs, fetching from database for cache misses -/// -/// This automatically: -/// 1. Checks IdCache for each DID -/// 2. For cache misses, queries database in batch -/// 3. Updates cache with results -/// 4. Returns HashMap of DID → actor_id -/// -/// # Returns -/// HashMap with all requested DIDs (omits DIDs not found in database) -pub async fn get_actor_ids_or_fetch( - pool: &Pool, - id_cache: &Arc, - dids: &[String], -) -> XrpcResult> { - if dids.is_empty() { - return Ok(HashMap::new()); - } - - // Try cache first - let cached = id_cache.get_actor_ids(dids).await; - let mut result: HashMap = cached - .into_iter() - .map(|(did, cached)| (did, cached.actor_id)) - .collect(); - - // Find cache misses - let missing: Vec<&str> = dids - .iter() - .filter(|did| !result.contains_key(did.as_str())) - .map(|s| s.as_str()) - .collect(); - - if missing.is_empty() { - return Ok(result); - } - - // Query database for misses - let mut conn = pool.get().await.map_err(|e| { - Error::new( - axum::http::StatusCode::INTERNAL_SERVER_ERROR, - "DatabaseError", - Some(format!("Failed to get database connection: {}", e)), - ) - })?; - - #[derive(diesel::QueryableByName)] - struct ActorRow { - #[diesel(sql_type = Integer)] - id: i32, - #[diesel(sql_type = Text)] - did: String, - #[diesel(sql_type = diesel::sql_types::Nullable)] - handle: Option, - } - - let db_results: Vec = diesel::sql_query( - "SELECT id, did, handle - FROM actors - WHERE did = ANY($1)" - ) - .bind::, _>(&missing) - .load(&mut conn) - .await - .map_err(|e| { - Error::new( - axum::http::StatusCode::INTERNAL_SERVER_ERROR, - "DatabaseError", - Some(format!("Failed to resolve actor IDs: {}", e)), - ) - })?; - - // Update cache and result - for row in db_results { - id_cache.set_actor_id( - row.did.clone(), - CachedActor { - actor_id: row.id, - is_allowlisted: false, // Allowlist concept removed - default to false - }, - ).await; - - id_cache.set_actor_data( - row.id, - CachedActorData { - did: row.did.clone(), - handle: row.handle, - }, - ).await; - - result.insert(row.did, row.id); - } - - Ok(result) -} diff --git a/parakeet/src/lib.rs b/parakeet/src/lib.rs index b11c2963..09f9ba43 100644 --- a/parakeet/src/lib.rs +++ b/parakeet/src/lib.rs @@ -10,7 +10,6 @@ pub mod common { //! Common infrastructure and utilities shared across the application pub mod auth; - pub mod cache_id_helpers; pub mod cache_listener; pub mod cache_timeline; pub mod errors; @@ -19,7 +18,6 @@ pub mod common { // Re-export commonly used items pub use auth::{AtpAcceptLabelers, AtpAuth, JwtVerifier}; - pub use cache_id_helpers::{get_actor_id_or_fetch, get_actor_dids_or_fetch, get_actor_ids_or_fetch}; pub use cache_listener::spawn_cache_listener; pub use cache_timeline::{AuthorFeedCache, TimelineCache}; pub use errors::{Error, XrpcResult}; diff --git a/parakeet/src/xrpc/app_bsky/actor.rs b/parakeet/src/xrpc/app_bsky/actor.rs index ff175a89..24e6f873 100644 --- a/parakeet/src/xrpc/app_bsky/actor.rs +++ b/parakeet/src/xrpc/app_bsky/actor.rs @@ -258,14 +258,17 @@ pub async fn get_suggestions( .into_response()); } - // Convert DIDs to actor_ids for efficient profile loading - let did_to_actor_id = crate::common::get_actor_ids_or_fetch( - &state.pool, - &state.id_cache, - &all_dids, - ).await.unwrap_or_default(); - - let actor_ids: Vec = did_to_actor_id.values().copied().collect(); + // Convert DIDs to actor_ids for efficient profile loading using ProfileEntity + let actor_ids = state.profile_entity.resolve_identifiers(&all_dids) + .await + .unwrap_or_default(); + + // Create a mapping for compatibility with existing code + let did_to_actor_id: std::collections::HashMap = all_dids + .iter() + .cloned() + .zip(actor_ids.iter().copied()) + .collect(); // Load profiles to check quality using ProfileEntity let profiles = state.profile_entity diff --git a/parakeet/src/xrpc/app_bsky/graph/thread_mutes.rs b/parakeet/src/xrpc/app_bsky/graph/thread_mutes.rs index 22be83cc..1821b263 100644 --- a/parakeet/src/xrpc/app_bsky/graph/thread_mutes.rs +++ b/parakeet/src/xrpc/app_bsky/graph/thread_mutes.rs @@ -18,12 +18,9 @@ pub async fn mute_thread( use crate::common::errors::Error; let mut conn = state.pool.get().await?; - // Resolve authenticated user's actor_id via IdCache - let actor_id = crate::common::get_actor_id_or_fetch( - &state.pool, - &state.id_cache, - &auth.0, - ).await?; + // Resolve authenticated user's actor_id using ProfileEntity + let actor_id = state.profile_entity.resolve_identifier(&auth.0).await + .map_err(|_| Error::actor_not_found(&auth.0))?; // Parse thread root URI and resolve to post_id let parts = form.root.strip_prefix("at://").ok_or_else(|| Error::invalid_request(Some("Invalid AT URI".into())))? @@ -39,11 +36,8 @@ pub async fn mute_thread( let rkey_bigint = parakeet_db::tid_util::decode_tid(rkey_str) .map_err(|_| Error::invalid_request(Some("Invalid TID in root URI".into())))?; - let root_post_actor_id = crate::common::get_actor_id_or_fetch( - &state.pool, - &state.id_cache, - root_did, - ).await?; + let root_post_actor_id = state.profile_entity.resolve_identifier(root_did).await + .map_err(|_| Error::actor_not_found(root_did))?; // Append to thread_mutes array (off-protocol, managed directly by AppView) // Deduplicates based on root_post_actor_id + root_post_rkey @@ -78,12 +72,9 @@ pub async fn unmute_thread( use crate::common::errors::Error; let mut conn = state.pool.get().await?; - // Resolve authenticated user's actor_id via IdCache - let actor_id = crate::common::get_actor_id_or_fetch( - &state.pool, - &state.id_cache, - &auth.0, - ).await?; + // Resolve authenticated user's actor_id using ProfileEntity + let actor_id = state.profile_entity.resolve_identifier(&auth.0).await + .map_err(|_| Error::actor_not_found(&auth.0))?; // Parse thread root URI and resolve to post_id let parts = form.root.strip_prefix("at://").ok_or_else(|| Error::invalid_request(Some("Invalid AT URI".into())))? @@ -99,11 +90,8 @@ pub async fn unmute_thread( let rkey_bigint = parakeet_db::tid_util::decode_tid(rkey_str) .map_err(|_| Error::invalid_request(Some("Invalid TID in root URI".into())))?; - let root_post_actor_id = crate::common::get_actor_id_or_fetch( - &state.pool, - &state.id_cache, - root_did, - ).await?; + let root_post_actor_id = state.profile_entity.resolve_identifier(root_did).await + .map_err(|_| Error::actor_not_found(root_did))?; // Remove from thread_mutes array (off-protocol, managed directly by AppView) diesel_async::RunQueryDsl::execute( diff --git a/parakeet/src/xrpc/app_bsky/notification.rs b/parakeet/src/xrpc/app_bsky/notification.rs index b361ad1e..f191ab5f 100644 --- a/parakeet/src/xrpc/app_bsky/notification.rs +++ b/parakeet/src/xrpc/app_bsky/notification.rs @@ -192,21 +192,21 @@ pub async fn list_notifications( } } - // Resolve actor_ids to DIDs using IdCache helper (with automatic database fallback) + // Resolve actor_ids to DIDs using ProfileEntity let actor_id_vec: Vec = actor_ids_to_resolve.into_iter().collect(); - let actor_did_map = crate::common::get_actor_dids_or_fetch( - &state.pool, - &state.id_cache, - &actor_id_vec, - ) - .await - .map_err(|_| { - Error::new( - axum::http::StatusCode::INTERNAL_SERVER_ERROR, + let profiles = state.profile_entity.get_profiles_by_ids(&actor_id_vec) + .await + .map_err(|_| { + Error::new( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, "InternalServerError".to_string(), Some("Database error resolving actor DIDs".to_string()), ) })?; + let actor_did_map: std::collections::HashMap = profiles + .into_iter() + .map(|actor| (actor.id, actor.did)) + .collect(); tracing::info!(" → Resolve actor_ids→DIDs (IdCache): {:.1}ms ({} actors)", resolve_start.elapsed().as_secs_f64() * 1000.0, actor_did_map.len()); // Extract author DIDs for profile hydration @@ -275,20 +275,20 @@ pub async fn list_notifications( let resolve_additional_start = std::time::Instant::now(); let actor_id_vec: Vec = additional_actor_ids.into_iter().collect(); - // Use IdCache helper for additional actors - let additional_dids = crate::common::get_actor_dids_or_fetch( - &state.pool, - &state.id_cache, - &actor_id_vec, - ) - .await - .map_err(|_| { - Error::new( - axum::http::StatusCode::INTERNAL_SERVER_ERROR, - "InternalServerError".to_string(), - Some("Database error resolving parent/root actor DIDs".to_string()), - ) - })?; + // Use ProfileEntity for additional actors + let additional_profiles = state.profile_entity.get_profiles_by_ids(&actor_id_vec) + .await + .map_err(|_| { + Error::new( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + "InternalServerError".to_string(), + Some("Database error resolving parent/root actor DIDs".to_string()), + ) + })?; + let additional_dids: std::collections::HashMap = additional_profiles + .into_iter() + .map(|actor| (actor.id, actor.did)) + .collect(); tracing::info!(" → Resolve parent/root actor_ids (IdCache): {:.1}ms ({} actors)", resolve_additional_start.elapsed().as_secs_f64() * 1000.0, additional_dids.len()); // Merge into actor_did_map diff --git a/parakeet/src/xrpc/app_bsky/unspecced/mod.rs b/parakeet/src/xrpc/app_bsky/unspecced/mod.rs index 08efb92e..b81e9e32 100644 --- a/parakeet/src/xrpc/app_bsky/unspecced/mod.rs +++ b/parakeet/src/xrpc/app_bsky/unspecced/mod.rs @@ -230,16 +230,19 @@ pub async fn get_suggested_users( return Ok(Json(GetSuggestedUsersResponse { actors: Vec::new() })); } - // Convert DIDs to actor_ids for profile loading - let actor_id_map = match crate::common::get_actor_ids_or_fetch( - &state.pool, - &state.id_cache, - &all_dids, - ).await { - Ok(map) => map, + // Convert DIDs to actor_ids for profile loading using ProfileEntity + let actor_ids = match state.profile_entity.resolve_identifiers(&all_dids).await { + Ok(ids) => ids, Err(_) => return Ok(Json(GetSuggestedUsersResponse { actors: Vec::new() })), }; + // Create a mapping for compatibility with existing code + let actor_id_map: std::collections::HashMap = all_dids + .iter() + .cloned() + .zip(actor_ids.iter().copied()) + .collect(); + let actor_ids: Vec = actor_id_map.values().copied().collect(); // Load profiles to check quality using ProfileEntity @@ -563,7 +566,7 @@ pub async fn get_suggested_starter_packs( // Hydrate starter packs maintaining order let (maybe_did, maybe_actor_id) = if let Some(auth) = maybe_auth { let did = auth.0.clone(); - let actor_id = crate::common::get_actor_id_or_fetch(&state.pool, &state.id_cache, &did).await.ok(); + let actor_id = state.profile_entity.resolve_identifier(&did).await.ok(); (Some(did), actor_id) } else { (None, None) diff --git a/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/other_replies.rs b/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/other_replies.rs index 9aafaefc..b81fc9ab 100644 --- a/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/other_replies.rs +++ b/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/other_replies.rs @@ -59,7 +59,7 @@ pub async fn get_post_thread_other_v2( let is_authenticated = maybe_auth.is_some(); let (maybe_did, maybe_actor_id) = if let Some(auth) = maybe_auth { let did = auth.0.clone(); - let actor_id = crate::common::get_actor_id_or_fetch(&state.pool, &state.id_cache, &did).await.ok(); + let actor_id = state.profile_entity.resolve_identifier(&did).await.ok(); (Some(did), actor_id) } else { (None, None) diff --git a/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/post_thread.rs b/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/post_thread.rs index c3e9ba59..1c1b2b12 100644 --- a/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/post_thread.rs +++ b/parakeet/src/xrpc/app_bsky/unspecced/thread_v2/post_thread.rs @@ -28,7 +28,7 @@ pub async fn get_post_thread_v2( let is_authenticated = maybe_auth.is_some(); let (maybe_did, maybe_actor_id) = if let Some(auth) = maybe_auth { let did = auth.0.clone(); - let actor_id = crate::common::get_actor_id_or_fetch(&state.pool, &state.id_cache, &did).await.ok(); + let actor_id = state.profile_entity.resolve_identifier(&did).await.ok(); (Some(did), actor_id) } else { (None, None) diff --git a/parakeet/src/xrpc/community_lexicon/bookmarks.rs b/parakeet/src/xrpc/community_lexicon/bookmarks.rs index f99f3d36..5a4deeae 100644 --- a/parakeet/src/xrpc/community_lexicon/bookmarks.rs +++ b/parakeet/src/xrpc/community_lexicon/bookmarks.rs @@ -30,12 +30,9 @@ pub async fn get_actor_bookmarks( let limit = query.limit.unwrap_or(50).clamp(1, 100); - // Resolve DID to actor_id via IdCache - let actor_id = crate::common::get_actor_id_or_fetch( - &state.pool, - &state.id_cache, - &auth.0, - ).await?; + // Resolve DID to actor_id using ProfileEntity + let actor_id = state.profile_entity.resolve_identifier(&auth.0).await + .map_err(|_| crate::common::errors::Error::actor_not_found(&auth.0))?; // Note: tags filtering not supported in current schema if query.tags.is_some() {