use std::str::FromStr; use anyhow::Context; use chrono::{DateTime, Utc}; use jacquard_common::types::{aturi::AtUri, cid::Cid as AtpCid, did::Did, nsid::Nsid, string::Tid}; use sqlx::SqlitePool; use sqlx::types::Json; use crate::records::{Record, RecordRefs}; const CURSOR_UPPER_BOUND: &str = "~"; struct RawRecord { uri: String, cid: String, collection: String, creator_did: String, created_at: DateTime, cursor: String, data: Json, target_did: Option, ref_cids: Json>, } impl TryFrom for Record { type Error = anyhow::Error; fn try_from(raw: RawRecord) -> anyhow::Result { Ok(Record { uri: AtUri::new_owned(&raw.uri).context("invalid uri")?, cid: AtpCid::from_str(&raw.cid).context("invalid cid in database")?, collection: Nsid::new_owned(&raw.collection).context("invalid collection nsid")?, creator_did: Did::new_owned(&raw.creator_did).context("invalid creator_did")?, created_at: raw.created_at, cursor: Tid::new(&raw.cursor).context("invalid cursor tid")?, data: raw.data.0, target_did: raw .target_did .map(|d| Did::new_owned(d).context("invalid target_did")) .transpose()?, refs: RecordRefs::from_stored( raw.ref_cids .0 .iter() .map(|s| AtpCid::from_str(s).unwrap()) .collect(), ), }) } } pub async fn insert_record(pool: &SqlitePool, record: &Record) -> sqlx::Result<()> { let uri = record.uri.as_ref(); let cid = record.cid.as_ref(); let collection = record.collection.as_ref(); let creator_did = record.creator_did.as_ref(); let cursor = record.cursor.as_ref(); let data = Json(&record.data); let ref_cid_strings = record.refs.to_cid_strings(); let ref_cids = Json(&ref_cid_strings); let target_did = record.target_did.as_ref().map(|d| d.as_ref()); sqlx::query!( "INSERT INTO record (uri, cid, collection, creator_did, created_at, cursor, data, target_did, ref_cids) \ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) \ ON CONFLICT (uri) DO UPDATE SET \ cid = excluded.cid, \ data = excluded.data, \ cursor = excluded.cursor, \ target_did = excluded.target_did, \ ref_cids = excluded.ref_cids", uri, cid, collection, creator_did, record.created_at, cursor, data, target_did, ref_cids ) .execute(pool) .await?; Ok(()) } pub async fn list_records_by_creator( pool: &SqlitePool, collection: &Nsid<'_>, creator_did: &Did<'_>, cursor: Option<&Tid>, limit: i64, ) -> anyhow::Result> { let collection_str: &str = collection; let creator_str: &str = creator_did; let cursor_str = cursor.map(|c| c.as_ref()).unwrap_or(CURSOR_UPPER_BOUND); let raw_records = sqlx::query_as!( RawRecord, r#"SELECT uri as "uri!", cid as "cid!", collection as "collection!", creator_did as "creator_did!", created_at as "created_at!: DateTime", cursor as "cursor!", data as "data!: Json", target_did, ref_cids as "ref_cids!: Json>" FROM record WHERE collection = ? AND creator_did = ? AND cursor < ? ORDER BY cursor DESC LIMIT ?"#, collection_str, creator_str, cursor_str, limit ) .fetch_all(pool) .await?; raw_records.into_iter().map(Record::try_from).collect() } pub async fn list_records_by_target( pool: &SqlitePool, collection: &Nsid<'_>, target_did: &Did<'_>, cursor: Option<&Tid>, limit: i64, ) -> anyhow::Result> { let collection_str: &str = collection; let target_str: &str = target_did; let cursor_str = cursor.map(|c| c.as_ref()).unwrap_or(CURSOR_UPPER_BOUND); let raw_records = sqlx::query_as!( RawRecord, r#"SELECT uri as "uri!", cid as "cid!", collection as "collection!", creator_did as "creator_did!", created_at as "created_at!: DateTime", cursor as "cursor!", data as "data!: Json", target_did, ref_cids as "ref_cids!: Json>" FROM record WHERE collection = ? AND target_did = ? AND cursor < ? ORDER BY cursor DESC LIMIT ?"#, collection_str, target_str, cursor_str, limit ) .fetch_all(pool) .await?; raw_records.into_iter().map(Record::try_from).collect() } pub async fn get_record_by_cid( pool: &SqlitePool, cid: &AtpCid<'_>, ) -> anyhow::Result> { let cid_str: &str = cid; let raw = sqlx::query_as!( RawRecord, r#"SELECT uri as "uri!", cid as "cid!", collection as "collection!", creator_did as "creator_did!", created_at as "created_at!: DateTime", cursor as "cursor!", data as "data!: Json", target_did, ref_cids as "ref_cids!: Json>" FROM record WHERE cid = ?"#, cid_str ) .fetch_optional(pool) .await?; raw.map(Record::try_from).transpose() }