//! `AgentRpc` — the Executor transport that talks to a remote `ops-agent`.
//!
//! Used for the macOS in-session sign step (the agent runs in the Aqua security
//! session where codesign can use the Developer ID key) and, later, the
//! resident observe plane. Non-signing, Linux/Windows work stays on
//! [`crate::SshExec`].
//!
//! Trust is the tailnet: the connection rides WireGuard, and the agent
//! independently re-checks the caller's identity via `whois`. This client also
//! enforces its own grant caller-side (the first half of double enforcement).

use crate::capability::{CapabilityDenied, CapabilitySet};
use crate::executor::{Executor, SyncOpts};
use crate::remote::{LogSink, RunOutput};
use crate::step::{Action, ObserveKind, Step};
use crate::wire::{Frame, HealthResponse, RunRequest};
use anyhow::{Context, Result};
use async_trait::async_trait;
use futures_util::StreamExt;
use std::path::Path;
use std::time::Duration;

/// A handle to one `ops-agent`, scoped to a caller-side capability set.
pub struct AgentRpc {
    base_url: String,
    host_label: String,
    caps: CapabilitySet,
    client: reqwest::Client,
}

/// How long to wait for the TCP connect before calling an agent unreachable.
///
/// Applies to every request, including `/run` and `/pull`: it bounds only
/// establishing the connection, never a live response body, so a multi-minute
/// build or a large artifact download is unaffected. The agent is one tailnet
/// hop away, so a connect that has not landed in this long is not going to.
const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);

/// Whole-request cap for the short control calls (`/health`), which return a
/// small JSON body immediately.
///
/// Deliberately NOT applied to the client as a whole: `reqwest`'s request
/// timeout runs until the response *body* has finished, so a client-level
/// timeout would sever `/run` mid-build and truncate `/pull` on a large DMG.
/// Those two are unbounded by design — a build legitimately runs for minutes —
/// and their liveness is covered by `CONNECT_TIMEOUT` plus the exit frame that
/// `run_streaming` requires before it will report success.
const HEALTH_TIMEOUT: Duration = Duration::from_secs(10);

impl AgentRpc {
    /// `base_url` is e.g. `http://mbp.tailnet:8765` (the agent listens on the
    /// tailnet interface). `host_label` is used only for audit messages.
    pub fn new(
        base_url: impl Into<String>,
        host_label: impl Into<String>,
        caps: CapabilitySet,
    ) -> Self {
        // A default-built client has no timeouts whatsoever, which left an agent
        // that accepts TCP and then never answers (a wedged process, or a path
        // that blackholes after the handshake) hanging a release indefinitely
        // with no output. An unreachable host only ever failed because the OS
        // gave up on the connect (~30s), which was luck, not design.
        let client = reqwest::Client::builder()
            .connect_timeout(CONNECT_TIMEOUT)
            .build()
            .expect("reqwest client with a connect timeout");
        Self {
            base_url: base_url.into(),
            host_label: host_label.into(),
            caps,
            client,
        }
    }

    /// `GET /health` — liveness plus the agent's own declared grant.
    pub async fn health(&self) -> Result<HealthResponse> {
        let resp = self
            .client
            .get(format!("{}/health", self.base_url))
            .timeout(HEALTH_TIMEOUT)
            .send()
            .await
            .with_context(|| {
                format!(
                    "GET /health ({}): no answer within {}s — the agent is unreachable, or it is \
                     accepting connections without answering them",
                    self.host_label,
                    HEALTH_TIMEOUT.as_secs()
                )
            })?
            .error_for_status()
            .context("agent /health status")?;
        resp.json().await.context("decoding /health")
    }
}

