//! keyspace schemas: the single table of truth for every keyspace.
//!
//! each keyspace is declared as one type implementing [`Schema`], carrying its
//! on-disk name, open-time tuning, optional zstd dictionary training config,
//! and debug key/value rendering. the registry (`db::registry`) derives the
//! by-name lookup, `/stats` enumeration, `/debug/*` dispatch, and dictionary
//! training from these types, so a keyspace cannot be half-registered.

use std::marker::PhantomData;

use fjall::{
    CompressionType, Keyspace, KeyspaceCreateOptions,
    config::{BlockSizePolicy, CompressionPolicy, RestartIntervalPolicy},
};
use miette::Result;
use serde_json::Value;

use super::keyspaces::OpenCx;

pub(super) const fn kb(v: u32) -> u32 {
    v * 1024
}

pub(super) const fn mb(v: u64) -> u64 {
    v * 1024 * 1024
}

/// zstd dictionary training configuration for a keyspace.
#[derive(Debug, Clone, Copy)]
pub struct TrainCfg {
    /// maximum dictionary size in bytes.
    pub dict_size: u32,
    /// number of data blocks to sample.
    pub samples: usize,
}

/// a keyspace schema: name, tuning, training, and debug rendering.
pub trait Schema {
    /// on-disk keyspace name. THIS SHOULD ALWAYS BE STABLE. DO NOT CHANGE.
    const NAME: &'static str;

    /// zstd dictionary training config; `None` disables training and
    /// dictionary loading for this keyspace.
    const TRAIN: Option<TrainCfg> = None;

    /// per-keyspace tuning applied at open.
    fn options(cx: &OpenCx) -> KeyspaceCreateOptions;

    /// render a raw value for `/debug/*` endpoints.
    fn render_value(value: &[u8]) -> Value {
        Value::String(hex::encode(value))
    }

    /// parse a user-supplied `/debug/*` key string into key bytes.
    /// `None` means the string is not a valid key for this keyspace.
    fn parse_debug_key(key: &str) -> Option<Vec<u8>> {
        Some(key.as_bytes().to_vec())
    }

    /// render raw key bytes for `/debug/iter` output.
    fn render_debug_key(key: &[u8]) -> String {
        String::from_utf8_lossy(key).into_owned()
    }
}

/// typed handle over a [`Keyspace`], parameterized by its [`Schema`].
///
/// zero-cost: one `Keyspace` handle plus a marker. cloning is as cheap as
/// cloning the underlying handle.
pub struct Ks<S: Schema> {
    inner: Keyspace,
    _s: PhantomData<fn() -> S>,
}

impl<S: Schema> Clone for Ks<S> {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
            _s: PhantomData,
        }
    }
}

impl<S: Schema> Ks<S> {
    /// open this keyspace with its schema tuning, recording the name for the
    /// open-time registry completeness check.
    pub(super) fn open(cx: &OpenCx) -> Result<Self> {
        let inner = cx.open_ks(S::NAME, S::options(cx))?;
        cx.record_opened(S::NAME);
        Ok(Self {
            inner,
            _s: PhantomData,
        })
    }

    /// an owned clone of the underlying keyspace handle.
    pub fn keyspace(&self) -> Keyspace {
        self.inner.clone()
    }

    /// the raw keyspace handle. escape hatch for migrations and debug paths.
    pub fn raw(&self) -> &Keyspace {
        &self.inner
    }

    /// point reads/writes mirror [`Keyspace`]'s signatures exactly, adding
    /// the fatal poison check every callsite previously had to remember.
    /// inherent methods win over the deref bridge, so all `db.<ks>.<op>()`
    /// callsites route through these.
    #[inline]
    pub fn get<K: AsRef<[u8]>>(&self, key: K) -> fjall::Result<Option<fjall::Slice>> {
        self.inner.get(key).inspect_err(super::check_poisoned)
    }

