From 30ffec8552fed311b63c9db43255cdbfd4cf6af7 Mon Sep 17 00:00:00 2001 From: Timothy Quilling Date: Tue, 9 Dec 2025 10:48:42 -0500 Subject: [PATCH] fix: use ids --- parakeet/src/db/feeds.rs | 14 ++--- parakeet/src/db/posts.rs | 9 ++- parakeet/src/hydration/posts/feed.rs | 56 +++++++++++++------ parakeet/src/hydration/profile/mod.rs | 24 -------- parakeet/src/hydration/starter_packs.rs | 48 ++++++++++++++-- .../src/xrpc/app_bsky/feed/get_timeline.rs | 38 +++++++------ .../src/xrpc/app_bsky/feed/posts/feeds.rs | 6 +- .../src/xrpc/app_bsky/feed/posts/helpers.rs | 2 +- 8 files changed, 122 insertions(+), 75 deletions(-) diff --git a/parakeet/src/db/feeds.rs b/parakeet/src/db/feeds.rs index d04a5bb2..1c4d3de6 100644 --- a/parakeet/src/db/feeds.rs +++ b/parakeet/src/db/feeds.rs @@ -145,8 +145,8 @@ pub struct AuthorFeedItem { pub uri: String, /// URI of the actual post (same as uri for posts, points to reposted post for reposts) pub item_uri: String, - /// DID of the actor (author or reposter) - pub actor_did: String, + /// Internal actor_id (author or reposter) - resolve to DID at API edge + pub actor_id: i32, /// Type of item: "post" or "repost" pub typ: String, /// Timestamp for sorting @@ -381,7 +381,7 @@ pub async fn get_author_feed( tracing::debug!(" → Application merge: {} items", merged.len()); - // Collect unique actor_ids from merged results + // Collect unique actor_ids from merged results (for URI construction) let mut actor_ids_needed: Vec = merged .iter() .flat_map(|row| vec![row.actor_id, row.post_author_id]) @@ -389,7 +389,7 @@ pub async fn get_author_feed( actor_ids_needed.sort_unstable(); actor_ids_needed.dedup(); - // Lookup actor DIDs using id_cache or fallback to DB + // Lookup actor DIDs using id_cache or fallback to DB (for URI construction only) let lookup_start = std::time::Instant::now(); let actor_id_to_did = if let Some(cache) = id_cache { // Try cache first @@ -464,7 +464,7 @@ pub async fn get_author_feed( tracing::warn!(" → Slow actor lookup: {:.1} ms", lookup_time); } - // Reconstruct URIs from actor_ids and rkeys + // Reconstruct URIs from actor_ids and rkeys (URIs still needed, but return actor_id too) let feed_items: Vec = merged .into_iter() .filter_map(|row| { @@ -492,7 +492,7 @@ pub async fn get_author_feed( Some(AuthorFeedItem { uri, item_uri, - actor_did: actor_did.clone(), + actor_id: row.actor_id, // Return actor_id instead of DID! typ: row.typ.to_string(), sort_at: parakeet_db::tid_util::tid_to_datetime(row.rkey), }) @@ -677,7 +677,7 @@ pub async fn get_list_feed( Some(AuthorFeedItem { uri: uri.clone(), item_uri: uri, // For list feeds, uri and item_uri are the same (no reposts) - actor_did: actor_did.clone(), + actor_id: row.actor_id, typ: "post".to_string(), sort_at: parakeet_db::tid_util::tid_to_datetime(row.rkey), }) diff --git a/parakeet/src/db/posts.rs b/parakeet/src/db/posts.rs index 0fe7b319..7f45f659 100644 --- a/parakeet/src/db/posts.rs +++ b/parakeet/src/db/posts.rs @@ -374,11 +374,13 @@ pub async fn get_post_ids_by_uris( pub async fn get_reposts_by_uris( conn: &mut AsyncPgConnection, uris: &[String], -) -> QueryResult> { - use diesel::sql_types::{Array, BigInt, Nullable, Timestamp}; +) -> QueryResult> { + use diesel::sql_types::{Array, BigInt, Integer, Nullable, Timestamp}; #[derive(QueryableByName)] struct RepostData { + #[diesel(sql_type = Integer)] + actor_id: i32, #[diesel(sql_type = Text)] did: String, #[diesel(sql_type = BigInt)] @@ -414,6 +416,7 @@ pub async fn get_reposts_by_uris( // Use UNNEST with arrays to safely query (did, rkey) tuples - prevents SQL injection let results: Vec = diesel::sql_query( "SELECT + r.actor_id, a.did, r.rkey, tid_timestamp(r.rkey) as created_at @@ -433,7 +436,7 @@ pub async fn get_reposts_by_uris( row.created_at.map(|created| { let encoded_rkey = parakeet_db::tid_util::encode_tid(row.rkey); let at_uri = format!("at://{}/app.bsky.feed.repost/{}", row.did, encoded_rkey); - (at_uri, (row.did, created)) + (at_uri, (row.actor_id, created)) // Return actor_id instead of DID }) }) .collect()) diff --git a/parakeet/src/hydration/posts/feed.rs b/parakeet/src/hydration/posts/feed.rs index fd363725..38770575 100644 --- a/parakeet/src/hydration/posts/feed.rs +++ b/parakeet/src/hydration/posts/feed.rs @@ -21,18 +21,37 @@ impl StatefulHydrator<'_> { .map(|item| item.post_uri().to_owned()) .collect::>(); - // Extract repost profiles early since they only depend on input posts - let repost_profiles = posts + // Extract repost author actor_ids (already in internal format!) + let repost_actor_ids: Vec = posts .iter() - .filter_map(RawFeedItem::repost_by) - .collect::>(); + .filter_map(RawFeedItem::repost_by_actor_id) + .collect::>() // Deduplicate + .into_iter() + .collect(); // Parallelize: hydrate posts and repost profiles concurrently let hydrate_main_start = std::time::Instant::now(); - let (mut posts_hyd, profiles_hydrated) = tokio::join!( - self.hydrate_posts_inner(post_uris), - self.hydrate_profiles_basic(repost_profiles) + let repost_count = repost_actor_ids.len(); + let (mut posts_hyd, profiles_by_id) = tokio::join!( + async { + let start = std::time::Instant::now(); + let result = self.hydrate_posts_inner(post_uris).await; + let elapsed = start.elapsed().as_secs_f64() * 1000.0; + tracing::info!(" ├─ Posts hydration: {:.1} ms", elapsed); + result + }, + async { + let start = std::time::Instant::now(); + let result = self.hydrate_profiles_by_id(repost_actor_ids).await; + let elapsed = start.elapsed().as_secs_f64() * 1000.0; + tracing::info!(" ├─ Repost profiles: {:.1} ms ({} profiles)", elapsed, repost_count); + if elapsed > 50.0 { + tracing::warn!(" ├─ Slow repost profiles: {:.1} ms", elapsed); + } + result + } ); + let main_time = hydrate_main_start.elapsed().as_secs_f64() * 1000.0; tracing::info!(" ├─ Main posts + repost profiles: {:.1} ms", main_time); if main_time > 100.0 { @@ -174,13 +193,16 @@ impl StatefulHydrator<'_> { }; let reason = match item { - RawFeedItem::Repost { uri, by, at, .. } => { - Some(FeedViewPostReason::Repost(Box::new(FeedReasonRepost { - by: profiles_hydrated.get(&by).cloned()?, - uri: Some(uri), - cid: None, - indexed_at: at, - }))) + RawFeedItem::Repost { uri, by_actor_id, at, .. } => { + // Look up profile by actor_id (internal key) + profiles_by_id.get(&by_actor_id).map(|profile| { + FeedViewPostReason::Repost(Box::new(FeedReasonRepost { + by: profile.clone(), + uri: Some(uri), + cid: None, + indexed_at: at, + })) + }) } RawFeedItem::Pin { .. } => Some(FeedViewPostReason::Pin), _ => None, @@ -231,7 +253,7 @@ pub enum RawFeedItem { Repost { uri: String, post: String, - by: String, + by_actor_id: i32, // Use actor_id internally, not DID at: chrono::DateTime, context: Option, }, @@ -245,9 +267,9 @@ impl RawFeedItem { } } - fn repost_by(&self) -> Option { + fn repost_by_actor_id(&self) -> Option { match self { - Self::Repost { by, .. } => Some(by.clone()), + Self::Repost { by_actor_id, .. } => Some(*by_actor_id), _ => None, } } diff --git a/parakeet/src/hydration/profile/mod.rs b/parakeet/src/hydration/profile/mod.rs index 7649b26a..2d1c48f3 100644 --- a/parakeet/src/hydration/profile/mod.rs +++ b/parakeet/src/hydration/profile/mod.rs @@ -30,30 +30,6 @@ impl super::StatefulHydrator<'_> { )) } - pub async fn hydrate_profiles_basic( - &self, - dids: Vec, - ) -> HashMap { - let labels = self.get_profile_label_many(&dids).await; - 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; - - 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(); - - let v = build_basic(profile_info, stats, labels, verif, viewer, &self.cdn); - (k, v) - }) - .collect() - } - /// Hydrate profiles by actor IDs instead of DIDs /// /// This is optimized for cases where we already have actor_id values (e.g., from posts) diff --git a/parakeet/src/hydration/starter_packs.rs b/parakeet/src/hydration/starter_packs.rs index f87d5339..d7b82471 100644 --- a/parakeet/src/hydration/starter_packs.rs +++ b/parakeet/src/hydration/starter_packs.rs @@ -147,7 +147,15 @@ impl StatefulHydrator<'_> { } } - let creators: Vec = result_with_uris.values().map(|enriched| enriched.owner.clone()).collect(); + let creator_dids: Vec = result_with_uris.values().map(|enriched| enriched.owner.clone()).collect(); + + // Resolve creator DIDs to actor_ids for optimized loading + let creator_actor_map = id_cache.get_actor_ids(&creator_dids).await; + let creator_actor_ids: Vec = creator_actor_map.values().map(|cached| cached.actor_id).collect(); + let creator_did_to_actor_id: std::collections::HashMap = creator_actor_map + .into_iter() + .map(|(did, cached)| (did, cached.actor_id)) + .collect(); // Parse list URIs and resolve DIDs to natural keys let mut list_natural_keys = Vec::new(); @@ -168,7 +176,19 @@ impl StatefulHydrator<'_> { } } - let creators = self.hydrate_profiles_basic(creators).await; + // Load profiles by actor_id (optimized) + let creators_by_id = self.hydrate_profiles_by_id(creator_actor_ids).await; + + // Convert back to DID-keyed for lookup + let creators: std::collections::HashMap = creator_dids + .iter() + .filter_map(|did| { + creator_did_to_actor_id.get(did) + .and_then(|actor_id| creators_by_id.get(actor_id)) + .map(|profile| (did.clone(), profile.clone())) + }) + .collect(); + let lists = self.loaders.list.load_many(list_natural_keys).await; result_with_uris @@ -271,7 +291,7 @@ impl StatefulHydrator<'_> { } } - let (creators, lists) = result_with_uris + let (creator_dids, lists): (Vec, Vec) = result_with_uris .values() .map(|enriched| (enriched.owner.clone(), enriched.list.clone())) .unzip(); @@ -281,7 +301,27 @@ impl StatefulHydrator<'_> { .flat_map(Vec::from) .collect(); - let creators = self.hydrate_profiles_basic(creators).await; + // Resolve creator DIDs to actor_ids for optimized loading + let creator_actor_map = id_cache.get_actor_ids(&creator_dids).await; + let creator_actor_ids: Vec = creator_actor_map.values().map(|cached| cached.actor_id).collect(); + let creator_did_to_actor_id: std::collections::HashMap = creator_actor_map + .into_iter() + .map(|(did, cached)| (did, cached.actor_id)) + .collect(); + + // Load profiles by actor_id (optimized) + let creators_by_id = self.hydrate_profiles_by_id(creator_actor_ids).await; + + // Convert back to DID-keyed for lookup + let creators: std::collections::HashMap = creator_dids + .iter() + .filter_map(|did| { + creator_did_to_actor_id.get(did) + .and_then(|actor_id| creators_by_id.get(actor_id)) + .map(|profile| (did.clone(), profile.clone())) + }) + .collect(); + let lists = self.hydrate_lists_basic(lists).await; let feeds = self.hydrate_feedgens(feeds).await; diff --git a/parakeet/src/xrpc/app_bsky/feed/get_timeline.rs b/parakeet/src/xrpc/app_bsky/feed/get_timeline.rs index e9aa00e7..9f91db4e 100644 --- a/parakeet/src/xrpc/app_bsky/feed/get_timeline.rs +++ b/parakeet/src/xrpc/app_bsky/feed/get_timeline.rs @@ -401,43 +401,49 @@ async fn hydrate_timeline_feed( .await .unwrap_or_default(); - // Batch resolve all actor_ids to DIDs for reposts - let mut all_actor_ids: Vec = reposts_results + // Batch resolve post actor_ids to DIDs for URI construction only + let mut post_actor_ids: Vec = reposts_results .iter() - .flat_map(|(post_actor_id, _, reposter_actor_id, _)| vec![*post_actor_id, *reposter_actor_id]) + .map(|(post_actor_id, _, _, _)| *post_actor_id) .collect(); - all_actor_ids.sort_unstable(); - all_actor_ids.dedup(); + post_actor_ids.sort_unstable(); + post_actor_ids.dedup(); let actor_id_to_did_map = crate::id_cache_helpers::get_actor_dids_or_fetch( &state.pool, &state.id_cache, - &all_actor_ids, + &post_actor_ids, ).await.unwrap_or_default(); // Group by post_uri and take the most recent repost for each post + // Keep reposter_actor_id as i32 (internal format)! for (post_actor_id, post_rkey, reposter_actor_id, indexed_at) in reposts_results { - if let (Some(post_did), Some(reposter_did)) = ( - actor_id_to_did_map.get(&post_actor_id), - actor_id_to_did_map.get(&reposter_actor_id) - ) { + if let Some(post_did) = actor_id_to_did_map.get(&post_actor_id) { let rkey_str = parakeet_db::tid_util::encode_tid(post_rkey); let post_uri = format!("at://{}/app.bsky.feed.post/{}", post_did, rkey_str); if let std::collections::hash_map::Entry::Vacant(e) = repost_data.entry(post_uri) { - let _ = e.insert((reposter_did.clone(), indexed_at)); + let _ = e.insert((reposter_actor_id, indexed_at)); // Store actor_id, not DID! } } } - // Batch hydrate all reposter profiles - let reposter_dids: Vec = repost_data.values().map(|(did, _)| did.clone()).collect(); - let profiles = hyd.hydrate_profiles_basic(reposter_dids).await; + // Batch hydrate all reposter profiles using optimized by_id loader + // Extract unique reposter actor_ids (already in internal format!) + let reposter_actor_ids: Vec = repost_data + .values() + .map(|(actor_id, _)| *actor_id) + .collect::>() + .into_iter() + .collect(); + + // Load profiles by actor_id (optimized, no DID conversion needed!) + let profiles_by_id = hyd.hydrate_profiles_by_id(reposter_actor_ids).await; // Convert the repost data to FeedViewPostReason let mut reason_map = HashMap::new(); - for (post_uri, (did, indexed_at)) in repost_data { - if let Some(profile) = profiles.get(&did) { + for (post_uri, (actor_id, indexed_at)) in repost_data { + if let Some(profile) = profiles_by_id.get(&actor_id) { drop(reason_map.insert( post_uri, FeedViewPostReason::Repost(Box::new(FeedReasonRepost { diff --git a/parakeet/src/xrpc/app_bsky/feed/posts/feeds.rs b/parakeet/src/xrpc/app_bsky/feed/posts/feeds.rs index 997ee4b3..f104b913 100644 --- a/parakeet/src/xrpc/app_bsky/feed/posts/feeds.rs +++ b/parakeet/src/xrpc/app_bsky/feed/posts/feeds.rs @@ -122,10 +122,10 @@ pub async fn get_feed( Some(SkeletonReason::Repost { repost }) => { repost_data .remove_entry(&repost) - .map(|(uri, (by, at))| RawFeedItem::Repost { + .map(|(uri, (by_actor_id, at))| RawFeedItem::Repost { uri, post: v.post, - by, + by_actor_id, at: at.and_utc(), context: v.feed_context, }) @@ -298,7 +298,7 @@ pub async fn get_author_feed( "repost" => Some(RawFeedItem::Repost { uri: item.uri, post: item.item_uri, - by: item.actor_did, + by_actor_id: item.actor_id, at: item.sort_at, context: None, }), diff --git a/parakeet/src/xrpc/app_bsky/feed/posts/helpers.rs b/parakeet/src/xrpc/app_bsky/feed/posts/helpers.rs index 83585200..09c4e4ea 100644 --- a/parakeet/src/xrpc/app_bsky/feed/posts/helpers.rs +++ b/parakeet/src/xrpc/app_bsky/feed/posts/helpers.rs @@ -62,7 +62,7 @@ pub(super) async fn get_feed_skeleton( pub(super) async fn get_skeleton_repost_data( conn: &mut AsyncPgConnection, reposts: Vec, -) -> HashMap { +) -> HashMap { crate::db::get_reposts_by_uris(conn, &reposts) .await .unwrap_or_default() -- 2.51.2