//! `ops-agent` — the on-host half of the executor.
//!
//! One binary; behavior is set entirely by its local config (its own grant +
//! which caller identities may reach it). It listens **only on the tailnet
//! interface** and, on every request, resolves the caller via the local
//! Tailscale LocalAPI `whois`, maps node/tags to a caller grant, and runs the
//! step under the **intersection** of that grant with its own — the agent-side
//! half of double enforcement. A buggy or compromised daemon cannot make this
//! agent exceed its local grant.
//!
//! macOS deployment is an Aqua LaunchAgent (`LimitLoadToSessionType = Aqua`) so
//! build+sign run in the GUI security session where codesign can use the key.

use crate::capability::CapabilitySet;
use crate::executor::Executor;
use crate::remote::LogSink;
use crate::step::Action;
use crate::transport::LocalExec;
use crate::wire::{Frame, HealthResponse, RunRequest};
use anyhow::{Context, Result};
use async_trait::async_trait;
use serde::Deserialize;
use std::net::{IpAddr, SocketAddr};
use std::path::PathBuf;
use std::sync::Arc;

/// The agent's local configuration (TOML).
#[derive(Clone, Debug, Deserialize)]
pub struct AgentConfig {
    /// Tailnet socket to bind — set this to the host's tailnet address so the
    /// agent is never reachable off the tailnet.
    pub listen: SocketAddr,
    /// What this host itself is allowed to do. The ceiling for every caller.
    #[serde(default)]
    pub grant: GrantConfig,
    /// Which caller identities may reach this agent, and the grant each
    /// implies. The effective grant is `caller.caps ∩ self.grant`.
    #[serde(default)]
    pub allow: Vec<CallerGrant>,
}

/// Grant tokens as they appear in config: `actuate = [...]`, `observe = [...]`.
#[derive(Clone, Debug, Default, Deserialize)]
pub struct GrantConfig {
    #[serde(default)]
    pub actuate: Vec<String>,
    #[serde(default)]
    pub observe: Vec<String>,
}

impl GrantConfig {
    pub fn to_caps(&self) -> CapabilitySet {
        CapabilitySet::from_tokens(&self.actuate, &self.observe)
    }
}

/// One allowed caller: a tailnet identity (a node name like `fw13` or a tag
/// like `tag:prod`) and the grant it implies.
#[derive(Clone, Debug, Deserialize)]
pub struct CallerGrant {
    pub identity: String,
    #[serde(default)]
    pub actuate: Vec<String>,
    #[serde(default)]
    pub observe: Vec<String>,
}

impl CallerGrant {
    fn to_caps(&self) -> CapabilitySet {
        CapabilitySet::from_tokens(&self.actuate, &self.observe)
    }
}

/// A resolved caller identity from `whois`: the peer's node name and its tags.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CallerIdentity {
    pub node: String,
    pub tags: Vec<String>,
}

impl CallerIdentity {
    /// Does this identity match a config `identity` string (a node name or a
    /// `tag:...`)?
    fn matches(&self, identity: &str) -> bool {
        self.node == identity || self.tags.iter().any(|t| t == identity)
    }
}

/// The outcome of authorizing a caller for an action.
#[derive(Debug, PartialEq, Eq)]
pub enum AuthDecision {
    /// Permitted; carries the effective (intersected) grant.
    Allow,
    /// The caller's identity is not in this agent's allow-list.
    UnknownCaller,
    /// Known caller, but the action is outside the effective grant.
    Denied,
}

/// Pure authorization core — no IO, fully unit-testable. The effective grant is
/// `caller_grant ∩ agent_grant`; the action must be permitted by it.
pub fn authorize(config: &AgentConfig, caller: &CallerIdentity, action: &Action) -> AuthDecision {
    let Some(entry) = config.allow.iter().find(|c| caller.matches(&c.identity)) else {
        return AuthDecision::UnknownCaller;
    };
    let effective = entry.to_caps().intersect(&config.grant.to_caps());
    if effective.permits(action) {
        AuthDecision::Allow
    } else {
        AuthDecision::Denied
    }
}

