dre-core 0.7.0

//! The run engine's parts around a Binding's run ([`crate::run`]):
//!
//! - the **plan**: which Bindings run, with every profile's entry checked ([`crate::run::plan`]);
//! - the **executor**: runs a plan on a worker thread, cancellable through a [`CancelToken`],
//!   reporting what happens as [`RunEvent`]s on a channel. The terminal UI, `--log-format json`,
//!   the log file and `run_results.json` are consumers; whoever drives the run decides how
//!   events look ([`dispatch`] replays them into a [`Ui`]);
//! - the **[`RunStore`]**: where a run's files go in the target path.

use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU8, Ordering};
use std::sync::mpsc::Sender;
use std::time::Duration;

use dre_protocol::host::LogSink;
use serde_json::{Map as JsonMap, Value as Json};

use crate::run::{BindingOutcome, BindingPlan, Level, RunTargets, ShownMessage, Ui};

/// Something that happened during a run.
#[derive(Debug, Clone)]
pub enum RunEvent {
    /// The run's target and each used profile's entry, once they're all known to exist.
    Targets(RunTargets),
    /// The number of Bindings about to run.
    Plan(usize),
    BindingStart {
        report: String,
        set: Option<String>,
    },
    /// The running Binding's log file (`runs/<run id>/dre.log`), once its run folder exists.
    BindingLog(PathBuf),
    /// The schedule a Binding runs under (if any) and every var it uses.
    BindingVars {
        schedule: Option<String>,
        schedule_vars: Option<JsonMap<String, Json>>,
        vars: JsonMap<String, Json>,
    },
    /// One step inside the current Binding.
    Step {
        level: Level,
        verb: String,
        detail: String,
        elapsed: Option<Duration>,
    },
    Warn(String),
    BindingEnd(BindingOutcome),
    /// A Binding compiled: what it would do.
    Compiled(BindingPlan),
    /// `--preview`: a rendered message, shown instead of sent.
    Message(ShownMessage),
    /// A line a plugin wrote to stderr: `(plugin, line)`.
    PluginLog(String, String),
    /// The full text of a statement sent to a source: `(label, sql)`.
    Sql(String, String),
    /// Bindings running at once (`threads:`): one Binding started or finished. Sent as it
    /// happens; the Binding's own events follow later, together, as a [`RunEvent::Block`].
    Running {
        label: String,
        running: bool,
    },
    /// A per-row output's progress. Sent as it happens, never held in a block.
    Rows(crate::run::RowProgress),
    /// One Binding's events, in order, when Bindings run at once: shown together so their lines
    /// don't interleave.
    Block(Vec<RunEvent>),
}

/// Where a running Binding sends its events. Cheap to clone, and `Send`: the executor's worker
/// and plugins' stderr readers hold one each.
#[derive(Clone)]
pub struct Events {
    tx: Sender<RunEvent>,
    /// When Bindings run at once, a Binding's events are held here and sent as one block.
    held: Option<std::sync::Arc<std::sync::Mutex<Vec<RunEvent>>>>,
}

impl Events {
    pub fn new(sender: Sender<RunEvent>) -> Events {
        Events {
            tx: sender,
            held: None,
        }
    }

    /// Events that are held until [`Events::flush`], for one of several Bindings running at once.
    pub fn held(&self) -> Events {
        Events {
            tx: self.tx.clone(),
            held: Some(Default::default()),
        }
    }

    /// Send an event. Nobody listening (the run was abandoned) isn't an error.
    pub fn emit(&self, e: RunEvent) {
        match &self.held {
            Some(h) => h.lock().unwrap().push(e),
            None => {
                let _ = self.tx.send(e);
            }
        }
    }

    /// Send an event now, even from held events.
    pub fn emit_now(&self, e: RunEvent) {
        let _ = self.tx.send(e);
    }

    /// Send the held events as one block.
    pub fn flush(&self) {
        if let Some(h) = &self.held {
            let events = std::mem::take(&mut *h.lock().unwrap());
            if !events.is_empty() {
                let _ = self.tx.send(RunEvent::Block(events));
            }
        }
    }

    pub fn step(&self, level: Level, verb: &str, detail: &str, elapsed: Option<Duration>) {
        self.emit(RunEvent::Step {
            level,
            verb: verb.to_string(),
            detail: detail.to_string(),
            elapsed,
        });
    }

    pub fn warn(&self, msg: &str) {
        self.emit(RunEvent::Warn(msg.to_string()));
    }

