feat: denormalize strategy; mutes, blocks, bookmarks

migrations/2025-12-08-012930_denormalize_user_preferences/down.sql

···
+-- Rollback Phase 4: Restore mutes, blocks, and bookmarks tables
+
+-- Recreate mutes table
+CREATE TABLE mutes (
+  actor_id INTEGER NOT NULL,
+  subject_actor_id INTEGER NOT NULL,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  PRIMARY KEY (actor_id, subject_actor_id)
+);
+
+-- Recreate blocks table
+CREATE TABLE blocks (
+  actor_id INTEGER NOT NULL,
+  subject_actor_id INTEGER NOT NULL,
+  rkey BIGINT NOT NULL,
+  PRIMARY KEY (actor_id, rkey)
+);
+
+-- Recreate bookmarks table
+CREATE TABLE bookmarks (
+  actor_id INTEGER NOT NULL,
+  rkey BIGINT NOT NULL,
+  post_actor_id INTEGER NOT NULL,
+  post_rkey BIGINT NOT NULL,
+  PRIMARY KEY (actor_id, rkey)
+);
+
+-- Restore mutes from mutes array
+INSERT INTO mutes (actor_id, subject_actor_id, created_at)
+SELECT
+  a.id,
+  (m).subject_actor_id,
+  (m).created_at
+FROM actors a, unnest(a.mutes) AS m
+WHERE a.mutes IS NOT NULL;
+
+-- Restore blocks from blocks array
+INSERT INTO blocks (actor_id, subject_actor_id, rkey)
+SELECT
+  a.id,
+  (b).subject_actor_id,
+  (b).rkey
+FROM actors a, unnest(a.blocks) AS b
+WHERE a.blocks IS NOT NULL;
+
+-- Restore bookmarks from bookmarks array
+INSERT INTO bookmarks (actor_id, rkey, post_actor_id, post_rkey)
+SELECT
+  a.id,
+  (b).rkey,
+  (b).post_actor_id,
+  (b).post_rkey
+FROM actors a, unnest(a.bookmarks) AS b
+WHERE a.bookmarks IS NOT NULL;
+
+-- Drop user preference arrays from actors
+ALTER TABLE actors
+  DROP COLUMN bookmarks,
+  DROP COLUMN blocks,
+  DROP COLUMN mutes;

migrations/2025-12-08-012930_denormalize_user_preferences/up.sql

···
+-- Phase 4: Denormalize user preferences (mutes, blocks, bookmarks) into actors table
+--
+-- This migration moves mutes, blocks, and bookmarks from separate tables into arrays
+-- on the actors table, similar to how follows were denormalized in Phase 3.
+--
+-- Benefits:
+-- - Single row lookup for all user preferences
+-- - Eliminates JOINs when checking if user has muted/blocked/bookmarked
+-- - Better cache locality (all user data in one row)
+-- - Simpler queries for user preference checks
+
+-- Add user preference arrays to actors table
+ALTER TABLE actors
+  ADD COLUMN mutes mute_record[],
+  ADD COLUMN blocks block_record[],
+  ADD COLUMN bookmarks bookmark_record[];
+
+-- Backfill mutes array from mutes table
+UPDATE actors a
+SET mutes = (
+  SELECT ARRAY_AGG(
+    ROW(m.subject_actor_id, m.created_at)::mute_record
+    ORDER BY m.created_at DESC
+  )
+  FROM mutes m
+  WHERE m.actor_id = a.id
+);
+
+-- Backfill blocks array from blocks table
+UPDATE actors a
+SET blocks = (
+  SELECT ARRAY_AGG(
+    ROW(b.subject_actor_id, b.rkey)::block_record
+    ORDER BY b.rkey DESC
+  )
+  FROM blocks b
+  WHERE b.actor_id = a.id
+);
+
+-- Backfill bookmarks array from bookmarks table
+UPDATE actors a
+SET bookmarks = (
+  SELECT ARRAY_AGG(
+    ROW(b.post_actor_id, b.post_rkey, b.rkey)::bookmark_record
+    ORDER BY b.rkey DESC
+  )
+  FROM bookmarks b
+  WHERE b.actor_id = a.id
+);
+
+-- Drop old tables
+DROP TABLE mutes;
+DROP TABLE blocks;
+DROP TABLE bookmarks;

