//! The durable topology-job lifecycle at the persistence layer: enqueue → claim
//! (CAS) → checkpoint → complete, plus the stale-run resume sweep. This is the
//! foundation that lets long-horizon topology runs survive worker restarts.

use cm_db::repo::{loops, research_topics, teams, topology_runs, users, workspaces};
use cm_domain::{
    AccessPolicy, Agent, AgentId, AgentStatus, Role, User, UserId, Workspace, WorkspaceId,
};
use serde_json::json;
use uuid::Uuid;

async fn seed_user(pool: &sqlx::PgPool, ws: WorkspaceId, email: &str) -> UserId {
    let user = User {
        id: UserId::new(),
        workspace_id: ws,
        email: email.into(),
        role: Role::Owner,
        display_name: "Owner".into(),
        created_at: time::OffsetDateTime::UNIX_EPOCH,
    };
    users::insert(pool, &user).await.unwrap();
    user.id
}

/// Enqueue a durable topology run with `research_topic_id` set. Kept in the
/// test file to avoid a production repo helper for the topic-scoped enqueue
/// path (nothing else in the codebase writes this column yet).
async fn enqueue_run_with_topic(
    pool: &sqlx::PgPool,
    workspace_id: WorkspaceId,
    task: &str,
    topic_id: Uuid,
) -> Uuid {
    let id = Uuid::now_v7();
    let graph = json!({"kind": "pipeline", "nodes": [{"id":"n","role":"r"}], "edges": []});
    sqlx::query!(
        "INSERT INTO topology_runs
            (id, workspace_id, task, kind, status, graph, tier, research_topic_id)
         VALUES ($1, $2, $3, 'run', 'queued', $4, 'team', $5)",
        id,
        workspace_id.as_uuid(),
        task,
        graph,
        topic_id,
    )
    .execute(pool)
    .await
    .unwrap();
    id
}

async fn seed_workspace(pool: &sqlx::PgPool) -> WorkspaceId {
    let ws = Workspace {
        id: WorkspaceId::new(),
        name: "Acme".into(),
        plan: "team".into(),
    };
    workspaces::insert(pool, &ws).await.unwrap();
    ws.id
}

#[tokio::test]
async fn durable_run_lifecycle_enqueue_claim_checkpoint_complete() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;

    let id = Uuid::now_v7();
    let graph = json!({"kind": "pipeline", "nodes": [{"id":"n1","role":"drafter"}], "edges": []});
    topology_runs::enqueue_run(&pool, id, ws, "write a haiku", &graph)
        .await
        .unwrap();

    // Status starts queued, no result yet.
    let st = topology_runs::status(&pool, id, ws).await.unwrap();
    assert_eq!(st.status, "queued");
    assert_eq!(st.kind, "run");
    assert!(st.result.is_none());

    // Claim flips it to running and returns the job + its graph.
    let claimed = topology_runs::claim_next_queued(&pool)
        .await
        .unwrap()
        .expect("a queued job to claim");
    assert_eq!(claimed.id, id);
    assert!(claimed.graph.is_some());
    assert!(claimed.checkpoint.is_none());

    // A second claim finds nothing (the job is no longer queued).
    assert!(topology_runs::claim_next_queued(&pool)
        .await
        .unwrap()
        .is_none());

    // Checkpoint mid-run progress.
    let progress = json!({"completed": 1, "outputs": ["draft"], "records": [], "totals": {}});
    topology_runs::checkpoint(&pool, id, &progress, 1)
        .await
        .unwrap();
    let st = topology_runs::status(&pool, id, ws).await.unwrap();
    assert_eq!(st.status, "running");
    assert_eq!(st.last_event_id, 1);
    assert!(st.checkpoint.is_some());

    // Complete with a result blob.
    let result = json!({"final_output": "a haiku", "totals": {"turns": 1}});
    topology_runs::complete(&pool, id, &result).await.unwrap();
    let st = topology_runs::status(&pool, id, ws).await.unwrap();
    assert_eq!(st.status, "completed");
    assert_eq!(st.result.unwrap()["final_output"], "a haiku");
}

