//! AIRGAP8 — the `airgap_events` telemetry stream.
//!
//! Design + LAW (persist-to-warehouse): every pack/fetch/unfold/install/start
//! boundary emits an [`AirgapEvent`] — the airgap twin of nornir's
//! `release_events` (`src/warehouse/release_events.rs`): one row per
//! `product × op × phase`, carrying `depends_on` so a reader can draw the live
//! bring-up DAG and see where a partial install stalled.
//!
//! This lean crate cannot pull nornir's Arrow/Iceberg stack, so it emits through
//! an [`EventSink`] seam:
//! - [`JsonlSink`] — the **real default**: appends one JSON line per event to a
//!   file UNDER THE WAREHOUSE ROOT (never `/tmp`), durable + greppable on an
//!   air-gapped box that has no warehouse server.
//! - nornir wires a `IcebergSink` over its `airgap_events` table when the fat CLI
//!   runs (see `src/warehouse/airgap_events.rs`) — same rows, real time-travel.

use std::path::{Path, PathBuf};
use std::sync::Mutex;

use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};

/// Canonical op tags (the pipeline stages).
pub mod op {
    pub const PACK: &str = "pack";
    pub const FETCH: &str = "fetch";
    pub const UNFOLD: &str = "unfold";
    pub const INSTALL: &str = "install";
    pub const START: &str = "start";
    /// One install-matrix cell's probe result (`matrix::ProbeReport` ingest).
    pub const MATRIX_PROBE: &str = "matrix_probe";
}

/// Canonical phase + status tags (mirror release_events).
pub mod phase {
    pub const START: &str = "start";
    pub const END: &str = "end";
    pub const SKIP: &str = "skip";
}
pub mod status {
    pub const OK: &str = "ok";
    pub const FAIL: &str = "fail";
    pub const RUNNING: &str = "running";
}

/// One row of the airgap-op DAG. `serde` so it ships over a thin RPC / lands in
/// the warehouse unchanged.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct AirgapEvent {
    /// Groups one install run's rows.
    pub run_id: String,
    /// Per-run monotonic ordering key (Iceberg/JSONL scan order is unspecified).
    pub seq: i64,
    /// Microseconds since the Unix epoch (UTC).
    pub ts_micros: i64,
    /// The product the op acts on (e.g. "holger"), or "" for a whole-lake op.
    pub product: String,
    /// `pack | fetch | unfold | install | start` (see [`op`]).
    pub op: String,
    /// `start | end | skip` (see [`phase`]).
    pub phase: String,
    /// `ok | fail | running` (see [`status`]).
    pub status: String,
    /// Short human message ("" when none — never a surprise null).
    pub detail: String,
    /// Products this op waited on (the dep-graph producers).
    pub depends_on: Vec<String>,
}

/// Where airgap events are durably written.
pub trait EventSink {
    fn append(&self, event: &AirgapEvent) -> Result<()>;
}

/// The real default sink: append-only JSON-lines under the warehouse root.
pub struct JsonlSink {
    path: PathBuf,
    lock: Mutex<()>,
}

impl JsonlSink {
    /// Open (create) `<warehouse_root>/airgap_events.jsonl`.
    pub fn under_warehouse(warehouse_root: &Path) -> Result<Self> {
        std::fs::create_dir_all(warehouse_root)
            .with_context(|| format!("create warehouse root {}", warehouse_root.display()))?;
        Ok(JsonlSink {
            path: warehouse_root.join("airgap_events.jsonl"),
            lock: Mutex::new(()),
        })
    }

    /// The backing file path (for tests / readers).
    pub fn path(&self) -> &Path {
        &self.path
    }
}

impl EventSink for JsonlSink {
    fn append(&self, event: &AirgapEvent) -> Result<()> {
        use std::io::Write;
        let line = serde_json::to_string(event)?;
        let _g = self.lock.lock().unwrap();
        let mut f = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.path)
            .with_context(|| format!("open {}", self.path.display()))?;
        writeln!(f, "{line}").context("append airgap_event")?;
        Ok(())
    }
}