    #[inline]
    pub fn contains_key<K: AsRef<[u8]>>(&self, key: K) -> fjall::Result<bool> {
        self.inner
            .contains_key(key)
            .inspect_err(super::check_poisoned)
    }

    #[inline]
    pub fn insert<K: Into<fjall::UserKey>, V: Into<fjall::UserValue>>(
        &self,
        key: K,
        value: V,
    ) -> fjall::Result<()> {
        self.inner
            .insert(key, value)
            .inspect_err(super::check_poisoned)
    }

    #[inline]
    pub fn remove<K: Into<fjall::UserKey>>(&self, key: K) -> fjall::Result<()> {
        self.inner.remove(key).inspect_err(super::check_poisoned)
    }
}

/// phase-1 migration bridge: existing callsites use raw `Keyspace` methods
/// directly. removed once callsites are ported to typed accessors.
impl<S: Schema> std::ops::Deref for Ks<S> {
    type Target = Keyspace;

    fn deref(&self) -> &Keyspace {
        &self.inner
    }
}

fn msgpack_or_hex<T: serde::de::DeserializeOwned + serde::Serialize>(value: &[u8]) -> Value {
    rmp_serde::from_slice::<T>(value)
        .ok()
        .and_then(|v| serde_json::to_value(v).ok())
        .unwrap_or_else(|| Value::String(hex::encode(value)))
}

#[cfg(any(feature = "indexer_stream", feature = "relay"))]
fn u64_be_key_render(key: &[u8]) -> String {
    key.try_into()
        .map(|arr| u64::from_be_bytes(arr).to_string())
        .unwrap_or_else(|_| "invalid_u64".to_string())
}

#[cfg(any(feature = "indexer_stream", feature = "relay"))]
fn u64_be_key_parse(key: &str) -> Option<Vec<u8>> {
    key.parse::<u64>().ok().map(|id| id.to_be_bytes().to_vec())
}

/// maps `{DID}` -> `RepoState` (msgpack)
pub struct Repos;

impl Schema for Repos {
    const NAME: &'static str = "repos";
    const TRAIN: Option<TrainCfg> = Some(TrainCfg {
        dict_size: kb(64),
        samples: 1000,
    });

    fn options(cx: &OpenCx) -> KeyspaceCreateOptions {
        KeyspaceCreateOptions::default()
            // crawler checks if a repo doesn't exist
            .expect_point_read_hits(false)
            .max_memtable_size(mb(cx.cfg.db_repos_memtable_size_mb))
            // did -> repo state, not gonna be compressable well because dids are random
            // and repo state doesnt have repeats really..
            // these block sizes work fine since we insert into repos constantly anyway
            // whenever we update anything related to a repo, so the repos that arent
            // being updated will be compacted away!
            .data_block_size_policy(BlockSizePolicy::new([kb(4), kb(4), kb(16), kb(64)]))
            .data_block_compression_policy(CompressionPolicy::new([
                CompressionType::None,
                CompressionType::None,
                (cx.compression)("repos", 3),
                (cx.compression)("repos", 5),
            ]))
            // did plc are random so the interval wont rlly matter
            .data_block_restart_interval_policy(RestartIntervalPolicy::new([2, 4]))
    }

    fn render_value(value: &[u8]) -> Value {
        // RepoState borrows from the input, so it can't go through the
        // DeserializeOwned helper
        rmp_serde::from_slice::<crate::types::RepoState>(value)
            .ok()
            .and_then(|v| serde_json::to_value(&v).ok())
            .unwrap_or_else(|| Value::String(hex::encode(value)))
    }
}

/// maps `rm|{DID}` -> `RepoMetadata` (msgpack)
pub struct RepoMetadata;

impl Schema for RepoMetadata {
    const NAME: &'static str = "repo_metadata";