#[tokio::test]
async fn stale_running_jobs_are_requeued_for_resume() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;

    let id = Uuid::now_v7();
    let graph = json!({"kind": "pipeline", "nodes": [{"id":"n1","role":"drafter"}], "edges": []});
    topology_runs::enqueue_run(&pool, id, ws, "task", &graph)
        .await
        .unwrap();
    // Claim it → running.
    topology_runs::claim_next_queued(&pool)
        .await
        .unwrap()
        .unwrap();

    // Not stale yet (just claimed) → sweep is a no-op.
    assert_eq!(topology_runs::requeue_stale(&pool, 60.0).await.unwrap(), 0);

    // With a zero threshold the running job counts as stale and is requeued,
    // so the next claim picks it up again (resume from checkpoint).
    assert_eq!(topology_runs::requeue_stale(&pool, 0.0).await.unwrap(), 1);
    let st = topology_runs::status(&pool, id, ws).await.unwrap();
    assert_eq!(st.status, "queued");
    assert!(topology_runs::claim_next_queued(&pool)
        .await
        .unwrap()
        .is_some());
}

#[tokio::test]
async fn cancel_transitions_only_active_runs() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;

    let id = Uuid::now_v7();
    let graph = json!({"kind": "pipeline", "nodes": [{"id":"n1","role":"drafter"}], "edges": []});
    topology_runs::enqueue_run(&pool, id, ws, "task", &graph)
        .await
        .unwrap();

    // A queued run cancels; the worker sees the new status (no workspace scope).
    assert!(topology_runs::cancel(&pool, id, ws).await.unwrap());
    assert_eq!(
        topology_runs::current_status(&pool, id)
            .await
            .unwrap()
            .as_deref(),
        Some("cancelled")
    );
    assert_eq!(
        topology_runs::status(&pool, id, ws).await.unwrap().status,
        "cancelled"
    );

    // Already terminal → cannot cancel again; wrong workspace → no-op.
    assert!(!topology_runs::cancel(&pool, id, ws).await.unwrap());
    let other = seed_workspace(&pool).await;
    let id2 = Uuid::now_v7();
    topology_runs::enqueue_run(&pool, id2, ws, "t", &graph)
        .await
        .unwrap();
    assert!(!topology_runs::cancel(&pool, id2, other).await.unwrap());
}

#[tokio::test]
async fn list_by_loop_returns_only_that_loops_iterations_newest_first() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;

    // A loop needs a valid `created_by` user in the same workspace.
    let user_id = seed_user(&pool, ws, "hop@acme.test").await;

    let graph = json!({"kind": "pipeline", "nodes": [{"id":"n1","role":"drafter"}], "edges": []});
    let triggers = json!({});
    let repeat = json!({"kind": "infinite"});
    let loop_id = loops::create(
        &pool,
        loops::NewLoop {
            workspace_id: ws.as_uuid(),
            title: "L",
            description: "d",
            graph: &graph,
            task_template: "task",
            triggers: &triggers,
            repeat_policy: &repeat,
            enabled: true,
            next_fire_at: None,
            webhook_token: None,
            webhook_signing_key: None,
            created_by: user_id.as_uuid(),
        },
    )
    .await
    .unwrap();

    // Three iterations of the target loop, plus a naked run + another loop's
    // iteration that should be filtered out.
    let it1 = loops::enqueue_iteration(&pool, loop_id, ws.as_uuid(), "t1", &graph, 1, None)
        .await
        .unwrap();
    let it2 = loops::enqueue_iteration(&pool, loop_id, ws.as_uuid(), "t2", &graph, 2, Some(it1))
        .await
        .unwrap();
    let it3 = loops::enqueue_iteration(&pool, loop_id, ws.as_uuid(), "t3", &graph, 3, Some(it2))
        .await
        .unwrap();
    topology_runs::enqueue_run(&pool, Uuid::now_v7(), ws, "naked", &graph)
        .await
        .unwrap();
    let other_loop = loops::create(
        &pool,
        loops::NewLoop {
            workspace_id: ws.as_uuid(),
            title: "L2",
            description: "d2",
            graph: &graph,
            task_template: "task2",
            triggers: &triggers,
            repeat_policy: &repeat,
            enabled: true,
            next_fire_at: None,
            webhook_token: None,
            webhook_signing_key: None,
            created_by: user_id.as_uuid(),
        },
    )
    .await
    .unwrap();
    let _other_it =
        loops::enqueue_iteration(&pool, other_loop, ws.as_uuid(), "other", &graph, 1, None)
            .await
            .unwrap();

    let rows = topology_runs::list_by_loop(&pool, ws, loop_id, 20)
        .await
        .unwrap();

    assert_eq!(rows.len(), 3, "only the target loop's iterations");
    // Newest iteration first.
    assert_eq!(rows[0].iteration, Some(3));
    assert_eq!(rows[0].id, it3);
    assert_eq!(rows[1].iteration, Some(2));
    assert_eq!(rows[1].id, it2);
    assert_eq!(rows[2].iteration, Some(1));
    assert_eq!(rows[2].id, it1);
    // Each iteration is still a durable run — finished_at is None until completion.
    assert!(rows.iter().all(|r| r.finished_at.is_none()));
    assert!(rows.iter().all(|r| r.kind == "run"));

    // Wrong workspace: nothing.
    let other_ws = seed_workspace(&pool).await;
    let cross = topology_runs::list_by_loop(&pool, other_ws, loop_id, 20)
        .await
        .unwrap();
    assert!(cross.is_empty(), "loops are scoped by workspace");
}

