clawmates/crates/cm-api/src/routes/slack.rs

//! Slack inbound events (§7.3: "Agent replies on @mention"). This route is
//! public — authenticity comes from Slack's request signature, which the
//! BROKER verifies against the stored signing secret (it never leaves the
//! broker). A verified @mention starts a real run in the agent's dedicated
//! Slack session; any reply the agent attempts is itself a gated outbound
//! post.

use axum::body::Bytes;
use axum::extract::State;
use axum::http::{HeaderMap, StatusCode};
use axum::Json;
use cm_domain::AgentId;
use serde_json::{json, Value};

use crate::AppState;

const SLACK_SESSION_TITLE: &str = "💬 Slack";

/// Finds the slack connection whose signing secret validates this request.
async fn verified_connection(
    state: &AppState,
    timestamp: &str,
    body: &str,
    signature: &str,
) -> Option<cm_db::repo::connections::AppConnection> {
    let socket = state.broker_socket.as_ref()?;
    let connections = sqlx::query!(
        r#"SELECT id, workspace_id, agent_id, provider, auth_type, status, secret_ref
           FROM app_connections WHERE provider = 'slack' AND status = 'connected'"#,
    )
    .fetch_all(&state.pool)
    .await
    .ok()?;
    for row in connections {
        let Some(secret_ref) = row.secret_ref else {
            continue;
        };
        let Ok(mut broker) = cm_secrets::BrokerClient::connect(socket).await else {
            return None;
        };
        if broker
            .verify_slack_signature(secret_ref, timestamp, body, signature)
            .await
            .unwrap_or(false)
        {
            return Some(cm_db::repo::connections::AppConnection {
                id: row.id,
                workspace_id: row.workspace_id,
                agent_id: row.agent_id,
                provider: row.provider,
                auth_type: row.auth_type,
                status: row.status,
                secret_ref: row.secret_ref,
            });
        }
    }
    None
}

/// POST /api/slack/events
pub async fn events(
    State(state): State<AppState>,
    headers: HeaderMap,
    body: Bytes,
) -> Result<Json<Value>, StatusCode> {
    let timestamp = headers
        .get("x-slack-request-timestamp")
        .and_then(|v| v.to_str().ok())
        .unwrap_or_default();
    let signature = headers
        .get("x-slack-signature")
        .and_then(|v| v.to_str().ok())
        .unwrap_or_default();
    let raw = String::from_utf8_lossy(&body).into_owned();

    let connection = verified_connection(&state, timestamp, &raw, signature)
        .await
        .ok_or(StatusCode::UNAUTHORIZED)?;

    let event: Value = serde_json::from_str(&raw).map_err(|_| StatusCode::BAD_REQUEST)?;
    // Slack's endpoint handshake.
    if event["type"] == "url_verification" {
        return Ok(Json(json!({ "challenge": event["challenge"] })));
    }
    if event["type"] == "event_callback" && event["event"]["type"] == "app_mention" {
        let text = event["event"]["text"]
            .as_str()
            .unwrap_or_default()
            .to_owned();
        let Some(agent_uuid) = connection.agent_id else {
            return Ok(Json(json!({ "ok": true })));
        };
        let agent_id = AgentId::from(agent_uuid);
        let Ok(agent) = cm_db::repo::agents::get(&state.pool, agent_id).await else {
            return Ok(Json(json!({ "ok": true })));
        };

        // One recognizable session per agent for Slack traffic.
        let session = match cm_db::repo::sessions::list_by_agent(&state.pool, agent_id)
            .await
            .ok()
            .and_then(|sessions| {
                sessions
                    .into_iter()
                    .find(|s| s.title == SLACK_SESSION_TITLE)
            }) {
            Some(existing) => existing,
            None => cm_db::repo::sessions::create(
                &state.pool,
                agent_id,
                agent.workspace_id,
                SLACK_SESSION_TITLE,
            )
            .await
            .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?,
        };
        // Fire-and-forget: Slack expects a fast 200; the run streams into
        // the session (and any outbound reply is gated as usual).
        let runtime = state.runtime.clone();
        tokio::spawn(async move {
            let _ = runtime.send_message(session.id, &text).await;
        });
    }
    Ok(Json(json!({ "ok": true })))
}