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

//! `GroupQueueServiceImpl` — wires `GroupQueueRepo` + the group-collection
//! registry underneath the `picloud_shared::GroupQueueService` trait that
//! scripts reach via the `queue::shared_collection("name")` Rhai handle
//! (§11.6 D3).
//!
//! Resolves the OWNING GROUP (kind `queue`) from `cx.app_id`'s chain (the
//! isolation boundary), enforces editor+ (`GroupQueueEnqueue`, fails closed on
//! anon — enqueue is a shared write), caps the payload size, then writes to the
//! group-keyed store. Consumption is out of band (competing per-descendant
//! materialized consumers, in the dispatcher).

use std::sync::Arc;

use async_trait::async_trait;
use chrono::Utc;
use picloud_shared::{
    EnqueueOpts, GroupId, GroupQueueError, GroupQueueService, QueueMessageId, SdkCallCx,
};

use crate::authz::{self, AuthzRepo, Capability};
use crate::group_collection_repo::GroupCollectionResolver;
use crate::group_queue_repo::{GroupQueueRepo, NewGroupQueueMessage};
use crate::queue_service::queue_max_payload_bytes_from_env;

/// The registry `kind` this service resolves.
const KIND_QUEUE: &str = "queue";

pub struct GroupQueueServiceImpl {
    repo: Arc<dyn GroupQueueRepo>,
    resolver: Arc<dyn GroupCollectionResolver>,
    authz: Arc<dyn AuthzRepo>,
    max_payload_bytes: usize,
}

impl GroupQueueServiceImpl {
    #[must_use]
    pub fn new(
        repo: Arc<dyn GroupQueueRepo>,
        resolver: Arc<dyn GroupCollectionResolver>,
        authz: Arc<dyn AuthzRepo>,
    ) -> Self {
        Self {
            repo,
            resolver,
            authz,
            max_payload_bytes: queue_max_payload_bytes_from_env(),
        }
    }

    async fn owning_group(
        &self,
        cx: &SdkCallCx,
        collection: &str,
    ) -> Result<GroupId, GroupQueueError> {
        if collection.is_empty() {
            return Err(GroupQueueError::InvalidCollection);
        }
        self.resolver
            .resolve_owning_group(cx.app_id, collection, KIND_QUEUE)
            .await
            .map_err(|e| GroupQueueError::Backend(e.to_string()))?
            .ok_or_else(|| GroupQueueError::CollectionNotShared(collection.to_string()))
    }

    async fn check_write(&self, cx: &SdkCallCx, group_id: GroupId) -> Result<(), GroupQueueError> {
        authz::script_gate_require_principal(
            &*self.authz,
            cx,
            Capability::GroupQueueEnqueue(group_id),
            || GroupQueueError::Forbidden,
            GroupQueueError::Backend,
        )
        .await
    }
}

#[async_trait]
impl GroupQueueService for GroupQueueServiceImpl {
    async fn enqueue(
        &self,
        cx: &SdkCallCx,
        collection: &str,
        payload: serde_json::Value,
        opts: EnqueueOpts,
    ) -> Result<QueueMessageId, GroupQueueError> {
        if collection.is_empty() {
            return Err(GroupQueueError::InvalidCollection);
        }
        // Size cap first (no DB) — a cheap reject before resolve/authz.
        let encoded_len = serde_json::to_vec(&payload)
            .map(|v| v.len())
            .map_err(|e| GroupQueueError::Backend(format!("encode payload: {e}")))?;
        if encoded_len > self.max_payload_bytes {
            return Err(GroupQueueError::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(GroupQueueError::InvalidOpts(
                "queue::enqueue: max_attempts must be in [1, 20]".into(),
            ));
        }
        if let Some(delay) = opts.delay_ms {
            if !(0..=86_400_000).contains(&delay) {
                return Err(GroupQueueError::InvalidOpts(
                    "queue::enqueue: delay_ms must be in [0, 86_400_000]".into(),
                ));
            }
        }
        let group_id = self.owning_group(cx, collection).await?;
        self.check_write(cx, group_id).await?;

        let deliver_after = opts
            .delay_ms
            .and_then(chrono::Duration::try_milliseconds)
            .map(|d| Utc::now() + d);
        self.repo
            .enqueue(NewGroupQueueMessage {
                group_id,
                collection: collection.to_string(),
                payload,
                deliver_after,
                max_attempts,
                enqueued_by_principal: cx.principal.as_ref().map(|p| p.user_id),
            })
            .await
            .map_err(|e| GroupQueueError::Backend(e.to_string()))
    }

    async fn depth(&self, cx: &SdkCallCx, collection: &str) -> Result<u64, GroupQueueError> {
        let group_id = self.owning_group(cx, collection).await?;
        // Depth is a read — reads are open (any subtree script).
        self.repo
            .depth(group_id, collection)
            .await
            .map_err(|e| GroupQueueError::Backend(e.to_string()))
    }

    async fn depth_pending(
        &self,
        cx: &SdkCallCx,
        collection: &str,
    ) -> Result<u64, GroupQueueError> {
        let group_id = self.owning_group(cx, collection).await?;
        self.repo
            .depth_pending(group_id, collection)
            .await
            .map_err(|e| GroupQueueError::Backend(e.to_string()))
    }
}