From 6991c129bb98f72f265470da0e282009ce11cb5e Mon Sep 17 00:00:00 2001 From: Mia Date: Tue, 17 Jun 2025 19:22:22 +0000 Subject: [PATCH] feat: app.bsky.actor.status --- migrations/2025-06-11-192947_statuses/down.sql | 1 + migrations/2025-06-11-192947_statuses/up.sql | 16 ++++++++++++++++ parakeet-db/src/models.rs | 22 ++++++++++++++++++++++ parakeet-db/src/schema.rs | 18 ++++++++++++++++++ parakeet/src/loaders.rs | 8 ++++++-- consumer/src/db/record.rs | 33 +++++++++++++++++++++++++++++++++ consumer/src/indexer/mod.rs | 6 ++++++ consumer/src/indexer/records.rs | 14 +++++++++++++- consumer/src/indexer/types.rs | 5 +++++ lexica/src/app_bsky/actor.rs | 53 +++++++++++++++++++++++++++++++++++++++++++++++++++++ parakeet/src/hydration/profile.rs | 108 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------------ consumer/src/db/sql/status_upsert.sql | 12 ++++++++++++ 12 file(s) changed, 269 insertion(s)(+), 27 deletion(s)(-) diff --git a/migrations/2025-06-11-192947_statuses/down.sql b/migrations/2025-06-11-192947_statuses/down.sql new file mode 100644 --- /dev/null +++ b/migrations/2025-06-11-192947_statuses/down.sql @@ -0,0 +1,1 @@ +drop table statuses; \ No newline at end of file diff --git a/migrations/2025-06-11-192947_statuses/up.sql b/migrations/2025-06-11-192947_statuses/up.sql new file mode 100644 --- /dev/null +++ b/migrations/2025-06-11-192947_statuses/up.sql @@ -0,0 +1,16 @@ +create table statuses +( + did text primary key references actors (did), + status text not null, + duration int, + record jsonb not null, + + embed_uri text, + embed_title text, + embed_description text, + thumb_mime_type text, + thumb_cid text, + + created_at timestamptz not null default now(), + indexed_at timestamp not null default now() +); \ No newline at end of file diff --git a/parakeet-db/src/models.rs b/parakeet-db/src/models.rs --- a/parakeet-db/src/models.rs +++ b/parakeet-db/src/models.rs @@ -718,3 +718,25 @@ #[diesel(treat_none_as_default_value = true)] pub indexed_at: Option, } + +#[derive(Clone, Debug, Queryable, Selectable, Identifiable)] +#[diesel(table_name = crate::schema::statuses)] +#[diesel(primary_key(did))] +#[diesel(check_for_backend(diesel::pg::Pg))] +pub struct Status { + pub did: String, + pub status: String, + pub duration: Option, + + pub record: serde_json::Value, + + pub embed_uri: Option, + pub embed_title: Option, + pub embed_description: Option, + pub thumb_mime_type: Option, + pub thumb_cid: Option, + + pub created_at: NaiveDateTime, + pub indexed_at: NaiveDateTime, +} + diff --git a/parakeet-db/src/schema.rs b/parakeet-db/src/schema.rs --- a/parakeet-db/src/schema.rs +++ b/parakeet-db/src/schema.rs @@ -303,6 +303,22 @@ } diesel::table! { + statuses (did) { + did -> Text, + status -> Text, + duration -> Nullable, + record -> Jsonb, + embed_uri -> Nullable, + embed_title -> Nullable, + embed_description -> Nullable, + thumb_mime_type -> Nullable, + thumb_cid -> Nullable, + created_at -> Timestamptz, + indexed_at -> Timestamp, + } +} + +diesel::table! { threadgates (at_uri) { at_uri -> Text, cid -> Text, @@ -349,6 +365,7 @@ diesel::joinable!(profiles -> actors (did)); diesel::joinable!(reposts -> actors (did)); diesel::joinable!(starterpacks -> actors (owner)); +diesel::joinable!(statuses -> actors (did)); diesel::joinable!(threadgates -> posts (post_uri)); diesel::joinable!(verification -> actors (verifier)); @@ -378,6 +395,7 @@ records, reposts, starterpacks, + statuses, threadgates, verification, ); diff --git a/parakeet/src/loaders.rs b/parakeet/src/loaders.rs --- a/parakeet/src/loaders.rs +++ b/parakeet/src/loaders.rs @@ -73,6 +73,7 @@ Option, bool, Option, + Option, ); impl BatchFn for ProfileLoader { async fn load(&mut self, keys: &[String]) -> HashMap { @@ -84,12 +85,14 @@ schema::chat_decls::table.on(schema::chat_decls::did.eq(schema::actors::did)), ) .left_join(schema::labelers::table.on(schema::labelers::did.eq(schema::actors::did))) + .left_join(schema::statuses::table.on(schema::statuses::did.eq(schema::actors::did))) .select(( schema::actors::did, schema::actors::handle, models::Profile::as_select(), schema::chat_decls::allow_incoming.nullable(), schema::labelers::cid.nullable(), + Option::::as_select(), )) .filter( schema::actors::did @@ -102,6 +105,7 @@ models::Profile, Option, Option, + Option, )>(&mut conn) .await; @@ -118,12 +122,12 @@ match res { Ok(res) => HashMap::from_iter(res.into_iter().map( - |(did, handle, profile, chat_decl, labeler_cid)| { + |(did, handle, profile, chat_decl, labeler_cid, status)| { let chat_decl = chat_decl.and_then(|v| ChatAllowIncoming::from_str(&v).ok()); let is_labeler = labeler_cid.is_some(); let maybe_stats = stats.remove(&did); - let val = (handle, profile, chat_decl, is_labeler, maybe_stats); + let val = (handle, profile, chat_decl, is_labeler, maybe_stats, status); (did, val) }, diff --git a/consumer/src/db/record.rs b/consumer/src/db/record.rs --- a/consumer/src/db/record.rs +++ b/consumer/src/db/record.rs @@ -586,6 +586,39 @@ .await } +pub async fn status_upsert( + conn: &mut C, + repo: &str, + rec: AppBskyActorStatus, +) -> PgExecResult { + let record = serde_json::to_value(&rec).unwrap(); + let thumb = rec.embed.as_ref().and_then(|v| v.external.thumb.clone()); + let thumb_mime = thumb.as_ref().map(|v| v.mime_type.clone()); + let thumb_cid = thumb.as_ref().map(|v| v.r#ref.to_string()); + + conn.execute( + include_str!("sql/status_upsert.sql"), + &[ + &repo, + &rec.status.to_string(), + &rec.duration_minutes, + &record, + &rec.embed.as_ref().map(|v| v.external.uri.clone()), + &rec.embed.as_ref().map(|v| v.external.title.clone()), + &rec.embed.as_ref().map(|v| v.external.description.clone()), + &thumb_mime, + &thumb_cid, + &rec.created_at, + ], + ) + .await +} + +pub async fn status_delete(conn: &mut C, did: &str) -> PgExecResult { + conn.execute("DELETE FROM statuses WHERE did=$1", &[&did]) + .await +} + pub async fn threadgate_upsert( conn: &mut C, at_uri: &str, diff --git a/consumer/src/indexer/mod.rs b/consumer/src/indexer/mod.rs --- a/consumer/src/indexer/mod.rs +++ b/consumer/src/indexer/mod.rs @@ -518,6 +518,11 @@ } } } + RecordTypes::AppBskyActorStatus(record) => { + if rkey == "self" { + db::status_upsert(conn, repo, record).await?; + } + } RecordTypes::AppBskyFeedGenerator(record) => { let labels = record.labels.clone(); let count = db::feedgen_upsert(conn, at_uri, repo, cid, record).await?; @@ -691,6 +696,7 @@ ) -> Result<(), tokio_postgres::Error> { match collection { CollectionType::BskyProfile => db::profile_delete(conn, repo).await?, + CollectionType::BskyStatus => db::status_delete(conn, repo).await?, CollectionType::BskyBlock => db::block_delete(conn, at_uri).await?, CollectionType::BskyFeedGen => { let count = db::feedgen_delete(conn, at_uri).await?; diff --git a/consumer/src/indexer/records.rs b/consumer/src/indexer/records.rs --- a/consumer/src/indexer/records.rs +++ b/consumer/src/indexer/records.rs @@ -1,7 +1,7 @@ use crate::utils; use chrono::{DateTime, Utc}; use ipld_core::cid::Cid; -use lexica::app_bsky::actor::ChatAllowIncoming; +use lexica::app_bsky::actor::{ChatAllowIncoming, Status}; use lexica::app_bsky::embed::AspectRatio; use lexica::app_bsky::labeler::LabelerPolicy; use lexica::app_bsky::richtext::FacetMain; @@ -41,6 +41,16 @@ pub joined_via_starter_pack: Option, pub pinned_post: Option, pub created_at: Option>, +} + +#[derive(Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct AppBskyActorStatus { + pub status: Status, + pub duration_minutes: Option, + pub embed: Option, + + pub created_at: DateTime, } #[derive(Clone, Debug, Deserialize, Serialize)] @@ -151,6 +161,8 @@ } #[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(tag = "$type")] +#[serde(rename = "app.bsky.embed.external")] pub struct AppBskyEmbedExternal { pub external: EmbedExternal, } diff --git a/consumer/src/indexer/types.rs b/consumer/src/indexer/types.rs --- a/consumer/src/indexer/types.rs +++ b/consumer/src/indexer/types.rs @@ -7,6 +7,8 @@ pub enum RecordTypes { #[serde(rename = "app.bsky.actor.profile")] AppBskyActorProfile(records::AppBskyActorProfile), + #[serde(rename = "app.bsky.actor.status")] + AppBskyActorStatus(records::AppBskyActorStatus), #[serde(rename = "app.bsky.feed.generator")] AppBskyFeedGenerator(records::AppBskyFeedGenerator), #[serde(rename = "app.bsky.feed.like")] @@ -42,6 +44,7 @@ #[derive(Debug, PartialOrd, PartialEq, Deserialize, Serialize)] pub enum CollectionType { BskyProfile, + BskyStatus, BskyFeedGen, BskyFeedLike, BskyFeedPost, @@ -64,6 +67,7 @@ pub(crate) fn from_str(input: &str) -> CollectionType { match input { "app.bsky.actor.profile" => CollectionType::BskyProfile, + "app.bsky.actor.status" => CollectionType::BskyStatus, "app.bsky.feed.generator" => CollectionType::BskyFeedGen, "app.bsky.feed.like" => CollectionType::BskyFeedLike, "app.bsky.feed.post" => CollectionType::BskyFeedPost, @@ -86,6 +90,7 @@ pub fn can_update(&self) -> bool { match self { CollectionType::BskyProfile => true, + CollectionType::BskyStatus => true, CollectionType::BskyFeedGen => true, CollectionType::BskyFeedLike => false, CollectionType::BskyFeedPost => false, diff --git a/lexica/src/app_bsky/actor.rs b/lexica/src/app_bsky/actor.rs --- a/lexica/src/app_bsky/actor.rs +++ b/lexica/src/app_bsky/actor.rs @@ -1,3 +1,4 @@ +use crate::app_bsky::embed::External; use crate::com_atproto::label::Label; use chrono::prelude::*; use serde::{Deserialize, Serialize}; @@ -52,6 +53,32 @@ } } +#[derive(Copy, Clone, Debug, Deserialize, Serialize)] +pub enum Status { + /// Advertises an account as currently offering live content. + #[serde(rename = "app.bsky.actor.status#live")] + Live, +} + +impl Display for Status { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Status::Live => write!(f, "app.bsky.actor.status#live"), + } + } +} + +impl FromStr for Status { + type Err = String; + + fn from_str(s: &str) -> Result { + match s { + "app.bsky.actor.status#live" => Ok(Status::Live), + x => Err(format!("Unrecognized variant {}", x)), + } + } +} + #[derive(Clone, Debug, Serialize)] #[serde(rename_all = "camelCase")] pub struct ProfileViewBasic { @@ -70,6 +97,8 @@ pub labels: Vec