From e36fd78c754ef2b6a68ddb257755f6a92212abd4 Mon Sep 17 00:00:00 2001 From: Timothy Quilling Date: Tue, 9 Dec 2025 16:28:56 -0500 Subject: [PATCH] fix: use denormalized threadgate --- parakeet/src/loaders/mod.rs | 2 +- parakeet/src/loaders/post.rs | 165 +++++++---------------- parakeet/tests/loaders_test.rs | 49 +------ parakeet/tests/sql/batch_loading_test.rs | 24 +--- 4 files changed, 52 insertions(+), 188 deletions(-) diff --git a/parakeet/src/loaders/mod.rs b/parakeet/src/loaders/mod.rs index 40946fe5..4b89202d 100644 --- a/parakeet/src/loaders/mod.rs +++ b/parakeet/src/loaders/mod.rs @@ -28,7 +28,7 @@ pub use feed::build_feedgens_batch_query; pub use labeler::{build_labeler_records_query, build_labels_query, build_labels_many_query}; 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, build_threadgates_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}; type CachingLoader = Loader>; diff --git a/parakeet/src/loaders/post.rs b/parakeet/src/loaders/post.rs index 0267009c..f514cc6f 100644 --- a/parakeet/src/loaders/post.rs +++ b/parakeet/src/loaders/post.rs @@ -159,6 +159,13 @@ pub struct PostWithComputed { pub reply_count: i32, #[diesel(sql_type = diesel::sql_types::Integer)] pub quote_count: i32, + // Threadgate denormalized fields (loaded inline to avoid separate query) + #[diesel(sql_type = diesel::sql_types::Nullable>>)] + pub threadgate_allow: Option>>, + #[diesel(sql_type = diesel::sql_types::Nullable>>)] + pub threadgate_hidden_actor_ids: Option>>, + #[diesel(sql_type = diesel::sql_types::Nullable>>)] + pub threadgate_hidden_rkeys: Option>>, } /// Build SQL query for batch loading posts with computed fields @@ -223,7 +230,11 @@ pub fn build_posts_batch_query() -> &'static str { p.like_count, p.repost_count, p.reply_count, - p.quote_count + p.quote_count, + -- Threadgate denormalized fields (loaded inline to avoid separate query) + p.threadgate_allow, + p.threadgate_hidden_actor_ids, + p.threadgate_hidden_rkeys FROM posts p WHERE p.rkey >= $3::bigint AND p.rkey <= $4::bigint @@ -254,26 +265,6 @@ pub fn build_actors_batch_query() -> &'static str { "SELECT id, did FROM actors WHERE id = ANY($1)" } -/// Build SQL query for batch loading threadgates from denormalized posts table -/// -/// DENORMALIZED: Threadgates are now stored directly in posts table columns. -/// This query reads threadgate_allow and threadgate_hidden_* arrays from posts. -/// Note: The threadgate's own (actor_id, rkey, cid) are no longer stored - we use -/// the post's identity as a proxy since threadgates are typically created by the post author. -/// -/// This function is public for testing purposes. -pub fn build_threadgates_batch_query() -> &'static str { - "SELECT p.actor_id, p.rkey, tid_timestamp(p.rkey) as created_at, - p.threadgate_allow, - p.threadgate_hidden_actor_ids, - p.threadgate_hidden_rkeys - FROM posts p - WHERE p.actor_id = ANY($1) - AND p.rkey = ANY($2) - AND (p.threadgate_allow IS NOT NULL - OR p.threadgate_hidden_actor_ids IS NOT NULL - OR p.threadgate_hidden_rkeys IS NOT NULL)" -} /// Facet data loaded from database #[allow(dead_code, reason = "facet_index used for SQL ordering but not in reconstruction logic")] @@ -963,6 +954,10 @@ impl BatchFn for PostLoader { p.facet_6, p.facet_7, p.facet_8, + // Threadgate data (loaded inline to avoid separate query) + p.threadgate_allow, + p.threadgate_hidden_actor_ids, + p.threadgate_hidden_rkeys, )) }) .collect(); @@ -972,7 +967,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); @@ -994,56 +989,14 @@ impl BatchFn for PostLoader { }; let facets_time = facets_start.elapsed().as_secs_f64() * 1000.0; - // Extract post natural keys for batch loading threadgates + // OPTIMIZATION: Process threadgates inline (no separate query!) + // Threadgate data is now included in the main posts query let threadgates_start = std::time::Instant::now(); - let post_keys: Vec<(i32, i64)> = posts_with_computed_data - .iter() - .map(|(post, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _)| (post.actor_id, post.rkey)) - .collect(); - - // Load threadgates from denormalized posts table columns (NO actors JOIN!) - - #[derive(diesel::QueryableByName)] - struct ThreadgateData { - #[diesel(sql_type = diesel::sql_types::Integer)] - actor_id: i32, - #[diesel(sql_type = diesel::sql_types::BigInt)] - rkey: i64, - #[diesel(sql_type = diesel::sql_types::Timestamptz)] - created_at: chrono::DateTime, - #[diesel(sql_type = diesel::sql_types::Nullable>>)] - threadgate_allow: Option>>, - #[diesel(sql_type = diesel::sql_types::Nullable>>)] - threadgate_hidden_actor_ids: Option>>, - #[diesel(sql_type = diesel::sql_types::Nullable>>)] - threadgate_hidden_rkeys: Option>>, - } - - let threadgates_data: Vec = if !post_keys.is_empty() { - // Split into separate arrays for query - let post_actor_ids: Vec = post_keys.iter().map(|(actor_id, _)| *actor_id).collect(); - let post_rkeys: Vec = post_keys.iter().map(|(_, rkey)| *rkey).collect(); - - diesel_async::RunQueryDsl::load( - diesel::sql_query(build_threadgates_batch_query()) - .bind::, _>(&post_actor_ids) - .bind::, _>(&post_rkeys), - &mut conn, - ) - .await - .unwrap_or_default() - } else { - vec![] - }; - - // OPTIMIZATION: Extract hidden_replies from denormalized arrays (no junction table query!) - // Build map of hidden reply URIs by post natural key, extracted from parallel arrays - let mut hidden_replies_by_post: HashMap<(i32, i64), Vec> = HashMap::new(); // Collect unique actor IDs from hidden replies for batch DID lookup let mut hidden_reply_actor_ids: Vec = Vec::new(); - for tg in &threadgates_data { - if let Some(actor_ids) = &tg.threadgate_hidden_actor_ids { + 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)); } } @@ -1051,7 +1004,7 @@ impl BatchFn for PostLoader { hidden_reply_actor_ids.dedup(); // Batch lookup actor DIDs for hidden replies using id_cache - let hidden_reply_actor_lookup: HashMap = if !hidden_reply_actor_ids.is_empty() { + let _hidden_reply_actor_lookup: HashMap = if !hidden_reply_actor_ids.is_empty() { self.1.get_actor_data_many(&hidden_reply_actor_ids) .await .into_iter() @@ -1061,54 +1014,25 @@ impl BatchFn for PostLoader { HashMap::new() }; - // Process each threadgate's hidden reply arrays - for tg in &threadgates_data { - if let (Some(actor_ids), Some(rkeys)) = ( - &tg.threadgate_hidden_actor_ids, - &tg.threadgate_hidden_rkeys, - ) { - // Parallel arrays - zip them together - let hidden_uris: Vec = actor_ids - .iter() - .zip(rkeys.iter()) - .filter_map(|(maybe_actor_id, maybe_rkey)| { - if let (Some(actor_id), Some(rkey)) = (maybe_actor_id, maybe_rkey) { - // Look up DID from cache - hidden_reply_actor_lookup.get(actor_id).map(|did| { - let encoded_rkey = parakeet_db::tid_util::encode_tid(*rkey); - format!("at://{}/app.bsky.feed.post/{}", did, encoded_rkey) - }) - } else { - None - } - }) - .collect(); - - if !hidden_uris.is_empty() { - hidden_replies_by_post.insert((tg.actor_id, tg.rkey), hidden_uris); + // 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)| { + // 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; } - } - } - - // Build threadgate map keyed by (actor_id, rkey) - NO URI construction! - let threadgate_map: HashMap<(i32, i64), EnrichedThreadgate> = threadgates_data - .into_iter() - .filter_map(|tg| { use parakeet_db::models::array_helpers::ThreadgateRuleArray; // Generate synthetic CID for threadgate (deterministic based on post natural key) // Since we no longer store the real threadgate CID, we create a consistent placeholder - let synthetic_cid = parakeet_db::cid_util::post_cid_string(tg.actor_id, tg.rkey); + let synthetic_cid = parakeet_db::cid_util::post_cid_string(post.actor_id, post.rkey); // TODO: Load allowed_lists from threadgate_allowed_lists junction table if needed let allowed_lists = None; - // Get hidden_replies for this threadgate using natural key - let hidden_replies_actor_ids = tg.threadgate_hidden_actor_ids.clone(); - let hidden_replies_rkeys = tg.threadgate_hidden_rkeys.clone(); - // Store hidden replies as (actor_id, rkey) pairs - URI construction deferred to hydration - let hidden_reply_keys: Vec<(i32, i64)> = if let (Some(actor_ids), Some(rkeys)) = (&hidden_replies_actor_ids, &hidden_replies_rkeys) { + let hidden_reply_keys: Vec<(i32, i64)> = if let (Some(actor_ids), Some(rkeys)) = (threadgate_hidden_actor_ids, threadgate_hidden_rkeys) { actor_ids.iter().zip(rkeys.iter()) .filter_map(|(maybe_actor_id, maybe_rkey)| { if let (Some(actor_id), Some(rkey)) = (maybe_actor_id, maybe_rkey) { @@ -1126,11 +1050,11 @@ impl BatchFn for PostLoader { // We need to store threadgate_allow and hidden_reply_keys for hydration let mut record = serde_json::json!({ "$type": "app.bsky.feed.threadgate", - "createdAt": tg.created_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true), + "createdAt": created_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true), }); // Add allow rules if present - if let Some(rules) = &tg.threadgate_allow { + if let Some(rules) = threadgate_allow { let rules_json: Vec = rules .iter() .filter_map(|r| r.as_ref()) @@ -1165,22 +1089,22 @@ impl BatchFn for PostLoader { // DENORMALIZED: Use post's natural key since threadgate identity is no longer stored let enriched = EnrichedThreadgate { threadgate: models::Threadgate { - actor_id: tg.actor_id, // Post's actor_id (threadgate typically owned by post author) - rkey: tg.rkey, // Post's rkey (synthetic - threadgate has own rkey in reality) + actor_id: post.actor_id, // Post's actor_id (threadgate typically owned by post author) + rkey: post.rkey, // Post's rkey (synthetic - threadgate has own rkey in reality) cid: synthetic_cid.as_bytes().to_vec(), // Synthetic CID bytes - allow: tg - .threadgate_allow - .map(|v| ThreadgateRuleArray(v.into_iter().flatten().collect())), - post_actor_id: tg.actor_id, // Same as actor_id (denormalized in post) - post_rkey: tg.rkey, // Same as rkey (denormalized in post) + allow: threadgate_allow + .as_ref() + .map(|v| ThreadgateRuleArray(v.iter().filter_map(|r| *r).collect())), + post_actor_id: post.actor_id, // Same as actor_id (denormalized in post) + post_rkey: post.rkey, // Same as rkey (denormalized in post) }, cid: synthetic_cid, record, allowed_lists, - actor_id: tg.actor_id, - rkey: tg.rkey, + actor_id: post.actor_id, + rkey: post.rkey, }; - Some(((tg.actor_id, tg.rkey), enriched)) + Some(((post.actor_id, post.rkey), enriched)) }) .collect(); let threadgates_time = threadgates_start.elapsed().as_secs_f64() * 1000.0; @@ -1219,6 +1143,9 @@ impl BatchFn for PostLoader { facet_6, facet_7, facet_8, + _threadgate_allow, + _threadgate_hidden_actor_ids, + _threadgate_hidden_rkeys, )| { // 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/tests/loaders_test.rs b/parakeet/tests/loaders_test.rs index 66e1fbff..06ff3815 100644 --- a/parakeet/tests/loaders_test.rs +++ b/parakeet/tests/loaders_test.rs @@ -272,52 +272,9 @@ async fn test_post_loader_threadgate_query_structure() -> eyre::Result<()> { Ok(()) } -#[tokio::test] -async fn test_post_loader_threadgate_query_with_data() -> 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 PostLoader threadgate query builder (no SQL duplication!) - let query = parakeet::loaders::build_threadgates_batch_query(); - - #[derive(diesel::QueryableByName)] - #[allow(dead_code)] - struct ThreadgateWithUri { - #[diesel(sql_type = diesel::sql_types::BigInt)] - id: i64, - #[diesel(sql_type = diesel::sql_types::Integer)] - actor_id: i32, - #[diesel(sql_type = diesel::sql_types::BigInt)] - rkey: i64, - #[diesel(sql_type = diesel::sql_types::Binary)] - cid: Vec, - #[diesel(sql_type = diesel::sql_types::Timestamptz)] - created_at: chrono::DateTime, - #[diesel(sql_type = diesel::sql_types::Nullable)] - post_id: Option, - #[diesel(sql_type = diesel::sql_types::Nullable>>)] - allow: Option>>, - #[diesel(sql_type = diesel::sql_types::Text)] - post_uri: String, - #[diesel(sql_type = diesel::sql_types::Text)] - at_uri: String, - #[diesel(sql_type = diesel::sql_types::Text)] - cid_str: String, - } - - let actor_ids: Vec = vec![]; // Empty for SQL structure test - let rkeys: Vec = vec![]; - - diesel::sql_query(query) - .bind::, _>(actor_ids) - .bind::, _>(rkeys) - .load::(&mut conn) - .await - .wrap_err("PostLoader threadgate query SQL failed")?; - - Ok(()) -} +// NOTE: Threadgate loading test removed - threadgates are now loaded inline with posts +// in build_posts_batch_query() to eliminate a separate database roundtrip. +// The post loading test (test_post_loader_with_data) now covers threadgate loading. // ============================================================================ // HANDLE LOADER TEST diff --git a/parakeet/tests/sql/batch_loading_test.rs b/parakeet/tests/sql/batch_loading_test.rs index d0568309..4553dba1 100644 --- a/parakeet/tests/sql/batch_loading_test.rs +++ b/parakeet/tests/sql/batch_loading_test.rs @@ -46,28 +46,8 @@ async fn test_post_loader_batch_query() -> eyre::Result<()> { Ok(()) } -/// Test threadgate batch loading query (from loaders/post.rs) -#[tokio::test] -async fn test_threadgate_loader_batch_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_threadgates_batch_query(); - - // Test with empty arrays (edge case) - diesel_async::RunQueryDsl::execute( - diesel::sql_query(query) - .bind::, _>(Vec::::new()) - .bind::, _>(Vec::::new()), - &mut conn, - ) - .await - .wrap_err("Threadgate batch loading query failed")?; - - Ok(()) -} +// 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) #[tokio::test] -- 2.51.2