//! 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;

/// 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) {
    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;
                }
            }
        }
    });
}

/// 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;
            }
        }
        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();

    let leaf = match ZeroClawDriveExecutor::from_env() {
        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;
            }
        }
    }
}

/// 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 {
            // 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
}