From 74437cc84fb20181ad4bfadb5a60cc28f8bb5826 Mon Sep 17 00:00:00 2001 From: Timothy Quilling Date: Thu, 04 Dec 2025 23:01:27 +0000 Subject: [PATCH] fix: try not to segfault postgresql --- migrations/2025-12-04-220007_replace_via_repost_arrays_with_jsonb/down.sql | 41 +++++++++++++++++++++++++++++++++++++++++ migrations/2025-12-04-220007_replace_via_repost_arrays_with_jsonb/up.sql | 34 ++++++++++++++++++++++++++++++++++ parakeet-db/src/models.rs | 47 ++++++++++++++--------------------------------- parakeet-db/src/schema.rs | 136 +--------------------------------------------------------------------------------------------------------------------------------------- consumer/src/database_writer/workers.rs | 236 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------------------------------------------------------------------------------------------------------------------------------------------- parakeet/src/db/likes.rs | 4 ++-- parakeet/src/loaders/post.rs | 3 +-- consumer/src/db/bulk_copy/mod.rs | 80 +++++++++++++++++++++++++++++++++++++++++--------------------------------------- consumer/src/db/operations/feed/like.rs | 71 +++++++++++++++++++++++++++++++++-------------------------------------- 9 file(s) changed, 252 insertion(s)(+), 400 deletion(s)(-) diff --git a/migrations/2025-12-04-220007_replace_via_repost_arrays_with_jsonb/down.sql b/migrations/2025-12-04-220007_replace_via_repost_arrays_with_jsonb/down.sql new file mode 100644 --- /dev/null +++ b/migrations/2025-12-04-220007_replace_via_repost_arrays_with_jsonb/down.sql @@ -0,0 +1,41 @@ +-- Revert JSONB back to parallel arrays + +-- Drop the GIN index +DROP INDEX IF EXISTS idx_posts_like_via_repost_data_gin; + +-- Add back the array columns +ALTER TABLE posts +ADD COLUMN like_via_repost_actors integer[], +ADD COLUMN like_via_repost_rkeys bigint[]; + +-- Migrate data back from JSONB to arrays (if needed) +-- This is complex because we need to reconstruct parallel arrays in the correct order +UPDATE posts +SET + like_via_repost_actors = ( + SELECT array_agg( + CASE + WHEN like_via_repost_data ? like_actor_ids[i]::text THEN + (like_via_repost_data->like_actor_ids[i]::text->>'actor_id')::integer + ELSE 0 + END + ORDER BY i + ) + FROM generate_series(1, array_length(like_actor_ids, 1)) i + ), + like_via_repost_rkeys = ( + SELECT array_agg( + CASE + WHEN like_via_repost_data ? like_actor_ids[i]::text THEN + (like_via_repost_data->like_actor_ids[i]::text->>'rkey')::bigint + ELSE 0 + END + ORDER BY i + ) + FROM generate_series(1, array_length(like_actor_ids, 1)) i + ) +WHERE like_via_repost_data IS NOT NULL; + +-- Drop the JSONB column +ALTER TABLE posts +DROP COLUMN like_via_repost_data; diff --git a/migrations/2025-12-04-220007_replace_via_repost_arrays_with_jsonb/up.sql b/migrations/2025-12-04-220007_replace_via_repost_arrays_with_jsonb/up.sql new file mode 100644 --- /dev/null +++ b/migrations/2025-12-04-220007_replace_via_repost_arrays_with_jsonb/up.sql @@ -0,0 +1,34 @@ +-- Replace parallel via_repost arrays with JSONB for simpler storage +-- This eliminates the need for array_fill() backfilling which was causing PostgreSQL crashes + +-- Add new JSONB column for via_repost data +-- Structure: {"30": {"actor_id": 15, "rkey": 999}, "45": {"actor_id": 20, "rkey": 1234}} +-- Key: liker's actor_id (as string) +-- Value: {actor_id: via_repost_actor_id, rkey: via_repost_rkey} +ALTER TABLE posts +ADD COLUMN like_via_repost_data JSONB; + +-- Migrate existing data from arrays to JSONB (if any exists) +-- This handles posts that already have via_repost arrays populated +UPDATE posts +SET like_via_repost_data = ( + SELECT jsonb_object_agg( + like_actor_ids[i]::text, + jsonb_build_object( + 'actor_id', like_via_repost_actors[i], + 'rkey', like_via_repost_rkeys[i] + ) + ) + FROM generate_series(1, array_length(like_actor_ids, 1)) i + WHERE like_via_repost_actors[i] IS NOT NULL + AND like_via_repost_actors[i] != 0 +) +WHERE like_via_repost_actors IS NOT NULL; + +-- Drop the old array columns +ALTER TABLE posts +DROP COLUMN like_via_repost_actors, +DROP COLUMN like_via_repost_rkeys; + +-- Add GIN index for efficient JSONB queries +CREATE INDEX idx_posts_like_via_repost_data_gin ON posts USING GIN (like_via_repost_data); diff --git a/parakeet-db/src/models.rs b/parakeet-db/src/models.rs --- a/parakeet-db/src/models.rs +++ b/parakeet-db/src/models.rs @@ -211,14 +211,13 @@ pub reply_count: Option, pub repost_count: Option, pub quote_count: Option, - // Embedded likes (parallel arrays indexed by position) - // NULL when post has no likes, otherwise arrays are aligned by index: + // Embedded likes // like_actor_ids[i] pairs with like_rkeys[i] - // like_via_repost_actors[i] == 0 means no via_repost for that like pub like_actor_ids: Option>, // [actor1, actor2, ...] (who liked) pub like_rkeys: Option>, // [rkey1, rkey2, ...] (when they liked) - pub like_via_repost_actors: Option>, // [0, actor2, 0, ...] (NULL if no likes have via_repost) - pub like_via_repost_rkeys: Option>, // [0, rkey2, 0, ...] (NULL if no likes have via_repost) + // Via repost tracking as JSONB: {"30": {"actor_id": 15, "rkey": 999}, ...} + // Key: liker's actor_id (as string), Value: {actor_id, rkey} of repost they came via + pub like_via_repost_data: Option, // Note: created_at derived from TID rkey via created_at() method } @@ -233,17 +232,17 @@ let actor_id = *actor_ids.get(idx)?; let rkey = *rkeys.get(idx)?; - let via_repost_actor_id = self - .like_via_repost_actors + // Extract via_repost from JSONB if it exists for this liker + let (via_repost_actor_id, via_repost_rkey) = self + .like_via_repost_data .as_ref() - .and_then(|arr| arr.get(idx).copied()) - .filter(|&id| id > 0); - - let via_repost_rkey = self - .like_via_repost_rkeys - .as_ref() - .and_then(|arr| arr.get(idx).copied()) - .filter(|&rk| rk > 0); + .and_then(|json| json.get(actor_id.to_string())) + .and_then(|data| { + let actor = data.get("actor_id")?.as_i64()? as i32; + let rkey = data.get("rkey")?.as_i64()?; + Some((Some(actor), Some(rkey))) + }) + .unwrap_or((None, None)); Some(PostLikeInfo { actor_id, @@ -394,22 +393,6 @@ // ============================================================================= // SOCIAL INTERACTIONS // ============================================================================= - -// Post Likes (99.98% of likes) -#[derive(Clone, Debug, Queryable, Selectable, Identifiable)] -#[diesel(table_name = crate::schema::post_likes)] -#[diesel(primary_key(actor_id, rkey))] -#[diesel(check_for_backend(diesel::pg::Pg))] -pub struct PostLike { - pub actor_id: i32, // PK part 1: FK to actors - pub rkey: i64, // PK part 2: TID as INT8 (also partition key for hypertable) - pub post_actor_id: i32, // Natural key reference to posts (hypertable → hypertable, NO FK) - NOT NULL - pub post_rkey: i64, // Natural key reference to posts - NOT NULL - pub via_repost_actor_id: Option, // Natural key reference to reposts (future-proof if reposts becomes hypertable) - pub via_repost_rkey: Option, - // Note: created_at derived from TID rkey via created_at() method - // Note: CID no longer stored - synthetic CIDs generated from (actor_id, rkey) -} // Feed Generator Likes (rare, ~0.01% of likes) #[derive(Clone, Debug, Queryable, Selectable, Identifiable)] @@ -857,7 +840,6 @@ impl_tid_rkey!(Post); impl_tid_rkey!(Postgate); -impl_tid_rkey!(PostLike); impl_tid_rkey!(FeedgenLike); impl_tid_rkey!(LabelerLike); impl_tid_rkey!(Repost); @@ -880,7 +862,6 @@ impl_tid_created_at!(Post); impl_tid_created_at!(Postgate); -impl_tid_created_at!(PostLike); impl_tid_created_at!(FeedgenLike); impl_tid_created_at!(LabelerLike); impl_tid_created_at!(Repost); diff --git a/parakeet-db/src/schema.rs b/parakeet-db/src/schema.rs --- a/parakeet-db/src/schema.rs +++ b/parakeet-db/src/schema.rs @@ -86,10 +86,6 @@ pub struct PostImageEmbed; #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)] - #[diesel(postgres_type(name = "post_like_embed"))] - pub struct PostLikeEmbed; - - #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)] #[diesel(postgres_type(name = "post_stat_type"))] pub struct PostStatType; @@ -478,17 +474,6 @@ } diesel::table! { - post_likes (actor_id, rkey) { - actor_id -> Int4, - rkey -> Int8, - post_actor_id -> Int4, - post_rkey -> Int8, - via_repost_actor_id -> Nullable, - via_repost_rkey -> Nullable, - } -} - -diesel::table! { postgate_detached (post_actor_id, post_rkey, detached_post_actor_id, detached_post_rkey) { post_actor_id -> Int4, post_rkey -> Int8, @@ -561,8 +546,7 @@ quote_count -> Nullable, like_actor_ids -> Nullable>, like_rkeys -> Nullable>, - like_via_repost_actors -> Nullable>, - like_via_repost_rkeys -> Nullable>, + like_via_repost_data -> Nullable, } } @@ -620,113 +604,6 @@ list_id -> Int8, search_vector -> Nullable, status -> RecordStatus, - } -} - -diesel::table! { - test_likes_baseline (actor_id, rkey) { - actor_id -> Int4, - rkey -> Int8, - post_actor_id -> Int4, - post_rkey -> Int8, - via_repost_actor_id -> Nullable, - via_repost_rkey -> Nullable, - } -} - -diesel::table! { - test_posts_baseline (actor_id, rkey) { - actor_id -> Int4, - rkey -> Int8, - like_count -> Nullable, - } -} - -diesel::table! { - use diesel::sql_types::*; - use super::sql_types::PostLikeEmbed; - - test_posts_composite (actor_id, rkey) { - actor_id -> Int4, - rkey -> Int8, - like_count -> Nullable, - likes -> Nullable>>, - } -} - -diesel::table! { - test_posts_empty_arrays (actor_id, rkey) { - actor_id -> Int4, - rkey -> Int8, - like_count -> Nullable, - like_actor_ids -> Array>, - like_rkeys -> Array>, - like_via_repost_actors -> Array>, - like_via_repost_rkeys -> Array>, - } -} - -diesel::table! { - test_posts_intarray (actor_id, rkey) { - actor_id -> Int4, - rkey -> Int8, - like_count -> Nullable, - likes_flat -> Nullable>>, - } -} - -diesel::table! { - test_posts_intarray_v2 (actor_id, rkey) { - actor_id -> Int4, - rkey -> Int8, - like_count -> Nullable, - likes_flat -> Nullable>>, - like_actor_ids -> Nullable>>, - } -} - -diesel::table! { - test_posts_jsonb (actor_id, rkey) { - actor_id -> Int4, - rkey -> Int8, - like_count -> Nullable, - likes -> Nullable, - } -} - -diesel::table! { - test_posts_null_arrays (actor_id, rkey) { - actor_id -> Int4, - rkey -> Int8, - like_count -> Nullable, - like_actor_ids -> Nullable>, - like_rkeys -> Nullable>, - like_via_repost_actors -> Nullable>, - like_via_repost_rkeys -> Nullable>, - } -} - -diesel::table! { - test_posts_parallel_arrays (actor_id, rkey) { - actor_id -> Int4, - rkey -> Int8, - like_count -> Nullable, - like_actor_ids -> Nullable>, - like_rkeys -> Nullable>, - like_via_repost_actors -> Nullable>, - like_via_repost_rkeys -> Nullable>, - } -} - -diesel::table! { - test_posts_parallel_opt (actor_id, rkey) { - actor_id -> Int4, - rkey -> Int8, - like_count -> Nullable, - like_actor_ids -> Nullable>, - like_rkeys -> Nullable>, - like_via_repost_actors -> Nullable>, - like_via_repost_rkeys -> Nullable>, } } @@ -824,7 +701,6 @@ mutes, notifications, post_aggregate_stats, - post_likes, postgate_detached, postgates, posts, @@ -832,16 +708,6 @@ spatial_ref_sys, starterpack_feeds, starterpacks, - test_likes_baseline, - test_posts_baseline, - test_posts_composite, - test_posts_empty_arrays, - test_posts_intarray, - test_posts_intarray_v2, - test_posts_jsonb, - test_posts_null_arrays, - test_posts_parallel_arrays, - test_posts_parallel_opt, thread_mutes, threadgate_allowed_lists, threadgate_hidden_replies, diff --git a/consumer/src/database_writer/workers.rs b/consumer/src/database_writer/workers.rs --- a/consumer/src/database_writer/workers.rs +++ b/consumer/src/database_writer/workers.rs @@ -327,60 +327,52 @@ } let start = std::time::Instant::now(); + let total_posts = deltas.len(); - // Build arrays for UNNEST - let mut actor_ids: Vec = Vec::with_capacity(deltas.len()); - let mut rkeys: Vec = Vec::with_capacity(deltas.len()); - let mut like_deltas: Vec = Vec::with_capacity(deltas.len()); - let mut reply_deltas: Vec = Vec::with_capacity(deltas.len()); - let mut repost_deltas: Vec = Vec::with_capacity(deltas.len()); - let mut quote_deltas: Vec = Vec::with_capacity(deltas.len()); + // Update posts individually (no FROM clause, no decompression limit issues) + // Prepare statement once for reuse + let stmt = conn.prepare_typed( + "UPDATE posts + SET + like_count = LEAST(32767, GREATEST(0, COALESCE(like_count, 0) + $3))::smallint, + reply_count = LEAST(32767, GREATEST(0, COALESCE(reply_count, 0) + $4))::smallint, + repost_count = LEAST(32767, GREATEST(0, COALESCE(repost_count, 0) + $5))::smallint, + quote_count = LEAST(32767, GREATEST(0, COALESCE(quote_count, 0) + $6))::smallint + WHERE actor_id = $1 AND rkey = $2", + &[ + tokio_postgres::types::Type::INT4, // actor_id + tokio_postgres::types::Type::INT8, // rkey + tokio_postgres::types::Type::INT4, // like_delta + tokio_postgres::types::Type::INT4, // reply_delta + tokio_postgres::types::Type::INT4, // repost_delta + tokio_postgres::types::Type::INT4, // quote_delta + ], + ).await?; + let mut updated_count = 0; for ((actor_id, rkey), delta) in deltas { - actor_ids.push(actor_id); - rkeys.push(rkey); - like_deltas.push(delta.like_delta); - reply_deltas.push(delta.reply_delta); - repost_deltas.push(delta.repost_delta); - quote_deltas.push(delta.quote_delta); + let rows = conn.execute( + &stmt, + &[ + &actor_id, + &rkey, + &delta.like_delta, + &delta.reply_delta, + &delta.repost_delta, + &delta.quote_delta, + ], + ).await?; + updated_count += rows; } - // Update posts with accumulated deltas - // Uses COALESCE to handle NULL (0) + delta = delta - let query = " - UPDATE posts - SET - like_count = LEAST(32767, GREATEST(0, COALESCE(like_count, 0) + d.like_delta))::smallint, - reply_count = LEAST(32767, GREATEST(0, COALESCE(reply_count, 0) + d.reply_delta))::smallint, - repost_count = LEAST(32767, GREATEST(0, COALESCE(repost_count, 0) + d.repost_delta))::smallint, - quote_count = LEAST(32767, GREATEST(0, COALESCE(quote_count, 0) + d.quote_delta))::smallint - FROM ( - SELECT - unnest($1::int[]) as actor_id, - unnest($2::bigint[]) as rkey, - unnest($3::int[]) as like_delta, - unnest($4::int[]) as reply_delta, - unnest($5::int[]) as repost_delta, - unnest($6::int[]) as quote_delta - ) d - WHERE posts.actor_id = d.actor_id AND posts.rkey = d.rkey"; - - match conn.execute(query, &[&actor_ids, &rkeys, &like_deltas, &reply_deltas, &repost_deltas, "e_deltas]).await { - Ok(rows) => { - let elapsed = start.elapsed(); - tracing::debug!( - posts_updated = rows, - deltas_count = actor_ids.len(), - duration_ms = elapsed.as_millis(), - "Batch updated post stats" - ); - Ok(()) - } - Err(e) => { - tracing::error!(error = ?e, "Failed to batch update post stats"); - Err(e.into()) - } - } + let elapsed = start.elapsed(); + tracing::debug!( + posts_updated = updated_count, + deltas_count = total_posts, + duration_ms = elapsed.as_millis(), + "Batch updated post stats" + ); + Ok(()) } /// Batch update actor stats in the database @@ -399,112 +391,54 @@ let start = std::time::Instant::now(); let total_actors = deltas.len(); - // Build arrays for all deltas - let mut actor_ids: Vec = Vec::with_capacity(total_actors); - let mut followers_deltas: Vec = Vec::with_capacity(total_actors); - let mut following_deltas: Vec = Vec::with_capacity(total_actors); - let mut posts_deltas: Vec = Vec::with_capacity(total_actors); - let mut lists_deltas: Vec = Vec::with_capacity(total_actors); - let mut feeds_deltas: Vec = Vec::with_capacity(total_actors); - let mut starterpacks_deltas: Vec = Vec::with_capacity(total_actors); - - for (actor_id, delta) in deltas { - actor_ids.push(actor_id); - followers_deltas.push(delta.followers_delta); - following_deltas.push(delta.following_delta); - posts_deltas.push(delta.posts_delta); - lists_deltas.push(delta.lists_delta); - feeds_deltas.push(delta.feeds_delta); - starterpacks_deltas.push(delta.starterpacks_delta); - } - - // Create temp table for deltas - conn.execute( - "CREATE TEMP TABLE actor_stats_deltas ( - actor_id integer NOT NULL, - followers_delta integer NOT NULL, - following_delta integer NOT NULL, - posts_delta integer NOT NULL, - lists_delta smallint NOT NULL, - feeds_delta smallint NOT NULL, - starterpacks_delta smallint NOT NULL - ) ON COMMIT DROP", - &[], + // Update actors individually (no FROM clause, no temp table, no decompression limit issues) + // Prepare statement once for reuse + let stmt = conn.prepare_typed( + "UPDATE actors + SET + followers_count = NULLIF(LEAST(2147483647, GREATEST(0, COALESCE(followers_count, 0) + $2)), 0), + following_count = NULLIF(LEAST(2147483647, GREATEST(0, COALESCE(following_count, 0) + $3)), 0), + posts_count = NULLIF(LEAST(2147483647, GREATEST(0, COALESCE(posts_count, 0) + $4)), 0), + lists_count = NULLIF(LEAST(32767, GREATEST(0, COALESCE(lists_count, 0) + $5))::smallint, 0), + feeds_count = NULLIF(LEAST(32767, GREATEST(0, COALESCE(feeds_count, 0) + $6))::smallint, 0), + starterpacks_count = NULLIF(LEAST(32767, GREATEST(0, COALESCE(starterpacks_count, 0) + $7))::smallint, 0) + WHERE id = $1", + &[ + tokio_postgres::types::Type::INT4, // actor_id + tokio_postgres::types::Type::INT4, // followers_delta + tokio_postgres::types::Type::INT4, // following_delta + tokio_postgres::types::Type::INT4, // posts_delta + tokio_postgres::types::Type::INT2, // lists_delta + tokio_postgres::types::Type::INT2, // feeds_delta + tokio_postgres::types::Type::INT2, // starterpacks_delta + ], ).await?; - // Binary COPY deltas into temp table - let copy_sink = conn - .copy_in("COPY actor_stats_deltas FROM STDIN (FORMAT binary)") - .await?; + let mut updated_count = 0; + for (actor_id, delta) in deltas { + let rows = conn.execute( + &stmt, + &[ + &actor_id, + &delta.followers_delta, + &delta.following_delta, + &delta.posts_delta, + &delta.lists_delta, + &delta.feeds_delta, + &delta.starterpacks_delta, + ], + ).await?; + updated_count += rows; + } - use tokio_postgres::types::Type; - use tokio_postgres::binary_copy::BinaryCopyInWriter; - - let writer = BinaryCopyInWriter::new( - copy_sink, - &[ - Type::INT4, // actor_id - Type::INT4, // followers_delta - Type::INT4, // following_delta - Type::INT4, // posts_delta - Type::INT2, // lists_delta - Type::INT2, // feeds_delta - Type::INT2, // starterpacks_delta - ], + let elapsed = start.elapsed(); + tracing::debug!( + actors_updated = updated_count, + deltas_count = total_actors, + duration_ms = elapsed.as_millis(), + "Batch updated actor stats" ); - - futures::pin_mut!(writer); - - for i in 0..actor_ids.len() { - writer - .as_mut() - .write(&[ - &actor_ids[i], - &followers_deltas[i], - &following_deltas[i], - &posts_deltas[i], - &lists_deltas[i], - &feeds_deltas[i], - &starterpacks_deltas[i], - ]) - .await?; - } - - writer.finish().await?; - - // Update actors with accumulated deltas - // CRITICAL: Filter on orderby column (id) using subquery to minimize decompression - // Actors table compressed with segmentby=status, orderby=id - // Using WHERE actors.id IN (SELECT ...) pushes down the filter before join - let query = " - UPDATE actors - SET - followers_count = NULLIF(LEAST(2147483647, GREATEST(0, COALESCE(followers_count, 0) + d.followers_delta)), 0), - following_count = NULLIF(LEAST(2147483647, GREATEST(0, COALESCE(following_count, 0) + d.following_delta)), 0), - posts_count = NULLIF(LEAST(2147483647, GREATEST(0, COALESCE(posts_count, 0) + d.posts_delta)), 0), - lists_count = NULLIF(LEAST(32767, GREATEST(0, COALESCE(lists_count, 0) + d.lists_delta))::smallint, 0), - feeds_count = NULLIF(LEAST(32767, GREATEST(0, COALESCE(feeds_count, 0) + d.feeds_delta))::smallint, 0), - starterpacks_count = NULLIF(LEAST(32767, GREATEST(0, COALESCE(starterpacks_count, 0) + d.starterpacks_delta))::smallint, 0) - FROM actor_stats_deltas d - WHERE actors.id = d.actor_id - AND actors.id IN (SELECT actor_id FROM actor_stats_deltas)"; - - match conn.execute(query, &[]).await { - Ok(rows) => { - let elapsed = start.elapsed(); - tracing::debug!( - actors_updated = rows, - deltas_count = total_actors, - duration_ms = elapsed.as_millis(), - "Batch updated actor stats" - ); - Ok(()) - } - Err(e) => { - tracing::error!(error = ?e, "Failed to batch update actor stats"); - Err(e.into()) - } - } + Ok(()) } /// Resolve an UnresolvedEvent by creating actor/post stubs and getting their IDs diff --git a/parakeet/src/db/likes.rs b/parakeet/src/db/likes.rs --- a/parakeet/src/db/likes.rs +++ b/parakeet/src/db/likes.rs @@ -1,9 +1,9 @@ //! Like state queries using the self-contained schema //! -//! Post likes are now embedded in posts table as parallel arrays: +//! Post likes are now embedded in posts table: //! - like_actor_ids: array of actor IDs who liked the post //! - like_rkeys: array of like rkeys (timestamps) -//! - like_via_repost_actors/rkeys: optional via-repost data +//! - like_via_repost_data: JSONB mapping liker actor_id to via-repost data //! //! Other likes still in separate tables: //! - feedgen_likes: likes on feed generators diff --git a/parakeet/src/loaders/post.rs b/parakeet/src/loaders/post.rs --- a/parakeet/src/loaders/post.rs +++ b/parakeet/src/loaders/post.rs @@ -904,8 +904,7 @@ // Embedded likes (not loaded for hydration - viewer state loaded separately) like_actor_ids: None, like_rkeys: None, - like_via_repost_actors: None, - like_via_repost_rkeys: None, + like_via_repost_data: None, }; // Encode TIDs using Rust utility functions diff --git a/consumer/src/db/bulk_copy/mod.rs b/consumer/src/db/bulk_copy/mod.rs --- a/consumer/src/db/bulk_copy/mod.rs +++ b/consumer/src/db/bulk_copy/mod.rs @@ -161,52 +161,54 @@ // Update posts one actor at a time to leverage segmentby compression for target_actor_id in target_actors { update_count += 1; - let rows = conn + + // Fetch staging rows for this post author + let staging_rows = conn .query( - "WITH target_posts AS ( - SELECT post_rkey, actor_id, rkey, via_repost_actor_id, via_repost_rkey - FROM post_likes_staging - WHERE post_actor_id = $1 - ) - UPDATE posts p - SET - like_count = COALESCE(like_count, 0) + 1, - like_actor_ids = COALESCE(like_actor_ids, ARRAY[]::integer[]) || tp.actor_id, - like_rkeys = COALESCE(like_rkeys, ARRAY[]::bigint[]) || tp.rkey, - like_via_repost_actors = CASE - WHEN tp.via_repost_actor_id IS NOT NULL OR like_via_repost_actors IS NOT NULL THEN - COALESCE( - like_via_repost_actors, - array_fill(0::integer, ARRAY[COALESCE(array_length(like_actor_ids, 1), 0)]) - ) || COALESCE(tp.via_repost_actor_id, 0) - ELSE NULL - END, - like_via_repost_rkeys = CASE - WHEN tp.via_repost_rkey IS NOT NULL OR like_via_repost_rkeys IS NOT NULL THEN - COALESCE( - like_via_repost_rkeys, - array_fill(0::bigint, ARRAY[COALESCE(array_length(like_rkeys, 1), 0)]) - ) || COALESCE(tp.via_repost_rkey, 0) - ELSE NULL - END - FROM target_posts tp - WHERE p.actor_id = $1 - AND p.rkey = tp.post_rkey - RETURNING tp.actor_id, p.actor_id as post_actor_id, p.rkey as post_rkey", + "SELECT actor_id, rkey, post_actor_id, post_rkey, via_repost_actor_id, via_repost_rkey + FROM post_likes_staging + WHERE post_actor_id = $1 + ORDER BY post_rkey, rkey", &[&target_actor_id], ) .await?; - for row in rows { + // Process each like individually (no aggregation needed - actor can only like post once) + for row in staging_rows { let actor_id: i32 = row.get(0); - let post_actor_id: i32 = row.get(1); - let post_rkey: i64 = row.get(2); + let rkey: i64 = row.get(1); + let post_actor_id: i32 = row.get(2); + let post_rkey: i64 = row.get(3); + let via_repost_actor_id: Option = row.get(4); + let via_repost_rkey: Option = row.get(5); - result.push(InsertedPostLike { - actor_id, - post_actor_id, - post_rkey, - }); + // Use same UPDATE pattern as individual like INSERT (consumer/src/db/operations/feed/like.rs) + let updated = conn + .execute( + "UPDATE posts + SET + like_actor_ids = COALESCE(like_actor_ids, ARRAY[]::integer[]) || $3::integer, + like_rkeys = COALESCE(like_rkeys, ARRAY[]::bigint[]) || $4::bigint, + like_via_repost_data = CASE + WHEN $5::integer IS NOT NULL THEN + COALESCE(like_via_repost_data, '{}'::jsonb) || + jsonb_build_object($3::text, jsonb_build_object('actor_id', $5::integer, 'rkey', $6::bigint)) + ELSE like_via_repost_data + END, + like_count = cardinality(like_actor_ids || $3::integer) + WHERE actor_id = $1 AND rkey = $2 + AND NOT ($3 = ANY(COALESCE(like_actor_ids, ARRAY[]::integer[])))", + &[&post_actor_id, &post_rkey, &actor_id, &rkey, &via_repost_actor_id, &via_repost_rkey], + ) + .await?; + + if updated > 0 { + result.push(InsertedPostLike { + actor_id, + post_actor_id, + post_rkey, + }); + } } } diff --git a/consumer/src/db/operations/feed/like.rs b/consumer/src/db/operations/feed/like.rs --- a/consumer/src/db/operations/feed/like.rs +++ b/consumer/src/db/operations/feed/like.rs @@ -136,24 +136,20 @@ .unwrap_or((0, 0)); // Update posts table to append to like arrays - // Uses conditional array concatenation to handle NULL arrays and via_repost + // Uses JSONB merge for via_repost data (no array_fill needed!) let stmt = conn .prepare_typed( "UPDATE posts SET - like_count = COALESCE(like_count, 0) + 1, like_actor_ids = COALESCE(like_actor_ids, ARRAY[]::integer[]) || $3::integer, like_rkeys = COALESCE(like_rkeys, ARRAY[]::bigint[]) || $4::bigint, - like_via_repost_actors = CASE - WHEN $5::integer > 0 OR like_via_repost_actors IS NOT NULL THEN - COALESCE(like_via_repost_actors, array_fill(0::integer, ARRAY[COALESCE(array_length(like_actor_ids, 1), 0)])) || $5::integer - ELSE NULL + like_via_repost_data = CASE + WHEN $5::integer > 0 THEN + COALESCE(like_via_repost_data, '{}'::jsonb) || + jsonb_build_object($3::text, jsonb_build_object('actor_id', $5::integer, 'rkey', $6::bigint)) + ELSE like_via_repost_data END, - like_via_repost_rkeys = CASE - WHEN $6::bigint > 0 OR like_via_repost_rkeys IS NOT NULL THEN - COALESCE(like_via_repost_rkeys, array_fill(0::bigint, ARRAY[COALESCE(array_length(like_rkeys, 1), 0)])) || $6::bigint - ELSE NULL - END + like_count = cardinality(like_actor_ids || $3::integer) WHERE actor_id = $1 AND rkey = $2 AND NOT ($3 = ANY(COALESCE(like_actor_ids, ARRAY[]::integer[]))) -- Idempotency: skip if already liked RETURNING 1", @@ -255,38 +251,37 @@ WHERE $2 = ANY(p.like_actor_ids) LIMIT 1 ), + new_arrays AS ( + SELECT + lp.post_actor_id, + lp.post_rkey, + (SELECT array_agg(val) + FROM unnest(p.like_actor_ids) WITH ORDINALITY AS t(val, idx) + WHERE idx != lp.pos) as new_actor_ids, + (SELECT array_agg(val) + FROM unnest(p.like_rkeys) WITH ORDINALITY AS t(val, idx) + WHERE idx != lp.pos) as new_rkeys, + (SELECT val::text + FROM unnest(p.like_actor_ids) WITH ORDINALITY AS t(val, idx) + WHERE idx = lp.pos) as removed_actor_id + FROM like_position lp + INNER JOIN posts p ON p.actor_id = lp.post_actor_id AND p.rkey = lp.post_rkey + WHERE lp.pos IS NOT NULL + AND $1 = ANY(p.like_rkeys) + ), updated_post AS ( UPDATE posts p SET - like_count = GREATEST(COALESCE(like_count, 0) - 1, 0), - like_actor_ids = ( - SELECT array_agg(val) - FROM unnest(p.like_actor_ids) WITH ORDINALITY AS t(val, idx) - WHERE idx != (SELECT pos FROM like_position WHERE p.actor_id = post_actor_id AND p.rkey = post_rkey) - ), - like_rkeys = ( - SELECT array_agg(val) - FROM unnest(p.like_rkeys) WITH ORDINALITY AS t(val, idx) - WHERE idx != (SELECT pos FROM like_position WHERE p.actor_id = post_actor_id AND p.rkey = post_rkey) - ), - like_via_repost_actors = CASE - WHEN p.like_via_repost_actors IS NOT NULL THEN - (SELECT array_agg(val) - FROM unnest(p.like_via_repost_actors) WITH ORDINALITY AS t(val, idx) - WHERE idx != (SELECT pos FROM like_position WHERE p.actor_id = post_actor_id AND p.rkey = post_rkey)) + like_actor_ids = na.new_actor_ids, + like_rkeys = na.new_rkeys, + like_via_repost_data = CASE + WHEN p.like_via_repost_data IS NOT NULL THEN + p.like_via_repost_data - na.removed_actor_id ELSE NULL END, - like_via_repost_rkeys = CASE - WHEN p.like_via_repost_rkeys IS NOT NULL THEN - (SELECT array_agg(val) - FROM unnest(p.like_via_repost_rkeys) WITH ORDINALITY AS t(val, idx) - WHERE idx != (SELECT pos FROM like_position WHERE p.actor_id = post_actor_id AND p.rkey = post_rkey)) - ELSE NULL - END - FROM like_position lp - WHERE p.actor_id = lp.post_actor_id AND p.rkey = lp.post_rkey - AND lp.pos IS NOT NULL - AND $1 = ANY(p.like_rkeys) -- Verify the rkey matches (deduplication) + like_count = cardinality(na.new_actor_ids) + FROM new_arrays na + WHERE p.actor_id = na.post_actor_id AND p.rkey = na.post_rkey RETURNING p.actor_id, p.rkey ) SELECT 'at://' || a.did || '/app.bsky.feed.post/' || i64_to_tid(up.rkey) as subject_uri -- tangled.sh