//! Rhai recipe engine + host-function API.
//!
//! A `(app, target)` resolves to a `.rhai` recipe composed from a shared step
//! vocabulary. The daemon embeds Rhai and registers the host functions recipes
//! call; the recipe is the orchestration, the host functions are the
//! privileged primitives (run a command, read a secret, collect artifacts,
//! publish). Recipes are otherwise sandboxed — no arbitrary FS/network except
//! through these functions — matching the Balanced Breakfast plugin model.
//!
//! Rhai is synchronous; the engine runs each recipe on a blocking thread
//! (`spawn_blocking`, see [`crate::runner`]) and host functions bridge to async
//! work via `Handle::block_on`. That is sound only off a runtime worker thread,
//! which `spawn_blocking` guarantees.

use crate::config::Config;
use crate::domain::{AppId, Status, Step, StepRunId, Target, Version};
use crate::events::{self, Event, EventTx};
use crate::ota::{OtaRegistry, Release};
use anyhow::{Context as _, Result};
use ops_core::live_log::LiveLog;
use ops_core::remote::RemoteHost;
use rhai::{Engine, EvalAltResult, Map};
use sqlx::SqlitePool;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use tokio::runtime::Handle;
use tokio::sync::Mutex as AsyncMutex;

/// The currently-open step within a recipe run: its DB row id, which step it
/// is, and the live-log sink that `sh`/`log` stream into.
struct StepState {
    run_id: StepRunId,
    step: Step,
    log: Arc<AsyncMutex<LiveLog>>,
}

/// Everything a recipe's host functions need, shared (Arc) into each closure.
pub struct RecipeCtx {
    pub app: AppId,
    pub version: Version,
    pub target: Target,
    pub target_run_id: i64,
    pub hosts: HashMap<String, RemoteHost>,
    pub pool: SqlitePool,
    pub events: EventTx,
    pub cfg: Arc<Config>,
    pub ota: Arc<OtaRegistry>,
    pub rt: Handle,
    current: Mutex<Option<StepState>>,
}

