feat: denormalization strategy; test driven development

consumer/src/db/gates/queries.rs

···
                 SELECT id FROM actors WHERE did = $2
             )
             SELECT
-                EXISTS(SELECT 1 FROM follows WHERE actor_id = (SELECT id FROM root_actor) AND subject_actor_id = (SELECT id FROM post_actor)) as following,
-                EXISTS(SELECT 1 FROM follows WHERE actor_id = (SELECT id FROM post_actor) AND subject_actor_id = (SELECT id FROM root_actor)) as followed",
+                EXISTS(
+                    SELECT 1 FROM actors a, unnest(COALESCE(a.following, ARRAY[]::follow_record[])) AS f
+                    WHERE a.id = (SELECT id FROM root_actor)
+                      AND (f).subject_actor_id = (SELECT id FROM post_actor)
+                ) as following,
+                EXISTS(
+                    SELECT 1 FROM actors a, unnest(COALESCE(a.following, ARRAY[]::follow_record[])) AS f
+                    WHERE a.id = (SELECT id FROM post_actor)
+                      AND (f).subject_actor_id = (SELECT id FROM root_actor)
+                ) as followed",
             &[&root_author, &post_author],
         )
         .await?;
···
                     SPLIT_PART(SUBSTRING(uri FROM 6), '/', 3) as rkey
                 FROM list_uris
             ),
