//! `Substrate` — integrator over L1-L5.

use std::path::PathBuf;

use ontoref_blobs::BlobStore;
use ontoref_commit::{CommitLayer, MergePolicy, StateRoot};
use ontoref_oplog::OpLog;
use ontoref_query::QueryEngine;
use ontoref_triples::TripleStore;
use ontoref_types::{canonical, operation::OpPayload, Operation};

use crate::error::{Error, Result};
use crate::receipt::Receipt;
use crate::snapshot::Snapshot;

/// Filename of the durable commit-layer snapshot under the substrate root.
const SNAPSHOT_FILE: &str = "commit-snapshot.bin";

/// Composed substrate: one project root containing the on-disk blobs, op log,
/// triple store, and the in-memory commit + query layers.
pub struct Substrate {
    root: PathBuf,
    blobs: BlobStore,
    oplog: OpLog,
    query: QueryEngine,
    commit: CommitLayer,
    /// Per-cell CRDT merge policy (ADR-041). Retained so the commit layer is
    /// rebuilt with the same folds after a snapshot restore or an ingest.
    policy: MergePolicy,
}

/// Outcome of [`Substrate::ingest_with`]: how many peer ops were admitted
/// and which were quarantined by the semantic validator, with the reason.
#[derive(Debug, Default, Clone)]
pub struct IngestReport {
    /// Count of newly-appended (admitted) operations.
    pub ingested: usize,
    /// Ops the validator rejected — `(op id, reason)`. Not appended, not
    /// applied; the local state is unaffected by them.
    pub rejected: Vec<(ontoref_types::OpId, String)>,
}

impl Substrate {
    /// Open or create a substrate rooted at `root`. The on-disk subdirectories
    /// (`blobs/`, `oplog.redb`, `triples.redb`) are created lazily; the
    /// commit layer is rebuilt from the op log so a process restart restores
    /// the same state root the previous process produced.
    pub fn open(root: impl Into<PathBuf>) -> Result<Self> {
        Self::open_with_policy(root, MergePolicy::default())
    }

    /// Like [`open`](Self::open) but installs a per-cell CRDT merge policy
    /// (ADR-041). The caller — which can read the catalog's domain
    /// declarations — derives the attribute→strategy map and passes it here,
    /// so `ontoref-core` stays free of the NCL catalog. An empty policy is
    /// identical to [`open`](Self::open) (all cells fold last-writer-wins).
    pub fn open_with_policy(root: impl Into<PathBuf>, policy: MergePolicy) -> Result<Self> {
        let root = root.into();
        std::fs::create_dir_all(&root).map_err(Error::Io)?;

        let blobs = BlobStore::open(root.join("blobs"))?;
        let oplog = OpLog::open(root.join("oplog.redb"))?;
        let triples = TripleStore::open(root.join("triples.redb"))?;

        // Restore the in-memory commit layer from the latest snapshot when
        // present, then replay only operations appended after its boundary
        // (bl-015). Without a snapshot this is a full replay — the prior
        // behaviour. The triple store is durable (redb), so snapshotted ops
        // are already materialised there and need no re-apply.
        let (mut commit, boundary) = match Snapshot::load(&root.join(SNAPSHOT_FILE))? {
            Some(snap) => (
                CommitLayer::from_cells(snap.cells).with_policy(policy.clone()),
                snap.boundary,
            ),
            None => (CommitLayer::new().with_policy(policy.clone()), None),
        };
        for op in oplog.all_ops_in_hlc_order()? {
            if let Some(b) = boundary {
                if (op.body.timestamp, op.id()) <= b {
                    continue;
                }
            }
            // Replay through L3 (idempotent: redb insert overwrites with the
            // same canonical bytes) and L4 to restore the in-memory state.
            triples.apply(&op)?;
            commit.apply(&op);
        }

        let query = QueryEngine::new(triples);
        Ok(Self {
            root,
            blobs,
            oplog,
            query,
            commit,
            policy,
        })
    }

    /// Apply an operation through the pipeline. Returns a `Receipt` carrying
    /// the op id, the canonical-encoded blob id, the current state root, and
    /// one Merkle witness per touched cell.
    pub fn apply(&mut self, op: &Operation) -> Result<Receipt> {
        let encoded = canonical::encode(op)?;
        let blob_id = self.blobs.put(&encoded)?;
        let op_id = self.oplog.append(op)?;
        self.query.apply(op)?;
        self.commit.apply(op);

        let state_root = self.commit.root();
        let witnesses = self.collect_witnesses(op)?;

        Ok(Receipt {
            blob_id,
            op_id,
            state_root,
            witnesses,
        })
    }

    /// Ingest operations produced by other actors (ADR-027 multi-actor
    /// convergence, bl-018). Each op is appended to the local oplog — which
    /// re-validates its signature, parent existence, and monotonic HLC
    /// (ADR-026 structural plane) — and, when newly added, folded into the
    /// triple store. The commit layer is then rebuilt in HLC order so the
    /// resulting `state_root` is the deterministic last-writer-wins-by-HLC
    /// merge: two actors that ingest each other's concurrent writes to the
    /// same cell converge to the identical root regardless of arrival order.
    /// Idempotent — re-ingesting a stored op is a no-op. Returns the count
    /// of newly-appended operations.
    ///
    /// # Errors
    ///
    /// [`Error::OpLog`] if an op fails append-time validation;
    /// [`Error::Triples`]/[`Error::Commit`] on the rebuild.
    pub fn ingest(&mut self, ops: &[Operation]) -> Result<usize> {
        let report = self.ingest_with(ops, |_| Ok(()))?;
        Ok(report.ingested)
    }