impl RecipeCtx {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        app: AppId,
        version: Version,
        target: Target,
        target_run_id: i64,
        hosts: HashMap<String, RemoteHost>,
        pool: SqlitePool,
        events: EventTx,
        cfg: Arc<Config>,
        ota: Arc<OtaRegistry>,
        rt: Handle,
    ) -> Self {
        Self {
            app,
            version,
            target,
            target_run_id,
            hosts,
            pool,
            events,
            cfg,
            ota,
            rt,
            current: Mutex::new(None),
        }
    }

    fn now() -> String {
        chrono::Utc::now().to_rfc3339()
    }

    /// `<logs_root>/<app>/<version>/<target-with-slash-as-dash>/<step>.log`.
    fn log_path(&self, step: Step) -> PathBuf {
        let target_dir = self.target.to_string().replace('/', "-");
        self.cfg
            .logs_root
            .join(self.app.as_str())
            .join(self.version.to_string())
            .join(target_dir)
            .join(format!("{}.log", step.as_str()))
    }

    /// Close the previous step (as `Ok`), open a new one: insert its DB row,
    /// open a live log whose chunks broadcast `StepLogChunk`, emit `StepStart`.
    fn begin_step(self: &Arc<Self>, step: Step) -> Result<()> {
        self.finish_step(Status::Ok)?;
        let me = self.clone();
        let run_id = self.rt.block_on(async move {
            let started = Self::now();
            let log_ref = me.log_path(step).to_string_lossy().into_owned();
            let id: i64 = sqlx::query_scalar(
                "INSERT INTO step_runs (target_run_id, step, status, log_ref, started_at)
                 VALUES (?, ?, 'running', ?, ?) RETURNING id",
            )
            .bind(me.target_run_id)
            .bind(step.as_str())
            .bind(&log_ref)
            .bind(&started)
            .fetch_one(&me.pool)
            .await
            .context("insert step_run")?;
            sqlx::query("UPDATE target_runs SET current_step = ? WHERE id = ?")
                .bind(step.as_str())
                .bind(me.target_run_id)
                .execute(&me.pool)
                .await
                .context("update current_step")?;
            anyhow::Ok(StepRunId(id))
        })?;

        // Live log: each chunk fans out as a StepLogChunk event keyed by run_id.
        let events = self.events.clone();
        let cb_run_id = run_id;
        let log = self.rt.block_on(LiveLog::open(
            self.log_path(step),
            Box::new(move |seq, text| {
                events::emit(
                    &events,
                    Event::StepLogChunk { run_id: cb_run_id, seq, text: text.to_string() },
                );
            }),
        ));

        events::emit(
            &self.events,
            Event::StepStart {
                run_id,
                app: self.app.clone(),
                version: self.version.clone(),
                target: self.target,
                step,
            },
        );

        *self.current.lock().unwrap() = Some(StepState {
            run_id,
            step,
            log: Arc::new(AsyncMutex::new(log)),
        });
        Ok(())
    }

    /// Finalize the open step (if any): close its log, stamp the DB row, emit
    /// `StepDone`. Idempotent when no step is open.
    pub fn finish_step(self: &Arc<Self>, status: Status) -> Result<()> {
        let st = self.current.lock().unwrap().take();
        let Some(st) = st else { return Ok(()) };
        let me = self.clone();
        self.rt.block_on(async move {
            // Drop all log refs so the sink can be owned + flushed.
            if let Ok(m) = Arc::try_unwrap(st.log) {
                m.into_inner().close().await;
            }
            let _ = sqlx::query(
                "UPDATE step_runs SET status = ?, finished_at = ? WHERE id = ?",
            )
            .bind(status.as_str())
            .bind(Self::now())
            .bind(st.run_id.0)
            .execute(&me.pool)
            .await;
        });
        events::emit(
            &self.events,
            Event::StepDone {
                run_id: st.run_id,
                app: self.app.clone(),
                target: self.target,
                step: st.step,
                status,
            },
        );
        Ok(())
    }

    /// Ensure a step is open; default to `Build` if a recipe runs a command
    /// before declaring one.
    fn ensure_step(self: &Arc<Self>) -> Result<Arc<AsyncMutex<LiveLog>>> {
        if self.current.lock().unwrap().is_none() {
            self.begin_step(Step::Build)?;
        }
        Ok(self.current.lock().unwrap().as_ref().unwrap().log.clone())
    }

    /// The step currently open, or `Build` as a default for failure
    /// attribution before any step was declared.
    pub fn current_step(&self) -> Step {
        self.current.lock().unwrap().as_ref().map(|s| s.step).unwrap_or(Step::Build)
    }

    fn host(&self, name: &str) -> Result<RemoteHost> {
        self.hosts
            .get(name)
            .cloned()
            .ok_or_else(|| anyhow::anyhow!("unknown build host `{name}` (not in topology)"))
    }

    /// Run `cmd` on `host`, streaming into the current step's log. Returns
    /// exit code + a tail of stdout for the recipe to branch on.
    fn run(self: &Arc<Self>, host: &str, cmd: &str) -> Result<(i32, String)> {
        let sink = self.ensure_step()?;
        let host = self.host(host)?;
        let cmd = cmd.to_string();
        let out = self.rt.block_on(async move { host.run_streaming(&cmd, sink).await })?;
        let code = out.status.code().unwrap_or(-1);
        let stdout = String::from_utf8_lossy(&out.stdout);
        let tail: String = stdout.chars().rev().take(2000).collect::<Vec<_>>().into_iter().rev().collect();
        Ok((code, tail))
    }
}

// ----- error bridging: anyhow -> Rhai runtime error -----

fn rhai_err(e: impl std::fmt::Display) -> Box<EvalAltResult> {
    Box::new(EvalAltResult::ErrorRuntime(e.to_string().into(), rhai::Position::NONE))
}

/// Read `tauri.conf.json`'s version for `app`, from its checkout on the daemon
/// host. Used by `version_of()` and the runner's default-version path.
pub fn version_from_tauri_conf(repo: &str) -> Result<Version> {
    let path = expand_tilde(repo).join("src-tauri").join("tauri.conf.json");
    let raw = std::fs::read_to_string(&path)
        .with_context(|| format!("reading {}", path.display()))?;
    let v: serde_json::Value = serde_json::from_str(&raw).context("parsing tauri.conf.json")?;
    let ver = v.get("version").and_then(|x| x.as_str()).context("no `version` in tauri.conf.json")?;
    Version::parse(ver).map_err(|e| anyhow::anyhow!(e))
}

/// Expand a leading `~/` to `$HOME`. Paths in the topology are written with `~`.
pub fn expand_tilde(p: &str) -> PathBuf {
    if let Some(rest) = p.strip_prefix("~/") {
        if let Ok(home) = std::env::var("HOME") {
            return Path::new(&home).join(rest);
        }
    }
    PathBuf::from(p)
}

