Merge branch 'feat-bookmarks' into 'main'

Bookmarks

See merge request parakeet-social/parakeet!21

consumer/src/db/record.rs

···
 use chrono::prelude::*;
 use deadpool_postgres::GenericClient;
 use ipld_core::cid::Cid;
+use lexica::community_lexicon::bookmarks::Bookmark;
 
 pub async fn record_upsert<C: GenericClient>(
     conn: &mut C,
···
 pub async fn record_delete<C: GenericClient>(conn: &mut C, at_uri: &str) -> PgExecResult {
     conn.execute("DELETE FROM records WHERE at_uri=$1", &[&at_uri])
         .await
+}
+
+pub async fn bookmark_upsert<C: GenericClient>(
+    conn: &mut C,
+    rkey: &str,
+    repo: &str,
+    rec: Bookmark,
+) -> PgExecResult {
+    // strip "at://" then break into parts by '/'
+    let rec_type = match rec.subject.strip_prefix("at://") {
+        Some(at_uri) => at_uri.split('/').collect::<Vec<_>>()[1],
+        None => "$uri",
+    };
+
+    conn.execute(
+        include_str!("sql/bookmarks_upsert.sql"),
+        &[&repo, &rkey, &rec.subject, &rec_type, &rec.tags, &rec.created_at],
+    )
+    .await
+}
+
+pub async fn bookmark_delete<C: GenericClient>(
+    conn: &mut C,
+    rkey: &str,
+    repo: &str,
+) -> PgExecResult {
+    conn.execute(
+        "DELETE FROM bookmarks WHERE rkey=$1 AND did=$2",
+        &[&rkey, &repo],
+    )
+    .await
 }
 
 pub async fn block_insert<C: GenericClient>(

consumer/src/db/sql/bookmarks_upsert.sql

···
+INSERT INTO bookmarks (did, rkey, subject, subject_type, tags, created_at)
+VALUES ($1, $2, $3, $4, $5, $6)
+ON CONFLICT (did, rkey) DO UPDATE SET subject=EXCLUDED.subject,
+                                              subject_type=EXCLUDED.subject_type,
+                                              tags=EXCLUDED.tags

consumer/src/indexer/mod.rs

···
                 redis::AsyncTypedCommands::del(rc, format!("profile#{repo}")).await?;
             }
         }
+        RecordTypes::CommunityLexiconBookmark(record) => {
+            db::bookmark_upsert(conn, rkey, repo, record).await?;
+        }
     }
 
     db::record_upsert(conn, at_uri, repo, cid).await?;
···
         CollectionType::ChatActorDecl => {
             redis::AsyncTypedCommands::del(rc, format!("profile#{repo}")).await?;
             db::chat_decl_delete(conn, repo).await?
+        }
+        CollectionType::CommunityLexiconBookmark => {
+            db::bookmark_delete(conn, rkey, repo).await?
         }
         _ => unreachable!(),
     };

consumer/src/indexer/types.rs

···
     AppBskyNotificationDeclaration(records::AppBskyNotificationDeclaration),
     #[serde(rename = "chat.bsky.actor.declaration")]
     ChatBskyActorDeclaration(records::ChatBskyActorDeclaration),
+    #[serde(rename = "community.lexicon.bookmarks.bookmark")]
+    CommunityLexiconBookmark(lexica::community_lexicon::bookmarks::Bookmark)
 }
 
 #[derive(Debug, PartialOrd, PartialEq, Deserialize, Serialize)]
···
     BskyLabelerService,
     BskyNotificationDeclaration,
     ChatActorDecl,
+    CommunityLexiconBookmark,
     Unsupported,
 }
 
···
             "app.bsky.labeler.service" => CollectionType::BskyLabelerService,
             "app.bsky.notification.declaration" => CollectionType::BskyNotificationDeclaration,
             "chat.bsky.actor.declaration" => CollectionType::ChatActorDecl,
+            "community.lexicon.bookmarks.bookmark" => CollectionType::CommunityLexiconBookmark,
             _ => CollectionType::Unsupported,
         }
     }
···
             CollectionType::BskyVerification => false,
             CollectionType::BskyLabelerService => true,
             CollectionType::BskyNotificationDeclaration => true,
+            CollectionType::CommunityLexiconBookmark => true,
             CollectionType::Unsupported => false,
         }
     }

lexica/src/app_bsky/bookmark.rs

