//! The [`Executor`] trait: a handle scoped to one `(host, capability set)`.
//!
//! Callers hold executors as `Arc<dyn Executor>` (one per node, built from
//! topology) so the concrete transport — [`crate::LocalExec`],
//! [`crate::SshExec`], or `AgentRpc` — is chosen per host without the caller
//! caring which.

use crate::capability::CapabilitySet;
use crate::remote::{LogSink, RunOutput};
use crate::step::{ObserveKind, Step};
use anyhow::Result;
use async_trait::async_trait;
use std::path::Path;
use tokio::process::Command;

/// Options controlling an rsync push/pull. Default = a plain `-az --partial`
/// mirror that never prunes the destination (safe for artifact collection).
/// Sando's release-dir deploy opts in to `delete` + `chmod` to keep its exact
/// behavior.
#[derive(Clone, Debug)]
pub struct SyncOpts {
    /// `--delete`: prune files on the destination that are gone from the
    /// source (a true mirror). Off by default.
    pub delete: bool,
    /// `--chmod=<spec>`: force destination permissions per-file.
    pub chmod: Option<String>,
    /// `-z`: compress in flight. On by default, which is right for source trees
    /// and release dirs. Turn it OFF for already-compressed payloads (a `.gz`
    /// dump, a `.dmg`) — `-z` then burns CPU on both ends to save ~nothing.
    pub compress: bool,
    /// `--partial`: keep a partially-transferred file so a retry can resume it.
    /// On by default (worth it for a multi-hundred-MB artifact over a flaky
    /// link). Turn it OFF when a truncated leftover is *dangerous* rather than
    /// merely useless — e.g. fetching a DB dump, where a resumed transfer could
    /// splice two different dumps into one plausible-looking file.
    pub partial: bool,
}

impl Default for SyncOpts {
    fn default() -> Self {
        Self {
            delete: false,
            chmod: None,
            compress: true,
            partial: true,
        }
    }
}

impl SyncOpts {
    /// The Sando release-dir mirror: prune stale assets and force exec bits the
    /// way `deploy.rs` always has.
    pub fn release_mirror() -> Self {
        Self {
            delete: true,
            chmod: Some("Du=rwx,Dgo=rx,Fu=rw,Fgo=r,F+X".into()),
            ..Self::default()
        }
    }

    /// For payloads that are already compressed (`.gz`/`.dmg`/`.zip`): skip
    /// `-z` so the transfer doesn't re-compress compressed bytes.
    pub fn precompressed() -> Self {
        Self {
            compress: false,
            ..Self::default()
        }
    }
}

/// A read-only host observation. v1 can synthesize these from SSH-streamed
/// commands; a resident `ops-agent` is a drop-in upgrade (E3). The variants
/// mirror the executor spec.
#[derive(Clone, Debug, PartialEq)]
pub enum ObserveEvent {
    ProcessExited {
        unit: String,
        code: i32,
    },
    ResourceSample {
        cpu: f64,
        rss: u64,
        disk: u64,
    },
    JournalLine {
        unit: String,
        line: String,
    },
    HealthChanged {
        check: String,
        from: String,
        to: String,
    },
}

/// A live stream of [`ObserveEvent`]s from a host's observe plane.
pub type EventStream = tokio::sync::mpsc::Receiver<ObserveEvent>;

#[async_trait]
pub trait Executor: Send + Sync {
    /// Run a typed step, streaming merged stdout/stderr into `sink`.
    ///
    /// Returns [`crate::CapabilityDenied`] (boxed into the error) if
    /// `step.action` is outside this executor's grant — checked *before* the
    /// command is dispatched.
    async fn run_streaming(&self, step: &Step, sink: &mut dyn LogSink) -> Result<RunOutput>;

