//! On-disk operation log backed by redb.

use std::path::Path;

use ontoref_types::{canonical, OpId, Operation};
use redb::{Database, ReadableTable, ReadableTableMetadata, TableDefinition};

use crate::error::{Error, Result};

const OPS: TableDefinition<&[u8; 32], &[u8]> = TableDefinition::new("ops");
const HEADS: TableDefinition<&[u8; 32], ()> = TableDefinition::new("heads");

/// Append-only operation log backed by a redb database file.
pub struct OpLog {
    db: Database,
}

impl OpLog {
    /// Open or create the log at `path`. The redb file is created on first use;
    /// tables are initialised inside a transaction so subsequent reads find
    /// them.
    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
        let db = Database::create(path).map_err(Error::from)?;
        // Touch both tables in a write transaction so they exist for reads.
        let txn = db.begin_write()?;
        {
            let _ = txn.open_table(OPS)?;
            let _ = txn.open_table(HEADS)?;
        }
        txn.commit()?;
        Ok(Self { db })
    }

    /// Append an operation. Validates signature and parent existence; updates
    /// the heads set. Idempotent: re-appending a stored operation is a no-op
    /// and returns the same id.
    pub fn append(&self, op: &Operation) -> Result<OpId> {
        let id = op.id();
        op.verify_signature().map_err(|_| Error::BadSignature(id))?;

        let encoded = canonical::encode(op)?;
        let txn = self.db.begin_write()?;
        {
            let mut ops = txn.open_table(OPS)?;

            if ops.get(&id.0)?.is_some() {
                return Ok(id);
            }

            // Validate every parent exists in the log.
            let mut parent_timestamps = Vec::with_capacity(op.body.parents.len());
            for parent in &op.body.parents {
                match ops.get(&parent.0)? {
                    Some(bytes) => {
                        let parent_op: Operation = canonical::decode(bytes.value())?;
                        parent_timestamps.push(parent_op.body.timestamp);
                    }
                    None => return Err(Error::UnknownParent(*parent)),
                }
            }

            // HLC must be strictly greater than every parent's timestamp.
            if parent_timestamps
                .iter()
                .any(|parent_ts| op.body.timestamp <= *parent_ts)
            {
                return Err(Error::NonMonotonicTimestamp(id));
            }

            ops.insert(&id.0, encoded.as_slice())?;

            let mut heads = txn.open_table(HEADS)?;
            for parent in &op.body.parents {
                heads.remove(&parent.0)?;
            }
            heads.insert(&id.0, ())?;
        }
        txn.commit()?;
        Ok(id)
    }

    /// Look up an operation by id. Returns `None` if the log has no such op.
    pub fn get(&self, id: &OpId) -> Result<Option<Operation>> {
        let txn = self.db.begin_read().map_err(Error::from)?;
        let ops = txn.open_table(OPS)?;
        match ops.get(&id.0)? {
            Some(bytes) => Ok(Some(canonical::decode(bytes.value())?)),
            None => Ok(None),
        }
    }

    /// Current DAG heads — every op id not referenced by any stored op as a
    /// parent. The order returned matches the byte-lex order of the ids
    /// (redb iterates keys in sorted order), so the result is deterministic.
    pub fn heads(&self) -> Result<Vec<OpId>> {
        let txn = self.db.begin_read().map_err(Error::from)?;
        let heads = txn.open_table(HEADS)?;
        let mut out = Vec::new();
        for entry in heads.iter()? {
            let (key, _) = entry?;
            out.push(OpId(*key.value()));
        }
        Ok(out)
    }

    /// Number of operations stored.
    pub fn len(&self) -> Result<u64> {
        let txn = self.db.begin_read().map_err(Error::from)?;
        let ops = txn.open_table(OPS)?;
        Ok(ops.len()?)
    }

    /// Whether the log holds no operations. Returns `Err` if the underlying
    /// read transaction fails — propagating the error is preferable to
    /// silently reporting "non-empty" when the DB is unreadable.
    pub fn is_empty(&self) -> Result<bool> {
        Ok(self.len()? == 0)
    }

    /// Return every stored operation, sorted by `(timestamp, op_id)`. The HLC
    /// is strictly monotonic across causal edges (ADR-018-style invariant
    /// enforced by `append`), so HLC order is a valid topological order; the
    /// `op_id` tie-break makes the order total across actors that share an
    /// HLC physical tick. Used by `Substrate::open` to rebuild the in-memory
    /// commit layer from durable state.
    pub fn all_ops_in_hlc_order(&self) -> Result<Vec<Operation>> {
        let txn = self.db.begin_read().map_err(Error::from)?;
        let ops_table = txn.open_table(OPS)?;
        let mut buffer: Vec<(Operation, OpId)> = Vec::new();
        for entry in ops_table.iter()? {
            let (_, bytes) = entry?;
            let op: Operation = canonical::decode(bytes.value())?;
            let id = op.id();
            buffer.push((op, id));
        }
        buffer.sort_by(|a, b| {
            a.0.body
                .timestamp
                .cmp(&b.0.body.timestamp)
                .then_with(|| a.1.cmp(&b.1))
        });
        Ok(buffer.into_iter().map(|(op, _)| op).collect())
    }
}
