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

    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}");
            }
            freeze_research_outcome(pool, id, &record.final_output).await;
        }
        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_transition_research_topic(pool, id).await;
}

/// If this run belongs to a research topic, snapshot the orchestrator's
/// final synthesis as a versioned `research_outcomes` row. The frontend
/// canvas reads `latest_outcome` for anything past `standby` so reviewers
/// see the produced draft rather than the original prompt. Best-effort:
/// a failure here logs but doesn't fail the run.
async fn freeze_research_outcome(pool: &PgPool, run_id: Uuid, final_output: &str) {
    let topic_id = match cm_db::repo::topology_runs::research_topic_id(pool, run_id).await {
        Ok(Some(id)) => id,
        Ok(None) => return,
        Err(e) => {
            eprintln!("topology_worker: research_topic_id({run_id}) failed: {e}");
            return;
        }
    };
    if final_output.trim().is_empty() {
        return;
    }
    if let Err(e) =
        cm_db::repo::research_outcomes::insert(pool, topic_id, final_output, Some(run_id)).await
    {
        eprintln!("topology_worker: research_outcomes::insert({run_id}) failed: {e}");
    }
}

/// Post-terminal hook: if this run belongs to a research topic and it was
/// the last sibling in flight, transition the topic `processing → reviewing`.
/// Best-effort — a DB hiccup here logs but doesn't fail the run.
async fn maybe_transition_research_topic(pool: &PgPool, id: Uuid) {
    match cm_db::repo::topology_runs::notify_run_completed(pool, id).await {
        Ok(true) => {
            // Left intentionally quiet on success; the UI polls the topic
            // status. Future: emit a run_event so live viewers see it flip.
        }
        Ok(false) => {}
        Err(e) => eprintln!("topology_worker: notify_run_completed({id}) failed: {e}"),
    }
    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 {
            // 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
}