    fn options(cx: &OpenCx) -> KeyspaceCreateOptions {
        KeyspaceCreateOptions::default()
            .expect_point_read_hits(true)
            .max_memtable_size(mb(cx.cfg.db_repos_memtable_size_mb / 2))
            // its did -> random u64 id + bool, not much to compress, very small
            .data_block_size_policy(BlockSizePolicy::new([kb(2), kb(4), kb(8)]))
            .data_block_compression_policy(CompressionPolicy::new([
                CompressionType::None,
                CompressionType::None,
                (cx.compression)("repos", 3),
            ]))
            // did plc are random so the interval wont rlly matter
            .data_block_restart_interval_policy(RestartIntervalPolicy::new([2, 4]))
    }

    fn render_value(value: &[u8]) -> Value {
        msgpack_or_hex::<crate::types::RepoMetadata>(value)
    }
}

/// per-relay cursor keys -> u64/i64 BE bytes
pub struct Cursors;

impl Schema for Cursors {
    const NAME: &'static str = "cursors";

    fn options(_cx: &OpenCx) -> KeyspaceCreateOptions {
        KeyspaceCreateOptions::default()
            // cursor point reads hit almost 100% of the time
            .expect_point_read_hits(true)
            .max_memtable_size(mb(4))
            // its just cursors...
            .data_block_size_policy(BlockSizePolicy::all(kb(1)))
            .data_block_compression_policy(CompressionPolicy::disabled())
            .data_block_restart_interval_policy(RestartIntervalPolicy::all(1))
    }

    fn render_value(value: &[u8]) -> Value {
        render_counter(value)
    }
}

/// maps `k|{NAME}` or `r|{DID}|{COL}` -> count (u64 BE), plus delta log
pub struct Counts;

impl Schema for Counts {
    const NAME: &'static str = "counts";

    fn options(_cx: &OpenCx) -> KeyspaceCreateOptions {
        KeyspaceCreateOptions::default()
            // count increments hit because counters are mostly pre-initialized
            .expect_point_read_hits(true)
            .max_memtable_size(mb(16))
            // the data is very small
            // this is at worst did|col -> u64, so its tiny (40 bytes)
            .data_block_size_policy(BlockSizePolicy::all(kb(2)))
            .data_block_compression_policy(CompressionPolicy::disabled())
            .data_block_restart_interval_policy(RestartIntervalPolicy::all(7))
    }

    fn render_value(value: &[u8]) -> Value {
        render_counter(value)
    }
}

fn render_counter(value: &[u8]) -> Value {
    if let Ok(arr) = value.try_into() {
        return Value::Number(u64::from_be_bytes(arr).into());
    }
    std::str::from_utf8(value)
        .map(|s| Value::String(s.to_owned()))
        .unwrap_or_else(|_| Value::String(hex::encode(value)))
}

/// filter config: mode key, signal/collection/exclude sets, pds meta
pub struct Filter;

impl Schema for Filter {
    const NAME: &'static str = "filter";

    fn options(_cx: &OpenCx) -> KeyspaceCreateOptions {
        // filter handles high-volume point reads (repo excludes) so it needs the bloom filter
        KeyspaceCreateOptions::default()
            .max_memtable_size(mb(16))
            // dids arent compressable so this is fine, and we have nothing as the value
            .data_block_size_policy(BlockSizePolicy::all(kb(1)))
            .data_block_compression_policy(CompressionPolicy::disabled())
            .data_block_restart_interval_policy(RestartIntervalPolicy::all(2))
    }
}

/// crawler + firehose source urls and retry entries
pub struct Crawler;

impl Schema for Crawler {
    const NAME: &'static str = "crawler";

    fn options(_cx: &OpenCx) -> KeyspaceCreateOptions {
        KeyspaceCreateOptions::default()
            // only iterators are used here
            .expect_point_read_hits(true)
            .max_memtable_size(mb(8))
            // did -> failed state, not very compressable
            .data_block_size_policy(BlockSizePolicy::all(kb(2)))
            .data_block_compression_policy(CompressionPolicy::disabled())
            .data_block_restart_interval_policy(RestartIntervalPolicy::all(2))
    }
}