    /// Like [`ingest`](Self::ingest) but runs `validate` on each peer op as
    /// a semantic gate before admitting it (bl-030). `oplog.append` already
    /// re-validates structurally (signature, parents, monotonic HLC —
    /// ADR-026 structural plane); `validate` is where a caller layers the
    /// SEMANTIC plane — domain validators that may inspect current state —
    /// so a CRDT merge cannot silently admit state a validator would have
    /// rejected (the bl-018 risk). An op whose `validate` returns `Err` is
    /// quarantined: not appended, not applied, recorded in the report.
    /// `ontoref-core` stays free of the operations layer — the validator is
    /// injected, so the substrate never depends on `ontoref-ops`.
    ///
    /// # Errors
    ///
    /// [`Error::OpLog`] if an admitted op fails append-time validation;
    /// [`Error::Triples`]/[`Error::Commit`] on the rebuild.
    pub fn ingest_with<F>(&mut self, ops: &[Operation], mut validate: F) -> Result<IngestReport>
    where
        F: FnMut(&Operation) -> std::result::Result<(), String>,
    {
        let mut report = IngestReport::default();
        for op in ops {
            if let Err(reason) = validate(op) {
                report.rejected.push((op.id(), reason));
                continue;
            }
            let before = self.oplog.len()?;
            self.oplog.append(op)?;
            if self.oplog.len()? > before {
                self.query.apply(op)?;
                report.ingested += 1;
            }
        }
        if report.ingested > 0 {
            // Re-derive the commit layer in HLC order; cell values resolve
            // to the highest-HLC assertion (HlcLastWriterWins), so the root
            // is independent of the order ops arrived across actors.
            self.commit = self.replay_commit()?;
        }
        Ok(report)
    }

    /// Rebuild the commit layer from the durable oplog in HLC order,
    /// restoring the latest snapshot first and replaying only ops after its
    /// boundary (bl-015). Used after [`ingest`](Self::ingest) folds peer ops
    /// in out of order.
    fn replay_commit(&self) -> Result<CommitLayer> {
        let (mut commit, boundary) = match Snapshot::load(&self.root.join(SNAPSHOT_FILE))? {
            Some(snap) => (
                CommitLayer::from_cells(snap.cells).with_policy(self.policy.clone()),
                snap.boundary,
            ),
            None => (CommitLayer::new().with_policy(self.policy.clone()), None),
        };
        for op in self.oplog.all_ops_in_hlc_order()? {
            if let Some(b) = boundary {
                if (op.body.timestamp, op.id()) <= b {
                    continue;
                }
            }
            commit.apply(&op);
        }
        Ok(commit)
    }

    fn collect_witnesses(&self, op: &Operation) -> Result<Vec<ontoref_commit::Witness>> {
        match &op.body.payload {
            // Noop, Witness, TierDescent, and KeySuccession change no cells;
            // Retract removes cells, so there is no included cell to prove
            // (a removal has no inclusion witness). All expose none.
            OpPayload::Noop { .. }
            | OpPayload::Witness { .. }
            | OpPayload::TierDescent { .. }
            | OpPayload::KeySuccession { .. }
            | OpPayload::Retract { .. } => Ok(Vec::new()),
            OpPayload::EntityAssert { attrs, entity, .. } => {
                let mut witnesses = Vec::with_capacity(attrs.len());
                for (attr, _) in attrs {
                    witnesses.push(self.commit.witness(entity, attr)?);
                }
                Ok(witnesses)
            }
        }
    }

    /// Current commit-layer state root without applying anything.
    pub fn state_root(&self) -> StateRoot {
        self.commit.root()
    }

    /// Persist a durable snapshot of the current commit layer (bl-015) to
    /// `<root>/commit-snapshot.bin`. A later [`Substrate::open`] restores it
    /// and replays only operations appended after the snapshot boundary, so
    /// recovery cost is bounded by ops-since-snapshot. The boundary is the
    /// last operation in the oplog's HLC order, which `open`/`apply` have
    /// already folded into the commit layer.
    ///
    /// # Errors
    ///
    /// [`Error::OpLog`] reading the order; [`Error::Canonical`]/[`Error::Io`]
    /// from the atomic, fsync'd write.
    pub fn write_snapshot(&self) -> Result<()> {
        let ops = self.oplog.all_ops_in_hlc_order()?;
        let boundary = ops.last().map(|op| (op.body.timestamp, op.id()));
        let snapshot = Snapshot {
            boundary,
            cells: self.commit.cells().clone(),
            state_root: self.commit.root(),
        };
        snapshot.write(&self.root.join(SNAPSHOT_FILE))
    }

    /// Register a subscription on the query engine.
    pub fn subscribe<F>(&mut self, handler: F) -> usize
    where
        F: FnMut(&Operation) + Send + 'static,
    {
        self.query.subscribe(handler)
    }

    /// Borrow the query engine (read-only views into L3 via L5).
    pub fn query(&self) -> &QueryEngine {
        &self.query
    }

    /// Borrow the op log (introspection, replay, debugging).
    pub fn oplog(&self) -> &OpLog {
        &self.oplog
    }

    /// Borrow the blob store (raw content-addressed retrieval).
    pub fn blobs(&self) -> &BlobStore {
        &self.blobs
    }

    /// On-disk root of this substrate.
    pub fn root(&self) -> &std::path::Path {
        &self.root
    }
}
