Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352use fjall::OwnedWriteBatch;
use jacquard_common::IntoStatic;use jacquard_common::types::did::Did;use miette::{Context, IntoDiagnostic, Result};use tracing::debug;
use crate::db::types::{DbAction, DbRkey, DbTid};use crate::db::{self, Db, Txn, keys};use crate::filter::FilterConfig;use crate::ingest::stream::Commit;use crate::ingest::validation::ValidatedCommit;use crate::state::AppState;use crate::types::{GaugeState, RepoState, RepoStatus, ResyncErrorKind, ResyncState};
#[cfg(feature = "indexer_stream")]use { crate::types::{AccountEvt, BroadcastEvent, IdentityEvt, MarshallableEvt}, jacquard_common::types::string::Handle, std::sync::atomic::Ordering,};pub(crate) mod backlink_ops;use record_events::RecordEvents;
pub mod record_events;
pub fn persist_to_resync_buffer( txn: &mut Txn<'_>, db: &Db, did: &Did, commit: &Commit,) -> Result<()> { if txn.lock_repo_and_is_excluded(did)? { return Ok(()); } let key = keys::resync_buffer_key(did, DbTid::from(&commit.rev)); let value = rmp_serde::to_vec_named(commit).into_diagnostic()?; txn.batch.insert(&db.indexer.resync_buffer, key, value); debug!( did = %did, seq = commit.seq, "buffered commit to resync_buffer" ); Ok(())}
// emitting identity is ephemeral// we dont replay these, consumers can just fetch identity themselves if they need it#[cfg(feature = "indexer_stream")]pub fn make_identity_event(db: &Db, evt: IdentityEvt) -> BroadcastEvent { let event_id = db.stream.next_event_id.fetch_add(1, Ordering::SeqCst); let marshallable = MarshallableEvt { id: event_id, kind: crate::types::EventType::Identity, record: None, identity: Some(evt), account: None, }; BroadcastEvent::Ephemeral(Box::new(marshallable))}
/// announce an identity the caller learned by re-resolving a doc rather than by/// receiving an event. ephemeral, like every other identity event.#[cfg(feature = "indexer_stream")]pub fn announce_identity(db: &Db, did: &Did, handle: Option<Handle>) { let evt = IdentityEvt { did: did.clone(), handle, }; let _ = db.stream.event_tx.send(make_identity_event(db, evt));}
#[cfg(feature = "indexer_stream")]pub fn make_account_event(db: &Db, evt: AccountEvt<'static>) -> BroadcastEvent { let event_id = db.stream.next_event_id.fetch_add(1, Ordering::SeqCst); let marshallable = MarshallableEvt { id: event_id, kind: crate::types::EventType::Account, record: None, identity: None, account: Some(evt), }; BroadcastEvent::Ephemeral(Box::new(marshallable))}
pub(crate) fn delete_repo( txn: &mut Txn<'_>, db: &Db, did: &Did, repo_state: &RepoState,) -> Result<()> { debug!(did = %did, "deleting repo");
let repo_key = keys::repo_key(did); let metadata_key = keys::repo_metadata_key(did);
let metadata_bytes = db.repo_metadata.get(&metadata_key).into_diagnostic()?; if let Some(metadata_bytes) = metadata_bytes { let metadata = db::deser_repo_meta(&metadata_bytes)?; txn.batch .remove(&db.indexer.pending, keys::pending_key(metadata.index_id)); }
// we don't delete from repos, relay uses it as a tombstone // todo: we should still delete it after some time txn.batch.remove(&db.indexer.resync, &repo_key); txn.batch.remove(&db.repo_metadata, &metadata_key);
// 2. delete from resync buffer let resync_prefix = keys::resync_buffer_prefix(did); for guard in db.indexer.resync_buffer.prefix(&resync_prefix) { let k = guard.key().into_diagnostic()?; txn.batch.remove(&db.indexer.resync_buffer, k); }
// 3. move record bodies to history, then delete the heads. history keeps // them resolvable for event replay; the synthetic death rev just needs to // sort strictly after every real rev we have seen (including future-skewed ones) let root_rev = repo_state.root.as_ref().map(|c| &c.rev); db::delete_repo_records(txn, db, did, root_rev)?;
// 5. remove backlinks for all records in this repo backlink_ops::delete_repo(&mut txn.batch, db, did)?;
Ok(())}
pub fn transition_repo<'s>( batch: &mut OwnedWriteBatch, db: &Db, lifecycle_transitions: &mut Vec<(Did, GaugeState)>, did: &Did, mut repo_state: RepoState<'s>, new_status: RepoStatus,) -> Result<RepoState<'s>> { debug!(did = %did, status = ?new_status, "updating repo status");
let repo_key = keys::repo_key(did); let metadata_key = keys::repo_metadata_key(did);
let metadata_bytes = db.repo_metadata.get(&metadata_key).into_diagnostic()?; if let Some(metadata_bytes) = metadata_bytes { let metadata = db::deser_repo_meta(&metadata_bytes)?; let pending_key = keys::pending_key(metadata.index_id);
// manage queues match &new_status { RepoStatus::Synced => { lifecycle_transitions.push((did.clone().into_static(), GaugeState::Synced)); batch.remove(&db.indexer.pending, pending_key.as_slice()); batch.remove(&db.indexer.resync, &repo_key); } RepoStatus::Error(msg) => { tracing::warn!("transitioning to error: {msg}"); lifecycle_transitions.push(( did.clone().into_static(), GaugeState::Resync(Some(ResyncErrorKind::Generic)), )); batch.remove(&db.indexer.pending, pending_key.as_slice()); // TODO: we need to make errors have kind instead of "message" in repo status // and then pass it to resync error kind let resync_state = crate::types::ResyncState::Error { kind: crate::types::ResyncErrorKind::Generic, retry_count: 0, next_retry: chrono::Utc::now().timestamp(), }; batch.insert( &db.indexer.resync, &repo_key, rmp_serde::to_vec(&resync_state).into_diagnostic()?, ); } RepoStatus::Deactivated | RepoStatus::Takendown | RepoStatus::Suspended => { lifecycle_transitions.push((did.clone().into_static(), GaugeState::Resync(None))); // this shouldnt be needed since a repo wont be in a pending state when it gets to any of these states // batch.remove(&db.pending, &pending_key); let resync_state = ResyncState::Gone { status: new_status.clone(), }; batch.insert( &db.indexer.resync, &repo_key, rmp_serde::to_vec(&resync_state).into_diagnostic()?, ); } RepoStatus::Deleted => { lifecycle_transitions.push((did.clone().into_static(), GaugeState::Synced)); // terminal state: remove from queues, no resync entry needed batch.remove(&db.indexer.pending, pending_key.as_slice()); batch.remove(&db.indexer.resync, &repo_key); } RepoStatus::Desynchronized | RepoStatus::Throttled => { lifecycle_transitions.push(( did.clone().into_static(), GaugeState::Resync(Some(ResyncErrorKind::Generic)), )); // like an error: remove from pending and schedule a resync attempt batch.remove(&db.indexer.pending, pending_key.as_slice()); let resync_state = crate::types::ResyncState::Error { kind: crate::types::ResyncErrorKind::Generic, retry_count: 0, next_retry: chrono::Utc::now().timestamp(), }; batch.insert( &db.indexer.resync, &repo_key, rmp_serde::to_vec(&resync_state).into_diagnostic()?, ); } } }
repo_state.active = matches!(new_status, RepoStatus::Synced | RepoStatus::Error(_)); repo_state.status = new_status; repo_state.touch(); batch.insert(&db.repos, repo_key, db::ser_repo_state(&repo_state)?);
Ok(repo_state)}
pub struct ApplyCommitResults<'s> { pub repo_state: RepoState<'s>, #[cfg_attr(not(feature = "indexer_stream"), allow(dead_code))] pub events: RecordEvents,}
pub fn apply_commit<'s>( txn: &mut crate::db::Txn<'_>, state: &AppState, mut repo_state: RepoState<'s>, validated: ValidatedCommit<'_>, filter: &FilterConfig,) -> Result<ApplyCommitResults<'s>> { let commit = validated.commit; let parsed = validated.parsed_blocks; let did = &commit.repo; debug!(did = %did, commit = %commit.commit, "applying commit");
repo_state.root = Some(validated.commit_obj.into()); repo_state.touch();
let mut record_txn = txn.records(state, &DbTid::from(&commit.rev), did)?;
for op in &commit.ops { let (collection, rkey) = parse_path(&op.path)?;
if !filter.matches_collection(collection) { continue; }
let rkey = DbRkey::new(rkey); let action = DbAction::try_from(op.action.as_str())?;
match action { DbAction::Create | DbAction::Update => { let Some(cid) = &op.cid else { continue; }; let cid_ipld = cid .to_ipld() .into_diagnostic() .wrap_err("expected valid cid from relay")?; let Some(bytes) = parsed.blocks.get(&cid_ipld) else { return Err(miette::miette!( "block {cid} not found in CAR for record {did}/{collection}/{rkey}" )); }; record_txn.put_record(collection, &rkey, &cid_ipld, bytes, action)?; } DbAction::Delete => { record_txn.delete_record(collection, &rkey)?; } } }
record_txn.update_repo_state(&repo_state)?; let events = record_txn.finish()?;
Ok(ApplyCommitResults { repo_state, events })}
pub fn parse_path(path: &str) -> Result<(&str, &str)> { let mut parts = path.splitn(2, '/'); let collection = parts.next().wrap_err("missing collection")?; let rkey = parts.next().wrap_err("missing rkey")?;
if collection.is_empty() || jacquard_common::types::nsid::Nsid::new(collection).is_err() { miette::bail!("invalid collection NSID: {collection}"); } if rkey.is_empty() || jacquard_common::types::string::Rkey::new(rkey).is_err() { miette::bail!("invalid record key (rkey): {rkey}"); }
Ok((collection, rkey))}
#[cfg(all(test, feature = "indexer_stream"))]mod tests { use super::*; use crate::config::Config; use jacquard_common::{CowStr, types::string::Handle};
fn ephemeral(event: BroadcastEvent) -> Box<MarshallableEvt<'static>> { let BroadcastEvent::Ephemeral(event) = event else { panic!("expected ephemeral event"); }; event }
#[test] fn ephemeral_identity_and_account_json_contracts() { let tmp = tempfile::tempdir().unwrap(); let db = Db::open(&Config { database_path: tmp.path().to_path_buf(), ..Default::default() }) .unwrap(); let did = Did::new_static("did:web:example.com").unwrap();
let identity = ephemeral(make_identity_event( &db, IdentityEvt { did: did.clone(), handle: Some(Handle::new_static("alice.test").unwrap()), }, )); let identity_json = serde_json::to_value(identity.as_ref()).unwrap(); assert_eq!(identity_json["type"], "identity"); assert_eq!(identity_json["identity"]["did"], "did:web:example.com"); assert_eq!(identity_json["identity"]["handle"], "alice.test"); assert!(identity_json.get("record").is_none()); assert!(identity_json.get("account").is_none());
let account = ephemeral(make_account_event( &db, AccountEvt { did, active: false, status: Some(CowStr::Borrowed("stinkpot")), }, )); let account_json = serde_json::to_value(account.as_ref()).unwrap(); assert_eq!(account_json["id"].as_u64(), Some(identity.id + 1)); assert_eq!(account_json["type"], "account"); assert_eq!(account_json["account"]["did"], "did:web:example.com"); assert_eq!(account_json["account"]["active"], false); assert_eq!(account_json["account"]["status"], "stinkpot"); assert!(account_json.get("record").is_none()); assert!(account_json.get("identity").is_none()); }}