fabi/PiCloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(shared-triggers): emit shared events from group services (M2.3)

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-07-01 20:33:09 +02:00
parent cf296cd2fd
commit 0210ace406
6 changed files with 381 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
crates/manager-core/src/group_docs_service.rs
View File

@@ -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<dyn GroupCollectionResolver>,
authz: Arc<dyn AuthzRepo>,
max_value_bytes: usize,
/// §11.6: fires `shared = true` docs triggers on a shared-collection write.
events: Arc<dyn ServiceEventEmitter>,
}
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<dyn ServiceEventEmitter>) -> Self {
self.events = events;
self
}
#[must_use]
pub fn with_max_value_bytes(
repo: Arc<dyn GroupDocsRepo>,
@@ -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<serde_json::Value>,
old_payload: Option<serde_json::Value>,
) {
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<bool, GroupDocsError> {
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(

68
crates/manager-core/src/group_files_service.rs
View File

@@ -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<dyn GroupCollectionResolver>,
authz: Arc<dyn AuthzRepo>,
max_file_size_bytes: usize,
/// §11.6: fires `shared = true` files triggers on a shared-collection write.
events: Arc<dyn ServiceEventEmitter>,
}
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<dyn ServiceEventEmitter>) -> 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(

74
crates/manager-core/src/group_kv_service.rs
View File

@@ -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<dyn GroupCollectionResolver>,
authz: Arc<dyn AuthzRepo>,
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<dyn ServiceEventEmitter>,
}
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<dyn ServiceEventEmitter>) -> Self {
self.events = events;
self
}
#[must_use]
pub fn with_max_value_bytes(
repo: Arc<dyn GroupKvRepo>,
@@ -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<serde_json::Value>,
) {
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<bool, GroupKvError> {
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<bool, GroupKvError> {

126
crates/manager-core/src/outbox_event_emitter.rs
View File

@@ -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::<Vec<_>>(),
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::<Vec<_>>(),
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::<FileMeta>(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::<Vec<_>>(),
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 {

28
crates/picloud/src/lib.rs
View File

@@ -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<dyn picloud_shared::GroupKvService> =
Arc::new(GroupKvServiceImpl::with_max_value_bytes(
let group_kv: Arc<dyn picloud_shared::GroupKvService> = 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<dyn picloud_shared::GroupDocsService> =
Arc::new(GroupDocsServiceImpl::with_max_value_bytes(
let group_docs: Arc<dyn picloud_shared::GroupDocsService> = 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<dyn DocsService> = 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
// `<root>/files/groups/<group_id>/...`, 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<dyn picloud_shared::GroupFilesService> =
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<dyn picloud_shared::GroupFilesService> = 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).

17
crates/shared/src/events.rs
View File

@@ -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