diff --git a/consumer/src/db/record.rs b/consumer/src/db/record.rs index bed87b37..23d8749e 100644 --- a/consumer/src/db/record.rs +++ b/consumer/src/db/record.rs @@ -586,6 +586,39 @@ pub async fn starter_pack_delete(conn: &mut C, at_uri: &str) - .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/db/sql/status_upsert.sql b/consumer/src/db/sql/status_upsert.sql new file mode 100644 index 00000000..71012475 --- /dev/null +++ b/consumer/src/db/sql/status_upsert.sql @@ -0,0 +1,12 @@ +INSERT INTO statuses (did, status, duration, record, embed_uri, embed_title, embed_description, thumb_mime_type, + thumb_cid, created_at) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) +ON CONFLICT (did) DO UPDATE SET status=EXCLUDED.status, + duration=EXCLUDED.duration, + record=EXCLUDED.record, + embed_uri=EXCLUDED.embed_uri, + embed_title=EXCLUDED.embed_title, + embed_description=EXCLUDED.embed_description, + thumb_mime_type=EXCLUDED.thumb_mime_type, + thumb_cid=EXCLUDED.thumb_cid, + indexed_at=NOW() \ No newline at end of file diff --git a/consumer/src/indexer/mod.rs b/consumer/src/indexer/mod.rs index 6b0951d3..caa5eb8a 100644 --- a/consumer/src/indexer/mod.rs +++ b/consumer/src/indexer/mod.rs @@ -518,6 +518,11 @@ pub async fn index_op( } } } + 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 @@ pub async fn index_op_delete( ) -> 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 index c4098c0c..a6b0c10e 100644 --- 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; @@ -43,6 +43,16 @@ pub struct AppBskyActorProfile { 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)] #[serde(untagged)] pub enum EmbedOuter { @@ -151,6 +161,8 @@ pub struct EmbedVideoCaptions { } #[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 index 05aa9d04..99a8abf2 100644 --- a/consumer/src/indexer/types.rs +++ b/consumer/src/indexer/types.rs @@ -7,6 +7,8 @@ use serde::{Deserialize, Serialize}; 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 @@ pub enum RecordTypes { #[derive(Debug, PartialOrd, PartialEq, Deserialize, Serialize)] pub enum CollectionType { BskyProfile, + BskyStatus, BskyFeedGen, BskyFeedLike, BskyFeedPost, @@ -64,6 +67,7 @@ impl CollectionType { 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 @@ impl CollectionType { 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 index 0116d1d2..01977e0f 100644 --- 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 @@ impl FromStr for ChatAllowIncoming { } } +#[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 struct ProfileViewBasic { pub labels: Vec