nornir 0.5.3

//! Iceberg writer/reader for the **workspace-populate STEP stream**
//! (`workspace_events`, EPIC #21 POPULATE 6-8).
//!
//! Where [`super::clone_events`] records the git *clone/fetch* outcome per
//! member, this records the per-STEP outcome of a
//! [`crate::monitor::republish_with`] POPULATE pass — one row per
//! `(member, step)` plus a workspace-level `populate` summary row (member `*`).
//! It makes "what did populate actually do, and did each phase succeed" a
//! first-class, persisted, remotely-readable stream instead of scattered
//! `eprintln!`s.
//!
//! Steps: `snapshot` · `deep-scan` · `test-surface` · `arch-generate` ·
//! `populate` (the workspace-level roll-up). Status: `ok` | `error` |
//! `skipped`. Same shape and read pattern as `clone_events`, so the fat/thin
//! renderers reuse one code path.
//!
//! Write path: [`WorkspaceEventLog`] accumulates rows during a populate pass;
//! [`append_workspace_events`] flushes them in one Iceberg snapshot.
//! Best-effort + non-fatal — a logging-row failure never aborts a populate.
//!
//! Read path: [`query_workspace_events`] scans the table, scopes by workspace,
//! newest-first.
//!
//! Schema: [`super::iceberg_schema::workspace_events`].

use std::sync::Arc;

use anyhow::{anyhow, Result};
use arrow::array::{Array, Int64Array, RecordBatch, StringArray, TimestampMicrosecondArray};
use chrono::{DateTime, TimeZone, Utc};
use iceberg::arrow::schema_to_arrow_schema;

use super::iceberg::{append_batch, IcebergWarehouse, TABLE_WORKSPACE_EVENTS};

// Column indices match `iceberg_schema::workspace_events` field order.
const COL_TS_MICROS: usize = 0;
const COL_WORKSPACE: usize = 1;
const COL_MEMBER: usize = 2;
const COL_STEP: usize = 3;
const COL_STATUS: usize = 4;
const COL_DETAIL: usize = 5;
const COL_ELAPSED_MS: usize = 6;

/// Member sentinel for a workspace-level (not per-member) event row.
pub const WORKSPACE_MEMBER: &str = "*";

/// Canonical populate-step tags (the `step` column).
pub mod step {
    /// Index snapshot into Iceberg (monitor step 3).
    pub const SNAPSHOT: &str = "snapshot";
    /// Transitive dep-closure deep scan (monitor step 5, opt-in).
    pub const DEEP_SCAN: &str = "deep-scan";
    /// Test-surface discovery/logging (monitor step 6).
    pub const TEST_SURFACE: &str = "test-surface";
    /// Architecture-wiring generate + historize (monitor step 7 — folded in).
    pub const ARCH_GENERATE: &str = "arch-generate";
    /// The workspace-level populate roll-up (monitor step 8, member `*`).
    pub const POPULATE: &str = "populate";
}

/// Canonical status tags (the `status` column).
pub mod status {
    pub const OK: &str = "ok";
    pub const ERROR: &str = "error";
    pub const SKIPPED: &str = "skipped";
}

/// One row of the `workspace_events` table — a single populate STEP's outcome
/// for one member (or the workspace-level `populate` roll-up, member `*`).
///
/// `serde` derives let a server ship the rows to a thin client (mirrors
/// [`super::clone_events::CloneEventRow`]).
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct WorkspaceEventRow {
    /// Event time, microseconds since the Unix epoch (UTC).
    pub ts_micros: i64,
    /// The workspace this populate belonged to.
    pub workspace: String,
    /// The member the step ran for, or [`WORKSPACE_MEMBER`] (`*`) for a
    /// workspace-level step (`populate`, or a whole-workspace `deep-scan`).
    pub member: String,
    /// The populate step (see [`step`]).
    pub step: String,
    /// `ok` | `error` | `skipped` (see [`status`]).
    pub status: String,
    /// Free-form detail: on `ok` a summary (snapshot id, node/edge counts, …);
    /// on `error` the error chain (`{e:#}`). Stored as "" never null.
    pub detail: String,
    /// Wall-clock duration of the step, milliseconds.
    pub elapsed_ms: i64,
}

/// An accumulating populate-step log. The populate pass pushes one row per step
/// as it runs, then [`flush`](Self::flush) appends them in a single Iceberg
/// snapshot at the end (best-effort, non-fatal).
#[derive(Debug, Default)]
pub struct WorkspaceEventLog {
    workspace: String,
    rows: Vec<WorkspaceEventRow>,
}

impl WorkspaceEventLog {
    pub fn new(workspace: impl Into<String>) -> Self {
        Self { workspace: workspace.into(), rows: Vec::new() }
    }

    /// Record one step outcome (stamped now, UTC micros).
    pub fn record(
        &mut self,
        member: impl Into<String>,
        step_: &str,
        status_: &str,
        detail: impl Into<String>,
        elapsed_ms: i64,
    ) {
        self.rows.push(WorkspaceEventRow {
            ts_micros: Utc::now().timestamp_micros(),
            workspace: self.workspace.clone(),
            member: member.into(),
            step: step_.to_string(),
            status: status_.to_string(),
            detail: detail.into(),
            elapsed_ms,
        });
    }