···
+use crate::app_bsky::feed::{BlockedAuthor, PostView};
+use crate::StrongRef;
+use chrono::prelude::*;
+use serde::Serialize;
+
+#[derive(Clone, Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct BookmarkView {
+    pub subject: StrongRef,
+    pub item: BookmarkViewItem,
+    pub created_at: DateTime<Utc>,
+}
+
+#[derive(Clone, Debug, Serialize)]
+#[serde(tag = "$type")]
+// This is technically the same as ReplyRefPost atm, but just in case...
+pub enum BookmarkViewItem {
+    #[serde(rename = "app.bsky.feed.defs#postView")]
+    Post(PostView),
+    #[serde(rename = "app.bsky.feed.defs#notFoundPost")]
+    NotFound {
+        uri: String,
+        #[serde(rename = "notFound")]
+        not_found: bool,
+    },
+    #[serde(rename = "app.bsky.feed.defs#blockedPost")]
+    Blocked {
+        uri: String,
+        blocked: bool,
+        author: BlockedAuthor,
+    },
+}

lexica/src/app_bsky/mod.rs

···
 use serde::Serialize;
 
 pub mod actor;
+pub mod bookmark;
 pub mod embed;
 pub mod feed;
 pub mod graph;

lexica/src/community_lexicon/bookmarks.rs

···
+use chrono::prelude::*;
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(tag = "$type")]
+#[serde(rename = "community.lexicon.bookmarks.bookmark")]
+#[serde(rename_all = "camelCase")]
+pub struct Bookmark {
+    pub subject: String,
+    #[serde(default)]
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    pub tags: Vec<String>,
+    pub created_at: DateTime<Utc>,
+}

lexica/src/community_lexicon/mod.rs

···
+pub mod bookmarks;

lexica/src/lib.rs

···
 
 pub mod app_bsky;
 pub mod com_atproto;
+pub mod community_lexicon;
 mod utils;
 
 #[derive(Clone, Debug, Serialize)]
···
     )]
     pub cid: Cid,
     pub uri: String,
+}
+
+impl StrongRef {
+    pub fn new_from_str(uri: String, cid: &str) -> Result<Self, cid::Error> {
+        let cid = cid.parse()?;
+        Ok(StrongRef { uri, cid })
+    }
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]

migrations/2025-09-02-190833_bookmarks/down.sql

···
+drop table bookmarks;

migrations/2025-09-02-190833_bookmarks/up.sql

···
+create table bookmarks
+(
+    did          text        not null references actors (did),
+    rkey         text,
+    subject      text        not null,
+    subject_cid  text,
+    subject_type text        not null,
+    tags         text[]      not null default ARRAY []::text[],
+
+    created_at   timestamptz not null default now(),
+
+    primary key (did, subject)
+);
+
+create index bookmarks_rkey_index on bookmarks (rkey);
+create index bookmarks_subject_index on bookmarks (subject);
+create index bookmarks_subject_type_index on bookmarks (subject_type);
+create index bookmarks_tags_index on bookmarks using gin (tags);
+create unique index bookmarks_rkey_ui on bookmarks (did, rkey);

parakeet-db/src/models.rs

···
     pub did: &'a str,
     pub list_uri: &'a str,
 }
+
+#[derive(Clone, Debug, Serialize, Deserialize, Queryable, Selectable, Identifiable)]
+#[diesel(table_name = crate::schema::bookmarks)]
+#[diesel(primary_key(did, subject, subject_cid))]
+#[diesel(check_for_backend(diesel::pg::Pg))]
+pub struct Bookmark {
+    pub did: String,
+    pub rkey: Option<String>,
+    pub subject: String,
+    pub subject_cid: Option<String>,
+    pub subject_type: String,
+    pub tags: Vec<Option<String>>,
+    pub created_at: DateTime<Utc>,
+}
+
+#[derive(Debug, Insertable, AsChangeset)]
+#[diesel(table_name = crate::schema::bookmarks)]
+#[diesel(check_for_backend(diesel::pg::Pg))]
+pub struct NewBookmark<'a> {
+    pub did: &'a str,
+    pub rkey: Option<String>,
+    pub subject: &'a str,
+    pub subject_cid: Option<String>,
+    pub subject_type: &'a str,
+    pub tags: Vec<String>,
+}

parakeet-db/src/schema.rs