/// Build a Rhai engine with the host API bound to `ctx`. Sandboxed: recipes
/// touch the outside world only through these functions.
pub fn build_engine(ctx: Arc<RecipeCtx>) -> Engine {
    let mut engine = Engine::new();
    // Defensive caps — recipes are first-party but bound the blast radius.
    engine.set_max_operations(5_000_000);
    engine.set_max_call_levels(64);
    engine.set_max_string_size(0);

    // --- step(name) ---
    {
        let ctx = ctx.clone();
        engine.register_fn("step", move |name: &str| -> Result<(), Box<EvalAltResult>> {
            let step: Step = name.parse().map_err(rhai_err)?;
            ctx.begin_step(step).map_err(rhai_err)
        });
    }

    // --- sh(host, cmd) -> #{ code, stdout_tail } ---
    {
        let ctx = ctx.clone();
        engine.register_fn("sh", move |host: &str, cmd: &str| -> Result<Map, Box<EvalAltResult>> {
            let (code, tail) = ctx.run(host, cmd).map_err(rhai_err)?;
            let mut m = Map::new();
            m.insert("code".into(), (code as i64).into());
            m.insert("stdout_tail".into(), tail.into());
            Ok(m)
        });
    }

    // --- sh_ok(host, cmd): run + assert exit 0 ---
    {
        let ctx = ctx.clone();
        engine.register_fn("sh_ok", move |host: &str, cmd: &str| -> Result<(), Box<EvalAltResult>> {
            let (code, _) = ctx.run(host, cmd).map_err(rhai_err)?;
            if code != 0 {
                return Err(rhai_err(format!("command on `{host}` exited {code}: {cmd}")));
            }
            Ok(())
        });
    }

    // --- log(msg): operator-visible line into the current step's tail ---
    {
        let ctx = ctx.clone();
        engine.register_fn("log", move |msg: &str| -> Result<(), Box<EvalAltResult>> {
            let sink = ctx.ensure_step().map_err(rhai_err)?;
            let line = format!("[recipe] {msg}\n");
            ctx.rt.block_on(async {
                use ops_core::remote::LogSink;
                sink.lock().await.write_chunk(line.as_bytes()).await;
            });
            Ok(())
        });
    }

    // --- version_of(app) -> string ---
    {
        let ctx = ctx.clone();
        engine.register_fn("version_of", move |app: &str| -> Result<String, Box<EvalAltResult>> {
            // Only the current app is in scope; cross-app reads aren't needed.
            if app != ctx.app.as_str() {
                return Err(rhai_err(format!("version_of: `{app}` is not the app being built")));
            }
            Ok(ctx.version.to_string())
        });
    }

    // --- secret(key) -> string (file under secrets_root; never logged) ---
    {
        let ctx = ctx.clone();
        engine.register_fn("secret", move |key: &str| -> Result<String, Box<EvalAltResult>> {
            // Guard against path traversal out of secrets_root.
            if key.contains("..") || key.starts_with('/') {
                return Err(rhai_err("secret key must be a relative path without `..`"));
            }
            let path = ctx.cfg.secrets_root.join(key);
            std::fs::read_to_string(&path)
                .map(|s| s.trim_end().to_string())
                .map_err(|e| rhai_err(format!("secret `{key}`: {e}")))
        });
    }

    // --- env(host, key) -> string ---
    {
        let ctx = ctx.clone();
        engine.register_fn("env", move |host: &str, key: &str| -> Result<String, Box<EvalAltResult>> {
            // Read via the shell so it works on remote hosts too.
            let (code, tail) = ctx
                .run(host, &format!("printf '%s' \"${{{key}}}\""))
                .map_err(rhai_err)?;
            if code != 0 {
                return Err(rhai_err(format!("env `{key}` on `{host}` failed")));
            }
            Ok(tail.trim().to_string())
        });
    }

    // --- collect(host, glob, app, version): pull artifacts to dist_root ---
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "collect",
            move |host: &str, glob: &str, app: &str, version: &str| -> Result<(), Box<EvalAltResult>> {
                ctx.collect(host, glob, app, version).map_err(rhai_err)
            },
        );
    }

    // --- publish(channel, app, target, version, artifact, meta) ---
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "publish",
            move |channel: &str, app: &str, target: &str, version: &str, artifact: &str, meta: Map| -> Result<String, Box<EvalAltResult>> {
                ctx.publish(channel, app, target, version, artifact, meta).map_err(rhai_err)
            },
        );
    }

    // --- macOS signing helpers (run on the named host; exercised once the Mac
    //     keychain blocker clears — see _private/docs/bento/design.md §3/§7) ---
    register_macos_fns(&mut engine, &ctx);

    engine
}