/// Reconstruct an `ExitStatus` from the raw exit code the agent reports, which
/// is a process exit code (0..=255); `.success()` and the non-zero distinction —
/// all a recipe branches on — are preserved exactly. A negative/out-of-range
/// code (e.g. the agent's `-1` "terminated by signal" sentinel) collapses into
/// the low byte, so it reads back as non-zero but not its original value.
///
/// Shared with the ssh path, which recovers a code the same way from its
/// [`crate::remote::RcSentinel`].
use crate::remote::exit_status_from_code as exit_status;

#[async_trait]
impl Executor for AgentRpc {
    async fn run_streaming(&self, step: &Step, sink: &mut dyn LogSink) -> Result<RunOutput> {
        // Caller-side enforcement (half 1 of 2). The agent re-checks against
        // its own grant (half 2) — this just fails fast before a round-trip.
        if !self.caps.permits(&step.action) {
            return Err(CapabilityDenied::new(&self.host_label, &step.action).into());
        }

        let resp = self
            .client
            .post(format!("{}/run", self.base_url))
            .json(&RunRequest { step: step.clone() })
            .send()
            .await
            .context("POST /run")?;
        if resp.status() == reqwest::StatusCode::FORBIDDEN {
            let body = resp.text().await.unwrap_or_default();
            anyhow::bail!("agent denied /run: {}", body.trim());
        }
        let resp = resp.error_for_status().context("agent /run status")?;

        let mut stream = resp.bytes_stream();
        let mut buf: Vec<u8> = Vec::new();
        let mut captured: Vec<u8> = Vec::new();
        let mut exit_code: Option<i32> = None;

        // A single NDJSON frame must fit in this much; a stream that sends this
        // many bytes with no newline is malformed (and would otherwise grow the
        // buffer unbounded — an OOM vector against the daemon).
        const MAX_FRAME_BYTES: usize = 16 * 1024 * 1024;
        while let Some(chunk) = stream.next().await {
            let chunk = chunk.context("reading /run stream")?;
            buf.extend_from_slice(&chunk);
            while let Some(nl) = buf.iter().position(|&b| b == b'\n') {
                let line: Vec<u8> = buf.drain(..=nl).collect();
                let line = &line[..line.len() - 1];
                if line.is_empty() {
                    continue;
                }
                let frame: Frame = serde_json::from_slice(line).with_context(|| {
                    format!("decoding agent frame: {}", String::from_utf8_lossy(line))
                })?;
                match frame {
                    Frame::Chunk { text } => {
                        crate::remote::push_bounded(
                            &mut captured,
                            text.as_bytes(),
                            crate::remote::OUTPUT_TAIL_CAP,
                        );
                        sink.write_chunk(text.as_bytes()).await;
                    }
                    Frame::Exit { code } => exit_code = Some(code),
                    Frame::Error { message } => anyhow::bail!("agent error: {message}"),
                    // Forward-compatible: a newer agent's unknown frame is ignored.
                    Frame::Unknown => {}
                }
            }
            // `buf` now holds at most one incomplete trailing line — bound it.
            anyhow::ensure!(
                buf.len() <= MAX_FRAME_BYTES,
                "agent /run frame exceeded {MAX_FRAME_BYTES} bytes without a newline",
            );
        }

        let code = exit_code.context("agent closed /run stream without an exit frame")?;
        Ok(RunOutput {
            status: exit_status(code),
            stdout: captured,
            stderr: Vec::new(),
        })
    }

