//! The P4 blocking property: Slack outbound is gated, and the approved
//! execution happens inside the broker (grant consumed there, bot token
//! never visible to the runtime). Real broker socket, real Postgres, real
//! local Slack-shaped receiver.

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

use axum::extract::State;
use axum::routing::post;
use cm_domain::{
    AccessPolicy, Agent, AgentId, AgentStatus, Role, RunState, User, UserId, Workspace, WorkspaceId,
};
use cm_llm::ScriptedProvider;
use cm_runtime::{RunEventBody, Runtime, RuntimeConfig};
use cm_safety::{approvals, Decision};
use cm_secrets::{BrokerClient, BrokerServer, FileKey};
use serde_json::{json, Value};
use tokio::sync::Mutex;

const SCENARIOS: &str = r##"
[[scenario]]
marker = "[[scenario:slack-post]]"

[[scenario.turns]]
events = [
    { type = "tool_use", name = "slack.post", input = { channel = "#general", text = "Q2 numbers are in." } },
]

[[scenario.turns]]
events = [
    { type = "text", text = " Posted to Slack." },
]
"##;

type Posts = Arc<Mutex<Vec<(String, Value)>>>;

/// A Slack-shaped receiver recording (authorization, body) pairs.
async fn spawn_slack_sink() -> (String, Posts) {
    let seen: Posts = Arc::new(Mutex::new(Vec::new()));
    let state = seen.clone();
    let app = axum::Router::new()
        .route(
            "/chat.postMessage",
            post(
                |State(seen): State<Posts>,
                 headers: axum::http::HeaderMap,
                 axum::Json(body): axum::Json<Value>| async move {
                    let auth = headers
                        .get("authorization")
                        .and_then(|v| v.to_str().ok())
                        .unwrap_or("")
                        .to_owned();
                    seen.lock().await.push((auth, body));
                    axum::Json(json!({"ok": true}))
                },
            ),
        )
        .with_state(state);
    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();
    });
    (format!("http://{addr}"), seen)
}

async fn spawn_broker(pool: sqlx::PgPool) -> std::path::PathBuf {
    let dir = std::env::temp_dir().join(format!("tc-slk-{}", uuid::Uuid::now_v7().simple()));
    std::fs::create_dir_all(&dir).unwrap();
    let key_path = dir.join("broker.key");
    FileKey::generate(&key_path).unwrap();
    let key = FileKey::load(&key_path).unwrap();
    let short = uuid::Uuid::now_v7().simple().to_string();
    let socket = std::path::PathBuf::from(format!("/tmp/tcs-{}.sock", &short[short.len() - 12..]));
    let server = BrokerServer::new(pool, key, socket.clone());
    tokio::spawn(async move {
        server.serve().await.unwrap();
    });
    for _ in 0..50 {
        if socket.exists() {
            break;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    socket
}

#[tokio::test]
async fn slack_post_blocks_then_the_broker_executes_exactly_once() {
    let pool = cm_testkit::test_pool().await;
    let (sink_url, posts) = spawn_slack_sink().await;
    let socket = spawn_broker(pool.clone()).await;

    // Seed workspace, owner, agent.
    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();
    let agent = Agent {
        id: AgentId::new(),
        workspace_id: ws.id,
        name: "Scout".into(),
        job_title: "Analyst".into(),
        system_prompt: String::new(),
        avatar: String::new(),
        accent: String::new(),
        wallpaper: String::new(),
        managed_by: owner.id,
        status: AgentStatus::Online,
    };
    cm_db::repo::agents::insert(&pool, &agent, &AccessPolicy::default())
        .await
        .unwrap();

    // Connect Slack: bot token stored through the broker, row references it.
    let mut client = BrokerClient::connect(&socket).await.unwrap();
    let secret_ref = client
        .store_secret(ws.id, "slack_bot_token", "xoxb-test-token")
        .await
        .unwrap();
    cm_db::repo::connections::insert(&pool, ws.id, Some(agent.id), "slack", "keys", secret_ref)
        .await
        .unwrap();

    let rt = Runtime::new(
        pool.clone(),
        Arc::new(ScriptedProvider::from_toml(SCENARIOS).unwrap()),
        RuntimeConfig {
            model: "scripted".into(),
            max_tokens: 1024,
            sandboxes: None,
            browser: None,
            terminals: None,
            broker_socket: Some(socket),
            slack_base_url: sink_url,
            providers: Default::default(),
        },
    );
    let session = cm_db::repo::sessions::create(&pool, agent.id, ws.id, "Slack")
        .await
        .unwrap();
    let started = rt
        .send_message(session.id, "post it [[scenario:slack-post]]")
        .await
        .unwrap();

    // Suspends with the outbound_message category; nothing posted.
    let mut rx = started.events;
    let mut suspended = false;
    while let Ok(envelope) = rx.recv().await {
        if matches!(envelope.event, RunEventBody::RunSuspended { .. }) {
            suspended = true;
            break;
        }
        if matches!(envelope.event, RunEventBody::Error { .. }) {
            break;
        }
    }
    assert!(suspended);
    assert!(posts.lock().await.is_empty(), "blocked while pending");

    // Approve → the broker posts with the bot token injected.
    let pending = approvals::list_pending(&pool, ws.id).await.unwrap();
    assert_eq!(
        pending[0].category,
        cm_domain::GatedCategory::OutboundMessage
    );
    approvals::decide(&pool, pending[0].id, owner.id, Decision::Approve)
        .await
        .unwrap();
    rt.resume_run(cm_safety::ResumeReady {
        run_id: started.run_id,
        approval_id: pending[0].id,
        approved: true,
    })
    .await
    .unwrap();

    for _ in 0..100 {
        let run = cm_db::repo::runs::get(&pool, started.run_id).await.unwrap();
        if run.state == RunState::Completed {
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    let observed = posts.lock().await;
    assert_eq!(observed.len(), 1, "exactly one post");
    assert_eq!(observed[0].0, "Bearer xoxb-test-token");
    assert_eq!(observed[0].1["channel"], "#general");
    assert_eq!(observed[0].1["text"], "Q2 numbers are in.");
    drop(observed);

    // The grant was consumed by the broker: replay refuses.
    let replay = cm_safety::grants::consume(&pool, pending[0].id).await;
    assert!(replay.is_err(), "grant must already be consumed");
}