refactor(groups): dedup the group-service value-size check + shared-emit tail

Two clean, local dedups in the group shared-collection services:

- Extract `GroupKvServiceImpl::check_value_size` — `set` and `set_if` inlined
  the identical encode-and-cap block; now one method (mirrors the sibling
  `group_docs_service::check_data_size`).
- Add `group_collection_repo::best_effort_emit_shared` — the KV/docs/files
  services each copied the same emit-and-log-on-error tail for shared-collection
  triggers. Centralize the tail (logging the event's own `source`/`op`); each
  service still builds its own `ServiceEvent` (payload shapes differ).

Pure refactor, pinned by the existing group-service unit tests. (The
`owning_group` resolver was evaluated for dedup too but left as-is: the five
error enums diverge — `GroupFilesError::InvalidCollection` carries a `String`,
`GroupPubsubError` lacks the variant, and files/pubsub skip the empty-check kv/
docs/queue do — so a shared conversion would be a behavior change, not a
refactor.)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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(

crates/manager-core/src/group_docs_service.rs

@@ -127,24 +127,20 @@ impl GroupDocsServiceImpl {
         payload: Option<serde_json::Value>,
         old_payload: Option<serde_json::Value>,
     ) {
-        if let Err(e) = self
+        crate::group_collection_repo::best_effort_emit_shared(
-            .events
+            &*self.events,
-            .emit_shared(
+            cx,
-                cx,
+            group_id,
-                group_id,
+            ServiceEvent {
-                ServiceEvent {
+                source: "docs",
-                    source: "docs",
+                op,
-                    op,
+                collection: Some(collection.to_string()),
-                    collection: Some(collection.to_string()),
+                key: Some(id.to_string()),
-                    key: Some(id.to_string()),
+                payload,
-                    payload,
+                old_payload,
-                    old_payload,
+            },
-                },
+        )
-            )
+        .await;
-            .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

crates/manager-core/src/group_files_service.rs

@@ -73,24 +73,20 @@ impl GroupFilesServiceImpl {
         meta: &FileMeta,
         old: Option<&FileMeta>,
     ) {
-        if let Err(e) = self
+        crate::group_collection_repo::best_effort_emit_shared(
-            .events
+            &*self.events,
-            .emit_shared(
+            cx,
-                cx,
+            group_id,
-                group_id,
+            ServiceEvent {
-                ServiceEvent {
+                source: "files",
-                    source: "files",
+                op,
-                    op,
+                collection: Some(collection.to_string()),
-                    collection: Some(collection.to_string()),
+                key: Some(meta.id.to_string()),
-                    key: Some(meta.id.to_string()),
+                payload: serde_json::to_value(meta).ok(),
-                    payload: serde_json::to_value(meta).ok(),
+                old_payload: old.and_then(|m| serde_json::to_value(m).ok()),
-                    old_payload: old.and_then(|m| serde_json::to_value(m).ok()),
+            },
-                },
+        )
-            )
+        .await;
-            .await
-        {
-            tracing::error!(error = %e, source = "files", op, event_emit_failure = true, "shared event emit failed");
-        }
     }
 
     /// The structural boundary: resolve the collection to its owning group

crates/manager-core/src/group_kv_service.rs

@@ -135,24 +135,36 @@ impl GroupKvServiceImpl {
         key: &str,
         payload: Option<serde_json::Value>,
     ) {
-        if let Err(e) = self
+        crate::group_collection_repo::best_effort_emit_shared(
-            .events
+            &*self.events,
-            .emit_shared(
+            cx,
-                cx,
+            group_id,
-                group_id,
+            ServiceEvent {
-                ServiceEvent {
+                source: "kv",
-                    source: "kv",
+                op,
-                    op,
+                collection: Some(collection.to_string()),
-                    collection: Some(collection.to_string()),
+                key: Some(key.to_string()),
-                    key: Some(key.to_string()),
+                payload,
-                    payload,
+                old_payload: None,
-                    old_payload: None,
+            },
-                },
+        )
-            )
+        .await;
-            .await
+    }
-        {
+
-            tracing::error!(error = %e, source = "kv", op, event_emit_failure = true, "shared event emit failed");
+    /// 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<usize, GroupKvError> {
+        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)
+        let encoded_len = self.check_value_size(&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,
-            });
-        }
         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<bool, GroupKvError> {
         let group_id = self.owning_group(cx, collection).await?;
-        let encoded_len = serde_json::to_vec(&new)
+        let encoded_len = self.check_value_size(&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,
-            });
-        }
         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)