/// maps `{DID} 00 {COL} 2f {RKey}` -> DAG-CBOR body or redacted CID marker.
#[cfg(feature = "indexer")]
pub struct Heads;

#[cfg(feature = "indexer")]
impl Schema for Heads {
    const NAME: &'static str = "heads";

    fn options(cx: &OpenCx) -> KeyspaceCreateOptions {
        KeyspaceCreateOptions::default()
            // point reads might miss when using getRecord
            // but we assume thats not going to happen often...
            // since this keyspace is big, turning off bloom filters will help a lot with memory/disk space,
            // but leaves point reads vulnerable to disk I/O misses under public query load
            .expect_point_read_hits(!cx.cfg.db_records_bloom_filters)
            .max_memtable_size(mb(cx.cfg.db_records_memtable_size_mb))
            // did 00 col 2f rkey -> record body (dag-cbor). bodies compress well
            // grouped by repo: author/time locality beats the old col|cid
            // content addressing (-56% measured on a live corpus, zstd:3@64k)
            .data_block_size_policy(BlockSizePolicy::new([kb(16), kb(64), kb(128)]))
            // compress from the first level: bulk loads (migrations) write in
            // sorted order, so their tables never overlap and compaction
            // trivial-moves them without ever re-compressing — a None first
            // level sticks to the whole tree (measured: 71MB stored vs 16MB
            // zstd:3 standalone on the same bodies). heads stay hot in the
            // memtable regardless. v10 migrates pre-v10 databases into this
            // keyspace because v9's `records` persisted "never compress" into
            // fjall's on-disk config, which no open-time schema change can
            // undo.
            .data_block_compression_policy(CompressionPolicy::new([
                (cx.compression)("heads", 3),
                (cx.compression)("heads", 3),
                (cx.compression)("heads", 3),
                (cx.compression)("heads", 5),
            ]))
            .data_block_restart_interval_policy(RestartIntervalPolicy::new([8, 16, 32]))
    }

    fn render_value(value: &[u8]) -> Value {
        serde_ipld_dagcbor::from_slice::<Value>(value)
            .unwrap_or_else(|_| Value::String(hex::encode(value)))
    }

    fn render_debug_key(key: &[u8]) -> String {
        // key is {trimmed did} 00 {collection} 2f {rkey}, show as "did|col|rkey"
        let Some(did_len) = super::keys::indexer::trimmed_did_len(key) else {
            return String::from_utf8_lossy(key).into_owned();
        };
        let Ok(did) = crate::db::types::TrimmedDid::try_from(&key[..did_len]) else {
            return String::from_utf8_lossy(key).into_owned();
        };
        match super::keys::indexer::split_record_suffix(&key[did_len + 1..]) {
            Ok((col, rkey)) => format!("{did}|{col}|{rkey}"),
            Err(_) => String::from_utf8_lossy(key).into_owned(),
        }
    }

    fn parse_debug_key(key: &str) -> Option<Vec<u8>> {
        // accept "did|col|rkey" and encode as a current record key. the
        // rendered did is trimmed (no `did:` prefix), so restore it
        let (did, rest) = key.split_once('|')?;
        let (col, rkey) = rest.split_once('|')?;
        let did_str = format!("did:{did}");
        let did = jacquard_common::types::string::Did::<&str>::new(&did_str).ok()?;
        let trimmed = crate::db::types::TrimmedDid::from_did_str(did.as_str());
        Some(super::keys::indexer::record_key_trimmed(
            &trimmed,
            col,
            &crate::db::types::DbRkey::new(rkey),
        ))
    }
}

/// superseded records: `{record key} 00 {death_rev}` -> body, CID marker, or tombstone.
///
/// written when an update/delete (live commit, backfill resync diff, or repo
/// removal) replaces or removes a head body in `records`; read only by stream
/// inflation of pre-deletion events. retention is the deployment's config:
/// permanent indexers keep everything, bounded retention is drop_range-shaped.
#[cfg(feature = "indexer")]
pub struct History;

