feat: denormalize engagements

consumer/src/database_writer/operations/executor.rs

···
 //! Operations return aggregate deltas ONLY when they actually affect rows in the database.
 //! This makes stats idempotent - duplicate events won't double-count.
 
-use super::{cache, ActorStatsDeltas, DatabaseOperation, PostStatsDeltas};
+use super::{cache, ActorStatsDeltas, DatabaseOperation};
 use eyre::WrapErr as _;
 use ipld_core::cid::Cid;
-
-/// Helper to create a single post stats delta
-fn like_delta(actor_id: i32, rkey: i64, delta: i32) -> Vec<((i32, i64), PostStatsDeltas)> {
-    vec![((actor_id, rkey), PostStatsDeltas { like_delta: delta, ..Default::default() })]
-}
-
-/// Helper to create repost delta
-fn repost_delta(actor_id: i32, rkey: i64, delta: i32) -> Vec<((i32, i64), PostStatsDeltas)> {
-    vec![((actor_id, rkey), PostStatsDeltas { repost_delta: delta, ..Default::default() })]
-}
-
-/// Helper to create reply delta
-fn reply_delta(actor_id: i32, rkey: i64, delta: i32) -> Vec<((i32, i64), PostStatsDeltas)> {
-    vec![((actor_id, rkey), PostStatsDeltas { reply_delta: delta, ..Default::default() })]
-}
-
-/// Helper to create quote delta
-fn quote_delta(actor_id: i32, rkey: i64, delta: i32) -> Vec<((i32, i64), PostStatsDeltas)> {
-    vec![((actor_id, rkey), PostStatsDeltas { quote_delta: delta, ..Default::default() })]
-}
 
 // Actor stats delta helpers for denormalized stats in actors table
 
···
                 if let Some(parent_uri) = parent_uri.as_ref() {
                     // Resolve parent URI to natural key
                     if let Ok(Some((parent_actor_id, parent_rkey))) = crate::db::id_resolution::resolve_post_uri(conn, parent_uri).await {
-                        deltas.extend(reply_delta(parent_actor_id, parent_rkey, 1));
-                        // Invalidate parent post cache since reply_count changed
+                        // Append to parent's reply arrays (array-only tracking)
+                        let _ = conn.execute(
+                            "UPDATE posts
+                             SET reply_actor_ids = COALESCE(reply_actor_ids, ARRAY[]::integer[]) || $3,
+                                 reply_rkeys = COALESCE(reply_rkeys, ARRAY[]::bigint[]) || $4
+                             WHERE (actor_id, rkey) = ($1, $2)
+                               AND NOT ($3 = ANY(COALESCE(reply_actor_ids, ARRAY[])))",
+                            &[&parent_actor_id, &parent_rkey, &actor_id, &rkey],
+                        ).await;
+                        // Invalidate parent post cache since reply array changed
                         cache_invalidations.push(cache::invalidate_post(parent_uri));
                     }
                 }
 
-                // If this quotes/embeds a post, increment that post's quote count
+                // If this quotes/embeds a post, append to that post's quote arrays
                 if let Some(quoted_uri) = embed_uri.as_ref() {
                     // Resolve quoted URI to natural key
                     if let Ok(Some((quoted_actor_id, quoted_rkey))) = crate::db::id_resolution::resolve_post_uri(conn, quoted_uri).await {
-                        deltas.extend(quote_delta(quoted_actor_id, quoted_rkey, 1));
-                        // Invalidate quoted post cache since quote_count changed
+                        // Append to quoted post's quote arrays (array-only tracking)
+                        let _ = conn.execute(
+                            "UPDATE posts
+                             SET quote_actor_ids = COALESCE(quote_actor_ids, ARRAY[]::integer[]) || $3,
+                                 quote_rkeys = COALESCE(quote_rkeys, ARRAY[]::bigint[]) || $4
+                             WHERE (actor_id, rkey) = ($1, $2)
+                               AND NOT ($3 = ANY(COALESCE(quote_actor_ids, ARRAY[])))",
+                            &[&quoted_actor_id, &quoted_rkey, &actor_id, &rkey],
+                        ).await;
+                        // Invalidate quoted post cache since quote array changed
                         cache_invalidations.push(cache::invalidate_post(quoted_uri));
                     }
                 }
···
             }
 
             if rows > 0 {
-                // Only send delta if we actually inserted a like (not a duplicate)
-                // Resolve subject URI to natural key
-                let deltas = if let Ok(Some((subject_actor_id, subject_rkey))) = crate::db::id_resolution::resolve_post_uri(conn, &subject_uri).await {
-                    like_delta(subject_actor_id, subject_rkey, 1)
-                } else {
-                    vec![]
-                };
+                // Note: like arrays are updated directly in db::like_insert via LikeArrayOp
+                // No deltas needed with array-only tracking
 
-                // Invalidate post cache since like_count changed
+                // Invalidate post cache since like array changed
                 let cache_invalidations = vec![cache::invalidate_post(&subject_uri)];
 
-                Ok((deltas, vec![], cache_invalidations))
+                Ok((vec![], vec![], cache_invalidations))
             } else {
                 Ok((vec![], vec![], vec![]))
             }
···
             }
 
             if rows > 0 {
-                // Only send delta if we actually inserted a repost
-                // Resolve subject URI to natural key
-                let deltas = if let Ok(Some((subject_actor_id, subject_rkey))) = crate::db::id_resolution::resolve_post_uri(conn, &subject_uri).await {
-                    repost_delta(subject_actor_id, subject_rkey, 1)
-                } else {
-                    vec![]
-                };
+                // Note: repost arrays will be updated in db::repost_insert (dual-write to reposts table + post arrays)
+                // No deltas needed with array-only tracking
 
-                // Invalidate post cache since repost_count changed
+                // Invalidate post cache since repost array changed
                 let cache_invalidations = vec![cache::invalidate_post(&subject_uri)];
 
-                Ok((deltas, vec![], cache_invalidations))
+                Ok((vec![], vec![], cache_invalidations))
             } else {
                 Ok((vec![], vec![], vec![]))
             }
