//! per-op stream/jetstream event emission for `apply_commit`.
//!
//! when `indexer_stream` is enabled the emitter stages a `StoredEvent` per
//! record op, plus its jetstream event, in the transaction's outbox; without
//! it the emitter is a zero-sized no-op, so `apply_commit` stays unconditional.
#[cfg(feature = "indexer_stream")]
mod enabled {
use bytes::Bytes;
use jacquard_common::CowStr;
use jacquard_common::IntoStatic;
use jacquard_common::types::cid::IpldCid;
use jacquard_common::types::did::Did;
use miette::{IntoDiagnostic, Result};
use crate::db::outbox::Outbox;
use crate::db::types::{DbAction, DbRkey, DbTid, TrimmedDid};
use crate::state::AppState;
use crate::types::{StoredData, StoredEvent};
/// one record op, as seen by the event emitter.
pub(crate) struct EmitOp<'a> {
pub(crate) did: &'a Did,
pub(crate) collection: &'a str,
pub(crate) rkey: &'a DbRkey,
pub(crate) action: DbAction,
/// record cid for create/update ops
pub(crate) cid: Option<IpldCid>,
/// raw record block for create/update ops
pub(crate) block: Option<&'a Bytes>,
}
#[derive(Clone, Copy)]
pub(crate) enum RecordEventOrigin {
Live,
Backfill,
}
pub(crate) struct RecordEmitter {
rev: DbTid,
ephemeral: bool,
only_index_links: bool,
live: bool,
should_broadcast_live: bool,
#[cfg(feature = "jetstream")]
should_stage_jetstream: bool,
}
impl RecordEmitter {
pub(crate) fn new(state: &AppState, commit_rev: &DbTid, origin: RecordEventOrigin) -> Self {
let live = matches!(origin, RecordEventOrigin::Live);
Self {
rev: *commit_rev,
ephemeral: state.ephemeral,
only_index_links: state.only_index_links,
live,
should_broadcast_live: live && state.db.stream.event_tx.receiver_count() > 0,
#[cfg(feature = "jetstream")]
should_stage_jetstream: live && state.db.jetstream.tx.receiver_count() > 0,
}
}
pub(crate) fn emit(&mut self, outbox: &mut Outbox, op: EmitOp<'_>) -> Result<()> {
// ephemeral events own their body bytes; event TTL deletes both
// atomically. permanent live tails can borrow the committing body
// while durable replay resolves its pointer from record storage.
let block_inline_for_event = op.block.filter(|_| self.ephemeral).cloned();
let inline_block = op
.block
.filter(|_| !self.ephemeral && self.should_broadcast_live && !self.only_index_links)
.cloned();
let data = block_inline_for_event
.map(StoredData::Block)
.or_else(|| {
op.cid
.filter(|_| !self.only_index_links)
.map(StoredData::Ptr)
})
.unwrap_or(StoredData::Nothing);
let did_trimmed = TrimmedDid::from(op.did);
let collection = CowStr::Borrowed(op.collection);
let evt = StoredEvent {
live: self.live,
did: did_trimmed.clone(),
rev: self.rev,
collection: collection.clone(),
rkey: op.rkey.clone(),
action: op.action,
data: data.clone(),
};
let row = rmp_serde::to_vec(&evt).into_diagnostic()?;
let live = self.should_broadcast_live.then(|| {
let stored = StoredEvent {
live: evt.live,
did: did_trimmed.clone().into_static(),
rev: evt.rev,
collection: collection.clone().into_static(),
rkey: op.rkey.clone(),
action: evt.action,
data: data.clone(),
};
(stored, inline_block.clone())
});
let event = outbox.stream.push_record(row, live);
#[cfg(feature = "jetstream")]
{
let json = self.should_stage_jetstream.then_some(()).and_then(|()| {
crate::jetstream::build_ephemeral_from_stored(
&did_trimmed,
self.rev.to_tid().as_str(),
op.action.as_str(),
collection.as_str(),
op.rkey,
&data,
inline_block.as_ref(),
self.live,
)
});
outbox.jetstream.push(
crate::types::StoredJetstreamEvent::Commit {
did: did_trimmed.into_static(),
collection: collection.into_static(),
event_id: event,
live: self.live,
},
json,
);
}
#[cfg(not(feature = "jetstream"))]
let _ = event;
Ok(())
}
}
}
#[cfg(feature = "indexer_stream")]
pub(crate) use enabled::*;
#[cfg(not(feature = "indexer_stream"))]
mod noop {
use bytes::Bytes;
use jacquard_common::types::cid::IpldCid;
use jacquard_common::types::did::Did;
use miette::Result;
use crate::db::outbox::Outbox;
use crate::db::types::{DbAction, DbRkey};
use crate::state::AppState;
#[derive(Clone, Copy)]
pub(crate) enum RecordEventOrigin {
Live,
Backfill,
}
#[allow(dead_code)]
pub(crate) struct EmitOp<'a> {
pub(crate) did: &'a Did,
pub(crate) collection: &'a str,
pub(crate) rkey: &'a DbRkey,
pub(crate) action: DbAction,
pub(crate) cid: Option<IpldCid>,
pub(crate) block: Option<&'a Bytes>,
}
pub(crate) struct RecordEmitter;
impl RecordEmitter {
#[inline(always)]
pub(crate) fn new(
_state: &AppState,
_commit_rev: &crate::db::types::DbTid,
_origin: RecordEventOrigin,
) -> Self {
Self
}
#[inline(always)]
pub(crate) fn emit(&mut self, _outbox: &mut Outbox, _op: EmitOp<'_>) -> Result<()> {
Ok(())
}
}
}
#[cfg(not(feature = "indexer_stream"))]
pub(crate) use noop::*;