    /// Record a workspace-level step (member `*`).
    pub fn record_workspace(
        &mut self,
        step_: &str,
        status_: &str,
        detail: impl Into<String>,
        elapsed_ms: i64,
    ) {
        self.record(WORKSPACE_MEMBER, step_, status_, detail, elapsed_ms);
    }

    /// The rows accumulated so far.
    pub fn rows(&self) -> &[WorkspaceEventRow] {
        &self.rows
    }

    /// True when a `member`/`step` pair recorded an `error`.
    pub fn any_error(&self) -> bool {
        self.rows.iter().any(|r| r.status == status::ERROR)
    }

    /// Append the accumulated rows to `workspace_events` in one snapshot.
    /// Best-effort + **non-fatal**: a warehouse hiccup is logged and swallowed
    /// so a step-logging failure never aborts an otherwise-successful populate.
    pub fn flush(&self, wh: &IcebergWarehouse) {
        if self.rows.is_empty() {
            return;
        }
        if let Err(e) = wh.block_on(append_workspace_events(wh, &self.rows)) {
            eprintln!(
                "   ⚠ workspace_events: dropped {} populate-step row(s) for `{}` (non-fatal): {e:#}",
                self.rows.len(),
                self.workspace,
            );
        }
    }
}

/// Append a batch of `workspace_events` rows (one Iceberg snapshot). Mirrors
/// [`super::clone_events::append_clone_events`].
pub async fn append_workspace_events(
    wh: &IcebergWarehouse,
    rows: &[WorkspaceEventRow],
) -> Result<()> {
    if rows.is_empty() {
        return Ok(());
    }
    // Blank/torn metadata JSON self-heals (drop+recreate the 0-row table).
    let table = wh
        .load_table_for_append(
            TABLE_WORKSPACE_EVENTS,
            super::iceberg_schema::workspace_events()?,
            &[],
        )
        .await?;
    let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);

    let cols: Vec<Arc<dyn Array>> = vec![
        Arc::new(
            TimestampMicrosecondArray::from(rows.iter().map(|r| r.ts_micros).collect::<Vec<_>>())
                .with_timezone("+00:00"),
        ),
        Arc::new(StringArray::from(rows.iter().map(|r| r.workspace.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(rows.iter().map(|r| r.member.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(rows.iter().map(|r| r.step.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(rows.iter().map(|r| r.status.clone()).collect::<Vec<_>>())),
        Arc::new(StringArray::from(rows.iter().map(|r| r.detail.clone()).collect::<Vec<_>>())),
        Arc::new(Int64Array::from(rows.iter().map(|r| r.elapsed_ms).collect::<Vec<_>>())),
    ];
    let batch = RecordBatch::try_new(arrow_schema, cols)?;
    append_batch(wh.catalog(), table, batch).await?;
    Ok(())
}

/// Which workspace events to read.
#[derive(Debug, Clone)]
pub enum WorkspaceEventSelector {
    /// Every event whose `workspace` matches.
    Workspace(String),
    /// Everything in the table.
    All,
}

/// Read workspace events, scoped by `sel`, returned **newest-first**
/// (`ts_micros` descending). Iceberg gives no scan order, so the reader sorts.
pub async fn query_workspace_events(
    wh: &IcebergWarehouse,
    sel: &WorkspaceEventSelector,
) -> Result<Vec<WorkspaceEventRow>> {
    let batches: Vec<RecordBatch> =
        super::iceberg::load_and_read_all(wh, TABLE_WORKSPACE_EVENTS).await?;

    let mut out: Vec<WorkspaceEventRow> = Vec::new();
    for b in &batches {
        let ts = col_ts(b, COL_TS_MICROS)?;
        let workspace = col_str(b, COL_WORKSPACE)?;
        let member = col_str(b, COL_MEMBER)?;
        let step_ = col_str(b, COL_STEP)?;
        let st = col_str(b, COL_STATUS)?;
        let detail = col_str(b, COL_DETAIL)?;
        let elapsed = col_i64(b, COL_ELAPSED_MS)?;
        for i in 0..b.num_rows() {
            let row = WorkspaceEventRow {
                ts_micros: ts.value(i),
                workspace: workspace.value(i).to_string(),
                member: member.value(i).to_string(),
                step: step_.value(i).to_string(),
                status: st.value(i).to_string(),
                detail: detail.value(i).to_string(),
                elapsed_ms: elapsed.value(i),
            };
            let keep = match sel {
                WorkspaceEventSelector::Workspace(w) => &row.workspace == w,
                WorkspaceEventSelector::All => true,
            };
            if keep {
                out.push(row);
            }
        }
    }
    out.sort_by(|a, b| b.ts_micros.cmp(&a.ts_micros));
    Ok(out)
}

/// Render populate step events human-readably: newest first, each line
/// `mark step member — detail (elapsed)`.
pub fn render_events(rows: &[WorkspaceEventRow]) -> String {
    if rows.is_empty() {
        return "(no workspace populate events recorded)\n".to_string();
    }
    let mut out = String::new();
    for r in rows {
        let mark = match r.status.as_str() {
            status::OK => "✓",
            status::ERROR => "✗",
            status::SKIPPED => "·",
            _ => "?",
        };
        let when = ts_to_rfc3339(r.ts_micros);
        let detail = if r.detail.is_empty() {
            String::new()
        } else {
            format!(" — {}", r.detail)
        };
        out.push_str(&format!(
            "{mark} {when}  {step:<14} {member}{detail} ({elapsed}ms)\n",
            step = r.step,
            member = r.member,
            elapsed = r.elapsed_ms,
        ));
    }
    out
}

// ─── column helpers ─────────────────────────────────────────────────────

fn col_str(b: &RecordBatch, idx: usize) -> Result<&StringArray> {
    b.column(idx)
        .as_any()
        .downcast_ref::<StringArray>()
        .ok_or_else(|| anyhow!("workspace_events col {idx} is not StringArray"))
}

fn col_i64(b: &RecordBatch, idx: usize) -> Result<&Int64Array> {
    b.column(idx)
        .as_any()
        .downcast_ref::<Int64Array>()
        .ok_or_else(|| anyhow!("workspace_events col {idx} is not Int64Array"))
}

fn col_ts(b: &RecordBatch, idx: usize) -> Result<&TimestampMicrosecondArray> {
    b.column(idx)
        .as_any()
        .downcast_ref::<TimestampMicrosecondArray>()
        .ok_or_else(|| anyhow!("workspace_events col {idx} is not TimestampMicrosecondArray"))
}

/// Format `ts_micros` as RFC3339 (UTC).
pub fn ts_to_rfc3339(ts_micros: i64) -> String {
    let dt: DateTime<Utc> = Utc.timestamp_micros(ts_micros).single().unwrap_or_else(Utc::now);
    dt.to_rfc3339()
}

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

    /// Inject-and-assert: a populate pass logs per-step + a workspace roll-up,
    /// flushes once, and reads back newest-first, scoped by workspace, with the
    /// (member, step, status, detail) round-tripping exactly.
    #[test]
    fn populate_steps_persist_and_read_back_scoped() {
        let dir = tempfile::tempdir().unwrap();
        let wh = IcebergWarehouse::open(dir.path()).unwrap();

        let mut log = WorkspaceEventLog::new("nordisk");
        log.record("facett", step::SNAPSHOT, status::OK, "snapshot-7", 12);
        log.record("facett", step::ARCH_GENERATE, status::OK, "nodes=40 edges=88", 30);
        log.record("korp", step::ARCH_GENERATE, status::ERROR, "scan failed: boom", 5);
        log.record_workspace(step::POPULATE, status::ERROR, "1/2 members clean", 47);
        assert!(log.any_error());

        log.flush(&wh);

        let rows = wh
            .block_on(query_workspace_events(&wh, &WorkspaceEventSelector::Workspace("nordisk".into())))
            .unwrap();
        assert_eq!(rows.len(), 4, "one row per recorded step");

        // The failed member/step is first-class and its error round-trips.
        let failed: Vec<_> = rows.iter().filter(|r| r.status == status::ERROR).collect();
        assert_eq!(failed.len(), 2, "korp arch + the populate roll-up");
        let korp = rows.iter().find(|r| r.member == "korp").unwrap();
        assert_eq!(korp.step, "arch-generate");
        assert!(korp.detail.contains("boom"), "error chain readable: {}", korp.detail);

        // The workspace-level roll-up row (member `*`).
        let rollup = rows.iter().find(|r| r.member == WORKSPACE_MEMBER).unwrap();
        assert_eq!(rollup.step, "populate");
        assert_eq!(rollup.status, "error");

        // Another workspace's scope sees nothing.
        let other = wh
            .block_on(query_workspace_events(&wh, &WorkspaceEventSelector::Workspace("holger".into())))
            .unwrap();
        assert!(other.is_empty(), "workspace scope isolates events");

        // Human render + JSON RPC contract.
        let txt = render_events(&rows);
        assert!(txt.contains("arch-generate") && txt.contains("korp") && txt.contains('✗'));
        let json = serde_json::to_string(&rows).unwrap();
        let back: Vec<WorkspaceEventRow> = serde_json::from_str(&json).unwrap();
        assert_eq!(back, rows);
    }

    #[test]
    fn empty_log_flush_is_a_noop() {
        let dir = tempfile::tempdir().unwrap();
        let wh = IcebergWarehouse::open(dir.path()).unwrap();
        let log = WorkspaceEventLog::new("nordisk");
        log.flush(&wh); // must not error / create rows
        let rows =
            wh.block_on(query_workspace_events(&wh, &WorkspaceEventSelector::All)).unwrap();
        assert!(rows.is_empty());
    }
}