···
                     if let Some(subject) = db::like_delete(conn, rkey, actor_id).await? {
                         match subject {
                             db::LikeSubject::Post { actor_id: post_actor_id, rkey: post_rkey } => {
-                                // Decrement post like_count
-                                deltas.extend(like_delta(post_actor_id, post_rkey, -1));
+                                // Note: like arrays are updated directly in db::like_delete via array_remove()
+                                // No deltas needed with array-only tracking
 
                                 // Invalidate post cache
                                 if let Ok(post_did) = db::actor_did_from_id(conn, post_actor_id).await {
···
                 CollectionType::BskyFeedRepost => {
                     // Returns post URI if deleted
                     if let Some(post_uri) = db::repost_delete(conn, rkey, actor_id).await? {
-                        // Resolve post URI to natural key for delta
-                        if let Ok(Some((post_actor_id, post_rkey))) = crate::db::id_resolution::resolve_post_uri(conn, &post_uri).await {
-                            deltas.extend(repost_delta(post_actor_id, post_rkey, -1));
-                        }
+                        // Note: repost arrays will be updated in db::repost_delete (remove from post arrays)
+                        // No deltas needed with array-only tracking
 
-                        // Invalidate post cache since repost_count changed
+                        // Invalidate post cache since repost array changed
                         cache_invalidations.push(cache::invalidate_post(&post_uri));
 
                         // Remove PostgreSQL notification (recipient = author of reposted post)
···
                         // Decrement actor's post count
                         actor_deltas.extend(actor_post_delta(actor_id, -1));
 
-                        // Decrement reply count for parent if this was a reply
+                        // Remove from parent's reply arrays if this was a reply
                         if let Some((parent_actor_id, parent_rkey)) = parent_key {
-                            deltas.extend(reply_delta(parent_actor_id, parent_rkey, -1));
+                            // Remove from reply arrays using array_remove()
+                            let _ = conn.execute(
+                                "UPDATE posts
+                                 SET reply_actor_ids = array_remove(reply_actor_ids, $3),
+                                     reply_rkeys = array_remove(reply_rkeys, $4)
+                                 WHERE (actor_id, rkey) = ($1, $2)",
+                                &[&parent_actor_id, &parent_rkey, &actor_id, &rkey],
+                            ).await;
 
-                            // Invalidate parent post cache since reply_count changed
+                            // Invalidate parent post cache since reply array changed
                             // Construct URI using actor cache
                             if let Ok(parent_did) = db::actor_did_from_id(conn, parent_actor_id).await {
                                 let parent_uri = format!("at://{}/app.bsky.feed.post/{}",
···
                             }
                         }
 
-                        // Decrement quote count if this quoted a record
+                        // Remove from embedded post's quote arrays if this quoted a record
                         if let Some((embed_actor_id, embed_rkey)) = embedded_key {
-                            deltas.extend(quote_delta(embed_actor_id, embed_rkey, -1));
+                            // Remove from quote arrays using array_remove()
+                            let _ = conn.execute(
+                                "UPDATE posts
+                                 SET quote_actor_ids = array_remove(quote_actor_ids, $3),
+                                     quote_rkeys = array_remove(quote_rkeys, $4)
+                                 WHERE (actor_id, rkey) = ($1, $2)",
+                                &[&embed_actor_id, &embed_rkey, &actor_id, &rkey],
+                            ).await;
 
-                            // Invalidate quoted post cache since quote_count changed
+                            // Invalidate quoted post cache since quote array changed
                             // Construct URI using actor cache
                             if let Ok(embed_did) = db::actor_did_from_id(conn, embed_actor_id).await {
                                 let embed_uri = format!("at://{}/app.bsky.feed.post/{}",
···
             Ok((vec![], vec![], vec![]))
         }
         DatabaseOperation::EnsureMinimumPostStats {
-            post_uri,
-            min_likes,
-            min_reposts,
-            min_quotes,
-            min_replies,
+            post_uri: _,
+            min_likes: _,
+            min_reposts: _,
+            min_quotes: _,
+            min_replies: _,
         } => {
-            // Resolve post_uri to post natural key
-            let (post_actor_id, post_rkey) = match crate::db::id_resolution::resolve_uri_to_id(conn, &post_uri).await {
-                Ok(crate::db::id_resolution::ResolvedId::Post(actor_id, rkey)) => (actor_id, rkey),
-                Ok(crate::db::id_resolution::ResolvedId::Actor(_)) => {
-                    tracing::error!("EnsureMinimumPostStats received actor DID instead of post URI: {}", post_uri);
-                    return Ok((vec![], vec![], vec![]));
-                }
-                Err(e) => {
-                    tracing::debug!("Post not found for EnsureMinimumPostStats (may not exist yet): {}: {}", post_uri, e);
-                    return Ok((vec![], vec![], vec![]));
-                }
-            };
-
-            // Get current stats from denormalized columns
-            let current_stats = conn.query_one(
-                "SELECT like_count, reply_count, repost_count, quote_count FROM posts WHERE actor_id = $1 AND rkey = $2",
-                &[&post_actor_id, &post_rkey],
-            ).await;
-
-            let (current_likes, current_replies, current_reposts, current_quotes) = match current_stats {
-                Ok(row) => {
-                    let likes: Option<i16> = row.get(0);
-                    let replies: Option<i16> = row.get(1);
-                    let reposts: Option<i16> = row.get(2);
-                    let quotes: Option<i16> = row.get(3);
-                    (
-                        likes.unwrap_or(0) as u64,
-                        replies.unwrap_or(0) as u64,
-                        reposts.unwrap_or(0) as u64,
-                        quotes.unwrap_or(0) as u64,
-                    )
-                }
-                Err(e) => {
-                    tracing::warn!("Failed to get post stats from database for {}: {}", post_uri, e);
-                    return Ok((vec![], vec![], vec![]));
-                }
-            };
-
-            // Calculate deltas needed to reach minimums and build PostStatsDeltas
-            let mut delta_likes = 0;
-            let mut delta_replies = 0;
-            let mut delta_reposts = 0;
-            let mut delta_quotes = 0;
-
-            if min_likes > current_likes {
-                delta_likes = (min_likes - current_likes) as i32;
-            }
-            if min_replies > current_replies {
-                delta_replies = (min_replies - current_replies) as i32;
-            }
-            if min_reposts > current_reposts {
-                delta_reposts = (min_reposts - current_reposts) as i32;
-            }
-            if min_quotes > current_quotes {
-                delta_quotes = (min_quotes - current_quotes) as i32;
-            }
-
-            // Return deltas if any are non-zero
-            if delta_likes != 0 || delta_replies != 0 || delta_reposts != 0 || delta_quotes != 0 {
-                let deltas = vec![((post_actor_id, post_rkey), PostStatsDeltas {
-                    like_delta: delta_likes,
-                    reply_delta: delta_replies,
-                    repost_delta: delta_reposts,
-                    quote_delta: delta_quotes,
-                })];
-                Ok((deltas, vec![], vec![]))
-            } else {
-                Ok((vec![], vec![], vec![]))
-            }
+            // DEPRECATED: EnsureMinimumPostStats is incompatible with array-only tracking.
+            // With arrays, we can't fabricate engagement entries to reach minimum counts
+            // (we'd need actual actor IDs and rkeys, not just synthetic counts).
+            // The Constellation enrichment feature that used this may need to be redesigned
+            // or removed entirely.
+            //
+            // For now, this operation is a no-op to avoid breaking callers.
+            // TODO: Remove this operation entirely once callers are updated.
+            Ok((vec![], vec![], vec![]))
         }
     }
 }

consumer/src/database_writer/operations/types.rs

···
     },
 }
 
-/// Post stats deltas for batch updating denormalized stats columns
-/// Accumulates deltas per post (actor_id, rkey) during operation processing
+/// Post stats deltas (DEPRECATED - now using array-only tracking)
+/// This struct is kept for backwards compatibility during migration but is no longer used.
+/// All functions should return empty HashMaps/Vecs for this type.
 #[derive(Default, Debug, Clone)]
 pub struct PostStatsDeltas {
-    pub like_delta: i32,
-    pub reply_delta: i32,
-    pub repost_delta: i32,
-    pub quote_delta: i32,
+    // All fields removed - engagement is now tracked via arrays only
 }
 
 /// Actor stats deltas for batch updating denormalized stats columns in actors table

consumer/src/database_writer/workers.rs

···
                         &pds_cache,
                     ).await {
                         Ok((stats_deltas, actor_stats_deltas)) => {
-                            // Batch update post stats
-                            if !stats_deltas.is_empty() {
-                                if let Err(e) = batch_update_post_stats(&mut conn_mut, stats_deltas).await {
-                                    tracing::error!(worker = name, error = ?e, repo = %repo, "Failed to batch update post stats after bulk COPY");
-                                    counter!("bulk_writer_stats_update_error").increment(1);
-                                }
-                            }
+                            // Note: Post stats (likes, replies, quotes, reposts) are now tracked via arrays,
+                            // updated directly during operation processing - no batch updates needed
+
                             // Batch update actor stats
                             if !actor_stats_deltas.is_empty() {
                                 if let Err(e) = batch_update_actor_stats(&mut conn_mut, actor_stats_deltas).await {
···
     let mut stats_deltas: HashMap<(i32, i64), super::PostStatsDeltas> = HashMap::new();
     let mut actor_stats_deltas: HashMap<i32, super::ActorStatsDeltas> = HashMap::new();
 
-    // Helper to accumulate deltas
-    let mut accumulate_delta = |actor_id: i32, rkey: i64, delta: super::PostStatsDeltas| {
-        stats_deltas.entry((actor_id, rkey))
-            .and_modify(|existing| {
-                existing.like_delta += delta.like_delta;
-                existing.reply_delta += delta.reply_delta;
-                existing.repost_delta += delta.repost_delta;
-                existing.quote_delta += delta.quote_delta;
-            })
-            .or_insert(delta);
+    // Helper to accumulate deltas (DEPRECATED - arrays updated directly now)
+    let mut accumulate_delta = |_actor_id: i32, _rkey: i64, _delta: super::PostStatsDeltas| {
+        // No-op: engagement is now tracked via arrays, updated directly during operations
     };
 
     // Helper to accumulate actor deltas
···
                 let op_elapsed = op_start.elapsed();
                 total_inserted += inserted.len();
 
-                // Generate deltas: increment like count for each liked post
-                for like in inserted {
-                    accumulate_delta(like.post_actor_id, like.post_rkey, super::PostStatsDeltas {
-                        like_delta: 1,
-                        ..Default::default()
-                    });
-                }
+                // Note: Like arrays are updated directly in bulk_copy, no deltas needed
+                // accumulate_delta is now a no-op
 
                 tracing::info!(
                     count = count,
···
             Ok(inserted) => {
                 total_inserted += inserted.len();
 
-                // Generate deltas: increment repost count for each reposted post
-                for repost in &inserted {
-                    accumulate_delta(repost.post_actor_id, repost.post_rkey, super::PostStatsDeltas {
-                        repost_delta: 1,
-                        ..Default::default()
-                    });
-                }
+                // Note: Repost arrays are updated directly in bulk_copy, no deltas needed
+                // accumulate_delta is now a no-op
 
                 tracing::info!(inserted = inserted.len(), "Bulk COPY reposts completed");
             }
···
 
                 // Generate deltas for posts
                 for post_data in &bulk_ops.posts {
-                    // Increment reply count on parent post
-                    if let (Some(parent_actor_id), Some(parent_rkey)) =
-                        (post_data.parent_post_actor_id, post_data.parent_post_rkey) {
-                        accumulate_delta(parent_actor_id, parent_rkey, super::PostStatsDeltas {
-                            reply_delta: 1,
-                            ..Default::default()
-                        });
-                    }
-
-                    // Increment quote count on embedded post
-                    if let (Some(embedded_actor_id), Some(embedded_rkey)) =
-                        (post_data.embedded_post_actor_id, post_data.embedded_post_rkey) {
-                        accumulate_delta(embedded_actor_id, embedded_rkey, super::PostStatsDeltas {
-                            quote_delta: 1,
-                            ..Default::default()
-                        });
-                    }
+                    // Note: Reply/quote arrays are updated directly in bulk_copy, no deltas needed
+                    // accumulate_delta is now a no-op
                 }
 
                 // Execute child table inserts
···
 /// Takes accumulated deltas and updates denormalized stats columns in posts table.
 /// Uses CASE-batched UPDATEs for efficiency (500 posts per UPDATE statement).
 /// Now uses the consolidated PostUpdate infrastructure.
-async fn batch_update_post_stats(
-    conn: &mut deadpool_postgres::Object,
-    deltas: std::collections::HashMap<(i32, i64), super::PostStatsDeltas>,
-) -> eyre::Result<()> {
-    if deltas.is_empty() {
-        return Ok(());
-    }
-
-    let start = std::time::Instant::now();
-    let total_posts = deltas.len();
-
-    // Convert deltas to Vec and sort by (actor_id, rkey) to ensure consistent lock ordering
-    // This prevents deadlocks when multiple workers update overlapping posts
-    let mut deltas_vec: Vec<((i32, i64), super::PostStatsDeltas)> = deltas.into_iter().collect();
-    deltas_vec.sort_by_key(|((actor_id, rkey), _)| (*actor_id, *rkey));
-
-    const BATCH_SIZE: usize = 500;
-    let batch_count = (deltas_vec.len() + BATCH_SIZE - 1) / BATCH_SIZE;
-
-    let mut total_updated = 0;
-
-    // Process deltas in batches using PostUpdate
-    for (batch_idx, chunk) in deltas_vec.chunks(BATCH_SIZE).enumerate() {
-        let batch_start = std::time::Instant::now();
-
-        // Convert to PostUpdate format
-        let mut count_deltas = std::collections::HashMap::new();
-        let mut post_keys = Vec::new();
-
-        for ((actor_id, rkey), delta) in chunk {
-            post_keys.push((*actor_id, *rkey));
-            count_deltas.insert(
-                (*actor_id, *rkey),
-                crate::db::operations::CountDeltas {
-                    like_count: Some(delta.like_delta),
-                    reply_count: Some(delta.reply_delta),
-                    repost_count: Some(delta.repost_delta),
-                    quote_count: Some(delta.quote_delta),
-                },
-            );
-        }
-
-        // Use consolidated PostUpdate API
-        let result = crate::db::operations::PostUpdate {
-            target: crate::db::operations::PostUpdateTarget::Batch { keys: post_keys },
-            count_deltas: Some(count_deltas),
-            ..Default::default()
-        }
-        .execute(conn)
-        .await?;
-
-        let updated = match result {
-            crate::db::operations::PostUpdateResult::Count(n) => n,
-            _ => unreachable!("PostUpdate with no RETURNING should return Count"),
-        };
-
-        total_updated += updated;
-
-        tracing::debug!(
-            batch = batch_idx + 1,
-            batch_size = chunk.len(),
-            updated = updated,
-            duration_ms = batch_start.elapsed().as_millis(),
-            "CASE-batched post stats UPDATE completed (via PostUpdate)"
-        );
-    }
-
-    let elapsed = start.elapsed();
-    tracing::debug!(
-        posts_updated = total_updated,
-        deltas_count = total_posts,
-        batches = batch_count,
-        batch_size = BATCH_SIZE,
-        duration_ms = elapsed.as_millis(),
-        "Batch updated post stats (CASE-batched via PostUpdate)"
-    );
-    Ok(())
-}
-
 /// Batch update actor stats in the database
 /// Updates denormalized stats columns in actors table (followers_count, following_count, posts_count, etc.)
 ///
···
             let mut actor_stats_deltas: std::collections::HashMap<i32, super::ActorStatsDeltas> = std::collections::HashMap::new();
             let mut event_cache_invalidations = Vec::new();
 
-            // Helper to accumulate deltas
-            let mut accumulate_delta = |actor_id: i32, rkey: i64, delta: super::PostStatsDeltas| {
-                stats_deltas.entry((actor_id, rkey))
-                    .and_modify(|existing| {
-                        existing.like_delta += delta.like_delta;
-                        existing.reply_delta += delta.reply_delta;
-                        existing.repost_delta += delta.repost_delta;
-                        existing.quote_delta += delta.quote_delta;
-                    })
-                    .or_insert(delta);
+            // Helper to accumulate deltas (DEPRECATED - arrays updated directly now)
+            let mut accumulate_delta = |_actor_id: i32, _rkey: i64, _delta: super::PostStatsDeltas| {
+                // No-op: engagement is now tracked via arrays, updated directly during operations
             };
 
             // Helper to accumulate actor deltas
···
                 );
             }
 
-            // Batch update post stats in the database
-            if !stats_deltas.is_empty() {
-                if let Err(e) = batch_update_post_stats(&mut conn, stats_deltas).await {
-                    tracing::error!(worker = name, "Failed to batch update post stats: {}", e);
-                    counter!("batch_writer_stats_update_error").increment(1);
-                }
-            }
+            // Note: Post stats (likes, replies, quotes, reposts) are now tracked via arrays,
+            // updated directly during operation processing - no batch updates needed
 
             // Batch update actor stats in the database
             if !actor_stats_deltas.is_empty() {

consumer/src/db/operations/feed.rs

···
 pub use helpers::{ensure_list_id, get_actor_id};
 pub use like::{like_delete, like_insert, LikeSubject};
 pub use post::{post_delete, post_insert};
-pub use post_update::{PostUpdate, PostUpdateTarget, PostUpdateReturning, PostUpdateResult, LikeArrayOp, CountDeltas, RecordDetachedOp};
+pub use post_update::{PostUpdate, PostUpdateTarget, PostUpdateReturning, PostUpdateResult, LikeArrayOp, RecordDetachedOp};
 pub use postgate::{postgate_delete, postgate_upsert};
 pub use repost::{repost_delete, repost_insert};
 pub use threadgate::{threadgate_delete, threadgate_get, threadgate_upsert};

consumer/src/db/operations/feed/like.rs

···
 ) -> Result<Option<LikeSubject>> {
     // Try removing from posts arrays first (most common case)
     // Use array_remove() - much simpler than manual array filtering!
+    // Note: like_count column removed in array-only migration
     let res = conn
         .query_opt(
             "UPDATE posts
              SET
                  like_actor_ids = array_remove(like_actor_ids, $2),
                  like_rkeys = array_remove(like_rkeys, $1),
-                 like_via_repost_data = like_via_repost_data - $2::text,
-                 like_count = cardinality(array_remove(like_actor_ids, $2))
+                 like_via_repost_data = like_via_repost_data - $2::text
              WHERE $2 = ANY(like_actor_ids)
                AND $1 = ANY(like_rkeys)
              RETURNING actor_id, rkey",

consumer/src/db/operations/feed/post_update.rs

···
     },
 }
 
-/// Count deltas for batch updates
-#[derive(Debug, Clone, Default)]
-pub struct CountDeltas {
-    pub like_count: Option<i32>,
-    pub reply_count: Option<i32>,
-    pub repost_count: Option<i32>,
-    pub quote_count: Option<i32>,
-}
-
 /// Postgate detach calculation
 #[derive(Debug, Clone)]
 pub enum RecordDetachedOp {
···
     /// Field updates (all optional)
     pub status: Option<PostStatus>,
     pub like_array_op: Option<LikeArrayOp>,
-    pub count_deltas: Option<HashMap<(i32, i64), CountDeltas>>,
     pub record_detached: Option<RecordDetachedOp>,
     pub tokens: Option<Vec<String>>,
 
···
             },
             status: None,
             like_array_op: None,
-            count_deltas: None,
             record_detached: None,
             tokens: None,
             returning: PostUpdateReturning::None,
···
                         param_idx += 3;
                     }
 
-                    // Update like_count
-                    clauses.push(format!(
-                        "like_count = cardinality(like_actor_ids || ${})",
-                        actor_param_idx
-                    ));
+                    // Note: like_count column removed in array-only migration
+                    // Count is now computed on-demand via array_length(like_actor_ids)
                 }
 
                 LikeArrayOp::AppendBatch { likes_by_post } => {
···
                         case_repost_data
                     ));
 
-                    clauses.push("like_count = cardinality(like_actor_ids) + 1".to_string());
+                    // Note: like_count column removed in array-only migration
+                    // Count is now computed on-demand via array_length(like_actor_ids)
                 }
 
                 LikeArrayOp::Remove {
···
                         actor_param_idx
                     ));
 
-                    // Update like_count
-                    clauses.push(format!(
-                        "like_count = cardinality(array_remove(like_actor_ids, ${}))",
-                        actor_param_idx
-                    ));
+                    // Note: like_count column removed in array-only migration
+                    // Count is now computed on-demand via array_length(like_actor_ids)
                 }
             }
-        }
-
-        // Count deltas (batch updates with CASE expressions)
-        if let Some(deltas_map) = &self.count_deltas {
-            // Build CASE expressions for each count field
-            // Format: like_count = CASE (actor_id, rkey) WHEN (1, 2) THEN LEAST(32767, GREATEST(0, COALESCE(like_count, 0) + delta)) WHEN ... END
-
-            let mut case_like = String::new();
-            let mut case_reply = String::new();
-            let mut case_repost = String::new();
-            let mut case_quote = String::new();
-
-            for (idx, ((post_actor_id, post_rkey), deltas)) in deltas_map.iter().enumerate() {
-                if idx > 0 {
-                    case_like.push(' ');
-                    case_reply.push(' ');
-                    case_repost.push(' ');
-                    case_quote.push(' ');
-                }
-
-                // Build CASE clauses with bounds checking (0-32767 for smallint)
-                if let Some(like_delta) = deltas.like_count {
-                    case_like.push_str(&format!(
-                        "WHEN ({}, {}) THEN LEAST(32767, GREATEST(0, COALESCE(like_count, 0) + {}))::smallint",
-                        post_actor_id, post_rkey, like_delta
-                    ));
-                } else {
-                    case_like.push_str(&format!(
-                        "WHEN ({}, {}) THEN like_count",
-                        post_actor_id, post_rkey
-                    ));
-                }
-
-                if let Some(reply_delta) = deltas.reply_count {
-                    case_reply.push_str(&format!(
-                        "WHEN ({}, {}) THEN LEAST(32767, GREATEST(0, COALESCE(reply_count, 0) + {}))::smallint",
-                        post_actor_id, post_rkey, reply_delta
-                    ));
-                } else {
-                    case_reply.push_str(&format!(
-                        "WHEN ({}, {}) THEN reply_count",
-                        post_actor_id, post_rkey
-                    ));
-                }
-
-                if let Some(repost_delta) = deltas.repost_count {
-                    case_repost.push_str(&format!(
-                        "WHEN ({}, {}) THEN LEAST(32767, GREATEST(0, COALESCE(repost_count, 0) + {}))::smallint",
-                        post_actor_id, post_rkey, repost_delta
-                    ));
-                } else {
-                    case_repost.push_str(&format!(
-                        "WHEN ({}, {}) THEN repost_count",
-                        post_actor_id, post_rkey
-                    ));
-                }
-
-                if let Some(quote_delta) = deltas.quote_count {
-                    case_quote.push_str(&format!(
-                        "WHEN ({}, {}) THEN LEAST(32767, GREATEST(0, COALESCE(quote_count, 0) + {}))::smallint",
-                        post_actor_id, post_rkey, quote_delta
-                    ));
-                } else {
-                    case_quote.push_str(&format!(
-                        "WHEN ({}, {}) THEN quote_count",
-                        post_actor_id, post_rkey
-                    ));
-                }
-            }
-
-            // Add SET clauses with CASE expressions
-            clauses.push(format!("like_count = CASE (actor_id, rkey) {} END", case_like));
-            clauses.push(format!("reply_count = CASE (actor_id, rkey) {} END", case_reply));
-            clauses.push(format!("repost_count = CASE (actor_id, rkey) {} END", case_repost));
-            clauses.push(format!("quote_count = CASE (actor_id, rkey) {} END", case_quote));
         }
 
         // Record detached (postgate)

consumer/tests/feed_operations_test.rs

···
     );
     assert_eq!(result.wrap_err("Operation failed")?, 1, "Should insert 1 like");
 
