From 4ed9fbc19410f7fadeea1e78a655ae791257f72a Mon Sep 17 00:00:00 2001 From: Mia Date: Sat, 5 Apr 2025 18:03:34 +0100 Subject: [PATCH] feat: likes and reposts --- consumer/src/indexer/db.rs | 48 ++++++++++ consumer/src/indexer/mod.rs | 8 ++ consumer/src/indexer/records.rs | 14 +++ consumer/src/indexer/types.rs | 10 ++ lexica/src/app_bsky/feed.rs | 8 ++ .../down.sql | 2 + .../up.sql | 25 +++++ parakeet-db/src/models.rs | 24 ++++- parakeet-db/src/schema.rs | 26 ++++++ parakeet/src/xrpc/app_bsky/feed/likes.rs | 92 +++++++++++++++++++ parakeet/src/xrpc/app_bsky/feed/mod.rs | 1 + parakeet/src/xrpc/app_bsky/feed/posts.rs | 64 +++++++++++++ parakeet/src/xrpc/app_bsky/mod.rs | 2 + 13 files changed, 323 insertions(+), 1 deletion(-) create mode 100644 migrations/2025-04-05-114428_likes_and_reposts/down.sql create mode 100644 migrations/2025-04-05-114428_likes_and_reposts/up.sql create mode 100644 parakeet/src/xrpc/app_bsky/feed/likes.rs diff --git a/consumer/src/indexer/db.rs b/consumer/src/indexer/db.rs index 6158fa1f..3100a820 100644 --- a/consumer/src/indexer/db.rs +++ b/consumer/src/indexer/db.rs @@ -619,3 +619,51 @@ pub async fn delete_threadgate(conn: &mut AsyncPgConnection, at_uri: &str) -> Qu .execute(conn) .await } + +pub async fn insert_like( + conn: &mut AsyncPgConnection, + did: &str, + at_uri: &str, + rec: records::AppBskyFeedLike, +) -> QueryResult { + let data = models::NewLike { + at_uri, + did, + subject: &rec.subject.uri, + subject_cid: rec.subject.cid.to_string(), + created_at: rec.created_at.naive_utc(), + }; + + diesel::insert_into(schema::likes::table).values(&data).execute(conn).await +} + +pub async fn delete_like(conn: &mut AsyncPgConnection, at_uri: &str) -> QueryResult { + diesel::delete(schema::likes::table) + .filter(schema::likes::at_uri.eq(at_uri)) + .execute(conn) + .await +} + +pub async fn insert_repost( + conn: &mut AsyncPgConnection, + did: &str, + at_uri: &str, + rec: records::AppBskyFeedRepost, +) -> QueryResult { + let data = models::NewRepost { + at_uri, + did, + post: &rec.subject.uri, + post_cid: rec.subject.cid.to_string(), + created_at: rec.created_at.naive_utc(), + }; + + diesel::insert_into(schema::reposts::table).values(&data).execute(conn).await +} + +pub async fn delete_repost(conn: &mut AsyncPgConnection, at_uri: &str) -> QueryResult { + diesel::delete(schema::reposts::table) + .filter(schema::reposts::at_uri.eq(at_uri)) + .execute(conn) + .await +} diff --git a/consumer/src/indexer/mod.rs b/consumer/src/indexer/mod.rs index 6abff5a3..57e089c8 100644 --- a/consumer/src/indexer/mod.rs +++ b/consumer/src/indexer/mod.rs @@ -356,6 +356,9 @@ pub async fn index_op( RecordTypes::AppBskyFeedGenerator(record) => { db::upsert_feedgen(conn, repo, cid, at_uri, record).await?; } + RecordTypes::AppBskyFeedLike(record) => { + db::insert_like(conn, repo, at_uri, record).await?; + } RecordTypes::AppBskyFeedPost(record) => { if let Some(records::AppBskyEmbed::RecordWithMedia(embed)) = &record.embed { if !embed.media.record_with_media_allowed() { @@ -380,6 +383,9 @@ pub async fn index_op( db::postgate_maintain_detaches(conn, &record.post, &record.detached_embedding_uris, disable_effective) .await?; } + RecordTypes::AppBskyFeedRepost(record) => { + db::insert_repost(conn, repo, at_uri, record).await?; + } RecordTypes::AppBskyFeedThreadgate(record) => { let split_aturi = record.post.rsplitn(4, '/').collect::>(); if repo != split_aturi[2] { @@ -429,8 +435,10 @@ pub async fn index_op_delete( CollectionType::BskyProfile => db::delete_profile(conn, repo).await?, CollectionType::BskyBlock => db::delete_block(conn, at_uri).await?, CollectionType::BskyFeedGen => db::delete_feedgen(conn, at_uri).await?, + CollectionType::BskyFeedLike => db::delete_like(conn, at_uri).await?, CollectionType::BskyFeedPost => db::delete_post(conn, at_uri).await?, CollectionType::BskyFeedPostgate => db::delete_postgate(conn, at_uri).await?, + CollectionType::BskyFeedRepost => db::delete_repost(conn, at_uri).await?, CollectionType::BskyFeedThreadgate => db::delete_threadgate(conn, at_uri).await?, CollectionType::BskyFollow => { if let Some(subject) = db::delete_follow(conn, at_uri).await? { diff --git a/consumer/src/indexer/records.rs b/consumer/src/indexer/records.rs index 84b00f56..e056a522 100644 --- a/consumer/src/indexer/records.rs +++ b/consumer/src/indexer/records.rs @@ -148,6 +148,13 @@ pub struct AppBskyFeedGenerator { pub created_at: DateTime, } +#[derive(Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct AppBskyFeedLike { + pub subject: StrongRef, + pub created_at: DateTime, +} + #[derive(Debug, Deserialize, Serialize)] #[serde(tag = "$type")] #[serde(rename = "app.bsky.feed.post")] @@ -194,6 +201,13 @@ impl PostgateEmbeddingRules { } } +#[derive(Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct AppBskyFeedRepost { + pub subject: StrongRef, + pub created_at: DateTime, +} + #[derive(Debug, Deserialize, Serialize)] #[serde(tag = "$type")] #[serde(rename = "app.bsky.feed.threadgate")] diff --git a/consumer/src/indexer/types.rs b/consumer/src/indexer/types.rs index 5754b857..d6d7a6ce 100644 --- a/consumer/src/indexer/types.rs +++ b/consumer/src/indexer/types.rs @@ -9,10 +9,14 @@ pub enum RecordTypes { AppBskyActorProfile(records::AppBskyActorProfile), #[serde(rename = "app.bsky.feed.generator")] AppBskyFeedGenerator(records::AppBskyFeedGenerator), + #[serde(rename = "app.bsky.feed.like")] + AppBskyFeedLike(records::AppBskyFeedLike), #[serde(rename = "app.bsky.feed.post")] AppBskyFeedPost(records::AppBskyFeedPost), #[serde(rename = "app.bsky.feed.postgate")] AppBskyFeedPostgate(records::AppBskyFeedPostgate), + #[serde(rename = "app.bsky.feed.repost")] + AppBskyFeedRepost(records::AppBskyFeedRepost), #[serde(rename = "app.bsky.feed.threadgate")] AppBskyFeedThreadgate(records::AppBskyFeedThreadgate), #[serde(rename = "app.bsky.graph.block")] @@ -31,8 +35,10 @@ pub enum RecordTypes { pub enum CollectionType { BskyProfile, BskyFeedGen, + BskyFeedLike, BskyFeedPost, BskyFeedPostgate, + BskyFeedRepost, BskyFeedThreadgate, BskyBlock, BskyFollow, @@ -47,8 +53,10 @@ impl CollectionType { match input { "app.bsky.actor.profile" => CollectionType::BskyProfile, "app.bsky.feed.generator" => CollectionType::BskyFeedGen, + "app.bsky.feed.like" => CollectionType::BskyFeedLike, "app.bsky.feed.post" => CollectionType::BskyFeedPost, "app.bsky.feed.postgate" => CollectionType::BskyFeedPostgate, + "app.bsky.feed.repost" => CollectionType::BskyFeedRepost, "app.bsky.feed.threadgate" => CollectionType::BskyFeedThreadgate, "app.bsky.graph.block" => CollectionType::BskyBlock, "app.bsky.graph.follow" => CollectionType::BskyFollow, @@ -63,8 +71,10 @@ impl CollectionType { match self { CollectionType::BskyProfile => true, CollectionType::BskyFeedGen => true, + CollectionType::BskyFeedLike => false, CollectionType::BskyFeedPost => false, CollectionType::BskyFeedPostgate => true, + CollectionType::BskyFeedRepost => false, CollectionType::BskyFeedThreadgate => true, CollectionType::BskyBlock => false, CollectionType::BskyFollow => false, diff --git a/lexica/src/app_bsky/feed.rs b/lexica/src/app_bsky/feed.rs index 05519f4e..b174ecf7 100644 --- a/lexica/src/app_bsky/feed.rs +++ b/lexica/src/app_bsky/feed.rs @@ -174,3 +174,11 @@ pub struct ThreadgateView { #[serde(skip_serializing_if = "Vec::is_empty")] pub lists: Vec, } + +#[derive(Clone, Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct Like { + pub actor: ProfileView, + pub created_at: DateTime, + pub indexed_at: NaiveDateTime, +} diff --git a/migrations/2025-04-05-114428_likes_and_reposts/down.sql b/migrations/2025-04-05-114428_likes_and_reposts/down.sql new file mode 100644 index 00000000..18180c87 --- /dev/null +++ b/migrations/2025-04-05-114428_likes_and_reposts/down.sql @@ -0,0 +1,2 @@ +drop table likes; +drop table reposts; \ No newline at end of file diff --git a/migrations/2025-04-05-114428_likes_and_reposts/up.sql b/migrations/2025-04-05-114428_likes_and_reposts/up.sql new file mode 100644 index 00000000..0e098418 --- /dev/null +++ b/migrations/2025-04-05-114428_likes_and_reposts/up.sql @@ -0,0 +1,25 @@ +create table likes +( + at_uri text primary key, + did text not null references actors (did), + subject text not null, + subject_cid text not null, + created_at timestamptz not null, + indexed_at timestamp not null default now() +); + +create index likes_did_index on likes using hash (did); +create index likes_subject_index on likes using hash (subject); + +create table reposts +( + at_uri text primary key, + did text not null references actors (did), + post text not null, + post_cid text not null, + created_at timestamptz not null, + indexed_at timestamp not null default now() +); + +create index reposts_did_index on reposts using hash (did); +create index reposts_post_index on reposts using hash (post); \ No newline at end of file diff --git a/parakeet-db/src/models.rs b/parakeet-db/src/models.rs index ee2e9e1a..b93ecaa5 100644 --- a/parakeet-db/src/models.rs +++ b/parakeet-db/src/models.rs @@ -523,4 +523,26 @@ pub struct UpsertThreadgate<'a> { pub record: serde_json::Value, pub created_at: NaiveDateTime, -} \ No newline at end of file +} + +#[derive(Insertable)] +#[diesel(table_name = crate::schema::likes)] +#[diesel(check_for_backend(diesel::pg::Pg))] +pub struct NewLike<'a> { + pub at_uri: &'a str, + pub did: &'a str, + pub subject: &'a str, + pub subject_cid: String, + pub created_at: NaiveDateTime, +} + +#[derive(Insertable)] +#[diesel(table_name = crate::schema::reposts)] +#[diesel(check_for_backend(diesel::pg::Pg))] +pub struct NewRepost<'a> { + pub at_uri: &'a str, + pub did: &'a str, + pub post: &'a str, + pub post_cid: String, + pub created_at: NaiveDateTime, +} diff --git a/parakeet-db/src/schema.rs b/parakeet-db/src/schema.rs index 97bb24e1..2046bf62 100644 --- a/parakeet-db/src/schema.rs +++ b/parakeet-db/src/schema.rs @@ -65,6 +65,17 @@ diesel::table! { } } +diesel::table! { + likes (at_uri) { + at_uri -> Text, + did -> Text, + subject -> Text, + subject_cid -> Text, + created_at -> Timestamptz, + indexed_at -> Timestamp, + } +} + diesel::table! { list_blocks (at_uri) { at_uri -> Text, @@ -203,6 +214,17 @@ diesel::table! { } } +diesel::table! { + reposts (at_uri) { + at_uri -> Text, + did -> Text, + post -> Text, + post_cid -> Text, + created_at -> Timestamptz, + indexed_at -> Timestamp, + } +} + diesel::table! { threadgates (at_uri) { at_uri -> Text, @@ -221,6 +243,7 @@ diesel::joinable!(backfill -> actors (repo)); diesel::joinable!(blocks -> actors (did)); diesel::joinable!(feedgens -> actors (owner)); diesel::joinable!(follows -> actors (did)); +diesel::joinable!(likes -> actors (did)); diesel::joinable!(list_blocks -> actors (did)); diesel::joinable!(lists -> actors (owner)); diesel::joinable!(post_embed_ext -> posts (post_uri)); @@ -231,6 +254,7 @@ diesel::joinable!(post_embed_video_captions -> posts (post_uri)); diesel::joinable!(postgates -> posts (post_uri)); diesel::joinable!(posts -> actors (did)); diesel::joinable!(profiles -> actors (did)); +diesel::joinable!(reposts -> actors (did)); diesel::joinable!(threadgates -> posts (post_uri)); diesel::allow_tables_to_appear_in_same_query!( @@ -240,6 +264,7 @@ diesel::allow_tables_to_appear_in_same_query!( feedgens, follow_stats, follows, + likes, list_blocks, list_items, lists, @@ -251,5 +276,6 @@ diesel::allow_tables_to_appear_in_same_query!( postgates, posts, profiles, + reposts, threadgates, ); diff --git a/parakeet/src/xrpc/app_bsky/feed/likes.rs b/parakeet/src/xrpc/app_bsky/feed/likes.rs new file mode 100644 index 00000000..a041d10a --- /dev/null +++ b/parakeet/src/xrpc/app_bsky/feed/likes.rs @@ -0,0 +1,92 @@ +use crate::xrpc::error::XrpcResult; +use crate::xrpc::{datetime_cursor, normalise_at_uri}; +use crate::{hydration, GlobalState}; +use axum::extract::{Query, State}; +use axum::Json; +use diesel::prelude::*; +use diesel_async::RunQueryDsl; +use lexica::app_bsky::feed::Like; +use parakeet_db::schema; +use serde::{Deserialize, Serialize}; + +// TODO: getActorLikes: once we get auth! + +#[derive(Debug, Deserialize)] +pub struct GetLikesQuery { + pub uri: String, + pub cid: Option, + pub limit: Option, + pub cursor: Option, +} + +#[derive(Debug, Serialize)] +pub struct GetLikesRes { + pub uri: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub cid: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub cursor: Option, + pub likes: Vec, +} + +pub async fn get_likes( + State(state): State, + Query(query): Query, +) -> XrpcResult> { + let mut conn = state.pool.get().await?; + + let uri = normalise_at_uri(&state.dataloaders, &query.uri).await?; + + let limit = query.limit.unwrap_or(50).clamp(1, 100); + + let mut likes_query = schema::likes::table + .select(( + schema::likes::created_at, + schema::likes::indexed_at, + schema::likes::did, + )) + .filter(schema::likes::subject.eq(&uri)) + .into_boxed(); + + if let Some(cid) = &query.cid { + likes_query = likes_query.filter(schema::likes::subject_cid.eq(cid)); + } + + if let Some(cursor) = datetime_cursor(query.cursor.as_ref()) { + likes_query = likes_query.filter(schema::likes::created_at.lt(cursor)); + } + + let results = likes_query + .order(schema::likes::created_at.desc()) + .limit(limit as i64) + .load::<(chrono::DateTime, chrono::NaiveDateTime, String)>(&mut conn) + .await?; + + let cursor = results + .last() + .map(|(last, _, _)| last.timestamp_millis().to_string()); + + let liker_dids = results.iter().map(|(_, _, actor)| actor.clone()).collect(); + + let profiles = hydration::hydrate_profiles(&state.dataloaders, liker_dids).await; + + let likes = results + .into_iter() + .filter_map(|(created_at, indexed_at, did)| { + let actor = profiles.get(&did)?.clone(); + + Some(Like { + actor, + created_at, + indexed_at, + }) + }) + .collect(); + + Ok(Json(GetLikesRes { + uri, + cid: query.cid, + cursor, + likes, + })) +} diff --git a/parakeet/src/xrpc/app_bsky/feed/mod.rs b/parakeet/src/xrpc/app_bsky/feed/mod.rs index d2bfe626..940479c9 100644 --- a/parakeet/src/xrpc/app_bsky/feed/mod.rs +++ b/parakeet/src/xrpc/app_bsky/feed/mod.rs @@ -1,2 +1,3 @@ pub mod feedgen; +pub mod likes; pub mod posts; diff --git a/parakeet/src/xrpc/app_bsky/feed/posts.rs b/parakeet/src/xrpc/app_bsky/feed/posts.rs index d1668e41..62adfec3 100644 --- a/parakeet/src/xrpc/app_bsky/feed/posts.rs +++ b/parakeet/src/xrpc/app_bsky/feed/posts.rs @@ -7,6 +7,7 @@ use axum::Json; use axum_extra::extract::Query as ExtraQuery; use diesel::prelude::*; use diesel_async::RunQueryDsl; +use lexica::app_bsky::actor::ProfileView; use lexica::app_bsky::feed::{ FeedViewPost, PostView, ThreadViewPost, ThreadViewPostType, ThreadgateView, }; @@ -363,6 +364,69 @@ pub async fn get_quotes( })) } +#[derive(Debug, Deserialize)] +pub struct GetRepostedByQuery { + pub uri: String, + pub cid: Option, + pub limit: Option, + pub cursor: Option, +} + +#[derive(Debug, Serialize)] +pub struct GetRepostedByRes { + pub uri: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub cid: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub cursor: Option, + pub reposted_by: Vec, +} + +pub async fn get_reposted_by( + State(state): State, + Query(query): Query, +) -> XrpcResult> { + let mut conn = state.pool.get().await?; + + let uri = normalise_at_uri(&state.dataloaders, &query.uri).await?; + + let limit = query.limit.unwrap_or(50).clamp(1, 100); + + let mut reposted_by_query = schema::reposts::table + .select((schema::reposts::created_at, schema::reposts::did)) + .filter(schema::reposts::post.eq(&uri)) + .into_boxed(); + + if let Some(cid) = &query.cid { + reposted_by_query = reposted_by_query.filter(schema::reposts::post_cid.eq(cid)); + } + + if let Some(cursor) = datetime_cursor(query.cursor.as_ref()) { + reposted_by_query = reposted_by_query.filter(schema::reposts::created_at.lt(cursor)); + } + + let results = reposted_by_query + .order(schema::reposts::created_at.desc()) + .limit(limit as i64) + .load::<(chrono::DateTime, String)>(&mut conn) + .await?; + + let cursor = results + .last() + .map(|(last, _)| last.timestamp_millis().to_string()); + + let reposter_dids = results.iter().map(|(_, actor)| actor.clone()).collect(); + + let profiles = hydration::hydrate_profiles(&state.dataloaders, reposter_dids).await; + + Ok(Json(GetRepostedByRes { + uri, + cid: query.cid, + cursor, + reposted_by: profiles.into_values().collect(), + })) +} + // TODO: getTimeline: once we get auth! #[diesel::dsl::auto_type] diff --git a/parakeet/src/xrpc/app_bsky/mod.rs b/parakeet/src/xrpc/app_bsky/mod.rs index 5a2d99ca..c5d6da9b 100644 --- a/parakeet/src/xrpc/app_bsky/mod.rs +++ b/parakeet/src/xrpc/app_bsky/mod.rs @@ -11,10 +11,12 @@ pub fn routes() -> Router { .route("/app.bsky.actor.getProfiles", get(actor::get_profiles)) .route("/app.bsky.feed.getActorFeeds", get(feed::feedgen::get_actor_feeds)) .route("/app.bsky.feed.getAuthorFeed", get(feed::posts::get_author_feed)) + .route("/app.bsky.feed.getLikes", get(feed::likes::get_likes)) .route("/app.bsky.feed.getListFeed", get(feed::posts::get_list_feed)) .route("/app.bsky.feed.getPostThread", get(feed::posts::get_post_thread)) .route("/app.bsky.feed.getPosts", get(feed::posts::get_posts)) .route("/app.bsky.feed.getQuotes", get(feed::posts::get_quotes)) + .route("/app.bsky.feed.getRepostedBy", get(feed::posts::get_reposted_by)) .route("/app.bsky.feed.getFeedGenerator", get(feed::feedgen::get_feed_generator)) .route("/app.bsky.feed.getFeedGenerators", get(feed::feedgen::get_feed_generators)) .route("/app.bsky.graph.getFollowers", get(graph::relations::get_followers)) -- 2.51.2