PiCloud/crates/manager-core/src/dead_letter_repo.rs

//! `DeadLetterRepo` — CRUD over the `dead_letters` table.
//!
//! The dispatcher writes new rows when an async trigger exhausts its
//! retry policy. Admin endpoints (commit 8) read for the dashboard
//! list view and write to mark rows resolved or replay them. The GC
//! sweeper (commit 10) deletes expired rows by `created_at`.

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use picloud_shared::{AppId, DeadLetterId, ScriptId, TriggerId};
use sqlx::PgPool;
use uuid::Uuid;

#[derive(Debug, thiserror::Error)]
pub enum DeadLetterRepoError {
    #[error("database error: {0}")]
    Db(#[from] sqlx::Error),

    #[error("dead-letter row not found: {0}")]
    NotFound(DeadLetterId),

    #[error("invalid resolution {0:?}")]
    InvalidResolution(String),
}

#[derive(Debug, Clone)]
pub struct NewDeadLetter {
    pub app_id: AppId,
    /// `outbox.id` that exhausted retries. Outbox row deleted at the
    /// same time.
    pub original_event_id: Uuid,
    pub source: String,
    pub op: String,
    pub trigger_id: Option<TriggerId>,
    pub script_id: Option<ScriptId>,
    pub payload: serde_json::Value,
    pub attempt_count: u32,
    pub first_attempt_at: DateTime<Utc>,
    pub last_attempt_at: DateTime<Utc>,
    pub last_error: String,
}

#[derive(Debug, Clone)]
pub struct DeadLetterRow {
    pub id: DeadLetterId,
    pub app_id: AppId,
    pub original_event_id: Uuid,
    pub source: String,
    pub op: String,
    pub trigger_id: Option<TriggerId>,
    pub script_id: Option<ScriptId>,
    pub payload: serde_json::Value,
    pub attempt_count: u32,
    pub first_attempt_at: DateTime<Utc>,
    pub last_attempt_at: DateTime<Utc>,
    pub last_error: String,
    pub created_at: DateTime<Utc>,
    pub resolved_at: Option<DateTime<Utc>>,
    pub resolution: Option<String>,
}

#[async_trait]
pub trait DeadLetterRepo: Send + Sync {
    /// Insert a new dead-letter row. Returns the assigned id.
    async fn insert(&self, row: NewDeadLetter) -> Result<DeadLetterId, DeadLetterRepoError>;

    async fn get(&self, id: DeadLetterId) -> Result<Option<DeadLetterRow>, DeadLetterRepoError>;

    /// Lookup for the dashboard list view. `unresolved_only=true`
    /// filters to `resolved_at IS NULL`.
    async fn list_for_app(
        &self,
        app_id: AppId,
        unresolved_only: bool,
        limit: i64,
        offset: i64,
    ) -> Result<Vec<DeadLetterRow>, DeadLetterRepoError>;

    /// Hot path for the dashboard's per-app unresolved-count badge.
    async fn unresolved_count(&self, app_id: AppId) -> Result<i64, DeadLetterRepoError>;

    /// Mark the row resolved with the given reason. The reason MUST
    /// be one of the four CHECK-constraint values
    /// (`replayed`, `ignored`, `handled_by_script`, `handler_failed`).
    async fn resolve(&self, id: DeadLetterId, reason: &str) -> Result<(), DeadLetterRepoError>;

    /// Retention sweep. Deletes rows with `created_at < older_than`
    /// up to `limit` at a time, using FOR UPDATE SKIP LOCKED to play
    /// nicely with concurrent dispatchers. Returns the count deleted.
    async fn gc(&self, older_than: DateTime<Utc>, limit: i64) -> Result<u64, DeadLetterRepoError>;
}

pub struct PostgresDeadLetterRepo {
    pool: PgPool,
}

impl PostgresDeadLetterRepo {
    #[must_use]
    pub fn new(pool: PgPool) -> Self {
        Self { pool }
    }
}

const ALLOWED_RESOLUTIONS: &[&str] =
    &["replayed", "ignored", "handled_by_script", "handler_failed"];

#[async_trait]
impl DeadLetterRepo for PostgresDeadLetterRepo {
    async fn insert(&self, row: NewDeadLetter) -> Result<DeadLetterId, DeadLetterRepoError> {
        let (id,): (Uuid,) = sqlx::query_as(
            "INSERT INTO dead_letters ( \
                app_id, original_event_id, source, op, trigger_id, script_id, \
                payload, attempt_count, first_attempt_at, last_attempt_at, last_error \
             ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) \
             RETURNING id",
        )
        .bind(row.app_id.into_inner())
        .bind(row.original_event_id)
        .bind(row.source)
        .bind(row.op)
        .bind(row.trigger_id.map(TriggerId::into_inner))
        .bind(row.script_id.map(ScriptId::into_inner))
        .bind(row.payload)
        .bind(i32::try_from(row.attempt_count).unwrap_or(0))
        .bind(row.first_attempt_at)
        .bind(row.last_attempt_at)
        .bind(row.last_error)
        .fetch_one(&self.pool)
        .await?;
        Ok(id.into())
    }