/// Resolve a peer IP to a tailnet identity by shelling `tailscale whois
/// --json`. Works identically under Headscale (it serves the same client CLI).
/// Runtime-only — not exercised by unit tests (no live tailnet in CI).
pub async fn tailscale_whois(peer: IpAddr) -> Result<CallerIdentity> {
    let out = tokio::process::Command::new("tailscale")
        .args(["whois", "--json", &peer.to_string()])
        .output()
        .await
        .context("spawning `tailscale whois`")?;
    anyhow::ensure!(
        out.status.success(),
        "tailscale whois {peer} failed: {}",
        String::from_utf8_lossy(&out.stderr)
    );
    #[derive(Deserialize)]
    struct Whois {
        #[serde(rename = "Node")]
        node: WhoisNode,
    }
    #[derive(Deserialize)]
    struct WhoisNode {
        #[serde(rename = "Name", default)]
        name: String,
        #[serde(rename = "Tags", default)]
        tags: Vec<String>,
    }
    let parsed: Whois = serde_json::from_slice(&out.stdout).context("parsing whois json")?;
    // `Name` is the MagicDNS FQDN (e.g. `fw13.tailnet.ts.net.`); reduce to the
    // bare hostname so config can say `fw13`.
    let node = parsed.node.name.trim_end_matches('.').split('.').next().unwrap_or("").to_string();
    Ok(CallerIdentity { node, tags: parsed.node.tags })
}

// ---------------------------------------------------------------------------
// HTTP layer (axum). Kept thin: it resolves identity, authorizes, and runs the
// step locally via `LocalExec`, streaming NDJSON frames back.
// ---------------------------------------------------------------------------

/// A resolver mapping a peer address to its tailnet identity. Boxed so tests
/// can inject a stub instead of shelling out to `tailscale`.
pub type WhoisResolver =
    Arc<dyn Fn(IpAddr) -> futures_util::future::BoxFuture<'static, Result<CallerIdentity>> + Send + Sync>;

/// Shared server state.
#[derive(Clone)]
pub struct AgentState {
    pub config: Arc<AgentConfig>,
    pub whois: WhoisResolver,
}

impl AgentState {
    /// Build state with the real `tailscale whois` resolver.
    pub fn new(config: AgentConfig) -> Self {
        Self {
            config: Arc::new(config),
            whois: Arc::new(|ip| Box::pin(tailscale_whois(ip))),
        }
    }
}

/// A [`LogSink`] that forwards each chunk as an NDJSON [`Frame::Chunk`] line
/// into the response body channel.
struct ChannelSink {
    tx: tokio::sync::mpsc::Sender<Result<axum::body::Bytes, std::io::Error>>,
}

#[async_trait]
impl LogSink for ChannelSink {
    async fn write_chunk(&mut self, bytes: &[u8]) {
        let text = String::from_utf8_lossy(bytes).into_owned();
        let line = Frame::Chunk { text }.to_line();
        let _ = self.tx.send(Ok(axum::body::Bytes::from(line))).await;
    }
}

/// Build the axum router. Serve it with
/// `.into_make_service_with_connect_info::<SocketAddr>()` so handlers see the
/// peer address.
pub fn router(state: AgentState) -> axum::Router {
    use axum::routing::{get, post};
    axum::Router::new()
        .route("/health", get(health))
        .route("/run", post(run))
        .route("/pull", get(pull))
        .with_state(state)
}

async fn health(
    axum::extract::State(state): axum::extract::State<AgentState>,
) -> axum::Json<HealthResponse> {
    let caps = state.config.grant.to_caps();
    axum::Json(HealthResponse {
        ok: true,
        actuate: caps.actuate_tokens().map(String::from).collect(),
        observe: caps.observe_kinds().map(|k| k.token()).collect(),
    })
}

