use 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::{db::outbox::StreamOutbox, types::IdentityEvt},
jacquard_common::types::string::Handle,
tracing::error,
};
pub(crate) mod backlink_ops;
pub mod record_events;
pub fn persist_to_resync_buffer(
txn: &mut Txn<'_>,
db: &Db,
did: &Did,
commit: &Commit,
) -> Result<()> {
if txn.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(())
}
/// drops what `did` buffered at or below `rev`, for a backfill writing a car at `rev`. the car
/// already has those commits, and the drain applying one on top would roll the car back
pub fn stage_buffer_covered_by(
batch: &mut OwnedWriteBatch,
db: &Db,
did: &Did,
rev: DbTid,
) -> Result<()> {
let covered = keys::resync_buffer_prefix(did)..=keys::resync_buffer_key(did, rev);
for guard in db.indexer.resync_buffer.range(covered) {
batch.remove(&db.indexer.resync_buffer, guard.key().into_diagnostic()?);
}
Ok(())
}
/// broadcast identity and account events staged by `stage`. these are
/// ephemeral: they take stream ids but are never persisted or replayed, so
/// consumers that need current state fetch it themselves.
#[cfg(feature = "indexer_stream")]
pub async fn announce(db: &Db, stage: impl FnOnce(&mut StreamOutbox) + Send + 'static) {
let result = db
.run(move |db| {
let mut txn = Txn::new(db);
stage(&mut txn.outbox.stream);
txn.commit()
})
.await;
if let Err(e) = result {
error!(err = %e, "failed to announce stream events");
}
}
/// announce an identity the caller learned by re-resolving a doc rather than by
/// receiving an event.
#[cfg(feature = "indexer_stream")]
pub async fn announce_identity(db: &Db, did: &Did, handle: Option<Handle>) {
let identity = IdentityEvt {
did: did.clone().into_static(),
handle,
};
announce(db, move |events| events.push_identity(identity)).await;
}
/// the first txn of a deleted account's wipe: takes the repo off every queue and forgets what
/// it buffered, then leaves the marker its owner shard moves the heads to history from, a
/// step at a time. a running erase's marker stays, since the erase forgets the heads anyway
pub(crate) fn start_wipe(
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)?;
db.indexer
.stage_pending_remove(&mut txn.batch, did, &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);
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);
}
if matches!(
db::read_bulk_job(db, did)?,
Some(crate::types::BulkJobV1::Erase)
) {
return Ok(());
}
// the death rev sorts after every rev we have seen, including future-skewed ones, and the
// marker keeps it so every step files the heads under the same one
let death_rev = db::death_rev(repo_state.root.as_ref().map(|c| &c.rev));
db::stage_bulk_job(
&mut txn.batch,
db,
did,
&crate::types::BulkJobV1::Wipe { death_rev },
)
}
pub fn transition_repo<'s>(
txn: &mut Txn<'_>,
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 db = txn.db;
let batch = &mut txn.batch;
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 mut metadata = db::deser_repo_meta(&metadata_bytes)?;
let pending_key = keys::pending_key(metadata.index_id);
// the retry worker only requeues these, any other status would strand a deferred refetch
let retries = repo_state.active
&& matches!(
new_status,
RepoStatus::Error(_) | RepoStatus::Desynchronized | RepoStatus::Throttled
);
if !retries && std::mem::take(&mut metadata.refetch_deferred) {
batch.insert(
&db.repo_metadata,
&metadata_key,
db::ser_repo_meta(&metadata)?,
);
}
// manage queues
match &new_status {
RepoStatus::Synced => {
lifecycle_transitions.push((did.clone().into_static(), GaugeState::Synced));
db.indexer.stage_pending_remove(batch, did, &pending_key);
batch.remove(&db.indexer.resync, &repo_key);
}
// the relay writes these for inactive accounts too, and a pds won't hand those over
// any more than a deactivated one, so a retry would only refetch into the same answer
RepoStatus::Error(_) | RepoStatus::Desynchronized | RepoStatus::Throttled
if !repo_state.active =>
{
stage_gone(batch, db, lifecycle_transitions, did, &new_status)?;
}
RepoStatus::Error(msg) => {
tracing::warn!("transitioning to error: {msg}");
lifecycle_transitions.push((
did.clone().into_static(),
GaugeState::Resync(Some(ResyncErrorKind::Generic)),
));
db.indexer.stage_pending_remove(batch, did, &pending_key);
// 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 => {
stage_gone(batch, db, lifecycle_transitions, did, &new_status)?;
}
RepoStatus::Deleted => {
lifecycle_transitions.push((did.clone().into_static(), GaugeState::Synced));
// terminal state: remove from queues, no resync entry needed
db.indexer.stage_pending_remove(batch, did, &pending_key);
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
db.indexer.stage_pending_remove(batch, did, &pending_key);
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 = match new_status {
RepoStatus::Synced => true,
RepoStatus::Error(_) => repo_state.active,
_ => false,
};
repo_state.status = new_status;
repo_state.touch();
txn.put_repo(did, &repo_state)?;
Ok(repo_state)
}
/// parks a repo whose pds won't hand it over until its account comes back. the queue is the
/// caller's to clear
fn stage_gone(
batch: &mut fjall::OwnedWriteBatch,
db: &Db,
lifecycle_transitions: &mut Vec<(Did, GaugeState)>,
did: &Did,
status: &RepoStatus,
) -> Result<()> {
lifecycle_transitions.push((did.clone().into_static(), GaugeState::Resync(None)));
let resync_state = ResyncState::Gone {
status: status.clone(),
};
batch.insert(
&db.indexer.resync,
keys::repo_key(did),
rmp_serde::to_vec(&resync_state).into_diagnostic()?,
);
Ok(())
}
pub struct ApplyCommitResults<'s> {
pub repo_state: RepoState<'s>,
}
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");
// the relay puts each commit's root on the row before it gets here, so the row can already
// be a commit or more ahead, and the relay checks the next commit's since against it
let root: crate::types::Commit = validated.commit_obj.into();
if repo_state
.root
.as_ref()
.is_none_or(|current| current.rev < root.rev)
{
repo_state.root = Some(root);
}
repo_state.touch();
// the shard commits its batch even when this fails, so everything the commit itself can get
// wrong is checked before the first write
let mut writes = Vec::with_capacity(commit.ops.len());
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())?;
let block = 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(block) = parsed.blocks.block(&cid_ipld) else {
return Err(miette::miette!(
"block {cid} not found in CAR for record {did}/{collection}/{rkey}"
));
};
let block = block
.checked()
.map_err(|e| miette::miette!("record {did}/{collection}/{rkey}: {e}"))?;
Some(block)
}
DbAction::Delete => None,
};
writes.push((collection, rkey, action, block));
}
let mut record_txn = txn.records(state, &DbTid::from(&commit.rev), did)?;
for (collection, rkey, action, block) in &writes {
match block {
Some(block) => record_txn.put_record(collection, rkey, block, *action)?,
None => record_txn.delete_record(collection, rkey)?,
}
}
record_txn.update_repo_state(&repo_state)?;
Ok(ApplyCommitResults { repo_state })
}
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 crate::types::{AccountEvt, BroadcastEvent, MarshallableEvt};
use jacquard_common::{CowStr, types::string::Handle};
use tokio::sync::broadcast;
fn next_ephemeral(
rx: &mut broadcast::Receiver<BroadcastEvent>,
) -> Box<MarshallableEvt<'static>> {
let Ok(BroadcastEvent::Ephemeral(event)) = rx.try_recv() else {
panic!("expected an ephemeral event");
};
event
}
#[tokio::test]
async 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 mut rx = db.stream.event_tx.subscribe();
let did = Did::new_static("did:web:example.com").unwrap();
announce_identity(&db, &did, Some(Handle::new_static("alice.test").unwrap())).await;
let identity = next_ephemeral(&mut rx);
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 = AccountEvt {
did,
active: false,
status: Some(CowStr::Borrowed("stinkpot")),
};
announce(&db, move |events| events.push_account(account)).await;
let account = next_ephemeral(&mut rx);
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());
// ephemeral events take ids but are never persisted
assert_eq!(db.stream.ids.head(), Some(account.id));
assert!(db.stream.event_head().unwrap().is_none());
}
}