From 0210ace406df75f5bc9a2fddbbfae11dfc6263d3 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Wed, 1 Jul 2026 20:33:09 +0200 Subject: [PATCH] feat(shared-triggers): emit shared events from group services (M2.3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ServiceEventEmitter gains emit_shared(cx, owning_group, event) (default no-op); OutboxEventEmitter implements it — matches shared triggers on the owning group and writes outbox rows stamped with the WRITER app_id (dispatch runs under the writer). GroupKv/Docs/FilesServiceImpl gain an events field (Noop default + with_events builder) and emit on set/create/update/delete; the host wires the outbox emitter via with_events. Closes the "group trigger has no app to watch" gap. Authoring + validation land in M2.4. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/manager-core/src/group_docs_service.rs | 96 ++++++++++++- .../manager-core/src/group_files_service.rs | 68 ++++++++-- crates/manager-core/src/group_kv_service.rs | 74 +++++++++- .../manager-core/src/outbox_event_emitter.rs | 126 +++++++++++++++++- crates/picloud/src/lib.rs | 28 ++-- crates/shared/src/events.rs | 17 ++- 6 files changed, 381 insertions(+), 28 deletions(-) diff --git a/crates/manager-core/src/group_docs_service.rs b/crates/manager-core/src/group_docs_service.rs index 35e9f2c..b5430ec 100644 --- a/crates/manager-core/src/group_docs_service.rs +++ b/crates/manager-core/src/group_docs_service.rs @@ -11,7 +11,8 @@ use std::sync::Arc; use async_trait::async_trait; use picloud_shared::{ - DocId, DocRow, DocsListPage, GroupDocsError, GroupDocsService, GroupId, SdkCallCx, + DocId, DocRow, DocsListPage, GroupDocsError, GroupDocsService, GroupId, NoopEventEmitter, + SdkCallCx, ServiceEvent, ServiceEventEmitter, }; use crate::authz::{self, AuthzRepo, Capability}; @@ -28,6 +29,8 @@ pub struct GroupDocsServiceImpl { resolver: Arc, authz: Arc, max_value_bytes: usize, + /// §11.6: fires `shared = true` docs triggers on a shared-collection write. + events: Arc, } impl GroupDocsServiceImpl { @@ -40,6 +43,13 @@ impl GroupDocsServiceImpl { Self::with_max_value_bytes(repo, resolver, authz, docs_max_value_bytes_from_env()) } + /// Wire the event emitter (§11.6 shared-collection triggers). + #[must_use] + pub fn with_events(mut self, events: Arc) -> Self { + self.events = events; + self + } + #[must_use] pub fn with_max_value_bytes( repo: Arc, @@ -48,6 +58,7 @@ impl GroupDocsServiceImpl { max_value_bytes: usize, ) -> Self { Self { + events: Arc::new(NoopEventEmitter), repo, resolver, authz, @@ -55,6 +66,39 @@ impl GroupDocsServiceImpl { } } + /// §11.6: fire `shared = true` docs triggers on the owning group. + /// Best-effort; an emit failure is logged, never surfaced. + #[allow(clippy::too_many_arguments)] // op/collection/id + new + old payloads + async fn emit_shared( + &self, + cx: &SdkCallCx, + group_id: GroupId, + op: &'static str, + collection: &str, + id: &str, + payload: Option, + old_payload: Option, + ) { + if let Err(e) = self + .events + .emit_shared( + cx, + group_id, + ServiceEvent { + source: "docs", + op, + collection: Some(collection.to_string()), + key: Some(id.to_string()), + payload, + old_payload, + }, + ) + .await + { + tracing::error!(error = %e, source = "docs", op, event_emit_failure = true, "shared event emit failed"); + } + } + /// The structural boundary: resolve the collection to its owning group /// (kind=`docs`) on the calling app's chain. `CollectionNotShared` when no /// ancestor group declares it. @@ -145,7 +189,18 @@ impl GroupDocsService for GroupDocsServiceImpl { validate_data(&data)?; self.check_data_size(&data)?; self.check_write(cx, group_id).await?; - Ok(self.repo.create(group_id, collection, data).await?.id) + let created = self.repo.create(group_id, collection, data.clone()).await?; + self.emit_shared( + cx, + group_id, + "create", + collection, + &created.id.to_string(), + Some(data), + None, + ) + .await; + Ok(created.id) } async fn get( @@ -198,8 +253,24 @@ impl GroupDocsService for GroupDocsServiceImpl { validate_data(&data)?; self.check_data_size(&data)?; self.check_write(cx, group_id).await?; - match self.repo.update(group_id, collection, id, data).await? { - Some(_) => Ok(()), + match self + .repo + .update(group_id, collection, id, data.clone()) + .await? + { + Some(prev) => { + self.emit_shared( + cx, + group_id, + "update", + collection, + &id.to_string(), + Some(data), + Some(prev), + ) + .await; + Ok(()) + } None => Err(GroupDocsError::NotFound), } } @@ -212,7 +283,22 @@ impl GroupDocsService for GroupDocsServiceImpl { ) -> Result { let group_id = self.owning_group(cx, collection).await?; self.check_write(cx, group_id).await?; - Ok(self.repo.delete(group_id, collection, id).await?.is_some()) + match self.repo.delete(group_id, collection, id).await? { + Some(prev) => { + self.emit_shared( + cx, + group_id, + "delete", + collection, + &id.to_string(), + None, + Some(prev), + ) + .await; + Ok(true) + } + None => Ok(false), + } } async fn list( diff --git a/crates/manager-core/src/group_files_service.rs b/crates/manager-core/src/group_files_service.rs index 41cc3c7..6884a04 100644 --- a/crates/manager-core/src/group_files_service.rs +++ b/crates/manager-core/src/group_files_service.rs @@ -12,7 +12,8 @@ use std::sync::Arc; use async_trait::async_trait; use picloud_shared::{ sanitize_stored_content_type, validate_files_collection, FileMeta, FileUpdate, FilesListPage, - GroupFilesError, GroupFilesService, GroupId, NewFile, SdkCallCx, + GroupFilesError, GroupFilesService, GroupId, NewFile, NoopEventEmitter, SdkCallCx, + ServiceEvent, ServiceEventEmitter, }; use uuid::Uuid; @@ -29,6 +30,8 @@ pub struct GroupFilesServiceImpl { resolver: Arc, authz: Arc, max_file_size_bytes: usize, + /// §11.6: fires `shared = true` files triggers on a shared-collection write. + events: Arc, } impl GroupFilesServiceImpl { @@ -44,6 +47,45 @@ impl GroupFilesServiceImpl { resolver, authz, max_file_size_bytes, + events: Arc::new(NoopEventEmitter), + } + } + + /// Wire the event emitter (§11.6 shared-collection triggers). + #[must_use] + pub fn with_events(mut self, events: Arc) -> Self { + self.events = events; + self + } + + /// §11.6: fire `shared = true` files triggers on the owning group. + /// Best-effort; an emit failure is logged, never surfaced. + async fn emit_shared( + &self, + cx: &SdkCallCx, + group_id: GroupId, + op: &'static str, + collection: &str, + meta: &FileMeta, + old: Option<&FileMeta>, + ) { + if let Err(e) = self + .events + .emit_shared( + cx, + group_id, + ServiceEvent { + source: "files", + op, + collection: Some(collection.to_string()), + key: Some(meta.id.to_string()), + payload: serde_json::to_value(meta).ok(), + old_payload: old.and_then(|m| serde_json::to_value(m).ok()), + }, + ) + .await + { + tracing::error!(error = %e, source = "files", op, event_emit_failure = true, "shared event emit failed"); } } @@ -118,7 +160,10 @@ impl GroupFilesService for GroupFilesServiceImpl { // shape checks pass (same as app files, audit 2026-06-11 C-2). new.content_type = sanitize_stored_content_type(&new.content_type); self.check_write(cx, group_id).await?; - Ok(self.repo.create(group_id, collection, new).await?.id) + let meta = self.repo.create(group_id, collection, new).await?; + self.emit_shared(cx, group_id, "create", collection, &meta, None) + .await; + Ok(meta.id) } async fn head( @@ -169,7 +214,11 @@ impl GroupFilesService for GroupFilesServiceImpl { return Err(GroupFilesError::NotFound); }; match self.repo.update(group_id, collection, uuid, upd).await? { - Some(FileUpdated { .. }) => Ok(()), + Some(FileUpdated { new, prev }) => { + self.emit_shared(cx, group_id, "update", collection, &new, Some(&prev)) + .await; + Ok(()) + } None => Err(GroupFilesError::NotFound), } } @@ -186,11 +235,14 @@ impl GroupFilesService for GroupFilesServiceImpl { let Some(uuid) = parse_id(id) else { return Ok(false); }; - Ok(self - .repo - .delete(group_id, collection, uuid) - .await? - .is_some()) + match self.repo.delete(group_id, collection, uuid).await? { + Some(meta) => { + self.emit_shared(cx, group_id, "delete", collection, &meta, Some(&meta)) + .await; + Ok(true) + } + None => Ok(false), + } } async fn list( diff --git a/crates/manager-core/src/group_kv_service.rs b/crates/manager-core/src/group_kv_service.rs index 700c01c..da4946c 100644 --- a/crates/manager-core/src/group_kv_service.rs +++ b/crates/manager-core/src/group_kv_service.rs @@ -22,7 +22,10 @@ use std::sync::Arc; use async_trait::async_trait; -use picloud_shared::{GroupId, GroupKvError, GroupKvService, KvListPage, SdkCallCx}; +use picloud_shared::{ + GroupId, GroupKvError, GroupKvService, KvListPage, NoopEventEmitter, SdkCallCx, ServiceEvent, + ServiceEventEmitter, +}; use crate::authz::{self, AuthzRepo, Capability}; use crate::group_collection_repo::GroupCollectionResolver; @@ -37,6 +40,10 @@ pub struct GroupKvServiceImpl { resolver: Arc, authz: Arc, max_value_bytes: usize, + /// §11.6: fires `shared = true` triggers on a shared-collection write. + /// Defaults to the noop emitter; the host wires the outbox emitter via + /// [`Self::with_events`]. + events: Arc, } impl GroupKvServiceImpl { @@ -49,6 +56,14 @@ impl GroupKvServiceImpl { Self::with_max_value_bytes(repo, resolver, authz, kv_max_value_bytes_from_env()) } + /// Wire the event emitter (§11.6 shared-collection triggers). Without this + /// the service uses the noop emitter and shared writes fire nothing. + #[must_use] + pub fn with_events(mut self, events: Arc) -> Self { + self.events = events; + self + } + #[must_use] pub fn with_max_value_bytes( repo: Arc, @@ -57,6 +72,7 @@ impl GroupKvServiceImpl { max_value_bytes: usize, ) -> Self { Self { + events: Arc::new(NoopEventEmitter), repo, resolver, authz, @@ -82,6 +98,38 @@ impl GroupKvServiceImpl { .ok_or_else(|| GroupKvError::CollectionNotShared(collection.to_string())) } + /// §11.6: fire `shared = true` kv triggers on the owning group. Best-effort + /// — an emit failure is logged, never surfaced (the write already + /// committed), matching the per-app `KvServiceImpl` behaviour. + async fn emit_shared( + &self, + cx: &SdkCallCx, + group_id: GroupId, + op: &'static str, + collection: &str, + key: &str, + payload: Option, + ) { + if let Err(e) = self + .events + .emit_shared( + cx, + group_id, + ServiceEvent { + source: "kv", + op, + collection: Some(collection.to_string()), + key: Some(key.to_string()), + payload, + old_payload: None, + }, + ) + .await + { + tracing::error!(error = %e, source = "kv", op, event_emit_failure = true, "shared event emit failed"); + } + } + async fn check_read(&self, cx: &SdkCallCx, group_id: GroupId) -> Result<(), GroupKvError> { authz::script_gate( &*self.authz, @@ -144,7 +192,22 @@ impl GroupKvService for GroupKvServiceImpl { }); } self.check_write(cx, group_id).await?; - self.repo.set(group_id, collection, key, value).await?; + // Determine insert vs update for the event op (best-effort, like the + // per-app path — the has+set is non-transactional but triggers are + // fire-and-forget). + let existed = self.repo.has(group_id, collection, key).await?; + self.repo + .set(group_id, collection, key, value.clone()) + .await?; + self.emit_shared( + cx, + group_id, + if existed { "update" } else { "insert" }, + collection, + key, + Some(value), + ) + .await; Ok(()) } @@ -156,7 +219,12 @@ impl GroupKvService for GroupKvServiceImpl { ) -> Result { let group_id = self.owning_group(cx, collection).await?; self.check_write(cx, group_id).await?; - Ok(self.repo.delete(group_id, collection, key).await?.is_some()) + let deleted = self.repo.delete(group_id, collection, key).await?.is_some(); + if deleted { + self.emit_shared(cx, group_id, "delete", collection, key, None) + .await; + } + Ok(deleted) } async fn has(&self, cx: &SdkCallCx, collection: &str, key: &str) -> Result { diff --git a/crates/manager-core/src/outbox_event_emitter.rs b/crates/manager-core/src/outbox_event_emitter.rs index a5c705c..e7a1d95 100644 --- a/crates/manager-core/src/outbox_event_emitter.rs +++ b/crates/manager-core/src/outbox_event_emitter.rs @@ -19,7 +19,7 @@ use std::sync::Arc; use async_trait::async_trait; use picloud_shared::{ - DocsEventOp, EmitError, FileMeta, FilesEventOp, KvEventOp, SdkCallCx, ServiceEvent, + DocsEventOp, EmitError, FileMeta, FilesEventOp, GroupId, KvEventOp, SdkCallCx, ServiceEvent, ServiceEventEmitter, TriggerEvent, }; @@ -51,6 +51,130 @@ impl ServiceEventEmitter for OutboxEventEmitter { _ => Ok(()), } } + + #[allow(clippy::too_many_lines)] // one match arm per source kind (kv/docs/files) + async fn emit_shared( + &self, + cx: &SdkCallCx, + owning_group: GroupId, + event: ServiceEvent, + ) -> Result<(), EmitError> { + // §11.6: a shared-collection write. Match `shared = true` triggers on + // the OWNING group; each fires under the WRITER app (`cx.app_id`), the + // same "group template runs under the firing app" model. + let source_kind = match event.source { + "kv" => OutboxSourceKind::Kv, + "docs" => OutboxSourceKind::Docs, + "files" => OutboxSourceKind::Files, + _ => return Ok(()), + }; + let Some(collection) = event.collection.clone() else { + return Ok(()); + }; + let (matches, trigger_event) = match event.source { + "kv" => { + let Some(op) = KvEventOp::from_wire(event.op) else { + return Ok(()); + }; + let m = self + .triggers + .list_matching_shared_kv(owning_group, &collection, op) + .await + .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?; + let ev = TriggerEvent::Kv { + op, + collection, + key: event.key.clone().unwrap_or_default(), + value: event.payload.clone(), + }; + ( + m.into_iter() + .map(|x| (x.trigger_id, x.script_id)) + .collect::>(), + ev, + ) + } + "docs" => { + let Some(op) = DocsEventOp::from_wire(event.op) else { + return Ok(()); + }; + let m = self + .triggers + .list_matching_shared_docs(owning_group, &collection, op) + .await + .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?; + let ev = TriggerEvent::Docs { + op, + collection, + id: event.key.clone().unwrap_or_default(), + data: event.payload.clone(), + prev_data: event.old_payload.clone(), + }; + ( + m.into_iter() + .map(|x| (x.trigger_id, x.script_id)) + .collect::>(), + ev, + ) + } + "files" => { + let Some(op) = FilesEventOp::from_wire(event.op) else { + return Ok(()); + }; + let Some(meta) = event + .payload + .clone() + .and_then(|v| serde_json::from_value::(v).ok()) + else { + return Ok(()); + }; + let m = self + .triggers + .list_matching_shared_files(owning_group, &collection, op) + .await + .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?; + let ev = TriggerEvent::Files { + op, + collection, + id: meta.id.to_string(), + name: meta.name, + content_type: meta.content_type, + size: meta.size, + checksum: meta.checksum, + prev: event.old_payload.clone(), + }; + ( + m.into_iter() + .map(|x| (x.trigger_id, x.script_id)) + .collect::>(), + ev, + ) + } + _ => return Ok(()), + }; + if matches.is_empty() { + return Ok(()); + } + let payload = serde_json::to_value(&trigger_event) + .map_err(|e| EmitError::Rejected(format!("event serialize: {e}")))?; + for (trigger_id, script_id) in matches { + self.outbox + .insert(NewOutboxRow { + app_id: cx.app_id, + source_kind, + trigger_id: Some(trigger_id), + script_id: Some(script_id), + reply_to: None, + payload: payload.clone(), + origin_principal: cx.principal.as_ref().map(|p| p.user_id), + trigger_depth: cx.trigger_depth.saturating_add(1), + root_execution_id: Some(cx.root_execution_id), + }) + .await + .map_err(|e| EmitError::Unavailable(format!("outbox insert: {e}")))?; + } + Ok(()) + } } impl OutboxEventEmitter { diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index d5bb473..f651193 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -161,21 +161,26 @@ pub async fn build_app( )); // §11.6 shared group collections (KV): a separate store keyed by the owning // group, resolved from cx.app_id's chain. Reuses the KV value-size cap. - let group_kv: Arc = - Arc::new(GroupKvServiceImpl::with_max_value_bytes( + let group_kv: Arc = Arc::new( + GroupKvServiceImpl::with_max_value_bytes( Arc::new(PostgresGroupKvRepo::new(pool.clone())), Arc::new(PostgresGroupCollectionResolver::new(pool.clone())), authz.clone(), picloud_manager_core::kv_service::kv_max_value_bytes_from_env(), - )); + ) + // §11.6 shared-collection triggers: fire `shared = true` triggers. + .with_events(events.clone()), + ); // §11.6 shared group collections (docs): group-keyed group_docs store. - let group_docs: Arc = - Arc::new(GroupDocsServiceImpl::with_max_value_bytes( + let group_docs: Arc = Arc::new( + GroupDocsServiceImpl::with_max_value_bytes( Arc::new(PostgresGroupDocsRepo::new(pool.clone())), Arc::new(PostgresGroupCollectionResolver::new(pool.clone())), authz.clone(), picloud_manager_core::docs_service::docs_max_value_bytes_from_env(), - )); + ) + .with_events(events.clone()), + ); let docs: Arc = Arc::new(DocsServiceImpl::with_max_value_bytes( docs_repo, authz.clone(), @@ -216,14 +221,17 @@ pub async fn build_app( )); // §11.6 shared group collections (files): group-keyed blobs under // `/files/groups//...`, resolved from cx.app_id's chain. - // Shares the per-file size cap; no events (no app to attribute a trigger to). - let group_files: Arc = - Arc::new(GroupFilesServiceImpl::new( + // Shares the per-file size cap; fires `shared = true` triggers under the + // writer app (§11.6 shared-collection triggers). + let group_files: Arc = Arc::new( + GroupFilesServiceImpl::new( Arc::new(FsGroupFilesRepo::new(pool.clone(), files_root.clone())), Arc::new(PostgresGroupCollectionResolver::new(pool.clone())), authz.clone(), files_max_size, - )); + ) + .with_events(events.clone()), + ); // v1.1.6 realtime: the in-process broadcaster is shared between the // publish path (PubsubServiceImpl fans out to SSE subscribers after // the durable outbox fan-out) and the SSE endpoint (subscribe side). diff --git a/crates/shared/src/events.rs b/crates/shared/src/events.rs index 1720925..9afec18 100644 --- a/crates/shared/src/events.rs +++ b/crates/shared/src/events.rs @@ -20,7 +20,7 @@ use async_trait::async_trait; use thiserror::Error; -use crate::SdkCallCx; +use crate::{GroupId, SdkCallCx}; /// Trait every stateful service depends on to emit events. The host /// binary constructs one instance and clones the Arc into each service. @@ -31,6 +31,21 @@ pub trait ServiceEventEmitter: Send + Sync { /// will return `Ok(())` once the event is durably persisted, the /// dispatcher reads it out-of-band. async fn emit(&self, cx: &SdkCallCx, event: ServiceEvent) -> Result<(), EmitError>; + + /// §11.6: publish an event for a write to a SHARED (group-owned) + /// collection. Unlike `emit` (which fans out over the writer app's own + + /// inherited triggers), this matches `shared = true` triggers on the + /// `owning_group` and runs the handler under the writer app (`cx.app_id`). + /// Default no-op so the noop emitter + any non-outbox impls drop it. + async fn emit_shared( + &self, + cx: &SdkCallCx, + owning_group: GroupId, + event: ServiceEvent, + ) -> Result<(), EmitError> { + let _ = (cx, owning_group, event); + Ok(()) + } } /// One service event. `source` and `op` are `&'static str` because they