#[cfg(feature = "indexer")]
impl Schema for History {
    const NAME: &'static str = "history";

    fn options(cx: &OpenCx) -> KeyspaceCreateOptions {
        KeyspaceCreateOptions::default()
            // seeks here are speculative and usually miss, then fall through to
            // the records head; bloom filters make those misses cheap
            .expect_point_read_hits(false)
            .max_memtable_size(mb(16))
            // entries are written once and only read on replay; big blocks and
            // compression at every level
            .data_block_size_policy(BlockSizePolicy::new([kb(64), kb(128)]))
            .data_block_compression_policy(CompressionPolicy::new([
                (cx.compression)("history", 3),
                (cx.compression)("history", 3),
                (cx.compression)("history", 3),
                (cx.compression)("history", 5),
            ]))
            .data_block_restart_interval_policy(RestartIntervalPolicy::new([8, 16, 32]))
    }

    fn render_value(value: &[u8]) -> Value {
        serde_ipld_dagcbor::from_slice::<Value>(value)
            .unwrap_or_else(|_| Value::String(hex::encode(value)))
    }

    fn render_debug_key(key: &[u8]) -> String {
        // key is {record key} 00 {death_rev: 8 BE}; the rev is always the tail
        if key.len() < 9 || key[key.len() - 9] != super::keys::indexer::REC_SEP {
            return String::from_utf8_lossy(key).into_owned();
        }
        let rev_bytes: [u8; 8] = key[key.len() - 8..].try_into().expect("length checked");
        let rev = crate::db::types::DbTid::new_from_bytes(rev_bytes);
        format!("{}@{rev}", Heads::render_debug_key(&key[..key.len() - 9]))
    }
}

/// durable operator-redaction identities: `{record key} 00 {cid}` -> empty.
///
/// a marker prevents a later replay or recurrence of the same record version
/// from restoring bytes that an operator explicitly erased. entries are sparse
/// and permanent; unlike ordinary history they are never TTL-filtered.
#[cfg(feature = "indexer")]
pub struct Redactions;

#[cfg(feature = "indexer")]
impl Schema for Redactions {
    const NAME: &'static str = "redactions";

    fn options(cx: &OpenCx) -> KeyspaceCreateOptions {
        KeyspaceCreateOptions::default()
            .expect_point_read_hits(false)
            .max_memtable_size(mb(4))
            .data_block_size_policy(BlockSizePolicy::new([kb(16), kb(64)]))
            .data_block_compression_policy(CompressionPolicy::new([
                (cx.compression)("redactions", 3),
                (cx.compression)("redactions", 3),
                (cx.compression)("redactions", 3),
                (cx.compression)("redactions", 5),
            ]))
            .data_block_restart_interval_policy(RestartIntervalPolicy::new([8, 16, 32]))
    }

    fn render_value(_value: &[u8]) -> Value {
        Value::Null
    }

    fn render_debug_key(key: &[u8]) -> String {
        let Some((record_key, cid)) = super::keys::indexer::split_record_cid_key(key) else {
            return String::from_utf8_lossy(key).into_owned();
        };
        format!("{}#{cid}", Heads::render_debug_key(record_key))
    }
}

/// record-scoped bodies retained solely for legacy stream events:
/// `{record key} 00 {cid}` -> dag-cbor body.
///
/// v10 materializes these from the frozen blocks CAS before deleting it. new
/// events never write this keyspace: their bodies resolve through the current
/// record head or ordinary death-keyed history.
#[cfg(feature = "indexer_stream")]
pub struct EventBodies;

#[cfg(feature = "indexer_stream")]
impl Schema for EventBodies {
    const NAME: &'static str = "event_bodies";