-    // Verify like was created in embedded arrays
+    // Verify like was created in embedded arrays (array-only tracking)
     let like_row = tx
         .query_one(
             "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
···
         .wrap_err("Failed to query like")?;
 
     let like_exists: i32 = like_row.get(0);
-    let like_count: Option<i32> = like_row.get(1);
-    let array_len: Option<i32> = like_row.get(2);
+    let array_len: Option<i32> = like_row.get(1);
 
     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(())
 }
···
         .query_one(
             "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
···
         .wrap_err("Failed to query post")?;
 
     let like_exists: i32 = like_row.get(0);
-    let like_count: Option<i32> = like_row.get(1);
-    let array_len: Option<i32> = like_row.get(2);
+    let array_len: Option<i32> = like_row.get(1);
 
     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(())
 }

migrations/2025-12-06-224848_array_only_engagement/down.sql

···
+-- Rollback: Restore count columns from arrays
+-- This migration reverses the array-only engagement tracking
+
+-- Step 1: Restore count columns
+ALTER TABLE posts
+  ADD COLUMN like_count integer DEFAULT 0 NOT NULL,
+  ADD COLUMN reply_count integer DEFAULT 0 NOT NULL,
+  ADD COLUMN repost_count integer DEFAULT 0 NOT NULL,
+  ADD COLUMN quote_count integer DEFAULT 0 NOT NULL;
+
+-- Step 2: Restore counts from arrays
+-- Use COALESCE to handle NULL arrays (treat as 0)
+UPDATE posts SET
+  like_count = COALESCE(array_length(like_actor_ids, 1), 0),
+  reply_count = COALESCE(array_length(reply_actor_ids, 1), 0),
+  repost_count = COALESCE(array_length(repost_actor_ids, 1), 0),
+  quote_count = COALESCE(array_length(quote_actor_ids, 1), 0);
+
+-- Step 3: Restore indexes on count columns
+CREATE INDEX idx_posts_like_count_desc ON posts (like_count DESC);
+CREATE INDEX idx_posts_engagement_score ON posts (
+  (like_count + reply_count * 2 + repost_count) DESC
+);
+
+-- Step 4: Drop array columns and indexes (keep like arrays from previous migration)
+DROP INDEX IF EXISTS idx_posts_reply_actor_ids_gin;
+DROP INDEX IF EXISTS idx_posts_quote_actor_ids_gin;
+DROP INDEX IF EXISTS idx_posts_repost_actor_ids_gin;
+
+ALTER TABLE posts
+  DROP COLUMN reply_actor_ids,
+  DROP COLUMN reply_rkeys,
+  DROP COLUMN quote_actor_ids,
+  DROP COLUMN quote_rkeys,
+  DROP COLUMN repost_actor_ids,
+  DROP COLUMN repost_rkeys;
+
+-- Note: We keep like_actor_ids and like_rkeys from the previous migration
+-- The like_count column is now restored and will be maintained alongside arrays

