diff --git a/parakeet/src/entity_cache.rs b/parakeet/src/entity_cache.rs index 459c4456..49fa6bec 100644 --- a/parakeet/src/entity_cache.rs +++ b/parakeet/src/entity_cache.rs @@ -123,8 +123,8 @@ impl ProfileCache { /// Get multiple profiles with caching /// - /// This is the recommended way to get multiple profiles, ensuring each one - /// uses the cache individually + /// This is the recommended way to get multiple profiles, using batch hydration + /// for all cache misses to avoid N+1 queries pub async fn get_or_hydrate_batch( &self, dids: Vec, @@ -132,18 +132,64 @@ impl ProfileCache { id_cache: &Arc, hydrator: &StatefulHydrator<'_>, ) -> Vec { - let mut profiles = Vec::with_capacity(dids.len()); + let mut cached_profiles = Vec::new(); + let mut uncached_requests = Vec::new(); + let mut did_to_actor_id = std::collections::HashMap::new(); + // First pass: check cache for all DIDs for did in dids { // Get actor_id for each DID if let Ok(actor_id) = id_cache_helpers::get_actor_id_or_fetch(pool, id_cache, &did).await { - if let Some(profile) = self.get_or_hydrate(actor_id, did, hydrator).await { - profiles.push(profile); + did_to_actor_id.insert(did.clone(), actor_id); + + // Check cache + if let Some(profile) = self.cache.get(&actor_id).await { + tracing::debug!(actor_id, "Profile cache hit in batch"); + cached_profiles.push(profile); + } else { + // Track cache misses for batch hydration + uncached_requests.push(did); } } } - profiles + // Batch hydrate all uncached profiles at once + if !uncached_requests.is_empty() { + let cache_hit_count = cached_profiles.len(); + let miss_count = uncached_requests.len(); + let hit_rate = if cache_hit_count + miss_count > 0 { + (cache_hit_count as f64 / (cache_hit_count + miss_count) as f64) * 100.0 + } else { + 0.0 + }; + + tracing::debug!( + "Profile cache batch: {} hits, {} misses ({:.1}% hit rate), hydrating misses", + cache_hit_count, miss_count, hit_rate + ); + + // Build a map of actor_id to DID for the uncached requests + let uncached_actor_ids: Vec<(i32, String)> = uncached_requests + .iter() + .filter_map(|did| { + did_to_actor_id.get(did).map(|&actor_id| (actor_id, did.clone())) + }) + .collect(); + + // Use the batch hydration method for all misses at once (passing actor_ids) + let hydrated = hydrator.hydrate_profiles_detailed_by_id(uncached_actor_ids.clone()).await; + + // Store in cache and collect results + for (actor_id, _did) in uncached_actor_ids { + if let Some(profile) = hydrated.get(&actor_id) { + // Cache the hydrated profile + self.cache.insert(actor_id, profile.clone()).await; + cached_profiles.push(profile.clone()); + } + } + } + + cached_profiles } pub async fn invalidate(&self, actor_id: i32) { diff --git a/parakeet/src/hydration/mod.rs b/parakeet/src/hydration/mod.rs index b59fd141..7b410ddf 100644 --- a/parakeet/src/hydration/mod.rs +++ b/parakeet/src/hydration/mod.rs @@ -176,38 +176,22 @@ impl StatefulHydrator<'_> { labels } - /// Get labels for multiple URIs using actor_ids (optimized version - avoids decompressing actors table) + /// Get labels for multiple URIs /// - /// This resolves labeler DIDs → actor_ids via IdCache first, avoiding the actors table join. + /// For now, fallback to the regular load_many function until we properly extract labels from loaded data async fn get_label_many_by_actor_ids( &self, uris: &[String], ) -> HashMap> { - // Resolve labeler DIDs → actor_ids via IdCache - let labeler_dids: Vec = self - .accept_labelers - .iter() - .map(|v| v.labeler.clone()) - .collect(); - - let mut labeler_actor_ids = Vec::new(); - for did in &labeler_dids { - if let Some(cached) = self.loaders.label.id_cache().get_actor_id(did).await { - labeler_actor_ids.push(cached.actor_id); - } - } - - if labeler_actor_ids.is_empty() { - return HashMap::new(); - } - + // TODO: Extract labels from already-loaded post/actor data instead of querying separately + // Posts and actors already have labels field loaded self.loaders .label - .load_many_by_actor_ids(uris, &labeler_actor_ids) + .load_many(uris, &self.accept_labelers) .await } - /// Get profile labels using actor_ids (optimized version) + /// Get profile labels async fn get_profile_label_many_by_actor_ids( &self, uris: &[String], @@ -218,6 +202,7 @@ impl StatefulHydrator<'_> { .map(|did| format!("at://{did}/app.bsky.actor.profile/self")), ); + // TODO: Extract labels from already-loaded actor data instead of querying separately let mut labels = self.get_label_many_by_actor_ids(&uris_full).await; for did in uris { diff --git a/parakeet/src/hydration/posts/mod.rs b/parakeet/src/hydration/posts/mod.rs index 6e588fa5..92f289a5 100644 --- a/parakeet/src/hydration/posts/mod.rs +++ b/parakeet/src/hydration/posts/mod.rs @@ -225,15 +225,65 @@ impl StatefulHydrator<'_> { }) .collect::>() }; + // Extract labels directly from the loaded posts instead of querying separately let labels_future = async { let start = std::time::Instant::now(); - let result = self.get_label_many_by_actor_ids(&post_uris).await; - let elapsed = start.elapsed().as_secs_f64() * 1000.0; - tracing::info!(" → Labels: {:.1} ms ({} posts)", elapsed, post_uri_count); - if elapsed > 10.0 { - tracing::warn!(" → Slow labels: {:.1} ms", elapsed); + + // Build a map of URI -> labels from the posts we already loaded + let mut labels_map: HashMap> = HashMap::new(); + + // Get allowed labeler actor IDs + let labeler_dids: Vec = self + .accept_labelers + .iter() + .map(|v| v.labeler.clone()) + .collect(); + + let mut labeler_actor_ids = Vec::new(); + for did in &labeler_dids { + if let Some(cached) = self.loaders.label.id_cache().get_actor_id(did).await { + labeler_actor_ids.push(cached.actor_id); + } } - result + + // Extract labels from each post + for (uri, (post, _, _)) in &posts_with_stats { + if let Some(post_labels) = &post.post.labels { + let mut labels_for_uri = Vec::new(); + + for label_opt in post_labels { + if let Some(label) = label_opt { + // Only include labels from allowed labelers + if labeler_actor_ids.contains(&label.labeler_actor_id) { + // Resolve labeler DID + if let Some(actor_data) = self.loaders.label.id_cache().get_actor_data(label.labeler_actor_id).await { + labels_for_uri.push(parakeet_db::models::Label { + labeler_actor_id: label.labeler_actor_id, + label: label.label.clone(), + uri: uri.clone(), + self_label: false, // Post labels are not self-labels + cid: None, // Not stored in denormalized structure + negated: label.negated, + expires: label.expires, + sig: None, // Not stored in denormalized structure + created_at: label.created_at, + labeler: actor_data.did, + }); + } + } + } + } + + if !labels_for_uri.is_empty() { + labels_map.insert(uri.clone(), labels_for_uri); + } + } + } + + let elapsed = start.elapsed().as_secs_f64() * 1000.0; + tracing::info!(" → Labels extracted from posts: {:.1} ms ({} posts)", elapsed, post_uri_count); + + labels_map }; let viewer_future = async { let start = std::time::Instant::now(); diff --git a/parakeet/src/hydration/profile/mod.rs b/parakeet/src/hydration/profile/mod.rs index 5fbaa6fd..a721e8c9 100644 --- a/parakeet/src/hydration/profile/mod.rs +++ b/parakeet/src/hydration/profile/mod.rs @@ -66,10 +66,17 @@ impl super::StatefulHydrator<'_> { } pub async fn hydrate_profile_basic(&self, did: String) -> Option { + // Convert DID to actor_id first + let actor_id = crate::id_cache_helpers::get_actor_id_or_fetch( + self.loaders.profile_state.pool(), + self.loaders.profile_state.id_cache(), + &did, + ).await.ok()?; + let viewer = self.get_profile_viewer_state(&did).await; let verif = self.loaders.verification.load(did.clone()).await; let stats = self.loaders.profile_stats.load(did.clone()).await; - let profile_info = self.loaders.profile.load(did.clone()).await?; + let profile_info = self.loaders.profile_by_id.load(actor_id).await?; // Extract and convert inline labels let labels = if let Some(ref label_records) = profile_info.8 { @@ -243,11 +250,25 @@ impl super::StatefulHydrator<'_> { } pub async fn hydrate_profile(&self, did: String) -> Option { - let labels = self.get_profile_label(&did).await; + // Convert DID to actor_id first + let actor_id = crate::id_cache_helpers::get_actor_id_or_fetch( + self.loaders.profile_state.pool(), + self.loaders.profile_state.id_cache(), + &did, + ).await.ok()?; + + let profile_info = self.loaders.profile_by_id.load(actor_id).await?; + + // Extract labels from loaded profile + let labels = if let Some(ref label_records) = profile_info.8 { + self.convert_actor_labels(&did, label_records).await + } else { + vec![] + }; + let viewer = self.get_profile_viewer_state(&did).await; let verif = self.loaders.verification.load(did.clone()).await; let stats = self.loaders.profile_stats.load(did.clone()).await; - let profile_info = self.loaders.profile.load(did).await?; Some(build_profile( profile_info, @@ -260,22 +281,49 @@ impl super::StatefulHydrator<'_> { } pub async fn hydrate_profiles(&self, dids: Vec) -> HashMap { - let labels = self.get_profile_label_many(&dids).await; + // Convert DIDs to actor_ids first + let actor_id_map = match crate::id_cache_helpers::get_actor_ids_or_fetch( + self.loaders.profile_state.pool(), + self.loaders.profile_state.id_cache(), + &dids, + ).await { + Ok(map) => map, + Err(_) => return HashMap::new(), + }; + + let actor_ids: Vec = actor_id_map.values().copied().collect(); + + let profiles = self.loaders.profile_by_id.load_many(actor_ids).await; + + // Extract labels from loaded profiles + let mut labels: HashMap> = HashMap::new(); + for (did, _, _, _, _, _, _, _, label_records) in profiles.values() { + if let Some(records) = label_records { + let converted = self.convert_actor_labels(did, records).await; + if !converted.is_empty() { + labels.insert(did.clone(), converted); + } + } + } + let viewers = self.get_profile_viewer_states(&dids).await; let verif = self.loaders.verification.load_many(dids.clone()).await; let stats = self.loaders.profile_stats.load_many(dids.clone()).await; - let profiles = self.loaders.profile.load_many(dids).await; + + // Convert actor_id-keyed results back to DID-keyed + let id_to_did: HashMap = actor_id_map.iter().map(|(did, id)| (*id, did.clone())).collect(); profiles .into_iter() - .map(|(k, profile_info)| { - let labels = labels.get(&k).cloned().unwrap_or_default(); - let verif = verif.get(&k).cloned(); - let viewer = viewers.get(&k).cloned(); - let stats = stats.get(&k).copied(); + .filter_map(|(actor_id, profile_info)| { + let did = id_to_did.get(&actor_id)?; + let labels = labels.get(did).cloned().unwrap_or_default(); + let verif = verif.get(did).cloned(); + let viewer = viewers.get(did).cloned(); + let stats = stats.get(did).copied(); let v = build_profile(profile_info, stats, labels, verif, viewer, &self.cdn); - (k, v) + Some((did.clone(), v)) }) .collect() } @@ -289,11 +337,25 @@ impl super::StatefulHydrator<'_> { note = "Use ProfileCache::get_or_hydrate() to ensure caching. Direct hydration bypasses the cache." )] pub async fn hydrate_profile_detailed(&self, did: String) -> Option { - let labels = self.get_profile_label(&did).await; + // Convert DID to actor_id first + let actor_id = crate::id_cache_helpers::get_actor_id_or_fetch( + self.loaders.profile_state.pool(), + self.loaders.profile_state.id_cache(), + &did, + ).await.ok()?; + + let profile_info = self.loaders.profile_by_id.load(actor_id).await?; + + // Extract labels from loaded profile + let labels = if let Some(ref label_records) = profile_info.8 { + self.convert_actor_labels(&did, label_records).await + } else { + vec![] + }; + let viewer = self.get_profile_viewer_state(&did).await; let verif = self.loaders.verification.load(did.clone()).await; let stats = self.loaders.profile_stats.load(did.clone()).await; - let profile_info = self.loaders.profile.load(did).await?; Some(build_detailed( profile_info, @@ -309,22 +371,95 @@ impl super::StatefulHydrator<'_> { &self, dids: Vec, ) -> HashMap { - let labels = self.get_profile_label_many(&dids).await; + // Convert DIDs to actor_ids first + let actor_id_map = match crate::id_cache_helpers::get_actor_ids_or_fetch( + self.loaders.profile_state.pool(), + self.loaders.profile_state.id_cache(), + &dids, + ).await { + Ok(map) => map, + Err(_) => return HashMap::new(), + }; + + let actor_ids: Vec = actor_id_map.values().copied().collect(); + + let profiles = self.loaders.profile_by_id.load_many(actor_ids).await; + + // Extract labels from loaded profiles + let mut labels: HashMap> = HashMap::new(); + for (did, _, _, _, _, _, _, _, label_records) in profiles.values() { + if let Some(records) = label_records { + let converted = self.convert_actor_labels(did, records).await; + if !converted.is_empty() { + labels.insert(did.clone(), converted); + } + } + } + let viewers = self.get_profile_viewer_states(&dids).await; let verif = self.loaders.verification.load_many(dids.clone()).await; let stats = self.loaders.profile_stats.load_many(dids.clone()).await; - let profiles = self.loaders.profile.load_many(dids).await; + + // Convert actor_id-keyed results back to DID-keyed + let id_to_did: HashMap = actor_id_map.iter().map(|(did, id)| (*id, did.clone())).collect(); + + profiles + .into_iter() + .filter_map(|(actor_id, profile_info)| { + let did = id_to_did.get(&actor_id)?; + let labels = labels.get(did).cloned().unwrap_or_default(); + let verif = verif.get(did).cloned(); + let viewer = viewers.get(did).cloned(); + let stats = stats.get(did).copied(); + + let v = build_detailed(profile_info, stats, labels, verif, viewer, &self.cdn); + Some((did.clone(), v)) + }) + .collect() + } + + /// Optimized version that takes actor_ids directly, avoiding DID lookups + pub async fn hydrate_profiles_detailed_by_id( + &self, + actor_ids_with_dids: Vec<(i32, String)>, + ) -> HashMap { + // Extract just the actor_ids for queries + let actor_ids: Vec = actor_ids_with_dids.iter().map(|(id, _)| *id).collect(); + let dids: Vec = actor_ids_with_dids.iter().map(|(_, did)| did.clone()).collect(); + + // Build a map of actor_id to DID for later lookups + let id_to_did: std::collections::HashMap = actor_ids_with_dids.into_iter().collect(); + + // Load data using actor_ids where possible + let profiles = self.loaders.profile_by_id.load_many(actor_ids.clone()).await; + let stats = self.loaders.profile_stats_by_id.load_many(&actor_ids).await; + + // Extract labels directly from loaded profiles + let mut labels: HashMap> = HashMap::new(); + for (actor_id, profile_info) in &profiles { + if let Some(did) = id_to_did.get(actor_id) { + if let Some(ref label_records) = profile_info.8 { + let converted_labels = self.convert_actor_labels(did, label_records).await; + if !converted_labels.is_empty() { + labels.insert(did.clone(), converted_labels); + } + } + } + } + let viewers = self.get_profile_viewer_states(&dids).await; + let verif = self.loaders.verification.load_many(dids).await; profiles .into_iter() - .map(|(k, profile_info)| { - let labels = labels.get(&k).cloned().unwrap_or_default(); - let verif = verif.get(&k).cloned(); - let viewer = viewers.get(&k).cloned(); - let stats = stats.get(&k).copied(); + .map(|(actor_id, profile_info)| { + let did = id_to_did.get(&actor_id).cloned().unwrap_or_default(); + let labels = labels.get(&did).cloned().unwrap_or_default(); + let verif = verif.get(&did).cloned(); + let viewer = viewers.get(&did).cloned(); + let stats = stats.get(&actor_id).copied(); let v = build_detailed(profile_info, stats, labels, verif, viewer, &self.cdn); - (k, v) + (actor_id, v) }) .collect() } diff --git a/parakeet/src/loaders/labeler.rs b/parakeet/src/loaders/labeler.rs index a0584ffc..e520ecdf 100644 --- a/parakeet/src/loaders/labeler.rs +++ b/parakeet/src/loaders/labeler.rs @@ -304,98 +304,6 @@ impl LabelLoader { .into_group_map_by(|v| v.uri.clone()) } - /// Load labels by URIs using actor_ids (optimized version - avoids decompressing actors table) - /// - /// This is an optimized version that resolves labeler DIDs → actor_ids via IdCache first, - /// then queries by actor_ids directly, avoiding the 2-30ms penalty from joining the - /// compressed actors table. - /// - /// Expected performance: 2-30ms → 0.5-2ms (5-60x faster) - pub async fn load_many_by_actor_ids( - &self, - uris: &[String], - labeler_actor_ids: &[i32], - ) -> HashMap> { - let mut conn = self.0.get().await.unwrap(); - - if labeler_actor_ids.is_empty() || uris.is_empty() { - return HashMap::new(); - } - - let uri_refs: Vec<&str> = uris.iter().map(|s| s.as_str()).collect(); - - #[derive(diesel::QueryableByName)] - struct LabelRowById { - #[diesel(sql_type = diesel::sql_types::Integer)] - labeler_actor_id: i32, - #[diesel(sql_type = diesel::sql_types::Text)] - label: String, - #[diesel(sql_type = diesel::sql_types::Text)] - uri: String, - #[diesel(sql_type = diesel::sql_types::Bool)] - self_label: bool, - #[diesel(sql_type = diesel::sql_types::Nullable)] - cid: Option>, - #[diesel(sql_type = diesel::sql_types::Bool)] - negated: bool, - #[diesel(sql_type = diesel::sql_types::Nullable)] - expires: Option>, - #[diesel(sql_type = diesel::sql_types::Nullable)] - sig: Option>, - #[diesel(sql_type = diesel::sql_types::Timestamptz)] - created_at: chrono::DateTime, - } - - let labels: Vec = diesel_async::RunQueryDsl::load( - diesel::sql_query(include_str!("../sql/labels_by_actor_ids.sql")) - .bind::, _>(&uri_refs) - .bind::, _>(labeler_actor_ids), - &mut conn, - ) - .await - .unwrap_or_else(|e| { - tracing::error!("label load by actor_ids failed: {e}"); - vec![] - }); - - // Collect unique labeler actor_ids to resolve - let labeler_ids: std::collections::HashSet = labels - .iter() - .map(|row| row.labeler_actor_id) - .collect(); - - // Batch resolve labeler actor_ids → DIDs using IdCache - let labeler_ids_vec: Vec = labeler_ids.into_iter().collect(); - let id_to_actor_data = self.1.get_actor_data_many(&labeler_ids_vec).await; - - // Build result with resolved labeler DIDs - labels - .into_iter() - .filter_map(|row| { - // Resolve labeler DID from IdCache - let labeler_did = match id_to_actor_data.get(&row.labeler_actor_id) { - Some(data) => data.did.clone(), - None => { - tracing::warn!("label: missing labeler DID for actor_id {}", row.labeler_actor_id); - return None; - } - }; - - Some(models::Label { - labeler_actor_id: row.labeler_actor_id, - label: row.label.clone(), - uri: row.uri.clone(), - self_label: row.self_label, - cid: row.cid, - negated: row.negated, - expires: row.expires, - sig: row.sig, - created_at: row.created_at, - labeler: labeler_did, - }) - }) - .into_group_map_by(|v| v.uri.clone()) - } /// Get the IdCache for DID → actor_id resolution pub fn id_cache(&self) -> ¶keet_db::id_cache::IdCache { diff --git a/parakeet/src/loaders/mod.rs b/parakeet/src/loaders/mod.rs index 8da75258..0fa5c652 100644 --- a/parakeet/src/loaders/mod.rs +++ b/parakeet/src/loaders/mod.rs @@ -20,7 +20,7 @@ pub use list::{EnrichedList, ListKey, ListLoader, ListLoaderRet, ListStateLoader pub use misc::{EnrichedStarterPack, EnrichedVerification, StarterPackKey, StarterPackLoader, StarterPackLoaderRet, VerificationLoader}; pub use post::{EnrichedThreadgate, HydratedPost, PostLoader, PostLoaderRet, PostStateLoader, PostWithComputed}; pub use profile::{ - EnrichedStatus, HandleLoader, Profile, ProfileByIdLoader, ProfileLoader, ProfileLoaderRet, ProfileStateLoader, ProfileStatsLoader, ProfileStatsByIdLoader, + EnrichedStatus, HandleLoader, Profile, ProfileByIdLoader, ProfileLoaderRet, ProfileStateLoader, ProfileStatsLoader, ProfileStatsByIdLoader, }; // Re-export query builder functions (for testing) @@ -29,7 +29,7 @@ pub use labeler::{build_labeler_records_query, build_labels_query, build_labels_ pub use list::build_lists_batch_query; pub use misc::{build_starterpack_feeds_query, build_starterpacks_batch_query, build_verifications_batch_query}; pub use post::{build_posts_batch_query, build_posts_by_natural_keys_batch_query}; -pub use profile::{build_profiles_batch_query, build_profiles_by_id_batch_query, build_statuses_batch_query}; +pub use profile::build_profiles_by_id_batch_query; type CachingLoader = Loader>; @@ -67,7 +67,6 @@ pub struct Dataloaders { pub like_state: LikeRecordLoader, pub posts: CachingLoader, pub post_state: PostStateLoader, - pub profile: CachingLoader, pub profile_by_id: CachingLoader, pub profile_stats: CachingLoader, pub profile_stats_by_id: ProfileStatsByIdLoader, @@ -96,7 +95,6 @@ impl Dataloaders { // Occasionally changed: Profile metadata can be updated // 1 hour TTL, 50k capacity for profiles - profile: new_plc_loader(ProfileLoader(pool.clone()), "profile:", 3600, 50_000), profile_by_id: new_plc_loader(ProfileByIdLoader(pool.clone(), id_cache.clone()), "profile_id:", 3600, 50_000), // 1 hour TTL, 10k capacity for feeds/lists/etc feedgen: new_plc_loader(FeedGenLoader(pool.clone()), "feedgen:", 3600, 10_000), diff --git a/parakeet/src/loaders/profile.rs b/parakeet/src/loaders/profile.rs index 6cc62a85..10ff0969 100644 --- a/parakeet/src/loaders/profile.rs +++ b/parakeet/src/loaders/profile.rs @@ -37,63 +37,6 @@ pub struct Status { pub thumb_cid: Option>, } -/// Build SQL query for batch loading profiles with actor metadata -/// -/// This function is public for testing purposes. -/// -/// SCHEMA CHANGE: profiles, chat_decls, and notif_decl tables have been consolidated into actors table -/// All data now comes from actors.profile_*, actors.chat_*, actors.notif_decl_* columns -pub fn build_profiles_batch_query() -> &'static str { - "SELECT - a.did, - a.handle, - a.account_created_at, - a.sync_state, - a.id as actor_id, - a.profile_cid as cid, - a.profile_avatar_cid as avatar_cid, - a.profile_banner_cid as banner_cid, - a.profile_display_name as display_name, - a.profile_description as description, - a.profile_pinned_post_rkey as pinned_post_rkey, - a.profile_joined_sp_id as joined_sp_id, - a.profile_pronouns as pronouns, - a.profile_website as website, - a.chat_allow_incoming as allow_incoming, - CASE WHEN a.labeler_cid IS NOT NULL THEN a.id ELSE NULL END as labeler_actor_id, - a.notif_decl_allow_subscriptions as allow_subscriptions, - a.labels - FROM actors a - WHERE a.did = ANY($1) - AND a.status = 'active'::actor_status" -} - -/// Build SQL query for batch loading statuses with embed URI reconstruction -/// -/// This function is public for testing purposes. -/// -/// SCHEMA CHANGE: statuses table has been consolidated into actors table -/// All data now comes from actors.status_* columns -pub fn build_statuses_batch_query() -> &'static str { - "SELECT - a.id as actor_id, - a.status_cid as cid, - a.status_created_at as created_at, - a.status_type as status, - a.status_duration as duration, - a.status_embed_post_actor_id as embed_post_actor_id, - a.status_embed_post_rkey as embed_post_rkey, - a.status_thumb_mime_type as thumb_mime_type, - a.status_thumb_cid as thumb_cid, - a.did, - (SELECT 'at://' || emb_a.did || '/app.bsky.feed.post/' || i64_to_tid(emb_p.rkey) - FROM posts emb_p - INNER JOIN actors emb_a ON emb_p.actor_id = emb_a.id - WHERE emb_p.actor_id = a.status_embed_post_actor_id AND emb_p.rkey = a.status_embed_post_rkey) as embed_uri - FROM actors a - WHERE a.did = ANY($1) - AND a.status_cid IS NOT NULL" -} // Enriched Status with reconstructed fields #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] @@ -172,6 +115,30 @@ impl std::ops::Deref for EnrichedStatus { } } +/// ProfileLoaderRet is the return type for both ProfileByIdLoader +/// +/// Tuple format: +/// - String: did - always present +/// - Option: handle - from actors table +/// - Option: account_created_at - from actors table +/// - Option: profile - optional (may not exist) +/// - Option: chat declaration +/// - bool: is_labeler +/// - Option: status +/// - Option: notification subscription settings +/// - Option>: labels - from actors table +pub type ProfileLoaderRet = ( + String, // did - always present (from actors table) + Option, // handle - from actors table + Option>, // account_created_at - from actors table + Option, // profile - optional (may not exist) + Option, + bool, + Option, + Option, + Option>, // labels - from actors table +); + pub struct HandleLoader(pub(super) Pool); impl BatchFn for HandleLoader { async fn load(&mut self, keys: &[String]) -> HashMap { @@ -198,208 +165,6 @@ impl BatchFn for HandleLoader { } } -pub struct ProfileLoader(pub(super) Pool); -pub type ProfileLoaderRet = ( - String, // did - always present (from actors table) - Option, // handle - from actors table - Option>, // account_created_at - from actors table - Option, // profile - optional (may not exist) - Option, - bool, - Option, - Option, - Option>, // labels - from actors table -); -impl BatchFn for ProfileLoader { - async fn load(&mut self, keys: &[String]) -> HashMap { - let overall_start = std::time::Instant::now(); - let mut conn = self.0.get().await.unwrap(); - - // Load basic actor/profile data with raw SQL to avoid Diesel DSL join complexity - let dids: Vec<&str> = keys.iter().map(|s| s.as_str()).collect(); - - #[derive(diesel::QueryableByName)] - #[allow(dead_code, reason = "Diesel QueryableByName requires all SQL columns even if unused")] - struct ActorRow { - #[diesel(sql_type = diesel::sql_types::Text)] - did: String, - #[diesel(sql_type = diesel::sql_types::Nullable)] - handle: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - account_created_at: Option>, - #[diesel(sql_type = parakeet_db::schema::sql_types::ActorSyncState)] - sync_state: parakeet_db::types::ActorSyncState, - #[diesel(sql_type = diesel::sql_types::Nullable)] - actor_id: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - cid: Option>, - #[diesel(sql_type = diesel::sql_types::Nullable)] - avatar_cid: Option>, - #[diesel(sql_type = diesel::sql_types::Nullable)] - banner_cid: Option>, - #[diesel(sql_type = diesel::sql_types::Nullable)] - display_name: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - description: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - pinned_post_rkey: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - joined_sp_id: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - pronouns: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - website: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - allow_incoming: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - labeler_actor_id: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - allow_subscriptions: Option, - #[diesel(sql_type = diesel::sql_types::Nullable>)] - labels: Option>, - } - - let profile_query_start = std::time::Instant::now(); - let res: Result, _> = diesel_async::RunQueryDsl::load( - diesel::sql_query(build_profiles_batch_query()) - .bind::, _>(&dids), - &mut conn, - ) - .await; - let profile_query_time = profile_query_start.elapsed().as_secs_f64() * 1000.0; - - match res { - Ok(res) => { - let profile_count = res.len(); - - // Load enriched statuses separately with raw SQL - let status_dids: Vec = res.iter().map(|row| row.did.clone()).collect(); - let status_query_start = std::time::Instant::now(); - let status_map = if !status_dids.is_empty() { - let status_dids_refs: Vec<&str> = status_dids.iter().map(|s| s.as_str()).collect(); - - #[derive(diesel::QueryableByName)] - struct StatusRow { - #[diesel(sql_type = diesel::sql_types::Integer)] - actor_id: i32, - #[diesel(sql_type = diesel::sql_types::Binary)] - cid: Vec, - #[diesel(sql_type = diesel::sql_types::Timestamptz)] - created_at: chrono::DateTime, - #[diesel(sql_type = parakeet_db::schema::sql_types::StatusType)] - status: parakeet_db::types::StatusType, - #[diesel(sql_type = diesel::sql_types::Nullable)] - duration: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - embed_post_actor_id: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - embed_post_rkey: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - thumb_mime_type: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - thumb_cid: Option>, - #[diesel(sql_type = diesel::sql_types::Text)] - did: String, - #[diesel(sql_type = diesel::sql_types::Nullable)] - embed_uri: Option, - } - - let statuses: Vec = diesel_async::RunQueryDsl::load( - diesel::sql_query(build_statuses_batch_query()) - .bind::, _>(&status_dids_refs), - &mut conn, - ) - .await - .unwrap_or_default(); - - statuses - .into_iter() - .map(|row| { - // Reconstruct the status record - let record = build_status_record( - &row.status, - &row.duration, - &row.embed_uri, - &None, // embed_title - not currently loaded - &None, // embed_description - not currently loaded - &row.thumb_mime_type, - &row.thumb_cid, - &row.created_at, - ); - - let enriched = EnrichedStatus { - status: Status { - actor_id: row.actor_id, - cid: row.cid, - created_at: row.created_at, - status: row.status, - duration: row.duration, - embed_post_actor_id: row.embed_post_actor_id, - embed_post_rkey: row.embed_post_rkey, - thumb_mime_type: row.thumb_mime_type, - thumb_cid: row.thumb_cid, - }, - did: row.did.clone(), - created_at: row.created_at, - record, - embed_uri: row.embed_uri, - embed_title: None, - embed_description: None, - }; - (row.did, enriched) - }) - .collect::>() - } else { - HashMap::new() - }; - let status_query_time = status_query_start.elapsed().as_secs_f64() * 1000.0; - let status_count = status_map.len(); - - let results: HashMap = HashMap::from_iter(res.into_iter().map( - |row| { - // Construct Profile if actor_id is present (cid can be None for empty profiles) - let profile = row.actor_id.map(|actor_id| Profile { - actor_id, - cid: row.cid, - avatar_cid: row.avatar_cid, - banner_cid: row.banner_cid, - display_name: row.display_name, - description: row.description, - pinned_post_rkey: row.pinned_post_rkey, - joined_sp_id: row.joined_sp_id, - pronouns: row.pronouns, - website: row.website, - }); - - let chat_decl = row.allow_incoming.and_then(|v| ChatAllowIncoming::from_str(&v.to_string()).ok()); - let notif_decl = row.allow_subscriptions.and_then(|v| ProfileAllowSubscriptions::from_str(&v.to_string()).ok()); - let is_labeler = row.labeler_actor_id.is_some(); - let status = status_map.get(&row.did).cloned(); - - let val = (row.did.clone(), row.handle, row.account_created_at, profile, chat_decl, is_labeler, status, notif_decl, row.labels); - - (row.did, val) - }, - )); - - let overall_time = overall_start.elapsed().as_secs_f64() * 1000.0; - - if overall_time > 15.0 || profile_query_time > 10.0 || status_query_time > 5.0 { - tracing::info!( - " → ProfileLoader: {:.1}ms total ({} profiles, {} statuses) | profile_query: {:.1}ms, status_query: {:.1}ms", - overall_time, profile_count, status_count, profile_query_time, status_query_time - ); - } - - results - } - Err(e) => { - tracing::error!("profile load failed: {e}"); - HashMap::new() - } - } - } -} /// Build SQL query for batch loading profiles by actor_id (uses consolidated actors table) /// @@ -878,6 +643,11 @@ impl ProfileStatsByIdLoader { pub struct ProfileStateLoader(pub(super) Pool, pub(super) std::sync::Arc); impl ProfileStateLoader { + /// Get the connection pool + pub fn pool(&self) -> &Pool { + &self.0 + } + /// Get single profile state (DID-based interface, internally uses actor_ids) pub async fn get(&self, viewer_did: &str, subject_did: &str) -> Option { let results = self.get_many(viewer_did, &vec![subject_did.to_string()]).await; @@ -978,7 +748,7 @@ impl ProfileStateLoader { } /// Get the IdCache for DID → actor_id resolution - pub fn id_cache(&self) -> ¶keet_db::id_cache::IdCache { + pub fn id_cache(&self) -> &std::sync::Arc { &self.1 } } diff --git a/parakeet/src/sql/labels_by_actor_ids.sql b/parakeet/src/sql/labels_by_actor_ids.sql deleted file mode 100644 index cd6862a7..00000000 --- a/parakeet/src/sql/labels_by_actor_ids.sql +++ /dev/null @@ -1,57 +0,0 @@ --- Phase 7: Labels denormalized into actors.labels[] and posts.labels[] arrays --- Optimized labels query that uses actor_ids instead of DIDs --- This avoids decompressing the actors table by using IdCache to resolve DIDs beforehand --- --- Parameters: --- $1: uris (text[]) - Array of URIs to fetch labels for --- $2: labeler_actor_ids (integer[]) - Array of allowed labeler actor IDs --- --- Performance: Maintained similar performance with denormalized structure --- Avoids: Separate labels table join --- --- Note: Caller must use IdCache to: --- 1. Resolve labeler DIDs → actor_ids before this query --- 2. Resolve labeler_actor_id → DID after this query --- --- Note: The new composite types (actor_label, post_label) do not include: --- - uri (reconstructed from actor/post) --- - self_label (always false for post labels, always true for actor self-labels) --- - cid (not stored in denormalized structure) --- - sig (not stored in denormalized structure) - --- Extract labels from actors table -SELECT - (label_record).labeler_actor_id as labeler_actor_id, - (label_record).label as label, - 'at://' || a.did as uri, - true as self_label, -- Actor labels are always self-labels - NULL::bytea as cid, -- Not stored in denormalized structure - (label_record).negated as negated, - (label_record).expires as expires, - NULL::bytea as sig, -- Not stored in denormalized structure - (label_record).created_at as created_at -FROM actors a, UNNEST(a.labels) as label_record -WHERE 'at://' || a.did = ANY($1) - AND (label_record).negated = false - AND ((label_record).labeler_actor_id = a.id OR (label_record).labeler_actor_id = ANY($2)) - -UNION ALL - --- Extract labels from posts table -SELECT - (label_record).labeler_actor_id as labeler_actor_id, - (label_record).label as label, - 'at://' || a.did || '/app.bsky.feed.post/' || parakeet_db.i64_to_tid(p.rkey) as uri, - false as self_label, -- Post labels are not self-labels - NULL::bytea as cid, -- Not stored in denormalized structure - (label_record).negated as negated, - (label_record).expires as expires, - NULL::bytea as sig, -- Not stored in denormalized structure - (label_record).created_at as created_at -FROM posts p -INNER JOIN actors a ON p.actor_id = a.id, UNNEST(p.labels) as label_record -WHERE 'at://' || a.did || '/app.bsky.feed.post/' || parakeet_db.i64_to_tid(p.rkey) = ANY($1) - AND (label_record).negated = false - AND (label_record).labeler_actor_id = ANY($2) - -ORDER BY created_at diff --git a/parakeet/src/xrpc/app_bsky/actor.rs b/parakeet/src/xrpc/app_bsky/actor.rs index 771875eb..998b47bc 100644 --- a/parakeet/src/xrpc/app_bsky/actor.rs +++ b/parakeet/src/xrpc/app_bsky/actor.rs @@ -9,6 +9,7 @@ use axum::Json; use axum_extra::extract::Query as ExtraQuery; use lexica::app_bsky::actor::ProfileViewDetailed; use serde::{Deserialize, Serialize}; +use std::collections::HashMap; #[derive(Debug, Deserialize)] pub struct ActorQuery { @@ -190,7 +191,17 @@ pub async fn search_actors( (None, None) }; let hyd = StatefulHydrator::new(&state.dataloaders, &state.cdn, &labelers, maybe_did, maybe_actor_id).await; - let mut profiles_map = hyd.hydrate_profiles_detailed(dids.clone()).await; + + // Use the cache's batch method instead of direct hydration + let profiles_vec = state.profile_cache + .get_or_hydrate_batch(dids.clone(), &state.pool, &state.id_cache, &hyd) + .await; + + // Convert Vec to HashMap for compatibility with existing code + let mut profiles_map: std::collections::HashMap = profiles_vec + .into_iter() + .map(|profile| (profile.did.clone(), profile)) + .collect(); // Maintain search result order (hydration returns HashMap) let actors: Vec = dids @@ -264,7 +275,17 @@ pub async fn search_actors_typeahead( (None, None) }; let hyd = StatefulHydrator::new(&state.dataloaders, &state.cdn, &labelers, maybe_did, maybe_actor_id).await; - let mut profiles_map = hyd.hydrate_profiles_detailed(dids.clone()).await; + + // Use the cache's batch method instead of direct hydration + let profiles_vec = state.profile_cache + .get_or_hydrate_batch(dids.clone(), &state.pool, &state.id_cache, &hyd) + .await; + + // Convert Vec to HashMap for compatibility with existing code + let mut profiles_map: std::collections::HashMap = profiles_vec + .into_iter() + .map(|profile| (profile.did.clone(), profile)) + .collect(); // Maintain priority order let actors: Vec = dids @@ -325,13 +346,32 @@ pub async fn get_suggestions( .into_response()); } + // Convert DIDs to actor_ids for efficient profile loading + let did_to_actor_id = crate::id_cache_helpers::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(); + // Load profiles to check quality (already sorted by follower count from SQL) - let profiles = state + let profiles_by_id = state .dataloaders - .profile - .load_many(all_dids.clone()) + .profile_by_id + .load_many(actor_ids) .await; + // Convert back to DID-keyed map for compatibility + let profiles: HashMap = profiles_by_id + .into_iter() + .filter_map(|(actor_id, profile)| { + did_to_actor_id.iter() + .find(|(_, id)| **id == actor_id) + .map(|(did, _)| (did.clone(), profile)) + }) + .collect(); + // Filter by quality, maintaining follower-count order from SQL query let ranked_dids: Vec = all_dids .into_iter() diff --git a/parakeet/src/xrpc/app_bsky/unspecced/mod.rs b/parakeet/src/xrpc/app_bsky/unspecced/mod.rs index 519dbd22..05c9ba36 100644 --- a/parakeet/src/xrpc/app_bsky/unspecced/mod.rs +++ b/parakeet/src/xrpc/app_bsky/unspecced/mod.rs @@ -250,8 +250,29 @@ 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::id_cache_helpers::get_actor_ids_or_fetch( + &state.pool, + &state.id_cache, + &all_dids, + ).await { + Ok(map) => map, + Err(_) => return Ok(Json(GetSuggestedUsersResponse { actors: Vec::new() })), + }; + + let actor_ids: Vec = actor_id_map.values().copied().collect(); + // Load profiles to check quality (already sorted by follower count from SQL) - let profiles = state.dataloaders.profile.load_many(all_dids.clone()).await; + let profiles_by_id = state.dataloaders.profile_by_id.load_many(actor_ids).await; + + // Convert actor_id-keyed profiles back to DID-keyed for easier lookup + let id_to_did: std::collections::HashMap = actor_id_map.iter().map(|(did, id)| (*id, did.clone())).collect(); + let profiles: std::collections::HashMap = profiles_by_id + .into_iter() + .filter_map(|(actor_id, profile_info)| { + id_to_did.get(&actor_id).map(|did| (did.clone(), profile_info)) + }) + .collect(); // Filter by quality, maintaining follower-count order from SQL query let ranked_dids: Vec = all_dids diff --git a/parakeet/tests/loaders_test.rs b/parakeet/tests/loaders_test.rs index 06ff3815..764e8d71 100644 --- a/parakeet/tests/loaders_test.rs +++ b/parakeet/tests/loaders_test.rs @@ -15,135 +15,9 @@ use eyre::WrapErr; // ============================================================================ // PROFILE LOADER TESTS // ============================================================================ -// ProfileLoader uses complex SQL with: -// - Queries consolidated actors table (profile_*, chat_*, notif_decl_* columns) -// - LEFT JOIN to labelers table (still separate) -// - Format string for DID list interpolation -// - Separate status query with embed reconstruction - -#[tokio::test] -async fn test_profile_loader_main_query_empty() -> eyre::Result<()> { - common::ensure_test_db_ready().await; - let pool = common::test_diesel_pool(); - let _conn = pool.get().await.wrap_err("Failed to get connection")?; - - // Simulate ProfileLoader's main query with empty DID list - let dids: Vec = vec![]; - - if dids.is_empty() { - // Empty case - should handle gracefully - return Ok(()); - } - - // This would be the actual query, but it won't execute with empty DIDs - // Just verifying the logic handles it - Ok(()) -} - -#[tokio::test] -async fn test_profile_loader_main_query_structure() -> eyre::Result<()> { - common::ensure_test_db_ready().await; - let pool = common::test_diesel_pool(); - let mut conn = pool.get().await.wrap_err("Failed to get connection")?; - - let dids = ["did:plc:test1".to_string(), "did:plc:test2".to_string()]; - - // Build the query using the actual ProfileLoader query builder (no SQL duplication!) - let dids_refs: Vec<&str> = dids.iter().map(|s| s.as_str()).collect(); - - let query = parakeet::loaders::build_profiles_batch_query(); - - #[derive(diesel::QueryableByName)] - #[allow(dead_code)] - struct ActorRow { - #[diesel(sql_type = diesel::sql_types::Text)] - did: String, - #[diesel(sql_type = diesel::sql_types::Nullable)] - handle: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - actor_id: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - cid: Option>, - #[diesel(sql_type = diesel::sql_types::Nullable)] - created_at: Option>, - #[diesel(sql_type = diesel::sql_types::Nullable)] - avatar_cid: Option>, - #[diesel(sql_type = diesel::sql_types::Nullable)] - banner_cid: Option>, - #[diesel(sql_type = diesel::sql_types::Nullable)] - display_name: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - description: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - pinned_post_id: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - joined_sp_id: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - pronouns: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - website: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - allow_incoming: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - labeler_actor_id: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - allow_subscriptions: Option, - } - - diesel::sql_query(query) - .bind::, _>(&dids_refs) - .load::(&mut conn) - .await - .wrap_err("ProfileLoader main query SQL failed")?; - - Ok(()) -} - -#[tokio::test] -async fn test_profile_loader_status_query_structure() -> eyre::Result<()> { - common::ensure_test_db_ready().await; - let pool = common::test_diesel_pool(); - let mut conn = pool.get().await.wrap_err("Failed to get connection")?; - - let dids = ["did:plc:test1".to_string()]; - let dids_refs: Vec<&str> = dids.iter().map(|s| s.as_str()).collect(); - - // Use the actual status query builder from ProfileLoader (no SQL duplication!) - let status_query = parakeet::loaders::build_statuses_batch_query(); - - #[derive(diesel::QueryableByName)] - #[allow(dead_code)] - struct StatusRow { - #[diesel(sql_type = diesel::sql_types::Integer)] - actor_id: i32, - #[diesel(sql_type = diesel::sql_types::Binary)] - cid: Vec, - #[diesel(sql_type = diesel::sql_types::Timestamptz)] - created_at: chrono::DateTime, - #[diesel(sql_type = parakeet_db::schema::sql_types::StatusType)] - status: parakeet_db::types::StatusType, - #[diesel(sql_type = diesel::sql_types::Nullable)] - duration: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - embed_post_id: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - thumb_mime_type: Option, - #[diesel(sql_type = diesel::sql_types::Nullable)] - thumb_cid: Option>, - #[diesel(sql_type = diesel::sql_types::Text)] - did: String, - #[diesel(sql_type = diesel::sql_types::Nullable)] - embed_uri: Option, - } - - diesel::sql_query(status_query) - .bind::, _>(&dids_refs) - .load::(&mut conn) - .await - .wrap_err("ProfileLoader status query SQL failed")?; - - Ok(()) -} +// ProfileByIdLoader uses optimized SQL querying by actor_id (primary key) +// instead of DID (btree index). Queries consolidated actors table with +// profile_*, chat_*, notif_decl_*, and status_* columns. #[tokio::test] async fn test_profile_by_id_loader_query_structure() -> eyre::Result<()> { diff --git a/parakeet/tests/sql/batch_loading_test.rs b/parakeet/tests/sql/batch_loading_test.rs index fd144a66..727851d2 100644 --- a/parakeet/tests/sql/batch_loading_test.rs +++ b/parakeet/tests/sql/batch_loading_test.rs @@ -49,46 +49,26 @@ async fn test_post_loader_batch_query() -> eyre::Result<()> { // NOTE: Threadgate loading test removed - threadgates are now loaded inline with posts // in build_posts_batch_query() to eliminate a separate database roundtrip -/// Test batch loading profiles (from loaders/profile.rs) +/// Test batch loading profiles by ID (from loaders/profile.rs) +/// Note: The old DID-based profile/status loaders have been removed as they were +/// inefficient (required btree index lookups). ProfileByIdLoader uses primary key lookups. #[tokio::test] -async fn test_batch_profiles_query() -> eyre::Result<()> { +async fn test_batch_profiles_by_id_query() -> eyre::Result<()> { common::ensure_test_db_ready().await; let pool = common::test_diesel_pool(); let mut conn = pool.get().await.wrap_err("Failed to get connection")?; // Use the ACTUAL query builder from the source code - let query = parakeet::loaders::build_profiles_batch_query(); - let dids = vec!["did:plc:test1", "did:plc:test2"]; + let query = parakeet::loaders::build_profiles_by_id_batch_query(); + let actor_ids = vec![1i32, 2i32]; diesel_async::RunQueryDsl::execute( diesel::sql_query(query) - .bind::, _>(&dids), - &mut conn, - ) - .await - .wrap_err("Batch profile loading query failed")?; - - Ok(()) -} - -/// Test batch loading statuses (from loaders/profile.rs) -#[tokio::test] -async fn test_batch_statuses_query() -> eyre::Result<()> { - common::ensure_test_db_ready().await; - let pool = common::test_diesel_pool(); - let mut conn = pool.get().await.wrap_err("Failed to get connection")?; - - // Use the ACTUAL query builder from the source code - let query = parakeet::loaders::build_statuses_batch_query(); - let dids = vec!["did:plc:test1"]; - - diesel_async::RunQueryDsl::execute( - diesel::sql_query(query) - .bind::, _>(&dids), + .bind::, _>(&actor_ids), &mut conn, ) .await - .wrap_err("Batch status loading query failed")?; + .wrap_err("Batch profile loading by ID query failed")?; Ok(()) }