parakeet-db/src/composite_types.rs

···
 use crate::schema::sql_types::{
     PostExtEmbed, PostVideoEmbed, PostImageEmbed, PostFacetEmbed, PostVideoCaption,
     LabelerDefRecord, FollowRecord,
-    // Note: Other composite types will be added when used in Phase 4-7:
-    // MuteRecord, BlockRecord, BookmarkRecord, ThreadMuteRecord,
-    // ListMuteRecord, ListBlockRecord, PostLabel, ActorLabel,
+    MuteRecord, BlockRecord, BookmarkRecord,  // Phase 4: User preferences
+    // Note: Other composite types will be added when used in Phase 5-7:
+    // ThreadMuteRecord, ListMuteRecord, ListBlockRecord, PostLabel, ActorLabel,
 };
 
 // Placeholder SQL types for composite types not yet used in tables
···
 mod placeholder_sql_types {
     use diesel::query_builder::QueryId;
     use diesel::sql_types::SqlType;
-
-    #[derive(QueryId, SqlType)]
-    #[diesel(postgres_type(name = "mute_record"))]
-    pub struct MuteRecord;
-
-    #[derive(QueryId, SqlType)]
-    #[diesel(postgres_type(name = "block_record"))]
-    pub struct BlockRecord;
-
-    #[derive(QueryId, SqlType)]
-    #[diesel(postgres_type(name = "bookmark_record"))]
-    pub struct BookmarkRecord;
 
     #[derive(QueryId, SqlType)]
     #[diesel(postgres_type(name = "thread_mute_record"))]

parakeet-db/src/models.rs

···
 //
 // =============================================================================
 
-use crate::composite_types::{Follow, LabelerDef};
+use crate::composite_types::{Block, Bookmark, Follow, LabelerDef, Mute};
 use crate::tid_util::{decode_tid, encode_tid, TidError};
 use crate::types::*;
 use chrono::prelude::*;
···
     // Social graph arrays (from follows table, denormalized - bidirectional)
     pub following: Option<Vec<Option<Follow>>>,  // Who this actor follows
     pub followers: Option<Vec<Option<Follow>>>,  // Who follows this actor
+    // User preference arrays (from mutes, blocks, bookmarks tables, denormalized)
+    pub mutes: Option<Vec<Option<Mute>>>,        // Muted actors
+    pub blocks: Option<Vec<Option<Block>>>,      // Blocked actors
+    pub bookmarks: Option<Vec<Option<Bookmark>>>, // Bookmarked posts
 }
 
 // AllowlistEntry model removed - allowlist table dropped in favor of actors.sync_state
···
 // Follow model removed - follow relationships now stored as follow_record[] arrays on actors table
 // Follow composite type is defined in composite_types.rs
 
-#[derive(Clone, Debug, Queryable, Selectable, Identifiable)]
-#[diesel(table_name = crate::schema::blocks)]
-#[diesel(primary_key(actor_id, rkey))]
-#[diesel(check_for_backend(diesel::pg::Pg))]
-pub struct Block {
-    pub actor_id: i32, // PK: FK to actors (blocker)
-    pub rkey: i64,     // PK: TID as INT8
-    pub subject_actor_id: i32, // FK to actors (blocked)
-                       // Note: created_at derived from TID rkey via created_at() method
-                       // Note: CID is synthetic, generated from actor_id + rkey
-}
-
-// Off-protocol social actions (have created_at because not in repo)
-
-#[derive(Clone, Debug, Queryable, Selectable)]
-#[diesel(table_name = crate::schema::mutes)]
-#[diesel(primary_key(actor_id, subject_actor_id))]
-#[diesel(check_for_backend(diesel::pg::Pg))]
-pub struct Mute {
-    pub actor_id: i32,             // PK: FK to actors
-    pub subject_actor_id: i32,     // PK: FK to actors
-    pub created_at: DateTime<Utc>, // Off-protocol: has own timestamp
-}
-
-#[derive(Clone, Debug, Queryable, Selectable)]
-#[diesel(table_name = crate::schema::bookmarks)]
-#[diesel(primary_key(actor_id, rkey))]
-#[diesel(check_for_backend(diesel::pg::Pg))]
-pub struct Bookmark {
-    pub actor_id: i32,      // PK: FK to actors
-    pub rkey: i64,          // PK: TID as INT8
-    pub post_actor_id: i32, // FK to posts (actor_id)
-    pub post_rkey: i64,     // FK to posts (rkey)
-                            // Note: Bookmarks can only reference posts
-                            // Note: created_at derived from TID rkey via created_at() method
-}
+// Block, Mute, Bookmark models removed - user preferences now stored as arrays on actors table
+// Composite types are defined in composite_types.rs
 
 // Profile, NotifDecl, ChatDecl, and Status structs removed - data now consolidated into Actor struct
 // See actors table columns: profile_*, status_*, chat_*, notif_decl_*, notif_seen_at, notif_unread_count