migrations/2025-12-06-224848_array_only_engagement/up.sql

···
+-- Array-Only Engagement Tracking Migration
+-- Replace {like,reply,repost,quote}_count columns with array-only tracking
+-- Compute counts on-demand via array_length()
+
+-- Step 1: Add new array columns for replies, quotes, and reposts
+-- (like arrays already exist from previous migration)
+ALTER TABLE posts
+  ADD COLUMN reply_actor_ids integer[],
+  ADD COLUMN reply_rkeys bigint[],
+  ADD COLUMN quote_actor_ids integer[],
+  ADD COLUMN quote_rkeys bigint[],
+  ADD COLUMN repost_actor_ids integer[],
+  ADD COLUMN repost_rkeys bigint[];
+
+-- Step 2: Backfill replies from posts.parent_post_* relationships
+-- Group by parent post and collect all reply actor_ids and rkeys
+UPDATE posts p
+SET
+  reply_actor_ids = COALESCE(r.actor_ids, ARRAY[]::integer[]),
+  reply_rkeys = COALESCE(r.rkeys, ARRAY[]::bigint[])
+FROM (
+  SELECT
+    parent_post_actor_id,
+    parent_post_rkey,
+    array_agg(actor_id ORDER BY rkey) as actor_ids,
+    array_agg(rkey ORDER BY rkey) as rkeys
+  FROM posts
+  WHERE parent_post_actor_id IS NOT NULL
+  GROUP BY parent_post_actor_id, parent_post_rkey
+) r
+WHERE p.actor_id = r.parent_post_actor_id
+  AND p.rkey = r.parent_post_rkey;
+
+-- Step 3: Backfill quotes from posts.embedded_post_* relationships
+-- Quotes are posts with embedded_post but NO parent (replies have parents)
+UPDATE posts p
+SET
+  quote_actor_ids = COALESCE(q.actor_ids, ARRAY[]::integer[]),
+  quote_rkeys = COALESCE(q.rkeys, ARRAY[]::bigint[])
+FROM (
+  SELECT
+    embedded_post_actor_id,
+    embedded_post_rkey,
+    array_agg(actor_id ORDER BY rkey) as actor_ids,
+    array_agg(rkey ORDER BY rkey) as rkeys
+  FROM posts
+  WHERE embedded_post_actor_id IS NOT NULL
+    AND parent_post_actor_id IS NULL  -- Quotes don't have parents (not replies)
+  GROUP BY embedded_post_actor_id, embedded_post_rkey
+) q
+WHERE p.actor_id = q.embedded_post_actor_id
+  AND p.rkey = q.embedded_post_rkey;
+
+-- Step 4: Backfill reposts from reposts table
+UPDATE posts p
+SET
+  repost_actor_ids = COALESCE(r.actor_ids, ARRAY[]::integer[]),
+  repost_rkeys = COALESCE(r.rkeys, ARRAY[]::bigint[])
+FROM (
+  SELECT
+    post_actor_id,
+    post_rkey,
+    array_agg(actor_id ORDER BY rkey) as actor_ids,
+    array_agg(rkey ORDER BY rkey) as rkeys
+  FROM reposts
+  GROUP BY post_actor_id, post_rkey
+) r
+WHERE p.actor_id = r.post_actor_id
+  AND p.rkey = r.post_rkey;
+
+-- Step 5: Create GIN indexes for array lookups
+-- These enable fast "who replied/quoted/reposted?" queries
+CREATE INDEX idx_posts_reply_actor_ids_gin ON posts USING GIN (reply_actor_ids);
+CREATE INDEX idx_posts_quote_actor_ids_gin ON posts USING GIN (quote_actor_ids);
+CREATE INDEX idx_posts_repost_actor_ids_gin ON posts USING GIN (repost_actor_ids);
+
+-- Step 6: Validate backfill (should return 0 rows with mismatches)
+-- This ensures arrays match the existing count columns before we drop them
+DO $$
+DECLARE
+  mismatches integer;
+  like_mismatches integer;
+  reply_mismatches integer;
+  quote_mismatches integer;
+  repost_mismatches integer;
+BEGIN
+  -- Check reply counts
+  SELECT COUNT(*) INTO reply_mismatches
+  FROM posts
+  WHERE COALESCE(array_length(reply_actor_ids, 1), 0) != COALESCE(reply_count, 0);
+
+  -- Check quote counts
+  SELECT COUNT(*) INTO quote_mismatches
+  FROM posts
+  WHERE COALESCE(array_length(quote_actor_ids, 1), 0) != COALESCE(quote_count, 0);
+
+  -- Check repost counts
+  SELECT COUNT(*) INTO repost_mismatches
+  FROM posts
+  WHERE COALESCE(array_length(repost_actor_ids, 1), 0) != COALESCE(repost_count, 0);
+
+  -- Check like counts (should already be correct from previous migration)
+  SELECT COUNT(*) INTO like_mismatches
+  FROM posts
+  WHERE COALESCE(array_length(like_actor_ids, 1), 0) != COALESCE(like_count, 0);
+
+  mismatches := like_mismatches + reply_mismatches + quote_mismatches + repost_mismatches;
+
+  IF mismatches > 0 THEN
+    -- WARNING instead of EXCEPTION: Count columns may be stale (that's why we're migrating!)
+    -- Arrays are built from actual FK relationships, so they're the source of truth.
+    RAISE WARNING 'Array backfill differs from old counts: % total (likes: %, replies: %, quotes: %, reposts: %). This is expected if counts drifted from reality.',
+      mismatches, like_mismatches, reply_mismatches, quote_mismatches, repost_mismatches;
+    RAISE NOTICE 'Arrays are built from actual relationships and will be the new source of truth.';
+  ELSE
+    RAISE NOTICE 'Backfill validation passed: all counts match arrays';
+    RAISE NOTICE '  - Like arrays: correct';
+    RAISE NOTICE '  - Reply arrays: correct';
+    RAISE NOTICE '  - Quote arrays: correct';
+    RAISE NOTICE '  - Repost arrays: correct';
+  END IF;
+END $$;
+
+-- Step 7: Drop old indexes that depend on count columns
+DROP INDEX IF EXISTS idx_posts_like_count_desc;
+DROP INDEX IF EXISTS idx_posts_engagement_score;
+
+-- Step 8: Drop count columns (arrays are now the single source of truth)
+ALTER TABLE posts
+  DROP COLUMN like_count,
+  DROP COLUMN reply_count,
+  DROP COLUMN repost_count,
+  DROP COLUMN quote_count;