#[tokio::test]
async fn notify_run_completed_transitions_topic_when_no_siblings_in_flight() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;
    let user_id = seed_user(&pool, ws, "researcher@acme.test").await;

    let topic = research_topics::create(
        &pool,
        ws.as_uuid(),
        "Topic",
        "desc",
        "spec",
        user_id.as_uuid(),
    )
    .await
    .unwrap();
    // The auto-transition only fires while the topic is `processing`.
    research_topics::set_status(&pool, topic, ws.as_uuid(), "processing")
        .await
        .unwrap();

    let run_id = enqueue_run_with_topic(&pool, ws, "task", topic).await;
    // Flip to a terminal state before calling — mirrors the worker order.
    let result = json!({"final_output": "done"});
    topology_runs::complete(&pool, run_id, &result)
        .await
        .unwrap();

    let transitioned = topology_runs::notify_run_completed(&pool, run_id)
        .await
        .unwrap();
    assert!(transitioned, "no siblings in flight → topic transitions");
    let t = research_topics::get(&pool, topic, ws.as_uuid())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(t.status, "reviewing");

    // Idempotent: a second call after the topic has already left `processing`
    // is a no-op.
    let again = topology_runs::notify_run_completed(&pool, run_id)
        .await
        .unwrap();
    assert!(
        !again,
        "second call is a no-op — topic is no longer processing"
    );
}

#[tokio::test]
async fn notify_run_completed_leaves_topic_processing_when_siblings_in_flight() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;
    let user_id = seed_user(&pool, ws, "researcher@acme.test").await;

    let topic = research_topics::create(
        &pool,
        ws.as_uuid(),
        "Topic",
        "desc",
        "spec",
        user_id.as_uuid(),
    )
    .await
    .unwrap();
    research_topics::set_status(&pool, topic, ws.as_uuid(), "processing")
        .await
        .unwrap();

    let done = enqueue_run_with_topic(&pool, ws, "first", topic).await;
    let _still_queued = enqueue_run_with_topic(&pool, ws, "second", topic).await;
    topology_runs::complete(&pool, done, &json!({"final_output": "x"}))
        .await
        .unwrap();

    let transitioned = topology_runs::notify_run_completed(&pool, done)
        .await
        .unwrap();
    assert!(!transitioned, "sibling still queued → hold");
    let t = research_topics::get(&pool, topic, ws.as_uuid())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(t.status, "processing");
}

#[tokio::test]
async fn notify_run_completed_ignores_runs_with_no_research_topic() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;

    let id = Uuid::now_v7();
    let graph = json!({"kind": "pipeline", "nodes": [{"id":"n","role":"r"}], "edges": []});
    topology_runs::enqueue_run(&pool, id, ws, "task", &graph)
        .await
        .unwrap();
    topology_runs::complete(&pool, id, &json!({}))
        .await
        .unwrap();

    let transitioned = topology_runs::notify_run_completed(&pool, id)
        .await
        .unwrap();
    assert!(
        !transitioned,
        "no research_topic_id → nothing to transition"
    );
}

