diff --git a/parakeet/src/hydration/posts/mod.rs b/parakeet/src/hydration/posts/mod.rs index 9b9ce3a9..3e2241e3 100644 --- a/parakeet/src/hydration/posts/mod.rs +++ b/parakeet/src/hydration/posts/mod.rs @@ -15,6 +15,42 @@ use crate::hydration::StatefulHydrator; use crate::loaders::EnrichedThreadgate; impl StatefulHydrator<'_> { + /// Convert denormalized post labels to full Label models + /// This resolves labeler_actor_id to DID and constructs URIs + async fn convert_post_labels( + &self, + post_uri: &str, + labels: &[parakeet_db::composite_types::PostLabelRecord], + ) -> Vec { + if labels.is_empty() { + return vec![]; + } + + // Collect unique labeler actor IDs + let labeler_ids: Vec = labels.iter().map(|l| l.labeler_actor_id).collect(); + + // Batch resolve labeler_actor_id → DID using IdCache + let labeler_data = self.loaders.post_state.id_cache().get_actor_data_many(&labeler_ids).await; + + // Convert each label + labels.iter().filter_map(|record| { + let labeler_did = labeler_data.get(&record.labeler_actor_id)?.did.clone(); + + Some(parakeet_db::models::Label { + labeler_actor_id: record.labeler_actor_id, + label: record.label.clone(), + uri: post_uri.to_string(), + self_label: false, // Post labels are never self-labels + cid: None, // Not stored in denormalized structure + negated: record.negated, + expires: record.expires, + sig: None, // Not stored in denormalized structure + created_at: record.created_at, + labeler: labeler_did, + }) + }).collect() + } + async fn hydrate_threadgate( &self, threadgate: Option, @@ -81,7 +117,13 @@ impl StatefulHydrator<'_> { let embed = self.hydrate_embed_from_post(&post).await; let author = self.hydrate_profile_basic(post.did.clone()).await?; let threadgate = self.hydrate_threadgate(threadgate, &post.did).await; - let labels = self.get_label(&post.at_uri).await; + // OPTIMIZED: Use inline labels from PostLoader instead of separate query + let labels = if let Some(ref label_records) = post.post.labels { + let unwrapped: Vec<_> = label_records.iter().filter_map(|opt| opt.as_ref()).cloned().collect(); + self.convert_post_labels(&post.at_uri, &unwrapped).await + } else { + vec![] + }; Some(build_postview(( post, author, labels, embed, threadgate, viewer, stats, @@ -95,7 +137,13 @@ impl StatefulHydrator<'_> { let (post, _threadgate, stats) = self.loaders.posts.load(post_uri).await?; let embed = self.hydrate_embed_from_post(&post).await; let author = self.hydrate_profile_basic(post.did.clone()).await?; - let labels = self.get_label(&post.at_uri).await; + // OPTIMIZED: Use inline labels from PostLoader instead of separate query + let labels = if let Some(ref label_records) = post.post.labels { + let unwrapped: Vec<_> = label_records.iter().filter_map(|opt| opt.as_ref()).cloned().collect(); + self.convert_post_labels(&post.at_uri, &unwrapped).await + } else { + vec![] + }; Some(build_postview(( post, diff --git a/parakeet/src/hydration/profile/builders.rs b/parakeet/src/hydration/profile/builders.rs index 9dd13242..0b8e2ce4 100644 --- a/parakeet/src/hydration/profile/builders.rs +++ b/parakeet/src/hydration/profile/builders.rs @@ -155,7 +155,7 @@ fn build_status(status: crate::loaders::EnrichedStatus, cdn: &BskyCdn) -> Option } pub(super) fn build_basic( - (did, handle, account_created_at, profile, chat_decl, is_labeler, status, notif_decl): ProfileLoaderRet, + (did, handle, account_created_at, profile, chat_decl, is_labeler, status, notif_decl, _labels): ProfileLoaderRet, stats: Option, labels: Vec, verifications: Option>, @@ -201,7 +201,7 @@ pub(super) fn build_basic( } pub(super) fn build_profile( - (did, handle, account_created_at, profile, chat_decl, is_labeler, status, notif_decl): ProfileLoaderRet, + (did, handle, account_created_at, profile, chat_decl, is_labeler, status, notif_decl, _labels): ProfileLoaderRet, stats: Option, labels: Vec, verifications: Option>, @@ -250,7 +250,7 @@ pub(super) fn build_profile( } pub(super) fn build_detailed( - (did, handle, account_created_at, profile, chat_decl, is_labeler, status, notif_decl): ProfileLoaderRet, + (did, handle, account_created_at, profile, chat_decl, is_labeler, status, notif_decl, _labels): ProfileLoaderRet, stats: Option, labels: Vec, verifications: Option>, diff --git a/parakeet/src/hydration/profile/mod.rs b/parakeet/src/hydration/profile/mod.rs index 2d1c48f3..408d0dbd 100644 --- a/parakeet/src/hydration/profile/mod.rs +++ b/parakeet/src/hydration/profile/mod.rs @@ -13,12 +13,70 @@ use builders::{build_basic, build_detailed, build_profile, build_viewer, build_v pub use verification::TRUSTED_VERIFIERS; impl super::StatefulHydrator<'_> { + /// Convert ActorLabelRecords to full Label models by resolving labeler DIDs + async fn convert_actor_labels( + &self, + did: &str, + labels: &[parakeet_db::composite_types::ActorLabelRecord], + ) -> Vec { + if labels.is_empty() { + return vec![]; + } + + // Collect unique labeler actor IDs + let labeler_ids: Vec = labels.iter().map(|l| l.labeler_actor_id).collect(); + + // Batch resolve labeler_actor_id → DID using IdCache + let labeler_data = self.loaders.profile_state.id_cache().get_actor_data_many(&labeler_ids).await; + + // Convert each label - use DID and profile URI as targets + let mut result = Vec::new(); + for record in labels { + if let Some(labeler_did_data) = labeler_data.get(&record.labeler_actor_id) { + // Add label for actor DID + result.push(parakeet_db::models::Label { + labeler_actor_id: record.labeler_actor_id, + label: record.label.clone(), + uri: did.to_string(), + self_label: false, // Actor labels from labelers are never self-labels + cid: None, // Not stored in denormalized structure + negated: record.negated, + expires: record.expires, + sig: None, // Not stored in denormalized structure + created_at: record.created_at, + labeler: labeler_did_data.did.clone(), + }); + + // Add label for profile URI + result.push(parakeet_db::models::Label { + labeler_actor_id: record.labeler_actor_id, + label: record.label.clone(), + uri: format!("at://{}/app.bsky.actor.profile/self", did), + self_label: false, + cid: None, + negated: record.negated, + expires: record.expires, + sig: None, + created_at: record.created_at, + labeler: labeler_did_data.did.clone(), + }); + } + } + result + } + pub async fn hydrate_profile_basic(&self, did: String) -> Option { - let labels = self.get_profile_label(&did).await; 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?; + let profile_info = self.loaders.profile.load(did.clone()).await?; + + // Extract and convert inline labels + let labels = if let Some(ref label_records) = profile_info.8 { + self.convert_actor_labels(&did, label_records).await + } else { + vec![] + }; Some(build_basic( profile_info, @@ -54,12 +112,20 @@ impl super::StatefulHydrator<'_> { let profiles = self.loaders.profile_by_id.load_many(unique_actor_ids.clone()).await; let profile_load_time = step_timer.elapsed().as_secs_f64() * 1000.0; - // Extract DIDs from loaded profiles for label/viewer/verif/stats lookups - let dids: Vec = profiles.values().map(|(did, _, _, _, _, _, _, _)| did.clone()).collect(); + // Extract DIDs from loaded profiles for viewer/verif/stats lookups + let dids: Vec = profiles.values().map(|(did, _, _, _, _, _, _, _, _)| did.clone()).collect(); - // OPTIMIZED: Use actor_ids for labels (avoids decompressing actors table for labelers) + // Convert inline labels for all profiles step_timer = std::time::Instant::now(); - let labels = self.get_profile_label_many_by_actor_ids(&dids).await; + let mut labels = HashMap::new(); + for (did, _, _, _, _, _, _, _, label_records) in profiles.values() { + let converted = if let Some(records) = label_records { + self.convert_actor_labels(did, records).await + } else { + vec![] + }; + labels.insert(did.clone(), converted); + } let labels_time = step_timer.elapsed().as_secs_f64() * 1000.0; // OPTIMIZED: Use actor_ids directly for viewer states (avoids decompressing actors table) @@ -113,7 +179,7 @@ impl super::StatefulHydrator<'_> { .into_iter() .filter_map(|(subject_actor_id, state)| { // Find the DID for this actor_id from the profiles we loaded - profiles.get(&subject_actor_id).map(|(subject_did, _, _, _, _, _, _, _)| { + profiles.get(&subject_actor_id).map(|(subject_did, _, _, _, _, _, _, _, _)| { // Resolve list URIs with resolved owner DIDs let list_block = state.list_block_ids().and_then(|(actor_id, rkey)| { id_to_actor_data.get(&actor_id).map(|data| { diff --git a/parakeet/src/loaders/post.rs b/parakeet/src/loaders/post.rs index f514cc6f..3e772b20 100644 --- a/parakeet/src/loaders/post.rs +++ b/parakeet/src/loaders/post.rs @@ -166,6 +166,9 @@ pub struct PostWithComputed { pub threadgate_hidden_actor_ids: Option>>, #[diesel(sql_type = diesel::sql_types::Nullable>>)] pub threadgate_hidden_rkeys: Option>>, + // Labels denormalized field (loaded inline to avoid separate query) + #[diesel(sql_type = diesel::sql_types::Nullable>)] + pub labels: Option>, } /// Build SQL query for batch loading posts with computed fields @@ -234,7 +237,9 @@ pub fn build_posts_batch_query() -> &'static str { -- Threadgate denormalized fields (loaded inline to avoid separate query) p.threadgate_allow, p.threadgate_hidden_actor_ids, - p.threadgate_hidden_rkeys + p.threadgate_hidden_rkeys, + -- Labels denormalized field (loaded inline to avoid separate query) + p.labels FROM posts p WHERE p.rkey >= $3::bigint AND p.rkey <= $4::bigint @@ -864,8 +869,8 @@ impl BatchFn for PostLoader { postgate_rules: None, postgate_detached_actor_ids: None, postgate_detached_rkeys: None, - // Phase 7: Labels (not loaded for hydration - will be loaded separately if needed) - labels: None, + // Phase 7: Labels (loaded inline from denormalized column) + labels: p.labels.as_ref().map(|labels| labels.iter().map(|l| Some(l.clone())).collect()), // Phase 8: Engagement counts (loaded from database, maintained by triggers) like_count: p.like_count, repost_count: p.repost_count, @@ -958,6 +963,8 @@ impl BatchFn for PostLoader { p.threadgate_allow, p.threadgate_hidden_actor_ids, p.threadgate_hidden_rkeys, + // Labels (loaded inline to avoid separate query) + p.labels, )) }) .collect(); @@ -967,7 +974,7 @@ impl BatchFn for PostLoader { // Collect all mention actor IDs from facets to batch lookup DIDs let facets_start = std::time::Instant::now(); let mut mention_actor_ids: Vec = Vec::new(); - for (_, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, facet_1, facet_2, facet_3, facet_4, facet_5, facet_6, facet_7, facet_8, _, _, _) in &posts_with_computed_data { + for (_, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, facet_1, facet_2, facet_3, facet_4, facet_5, facet_6, facet_7, facet_8, _, _, _, _) in &posts_with_computed_data { for facet in [facet_1, facet_2, facet_3, facet_4, facet_5, facet_6, facet_7, facet_8].iter().filter_map(|f| f.as_ref()) { if let Some(actor_id) = facet.mention_actor_id { mention_actor_ids.push(actor_id); @@ -995,7 +1002,7 @@ impl BatchFn for PostLoader { // Collect unique actor IDs from hidden replies for batch DID lookup let mut hidden_reply_actor_ids: Vec = Vec::new(); - for (_post, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, threadgate_hidden_actor_ids, _) in &posts_with_computed_data { + for (_post, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, threadgate_hidden_actor_ids, _, _) in &posts_with_computed_data { if let Some(actor_ids) = threadgate_hidden_actor_ids { hidden_reply_actor_ids.extend(actor_ids.iter().filter_map(|&id| id)); } @@ -1017,7 +1024,7 @@ impl BatchFn for PostLoader { // Build threadgate map from inline data keyed by (actor_id, rkey) let threadgate_map: HashMap<(i32, i64), EnrichedThreadgate> = posts_with_computed_data .iter() - .filter_map(|(post, _, _, _, _, _, _, _, created_at, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, threadgate_allow, threadgate_hidden_actor_ids, threadgate_hidden_rkeys)| { + .filter_map(|(post, _, _, _, _, _, _, _, created_at, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, threadgate_allow, threadgate_hidden_actor_ids, threadgate_hidden_rkeys, _labels)| { // Only process posts that have threadgate data if threadgate_allow.is_none() && threadgate_hidden_actor_ids.is_none() && threadgate_hidden_rkeys.is_none() { return None; @@ -1146,6 +1153,7 @@ impl BatchFn for PostLoader { _threadgate_allow, _threadgate_hidden_actor_ids, _threadgate_hidden_rkeys, + labels, )| { // OPTIMIZED: Process facets from inline composite fields (no separate query) let facet_vec = vec![facet_1, facet_2, facet_3, facet_4, facet_5, facet_6, facet_7, facet_8]; diff --git a/parakeet/src/loaders/profile.rs b/parakeet/src/loaders/profile.rs index 5b640a47..719ed481 100644 --- a/parakeet/src/loaders/profile.rs +++ b/parakeet/src/loaders/profile.rs @@ -61,7 +61,8 @@ pub fn build_profiles_batch_query() -> &'static str { 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.notif_decl_allow_subscriptions as allow_subscriptions, + a.labels FROM actors a WHERE a.did = ANY($1) AND a.status = 'active'::actor_status" @@ -207,6 +208,7 @@ pub type ProfileLoaderRet = ( bool, Option, Option, + Option>, // labels - from actors table ); impl BatchFn for ProfileLoader { async fn load(&mut self, keys: &[String]) -> HashMap { @@ -253,6 +255,8 @@ impl BatchFn for ProfileLoader { 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(); @@ -374,7 +378,7 @@ impl BatchFn for ProfileLoader { 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); + let val = (row.did.clone(), row.handle, row.account_created_at, profile, chat_decl, is_labeler, status, notif_decl, row.labels); (row.did, val) }, @@ -424,7 +428,8 @@ pub fn build_profiles_by_id_batch_query() -> &'static str { a.status_embed_post_actor_id, a.status_embed_post_rkey, a.status_thumb_mime_type, - a.status_thumb_cid + a.status_thumb_cid, + a.labels FROM actors a WHERE a.id = ANY($1)" } @@ -532,6 +537,8 @@ impl BatchFn for ProfileByIdLoader { status_thumb_mime_type: Option, #[diesel(sql_type = diesel::sql_types::Nullable)] status_thumb_cid: Option>, + #[diesel(sql_type = diesel::sql_types::Nullable>)] + labels: Option>, } let profile_query_start = std::time::Instant::now(); @@ -654,7 +661,7 @@ impl BatchFn for ProfileByIdLoader { let status = status_map.get(&row.actor_id).cloned(); // Note: account_created_at is None since we don't query actors table - let val = (cached_data.did.clone(), cached_data.handle.clone(), None, profile, chat_decl, is_labeler, status, notif_decl); + let val = (cached_data.did.clone(), cached_data.handle.clone(), None, profile, chat_decl, is_labeler, status, notif_decl, row.labels); Some((row.actor_id, val)) },