From eab5c41736b83a29587b5f30a6138dcfa7b05a39 Mon Sep 17 00:00:00 2001 From: Mia Date: Thu, 11 Sep 2025 18:48:43 +0000 Subject: [PATCH] feat: bookmarks --- consumer/src/db/record.rs | 32 ++++ consumer/src/db/sql/bookmarks_upsert.sql | 5 + consumer/src/indexer/mod.rs | 6 + consumer/src/indexer/types.rs | 5 + lexica/src/app_bsky/bookmark.rs | 32 ++++ lexica/src/app_bsky/mod.rs | 1 + lexica/src/community_lexicon/bookmarks.rs | 14 ++ lexica/src/community_lexicon/mod.rs | 1 + lexica/src/lib.rs | 8 + .../2025-09-02-190833_bookmarks/down.sql | 1 + migrations/2025-09-02-190833_bookmarks/up.sql | 19 +++ parakeet-db/src/models.rs | 26 ++++ parakeet-db/src/schema.rs | 14 ++ parakeet/src/xrpc/app_bsky/bookmark.rs | 146 ++++++++++++++++++ parakeet/src/xrpc/app_bsky/mod.rs | 4 + .../src/xrpc/community_lexicon/bookmarks.rs | 69 +++++++++ parakeet/src/xrpc/community_lexicon/mod.rs | 10 ++ parakeet/src/xrpc/mod.rs | 2 + 18 files changed, 395 insertions(+) create mode 100644 consumer/src/db/sql/bookmarks_upsert.sql create mode 100644 lexica/src/app_bsky/bookmark.rs create mode 100644 lexica/src/community_lexicon/bookmarks.rs create mode 100644 lexica/src/community_lexicon/mod.rs create mode 100644 migrations/2025-09-02-190833_bookmarks/down.sql create mode 100644 migrations/2025-09-02-190833_bookmarks/up.sql create mode 100644 parakeet/src/xrpc/app_bsky/bookmark.rs create mode 100644 parakeet/src/xrpc/community_lexicon/bookmarks.rs create mode 100644 parakeet/src/xrpc/community_lexicon/mod.rs diff --git a/consumer/src/db/record.rs b/consumer/src/db/record.rs index 6d9b632b..bca14047 100644 --- a/consumer/src/db/record.rs +++ b/consumer/src/db/record.rs @@ -4,6 +4,7 @@ use crate::utils::{blob_ref, strongref_to_parts}; use chrono::prelude::*; use deadpool_postgres::GenericClient; use ipld_core::cid::Cid; +use lexica::community_lexicon::bookmarks::Bookmark; pub async fn record_upsert( conn: &mut C, @@ -22,6 +23,37 @@ pub async fn record_delete(conn: &mut C, at_uri: &str) -> PgEx .await } +pub async fn bookmark_upsert( + 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::>()[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( + 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( conn: &mut C, rkey: &str, diff --git a/consumer/src/db/sql/bookmarks_upsert.sql b/consumer/src/db/sql/bookmarks_upsert.sql new file mode 100644 index 00000000..facc5172 --- /dev/null +++ b/consumer/src/db/sql/bookmarks_upsert.sql @@ -0,0 +1,5 @@ +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 \ No newline at end of file diff --git a/consumer/src/indexer/mod.rs b/consumer/src/indexer/mod.rs index b207bfdc..a813f845 100644 --- a/consumer/src/indexer/mod.rs +++ b/consumer/src/indexer/mod.rs @@ -723,6 +723,9 @@ pub async fn index_op( 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?; @@ -833,6 +836,9 @@ pub async fn index_op_delete( 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!(), }; diff --git a/consumer/src/indexer/types.rs b/consumer/src/indexer/types.rs index a1d1aaf1..795b9e5d 100644 --- a/consumer/src/indexer/types.rs +++ b/consumer/src/indexer/types.rs @@ -41,6 +41,8 @@ pub enum RecordTypes { 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)] @@ -63,6 +65,7 @@ pub enum CollectionType { BskyLabelerService, BskyNotificationDeclaration, ChatActorDecl, + CommunityLexiconBookmark, Unsupported, } @@ -87,6 +90,7 @@ impl CollectionType { "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, } } @@ -111,6 +115,7 @@ impl CollectionType { CollectionType::BskyVerification => false, CollectionType::BskyLabelerService => true, CollectionType::BskyNotificationDeclaration => true, + CollectionType::CommunityLexiconBookmark => true, CollectionType::Unsupported => false, } } diff --git a/lexica/src/app_bsky/bookmark.rs b/lexica/src/app_bsky/bookmark.rs new file mode 100644 index 00000000..24d67cea --- /dev/null +++ b/lexica/src/app_bsky/bookmark.rs @@ -0,0 +1,32 @@ +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, +} + +#[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, + }, +} diff --git a/lexica/src/app_bsky/mod.rs b/lexica/src/app_bsky/mod.rs index 1ae1b370..4505c3d1 100644 --- a/lexica/src/app_bsky/mod.rs +++ b/lexica/src/app_bsky/mod.rs @@ -1,6 +1,7 @@ use serde::Serialize; pub mod actor; +pub mod bookmark; pub mod embed; pub mod feed; pub mod graph; diff --git a/lexica/src/community_lexicon/bookmarks.rs b/lexica/src/community_lexicon/bookmarks.rs new file mode 100644 index 00000000..f5cda6ea --- /dev/null +++ b/lexica/src/community_lexicon/bookmarks.rs @@ -0,0 +1,14 @@ +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, + pub created_at: DateTime, +} \ No newline at end of file diff --git a/lexica/src/community_lexicon/mod.rs b/lexica/src/community_lexicon/mod.rs new file mode 100644 index 00000000..3f736c29 --- /dev/null +++ b/lexica/src/community_lexicon/mod.rs @@ -0,0 +1 @@ +pub mod bookmarks; \ No newline at end of file diff --git a/lexica/src/lib.rs b/lexica/src/lib.rs index 7c1a9188..54974c80 100644 --- a/lexica/src/lib.rs +++ b/lexica/src/lib.rs @@ -5,6 +5,7 @@ pub use utils::LinkRef; pub mod app_bsky; pub mod com_atproto; +pub mod community_lexicon; mod utils; #[derive(Clone, Debug, Serialize)] @@ -23,6 +24,13 @@ pub struct StrongRef { pub uri: String, } +impl StrongRef { + pub fn new_from_str(uri: String, cid: &str) -> Result { + let cid = cid.parse()?; + Ok(StrongRef { uri, cid }) + } +} + #[derive(Clone, Debug, Deserialize, Serialize)] #[serde(tag = "$type")] #[serde(rename = "blob")] diff --git a/migrations/2025-09-02-190833_bookmarks/down.sql b/migrations/2025-09-02-190833_bookmarks/down.sql new file mode 100644 index 00000000..e3378df5 --- /dev/null +++ b/migrations/2025-09-02-190833_bookmarks/down.sql @@ -0,0 +1 @@ +drop table bookmarks; \ No newline at end of file diff --git a/migrations/2025-09-02-190833_bookmarks/up.sql b/migrations/2025-09-02-190833_bookmarks/up.sql new file mode 100644 index 00000000..137da209 --- /dev/null +++ b/migrations/2025-09-02-190833_bookmarks/up.sql @@ -0,0 +1,19 @@ +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); diff --git a/parakeet-db/src/models.rs b/parakeet-db/src/models.rs index 1cbb2a9a..71a3a10d 100644 --- a/parakeet-db/src/models.rs +++ b/parakeet-db/src/models.rs @@ -383,3 +383,29 @@ pub struct NewListMute<'a> { 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, + pub subject: String, + pub subject_cid: Option, + pub subject_type: String, + pub tags: Vec>, + pub created_at: DateTime, +} + +#[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, + pub subject: &'a str, + pub subject_cid: Option, + pub subject_type: &'a str, + pub tags: Vec, +} \ No newline at end of file diff --git a/parakeet-db/src/schema.rs b/parakeet-db/src/schema.rs index 7153a336..8e9e3bb4 100644 --- a/parakeet-db/src/schema.rs +++ b/parakeet-db/src/schema.rs @@ -42,6 +42,18 @@ diesel::table! { } } +diesel::table! { + bookmarks (did, subject) { + did -> Text, + rkey -> Nullable, + subject -> Text, + subject_cid -> Nullable, + subject_type -> Text, + tags -> Array>, + created_at -> Timestamptz, + } +} + diesel::table! { chat_decls (did) { did -> Text, @@ -375,6 +387,7 @@ diesel::table! { 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)); @@ -405,6 +418,7 @@ diesel::allow_tables_to_appear_in_same_query!( backfill, backfill_jobs, blocks, + bookmarks, chat_decls, feedgens, follows, diff --git a/parakeet/src/xrpc/app_bsky/bookmark.rs b/parakeet/src/xrpc/app_bsky/bookmark.rs new file mode 100644 index 00000000..229393b9 --- /dev/null +++ b/parakeet/src/xrpc/app_bsky/bookmark.rs @@ -0,0 +1,146 @@ +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, + auth: AtpAuth, + Json(form): Json, +) -> XrpcResult<()> { + let mut conn = state.pool.get().await?; + + // strip "at://" then break into parts by '/' + let parts = form.uri[5..].split('/').collect::>(); + + 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, + auth: AtpAuth, + Json(form): Json, +) -> 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, + bookmarks: Vec, +} + +pub async fn get_bookmarks( + State(state): State, + AtpAcceptLabelers(labelers): AtpAcceptLabelers, + auth: AtpAuth, + Query(query): Query, +) -> XrpcResult> { + 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 })) +} diff --git a/parakeet/src/xrpc/app_bsky/mod.rs b/parakeet/src/xrpc/app_bsky/mod.rs index 44a33eb3..8c5a9b84 100644 --- a/parakeet/src/xrpc/app_bsky/mod.rs +++ b/parakeet/src/xrpc/app_bsky/mod.rs @@ -2,6 +2,7 @@ use axum::routing::{get, post}; use axum::Router; mod actor; +mod bookmark; mod feed; mod graph; mod labeler; @@ -14,6 +15,9 @@ pub fn routes() -> Router { // 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)) diff --git a/parakeet/src/xrpc/community_lexicon/bookmarks.rs b/parakeet/src/xrpc/community_lexicon/bookmarks.rs new file mode 100644 index 00000000..419bacef --- /dev/null +++ b/parakeet/src/xrpc/community_lexicon/bookmarks.rs @@ -0,0 +1,69 @@ +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>, + pub limit: Option, + pub cursor: Option, +} + +#[derive(Debug, Serialize)] +pub struct GetActorBookmarksRes { + #[serde(skip_serializing_if = "Option::is_none")] + cursor: Option, + bookmarks: Vec, +} + +pub async fn get_actor_bookmarks( + State(state): State, + auth: AtpAuth, + Query(query): Query, +) -> XrpcResult> { + 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 })) +} diff --git a/parakeet/src/xrpc/community_lexicon/mod.rs b/parakeet/src/xrpc/community_lexicon/mod.rs new file mode 100644 index 00000000..a78e3880 --- /dev/null +++ b/parakeet/src/xrpc/community_lexicon/mod.rs @@ -0,0 +1,10 @@ +use axum::routing::get; +use axum::Router; + +pub mod bookmarks; + +#[rustfmt::skip] +pub fn routes() -> Router { + Router::new() + .route("/community.lexicon.bookmarks.getActorBookmarks", get(bookmarks::get_actor_bookmarks)) +} diff --git a/parakeet/src/xrpc/mod.rs b/parakeet/src/xrpc/mod.rs index 70af1cc4..d85753ea 100644 --- a/parakeet/src/xrpc/mod.rs +++ b/parakeet/src/xrpc/mod.rs @@ -8,6 +8,7 @@ use std::str::FromStr; mod app_bsky; pub mod cdn; mod com_atproto; +mod community_lexicon; mod error; pub mod extract; pub mod jwt; @@ -16,6 +17,7 @@ pub fn xrpc_routes() -> Router { Router::new() .merge(app_bsky::routes()) .merge(com_atproto::routes()) + .merge(community_lexicon::routes()) } fn datetime_cursor(cursor: Option<&String>) -> Option> { -- 2.51.2