//! 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<Database>,
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<Vec<&'static str>>,
pub(super) poison: super::Poison,
}
impl OpenCx<'_> {
pub(super) fn open_ks(&self, name: &str, opts: KeyspaceCreateOptions) -> Result<Keyspace> {
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<schema::Heads>,
/// superseded bodies/markers: `{record key} 00 {death_rev}` -> body/CID/empty
pub(super) history: Ks<schema::History>,
/// operator-redacted versions: `{record key} 00 {cid}` -> empty
pub(super) redactions: Ks<schema::Redactions>,
/// backfill queue of `{ID}` -> `{DID}`
pub(super) pending: Ks<schema::Pending>,
/// 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<schema::PendingDids>,
/// per-repo resync/retry state
pub resync: Ks<schema::Resync>,
/// live events buffered during backfill
pub resync_buffer: Ks<schema::ResyncBuffer>,
}
#[cfg(feature = "indexer")]
impl IndexerDb {
pub(super) fn open(cx: &OpenCx) -> Result<Self> {
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<K: AsRef<[u8]>>(&self, key: K) -> fjall::Result<Option<fjall::Slice>> {
self.heads.get(key)
}
pub(crate) fn head_prefix<K: AsRef<[u8]>>(&self, prefix: K) -> fjall::Iter {
self.heads.prefix(prefix)
}
pub(crate) fn head_range<K, R>(&self, range: R) -> fjall::Iter
where
K: AsRef<[u8]>,
R: std::ops::RangeBounds<K>,
{
self.heads.range(range)
}
// read-side only: stream inflation is today's sole history reader
#[cfg(feature = "indexer_stream")]
pub(crate) fn history_range<K, R>(&self, range: R) -> fjall::Iter
where
K: AsRef<[u8]>,
R: std::ops::RangeBounds<K>,
{
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<K, V>(&self, batch: &mut fjall::OwnedWriteBatch, key: K, value: V)
where
K: Into<fjall::UserKey>,
V: Into<fjall::UserValue>,
{
batch.insert(&self.heads, key, value);
}
#[cfg(test)]
#[allow(dead_code)]
pub(crate) fn stage_history<K, V>(&self, batch: &mut fjall::OwnedWriteBatch, key: K, value: V)
where
K: Into<fjall::UserKey>,
V: Into<fjall::UserValue>,
{
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::<std::collections::BTreeSet<_>>();
let actual = self
.pending_dids
.iter()
.map(|guard| guard.key().unwrap().to_vec())
.collect::<std::collections::BTreeSet<_>>();
assert_eq!(actual, expected, "pending_dids drifted from pending");
}
pub(crate) fn is_pending(&self, pending_key: &[u8]) -> fjall::Result<bool> {
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<K, R>(&self, range: R) -> fjall::Iter
where
K: AsRef<[u8]>,
R: std::ops::RangeBounds<K>,
{
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<schema::Events>,
/// 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<schema::EventBodies>,
pub(crate) event_tx: tokio::sync::broadcast::Sender<crate::types::BroadcastEvent>,
/// event ids, taken only by sequenced commits
pub(crate) ids: Arc<Sequence>,
/// a wiped db or another node has its own id, and their cursors never match here.
id: Arc<std::sync::OnceLock<String>>,
chunks: Arc<std::sync::OnceLock<super::stream_chunks::StreamChunks>>,
}
#[cfg(feature = "indexer_stream")]
impl StreamDb {
pub(super) fn open(cx: &OpenCx) -> Result<Self> {
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<K: AsRef<[u8]>>(&self, key: K) -> fjall::Result<Option<fjall::Slice>> {
self.events.get(key)
}
pub(crate) fn event_range<K, R>(&self, range: R) -> fjall::Iter
where
K: AsRef<[u8]>,
R: std::ops::RangeBounds<K>,
{
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<K, V>(&self, batch: &mut fjall::OwnedWriteBatch, key: K, value: V)
where
K: Into<fjall::UserKey>,
V: Into<fjall::UserValue>,
{
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<Option<u64>> {
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<schema::Cursors>,
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<schema::JetstreamEvents>,
pub(crate) tx: tokio::sync::broadcast::Sender<crate::types::JetstreamBroadcast>,
/// key ids and timestamps, taken only by sequenced commits
pub(crate) ids: Arc<Sequence>,
pub(crate) clock: Arc<Clock>,
}
#[cfg(feature = "jetstream")]
impl JetstreamDb {
pub(super) fn open(cx: &OpenCx) -> Result<Self> {
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<Option<crate::types::JetstreamPosition>> {
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<schema::Cursors>) -> 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<schema::RelayEvents>,
/// relay seqs, taken only by sequenced commits
pub(crate) seqs: Arc<Sequence>,
pub(crate) broadcast_tx: tokio::sync::broadcast::Sender<crate::types::RelayBroadcast>,
}
#[cfg(feature = "relay")]
impl RelayDb {
pub(super) fn open(cx: &OpenCx) -> Result<Self> {
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<schema::Cursors>) -> 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);
}
}