//! Background worker that drains durable topology run jobs.
//!
//! A `POST /api/topologies/run` enqueues a job (`topology_runs` row, status
//! `queued`); this loop claims it, drives the topology turn-by-turn via the
//! ZeroClaw runtime, and checkpoints the [`RunProgress`] after every step. If
//! the worker (or the whole server) dies mid-run, the row is left `running`;
//! the stale sweep requeues it and the next claim resumes it from the last
//! checkpointed step — so long-horizon runs survive restarts.
//!
//! This reuses the agent-run durability pattern (claim CAS, checkpoint, resume
//! sweep) without coupling topology runs to the chat-session schema.

use std::sync::Arc;
use std::time::Duration;

use cm_domain::WorkspaceId;
use cm_orchestrator::{execute_resumable, OrchestratorError, RunProgress, RunRecord, TurnExecutor};
use cm_topology::TopologyGraph;
use sqlx::PgPool;
use uuid::Uuid;

use crate::recursive_exec::{SubTopologyExecutor, Tier};
use crate::topology_exec::ZeroClawDriveExecutor;

/// Requeue a `running` job whose worker hasn't checkpointed within this window.
const STALE_AFTER_SECS: f64 = 180.0;

/// Maximum age a `running` run may spend WITHOUT journaling any step
/// records before the reaper kills its container and fails it. 15 min
/// is generous: healthy first-step latency is typically 5–60s; anything
/// past this is a stuck container (usually a wedged provider CLI).
const REAP_STUCK_AFTER_SECS: i64 = 15 * 60;

/// Spawn the durable topology job worker. Polls for queued jobs every `poll`
/// interval; runs each to completion (or failure), checkpointing per step.
pub fn spawn(pool: PgPool, runtime: cm_runtime::Runtime, poll: Duration) {
    // Fire the stuck-container reaper on its own cadence — checking
    // once a minute is plenty and keeps this off the hot claim loop.
    let reaper_pool = pool.clone();
    tokio::spawn(async move {
        let mut ticker = tokio::time::interval(Duration::from_secs(60));
        loop {
            ticker.tick().await;
            if let Err(e) = reap_stuck_runs(&reaper_pool).await {
                eprintln!("topology_worker::reaper: reap failed: {e}");
            }
        }
    });

    tokio::spawn(async move {
        loop {
            // Recover jobs orphaned by a dead worker before claiming new ones.
            if let Err(e) = cm_db::repo::topology_runs::requeue_stale(&pool, STALE_AFTER_SECS).await
            {
                eprintln!("topology_worker: requeue_stale failed: {e}");
            }
            match cm_db::repo::topology_runs::claim_next_queued(&pool).await {
                Ok(Some(job)) => run_job(&pool, &runtime, job).await,
                Ok(None) => tokio::time::sleep(poll).await,
                Err(e) => {
                    eprintln!("topology_worker: claim failed: {e}");
                    tokio::time::sleep(poll).await;
                }
            }
        }
    });
}

/// Mark `running` mission-bound topology_runs that have been alive past
/// [`REAP_STUCK_AFTER_SECS`] without journaling a single step record as
/// `failed`, with a diagnostic error so the user sees WHY instead of an
/// infinitely-spinning pipeline.
///
/// Only reaps runs bound to a mission — non-mission runs (raw API-driven
/// topology runs) are left to the existing stale-checkpoint requeuer.
async fn reap_stuck_runs(pool: &PgPool) -> Result<(), sqlx::Error> {
    use sqlx::Row;
    let rows: Vec<sqlx::postgres::PgRow> = sqlx::query(
        "SELECT id, mission_id
           FROM topology_runs
          WHERE status = 'running'
            AND mission_id IS NOT NULL
            AND created_at < now() - make_interval(secs => $1::float)
            AND coalesce(jsonb_array_length(coalesce(checkpoint->'records', '[]'::jsonb)), 0) = 0",
    )
    .bind(REAP_STUCK_AFTER_SECS as f64)
    .fetch_all(pool)
    .await?;

    for row in rows {
        let id: Uuid = row.get("id");
        let mission_id: Uuid = row.get("mission_id");
        eprintln!(
            "topology_worker::reaper: reaping stuck run run_id={} mission_id={} (no step records after {}s)",
            id, mission_id, REAP_STUCK_AFTER_SECS,
        );
        let _ = cm_db::repo::topology_runs::fail(
            pool,
            id,
            &format!("reaped: no step records after {}s", REAP_STUCK_AFTER_SECS),
        )
        .await;
    }
    Ok(())
}

