use jacquard_common::{CowStr, types::string::Handle};
use serde::{Deserialize, Serialize};
use crate::db::types::TrimmedDid;
use crate::ingest::stream::Datetime;
#[inline]
#[allow(dead_code)]
pub(crate) fn default_true() -> bool {
true
}
/// a stored jetstream event. `Id` is how it refers to the upstream row its
/// body resolves through (a `/stream` event id, or a relay seq); an outbox
/// stages it with a reference that commit swaps for the committed id.
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
#[serde(bound(
serialize = "Id: Serialize",
deserialize = "'i: 'de, Id: Deserialize<'de>"
))]
pub(crate) enum StoredJetstreamEvent<'i, Id = u64> {
#[cfg(feature = "indexer_stream")]
Commit {
#[serde(borrow)]
did: TrimmedDid<'i>,
#[serde(borrow)]
collection: CowStr<'i>,
event_id: Id,
#[serde(default = "default_true")]
live: bool,
},
#[cfg(feature = "relay")]
RelayCommit {
#[serde(borrow)]
did: TrimmedDid<'i>,
#[serde(borrow)]
collection: CowStr<'i>,
relay_seq: Id,
op_index: u32,
},
#[cfg(feature = "relay")]
RelayAccount {
#[serde(borrow)]
did: TrimmedDid<'i>,
relay_seq: Id,
},
#[cfg(feature = "relay")]
RelayIdentity {
#[serde(borrow)]
did: TrimmedDid<'i>,
relay_seq: Id,
},
Account {
#[serde(borrow)]
did: TrimmedDid<'i>,
active: bool,
#[serde(borrow)]
status: Option<CowStr<'i>>,
seq: i64,
time: Datetime,
},
Identity {
#[serde(borrow)]
did: TrimmedDid<'i>,
handle: Option<Handle>,
seq: i64,
time: Datetime,
},
}
/// a jetstream event's place in commit order, which is also its key in the
/// events keyspace.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub(crate) struct JetstreamPosition {
pub(crate) time_us: u64,
pub(crate) id: u64,
}
impl JetstreamPosition {
pub(crate) fn key(self) -> [u8; 16] {
crate::db::keys::jetstream_event_key(self.time_us, self.id)
}
/// the first key after this position.
pub(crate) fn key_after(self) -> [u8; 16] {
match self.id.checked_add(1) {
Some(id) => crate::db::keys::jetstream_event_key(self.time_us, id),
None => crate::db::keys::jetstream_event_key(self.time_us.saturating_add(1), 0),
}
}
/// the last position before any event at or after `time_us`.
pub(crate) fn before(time_us: u64) -> Option<Self> {
time_us.checked_sub(1).map(|time_us| Self {
time_us,
id: u64::MAX,
})
}
}
#[derive(Clone, Debug)]
pub(crate) enum JetstreamBroadcast {
/// every event through this position is committed, and the ones since
/// the previous broadcast are backfilled commits, which are never
/// broadcast themselves.
Historical(JetstreamPosition),
Live(JetstreamLive),
}
#[derive(Clone, Debug)]
pub(crate) struct JetstreamLive {
pub(crate) position: JetstreamPosition,
pub(crate) event: StoredJetstreamEvent<'static>,
/// pre-serialized json, when a subscriber was listening as the event was
/// staged. otherwise it is rendered from the database.
pub(crate) json: Option<bytes::Bytes>,
}
impl<'i, Id> StoredJetstreamEvent<'i, Id> {
pub(crate) fn is_live(&self) -> bool {
match self {
#[cfg(feature = "indexer_stream")]
Self::Commit { live, .. } => *live,
_ => true,
}
}
pub(crate) fn did(&self) -> &TrimmedDid<'i> {
match self {
#[cfg(feature = "indexer_stream")]
Self::Commit { did, .. } => did,
#[cfg(feature = "relay")]
Self::RelayCommit { did, .. }
| Self::RelayAccount { did, .. }
| Self::RelayIdentity { did, .. } => did,
Self::Account { did, .. } | Self::Identity { did, .. } => did,
}
}
pub(crate) fn collection(&self) -> Option<&str> {
match self {
#[cfg(feature = "indexer_stream")]
Self::Commit { collection, .. } => Some(collection.as_str()),
#[cfg(feature = "relay")]
Self::RelayCommit { collection, .. } => Some(collection.as_str()),
#[cfg(feature = "relay")]
Self::RelayAccount { .. } | Self::RelayIdentity { .. } => None,
Self::Account { .. } | Self::Identity { .. } => None,
}
}
/// swap this event's upstream reference for what `id` maps it to.
pub(crate) fn resolve<T>(self, id: impl FnOnce(Id) -> T) -> StoredJetstreamEvent<'i, T> {
match self {
#[cfg(feature = "indexer_stream")]
Self::Commit {
did,
collection,
event_id,
live,
} => StoredJetstreamEvent::Commit {
did,
collection,
event_id: id(event_id),
live,
},
#[cfg(feature = "relay")]
Self::RelayCommit {
did,
collection,
relay_seq,
op_index,
} => StoredJetstreamEvent::RelayCommit {
did,
collection,
relay_seq: id(relay_seq),
op_index,
},
#[cfg(feature = "relay")]
Self::RelayAccount { did, relay_seq } => StoredJetstreamEvent::RelayAccount {
did,
relay_seq: id(relay_seq),
},
#[cfg(feature = "relay")]
Self::RelayIdentity { did, relay_seq } => StoredJetstreamEvent::RelayIdentity {
did,
relay_seq: id(relay_seq),
},
Self::Account {
did,
active,
status,
seq,
time,
} => StoredJetstreamEvent::Account {
did,
active,
status,
seq,
time,
},
Self::Identity {
did,
handle,
seq,
time,
} => StoredJetstreamEvent::Identity {
did,
handle,
seq,
time,
},
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use jacquard_common::types::string::Did;
fn did() -> TrimmedDid<'static> {
let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
TrimmedDid::from(&did).into_static()
}
fn time() -> Datetime {
serde_json::from_str(r#""2026-08-10T12:34:56Z""#).unwrap()
}
fn assert_fixture(event: &StoredJetstreamEvent<'_>, golden: &str) {
let encoded = rmp_serde::to_vec(event).unwrap();
assert_eq!(crate::util::hex::encode(&encoded), golden);
let decoded: StoredJetstreamEvent<'_> = rmp_serde::from_slice(&encoded).unwrap();
assert_eq!(&decoded, event);
}
#[test]
fn common_jetstream_event_golden_fixtures() {
assert_fixture(
&StoredJetstreamEvent::Account {
did: did(),
active: false,
status: Some("stinkpot".into()),
seq: 42,
time: time(),
},
"81a74163636f756e7495c4100125aa8fb6f9c3a8df64f789e5ee3919c2a87374696e6b706f742ab4323032362d30382d31305431323a33343a35365a",
);
assert_fixture(
&StoredJetstreamEvent::Identity {
did: did(),
handle: Some(Handle::new_static("alice.test").unwrap()),
seq: 42,
time: time(),
},
"81a84964656e7469747994c4100125aa8fb6f9c3a8df64f789e5ee3919aa616c6963652e746573742ab4323032362d30382d31305431323a33343a35365a",
);
}
#[cfg(feature = "indexer_stream")]
#[test]
fn indexer_jetstream_event_golden_fixture() {
let event = StoredJetstreamEvent::Commit {
did: did(),
collection: "app.bsky.feed.post".into(),
event_id: 42,
live: false,
};
assert_fixture(
&event,
"81a6436f6d6d697494c4100125aa8fb6f9c3a8df64f789e5ee3919b26170702e62736b792e666565642e706f73742ac2",
);
assert!(!event.is_live());
assert_eq!(event.collection(), Some("app.bsky.feed.post"));
}
#[cfg(feature = "relay")]
#[test]
fn relay_jetstream_event_golden_fixtures() {
for (event, golden) in [
(
StoredJetstreamEvent::RelayCommit {
did: did(),
collection: "app.bsky.feed.post".into(),
relay_seq: 42,
op_index: 7,
},
"81ab52656c6179436f6d6d697494c4100125aa8fb6f9c3a8df64f789e5ee3919b26170702e62736b792e666565642e706f73742a07",
),
(
StoredJetstreamEvent::RelayAccount {
did: did(),
relay_seq: 42,
},
"81ac52656c61794163636f756e7492c4100125aa8fb6f9c3a8df64f789e5ee39192a",
),
(
StoredJetstreamEvent::RelayIdentity {
did: did(),
relay_seq: 42,
},
"81ad52656c61794964656e7469747992c4100125aa8fb6f9c3a8df64f789e5ee39192a",
),
] {
assert_fixture(&event, golden);
}
}
}