From 7c6baa297079ba12b4f0f17da732c2fa8bc7df5e Mon Sep 17 00:00:00 2001 From: Mia Date: Tue, 28 Jan 2025 20:47:11 +0000 Subject: [PATCH] #account and #identity indexing --- consumer/src/firehose/types.rs | 11 +++ consumer/src/indexer/db.rs | 27 +++++++ consumer/src/indexer/mod.rs | 27 ++++++- migrations/2025-01-26-171756_actors/down.sql | 1 + migrations/2025-01-26-171756_actors/up.sql | 15 ++++ parakeet-db/src/lib.rs | 1 + parakeet-db/src/models.rs | 27 +++++++ parakeet-db/src/schema.rs | 11 +++ parakeet-db/src/types.rs | 81 ++++++++++++++++++++ 9 files changed, 199 insertions(+), 2 deletions(-) create mode 100644 consumer/src/indexer/db.rs create mode 100644 migrations/2025-01-26-171756_actors/down.sql create mode 100644 migrations/2025-01-26-171756_actors/up.sql create mode 100644 parakeet-db/src/types.rs diff --git a/consumer/src/firehose/types.rs b/consumer/src/firehose/types.rs index e8067456..54de61af 100644 --- a/consumer/src/firehose/types.rs +++ b/consumer/src/firehose/types.rs @@ -59,6 +59,17 @@ impl AtpAccountStatus { } } +impl Into for AtpAccountStatus { + fn into(self) -> parakeet_db::types::ActorStatus { + match self { + AtpAccountStatus::Takendown => parakeet_db::types::ActorStatus::Takendown, + AtpAccountStatus::Suspended => parakeet_db::types::ActorStatus::Suspended, + AtpAccountStatus::Deleted => parakeet_db::types::ActorStatus::Deleted, + AtpAccountStatus::Deactivated => parakeet_db::types::ActorStatus::Deactivated, + } + } +} + #[derive(Debug, Deserialize)] pub struct AtpAccountEvent { pub seq: u64, diff --git a/consumer/src/indexer/db.rs b/consumer/src/indexer/db.rs new file mode 100644 index 00000000..4c3e6bad --- /dev/null +++ b/consumer/src/indexer/db.rs @@ -0,0 +1,27 @@ +use chrono::prelude::*; +use diesel::prelude::*; +use diesel_async::{AsyncPgConnection, RunQueryDsl}; +use parakeet_db::{models, schema, types}; + +pub async fn upsert_actor( + conn: &mut AsyncPgConnection, + did: &str, + handle: Option>, + status: Option, + time: DateTime, +) -> QueryResult { + let data = models::NewActor { + did, + handle, + status, + last_indexed: Some(time.naive_utc()), + }; + + diesel::insert_into(schema::actors::table) + .values(&data) + .on_conflict(schema::actors::did) + .do_update() + .set(&data) + .execute(conn) + .await +} diff --git a/consumer/src/indexer/mod.rs b/consumer/src/indexer/mod.rs index e14c2a61..3292cbce 100644 --- a/consumer/src/indexer/mod.rs +++ b/consumer/src/indexer/mod.rs @@ -4,10 +4,12 @@ use diesel_async::pooled_connection::deadpool::Pool; use diesel_async::AsyncPgConnection; use futures::StreamExt; use ipld_core::cid::Cid; +use parakeet_db::types::ActorStatus; use std::collections::HashMap; use tokio::sync::mpsc::Receiver; -use tracing::Instrument; +use tracing::{instrument, Instrument}; +mod db; mod types; pub async fn relay_indexer( @@ -44,11 +46,32 @@ pub async fn relay_indexer( Ok(()) } -async fn index_identity(conn: &mut AsyncPgConnection, identity: AtpIdentityEvent) -> eyre::Result<()> { +#[instrument(skip_all, fields(seq = identity.seq, repo=identity.did))] +async fn index_identity( + conn: &mut AsyncPgConnection, + identity: AtpIdentityEvent, +) -> eyre::Result<()> { + db::upsert_actor( + conn, + &identity.did, + Some(identity.handle), + None, + identity.time, + ) + .await?; + Ok(()) } +#[instrument(skip_all, fields(seq = account.seq, repo=account.did))] async fn index_account(conn: &mut AsyncPgConnection, account: AtpAccountEvent) -> eyre::Result<()> { + let status = account + .status + .map(|status| status.into()) + .unwrap_or(ActorStatus::Active); + + db::upsert_actor(conn, &account.did, None, Some(status), account.time).await?; + Ok(()) } diff --git a/migrations/2025-01-26-171756_actors/down.sql b/migrations/2025-01-26-171756_actors/down.sql new file mode 100644 index 00000000..696f90a1 --- /dev/null +++ b/migrations/2025-01-26-171756_actors/down.sql @@ -0,0 +1 @@ +drop table actors; \ No newline at end of file diff --git a/migrations/2025-01-26-171756_actors/up.sql b/migrations/2025-01-26-171756_actors/up.sql new file mode 100644 index 00000000..4cb4e273 --- /dev/null +++ b/migrations/2025-01-26-171756_actors/up.sql @@ -0,0 +1,15 @@ +create table actors +( + did text primary key, + handle text, + + -- active / takendown / suspended / deleted / deactivated + status text not null default 'active', + -- synced / dirty / processing + sync_state text not null default 'dirty', + + repo_rev text, + repo_cid text, + + last_indexed timestamp +); diff --git a/parakeet-db/src/lib.rs b/parakeet-db/src/lib.rs index d5cbad7e..3d8380a0 100644 --- a/parakeet-db/src/lib.rs +++ b/parakeet-db/src/lib.rs @@ -1,2 +1,3 @@ pub mod models; pub mod schema; +pub mod types; diff --git a/parakeet-db/src/models.rs b/parakeet-db/src/models.rs index e69de29b..13d0142a 100644 --- a/parakeet-db/src/models.rs +++ b/parakeet-db/src/models.rs @@ -0,0 +1,27 @@ +use crate::types::*; +use chrono::NaiveDateTime; +use diesel::prelude::*; + +#[derive(Debug, Queryable, Selectable, Identifiable)] +#[diesel(table_name = crate::schema::actors)] +#[diesel(primary_key(did))] +#[diesel(check_for_backend(diesel::pg::Pg))] +pub struct Actor { + pub did: String, + pub handle: Option, + pub status: ActorStatus, + pub sync_state: ActorSyncState, + pub repo_rev: Option, + pub repo_cid: Option, + pub last_indexed: Option, +} + +#[derive(Insertable, AsChangeset)] +#[diesel(table_name = crate::schema::actors)] +#[diesel(check_for_backend(diesel::pg::Pg))] +pub struct NewActor<'a> { + pub did: &'a str, + pub handle: Option>, + pub status: Option, + pub last_indexed: Option, +} diff --git a/parakeet-db/src/schema.rs b/parakeet-db/src/schema.rs index 4dc25e28..07ad7b5e 100644 --- a/parakeet-db/src/schema.rs +++ b/parakeet-db/src/schema.rs @@ -1,2 +1,13 @@ // @generated automatically by Diesel CLI. +diesel::table! { + actors (did) { + did -> Text, + handle -> Nullable, + status -> Text, + sync_state -> Text, + repo_rev -> Nullable, + repo_cid -> Nullable, + last_indexed -> Nullable, + } +} diff --git a/parakeet-db/src/types.rs b/parakeet-db/src/types.rs new file mode 100644 index 00000000..0a37a42e --- /dev/null +++ b/parakeet-db/src/types.rs @@ -0,0 +1,81 @@ +use diesel::backend::Backend; +use diesel::deserialize::FromSql; +use diesel::pg::Pg; +use diesel::serialize::{Output, ToSql}; +use diesel::{AsExpression, FromSqlRow}; + +#[derive(Debug, AsExpression, FromSqlRow)] +#[diesel(sql_type = diesel::sql_types::Text)] +pub enum ActorStatus { + Active, + Takendown, + Suspended, + Deleted, + Deactivated, +} + +impl FromSql for ActorStatus +where + DB: Backend, + String: FromSql, +{ + fn from_sql(bytes: DB::RawValue<'_>) -> diesel::deserialize::Result { + match String::from_sql(bytes)?.as_str() { + "active" => Ok(ActorStatus::Active), + "takendown" => Ok(ActorStatus::Takendown), + "suspended" => Ok(ActorStatus::Suspended), + "deleted" => Ok(ActorStatus::Deleted), + "deactivated" => Ok(ActorStatus::Deactivated), + x => Err(format!("Unrecognized variant {}", x).into()), + } + } +} + +impl ToSql for ActorStatus { + fn to_sql<'b>(&'b self, out: &mut Output<'b, '_, Pg>) -> diesel::serialize::Result { + let val = match self { + ActorStatus::Active => "active", + ActorStatus::Takendown => "takendown", + ActorStatus::Suspended => "suspended", + ActorStatus::Deleted => "deleted", + ActorStatus::Deactivated => "deactivated", + }; + + >::to_sql(val, out) + } +} + +#[derive(Debug, AsExpression, FromSqlRow)] +#[diesel(sql_type = diesel::sql_types::Text)] +pub enum ActorSyncState { + Synced, + Dirty, + Processing, +} + +impl FromSql for ActorSyncState +where + DB: Backend, + String: FromSql, +{ + fn from_sql(bytes: DB::RawValue<'_>) -> diesel::deserialize::Result { + match String::from_sql(bytes)?.as_str() { + "synced" => Ok(ActorSyncState::Synced), + "dirty" => Ok(ActorSyncState::Dirty), + "processing" => Ok(ActorSyncState::Processing), + x => Err(format!("Unrecognized variant {}", x).into()), + } + } +} + +impl ToSql for ActorSyncState { + fn to_sql<'b>(&'b self, out: &mut Output<'b, '_, Pg>) -> diesel::serialize::Result { + let val = match self { + ActorSyncState::Synced => "synced", + ActorSyncState::Dirty => "dirty", + ActorSyncState::Processing => "processing", + }; + + >::to_sql(val, out) + } +} -- 2.51.2