/// Drive one claimed job to a terminal state, persisting checkpoints as it goes.
async fn run_job(
    pool: &PgPool,
    runtime: &cm_runtime::Runtime,
    job: cm_db::repo::topology_runs::ClaimedTopologyRun,
) {
    let id = job.id;

    // Swarm runs aren't graph topologies — the `graph` JSONB holds the swarm
    // config. Branch before the graph parse and run the self-verifying loop.
    if job.tier == "swarm" {
        let cfg = job.graph.clone().unwrap_or(serde_json::Value::Null);
        let swarm_job: crate::swarm::SwarmJob = serde_json::from_value(cfg).unwrap_or_default();
        let result = crate::swarm::run_swarm_job(pool, runtime, id, swarm_job, &job.task).await;
        match result {
            Ok(record) => {
                let value = serde_json::to_value(&record).unwrap_or(serde_json::Value::Null);
                if let Err(e) = cm_db::repo::topology_runs::complete(pool, id, &value).await {
                    eprintln!("topology_worker: swarm complete({id}) failed: {e}");
                }
            }
            Err(e) => {
                let _ = cm_db::repo::topology_runs::fail(pool, id, &e).await;
            }
        }
        maybe_teardown_ephemeral_team(pool, id).await;
        return;
    }

    let Some(graph) = job
        .graph
        .as_ref()
        .and_then(|g| serde_json::from_value::<TopologyGraph>(g.clone()).ok())
    else {
        let _ = cm_db::repo::topology_runs::fail(pool, id, "missing or invalid graph").await;
        return;
    };

    // Resume from the last checkpoint, or start fresh.
    let progress: RunProgress = job
        .checkpoint
        .and_then(|c| serde_json::from_value(c).ok())
        .unwrap_or_default();

    // Missions-era runs drive through the shared, env-derived ZeroClaw
    // gateway — per-claw provisioning happens ahead of time via
    // `RuntimeProvisioner` (see `mission_orchestrator::on_launch`), so
    // there's no per-run container/gateway resolution left to do here.
    let leaf_result = ZeroClawDriveExecutor::from_env();
    let leaf = match leaf_result {
        Ok(e) => e,
        Err(e) => {
            let _ = cm_db::repo::topology_runs::fail(pool, id, &e).await;
            return;
        }
    };

    // Select the executor by deploy tier: `team` drives claws directly; the
    // upper tiers drive the recursive sub-topology executor (which runs each
    // child tier's graph, all the way down to the same leaf claw executor).
    let result = match job.tier.as_str() {
        "company" | "org" => {
            let tier = if job.tier == "org" {
                Tier::Org
            } else {
                Tier::Company
            };
            let exec = SubTopologyExecutor::new(
                pool.clone(),
                WorkspaceId::from(job.workspace_id),
                tier,
                id,
                Arc::new(leaf),
            );
            drive(pool, id, &graph, &job.task, progress, &exec).await
        }
        _ => drive(pool, id, &graph, &job.task, progress, &leaf).await,
    };

    match result {
        Ok(record) => {
            let value = serde_json::to_value(&record).unwrap_or(serde_json::Value::Null);
            if let Err(e) = cm_db::repo::topology_runs::complete(pool, id, &value).await {
                eprintln!("topology_worker: complete({id}) failed: {e}");
            }
        }
        Err(e) => {
            // Don't clobber a cancellation (or any already-terminal state) with `failed`.
            let terminal = matches!(
                cm_db::repo::topology_runs::current_status(pool, id).await,
                Ok(Some(ref s)) if s == "cancelled" || s == "completed" || s == "failed"
            );
            if !terminal {
                let _ = cm_db::repo::topology_runs::fail(pool, id, &format!("{e}")).await;
            }
        }
    }
    maybe_teardown_ephemeral_team(pool, id).await;
}

