//! The §15 chain over HTTP: gateway suspends, the approvals API decides,
//! and the gateway resume streams the continuation.

use std::sync::Arc;

use cm_api::AppState;
use cm_auth::AuthService;
use cm_domain::{Role, User, UserId, Workspace, WorkspaceId};
use cm_llm::ScriptedProvider;
use cm_runtime::{Runtime, RuntimeConfig};
use eventsource_stream::Eventsource;
use futures::StreamExt;
use serde_json::{json, Value};

const SCENARIOS: &str = r#"
[[scenario]]
marker = "[[scenario:gated-email]]"

[[scenario.turns]]
events = [
    { type = "text", text = "I'll send that email." },
    { type = "tool_use", name = "email.send", input = { to = "ceo@example.com", subject = "Q2", body = "Revenue is up." } },
]

[[scenario.turns]]
events = [
    { type = "text", text = " The email step is finished." },
]
"#;

struct TestServer {
    base: String,
    client: reqwest::Client,
}

async fn serve(pool: sqlx::PgPool) -> TestServer {
    let runtime = Runtime::new(
        pool.clone(),
        Arc::new(ScriptedProvider::from_toml(SCENARIOS).unwrap()),
        RuntimeConfig::basic("scripted", 1024),
    );
    let app = cm_api::router(AppState::new(pool, runtime));
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        axum::serve(listener, app).await.unwrap();
    });
    TestServer {
        base: format!("http://{addr}"),
        client: reqwest::Client::new(),
    }
}

async fn seed_and_login(pool: &sqlx::PgPool, server: &TestServer) -> (String, String) {
    let ws = Workspace {
        id: WorkspaceId::new(),
        name: "Acme".into(),
        plan: "team".into(),
    };
    cm_db::repo::workspaces::insert(pool, &ws).await.unwrap();
    let owner = User {
        id: UserId::new(),
        workspace_id: ws.id,
        email: format!("{}@acme.test", UserId::new()),
        role: Role::Owner,
        display_name: "Owner".into(),
        created_at: time::OffsetDateTime::UNIX_EPOCH,
    };
    cm_db::repo::users::insert(pool, &owner).await.unwrap();
    AuthService::new(pool.clone())
        .set_password(owner.id, "pw")
        .await
        .unwrap();
    let token = server
        .client
        .post(format!("{}/api/auth/login", server.base))
        .json(&json!({"email": owner.email, "password": "pw"}))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap()["token"]
        .as_str()
        .unwrap()
        .to_owned();
    let claw: Value = server
        .client
        .post(format!("{}/api/claws", server.base))
        .bearer_auth(&token)
        .json(&json!({"name": "Scout", "job_title": "Analyst"}))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    (token, claw["id"].as_str().unwrap().to_owned())
}

async fn collect_sse(response: reqwest::Response) -> Vec<(String, String, Value)> {
    let mut events = Vec::new();
    let mut stream = response.bytes_stream().eventsource();
    while let Some(event) = stream.next().await {
        let event = event.unwrap();
        let data: Value = serde_json::from_str(&event.data).unwrap();
        let name = event.event.clone();
        let done = matches!(name.as_str(), "run_completed" | "error" | "run_suspended");
        events.push((event.id, name, data));
        if done {
            break;
        }
    }
    events
}

