diff --git a/crates/executor-core/tests/module_redaction_logging.rs b/crates/executor-core/tests/module_redaction_logging.rs index 4330b93..646f5ef 100644 --- a/crates/executor-core/tests/module_redaction_logging.rs +++ b/crates/executor-core/tests/module_redaction_logging.rs @@ -104,6 +104,7 @@ async fn original_backend_error_is_logged_at_error_level() { Arc::new(picloud_shared::NoopSecretsService), Arc::new(picloud_shared::NoopEmailService), Arc::new(picloud_shared::NoopUsersService), + Arc::new(picloud_shared::NoopQueueService), ); let engine = Engine::new(Limits::default(), services); diff --git a/crates/executor-core/tests/modules.rs b/crates/executor-core/tests/modules.rs index 9f63847..3118950 100644 --- a/crates/executor-core/tests/modules.rs +++ b/crates/executor-core/tests/modules.rs @@ -102,6 +102,7 @@ fn services_with(modules: Arc) -> Services { Arc::new(picloud_shared::NoopSecretsService), Arc::new(picloud_shared::NoopEmailService), Arc::new(picloud_shared::NoopUsersService), + Arc::new(picloud_shared::NoopQueueService), ) } diff --git a/crates/executor-core/tests/sdk_docs.rs b/crates/executor-core/tests/sdk_docs.rs index 87d8553..b99353f 100644 --- a/crates/executor-core/tests/sdk_docs.rs +++ b/crates/executor-core/tests/sdk_docs.rs @@ -233,6 +233,7 @@ fn make_engine() -> Arc { Arc::new(picloud_shared::NoopSecretsService), Arc::new(picloud_shared::NoopEmailService), Arc::new(picloud_shared::NoopUsersService), + Arc::new(picloud_shared::NoopQueueService), ); Arc::new(Engine::new(Limits::default(), services)) } diff --git a/crates/executor-core/tests/sdk_email.rs b/crates/executor-core/tests/sdk_email.rs index 99d4c67..33bde64 100644 --- a/crates/executor-core/tests/sdk_email.rs +++ b/crates/executor-core/tests/sdk_email.rs @@ -42,6 +42,7 @@ fn engine_with(rec: Arc) -> Arc { Arc::new(picloud_shared::NoopSecretsService), rec, Arc::new(picloud_shared::NoopUsersService), + Arc::new(picloud_shared::NoopQueueService), ); Arc::new(Engine::new(Limits::default(), services)) } diff --git a/crates/executor-core/tests/sdk_files.rs b/crates/executor-core/tests/sdk_files.rs index 61fce83..9f35a79 100644 --- a/crates/executor-core/tests/sdk_files.rs +++ b/crates/executor-core/tests/sdk_files.rs @@ -170,6 +170,7 @@ fn make_engine() -> Arc { Arc::new(picloud_shared::NoopSecretsService), Arc::new(picloud_shared::NoopEmailService), Arc::new(picloud_shared::NoopUsersService), + Arc::new(picloud_shared::NoopQueueService), ); Arc::new(Engine::new(Limits::default(), services)) } diff --git a/crates/executor-core/tests/sdk_http.rs b/crates/executor-core/tests/sdk_http.rs index 9f944a6..be4850d 100644 --- a/crates/executor-core/tests/sdk_http.rs +++ b/crates/executor-core/tests/sdk_http.rs @@ -93,6 +93,7 @@ fn engine_with(http: Arc) -> Arc { Arc::new(picloud_shared::NoopSecretsService), Arc::new(picloud_shared::NoopEmailService), Arc::new(picloud_shared::NoopUsersService), + Arc::new(picloud_shared::NoopQueueService), ); Arc::new(Engine::new(Limits::default(), services)) } diff --git a/crates/executor-core/tests/sdk_kv.rs b/crates/executor-core/tests/sdk_kv.rs index dfc1a33..458427d 100644 --- a/crates/executor-core/tests/sdk_kv.rs +++ b/crates/executor-core/tests/sdk_kv.rs @@ -112,6 +112,7 @@ fn make_engine() -> Arc { Arc::new(picloud_shared::NoopSecretsService), Arc::new(picloud_shared::NoopEmailService), Arc::new(picloud_shared::NoopUsersService), + Arc::new(picloud_shared::NoopQueueService), ); Arc::new(Engine::new(Limits::default(), services)) } diff --git a/crates/executor-core/tests/sdk_pubsub.rs b/crates/executor-core/tests/sdk_pubsub.rs index 628a9e0..7bb87c2 100644 --- a/crates/executor-core/tests/sdk_pubsub.rs +++ b/crates/executor-core/tests/sdk_pubsub.rs @@ -50,6 +50,7 @@ fn make_engine(svc: Arc) -> Arc { Arc::new(picloud_shared::NoopSecretsService), Arc::new(picloud_shared::NoopEmailService), Arc::new(picloud_shared::NoopUsersService), + Arc::new(picloud_shared::NoopQueueService), ); Arc::new(Engine::new(Limits::default(), services)) } diff --git a/crates/executor-core/tests/sdk_secrets.rs b/crates/executor-core/tests/sdk_secrets.rs index 53c70ce..5306106 100644 --- a/crates/executor-core/tests/sdk_secrets.rs +++ b/crates/executor-core/tests/sdk_secrets.rs @@ -103,6 +103,7 @@ fn make_engine() -> Arc { Arc::new(InMemorySecrets::default()), Arc::new(picloud_shared::NoopEmailService), Arc::new(picloud_shared::NoopUsersService), + Arc::new(picloud_shared::NoopQueueService), ); Arc::new(Engine::new(Limits::default(), services)) } diff --git a/crates/executor-core/tests/sdk_subscriber_token.rs b/crates/executor-core/tests/sdk_subscriber_token.rs index f365cc1..cf38c79 100644 --- a/crates/executor-core/tests/sdk_subscriber_token.rs +++ b/crates/executor-core/tests/sdk_subscriber_token.rs @@ -97,6 +97,7 @@ fn make_engine() -> Arc { Arc::new(picloud_shared::NoopSecretsService), Arc::new(picloud_shared::NoopEmailService), Arc::new(picloud_shared::NoopUsersService), + Arc::new(picloud_shared::NoopQueueService), ); Arc::new(Engine::new(Limits::default(), services)) } diff --git a/crates/manager-core/src/authz.rs b/crates/manager-core/src/authz.rs index 2258eb6..1b46c47 100644 --- a/crates/manager-core/src/authz.rs +++ b/crates/manager-core/src/authz.rs @@ -89,6 +89,12 @@ pub enum Capability { /// (v1.1.5). Maps to `script:write` on API keys (a publish is a /// write that fans out to subscribers). Granted to `editor`+. AppPubsubPublish(AppId), + /// Enqueue a message onto this app's queue from a script (v1.1.9). + /// Maps to `script:write` on API keys (an enqueue is a write that + /// fans out to the registered consumer). Granted to `editor`+. + /// `depth` / `depth_pending` are read-only inspection and don't gate + /// — scripts in the app can always see their own queue depths. + AppQueueEnqueue(AppId), /// Read a decrypted secret from this app's secrets store (v1.1.7). /// Same trust shape as KV/docs/files read — granted to `viewer`+, /// maps to `script:read` on API keys. Honors the seven-scope @@ -156,6 +162,7 @@ impl Capability { | Self::AppFilesRead(id) | Self::AppFilesWrite(id) | Self::AppPubsubPublish(id) + | Self::AppQueueEnqueue(id) | Self::AppSecretsRead(id) | Self::AppSecretsWrite(id) | Self::AppEmailSend(id) @@ -191,6 +198,7 @@ impl Capability { | Self::AppHttpRequest(_) | Self::AppFilesWrite(_) | Self::AppPubsubPublish(_) + | Self::AppQueueEnqueue(_) | Self::AppSecretsWrite(_) | Self::AppEmailSend(_) | Self::AppUsersWrite(_) @@ -358,6 +366,7 @@ const fn role_satisfies(role: AppRole, cap: Capability) -> bool { | Capability::AppHttpRequest(_) | Capability::AppFilesWrite(_) | Capability::AppPubsubPublish(_) + | Capability::AppQueueEnqueue(_) | Capability::AppSecretsWrite(_) | Capability::AppEmailSend(_) | Capability::AppUsersWrite(_) diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index b80e976..00387d2 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -56,6 +56,7 @@ pub mod principal_resolver; pub mod pubsub_repo; pub mod pubsub_service; pub mod queue_repo; +pub mod queue_service; pub mod realtime_authority; pub mod repo; pub mod route_admin; diff --git a/crates/manager-core/src/queue_service.rs b/crates/manager-core/src/queue_service.rs new file mode 100644 index 0000000..c21e7f5 --- /dev/null +++ b/crates/manager-core/src/queue_service.rs @@ -0,0 +1,325 @@ +//! `QueueServiceImpl` — wires `QueueRepo` underneath the +//! `picloud_shared::QueueService` trait scripts see via the Rhai bridge. +//! +//! Mirrors the other stateful services: script-as-gate authz +//! (`AppQueueEnqueue`, skipped when `cx.principal` is `None`), with the +//! backend doing the actual write. No `ServiceEventEmitter` here — +//! enqueue is observable via `queue:receive` triggers (commit 6). + +use std::sync::Arc; + +use async_trait::async_trait; +use chrono::Utc; +use picloud_shared::{EnqueueOpts, QueueError, QueueMessageId, QueueService, SdkCallCx}; + +use crate::authz::{self, AuthzRepo, Capability}; +use crate::queue_repo::{NewQueueMessage, QueueRepo}; + +/// Production impl: authz gate → repo. Trivial wrapper. +pub struct QueueServiceImpl { + repo: Arc, + authz: Arc, +} + +impl QueueServiceImpl { + #[must_use] + pub fn new(repo: Arc, authz: Arc) -> Self { + Self { repo, authz } + } +} + +#[async_trait] +impl QueueService for QueueServiceImpl { + async fn enqueue( + &self, + cx: &SdkCallCx, + queue_name: &str, + payload: serde_json::Value, + opts: EnqueueOpts, + ) -> Result { + if queue_name.is_empty() { + return Err(QueueError::EmptyName); + } + let max_attempts = opts.max_attempts.unwrap_or(3); + if !(1..=20).contains(&max_attempts) { + return Err(QueueError::InvalidOpts( + "queue::enqueue: max_attempts must be in [1, 20]".into(), + )); + } + if let Some(delay) = opts.delay_ms { + // 24h cap on delay; matches what a cron tick would do anyway. + if !(0..=86_400_000).contains(&delay) { + return Err(QueueError::InvalidOpts( + "queue::enqueue: delay_ms must be in [0, 86_400_000]".into(), + )); + } + } + + // Script-as-gate authz: anonymous public-HTTP scripts skip the + // check (cx.principal is None); authenticated callers must hold + // AppQueueEnqueue. + if let Some(principal) = cx.principal.as_ref() { + authz::require( + &*self.authz, + principal, + Capability::AppQueueEnqueue(cx.app_id), + ) + .await + .map_err(|_| QueueError::Rejected("forbidden".into()))?; + } + + let deliver_after = opts + .delay_ms + .and_then(chrono::Duration::try_milliseconds) + .map(|d| Utc::now() + d); + + // Forensic only — never used for authz at consume-time (the + // queue:receive trigger runs as the registering principal per + // design notes §4). + let enqueued_by_principal = cx.principal.as_ref().map(|p| p.user_id); + + self.repo + .enqueue(NewQueueMessage { + app_id: cx.app_id, + queue_name: queue_name.to_string(), + payload, + deliver_after, + max_attempts, + enqueued_by_principal, + }) + .await + .map_err(|e| QueueError::Unavailable(e.to_string())) + } + + async fn depth(&self, cx: &SdkCallCx, queue_name: &str) -> Result { + if queue_name.is_empty() { + return Err(QueueError::EmptyName); + } + self.repo + .depth(cx.app_id, queue_name) + .await + .map_err(|e| QueueError::Unavailable(e.to_string())) + } + + async fn depth_pending(&self, cx: &SdkCallCx, queue_name: &str) -> Result { + if queue_name.is_empty() { + return Err(QueueError::EmptyName); + } + self.repo + .depth_pending(cx.app_id, queue_name) + .await + .map_err(|e| QueueError::Unavailable(e.to_string())) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::authz::AuthzError; + use picloud_shared::{AppId, AppRole, ExecutionId, RequestId, ScriptId}; + + struct AlwaysAllowAuthz; + #[async_trait] + impl AuthzRepo for AlwaysAllowAuthz { + async fn membership( + &self, + _user_id: picloud_shared::UserId, + _app_id: AppId, + ) -> Result, AuthzError> { + Ok(Some(AppRole::Editor)) + } + } + + struct CapturingRepo { + last: tokio::sync::Mutex>, + } + #[async_trait] + impl QueueRepo for CapturingRepo { + async fn enqueue( + &self, + msg: NewQueueMessage, + ) -> Result { + let id = QueueMessageId::new(); + *self.last.lock().await = Some(msg); + Ok(id) + } + async fn claim( + &self, + _app_id: AppId, + _queue_name: &str, + ) -> Result, crate::queue_repo::QueueRepoError> + { + Ok(None) + } + async fn ack( + &self, + _message_id: QueueMessageId, + _claim_token: uuid::Uuid, + ) -> Result { + Ok(true) + } + async fn nack( + &self, + _message_id: QueueMessageId, + _claim_token: uuid::Uuid, + _retry_delay: chrono::Duration, + ) -> Result { + Ok(true) + } + async fn reclaim_visibility_timeouts( + &self, + ) -> Result { + Ok(0) + } + async fn depth( + &self, + _app_id: AppId, + _queue_name: &str, + ) -> Result { + Ok(42) + } + async fn depth_pending( + &self, + _app_id: AppId, + _queue_name: &str, + ) -> Result { + Ok(7) + } + async fn list_for_app( + &self, + _app_id: AppId, + ) -> Result, crate::queue_repo::QueueRepoError> + { + Ok(vec![]) + } + async fn dead_letter( + &self, + _message_id: QueueMessageId, + _claim_token: uuid::Uuid, + _app_id: AppId, + _queue_name: &str, + _trigger_id: Option, + _script_id: Option, + _attempt: u32, + _first_attempt_at: chrono::DateTime, + _last_error: &str, + ) -> Result { + Ok(picloud_shared::DeadLetterId::new()) + } + } + + fn anon_cx() -> SdkCallCx { + let exec = ExecutionId::new(); + SdkCallCx { + app_id: AppId::new(), + script_id: ScriptId::new(), + principal: None, + execution_id: exec, + request_id: RequestId::new(), + trigger_depth: 0, + root_execution_id: exec, + is_dead_letter_handler: false, + event: None, + } + } + + #[tokio::test] + async fn empty_queue_name_rejects() { + let repo = Arc::new(CapturingRepo { + last: tokio::sync::Mutex::new(None), + }); + let svc = QueueServiceImpl::new(repo.clone(), Arc::new(AlwaysAllowAuthz)); + let cx = anon_cx(); + let err = svc + .enqueue(&cx, "", serde_json::Value::Null, EnqueueOpts::default()) + .await + .unwrap_err(); + assert!(matches!(err, QueueError::EmptyName)); + } + + #[tokio::test] + async fn invalid_max_attempts_rejects() { + let repo = Arc::new(CapturingRepo { + last: tokio::sync::Mutex::new(None), + }); + let svc = QueueServiceImpl::new(repo, Arc::new(AlwaysAllowAuthz)); + let cx = anon_cx(); + let err = svc + .enqueue( + &cx, + "x", + serde_json::Value::Null, + EnqueueOpts { + delay_ms: None, + max_attempts: Some(0), + }, + ) + .await + .unwrap_err(); + assert!(matches!(err, QueueError::InvalidOpts(_))); + let err = svc + .enqueue( + &cx, + "x", + serde_json::Value::Null, + EnqueueOpts { + delay_ms: None, + max_attempts: Some(21), + }, + ) + .await + .unwrap_err(); + assert!(matches!(err, QueueError::InvalidOpts(_))); + } + + #[tokio::test] + async fn invalid_delay_rejects() { + let repo = Arc::new(CapturingRepo { + last: tokio::sync::Mutex::new(None), + }); + let svc = QueueServiceImpl::new(repo, Arc::new(AlwaysAllowAuthz)); + let cx = anon_cx(); + let err = svc + .enqueue( + &cx, + "x", + serde_json::Value::Null, + EnqueueOpts { + delay_ms: Some(-1), + max_attempts: None, + }, + ) + .await + .unwrap_err(); + assert!(matches!(err, QueueError::InvalidOpts(_))); + } + + #[tokio::test] + async fn anon_principal_skips_authz_and_writes() { + let repo = Arc::new(CapturingRepo { + last: tokio::sync::Mutex::new(None), + }); + let svc = QueueServiceImpl::new(repo.clone(), Arc::new(AlwaysAllowAuthz)); + let cx = anon_cx(); + let payload = serde_json::json!({ "x": 1 }); + svc.enqueue(&cx, "jobs", payload.clone(), EnqueueOpts::default()) + .await + .unwrap(); + let captured = repo.last.lock().await.clone().unwrap(); + assert_eq!(captured.queue_name, "jobs"); + assert_eq!(captured.payload, payload); + assert_eq!(captured.max_attempts, 3); + assert!(captured.deliver_after.is_none()); + } + + #[tokio::test] + async fn depth_passes_through() { + let repo = Arc::new(CapturingRepo { + last: tokio::sync::Mutex::new(None), + }); + let svc = QueueServiceImpl::new(repo, Arc::new(AlwaysAllowAuthz)); + let cx = anon_cx(); + assert_eq!(svc.depth(&cx, "any").await.unwrap(), 42); + assert_eq!(svc.depth_pending(&cx, "any").await.unwrap(), 7); + } +} diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index f9312a2..74cbefd 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -261,6 +261,19 @@ pub async fn build_app( app_secrets_repo.clone(), users.clone(), )); + // v1.1.9 durable per-app queues. Producers write to queue_messages + // via QueueService; the dispatcher's queue arm (commit 6) consumes + // claimed messages and fires the registered queue:receive trigger. + let queue_repo: Arc = + Arc::new(picloud_manager_core::queue_repo::PostgresQueueRepo::new( + pool.clone(), + )); + let queue: Arc = Arc::new( + picloud_manager_core::queue_service::QueueServiceImpl::new( + queue_repo.clone(), + authz.clone(), + ), + ); let services = Services::new( kv, docs, @@ -273,6 +286,7 @@ pub async fn build_app( secrets, email, users.clone(), + queue, ); let engine = Arc::new(Engine::new(Limits::default(), services)); diff --git a/crates/shared/src/services.rs b/crates/shared/src/services.rs index 8e16c88..93f9e40 100644 --- a/crates/shared/src/services.rs +++ b/crates/shared/src/services.rs @@ -23,8 +23,8 @@ use crate::{ DeadLetterService, DocsService, EmailService, FilesService, HttpService, KvService, ModuleSource, NoopDeadLetterService, NoopDocsService, NoopEmailService, NoopEventEmitter, NoopFilesService, NoopHttpService, NoopKvService, NoopModuleSource, NoopPubsubService, - NoopSecretsService, NoopUsersService, PubsubService, SecretsService, ServiceEventEmitter, - UsersService, + NoopQueueService, NoopSecretsService, NoopUsersService, PubsubService, QueueService, + SecretsService, ServiceEventEmitter, UsersService, }; /// SDK service bundle. See module docs for the lifecycle and the v1.1.x @@ -95,6 +95,13 @@ pub struct Services { /// session tokens in the picloud binary; `NoopUsersService` in /// tests that don't exercise users. pub users: Arc, + + /// Durable per-app named queues (v1.1.9). Scripts get + /// `queue::{enqueue,depth,depth_pending}`. Backed by Postgres in + /// the picloud binary; `NoopQueueService` in tests that don't + /// touch queues. Consumers register via `queue:receive` triggers + /// (one per `(app_id, queue_name)`). + pub queue: Arc, } impl Services { @@ -115,6 +122,7 @@ impl Services { secrets: Arc, email: Arc, users: Arc, + queue: Arc, ) -> Self { Self { kv, @@ -128,6 +136,7 @@ impl Services { secrets, email, users, + queue, } } @@ -150,6 +159,7 @@ impl Services { Arc::new(NoopSecretsService), Arc::new(NoopEmailService), Arc::new(NoopUsersService), + Arc::new(NoopQueueService), ) } }