diff --git a/crates/manager-core/src/group_collection_repo.rs b/crates/manager-core/src/group_collection_repo.rs index 15eb86c..0667369 100644 --- a/crates/manager-core/src/group_collection_repo.rs +++ b/crates/manager-core/src/group_collection_repo.rs @@ -13,11 +13,29 @@ //! name does not resolve. use async_trait::async_trait; -use picloud_shared::{AppId, GroupId, ScriptOwner}; +use picloud_shared::{AppId, GroupId, ScriptOwner, SdkCallCx, ServiceEvent, ServiceEventEmitter}; use sqlx::{PgPool, Postgres, Transaction}; use crate::config_resolver::CHAIN_LEVELS_CTE; +/// §11.6: fire a `shared = true` trigger event on the owning group, best-effort. +/// An emit failure is logged (with the event's `source`/`op`) and swallowed — +/// the write already committed, and shared triggers are fire-and-forget, so a +/// dropped emit must never fail the caller. The single home for the tail the +/// group KV/docs/files services previously each copied; each still builds its +/// own `ServiceEvent` (payload shapes differ) and calls this. +pub(crate) async fn best_effort_emit_shared( + events: &dyn ServiceEventEmitter, + cx: &SdkCallCx, + group_id: GroupId, + event: ServiceEvent, +) { + let (source, op) = (event.source, event.op); + if let Err(e) = events.emit_shared(cx, group_id, event).await { + tracing::error!(error = %e, source, op, event_emit_failure = true, "shared event emit failed"); + } +} + /// List **all** shared-collection declarations at `owner` as `(name, kind)` /// pairs, sorted. Used by the kind-aware apply diff and `collections ls`. pub async fn list_all_for_owner( diff --git a/crates/manager-core/src/group_docs_service.rs b/crates/manager-core/src/group_docs_service.rs index 9ea6607..efccde4 100644 --- a/crates/manager-core/src/group_docs_service.rs +++ b/crates/manager-core/src/group_docs_service.rs @@ -127,24 +127,20 @@ impl GroupDocsServiceImpl { 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"); - } + crate::group_collection_repo::best_effort_emit_shared( + &*self.events, + cx, + group_id, + ServiceEvent { + source: "docs", + op, + collection: Some(collection.to_string()), + key: Some(id.to_string()), + payload, + old_payload, + }, + ) + .await; } /// The structural boundary: resolve the collection to its owning group diff --git a/crates/manager-core/src/group_files_service.rs b/crates/manager-core/src/group_files_service.rs index 062fcd4..793d0a8 100644 --- a/crates/manager-core/src/group_files_service.rs +++ b/crates/manager-core/src/group_files_service.rs @@ -73,24 +73,20 @@ impl GroupFilesServiceImpl { 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"); - } + crate::group_collection_repo::best_effort_emit_shared( + &*self.events, + 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; } /// The structural boundary: resolve the collection to its owning group diff --git a/crates/manager-core/src/group_kv_service.rs b/crates/manager-core/src/group_kv_service.rs index 4221457..38f4a02 100644 --- a/crates/manager-core/src/group_kv_service.rs +++ b/crates/manager-core/src/group_kv_service.rs @@ -135,24 +135,36 @@ impl GroupKvServiceImpl { 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"); + crate::group_collection_repo::best_effort_emit_shared( + &*self.events, + cx, + group_id, + ServiceEvent { + source: "kv", + op, + collection: Some(collection.to_string()), + key: Some(key.to_string()), + payload, + old_payload: None, + }, + ) + .await; + } + + /// Encode `value` and enforce the per-value byte cap, returning its encoded + /// length (both callers need it for the byte-quota projection). Mirrors the + /// sibling `group_docs_service::check_data_size`. + fn check_value_size(&self, value: &serde_json::Value) -> Result { + let encoded_len = serde_json::to_vec(value) + .map(|v| v.len()) + .map_err(|e| GroupKvError::Backend(format!("encode value: {e}")))?; + if encoded_len > self.max_value_bytes { + return Err(GroupKvError::ValueTooLarge { + limit: self.max_value_bytes, + actual: encoded_len, + }); } + Ok(encoded_len) } async fn check_read(&self, cx: &SdkCallCx, group_id: GroupId) -> Result<(), GroupKvError> { @@ -207,15 +219,7 @@ impl GroupKvService for GroupKvServiceImpl { value: serde_json::Value, ) -> Result<(), GroupKvError> { let group_id = self.owning_group(cx, collection).await?; - let encoded_len = serde_json::to_vec(&value) - .map(|v| v.len()) - .map_err(|e| GroupKvError::Backend(format!("encode value: {e}")))?; - if encoded_len > self.max_value_bytes { - return Err(GroupKvError::ValueTooLarge { - limit: self.max_value_bytes, - actual: encoded_len, - }); - } + let encoded_len = self.check_value_size(&value)?; self.check_write(cx, group_id).await?; // Determine insert vs update for the event op + quota deltas (best-effort, // like the per-app path — the get+set is non-transactional but triggers @@ -282,15 +286,7 @@ impl GroupKvService for GroupKvServiceImpl { new: serde_json::Value, ) -> Result { let group_id = self.owning_group(cx, collection).await?; - let encoded_len = serde_json::to_vec(&new) - .map(|v| v.len()) - .map_err(|e| GroupKvError::Backend(format!("encode value: {e}")))?; - if encoded_len > self.max_value_bytes { - return Err(GroupKvError::ValueTooLarge { - limit: self.max_value_bytes, - actual: encoded_len, - }); - } + let encoded_len = self.check_value_size(&new)?; self.check_write(cx, group_id).await?; // Same quota rules as `set` (best-effort pre-check; the swap itself is // atomic in the repo). A swap that inserts a NEW key (expected absent)