parakeet-db/src/models.rs

···
     pub facet_7: Option<crate::composite_types::FacetEmbed>,
     pub facet_8: Option<crate::composite_types::FacetEmbed>,
     pub mentions: Option<Vec<Option<i32>>>,  // Array of actor_ids
-    // Aggregate stats (denormalized for performance, NULL = 0)
-    // INTEGER in PostgreSQL = Option<i32> in Rust (changed from SMALLINT to prevent overflow)
-    pub like_count: Option<i32>,
-    pub reply_count: Option<i32>,
-    pub repost_count: Option<i32>,
-    pub quote_count: Option<i32>,
-    // Embedded likes
-    //   like_actor_ids[i] pairs with like_rkeys[i]
+    // Embedded engagement arrays (array-only tracking, compute counts via array_length())
+    // Likes: like_actor_ids[i] pairs with like_rkeys[i]
     pub like_actor_ids: Option<Vec<i32>>,           // [actor1, actor2, ...] (who liked)
     pub like_rkeys: Option<Vec<i64>>,               // [rkey1, rkey2, ...] (when they liked)
     // 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<serde_json::Value>,
-                                                   // Note: created_at derived from TID rkey via created_at() method
+    // Replies: reply_actor_ids[i] pairs with reply_rkeys[i]
+    pub reply_actor_ids: Option<Vec<i32>>,          // [actor1, actor2, ...] (who replied)
+    pub reply_rkeys: Option<Vec<i64>>,              // [rkey1, rkey2, ...] (when they replied)
+    // Quotes: quote_actor_ids[i] pairs with quote_rkeys[i]
+    pub quote_actor_ids: Option<Vec<i32>>,          // [actor1, actor2, ...] (who quoted)
+    pub quote_rkeys: Option<Vec<i64>>,              // [rkey1, rkey2, ...] (when they quoted)
+    // Reposts: repost_actor_ids[i] pairs with repost_rkeys[i]
+    pub repost_actor_ids: Option<Vec<i32>>,         // [actor1, actor2, ...] (who reposted)
+    pub repost_rkeys: Option<Vec<i64>>,             // [rkey1, rkey2, ...] (when they reposted)
+                                                    // Note: created_at derived from TID rkey via created_at() method
 }
 
 impl Post {
···
             .map(|v| v.len())
             .unwrap_or(0)
     }