    /// Progress is live even when the Binding's other events are held for a block.
    pub fn rows(&self, progress: crate::run::RowProgress) {
        self.emit_now(RunEvent::Rows(progress));
    }

    pub fn compiled(&self, plan: &BindingPlan) {
        self.emit(RunEvent::Compiled(plan.clone()));
    }

    pub fn message(&self, message: &ShownMessage) {
        self.emit(RunEvent::Message(message.clone()));
    }

    pub fn binding_vars(
        &self,
        schedule: Option<&str>,
        schedule_vars: Option<&JsonMap<String, Json>>,
        vars: &JsonMap<String, Json>,
    ) {
        self.emit(RunEvent::BindingVars {
            schedule: schedule.map(str::to_string),
            schedule_vars: schedule_vars.cloned(),
            vars: vars.clone(),
        });
    }

    /// Where plugin stderr goes: into the event stream.
    pub fn plugin_log(&self) -> LogSink {
        let events = self.clone();
        Arc::new(move |plugin: &str, line: &str| events.emit(RunEvent::PluginLog(plugin.into(), line.into())))
    }

    /// Where every statement's full text goes: into the event stream.
    pub fn sql_log(&self) -> LogSink {
        let events = self.clone();
        Arc::new(move |label: &str, sql: &str| events.emit(RunEvent::Sql(label.into(), sql.into())))
    }
}

/// Replay an event into a [`Ui`]. `plugin_log` and `sql_log` are the UI's own sinks, taken once
/// before the run.
pub fn dispatch(ui: &mut dyn Ui, e: RunEvent, plugin_log: &LogSink, sql_log: &LogSink) {
    match e {
        RunEvent::Targets(t) => ui.targets(&t),
        RunEvent::Plan(n) => ui.plan(n),
        RunEvent::BindingStart { report, set } => ui.binding_start(&report, set.as_deref()),
        RunEvent::BindingLog(path) => ui.binding_log(&path),
        RunEvent::BindingVars {
            schedule,
            schedule_vars,
            vars,
        } => ui.binding_vars(schedule.as_deref(), schedule_vars.as_ref(), &vars),
        RunEvent::Step {
            level,
            verb,
            detail,
            elapsed,
        } => ui.step(level, &verb, &detail, elapsed),
        RunEvent::Warn(msg) => ui.warn(&msg),
        RunEvent::BindingEnd(o) => ui.binding_end(&o),
        RunEvent::Compiled(p) => ui.compiled(&p),
        RunEvent::Message(m) => ui.message(&m),
        RunEvent::PluginLog(plugin, line) => plugin_log(&plugin, &line),
        RunEvent::Sql(label, sql) => sql_log(&label, &sql),
        RunEvent::Running { label, running } => ui.running(&label, running),
        RunEvent::Rows(p) => ui.rows(&p),
        RunEvent::Block(events) => {
            for e in events {
                dispatch(ui, e, plugin_log, sql_log);
            }
        }
    }
}

/// Why a run was asked to stop.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CancelReason {
    /// Ctrl-C (SIGINT).
    Interrupt,
    /// SIGTERM, or on Windows Ctrl-Break, closing the console, logging off or shutting down.
    Terminate,
    /// The run took longer than its timeout (`--timeout`, `DRE_RUN_TIMEOUT`, `flags: run_timeout`).
    Timeout,
}

impl CancelReason {
    /// The conventional exit code: 130 after an interrupt, 143 after a termination, 124 after a
    /// timeout (as `timeout(1)`).
    pub fn exit_code(self) -> u8 {
        match self {
            CancelReason::Interrupt => 130,
            CancelReason::Terminate => 143,
            CancelReason::Timeout => 124,
        }
    }

    fn code(self) -> u8 {
        match self {
            CancelReason::Interrupt => 1,
            CancelReason::Terminate => 2,
            CancelReason::Timeout => 3,
        }
    }
}

/// Asks a run to stop. Clones share the request; the first reason given sticks.
#[derive(Debug, Clone, Default)]
pub struct CancelToken(Arc<AtomicU8>);

impl CancelToken {
    pub fn new() -> CancelToken {
        CancelToken::default()
    }

    /// Cancel as an interrupt (Ctrl-C).
    pub fn cancel(&self) {
        self.cancel_with(CancelReason::Interrupt);
    }

    pub fn cancel_with(&self, reason: CancelReason) {
        let _ = self
            .0
            .compare_exchange(0, reason.code(), Ordering::SeqCst, Ordering::SeqCst);
    }

