diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index 2826a1d..b80e976 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -55,6 +55,7 @@ pub mod outbox_repo; pub mod principal_resolver; pub mod pubsub_repo; pub mod pubsub_service; +pub mod queue_repo; pub mod realtime_authority; pub mod repo; pub mod route_admin; diff --git a/crates/manager-core/src/queue_repo.rs b/crates/manager-core/src/queue_repo.rs new file mode 100644 index 0000000..2c4b032 --- /dev/null +++ b/crates/manager-core/src/queue_repo.rs @@ -0,0 +1,463 @@ +//! `QueueRepo` — CRUD over the `queue_messages` table. +//! +//! Producer path: `enqueue(...)` writes one row. Dispatcher hot path: +//! `claim(...)` does the single-round-trip atomic claim +//! (`UPDATE ... WHERE id = (SELECT ... FOR UPDATE SKIP LOCKED)`). +//! Outcome: `ack` deletes the row, `nack` clears the claim with a +//! `deliver_after` retry delay, `dead_letter` moves the payload to +//! `dead_letters` and deletes the queue row. +//! +//! `reclaim_visibility_timeouts()` is the periodic safety net — a +//! crashed dispatcher (or one whose handler hung past the visibility +//! timeout) loses its claim and the message becomes claimable again. +//! +//! All methods take `app_id` explicitly; the service layer derives it +//! from `cx.app_id` and passes it in — repo never reads cx. + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use picloud_shared::{AdminUserId, AppId, QueueMessageId, TriggerId}; +use sqlx::PgPool; +use uuid::Uuid; + +use crate::dead_letter_repo::{DeadLetterRepoError, NewDeadLetter}; + +#[derive(Debug, thiserror::Error)] +pub enum QueueRepoError { + #[error("database error: {0}")] + Db(#[from] sqlx::Error), + + /// JSON serialization failure on the payload (extremely rare; only + /// surfaces if a non-serializable Value somehow reached the repo). + #[error("payload serialization: {0}")] + Payload(String), + + /// Surfaces from the dead_letter() helper if the dead-letter insert + /// fails. + #[error("dead-letter write: {0}")] + DeadLetter(String), +} + +/// Insert payload — what `QueueService::enqueue` hands the repo. +#[derive(Debug, Clone)] +pub struct NewQueueMessage { + pub app_id: AppId, + pub queue_name: String, + pub payload: serde_json::Value, + /// `Some` → defer until `NOW() + delay`; `None` → immediate. + pub deliver_after: Option>, + pub max_attempts: u32, + pub enqueued_by_principal: Option, +} + +/// One claimed message ready for handler dispatch. +#[derive(Debug, Clone)] +pub struct ClaimedMessage { + pub id: QueueMessageId, + pub app_id: AppId, + pub queue_name: String, + pub payload: serde_json::Value, + pub enqueued_at: DateTime, + pub attempt: u32, + pub max_attempts: u32, + pub claim_token: Uuid, +} + +/// Aggregate counts for the dashboard's read-only queue overview. +#[derive(Debug, Clone, Copy, Default)] +pub struct QueueStats { + /// `COUNT(*)` for `(app_id, queue_name)`. + pub total: u64, + /// Claimable now — `claim_token IS NULL AND (deliver_after IS NULL OR <= NOW())`. + pub pending: u64, + /// Currently leased to a dispatcher — `claim_token IS NOT NULL`. + pub claimed: u64, +} + +#[async_trait] +pub trait QueueRepo: Send + Sync { + /// Insert one row. Returns the new message id. + async fn enqueue(&self, msg: NewQueueMessage) -> Result; + + /// Atomic claim of a single ready message from `(app_id, queue_name)`. + /// Returns `Ok(None)` if nothing is claimable. + async fn claim( + &self, + app_id: AppId, + queue_name: &str, + ) -> Result, QueueRepoError>; + + /// Handler succeeded: delete the row IFF the claim_token matches + /// (so a stale dispatcher whose lease expired doesn't accidentally + /// delete a re-claimed message). Returns `true` if 1 row deleted. + async fn ack( + &self, + message_id: QueueMessageId, + claim_token: Uuid, + ) -> Result; + + /// Handler threw: clear the claim, defer redelivery by `retry_delay`. + /// IFF the claim_token matches. Returns `true` on success. + async fn nack( + &self, + message_id: QueueMessageId, + claim_token: Uuid, + retry_delay: chrono::Duration, + ) -> Result; + + /// Periodic safety net: clear claims whose `claimed_at` exceeds the + /// per-queue `visibility_timeout_secs` (joined from + /// `queue_trigger_details`). Returns the number of rows reclaimed. + async fn reclaim_visibility_timeouts(&self) -> Result; + + /// `queue::depth(name)` — total rows in the queue. + async fn depth(&self, app_id: AppId, queue_name: &str) -> Result; + + /// `queue::depth_pending(name)` — currently claimable rows. + async fn depth_pending(&self, app_id: AppId, queue_name: &str) + -> Result; + + /// Dashboard read-only view: every distinct queue name in `app_id` + /// with its aggregate counts. + async fn list_for_app( + &self, + app_id: AppId, + ) -> Result, QueueRepoError>; + + /// Dispatcher dead-letter helper: move a claimed message to + /// `dead_letters` and delete it from `queue_messages`, in one + /// transaction. The dispatcher's existing `fan_out_dead_letter` + /// fires registered `dead_letter` triggers off the new row. + #[allow(clippy::too_many_arguments)] + async fn dead_letter( + &self, + message_id: QueueMessageId, + claim_token: Uuid, + app_id: AppId, + queue_name: &str, + trigger_id: Option, + script_id: Option, + attempt: u32, + first_attempt_at: DateTime, + last_error: &str, + ) -> Result; +} + +pub struct PostgresQueueRepo { + pool: PgPool, +} + +impl PostgresQueueRepo { + #[must_use] + pub fn new(pool: PgPool) -> Self { + Self { pool } + } +} + +#[async_trait] +impl QueueRepo for PostgresQueueRepo { + async fn enqueue(&self, msg: NewQueueMessage) -> Result { + let (id,): (Uuid,) = sqlx::query_as( + "INSERT INTO queue_messages ( \ + app_id, queue_name, payload, deliver_after, \ + max_attempts, enqueued_by_principal \ + ) VALUES ($1, $2, $3, $4, $5, $6) \ + RETURNING id", + ) + .bind(msg.app_id.into_inner()) + .bind(&msg.queue_name) + .bind(&msg.payload) + .bind(msg.deliver_after) + .bind(i32::try_from(msg.max_attempts).unwrap_or(3)) + .bind(msg.enqueued_by_principal.map(AdminUserId::into_inner)) + .fetch_one(&self.pool) + .await?; + Ok(id.into()) + } + + async fn claim( + &self, + app_id: AppId, + queue_name: &str, + ) -> Result, QueueRepoError> { + let token = Uuid::new_v4(); + let row: Option = sqlx::query_as( + "UPDATE queue_messages \ + SET claim_token = $3, claimed_at = NOW(), attempt = attempt + 1 \ + WHERE id = ( \ + SELECT id FROM queue_messages \ + WHERE app_id = $1 AND queue_name = $2 \ + AND claim_token IS NULL \ + AND (deliver_after IS NULL OR deliver_after <= NOW()) \ + ORDER BY enqueued_at \ + FOR UPDATE SKIP LOCKED \ + LIMIT 1 \ + ) \ + RETURNING id, app_id, queue_name, payload, enqueued_at, attempt, max_attempts", + ) + .bind(app_id.into_inner()) + .bind(queue_name) + .bind(token) + .fetch_optional(&self.pool) + .await?; + Ok(row.map(|r| ClaimedMessage { + id: r.id.into(), + app_id: r.app_id.into(), + queue_name: r.queue_name, + payload: r.payload, + enqueued_at: r.enqueued_at, + attempt: u32::try_from(r.attempt).unwrap_or(1), + max_attempts: u32::try_from(r.max_attempts).unwrap_or(3), + claim_token: token, + })) + } + + async fn ack( + &self, + message_id: QueueMessageId, + claim_token: Uuid, + ) -> Result { + let res = sqlx::query( + "DELETE FROM queue_messages WHERE id = $1 AND claim_token = $2", + ) + .bind(message_id.into_inner()) + .bind(claim_token) + .execute(&self.pool) + .await?; + Ok(res.rows_affected() == 1) + } + + async fn nack( + &self, + message_id: QueueMessageId, + claim_token: Uuid, + retry_delay: chrono::Duration, + ) -> Result { + let next = Utc::now() + retry_delay; + let res = sqlx::query( + "UPDATE queue_messages \ + SET claim_token = NULL, claimed_at = NULL, deliver_after = $3 \ + WHERE id = $1 AND claim_token = $2", + ) + .bind(message_id.into_inner()) + .bind(claim_token) + .bind(next) + .execute(&self.pool) + .await?; + Ok(res.rows_affected() == 1) + } + + async fn reclaim_visibility_timeouts(&self) -> Result { + let res = sqlx::query( + "UPDATE queue_messages m \ + SET claim_token = NULL, claimed_at = NULL \ + FROM triggers t \ + JOIN queue_trigger_details d ON d.trigger_id = t.id \ + WHERE m.app_id = t.app_id \ + AND m.queue_name = d.queue_name \ + AND m.claim_token IS NOT NULL \ + AND m.claimed_at < NOW() - (d.visibility_timeout_secs || ' seconds')::INTERVAL \ + AND t.kind = 'queue' AND t.enabled = TRUE", + ) + .execute(&self.pool) + .await?; + Ok(res.rows_affected()) + } + + async fn depth(&self, app_id: AppId, queue_name: &str) -> Result { + let (n,): (i64,) = sqlx::query_as( + "SELECT COUNT(*) FROM queue_messages WHERE app_id = $1 AND queue_name = $2", + ) + .bind(app_id.into_inner()) + .bind(queue_name) + .fetch_one(&self.pool) + .await?; + Ok(u64::try_from(n).unwrap_or(0)) + } + + async fn depth_pending( + &self, + app_id: AppId, + queue_name: &str, + ) -> Result { + let (n,): (i64,) = sqlx::query_as( + "SELECT COUNT(*) FROM queue_messages \ + WHERE app_id = $1 AND queue_name = $2 \ + AND claim_token IS NULL \ + AND (deliver_after IS NULL OR deliver_after <= NOW())", + ) + .bind(app_id.into_inner()) + .bind(queue_name) + .fetch_one(&self.pool) + .await?; + Ok(u64::try_from(n).unwrap_or(0)) + } + + async fn list_for_app( + &self, + app_id: AppId, + ) -> Result, QueueRepoError> { + let rows: Vec = sqlx::query_as( + "SELECT queue_name, \ + COUNT(*) AS total, \ + COUNT(*) FILTER ( \ + WHERE claim_token IS NULL \ + AND (deliver_after IS NULL OR deliver_after <= NOW()) \ + ) AS pending, \ + COUNT(*) FILTER (WHERE claim_token IS NOT NULL) AS claimed \ + FROM queue_messages \ + WHERE app_id = $1 \ + GROUP BY queue_name \ + ORDER BY queue_name", + ) + .bind(app_id.into_inner()) + .fetch_all(&self.pool) + .await?; + Ok(rows + .into_iter() + .map(|r| { + ( + r.queue_name, + QueueStats { + total: u64::try_from(r.total).unwrap_or(0), + pending: u64::try_from(r.pending).unwrap_or(0), + claimed: u64::try_from(r.claimed).unwrap_or(0), + }, + ) + }) + .collect()) + } + + async fn dead_letter( + &self, + message_id: QueueMessageId, + claim_token: Uuid, + app_id: AppId, + queue_name: &str, + trigger_id: Option, + script_id: Option, + attempt: u32, + first_attempt_at: DateTime, + last_error: &str, + ) -> Result { + let mut tx = self.pool.begin().await?; + // Pull the row inside the transaction; we need the original + // payload for the dead_letters row. Filtered by claim_token so + // a lost lease can't dead-letter a re-claimed message. + let row: QueueRowMinimal = sqlx::query_as( + "SELECT id, payload, enqueued_at FROM queue_messages \ + WHERE id = $1 AND claim_token = $2", + ) + .bind(message_id.into_inner()) + .bind(claim_token) + .fetch_one(&mut *tx) + .await?; + + let payload = serde_json::json!({ + "source": "queue", + "queue_name": queue_name, + "message": row.payload, + "enqueued_at": row.enqueued_at, + "attempt": attempt, + "message_id": row.id.to_string(), + }); + + let dl_id = picloud_shared::DeadLetterId::new(); + sqlx::query( + "INSERT INTO dead_letters ( \ + id, 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, 'queue', 'receive', $4, $5, $6, $7, $8, NOW(), $9)", + ) + .bind(dl_id.into_inner()) + .bind(app_id.into_inner()) + .bind(row.id) + .bind(trigger_id.map(TriggerId::into_inner)) + .bind(script_id.map(picloud_shared::ScriptId::into_inner)) + .bind(&payload) + .bind(i32::try_from(attempt).unwrap_or(0)) + .bind(first_attempt_at) + .bind(last_error) + .execute(&mut *tx) + .await?; + + sqlx::query("DELETE FROM queue_messages WHERE id = $1 AND claim_token = $2") + .bind(message_id.into_inner()) + .bind(claim_token) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + Ok(dl_id) + } +} + +/// Bridge between this repo and the existing `DeadLetterRepo`. Not used +/// by `dead_letter()` directly (that path inlines its own INSERT for one- +/// transaction atomicity), but exposed for callers that already have a +/// `DeadLetterRepo` and want to keep one writer. +#[must_use] +pub fn build_new_dead_letter_from_claimed( + msg: &ClaimedMessage, + trigger_id: Option, + script_id: Option, + last_error: String, + first_attempt_at: DateTime, +) -> NewDeadLetter { + let payload = serde_json::json!({ + "source": "queue", + "queue_name": msg.queue_name, + "message": msg.payload, + "enqueued_at": msg.enqueued_at, + "attempt": msg.attempt, + "message_id": msg.id.to_string(), + }); + NewDeadLetter { + app_id: msg.app_id, + original_event_id: msg.id.into_inner(), + source: "queue".into(), + op: "receive".into(), + trigger_id, + script_id, + payload, + attempt_count: msg.attempt, + first_attempt_at, + last_attempt_at: Utc::now(), + last_error, + } +} + +/// Surface to bridge DeadLetterRepoError → QueueRepoError when callers +/// route the dead-letter write through the standard repo instead of the +/// inline `dead_letter()` path. +impl From for QueueRepoError { + fn from(e: DeadLetterRepoError) -> Self { + Self::DeadLetter(e.to_string()) + } +} + +#[derive(sqlx::FromRow)] +struct ClaimedRow { + id: Uuid, + app_id: Uuid, + queue_name: String, + payload: serde_json::Value, + enqueued_at: DateTime, + attempt: i32, + max_attempts: i32, +} + +#[derive(sqlx::FromRow)] +struct QueueAggRow { + queue_name: String, + total: i64, + pending: i64, + claimed: i64, +} + +#[derive(sqlx::FromRow)] +struct QueueRowMinimal { + id: Uuid, + payload: serde_json::Value, + enqueued_at: DateTime, +} diff --git a/crates/shared/src/ids.rs b/crates/shared/src/ids.rs index f599a91..6204c72 100644 --- a/crates/shared/src/ids.rs +++ b/crates/shared/src/ids.rs @@ -56,3 +56,4 @@ id_type!(ApiKeyId); id_type!(TriggerId); id_type!(AppUserId); id_type!(InvitationId); +id_type!(QueueMessageId); diff --git a/crates/shared/src/lib.rs b/crates/shared/src/lib.rs index a8eaaaa..c55db9a 100644 --- a/crates/shared/src/lib.rs +++ b/crates/shared/src/lib.rs @@ -23,6 +23,7 @@ pub mod log_sink; pub mod modules; pub mod outbox_writer; pub mod pubsub; +pub mod queue; pub mod realtime; pub mod realtime_authority; pub mod route; @@ -53,8 +54,8 @@ pub use files::{ }; pub use http::{HttpError, HttpRequest, HttpResponse, HttpService, NoopHttpService}; pub use ids::{ - AdminUserId, ApiKeyId, AppId, AppUserId, ExecutionId, InvitationId, RequestId, ScriptId, - TriggerId, + AdminUserId, ApiKeyId, AppId, AppUserId, ExecutionId, InvitationId, QueueMessageId, RequestId, + ScriptId, TriggerId, }; pub use inbox::{ InboxDeliveryOutcome, InboxFailureKind, InboxResolver, InboxResult, NoopInboxResolver, @@ -66,6 +67,7 @@ pub use outbox_writer::{HttpDispatchPayload, NewHttpOutbox, OutboxWriter, Outbox pub use pubsub::{ topic_matches, validate_topic_pattern, NoopPubsubService, PubsubError, PubsubService, }; +pub use queue::{EnqueueOpts, NoopQueueService, QueueError, QueueService}; pub use realtime::{BroadcasterError, NoopRealtimeBroadcaster, RealtimeBroadcaster, RealtimeEvent}; pub use realtime_authority::{DenyAllRealtimeAuthority, RealtimeAuthority, SubscribeDenied}; pub use route::{DispatchMode, HostKind, PathKind, Route}; diff --git a/crates/shared/src/queue.rs b/crates/shared/src/queue.rs new file mode 100644 index 0000000..a4c1c2c --- /dev/null +++ b/crates/shared/src/queue.rs @@ -0,0 +1,103 @@ +//! `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, + /// 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, +} + +#[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; + + /// 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; + + /// 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; +} + +#[derive(Debug, Error)] +pub enum QueueError { + #[error("queue_name must not be empty")] + EmptyName, + + /// 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), + + /// Backend unavailable (Postgres down, etc.). + #[error("queue backend error: {0}")] + Unavailable(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 { + Err(QueueError::Unavailable("queue is not wired in".into())) + } + + async fn depth(&self, _cx: &SdkCallCx, _queue_name: &str) -> Result { + Err(QueueError::Unavailable("queue is not wired in".into())) + } + + async fn depth_pending(&self, _cx: &SdkCallCx, _queue_name: &str) -> Result { + Err(QueueError::Unavailable("queue is not wired in".into())) + } +}