use crate::config::Config;
use crate::domain::{AppId, Target};
use crate::events::EventTx;
use crate::ota::OtaRegistry;
use crate::topology::{Host, HostTransport, Topology};
use metrics_exporter_prometheus::PrometheusHandle;
use ops_exec::{AgentRpc, CapabilitySet, Executor, LocalExec, SshExec};
use sqlx::SqlitePool;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use tokio::sync::Mutex;
use tokio::task::AbortHandle;

/// Per-host executors keyed by host name, built once from the topology at
/// startup. The recipe engine looks a host's executor up here instead of
/// constructing ssh/scp invocations inline — capability-scoped, transport
/// chosen per host (local / ssh / in-session agent). Mirrors Sando's
/// `ExecutorMap`.
pub type ExecutorMap = HashMap<String, Arc<dyn Executor>>;

/// One occupant of the latest-wins guard: the owning `build_id`, an
/// [`AbortHandle`] for the spawned target task (stops the async wrapper), and a
/// cooperative `cancel` flag the recipe checks at step boundaries and before
/// publish (stops the *blocking* Rhai body, which `abort()` alone cannot reach).
#[derive(Clone)]
pub struct ActiveSlot {
    pub build_id: i64,
    pub abort: AbortHandle,
    pub cancel: Arc<AtomicBool>,
}

/// Latest-wins guard map: per `(app, target)`, the in-flight occupant. A newer
/// build supersedes the slot (setting the prior `cancel` and aborting its
/// handle); a finishing build reaps only the slots it still owns.
pub type ActiveBuilds = Arc<Mutex<HashMap<(AppId, Target), ActiveSlot>>>;

/// One serialization lock per build host. A target holds its host's lock across
/// the whole recipe run, so two targets that land on the same host never build
/// concurrently in one checkout. Without it, goingson's `macos/aarch64` and
/// `ios/universal` — both on mbp — run at once in `~/Code/Apps/goingson`:
/// concurrent `git pull`, a shared `target/`, `release-ios.sh` rewriting
/// `project.yml` mid-build, and (worst) a fixed-path build keychain that the
/// second build deletes out from under the first's codesign.
pub type HostLocks = Arc<HashMap<String, Arc<Mutex<()>>>>;

/// Build one serialization lock per host in the topology. Hosts are fixed at
/// load, so the map is built once and never mutated.
pub fn build_host_locks(topo: &Topology) -> HostLocks {
    Arc::new(
        topo.hosts
            .iter()
            .map(|h| (h.name.clone(), Arc::new(Mutex::new(()))))
            .collect(),
    )
}

#[derive(Clone)]
pub struct AppState {
    pub pool: SqlitePool,
    pub topo: Arc<Topology>,
    pub cfg: Arc<Config>,
    pub prom: PrometheusHandle,
    pub events: EventTx,
    pub ota: Arc<OtaRegistry>,
    /// One capability-scoped [`Executor`] per build host, from the topology —
    /// the transport that *runs steps* (may be the in-session agent).
    pub executors: Arc<ExecutorMap>,
    /// One transport per build host for *moving artifacts* off it. Never the
    /// agent — see [`build_sync`].
    pub syncs: Arc<ExecutorMap>,
    /// Bearer token required on the build-triggering routes (`/build`,
    /// `/retry`). Sourced from `BENTO_API_TOKEN` (systemd EnvironmentFile).
    /// `None` = unauthenticated, which main() permits only on a loopback bind
    /// (CF2).
    pub api_token: Option<Arc<str>>,
    /// Single-slot guard per `(app, target)`: a newer build for the same
    /// target aborts the in-flight one (latest request wins), mirroring
    /// Sando's `active_build`. Other targets keep running — that's the fan-out.
    /// The value carries the owning `build_id` so a finished build reaps only
    /// its own slots and never a superseding build's handle.
    pub active: ActiveBuilds,
    /// One lock per host, held for a target's whole recipe run so two targets on
    /// the same host serialize instead of corrupting one shared checkout +
    /// keychain. See [`HostLocks`].
    pub host_locks: HostLocks,
    /// Cached answers to "is this app's version downloadable from MNW".
    ///
    /// `/status.json` is polled every few seconds and the probe talks to
    /// production, so the answer is cached per `(app, version)`: a viewer
    /// refreshing its board must not turn into sustained traffic against the
    /// endpoint real users' updaters poll.
    pub distribution: DistributionCache,
    /// HTTP client for the distribution probe. One client, so the connection
    /// pool is reused across polls rather than rebuilt per request.
    pub http: reqwest::Client,
    /// MNW base URL the probe reads. Same value the `tauri-mnw` backend
    /// publishes to, so the check and the publish can never disagree about
    /// which host they mean.
    pub mnw_base_url: Arc<str>,
}