/// Mints a run id + a monotonic `seq` and emits events to a sink. Best-effort +
/// non-fatal (mirrors `ReleaseEventEmitter`): a telemetry write must never abort
/// an otherwise-successful install.
pub struct EventEmitter<'a> {
    run_id: String,
    seq: std::cell::Cell<i64>,
    sink: &'a dyn EventSink,
}

impl<'a> EventEmitter<'a> {
    pub fn new(run_id: impl Into<String>, sink: &'a dyn EventSink) -> Self {
        EventEmitter {
            run_id: run_id.into(),
            seq: std::cell::Cell::new(0),
            sink,
        }
    }

    /// Emit one event (non-fatal: logs to stderr on sink error, does not abort).
    pub fn emit(
        &self,
        product: &str,
        op: &str,
        phase: &str,
        status: &str,
        detail: &str,
        depends_on: &[String],
    ) {
        let seq = self.seq.get();
        self.seq.set(seq + 1);
        let ev = AirgapEvent {
            run_id: self.run_id.clone(),
            seq,
            ts_micros: now_micros(),
            product: product.to_string(),
            op: op.to_string(),
            phase: phase.to_string(),
            status: status.to_string(),
            detail: detail.to_string(),
            depends_on: depends_on.to_vec(),
        };
        if let Err(e) = self.sink.append(&ev) {
            eprintln!("   ⚠ airgap_events: dropped {op}/{phase} for `{product}` (non-fatal): {e:#}");
        }
    }
}

/// Read all events back from a [`JsonlSink`] file (for status / verification).
pub fn read_jsonl(path: &Path) -> Result<Vec<AirgapEvent>> {
    let text = std::fs::read_to_string(path)
        .with_context(|| format!("read {}", path.display()))?;
    let mut out: Vec<AirgapEvent> = Vec::new();
    for line in text.lines().filter(|l| !l.trim().is_empty()) {
        out.push(serde_json::from_str(line).context("parse airgap_event line")?);
    }
    out.sort_by_key(|e| e.seq);
    Ok(out)
}

fn now_micros() -> i64 {
    chrono::Utc::now().timestamp_micros()
}

#[cfg(test)]
mod tests {
    use super::*;

    /// LAW 1 (inject-assert) + persist-to-warehouse: emit a real install DAG to a
    /// JSONL sink UNDER the warehouse root, read it back, assert the rows carry
    /// the injected op/phase/status/depends_on and are seq-ordered.
    #[test]
    fn emits_install_dag_to_warehouse_jsonl_and_reads_back() {
        let wh = tempfile::tempdir().unwrap();
        let sink = JsonlSink::under_warehouse(wh.path()).unwrap();
        // Sink lives under the warehouse root, NOT /tmp.
        assert!(sink.path().starts_with(wh.path()));
        assert!(sink.path().ends_with("airgap_events.jsonl"));

        let em = EventEmitter::new("run-42", &sink);
        em.emit("holger", op::INSTALL, phase::START, status::RUNNING, "", &[]);
        em.emit("holger", op::INSTALL, phase::END, status::OK, "installed", &[]);
        em.emit(
            "nornir-server",
            op::START,
            phase::END,
            status::OK,
            "healthy",
            &["holger".to_string()],
        );

        let rows = read_jsonl(sink.path()).unwrap();
        assert_eq!(rows.len(), 3);
        assert_eq!(rows[0].run_id, "run-42");
        assert_eq!((rows[0].seq, rows[1].seq, rows[2].seq), (0, 1, 2));
        assert_eq!(rows[1].product, "holger");
        assert_eq!(rows[1].op, op::INSTALL);
        assert_eq!(rows[1].phase, phase::END);
        assert_eq!(rows[1].status, status::OK);
        assert_eq!(rows[1].detail, "installed");
        // The bring-up edge is recorded (the live DAG).
        assert_eq!(rows[2].depends_on, vec!["holger"]);
        assert!(rows[2].ts_micros > 0);
    }
}
