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

//! `PostgresDeadLetterService` — replaces `NoopDeadLetterService` in
//! v1.1.1's `Services` bundle. Implements `replay` (re-enqueue the
//! original event into the outbox + mark the DL row replayed) and
//! `resolve` (close the row out with a reason).
//!
//! Both methods are gated by `Capability::AppDeadLetterManage(AppId)`
//! evaluated against `cx.principal`. Public-HTTP scripts with
//! `principal: None` fail the check — design notes §4: managing
//! dead letters is an admin act.

use std::sync::Arc;

use async_trait::async_trait;
use picloud_shared::{DeadLetterError, DeadLetterId, DeadLetterService, SdkCallCx};

use crate::authz::{self, AuthzRepo, Capability};
use crate::dead_letter_repo::{DeadLetterRepo, DeadLetterRepoError, DeadLetterRow};
use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxSourceKind};

pub struct PostgresDeadLetterService {
    repo: Arc<dyn DeadLetterRepo>,
    outbox: Arc<dyn OutboxRepo>,
    authz: Arc<dyn AuthzRepo>,
}

impl PostgresDeadLetterService {
    #[must_use]
    pub fn new(
        repo: Arc<dyn DeadLetterRepo>,
        outbox: Arc<dyn OutboxRepo>,
        authz: Arc<dyn AuthzRepo>,
    ) -> Self {
        Self {
            repo,
            outbox,
            authz,
        }
    }

    async fn require_dl_capability(&self, cx: &SdkCallCx) -> Result<(), DeadLetterError> {
        let Some(ref principal) = cx.principal else {
            return Err(DeadLetterError::Forbidden);
        };
        authz::require(
            &*self.authz,
            principal,
            Capability::AppDeadLetterManage(cx.app_id),
        )
        .await
        .map_err(|_| DeadLetterError::Forbidden)
    }

    async fn load_row(&self, id: DeadLetterId) -> Result<DeadLetterRow, DeadLetterError> {
        self.repo
            .get(id)
            .await
            .map_err(map_repo_err)?
            .ok_or(DeadLetterError::NotFound)
    }
}

#[async_trait]
impl DeadLetterService for PostgresDeadLetterService {
    async fn replay(&self, cx: &SdkCallCx, id: DeadLetterId) -> Result<(), DeadLetterError> {
        self.require_dl_capability(cx).await?;
        let row = self.load_row(id).await?;
        if row.app_id != cx.app_id {
            // Cross-app — treat as not-found to avoid leaking
            // information about other apps' dead letters.
            return Err(DeadLetterError::NotFound);
        }

        let source_kind = OutboxSourceKind::from_wire(&row.source).unwrap_or(OutboxSourceKind::Kv);
        self.outbox
            .insert(NewOutboxRow {
                app_id: row.app_id,
                source_kind,
                trigger_id: row.trigger_id,
                script_id: row.script_id,
                reply_to: None,
                payload: row.payload.clone(),
                origin_principal: None,
                trigger_depth: 0,
                root_execution_id: None,
            })
            .await
            .map_err(|e| DeadLetterError::Backend(e.to_string()))?;

        self.repo
            .resolve(id, "replayed")
            .await
            .map_err(map_repo_err)?;
        Ok(())
    }

    async fn resolve(
        &self,
        cx: &SdkCallCx,
        id: DeadLetterId,
        reason: &str,
    ) -> Result<(), DeadLetterError> {
        self.require_dl_capability(cx).await?;
        let row = self.load_row(id).await?;
        if row.app_id != cx.app_id {
            return Err(DeadLetterError::NotFound);
        }
        self.repo.resolve(id, reason).await.map_err(map_repo_err)?;
        Ok(())
    }
}

fn map_repo_err(e: DeadLetterRepoError) -> DeadLetterError {
    match e {
        DeadLetterRepoError::NotFound(_) => DeadLetterError::NotFound,
        DeadLetterRepoError::InvalidResolution(s) => DeadLetterError::InvalidResolution(s),
        DeadLetterRepoError::Db(e) => DeadLetterError::Backend(e.to_string()),
    }
}