/// `(app, version)` to its last probe result and when that was taken.
pub type DistributionCache =
    Arc<Mutex<HashMap<(String, String), (std::time::Instant, crate::ota::Distribution)>>>;

/// How long a distribution answer stays good.
///
/// Generous on purpose. The thing being watched is a manual upload, which
/// happens on human timescales, so a minute of staleness costs nothing and
/// keeps the poll off MNW's back.
pub const DISTRIBUTION_TTL: std::time::Duration = std::time::Duration::from_mins(1);

/// Build one host's EXEC executor: `LocalExec` for `ssh = "local"`, `AgentRpc`
/// for an agent-transport host (macOS in-session signing), `SshExec` otherwise —
/// each granted exactly the host's declared capabilities.
///
/// This is the transport for *running steps*. Moving artifacts uses
/// [`build_sync`] instead; see it for why the two are not the same thing.
pub fn build_executor(host: &Host) -> Arc<dyn Executor> {
    let caps = CapabilitySet::from_tokens(&host.actuate, &host.observe);
    match host.transport {
        HostTransport::Agent => {
            // validate() guarantees agent_url is set for agent hosts.
            let url = host.agent_url.clone().unwrap_or_default();
            Arc::new(AgentRpc::new(url, host.name.clone(), caps))
        }
        HostTransport::Ssh if host.ssh == "local" || host.ssh.is_empty() => {
            Arc::new(LocalExec::new(caps))
        }
        HostTransport::Ssh => Arc::new(SshExec::new(host.ssh.clone(), caps)),
    }
}

/// Build one host's SYNC transport — how the daemon moves *artifacts* off it.
/// Always `LocalExec`/`SshExec`, **never `AgentRpc`, even for an agent host.**
///
/// The agent is an *execution* transport: it exists so codesign runs in the Aqua
/// session where the Developer ID key is usable (design §7 "THE WALL"). It is
/// deliberately a poor artifact mover — `/pull` is confined to the agent's
/// narrow `pull_root` (mbp: `/Users/max/Dist`), which is what keeps an
/// allow-listed caller from reading `~/.tauri/passwords.env`. Build artifacts
/// live in the *repo checkout*, outside that root, so routing collect through
/// the agent would either 404 or force `pull_root` wide enough to undo the
/// confinement. `AgentRpc::push_dir`/`pull_glob` say the same thing in their
/// refusals: bulk data moves over ssh/rsync, not the agent.
///
/// So a mac host has two transports at once: `AgentRpc` to sign, `SshExec` to
/// fetch what it signed. Both reach the same box; only the privilege differs.
///
/// The sync transport is confined to the host's declared `pull_root` and gated
/// on its `observe:artifact` grant (`ops_exec::gate_pull`). Both are
/// fail-closed: a host that declares no `pull_root` collects nothing. That
/// confinement is what keeps a `collect(host, '…/.tauri/passwords.env')` from
/// depositing the notary credential into `dist_root`.
pub fn build_sync(host: &Host) -> Arc<dyn Executor> {
    let caps = CapabilitySet::from_tokens(&host.actuate, &host.observe);
    let roots = host.artifact_roots();
    if host.ssh == "local" || host.ssh.is_empty() {
        Arc::new(LocalExec::new(caps).with_pull_roots(roots))
    } else {
        Arc::new(SshExec::new(host.ssh.clone(), caps).with_pull_roots(roots))
    }
}

/// Build the executor for a service's deploy destination.
///
/// Deliberately NOT a build host, and never in the topology's host list: it is
/// granted `deploy` + `restart` and nothing else, so the same executor that
/// installs pom's binary cannot be handed a `build` step, and a compromised
/// recipe cannot turn the production box into a build host. `Action::Deploy` and
/// `Action::Restart` already exist in `ops_exec` for Sando's promotions; this is
/// the same grant reaching the same kind of destination.
///
/// It gets no `pull_root`, so the artifact-collection plane is closed on it in
/// both directions: `collect()` from a deploy host is refused fail-closed, which
/// is right — a service host produces nothing Bento should be fetching.
pub fn build_deploy_executor(d: &crate::topology::DeployTarget) -> Arc<dyn Executor> {
    let caps = CapabilitySet::from_tokens(["deploy", "restart"], ["build-log"]);
    if d.host == "local" || d.host.is_empty() {
        return Arc::new(LocalExec::new(caps));
    }
    Arc::new(SshExec::new(d.host.clone(), caps).with_port(d.port))
}

