From 714f6dae18e83abf2c1b1b5418b92a7a912da6d7 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Sat, 6 Jun 2026 22:36:50 +0200 Subject: [PATCH] =?UTF-8?q?feat(v1.1.9):=20admin=20HTTP=20endpoints=20?= =?UTF-8?q?=E2=80=94=20queue=20triggers=20+=20read-only=20queues?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit triggers_api.rs adds: POST /api/v1/admin/apps/{id}/triggers/queue body: { script_id, queue_name, visibility_timeout_secs?, dispatch_mode?, retry_max_attempts?, retry_backoff?, retry_base_ms? } cap: AppManageTriggers 409-equiv via the repo's "queue 'X' already has a consumer" error (TriggerRepoError::Invalid → 422 Invalid in the existing mapping) queues_api.rs (new) adds read-only inspection endpoints: GET /api/v1/admin/apps/{id}/queues -> [{queue_name, total, pending, claimed}] GET /api/v1/admin/apps/{id}/queues/{queue_name} -> {queue_name, total, pending, claimed, consumer: Some(QueueConsumer) | None} QueueConsumer = { trigger_id, script_id, script_name, visibility_timeout_secs, last_fired_at } Capability: AppLogRead (same trust tier as execution-log read — these are read-only operational views, not config changes). No mutating queue endpoints — enqueue lives on the SDK side; purge / requeue / delete-message stays at v1.2. picloud/lib.rs wires the queues router into the guarded /admin chain. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/manager-core/src/lib.rs | 1 + crates/manager-core/src/queues_api.rs | 205 ++++++++++++++++++++++++ crates/manager-core/src/triggers_api.rs | 60 +++++++ crates/picloud/src/lib.rs | 13 +- 4 files changed, 278 insertions(+), 1 deletion(-) create mode 100644 crates/manager-core/src/queues_api.rs diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index 5373c3c..8b47a28 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -58,6 +58,7 @@ pub mod pubsub_service; pub mod invoke_service; pub mod queue_repo; pub mod queue_service; +pub mod queues_api; pub mod realtime_authority; pub mod repo; pub mod route_admin; diff --git a/crates/manager-core/src/queues_api.rs b/crates/manager-core/src/queues_api.rs new file mode 100644 index 0000000..53249e1 --- /dev/null +++ b/crates/manager-core/src/queues_api.rs @@ -0,0 +1,205 @@ +//! v1.1.9. Read-only admin endpoints for queues. +//! +//! - `GET /apps/{app_id}/queues` — list all queue names in the app +//! with `(total, pending, claimed)` counts. +//! - `GET /apps/{app_id}/queues/{queue_name}` — per-queue drill-down: +//! stats + the registered consumer trigger (if any). +//! +//! No mutating endpoints — purge / requeue is v1.2. +//! Capability: `AppLogRead` (read-only inspection of operational state, +//! same tier as execution logs). + +use std::sync::Arc; + +use axum::{ + extract::{Path, State}, + routing::get, + Extension, Json, Router, +}; +use picloud_shared::{AppId, Principal}; +use serde::Serialize; + +use crate::app_repo::AppRepository; +use crate::authz::{self, AuthzRepo, Capability}; +use crate::queue_repo::{QueueRepo, QueueStats}; +use crate::repo::ScriptRepository; +use crate::trigger_repo::TriggerRepo; + +#[derive(Clone)] +pub struct QueuesState { + pub queues: Arc, + pub triggers: Arc, + pub apps: Arc, + pub authz: Arc, + pub scripts: Arc, +} + +#[derive(Debug, thiserror::Error)] +pub enum QueuesApiError { + #[error("forbidden")] + Forbidden, + #[error("app not found")] + AppNotFound, + #[error("repo: {0}")] + Repo(String), +} + +impl axum::response::IntoResponse for QueuesApiError { + fn into_response(self) -> axum::response::Response { + let status = match self { + Self::Forbidden => axum::http::StatusCode::FORBIDDEN, + Self::AppNotFound => axum::http::StatusCode::NOT_FOUND, + Self::Repo(_) => axum::http::StatusCode::INTERNAL_SERVER_ERROR, + }; + (status, self.to_string()).into_response() + } +} + +pub fn queues_router(state: QueuesState) -> Router { + Router::new() + .route("/apps/{app_id}/queues", get(list_queues)) + .route("/apps/{app_id}/queues/{queue_name}", get(get_queue)) + .with_state(state) +} + +#[derive(Debug, Serialize)] +pub struct QueueSummary { + pub queue_name: String, + pub total: u64, + pub pending: u64, + pub claimed: u64, +} + +#[derive(Debug, Serialize)] +pub struct QueueDetail { + pub queue_name: String, + pub total: u64, + pub pending: u64, + pub claimed: u64, + /// `Some` if a queue:receive trigger is registered for this queue. + pub consumer: Option, +} + +#[derive(Debug, Serialize)] +pub struct QueueConsumer { + pub trigger_id: picloud_shared::TriggerId, + pub script_id: picloud_shared::ScriptId, + pub script_name: String, + pub visibility_timeout_secs: u32, + pub last_fired_at: Option>, +} + +async fn list_queues( + State(s): State, + Extension(principal): Extension, + Path(app_id): Path, +) -> Result>, QueuesApiError> { + ensure_app(&*s.apps, app_id).await?; + require_log_read(&*s.authz, &principal, app_id).await?; + let rows = s + .queues + .list_for_app(app_id) + .await + .map_err(|e| QueuesApiError::Repo(e.to_string()))?; + Ok(Json( + rows.into_iter() + .map(|(name, stats)| QueueSummary { + queue_name: name, + total: stats.total, + pending: stats.pending, + claimed: stats.claimed, + }) + .collect(), + )) +} + +async fn get_queue( + State(s): State, + Extension(principal): Extension, + Path((app_id, queue_name)): Path<(AppId, String)>, +) -> Result, QueuesApiError> { + ensure_app(&*s.apps, app_id).await?; + require_log_read(&*s.authz, &principal, app_id).await?; + + let total = s + .queues + .depth(app_id, &queue_name) + .await + .map_err(|e| QueuesApiError::Repo(e.to_string()))?; + let pending = s + .queues + .depth_pending(app_id, &queue_name) + .await + .map_err(|e| QueuesApiError::Repo(e.to_string()))?; + // Claimed = total - pending - (delayed-not-yet-due). The delayed + // portion can't be teased apart without an extra query; we report + // (total - pending) as a useful approximation. (For v1.2 we can + // surface a third count via list_for_app's QueueStats which has it.) + let claimed = total.saturating_sub(pending); + + // Find a registered consumer for this queue. + let consumers = s + .triggers + .list_active_queue_consumers() + .await + .map_err(|e| QueuesApiError::Repo(e.to_string()))?; + let consumer_record = consumers + .into_iter() + .find(|c| c.app_id == app_id && c.queue_name == queue_name); + let consumer = if let Some(c) = consumer_record { + let script = s + .scripts + .get(c.script_id) + .await + .map_err(|e| QueuesApiError::Repo(e.to_string()))?; + let script_name = script.map_or_else(|| "".to_string(), |s| s.name); + // Pull last_fired_at from the trigger row (it's on the detail + // table — exposed via `Trigger.details` / TriggerDetails::Queue). + let trigger = s + .triggers + .get(c.trigger_id) + .await + .map_err(|e| QueuesApiError::Repo(e.to_string()))?; + let last_fired_at = match trigger.map(|t| t.details) { + Some(crate::trigger_repo::TriggerDetails::Queue { last_fired_at, .. }) => last_fired_at, + _ => None, + }; + Some(QueueConsumer { + trigger_id: c.trigger_id, + script_id: c.script_id, + script_name, + visibility_timeout_secs: c.visibility_timeout_secs, + last_fired_at, + }) + } else { + None + }; + + let _ = QueueStats::default(); + Ok(Json(QueueDetail { + queue_name, + total, + pending, + claimed, + consumer, + })) +} + +async fn ensure_app(apps: &dyn AppRepository, app_id: AppId) -> Result<(), QueuesApiError> { + apps.get_by_id(app_id) + .await + .map_err(|e| QueuesApiError::Repo(e.to_string()))? + .ok_or(QueuesApiError::AppNotFound)?; + Ok(()) +} + +async fn require_log_read( + authz: &dyn AuthzRepo, + principal: &Principal, + app_id: AppId, +) -> Result<(), QueuesApiError> { + authz::require(authz, principal, Capability::AppLogRead(app_id)) + .await + .map_err(|_| QueuesApiError::Forbidden)?; + Ok(()) +} diff --git a/crates/manager-core/src/triggers_api.rs b/crates/manager-core/src/triggers_api.rs index 2d2326a..a56bd5b 100644 --- a/crates/manager-core/src/triggers_api.rs +++ b/crates/manager-core/src/triggers_api.rs @@ -72,6 +72,7 @@ pub fn triggers_router(state: TriggersState) -> Router { post(create_dl_trigger), ) .route("/apps/{app_id}/triggers/email", post(create_email_trigger)) + .route("/apps/{app_id}/triggers/queue", post(create_queue_trigger)) .route( "/apps/{app_id}/triggers/{trigger_id}", delete(delete_trigger), @@ -376,6 +377,26 @@ pub struct CreatePubsubTriggerRequest { pub retry_base_ms: Option, } +/// v1.1.9. Inbound shape for a queue:receive trigger create. +/// `visibility_timeout_secs` defaults to TriggerConfig's per-queue +/// default if omitted; retry fields default to the standard trigger +/// retry policy. +#[derive(Debug, Deserialize, Serialize)] +pub struct CreateQueueTriggerRequest { + pub script_id: ScriptId, + pub queue_name: String, + #[serde(default)] + pub visibility_timeout_secs: Option, + #[serde(default = "default_dispatch")] + pub dispatch_mode: TriggerDispatchMode, + #[serde(default)] + pub retry_max_attempts: Option, + #[serde(default)] + pub retry_backoff: Option, + #[serde(default)] + pub retry_base_ms: Option, +} + async fn create_pubsub_trigger( State(s): State, Extension(principal): Extension, @@ -527,6 +548,45 @@ async fn create_email_trigger( Ok((StatusCode::CREATED, Json(created))) } +async fn create_queue_trigger( + State(s): State, + Extension(principal): Extension, + Path(app_id): Path, + Json(input): Json, +) -> Result<(StatusCode, Json), TriggersApiError> { + ensure_app_exists(&*s.apps, app_id).await?; + require( + s.authz.as_ref(), + &principal, + Capability::AppManageTriggers(app_id), + ) + .await?; + + if input.queue_name.trim().is_empty() { + return Err(TriggersApiError::Invalid( + "queue_name must not be empty".into(), + )); + } + validate_trigger_target(&*s.scripts, app_id, input.script_id).await?; + + let req = crate::trigger_repo::CreateQueueTrigger { + script_id: input.script_id, + queue_name: input.queue_name, + visibility_timeout_secs: input + .visibility_timeout_secs + .unwrap_or(s.config.queue_default_visibility_timeout_secs), + dispatch_mode: input.dispatch_mode, + retry_max_attempts: input + .retry_max_attempts + .unwrap_or(s.config.retry_max_attempts), + retry_backoff: input.retry_backoff.unwrap_or(s.config.retry_backoff), + retry_base_ms: input.retry_base_ms.unwrap_or(s.config.retry_base_ms), + registered_by_principal: principal.user_id, + }; + let created = s.triggers.create_queue_trigger(app_id, req).await?; + Ok((StatusCode::CREATED, Json(created))) +} + async fn delete_trigger( State(s): State, Extension(principal): Extension, diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index 162b74f..f07594b 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -432,6 +432,8 @@ pub async fn build_app( config: trigger_config, master_key: master_key.clone(), }; + // v1.1.9: keep a clone for the queues-api state (built later). + let trigger_repo_for_queues = trigger_repo.clone(); // v1.1.7 public inbound-email receiver. Outside the admin auth layer // (the URL + per-trigger HMAC secret are the security boundary). let email_inbound_state = EmailInboundState { @@ -464,7 +466,7 @@ pub async fn build_app( max_value_bytes: secrets_max_value_bytes, }; let apps_state = AppsState { - apps: apps_repo, + apps: apps_repo.clone(), domains: domains_repo, routes: route_repo, domain_table: app_domain_table.clone(), @@ -513,6 +515,15 @@ pub async fn build_app( )) .merge(api_keys_router(api_keys_state)) .merge(triggers_router(triggers_state)) + .merge(picloud_manager_core::queues_api::queues_router( + picloud_manager_core::queues_api::QueuesState { + queues: queue_repo.clone(), + triggers: trigger_repo_for_queues.clone(), + apps: apps_repo.clone(), + authz: authz.clone(), + scripts: Arc::new(PostgresScriptRepoHandle(script_repo.clone())), + }, + )) .merge(files_admin_router(files_admin_state)) .merge(topics_router(topics_state)) .merge(secrets_router(secrets_state))