    /// Pull ONE FILE a prior step produced back to the caller: `remote` names
    /// the file on the host, `local` the destination path.
    ///
    /// Split from [`Executor::pull_dir`] because the transports genuinely
    /// differ here, and conflating them is a silent foot-gun: the rsync
    /// transports need a trailing slash for directory-contents semantics and
    /// must NOT have one for a file, while [`crate::AgentRpc`] streams a single
    /// file over HTTP and cannot do directories at all. One `pull` taking either
    /// shape meant a config-level transport swap (`HostTransport::Agent` →
    /// `Ssh`) could turn a working file pull into `rsync host:/x/App.dmg/`.
    ///
    /// Contract is file→file. Pointing this at a *directory* is not checked and
    /// not an error on the rsync transports (`-a` implies `-r`, so the dir is
    /// copied nested under `local`); `AgentRpc` will fail on it. Use
    /// [`Executor::pull_dir`] when you mean a directory.
    async fn pull_file(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()>;

    /// Pull the CONTENTS OF A DIRECTORY back to the caller (rsync from the host
    /// into `local`). See [`Executor::pull_file`] for why this is separate.
    async fn pull_dir(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()>;

    /// Pull every file matching a shell GLOB into the directory `local_dir`.
    ///
    /// The artifact-collection shape: a recipe knows `…/bundle/msi/*.msi`, not
    /// the exact filenames. Matching zero files is an error — a collect that
    /// silently gathers nothing is how an empty release ships.
    ///
    /// `remote_glob` is a `str`, not a `Path`, because it is a *pattern*: the
    /// wildcard must survive to whatever expands it, and `Path` invites callers
    /// to `join`/normalize it. Expansion differs per transport — the ssh
    /// transport lets the REMOTE shell expand, the local one expands in-process
    /// — so implementations must not assume a shell is involved.
    async fn pull_glob(&self, remote_glob: &str, local_dir: &Path, opts: &SyncOpts) -> Result<()>;

    /// Push the contents of a local directory to the host (rsync into `remote`).
    ///
    /// Directory-only: both rsync transports append a trailing slash, and
    /// `AgentRpc` refuses by design (bulk data onto a host goes over
    /// SshExec/rsync or a `git pull` step, not the agent's exec surface).
    async fn push_dir(&self, local: &Path, remote: &Path, opts: &SyncOpts) -> Result<()>;

    /// Subscribe to the host's observe stream; `None` if no observe capability
    /// (or, in v1, if no resident observer is wired — E3).
    fn observe(&self) -> Option<EventStream> {
        None
    }

    /// Optional readiness check, run before dispatching real work to this host.
    /// The default is a no-op: a `LocalExec`/`SshExec` host is ready if it is
    /// reachable, and the first command surfaces any failure. `AgentRpc`
    /// overrides this to hit `/health`, so an agent-host build fails fast with a
    /// clear "is ops-agent running in the Aqua session?" message instead of
    /// erroring opaquely on the first `/run` dispatch.
    async fn preflight(&self) -> Result<()> {
        Ok(())
    }

    /// The capability set this executor was granted (introspection / audit).
    fn capabilities(&self) -> &CapabilitySet;

    /// Convenience: does this executor's grant cover `kind`?
    fn can_observe(&self, kind: &ObserveKind) -> bool {
        self.capabilities().permits_observe(kind)
    }
}

/// Spawn `cmd`, draining stdout+stderr concurrently into the single
/// `&mut dyn LogSink`, and return the exit status plus full captured bytes.
///
/// Unlike [`crate::remote::RemoteHost::run_streaming`] (which shares the sink
/// across two spawned tasks via `Arc<Mutex<_>>`), this drains both pipes in one
/// task with `select!` so it can take a borrowed `&mut dyn` sink — exactly the
/// shape the [`Executor`] trait exposes.
///
/// `sentinel` must be the one [`crate::remote::RemoteHost::command_for`] handed
/// back with `cmd`: `Some` for a remote host, whose reported status is not
/// trustworthy (see [`crate::remote::RcSentinel`]), `None` for local.  `host`
/// only names the target in errors.
pub(crate) async fn run_command_into_sink(
    mut cmd: Command,
    sink: &mut dyn LogSink,
    sentinel: Option<crate::remote::RcSentinel>,
    host: &str,
) -> Result<RunOutput> {
    use std::process::Stdio;
    use tokio::io::AsyncReadExt;

    cmd.stdout(Stdio::piped());
    cmd.stderr(Stdio::piped());
    cmd.kill_on_drop(true);
    let mut child = cmd
        .spawn()
        .map_err(|e| anyhow::anyhow!("spawning command: {e}"))?;

    let mut out = child.stdout.take();
    let mut err = child.stderr.take();
    let mut stdout_buf = Vec::new();
    let mut stderr_buf = Vec::new();
    let mut ob = [0u8; 4096];
    let mut eb = [0u8; 4096];
    let mut out_done = out.is_none();
    let mut err_done = err.is_none();
    // Only stdout carries the sentinel; stderr streams untouched.
    let mut filter = crate::remote::RcFilter::new(sentinel.clone());

    while !(out_done && err_done) {
        tokio::select! {
            r = async { out.as_mut().unwrap().read(&mut ob).await }, if !out_done => {
                match r {
                    Ok(0) | Err(_) => out_done = true,
                    Ok(n) => {
                        let chunk = filter.feed(&ob[..n]);
                        if !chunk.is_empty() {
                            crate::remote::push_bounded(&mut stdout_buf, &chunk, crate::remote::OUTPUT_TAIL_CAP);
                            sink.write_chunk(&chunk).await;
                        }
                    }
                }
            }
            r = async { err.as_mut().unwrap().read(&mut eb).await }, if !err_done => {
                match r {
                    Ok(0) | Err(_) => err_done = true,
                    Ok(n) => { crate::remote::push_bounded(&mut stderr_buf, &eb[..n], crate::remote::OUTPUT_TAIL_CAP); sink.write_chunk(&eb[..n]).await; }
                }
            }
        }
    }

    let (rest, code) = filter.finish();
    if !rest.is_empty() {
        crate::remote::push_bounded(&mut stdout_buf, &rest, crate::remote::OUTPUT_TAIL_CAP);
        sink.write_chunk(&rest).await;
    }

    let status = child
        .wait()
        .await
        .map_err(|e| anyhow::anyhow!("waiting on child: {e}"))?;
    let status = match sentinel {
        Some(_) => crate::remote::resolve_remote_status(host, status, code)?,
        None => status,
    };
    Ok(RunOutput {
        status,
        stdout: stdout_buf,
        stderr: stderr_buf,
    })
}