From 9626d15863db21d72ce28c313ce1f7915f8a888d Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Thu, 2 Jul 2026 21:48:59 +0200 Subject: [PATCH] feat(shared-topics): GroupPubsubService + shared publish fan-out + SDK handle (D2) The runtime for shared topics: - pubsub_repo: fan_out_publish gains `AND t.shared = FALSE` (a shared trigger never fires on a per-app publish); new fan_out_shared_publish matches `shared = true` pubsub triggers on the resolved owning group, each delivery stamped with the writer app_id (M2 model). - GroupPubsubService trait (shared) + GroupPubsubServiceImpl: resolve the owning group (kind='topic') from cx.app_id's chain, require editor+ (GroupPubsubPublish, fails closed for anon), size-cap, fan out. - SDK: `pubsub::shared_topic("name")` -> GroupTopicHandle with `.publish(subtopic, msg)` / `.publish(msg)`; wired through Services + the picloud binary (reuses the one PubsubRepo). - authz: Capability::GroupPubsubPublish(GroupId), editor+ / script:write. The owning-group chain walk is the isolation boundary; a sibling-subtree app never resolves the topic. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/executor-core/src/sdk/pubsub.rs | 64 +++++++++ crates/manager-core/src/authz.rs | 14 +- .../manager-core/src/group_pubsub_service.rs | 128 ++++++++++++++++++ crates/manager-core/src/lib.rs | 2 + crates/manager-core/src/pubsub_repo.rs | 99 +++++++++++--- crates/manager-core/src/pubsub_service.rs | 11 ++ crates/picloud/src/lib.rs | 43 +++--- crates/shared/src/group_pubsub.rs | 84 ++++++++++++ crates/shared/src/lib.rs | 2 + crates/shared/src/services.rs | 26 +++- 10 files changed, 430 insertions(+), 43 deletions(-) create mode 100644 crates/manager-core/src/group_pubsub_service.rs create mode 100644 crates/shared/src/group_pubsub.rs diff --git a/crates/executor-core/src/sdk/pubsub.rs b/crates/executor-core/src/sdk/pubsub.rs index 9458db5..0b56cdb 100644 --- a/crates/executor-core/src/sdk/pubsub.rs +++ b/crates/executor-core/src/sdk/pubsub.rs @@ -69,7 +69,71 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc Result> { + if name.is_empty() { + return Err("pubsub::shared_topic name must not be empty".into()); + } + Ok(GroupTopicHandle { + namespace: name.to_string(), + service: group_svc.clone(), + cx: cx.clone(), + }) + }, + ); + } engine.register_static_module("pubsub", module.into()); + engine.register_type_with_name::("GroupTopicHandle"); + register_shared_publish(engine); +} + +/// §11.6 D2 handle returned by `pubsub::shared_topic("name")`. `.publish(sub, +/// msg)` publishes `name.sub` to the group shared topic; `.publish(msg)` +/// publishes to `name` directly. +#[derive(Clone)] +pub struct GroupTopicHandle { + namespace: String, + service: Arc, + cx: Arc, +} + +fn shared_publish( + h: &mut GroupTopicHandle, + subtopic: &str, + message: Dynamic, +) -> Result<(), Box> { + let json = message_to_json(&message); + let service = h.service.clone(); + let cx = h.cx.clone(); + let namespace = h.namespace.clone(); + let subtopic = subtopic.to_string(); + block_on("pubsub", async move { + service.publish(&cx, &namespace, &subtopic, json).await + }) + // The fan-out count isn't surfaced to scripts (fire-and-forget, like the + // per-app publish). + .map(|_n: u32| ()) +} + +fn register_shared_publish(engine: &mut RhaiEngine) { + engine.register_fn( + "publish", + |h: &mut GroupTopicHandle, subtopic: &str, message: Dynamic| { + shared_publish(h, subtopic, message) + }, + ); + // `.publish(msg)` with no subtopic → publish to the namespace directly. + engine.register_fn("publish", |h: &mut GroupTopicHandle, message: Dynamic| { + shared_publish(h, "", message) + }); } /// Interpret the optional `ttl` argument: `()` → use the default, diff --git a/crates/manager-core/src/authz.rs b/crates/manager-core/src/authz.rs index d344ce1..52fca83 100644 --- a/crates/manager-core/src/authz.rs +++ b/crates/manager-core/src/authz.rs @@ -166,6 +166,10 @@ pub enum Capability { /// Write a group-owned shared FILES collection (§11.6). editor+ on the owning /// group; fails closed for an anonymous principal, like `GroupKvWrite`. GroupFilesWrite(GroupId), + /// Publish to a group-owned shared TOPIC (§11.6 D2). editor+ on the owning + /// group; fails closed for an anonymous principal, like `GroupKvWrite` — a + /// publish is a write to shared state. + GroupPubsubPublish(GroupId), /// Send an outbound email from a script in this app (v1.1.7). Maps /// to `script:write` on API keys (sending mail is an outbound /// side-effect like an HTTP request). Granted to `editor`+. @@ -238,7 +242,8 @@ impl Capability { | Self::GroupDocsRead(_) | Self::GroupDocsWrite(_) | Self::GroupFilesRead(_) - | Self::GroupFilesWrite(_) => None, + | Self::GroupFilesWrite(_) + | Self::GroupPubsubPublish(_) => None, Self::AppRead(id) | Self::AppWriteScript(id) | Self::AppWriteRoute(id) @@ -315,6 +320,7 @@ impl Capability { | Self::GroupKvWrite(_) | Self::GroupDocsWrite(_) | Self::GroupFilesWrite(_) + | Self::GroupPubsubPublish(_) | Self::AppInvoke(_) => Scope::ScriptWrite, Self::AppWriteRoute(_) => Scope::RouteWrite, Self::AppManageDomains(_) => Scope::DomainManage, @@ -536,7 +542,8 @@ async fn role_grants( | Capability::GroupDocsRead(g) | Capability::GroupDocsWrite(g) | Capability::GroupFilesRead(g) - | Capability::GroupFilesWrite(g) => { + | Capability::GroupFilesWrite(g) + | Capability::GroupPubsubPublish(g) => { group_member_grants(repo, principal.user_id, cap, g).await } // Creating a root-level group is an instance act — members @@ -610,7 +617,8 @@ const fn group_role_satisfies(role: AppRole, cap: Capability) -> bool { | Capability::GroupScriptsWrite(_) | Capability::GroupKvWrite(_) | Capability::GroupDocsWrite(_) - | Capability::GroupFilesWrite(_) => { + | Capability::GroupFilesWrite(_) + | Capability::GroupPubsubPublish(_) => { matches!(role, AppRole::Editor | AppRole::AppAdmin) } // group_admin manages the group + reads secret VALUES (the diff --git a/crates/manager-core/src/group_pubsub_service.rs b/crates/manager-core/src/group_pubsub_service.rs new file mode 100644 index 0000000..affafd4 --- /dev/null +++ b/crates/manager-core/src/group_pubsub_service.rs @@ -0,0 +1,128 @@ +//! `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, + resolver: Arc, + authz: Arc, + max_message_bytes: usize, +} + +impl GroupPubsubServiceImpl { + #[must_use] + pub fn new( + repo: Arc, + resolver: Arc, + authz: Arc, + ) -> 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 { + 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 { + 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())) + } +} diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index 763eae0..c2bea20 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -59,6 +59,7 @@ pub mod group_files_service; pub mod group_kv_repo; pub mod group_kv_service; pub mod group_members_repo; +pub mod group_pubsub_service; pub mod group_quota; pub mod group_repo; pub mod group_scripts_api; @@ -198,6 +199,7 @@ pub use group_members_repo::{ GroupMembersRepository, GroupMembersRepositoryError, GroupMembershipDetail, GroupMembershipRow, PostgresGroupMembersRepository, }; +pub use group_pubsub_service::GroupPubsubServiceImpl; pub use group_repo::{ GroupChildCounts, GroupRepository, GroupRepositoryError, PostgresGroupRepository, ROOT_GROUP_SLUG, diff --git a/crates/manager-core/src/pubsub_repo.rs b/crates/manager-core/src/pubsub_repo.rs index 6a589dd..6b20953 100644 --- a/crates/manager-core/src/pubsub_repo.rs +++ b/crates/manager-core/src/pubsub_repo.rs @@ -12,7 +12,7 @@ //! becomes a hot path. use async_trait::async_trait; -use picloud_shared::{topic_matches, AdminUserId, AppId, ExecutionId}; +use picloud_shared::{topic_matches, AdminUserId, AppId, ExecutionId, GroupId}; use sqlx::PgPool; use uuid::Uuid; @@ -47,6 +47,20 @@ pub trait PubsubRepo: Send + Sync { topic: &str, event_payload: serde_json::Value, ) -> Result; + + /// §11.6 D2: fan out a SHARED-topic publish to every `shared = true` pubsub + /// trigger on `owning_group` (the group resolved from the shared-topic + /// declaration), inserting one outbox row each stamped with the WRITER + /// `ctx.app_id` — so the handler runs under the publishing app, matching the + /// M2 shared-collection trigger model. No chain union (the owning group is + /// already resolved) and no suppression (shared triggers aren't suppressible). + async fn fan_out_shared_publish( + &self, + ctx: PublishCtx, + owning_group: GroupId, + topic: &str, + event_payload: serde_json::Value, + ) -> Result; } pub struct PostgresPubsubRepo { @@ -92,7 +106,7 @@ impl PubsubRepo for PostgresPubsubRepo { FROM triggers t \ JOIN pubsub_trigger_details d ON d.trigger_id = t.id \ JOIN chain c ON (t.app_id = c.app_owner OR t.group_id = c.group_owner) \ - WHERE t.kind = 'pubsub' AND t.enabled = TRUE{ANTIJOIN}", + WHERE t.kind = 'pubsub' AND t.enabled = TRUE AND t.shared = FALSE{ANTIJOIN}", ANTIJOIN = crate::trigger_repo::TRIGGER_SUPPRESSION_ANTIJOIN, )) .bind(ctx.app_id.into_inner()) @@ -104,21 +118,7 @@ impl PubsubRepo for PostgresPubsubRepo { if !topic_matches(&r.topic_pattern, topic) { continue; } - sqlx::query( - "INSERT INTO outbox ( \ - app_id, source_kind, trigger_id, script_id, reply_to, \ - payload, origin_principal, trigger_depth, root_execution_id \ - ) VALUES ($1, 'pubsub', $2, $3, NULL, $4, $5, $6, $7)", - ) - .bind(ctx.app_id.into_inner()) - .bind(r.id) - .bind(r.script_id) - .bind(&event_payload) - .bind(ctx.origin_principal.map(AdminUserId::into_inner)) - .bind(i32::try_from(ctx.trigger_depth.saturating_add(1)).unwrap_or(1)) - .bind(ctx.root_execution_id.into_inner()) - .execute(&mut *tx) - .await?; + insert_pubsub_outbox_row(&mut tx, &ctx, r.id, r.script_id, &event_payload).await?; written += 1; } @@ -126,4 +126,69 @@ impl PubsubRepo for PostgresPubsubRepo { tx.commit().await?; Ok(written) } + + async fn fan_out_shared_publish( + &self, + ctx: PublishCtx, + owning_group: GroupId, + topic: &str, + event_payload: serde_json::Value, + ) -> Result { + let mut tx = self.pool.begin().await?; + + // §11.6 D2: only `shared = true` pubsub triggers on the RESOLVED owning + // group — the shared-topic namespace boundary (a per-app or non-owning + // trigger never matches a shared publish). Topic-pattern match in Rust + // like the per-app path. + let rows: Vec = sqlx::query_as( + "SELECT t.id, t.script_id, d.topic_pattern \ + FROM triggers t \ + JOIN pubsub_trigger_details d ON d.trigger_id = t.id \ + WHERE t.kind = 'pubsub' AND t.enabled = TRUE AND t.shared = TRUE \ + AND t.group_id = $1", + ) + .bind(owning_group.into_inner()) + .fetch_all(&mut *tx) + .await?; + + let mut written: u32 = 0; + for r in rows { + if !topic_matches(&r.topic_pattern, topic) { + continue; + } + insert_pubsub_outbox_row(&mut tx, &ctx, r.id, r.script_id, &event_payload).await?; + written += 1; + } + + tx.commit().await?; + Ok(written) + } +} + +/// Insert one pubsub delivery row, stamped with the writer `ctx.app_id` so the +/// handler runs under the publishing app. Shared by the per-app + shared +/// fan-outs. +async fn insert_pubsub_outbox_row( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + ctx: &PublishCtx, + trigger_id: Uuid, + script_id: Uuid, + event_payload: &serde_json::Value, +) -> Result<(), PubsubRepoError> { + sqlx::query( + "INSERT INTO outbox ( \ + app_id, source_kind, trigger_id, script_id, reply_to, \ + payload, origin_principal, trigger_depth, root_execution_id \ + ) VALUES ($1, 'pubsub', $2, $3, NULL, $4, $5, $6, $7)", + ) + .bind(ctx.app_id.into_inner()) + .bind(trigger_id) + .bind(script_id) + .bind(event_payload) + .bind(ctx.origin_principal.map(AdminUserId::into_inner)) + .bind(i32::try_from(ctx.trigger_depth.saturating_add(1)).unwrap_or(1)) + .bind(ctx.root_execution_id.into_inner()) + .execute(&mut **tx) + .await?; + Ok(()) } diff --git a/crates/manager-core/src/pubsub_service.rs b/crates/manager-core/src/pubsub_service.rs index 4b35ee2..06d379f 100644 --- a/crates/manager-core/src/pubsub_service.rs +++ b/crates/manager-core/src/pubsub_service.rs @@ -386,6 +386,17 @@ mod tests { self.written.lock().unwrap().extend(staged); Ok(u32::try_from(n).unwrap_or(u32::MAX)) } + + async fn fan_out_shared_publish( + &self, + _ctx: PublishCtx, + _owning_group: picloud_shared::GroupId, + _topic: &str, + _event_payload: serde_json::Value, + ) -> Result { + // The per-app PubsubService tests don't exercise the shared path. + Ok(0) + } } #[derive(Default)] diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index ee55471..f7f7f42 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -20,22 +20,23 @@ use picloud_manager_core::{ AppsState, AuthState, AuthzRepo, DeadLetterRepo, DeadLettersState, DevEmailState, Dispatcher, DocsServiceImpl, EmailInboundState, EmailServiceImpl, FilesAdminState, FilesConfig, FilesServiceImpl, FsFilesRepo, FsGroupFilesRepo, GroupDocsServiceImpl, GroupFilesServiceImpl, - GroupKvServiceImpl, GroupMembersRepository, GroupRepository, GroupsState, HttpConfig, - HttpServiceImpl, InboundNonceDedup, KvAdminState, KvServiceImpl, OutboxEventEmitter, - OutboxRepo, PostgresAbandonedRepo, PostgresAdminSessionRepository, PostgresAdminUserRepository, - PostgresApiKeyRepository, PostgresAppDomainRepository, PostgresAppMembersRepository, - PostgresAppRepository, PostgresAppSecretsRepo, PostgresAppUserInvitationRepo, - PostgresAppUserPasswordResetRepo, PostgresAppUserRepository, PostgresAppUserRoleRepo, - PostgresAppUserSessionRepository, PostgresAppUserVerificationRepo, PostgresDeadLetterRepo, - PostgresDeadLetterService, PostgresDocsRepo, PostgresExecutionLogRepository, - PostgresExecutionLogSink, PostgresGroupCollectionResolver, PostgresGroupDocsRepo, - PostgresGroupKvRepo, PostgresGroupMembersRepository, PostgresGroupRepository, PostgresKvRepo, - PostgresOutboxRepo, PostgresPubsubRepo, PostgresRouteRepository, PostgresScriptRepository, - PostgresSecretsRepo, PostgresTopicRepo, PostgresTriggerRepo, PostgresVarsRepo, - PrincipalResolver, PubsubServiceImpl, RealtimeAuthorityImpl, RepoResolver, RouteAdminState, - SandboxCeiling, ScriptRepository, SecretsConfig, SecretsServiceImpl, SecretsState, - SubscriberTokenConfig, TopicRepo, TopicsState, TriggerConfig, TriggerRepo, TriggersState, - UsersServiceConfig, UsersServiceImpl, VarsApiState, VarsServiceImpl, + GroupKvServiceImpl, GroupMembersRepository, GroupPubsubServiceImpl, GroupRepository, + GroupsState, HttpConfig, HttpServiceImpl, InboundNonceDedup, KvAdminState, KvServiceImpl, + OutboxEventEmitter, OutboxRepo, PostgresAbandonedRepo, PostgresAdminSessionRepository, + PostgresAdminUserRepository, PostgresApiKeyRepository, PostgresAppDomainRepository, + PostgresAppMembersRepository, PostgresAppRepository, PostgresAppSecretsRepo, + PostgresAppUserInvitationRepo, PostgresAppUserPasswordResetRepo, PostgresAppUserRepository, + PostgresAppUserRoleRepo, PostgresAppUserSessionRepository, PostgresAppUserVerificationRepo, + PostgresDeadLetterRepo, PostgresDeadLetterService, PostgresDocsRepo, + PostgresExecutionLogRepository, PostgresExecutionLogSink, PostgresGroupCollectionResolver, + PostgresGroupDocsRepo, PostgresGroupKvRepo, PostgresGroupMembersRepository, + PostgresGroupRepository, PostgresKvRepo, PostgresOutboxRepo, PostgresPubsubRepo, + PostgresRouteRepository, PostgresScriptRepository, PostgresSecretsRepo, PostgresTopicRepo, + PostgresTriggerRepo, PostgresVarsRepo, PrincipalResolver, PubsubServiceImpl, + RealtimeAuthorityImpl, RepoResolver, RouteAdminState, SandboxCeiling, ScriptRepository, + SecretsConfig, SecretsServiceImpl, SecretsState, SubscriberTokenConfig, TopicRepo, TopicsState, + TriggerConfig, TriggerRepo, TriggersState, UsersServiceConfig, UsersServiceImpl, VarsApiState, + VarsServiceImpl, }; use picloud_orchestrator_core::realtime::DEFAULT_GC_INTERVAL_SECS; use picloud_orchestrator_core::routing::{AppDomainTable, RouteTable}; @@ -260,6 +261,13 @@ pub async fn build_app( // the same dispatcher as every other async trigger) AND, best-effort, // to in-process SSE subscribers. let pubsub_repo = Arc::new(PostgresPubsubRepo::new(pool.clone())); + // §11.6 D2: the shared-topic service reuses the same fan-out repo. + let group_pubsub: Arc = + Arc::new(GroupPubsubServiceImpl::new( + pubsub_repo.clone(), + Arc::new(PostgresGroupCollectionResolver::new(pool.clone())), + authz.clone(), + )); let pubsub: Arc = Arc::new( PubsubServiceImpl::new(pubsub_repo, authz.clone()) .with_max_message_bytes( @@ -384,7 +392,8 @@ pub async fn build_app( ) .with_group_kv(group_kv) .with_group_docs(group_docs) - .with_group_files(group_files); + .with_group_files(group_files) + .with_group_pubsub(group_pubsub); // v1.1.9: keep the invoke depth bound aligned with the dispatcher's // trigger-depth bound (same counter under the hood). let engine_limits = Limits { diff --git a/crates/shared/src/group_pubsub.rs b/crates/shared/src/group_pubsub.rs new file mode 100644 index 0000000..a1a25f0 --- /dev/null +++ b/crates/shared/src/group_pubsub.rs @@ -0,0 +1,84 @@ +//! `GroupPubsubService` — the §11.6 D2 shared group-TOPIC publish contract. +//! +//! The pub/sub counterpart to [`crate::GroupKvService`]. A script reaches it via +//! the explicit `pubsub::shared_topic("name")` handle (distinct from the app- +//! private `pubsub::publish_durable`). A shared topic is **storeless**: there is +//! no message log — a publish fans out to the `shared = true` pubsub triggers on +//! the OWNING GROUP (resolved by walking the calling app's ancestor chain for +//! the nearest group that declares the topic shared). That ancestry walk is the +//! isolation boundary — a foreign app's chain never contains the owning group, +//! so the topic does not resolve. The trait takes **no** group id (owner +//! derivation stays inside the impl), matching `GroupKvService`. +//! +//! Trust model (§11.6): a publish is a WRITE to shared state, so it requires an +//! authenticated principal with editor+ on the owning group (fails closed for +//! an anonymous caller), matching `GroupKvService` writes. + +use async_trait::async_trait; +use thiserror::Error; + +use crate::SdkCallCx; + +#[async_trait] +pub trait GroupPubsubService: Send + Sync { + /// Publish `message` to the group shared topic `namespace` (optionally under + /// `subtopic`, so `("events", "created")` publishes `events.created`). + /// Resolves the owning group (kind `topic`) from `cx.app_id`'s chain, + /// requires editor+, and fans out to `shared = true` pubsub triggers on that + /// group. Returns the number of delivery rows written (0 when no trigger + /// matched — the publish still succeeds). + async fn publish( + &self, + cx: &SdkCallCx, + namespace: &str, + subtopic: &str, + message: serde_json::Value, + ) -> Result; +} + +/// Stub for test bundles that don't exercise shared topics. Every call errors so +/// accidental use surfaces clearly. +#[derive(Debug, Default, Clone, Copy)] +pub struct NoopGroupPubsubService; + +#[async_trait] +impl GroupPubsubService for NoopGroupPubsubService { + async fn publish( + &self, + _cx: &SdkCallCx, + _namespace: &str, + _subtopic: &str, + _message: serde_json::Value, + ) -> Result { + Err(GroupPubsubError::Backend( + "group pubsub is not wired in".into(), + )) + } +} + +/// Failure modes surfaced to the Rhai bridge. +#[derive(Debug, Error)] +pub enum GroupPubsubError { + /// Empty topic namespace; rejected at the SDK boundary. + #[error("shared topic name must not be empty")] + InvalidTopic, + + /// No group on the calling app's ancestor chain declares this topic shared — + /// the structural "not shared with you" boundary. Also the outcome for a + /// foreign app naming another subtree's topic. + #[error("topic {0:?} is not a shared group topic for this app")] + CollectionNotShared(String), + + /// Caller lacked the required capability. A publish always needs an + /// authenticated editor+ on the owning group (fails closed for anon). + #[error("forbidden")] + Forbidden, + + /// Message exceeds the per-message JSON-encoded size cap. + #[error("message too large: {actual} bytes exceeds the {limit}-byte cap")] + MessageTooLarge { limit: usize, actual: usize }, + + /// Storage / backend failure. + #[error("backend error: {0}")] + Backend(String), +} diff --git a/crates/shared/src/lib.rs b/crates/shared/src/lib.rs index 924af55..ac94c0f 100644 --- a/crates/shared/src/lib.rs +++ b/crates/shared/src/lib.rs @@ -19,6 +19,7 @@ pub mod group; pub mod group_docs; pub mod group_files; pub mod group_kv; +pub mod group_pubsub; pub mod http; pub mod ids; pub mod inbox; @@ -71,6 +72,7 @@ pub use group::Group; pub use group_docs::{GroupDocsError, GroupDocsService, NoopGroupDocsService}; pub use group_files::{GroupFilesError, GroupFilesService, NoopGroupFilesService}; pub use group_kv::{GroupKvError, GroupKvService, NoopGroupKvService}; +pub use group_pubsub::{GroupPubsubError, GroupPubsubService, NoopGroupPubsubService}; pub use http::{HttpError, HttpRequest, HttpResponse, HttpService, NoopHttpService}; pub use ids::{ AdminUserId, ApiKeyId, AppId, AppUserId, ExecutionId, GroupId, InvitationId, QueueMessageId, diff --git a/crates/shared/src/services.rs b/crates/shared/src/services.rs index e0f17de..4a5c1dc 100644 --- a/crates/shared/src/services.rs +++ b/crates/shared/src/services.rs @@ -21,12 +21,12 @@ use std::sync::Arc; use crate::{ DeadLetterService, DocsService, EmailService, FilesService, GroupDocsService, - GroupFilesService, GroupKvService, HttpService, InvokeService, KvService, ModuleSource, - NoopDeadLetterService, NoopDocsService, NoopEmailService, NoopEventEmitter, NoopFilesService, - NoopGroupDocsService, NoopGroupFilesService, NoopGroupKvService, NoopHttpService, - NoopInvokeService, NoopKvService, NoopModuleSource, NoopPubsubService, NoopQueueService, - NoopSecretsService, NoopUsersService, NoopVarsService, PubsubService, QueueService, - SecretsService, ServiceEventEmitter, UsersService, VarsService, + GroupFilesService, GroupKvService, GroupPubsubService, HttpService, InvokeService, KvService, + ModuleSource, NoopDeadLetterService, NoopDocsService, NoopEmailService, NoopEventEmitter, + NoopFilesService, NoopGroupDocsService, NoopGroupFilesService, NoopGroupKvService, + NoopGroupPubsubService, NoopHttpService, NoopInvokeService, NoopKvService, NoopModuleSource, + NoopPubsubService, NoopQueueService, NoopSecretsService, NoopUsersService, NoopVarsService, + PubsubService, QueueService, SecretsService, ServiceEventEmitter, UsersService, VarsService, }; /// SDK service bundle. See module docs for the lifecycle and the v1.1.x @@ -133,6 +133,12 @@ pub struct Services { /// `files::shared_collection(name)`. Wired via [`Services::with_group_files`]; /// defaults to `NoopGroupFilesService`. pub group_files: Arc, + + /// Shared cross-app group TOPICS (§11.6 D2). Scripts get + /// `pubsub::shared_topic(name)` — publish to a group-scoped topic watched by + /// `shared = true` group pubsub triggers. Wired via + /// [`Services::with_group_pubsub`]; defaults to `NoopGroupPubsubService`. + pub group_pubsub: Arc, } impl Services { @@ -178,6 +184,7 @@ impl Services { group_kv: Arc::new(NoopGroupKvService), group_docs: Arc::new(NoopGroupDocsService), group_files: Arc::new(NoopGroupFilesService), + group_pubsub: Arc::new(NoopGroupPubsubService), } } @@ -203,6 +210,13 @@ impl Services { self } + /// Set the §11.6 D2 shared group-TOPIC service (picloud binary; tests leave noop). + #[must_use] + pub fn with_group_pubsub(mut self, group_pubsub: Arc) -> Self { + self.group_pubsub = group_pubsub; + self + } + /// All-noop bundle for tests that build an `Engine` but don't /// exercise the stateful services. Returns the same shape as /// `Services::new` so callers can't accidentally rely on a stub