+
+    // Count helpers (compute from array lengths, replacing old count columns)
+
+    /// Get like count from array length
+    pub fn like_count(&self) -> usize {
+        self.like_actor_ids.as_ref().map_or(0, |v| v.len())
+    }
+
+    /// Get reply count from array length
+    pub fn reply_count(&self) -> usize {
+        self.reply_actor_ids.as_ref().map_or(0, |v| v.len())
+    }
+
+    /// Get quote count from array length
+    pub fn quote_count(&self) -> usize {
+        self.quote_actor_ids.as_ref().map_or(0, |v| v.len())
+    }
+
+    /// Get repost count from array length
+    pub fn repost_count(&self) -> usize {
+        self.repost_actor_ids.as_ref().map_or(0, |v| v.len())
+    }
 }
 
 /// Represents a single like extracted from post arrays
···
 impl PostStats {
     pub fn from_post(post: &Post) -> Self {
         Self {
-            likes: post.like_count.unwrap_or(0) as i32,
-            replies: post.reply_count.unwrap_or(0) as i32,
-            reposts: post.repost_count.unwrap_or(0) as i32,
-            quotes: post.quote_count.unwrap_or(0) as i32,
+            likes: post.like_count() as i32,
+            replies: post.reply_count() as i32,
+            reposts: post.repost_count() as i32,
+            quotes: post.quote_count() as i32,
         }
     }
 

parakeet-db/src/schema.rs

···
 }
 
 diesel::table! {
-    post_likes (actor_id, rkey) {
-        actor_id -> Int4,
-        rkey -> Int8,
-        post_actor_id -> Int4,
-        post_rkey -> Int8,
-        via_repost_actor_id -> Nullable<Int4>,
-        via_repost_rkey -> Nullable<Int8>,
-    }
-}
-
-diesel::table! {
     postgate_detached (post_actor_id, post_rkey, detached_post_actor_id, detached_post_rkey) {
         post_actor_id -> Int4,
         post_rkey -> Int8,
···
         facet_7 -> Nullable<PostFacetEmbed>,
         facet_8 -> Nullable<PostFacetEmbed>,
         mentions -> Nullable<Array<Nullable<Int4>>>,
-        like_count -> Nullable<Int4>,
-        reply_count -> Nullable<Int4>,
-        repost_count -> Nullable<Int4>,
-        quote_count -> Nullable<Int4>,
         like_actor_ids -> Nullable<Array<Int4>>,
         like_rkeys -> Nullable<Array<Int8>>,
         like_via_repost_data -> Nullable<Jsonb>,
+        reply_actor_ids -> Nullable<Array<Int4>>,
+        reply_rkeys -> Nullable<Array<Int8>>,
+        quote_actor_ids -> Nullable<Array<Int4>>,
+        quote_rkeys -> Nullable<Array<Int8>>,
+        repost_actor_ids -> Nullable<Array<Int4>>,
+        repost_rkeys -> Nullable<Array<Int8>>,
     }
 }
 
···
     mutes,
     notifications,
     post_aggregate_stats,
-    post_likes,
     postgate_detached,
     postgates,
     posts,

parakeet/src/loaders/mod.rs

···
 pub use labeler::{EnrichedLabeler, LabelLoader, LabelServiceLoader, LabelServiceLoaderRet};
 pub use list::{EnrichedList, ListLoader, ListLoaderRet, ListStateLoader};
 pub use misc::{EnrichedStarterPack, EnrichedVerification, StarterPackLoader, StarterPackLoaderRet, VerificationLoader};