···
 }
 
 diesel::table! {
+    bookmarks (did, subject) {
+        did -> Text,
+        rkey -> Nullable<Text>,
+        subject -> Text,
+        subject_cid -> Nullable<Text>,
+        subject_type -> Text,
+        tags -> Array<Nullable<Text>>,
+        created_at -> Timestamptz,
+    }
+}
+
+diesel::table! {
     chat_decls (did) {
         did -> Text,
         allow_incoming -> Text,
···
 
 diesel::joinable!(backfill -> actors (repo));
 diesel::joinable!(blocks -> actors (did));
+diesel::joinable!(bookmarks -> actors (did));
 diesel::joinable!(chat_decls -> actors (did));
 diesel::joinable!(feedgens -> actors (owner));
 diesel::joinable!(follows -> actors (did));
···
     backfill,
     backfill_jobs,
     blocks,
+    bookmarks,
     chat_decls,
     feedgens,
     follows,

parakeet/src/xrpc/app_bsky/bookmark.rs

···
+use crate::hydration::StatefulHydrator;
+use crate::xrpc::error::XrpcResult;
+use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth};
+use crate::xrpc::{datetime_cursor, CursorQuery};
+use crate::GlobalState;
+use axum::extract::{Query, State};
+use axum::Json;
+use diesel::prelude::*;
+use diesel_async::RunQueryDsl;
+use lexica::app_bsky::bookmark::{BookmarkView, BookmarkViewItem};
+use parakeet_db::{models, schema};
+use serde::{Deserialize, Serialize};
+use lexica::StrongRef;
+
+const BSKY_ALLOWED_TYPES: &[&str] = &["app.bsky.feed.post"];
+
+#[derive(Debug, Deserialize)]
+pub struct CreateBookmarkReq {
+    pub uri: String,
+    pub cid: String,
+}
+
+pub async fn create_bookmark(
+    State(state): State<GlobalState>,
+    auth: AtpAuth,
+    Json(form): Json<CreateBookmarkReq>,
+) -> XrpcResult<()> {
+    let mut conn = state.pool.get().await?;
+
+    // strip "at://" then break into parts by '/'
+    let parts = form.uri[5..].split('/').collect::<Vec<_>>();
+
+    let data = models::NewBookmark {
+        did: &auth.0,
+        rkey: None,
+        subject: &form.uri,
+        subject_cid: Some(form.cid),
+        subject_type: &parts[1],
+        tags: vec![],
+    };
+
+    diesel::insert_into(schema::bookmarks::table)
+        .values(&data)
+        .on_conflict_do_nothing()
+        .execute(&mut conn)
+        .await?;
+
+    Ok(())
+}
+
+#[derive(Debug, Deserialize)]
+pub struct DeleteBookmarkReq {
+    pub uri: String,
+}
+
+pub async fn delete_bookmark(
+    State(state): State<GlobalState>,
+    auth: AtpAuth,
+    Json(form): Json<DeleteBookmarkReq>,
+) -> XrpcResult<()> {
+    let mut conn = state.pool.get().await?;
+
+    diesel::delete(schema::bookmarks::table)
+        .filter(
+            schema::bookmarks::did
+                .eq(&auth.0)
+                .and(schema::bookmarks::subject.eq(&form.uri)),
+        )
+        .execute(&mut conn)
+        .await?;
+
+    Ok(())
+}
+
+#[derive(Debug, Serialize)]
+pub struct GetBookmarksRes {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    cursor: Option<String>,
+    bookmarks: Vec<BookmarkView>,
+}
+
+pub async fn get_bookmarks(
+    State(state): State<GlobalState>,
+    AtpAcceptLabelers(labelers): AtpAcceptLabelers,
+    auth: AtpAuth,
+    Query(query): Query<CursorQuery>,
+) -> XrpcResult<Json<GetBookmarksRes>> {
+    let mut conn = state.pool.get().await?;
+    let did = auth.0.clone();
+    let hyd = StatefulHydrator::new(&state.dataloaders, &state.cdn, &labelers, Some(auth));
+
+    let limit = query.limit.unwrap_or(50).clamp(1, 100);
+
+    let mut bookmarks_query = schema::bookmarks::table
+        .select(models::Bookmark::as_select())
+        .filter(schema::bookmarks::did.eq(&did))
+        .filter(schema::bookmarks::subject_type.eq_any(BSKY_ALLOWED_TYPES))
+        .into_boxed();
+
+    if let Some(cursor) = datetime_cursor(query.cursor.as_ref()) {
+        bookmarks_query = bookmarks_query.filter(schema::bookmarks::created_at.lt(cursor));
+    }
+
+    let results = bookmarks_query
+        .order(schema::bookmarks::created_at.desc())
+        .limit(limit as i64)
+        .load(&mut conn)
+        .await?;
+
+    let cursor = results
+        .last()
+        .map(|bm| bm.created_at.timestamp_millis().to_string());
+
+    let uris = results.iter().map(|bm| bm.subject.clone()).collect();
+
+    let mut posts = hyd.hydrate_posts(uris).await;
+
+    let bookmarks = results
+        .into_iter()
+        .filter_map(|bookmark| {
+            let maybe_item = posts.remove(&bookmark.subject);
+            let maybe_cid = maybe_item.as_ref().map(|v| v.cid.clone());
+
+            // ensure that either the cid is set in the bookmark record *or* in the post record
+            // otherwise just ditch. we should have one.
+            let cid = bookmark.subject_cid.or(maybe_cid)?;
+
+            let item = maybe_item.map(BookmarkViewItem::Post).unwrap_or(
+                BookmarkViewItem::NotFound {
+                    uri: bookmark.subject.clone(),
+                    not_found: true,
+                },
+            );
+
+            let subject = StrongRef::new_from_str(bookmark.subject, &cid).ok()?;
+
+            Some(BookmarkView {
+                subject,
+                item,
+                created_at: bookmark.created_at,
+            })
+        })
+        .collect();
+
+    Ok(Json(GetBookmarksRes { cursor, bookmarks }))
+}

parakeet/src/xrpc/app_bsky/mod.rs

···
 use axum::Router;
 
 mod actor;
+mod bookmark;
 mod feed;
 mod graph;
 mod labeler;
···
         // TODO: app.bsky.actor.getSuggestions (recs)
         // TODO: app.bsky.actor.searchActor (search)
         // TODO: app.bsky.actor.searchActorTypeahead (search)
+        .route("/app.bsky.bookmark.createBookmark", post(bookmark::create_bookmark))
+        .route("/app.bsky.bookmark.deleteBookmark", post(bookmark::delete_bookmark))
+        .route("/app.bsky.bookmark.getBookmarks", get(bookmark::get_bookmarks))
         .route("/app.bsky.feed.getActorFeeds", get(feed::feedgen::get_actor_feeds))
         .route("/app.bsky.feed.getActorLikes", get(feed::likes::get_actor_likes))
         .route("/app.bsky.feed.getAuthorFeed", get(feed::posts::get_author_feed))

parakeet/src/xrpc/community_lexicon/bookmarks.rs

···
+use crate::xrpc::datetime_cursor;
+use crate::xrpc::error::XrpcResult;
+use crate::xrpc::extract::AtpAuth;
+use crate::GlobalState;
+use axum::extract::{Query, State};
+use axum::Json;
+use diesel::prelude::*;
+use diesel_async::RunQueryDsl;
+use lexica::community_lexicon::bookmarks::Bookmark;
+use parakeet_db::{models, schema};
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Deserialize)]
+pub struct BookmarkCursorQuery {
+    pub tags: Option<Vec<String>>,
+    pub limit: Option<u8>,
+    pub cursor: Option<String>,
+}
+
+#[derive(Debug, Serialize)]
+pub struct GetActorBookmarksRes {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    cursor: Option<String>,
+    bookmarks: Vec<Bookmark>,
+}
+
+pub async fn get_actor_bookmarks(
+    State(state): State<GlobalState>,
+    auth: AtpAuth,
+    Query(query): Query<BookmarkCursorQuery>,
+) -> XrpcResult<Json<GetActorBookmarksRes>> {
+    let mut conn = state.pool.get().await?;
+
+    let limit = query.limit.unwrap_or(50).clamp(1, 100);
+
+    let mut bookmarks_query = schema::bookmarks::table
+        .select(models::Bookmark::as_select())
+        .filter(schema::bookmarks::did.eq(&auth.0))
+        .into_boxed();
+
+    if let Some(cursor) = datetime_cursor(query.cursor.as_ref()) {
+        bookmarks_query = bookmarks_query.filter(schema::bookmarks::created_at.lt(cursor));
+    }
+
+    if let Some(tags) = query.tags {
+        bookmarks_query = bookmarks_query.filter(schema::bookmarks::tags.contains(tags));
+    }
+
+    let results = bookmarks_query
+        .order(schema::bookmarks::created_at.desc())
+        .limit(limit as i64)
+        .load(&mut conn)
+        .await?;
+
+    let cursor = results
+        .last()
+        .map(|bm| bm.created_at.timestamp_millis().to_string());
+
+    let bookmarks = results
+        .into_iter()
+        .map(|bookmark| Bookmark {
+            subject: bookmark.subject,
+            tags: bookmark.tags.into_iter().flatten().collect(),
+            created_at: bookmark.created_at,
+        })
+        .collect();
+
+    Ok(Json(GetActorBookmarksRes { cursor, bookmarks }))
+}

parakeet/src/xrpc/community_lexicon/mod.rs

···
+use axum::routing::get;
+use axum::Router;
+
+pub mod bookmarks;
+
+#[rustfmt::skip]
+pub fn routes() -> Router<crate::GlobalState> {
+    Router::new()
+        .route("/community.lexicon.bookmarks.getActorBookmarks", get(bookmarks::get_actor_bookmarks))
+}

parakeet/src/xrpc/mod.rs

···
 mod app_bsky;
 pub mod cdn;
 mod com_atproto;
+mod community_lexicon;
 mod error;
 pub mod extract;
 pub mod jwt;
···
     Router::new()
         .merge(app_bsky::routes())
         .merge(com_atproto::routes())
+        .merge(community_lexicon::routes())
 }
 
 fn datetime_cursor(cursor: Option<&String>) -> Option<chrono::DateTime<chrono::Utc>> {