-            list_ids AS (
-                SELECT l.id
+            list_keys AS (
+                SELECT a.id as list_owner_actor_id, lp.rkey as list_rkey
                 FROM list_parts lp
                 INNER JOIN actors a ON a.did = lp.did
                 INNER JOIN lists l ON l.actor_id = a.id AND l.rkey = lp.rkey
···
             SELECT count(*)
             FROM list_items li
             INNER JOIN actors a ON a.did = $2
-            WHERE li.list_id IN (SELECT id FROM list_ids)
-              AND li.subject_actor_id = a.id",
+            INNER JOIN list_keys lk ON li.list_owner_actor_id = lk.list_owner_actor_id
+                                   AND li.list_rkey = lk.list_rkey
+            WHERE li.subject_actor_id = a.id",
             &[&allow_lists, &post_author],
         )
         .await?;

consumer/src/db/mod.rs

···
 ) -> Result<bool> {
     let row = conn
         .query_opt(
-            "SELECT 1 FROM thread_mutes
-             WHERE actor_id = $1
-               AND root_post_actor_id = $2
-               AND root_post_rkey = $3",
+            "SELECT 1 FROM actors, unnest(COALESCE(thread_mutes, ARRAY[]::thread_mute_record[])) AS tm
+             WHERE id = $1
+               AND (tm).root_post_actor_id = $2
+               AND (tm).root_post_rkey = $3",
             &[&recipient_actor_id, &root_post_actor_id, &root_post_rkey],
         )
         .await?;

consumer/src/db/operations/graph.rs

···
     let (table_id, key_id) = crate::database_writer::locking::actor_record_lock("follows", actor_id, rkey);
     crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?;
 
+    // Check if exact same follow already exists (same subject_actor_id and rkey)
+    let already_exists = conn
+        .query_opt(
+            "SELECT 1 FROM actors, unnest(COALESCE(following, ARRAY[]::follow_record[])) AS f
+             WHERE id = $1 AND (f).subject_actor_id = $2 AND (f).rkey = $3",
+            &[&actor_id, &subject_actor_id, &rkey],
+        )
+        .await?
+        .is_some();
+
+    if already_exists {
+        return Ok(0); // Duplicate, no change needed
+    }
+
     // BIDIRECTIONAL UPDATE: Update both following[] and followers[] arrays
-    // 1. Add to actor's following array
-    // 2. Add to subject_actor's followers array
+    // 1. Add to actor's following array - stores (subject_actor_id, rkey)
+    // 2. Add to subject_actor's followers array - stores (follower_actor_id, rkey) using same follow_record type
     // Returns number of rows updated (should be 2 if both actors exist)
     let rows = conn
         .execute(
-            "WITH follow_record AS (
-                SELECT ROW($2, $3)::follow_record as rec
+            "WITH follow_rec AS (
+                SELECT ROW($3, $2)::follow_record as rec  -- (subject_actor_id, rkey)
             ),
             update_following AS (
                 UPDATE actors
                 SET following = COALESCE(following, ARRAY[]::follow_record[]) ||
-                                (SELECT ARRAY[rec] FROM follow_record)
+                                (SELECT ARRAY[rec] FROM follow_rec)
                 WHERE id = $1
                   AND NOT EXISTS (
                     SELECT 1 FROM unnest(following) f WHERE (f).rkey = $2
···
             update_followers AS (
                 UPDATE actors
                 SET followers = COALESCE(followers, ARRAY[]::follow_record[]) ||
-                                ARRAY[ROW($1, $2)::follow_record]
+                                ARRAY[ROW($1, $2)::follow_record]  -- (follower_actor_id, rkey)
                 WHERE id = $3
                   AND NOT EXISTS (
-                    SELECT 1 FROM unnest(followers) f WHERE (f).actor_id = $1 AND (f).rkey = $2
+                    SELECT 1 FROM unnest(followers) f WHERE (f).subject_actor_id = $1 AND (f).rkey = $2
                   )
                 RETURNING 1
             )
···
         .await?;
 
     // Return 1 if at least one array was updated (follow was created)
-    // Return 0 if neither was updated (duplicate follow)
+    // Return 0 if neither was updated (shouldn't happen since we checked already_exists)
     Ok(if rows > 0 { 1 } else { 0 })
 }
 
···
                 UPDATE actors
                 SET followers = ARRAY(
                     SELECT f FROM unnest(followers) AS f
-                    WHERE NOT ((f).actor_id = $2 AND (f).rkey = $1)
+                    WHERE NOT ((f).subject_actor_id = $2 AND (f).rkey = $1)
                 )
                 WHERE id = (SELECT subject_actor_id FROM follow_to_delete)
                   AND EXISTS (
                     SELECT 1 FROM unnest(followers) f
-                    WHERE (f).actor_id = $2 AND (f).rkey = $1
+                    WHERE (f).subject_actor_id = $2 AND (f).rkey = $1
                   )
                 RETURNING 1
             )
···
 
     let (list_did, _collection, list_rkey) = (parts[0], parts[1], parts[2]);
 
-    // Resolve list owner DID to actor_id
-    let list_actor_id: i32 = conn.query_one(
-        "INSERT INTO actors (did, handle, created_at)
-         VALUES ($1, NULL, NOW())
-         ON CONFLICT (did) DO UPDATE SET did = EXCLUDED.did
-         RETURNING id",
-        &[&list_did],
-    )
-    .await?
-    .get(0);
+    // Resolve list owner DID to actor_id (creates stub actor if needed)
+    let (list_actor_id, _, _) = super::feed::get_actor_id(conn, list_did).await?;
 
     // Acquire advisory lock on record to prevent deadlocks
     let (table_id, key_id) = crate::database_writer::locking::actor_record_lock("list_blocks", actor_id, rkey);
     crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?;
 
     // Append to list_blocks array with natural key deduplication
+    // list_block_record has fields: (list_actor_id, list_rkey, rkey, cid)
     // Returns number of rows updated (1 if inserted, 0 if duplicate)
     let rows = conn.execute(
         "UPDATE actors
···
          WHERE id = $1
            AND NOT EXISTS (
              SELECT 1 FROM unnest(list_blocks) lb
-             WHERE (lb).rkey = $2
+             WHERE (lb).rkey = $4
            )",
         &[
             &actor_id,
-            &rkey,
-            &cid_digest,
-            &list_actor_id,
-            &list_rkey,
+            &list_actor_id,  // $2: list_actor_id
+            &list_rkey,      // $3: list_rkey
+            &rkey,           // $4: rkey
+            &cid_digest,     // $5: cid
         ],
     )
     .await
···
     let (table_id, key_id) = crate::database_writer::locking::actor_record_lock("list_items", actor_id, rkey);
     crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?;
 
-    // Resolve list_id by creating stub list if needed
-    let list_id = super::feed::ensure_list_id(conn, &rec.list).await?;
+    // Resolve list natural keys by creating stub list if needed
+    let (list_owner_actor_id, list_rkey) = super::feed::ensure_list_natural_key(conn, &rec.list).await?;
 
     conn.execute(
         include_str!("../sql/list_item_upsert.sql"),
···
             &cid_digest,
             // Note: created_at (param $4) removed - derived from TID rkey
             &rkey,
-            &list_id, // Pass resolved list_id directly
+            &list_owner_actor_id, // Pass resolved list_owner_actor_id (may be NULL)
+            &list_rkey,           // Pass resolved list_rkey (may be NULL)
         ],
     )
     .await

consumer/src/db/record_exists/queries.rs