    pub fn is_cancelled(&self) -> bool {
        self.reason().is_some()
    }

    pub fn reason(&self) -> Option<CancelReason> {
        match self.0.load(Ordering::SeqCst) {
            1 => Some(CancelReason::Interrupt),
            2 => Some(CancelReason::Terminate),
            3 => Some(CancelReason::Timeout),
            _ => None,
        }
    }
}

/// Where a run's files go in the target path, for one Binding (`<report>/<binding>`):
/// `compiled/` (rendered SQL), `run/` (outputs and `run_results.json`) and `schema/` (the schema
/// snapshot drift is checked against).
#[derive(Debug, Clone)]
pub struct RunStore {
    target: PathBuf,
    /// Bindings running at once: each destination path a Binding delivers to, and which Binding,
    /// so two never write the same file.
    claims: Option<std::sync::Arc<std::sync::Mutex<std::collections::HashMap<String, String>>>>,
}

impl RunStore {
    pub fn new(target: &Path) -> RunStore {
        RunStore {
            target: target.to_path_buf(),
            claims: None,
        }
    }

    /// A store for Bindings running at once, which refuses a destination path two of them use.
    pub fn concurrent(target: &Path) -> RunStore {
        RunStore {
            target: target.to_path_buf(),
            claims: Some(Default::default()),
        }
    }

    /// Claim `profile`'s `path` for `binding`; the Binding that already has it, if another.
    pub fn claim(&self, profile: &str, path: &str, binding: &str) -> Option<String> {
        let mut claims = self.claims.as_ref()?.lock().unwrap();
        let owner = claims
            .entry(format!("{profile}\u{0}{path}"))
            .or_insert_with(|| binding.to_string());
        (owner != binding).then(|| owner.clone())
    }

    fn binding(&self, kind: &str, report: &str, binding: &str) -> PathBuf {
        self.target.join(kind).join(report).join(binding)
    }

    /// Where a Binding's rendered SQL goes.
    pub fn compiled_dir(&self, report: &str, binding: &str) -> PathBuf {
        self.binding("compiled", report, binding)
    }

    /// Where a Binding's outputs and `run_results.json` go.
    pub fn run_dir(&self, report: &str, binding: &str) -> PathBuf {
        self.binding("run", report, binding)
    }

    /// Where a Binding's schema snapshot goes.
    pub fn schema_dir(&self, report: &str, binding: &str) -> PathBuf {
        self.binding("schema", report, binding)
    }

    /// Write a Binding's `run_results.json` into its run folder (secrets masked).
    pub fn write_results(
        &self,
        run_dir: &Path,
        results: &crate::run_results::RunResults,
    ) -> std::io::Result<()> {
        std::fs::create_dir_all(run_dir)?;
        crate::fs::write_atomic(
            &run_dir.join("run_results.json"),
            (crate::secrets::to_json_pretty(results)? + "\n").as_bytes(),
        )
    }
}

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

    #[test]
    fn a_cancel_is_seen_by_every_clone() {
        let a = CancelToken::new();
        let b = a.clone();
        assert!(!b.is_cancelled());
        a.cancel_with(CancelReason::Terminate);
        assert!(b.is_cancelled());
        // The first reason sticks.
        b.cancel();
        assert_eq!(a.reason(), Some(CancelReason::Terminate));
        assert_eq!(CancelReason::Terminate.exit_code(), 143);
    }

    #[test]
    fn the_store_lays_out_a_bindings_folders() {
        let s = RunStore::new(Path::new("/t"));
        assert_eq!(s.run_dir("daily", "default"), Path::new("/t/run/daily/default"));
        assert_eq!(s.compiled_dir("daily", "eu"), Path::new("/t/compiled/daily/eu"));
        assert_eq!(s.schema_dir("daily", "eu"), Path::new("/t/schema/daily/eu"));
    }

    #[test]
    fn events_arrive_in_order_and_log_sinks_feed_the_stream() {
        let (tx, rx) = std::sync::mpsc::channel();
        let events = Events::new(tx);
        events.warn("one");
        events.sql_log()("q.sql:1", "select 1");
        events.plugin_log()("duckdb", "hello");
        drop(events);
        let got: Vec<String> = rx.iter().map(|e| format!("{e:?}")).collect();
        assert_eq!(got.len(), 3);
        assert!(got[0].starts_with("Warn") && got[1].starts_with("Sql") && got[2].starts_with("PluginLog"));
    }
}