feat(v1.1.9): admin HTTP endpoints — queue triggers + read-only queues

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) <noreply@anthropic.com>

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;

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<dyn QueueRepo>,
+    pub triggers: Arc<dyn TriggerRepo>,
+    pub apps: Arc<dyn AppRepository>,
+    pub authz: Arc<dyn AuthzRepo>,
+    pub scripts: Arc<dyn ScriptRepository>,
+}
+
+#[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<QueueConsumer>,
+}
+
+#[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<chrono::DateTime<chrono::Utc>>,
+}
+
+async fn list_queues(
+    State(s): State<QueuesState>,
+    Extension(principal): Extension<Principal>,
+    Path(app_id): Path<AppId>,
+) -> Result<Json<Vec<QueueSummary>>, 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<QueuesState>,
+    Extension(principal): Extension<Principal>,
+    Path((app_id, queue_name)): Path<(AppId, String)>,
+) -> Result<Json<QueueDetail>, 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(|| "<missing>".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(())
+}

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<u32>,
 }
 
+/// 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<u32>,
+    #[serde(default = "default_dispatch")]
+    pub dispatch_mode: TriggerDispatchMode,
+    #[serde(default)]
+    pub retry_max_attempts: Option<u32>,
+    #[serde(default)]
+    pub retry_backoff: Option<BackoffShape>,
+    #[serde(default)]
+    pub retry_base_ms: Option<u32>,
+}
+
 async fn create_pubsub_trigger(
     State(s): State<TriggersState>,
     Extension(principal): Extension<Principal>,
@@ -527,6 +548,45 @@ async fn create_email_trigger(
     Ok((StatusCode::CREATED, Json(created)))
 }
 
+async fn create_queue_trigger(
+    State(s): State<TriggersState>,
+    Extension(principal): Extension<Principal>,
+    Path(app_id): Path<AppId>,
+    Json(input): Json<CreateQueueTriggerRequest>,
+) -> Result<(StatusCode, Json<Trigger>), 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<TriggersState>,
     Extension(principal): Extension<Principal>,

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))