From 8efb05f2b620692d7f1bf8fbfc3d45d09da41060 Mon Sep 17 00:00:00 2001 From: Timothy Quilling Date: Tue, 18 Nov 2025 19:26:41 -0500 Subject: [PATCH] chore: finish splitting likes --- consumer/src/database_writer/bulk_types.rs | 37 +++++++++++++++------- consumer/src/database_writer/workers.rs | 16 +++------- consumer/src/db/bulk_copy/mod.rs | 31 ++++++++---------- consumer/tests/bulk_copy_queries_test.rs | 5 ++- parakeet/src/sql/post_state.sql.old | 37 ---------------------- 5 files changed, 45 insertions(+), 81 deletions(-) delete mode 100644 parakeet/src/sql/post_state.sql.old diff --git a/consumer/src/database_writer/bulk_types.rs b/consumer/src/database_writer/bulk_types.rs index ca7d662c..e73f4c31 100644 --- a/consumer/src/database_writer/bulk_types.rs +++ b/consumer/src/database_writer/bulk_types.rs @@ -217,20 +217,33 @@ pub struct PostCopyData { /// Result of a successful COPY operation for likes /// -/// Returned from COPY operation via RETURNING clause. -/// Used for aggregate delta generation. +/// Result of inserting a post like via bulk COPY #[derive(Debug)] -pub struct InsertedLike { - /// The actor_id who liked +pub struct InsertedPostLike { + /// The actor_id who liked the post pub actor_id: i32, - /// The subject type that was liked ("post", "feedgen", "labeler") - pub subject_type: String, - /// For post likes: the actor_id of the post (natural key part 1) - /// For feedgen/labeler likes: the synthetic ID - pub subject_id: i64, - /// For post likes: the rkey of the post (natural key part 2) - /// For feedgen/labeler likes: None - pub subject_rkey: Option, + /// The post's actor_id (natural key part 1) + pub post_actor_id: i32, + /// The post's rkey (natural key part 2) + pub post_rkey: i64, +} + +/// Result of inserting a feedgen like via bulk COPY +#[derive(Debug)] +pub struct InsertedFeedgenLike { + /// The actor_id who liked the feedgen + pub actor_id: i32, + /// The feedgen's synthetic ID + pub feedgen_id: i64, +} + +/// Result of inserting a labeler like via bulk COPY +#[derive(Debug)] +pub struct InsertedLabelerLike { + /// The actor_id who liked the labeler + pub actor_id: i32, + /// The labeler's actor_id + pub labeler_actor_id: i32, } /// Result of a successful COPY operation for follows diff --git a/consumer/src/database_writer/workers.rs b/consumer/src/database_writer/workers.rs index e532519f..33459f0e 100644 --- a/consumer/src/database_writer/workers.rs +++ b/consumer/src/database_writer/workers.rs @@ -68,17 +68,11 @@ async fn execute_bulk_operations( // Generate deltas: increment like count for each liked post use super::operations::AggregateTarget; for like in inserted { - // Only generate deltas for post likes (which have subject_rkey) - // Feedgen and labeler likes don't affect post aggregates - if like.subject_type == "post" { - if let Some(rkey) = like.subject_rkey { - deltas.push(AggregateDelta { - target: AggregateTarget::Post(like.subject_id as i32, rkey), - delta_type: AggregateType::Like, - delta: 1, - }); - } - } + deltas.push(AggregateDelta { + target: AggregateTarget::Post(like.post_actor_id, like.post_rkey), + delta_type: AggregateType::Like, + delta: 1, + }); } } Err(e) => { diff --git a/consumer/src/db/bulk_copy/mod.rs b/consumer/src/db/bulk_copy/mod.rs index 5d95dfe5..13ce8ace 100644 --- a/consumer/src/db/bulk_copy/mod.rs +++ b/consumer/src/db/bulk_copy/mod.rs @@ -40,9 +40,9 @@ pub mod queries; use crate::database_writer::bulk_types::{ - BlockCopyData, FeedgenLikeCopyData, FollowCopyData, InsertedBlock, InsertedFollow, - InsertedLike, InsertedPost, InsertedRepost, LabelerLikeCopyData, PostCopyData, - PostLikeCopyData, RepostCopyData, + BlockCopyData, FeedgenLikeCopyData, FollowCopyData, InsertedBlock, InsertedFeedgenLike, + InsertedFollow, InsertedLabelerLike, InsertedPost, InsertedPostLike, InsertedRepost, + LabelerLikeCopyData, PostCopyData, PostLikeCopyData, RepostCopyData, }; use crate::Result; use deadpool_postgres::{GenericClient, Transaction}; @@ -86,7 +86,7 @@ use tokio_postgres::types::Type; pub async fn copy_post_likes( conn: &Transaction<'_>, likes: Vec, -) -> Result> { +) -> Result> { if likes.is_empty() { return Ok(Vec::new()); } @@ -163,11 +163,10 @@ pub async fn copy_post_likes( let post_actor_id: i32 = row.get(1); let post_rkey: i64 = row.get(2); - result.push(InsertedLike { + result.push(InsertedPostLike { actor_id, - subject_type: "post".to_string(), - subject_id: post_actor_id as i64, - subject_rkey: Some(post_rkey), + post_actor_id, + post_rkey, }); } @@ -193,7 +192,7 @@ pub async fn copy_post_likes( pub async fn copy_feedgen_likes( conn: &Transaction<'_>, likes: Vec, -) -> Result> { +) -> Result> { if likes.is_empty() { return Ok(Vec::new()); } @@ -255,11 +254,9 @@ pub async fn copy_feedgen_likes( let actor_id: i32 = row.get(0); let feedgen_id: i64 = row.get(1); - result.push(InsertedLike { + result.push(InsertedFeedgenLike { actor_id, - subject_type: "feedgen".to_string(), - subject_id: feedgen_id, - subject_rkey: None, + feedgen_id, }); } @@ -285,7 +282,7 @@ pub async fn copy_feedgen_likes( pub async fn copy_labeler_likes( conn: &Transaction<'_>, likes: Vec, -) -> Result> { +) -> Result> { if likes.is_empty() { return Ok(Vec::new()); } @@ -346,11 +343,9 @@ pub async fn copy_labeler_likes( let actor_id: i32 = row.get(0); let labeler_actor_id: i32 = row.get(1); - result.push(InsertedLike { + result.push(InsertedLabelerLike { actor_id, - subject_type: "labeler".to_string(), - subject_id: labeler_actor_id as i64, - subject_rkey: None, + labeler_actor_id, }); } diff --git a/consumer/tests/bulk_copy_queries_test.rs b/consumer/tests/bulk_copy_queries_test.rs index 59b59eca..77cb9069 100644 --- a/consumer/tests/bulk_copy_queries_test.rs +++ b/consumer/tests/bulk_copy_queries_test.rs @@ -357,9 +357,8 @@ async fn test_insert_post_likes_from_staging_with_data() -> eyre::Result<()> { let likes = result.unwrap(); assert_eq!(likes.len(), 1); assert_eq!(likes[0].actor_id, actor_id); - assert_eq!(likes[0].subject_type, "post"); - // Natural key fields (actor_id part in subject_id, rkey in subject_rkey) - assert!(likes[0].subject_rkey.is_some()); + assert_eq!(likes[0].post_actor_id, post_actor_id); + assert_eq!(likes[0].post_rkey, post_rkey); txn.rollback().await?; Ok(()) diff --git a/parakeet/src/sql/post_state.sql.old b/parakeet/src/sql/post_state.sql.old deleted file mode 100644 index f7b8f365..00000000 --- a/parakeet/src/sql/post_state.sql.old +++ /dev/null @@ -1,37 +0,0 @@ --- Get post state (likes, reposts, bookmarks, postgates, pinned status) for a viewer --- $1 = viewer DID --- $2 = array of post dids --- $3 = array of post rkeys (bigint) --- --- OPTIMIZED: Use CTE to look up viewer actor_id once (was repeated 3 times in LEFT JOINs) -WITH viewer_info AS ( - SELECT - a.id as viewer_id, - pa.did as pinned_did, - pp.rkey as pinned_rkey, - pp.cid as pinned_cid - FROM actors a - LEFT JOIN profiles prof ON prof.actor_id = a.id - LEFT JOIN posts pp ON prof.pinned_post_id = pp.id - LEFT JOIN actors pa ON pp.actor_id = pa.id - WHERE a.did = $1 -) -SELECT - a.did, - p.actor_id, - p.rkey, - p.cid, - l.rkey as like_rkey, - r.rkey as repost_rkey, - b.actor_id IS NOT NULL as bookmarked, - COALESCE(pg.rules && ARRAY['app.bsky.feed.postgate#disableRule']::postgate_rule[], false) as embed_disabled, - COALESCE(a.did = v.pinned_did AND p.rkey = v.pinned_rkey, false) as pinned -FROM posts p -INNER JOIN actors a ON p.actor_id = a.id -INNER JOIN unnest($2::text[], $3::bigint[]) AS lookup(lookup_did, lookup_rkey) - ON a.did = lookup.lookup_did AND p.rkey = lookup.lookup_rkey -CROSS JOIN viewer_info v -LEFT JOIN post_likes l ON l.post_id = p.id AND l.actor_id = v.viewer_id -LEFT JOIN reposts r ON r.post_id = p.id AND r.actor_id = v.viewer_id -LEFT JOIN bookmarks b ON b.subject_type = 'post' AND b.subject_id = p.id AND b.actor_id = v.viewer_id -LEFT JOIN postgates pg ON pg.post_id = p.id; -- 2.51.2