diff --git a/consumer/src/db/bulk_copy/mod.rs b/consumer/src/db/bulk_copy/mod.rs index 13ce8ace..24bd03c3 100644 --- a/consumer/src/db/bulk_copy/mod.rs +++ b/consumer/src/db/bulk_copy/mod.rs @@ -51,38 +51,25 @@ use tokio_postgres::types::Type; /// Bulk COPY post likes into the database /// +/// Post likes are now embedded in the posts table as parallel arrays +/// (like_actor_ids, like_rkeys, like_via_repost_actors, like_via_repost_rkeys). +/// /// This function: /// 1. Creates a temporary staging table /// 2. Streams post likes into staging using PostgreSQL's Binary COPY protocol -/// 3. Moves staging data to main table with ON CONFLICT DO NOTHING -/// 4. Returns list of actually-inserted likes (for delta generation) -/// -/// # Schema -/// -/// ```sql -/// post_likes(actor_id, rkey, post_actor_id, post_rkey, via_repost_actor_id, via_repost_rkey) -/// ``` +/// 3. Groups likes by post and filters out duplicates (actors already in arrays) +/// 4. UPDATEs posts table to append new likes to arrays in bulk +/// 5. Returns list of actually-inserted likes (for delta generation) /// /// # Performance /// -/// - 1,000 likes: ~2-3ms (vs 6-7ms multi-row INSERT, vs 3-15s individual) -/// - 10,000 likes: ~15-30ms (unlimited - no batching needed) -/// -/// # Binary COPY -/// -/// Uses PostgreSQL's Binary COPY protocol which: -/// - Has no parameter limits (streaming) -/// - Is faster than multi-row INSERT (2-3x) -/// - Automatically serializes via ToSql trait +/// Binary COPY into staging: ~2-3ms for 1,000 likes, ~15-30ms for 10,000 likes +/// Bulk array UPDATE: Similar performance to old INSERT approach /// /// # Idempotency /// -/// Uses ON CONFLICT (actor_id, rkey) DO NOTHING to skip duplicates. -/// -/// # Note on Type Signature -/// -/// This function requires `Transaction` instead of `GenericClient` because `copy_in` -/// is not part of the GenericClient trait. Binary COPY requires the concrete transaction type. +/// Filters out likes where actor_id is already in the post's like_actor_ids array. +/// Only new likes are appended to arrays. pub async fn copy_post_likes( conn: &Transaction<'_>, likes: Vec, @@ -91,7 +78,7 @@ pub async fn copy_post_likes( return Ok(Vec::new()); } - tracing::debug!(count = likes.len(), "Starting bulk COPY for post_likes"); + tracing::debug!(count = likes.len(), "Starting bulk COPY for post_likes (embedded arrays)"); let start = std::time::Instant::now(); // Step 1: Create temporary staging table @@ -136,23 +123,56 @@ pub async fn copy_post_likes( writer.finish().await?; - // Step 3: Insert from staging to main table with RETURNING for deltas + // Step 3: Bulk UPDATE posts table to append likes to arrays + // IMPORTANT: Bulk COPY only happens during repo backfills where all likes + // have the same actor_id (the repo owner). This means each post gets at most + // ONE like from this batch, so no grouping is needed - it's 1:1 post to like. + // + // This query: + // 1. Filters out likes where actor_id is already in the post's like_actor_ids array + // 2. UPDATEs posts to append single like per post (no GROUP BY needed) + // 3. Returns inserted likes for delta generation let rows = conn .query( - "WITH inserted AS ( - INSERT INTO post_likes (actor_id, rkey, post_actor_id, post_rkey, via_repost_actor_id, via_repost_rkey) - SELECT actor_id, rkey, post_actor_id, post_rkey, via_repost_actor_id, via_repost_rkey - FROM post_likes_staging - ON CONFLICT (actor_id, rkey) DO NOTHING - RETURNING actor_id, post_actor_id, post_rkey + "WITH new_likes AS ( + -- Filter out likes where actor is already in the post's like_actor_ids array + -- During repo backfill, all likes have same actor_id, so this is simple deduplication + SELECT s.actor_id, s.rkey, s.post_actor_id, s.post_rkey, + s.via_repost_actor_id, s.via_repost_rkey + FROM post_likes_staging s + INNER JOIN posts p ON p.actor_id = s.post_actor_id AND p.rkey = s.post_rkey + WHERE NOT (s.actor_id = ANY(COALESCE(p.like_actor_ids, ARRAY[]::integer[]))) + ), + updated AS ( + -- Update posts to append single like per post + -- No GROUP BY needed: repo backfill = one actor, many posts, 1:1 relationship + UPDATE posts p + SET + like_count = COALESCE(like_count, 0) + 1, + like_actor_ids = COALESCE(like_actor_ids, ARRAY[]::integer[]) || nl.actor_id, + like_rkeys = COALESCE(like_rkeys, ARRAY[]::bigint[]) || nl.rkey, + like_via_repost_actors = CASE + WHEN nl.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(nl.via_repost_actor_id, 0) + ELSE NULL + END, + like_via_repost_rkeys = CASE + WHEN nl.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(nl.via_repost_rkey, 0) + ELSE NULL + END + FROM new_likes nl + WHERE p.actor_id = nl.post_actor_id AND p.rkey = nl.post_rkey + RETURNING nl.actor_id, nl.post_actor_id, nl.post_rkey ) - SELECT - i.actor_id, - i.post_actor_id, - i.post_rkey, - 'at://' || a.did || '/app.bsky.feed.post/' || i64_to_tid(i.post_rkey) as subject_uri - FROM inserted i - INNER JOIN actors a ON i.post_actor_id = a.id", + SELECT actor_id, post_actor_id, post_rkey + FROM updated", &[], ) .await?; @@ -176,7 +196,7 @@ pub async fn copy_post_likes( inserted = result.len(), duplicates = likes.len() - result.len(), duration_ms = elapsed.as_millis(), - "Bulk COPY post_likes completed" + "Bulk COPY post_likes (embedded arrays) completed" ); Ok(result) diff --git a/consumer/src/db/operations/feed/like.rs b/consumer/src/db/operations/feed/like.rs index 12ad0c1a..d95b457b 100644 --- a/consumer/src/db/operations/feed/like.rs +++ b/consumer/src/db/operations/feed/like.rs @@ -120,7 +120,8 @@ pub async fn like_insert( }; let subject_ms = subject_start.elapsed().as_millis(); - // Insert like with resolved IDs into the appropriate table + // Insert like with resolved IDs + // For posts: Embed in posts table arrays instead of separate post_likes table // Note: created_at is derived from TID rkey // Note: cid column has been dropped - we generate synthetic CIDs for notifications let insert_start = std::time::Instant::now(); @@ -128,26 +129,46 @@ pub async fn like_insert( "post" => { // Extract post natural key components let (post_actor_id, post_rkey) = subject_post_key.expect("post likes must have post_key"); + + // Extract via_repost natural key components (or 0 for NULL) + let (via_repost_actor_id, via_repost_rkey) = via_repost_key + .map(|(a, r)| (a, r)) + .unwrap_or((0, 0)); + + // Update posts table to append to like arrays + // Uses conditional array concatenation to handle NULL arrays and via_repost let stmt = conn .prepare_typed( - "INSERT INTO post_likes (actor_id, rkey, post_actor_id, post_rkey, via_repost_actor_id, via_repost_rkey) - VALUES ($1, $2, $3, $4, $5, $6) - ON CONFLICT (actor_id, rkey) DO NOTHING", + "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 + 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 + WHERE actor_id = $1 AND rkey = $2 + AND NOT ($3 = ANY(COALESCE(like_actor_ids, ARRAY[]::integer[]))) -- Idempotency: skip if already liked + RETURNING 1", &[ - tokio_postgres::types::Type::INT4, - tokio_postgres::types::Type::INT8, - tokio_postgres::types::Type::INT4, - tokio_postgres::types::Type::INT8, - tokio_postgres::types::Type::INT4, - tokio_postgres::types::Type::INT8, + tokio_postgres::types::Type::INT4, // $1 post_actor_id + tokio_postgres::types::Type::INT8, // $2 post_rkey + tokio_postgres::types::Type::INT4, // $3 liker actor_id + tokio_postgres::types::Type::INT8, // $4 liker rkey + tokio_postgres::types::Type::INT4, // $5 via_repost_actor_id + tokio_postgres::types::Type::INT8, // $6 via_repost_rkey ], ) .await?; - // Extract via_repost natural key components - let (via_repost_actor_id, via_repost_rkey) = via_repost_key - .map(|(a, r)| (Some(a), Some(r))) - .unwrap_or((None, None)); - conn.execute(&stmt, &[&actor_id, &rkey, &post_actor_id, &post_rkey, &via_repost_actor_id, &via_repost_rkey]) + + conn.execute(&stmt, &[&post_actor_id, &post_rkey, &actor_id, &rkey, &via_repost_actor_id, &via_repost_rkey]) .await? } "feedgen" => { @@ -214,22 +235,63 @@ pub async fn like_insert( /// /// Returns the AT URI of the liked subject (post, feedgen, or labeler) if found. /// Returns None if the like didn't exist. +/// +/// For posts: Removes like from embedded arrays in posts table pub async fn like_delete( conn: &C, rkey: i64, actor_id: i32, ) -> Result> { - // Try deleting from post_likes first (most common case) + // Try removing from posts arrays first (most common case) + // Find the position of the like in the arrays and remove it let res = conn .query_opt( - "WITH deleted_like AS ( - DELETE FROM post_likes - WHERE rkey=$1 AND actor_id=$2 - RETURNING post_actor_id, post_rkey + "WITH like_position AS ( + SELECT + p.actor_id as post_actor_id, + p.rkey as post_rkey, + array_position(p.like_actor_ids, $2) as pos + FROM posts p + WHERE $2 = ANY(p.like_actor_ids) + LIMIT 1 + ), + 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)) + 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) + RETURNING p.actor_id, p.rkey ) - SELECT 'at://' || a.did || '/app.bsky.feed.post/' || i64_to_tid(dl.post_rkey) as subject_uri - FROM deleted_like dl - INNER JOIN actors a ON a.id = dl.post_actor_id", + SELECT 'at://' || a.did || '/app.bsky.feed.post/' || i64_to_tid(up.rkey) as subject_uri + FROM updated_post up + INNER JOIN actors a ON a.id = up.actor_id", &[&rkey, &actor_id], ) .await?; diff --git a/consumer/src/db/record_exists/queries.rs b/consumer/src/db/record_exists/queries.rs index df7c7f1c..652583ba 100644 --- a/consumer/src/db/record_exists/queries.rs +++ b/consumer/src/db/record_exists/queries.rs @@ -30,9 +30,10 @@ pub async fn post_exists(conn: &C, did: &str, rkey: i64) -> Qu pub async fn like_exists(conn: &C, did: &str, rkey: i64) -> QueryResult { let row = conn .query_opt( - "SELECT 1 FROM post_likes l - INNER JOIN actors a ON l.actor_id = a.id - WHERE a.did = $1 AND l.rkey = $2", + "SELECT 1 FROM posts p + INNER JOIN actors a ON a.id = ANY(p.like_actor_ids) + WHERE a.did = $1 AND $2 = ANY(p.like_rkeys) + AND p.like_rkeys[array_position(p.like_actor_ids, a.id)] = $2", &[&did, &rkey], ) .await?; diff --git a/consumer/tests/feed_operations_test.rs b/consumer/tests/feed_operations_test.rs index 1e40290f..cc6405c9 100644 --- a/consumer/tests/feed_operations_test.rs +++ b/consumer/tests/feed_operations_test.rs @@ -513,21 +513,28 @@ async fn test_like_insert_post_subject() -> eyre::Result<()> { ); assert_eq!(result.wrap_err("Operation failed")?, 1, "Should insert 1 like"); - // Verify like was created + // Verify like was created in embedded arrays let like_row = tx .query_one( - "SELECT COUNT(*) - FROM post_likes l - INNER JOIN actors a ON l.actor_id = a.id - WHERE a.did = $1 AND l.rkey = tid_to_i64($2)", - &[&"did:plc:liker", &"3l7mkz4lmk23y"], + "SELECT + CASE WHEN $2 = ANY(p.like_actor_ids) THEN 1 ELSE 0 END as like_exists, + p.like_count, + array_length(p.like_actor_ids, 1) as array_len + FROM posts p + INNER JOIN actors pa ON p.actor_id = pa.id + WHERE pa.did = $1 AND p.rkey = tid_to_i64($3)", + &[&"did:plc:postauthor", &liker_actor_id, &"3l7mkz4lmk23a"], ) .await .wrap_err("Failed to query like")?; - let count: i64 = like_row.get(0); + let like_exists: i32 = like_row.get(0); + let like_count: Option = like_row.get(1); + let array_len: Option = like_row.get(2); - assert_eq!(count, 1, "Should have 1 like"); + assert_eq!(like_exists, 1, "Liker should be in like_actor_ids array"); + assert_eq!(like_count, Some(1), "like_count should be 1"); + assert_eq!(array_len, Some(1), "like_actor_ids array should have 1 element"); Ok(()) } @@ -609,20 +616,28 @@ async fn test_like_delete() -> eyre::Result<()> { "Should return subject URI" ); - // Verify like was deleted - let like_exists: bool = tx + // Verify like was deleted from embedded arrays + let like_row = tx .query_one( - "SELECT EXISTS( - SELECT 1 FROM post_likes l - INNER JOIN actors a ON l.actor_id = a.id - WHERE a.did = $1 AND l.rkey = tid_to_i64($2) - )", - &[&"did:plc:liker3", &"3l7mkz4lmk242"], + "SELECT + CASE WHEN $2 = ANY(p.like_actor_ids) THEN 1 ELSE 0 END as like_exists, + p.like_count, + array_length(p.like_actor_ids, 1) as array_len + FROM posts p + INNER JOIN actors pa ON p.actor_id = pa.id + WHERE pa.did = $1 AND p.rkey = tid_to_i64($3)", + &[&"did:plc:postauthor2", &liker3_actor_id, &"3l7mkz4lmk23c"], ) .await - .wrap_err("Failed to query likes")? - .get(0); - assert!(!like_exists, "Like should be deleted"); + .wrap_err("Failed to query post")?; + + let like_exists: i32 = like_row.get(0); + let like_count: Option = like_row.get(1); + let array_len: Option = like_row.get(2); + + assert_eq!(like_exists, 0, "Liker should NOT be in like_actor_ids array"); + assert_eq!(like_count, Some(0), "like_count should be 0 after deletion"); + assert!(array_len.is_none() || array_len == Some(0), "like_actor_ids array should be empty or NULL"); Ok(()) } diff --git a/migrations/2025-12-04-184457_add_likes_arrays_to_posts/down.sql b/migrations/2025-12-04-184457_add_likes_arrays_to_posts/down.sql new file mode 100644 index 00000000..9cf29ebb --- /dev/null +++ b/migrations/2025-12-04-184457_add_likes_arrays_to_posts/down.sql @@ -0,0 +1,12 @@ +-- Rollback: Remove array columns for embedded likes from posts table +-- WARNING: This will delete all embedded likes data! +-- Ensure post_likes table still exists if you need to restore data + +ALTER TABLE posts +DROP COLUMN IF EXISTS like_actor_ids, +DROP COLUMN IF EXISTS like_rkeys, +DROP COLUMN IF EXISTS like_via_repost_actors, +DROP COLUMN IF EXISTS like_via_repost_rkeys; + +-- Note: We don't drop intarray extension as other tables might use it +-- DROP EXTENSION IF EXISTS intarray; diff --git a/migrations/2025-12-04-184457_add_likes_arrays_to_posts/up.sql b/migrations/2025-12-04-184457_add_likes_arrays_to_posts/up.sql new file mode 100644 index 00000000..17028d48 --- /dev/null +++ b/migrations/2025-12-04-184457_add_likes_arrays_to_posts/up.sql @@ -0,0 +1,18 @@ +-- Add array columns for embedded likes to posts table +-- This migration only adds the schema changes +-- Data backfill from post_likes table will be done separately via one-shot migration + +ALTER TABLE posts +ADD COLUMN like_actor_ids integer[], +ADD COLUMN like_rkeys bigint[], +ADD COLUMN like_via_repost_actors integer[], +ADD COLUMN like_via_repost_rkeys bigint[]; + +-- Enable intarray extension for optimized array operations +CREATE EXTENSION IF NOT EXISTS intarray; + +-- Note: The columns are nullable by default +-- NULL means the post has no likes +-- Arrays are populated with likes data in index position alignment: +-- like_actor_ids[i] pairs with like_rkeys[i] +-- like_via_repost_actors[i] == 0 means no via_repost for that like diff --git a/parakeet-db/src/models.rs b/parakeet-db/src/models.rs index cde6f966..482c34c5 100644 --- a/parakeet-db/src/models.rs +++ b/parakeet-db/src/models.rs @@ -211,9 +211,99 @@ pub struct Post { 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: + // 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) // Note: created_at derived from TID rkey via created_at() method } +impl Post { + /// Get a specific like by index position + /// + /// Returns None if index is out of bounds or arrays are NULL + pub fn get_like(&self, idx: usize) -> Option { + let actor_ids = self.like_actor_ids.as_ref()?; + let rkeys = self.like_rkeys.as_ref()?; + + let actor_id = *actor_ids.get(idx)?; + let rkey = *rkeys.get(idx)?; + + let via_repost_actor_id = self + .like_via_repost_actors + .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); + + Some(PostLikeInfo { + actor_id, + rkey, + via_repost_actor_id, + via_repost_rkey, + }) + } + + /// Check if an actor has liked this post + /// + /// Returns Some(like_rkey) if the actor has liked, None otherwise + pub fn actor_liked(&self, actor_id: i32) -> Option { + let actor_ids = self.like_actor_ids.as_ref()?; + let rkeys = self.like_rkeys.as_ref()?; + + actor_ids + .iter() + .position(|&id| id == actor_id) + .and_then(|idx| rkeys.get(idx).copied()) + } + + /// Get iterator over all likes + /// + /// Returns empty iterator if post has no likes + pub fn likes(&self) -> impl Iterator + '_ { + let count = self + .like_actor_ids + .as_ref() + .map(|v| v.len()) + .unwrap_or(0); + + (0..count).filter_map(move |i| self.get_like(i)) + } + + /// Get number of likes from array length (should match like_count) + pub fn likes_len(&self) -> usize { + self.like_actor_ids + .as_ref() + .map(|v| v.len()) + .unwrap_or(0) + } +} + +/// Represents a single like extracted from post arrays +#[derive(Debug, Clone)] +pub struct PostLikeInfo { + pub actor_id: i32, + pub rkey: i64, + pub via_repost_actor_id: Option, + pub via_repost_rkey: Option, +} + +impl PostLikeInfo { + /// Get created_at timestamp from the like's TID rkey + pub fn created_at(&self) -> DateTime { + crate::tid_util::tid_to_datetime(self.rkey) + } +} + // Stats struct for API responses (replaces parakeet_index::PostStats) #[derive(Debug, Clone, Copy, Serialize, Deserialize)] pub struct PostStats { diff --git a/parakeet-db/src/schema.rs b/parakeet-db/src/schema.rs index 43ab0e8f..a8f5d5b1 100644 --- a/parakeet-db/src/schema.rs +++ b/parakeet-db/src/schema.rs @@ -85,6 +85,10 @@ pub mod sql_types { #[diesel(postgres_type(name = "post_image_embed"))] 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; @@ -555,6 +559,10 @@ diesel::table! { reply_count -> Nullable, repost_count -> Nullable, quote_count -> Nullable, + like_actor_ids -> Nullable>, + like_rkeys -> Nullable>, + like_via_repost_actors -> Nullable>, + like_via_repost_rkeys -> Nullable>, } } @@ -615,6 +623,113 @@ diesel::table! { } } +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>, + } +} + diesel::table! { thread_mutes (actor_id, root_post_actor_id, root_post_rkey) { actor_id -> Int4, @@ -717,6 +832,16 @@ diesel::allow_tables_to_appear_in_same_query!( 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/parakeet/src/db/likes.rs b/parakeet/src/db/likes.rs index 584a7fe4..c58855ac 100644 --- a/parakeet/src/db/likes.rs +++ b/parakeet/src/db/likes.rs @@ -1,7 +1,11 @@ //! Like state queries using the self-contained schema //! -//! Likes split into three tables for proper FK constraints: -//! - post_likes: likes on posts (99.98% of likes) +//! Post likes are now embedded in posts table as parallel arrays: +//! - 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 +//! +//! Other likes still in separate tables: //! - feedgen_likes: likes on feed generators //! - labeler_likes: likes on labeler services @@ -32,18 +36,18 @@ pub async fn get_like_state( let subject_rkey = parakeet_db::tid_util::decode_tid(subject_rkey_base32) .map_err(|_| diesel::result::Error::NotFound)?; - // Query post_likes table with proper FK joins + // Query posts table and check if actor_id is in like_actor_ids array + // Find the array position to get the matching rkey from like_rkeys array diesel_async::RunQueryDsl::get_result( diesel::sql_query( - "SELECT a.did, l.rkey - FROM post_likes l - INNER JOIN actors a ON l.actor_id = a.id - INNER JOIN posts p ON l.post_actor_id = p.actor_id AND l.post_rkey = p.rkey + "SELECT a.did, p.like_rkeys[array_position(p.like_actor_ids, a.id)] as rkey + FROM posts p INNER JOIN actors pa ON p.actor_id = pa.id - WHERE a.did = $1 + INNER JOIN actors a ON a.did = $1 + WHERE pa.did = $2 + AND p.rkey = $3 AND p.status = 'complete' - AND pa.did = $2 - AND p.rkey = $3" + AND a.id = ANY(p.like_actor_ids)" ) .bind::(did) .bind::(subject_did) @@ -98,22 +102,22 @@ pub async fn get_like_states( return Ok(Vec::new()); } - // Query post_likes with URI reconstruction + // Query posts table and check if actor_id is in like_actor_ids arrays + // Use unnest to match lookup DIDs/rkeys with posts let results: Vec = diesel_async::RunQueryDsl::load( diesel::sql_query( "SELECT pa.did as subject_did, p.rkey as subject_rkey, a.did, - l.rkey - FROM post_likes l - INNER JOIN actors a ON l.actor_id = a.id - INNER JOIN posts p ON l.post_actor_id = p.actor_id AND l.post_rkey = p.rkey + p.like_rkeys[array_position(p.like_actor_ids, a.id)] as rkey + FROM posts p INNER JOIN actors pa ON p.actor_id = pa.id + INNER JOIN actors a ON a.did = $1 INNER JOIN unnest($2::text[], $3::bigint[]) AS lookup(lookup_did, lookup_rkey) ON pa.did = lookup.lookup_did AND p.rkey = lookup.lookup_rkey - WHERE a.did = $1 - AND p.status = 'complete'" + WHERE p.status = 'complete' + AND a.id = ANY(p.like_actor_ids)" ) .bind::(did) .bind::, _>(&subject_dids) @@ -149,19 +153,25 @@ pub async fn get_actor_likes( subject_did: String, #[diesel(sql_type = BigInt)] subject_rkey: i64, + #[diesel(sql_type = BigInt)] + like_rkey: i64, } - // Use .bind() for cursor parameter to prevent SQL injection + // Query posts where actor is in like_actor_ids array + // Extract like rkey using array_position, use tid_timestamp for ordering/cursor diesel::sql_query( - "SELECT tid_timestamp(l.rkey) as created_at, pa.did as subject_did, p.rkey as subject_rkey - FROM post_likes l - INNER JOIN actors a ON l.actor_id = a.id - INNER JOIN posts p ON l.post_actor_id = p.actor_id AND l.post_rkey = p.rkey + "SELECT + tid_timestamp(p.like_rkeys[array_position(p.like_actor_ids, a.id)]) as created_at, + pa.did as subject_did, + p.rkey as subject_rkey, + p.like_rkeys[array_position(p.like_actor_ids, a.id)] as like_rkey + FROM posts p INNER JOIN actors pa ON p.actor_id = pa.id - WHERE a.did = $1 - AND p.status = 'complete' - AND ($2::timestamptz IS NULL OR tid_timestamp(l.rkey) < $2) - ORDER BY l.rkey DESC + INNER JOIN actors a ON a.did = $1 + WHERE p.status = 'complete' + AND p.like_actor_ids @> ARRAY[a.id] + AND ($2::timestamptz IS NULL OR tid_timestamp(p.like_rkeys[array_position(p.like_actor_ids, a.id)]) < $2) + ORDER BY like_rkey DESC LIMIT $3" ) .bind::(actor_did) @@ -209,18 +219,24 @@ pub async fn get_post_likes( let post_rkey = parakeet_db::tid_util::decode_tid(post_rkey_base32) .map_err(|_| diesel::result::Error::NotFound)?; - // Use .bind() for cursor parameter to prevent SQL injection + // Unnest like arrays to get individual likes + // Use tid_timestamp on like_rkey for ordering and cursor diesel::sql_query( - "SELECT tid_timestamp(l.rkey) as created_at, a.did as liker_did - FROM post_likes l - INNER JOIN actors a ON l.actor_id = a.id - INNER JOIN posts p ON l.post_actor_id = p.actor_id AND l.post_rkey = p.rkey + "SELECT + tid_timestamp(like_rkey) as created_at, + a.did as liker_did + FROM posts p INNER JOIN actors pa ON p.actor_id = pa.id + CROSS JOIN LATERAL unnest( + COALESCE(p.like_actor_ids, ARRAY[]::integer[]), + COALESCE(p.like_rkeys, ARRAY[]::bigint[]) + ) AS like_data(actor_id, like_rkey) + INNER JOIN actors a ON a.id = like_data.actor_id WHERE pa.did = $1 AND p.rkey = $2 AND p.status = 'complete' - AND ($3::timestamptz IS NULL OR tid_timestamp(l.rkey) < $3) - ORDER BY l.rkey DESC + AND ($3::timestamptz IS NULL OR tid_timestamp(like_data.like_rkey) < $3) + ORDER BY like_data.like_rkey DESC LIMIT $4" ) .bind::(post_did) diff --git a/parakeet/src/db/notification_records.rs b/parakeet/src/db/notification_records.rs index 1cd3c76f..9a7ed7db 100644 --- a/parakeet/src/db/notification_records.rs +++ b/parakeet/src/db/notification_records.rs @@ -25,14 +25,16 @@ pub async fn get_like_record( let rkey_bigint = parakeet_db::tid_util::decode_tid(rkey).map_err(|_| diesel::result::Error::NotFound)?; + // Query posts where the actor's like rkey exists in like_rkeys array diesel::sql_query( "SELECT pa.did as subject_did, p.rkey as subject_rkey, - tid_timestamp(l.rkey) as created_at - FROM post_likes l - INNER JOIN actors a ON l.actor_id = a.id - INNER JOIN posts p ON l.post_actor_id = p.actor_id AND l.post_rkey = p.rkey + tid_timestamp($2::bigint) as created_at + FROM posts p INNER JOIN actors pa ON p.actor_id = pa.id - WHERE a.did = $1 AND l.rkey = $2", + INNER JOIN actors a ON a.did = $1 + WHERE a.id = ANY(p.like_actor_ids) + AND $2 = ANY(p.like_rkeys) + AND p.like_rkeys[array_position(p.like_actor_ids, a.id)] = $2", ) .bind::(did) .bind::(rkey_bigint) diff --git a/parakeet/src/loaders/post.rs b/parakeet/src/loaders/post.rs index 0fe493bf..eafdcb93 100644 --- a/parakeet/src/loaders/post.rs +++ b/parakeet/src/loaders/post.rs @@ -901,6 +901,11 @@ impl BatchFn for PostLoader { reply_count: p.reply_count, repost_count: p.repost_count, quote_count: p.quote_count, + // 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, }; // Encode TIDs using Rust utility functions diff --git a/parakeet/src/sql/post_state.sql b/parakeet/src/sql/post_state.sql index c3cd848c..3ad5431e 100644 --- a/parakeet/src/sql/post_state.sql +++ b/parakeet/src/sql/post_state.sql @@ -42,14 +42,15 @@ viewer_pinned AS ( LIMIT 1 ), -- Query viewer's likes for these specific posts only --- Uses rkey range to enable chunk exclusion on post_likes +-- Check if viewer actor_id is in like_actor_ids array, extract matching like rkey viewer_likes AS ( - SELECT pl.post_actor_id, pl.post_rkey, pl.rkey as like_rkey - FROM post_likes pl - INNER JOIN target_posts tp ON pl.post_actor_id = tp.actor_id AND pl.post_rkey = tp.rkey - WHERE pl.actor_id = $1 - AND pl.post_rkey >= $4 - AND pl.post_rkey <= $5 + SELECT tp.actor_id as post_actor_id, tp.rkey as post_rkey, + p.like_rkeys[array_position(p.like_actor_ids, $1)] as like_rkey + FROM target_posts tp + INNER JOIN posts p ON p.actor_id = tp.actor_id AND p.rkey = tp.rkey + WHERE p.like_actor_ids @> ARRAY[$1] + AND tp.rkey >= $4 + AND tp.rkey <= $5 ), -- Query viewer's reposts for these specific posts only -- Uses rkey range to enable chunk exclusion on reposts