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

//! Loop endpoints — CRUD, enable/disable, immediate-run, and the public
//! webhook receiver.
//!
//!   GET    /api/loops                 list workspace's loops
//!   POST   /api/loops                 create
//!   GET    /api/loops/:id             detail
//!   PATCH  /api/loops/:id             update definition
//!   DELETE /api/loops/:id             delete
//!   POST   /api/loops/:id/run         trigger one iteration NOW (bypass schedule)
//!   POST   /api/loops/:id/enable      set enabled=true; recomputes next_fire_at
//!   POST   /api/loops/:id/disable     set enabled=false
//!   POST   /webhooks/loops/:token     public; HMAC-SHA256-verified via
//!                                     X-Loop-Signature: sha256=<hex>

use axum::body::Bytes;
use axum::extract::{Path, State};
use axum::http::{HeaderMap, StatusCode};
use axum::Json;
use base64::Engine;
use cm_runtime::scheduling::next_occurrence;
use hmac::{Hmac, Mac};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use time::OffsetDateTime;
use uuid::Uuid;

use crate::{ApiError, AppState, Authed};

#[derive(Deserialize)]
pub struct CreateLoopRequest {
    pub title: String,
    pub description: String,
    pub graph: Value,
    pub task_template: String,
    /// {cron?: '0 */6 * * *', on_completion?: bool, webhook_enabled?: bool}
    #[serde(default)]
    pub triggers: Value,
    /// {kind: 'infinite' | 'iters' | 'until', n?: int}
    #[serde(default = "default_repeat")]
    pub repeat_policy: Value,
    #[serde(default)]
    pub agents: Vec<AgentSlotInput>,
    #[serde(default)]
    pub teams: Vec<Uuid>,
    #[serde(default)]
    pub orgs: Vec<Uuid>,
}
fn default_repeat() -> Value {
    serde_json::json!({"kind": "infinite"})
}

#[derive(Deserialize)]
pub struct AgentSlotInput {
    pub agent_id: Uuid,
    #[serde(default)]
    pub role_slot: Option<String>,
}

#[derive(Serialize)]
pub struct LoopCreated {
    pub id: Uuid,
    /// Set when `triggers.webhook_enabled == true`. The full URL is
    /// `<origin>/webhooks/loops/<webhook_token>`; the signing key is
    /// returned exactly once at creation and never surfaced again.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub webhook_token: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub webhook_signing_key: Option<String>,
}

fn parse_triggers(v: &Value) -> Option<Triggers> {
    serde_json::from_value(v.clone()).ok()
}

#[derive(Deserialize)]
struct Triggers {
    #[serde(default)]
    cron: Option<String>,
    #[serde(default)]
    #[allow(dead_code)]
    on_completion: bool,
    #[serde(default)]
    webhook_enabled: bool,
}

fn make_webhook_material() -> (String, String) {
    // 24 bytes ≈ 192 bits of entropy each; URL-safe base64 for the token,
    // standard base64 for the signing key.
    let mut token_buf = [0u8; 24];
    let mut key_buf = [0u8; 24];
    // getrandom is already in the dep tree via base64/hmac/etc; failure
    // (broken kernel RNG) is fatal enough that unwrapping is fine here.
    getrandom::getrandom(&mut token_buf).expect("OS RNG");
    getrandom::getrandom(&mut key_buf).expect("OS RNG");
    let token = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(token_buf);
    let key = base64::engine::general_purpose::STANDARD_NO_PAD.encode(key_buf);
    (token, key)
}

fn compute_next_fire(triggers: &Value) -> Option<OffsetDateTime> {
    let t = parse_triggers(triggers)?;
    let pattern = t.cron?;
    if pattern.trim().is_empty() {
        return None;
    }
    next_occurrence(pattern.trim(), OffsetDateTime::now_utc()).ok()
}

pub async fn create_loop(
    State(state): State<AppState>,
    Authed(user): Authed,
    Json(body): Json<CreateLoopRequest>,
) -> Result<(StatusCode, Json<LoopCreated>), ApiError> {
    if body.title.trim().is_empty() || body.task_template.trim().is_empty() {
        return Err(ApiError::BadRequest);
    }
    let webhook_enabled = parse_triggers(&body.triggers)
        .map(|t| t.webhook_enabled)
        .unwrap_or(false);
    let (webhook_token, webhook_signing_key) = if webhook_enabled {
        let (t, k) = make_webhook_material();
        (Some(t), Some(k))
    } else {
        (None, None)
    };
    let next_fire_at = compute_next_fire(&body.triggers);

    let id = cm_db::repo::loops::create(
        &state.pool,
        cm_db::repo::loops::NewLoop {
            workspace_id: user.workspace_id.as_uuid(),
            title: body.title.trim(),
            description: body.description.trim(),
            graph: &body.graph,
            task_template: body.task_template.trim(),
            triggers: &body.triggers,
            repeat_policy: &body.repeat_policy,
            enabled: true,
            next_fire_at,
            webhook_token: webhook_token.as_deref(),
            webhook_signing_key: webhook_signing_key.as_deref(),
            created_by: user.user_id.as_uuid(),
        },
    )
    .await?;

    apply_staffing(&state.pool, id, &body.agents, &body.teams, &body.orgs).await?;

    Ok((
        StatusCode::CREATED,
        Json(LoopCreated {
            id,
            webhook_token,
            webhook_signing_key,
        }),
    ))
}

