//! 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 serde_json::json; 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 { // Match the JSON envelope shape used by every other admin api // file (`{"error": "..."}`); previously this returned a // plain-text body inconsistent with siblings. let (status, body) = match &self { Self::Forbidden => ( axum::http::StatusCode::FORBIDDEN, json!({ "error": self.to_string() }), ), Self::AppNotFound => ( axum::http::StatusCode::NOT_FOUND, json!({ "error": self.to_string() }), ), Self::Repo(e) => { tracing::error!(error = %e, "queues admin backend error"); ( axum::http::StatusCode::INTERNAL_SERVER_ERROR, json!({ "error": "internal error" }), ) } }; (status, Json(body)).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(id_or_slug): Path, ) -> Result>, QueuesApiError> { let app_id = resolve_app(&*s.apps, &id_or_slug).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((id_or_slug, queue_name)): Path<(String, String)>, ) -> Result, QueuesApiError> { let app_id = resolve_app(&*s.apps, &id_or_slug).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 resolve_app(apps: &dyn AppRepository, ident: &str) -> Result { crate::app_repo::resolve_app(apps, ident) .await .map_err(|e| QueuesApiError::Repo(e.to_string()))? .map(|l| l.app.id) .ok_or(QueuesApiError::AppNotFound) } 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(()) } // In-process unit tests for queues_api are intentionally not added. // Constructing a `QueuesState` requires mocking five traits // (`AppRepository`, `QueueRepo`, `TriggerRepo`, `ScriptRepository`, // `AuthzRepo`) totalling 30+ methods just to exercise the resolver // path. Integration tests in `crates/picloud/tests/` and the // dashboard's `tests/e2e/navigation/tabs.spec.ts` already cover both // the UUID and slug entry points end-to-end. The same applies to // `files_api.rs`, `secrets_api.rs`, and `dead_letters_api.rs` — all // three have zero in-process tests and would each need a similarly // wide mock surface. Adding a shared mock-repo crate is the right // long-term fix.