/// Build the full host name -> exec-executor map from the topology.
pub fn build_executors(topo: &Topology) -> ExecutorMap {
    topo.hosts
        .iter()
        .map(|h| (h.name.clone(), build_executor(h)))
        .collect()
}

/// Build the full host name -> sync-transport map from the topology.
pub fn build_syncs(topo: &Topology) -> ExecutorMap {
    topo.hosts
        .iter()
        .map(|h| (h.name.clone(), build_sync(h)))
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;
    use ops_exec::{Action, SyncOpts};

    fn host(toml_host: &str) -> Host {
        // Only a host is under test here, so parse just the `[[host]]` table.
        // Going through `Topology` would drag in an app pointer and a manifest
        // on disk for no reason.
        #[derive(serde::Deserialize)]
        struct Hosts {
            #[serde(rename = "host")]
            hosts: Vec<Host>,
        }
        let parsed: Hosts = toml::from_str(toml_host).unwrap();
        parsed.hosts.into_iter().next().unwrap()
    }

    #[test]
    fn build_host_executor_permits_build_not_sign() {
        let h = host("[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]");
        let exec = build_executor(&h);
        assert!(exec.capabilities().permits(&Action::Build));
        assert!(exec.capabilities().permits(&Action::Package));
        assert!(!exec.capabilities().permits(&Action::Sign));
    }

    /// The load-bearing property of the two-plane split: a mac host signs over
    /// the agent but is COLLECTED FROM over ssh. If this ever regresses to one
    /// transport, collect hits `AgentRpc::pull_glob` (refused by design) or
    /// forces `pull_root` wide enough to expose `~/.tauri/passwords.env`.
    #[tokio::test]
    async fn agent_host_syncs_over_ssh_never_the_agent() {
        let h = host(
            "[[host]]\nname = \"mbp\"\nssh = \"mbp\"\ntargets = [\"macos/aarch64\"]\n\
             transport = \"agent\"\nagent_url = \"http://mbp:8765\"\n\
             actuate = [\"build\", \"sign\", \"notarize\", \"staple\"]\n\
             observe = [\"build-log\", \"gatekeeper\", \"artifact\"]\n\
             pull_root = \"/nonexistent\"",
        );
        // The sync transport must not be the agent. AgentRpc refuses pull_glob
        // by design, so a non-refusing error proves we got an ssh transport.
        // The host declares `artifact` + a `pull_root` so the pull clears the
        // fail-closed sync gate and reaches the actual ssh rsync (which then
        // fails on the unreachable host / no match) rather than the gate.
        let sync = build_sync(&h);
        let err = sync
            .pull_glob(
                "/nonexistent/*.dmg",
                std::path::Path::new("/tmp"),
                &SyncOpts::default(),
            )
            .await
            .expect_err("nothing matches, so this must error either way");
        assert!(
            !err.to_string().contains("unsupported by design"),
            "sync transport for an agent host must NOT be AgentRpc: {err}"
        );
        assert!(
            !err.to_string().contains("no artifact root declared"),
            "the declared pull_root must let the pull reach the ssh transport: {err}"
        );
        // ...while the exec transport for the same host still is the agent.
        let exec = build_executor(&h);
        let err = exec
            .pull_glob(
                "/x/*.dmg",
                std::path::Path::new("/tmp"),
                &SyncOpts::default(),
            )
            .await
            .expect_err("AgentRpc has no glob form");
        assert!(
            err.to_string().contains("unsupported by design"),
            "exec plane is the agent: {err}"
        );
    }

    /// pom's case: one host, two trees. The singular and the plural are unioned,
    /// so a topology that adds `pull_roots` keeps whatever `pull_root` said.
    #[tokio::test]
    async fn sync_pull_accepts_every_declared_root() {
        let h = host(
            "[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]\n\
             pull_root = \"/home/max/Code/Apps\"\n\
             pull_roots = [\"/home/max/Code/MNW\"]",
        );
        assert_eq!(
            h.artifact_roots(),
            vec![
                std::path::PathBuf::from("/home/max/Code/Apps"),
                std::path::PathBuf::from("/home/max/Code/MNW"),
            ],
        );
        let sync = build_sync(&h);
        // The refusal must still name the secret-bearing sibling as out of
        // bounds: adding MNW widens the gate by MNW, not by ~/Code.
        let err = sync
            .pull_file(
                std::path::Path::new("/home/max/Code/_private/apple/notary.p8"),
                std::path::Path::new("/tmp/out"),
                &SyncOpts::default(),
            )
            .await
            .expect_err("_private is under neither root");
        assert!(
            err.to_string()
                .contains("escapes every declared artifact root"),
            "{err}"
        );
    }

    #[test]
    fn local_host_syncs_locally() {
        let h = host("[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]");
        // A local host's sync transport still carries the host's grant.
        assert!(build_sync(&h).capabilities().permits(&Action::Build));
    }

    /// The finding this fix closes, end to end through `build_sync`: a host with
    /// a declared `pull_root` collects artifacts under it but refuses a path
    /// outside it (the `~/.tauri/passwords.env` exfil). Confinement is lexical,
    /// so no real filesystem is needed.
    #[tokio::test]
    async fn sync_pull_is_confined_to_declared_root() {
        let h = host(
            "[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]\n\
             pull_root = \"/home/max/Code/Apps\"",
        );
        let sync = build_sync(&h);
        // A path outside the declared root is refused before any rsync.
        let err = sync
            .pull_file(
                std::path::Path::new("/home/max/.tauri/passwords.env"),
                std::path::Path::new("/tmp/out"),
                &SyncOpts::default(),
            )
            .await
            .expect_err("a path outside pull_root must be denied");
        assert!(
            err.to_string()
                .contains("escapes every declared artifact root"),
            "{err}"
        );

        // A host with NO declared root collects nothing at all (fail-closed).
        let bare = host("[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]");
        let err = build_sync(&bare)
            .pull_file(
                std::path::Path::new("/home/max/Code/Apps/goingson/x.dmg"),
                std::path::Path::new("/tmp/out"),
                &SyncOpts::default(),
            )
            .await
            .expect_err("no pull_root ⇒ no pulls");
        assert!(
            err.to_string().contains("no artifact root declared"),
            "{err}"
        );
    }

    /// One serialization lock per host, and they are independent: holding mbp's
    /// lock (a macOS + iOS build serialize behind it) leaves fw13 free to build
    /// in parallel. This is the fan-out-across-hosts / serialize-within-a-host
    /// property the runner relies on.
    #[tokio::test]
    async fn host_locks_are_one_per_host_and_independent() {
        use crate::topology::Topology;
        let dir = tempfile::tempdir().unwrap();
        let repo = dir.path().join("goingson");
        std::fs::create_dir_all(&repo).unwrap();
        std::fs::write(repo.join("bento.toml"), "targets = [\"macos/aarch64\"]\n").unwrap();
        let topo = Topology::from_str_for_tests(&format!(
            "[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]\n\
             [[host]]\nname = \"mbp\"\nssh = \"mbp\"\ntargets = [\"macos/aarch64\"]\n\
             [app.goingson]\nrepo = \"{}\"\n",
            repo.display()
        ))
        .unwrap();
        let locks = build_host_locks(&topo);
        assert_eq!(locks.len(), 2);
        let mbp = locks.get("mbp").expect("mbp lock").clone();
        let _held = mbp.clone().lock_owned().await;
        // Same host: a second acquire cannot proceed while the first is held.
        assert!(mbp.try_lock().is_err(), "same-host builds must serialize");
        // Different host: unaffected — it builds in parallel.
        assert!(
            locks.get("fw13").expect("fw13 lock").try_lock().is_ok(),
            "a different host must not be blocked"
        );
    }

    #[test]
    fn agent_host_executor_permits_sign() {
        // The mac host: agent transport, widened grant. Proves AgentRpc is
        // constructed (no panic) and carries the sign capability that the
        // SSH/local transports' default grant does not.
        let h = host(
            "[[host]]\nname = \"mbp\"\nssh = \"mbp\"\ntargets = [\"macos/aarch64\"]\n\
             transport = \"agent\"\nagent_url = \"http://mbp:8765\"\n\
             actuate = [\"build\", \"sign\", \"notarize\", \"staple\"]",
        );
        let exec = build_executor(&h);
        assert!(exec.capabilities().permits(&Action::Sign));
        assert!(exec.capabilities().permits(&Action::Notarize));
        assert!(exec.capabilities().permits(&Action::Build));
    }
}