async fn apply_staffing(
    pool: &sqlx::PgPool,
    loop_id: Uuid,
    agents: &[AgentSlotInput],
    teams: &[Uuid],
    orgs: &[Uuid],
) -> Result<(), ApiError> {
    let slots: Vec<cm_db::repo::loops::AgentSlot> = agents
        .iter()
        .map(|a| cm_db::repo::loops::AgentSlot {
            agent_id: a.agent_id,
            role_slot: a.role_slot.clone(),
        })
        .collect();
    cm_db::repo::loops::set_agents(pool, loop_id, &slots).await?;
    cm_db::repo::loops::set_teams(pool, loop_id, teams).await?;
    cm_db::repo::loops::set_orgs(pool, loop_id, orgs).await?;
    Ok(())
}

#[derive(Serialize)]
pub struct LoopWithStaffing {
    #[serde(flatten)]
    pub inner: cm_db::repo::loops::Loop,
    pub agents: Vec<cm_db::repo::loops::AgentSlot>,
    pub teams: Vec<Uuid>,
    pub orgs: Vec<Uuid>,
}

async fn hydrate_staffing(
    pool: &sqlx::PgPool,
    inner: cm_db::repo::loops::Loop,
) -> Result<LoopWithStaffing, ApiError> {
    let id = inner.id;
    let agents = cm_db::repo::loops::agents(pool, id).await?;
    let teams = cm_db::repo::loops::teams(pool, id).await?;
    let orgs = cm_db::repo::loops::orgs(pool, id).await?;
    Ok(LoopWithStaffing {
        inner,
        agents,
        teams,
        orgs,
    })
}

pub async fn list_loops(
    State(state): State<AppState>,
    Authed(user): Authed,
) -> Result<Json<Vec<LoopWithStaffing>>, ApiError> {
    let loops = cm_db::repo::loops::list(&state.pool, user.workspace_id.as_uuid()).await?;
    let mut out = Vec::with_capacity(loops.len());
    for l in loops {
        out.push(hydrate_staffing(&state.pool, l).await?);
    }
    Ok(Json(out))
}

pub async fn get_loop(
    State(state): State<AppState>,
    Authed(user): Authed,
    Path(id): Path<Uuid>,
) -> Result<Json<LoopWithStaffing>, ApiError> {
    let inner = cm_db::repo::loops::get(&state.pool, id, user.workspace_id.as_uuid())
        .await?
        .ok_or(ApiError::NotFound)?;
    Ok(Json(hydrate_staffing(&state.pool, inner).await?))
}

#[derive(Deserialize)]
pub struct UpdateLoopRequest {
    pub title: String,
    pub description: String,
    pub graph: Value,
    pub task_template: String,
    pub triggers: Value,
    pub repeat_policy: Value,
    #[serde(default)]
    pub agents: Vec<AgentSlotInput>,
    #[serde(default)]
    pub teams: Vec<Uuid>,
    #[serde(default)]
    pub orgs: Vec<Uuid>,
}

pub async fn patch_loop(
    State(state): State<AppState>,
    Authed(user): Authed,
    Path(id): Path<Uuid>,
    Json(body): Json<UpdateLoopRequest>,
) -> Result<StatusCode, ApiError> {
    if body.title.trim().is_empty() || body.task_template.trim().is_empty() {
        return Err(ApiError::BadRequest);
    }
    cm_db::repo::loops::get(&state.pool, id, user.workspace_id.as_uuid())
        .await?
        .ok_or(ApiError::NotFound)?;
    let next_fire_at = compute_next_fire(&body.triggers);
    cm_db::repo::loops::update(
        &state.pool,
        id,
        user.workspace_id.as_uuid(),
        cm_db::repo::loops::UpdateLoop {
            title: body.title.trim(),
            description: body.description.trim(),
            graph: &body.graph,
            task_template: body.task_template.trim(),
            triggers: &body.triggers,
            repeat_policy: &body.repeat_policy,
            next_fire_at,
        },
    )
    .await?;
    apply_staffing(&state.pool, id, &body.agents, &body.teams, &body.orgs).await?;
    Ok(StatusCode::NO_CONTENT)
}

