use std::io::Write as _;
use std::marker::PhantomData;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use bytes::Bytes;
use tokio::sync::mpsc;
use tracing::{debug, error};
use crate::db::types::{DbRkey, DbTid, TrimmedDid};
use crate::db::{self, keys};
use crate::state::AppState;
use crate::types::{
BroadcastEvent, LiveRecordEvent, MarshallableEvt, RecordEvt, StoredData, StoredEvent,
record_json, write_record_json,
};
use jacquard_common::types::cid::IpldCid;
use jacquard_common::types::nsid::Nsid;
use jacquard_common::types::string::Rkey;
use jacquard_common::{CowStr, IntoStatic};
use super::{ReplayChunk, ReplayFailed, Start, StreamOptions, StreamSource, run_stream};
use crate::control::{Event, StreamError};
/// what a subscriber's channel carries for each event: the decoded event for
/// [`crate::control::Hydrant::subscribe`], or the frame `/stream` sends
pub(crate) trait Form {
type Output: Send + 'static;
fn stored(state: &AppState, id: u64, stored: StoredEvent<'_>) -> Option<Self::Output>;
fn live(state: &AppState, event: &LiveRecordEvent) -> Option<Self::Output>;
fn ephemeral(event: Event) -> Option<Self::Output>;
/// about how many bytes the output takes on the wire, see [`StreamSource::frame_bytes`]
fn frame_bytes(output: &Self::Output) -> usize;
}
/// the decoded [`Event`]
pub(crate) struct Decoded;
impl Form for Decoded {
type Output = Event;
fn stored(state: &AppState, id: u64, stored: StoredEvent<'_>) -> Option<Event> {
stored_to_event(state, id, stored, None)
}
fn live(state: &AppState, event: &LiveRecordEvent) -> Option<Event> {
match event.stored() {
Some((stored, inline_block)) => stored_to_event(state, event.id, stored, inline_block),
// a `/stream` reader swapped the event for its frame already
None => read_event(state, event.id),
}
}
fn ephemeral(event: Event) -> Option<Event> {
Some(event)
}
/// decoded events only go to in-process readers, which are never capped, so nothing pays
/// for these
fn frame_bytes(_: &Event) -> usize {
0
}
}
/// the event's `/stream` frame. a live event's is built once and shared by every reader
pub(crate) struct Frame;
impl Form for Frame {
type Output = Bytes;
fn stored(state: &AppState, id: u64, stored: StoredEvent<'_>) -> Option<Bytes> {
stored_to_frame(state, id, stored, None)
}
fn live(state: &AppState, event: &LiveRecordEvent) -> Option<Bytes> {
event.frame(|stored, inline_block| stored_to_frame(state, event.id, stored, inline_block))
}
fn ephemeral(event: Event) -> Option<Bytes> {
serde_json::to_vec(&event)
.inspect_err(|e| error!(err = %e, id = event.id, "failed to serialize event"))
.ok()
.map(Bytes::from)
}
fn frame_bytes(frame: &Bytes) -> usize {
frame.len()
}
}
pub(crate) fn event_stream_thread<F: Form>(
state: Arc<AppState>,
tx: mpsc::Sender<Result<F::Output, StreamError>>,
cursor: Option<u64>,
opts: StreamOptions,
) {
let event_rx = state.db.stream.event_tx.subscribe();
let head = state.db.stream.ids.head();
let start = match cursor {
Some(cursor) => Start::replay(cursor.checked_sub(1), head),
None => Start::live(head),
};
let source = EventSource::<F> {
state,
form: PhantomData,
};
run_stream(tx, event_rx, start, opts, source);
}
struct EventSource<F> {
state: Arc<AppState>,
form: PhantomData<F>,
}
impl<F: Form> StreamSource for EventSource<F> {
type Broadcast = BroadcastEvent;
type Output = F::Output;
fn read(
&mut self,
after: Option<u64>,
through: Option<u64>,
limit: usize,
) -> Result<ReplayChunk<F::Output, u64>, ReplayFailed> {
read_event_replay_chunk(&self.state, after, through, limit, F::stored)
}
fn written(&self) -> u64 {
self.state.db.stream.ids.next()
}
fn frame_bytes(output: &F::Output) -> usize {
F::frame_bytes(output)
}
fn render(&mut self, event: BroadcastEvent) -> Option<F::Output> {
match event {
BroadcastEvent::Persisted(_) => None,
BroadcastEvent::LiveRecord(event) => F::live(&self.state, &event),
BroadcastEvent::Ephemeral(event) => F::ephemeral(*event),
}
}
}
fn read_event_replay_chunk<T>(
state: &AppState,
after: Option<u64>,
through: Option<u64>,
limit: usize,
render: impl Fn(&AppState, u64, StoredEvent<'_>) -> Option<T>,
) -> Result<ReplayChunk<T, u64>, ReplayFailed> {
let start = after.map_or(0, |id| id.saturating_add(1));
let end = through.unwrap_or(u64::MAX);
if start > end {
return Ok(ReplayChunk {
events: Vec::new(),
last_seen: after,
exhausted: true,
scanned: 0,
});
}
let rows = state
.db
.stream
.event_range(keys::event_key(start)..=keys::event_key(end));
ReplayChunk::read(
rows,
after,
limit,
|key| {
let id = key.try_into().map(u64::from_be_bytes);
id.inspect_err(|_| error!("failed to parse event id")).ok()
},
|id, value| {
let stored = rmp_serde::from_slice::<StoredEvent>(value)
.inspect_err(|e| error!(err = %e, "failed to deserialize stored event"))
.ok()?;
render(state, id, stored)
},
)
}
/// the event `id` as a replay would read it
fn read_event(state: &AppState, id: u64) -> Option<Event> {
let row = state
.db
.stream
.event(keys::event_key(id))
.inspect_err(|e| error!(err = %e, id, "failed to read a live event back"))
.ok()??;
let stored = rmp_serde::from_slice::<StoredEvent>(&row)
.inspect_err(|e| error!(err = %e, id, "failed to deserialize stored event"))
.ok()?;
stored_to_event(state, id, stored, None)
}
pub(crate) fn stored_to_event(
state: &AppState,
id: u64,
stored: StoredEvent<'_>,
inline_block: Option<Bytes>,
) -> Option<Event> {
let (cid, body) = event_body(state, id, &stored, inline_block, Inline::IfStillStored);
let record = body.and_then(|body| {
record_json(&body)
.inspect_err(|e| error!(err = %e, id, "cant parse record body"))
.ok()
});
let StoredEvent {
live,
did,
rev,
collection,
rkey,
action,
data: _,
} = stored;
Some(MarshallableEvt {
id,
kind: crate::types::EventType::Record,
record: Some(RecordEvt {
live,
did: did.to_did(),
rev: rev.to_tid(),
// SAFETY: nothing unsound rests on these being valid, they're strings. the db is
// trusted so a replay doesn't check each again, but a row written before ingest
// checked them can hold one that isn't, and `Nsid::name` panics on that
collection: unsafe { Nsid::unchecked(smol_str::SmolStr::new(collection.as_str())) },
// SAFETY: as for the collection
rkey: unsafe { Rkey::unchecked(rkey.to_smolstr()) },
action: CowStr::Borrowed(action.as_str()),
record,
cid,
}),
identity: None,
account: None,
})
}
/// the event's `/stream` frame: the same bytes `serde_json::to_vec` writes for what
/// [`stored_to_event`] returns, with the body transcoded straight into it
pub(crate) fn stored_to_frame(
state: &AppState,
id: u64,
stored: StoredEvent<'_>,
inline_block: Option<Bytes>,
) -> Option<Bytes> {
let (cid, body) = event_body(state, id, &stored, inline_block, Inline::AsCommitted);
let mut out = Vec::with_capacity(256 + body.as_ref().map_or(0, |body| body.len() * 2));
out.extend_from_slice(br#"{"id":"#);
write_json(&mut out, &id);
out.extend_from_slice(br#","type":"record","record":{"live":"#);
write_json(&mut out, &stored.live);
out.extend_from_slice(br#","did":"#);
write_json(&mut out, stored.did.to_did().as_str());
out.extend_from_slice(br#","rev":"#);
write_json(&mut out, stored.rev.to_tid().as_str());
out.extend_from_slice(br#","collection":"#);
write_json(&mut out, stored.collection.as_str());
out.extend_from_slice(br#","rkey":"#);
write_json(&mut out, stored.rkey.to_smolstr().as_str());
out.extend_from_slice(br#","action":"#);
write_json(&mut out, stored.action.as_str());
if let Some(body) = body {
let mark = out.len();
out.extend_from_slice(br#","record":"#);
if let Err(e) = write_record_json(&body, &mut out) {
error!(err = %e, id, "cant parse record body");
out.truncate(mark);
}
}
if let Some(cid) = cid {
out.extend_from_slice(br#","cid":""#);
write!(out, "{cid}").expect("a vec takes any write");
out.push(b'"');
}
out.extend_from_slice(b"}}");
// a live frame can sit in many readers' queues, so it shouldn't keep the guess's slack
out.shrink_to_fit();
Some(Bytes::from(out))
}
pub(crate) fn stored_to_chunk_value(state: &AppState, id: u64, stored: StoredEvent<'_>) -> Vec<u8> {
let (cid, body) = event_body(state, id, &stored, None, Inline::AsCommitted);
// re-encoding can change the hash, and the stored cid goes stale.
let cid = cid.filter(|cid| {
body.as_ref()
.is_none_or(|body| crate::car::compute_cid(body) != *cid)
});
let (did, rev, rkey) = (
stored.did.to_did(),
stored.rev.to_tid(),
stored.rkey.to_smolstr(),
);
let value = ChunkValue {
cid,
did: did.as_str(),
rev: rev.as_str(),
live: stored.live,
rkey: rkey.as_str(),
action: stored.action.as_str(),
record: body.as_deref().map(serde_bytes::Bytes::new),
collection: stored.collection.as_str(),
};
serde_ipld_dagcbor::to_vec(&value).expect("a chunk value encodes")
}
#[derive(serde::Serialize)]
struct ChunkValue<'a> {
#[serde(skip_serializing_if = "Option::is_none")]
cid: Option<IpldCid>,
did: &'a str,
rev: &'a str,
live: bool,
rkey: &'a str,
action: &'a str,
#[serde(skip_serializing_if = "Option::is_none")]
record: Option<&'a serde_bytes::Bytes>,
collection: &'a str,
}
fn write_json(out: &mut Vec<u8>, value: &(impl serde::Serialize + ?Sized)) {
serde_json::to_writer(out, value).expect("a vec takes any write");
}
/// whether a live event's inline body goes out as it is
#[derive(Clone, Copy)]
enum Inline {
/// only while the db still has it, so a redaction that finished since the commit holds
IfStillStored,
/// as the commit had it. the frame is built once for every reader, see the redaction
/// note in docs/api/stream.md
AsCommitted,
}
/// an event's record cid and body, `None` for a delete or a body that isn't there anymore
fn event_body(
state: &AppState,
id: u64,
stored: &StoredEvent<'_>,
inline_block: Option<Bytes>,
inline: Inline,
) -> (Option<IpldCid>, Option<Body>) {
match &stored.data {
StoredData::Ptr(cid) => {
let resolve = || {
resolve_event_body(
state,
id,
&stored.did,
stored.collection.as_str(),
&stored.rkey,
&stored.rev,
cid,
)
};
let body = match (inline_block, inline) {
(Some(block), Inline::AsCommitted) => Some(Body::Inline(block)),
(Some(block), Inline::IfStillStored) => resolve().map(|_| Body::Inline(block)),
(None, _) => resolve().map(Body::Stored),
};
if body.is_none() {
if state.history_ttl.is_some() {
// bounded history: the body was most likely trimmed by
// retention, which is expected, not a storage bug
state.history_trim_misses.fetch_add(1, Ordering::Relaxed);
debug!(id, "record body not found for event (bounded history)");
} else {
debug!(id, "record body unavailable for event");
}
}
(Some(*cid), body)
}
StoredData::Block(block) => {
let cid = crate::car::compute_cid(block);
(Some(cid), Some(Body::Inline(block.clone())))
}
StoredData::Nothing => (None, None),
}
}
enum Body {
Inline(Bytes),
Stored(fjall::Slice),
}
impl std::ops::Deref for Body {
type Target = [u8];
fn deref(&self) -> &[u8] {
match self {
Self::Inline(bytes) => bytes,
Self::Stored(slice) => slice,
}
}
}
/// tries the current head, then the history after this event's rev, then the pre-v10 archive.
/// a body only counts if it matches the event's cid
pub(crate) fn resolve_event_body(
state: &AppState,
id: u64,
did: &TrimmedDid,
collection: &str,
rkey: &DbRkey,
rev: &DbTid,
cid: &IpldCid,
) -> Option<fjall::Slice> {
let key = keys::record_key_trimmed(did, collection, rkey);
match state.db.resolve_event_record_body(&key, rev, cid) {
Ok(body) => body,
Err(e) => {
error!(err = %e, id, "cant resolve event record body");
state.db.poison.check_report(&e);
None
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::Config;
use jacquard_common::types::string::Tid;
use tempfile::TempDir;
const DID: &str = "did:plc:ewvi7nxzyoun6zhxrhs64oiz";
const COL: &str = "app.bsky.feed.post";
const RKEY: &str = "3kzbif5moe22m";
fn test_state() -> (TempDir, AppState) {
let tmp = tempfile::tempdir().unwrap();
let mut config = Config::default();
config.database_path = tmp.path().to_path_buf();
let state = AppState::new(&config).unwrap();
(tmp, state)
}
fn rev(s: &str) -> DbTid {
DbTid::from(&Tid::new(s).unwrap())
}
fn resolve(state: &AppState, id: u64, event_rev: &DbTid, cid: &IpldCid) -> Option<Vec<u8>> {
let did_full = jacquard_common::types::string::Did::new_static(DID).unwrap();
let did = TrimmedDid::from(&did_full);
resolve_event_body(state, id, &did, COL, &DbRkey::new(RKEY), event_rev, cid)
.map(|b| b.to_vec())
}
fn record_key() -> Vec<u8> {
let did = jacquard_common::types::string::Did::new_static(DID).unwrap();
keys::record_key(&did, COL, &DbRkey::new(RKEY))
}
fn post_event(collection: &'static str) -> StoredEvent<'static> {
let did = jacquard_common::types::string::Did::new_static(DID).unwrap();
let body =
serde_ipld_dagcbor::to_vec(&serde_json::json!({"$type": COL, "text": "stinkpot"}))
.unwrap();
StoredEvent {
live: true,
did: TrimmedDid::from(&did).into_static(),
rev: rev("3kzbif5moe22m"),
collection: CowStr::Borrowed(collection),
rkey: DbRkey::new(RKEY),
action: crate::db::types::DbAction::Create,
data: StoredData::Block(bytes::Bytes::from(body)),
}
}
#[test]
fn a_stored_event_replays_byte_for_byte_like_it_did_when_checked() {
let (_tmp, state) = test_state();
let event = stored_to_event(&state, 7, post_event(COL), None).unwrap();
assert_eq!(
serde_json::to_string(&event).unwrap(),
r#"{"id":7,"type":"record","record":{"live":true,"did":"did:plc:ewvi7nxzyoun6zhxrhs64oiz","rev":"3kzbif5moe22m","collection":"app.bsky.feed.post","rkey":"3kzbif5moe22m","action":"create","record":{"text":"stinkpot","$type":"app.bsky.feed.post"},"cid":"bafyreifnnc6tdkmifbsnopak4maqkyf2lxqst4dmsgyjjvf7vhqq7grtae"}}"#
);
}
/// rows written before ingest checked collections may hold one no ingest would take today.
/// replay trusts the db, so they stream as they are instead of being skipped
#[test]
fn a_stored_collection_isnt_checked_again() {
let (_tmp, state) = test_state();
let event = stored_to_event(&state, 7, post_event("not an nsid"), None).unwrap();
assert_eq!(event.record.unwrap().collection.as_str(), "not an nsid");
}
fn ptr_event(cid: IpldCid) -> StoredEvent<'static> {
StoredEvent {
data: StoredData::Ptr(cid),
..post_event(COL)
}
}
fn body() -> Vec<u8> {
serde_ipld_dagcbor::to_vec(&serde_json::json!({"$type": COL, "text": "stinkpot"})).unwrap()
}
/// `body` as the record's head, which is where a live event's pointer resolves
fn store_body(state: &AppState, body: &[u8]) -> IpldCid {
let mut batch = state.db.inner.batch();
state.db.indexer.stage_head(&mut batch, record_key(), body);
batch.commit().unwrap();
crate::car::compute_cid(body)
}
fn as_json(state: &AppState, id: u64, stored: StoredEvent<'_>) -> Vec<u8> {
serde_json::to_vec(&stored_to_event(state, id, stored, None).unwrap()).unwrap()
}
#[test]
fn a_frame_is_the_events_json() {
let (_tmp, state) = test_state();
let stored = store_body(&state, &body());
let missing = crate::car::compute_cid(b"never stored");
let cases = [
post_event(COL),
ptr_event(stored),
ptr_event(missing),
StoredEvent {
live: false,
collection: CowStr::Borrowed("not an nsid"),
..post_event(COL)
},
StoredEvent {
action: crate::db::types::DbAction::Delete,
data: StoredData::Nothing,
..post_event(COL)
},
StoredEvent {
data: StoredData::Block(bytes::Bytes::from_static(b"\xff not cbor")),
..post_event(COL)
},
];
for stored in cases {
let frame = stored_to_frame(&state, 7, stored.clone(), None).unwrap();
assert_eq!(
String::from_utf8_lossy(&frame),
String::from_utf8_lossy(&as_json(&state, 7, stored))
);
}
}
#[test]
fn a_live_frame_is_built_once_and_shared() {
let (_tmp, state) = test_state();
let body = body();
let cid = crate::car::compute_cid(&body);
let event = LiveRecordEvent::new(7, ptr_event(cid), Some(bytes::Bytes::from(body)));
let first = Frame::live(&state, &event).unwrap();
let second = Frame::live(&state, &event).unwrap();
assert_eq!(
first.as_ptr(),
second.as_ptr(),
"the second reader built its own"
);
assert!(
event.stored().is_none(),
"the event and its body are still held"
);
// the body goes out as the commit had it, without a db read
assert!(first.windows(8).any(|w| w == b"stinkpot"));
}
#[test]
fn an_ephemeral_frame_is_the_events_json() {
let identity = MarshallableEvt {
id: 9,
kind: crate::types::EventType::Identity,
record: None,
identity: Some(crate::types::IdentityEvt {
did: jacquard_common::types::string::Did::new_static(DID).unwrap(),
handle: None,
}),
account: None,
};
let json = serde_json::to_vec(&identity).unwrap();
assert_eq!(Frame::ephemeral(identity).unwrap().to_vec(), json);
}
/// an in-process reader shares the broadcast with `/stream` readers, so it can find the
/// event already swapped for its frame
#[test]
fn an_in_process_reader_after_the_swap_reads_the_event_back() {
let (_tmp, state) = test_state();
let body = body();
let cid = store_body(&state, &body);
let mut batch = state.db.inner.batch();
let row = rmp_serde::to_vec(&ptr_event(cid)).unwrap();
state
.db
.stream
.stage_event(&mut batch, keys::event_key(7), row);
batch.commit().unwrap();
let event = LiveRecordEvent::new(7, ptr_event(cid), Some(bytes::Bytes::from(body)));
let frame = Frame::live(&state, &event).unwrap();
let decoded = Decoded::live(&state, &event).expect("the event from the db");
assert_eq!(serde_json::to_vec(&decoded).unwrap(), frame.to_vec());
}
#[test]
fn legacy_body_resolves_from_compatibility_archive() {
let (_tmp, state) = test_state();
let cid = crate::car::compute_cid(b"legacy body");
let mut batch = state.db.inner.batch();
batch.insert(
&state.db.stream.event_bodies,
keys::event_body_key(&record_key(), &cid),
b"legacy body",
);
batch.commit().unwrap();
assert_eq!(
resolve(&state, 41, &rev("3kzbif5moe22m"), &cid),
Some(b"legacy body".to_vec())
);
}
#[test]
fn falls_to_head_when_record_never_died() {
let (_tmp, state) = test_state();
let mut batch = state.db.inner.batch();
state
.db
.indexer
.stage_head(&mut batch, record_key(), b"head body".as_slice());
batch.commit().unwrap();
let cid = crate::car::compute_cid(b"head body");
assert_eq!(
resolve(&state, 42, &rev("3kzbif5moe22m"), &cid),
Some(b"head body".to_vec())
);
}
#[test]
fn serves_the_body_killed_after_the_event() {
let (_tmp, state) = test_state();
// the event wrote "old body" at rev m; the record died at rev n > m
let mut batch = state.db.inner.batch();
state.db.indexer.stage_history(
&mut batch,
keys::history_key(&record_key(), &rev("3kzbif5mof33m")),
b"old body".as_slice(),
);
state
.db
.indexer
.stage_head(&mut batch, record_key(), b"new body".as_slice());
batch.commit().unwrap();
let cid = crate::car::compute_cid(b"old body");
assert_eq!(
resolve(&state, 42, &rev("3kzbif5moe22m"), &cid),
Some(b"old body".to_vec())
);
}
#[test]
fn deaths_before_the_event_are_not_served() {
let (_tmp, state) = test_state();
// the only death predates the event: the event's body must be the head
let mut batch = state.db.inner.batch();
state.db.indexer.stage_history(
&mut batch,
keys::history_key(&record_key(), &rev("3kzbif5mod22m")),
b"ancient body".as_slice(),
);
state
.db
.indexer
.stage_head(&mut batch, record_key(), b"event body".as_slice());
batch.commit().unwrap();
let cid = crate::car::compute_cid(b"event body");
assert_eq!(
resolve(&state, 42, &rev("3kzbif5moe22m"), &cid),
Some(b"event body".to_vec())
);
}
#[test]
fn delete_tombstone_resolves_to_no_body() {
let (_tmp, state) = test_state();
// a delete of a never-seen record leaves an empty tombstone: the
// record was already deleted at that point, so there is no body
let mut batch = state.db.inner.batch();
state.db.indexer.stage_history(
&mut batch,
keys::history_key(&record_key(), &rev("3kzbif5mof33m")),
&[],
);
batch.commit().unwrap();
let cid = crate::car::compute_cid(b"anything");
assert_eq!(resolve(&state, 42, &rev("3kzbif5moe22m"), &cid), None);
}
#[test]
fn block_event_inflates_with_inline_body() {
let (_tmp, state) = test_state();
let record_val = serde_json::json!({
"$type": "app.bsky.feed.post",
"text": "hello"
});
let record_cbor = serde_ipld_dagcbor::to_vec(&record_val).unwrap();
let expected_cid = crate::car::compute_cid(&record_cbor);
let did_full = jacquard_common::types::string::Did::new_static(DID).unwrap();
let stored = StoredEvent {
live: false,
did: TrimmedDid::from(&did_full),
rev: rev("3kzbif5moe22m"),
collection: CowStr::Borrowed(COL),
rkey: DbRkey::new(RKEY),
action: crate::db::types::DbAction::Create,
data: StoredData::Block(bytes::Bytes::from(record_cbor)),
};
let serialized = rmp_serde::to_vec(&stored).unwrap();
let deserialized: StoredEvent = rmp_serde::from_slice(&serialized).unwrap();
let event = stored_to_event(&state, 100, deserialized, None).unwrap();
let rec = event.record.unwrap();
assert_eq!(rec.cid, Some(expected_cid));
assert!(rec.record.is_some());
}
#[test]
fn persisted_eight_byte_string_rkey_replays_verbatim() {
let tmp = tempfile::tempdir().unwrap();
let config = Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
};
let event_id = 153_752;
{
let state = AppState::new(&config).unwrap();
let did = jacquard_common::types::string::Did::new_static(DID).unwrap();
let stored = StoredEvent {
live: false,
did: TrimmedDid::from(&did),
rev: rev("3kzbif5moe22m"),
collection: CowStr::Borrowed(COL),
rkey: DbRkey::Str("stinkpot".into()),
action: crate::db::types::DbAction::Create,
data: StoredData::Block(bytes::Bytes::from_static(b"\xa0")),
};
let serialized = rmp_serde::to_vec(&stored).unwrap();
assert_eq!(
crate::util::hex::encode(&serialized),
"97c2c4100125aa8fb6f9c3a8df64f789e5ee3919c408187cee58e5450012b26170702e62736b792e666565642e706f7374a87374696e6b706f74a643726561746581a5426c6f636bc401a0"
);
let mut batch = state.db.inner.batch();
state
.db
.stream
.stage_event(&mut batch, keys::event_key(event_id), serialized);
batch.commit().unwrap();
state.db.persist().unwrap();
}
let state = AppState::new(&config).unwrap();
let replay =
read_event_replay_chunk(&state, None, Some(event_id), 8, Decoded::stored).unwrap();
assert_eq!(replay.last_seen, Some(event_id));
assert_eq!(replay.events.len(), 1);
let json = serde_json::to_value(&replay.events[0]).unwrap();
assert_eq!(json["id"], event_id);
assert_eq!(json["type"], "record");
assert_eq!(json["record"]["did"], DID);
assert_eq!(json["record"]["collection"], COL);
assert_eq!(json["record"]["rkey"], "stinkpot");
assert_eq!(json["record"]["record"], serde_json::json!({}));
}
#[test]
fn legacy_string_trimmed_did_replays_after_reopen() {
#[derive(serde::Serialize)]
struct LegacyStoredEvent<'a> {
live: bool,
did: &'a str,
rev: DbTid,
collection: &'a str,
rkey: &'a str,
action: crate::db::types::DbAction,
data: StoredData,
}
let tmp = tempfile::tempdir().unwrap();
let config = Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
};
let event_id = 153_753;
{
let state = AppState::new(&config).unwrap();
let stored = LegacyStoredEvent {
live: false,
did: DID.strip_prefix("did:").unwrap(),
rev: rev("3kzbif5moe22m"),
collection: COL,
rkey: "stinkpot",
action: crate::db::types::DbAction::Create,
data: StoredData::Block(bytes::Bytes::from_static(b"\xa0")),
};
let serialized = rmp_serde::to_vec(&stored).unwrap();
assert_eq!(
crate::util::hex::encode(&serialized),
"97c2bc706c633a65777669376e787a796f756e367a687872687336346f697ac408187cee58e5450012b26170702e62736b792e666565642e706f7374a87374696e6b706f74a643726561746581a5426c6f636bc401a0"
);
let mut batch = state.db.inner.batch();
state
.db
.stream
.stage_event(&mut batch, keys::event_key(event_id), serialized);
batch.commit().unwrap();
state.db.persist().unwrap();
}
let state = AppState::new(&config).unwrap();
let replay =
read_event_replay_chunk(&state, None, Some(event_id), 1, Decoded::stored).unwrap();
assert_eq!(replay.last_seen, Some(event_id));
assert_eq!(replay.events.len(), 1);
let json = serde_json::to_value(&replay.events[0]).unwrap();
assert_eq!(json["record"]["did"], DID);
assert_eq!(json["record"]["rkey"], "stinkpot");
}
#[test]
fn redacted_head_suppresses_queued_inline_body_but_keeps_cid_metadata() {
let (_tmp, state) = test_state();
let body = serde_ipld_dagcbor::to_vec(&serde_json::json!({"text": "private"})).unwrap();
let cid = crate::car::compute_cid(&body);
let mut batch = state.db.inner.batch();
state
.db
.indexer
.stage_head(&mut batch, record_key(), cid.to_bytes());
batch.commit().unwrap();
let did_full = jacquard_common::types::string::Did::new_static(DID).unwrap();
let stored = StoredEvent {
live: true,
did: TrimmedDid::from(&did_full),
rev: rev("3kzbif5moe22m"),
collection: CowStr::Borrowed(COL),
rkey: DbRkey::new(RKEY),
action: crate::db::types::DbAction::Create,
data: StoredData::Ptr(cid),
};
let event = stored_to_event(&state, 42, stored, Some(bytes::Bytes::from(body))).unwrap();
let record = event.record.unwrap();
assert_eq!(record.cid, Some(cid));
assert!(record.record.is_none());
}
#[test]
fn history_body_cid_mismatch_returns_none() {
let (_tmp, mut state) = test_state();
state.history_ttl = Some(std::time::Duration::from_secs(3600));
let body1 = serde_ipld_dagcbor::to_vec(&serde_json::json!({"text": "body 1"})).unwrap();
let body2 = serde_ipld_dagcbor::to_vec(&serde_json::json!({"text": "body 2"})).unwrap();
let cid1 = crate::car::compute_cid(&body1);
let cid2 = crate::car::compute_cid(&body2);
assert_ne!(cid1, cid2);
let mut batch = state.db.inner.batch();
state.db.indexer.stage_history(
&mut batch,
keys::history_key(&record_key(), &rev("3kzbif5mof33m")),
&body2,
);
batch.commit().unwrap();
let resolved = resolve(&state, 42, &rev("3kzbif5moe22m"), &cid1);
assert_eq!(resolved, None);
let did_full = jacquard_common::types::string::Did::new_static(DID).unwrap();
let stored = StoredEvent {
live: false,
did: TrimmedDid::from(&did_full),
rev: rev("3kzbif5moe22m"),
collection: CowStr::Borrowed(COL),
rkey: DbRkey::new(RKEY),
action: crate::db::types::DbAction::Create,
data: StoredData::Ptr(cid1),
};
let initial_misses = state.history_trim_misses.load(Ordering::Relaxed);
let event = stored_to_event(&state, 42, stored, None).unwrap();
let record = event.record.unwrap();
assert_eq!(record.cid, Some(cid1));
assert!(record.record.is_none());
assert_eq!(
state.history_trim_misses.load(Ordering::Relaxed),
initial_misses + 1
);
}
#[test]
fn head_body_cid_mismatch_returns_none() {
let (_tmp, mut state) = test_state();
state.history_ttl = Some(std::time::Duration::from_secs(3600));
let body1 = serde_ipld_dagcbor::to_vec(&serde_json::json!({"text": "body 1"})).unwrap();
let body2 = serde_ipld_dagcbor::to_vec(&serde_json::json!({"text": "body 2"})).unwrap();
let cid1 = crate::car::compute_cid(&body1);
let cid2 = crate::car::compute_cid(&body2);
assert_ne!(cid1, cid2);
let mut batch = state.db.inner.batch();
state
.db
.indexer
.stage_head(&mut batch, record_key(), &body2);
batch.commit().unwrap();
let resolved = resolve(&state, 42, &rev("3kzbif5moe22m"), &cid1);
assert_eq!(resolved, None);
let did_full = jacquard_common::types::string::Did::new_static(DID).unwrap();
let stored = StoredEvent {
live: false,
did: TrimmedDid::from(&did_full),
rev: rev("3kzbif5moe22m"),
collection: CowStr::Borrowed(COL),
rkey: DbRkey::new(RKEY),
action: crate::db::types::DbAction::Create,
data: StoredData::Ptr(cid1),
};
let initial_misses = state.history_trim_misses.load(Ordering::Relaxed);
let event = stored_to_event(&state, 42, stored, None).unwrap();
let record = event.record.unwrap();
assert_eq!(record.cid, Some(cid1));
assert!(record.record.is_none());
assert_eq!(
state.history_trim_misses.load(Ordering::Relaxed),
initial_misses + 1
);
}
}