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

//! `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};

/// Default per-message JSON-encoded payload cap (256 KB). Override with
/// `PICLOUD_QUEUE_MAX_PAYLOAD_BYTES`.
pub const DEFAULT_QUEUE_MAX_PAYLOAD_BYTES: usize = 256 * 1024;

/// Read `PICLOUD_QUEUE_MAX_PAYLOAD_BYTES`; invalid values fall back to
/// the conservative default with a warning.
#[must_use]
pub fn queue_max_payload_bytes_from_env() -> usize {
    if let Ok(v) = std::env::var("PICLOUD_QUEUE_MAX_PAYLOAD_BYTES") {
        match v.trim().parse::<usize>() {
            Ok(n) if n > 0 => return n,
            _ => tracing::warn!(
                value = %v,
                "ignoring invalid PICLOUD_QUEUE_MAX_PAYLOAD_BYTES (want a positive integer)"
            ),
        }
    }
    DEFAULT_QUEUE_MAX_PAYLOAD_BYTES
}

/// Production impl: authz gate → repo. Trivial wrapper.
pub struct QueueServiceImpl {
    repo: Arc<dyn QueueRepo>,
    authz: Arc<dyn AuthzRepo>,
    max_payload_bytes: usize,
}

impl QueueServiceImpl {
    #[must_use]
    pub fn new(repo: Arc<dyn QueueRepo>, authz: Arc<dyn AuthzRepo>) -> Self {
        Self::with_max_payload_bytes(repo, authz, DEFAULT_QUEUE_MAX_PAYLOAD_BYTES)
    }

    #[must_use]
    pub fn with_max_payload_bytes(
        repo: Arc<dyn QueueRepo>,
        authz: Arc<dyn AuthzRepo>,
        max_payload_bytes: usize,
    ) -> Self {
        Self {
            repo,
            authz,
            max_payload_bytes,
        }
    }
}

