//! Per-mission ZeroClaw runtime container lifecycle.
//!
//! C3 workspace-isolation model: every mission gets its own ZeroClaw
//! daemon container, so agents' sandboxed filesystem is scoped to that
//! mission's repo checkout instead of the shared `/zeroclaw-data/
//! workspace` on the singleton `clawmates-runtime` daemon.
//!
//! Container naming: `cm-runtime-mission-{first 12 chars of mission uuid}`.
//! Endpoint: `http://<container_name>:42617` (well-known ZeroClaw port,
//! reachable over the `clawmates_core` docker network).
//!
//! Lifecycle:
//!   - `ensure_container(mission_id)` — idempotent; spawns the container
//!     if not present, returns its endpoint. Called from
//!     `mission_orchestrator::on_launch` and (as a fallback for
//!     pre-C3 missions) `phase_runner::launch_phase`.
//!   - `teardown_container(mission_id)` — force-removes the container.
//!     Called by the sweeper (slice 3) N minutes after a mission
//!     reaches a terminal state, so the operator has a window to
//!     re-open the ClawmateS UI and pull the last checkpoint before
//!     the daemon disappears.
//!
//! State: `missions.runtime_container_name` and `missions.runtime_endpoint`
//! carry the current binding (both null when torn down or never
//! provisioned).

use bollard::exec::{CreateExecOptions, StartExecResults};
use bollard::models::{
    ContainerCreateBody, EndpointSettings, HostConfig, Mount, MountTypeEnum, NetworkConnectRequest,
};
use bollard::query_parameters::{
    CreateContainerOptions, InspectContainerOptions, RemoveContainerOptions, StartContainerOptions,
};
use bollard::Docker;
use futures::StreamExt;
use std::collections::HashMap;
use uuid::Uuid;

/// Docker image the per-mission runtime uses. Matches the current
/// singleton `clawmates-runtime` image; can be overridden per-deploy
/// via `CLAWMATES_RUNTIME_IMAGE`.
const DEFAULT_IMAGE: &str = "clawmates-runtime:sync";

/// Well-known ZeroClaw gateway port.
const GATEWAY_PORT: u16 = 42617;

/// Docker networks the runtime container must be attached to.
/// - `clawmates_core`: talks to the server + database
/// - `clawmates_edge`: has egress for outbound provider calls
const CORE_NETWORK: &str = "clawmates_core";
const EDGE_NETWORK: &str = "clawmates_edge";

/// Host path (as seen by the docker engine, NOT the server container)
/// where the mission's checkouts live. Matches the mount source used
/// by `clawmates-runtime.service`.
const MISSIONS_HOST_ROOT: &str = "/var/lib/clawmates-missions";

/// Deterministic docker container name for a mission's runtime.
/// Uses the full UUID hex — UUIDv7 encodes time in the leading bytes,
/// so a short prefix isn't guaranteed unique across missions minted
/// in the same second. Docker permits up to 253 characters in a name,
/// so the extra length is free.
pub fn container_name(mission_id: Uuid) -> String {
    format!("cm-runtime-mission-{}", mission_id.simple())
}

/// Endpoint URL the topology_worker's ZeroClawDriveExecutor will dial.
/// Uses the container name as hostname — resolves within the shared
/// `clawmates_core` docker network.
pub fn endpoint_url(container_name: &str) -> String {
    format!("http://{}:{}", container_name, GATEWAY_PORT)
}

pub struct MissionRuntimeProvisioner {
    docker: Docker,
    image: String,
}

/// What `ensure_container` returns: everything the caller needs to
/// point a topology_worker at this mission's fresh gateway.
#[derive(Debug, Clone)]
pub struct EnsuredContainer {
    pub endpoint: String,
    /// One-time pairing code minted by the daemon at boot; may be
    /// None on the reuse-existing path when we couldn't scrape it
    /// back (log rotation). Callers keep the previously-persisted
    /// value in that case.
    pub pairing_code: Option<String>,
}

