use fjall::OwnedWriteBatch;
use miette::Result;
#[cfg(feature = "jetstream")]
use smol_str::ToSmolStr;
use url::Url;
use crate::db::outbox::{Outbox, RelayFrameRef, RelayOutbox};
#[cfg(feature = "jetstream")]
use crate::db::types::TrimmedDid;
use crate::ingest::stream::{Account, Commit, Identity, Sync};
use crate::state::AppState;
use crate::types::RepoState;
#[cfg(feature = "jetstream")]
use crate::types::StoredJetstreamEvent;
#[cfg(feature = "jetstream")]
use jacquard_common::CowStr;
/// the relay is the last stage, so a frame is done once its relay commit lands, and the
/// watermark is the source's cursor
pub(crate) const PUBLISHES_WATERMARK: bool = true;
/// per-worker seed from which each shard builds its sink.
#[derive(Clone, Default)]
pub(crate) struct SinkSeed;
/// re-sequences validated events and re-emits them on the relay stream,
/// plus jetstream events when that feature is enabled.
pub(crate) struct EventSink {
stats: std::sync::Arc<crate::ingest::firehose_stats::RelayShardStats>,
/// the current message's frames, moved into its transaction on commit
outbox: Outbox,
}
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 {
stats,
outbox: Outbox::default(),
}
}
/// 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) {
self.outbox = Outbox::default();
}
pub(crate) fn account_snapshot(&self, _repo_state: &RepoState) -> AccountSnapshot {
AccountSnapshot
}
/// queue a frame; it gets its seq, row and broadcast when the message's
/// transaction commits.
fn queue_emit(
&mut self,
push: impl FnOnce(&mut RelayOutbox) -> RelayFrameRef,
) -> RelayFrameRef {
let started = crate::ingest::firehose_stats::StatsInstant::now();
let frame = push(&mut self.outbox.relay);
self.stats.record_queue_emit(started.elapsed());
frame
}
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<()> {
#[cfg(feature = "jetstream")]
let jetstream = {
// skip car parse and record materialization when no jetstream subscribers are
// connected — the stream thread re-reads from relay_events as a fallback.
let has_subscribers = state.db.jetstream.tx.receiver_count() > 0;
let did = TrimmedDid::from(&commit.repo).into_static();
commit
.ops
.iter()
.enumerate()
.filter(|(_, op)| matches!(op.action.as_str(), "create" | "update" | "delete"))
.filter_map(|(idx, op)| Some((idx as u32, split_collection(&op.path)?)))
.map(|(op_index, collection)| {
let json = has_subscribers.then_some(()).and_then(|()| {
build_relay_commit_ephemeral(&commit, op_index, &collection, &parsed_blocks)
});
(did.clone(), collection, op_index, json)
})
.collect::<Vec<_>>()
};
#[cfg(not(feature = "jetstream"))]
let _ = (state, &parsed_blocks);
let _frame = self.queue_emit(|relay| relay.push_commit(commit));
#[cfg(feature = "jetstream")]
for (did, collection, op_index, json) in jetstream {
self.outbox.jetstream.push(
StoredJetstreamEvent::RelayCommit {
did,
collection,
relay_seq: _frame,
op_index,
},
json,
);
}
Ok(())
}
pub(crate) fn sync(
&mut self,
_state: &AppState,
_batch: &mut OwnedWriteBatch,
_firehose: &Url,
sync: Sync<'static>,
) -> Result<()> {
self.queue_emit(|relay| relay.push_sync(sync));
Ok(())
}
pub(crate) fn identity(
&mut self,
_state: &AppState,
_batch: &mut OwnedWriteBatch,
_firehose: &Url,
identity: Identity,
_changed: bool,
_own_frame: bool,
) -> Result<()> {
#[cfg(feature = "jetstream")]
let did = TrimmedDid::from(&identity.did).into_static();
let _frame = self.queue_emit(|relay| relay.push_identity(identity));
#[cfg(feature = "jetstream")]
self.outbox.jetstream.push(
StoredJetstreamEvent::RelayIdentity {
did,
relay_seq: _frame,
},
None,
);
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<()> {
#[cfg(feature = "jetstream")]
let did = TrimmedDid::from(&account.did).into_static();
let _frame = self.queue_emit(|relay| relay.push_account(account));
#[cfg(feature = "jetstream")]
self.outbox.jetstream.push(
StoredJetstreamEvent::RelayAccount {
did,
relay_seq: _frame,
},
None,
);
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,
mut txn: crate::db::Txn<'_>,
) -> Result<(Staged, crate::db::TxnCommitTimings)> {
txn.outbox = std::mem::take(&mut self.outbox);
let (_, timings) = txn.commit_with(|_| Ok(()))?;
Ok((Staged, timings))
}
/// frames are broadcast as their transaction commits.
pub(crate) fn flush(&mut self, _state: &AppState, _staged: Staged) {}
}
#[cfg(feature = "jetstream")]
fn split_collection(path: &str) -> Option<CowStr<'static>> {
path.split_once('/')
.map(|(collection, _)| CowStr::Owned(collection.to_smolstr()))
}
#[cfg(feature = "jetstream")]
fn build_relay_commit_ephemeral(
commit: &Commit,
op_index: u32,
collection: &CowStr<'static>,
car: &crate::car::CommitCar,
) -> Option<crate::jetstream::JetstreamEphemeral> {
let op = commit.ops.get(op_index as usize)?;
let (_, rkey) = op.path.split_once('/')?;
let action = op.action.as_str();
let (record, cid) = if matches!(action, "create" | "update") {
let cid_link = op.cid.as_ref()?;
let cid_ipld = cid_link.to_ipld().ok()?;
let block = car.blocks.get(&cid_ipld)?;
let record = crate::types::record_json(block).ok()?;
(Some(record), Some(cid_link.to_string()))
} else {
(None, None)
};
Some(crate::jetstream::JetstreamEphemeral {
did: commit.repo.as_str().to_string(),
rev: commit.rev.as_str().to_string(),
operation: action.to_string(),
collection: collection.as_str().to_string(),
rkey: rkey.to_string(),
record,
cid,
live: true,
})
}
#[cfg(all(test, feature = "jetstream"))]
mod tests {
use std::collections::BTreeMap;
use jacquard_common::{
CowStr,
types::{
cid::CidLink,
string::{Did, Tid},
},
};
use super::*;
use crate::ingest::stream::{Datetime, RepoOp, RepoOpAction};
#[test]
fn relay_ephemeral_projection_preserves_record_representation() {
let body = serde_ipld_dagcbor::to_vec(&serde_json::json!({
"$type": "app.bsky.feed.post",
"text": "stinkpot",
}))
.unwrap();
let cid = crate::car::compute_cid(&body);
let cid_link = CidLink::from(cid);
let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
let rev: Tid = "3kzbif5moe22m".parse().unwrap();
let commit = Commit {
blobs: Vec::new(),
blocks: bytes::Bytes::new(),
commit: cid_link.clone(),
ops: vec![RepoOp {
action: RepoOpAction::Create,
cid: Some(cid_link),
path: CowStr::Borrowed("app.bsky.feed.post/stinkpot"),
prev: None,
}],
prev_data: None,
rebase: false,
repo: did,
rev: rev.clone(),
seq: 42,
since: None,
time: Datetime::try_from("2026-08-10T12:34:56Z".to_owned()).unwrap(),
too_big: false,
};
let car = crate::car::CommitCar {
root: cid,
blocks: crate::car::Blocks::unverified(BTreeMap::from([(cid, body.into())])),
};
let event =
build_relay_commit_ephemeral(&commit, 0, &CowStr::Borrowed("app.bsky.feed.post"), &car)
.unwrap();
assert_eq!(event.did, commit.repo.as_str());
assert_eq!(event.rev, rev.as_str());
assert_eq!(event.operation, "create");
assert_eq!(event.collection, "app.bsky.feed.post");
assert_eq!(event.rkey, "stinkpot");
assert_eq!(event.cid.as_deref(), Some(cid.to_string().as_str()));
assert_eq!(
serde_json::from_str::<serde_json::Value>(event.record.unwrap().get()).unwrap(),
serde_json::json!({
"$type": "app.bsky.feed.post",
"text": "stinkpot",
})
);
}
}