parakeet-db/src/schema.rs

···
     pub struct ActorSyncState;
 
     #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)]
+    #[diesel(postgres_type(name = "block_record"))]
+    pub struct BlockRecord;
+
+    #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)]
+    #[diesel(postgres_type(name = "bookmark_record"))]
+    pub struct BookmarkRecord;
+
+    #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "caption_mime_type"))]
     pub struct CaptionMimeType;
 
···
     pub struct ImageMimeType;
 
     #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)]
+    #[diesel(postgres_type(name = "labeler_def_record"))]
+    pub struct LabelerDefRecord;
+
+    #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)]
+    #[diesel(postgres_type(name = "labeler_status"))]
+    pub struct LabelerStatus;
+
+    #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "label_blurs"))]
     pub struct LabelBlurs;
 
···
     pub struct LabelSeverity;
 
     #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)]
-    #[diesel(postgres_type(name = "labeler_def_record"))]
-    pub struct LabelerDefRecord;
-
-    #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)]
-    #[diesel(postgres_type(name = "labeler_status"))]
-    pub struct LabelerStatus;
-
-    #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "language_code"))]
     pub struct LanguageCode;
 
     #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "list_type"))]
     pub struct ListType;
+
+    #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)]
+    #[diesel(postgres_type(name = "mute_record"))]
+    pub struct MuteRecord;
 
     #[derive(diesel::query_builder::QueryId, diesel::sql_types::SqlType)]
     #[diesel(postgres_type(name = "notif_allow_subscriptions"))]
···
     use super::sql_types::LabelerStatus;
     use super::sql_types::LabelerDefRecord;
     use super::sql_types::FollowRecord;
+    use super::sql_types::MuteRecord;
+    use super::sql_types::BlockRecord;
+    use super::sql_types::BookmarkRecord;
 
     actors (id) {
         id -> Int4,
···
         labeler_defs -> Nullable<Array<Nullable<LabelerDefRecord>>>,
         following -> Nullable<Array<Nullable<FollowRecord>>>,
         followers -> Nullable<Array<Nullable<FollowRecord>>>,
+        mutes -> Nullable<Array<Nullable<MuteRecord>>>,
+        blocks -> Nullable<Array<Nullable<BlockRecord>>>,
+        bookmarks -> Nullable<Array<Nullable<BookmarkRecord>>>,
     }
 }
 
···
         scheduled_at -> Timestamptz,
         started_at -> Nullable<Timestamptz>,
         completed_at -> Nullable<Timestamptz>,
-    }
-}
-
-diesel::table! {
-    blocks (actor_id, rkey) {
-        actor_id -> Int4,
-        rkey -> Int8,
-        subject_actor_id -> Int4,
-    }
-}
-
-diesel::table! {
-    bookmarks (actor_id, rkey) {
-        actor_id -> Int4,
-        rkey -> Int8,
-        post_actor_id -> Int4,
-        post_rkey -> Int8,
     }
 }
 
···
         description_facets -> Nullable<Jsonb>,
         avatar_cid -> Nullable<Bytea>,
         status -> RecordStatus,
-    }
-}
-
-diesel::table! {
-    mutes (actor_id, subject_actor_id) {
-        actor_id -> Int4,
-        subject_actor_id -> Int4,
-        created_at -> Timestamptz,
     }
 }
 
···
     _diesel_schema_inference,
     actors,
     backfill_jobs,
-    blocks,
-    bookmarks,
     constellation_enrichment_queue,
     feedgen_likes,
     feedgens,
···
     list_items,
     list_mutes,
     lists,