impl MissionRuntimeProvisioner {
    /// Connect to the docker engine. Honors `DOCKER_HOST` (set in
    /// compose to the socket proxy) and falls back to the local
    /// socket. Returns None when docker is unreachable, so callers
    /// can degrade gracefully (missions still launch, just against
    /// the shared runtime).
    pub fn from_env() -> Option<MissionRuntimeProvisioner> {
        let docker = if let Ok(host) = std::env::var("DOCKER_HOST") {
            Docker::connect_with_http(&host, 30, bollard::API_DEFAULT_VERSION).ok()?
        } else {
            Docker::connect_with_local_defaults().ok()?
        };
        let image =
            std::env::var("CLAWMATES_RUNTIME_IMAGE").unwrap_or_else(|_| DEFAULT_IMAGE.to_string());
        Some(MissionRuntimeProvisioner { docker, image })
    }

    /// Idempotent: returns the endpoint URL, creating the container
    /// on first call. If the container exists but is stopped, starts
    /// it. If it exists and is running, returns its endpoint.
    pub async fn ensure_container(&self, mission_id: Uuid) -> Result<EnsuredContainer, String> {
        let name = container_name(mission_id);
        // Fast path: already running.
        if let Ok(inspect) = self
            .docker
            .inspect_container(&name, None::<InspectContainerOptions>)
            .await
        {
            let running = inspect
                .state
                .as_ref()
                .and_then(|s| s.running)
                .unwrap_or(false);
            if running {
                // Reuse; try to re-scrape the pairing code from logs,
                // but it may have rotated out — caller falls back to
                // the previously-persisted value in that case.
                let pairing_code = self.scrape_pairing_code(&name).await;
                return Ok(EnsuredContainer {
                    endpoint: endpoint_url(&name),
                    pairing_code,
                });
            }
            // Exists but not running — remove + recreate below rather
            // than trying to restart a dirty-state container.
            let _ = self
                .docker
                .remove_container(
                    &name,
                    Some(RemoveContainerOptions {
                        force: true,
                        ..Default::default()
                    }),
                )
                .await;
        }

        // Create fresh. Ensure the bind source exists first — research-
        // only missions (no repo checkout) still need the directory
        // present or docker start fails with EACCES/ENOENT.
        let mission_dir = format!("{MISSIONS_HOST_ROOT}/{mission_id}");
        let _ = tokio::fs::create_dir_all(&mission_dir).await;
        let mounts = vec![
            // Mount just this mission's directory. Agents can navigate
            // its `/repo` subdir but never see other missions'.
            // `/zeroclaw-data` is intentionally NOT bind-mounted — each
            // per-mission gateway boots with a fresh state directory
            // and mints its own pairing code, scraped in
            // `wait_for_pairing_code` below.
            Mount {
                target: Some("/mission".to_string()),
                source: Some(mission_dir.clone()),
                typ: Some(MountTypeEnum::BIND),
                read_only: Some(false),
                ..Default::default()
            },
        ];

        let host_config = HostConfig {
            mounts: Some(mounts),
            restart_policy: Some(bollard::models::RestartPolicy {
                name: Some(bollard::models::RestartPolicyNameEnum::UNLESS_STOPPED),
                ..Default::default()
            }),
            network_mode: Some(CORE_NETWORK.to_string()),
            ..Default::default()
        };

        let mut env = vec![
            format!("ZEROCLAW_GATEWAY_PORT={GATEWAY_PORT}"),
            "ZEROCLAW_WORKSPACE=/mission/repo".to_string(),
            format!("CM_MISSION_ID={mission_id}"),
        ];
        for key in [
            "ANTHROPIC_API_KEY",
            "GEMINI_API_KEY",
            "GROQ_API_KEY",
            "OPENAI_API_KEY",
        ] {
            if let Ok(v) = std::env::var(key) {
                env.push(format!("{key}={v}"));
            }
        }

        let mut labels = HashMap::new();
        labels.insert("clawmates.role".to_string(), "mission-runtime".to_string());
        labels.insert("clawmates.mission_id".to_string(), mission_id.to_string());

        let config = ContainerCreateBody {
            image: Some(self.image.clone()),
            cmd: Some(vec![
                "daemon".to_string(),
                "--host".to_string(),
                "0.0.0.0".to_string(),
            ]),
            env: Some(env),
            host_config: Some(host_config),
            labels: Some(labels),
            ..Default::default()
        };

        self.docker
            .create_container(
                Some(CreateContainerOptions {
                    name: Some(name.clone()),
                    ..Default::default()
                }),
                config,
            )
            .await
            .map_err(|e| format!("create mission runtime container: {e}"))?;

        // Attach to the edge network for outbound provider egress.
        let _ = self
            .docker
            .connect_network(
                EDGE_NETWORK,
                NetworkConnectRequest {
                    container: Some(name.clone()),
                    endpoint_config: Some(EndpointSettings::default()),
                },
            )
            .await;

        self.docker
            .start_container(&name, None::<StartContainerOptions>)
            .await
            .map_err(|e| format!("start mission runtime container: {e}"))?;

        // Give the daemon a moment to print its boot banner, then
        // scrape the pairing code. Poll with a short deadline so a
        // slow boot doesn't hang the launch — the topology_worker
        // will retry pairing later if we came up empty.
        let pairing_code = self.wait_for_pairing_code(&name).await;
        Ok(EnsuredContainer {
            endpoint: endpoint_url(&name),
            pairing_code,
        })
    }

