From a2360a9464596b4e625d8262b0a192fcfcb0753c Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Tue, 14 Jul 2026 19:20:41 +0200 Subject: [PATCH] refactor(outbox): make the trigger fan-out connection-scoped MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `OutboxEventEmitter` resolved matching triggers and inserted outbox rows through `Arc` / `Arc`, i.e. against the pool — each query on whatever connection it happened to get. That makes a transactional outbox impossible: the data write and the outbox rows can never share a transaction, so a crash (or an outbox error) between them silently loses the trigger event. Take the emitter down to a connection instead of a pool: * `outbox_repo::insert_on(exec, row)` and `trigger_repo::list_matching_on` / `list_matching_shared_on(exec, ...)` are generic over `PgExecutor`, so the same SQL serves a pooled connection or a `&mut *tx`. The repo trait methods delegate to them — the SQL keeps exactly one home. * `emit_on` / `emit_shared_on` take a `&mut PgConnection` and run the whole fan-out on it. The `ServiceEventEmitter` impl acquires one pooled connection and calls them, so behaviour is unchanged today; a caller holding a transaction can now pass `&mut *tx` and have the outbox rows commit with the write. * `OutboxEventEmitter::new` takes the `PgPool` directly (it was only ever constructed once, in the host wiring). Also collapses the three copy-pasted per-app match queries (kv/docs/files differ only in the `kind` discriminator and detail table) and the three `emit_*` bodies into one `plan()` + one match fn, so the suppression anti-join, the chain walk, and the empty-ops-means-any-op semantic each exist once rather than three times. No behaviour change — pure refactor. It is the seam the transactional write lands on next. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../manager-core/src/outbox_event_emitter.rs | 473 +++++++----------- crates/manager-core/src/outbox_repo.rs | 49 +- crates/manager-core/src/trigger_repo.rs | 329 ++++++------ crates/picloud/src/lib.rs | 5 +- 4 files changed, 351 insertions(+), 505 deletions(-) diff --git a/crates/manager-core/src/outbox_event_emitter.rs b/crates/manager-core/src/outbox_event_emitter.rs index e7a1d95..3bf4545 100644 --- a/crates/manager-core/src/outbox_event_emitter.rs +++ b/crates/manager-core/src/outbox_event_emitter.rs @@ -1,139 +1,135 @@ -//! `OutboxEventEmitter` — the real `ServiceEventEmitter` that replaces -//! v1.1.0's `NoopEventEmitter` once the triggers framework lands. +//! `OutboxEventEmitter` — the real `ServiceEventEmitter` behind every stateful +//! service. On each collection mutation (KV / docs / files) it looks up the +//! triggers the event matches and writes one outbox row per match; the +//! dispatcher reads them back out-of-band. //! -//! On each `emit` (a KV mutation, future doc/file/pubsub event, etc.): -//! 1. Look up matching triggers for the event's (app_id, source, op, -//! collection) tuple via `TriggerRepo::list_matching_*`. -//! 2. For each match, write one outbox row carrying the event payload -//! serialized as a `TriggerEvent`. +//! **The fan-out is connection-scoped, not pool-scoped.** [`emit_on`] and +//! [`emit_shared_on`] take a `&mut PgConnection`, so a caller that already holds +//! a transaction can pass `&mut *tx` and have the outbox rows commit *with* the +//! data write that produced them — the transactional-outbox pattern. That closes +//! the window where a row committed but its trigger never fired because the +//! outbox insert failed (or the process died) immediately afterwards. Services +//! that write through `crate::atomic_write` take that path; the +//! `ServiceEventEmitter` trait impl below is the non-transactional entry point, +//! kept for the callers (and the tests) that only need best-effort emission. //! -//! Defaults applied at write time so `OutboxRow.payload` carries -//! everything the dispatcher needs to reconstruct the executor -//! invocation without joining back to the trigger row. -//! -//! Non-KV `ServiceEvent` sources are silently dropped in v1.1.1 — the -//! dispatcher only knows how to fire KV triggers this release. Future -//! sources (docs/files/pubsub) add their own dispatch arm. - -use std::sync::Arc; +//! Non-collection `ServiceEvent` sources are silently dropped: the SDK calls +//! `events.emit(...)` unconditionally for forward compat, and every source the +//! dispatcher has an arm for is handled here. use async_trait::async_trait; use picloud_shared::{ DocsEventOp, EmitError, FileMeta, FilesEventOp, GroupId, KvEventOp, SdkCallCx, ServiceEvent, ServiceEventEmitter, TriggerEvent, }; +use sqlx::{PgConnection, PgPool}; -use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxSourceKind}; -use crate::trigger_repo::TriggerRepo; +use crate::outbox_repo::{insert_on, NewOutboxRow, OutboxSourceKind}; +use crate::trigger_repo::{list_matching_on, list_matching_shared_on, KvMatchRow}; pub struct OutboxEventEmitter { - triggers: Arc, - outbox: Arc, + pool: PgPool, } impl OutboxEventEmitter { #[must_use] - pub fn new(triggers: Arc, outbox: Arc) -> Self { - Self { triggers, outbox } + pub fn new(pool: PgPool) -> Self { + Self { pool } } } #[async_trait] impl ServiceEventEmitter for OutboxEventEmitter { async fn emit(&self, cx: &SdkCallCx, event: ServiceEvent) -> Result<(), EmitError> { - match event.source { - "kv" => self.emit_kv(cx, event).await, - "docs" => self.emit_docs(cx, event).await, - "files" => self.emit_files(cx, event).await, - // Future sources land here. For now, silently drop — the - // SDK calls `events.emit(...)` unconditionally for forward - // compat, so swallowing without an error is correct. - _ => Ok(()), - } + let mut conn = self + .pool + .acquire() + .await + .map_err(|e| EmitError::Unavailable(format!("acquire connection: {e}")))?; + emit_on(&mut conn, cx, &event).await } - #[allow(clippy::too_many_lines)] // one match arm per source kind (kv/docs/files) async fn emit_shared( &self, cx: &SdkCallCx, owning_group: GroupId, event: ServiceEvent, ) -> Result<(), EmitError> { - // §11.6: a shared-collection write. Match `shared = true` triggers on - // the OWNING group; each fires under the WRITER app (`cx.app_id`), the - // same "group template runs under the firing app" model. - let source_kind = match event.source { - "kv" => OutboxSourceKind::Kv, - "docs" => OutboxSourceKind::Docs, - "files" => OutboxSourceKind::Files, - _ => return Ok(()), - }; - let Some(collection) = event.collection.clone() else { - return Ok(()); - }; - let (matches, trigger_event) = match event.source { - "kv" => { - let Some(op) = KvEventOp::from_wire(event.op) else { - return Ok(()); - }; - let m = self - .triggers - .list_matching_shared_kv(owning_group, &collection, op) - .await - .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?; - let ev = TriggerEvent::Kv { + let mut conn = self + .pool + .acquire() + .await + .map_err(|e| EmitError::Unavailable(format!("acquire connection: {e}")))?; + emit_shared_on(&mut conn, cx, owning_group, &event).await + } +} + +/// Everything the fan-out needs, derived from a `ServiceEvent` with no I/O: the +/// trigger-table coordinates to match on, and the `TriggerEvent` the handler +/// will see as `ctx.event`. `None` for a source or op the dispatcher has no arm +/// for — the caller drops the event quietly. +struct EventPlan { + source_kind: OutboxSourceKind, + /// `triggers.kind` discriminator. A literal — never user data, since it is + /// interpolated into the match SQL. + kind: &'static str, + /// Per-kind detail table. A literal, for the same reason. + detail_table: &'static str, + collection: String, + op_str: &'static str, + trigger_event: TriggerEvent, +} + +fn plan(event: &ServiceEvent) -> Option { + // Collection events always carry a collection; defensively skip if not. + let collection = event.collection.clone()?; + let key = event.key.clone().unwrap_or_default(); + match event.source { + "kv" => { + let op = KvEventOp::from_wire(event.op)?; + Some(EventPlan { + source_kind: OutboxSourceKind::Kv, + kind: "kv", + detail_table: "kv_trigger_details", + collection: collection.clone(), + op_str: op.as_str(), + trigger_event: TriggerEvent::Kv { op, collection, - key: event.key.clone().unwrap_or_default(), + key, value: event.payload.clone(), - }; - ( - m.into_iter() - .map(|x| (x.trigger_id, x.script_id)) - .collect::>(), - ev, - ) - } - "docs" => { - let Some(op) = DocsEventOp::from_wire(event.op) else { - return Ok(()); - }; - let m = self - .triggers - .list_matching_shared_docs(owning_group, &collection, op) - .await - .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?; - let ev = TriggerEvent::Docs { + }, + }) + } + "docs" => { + let op = DocsEventOp::from_wire(event.op)?; + Some(EventPlan { + source_kind: OutboxSourceKind::Docs, + kind: "docs", + detail_table: "docs_trigger_details", + collection: collection.clone(), + op_str: op.as_str(), + trigger_event: TriggerEvent::Docs { op, collection, - id: event.key.clone().unwrap_or_default(), + id: key, data: event.payload.clone(), prev_data: event.old_payload.clone(), - }; - ( - m.into_iter() - .map(|x| (x.trigger_id, x.script_id)) - .collect::>(), - ev, - ) - } - "files" => { - let Some(op) = FilesEventOp::from_wire(event.op) else { - return Ok(()); - }; - let Some(meta) = event - .payload - .clone() - .and_then(|v| serde_json::from_value::(v).ok()) - else { - return Ok(()); - }; - let m = self - .triggers - .list_matching_shared_files(owning_group, &collection, op) - .await - .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?; - let ev = TriggerEvent::Files { + }, + }) + } + "files" => { + let op = FilesEventOp::from_wire(event.op)?; + // The payload is the `FileMeta` JSON the files service emitted — + // never the blob bytes. + let meta: FileMeta = serde_json::from_value(event.payload.clone()?).ok()?; + Some(EventPlan { + source_kind: OutboxSourceKind::Files, + kind: "files", + detail_table: "files_trigger_details", + collection: collection.clone(), + op_str: op.as_str(), + trigger_event: TriggerEvent::Files { op, collection, id: meta.id.to_string(), @@ -142,206 +138,87 @@ impl ServiceEventEmitter for OutboxEventEmitter { size: meta.size, checksum: meta.checksum, prev: event.old_payload.clone(), - }; - ( - m.into_iter() - .map(|x| (x.trigger_id, x.script_id)) - .collect::>(), - ev, - ) - } - _ => return Ok(()), - }; - if matches.is_empty() { - return Ok(()); + }, + }) } - let payload = serde_json::to_value(&trigger_event) - .map_err(|e| EmitError::Rejected(format!("event serialize: {e}")))?; - for (trigger_id, script_id) in matches { - self.outbox - .insert(NewOutboxRow { - app_id: cx.app_id, - source_kind, - trigger_id: Some(trigger_id), - script_id: Some(script_id), - reply_to: None, - payload: payload.clone(), - origin_principal: cx.principal.as_ref().map(|p| p.user_id), - trigger_depth: cx.trigger_depth.saturating_add(1), - root_execution_id: Some(cx.root_execution_id), - }) - .await - .map_err(|e| EmitError::Unavailable(format!("outbox insert: {e}")))?; - } - Ok(()) + _ => None, } } -impl OutboxEventEmitter { - async fn emit_kv(&self, cx: &SdkCallCx, event: ServiceEvent) -> Result<(), EmitError> { - let Some(op) = KvEventOp::from_wire(event.op) else { - return Ok(()); // unknown op — drop quietly - }; - let Some(collection) = event.collection.clone() else { - return Ok(()); // KV events always carry a collection — defensively skip - }; - let key = event.key.clone().unwrap_or_default(); - - let matches = self - .triggers - .list_matching_kv(cx.app_id, &collection, op) - .await - .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?; - - if matches.is_empty() { - return Ok(()); - } - - // Serialize the originating event as a TriggerEvent so the - // dispatcher can hand it to the script as `ctx.event` without - // round-tripping back to the trigger row. - let trigger_event = TriggerEvent::Kv { - op, - collection, - key, - value: event.payload.clone(), - }; - let payload = serde_json::to_value(&trigger_event) - .map_err(|e| EmitError::Rejected(format!("event serialize: {e}")))?; - - for m in matches { - self.outbox - .insert(NewOutboxRow { - app_id: cx.app_id, - source_kind: OutboxSourceKind::Kv, - trigger_id: Some(m.trigger_id), - script_id: Some(m.script_id), - reply_to: None, - payload: payload.clone(), - origin_principal: cx.principal.as_ref().map(|p| p.user_id), - trigger_depth: cx.trigger_depth.saturating_add(1), - root_execution_id: Some(cx.root_execution_id), - }) - .await - .map_err(|e| EmitError::Unavailable(format!("outbox insert: {e}")))?; - } - Ok(()) - } - - /// v1.1.2. Mirrors `emit_kv` — fan out a docs mutation across - /// matching docs triggers + write one outbox row each. The - /// `prev_data` change-data-capture surface is preserved from the - /// `ServiceEvent.old_payload` field (set by `DocsServiceImpl` on - /// update and delete; `None` for create). - async fn emit_docs(&self, cx: &SdkCallCx, event: ServiceEvent) -> Result<(), EmitError> { - let Some(op) = DocsEventOp::from_wire(event.op) else { - return Ok(()); - }; - let Some(collection) = event.collection.clone() else { - return Ok(()); - }; - let id = event.key.clone().unwrap_or_default(); - - let matches = self - .triggers - .list_matching_docs(cx.app_id, &collection, op) - .await - .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?; - - if matches.is_empty() { - return Ok(()); - } - - let trigger_event = TriggerEvent::Docs { - op, - collection, - id, - data: event.payload.clone(), - prev_data: event.old_payload.clone(), - }; - let payload = serde_json::to_value(&trigger_event) - .map_err(|e| EmitError::Rejected(format!("event serialize: {e}")))?; - - for m in matches { - self.outbox - .insert(NewOutboxRow { - app_id: cx.app_id, - source_kind: OutboxSourceKind::Docs, - trigger_id: Some(m.trigger_id), - script_id: Some(m.script_id), - reply_to: None, - payload: payload.clone(), - origin_principal: cx.principal.as_ref().map(|p| p.user_id), - trigger_depth: cx.trigger_depth.saturating_add(1), - root_execution_id: Some(cx.root_execution_id), - }) - .await - .map_err(|e| EmitError::Unavailable(format!("outbox insert: {e}")))?; - } - Ok(()) - } - - /// v1.1.5. Fan out a files mutation across matching files triggers. - /// The `ServiceEvent.payload` is the file **metadata** (never the - /// blob bytes); `old_payload` is the prior metadata (the deleted - /// row's metadata on delete). The `TriggerEvent::Files` carries the - /// metadata fields explicitly + `prev` for the change-data-capture - /// surface. - async fn emit_files(&self, cx: &SdkCallCx, event: ServiceEvent) -> Result<(), EmitError> { - let Some(op) = FilesEventOp::from_wire(event.op) else { - return Ok(()); - }; - let Some(collection) = event.collection.clone() else { - return Ok(()); - }; - // The payload is the FileMeta JSON the FilesServiceImpl emitted. - let Some(meta) = event - .payload - .clone() - .and_then(|v| serde_json::from_value::(v).ok()) - else { - return Ok(()); - }; - - let matches = self - .triggers - .list_matching_files(cx.app_id, &collection, op) - .await - .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?; - - if matches.is_empty() { - return Ok(()); - } - - let trigger_event = TriggerEvent::Files { - op, - collection, - id: meta.id.to_string(), - name: meta.name, - content_type: meta.content_type, - size: meta.size, - checksum: meta.checksum, - prev: event.old_payload.clone(), - }; - let payload = serde_json::to_value(&trigger_event) - .map_err(|e| EmitError::Rejected(format!("event serialize: {e}")))?; - - for m in matches { - self.outbox - .insert(NewOutboxRow { - app_id: cx.app_id, - source_kind: OutboxSourceKind::Files, - trigger_id: Some(m.trigger_id), - script_id: Some(m.script_id), - reply_to: None, - payload: payload.clone(), - origin_principal: cx.principal.as_ref().map(|p| p.user_id), - trigger_depth: cx.trigger_depth.saturating_add(1), - root_execution_id: Some(cx.root_execution_id), - }) - .await - .map_err(|e| EmitError::Unavailable(format!("outbox insert: {e}")))?; - } - Ok(()) - } +/// Fan a collection mutation out over the writing app's own + inherited +/// triggers. Pass `&mut *tx` to have the outbox rows commit with the write. +pub(crate) async fn emit_on( + conn: &mut PgConnection, + cx: &SdkCallCx, + event: &ServiceEvent, +) -> Result<(), EmitError> { + let Some(p) = plan(event) else { return Ok(()) }; + let matches = list_matching_on( + &mut *conn, + p.kind, + p.detail_table, + cx.app_id, + &p.collection, + p.op_str, + ) + .await + .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?; + write_outbox_rows(conn, cx, &p, matches).await +} + +/// §11.6: fan a SHARED-collection write out over the `shared = true` triggers on +/// the OWNING group. Each fires under the WRITER app (`cx.app_id`) — the same +/// "a group template runs under the firing app" model as an inherited trigger. +pub(crate) async fn emit_shared_on( + conn: &mut PgConnection, + cx: &SdkCallCx, + owning_group: GroupId, + event: &ServiceEvent, +) -> Result<(), EmitError> { + let Some(p) = plan(event) else { return Ok(()) }; + let matches = list_matching_shared_on( + &mut *conn, + p.kind, + p.detail_table, + owning_group, + &p.collection, + p.op_str, + ) + .await + .map_err(|e| EmitError::Unavailable(format!("trigger lookup: {e}")))?; + write_outbox_rows(conn, cx, &p, matches).await +} + +async fn write_outbox_rows( + conn: &mut PgConnection, + cx: &SdkCallCx, + p: &EventPlan, + matches: Vec, +) -> Result<(), EmitError> { + if matches.is_empty() { + return Ok(()); + } + // Serialize the originating event once so the dispatcher can hand it to the + // handler as `ctx.event` without joining back to the trigger row. + let payload = serde_json::to_value(&p.trigger_event) + .map_err(|e| EmitError::Rejected(format!("event serialize: {e}")))?; + for m in matches { + insert_on( + &mut *conn, + NewOutboxRow { + app_id: cx.app_id, + source_kind: p.source_kind, + trigger_id: Some(m.id.into()), + script_id: Some(m.script_id.into()), + reply_to: None, + payload: payload.clone(), + origin_principal: cx.principal.as_ref().map(|pr| pr.user_id), + trigger_depth: cx.trigger_depth.saturating_add(1), + root_execution_id: Some(cx.root_execution_id), + }, + ) + .await + .map_err(|e| EmitError::Unavailable(format!("outbox insert: {e}")))?; + } + Ok(()) } diff --git a/crates/manager-core/src/outbox_repo.rs b/crates/manager-core/src/outbox_repo.rs index 1810a45..7da669b 100644 --- a/crates/manager-core/src/outbox_repo.rs +++ b/crates/manager-core/src/outbox_repo.rs @@ -145,28 +145,39 @@ impl PostgresOutboxRepo { } } +/// Insert one outbox row on an arbitrary executor — a pooled connection, or a +/// `&mut *tx` so the row commits atomically with the data write that produced +/// it (the transactional outbox; see `outbox_event_emitter`). The `insert` +/// trait method delegates here, so the INSERT has exactly one home. +pub(crate) async fn insert_on<'c, E>(exec: E, row: NewOutboxRow) -> Result +where + E: sqlx::PgExecutor<'c>, +{ + let (id,): (Uuid,) = sqlx::query_as( + "INSERT INTO outbox ( \ + app_id, source_kind, trigger_id, script_id, reply_to, \ + payload, origin_principal, trigger_depth, root_execution_id \ + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) \ + RETURNING id", + ) + .bind(row.app_id.into_inner()) + .bind(row.source_kind.as_str()) + .bind(row.trigger_id.map(TriggerId::into_inner)) + .bind(row.script_id.map(ScriptId::into_inner)) + .bind(row.reply_to) + .bind(row.payload) + .bind(row.origin_principal.map(AdminUserId::into_inner)) + .bind(i32::try_from(row.trigger_depth).unwrap_or(0)) + .bind(row.root_execution_id.map(ExecutionId::into_inner)) + .fetch_one(exec) + .await?; + Ok(id) +} + #[async_trait] impl OutboxRepo for PostgresOutboxRepo { async fn insert(&self, row: NewOutboxRow) -> Result { - let (id,): (Uuid,) = sqlx::query_as( - "INSERT INTO outbox ( \ - app_id, source_kind, trigger_id, script_id, reply_to, \ - payload, origin_principal, trigger_depth, root_execution_id \ - ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) \ - RETURNING id", - ) - .bind(row.app_id.into_inner()) - .bind(row.source_kind.as_str()) - .bind(row.trigger_id.map(TriggerId::into_inner)) - .bind(row.script_id.map(ScriptId::into_inner)) - .bind(row.reply_to) - .bind(row.payload) - .bind(row.origin_principal.map(AdminUserId::into_inner)) - .bind(i32::try_from(row.trigger_depth).unwrap_or(0)) - .bind(row.root_execution_id.map(ExecutionId::into_inner)) - .fetch_one(&self.pool) - .await?; - Ok(id) + insert_on(&self.pool, row).await } async fn claim_due( diff --git a/crates/manager-core/src/trigger_repo.rs b/crates/manager-core/src/trigger_repo.rs index 435da7a..cc59e21 100644 --- a/crates/manager-core/src/trigger_repo.rs +++ b/crates/manager-core/src/trigger_repo.rs @@ -605,39 +605,95 @@ 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()) - } +// ---------------------------------------------------------------------------- +// Connection-scoped event matching +// +// The three collection-event kinds (kv / docs / files) match identically apart +// from the `triggers.kind` discriminator and their detail table, so both shapes +// — the per-app CHAIN walk and the §11.6 SHARED-collection lookup — live here +// once, parameterized on an executor. +// +// Taking an executor rather than `&self.pool` is what lets the transactional +// outbox work: `outbox_event_emitter` runs the match on the SAME connection +// (`&mut *tx`) as the data write it is fanning out, so the write and its outbox +// rows commit together or not at all. The `list_matching_*` trait methods below +// pass `&self.pool` and behave exactly as before. +// +// `kind` and `detail_table` are interpolated into the SQL, so they must stay +// hard-coded literals from the call sites below — never user data. +// ---------------------------------------------------------------------------- + +/// Glob + op filtering, in Rust rather than SQL. **Critical**: an empty `ops` +/// array means "any op", which a SQL `$op = ANY(ops)` predicate would silently +/// exclude. +fn filter_match_rows(rows: Vec, collection: &str, op_str: &str) -> Vec { + 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() +} + +/// Per-app match: the firing app's OWN triggers plus any inherited from an +/// ancestor group, minus the ones a suppression on its chain declines. +pub(crate) async fn list_matching_on<'c, E>( + exec: E, + kind: &str, + detail_table: &str, + app_id: AppId, + collection: &str, + op_str: &str, +) -> Result, TriggerRepoError> +where + E: sqlx::PgExecutor<'c>, +{ + 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 {detail_table} 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 = '{kind}' AND t.enabled = TRUE \ + AND t.shared = FALSE{TRIGGER_SUPPRESSION_ANTIJOIN}" + )) + .bind(app_id.into_inner()) + .fetch_all(exec) + .await?; + Ok(filter_match_rows(rows, collection, op_str)) +} + +/// §11.6 shared-collection match: enabled `shared = true` triggers on the +/// OWNING group. No chain and no suppression anti-join — a shared trigger is +/// declared on the group that owns the collection and fires under the writer. +pub(crate) async fn list_matching_shared_on<'c, E>( + exec: E, + kind: &str, + detail_table: &str, + owning_group: GroupId, + collection: &str, + op_str: &str, +) -> Result, TriggerRepoError> +where + E: sqlx::PgExecutor<'c>, +{ + 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(exec) + .await?; + Ok(filter_match_rows(rows, collection, op_str)) } /// Insert a trigger (parent row + per-kind detail) within an existing @@ -1483,48 +1539,16 @@ impl TriggerRepo for PostgresTriggerRepo { collection: &str, op: KvEventOp, ) -> Result, TriggerRepoError> { - // Fetch all enabled KV triggers for the app — glob matching - // 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(&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 \ - 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 \ - AND t.shared = FALSE{TRIGGER_SUPPRESSION_ANTIJOIN}" - )) - .bind(app_id.into_inner()) - .fetch_all(&self.pool) + let rows = list_matching_on( + &self.pool, + "kv", + "kv_trigger_details", + app_id, + collection, + op.as_str(), + ) .await?; - - let op_str = op.as_str(); - let mut out = Vec::new(); - for r in rows { - if !collection_matches(&r.collection_glob, collection) { - continue; - } - let any_op = r.ops.is_empty(); - if !any_op && !r.ops.iter().any(|o| o == op_str) { - continue; - } - out.push(KvTriggerMatch { - trigger_id: r.id.into(), - script_id: r.script_id.into(), - dispatch_mode: dispatch_from_str(&r.dispatch_mode), - retry_max_attempts: u32::try_from(r.retry_max_attempts).unwrap_or(3), - retry_backoff: BackoffShape::from_wire(&r.retry_backoff) - .unwrap_or(BackoffShape::Exponential), - retry_base_ms: u32::try_from(r.retry_base_ms).unwrap_or(1000), - registered_by_principal: r.registered_by_principal.into(), - }); - } - Ok(out) + Ok(rows.into_iter().map(KvMatchRow::into_kv).collect()) } async fn list_matching_docs( @@ -1533,49 +1557,16 @@ impl TriggerRepo for PostgresTriggerRepo { collection: &str, op: DocsEventOp, ) -> Result, TriggerRepoError> { - // Mirrors list_matching_kv: pull every enabled docs trigger, - // filter glob + ops in Rust. **Critical**: do NOT push the - // 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(&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 \ - 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 \ - AND t.shared = FALSE{TRIGGER_SUPPRESSION_ANTIJOIN}" - )) - .bind(app_id.into_inner()) - .fetch_all(&self.pool) + let rows = list_matching_on( + &self.pool, + "docs", + "docs_trigger_details", + app_id, + collection, + op.as_str(), + ) .await?; - - let op_str = op.as_str(); - let mut out = Vec::new(); - for r in rows { - if !collection_matches(&r.collection_glob, collection) { - continue; - } - let any_op = r.ops.is_empty(); - if !any_op && !r.ops.iter().any(|o| o == op_str) { - continue; - } - out.push(DocsTriggerMatch { - trigger_id: r.id.into(), - script_id: r.script_id.into(), - dispatch_mode: dispatch_from_str(&r.dispatch_mode), - retry_max_attempts: u32::try_from(r.retry_max_attempts).unwrap_or(3), - retry_backoff: BackoffShape::from_wire(&r.retry_backoff) - .unwrap_or(BackoffShape::Exponential), - retry_base_ms: u32::try_from(r.retry_base_ms).unwrap_or(1000), - registered_by_principal: r.registered_by_principal.into(), - }); - } - Ok(out) + Ok(rows.into_iter().map(KvMatchRow::into_docs).collect()) } async fn list_matching_files( @@ -1584,46 +1575,16 @@ impl TriggerRepo for PostgresTriggerRepo { collection: &str, op: FilesEventOp, ) -> 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(&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 \ - 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 \ - AND t.shared = FALSE{TRIGGER_SUPPRESSION_ANTIJOIN}" - )) - .bind(app_id.into_inner()) - .fetch_all(&self.pool) + let rows = list_matching_on( + &self.pool, + "files", + "files_trigger_details", + app_id, + collection, + op.as_str(), + ) .await?; - - let op_str = op.as_str(); - let mut out = Vec::new(); - for r in rows { - if !collection_matches(&r.collection_glob, collection) { - continue; - } - let any_op = r.ops.is_empty(); - if !any_op && !r.ops.iter().any(|o| o == op_str) { - continue; - } - out.push(FilesTriggerMatch { - trigger_id: r.id.into(), - script_id: r.script_id.into(), - dispatch_mode: dispatch_from_str(&r.dispatch_mode), - retry_max_attempts: u32::try_from(r.retry_max_attempts).unwrap_or(3), - retry_backoff: BackoffShape::from_wire(&r.retry_backoff) - .unwrap_or(BackoffShape::Exponential), - retry_base_ms: u32::try_from(r.retry_base_ms).unwrap_or(1000), - registered_by_principal: r.registered_by_principal.into(), - }); - } - Ok(out) + Ok(rows.into_iter().map(KvMatchRow::into_files).collect()) } async fn list_matching_shared_kv( @@ -1632,15 +1593,15 @@ impl TriggerRepo for PostgresTriggerRepo { collection: &str, op: KvEventOp, ) -> Result, TriggerRepoError> { - let rows = self - .shared_match_rows( - "kv", - "kv_trigger_details", - owning_group, - collection, - op.as_str(), - ) - .await?; + let rows = list_matching_shared_on( + &self.pool, + "kv", + "kv_trigger_details", + owning_group, + collection, + op.as_str(), + ) + .await?; Ok(rows.into_iter().map(KvMatchRow::into_kv).collect()) } @@ -1650,15 +1611,15 @@ impl TriggerRepo for PostgresTriggerRepo { collection: &str, op: DocsEventOp, ) -> Result, TriggerRepoError> { - let rows = self - .shared_match_rows( - "docs", - "docs_trigger_details", - owning_group, - collection, - op.as_str(), - ) - .await?; + let rows = list_matching_shared_on( + &self.pool, + "docs", + "docs_trigger_details", + owning_group, + collection, + op.as_str(), + ) + .await?; Ok(rows.into_iter().map(KvMatchRow::into_docs).collect()) } @@ -1668,15 +1629,15 @@ impl TriggerRepo for PostgresTriggerRepo { collection: &str, op: FilesEventOp, ) -> Result, TriggerRepoError> { - let rows = self - .shared_match_rows( - "files", - "files_trigger_details", - owning_group, - collection, - op.as_str(), - ) - .await?; + let rows = list_matching_shared_on( + &self.pool, + "files", + "files_trigger_details", + owning_group, + collection, + op.as_str(), + ) + .await?; Ok(rows.into_iter().map(KvMatchRow::into_files).collect()) } @@ -2169,9 +2130,9 @@ struct DlDetailRow { } #[derive(sqlx::FromRow)] -struct KvMatchRow { - id: Uuid, - script_id: Uuid, +pub(crate) struct KvMatchRow { + pub(crate) id: Uuid, + pub(crate) script_id: Uuid, dispatch_mode: String, retry_max_attempts: i32, retry_backoff: String, diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index fe9a114..ff041b0 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -169,10 +169,7 @@ pub async fn build_app( // dispatcher. let kv_repo = Arc::new(PostgresKvRepo::new(pool.clone())); let docs_repo = Arc::new(PostgresDocsRepo::new(pool.clone())); - let events: Arc = Arc::new(OutboxEventEmitter::new( - trigger_repo.clone(), - outbox_repo.clone(), - )); + let events: Arc = Arc::new(OutboxEventEmitter::new(pool.clone())); let kv: Arc = Arc::new(KvServiceImpl::with_max_value_bytes( kv_repo.clone(), authz.clone(),