impl RecipeCtx {
    fn collect(self: &Arc<Self>, host: &str, glob: &str, app: &str, version: &str) -> Result<()> {
        let dest = self.cfg.dist_root.join(app).join(version);
        let dest_s = dest.to_string_lossy().into_owned();
        let h = self.host(host)?;
        // The daemon does the transfer locally: cp for a local host, scp for a
        // remote one. (Remote glob expansion happens on the remote shell.)
        let cmd = if h.is_local() {
            let g = ops_core::remote::sh_quote(glob);
            let d = ops_core::remote::sh_quote(&dest_s);
            format!("mkdir -p {d} && cp -vR {g} {d}/")
        } else {
            let d = ops_core::remote::sh_quote(&dest_s);
            format!(
                "mkdir -p {d} && scp -r {flags} {tgt}:{glob} {d}/",
                flags = ops_core::remote::SSH_FLAGS.join(" "),
                tgt = h.ssh_target(),
            )
        };
        // The daemon always runs the transfer itself (local cp or local scp),
        // regardless of which host built the artifact.
        let sink = self.ensure_step()?;
        let local = RemoteHost::new("local");
        let out = self.rt.block_on(async move { local.run_streaming(&cmd, sink).await })?;
        if !out.success() {
            anyhow::bail!("collect failed (exit {:?})", out.status.code());
        }
        // Best-effort size accounting for the event.
        events::emit(
            &self.events,
            Event::ArtifactCollected {
                app: self.app.clone(),
                target: self.target,
                path: dest_s,
                bytes: dir_size(&dest).unwrap_or(0),
            },
        );
        Ok(())
    }

    fn publish(
        self: &Arc<Self>,
        channel: &str,
        app: &str,
        target: &str,
        version: &str,
        artifact: &str,
        meta: Map,
    ) -> Result<String> {
        let backend = self
            .ota
            .get(channel)
            .ok_or_else(|| anyhow::anyhow!("unknown publish channel `{channel}`"))?;
        let target: Target = target.parse().map_err(|e: String| anyhow::anyhow!(e))?;
        let version = Version::parse(version).map_err(|e| anyhow::anyhow!(e))?;
        let app = AppId::new(app);
        let notes = meta.get("notes").and_then(|v| v.clone().into_string().ok()).unwrap_or_default();
        // Resolve the artifact relative to the collected dist dir if not absolute.
        let artifact_path = {
            let p = PathBuf::from(artifact);
            if p.is_absolute() {
                p
            } else {
                self.cfg.dist_root.join(app.as_str()).join(version.to_string()).join(artifact)
            }
        };
        let rel = Release { app: &app, target, version: &version, notes };
        let receipt = backend
            .publish(&rel, &artifact_path)
            .with_context(|| format!("publish to `{channel}`"))?;
        // Record for idempotency / monotonicity.
        let me = self.clone();
        let (app_s, target_s, ver_s, chan_s) =
            (app.to_string(), target.to_string(), version.to_string(), channel.to_string());
        self.rt.block_on(async move {
            let _ = sqlx::query(
                "INSERT OR IGNORE INTO releases (app, target, version, channel, published_at)
                 VALUES (?, ?, ?, ?, ?)",
            )
            .bind(app_s)
            .bind(target_s)
            .bind(ver_s)
            .bind(chan_s)
            .bind(Self::now())
            .execute(&me.pool)
            .await;
        });
        events::emit(
            &self.events,
            Event::PublishOk { app: self.app.clone(), target: self.target, channel: channel.to_string() },
        );
        Ok(receipt)
    }
}

fn dir_size(p: &Path) -> Option<i64> {
    let mut total = 0i64;
    for entry in std::fs::read_dir(p).ok()? {
        let entry = entry.ok()?;
        let md = entry.metadata().ok()?;
        if md.is_file() {
            total += md.len() as i64;
        }
    }
    Some(total)
}