pub async fn delete_loop(
    State(state): State<AppState>,
    Authed(user): Authed,
    Path(id): Path<Uuid>,
) -> Result<StatusCode, ApiError> {
    cm_db::repo::loops::delete(&state.pool, id, user.workspace_id.as_uuid()).await?;
    Ok(StatusCode::NO_CONTENT)
}

pub async fn enable_loop(
    State(state): State<AppState>,
    Authed(user): Authed,
    Path(id): Path<Uuid>,
) -> Result<StatusCode, ApiError> {
    cm_db::repo::loops::set_enabled(&state.pool, id, user.workspace_id.as_uuid(), true).await?;
    Ok(StatusCode::NO_CONTENT)
}

pub async fn disable_loop(
    State(state): State<AppState>,
    Authed(user): Authed,
    Path(id): Path<Uuid>,
) -> Result<StatusCode, ApiError> {
    cm_db::repo::loops::set_enabled(&state.pool, id, user.workspace_id.as_uuid(), false).await?;
    Ok(StatusCode::NO_CONTENT)
}

#[derive(Serialize)]
pub struct RunTriggered {
    pub run_id: Uuid,
    pub iteration: i32,
}

/// `POST /api/loops/:id/run` — enqueue one iteration NOW, bypassing the
/// scheduler and any trigger config. Iteration counter continues from
/// wherever it was; parent_run_id chains to whatever last_run_id points at.
pub async fn run_now(
    State(state): State<AppState>,
    Authed(user): Authed,
    Path(id): Path<Uuid>,
) -> Result<Json<RunTriggered>, ApiError> {
    let l = cm_db::repo::loops::get(&state.pool, id, user.workspace_id.as_uuid())
        .await?
        .ok_or(ApiError::NotFound)?;
    let iter = cm_db::repo::loops::next_iteration(&state.pool, l.id).await?;
    let run_id = cm_db::repo::loops::enqueue_iteration(
        &state.pool,
        l.id,
        l.workspace_id,
        &l.task_template,
        &l.graph,
        iter,
        l.last_run_id,
    )
    .await?;
    cm_db::repo::loops::mark_fired(&state.pool, l.id, run_id, l.next_fire_at).await?;
    Ok(Json(RunTriggered {
        run_id,
        iteration: iter,
    }))
}

/// `POST /webhooks/loops/:token` — public, HMAC-verified. Enqueues one
/// iteration on the loop that owns `token`. Returns 202 + `{run_id}` on
/// success, 401 on missing/bad signature, 404 on unknown token.
pub async fn webhook_receive(
    State(state): State<AppState>,
    Path(token): Path<String>,
    headers: HeaderMap,
    body: Bytes,
) -> (StatusCode, Json<Value>) {
    let Ok(Some((_id, _ws, key, l))) =
        cm_db::repo::loops::get_by_webhook_token(&state.pool, &token).await
    else {
        return (StatusCode::NOT_FOUND, Json(Value::Null));
    };

    let Some(sig_header) = headers
        .get("X-Loop-Signature")
        .and_then(|v| v.to_str().ok())
    else {
        return (StatusCode::UNAUTHORIZED, Json(Value::Null));
    };
    let Some(provided) = sig_header.strip_prefix("sha256=") else {
        return (StatusCode::UNAUTHORIZED, Json(Value::Null));
    };
    if !verify_hmac(&key, &body, provided) {
        return (StatusCode::UNAUTHORIZED, Json(Value::Null));
    }

    let iter = match cm_db::repo::loops::next_iteration(&state.pool, l.id).await {
        Ok(n) => n,
        Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(Value::Null)),
    };
    let run_id = match cm_db::repo::loops::enqueue_iteration(
        &state.pool,
        l.id,
        l.workspace_id,
        &l.task_template,
        &l.graph,
        iter,
        l.last_run_id,
    )
    .await
    {
        Ok(r) => r,
        Err(_) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(Value::Null)),
    };
    let _ = cm_db::repo::loops::mark_fired(&state.pool, l.id, run_id, None).await;

    (
        StatusCode::ACCEPTED,
        Json(serde_json::json!({"run_id": run_id, "iteration": iter})),
    )
}

fn verify_hmac(key: &str, body: &[u8], provided_hex: &str) -> bool {
    let Ok(mut mac) = Hmac::<sha2::Sha256>::new_from_slice(key.as_bytes()) else {
        return false;
    };
    mac.update(body);
    let expected = hex::encode(mac.finalize().into_bytes());
    if expected.len() != provided_hex.len() {
        return false;
    }
    // Constant-time compare.
    expected
        .bytes()
        .zip(provided_hex.bytes())
        .fold(0u8, |acc, (a, b)| acc | (a ^ b))
        == 0
}