async fn run(
    axum::extract::State(state): axum::extract::State<AgentState>,
    axum::extract::ConnectInfo(peer): axum::extract::ConnectInfo<SocketAddr>,
    axum::Json(req): axum::Json<RunRequest>,
) -> axum::response::Response {
    use axum::http::StatusCode;
    use axum::response::IntoResponse;

    // Identity → authorization (agent-side enforcement).
    let identity = match (state.whois)(peer.ip()).await {
        Ok(id) => id,
        Err(e) => {
            return (StatusCode::FORBIDDEN, format!("whois failed: {e}")).into_response();
        }
    };
    match authorize(&state.config, &identity, &req.step.action) {
        AuthDecision::Allow => {}
        AuthDecision::UnknownCaller => {
            return (StatusCode::FORBIDDEN, format!("unknown caller: {}", identity.node)).into_response();
        }
        AuthDecision::Denied => {
            return (
                StatusCode::FORBIDDEN,
                format!("action `{:?}` denied for {}", req.step.action, identity.node),
            )
                .into_response();
        }
    }

    // Run locally under the effective grant, streaming NDJSON frames back.
    let effective = state
        .config
        .allow
        .iter()
        .find(|c| identity.matches(&c.identity))
        .map(|c| c.to_caps().intersect(&state.config.grant.to_caps()))
        .unwrap_or_default();

    let (tx, rx) = tokio::sync::mpsc::channel::<Result<axum::body::Bytes, std::io::Error>>(64);
    tokio::spawn(async move {
        let exec = LocalExec::new(effective);
        let mut sink = ChannelSink { tx: tx.clone() };
        let terminal = match exec.run_streaming(&req.step, &mut sink).await {
            Ok(out) => Frame::Exit { code: out.status.code().unwrap_or(-1) },
            Err(e) => Frame::Error { message: format!("{e:#}") },
        };
        let _ = tx.send(Ok(axum::body::Bytes::from(terminal.to_line()))).await;
    });

    let stream = tokio_stream::wrappers::ReceiverStream::new(rx);
    axum::body::Body::from_stream(stream).into_response()
}

#[derive(Deserialize)]
struct PullQuery {
    path: PathBuf,
}

async fn pull(
    axum::extract::State(_state): axum::extract::State<AgentState>,
    axum::extract::Query(q): axum::extract::Query<PullQuery>,
) -> axum::response::Response {
    use axum::http::StatusCode;
    use axum::response::IntoResponse;
    match tokio::fs::read(&q.path).await {
        Ok(bytes) => bytes.into_response(),
        Err(e) => (StatusCode::NOT_FOUND, format!("pull {}: {e}", q.path.display())).into_response(),
    }
}

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

    fn cfg() -> AgentConfig {
        AgentConfig {
            listen: "127.0.0.1:0".parse().unwrap(),
            // This host (mbp-like) may build/sign/notarize/staple.
            grant: GrantConfig {
                actuate: vec!["build".into(), "sign".into(), "notarize".into(), "staple".into()],
                observe: vec!["build-log".into()],
            },
            allow: vec![CallerGrant {
                identity: "fw13".into(),
                actuate: vec!["build".into(), "sign".into(), "notarize".into(), "staple".into()],
                observe: vec![],
            }],
        }
    }

    fn fw13() -> CallerIdentity {
        CallerIdentity { node: "fw13".into(), tags: vec![] }
    }

    #[test]
    fn known_caller_granted_action_is_allowed() {
        assert_eq!(authorize(&cfg(), &fw13(), &Action::Sign), AuthDecision::Allow);
    }

    #[test]
    fn unknown_caller_is_rejected() {
        let stranger = CallerIdentity { node: "laptop-x".into(), tags: vec![] };
        assert_eq!(authorize(&cfg(), &stranger, &Action::Sign), AuthDecision::UnknownCaller);
    }

    #[test]
    fn action_outside_agent_grant_is_denied_even_if_caller_asks() {
        // Caller is granted only what config lists; deploy is not in either set.
        assert_eq!(authorize(&cfg(), &fw13(), &Action::Deploy), AuthDecision::Denied);
    }

    #[test]
    fn intersection_floors_a_too_broad_caller() {
        // A caller granted `deploy` but the agent host only grants build/sign:
        // deploy must be denied (agent grant is the ceiling).
        let mut c = cfg();
        c.allow[0].actuate.push("deploy".into());
        assert_eq!(authorize(&c, &fw13(), &Action::Deploy), AuthDecision::Denied);
        assert_eq!(authorize(&c, &fw13(), &Action::Sign), AuthDecision::Allow);
    }

    #[test]
    fn tag_identity_matches() {
        let mut c = cfg();
        c.allow[0].identity = "tag:builder".into();
        let tagged = CallerIdentity { node: "whatever".into(), tags: vec!["tag:builder".into()] };
        assert_eq!(authorize(&c, &tagged, &Action::Sign), AuthDecision::Allow);
    }
}