-    mutes,
     notifications,
     post_aggregate_stats,
     posts,

parakeet/src/db/bookmarks.rs

···
 /// Get bookmarks for a user with cursor pagination
 ///
 /// Returns list of (created_at, subject_uri, cid_str) tuples
+///
+/// OPTIMIZED: Reads from denormalized bookmarks array (single row lookup!)
 pub async fn get_user_bookmarks(
     conn: &mut AsyncPgConnection,
     actor_id: i32,
···
         cid: Vec<u8>,
     }
 
-    // Use .bind() for cursor parameter to prevent SQL injection
     diesel::sql_query(
-        "SELECT tid_timestamp(b.rkey) as created_at,
+        "SELECT tid_timestamp((b).rkey) as created_at,
                 a.did,
                 p.rkey,
                 p.cid
-         FROM bookmarks b
-         INNER JOIN posts p ON b.post_actor_id = p.actor_id AND b.post_rkey = p.rkey
+         FROM actors, unnest(bookmarks) AS b
+         INNER JOIN posts p ON (b).post_actor_id = p.actor_id AND (b).post_rkey = p.rkey
          INNER JOIN actors a ON p.actor_id = a.id
-         WHERE b.actor_id = $1
+         WHERE actors.id = $1
            AND p.status = 'complete'
-           AND ($2::timestamptz IS NULL OR tid_timestamp(b.rkey) < $2)
-         ORDER BY b.rkey DESC
+           AND ($2::timestamptz IS NULL OR tid_timestamp((b).rkey) < $2)
+         ORDER BY (b).rkey DESC
          LIMIT $3"
     )
     .bind::<Integer, _>(actor_id)

parakeet/src/db/graph.rs

···
 /// Get muted accounts for a user with cursor pagination
 ///
 /// Returns list of (created_at, subject_did) tuples
+///
+/// OPTIMIZED: Reads from denormalized mutes array (single row lookup!)
 pub async fn get_user_mutes(
     conn: &mut AsyncPgConnection,
     actor_id: i32,
···
         subject_did: String,
     }
 
-    // Use .bind() for cursor parameter to prevent SQL injection
     diesel::sql_query(
-        "SELECT m.created_at, subject.did as subject_did
-         FROM mutes m
-         INNER JOIN actors subject ON m.subject_actor_id = subject.id
-         WHERE m.actor_id = $1
-           AND ($2::timestamptz IS NULL OR m.created_at < $2)
-         ORDER BY m.created_at DESC
+        "SELECT (m).created_at, subject.did as subject_did
+         FROM actors, unnest(mutes) AS m
+         INNER JOIN actors subject ON (m).subject_actor_id = subject.id
+         WHERE actors.id = $1
+           AND ($2::timestamptz IS NULL OR (m).created_at < $2)
+         ORDER BY (m).created_at DESC
          LIMIT $3"
     )
     .bind::<Integer, _>(actor_id)
···
 /// Get blocked accounts for a user with cursor pagination
 ///
 /// Returns list of (created_at, subject_actor_id) tuples
+///
+/// OPTIMIZED: Reads from denormalized blocks array (single row lookup!)
 pub async fn get_user_blocks(
     conn: &mut AsyncPgConnection,
     actor_id: i32,
···
     }
 
     diesel::sql_query(
-        "SELECT DISTINCT ON (b.subject_actor_id) tid_timestamp(b.rkey) as created_at, b.subject_actor_id
-         FROM blocks b
-         WHERE b.actor_id = $1
-           AND ($2::timestamptz IS NULL OR tid_timestamp(b.rkey) < $2)
-         ORDER BY b.subject_actor_id, b.rkey DESC
+        "SELECT tid_timestamp((b).rkey) as created_at, (b).subject_actor_id
+         FROM actors, unnest(blocks) AS b
+         WHERE id = $1
+           AND ($2::timestamptz IS NULL OR tid_timestamp((b).rkey) < $2)
+         ORDER BY (b).rkey DESC
          LIMIT $3"
     )
     .bind::<Integer, _>(actor_id)

parakeet/src/xrpc/app_bsky/bookmark.rs

···
     // Generate rkey (TID) for the bookmark from current timestamp
     let rkey = chrono::Utc::now().timestamp_micros();
 