    async fn pull_file(&self, remote: &Path, local: &Path, _opts: &SyncOpts) -> Result<()> {
        // Caller-side enforcement (half 1 of 2), same as `run_streaming`. `/pull`
        // is observe-plane: it needs the `artifact` grant. Failing fast here
        // matters more than elsewhere — a pull is the *last* step of a release,
        // so an ungranted caller would otherwise learn it after a full build and
        // an Apple notary round trip.
        let action = Action::Observe(ObserveKind::Artifact);
        if !self.caps.permits(&action) {
            return Err(CapabilityDenied::new(&self.host_label, &action).into());
        }

        let resp = self
            .client
            .get(format!("{}/pull", self.base_url))
            .query(&[("path", remote.to_string_lossy().as_ref())])
            .send()
            .await
            .context("GET /pull")?;
        // Surface the agent's reason (ungranted, /pull disabled, outside
        // pull_root); `error_for_status` alone would reduce it to a bare 403.
        if resp.status() == reqwest::StatusCode::FORBIDDEN {
            let body = resp.text().await.unwrap_or_default();
            anyhow::bail!("agent denied /pull: {}", body.trim());
        }
        let resp = resp.error_for_status().context("agent /pull status")?;
        // Stream the body straight to disk in chunks — a signed .app/.dmg can be
        // hundreds of MB, so never buffer the whole artifact in the daemon's heap
        // (the agent's /pull already streams; this is the matching client half).
        if let Some(parent) = local.parent() {
            tokio::fs::create_dir_all(parent).await.ok();
        }
        let mut file = tokio::fs::File::create(local)
            .await
            .context("creating pulled artifact")?;
        let mut stream = resp.bytes_stream();
        while let Some(chunk) = stream.next().await {
            let chunk = chunk.context("reading /pull body")?;
            tokio::io::AsyncWriteExt::write_all(&mut file, &chunk)
                .await
                .context("writing pulled artifact")?;
        }
        tokio::io::AsyncWriteExt::flush(&mut file)
            .await
            .context("flushing pulled artifact")?;
        Ok(())
    }

    async fn pull_dir(&self, _remote: &Path, _local: &Path, _opts: &SyncOpts) -> Result<()> {
        // `/pull` serves exactly one file per request (a streamed HTTP body);
        // there is no directory form, and inventing one would mean walking the
        // host's tree over the agent's exec surface. Artifact retrieval is
        // per-file by design — collect a directory with SshExec/rsync.
        anyhow::bail!(
            "AgentRpc::pull_dir is unsupported by design; the agent serves one file per \
             /pull — use pull_file, or SshExec/rsync for a directory"
        )
    }

    async fn pull_glob(
        &self,
        _remote_glob: &str,
        _local_dir: &Path,
        _opts: &SyncOpts,
    ) -> Result<()> {
        // `/pull` takes one concrete path and has no expansion surface — and
        // giving it one would mean the agent globbing its own filesystem on a
        // caller's behalf. Artifact collection over the agent is per-file.
        anyhow::bail!(
            "AgentRpc::pull_glob is unsupported by design; the agent serves one concrete \
             path per /pull — resolve the glob on the host and use pull_file, or collect \
             with SshExec/rsync"
        )
    }

    async fn push_dir(&self, _local: &Path, _remote: &Path, _opts: &SyncOpts) -> Result<()> {
        // The agent transport is for in-session *execution*. Bulk data movement
        // onto the host uses SshExec/rsync or a `git pull` step inside the
        // recipe — keeping the agent's surface small (one open port, exec only).
        anyhow::bail!(
            "AgentRpc::push_dir is unsupported by design; move source/data with SshExec/rsync \
             or a `git pull` step, not the agent"
        )
    }

    async fn preflight(&self) -> Result<()> {
        let health = self.health().await.with_context(|| {
            format!(
                "agent /health failed for `{}` — is ops-agent running in the Aqua session on the build host?",
                self.host_label,
            )
        })?;
        // Detect a skewed agent here rather than mid-stream. version 0 = a legacy
        // agent that predates the field (HealthResponse.version defaults to 0);
        // tolerate it, but refuse a future major we don't speak.
        anyhow::ensure!(
            health.version == 0 || health.version == crate::wire::PROTOCOL_VERSION,
            "agent `{}` speaks wire protocol v{}, but this client expects v{} — upgrade the mismatched side",
            self.host_label,
            health.version,
            crate::wire::PROTOCOL_VERSION,
        );
        Ok(())
    }

    fn capabilities(&self) -> &CapabilitySet {
        &self.caps
    }
}