feat(shared-triggers): match-query split for shared vs per-app (M2.2)

The per-app list_matching_kv/docs/files gain `AND t.shared = FALSE` (a shared
trigger never matches a per-app write). New list_matching_shared_{kv,docs,files}
select enabled `shared = TRUE` triggers on the OWNING group (glob+op filtered
in Rust), returning the same match types — no chain, no suppression anti-join.
Trait defaults return empty so non-Postgres impls degrade. Emission wiring in
M2.3.

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

crates/manager-core/src/trigger_repo.rs

@@ -485,6 +485,39 @@ pub trait TriggerRepo: Send + Sync {
         op: FilesEventOp,
     ) -> Result<Vec<FilesTriggerMatch>, TriggerRepoError>;
 
+    /// §11.6 shared-collection fan-out: enabled `shared = true` triggers on the
+    /// OWNING group (the group that declares the shared collection), glob-matched
+    /// in Rust. The dispatch runs under the writing app; these methods key on the
+    /// owning group, not the writer's chain. Default empty so non-Postgres impls
+    /// degrade to "no shared triggers".
+    async fn list_matching_shared_kv(
+        &self,
+        owning_group: GroupId,
+        collection: &str,
+        op: KvEventOp,
+    ) -> Result<Vec<KvTriggerMatch>, TriggerRepoError> {
+        let _ = (owning_group, collection, op);
+        Ok(Vec::new())
+    }
+    async fn list_matching_shared_docs(
+        &self,
+        owning_group: GroupId,
+        collection: &str,
+        op: DocsEventOp,
+    ) -> Result<Vec<DocsTriggerMatch>, TriggerRepoError> {
+        let _ = (owning_group, collection, op);
+        Ok(Vec::new())
+    }
+    async fn list_matching_shared_files(
+        &self,
+        owning_group: GroupId,
+        collection: &str,
+        op: FilesEventOp,
+    ) -> Result<Vec<FilesTriggerMatch>, TriggerRepoError> {
+        let _ = (owning_group, collection, op);
+        Ok(Vec::new())
+    }
+
     /// Dispatcher hot path for dead-letter fan-out. Filters: source
     /// (or any-source), originating trigger_id (or any), originating
     /// script_id (or any). Each filter is "match OR is_null".
@@ -538,6 +571,39 @@ impl PostgresTriggerRepo {
     pub fn new(pool: PgPool) -> Self {
         Self { pool }
     }
+
+    /// §11.6 shared-collection match: enabled `shared = true` triggers on the
+    /// OWNING group, glob + op filtered in Rust (like the per-app path). `kind`
+    /// and `detail_table` are hard-coded literals from the three call sites (no
+    /// injection). No chain / no suppression anti-join — a shared trigger is
+    /// declared on the group that owns the collection and fires under the writer.
+    async fn shared_match_rows(
+        &self,
+        kind: &str,
+        detail_table: &str,
+        owning_group: GroupId,
+        collection: &str,
+        op_str: &str,
+    ) -> Result<Vec<KvMatchRow>, TriggerRepoError> {
+        let rows: Vec<KvMatchRow> = sqlx::query_as(&format!(
+            "SELECT t.id, t.script_id, t.dispatch_mode, \
+                    t.retry_max_attempts, t.retry_backoff, t.retry_base_ms, \
+                    t.registered_by_principal, \
+                    d.collection_glob, d.ops \
+             FROM triggers t \
+             JOIN {detail_table} d ON d.trigger_id = t.id \
+             WHERE t.kind = '{kind}' AND t.enabled = TRUE \
+               AND t.shared = TRUE AND t.group_id = $1"
+        ))
+        .bind(owning_group.into_inner())
+        .fetch_all(&self.pool)
+        .await?;
+        Ok(rows
+            .into_iter()
+            .filter(|r| collection_matches(&r.collection_glob, collection))
+            .filter(|r| r.ops.is_empty() || r.ops.iter().any(|o| o == op_str))
+            .collect())
+    }
 }
 
 /// Insert a trigger (parent row + per-kind detail) within an existing
@@ -1353,7 +1419,8 @@ impl TriggerRepo for PostgresTriggerRepo {
              FROM triggers t \
              JOIN kv_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 = 'kv' AND t.enabled = TRUE{TRIGGER_SUPPRESSION_ANTIJOIN}"
+             WHERE t.kind = 'kv' AND t.enabled = TRUE \
+               AND t.shared = FALSE{TRIGGER_SUPPRESSION_ANTIJOIN}"
         ))
         .bind(app_id.into_inner())
         .fetch_all(&self.pool)