/// macOS signing/notarization host functions. Thin wrappers over the right
/// shell incantations, run on the named host. Not exercised this session (the
/// Mac keychain blocker, design §3, is uncleared) but staged so the macOS
/// recipe runs unmodified once it is.
fn register_macos_fns(engine: &mut Engine, ctx: &Arc<RecipeCtx>) {
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "verify_gatekeeper",
            move |host: &str, path: &str| -> Result<bool, Box<EvalAltResult>> {
                let (_, tail) = ctx
                    .run(host, &format!("spctl --assess -vv --type install {} 2>&1 || true", ops_core::remote::sh_quote(path)))
                    .map_err(rhai_err)?;
                Ok(tail.contains("source=Notarized Developer ID"))
            },
        );
    }
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "codesign",
            move |host: &str, identity: &str, path: &str| -> Result<(), Box<EvalAltResult>> {
                let cmd = format!(
                    "codesign --force --options runtime --timestamp --sign {} {}",
                    ops_core::remote::sh_quote(identity),
                    ops_core::remote::sh_quote(path),
                );
                let (code, _) = ctx.run(host, &cmd).map_err(rhai_err)?;
                if code != 0 {
                    return Err(rhai_err("codesign failed"));
                }
                Ok(())
            },
        );
    }
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "staple",
            move |host: &str, path: &str| -> Result<(), Box<EvalAltResult>> {
                let (code, _) = ctx
                    .run(host, &format!("xcrun stapler staple {}", ops_core::remote::sh_quote(path)))
                    .map_err(rhai_err)?;
                if code != 0 {
                    return Err(rhai_err("stapler failed"));
                }
                Ok(())
            },
        );
    }
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "notarize",
            move |host: &str, path: &str| -> Result<String, Box<EvalAltResult>> {
                ctx.notarize(host, path).map_err(rhai_err)
            },
        );
    }
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "keychain_open",
            move |host: &str, name: &str| -> Result<(), Box<EvalAltResult>> {
                // The full build-keychain lifecycle lives in dist/build-keychain.sh
                // (design §7); this drives it by name so the recipe stays short.
                let (code, _) = ctx
                    .run(host, &format!(". ~/.tauri/passwords.env && ./dist/build-keychain.sh open {}", ops_core::remote::sh_quote(name)))
                    .map_err(rhai_err)?;
                if code != 0 {
                    return Err(rhai_err("keychain_open failed"));
                }
                Ok(())
            },
        );
    }
    {
        let ctx = ctx.clone();
        engine.register_fn(
            "keychain_close",
            move |host: &str, name: &str| -> Result<(), Box<EvalAltResult>> {
                let _ = ctx.run(host, &format!("./dist/build-keychain.sh close {}", ops_core::remote::sh_quote(name)));
                Ok(())
            },
        );
    }
}

impl RecipeCtx {
    /// `xcrun notarytool submit --wait` with bounded retry (the one flaky,
    /// network-bound step). Emits `NotarizeRetry` per attempt.
    fn notarize(self: &Arc<Self>, host: &str, path: &str) -> Result<String> {
        const MAX_ATTEMPTS: u32 = 3;
        let cmd = format!(
            ". ~/.tauri/passwords.env && xcrun notarytool submit {} \
             --key \"$NOTARY_KEY\" --key-id \"$NOTARY_KEY_ID\" --issuer \"$NOTARY_ISSUER\" \
             --wait --output-format json",
            ops_core::remote::sh_quote(path),
        );
        let mut last = String::new();
        for attempt in 1..=MAX_ATTEMPTS {
            let (code, tail) = self.run(host, &cmd)?;
            if code == 0 && tail.contains("\"status\":\"Accepted\"") {
                return Ok(tail);
            }
            last = tail;
            if attempt < MAX_ATTEMPTS {
                events::emit(
                    &self.events,
                    Event::NotarizeRetry {
                        app: self.app.clone(),
                        target: self.target,
                        attempt,
                        reason: format!("exit {code}"),
                    },
                );
                self.rt.block_on(tokio::time::sleep(std::time::Duration::from_secs(15)));
            }
        }
        anyhow::bail!("notarization failed after {MAX_ATTEMPTS} attempts: {last}")
    }
}

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

    #[test]
    fn expand_tilde_handles_home() {
        unsafe { std::env::set_var("HOME", "/home/test") };
        assert_eq!(expand_tilde("~/Code/x"), PathBuf::from("/home/test/Code/x"));
        assert_eq!(expand_tilde("/abs/path"), PathBuf::from("/abs/path"));
    }
}