clawmates/crates/cm-db/src/repo/run_events.rs

use sqlx::PgPool;
use uuid::Uuid;

use crate::DbError;

/// One persisted gateway event (§13): the journal the SSE stream and
/// reconnect replay both read from.
#[derive(Debug, Clone, PartialEq)]
pub struct RunEvent {
    pub run_id: Uuid,
    pub seq: i64,
    pub event_type: String,
    pub payload: serde_json::Value,
}

/// Persists an event. Called BEFORE the event is emitted to any client so
/// the journal is always at least as complete as what observers saw.
pub async fn append(
    pool: &PgPool,
    run_id: Uuid,
    seq: i64,
    event_type: &str,
    payload: serde_json::Value,
) -> Result<(), DbError> {
    sqlx::query!(
        "INSERT INTO run_events (run_id, seq, event_type, payload)
         VALUES ($1, $2, $3, $4)",
        run_id,
        seq,
        event_type,
        payload,
    )
    .execute(pool)
    .await?;
    Ok(())
}

/// Events after a client's `resumeFrom` offset, in order.
pub async fn list_after(
    pool: &PgPool,
    run_id: Uuid,
    after_seq: i64,
) -> Result<Vec<RunEvent>, DbError> {
    let rows = sqlx::query!(
        "SELECT run_id, seq, event_type, payload
         FROM run_events WHERE run_id = $1 AND seq > $2 ORDER BY seq",
        run_id,
        after_seq,
    )
    .fetch_all(pool)
    .await?;
    Ok(rows
        .into_iter()
        .map(|row| RunEvent {
            run_id: row.run_id,
            seq: row.seq,
            event_type: row.event_type,
            payload: row.payload,
        })
        .collect())
}