From cf296cd2fd529a0aab928dc2c559a895d9a8361f Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Wed, 1 Jul 2026 20:21:24 +0200 Subject: [PATCH] feat(shared-triggers): match-query split for shared vs per-app (M2.2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- crates/manager-core/src/trigger_repo.rs | 170 +++++++++++++++++++++++- 1 file changed, 167 insertions(+), 3 deletions(-) diff --git a/crates/manager-core/src/trigger_repo.rs b/crates/manager-core/src/trigger_repo.rs index d3782bb..ac721c3 100644 --- a/crates/manager-core/src/trigger_repo.rs +++ b/crates/manager-core/src/trigger_repo.rs @@ -485,6 +485,39 @@ pub trait TriggerRepo: Send + Sync { op: FilesEventOp, ) -> Result, 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, TriggerRepoError> { + let _ = (owning_group, collection, op); + Ok(Vec::new()) + } + async fn list_matching_shared_docs( + &self, + owning_group: GroupId, + collection: &str, + op: DocsEventOp, + ) -> Result, TriggerRepoError> { + let _ = (owning_group, collection, op); + Ok(Vec::new()) + } + async fn list_matching_shared_files( + &self, + owning_group: GroupId, + collection: &str, + op: FilesEventOp, + ) -> Result, 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, TriggerRepoError> { + let rows: Vec = 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, 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, 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, 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, } +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,