-pub use post::{EnrichedThreadgate, HydratedPost, PostLoader, PostLoaderRet, PostStateLoader};
+pub use post::{EnrichedThreadgate, HydratedPost, PostLoader, PostLoaderRet, PostStateLoader, PostWithComputed};
 pub use profile::{
     EnrichedStatus, HandleLoader, Profile, ProfileByIdLoader, ProfileLoader, ProfileLoaderRet, ProfileStateLoader, ProfileStatsLoader,
 };

parakeet/src/loaders/post.rs

···
     }
 }
 
+/// Struct to capture SQL query results from build_posts_batch_query()
+///
+/// This struct MUST match the SELECT columns in build_posts_batch_query() exactly.
+/// It's public for testing - tests can use this to validate the SQL query is executable.
+/// If the query changes but this struct doesn't (or vice versa), tests will fail.
+#[derive(diesel::QueryableByName)]
+#[allow(dead_code, reason = "Diesel QueryableByName requires all SQL columns even if unused")]
+pub struct PostWithComputed {
+    #[diesel(sql_type = diesel::sql_types::Integer)]
+    pub actor_id: i32,
+    #[diesel(sql_type = diesel::sql_types::BigInt)]
+    pub rkey: i64,
+    #[diesel(sql_type = diesel::sql_types::Binary)]
+    pub cid: Vec<u8>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Binary>)]
+    pub content: Option<Vec<u8>>,
+    #[diesel(sql_type = diesel::sql_types::Array<diesel::sql_types::Nullable<parakeet_db::schema::sql_types::LanguageCode>>)]
+    pub langs: Vec<Option<parakeet_db::types::LanguageCode>>,
+    #[diesel(sql_type = diesel::sql_types::Array<diesel::sql_types::Nullable<diesel::sql_types::Text>>)]
+    pub tags: Vec<Option<String>>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
+    pub parent_post_actor_id: Option<i32>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
+    pub parent_post_rkey: Option<i64>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
+    pub root_post_actor_id: Option<i32>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
+    pub root_post_rkey: Option<i64>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::EmbedType>)]
+    pub embed_type: Option<parakeet_db::types::EmbedType>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::EmbedType>)]
+    pub embed_subtype: Option<parakeet_db::types::EmbedType>,
+    #[diesel(sql_type = diesel::sql_types::Bool)]
+    pub violates_threadgate: bool,
+    #[diesel(sql_type = parakeet_db::schema::sql_types::PostStatus)]
+    pub status: parakeet_db::types::PostStatus,
+    #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostVideoEmbed>)]
+    pub video_embed: Option<parakeet_db::composite_types::VideoEmbed>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostExtEmbed>)]
+    pub ext_embed: Option<parakeet_db::composite_types::ExtEmbed>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostImageEmbed>)]
+    pub image_1: Option<parakeet_db::composite_types::ImageEmbed>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostImageEmbed>)]
+    pub image_2: Option<parakeet_db::composite_types::ImageEmbed>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostImageEmbed>)]
+    pub image_3: Option<parakeet_db::composite_types::ImageEmbed>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostImageEmbed>)]
+    pub image_4: Option<parakeet_db::composite_types::ImageEmbed>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
+    pub embedded_post_actor_id: Option<i32>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
+    pub embedded_post_rkey: Option<i64>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Bool>)]
+    pub record_detached: Option<bool>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostFacetEmbed>)]
+    pub facet_1: Option<parakeet_db::composite_types::FacetEmbed>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostFacetEmbed>)]
+    pub facet_2: Option<parakeet_db::composite_types::FacetEmbed>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostFacetEmbed>)]
+    pub facet_3: Option<parakeet_db::composite_types::FacetEmbed>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostFacetEmbed>)]
+    pub facet_4: Option<parakeet_db::composite_types::FacetEmbed>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostFacetEmbed>)]
+    pub facet_5: Option<parakeet_db::composite_types::FacetEmbed>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostFacetEmbed>)]
+    pub facet_6: Option<parakeet_db::composite_types::FacetEmbed>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostFacetEmbed>)]
+    pub facet_7: Option<parakeet_db::composite_types::FacetEmbed>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostFacetEmbed>)]
+    pub facet_8: Option<parakeet_db::composite_types::FacetEmbed>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Array<diesel::sql_types::Integer>>)]
+    pub like_actor_ids: Option<Vec<i32>>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Array<diesel::sql_types::BigInt>>)]
+    pub like_rkeys: Option<Vec<i64>>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Jsonb>)]
+    pub like_via_repost_data: Option<serde_json::Value>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Array<diesel::sql_types::Integer>>)]
+    pub reply_actor_ids: Option<Vec<i32>>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Array<diesel::sql_types::BigInt>>)]
+    pub reply_rkeys: Option<Vec<i64>>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Array<diesel::sql_types::Integer>>)]
+    pub quote_actor_ids: Option<Vec<i32>>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Array<diesel::sql_types::BigInt>>)]
+    pub quote_rkeys: Option<Vec<i64>>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Array<diesel::sql_types::Integer>>)]
+    pub repost_actor_ids: Option<Vec<i32>>,
+    #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Array<diesel::sql_types::BigInt>>)]
+    pub repost_rkeys: Option<Vec<i64>>,
+}
+
 /// Build SQL query for batch loading posts with computed fields
 ///
 /// This function is public for testing purposes.
···
             p.facet_6,
             p.facet_7,
             p.facet_8,
-            -- Aggregate stats (denormalized columns)
-            p.like_count,
-            p.reply_count,
-            p.repost_count,
-            p.quote_count
+            -- Engagement arrays (array-only tracking)
+            p.like_actor_ids,
+            p.like_rkeys,
+            p.like_via_repost_data,
+            p.reply_actor_ids,
+            p.reply_rkeys,
+            p.quote_actor_ids,
+            p.quote_rkeys,
+            p.repost_actor_ids,
+            p.repost_rkeys
          FROM posts p
          INNER JOIN unnest($1::integer[], $2::bigint[]) AS lookup(lookup_actor_id, lookup_rkey)
             ON p.actor_id = lookup.lookup_actor_id AND p.rkey = lookup.lookup_rkey
···
 
         let query = build_posts_batch_query();
 
