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

//! `GroupPubsubServiceImpl` — wires the group-collection registry + the pubsub
//! fan-out repo underneath the `picloud_shared::GroupPubsubService` trait that
//! scripts reach via the `pubsub::shared_topic("name")` Rhai handle (§11.6 D2).
//!
//! A shared topic is storeless — a publish resolves the OWNING GROUP (the
//! nearest ancestor group declaring the topic shared, kind `topic`), then fans
//! out to that group's `shared = true` pubsub triggers, each delivery stamped
//! with the WRITER `cx.app_id` so the handler runs under the publishing app
//! (matching the M2 shared-collection trigger model). Owner resolution is the
//! isolation boundary — a foreign app's chain never reaches the owning group.
//!
//! Trust model: a publish is a WRITE, so it uses `script_gate_require_principal`
//! (anonymous fails closed — shared publish always needs an authenticated
//! editor+ on the owning group).

use std::sync::Arc;

use async_trait::async_trait;
use picloud_shared::{GroupId, GroupPubsubError, GroupPubsubService, SdkCallCx, TriggerEvent};

use crate::authz::{self, AuthzRepo, Capability};
use crate::group_collection_repo::GroupCollectionResolver;
use crate::pubsub_repo::{PublishCtx, PubsubRepo};
use crate::pubsub_service::pubsub_max_message_bytes_from_env;

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

pub struct GroupPubsubServiceImpl {
    repo: Arc<dyn PubsubRepo>,
    resolver: Arc<dyn GroupCollectionResolver>,
    authz: Arc<dyn AuthzRepo>,
    max_message_bytes: usize,
}

impl GroupPubsubServiceImpl {
    #[must_use]
    pub fn new(
        repo: Arc<dyn PubsubRepo>,
        resolver: Arc<dyn GroupCollectionResolver>,
        authz: Arc<dyn AuthzRepo>,
    ) -> Self {
        Self {
            repo,
            resolver,
            authz,
            max_message_bytes: pubsub_max_message_bytes_from_env(),
        }
    }

    /// The structural boundary: resolve the topic namespace to its owning group
    /// on the calling app's chain. `CollectionNotShared` when no ancestor group
    /// declares it — the same outcome a foreign app gets, by construction.
    async fn owning_group(
        &self,
        cx: &SdkCallCx,
        namespace: &str,
    ) -> Result<GroupId, GroupPubsubError> {
        self.resolver
            .resolve_owning_group(cx.app_id, namespace, KIND_TOPIC)
            .await
            .map_err(|e| GroupPubsubError::Backend(e.to_string()))?
            .ok_or_else(|| GroupPubsubError::CollectionNotShared(namespace.to_string()))
    }

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

#[async_trait]
impl GroupPubsubService for GroupPubsubServiceImpl {
    async fn publish(
        &self,
        cx: &SdkCallCx,
        namespace: &str,
        subtopic: &str,
        message: serde_json::Value,
    ) -> Result<u32, GroupPubsubError> {
        if namespace.trim().is_empty() {
            return Err(GroupPubsubError::InvalidTopic);
        }
        // Size cap first (no DB) — a cheap reject before the resolve/authz work.
        let encoded_len = serde_json::to_vec(&message)
            .map(|v| v.len())
            .map_err(|e| GroupPubsubError::Backend(format!("encode message: {e}")))?;
        if encoded_len > self.max_message_bytes {
            return Err(GroupPubsubError::MessageTooLarge {
                limit: self.max_message_bytes,
                actual: encoded_len,
            });
        }
        let group_id = self.owning_group(cx, namespace).await?;
        self.check_write(cx, group_id).await?;

        // The full topic is `namespace` or `namespace.subtopic`; shared triggers
        // match it with the same `topic_matches` semantics as the per-app path.
        let topic = if subtopic.trim().is_empty() {
            namespace.to_string()
        } else {
            format!("{namespace}.{subtopic}")
        };
        let event = TriggerEvent::Pubsub {
            topic: topic.clone(),
            message,
            published_at: chrono::Utc::now(),
        };
        let payload = serde_json::to_value(&event)
            .map_err(|e| GroupPubsubError::Backend(format!("event serialize: {e}")))?;
        let publish_ctx = PublishCtx {
            app_id: cx.app_id,
            origin_principal: cx.principal.as_ref().map(|p| p.user_id),
            trigger_depth: cx.trigger_depth,
            root_execution_id: cx.root_execution_id,
        };
        self.repo
            .fan_out_shared_publish(publish_ctx, group_id, &topic, payload)
            .await
            .map_err(|e| GroupPubsubError::Backend(e.to_string()))
    }
}