/// Starts the gated run and returns (session_key, approval_id, last_seq).
async fn suspend_gated_run(
    server: &TestServer,
    token: &str,
    claw_id: &str,
) -> (String, String, i64) {
    let session: Value = server
        .client
        .post(format!("{}/api/sessions", server.base))
        .bearer_auth(token)
        .json(&json!({"clawId": claw_id, "title": "Email"}))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let session_key = session["sessionKey"].as_str().unwrap().to_owned();

    let events = collect_sse(
        server
            .client
            .post(format!("{}/api/gateway?clawId={claw_id}", server.base))
            .bearer_auth(token)
            .json(&json!({
                "sessionKey": session_key,
                "message": "send it [[scenario:gated-email]]"
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    let (last_id, name, data) = events.last().unwrap();
    assert_eq!(name, "run_suspended");
    let approval_id = data["approval_id"].as_str().unwrap().to_owned();
    (session_key, approval_id, last_id.parse().unwrap())
}

#[tokio::test]
async fn full_chain_over_http_executes_only_after_approval() {
    let pool = cm_testkit::test_pool().await;
    let server = serve(pool.clone()).await;
    let (token, claw_id) = seed_and_login(&pool, &server).await;
    let (session_key, approval_id, last_seq) = suspend_gated_run(&server, &token, &claw_id).await;

    // Queued: the approvals API lists it with the exact preview.
    let queue: Value = server
        .client
        .get(format!("{}/api/approvals", server.base))
        .bearer_auth(&token)
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let pending = queue.as_array().unwrap();
    assert_eq!(pending.len(), 1);
    assert_eq!(pending[0]["id"], approval_id);
    assert_eq!(
        pending[0]["preview"]["summary"],
        "Send email to ceo@example.com"
    );

    // Blocked while pending.
    let outbox = sqlx::query_scalar::<_, i64>("SELECT count(*) FROM outbox")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(outbox, 0);

    // Approve over HTTP.
    let approve = server
        .client
        .post(format!(
            "{}/api/approvals/{approval_id}/approve",
            server.base
        ))
        .bearer_auth(&token)
        .send()
        .await
        .unwrap();
    assert_eq!(approve.status(), 200);

    // Double-decide is a conflict.
    let again = server
        .client
        .post(format!(
            "{}/api/approvals/{approval_id}/approve",
            server.base
        ))
        .bearer_auth(&token)
        .send()
        .await
        .unwrap();
    assert_eq!(again.status(), 409);

    // Re-attach the gateway from where we left off: the continuation
    // streams the approved execution through to completion.
    let continuation = collect_sse(
        server
            .client
            .post(format!("{}/api/gateway?clawId={claw_id}", server.base))
            .bearer_auth(&token)
            .json(&json!({"sessionKey": session_key, "resumeFrom": last_seq}))
            .send()
            .await
            .unwrap(),
    )
    .await;
    let kinds: Vec<&str> = continuation.iter().map(|(_, n, _)| n.as_str()).collect();
    assert!(kinds.contains(&"step_finished"));
    assert_eq!(*kinds.last().unwrap(), "run_completed");

    let outbox = sqlx::query_scalar::<_, i64>("SELECT count(*) FROM outbox")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(outbox, 1);

    // The queue is clear again.
    let queue: Value = server
        .client
        .get(format!("{}/api/approvals", server.base))
        .bearer_auth(&token)
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert!(queue.as_array().unwrap().is_empty());
}

#[tokio::test]
async fn reject_over_http_executes_nothing() {
    let pool = cm_testkit::test_pool().await;
    let server = serve(pool.clone()).await;
    let (token, claw_id) = seed_and_login(&pool, &server).await;
    let (session_key, approval_id, last_seq) = suspend_gated_run(&server, &token, &claw_id).await;

    let reject = server
        .client
        .post(format!(
            "{}/api/approvals/{approval_id}/reject",
            server.base
        ))
        .bearer_auth(&token)
        .send()
        .await
        .unwrap();
    assert_eq!(reject.status(), 200);

    let continuation = collect_sse(
        server
            .client
            .post(format!("{}/api/gateway?clawId={claw_id}", server.base))
            .bearer_auth(&token)
            .json(&json!({"sessionKey": session_key, "resumeFrom": last_seq}))
            .send()
            .await
            .unwrap(),
    )
    .await;
    assert_eq!(continuation.last().unwrap().1, "run_completed");

    let outbox = sqlx::query_scalar::<_, i64>("SELECT count(*) FROM outbox")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(outbox, 0);
}

#[tokio::test]
async fn approvals_are_tenant_isolated() {
    let pool = cm_testkit::test_pool().await;
    let server = serve(pool.clone()).await;
    let (token, claw_id) = seed_and_login(&pool, &server).await;
    let (other_token, _) = seed_and_login(&pool, &server).await;
    let (_, approval_id, _) = suspend_gated_run(&server, &token, &claw_id).await;

    // Another workspace sees an empty queue and cannot decide.
    let queue: Value = server
        .client
        .get(format!("{}/api/approvals", server.base))
        .bearer_auth(&other_token)
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert!(queue.as_array().unwrap().is_empty());

    let foreign = server
        .client
        .post(format!(
            "{}/api/approvals/{approval_id}/approve",
            server.base
        ))
        .bearer_auth(&other_token)
        .send()
        .await
        .unwrap();
    assert_eq!(foreign.status(), 404);
}