-        // Define a struct to capture the SQL results (simplified - no LEFT JOIN fields)
-        #[derive(diesel::QueryableByName)]
-        #[allow(dead_code, reason = "Diesel QueryableByName requires all SQL columns even if unused")]
-        struct PostWithComputed {
-            // Actual fields from posts table
-            #[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<u8>,
-            // Note: created_at computed from rkey in Rust via tid_to_datetime()
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Binary>)]
-            content: Option<Vec<u8>>, // Compressed content (BYTEA)
-            #[diesel(sql_type = diesel::sql_types::Array<diesel::sql_types::Nullable<parakeet_db::schema::sql_types::LanguageCode>>)]
-            langs: Vec<Option<parakeet_db::types::LanguageCode>>,
-            #[diesel(sql_type = diesel::sql_types::Array<diesel::sql_types::Nullable<diesel::sql_types::Text>>)]
-            tags: Vec<Option<String>>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
-            parent_post_actor_id: Option<i32>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
-            parent_post_rkey: Option<i64>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
-            root_post_actor_id: Option<i32>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
-            root_post_rkey: Option<i64>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::EmbedType>)]
-            embed_type: Option<parakeet_db::types::EmbedType>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::EmbedType>)]
-            embed_subtype: Option<parakeet_db::types::EmbedType>,
-            #[diesel(sql_type = diesel::sql_types::Bool)]
-            violates_threadgate: bool,
-            #[diesel(sql_type = parakeet_db::schema::sql_types::PostStatus)]
-            status: parakeet_db::types::PostStatus,
-            // Embed composite fields
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostVideoEmbed>)]
-            video_embed: Option<parakeet_db::composite_types::VideoEmbed>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostExtEmbed>)]
-            ext_embed: Option<parakeet_db::composite_types::ExtEmbed>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostImageEmbed>)]
-            image_1: Option<parakeet_db::composite_types::ImageEmbed>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostImageEmbed>)]
-            image_2: Option<parakeet_db::composite_types::ImageEmbed>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostImageEmbed>)]
-            image_3: Option<parakeet_db::composite_types::ImageEmbed>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostImageEmbed>)]
-            image_4: Option<parakeet_db::composite_types::ImageEmbed>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
-            embedded_post_actor_id: Option<i32>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
-            embedded_post_rkey: Option<i64>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Bool>)]
-            record_detached: Option<bool>,
-            // Facet composite fields (loaded inline to avoid separate query)
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostFacetEmbed>)]
-            facet_1: Option<parakeet_db::composite_types::FacetEmbed>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostFacetEmbed>)]
-            facet_2: Option<parakeet_db::composite_types::FacetEmbed>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostFacetEmbed>)]
-            facet_3: Option<parakeet_db::composite_types::FacetEmbed>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostFacetEmbed>)]
-            facet_4: Option<parakeet_db::composite_types::FacetEmbed>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostFacetEmbed>)]
-            facet_5: Option<parakeet_db::composite_types::FacetEmbed>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostFacetEmbed>)]
-            facet_6: Option<parakeet_db::composite_types::FacetEmbed>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostFacetEmbed>)]
-            facet_7: Option<parakeet_db::composite_types::FacetEmbed>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::PostFacetEmbed>)]
-            facet_8: Option<parakeet_db::composite_types::FacetEmbed>,
-            // Aggregate stats (denormalized columns - changed from SmallInt to Integer)
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
-            like_count: Option<i32>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
-            reply_count: Option<i32>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
-            repost_count: Option<i32>,
-            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
-            quote_count: Option<i32>,
-            // Note: author_did comes from posts_with_keys, not from query result
-        }
+        // PostWithComputed struct is now defined at module level for reuse in tests
 
         let posts_query_start = std::time::Instant::now();
         let posts_with_computed: Vec<PostWithComputed> = diesel_async::RunQueryDsl::load(
···
                     facet_7: None,
                     facet_8: None,
                     mentions: None,
-                    // Denormalized stats (loaded from columns)
-                    like_count: p.like_count,
-                    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_data: None,
+                    // Engagement arrays (array-only tracking, counts computed via helper methods)
+                    like_actor_ids: p.like_actor_ids.clone(),
+                    like_rkeys: p.like_rkeys.clone(),
+                    like_via_repost_data: p.like_via_repost_data.clone(),
+                    reply_actor_ids: p.reply_actor_ids.clone(),
+                    reply_rkeys: p.reply_rkeys.clone(),
+                    quote_actor_ids: p.quote_actor_ids.clone(),
+                    quote_rkeys: p.quote_rkeys.clone(),
+                    repost_actor_ids: p.repost_actor_ids.clone(),
+                    repost_rkeys: p.repost_rkeys.clone(),
                 };
 
                 // Encode TIDs using Rust utility functions
···
                     p.facet_6,
                     p.facet_7,
                     p.facet_8,
-                    // Stats data (denormalized columns)
-                    p.like_count,
-                    p.reply_count,
-                    p.repost_count,
-                    p.quote_count,
                 ))
             })
             .collect();
···
         // Collect all mention actor IDs from facets to batch lookup DIDs
         let facets_start = std::time::Instant::now();
         let mut mention_actor_ids: Vec<i32> = 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);
···
         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))
+            .map(|(post, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _, _)| (post.actor_id, post.rkey))
             .collect();
 
         // Load threadgates separately - need to reconstruct post URIs from post_id FK
···
                     facet_6,
                     facet_7,
                     facet_8,
-                    like_count,
-                    reply_count,
-                    repost_count,
-                    quote_count,
                 )| {
                     // 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];
···
                         &created_at,
                     );
 
-                    // Create PostStats from denormalized columns
+                    // Create PostStats from array lengths using helper methods
                     let stats = parakeet_db::models::PostStats {
-                        likes: like_count.unwrap_or(0),
-                        replies: reply_count.unwrap_or(0),
-                        reposts: repost_count.unwrap_or(0),
-                        quotes: quote_count.unwrap_or(0),
+                        likes: post.like_count() as i32,
+                        replies: post.reply_count() as i32,
+                        reposts: post.repost_count() as i32,
+                        quotes: post.quote_count() as i32,
                     };
 
                     let hydrated_post = HydratedPost {

parakeet/tests/loaders_test.rs

···
     // Use the actual PostLoader query builder (no SQL duplication!)
     let query = parakeet::loaders::build_posts_batch_query();
 
-    #[derive(diesel::QueryableByName)]
-    #[allow(dead_code)]
-    struct PostWithComputed {
-        #[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<u8>,
-        #[diesel(sql_type = diesel::sql_types::Timestamptz)]
-        created_at: chrono::DateTime<chrono::Utc>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Binary>)]
-        content: Option<Vec<u8>>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::SmallInt>)]
-        content_version: Option<i16>,
-        #[diesel(sql_type = diesel::sql_types::Array<diesel::sql_types::Nullable<parakeet_db::schema::sql_types::LanguageCode>>)]
-        langs: Vec<Option<parakeet_db::types::LanguageCode>>,
-        #[diesel(sql_type = diesel::sql_types::Array<diesel::sql_types::Nullable<diesel::sql_types::Text>>)]
-        tags: Vec<Option<String>>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
-        parent_post_id: Option<i64>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
-        root_post_id: Option<i64>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::EmbedType>)]
-        embed_type: Option<parakeet_db::types::EmbedType>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<parakeet_db::schema::sql_types::EmbedType>)]
-        embed_subtype: Option<parakeet_db::types::EmbedType>,
-        #[diesel(sql_type = diesel::sql_types::Bool)]
-        violates_threadgate: bool,
-        #[diesel(sql_type = parakeet_db::schema::sql_types::PostStatus)]
-        status: parakeet_db::types::PostStatus,
-        #[diesel(sql_type = diesel::sql_types::Text)]
-        author_did: String,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
-        parent_did: Option<String>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
-        parent_rkey: Option<i64>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Binary>)]
-        parent_cid_bytes: Option<Vec<u8>>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
-        root_did: Option<String>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::BigInt>)]
-        root_rkey: Option<i64>,
-        #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Binary>)]
-        root_cid_bytes: Option<Vec<u8>>,
-    }
-
+    // Use the production PostWithComputed struct (no duplication!)
+    // If SQL query changes but struct doesn't match, this test will fail
     let actor_ids: Vec<i32> = vec![];
     let min_rkey = 0i64;
     let max_rkey = i64::MAX;
···
         .bind::<diesel::sql_types::Array<diesel::sql_types::BigInt>, _>(&rkeys)
         .bind::<diesel::sql_types::BigInt, _>(min_rkey)
         .bind::<diesel::sql_types::BigInt, _>(max_rkey)
-        .load::<PostWithComputed>(&mut conn)
+        .load::<parakeet::loaders::PostWithComputed>(&mut conn)
         .await
         .wrap_err("PostLoader main query SQL failed")?;