@@ -1403,7 +1470,8 @@ impl TriggerRepo for PostgresTriggerRepo {
              FROM triggers t \
              JOIN docs_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 = 'docs' AND t.enabled = TRUE{TRIGGER_SUPPRESSION_ANTIJOIN}"
+             WHERE t.kind = 'docs' AND t.enabled = TRUE \
+               AND t.shared = FALSE{TRIGGER_SUPPRESSION_ANTIJOIN}"
         ))
         .bind(app_id.into_inner())
         .fetch_all(&self.pool)
@@ -1450,7 +1518,8 @@ impl TriggerRepo for PostgresTriggerRepo {
              FROM triggers t \
              JOIN files_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 = 'files' AND t.enabled = TRUE{TRIGGER_SUPPRESSION_ANTIJOIN}"
+             WHERE t.kind = 'files' AND t.enabled = TRUE \
+               AND t.shared = FALSE{TRIGGER_SUPPRESSION_ANTIJOIN}"
         ))
         .bind(app_id.into_inner())
         .fetch_all(&self.pool)
@@ -1480,6 +1549,60 @@ impl TriggerRepo for PostgresTriggerRepo {
         Ok(out)
     }
 
+    async fn list_matching_shared_kv(
+        &self,
+        owning_group: GroupId,
+        collection: &str,
+        op: KvEventOp,
+    ) -> Result<Vec<KvTriggerMatch>, TriggerRepoError> {
+        let rows = self
+            .shared_match_rows(
+                "kv",
+                "kv_trigger_details",
+                owning_group,
+                collection,
+                op.as_str(),
+            )
+            .await?;
+        Ok(rows.into_iter().map(KvMatchRow::into_kv).collect())
+    }
+
+    async fn list_matching_shared_docs(
+        &self,
+        owning_group: GroupId,
+        collection: &str,
+        op: DocsEventOp,
+    ) -> Result<Vec<DocsTriggerMatch>, TriggerRepoError> {
+        let rows = self
+            .shared_match_rows(
+                "docs",
+                "docs_trigger_details",
+                owning_group,
+                collection,
+                op.as_str(),
+            )
+            .await?;
+        Ok(rows.into_iter().map(KvMatchRow::into_docs).collect())
+    }
+
+    async fn list_matching_shared_files(
+        &self,
+        owning_group: GroupId,
+        collection: &str,
+        op: FilesEventOp,
+    ) -> Result<Vec<FilesTriggerMatch>, TriggerRepoError> {
+        let rows = self
+            .shared_match_rows(
+                "files",
+                "files_trigger_details",
+                owning_group,
+                collection,
+                op.as_str(),
+            )
+            .await?;
+        Ok(rows.into_iter().map(KvMatchRow::into_files).collect())
+    }
+
     async fn list_matching_dead_letter(
         &self,
         app_id: AppId,
@@ -1950,6 +2073,47 @@ struct KvMatchRow {
     ops: Vec<String>,
 }
 
+impl KvMatchRow {
+    fn into_kv(self) -> KvTriggerMatch {
+        KvTriggerMatch {
+            trigger_id: self.id.into(),
+            script_id: self.script_id.into(),
+            dispatch_mode: dispatch_from_str(&self.dispatch_mode),
+            retry_max_attempts: u32::try_from(self.retry_max_attempts).unwrap_or(3),
+            retry_backoff: BackoffShape::from_wire(&self.retry_backoff)
+                .unwrap_or(BackoffShape::Exponential),
+            retry_base_ms: u32::try_from(self.retry_base_ms).unwrap_or(1000),
+            registered_by_principal: self.registered_by_principal.into(),
+        }
+    }
+
+    fn into_docs(self) -> DocsTriggerMatch {
+        let m = self.into_kv();
+        DocsTriggerMatch {
+            trigger_id: m.trigger_id,
+            script_id: m.script_id,
+            dispatch_mode: m.dispatch_mode,
+            retry_max_attempts: m.retry_max_attempts,
+            retry_backoff: m.retry_backoff,
+            retry_base_ms: m.retry_base_ms,
+            registered_by_principal: m.registered_by_principal,
+        }
+    }
+
+    fn into_files(self) -> FilesTriggerMatch {
+        let m = self.into_kv();
+        FilesTriggerMatch {
+            trigger_id: m.trigger_id,
+            script_id: m.script_id,
+            dispatch_mode: m.dispatch_mode,
+            retry_max_attempts: m.retry_max_attempts,
+            retry_backoff: m.retry_backoff,
+            retry_base_ms: m.retry_base_ms,
+            registered_by_principal: m.registered_by_principal,
+        }
+    }
+}
+
 #[derive(sqlx::FromRow)]
 struct DlMatchRow {
     id: Uuid,