use bytes::Bytes;
use jacquard_common::types::cid::IpldCid;
use jacquard_common::types::nsid::Nsid;
use jacquard_common::types::string::{Did, Rkey};
use jacquard_common::types::tid::Tid;
use jacquard_common::{CowStr, types::string::Handle};
use serde::{Deserialize, Serialize, Serializer};
use serde_json::value::RawValue;
use std::fmt::Debug;
use crate::db::types::{DbAction, DbRkey, DbTid, TrimmedDid};
#[derive(Debug, Serialize, Clone)]
pub enum EventType {
Record,
Identity,
Account,
}
impl AsRef<str> for EventType {
fn as_ref(&self) -> &str {
match self {
Self::Record => "record",
Self::Identity => "identity",
Self::Account => "account",
}
}
}
fn event_type_ser_str<S: Serializer>(v: &EventType, s: S) -> Result<S::Ok, S::Error> {
s.serialize_str(v.as_ref())
}
#[derive(Debug, Serialize, Clone)]
pub struct MarshallableEvt<'i> {
pub id: u64,
#[serde(rename = "type")]
#[serde(serialize_with = "event_type_ser_str")]
pub kind: EventType,
#[serde(borrow)]
#[serde(skip_serializing_if = "Option::is_none")]
pub record: Option<RecordEvt<'i>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub identity: Option<IdentityEvt>,
#[serde(borrow)]
#[serde(skip_serializing_if = "Option::is_none")]
pub account: Option<AccountEvt<'i>>,
}
#[derive(Debug, Serialize, Clone)]
pub struct RecordEvt<'i> {
pub live: bool,
pub did: Did,
pub rev: Tid,
pub collection: Nsid,
pub rkey: Rkey,
pub action: CowStr<'i>,
#[serde(skip_serializing_if = "Option::is_none")]
pub record: Option<std::sync::Arc<RawValue>>,
#[serde(skip_serializing_if = "Option::is_none")]
#[serde(serialize_with = "crate::util::opt_cid_serialize_str")]
pub cid: Option<IpldCid>,
}
#[derive(Debug, Serialize, Clone)]
pub struct IdentityEvt {
pub did: Did,
#[serde(skip_serializing_if = "Option::is_none")]
pub handle: Option<Handle>,
}
#[derive(Debug, Serialize, Clone)]
pub struct AccountEvt<'i> {
pub did: Did,
pub active: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub status: Option<CowStr<'i>>,
}
#[derive(Serialize, Deserialize, Clone, Default)]
pub(crate) enum StoredData {
#[default]
Nothing,
Ptr(IpldCid),
/// Ephemeral create/update body. The event is its sole durable owner, so
/// event TTL removes the identity and bytes together.
#[serde(with = "jacquard_common::serde_bytes_helper")]
Block(Bytes),
}
impl StoredData {
pub fn is_nothing(&self) -> bool {
matches!(self, StoredData::Nothing)
}
}
impl Debug for StoredData {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Nothing => f.write_str("nothing"),
Self::Ptr(cid) => write!(f, "{cid}"),
Self::Block(_) => f.write_str("<block>"),
}
}
}
#[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(bound(deserialize = "'i: 'de"))]
pub(crate) struct StoredEvent<'i> {
#[serde(default)]
pub live: bool,
#[serde(borrow)]
pub did: TrimmedDid<'i>,
pub rev: DbTid,
#[serde(borrow)]
pub collection: CowStr<'i>,
pub rkey: DbRkey,
pub action: DbAction,
#[serde(default)]
#[serde(skip_serializing_if = "StoredData::is_nothing")]
pub data: StoredData,
}
/// a durable record event that is also emitted on the in-memory stream after commit.
///
/// the first `/stream` reader turns it into its frame, which every other reader shares, and
/// from then on only the frame is kept. so a slow reader's backlog holds one frame per event
/// instead of the event, its body and the commit's car the body is a slice of. a reader that
/// wants the decoded event after that reads it from the db like a replay does.
#[derive(Debug)]
pub(crate) struct LiveRecordEvent {
pub id: u64,
live: parking_lot::Mutex<Live>,
}
#[derive(Debug)]
enum Live {
/// `inline_block` is the body the commit had, so a live reader needn't read it back
Stored {
stored: StoredEvent<'static>,
inline_block: Option<Bytes>,
},
Frame(Bytes),
}
impl LiveRecordEvent {
pub(crate) fn new(id: u64, stored: StoredEvent<'static>, inline_block: Option<Bytes>) -> Self {
Self {
id,
live: parking_lot::Mutex::new(Live::Stored {
stored,
inline_block,
}),
}
}
/// the event and its inline body, unless a reader already swapped them for the frame
pub(crate) fn stored(&self) -> Option<(StoredEvent<'static>, Option<Bytes>)> {
match &*self.live.lock() {
Live::Stored {
stored,
inline_block,
} => Some((stored.clone(), inline_block.clone())),
Live::Frame(_) => None,
}
}
/// the event's frame, which `build` makes from the event the first time. the lock isn't
/// held while it builds, so two readers can race to build it, and the first one in wins
pub(crate) fn frame(
&self,
build: impl FnOnce(StoredEvent<'static>, Option<Bytes>) -> Option<Bytes>,
) -> Option<Bytes> {
let (stored, inline_block) = match &*self.live.lock() {
Live::Frame(frame) => return Some(frame.clone()),
Live::Stored {
stored,
inline_block,
} => (stored.clone(), inline_block.clone()),
};
let built = build(stored, inline_block)?;
let mut live = self.live.lock();
match &*live {
Live::Frame(frame) => Some(frame.clone()),
Live::Stored { .. } => {
*live = Live::Frame(built.clone());
Some(built)
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn stream_event_json_envelopes_cover_every_variant() {
let did = Did::new_static("did:web:example.com").unwrap();
let cid = crate::car::compute_cid(b"body");
let record = MarshallableEvt {
id: 1,
kind: EventType::Record,
record: Some(RecordEvt {
live: false,
did: did.clone(),
rev: Tid::new("3jzfcijpj2z2a").unwrap(),
collection: Nsid::new_static("app.bsky.feed.post").unwrap(),
rkey: Rkey::new_static("stinkpot").unwrap(),
action: CowStr::Borrowed("create"),
record: None,
cid: Some(cid),
}),
identity: None,
account: None,
};
assert_eq!(
serde_json::to_value(&record).unwrap(),
serde_json::json!({
"id": 1,
"type": "record",
"record": {
"live": false,
"did": "did:web:example.com",
"rev": "3jzfcijpj2z2a",
"collection": "app.bsky.feed.post",
"rkey": "stinkpot",
"action": "create",
"cid": cid.to_string(),
},
})
);
let identity = MarshallableEvt {
id: 2,
kind: EventType::Identity,
record: None,
identity: Some(IdentityEvt {
did: did.clone(),
handle: None,
}),
account: None,
};
assert_eq!(
serde_json::to_value(&identity).unwrap(),
serde_json::json!({
"id": 2,
"type": "identity",
"identity": {"did": "did:web:example.com"},
})
);
let account = MarshallableEvt {
id: 3,
kind: EventType::Account,
record: None,
identity: None,
account: Some(AccountEvt {
did,
active: true,
status: None,
}),
};
assert_eq!(
serde_json::to_value(&account).unwrap(),
serde_json::json!({
"id": 3,
"type": "account",
"account": {
"did": "did:web:example.com",
"active": true,
},
})
);
}
#[test]
fn stored_data_persisted_golden_fixtures() {
let cid = crate::car::compute_cid(b"body");
let variants = [
StoredData::Nothing,
StoredData::Ptr(cid),
StoredData::Block(Bytes::from_static(b"stinkpot")),
];
let encoded = variants
.iter()
.map(|value| rmp_serde::to_vec(value).unwrap())
.collect::<Vec<_>>();
assert_eq!(
encoded.iter().map(crate::util::hex::encode).collect::<Vec<_>>(),
[
"a74e6f7468696e67",
"81a3507472c42401711220230d8358dc8e8890b4c58deeb62912ee2f20357ae92a5cc861b98e68fe31acb5",
"81a5426c6f636bc4087374696e6b706f74",
]
);
assert!(matches!(
rmp_serde::from_slice::<StoredData>(&encoded[0]).unwrap(),
StoredData::Nothing
));
assert!(matches!(
rmp_serde::from_slice::<StoredData>(&encoded[1]).unwrap(),
StoredData::Ptr(decoded) if decoded == cid
));
assert!(matches!(
rmp_serde::from_slice::<StoredData>(&encoded[2]).unwrap(),
StoredData::Block(decoded) if decoded.as_ref() == b"stinkpot"
));
}
}