    fn options(cx: &OpenCx) -> KeyspaceCreateOptions {
        KeyspaceCreateOptions::default()
            .expect_point_read_hits(false)
            .max_memtable_size(mb(16))
            .data_block_size_policy(BlockSizePolicy::new([kb(64), kb(128)]))
            .data_block_compression_policy(CompressionPolicy::new([
                (cx.compression)("event_bodies", 3),
                (cx.compression)("event_bodies", 3),
                (cx.compression)("event_bodies", 3),
                (cx.compression)("event_bodies", 5),
            ]))
            .data_block_restart_interval_policy(RestartIntervalPolicy::new([8, 16, 32]))
    }

    fn render_value(value: &[u8]) -> Value {
        serde_ipld_dagcbor::from_slice::<Value>(value)
            .unwrap_or_else(|_| Value::String(hex::encode(value)))
    }

    fn render_debug_key(key: &[u8]) -> String {
        let Some((record_key, cid)) = super::keys::indexer::split_record_cid_key(key) else {
            return String::from_utf8_lossy(key).into_owned();
        };
        format!("{}#{cid}", Heads::render_debug_key(record_key))
    }
}

/// backfill queue of `{ID}` (u64 BE) -> empty
#[cfg(feature = "indexer")]
pub struct Pending;

#[cfg(feature = "indexer")]
impl Schema for Pending {
    const NAME: &'static str = "pending";

    fn options(_cx: &OpenCx) -> KeyspaceCreateOptions {
        KeyspaceCreateOptions::default()
            // iterated over as a queue, no point reads are used so bloom filters are disabled
            .expect_point_read_hits(true)
            .max_memtable_size(mb(8))
            // its just index of id (int) -> did, and dids arent compressable (especially with the ids being random)
            .data_block_size_policy(BlockSizePolicy::all(kb(8)))
            // and we'll transition from pending to synced anyway, no point trying to compress
            .data_block_compression_policy(CompressionPolicy::disabled())
            // ids are sequential and share prefix so we can use large interval to save space
            .data_block_restart_interval_policy(RestartIntervalPolicy::all(64))
    }

    fn render_value(_value: &[u8]) -> Value {
        Value::Null
    }
}

/// per-repo resync/retry state, `{DID}` -> `ResyncState` (msgpack)
#[cfg(feature = "indexer")]
pub struct Resync;

#[cfg(feature = "indexer")]
impl Schema for Resync {
    const NAME: &'static str = "resync";

    fn options(_cx: &OpenCx) -> KeyspaceCreateOptions {
        KeyspaceCreateOptions::default()
            // we only point read in backfill when we check for existing resync state
            // ...and also in repos api. so we can disable bloom filters
            .expect_point_read_hits(true)
            .max_memtable_size(mb(8))
            // did -> error state, so its gonna be basically random, cant compress well
            .data_block_size_policy(BlockSizePolicy::all(kb(4)))
            // and we arent going to have many of these anyway, no point trying
            .data_block_compression_policy(CompressionPolicy::disabled())
            .data_block_restart_interval_policy(RestartIntervalPolicy::all(4))
    }

    fn render_value(value: &[u8]) -> Value {
        msgpack_or_hex::<crate::types::ResyncState>(value)
    }
}

/// live events buffered during backfill, `{DID}|{Rev}` -> commit (msgpack)
#[cfg(feature = "indexer")]
pub struct ResyncBuffer;

#[cfg(feature = "indexer")]
impl Schema for ResyncBuffer {
    const NAME: &'static str = "resync_buffer";

    fn options(_cx: &OpenCx) -> KeyspaceCreateOptions {
        KeyspaceCreateOptions::default()
            // iterated during backfill, no point reads
            .expect_point_read_hits(true)
            .max_memtable_size(mb(16))
            .data_block_size_policy(BlockSizePolicy::all(kb(32)))
            // dont have to compress here since resync buffer will be emptied at some point anyway
            .data_block_compression_policy(CompressionPolicy::disabled())
            .data_block_restart_interval_policy(RestartIntervalPolicy::all(16))
    }
}