    async fn get(&self, id: DeadLetterId) -> Result<Option<DeadLetterRow>, DeadLetterRepoError> {
        let row: Option<DeadLetterRowRaw> = sqlx::query_as(
            "SELECT id, app_id, original_event_id, source, op, trigger_id, script_id, \
                    payload, attempt_count, first_attempt_at, last_attempt_at, \
                    last_error, created_at, resolved_at, resolution \
             FROM dead_letters WHERE id = $1",
        )
        .bind(id.into_inner())
        .fetch_optional(&self.pool)
        .await?;
        Ok(row.map(DeadLetterRowRaw::into_row))
    }

    async fn list_for_app(
        &self,
        app_id: AppId,
        unresolved_only: bool,
        limit: i64,
        offset: i64,
    ) -> Result<Vec<DeadLetterRow>, DeadLetterRepoError> {
        let rows: Vec<DeadLetterRowRaw> = sqlx::query_as(
            "SELECT id, app_id, original_event_id, source, op, trigger_id, script_id, \
                    payload, attempt_count, first_attempt_at, last_attempt_at, \
                    last_error, created_at, resolved_at, resolution \
             FROM dead_letters \
             WHERE app_id = $1 \
               AND ($2::bool = FALSE OR resolved_at IS NULL) \
             ORDER BY created_at DESC \
             LIMIT $3 OFFSET $4",
        )
        .bind(app_id.into_inner())
        .bind(unresolved_only)
        .bind(limit)
        .bind(offset)
        .fetch_all(&self.pool)
        .await?;
        Ok(rows.into_iter().map(DeadLetterRowRaw::into_row).collect())
    }

    async fn unresolved_count(&self, app_id: AppId) -> Result<i64, DeadLetterRepoError> {
        let (count,): (i64,) = sqlx::query_as(
            "SELECT COUNT(*) FROM dead_letters \
             WHERE app_id = $1 AND resolved_at IS NULL",
        )
        .bind(app_id.into_inner())
        .fetch_one(&self.pool)
        .await?;
        Ok(count)
    }

    async fn resolve(&self, id: DeadLetterId, reason: &str) -> Result<(), DeadLetterRepoError> {
        if !ALLOWED_RESOLUTIONS.contains(&reason) {
            return Err(DeadLetterRepoError::InvalidResolution(reason.to_string()));
        }
        let res = sqlx::query(
            "UPDATE dead_letters \
                SET resolution = $2, resolved_at = NOW() \
                WHERE id = $1",
        )
        .bind(id.into_inner())
        .bind(reason)
        .execute(&self.pool)
        .await?;
        if res.rows_affected() == 0 {
            return Err(DeadLetterRepoError::NotFound(id));
        }
        Ok(())
    }

    async fn gc(&self, older_than: DateTime<Utc>, limit: i64) -> Result<u64, DeadLetterRepoError> {
        // Tombstones picked under FOR UPDATE SKIP LOCKED so concurrent
        // sweepers (cluster mode) don't fight each other.
        let res = sqlx::query(
            "DELETE FROM dead_letters \
                WHERE id IN ( \
                    SELECT id FROM dead_letters \
                    WHERE created_at < $1 \
                    FOR UPDATE SKIP LOCKED \
                    LIMIT $2 \
                )",
        )
        .bind(older_than)
        .bind(limit)
        .execute(&self.pool)
        .await?;
        Ok(res.rows_affected())
    }
}

#[derive(sqlx::FromRow)]
struct DeadLetterRowRaw {
    id: Uuid,
    app_id: Uuid,
    original_event_id: Uuid,
    source: String,
    op: String,
    trigger_id: Option<Uuid>,
    script_id: Option<Uuid>,
    payload: serde_json::Value,
    attempt_count: i32,
    first_attempt_at: DateTime<Utc>,
    last_attempt_at: DateTime<Utc>,
    last_error: String,
    created_at: DateTime<Utc>,
    resolved_at: Option<DateTime<Utc>>,
    resolution: Option<String>,
}

impl DeadLetterRowRaw {
    fn into_row(self) -> DeadLetterRow {
        DeadLetterRow {
            id: self.id.into(),
            app_id: self.app_id.into(),
            original_event_id: self.original_event_id,
            source: self.source,
            op: self.op,
            trigger_id: self.trigger_id.map(Into::into),
            script_id: self.script_id.map(Into::into),
            payload: self.payload,
            attempt_count: u32::try_from(self.attempt_count).unwrap_or(0),
            first_attempt_at: self.first_attempt_at,
            last_attempt_at: self.last_attempt_at,
            last_error: self.last_error,
            created_at: self.created_at,
            resolved_at: self.resolved_at,
            resolution: self.resolution,
        }
    }
}