PiCloud/crates/shared/src/queue.rs

//! `QueueService` — the v1.1.9 durable queue contract.
//!
//! `queue::enqueue(name, message, opts?)` writes one row to
//! `queue_messages`; the dispatcher's queue arm claims it later via
//! `FOR UPDATE SKIP LOCKED` and fires the registered `queue:receive`
//! trigger (commit 6). On handler success the row is deleted (ack);
//! on throw it's nacked (claim cleared, retry per the trigger's policy
//! with `deliver_after = NOW() + backoff`); when `attempt >= max_attempts`
//! the message is moved to `dead_letters`.
//!
//! No `peek`/`dequeue`/`purge` script-side surface — consumers are
//! triggers. Exposing manual dequeue would force PiCloud to expose
//! visibility-timeout + nack semantics to script-land, which is the
//! whole reason queue is trigger-based.
//!
//! Identity tuple: `(cx.app_id, queue_name)`. Cross-app isolation is
//! enforced because every method derives `app_id` from `cx.app_id` —
//! never from a script argument.

use async_trait::async_trait;
use thiserror::Error;

use crate::{QueueMessageId, SdkCallCx};

/// Optional per-enqueue overrides passed to [`QueueService::enqueue`].
#[derive(Debug, Clone, Copy, Default)]
pub struct EnqueueOpts {
    /// Defer delivery until `NOW() + delay_ms`. Default: immediate.
    pub delay_ms: Option<i64>,
    /// Override the default `max_attempts` (3) for THIS message. Clamped
    /// to `[1, 20]` at the SDK boundary so a script can't enqueue a
    /// retry-forever message.
    pub max_attempts: Option<u32>,
}

#[async_trait]
pub trait QueueService: Send + Sync {
    /// Enqueue one message onto `(cx.app_id, queue_name)`. Returns the
    /// new message id (forensic; not surfaced to scripts in the
    /// script-side return).
    async fn enqueue(
        &self,
        cx: &SdkCallCx,
        queue_name: &str,
        payload: serde_json::Value,
        opts: EnqueueOpts,
    ) -> Result<QueueMessageId, QueueError>;

    /// Total queued messages (claimed + unclaimed + retrying-but-not-deleted)
    /// for `(cx.app_id, queue_name)`. Surfaces as `queue::depth(name)`.
    async fn depth(&self, cx: &SdkCallCx, queue_name: &str) -> Result<u64, QueueError>;

    /// Currently-claimable count — `claim_token IS NULL` AND `deliver_after`
    /// has passed. Surfaces as `queue::depth_pending(name)`. Diverges from
    /// `depth` for delayed / in-flight messages.
    async fn depth_pending(&self, cx: &SdkCallCx, queue_name: &str) -> Result<u64, QueueError>;
}

#[derive(Debug, Error)]
pub enum QueueError {
    #[error("queue_name must not be empty")]
    EmptyName,

    /// Caller principal lacked the required capability. Only raised when
    /// `cx.principal.is_some()` (script-as-gate; public HTTP skips it).
    /// Sibling shape across KvError / DocsError / PubsubError — uniform
    /// for 403 translation at the HTTP boundary.
    #[error("forbidden")]
    Forbidden,

    /// Payload could not be serialized (e.g. a Rhai FnPtr / closure
    /// captured into the message). Rejected at the SDK boundary; surface
    /// the wording verbatim so scripts see the documented message.
    #[error("queue rejected: {0}")]
    Rejected(String),

    /// max_attempts (or delay_ms) outside the allowed bounds.
    #[error("{0}")]
    InvalidOpts(String),

    /// JSON-encoded payload exceeded the configured `max_payload_bytes`
    /// cap (`PICLOUD_QUEUE_MAX_PAYLOAD_BYTES`). Defends against script
    /// queue::enqueue calls filling disk with multi-MB payloads.
    #[error("queue: payload too large ({actual} bytes; limit {limit})")]
    PayloadTooLarge { limit: usize, actual: usize },

    /// Backend failure (Postgres unavailable, etc.). Named `Backend` to
    /// align with KvError / DocsError / FilesError / SecretsError.
    #[error("queue backend error: {0}")]
    Backend(String),
}

/// Test-only stub: every call errors `Unavailable`. Used by harnesses
/// that build a `Services` bundle without a database.
#[derive(Debug, Default, Clone, Copy)]
pub struct NoopQueueService;

#[async_trait]
impl QueueService for NoopQueueService {
    async fn enqueue(
        &self,
        _cx: &SdkCallCx,
        _queue_name: &str,
        _payload: serde_json::Value,
        _opts: EnqueueOpts,
    ) -> Result<QueueMessageId, QueueError> {
        Err(QueueError::Backend("queue is not wired in".into()))
    }

    async fn depth(&self, _cx: &SdkCallCx, _queue_name: &str) -> Result<u64, QueueError> {
        Err(QueueError::Backend("queue is not wired in".into()))
    }

    async fn depth_pending(&self, _cx: &SdkCallCx, _queue_name: &str) -> Result<u64, QueueError> {
        Err(QueueError::Backend("queue is not wired in".into()))
    }
}