    /// Poll the daemon's localhost admin endpoint until it responds
    /// with a fresh pairing code. Boot takes ~1-2s; deadline is 15s.
    /// Returns None on timeout so callers see a clear paired=false
    /// signal in the mission binding log.
    async fn wait_for_pairing_code(&self, name: &str) -> Option<String> {
        let deadline = std::time::Duration::from_secs(15);
        let start = std::time::Instant::now();
        while start.elapsed() < deadline {
            if let Some(code) = self.mint_pairing_code(name).await {
                return Some(code);
            }
            tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        }
        None
    }

    /// `docker exec` into the container and hit the local admin
    /// endpoint that mints a fresh pairing code. This works whether
    /// the daemon booted "already paired" (no code in the log) or
    /// "pairing required" (code in the log) — both surfaces mint on
    /// demand.
    async fn mint_pairing_code(&self, name: &str) -> Option<String> {
        let exec = self
            .docker
            .create_exec(
                name,
                CreateExecOptions {
                    cmd: Some(
                        [
                            "curl",
                            "-fs",
                            "-X",
                            "POST",
                            "http://127.0.0.1:42617/admin/paircode/new",
                        ]
                        .iter()
                        .map(|s| s.to_string())
                        .collect(),
                    ),
                    attach_stdout: Some(true),
                    attach_stderr: Some(true),
                    ..Default::default()
                },
            )
            .await
            .ok()?;
        let started = self.docker.start_exec(&exec.id, None).await.ok()?;
        let StartExecResults::Attached { mut output, .. } = started else {
            return None;
        };
        let mut buf = String::new();
        while let Some(chunk) = output.next().await {
            if let Ok(c) = chunk {
                buf.push_str(&c.to_string());
                if buf.len() > 8_000 {
                    break;
                }
            }
        }
        extract_pairing_code_from_json(&buf)
    }
}

/// Parse the JSON `{ "pairing_code": "NNNNNN", ... }` body from the
/// admin/paircode/new endpoint.
fn extract_pairing_code_from_json(body: &str) -> Option<String> {
    let v: serde_json::Value = serde_json::from_str(body.trim()).ok()?;
    v.get("pairing_code")
        .and_then(|x| x.as_str())
        .filter(|s| !s.is_empty())
        .map(String::from)
}