/// Post-terminal hook: if this run's team is `ephemeral` and no siblings are
/// still in flight, deprovision every bound claw on the ZeroClaw daemon,
/// delete the claw rows, and delete the team row. Best-effort — a failure to
/// tear down leaves the team intact and logs; a future sweep can retry.
async fn maybe_teardown_ephemeral_team(pool: &PgPool, id: Uuid) {
    let teardown = match cm_db::repo::topology_runs::check_ephemeral_teardown(pool, id).await {
        Ok(Some(t)) => t,
        Ok(None) => return,
        Err(e) => {
            eprintln!("topology_worker: check_ephemeral_teardown({id}) failed: {e}");
            return;
        }
    };
    // Deprovision each claw on the daemon before deleting rows — if the daemon
    // side fails we still delete our rows (the daemon can be swept for orphans
    // by the fleet-reconcile timer). This is the trade cm-api owns everywhere:
    // Postgres is authoritative, the daemon config is a cache.
    if let Some(prov) = crate::runtime_provision::RuntimeProvisioner::from_env() {
        for cid in &teardown.claw_ids {
            if let Err(e) = prov.deprovision_claw(*cid).await {
                eprintln!("topology_worker: deprovision_claw({cid}) failed: {e}");
            }
        }
    }
    for cid in &teardown.claw_ids {
        if let Err(e) = cm_db::repo::agents::hard_purge(pool, cm_domain::AgentId::from(*cid)).await
        {
            eprintln!("topology_worker: agents::hard_purge({cid}) failed: {e}");
        }
    }
    if let Err(e) = cm_db::repo::teams::delete_team(
        pool,
        teardown.team_id,
        cm_domain::WorkspaceId::from(teardown.workspace_id),
    )
    .await
    {
        eprintln!(
            "topology_worker: teams::delete_team({}) failed: {e}",
            teardown.team_id
        );
    }
}

/// Drive a graph to completion with the durable per-step checkpoint +
/// cancellation closure, generic over the executor so the team (leaf) and
/// company/org (recursive) tiers share the same outer durability logic. The
/// checkpoint here is parent-node-level (coarse resume); the recursive executor
/// additionally touches `updated_at` from each inner leaf step to stay alive.
async fn drive<E: TurnExecutor>(
    pool: &PgPool,
    id: Uuid,
    graph: &TopologyGraph,
    task: &str,
    progress: RunProgress,
    executor: &E,
) -> Result<RunRecord, OrchestratorError> {
    let pool_cb = pool.clone();
    execute_resumable(graph, task, executor, progress, move |snap| {
        let pool = pool_cb.clone();
        async move {
            // 2026-07-15: verbose per-step trace so `docker logs
            // clawmates_server_1` shows which topology node just fired,
            // its role, output size, tokens, and gated-action count.
            // Cheap (one info! per step) and gives us the missing
            // "topology is flowing" signal without needing to open the
            // canvas.
            if let Some(last) = snap.records.last() {
                let phase = match &last.phase {
                    cm_orchestrator::StepPhase::Plan => "plan",
                    cm_orchestrator::StepPhase::Work => "work",
                    cm_orchestrator::StepPhase::Synth => "synth",
                    cm_orchestrator::StepPhase::Aggregate => "aggregate",
                };
                eprintln!(
                    "topology_worker::step run_id={} step={} node={} role={} phase={} output_bytes={} tokens={} gated={}",
                    id,
                    snap.completed,
                    last.node_id,
                    last.role,
                    phase,
                    last.output.len(),
                    last.tokens,
                    last.gated.len(),
                );
            }
            // Best-effort checkpoint: a failed write just means we re-run the
            // step on resume (idempotent — topology turns are pure reads here).
            if let Ok(v) = serde_json::to_value(&snap) {
                let _ =
                    cm_db::repo::topology_runs::checkpoint(&pool, id, &v, snap.completed as i64)
                        .await;
            }
            // Honor cancellation at the step boundary: stop before the next turn.
            if matches!(
                cm_db::repo::topology_runs::current_status(&pool, id).await,
                Ok(Some(ref s)) if s == "cancelled"
            ) {
                return Err(OrchestratorError::Executor("run cancelled".into()));
            }
            Ok(())
        }
    })
    .await
}