/// maps `{ID}` (u64 BE) -> `StoredEvent`, the source for the json stream api
#[cfg(feature = "indexer_stream")]
pub struct Events;

#[cfg(feature = "indexer_stream")]
impl Schema for Events {
    const NAME: &'static str = "events";
    const TRAIN: Option<TrainCfg> = Some(TrainCfg {
        dict_size: kb(64),
        samples: 1000,
    });

    fn options(cx: &OpenCx) -> KeyspaceCreateOptions {
        KeyspaceCreateOptions::default()
            // only iterators are used here, no point reads
            .expect_point_read_hits(true)
            .max_memtable_size(mb(cx.cfg.db_events_memtable_size_mb))
            // the compression here wont be quite as good since events are quite random
            // eg. by many different repos and different records etc.
            // since its sequential we should still go with bigger block size though
            // backfills will be sequential though...
            .data_block_size_policy(
                cx.cfg
                    .ephemeral
                    .then(|| BlockSizePolicy::new([kb(64), kb(128), kb(256)]))
                    .unwrap_or_else(|| BlockSizePolicy::new([kb(128), kb(256)])),
            )
            // monotonic {ID} keys are trivial-moved down (never rewritten), so the L0
            // policy is permanent; the live tail is served uncompressed from the memtable
            .data_block_compression_policy(
                cx.cfg
                    .ephemeral
                    .then(|| {
                        CompressionPolicy::new([
                            CompressionType::None,
                            (cx.compression)("events", 3),
                        ])
                    })
                    .unwrap_or_else(|| {
                        CompressionPolicy::new([
                            (cx.compression)("events", 3),
                            (cx.compression)("events", 3),
                            (cx.compression)("events", 3),
                            (cx.compression)("events", 5),
                        ])
                    }),
            )
            // ids are int, we can prefix truncate a lot
            .data_block_restart_interval_policy(RestartIntervalPolicy::new([64, 128]))
    }

    fn render_value(value: &[u8]) -> Value {
        // StoredEvent borrows from the input, so it can't go through the
        // DeserializeOwned helper
        rmp_serde::from_slice::<crate::types::StoredEvent>(value)
            .ok()
            .and_then(|v| serde_json::to_value(&v).ok())
            .unwrap_or_else(|| Value::String(hex::encode(value)))
    }

    fn parse_debug_key(key: &str) -> Option<Vec<u8>> {
        u64_be_key_parse(key)
    }

    fn render_debug_key(key: &[u8]) -> String {
        u64_be_key_render(key)
    }
}

/// maps `{time_us}|{ID}` (16 bytes) -> jetstream event data
#[cfg(feature = "jetstream")]
pub struct JetstreamEvents;

#[cfg(feature = "jetstream")]
impl Schema for JetstreamEvents {
    const NAME: &'static str = "jetstream_events";
    const TRAIN: Option<TrainCfg> = Some(TrainCfg {
        dict_size: kb(64),
        samples: 1000,
    });

    fn options(cx: &OpenCx) -> KeyspaceCreateOptions {
        KeyspaceCreateOptions::default()
            // time-ordered append-only stream metadata, only iterated for replay.
            .expect_point_read_hits(true)
            .max_memtable_size(mb(cx.cfg.db_events_memtable_size_mb))
            .data_block_size_policy(BlockSizePolicy::new([kb(128), kb(256)]))
            .data_block_compression_policy(CompressionPolicy::new([
                (cx.compression)("jetstream_events", 3),
                (cx.compression)("jetstream_events", 3),
                (cx.compression)("jetstream_events", 5),
            ]))
            .data_block_restart_interval_policy(RestartIntervalPolicy::new([64, 128]))
    }

    fn render_value(value: &[u8]) -> Value {
        // StoredJetstreamEvent borrows from the input, so it can't go through
        // the DeserializeOwned helper
        rmp_serde::from_slice::<crate::types::StoredJetstreamEvent>(value)
            .ok()
            .and_then(|v| serde_json::to_value(&v).ok())
            .unwrap_or_else(|| Value::String(hex::encode(value)))
    }
}