···
     Ok(row.is_some())
 }
 
-/// Check if a follow exists
+/// Check if a follow exists in the actor's following[] array
 pub async fn follow_exists<C: GenericClient>(conn: &C, did: &str, rkey: i64) -> QueryResult<bool> {
     let row = conn
         .query_opt(
-            "SELECT 1 FROM follows f
-             INNER JOIN actors a ON f.actor_id = a.id
-             WHERE a.did = $1 AND f.rkey = $2",
+            "SELECT 1 FROM actors a, unnest(COALESCE(a.following, ARRAY[]::follow_record[])) AS f
+             WHERE a.did = $1 AND (f).rkey = $2",
             &[&did, &rkey],
         )
         .await?;
     Ok(row.is_some())
 }
 
-/// Check if a block exists
+/// Check if a block exists in the actor's blocks[] array
 pub async fn block_exists<C: GenericClient>(conn: &C, did: &str, rkey: i64) -> QueryResult<bool> {
     let row = conn
         .query_opt(
-            "SELECT 1 FROM blocks b
-             INNER JOIN actors a ON b.actor_id = a.id
-             WHERE a.did = $1 AND b.rkey = $2",
+            "SELECT 1 FROM actors a, unnest(COALESCE(a.blocks, ARRAY[]::block_record[])) AS b
+             WHERE a.did = $1 AND (b).rkey = $2",
             &[&did, &rkey],
         )
         .await?;
···
 
 /// Check if a labeler exists and should skip fetching
 ///
+/// DENORMALIZED: Labelers are now stored directly on actors table (labeler_status, labeler_cid, etc)
+///
 /// Returns true if the labeler exists with any status except 'stub'.
 /// - stub: Return false (needs fetching)
 /// - missing: Return true (permanently unfetchable, skip)
···
 pub async fn labeler_exists<C: GenericClient>(conn: &C, did: &str) -> QueryResult<bool> {
     let row = conn
         .query_opt(
-            "SELECT 1 FROM labelers l
-             INNER JOIN actors a ON l.actor_id = a.id
-             WHERE a.did = $1 AND l.status != 'stub'::labeler_status",
+            "SELECT 1 FROM actors
+             WHERE did = $1 AND labeler_status IS NOT NULL AND labeler_status != 'stub'::labeler_status",
             &[&did],
         )
         .await?;
···
     Ok(row.is_some())
 }
 
-/// Check if a bookmark exists
+/// Check if a bookmark exists in the actor's bookmarks[] array
 pub async fn bookmark_exists<C: GenericClient>(
     conn: &C,
     did: &str,
···
 ) -> QueryResult<bool> {
     let row = conn
         .query_opt(
-            "SELECT 1 FROM bookmarks b
-             INNER JOIN actors a ON b.actor_id = a.id
-             WHERE a.did = $1 AND b.rkey = $2",
+            "SELECT 1 FROM actors a, unnest(COALESCE(a.bookmarks, ARRAY[]::bookmark_record[])) AS b
+             WHERE a.did = $1 AND (b).rkey = $2",
             &[&did, &rkey],
         )
         .await?;

consumer/src/db/sql/list_item_upsert.sql

···
--- Insert list_item with self-contained schema (no records table)
+-- Insert list_item with natural keys (no synthetic list_id)
 -- This prevents foreign key violations when indexing list items
--- Parameters: $1=actor_id(INT4), $2=subject_actor_id(i32), $3=cid(bytea), $4=rkey, $5=list_id(INT8)
+-- Parameters: $1=actor_id(INT4), $2=subject_actor_id(i32), $3=cid(bytea), $4=rkey,
+--             $5=list_owner_actor_id(INT4), $6=list_rkey(TEXT)
 -- Note: created_at is derived from TID rkey
 -- Note: actor_id is provided by caller after calling get_actor_id (ensures zero sequence waste)
-INSERT INTO list_items (actor_id, rkey, cid, list_id, subject_actor_id)
+INSERT INTO list_items (actor_id, rkey, cid, list_owner_actor_id, list_rkey, subject_actor_id)
 SELECT
     $1,  -- actor_id (provided by caller)
     $4,  -- rkey (INT8)
     $3,  -- cid (embedded, already bytea)
-    $5,  -- list_id (already resolved, may be stub)
+    $5,  -- list_owner_actor_id (already resolved, may be NULL)
+    $6,  -- list_rkey (already resolved, may be NULL)
     $2::int4  -- subject_actor_id (already resolved, cast to int4)
 WHERE $1::int4 IS NOT NULL  -- Only insert if owner exists
   AND $2::int4 IS NOT NULL  -- Only insert if subject exists
 ON CONFLICT (actor_id, rkey) DO UPDATE SET
     cid=EXCLUDED.cid,
-    list_id=EXCLUDED.list_id,
+    list_owner_actor_id=EXCLUDED.list_owner_actor_id,
+    list_rkey=EXCLUDED.list_rkey,
     subject_actor_id=EXCLUDED.subject_actor_id

consumer/src/workers/stub_resolution/queries.rs

···
 
 /// Find stub labelers and return their AT URIs for fetching
 ///
+/// DENORMALIZED: Labelers are now stored directly on actors table (labeler_status, labeler_cid, etc)
+///
 /// This query finds labelers with status='stub', constructs their AT URIs,
 /// and returns them for enqueuing to the fetch queue.
 /// Limits to 100 records per batch to avoid overwhelming the fetch queue.
 pub async fn find_stub_labelers<C: GenericClient>(conn: &C) -> QueryResult<Vec<String>> {
     let rows = conn
         .query(
-            "SELECT 'at://' || a.did || '/app.bsky.labeler.service/self' as uri
-             FROM labelers l
-             INNER JOIN actors a ON l.actor_id = a.id
-             WHERE l.status = 'stub'::labeler_status
-             ORDER BY l.created_at ASC
+            "SELECT 'at://' || did || '/app.bsky.labeler.service/self' as uri
+             FROM actors
+             WHERE labeler_status = 'stub'::labeler_status
+             ORDER BY labeler_created_at ASC
              LIMIT 100",
             &[],
         )

consumer/tests/graph_operations_test.rs

···
         .get(0);
     assert_eq!(target_count, 1, "Target actor should be created");
 
-    // Verify follow relationship exists
+    // Verify follow relationship exists in following[] array
     let follow_count: i64 = tx
         .query_one(
-            "SELECT COUNT(*) FROM follows f
-             INNER JOIN actors a ON f.actor_id = a.id
-             WHERE a.did = $1 AND f.rkey = tid_to_i64($2)",
+            "SELECT COUNT(*) FROM actors a, unnest(a.following) AS f
+             WHERE a.did = $1 AND (f).rkey = tid_to_i64($2)",
             &[&"did:plc:user123", &"3l7mkz4lmk245"],
         )
         .await
···
         "Should return deleted target DID"
     );
 
-    // Verify follow was deleted
+    // Verify follow was deleted from following[] array
     let follow_count: i64 = tx
         .query_one(
-            "SELECT COUNT(*) FROM follows f
-             INNER JOIN actors a ON f.actor_id = a.id
-             WHERE a.did = $1 AND f.rkey = tid_to_i64($2)",
+            "SELECT COUNT(*) FROM actors a, unnest(COALESCE(a.following, ARRAY[]::follow_record[])) AS f
+             WHERE a.did = $1 AND (f).rkey = tid_to_i64($2)",
             &[&"did:plc:user789", &"3l7mkz4lmk247"],
         )
         .await
···
         .get(0);
     assert!(blocked_exists, "Blocked actor should exist");
 
-    // Verify block relationship
+    // Verify block relationship in blocks[] array
     let block_count: i64 = tx
         .query_one(
-            "SELECT COUNT(*) FROM blocks b
-             INNER JOIN actors a ON b.actor_id = a.id
+            "SELECT COUNT(*) FROM actors a, unnest(a.blocks) AS b
              WHERE a.did = $1",
             &[&"did:plc:blocker123"],
         )
···
         .get(0);
     assert!(item_exists, "List item should exist");
 
-    // Verify list_id is set (not pending)
-    let list_id_is_set: bool = tx
+    // Verify list natural keys are set (not NULL)
+    let list_keys_are_set: bool = tx
         .query_one(
-            "SELECT li.list_id IS NOT NULL
+            "SELECT li.list_owner_actor_id IS NOT NULL AND li.list_rkey IS NOT NULL
              FROM list_items li
              INNER JOIN actors a ON li.actor_id = a.id
              WHERE a.did = $1 AND li.rkey = tid_to_i64($2)",
···
         .await
         .wrap_err("Failed to query list_items")?
         .get(0);
-    assert!(list_id_is_set, "List ID should be set (not pending)");
+    assert!(list_keys_are_set, "List natural keys should be set (not NULL)");
     Ok(())
 }
 
···
     );
     assert_eq!(result.wrap_err("Operation failed")?, 1, "Should insert 1 row");
 
-    // Verify list_id is set (not NULL) - stub list was created
-    let list_id_is_set: bool = tx
+    // Verify list natural keys are set (not NULL) - stub list was created
+    let list_keys_are_set: bool = tx
         .query_one(
-            "SELECT li.list_id IS NOT NULL
+            "SELECT li.list_owner_actor_id IS NOT NULL AND li.list_rkey IS NOT NULL
              FROM list_items li
              INNER JOIN actors a ON li.actor_id = a.id
              WHERE a.did = $1 AND li.rkey = tid_to_i64($2)",
···
         .await
         .wrap_err("Failed to query list_items")?
         .get(0);
-    assert!(list_id_is_set, "List ID should be set (stub list created)");
+    assert!(list_keys_are_set, "List natural keys should be set (stub list created)");
 
     // Verify the stub list was created with correct status
     let stub_list_row = tx
···
     );
     assert_eq!(result.wrap_err("Operation failed")?, 1, "Should insert 1 row");
 
-    // Verify list block exists
+    // Verify list block exists in list_blocks[] array
     let block_exists: bool = tx
         .query_one(
             "SELECT EXISTS(
-                SELECT 1 FROM list_blocks lb
-                INNER JOIN actors a ON lb.actor_id = a.id
-                WHERE a.did = $1 AND lb.rkey = tid_to_i64($2)
+                SELECT 1 FROM actors a, unnest(a.list_blocks) AS lb
+                WHERE a.did = $1 AND (lb).rkey = tid_to_i64($2)
             )",
             &[&"did:plc:blocker7", &"3l7mkz4lmk24m"],
         )
