//! sink for builds with neither indexer nor relay: validated events update
//! repo state but are not forwarded anywhere.
use fjall::OwnedWriteBatch;
use miette::Result;
use url::Url;
use crate::ingest::stream::{Account, Commit, Identity, Sync};
use crate::state::AppState;
use crate::types::RepoState;
/// nothing consumes the events, so nothing ever moved the cursor of a bare build
pub(crate) const PUBLISHES_WATERMARK: bool = false;
/// per-worker seed from which each shard builds its sink.
#[derive(Clone, Default)]
pub(crate) struct SinkSeed;
pub(crate) struct EventSink;
pub(crate) struct Staged;
pub(crate) struct AccountSnapshot;
impl EventSink {
pub(crate) fn new(
_seed: &SinkSeed,
_stats: std::sync::Arc<crate::ingest::firehose_stats::RelayShardStats>,
) -> Self {
Self
}
/// nothing downstream marks frames done, so the relay keeps every ticket
pub(crate) fn hand_off(
&mut self,
ticket: crate::ingest::cursor::Ticket,
) -> Option<crate::ingest::cursor::Ticket> {
Some(ticket)
}
pub(crate) fn discard(&mut self) {}
pub(crate) fn account_snapshot(&self, _repo_state: &RepoState) -> AccountSnapshot {
AccountSnapshot
}
pub(crate) fn commit(
&mut self,
_state: &AppState,
_batch: &mut OwnedWriteBatch,
_firehose: &Url,
_commit: Commit<'static>,
_chain_break: bool,
_parsed_blocks: crate::car::CommitCar,
) -> Result<()> {
Ok(())
}
pub(crate) fn sync(
&mut self,
_state: &AppState,
_batch: &mut OwnedWriteBatch,
_firehose: &Url,
_sync: Sync<'static>,
) -> Result<()> {
Ok(())
}
pub(crate) fn identity(
&mut self,
_state: &AppState,
_batch: &mut OwnedWriteBatch,
_firehose: &Url,
_identity: Identity,
_changed: bool,
_own_frame: bool,
) -> Result<()> {
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub(crate) fn account(
&mut self,
_state: &AppState,
_batch: &mut OwnedWriteBatch,
_firehose: &Url,
_account: Account<'static>,
_repo_state: &RepoState,
_snapshot: AccountSnapshot,
_was_active: bool,
) -> Result<()> {
Ok(())
}
/// puts the row the message staged on `txn`, the relay shard being its only writer here
pub(crate) fn stage_row(
&mut self,
state: &AppState,
txn: &mut crate::db::Txn<'_>,
did: &jacquard_common::types::did::Did,
_loaded: Option<RepoState<'static>>,
staged: Option<RepoState<'static>>,
_firehose: Option<&Url>,
) -> Result<()> {
match staged {
Some(row) => super::put_row(state, txn, did, &row),
None => Ok(()),
}
}
pub(crate) fn commit_txn(
&mut self,
_state: &AppState,
txn: crate::db::Txn<'_>,
) -> Result<(Staged, crate::db::TxnCommitTimings)> {
let (_, timings) = txn.commit_with(|_| Ok(()))?;
Ok((Staged, timings))
}
pub(crate) fn flush(&mut self, _state: &AppState, _staged: Staged) {}
}