-    let _ = diesel_async::RunQueryDsl::execute(
-        diesel::insert_into(schema::bookmarks::table) // TODO: Consider communicating back to consumer worker using PostgreSQL queue
-            .values((
-                schema::bookmarks::actor_id.eq(actor_id),
-                schema::bookmarks::rkey.eq(rkey),
-                schema::bookmarks::post_actor_id.eq(post_actor_id),
-                schema::bookmarks::post_rkey.eq(post_rkey),
-            ))
-            .on_conflict((schema::bookmarks::actor_id, schema::bookmarks::post_actor_id, schema::bookmarks::post_rkey))
-            .do_nothing(),
+    // Append to bookmarks array (off-protocol, managed directly by AppView)
+    // Deduplicates based on post_actor_id + post_rkey
+    diesel_async::RunQueryDsl::execute(
+        diesel::sql_query(
+            "UPDATE actors
+             SET bookmarks = COALESCE(bookmarks, ARRAY[]::bookmark_record[]) ||
+                             ARRAY[ROW($2, $3, $4)::bookmark_record]
+             WHERE id = $1
+               AND NOT EXISTS (
+                 SELECT 1 FROM unnest(bookmarks) b
+                 WHERE (b).post_actor_id = $2 AND (b).post_rkey = $3
+               )"
+        )
+        .bind::<diesel::sql_types::Integer, _>(actor_id)
+        .bind::<diesel::sql_types::Integer, _>(post_actor_id)
+        .bind::<diesel::sql_types::BigInt, _>(post_rkey)
+        .bind::<diesel::sql_types::BigInt, _>(rkey),
         &mut conn,
     )
     .await?;
···
     )
     .await?;
 
-    let _ = diesel_async::RunQueryDsl::execute(
-        diesel::delete(schema::bookmarks::table).filter( // TODO: Consider communicating back to consumer worker using PostgreSQL queue
-            schema::bookmarks::actor_id
-                .eq(actor_id)
-                .and(schema::bookmarks::post_actor_id.eq(post_actor_id))
-                .and(schema::bookmarks::post_rkey.eq(post_rkey)),
-        ),
+    // Remove from bookmarks array (off-protocol, managed directly by AppView)
+    diesel_async::RunQueryDsl::execute(
+        diesel::sql_query(
+            "UPDATE actors
+             SET bookmarks = ARRAY(
+               SELECT b FROM unnest(bookmarks) AS b
+               WHERE NOT ((b).post_actor_id = $2 AND (b).post_rkey = $3)
+             )
+             WHERE id = $1"
+        )
+        .bind::<diesel::sql_types::Integer, _>(actor_id)
+        .bind::<diesel::sql_types::Integer, _>(post_actor_id)
+        .bind::<diesel::sql_types::BigInt, _>(post_rkey),
         &mut conn,
     )
     .await?;

parakeet/src/xrpc/app_bsky/graph/mutes.rs

···
     )
     .await?;
 
-    let _ = diesel_async::RunQueryDsl::execute(
-        diesel::insert_into(schema::mutes::table) // TODO: Consider communicating back to consumer worker using PostgreSQL queue
-            .values((
-                schema::mutes::actor_id.eq(actor_id),
-                schema::mutes::subject_actor_id.eq(subject_actor_id),
-                schema::mutes::created_at.eq(chrono::Utc::now()),
-            ))
-            .on_conflict((schema::mutes::actor_id, schema::mutes::subject_actor_id))
-            .do_nothing(),
+    // Append to mutes array (off-protocol, managed directly by AppView)
+    // Deduplicates based on subject_actor_id
+    let created_at = chrono::Utc::now();
+    diesel_async::RunQueryDsl::execute(
+        diesel::sql_query(
+            "UPDATE actors
+             SET mutes = COALESCE(mutes, ARRAY[]::mute_record[]) ||
+                         ARRAY[ROW($2, $3)::mute_record]
+             WHERE id = $1
+               AND NOT EXISTS (
+                 SELECT 1 FROM unnest(mutes) m
+                 WHERE (m).subject_actor_id = $2
+               )"
+        )
+        .bind::<diesel::sql_types::Integer, _>(actor_id)
+        .bind::<diesel::sql_types::Integer, _>(subject_actor_id)
+        .bind::<diesel::sql_types::Timestamptz, _>(created_at),
         &mut conn,
     )
     .await?;
···
     )
     .await?;
 
