From 31651cec165477b75e433ce7366bd39a1983049e Mon Sep 17 00:00:00 2001 From: Timothy Quilling Date: Sun, 7 Dec 2025 20:52:35 -0500 Subject: [PATCH] feat: denormalize strategy; mutes, blocks, bookmarks --- .../down.sql | 60 ++++++++++++++++++ .../up.sql | 54 ++++++++++++++++ parakeet-db/src/composite_types.rs | 18 +----- parakeet-db/src/models.rs | 44 +++---------- parakeet-db/src/schema.rs | 62 ++++++++----------- parakeet/src/db/bookmarks.rs | 15 ++--- parakeet/src/db/graph.rs | 27 ++++---- parakeet/src/xrpc/app_bsky/bookmark.rs | 47 +++++++++----- parakeet/src/xrpc/app_bsky/graph/mutes.rs | 44 ++++++++----- .../src/xrpc/community_lexicon/bookmarks.rs | 53 +++++++++++----- 10 files changed, 268 insertions(+), 156 deletions(-) create mode 100644 migrations/2025-12-08-012930_denormalize_user_preferences/down.sql create mode 100644 migrations/2025-12-08-012930_denormalize_user_preferences/up.sql diff --git a/migrations/2025-12-08-012930_denormalize_user_preferences/down.sql b/migrations/2025-12-08-012930_denormalize_user_preferences/down.sql new file mode 100644 index 00000000..faf4c661 --- /dev/null +++ b/migrations/2025-12-08-012930_denormalize_user_preferences/down.sql @@ -0,0 +1,60 @@ +-- 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; diff --git a/migrations/2025-12-08-012930_denormalize_user_preferences/up.sql b/migrations/2025-12-08-012930_denormalize_user_preferences/up.sql new file mode 100644 index 00000000..97a75bc6 --- /dev/null +++ b/migrations/2025-12-08-012930_denormalize_user_preferences/up.sql @@ -0,0 +1,54 @@ +-- 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; diff --git a/parakeet-db/src/composite_types.rs b/parakeet-db/src/composite_types.rs index 75c92757..51b41d26 100644 --- a/parakeet-db/src/composite_types.rs +++ b/parakeet-db/src/composite_types.rs @@ -26,9 +26,9 @@ 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 @@ -38,18 +38,6 @@ 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"))] pub struct ThreadMuteRecord; diff --git a/parakeet-db/src/models.rs b/parakeet-db/src/models.rs index 73ab13b9..2c2584a4 100644 --- a/parakeet-db/src/models.rs +++ b/parakeet-db/src/models.rs @@ -34,7 +34,7 @@ // // ============================================================================= -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::*; @@ -164,6 +164,10 @@ pub struct Actor { // Social graph arrays (from follows table, denormalized - bidirectional) pub following: Option>>, // Who this actor follows pub followers: Option>>, // Who follows this actor + // User preference arrays (from mutes, blocks, bookmarks tables, denormalized) + pub mutes: Option>>, // Muted actors + pub blocks: Option>>, // Blocked actors + pub bookmarks: Option>>, // Bookmarked posts } // AllowlistEntry model removed - allowlist table dropped in favor of actors.sync_state @@ -450,42 +454,8 @@ pub struct Repost { // 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, // 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 diff --git a/parakeet-db/src/schema.rs b/parakeet-db/src/schema.rs index 7711d892..641bc2c8 100644 --- a/parakeet-db/src/schema.rs +++ b/parakeet-db/src/schema.rs @@ -9,6 +9,14 @@ pub mod sql_types { #[diesel(postgres_type(name = "actor_sync_state"))] 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; @@ -41,6 +49,14 @@ pub mod sql_types { #[diesel(postgres_type(name = "image_mime_type"))] 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; @@ -53,14 +69,6 @@ pub mod sql_types { #[diesel(postgres_type(name = "label_severity"))] 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; @@ -69,6 +77,10 @@ pub mod sql_types { #[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"))] pub struct NotifAllowSubscriptions; @@ -176,6 +188,9 @@ diesel::table! { 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, @@ -228,6 +243,9 @@ diesel::table! { labeler_defs -> Nullable>>, following -> Nullable>>, followers -> Nullable>>, + mutes -> Nullable>>, + blocks -> Nullable>>, + bookmarks -> Nullable>>, } } @@ -244,23 +262,6 @@ diesel::table! { } } -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, - } -} - diesel::table! { constellation_enrichment_queue (id) { id -> Int8, @@ -403,14 +404,6 @@ diesel::table! { } } -diesel::table! { - mutes (actor_id, subject_actor_id) { - actor_id -> Int4, - subject_actor_id -> Int4, - created_at -> Timestamptz, - } -} - diesel::table! { use diesel::sql_types::*; use super::sql_types::NotificationReason; @@ -607,8 +600,6 @@ diesel::allow_tables_to_appear_in_same_query!( _diesel_schema_inference, actors, backfill_jobs, - blocks, - bookmarks, constellation_enrichment_queue, feedgen_likes, feedgens, @@ -620,7 +611,6 @@ diesel::allow_tables_to_appear_in_same_query!( list_items, list_mutes, lists, - mutes, notifications, post_aggregate_stats, posts, diff --git a/parakeet/src/db/bookmarks.rs b/parakeet/src/db/bookmarks.rs index 2acb97fb..8a958eda 100644 --- a/parakeet/src/db/bookmarks.rs +++ b/parakeet/src/db/bookmarks.rs @@ -8,6 +8,8 @@ use diesel_async::{AsyncPgConnection, RunQueryDsl}; /// 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, @@ -28,19 +30,18 @@ pub async fn get_user_bookmarks( cid: Vec, } - // 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::(actor_id) diff --git a/parakeet/src/db/graph.rs b/parakeet/src/db/graph.rs index 87db750d..dd611a26 100644 --- a/parakeet/src/db/graph.rs +++ b/parakeet/src/db/graph.rs @@ -7,6 +7,8 @@ use diesel_async::{AsyncPgConnection, RunQueryDsl}; /// 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, @@ -21,14 +23,13 @@ pub async fn get_user_mutes( 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::(actor_id) @@ -72,6 +73,8 @@ pub async fn get_list_id_by_uri( /// 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, @@ -87,11 +90,11 @@ pub async fn get_user_blocks( } 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::(actor_id) diff --git a/parakeet/src/xrpc/app_bsky/bookmark.rs b/parakeet/src/xrpc/app_bsky/bookmark.rs index 47a26d27..56eadd90 100644 --- a/parakeet/src/xrpc/app_bsky/bookmark.rs +++ b/parakeet/src/xrpc/app_bsky/bookmark.rs @@ -69,16 +69,23 @@ pub async fn create_bookmark( // 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::(actor_id) + .bind::(post_actor_id) + .bind::(post_rkey) + .bind::(rkey), &mut conn, ) .await?; @@ -137,13 +144,19 @@ pub async fn delete_bookmark( ) .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::(actor_id) + .bind::(post_actor_id) + .bind::(post_rkey), &mut conn, ) .await?; diff --git a/parakeet/src/xrpc/app_bsky/graph/mutes.rs b/parakeet/src/xrpc/app_bsky/graph/mutes.rs index 0fe8735d..205d096a 100644 --- a/parakeet/src/xrpc/app_bsky/graph/mutes.rs +++ b/parakeet/src/xrpc/app_bsky/graph/mutes.rs @@ -96,15 +96,23 @@ pub async fn mute_actor( ) .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::(actor_id) + .bind::(subject_actor_id) + .bind::(created_at), &mut conn, ) .await?; @@ -190,12 +198,18 @@ pub async fn unmute_actor( ) .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::(actor_id) + .bind::(subject_actor_id), &mut conn, ) .await?; diff --git a/parakeet/src/xrpc/community_lexicon/bookmarks.rs b/parakeet/src/xrpc/community_lexicon/bookmarks.rs index b02ebe09..eccf2667 100644 --- a/parakeet/src/xrpc/community_lexicon/bookmarks.rs +++ b/parakeet/src/xrpc/community_lexicon/bookmarks.rs @@ -6,7 +6,7 @@ use axum::extract::{Query, State}; 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)] @@ -41,34 +41,53 @@ pub async fn get_actor_bookmarks( ) .await?; - let mut bookmarks_query = schema::bookmarks::table - .select(models::Bookmark::as_select()) - .filter(schema::bookmarks::actor_id.eq(actor_id)) - .into_boxed(); - - 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)); - } - // 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}; + + #[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, + } + + impl BookmarkRecord { + fn created_at(&self) -> chrono::DateTime { + 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::(actor_id) + .bind::, _>(cursor_tid) + .bind::(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.) -- 2.51.2