nordisk/edda

crates/nornir-release-core/src/progress.rs

//! Live progress events for a release run.
//!
//! `run_cargo_test` and `run_release_run` already log human-readable
//! lines to stdout/log file. That's fine for tailing, but useless to a
//! GUI: every consumer would have to re-parse the prose.
//!
//! Instead we emit a parallel stream of structured events as newline-
//! delimited JSON into `<log_dir>/release-run-<ts>.events.ndjson`.
//! Append-only, line-atomic, no schema migrations needed (additive
//! variants only). Consumers:
//!   * `nornir-server` exposes `GET /release/progress` as Server-Sent
//!     Events by tailing the latest file.
//!   * `nornir viz` opens the same file directly when no server is
//!     reachable (single-machine dev loop).
//!
//! Why a file and not a tokio channel? Because the release run lives
//! in the CLI process while the server lives in another process; a
//! file is the simplest shared bus that survives a server restart and
//! also works with zero server running.
//!
//! Why ndjson? Each line is a complete JSON value, so a tailing reader
//! can resume from any byte offset and still parse the next line. SSE
//! frames map 1:1 to lines, so the server forwards bytes verbatim.

use std::fs::{File, OpenOptions};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::Mutex;

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

/// One event in the release-run progress stream. All variants share
/// `ts` (RFC3339) so a consumer can sort/filter purely from JSON.
///
/// Additive contract: new fields must be `#[serde(default)]` and new
/// variants get an explicit `#[serde(rename = "...")]` tag so older
/// readers can `match _ => skip`.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "kind")]
pub enum ReleaseEvent {
    #[serde(rename = "run_start")]
    RunStart {
        ts: DateTime<Utc>,
        run_id: String,
        workspace: String,
    },
    #[serde(rename = "repo_start")]
    RepoStart {
        ts: DateTime<Utc>,
        repo: String,
        sha: String,
    },
    #[serde(rename = "phase_start")]
    PhaseStart {
        ts: DateTime<Utc>,
        repo: String,
        /// One of `build`, `test`, `publish`, `snapshot`.
        phase: String,
    },
    #[serde(rename = "phase_end")]
    PhaseEnd {
        ts: DateTime<Utc>,
        repo: String,
        phase: String,
        ok: bool,
        duration_ms: u64,
    },
    #[serde(rename = "binary_start")]
    BinaryStart {
        ts: DateTime<Utc>,
        repo: String,
        /// The full `Running ...` text cargo prints, minus the prefix.
        binary: String,
    },
    #[serde(rename = "test_pass")]
    TestPass {
        ts: DateTime<Utc>,
        repo: String,
        binary: String,
        name: String,
    },
    #[serde(rename = "test_fail")]
    TestFail {
        ts: DateTime<Utc>,
        repo: String,
        binary: String,
        name: String,
    },
    #[serde(rename = "binary_done")]
    BinaryDone {
        ts: DateTime<Utc>,
        repo: String,
        binary: String,
        passed: u32,
        failed: u32,
    },
    #[serde(rename = "repo_end")]
    RepoEnd {
        ts: DateTime<Utc>,
        repo: String,
        ok: bool,
    },
    #[serde(rename = "run_end")]
    RunEnd {
        ts: DateTime<Utc>,
        run_id: String,
        ok: bool,
    },
}

impl ReleaseEvent {
    pub fn timestamp(&self) -> DateTime<Utc> {
        match self {
            Self::RunStart { ts, .. }
            | Self::RepoStart { ts, .. }
            | Self::PhaseStart { ts, .. }
            | Self::PhaseEnd { ts, .. }
            | Self::BinaryStart { ts, .. }
            | Self::TestPass { ts, .. }
            | Self::TestFail { ts, .. }
            | Self::BinaryDone { ts, .. }
            | Self::RepoEnd { ts, .. }
            | Self::RunEnd { ts, .. } => *ts,
        }
    }
}

/// Append-only writer. Cheap to clone (wraps `Arc<Mutex<File>>`),
/// safe to share across the cargo-stdout reader thread + the async
/// pipeline driver.
#[derive(Clone, Debug)]
pub struct ProgressWriter {
    inner: std::sync::Arc<Mutex<File>>,
    path: PathBuf,
}

impl ProgressWriter {
    /// Opens (creates + appends) `<log_dir>/release-run-<run_id>.events.ndjson`.
    pub fn open(log_dir: &Path, run_id: &str) -> Result<Self> {
        std::fs::create_dir_all(log_dir)
            .with_context(|| format!("mkdir {}", log_dir.display()))?;
        let path = log_dir.join(format!("release-run-{run_id}.events.ndjson"));
        let f = OpenOptions::new()
            .create(true)
            .append(true)
            .open(&path)
            .with_context(|| format!("open {}", path.display()))?;
        Ok(Self { inner: std::sync::Arc::new(Mutex::new(f)), path })
    }

    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Best-effort: progress events are diagnostics; if the disk is
    /// full we still want the release run to continue, so failures
    /// here are logged and dropped.
    pub fn emit(&self, ev: &ReleaseEvent) {
        let line = match serde_json::to_string(ev) {
            Ok(s) => s,
            Err(e) => {
                eprintln!("progress: encode event failed: {e}");
                return;
            }
        };
        let mut guard = match self.inner.lock() {
            Ok(g) => g,
            Err(p) => p.into_inner(),
        };
        if let Err(e) = writeln!(*guard, "{line}") {
            eprintln!("progress: write {} failed: {e}", self.path.display());
        }
        let _ = guard.flush();
    }
}

/// No-op writer used when the caller hasn't wired one up — keeps the
/// emission sites unconditional.
#[derive(Clone, Default)]
pub struct NoopProgress;

impl NoopProgress {
    pub fn emit(&self, _ev: &ReleaseEvent) {}
}

/// Either a real writer or the no-op. Lets `run_cargo_test` accept an
/// `Option<&ProgressWriter>` without making every call site allocate.
pub fn now() -> DateTime<Utc> {
    Utc::now()
}