PiCloud/crates/manager-core/src/queues_api.rs

//! 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<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 {
        // 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<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(id_or_slug): Path<String>,
) -> Result<Json<Vec<QueueSummary>>, 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<QueuesState>,
    Extension(principal): Extension<Principal>,
    Path((id_or_slug, queue_name)): Path<(String, String)>,
) -> Result<Json<QueueDetail>, 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(|| "<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 resolve_app(apps: &dyn AppRepository, ident: &str) -> Result<AppId, QueuesApiError> {
    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.