//! per-mode keyspace groups. this is the single place where mode-specific //! database state is declared: each feature contributes one group struct, //! opened and initialized here, and `Db` embeds each group behind one cfg. //! //! keyspace names, tuning, and debug rendering live in `db::schema`; the //! by-name and enumeration tables are generated in `db::registry`. use fjall::{CompressionType, Database, Keyspace, KeyspaceCreateOptions}; use miette::{IntoDiagnostic, Result}; use std::cell::RefCell; use std::sync::Arc; use crate::config::Config; #[cfg(feature = "indexer")] use jacquard_common::types::did::Did; #[cfg(any( feature = "indexer", feature = "indexer_stream", feature = "jetstream", feature = "relay" ))] use super::schema::{self, Ks}; #[cfg(feature = "jetstream")] use super::sequencer::Clock; #[cfg(any(feature = "indexer_stream", feature = "relay"))] use super::sequencer::Sequence; #[cfg(any(feature = "indexer_stream", feature = "jetstream", feature = "relay"))] use miette::Context; /// everything a keyspace needs to open itself. /// /// nameable so `Schema::options` can reference it in its signature, but all /// fields and methods are `pub(super)`: it cannot be constructed or used /// outside `db`. pub struct OpenCx<'a> { pub(super) db: &'a Arc, pub(super) cfg: &'a Config, pub(super) compression: &'a dyn Fn(&str, i32) -> CompressionType, /// names opened so far, checked against the registry at the end of open. pub(super) opened: RefCell>, pub(super) poison: super::Poison, } impl OpenCx<'_> { pub(super) fn open_ks(&self, name: &str, opts: KeyspaceCreateOptions) -> Result { self.db.keyspace(name, move || opts).into_diagnostic() } pub(super) fn record_opened(&self, name: &'static str) { self.opened.borrow_mut().push(name); } } #[cfg(feature = "indexer")] #[derive(Clone)] pub struct IndexerDb { /// maps `{DID} 00 {COL} 2f {RKey}` -> DAG-CBOR body or redacted CID marker pub(super) heads: Ks, /// superseded bodies/markers: `{record key} 00 {death_rev}` -> body/CID/empty pub(super) history: Ks, /// operator-redacted versions: `{record key} 00 {cid}` -> empty pub(super) redactions: Ks, /// backfill queue of `{ID}` -> `{DID}` pub(super) pending: Ks, /// the same queue in did order, `{DID} 00 {ID}` -> empty. both are written /// only by `stage_pending_insert` and `stage_pending_remove`, which keep them /// in step pub(super) pending_dids: Ks, /// per-repo resync/retry state pub resync: Ks, /// live events buffered during backfill pub resync_buffer: Ks, } #[cfg(feature = "indexer")] impl IndexerDb { pub(super) fn open(cx: &OpenCx) -> Result { Ok(Self { heads: Ks::open(cx)?, history: Ks::open(cx)?, redactions: Ks::open(cx)?, pending: Ks::open(cx)?, pending_dids: Ks::open(cx)?, resync: Ks::open(cx)?, resync_buffer: Ks::open(cx)?, }) } pub(crate) fn head>(&self, key: K) -> fjall::Result> { self.heads.get(key) } pub(crate) fn head_prefix>(&self, prefix: K) -> fjall::Iter { self.heads.prefix(prefix) } pub(crate) fn head_range(&self, range: R) -> fjall::Iter where K: AsRef<[u8]>, R: std::ops::RangeBounds, { self.heads.range(range) } // read-side only: stream inflation is today's sole history reader #[cfg(feature = "indexer_stream")] pub(crate) fn history_range(&self, range: R) -> fjall::Iter where K: AsRef<[u8]>, R: std::ops::RangeBounds, { self.history.range(range) } /// read from table and memtable metadata, so entries that expired, were /// erased or overwritten still count until compaction reaches them pub(crate) fn approximate_history_count(&self) -> usize { self.history.approximate_len() } #[cfg(any(test, feature = "indexer_stream"))] pub(crate) fn stage_head(&self, batch: &mut fjall::OwnedWriteBatch, key: K, value: V) where K: Into, V: Into, { batch.insert(&self.heads, key, value); } #[cfg(test)] #[allow(dead_code)] pub(crate) fn stage_history(&self, batch: &mut fjall::OwnedWriteBatch, key: K, value: V) where K: Into, V: Into, { batch.insert(&self.history, key, value); } /// queues a backfill of `did` under `pending_key` pub(crate) fn stage_pending_insert( &self, batch: &mut fjall::OwnedWriteBatch, did: &Did, pending_key: &[u8], ) { let repo_key = super::keys::repo_key(did); let did_key = super::keys::pending_did_key(&repo_key, pending_key); batch.insert(&self.pending_dids, did_key, []); batch.insert(&self.pending, pending_key, repo_key); } /// drops `did`'s queue entry under `pending_key`. the did has to be the /// entry's own, or its did-order copy stays behind pub(crate) fn stage_pending_remove( &self, batch: &mut fjall::OwnedWriteBatch, did: &Did, pending_key: &[u8], ) { let repo_key = super::keys::repo_key(did); batch.remove( &self.pending_dids, super::keys::pending_did_key(&repo_key, pending_key), ); batch.remove(&self.pending, pending_key); } /// panics unless `pending_dids` holds exactly the did-order copy of every /// `pending` entry #[cfg(test)] pub(crate) fn assert_pending_dids_match(&self) { let expected = self .pending .iter() .map(|guard| { let (pending_key, repo_key) = guard.into_inner().unwrap(); super::keys::pending_did_key(&repo_key, &pending_key) }) .collect::>(); let actual = self .pending_dids .iter() .map(|guard| guard.key().unwrap().to_vec()) .collect::>(); assert_eq!(actual, expected, "pending_dids drifted from pending"); } pub(crate) fn is_pending(&self, pending_key: &[u8]) -> fjall::Result { self.pending.contains_key(pending_key) } /// the queue in `pending_key` order, each entry's value is its repo key pub(crate) fn pending_iter(&self) -> fjall::Iter { self.pending.iter() } /// the queue in did order, keys are `pending_did_key`s pub(crate) fn pending_did_range(&self, range: R) -> fjall::Iter where K: AsRef<[u8]>, R: std::ops::RangeBounds, { self.pending_dids.range(range) } } #[cfg(feature = "indexer_stream")] #[derive(Clone)] pub struct StreamDb { /// maps `{ID}` (u64 BE) -> `StoredEvent`, the source for the json stream api pub(super) events: Ks, /// bodies needed only by legacy events whose record is no longer the /// current head: `{record key} 00 {cid}` -> raw DAG-CBOR body pub(crate) event_bodies: Ks, pub(crate) event_tx: tokio::sync::broadcast::Sender, /// event ids, taken only by sequenced commits pub(crate) ids: Arc, /// a wiped db or another node has its own id, and their cursors never match here. id: Arc>, chunks: Arc>, } #[cfg(feature = "indexer_stream")] impl StreamDb { pub(super) fn open(cx: &OpenCx) -> Result { let (event_tx, _) = tokio::sync::broadcast::channel(512); Ok(Self { events: Ks::open(cx)?, event_bodies: Ks::open(cx)?, event_tx, ids: Arc::new(Sequence::new(super::keys::STREAM_EVENT_IDS_MARK)), id: Arc::default(), chunks: Arc::default(), }) } pub(crate) fn id(&self) -> &str { self.id.get().map_or("", String::as_str) } pub(crate) fn chunks(&self) -> Option<&super::stream_chunks::StreamChunks> { self.chunks.get() } pub(crate) fn event>(&self, key: K) -> fjall::Result> { self.events.get(key) } pub(crate) fn event_range(&self, range: R) -> fjall::Iter where K: AsRef<[u8]>, R: std::ops::RangeBounds, { self.events.range(range) } /// write an event row with a chosen id, for test fixtures. real events /// take their ids from the outbox when they commit. #[cfg(test)] pub(crate) fn stage_event(&self, batch: &mut fjall::OwnedWriteBatch, key: K, value: V) where K: Into, V: Into, { batch.insert(&self.events, key, value); } pub(crate) fn approximate_event_count(&self) -> usize { self.events.approximate_len() } pub(crate) fn event_head(&self) -> Result> { let Some(guard) = self.events.iter().next_back() else { return Ok(None); }; let key = guard.key().into_diagnostic()?; let id = u64::from_be_bytes( key.as_ref() .try_into() .into_diagnostic() .wrap_err("expected stream event ID to be 8 bytes")?, ); Ok(Some(id)) } pub(super) fn init( &self, marks: &Ks, cfg: &crate::config::Config, ) -> Result<()> { self.ids.resume_after(marks, self.event_head()?)?; let id = match marks.get(super::keys::STREAM_ID).into_diagnostic()? { Some(id) => String::from_utf8_lossy(&id).into_owned(), None => { let id = crate::util::hex::encode(rand::random::<[u8; 16]>()); marks .insert(super::keys::STREAM_ID, id.as_bytes()) .into_diagnostic()?; id } }; // sealed chunks can't follow ephemeral rows or trimmed history. let dir = cfg.database_path.join("stream-chunks"); if cfg.stream_chunk_span > 0 && !cfg.ephemeral && cfg.history_ttl.is_none() { let chunks = super::stream_chunks::StreamChunks::open( dir, cfg.stream_chunk_span, &id, marks.clone(), )?; let _ = self.chunks.set(chunks); } else if dir.exists() { // nothing marks what a redaction reaches, and old chunks would miss it. std::fs::remove_dir_all(&dir).into_diagnostic()?; } let _ = self.id.set(id); Ok(()) } } #[cfg(feature = "jetstream")] #[derive(Clone)] pub(crate) struct JetstreamDb { /// maps `{time_us}|{ID}` (16 bytes) -> jetstream event data pub(crate) events: Ks, pub(crate) tx: tokio::sync::broadcast::Sender, /// key ids and timestamps, taken only by sequenced commits pub(crate) ids: Arc, pub(crate) clock: Arc, } #[cfg(feature = "jetstream")] impl JetstreamDb { pub(super) fn open(cx: &OpenCx) -> Result { let (tx, _) = tokio::sync::broadcast::channel(512); Ok(Self { events: Ks::open(cx)?, tx, ids: Arc::new(Sequence::new(super::keys::JETSTREAM_IDS_MARK)), clock: Arc::new(Clock::new(super::keys::JETSTREAM_TIME_US_MARK)), }) } /// the last committed event's position. pub(crate) fn head(&self) -> Result> { let Some(guard) = self.events.iter().next_back() else { return Ok(None); }; let key = guard.key().into_diagnostic()?; let (time_us, id) = super::keys::parse_jetstream_event_key(&key)?; Ok(Some(crate::types::JetstreamPosition { time_us, id })) } /// resume jetstream ids and time watermark after the last stored event, /// or the last ones taken. pub(super) fn init(&self, marks: &Ks) -> Result<()> { let head = self.head()?; self.ids.resume_after(marks, head.map(|head| head.id))?; self.clock .resume_after(marks, head.map(|head| head.time_us)) } } #[cfg(feature = "relay")] #[derive(Clone)] pub(crate) struct RelayDb { /// maps `{SEQ}` (u64 BE) -> re-encoded relay frame pub(crate) events: Ks, /// relay seqs, taken only by sequenced commits pub(crate) seqs: Arc, pub(crate) broadcast_tx: tokio::sync::broadcast::Sender, } #[cfg(feature = "relay")] impl RelayDb { pub(super) fn open(cx: &OpenCx) -> Result { let (broadcast_tx, _) = tokio::sync::broadcast::channel(512); Ok(Self { events: Ks::open(cx)?, seqs: Arc::new(Sequence::new(super::keys::RELAY_SEQS_MARK)), broadcast_tx, }) } /// resume relay sequence numbers after the last stored frame, or the /// last one taken. pub(super) fn init(&self, marks: &Ks) -> Result<()> { let last_relay_seq = self .events .iter() .next_back() .map(|guard| { let key = guard.key().into_diagnostic()?; key.as_ref() .try_into() .map(u64::from_be_bytes) .into_diagnostic() .wrap_err("relay_events: invalid key length") }) .transpose()?; self.seqs.resume_after(marks, last_relay_seq) } } #[cfg(all(test, feature = "relay"))] mod tests { use crate::config::Config; use crate::db::Db; #[test] fn relay_seqs_resume_after_the_last_big_endian_frame_key() { let tmp = tempfile::tempdir().unwrap(); let cfg = Config { database_path: tmp.path().to_path_buf(), ..Default::default() }; { let db = Db::open(&cfg).unwrap(); db.relay .events .insert([0, 0, 0, 0, 0, 0, 1, 2], b"frame") .unwrap(); db.persist().unwrap(); } let db = Db::open(&cfg).unwrap(); assert_eq!(db.relay.seqs.next(), 259); } }