Something went wrong. Try again.
relay filter/appview bootstrap
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157use std::sync::Arc;
use chrono::{DateTime, Utc};use futures::StreamExt;use hydrant::control::Hydrant;use hydrant::types::RecordEvt;use jacquard_common::types::{aturi::AtUri, cid::Cid as AtpCid, did::Did, nsid::Nsid};
use crate::AppState;use crate::db;use crate::records::{ CHANNEL_COLLECTION, INVITE_COLLECTION, LATTICE_COLLECTION, MEMBERSHIP_COLLECTION, Record, RecordRefs, SHARD_COLLECTION, new_cursor,};
const ACTION_DELETE: &str = "delete";
enum KnownCollection { Lattice, Shard, Channel, Invite, Membership,}
impl KnownCollection { fn parse(nsid: &str) -> Option<Self> { match nsid { s if s == LATTICE_COLLECTION.as_ref() => Some(Self::Lattice), s if s == SHARD_COLLECTION.as_ref() => Some(Self::Shard), s if s == CHANNEL_COLLECTION.as_ref() => Some(Self::Channel), s if s == INVITE_COLLECTION.as_ref() => Some(Self::Invite), s if s == MEMBERSHIP_COLLECTION.as_ref() => Some(Self::Membership), _ => None, } }}
fn parse_datetime(s: &str) -> Option<DateTime<Utc>> { DateTime::parse_from_rfc3339(s) .map(|dt| dt.with_timezone(&Utc)) .ok()}
fn build_uri(did: &Did<'_>, collection: &Nsid<'_>, rkey: &str) -> anyhow::Result<AtUri<'static>> { AtUri::from_parts_owned(did.as_ref(), collection.as_ref(), rkey) .map_err(|e| anyhow::anyhow!("invalid AT-URI: {e}"))}
fn extract_cid(record: &serde_json::Value, field: &str) -> Option<AtpCid<'static>> { record .get(field) .and_then(|v| v.get("cid")) .and_then(|v| v.as_str()) .map(|s| s.parse::<AtpCid>().unwrap())}
async fn process_record_event(state: &Arc<AppState>, evt: &RecordEvt<'_>) -> anyhow::Result<()> { if evt.action.as_ref() == ACTION_DELETE { return Ok(()); }
let Some(record) = &evt.record else { return Ok(()); };
let Some(collection_kind) = KnownCollection::parse(evt.collection.as_ref()) else { return Ok(()); };
let Some(cid) = &evt.cid else { return Ok(()); }; let event_cid = AtpCid::from(*cid);
let pool = &state.pool; let now = Utc::now(); let creator_did = Did::new_owned(evt.did.as_ref())?; let collection = Nsid::new_owned(evt.collection.as_ref())?; let uri = build_uri(&creator_did, &collection, evt.rkey.as_ref())?;
let created_at = record .get("createdAt") .and_then(|v| v.as_str()) .and_then(parse_datetime) .unwrap_or(now);
let (target_did, refs) = match collection_kind { KnownCollection::Lattice | KnownCollection::Shard | KnownCollection::Channel => { (None, RecordRefs::default()) } KnownCollection::Invite => { let recipient = record .get("recipient") .and_then(|v| v.as_str()) .map(Did::new_owned) .transpose()?;
let refs = RecordRefs { channel: extract_cid(record, "channel"), ..Default::default() };
(recipient, refs) } KnownCollection::Membership => { let refs = RecordRefs { channel: extract_cid(record, "channel"), invite: extract_cid(record, "invite"), };
(Some(creator_did.clone()), refs) } };
db::insert_record( pool, &Record { uri, cid: event_cid, collection, creator_did, created_at, cursor: new_cursor(), data: record.clone(), target_did, refs, }, ) .await?;
if let Ok(json) = serde_json::to_string(record) { let _ = state.tx.send(json); }
Ok(())}
pub async fn run(hydrant: Hydrant, state: Arc<AppState>) { tracing::info!("Subscribed to hydrant event stream");
hydrant .subscribe(None) .filter_map(|event| std::future::ready(event.record)) .for_each(|evt| { let state = &state; async move { if let Err(e) = process_record_event(state, &evt).await { tracing::error!(error = %e, "Failed to process hydrant event"); } } }) .await;
tracing::warn!("Hydrant event stream ended");}