···
         .get(0);
     assert!(block_exists, "List block should exist");
 
-    // Verify list_id is set
-    let list_id_is_set: bool = tx
+    // Verify list natural keys are set
+    let list_keys_are_set: bool = tx
         .query_one(
-            "SELECT lb.list_id IS NOT NULL
-             FROM list_blocks lb
-             INNER JOIN actors a ON lb.actor_id = a.id
-             WHERE a.did = $1 AND lb.rkey = tid_to_i64($2)",
+            "SELECT (lb).list_actor_id IS NOT NULL AND (lb).list_rkey IS NOT NULL
+             FROM actors a, unnest(a.list_blocks) AS lb
+             WHERE a.did = $1 AND (lb).rkey = tid_to_i64($2)",
             &[&"did:plc:blocker7", &"3l7mkz4lmk24m"],
         )
         .await
         .wrap_err("Failed to query list_blocks")?
         .get(0);
-    assert!(list_id_is_set, "List ID should be set");
+    assert!(list_keys_are_set, "List natural keys should be set");
     Ok(())
 }
 
···
     );
     assert_eq!(result.unwrap(), 1, "Should delete 1 row");
 
-    // Verify list block was deleted
+    // Verify list block was deleted from list_blocks[] array
     let block_exists: bool = tx
         .query_one(
             "SELECT EXISTS(
-                SELECT 1 FROM list_blocks lb
-                INNER JOIN actors a ON lb.actor_id = a.id
-                WHERE a.did = $1 AND lb.rkey = tid_to_i64($2)
+                SELECT 1 FROM actors a, unnest(COALESCE(a.list_blocks, ARRAY[]::list_block_record[])) AS lb
+                WHERE a.did = $1 AND (lb).rkey = tid_to_i64($2)
             )",
             &[&"did:plc:blocker8", &"3l7mkz4lmk24o"],
         )

consumer/tests/notification_test.rs

···
         .wrap_err("Failed to get actor ID")?
         .get(0);
 
-    // Insert a thread mute using natural keys (root_post_actor_id, root_post_rkey)
-    // This is the only place we manually INSERT as there's no function for it
+    // Insert a thread mute into the actors.thread_mutes[] array
+    // thread_mute_record has fields: (root_post_actor_id, root_post_rkey, created_at)
     tx.execute(
-        "INSERT INTO thread_mutes (actor_id, root_post_actor_id, root_post_rkey, created_at) VALUES ($1, $2, $3, NOW())",
+        "UPDATE actors
+         SET thread_mutes = COALESCE(thread_mutes, ARRAY[]::thread_mute_record[]) ||
+                            ARRAY[ROW($2, $3, NOW())::thread_mute_record]
+         WHERE id = $1",
         &[&actor_id, &rootpostauthor_id, &rkey_i64],
     )
     .await