//! 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);
    }
}