use bytes::Bytes;
use std::sync::Arc;
use tracing::error;
use crate::db::keys;
use crate::state::AppState;
use crate::types::{JetstreamBroadcast, JetstreamPosition, StoredJetstreamEvent};
#[cfg(feature = "relay")]
use crate::ingest::stream::{SubscribeReposMessage, decode_frame};
#[cfg(feature = "indexer_stream")]
use crate::types::StoredEvent;
use super::{ReplayChunk, ReplayFailed, Start, StreamOptions, StreamSource, run_stream};
use crate::control::{JetstreamFilter, JetstreamStreamError};
#[cfg(feature = "indexer_stream")]
use super::indexer::stored_to_event;
pub(crate) fn jetstream_stream_thread(
state: Arc<AppState>,
tx: tokio::sync::mpsc::Sender<Result<Bytes, JetstreamStreamError>>,
cursor: Option<i64>,
filter: JetstreamFilter,
opts: StreamOptions,
) {
let event_rx = state.db.jetstream.tx.subscribe();
let head = match state.db.jetstream.head() {
Ok(head) => head,
Err(e) => {
error!(err = %e, "failed to read the Jetstream head");
return;
}
};
// a cursor past the head means live, like a future cursor upstream. the
// head decides rather than the wall clock, which committed time_us can
// run ahead of.
let replay_from = cursor
.map(|cursor| cursor.max(0) as u64)
.filter(|cursor| head.is_some_and(|head| *cursor <= head.time_us));
let start = match replay_from {
Some(cursor) => Start::replay(JetstreamPosition::before(cursor), head),
None => Start::live(head),
};
let source = JetstreamSource {
state,
filter,
head,
};
run_stream(tx, event_rx, start, opts, source);
}
struct JetstreamSource {
state: Arc<AppState>,
filter: JetstreamFilter,
/// the head when the subscriber connected. rows after it belong to the
/// live tail, even when a lag has them read from the db.
head: Option<JetstreamPosition>,
}
impl JetstreamSource {
/// whether a stored row goes out: what the filter wants, except backfills
/// past the head for subscribers that don't tail them.
fn sends(&self, position: JetstreamPosition, event: &StoredJetstreamEvent<'_>) -> bool {
self.filter.wants(event)
&& (event.is_live() || Some(position) <= self.head || self.filter.tails_historical())
}
}
impl StreamSource for JetstreamSource {
type Broadcast = JetstreamBroadcast;
type Output = Bytes;
fn read(
&mut self,
after: Option<JetstreamPosition>,
through: Option<JetstreamPosition>,
limit: usize,
) -> Result<ReplayChunk<Bytes, JetstreamPosition>, ReplayFailed> {
read_jetstream_replay_chunk(&self.state, after, through, limit, |position, event| {
self.sends(position, event)
})
}
fn written(&self) -> u64 {
self.state.db.jetstream.ids.next()
}
fn frame_bytes(frame: &Bytes) -> usize {
frame.len()
}
fn render(&mut self, event: JetstreamBroadcast) -> Option<Bytes> {
let JetstreamBroadcast::Live(live) = event else {
return None;
};
if !self.filter.wants(&live.event) {
return None;
}
// json built at the commit goes out as it is, even past a redaction that completed
// since, see the redaction note in docs/api/jetstream.md
live.json.or_else(|| {
stored_event_to_bytes(&self.state, live.position.time_us as i64, &live.event)
})
}
/// markers only cover backfilled commits, which the tail sends only to
/// subscribers that asked for them, so the rest skip the read.
fn wants_marker(&self, _marker: &JetstreamBroadcast) -> bool {
self.filter.tails_historical()
}
}
fn read_jetstream_replay_chunk(
state: &AppState,
after: Option<JetstreamPosition>,
through: Option<JetstreamPosition>,
limit: usize,
sends: impl Fn(JetstreamPosition, &StoredJetstreamEvent<'_>) -> bool,
) -> Result<ReplayChunk<Bytes, JetstreamPosition>, ReplayFailed> {
let start = after.map_or([0; 16], JetstreamPosition::key_after);
let rows = match through {
Some(through) => state.db.jetstream.events.range(start..=through.key()),
None => state.db.jetstream.events.range(start..),
};
ReplayChunk::read(
rows,
after,
limit,
|key| {
let (time_us, id) = keys::parse_jetstream_event_key(key)
.inspect_err(|e| error!(err = %e, "failed to parse Jetstream event key"))
.ok()?;
Some(JetstreamPosition { time_us, id })
},
|position, value| {
let event = rmp_serde::from_slice::<StoredJetstreamEvent>(value)
.inspect_err(|e| error!(err = %e, "failed to deserialize Jetstream event"))
.ok()?;
if !sends(position, &event) {
return None;
}
stored_event_to_bytes(state, position.time_us as i64, &event)
},
)
}
#[derive(serde::Serialize)]
pub(crate) struct JetstreamEvent<'a> {
pub(crate) did: &'a str,
pub(crate) time_us: i64,
#[serde(flatten)]
pub(crate) payload: JetstreamPayload<'a>,
}
#[derive(serde::Serialize)]
#[serde(tag = "kind", rename_all = "lowercase")]
pub(crate) enum JetstreamPayload<'a> {
Commit {
commit: JetstreamCommit<'a>,
},
Identity {
#[serde(borrow)]
identity: JetstreamIdentity<'a>,
},
Account {
#[serde(borrow)]
account: JetstreamAccount<'a>,
},
}
#[derive(serde::Serialize)]
pub(crate) struct JetstreamCommit<'a> {
pub(crate) rev: &'a str,
pub(crate) operation: &'a str,
pub(crate) collection: &'a str,
pub(crate) rkey: &'a str,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) record: Option<&'a serde_json::value::RawValue>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) cid: Option<String>,
pub(crate) live: bool,
}
#[derive(serde::Serialize)]
pub(crate) struct JetstreamIdentity<'a> {
pub(crate) did: String,
pub(crate) seq: i64,
pub(crate) time: &'a crate::ingest::stream::Datetime,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) handle: Option<String>,
}
#[derive(serde::Serialize)]
pub(crate) struct JetstreamAccount<'a> {
pub(crate) active: bool,
pub(crate) did: String,
pub(crate) seq: i64,
pub(crate) time: &'a crate::ingest::stream::Datetime,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) status: Option<String>,
}
/// render a stored event from the database, for a replay or a live event
/// broadcast without its json.
fn stored_event_to_bytes(
state: &AppState,
time_us: i64,
event: &StoredJetstreamEvent<'_>,
) -> Option<Bytes> {
match event {
#[cfg(feature = "indexer_stream")]
StoredJetstreamEvent::Commit { event_id, live, .. } => {
let bytes = state.db.stream.event(keys::event_key(*event_id)).ok()??;
let stored: StoredEvent = rmp_serde::from_slice(&bytes).ok()?;
let evt = stored_to_event(state, *event_id, stored, None)?;
let rec = evt.record?;
let did_str = rec.did.as_str();
// serialize the record to raw json so it embeds without re-encoding.
let record_raw = rec.record.as_deref();
let json_event = JetstreamEvent {
did: did_str,
time_us,
payload: JetstreamPayload::Commit {
commit: JetstreamCommit {
rev: rec.rev.as_str(),
operation: rec.action.as_str(),
collection: rec.collection.as_str(),
rkey: rec.rkey.as_str(),
record: record_raw,
cid: rec.cid.map(|cid| cid.to_string()),
live: *live,
},
},
};
serde_json::to_vec(&json_event).ok().map(Bytes::from)
}
#[cfg(feature = "relay")]
StoredJetstreamEvent::RelayCommit {
relay_seq,
op_index,
..
} => {
let frame = state
.db
.relay
.events
.get(keys::relay_event_key(*relay_seq))
.ok()??;
let SubscribeReposMessage::Commit(commit) = decode_frame(frame.as_ref()).ok()? else {
return None;
};
let op = commit.ops.get(*op_index as usize)?;
let (collection, rkey) = op.path.split_once('/')?;
let action = op.action.as_str();
let mut record_raw = None;
let cid = if matches!(action, "create" | "update") {
let cid = op.cid.as_ref()?;
let cid_ipld = cid.to_ipld().ok()?;
let parsed = crate::car::parse_commit_car(commit.blocks.clone(), false).ok()?;
let block = parsed.blocks.get(&cid_ipld)?;
record_raw = Some(crate::types::record_json(block).ok()?);
Some(cid.to_string())
} else {
None
};
let json_event = JetstreamEvent {
did: commit.repo.as_str(),
time_us,
payload: JetstreamPayload::Commit {
commit: JetstreamCommit {
rev: commit.rev.as_str(),
operation: action,
collection,
rkey,
record: record_raw.as_deref(),
cid,
live: true,
},
},
};
serde_json::to_vec(&json_event).ok().map(Bytes::from)
}
#[cfg(feature = "relay")]
StoredJetstreamEvent::RelayAccount { relay_seq, .. } => {
let frame = state
.db
.relay
.events
.get(keys::relay_event_key(*relay_seq))
.ok()??;
let SubscribeReposMessage::Account(account) = decode_frame(frame.as_ref()).ok()? else {
return None;
};
let did_str = account.did.as_str();
let status = account.status.as_ref().map(|s| {
use jacquard_common::{IntoStatic, cowstr::ToCowStr};
s.to_cowstr().into_static().as_str().to_string()
});
let json_event = JetstreamEvent {
did: did_str,
time_us,
payload: JetstreamPayload::Account {
account: JetstreamAccount {
active: account.active,
did: did_str.to_string(),
seq: account.seq,
time: &account.time,
status,
},
},
};
serde_json::to_vec(&json_event).ok().map(Bytes::from)
}
#[cfg(feature = "relay")]
StoredJetstreamEvent::RelayIdentity { relay_seq, .. } => {
let frame = state
.db
.relay
.events
.get(keys::relay_event_key(*relay_seq))
.ok()??;
let SubscribeReposMessage::Identity(identity) = decode_frame(frame.as_ref()).ok()?
else {
return None;
};
let did_str = identity.did.as_str();
let handle = identity.handle.as_ref().map(|h| h.as_str().to_string());
let json_event = JetstreamEvent {
did: did_str,
time_us,
payload: JetstreamPayload::Identity {
identity: JetstreamIdentity {
did: did_str.to_string(),
seq: identity.seq,
time: &identity.time,
handle,
},
},
};
serde_json::to_vec(&json_event).ok().map(Bytes::from)
}
StoredJetstreamEvent::Account {
did,
active,
status,
seq,
time,
} => {
let did_str = did.to_did();
let status = status.as_ref().map(|s| s.as_str().to_string());
let json_event = JetstreamEvent {
did: did_str.as_str(),
time_us,
payload: JetstreamPayload::Account {
account: JetstreamAccount {
active: *active,
did: did_str.as_str().to_string(),
seq: *seq,
time,
status,
},
},
};
serde_json::to_vec(&json_event).ok().map(Bytes::from)
}
StoredJetstreamEvent::Identity {
did,
handle,
seq,
time,
} => {
let did_str = did.to_did();
let handle = handle.as_ref().map(|h| h.as_str().to_string());
let json_event = JetstreamEvent {
did: did_str.as_str(),
time_us,
payload: JetstreamPayload::Identity {
identity: JetstreamIdentity {
did: did_str.as_str().to_string(),
seq: *seq,
time,
handle,
},
},
};
serde_json::to_vec(&json_event).ok().map(Bytes::from)
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn positions_order_like_their_keys() {
let positions = [
JetstreamPosition {
time_us: 1,
id: u64::MAX,
},
JetstreamPosition { time_us: 2, id: 0 },
JetstreamPosition { time_us: 2, id: 7 },
JetstreamPosition { time_us: 3, id: 1 },
];
for pair in positions.windows(2) {
assert!(pair[0] < pair[1]);
assert!(pair[0].key() < pair[1].key());
assert!(pair[0].key_after() <= pair[1].key());
assert!(pair[0].key() < pair[0].key_after());
}
}
#[test]
fn a_cursor_starts_before_its_first_event() {
let before = JetstreamPosition::before(5).unwrap();
assert!(before.key() < JetstreamPosition { time_us: 5, id: 0 }.key());
assert_eq!(before.key_after(), keys::jetstream_event_key(5, 0));
assert!(JetstreamPosition::before(0).is_none());
}
}