#[async_trait]
impl QueueService for QueueServiceImpl {
    async fn enqueue(
        &self,
        cx: &SdkCallCx,
        queue_name: &str,
        payload: serde_json::Value,
        opts: EnqueueOpts,
    ) -> Result<QueueMessageId, QueueError> {
        if queue_name.is_empty() {
            return Err(QueueError::EmptyName);
        }
        // Reject oversized payloads BEFORE the authz check so an
        // anonymous public-script DoS doesn't pay the membership lookup.
        let encoded_len = serde_json::to_vec(&payload)
            .map(|v| v.len())
            .map_err(|e| QueueError::Rejected(format!("encode payload: {e}")))?;
        if encoded_len > self.max_payload_bytes {
            return Err(QueueError::PayloadTooLarge {
                limit: self.max_payload_bytes,
                actual: encoded_len,
            });
        }
        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.
        authz::script_gate(
            &*self.authz,
            cx,
            Capability::AppQueueEnqueue(cx.app_id),
            || QueueError::Forbidden,
            QueueError::Backend,
        )
        .await?;

        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::Backend(e.to_string()))
    }

    async fn depth(&self, cx: &SdkCallCx, queue_name: &str) -> Result<u64, QueueError> {
        if queue_name.is_empty() {
            return Err(QueueError::EmptyName);
        }
        self.repo
            .depth(cx.app_id, queue_name)
            .await
            .map_err(|e| QueueError::Backend(e.to_string()))
    }

    async fn depth_pending(&self, cx: &SdkCallCx, queue_name: &str) -> Result<u64, QueueError> {
        if queue_name.is_empty() {
            return Err(QueueError::EmptyName);
        }
        self.repo
            .depth_pending(cx.app_id, queue_name)
            .await
            .map_err(|e| QueueError::Backend(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<Option<AppRole>, AuthzError> {
            Ok(Some(AppRole::Editor))
        }
    }

    struct CapturingRepo {
        last: tokio::sync::Mutex<Option<NewQueueMessage>>,
    }
    #[async_trait]
    impl QueueRepo for CapturingRepo {
        async fn enqueue(
            &self,
            msg: NewQueueMessage,
        ) -> Result<QueueMessageId, crate::queue_repo::QueueRepoError> {
            let id = QueueMessageId::new();
            *self.last.lock().await = Some(msg);
            Ok(id)
        }
        async fn claim(
            &self,
            _app_id: AppId,
            _queue_name: &str,
        ) -> Result<Option<crate::queue_repo::ClaimedMessage>, crate::queue_repo::QueueRepoError>
        {
            Ok(None)
        }
        async fn ack(
            &self,
            _message_id: QueueMessageId,
            _claim_token: uuid::Uuid,
        ) -> Result<bool, crate::queue_repo::QueueRepoError> {
            Ok(true)
        }
        async fn nack(
            &self,
            _message_id: QueueMessageId,
            _claim_token: uuid::Uuid,
            _retry_delay: chrono::Duration,
        ) -> Result<bool, crate::queue_repo::QueueRepoError> {
            Ok(true)
        }
        async fn release(
            &self,
            _message_id: QueueMessageId,
            _claim_token: uuid::Uuid,
            _retry_delay: chrono::Duration,
        ) -> Result<bool, crate::queue_repo::QueueRepoError> {
            Ok(true)
        }
        async fn reclaim_visibility_timeouts(
            &self,
        ) -> Result<u64, crate::queue_repo::QueueRepoError> {
            Ok(0)
        }
        async fn depth(
            &self,
            _app_id: AppId,
            _queue_name: &str,
        ) -> Result<u64, crate::queue_repo::QueueRepoError> {
            Ok(42)
        }
        async fn depth_pending(
            &self,
            _app_id: AppId,
            _queue_name: &str,
        ) -> Result<u64, crate::queue_repo::QueueRepoError> {
            Ok(7)
        }
        async fn list_for_app(
            &self,
            _app_id: AppId,
        ) -> Result<Vec<(String, crate::queue_repo::QueueStats)>, 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<picloud_shared::TriggerId>,
            _script_id: Option<ScriptId>,
            _attempt: u32,
            _first_attempt_at: chrono::DateTime<chrono::Utc>,
            _last_error: &str,
        ) -> Result<picloud_shared::DeadLetterId, crate::queue_repo::QueueRepoError> {
            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,
        }
    }

    /// An AUTHENTICATED member with no app role. `DenyAllAuthz` denies it — unlike
    /// an anon cx, which passes `script_gate` regardless, so only an authenticated-
    /// denied cx can distinguish "size check ran before authz" from "after".
    fn member_cx() -> SdkCallCx {
        use picloud_shared::{InstanceRole, Principal, UserId};
        let exec = ExecutionId::new();
        SdkCallCx {
            app_id: AppId::new(),
            script_id: ScriptId::new(),
            principal: Some(Principal {
                user_id: UserId::new(),
                instance_role: InstanceRole::Member,
                scopes: None,
                app_binding: 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 oversized_payload_is_rejected_before_authz() {
        // DenyAllAuthz + an authenticated cx makes the ORDERING observable: a
        // size-first service returns PayloadTooLarge, an authz-first one would
        // return Forbidden. (The previous version used AlwaysAllow + anon, which
        // caught a DROPPED cap but not a REORDERED one.)
        let repo = Arc::new(CapturingRepo {
            last: tokio::sync::Mutex::new(None),
        });
        let svc = QueueServiceImpl::with_max_payload_bytes(repo, Arc::new(DenyAllAuthz), 16);
        let cx = member_cx();
        let payload = serde_json::json!({ "blob": "x".repeat(200) });
        let err = svc
            .enqueue(&cx, "jobs", payload, EnqueueOpts::default())
            .await
            .unwrap_err();
        assert!(
            matches!(err, QueueError::PayloadTooLarge { .. }),
            "an oversized payload must be PayloadTooLarge before authz; an authz-first \
             order would return Forbidden for this denied cx. got {err:?}"
        );
        // Control: an under-cap payload with the same cx is Forbidden — confirming
        // the cx is authz-denied, so the case above genuinely bypassed authz.
        let err = svc
            .enqueue(
                &cx,
                "jobs",
                serde_json::json!({ "ok": 1 }),
                EnqueueOpts::default(),
            )
            .await
            .unwrap_err();
        assert!(
            matches!(err, QueueError::Forbidden),
            "the control confirms this cx is authz-denied; got {err:?}"
        );
    }

    struct DenyAllAuthz;
    #[async_trait]
    impl AuthzRepo for DenyAllAuthz {
        async fn membership(
            &self,
            _user_id: picloud_shared::UserId,
            _app_id: AppId,
        ) -> Result<Option<AppRole>, AuthzError> {
            Ok(None)
        }
    }

    #[tokio::test]
    async fn authed_principal_denied_returns_forbidden_variant() {
        let repo = Arc::new(CapturingRepo {
            last: tokio::sync::Mutex::new(None),
        });
        let svc = QueueServiceImpl::new(repo, Arc::new(DenyAllAuthz));
        let err = svc
            .enqueue(
                &member_cx(),
                "x",
                serde_json::Value::Null,
                EnqueueOpts::default(),
            )
            .await
            .unwrap_err();
        assert!(matches!(err, QueueError::Forbidden));
    }

    #[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);
    }
}