use fjall::OwnedWriteBatch; use fjall::Slice; #[cfg(feature = "backlinks")] use jacquard_common::Data; use jacquard_common::IntoStatic; use jacquard_common::types::did::Did; use miette::{Context, IntoDiagnostic, Result}; use std::collections::HashMap; use tracing::debug; use crate::db::types::{DbAction, DbRkey, DbTid, TrimmedDid}; use crate::db::{self, Db, 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, LiveRecordEvent, MarshallableEvt, StoredData, StoredEvent, }, jacquard_common::CowStr, std::sync::atomic::Ordering, }; #[cfg(feature = "jetstream")] use crate::types::{JetstreamBroadcast, StoredJetstreamEvent}; pub fn persist_to_resync_buffer(db: &Db, did: &Did, commit: &Commit) -> Result<()> { let key = keys::resync_buffer_key(did, DbTid::from(&commit.rev)); let value = rmp_serde::to_vec_named(commit).into_diagnostic()?; db.resync_buffer.insert(key, value).into_diagnostic()?; 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<'static>) -> BroadcastEvent { let event_id = db.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)) } #[cfg(feature = "indexer_stream")] pub fn make_account_event(db: &Db, evt: AccountEvt<'static>) -> BroadcastEvent { let event_id = db.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 fn delete_repo( batch: &mut OwnedWriteBatch, 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)?; batch.remove(&db.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 batch.remove(&db.resync, &repo_key); batch.remove(&db.repo_metadata, &metadata_key); // 2. delete from resync buffer let resync_prefix = keys::resync_buffer_prefix(did); for guard in db.resync_buffer.prefix(&resync_prefix) { let k = guard.key().into_diagnostic()?; batch.remove(&db.resync_buffer, k); } // 3. delete from records // todo: figure out how we want to handle the blocks associated with these records // without delving too much into gc madness we had before let records_prefix = keys::record_prefix_did(did); for guard in db.records.prefix(&records_prefix) { let (k, _cid_bytes) = guard.into_inner().into_diagnostic()?; batch.remove(&db.records, k); } // 4. reset collection counts let mut count_prefix = Vec::new(); count_prefix.push(b'r'); count_prefix.push(keys::SEP); TrimmedDid::from(did).write_to_vec(&mut count_prefix); count_prefix.push(keys::SEP); for guard in db.counts.prefix(&count_prefix) { let k = guard.key().into_diagnostic()?; batch.remove(&db.counts, k); } // 5. remove backlinks for all records in this repo #[cfg(feature = "backlinks")] crate::backlinks::store::delete_repo(batch, &db.backlinks, did)?; Ok(()) } pub fn transition_repo<'s>( batch: &mut OwnedWriteBatch, db: &Db, lifecycle_transitions: &mut Vec<(Did<'static>, GaugeState)>, did: &Did, mut repo_state: RepoState<'s>, new_status: RepoStatus, ) -> Result> { 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.pending, pending_key.as_slice()); // we dont have to remove from resync here because it has to transition resync -> pending first } RepoStatus::Error(msg) => { tracing::warn!("transitioning to error: {msg}"); lifecycle_transitions.push(( did.clone().into_static(), GaugeState::Resync(Some(ResyncErrorKind::Generic)), )); batch.remove(&db.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.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.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.pending, pending_key.as_slice()); batch.remove(&db.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.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.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(); Ok(repo_state) } pub struct ApplyCommitResults<'s> { pub repo_state: RepoState<'s>, pub records_delta: i64, pub blocks_count: i64, #[cfg(feature = "indexer_stream")] pub live_events: Vec, #[cfg(feature = "indexer_stream")] pub last_event_id: Option, #[cfg(feature = "jetstream")] pub jetstream_events: Vec, } pub fn apply_commit<'s>( batch: &mut OwnedWriteBatch, state: &AppState, mut repo_state: RepoState<'s>, validated: ValidatedCommit<'_>, filter: &FilterConfig, ) -> Result> { let db = &state.db; let ephemeral = state.ephemeral; let only_index_links = state.only_index_links; 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(); // 2. iterate ops and update records index let mut records_delta = 0; let mut blocks_count = 0; let mut collection_deltas: HashMap<&str, i64> = HashMap::new(); #[cfg(feature = "indexer_stream")] let rev = DbTid::from(&commit.rev); #[cfg(feature = "indexer_stream")] let should_broadcast_live = db.event_tx.receiver_count() > 0; #[cfg(feature = "indexer_stream")] let mut live_events = Vec::new(); #[cfg(feature = "indexer_stream")] let mut last_event_id = None; #[cfg(feature = "jetstream")] let mut jetstream_events = Vec::new(); for op in &commit.ops { let (collection, rkey) = parse_path(&op.path)?; if !filter.matches_collection(collection) { continue; } let rkey = DbRkey::new(rkey); let db_key = keys::record_key(did, collection, &rkey); let action = DbAction::try_from(op.action.as_str())?; #[cfg(feature = "indexer_stream")] let mut cid_for_event: Option = None; #[cfg(feature = "indexer_stream")] let mut block_inline_for_event: Option = None; #[cfg(feature = "indexer_stream")] let mut inline_block: Option = None; 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")?; #[cfg(feature = "indexer_stream")] { cid_for_event = Some(cid_ipld); } let Some(bytes) = parsed.blocks.get(&cid_ipld) else { return Err(miette::miette!( "block {cid} not found in CAR for record {did}/{collection}/{rkey}" )); }; let cid_raw = cid_ipld.to_bytes(); let block_key = Slice::from(keys::block_key(collection, &cid_raw)); blocks_count += 1; if !ephemeral { if !only_index_links { batch.insert(&db.blocks, block_key.clone(), bytes.as_ref()); } batch.insert(&db.records, db_key.clone(), cid_raw); // accumulate counts if action == DbAction::Create { records_delta += 1; *collection_deltas.entry(collection).or_default() += 1; } #[cfg(feature = "backlinks")] if let Ok(value) = serde_ipld_dagcbor::from_slice::(bytes.as_ref()) { crate::backlinks::store::index_record( batch, &db.backlinks, did.as_str(), collection, &rkey.to_smolstr(), &value, )?; } #[cfg(feature = "indexer_stream")] if should_broadcast_live && !only_index_links { // inline record bytes for live tailing so we don't have to load from blocks. inline_block = Some(bytes.clone()); } } else { #[cfg(feature = "indexer_stream")] { // in ephemeral mode, the event payload is the only place we persist the record. block_inline_for_event = Some(bytes.clone()); } } } DbAction::Delete => { if !ephemeral { batch.remove(&db.records, db_key); // accumulate counts records_delta -= 1; *collection_deltas.entry(collection).or_default() -= 1; #[cfg(feature = "backlinks")] crate::backlinks::store::delete_record( batch, &db.backlinks, did.as_str(), collection, &rkey.to_smolstr(), )?; } } }; #[cfg(feature = "indexer_stream")] { let data = block_inline_for_event .clone() .map(StoredData::Block) .or_else(|| { (!only_index_links) .then(|| cid_for_event.map(StoredData::Ptr)) .flatten() }) .unwrap_or(StoredData::Nothing); let event_id = db.next_event_id.fetch_add(1, Ordering::SeqCst); last_event_id = Some(event_id); let did_trimmed = TrimmedDid::from(did); let collection = CowStr::Borrowed(collection); let evt = StoredEvent { live: true, did: did_trimmed.clone(), rev, collection: collection.clone(), rkey: rkey.clone(), action, data: data.clone(), }; let bytes = rmp_serde::to_vec(&evt).into_diagnostic()?; batch.insert(&db.events, keys::event_key(event_id), bytes); #[cfg(feature = "jetstream")] { let jetstream = StoredJetstreamEvent::Commit { did: did_trimmed.clone().into_static(), collection: collection.clone().into_static(), event_id, live: true, }; let ephemeral = crate::jetstream::build_ephemeral_from_stored( did_trimmed.to_did().as_str(), rev.to_tid().as_str(), action.as_str(), collection.as_str(), rkey.to_smolstr().as_str(), &data, inline_block.as_ref(), true, ); jetstream_events.push(crate::jetstream::stage_event( batch, db, jetstream, ephemeral, )?); } if should_broadcast_live { live_events.push(LiveRecordEvent { id: event_id, stored: StoredEvent { live: evt.live, did: did_trimmed.into_static(), rev: evt.rev, collection: collection.into_static(), rkey, action: evt.action, data, }, inline_block, }); } } } // update counts if !ephemeral { for (col, delta) in collection_deltas { db::update_record_count(batch, db, did, col, delta)?; } } Ok(ApplyCommitResults { repo_state, records_delta, blocks_count, #[cfg(feature = "indexer_stream")] live_events, #[cfg(feature = "indexer_stream")] last_event_id, #[cfg(feature = "jetstream")] jetstream_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")?; Ok((collection, rkey)) }