#account and #identity indexing

consumer/src/firehose/types.rs

···
     }
 }
 
+impl Into<parakeet_db::types::ActorStatus> 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,

consumer/src/indexer/db.rs

···
+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<Option<String>>,
+    status: Option<types::ActorStatus>,
+    time: DateTime<Utc>,
+) -> QueryResult<usize> {
+    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
+}

consumer/src/indexer/mod.rs

···
 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(
···
     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(())
 }
 

migrations/2025-01-26-171756_actors/down.sql

···
+drop table actors;

migrations/2025-01-26-171756_actors/up.sql

···
+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
+);

parakeet-db/src/lib.rs

···
 pub mod models;
 pub mod schema;
+pub mod types;

parakeet-db/src/models.rs

···
+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<String>,
+    pub status: ActorStatus,
+    pub sync_state: ActorSyncState,
+    pub repo_rev: Option<String>,
+    pub repo_cid: Option<String>,
+    pub last_indexed: Option<NaiveDateTime>,
+}
+
+#[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<Option<String>>,
+    pub status: Option<ActorStatus>,
+    pub last_indexed: Option<NaiveDateTime>,
+}

parakeet-db/src/schema.rs

···
 // @generated automatically by Diesel CLI.
 
+diesel::table! {
+    actors (did) {
+        did -> Text,
+        handle -> Nullable<Text>,
+        status -> Text,
+        sync_state -> Text,
+        repo_rev -> Nullable<Text>,
+        repo_cid -> Nullable<Text>,
+        last_indexed -> Nullable<Timestamp>,
+    }
+}

parakeet-db/src/types.rs

···
+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<DB> FromSql<diesel::sql_types::Text, DB> for ActorStatus
+where
+    DB: Backend,
+    String: FromSql<diesel::sql_types::Text, DB>,
+{
+    fn from_sql(bytes: DB::RawValue<'_>) -> diesel::deserialize::Result<Self> {
+        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<diesel::sql_types::Text, Pg> 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",
+        };
+
+        <str as ToSql<diesel::sql_types::Text, Pg>>::to_sql(val, out)
+    }
+}
+
+#[derive(Debug, AsExpression, FromSqlRow)]
+#[diesel(sql_type = diesel::sql_types::Text)]
+pub enum ActorSyncState {
+    Synced,
+    Dirty,
+    Processing,
+}
+
+impl<DB> FromSql<diesel::sql_types::Text, DB> for ActorSyncState
+where
+    DB: Backend,
+    String: FromSql<diesel::sql_types::Text, DB>,
+{
+    fn from_sql(bytes: DB::RawValue<'_>) -> diesel::deserialize::Result<Self> {
+        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<diesel::sql_types::Text, Pg> 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",
+        };
+
+        <str as ToSql<diesel::sql_types::Text, Pg>>::to_sql(val, out)
+    }
+}