-    let _ = diesel_async::RunQueryDsl::execute(
-        diesel::delete(schema::mutes::table).filter( // TODO: Consider communicating back to consumer worker using PostgreSQL queue
-            schema::mutes::actor_id
-                .eq(actor_id)
-                .and(schema::mutes::subject_actor_id.eq(subject_actor_id)),
-        ),
+    // Remove from mutes array (off-protocol, managed directly by AppView)
+    diesel_async::RunQueryDsl::execute(
+        diesel::sql_query(
+            "UPDATE actors
+             SET mutes = ARRAY(
+               SELECT m FROM unnest(mutes) AS m
+               WHERE NOT ((m).subject_actor_id = $2)
+             )
+             WHERE id = $1"
+        )
+        .bind::<diesel::sql_types::Integer, _>(actor_id)
+        .bind::<diesel::sql_types::Integer, _>(subject_actor_id),
         &mut conn,
     )
     .await?;

parakeet/src/xrpc/community_lexicon/bookmarks.rs

···
 use axum::Json;
 use diesel::prelude::*;
 use lexica::community_lexicon::bookmarks::Bookmark;
-use parakeet_db::{models, schema};
+use parakeet_db::schema;
 use serde::{Deserialize, Serialize};
 
 #[derive(Debug, Deserialize)]
···
     )
     .await?;
 
-    let mut bookmarks_query = schema::bookmarks::table
-        .select(models::Bookmark::as_select())
-        .filter(schema::bookmarks::actor_id.eq(actor_id))
-        .into_boxed();
+    // Note: tags filtering not supported in current schema
+    if query.tags.is_some() {
+        tracing::warn!("tags filtering requested but not supported in current schema");
+    }
+
+    // Read from bookmarks array
+    use diesel::sql_types::{BigInt, Integer, Nullable};
 
-    if let Some(cursor) = datetime_cursor(query.cursor.as_ref()) {
-        // Convert timestamp cursor to TID for rkey-based pagination
-        let cursor_micros = cursor.timestamp_micros();
-        let cursor_tid = cursor_micros << 10; // Shift left 10 bits to create approximate TID
-        bookmarks_query = bookmarks_query.filter(schema::bookmarks::rkey.lt(cursor_tid));
+    #[derive(diesel::QueryableByName)]
+    struct BookmarkRecord {
+        #[diesel(sql_type = Integer)]
+        post_actor_id: i32,
+        #[diesel(sql_type = BigInt)]
+        post_rkey: i64,
+        #[diesel(sql_type = BigInt)]
+        rkey: i64,
     }
 
-    // Note: tags filtering not supported in current schema
-    if query.tags.is_some() {
-        tracing::warn!("tags filtering requested but not supported in current schema");
+    impl BookmarkRecord {
+        fn created_at(&self) -> chrono::DateTime<chrono::Utc> {
+            parakeet_db::tid_util::tid_to_datetime(self.rkey)
+        }
     }
 
+    let cursor_tid = query.cursor.as_ref()
+        .and_then(|c| datetime_cursor(Some(c)))
+        .map(|dt| dt.timestamp_micros() << 10);
+
     let results = diesel_async::RunQueryDsl::load(
-        bookmarks_query
-            .order(schema::bookmarks::rkey.desc())
-            .limit(i64::from(limit)),
+        diesel::sql_query(
+            "SELECT (b).post_actor_id, (b).post_rkey, (b).rkey
+             FROM actors, unnest(bookmarks) AS b
+             WHERE id = $1
+               AND ($2::bigint IS NULL OR (b).rkey < $2)
+             ORDER BY (b).rkey DESC
+             LIMIT $3"
+        )
+        .bind::<Integer, _>(actor_id)
+        .bind::<Nullable<BigInt>, _>(cursor_tid)
+        .bind::<BigInt, _>(i64::from(limit)),
         &mut conn,
     )
     .await?;
 
     let cursor = results
         .last()
-        .map(|bm| bm.created_at().timestamp_millis().to_string());
+        .map(|bm: &BookmarkRecord| bm.created_at().timestamp_millis().to_string());
 
     // Reconstruct post AT URIs from posts table using natural keys
     // Note: Bookmarks can only reference posts (not feedgens, lists, etc.)