impl MissionRuntimeProvisioner {
    /// Force-remove the mission's runtime container. Idempotent.
    pub async fn teardown_container(&self, mission_id: Uuid) -> Result<(), String> {
        let name = container_name(mission_id);
        self.docker
            .remove_container(
                &name,
                Some(RemoveContainerOptions {
                    force: true,
                    ..Default::default()
                }),
            )
            .await
            .map_err(|e| format!("remove mission runtime container: {e}"))?;
        Ok(())
    }
}

/// Background sweeper: force-remove runtime containers for missions
/// that reached a terminal state ≥ `grace` ago. Keeps the container
/// around briefly after `completed`/`failed`/`cancelled` so the
/// operator can re-open the UI and pull the last checkpoint before
/// the daemon disappears. Runs on the same cadence as the phase
/// runner (10s) with a much longer per-mission grace.
pub fn spawn_sweeper(pool: sqlx::PgPool, grace: std::time::Duration) {
    tokio::spawn(async move {
        // Wait past the phase_runner boot so we don't fight over
        // just-launched missions.
        tokio::time::sleep(std::time::Duration::from_secs(30)).await;
        let mut ticker = tokio::time::interval(std::time::Duration::from_secs(60));
        ticker.tick().await;
        loop {
            ticker.tick().await;
            if let Err(e) = sweep_once(&pool, grace).await {
                eprintln!("mission_runtime::sweeper: sweep failed: {e}");
            }
        }
    });
}

async fn sweep_once(pool: &sqlx::PgPool, grace: std::time::Duration) -> Result<(), String> {
    use sqlx::Row;
    let grace_secs = grace.as_secs() as f64;
    let rows = sqlx::query(
        "SELECT id, workspace_id, runtime_container_name
           FROM missions
          WHERE status IN ('completed', 'failed', 'cancelled')
            AND runtime_endpoint IS NOT NULL
            AND completed_at IS NOT NULL
            AND completed_at < now() - make_interval(secs => $1::float)
          LIMIT 20",
    )
    .bind(grace_secs)
    .fetch_all(pool)
    .await
    .map_err(|e| format!("query terminal missions: {e}"))?;
    if rows.is_empty() {
        return Ok(());
    }
    let Some(prov) = MissionRuntimeProvisioner::from_env() else {
        return Ok(());
    };
    for row in rows {
        let id: Uuid = row.get("id");
        let workspace_id: Uuid = row.get("workspace_id");
        if let Err(e) = prov.teardown_container(id).await {
            // A not-found is expected when the container was already
            // reaped by a docker restart or a manual op; log at info
            // level (via eprintln) and clear the binding anyway so the
            // sweeper doesn't retry forever.
            eprintln!("mission_runtime::sweeper: teardown mission {id}: {e}");
        }
        if let Err(e) =
            cm_db::repo::missions::set_runtime_binding(pool, id, workspace_id, None, None, None)
                .await
        {
            eprintln!("mission_runtime::sweeper: clear binding for {id}: {e}");
        }
    }
    Ok(())
}

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

    #[test]
    fn container_name_is_stable_and_prefixed() {
        let id = Uuid::parse_str("019f84a0-f2a2-7bd0-be9b-86713ec73693").unwrap();
        let name = container_name(id);
        assert_eq!(name, "cm-runtime-mission-019f84a0f2a27bd0be9b86713ec73693");
        // Determinism: same input → same output.
        assert_eq!(name, container_name(id));
    }

    #[test]
    fn container_names_differ_across_missions() {
        // Two UUIDs differing only in the trailing hex char — full-uuid
        // naming must distinguish them (UUIDv7's timestamp shares
        // leading bytes for missions minted in the same second).
        let a = container_name(Uuid::parse_str("019f84a0-f2a2-7bd0-be9b-86713ec73693").unwrap());
        let b = container_name(Uuid::parse_str("019f84a0-f2a2-7bd0-be9b-86713ec7369f").unwrap());
        assert_ne!(a, b);
    }

    #[test]
    fn endpoint_url_uses_gateway_port() {
        let url = endpoint_url("cm-runtime-mission-abc");
        assert_eq!(url, "http://cm-runtime-mission-abc:42617");
    }
}