/// maps `{SEQ}` (u64 BE) -> re-encoded relay frame
#[cfg(feature = "relay")]
pub struct RelayEvents;

#[cfg(feature = "relay")]
impl Schema for RelayEvents {
    const NAME: &'static str = "relay_events";

    fn options(cx: &OpenCx) -> KeyspaceCreateOptions {
        KeyspaceCreateOptions::default()
            // only iterated for cursor replay
            .expect_point_read_hits(true)
            .max_memtable_size(mb(cx.cfg.db_events_memtable_size_mb))
            .data_block_size_policy(BlockSizePolicy::new([kb(128), kb(256)]))
            // monotonic {SEQ} keys are trivial-moved down (never rewritten), so L0
            // compression is permanent; the live tail is served from the memtable
            .data_block_compression_policy(CompressionPolicy::new([
                (cx.compression)("relay_events", 3),
                (cx.compression)("relay_events", 3),
                (cx.compression)("relay_events", 3),
                (cx.compression)("relay_events", 5),
            ]))
            .data_block_restart_interval_policy(RestartIntervalPolicy::new([64, 128]))
    }

    fn parse_debug_key(key: &str) -> Option<Vec<u8>> {
        u64_be_key_parse(key)
    }

    fn render_debug_key(key: &[u8]) -> String {
        u64_be_key_render(key)
    }
}

/// reverse index of record references
#[cfg(feature = "backlinks")]
pub struct Backlinks;

#[cfg(feature = "backlinks")]
impl Schema for Backlinks {
    const NAME: &'static str = "backlinks";
    const TRAIN: Option<TrainCfg> = Some(TrainCfg {
        dict_size: kb(64),
        samples: 1000,
    });

    fn options(cx: &OpenCx) -> KeyspaceCreateOptions {
        KeyspaceCreateOptions::default()
            // lets assume we hit backlinks, getBacklinks will use iterator anyway
            // so we can disable bloom filter okay
            .expect_point_read_hits(true)
            .max_memtable_size(mb(cx.cfg.db_records_memtable_size_mb))
            // same as records basically
            .data_block_size_policy(BlockSizePolicy::new([kb(16), kb(32)]))
            .data_block_compression_policy(CompressionPolicy::new([
                CompressionType::None,
                (cx.compression)("backlinks", 3),
            ]))
            .data_block_restart_interval_policy(RestartIntervalPolicy::new([16, 32]))
    }
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeSet;

    #[test]
    fn keyspace_names_match_the_compiled_storage_schema() {
        let tmp = tempfile::tempdir().unwrap();
        let db = crate::db::Db::open(&crate::config::Config {
            database_path: tmp.path().to_path_buf(),
            ..Default::default()
        })
        .unwrap();

        let expected = BTreeSet::from([
            "repos",
            "repo_metadata",
            "cursors",
            "counts",
            "filter",
            "crawler",
        ]);
        #[cfg(any(
            feature = "backlinks",
            feature = "indexer",
            feature = "indexer_stream",
            feature = "jetstream",
            feature = "relay"
        ))]
        let mut expected = expected;
        #[cfg(feature = "backlinks")]
        expected.insert("backlinks");
        #[cfg(feature = "indexer")]
        expected.extend([
            "heads",
            "history",
            "redactions",
            "pending",
            "resync",
            "resync_buffer",
        ]);
        #[cfg(feature = "indexer_stream")]
        expected.extend(["events", "event_bodies"]);
        #[cfg(feature = "jetstream")]
        expected.insert("jetstream_events");
        #[cfg(feature = "relay")]
        expected.insert("relay_events");

        let actual = db
            .all_keyspaces()
            .into_iter()
            .map(|(name, _)| name)
            .collect::<BTreeSet<_>>();
        assert_eq!(actual, expected);
    }
}