From 72802de64484e20664c555d6c9b66b504fea5773 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Tue, 30 Jun 2026 20:43:01 +0200 Subject: [PATCH] =?UTF-8?q?feat(modules):=20live=20chain-union=20dispatch?= =?UTF-8?q?=20for=20group=20trigger=20templates=20(=C2=A711=20tail=20T3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The runtime heart: a descendant app's event now fires its ancestor groups' trigger templates, resolved LIVE (no materialization). - trigger_repo: list_matching_kv/docs/files prepend CHAIN_LEVELS_CTE and `JOIN chain c ON (t.app_id = c.app_owner OR t.group_id = c.group_owner)` (bind $1 = app_id), so each returns the app's own triggers PLUS ancestor-group templates in one statement. NULL-comparison semantics keep app/group rows from cross-matching; each row matches at exactly one chain depth (no dup fan-out). - pubsub_repo: same union on the publish fan-out query. - The outbox row stamps the firing app_id + the (group-owned) script_id; the dispatcher's existing script_invocable() (app-owned OR invocable via chain membership) already runs a group handler under a descendant app โ€” the Phase-4 inheriting-app boundary, unchanged. Glob/ops/topic filtering in Rust is untouched. 404 manager-core unit tests green; clippy -D clean. Cross-subtree firing is journey-proven in T4. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/manager-core/src/pubsub_repo.rs | 16 +++++++++---- crates/manager-core/src/trigger_repo.rs | 31 +++++++++++++++---------- 2 files changed, 31 insertions(+), 16 deletions(-) diff --git a/crates/manager-core/src/pubsub_repo.rs b/crates/manager-core/src/pubsub_repo.rs index 58a1f52..8613bd6 100644 --- a/crates/manager-core/src/pubsub_repo.rs +++ b/crates/manager-core/src/pubsub_repo.rs @@ -16,6 +16,8 @@ use picloud_shared::{topic_matches, AdminUserId, AppId, ExecutionId}; use sqlx::PgPool; use uuid::Uuid; +use crate::config_resolver::CHAIN_LEVELS_CTE; + #[derive(Debug, thiserror::Error)] pub enum PubsubRepoError { #[error("database error: {0}")] @@ -78,12 +80,18 @@ impl PubsubRepo for PostgresPubsubRepo { // Load all enabled pubsub triggers for the app; filter by topic // pattern in Rust (keeps the query simple, honours the // empty/`*`/prefix semantics without teaching SQL about globs). - let rows: Vec = sqlx::query_as( - "SELECT t.id, t.script_id, d.topic_pattern \ + // ยง11 tail: the chain union picks up the app's own pubsub triggers AND + // ancestor-group pubsub TEMPLATES (live, no materialization). The outbox + // row below stamps the firing `ctx.app_id`, so a template runs under the + // publishing app โ€” the inheriting-app boundary. + let rows: Vec = sqlx::query_as(&format!( + "{CHAIN_LEVELS_CTE} \ + SELECT t.id, t.script_id, d.topic_pattern \ FROM triggers t \ JOIN pubsub_trigger_details d ON d.trigger_id = t.id \ - WHERE t.app_id = $1 AND t.kind = 'pubsub' AND t.enabled = TRUE", - ) + JOIN chain c ON (t.app_id = c.app_owner OR t.group_id = c.group_owner) \ + WHERE t.kind = 'pubsub' AND t.enabled = TRUE" + )) .bind(ctx.app_id.into_inner()) .fetch_all(&mut *tx) .await?; diff --git a/crates/manager-core/src/trigger_repo.rs b/crates/manager-core/src/trigger_repo.rs index adf232c..e3d0e90 100644 --- a/crates/manager-core/src/trigger_repo.rs +++ b/crates/manager-core/src/trigger_repo.rs @@ -14,6 +14,7 @@ use serde::{Deserialize, Serialize}; use sqlx::PgPool; use uuid::Uuid; +use crate::config_resolver::CHAIN_LEVELS_CTE; use crate::trigger_config::BackoffShape; #[derive(Debug, thiserror::Error)] @@ -1304,15 +1305,17 @@ impl TriggerRepo for PostgresTriggerRepo { // happens in Rust so we don't have to teach the query about // `*` and `prefix:*`. Sets are tiny in practice (one app's // worth of triggers, usually a handful). - let rows: Vec = sqlx::query_as( - "SELECT t.id, t.script_id, t.dispatch_mode, \ + let rows: Vec = sqlx::query_as(&format!( + "{CHAIN_LEVELS_CTE} \ + 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 kv_trigger_details d ON d.trigger_id = t.id \ - WHERE t.app_id = $1 AND t.kind = 'kv' AND t.enabled = TRUE", - ) + 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" + )) .bind(app_id.into_inner()) .fetch_all(&self.pool) .await?; @@ -1352,15 +1355,17 @@ impl TriggerRepo for PostgresTriggerRepo { // ops check into SQL (`WHERE $op = ANY(ops)`) โ€” that would // exclude rows with `ops = '{}'` from the results, breaking // the empty-array-means-any-op semantic. - let rows: Vec = sqlx::query_as( - "SELECT t.id, t.script_id, t.dispatch_mode, \ + let rows: Vec = sqlx::query_as(&format!( + "{CHAIN_LEVELS_CTE} \ + 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 docs_trigger_details d ON d.trigger_id = t.id \ - WHERE t.app_id = $1 AND t.kind = 'docs' AND t.enabled = TRUE", - ) + 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" + )) .bind(app_id.into_inner()) .fetch_all(&self.pool) .await?; @@ -1397,15 +1402,17 @@ impl TriggerRepo for PostgresTriggerRepo { ) -> Result, TriggerRepoError> { // Mirrors list_matching_kv: pull every enabled files trigger, // filter glob + ops in Rust (empty ops array means "any op"). - let rows: Vec = sqlx::query_as( - "SELECT t.id, t.script_id, t.dispatch_mode, \ + let rows: Vec = sqlx::query_as(&format!( + "{CHAIN_LEVELS_CTE} \ + 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 files_trigger_details d ON d.trigger_id = t.id \ - WHERE t.app_id = $1 AND t.kind = 'files' AND t.enabled = TRUE", - ) + 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" + )) .bind(app_id.into_inner()) .fetch_all(&self.pool) .await?;