async fn seed_team(
    pool: &sqlx::PgPool,
    ws: WorkspaceId,
    user_id: UserId,
    lifecycle: &str,
    claw_count: usize,
) -> (Uuid, Vec<Uuid>) {
    let team_id = Uuid::now_v7();
    let graph = json!({"kind": "hub_spoke", "nodes": [], "edges": []});
    teams::insert_team_with_lifecycle(pool, team_id, ws, "T", "hub_spoke", &graph, lifecycle)
        .await
        .unwrap();
    let mut claws = Vec::with_capacity(claw_count);
    for i in 0..claw_count {
        let agent = Agent {
            id: AgentId::new(),
            workspace_id: ws,
            name: format!("claw{i}"),
            job_title: "worker".into(),
            system_prompt: String::new(),
            avatar: String::new(),
            accent: String::new(),
            wallpaper: String::new(),
            managed_by: user_id,
            status: AgentStatus::Online,
        };
        cm_db::repo::agents::insert(pool, &agent, &AccessPolicy::default())
            .await
            .unwrap();
        teams::add_member(
            pool,
            team_id,
            &format!("n{i}"),
            agent.id.as_uuid(),
            "worker",
        )
        .await
        .unwrap();
        claws.push(agent.id.as_uuid());
    }
    (team_id, claws)
}

#[tokio::test]
async fn check_ephemeral_teardown_returns_claws_when_no_siblings_left() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;
    let user_id = seed_user(&pool, ws, "eph@acme.test").await;
    let (team_id, claws) = seed_team(&pool, ws, user_id, "ephemeral", 3).await;

    let graph = json!({"kind": "hub_spoke", "nodes": [], "edges": []});
    let run_id = Uuid::now_v7();
    topology_runs::enqueue_run_for_team(&pool, run_id, ws, "task", &graph, team_id)
        .await
        .unwrap();
    topology_runs::complete(&pool, run_id, &json!({}))
        .await
        .unwrap();

    let teardown = topology_runs::check_ephemeral_teardown(&pool, run_id)
        .await
        .unwrap()
        .expect("ephemeral team, no siblings — should return teardown");
    assert_eq!(teardown.team_id, team_id);
    assert_eq!(teardown.workspace_id, ws.as_uuid());
    let mut got = teardown.claw_ids.clone();
    let mut want = claws.clone();
    got.sort();
    want.sort();
    assert_eq!(got, want, "returns every bound claw");
}

#[tokio::test]
async fn check_ephemeral_teardown_holds_when_siblings_in_flight() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;
    let user_id = seed_user(&pool, ws, "eph@acme.test").await;
    let (team_id, _) = seed_team(&pool, ws, user_id, "ephemeral", 2).await;

    let graph = json!({"kind": "hub_spoke", "nodes": [], "edges": []});
    let done = Uuid::now_v7();
    let still_queued = Uuid::now_v7();
    topology_runs::enqueue_run_for_team(&pool, done, ws, "first", &graph, team_id)
        .await
        .unwrap();
    topology_runs::enqueue_run_for_team(&pool, still_queued, ws, "second", &graph, team_id)
        .await
        .unwrap();
    topology_runs::complete(&pool, done, &json!({}))
        .await
        .unwrap();

    let teardown = topology_runs::check_ephemeral_teardown(&pool, done)
        .await
        .unwrap();
    assert!(teardown.is_none(), "sibling still queued → hold teardown");
}

#[tokio::test]
async fn check_ephemeral_teardown_ignores_permanent_teams() {
    let pool = cm_testkit::test_pool().await;
    let ws = seed_workspace(&pool).await;
    let user_id = seed_user(&pool, ws, "perm@acme.test").await;
    let (team_id, _) = seed_team(&pool, ws, user_id, "permanent", 1).await;

    let graph = json!({"kind": "hub_spoke", "nodes": [], "edges": []});
    let run_id = Uuid::now_v7();
    topology_runs::enqueue_run_for_team(&pool, run_id, ws, "task", &graph, team_id)
        .await
        .unwrap();
    topology_runs::complete(&pool, run_id, &json!({}))
        .await
        .unwrap();

    let teardown = topology_runs::check_ephemeral_teardown(&pool, run_id)
        .await
        .unwrap();
    assert!(teardown.is_none(), "permanent teams are never torn down");
}