4d2eed4e81
Completes §4.5 templates (M4a shipped routes). A group declares a trigger once;
the tree apply fans it out into one concrete per-app_id trigger on each
descendant app, reusing the M4a expansion engine over the trigger insert path.
- Migration 0054: `trigger_templates` table (group-owned; the whole
BundleTrigger wire object stored as `spec` JSONB, placeholders unresolved) +
`triggers.from_template` provenance column + index.
- `template_repo.rs`: trigger-template CRUD + chain-load.
- Manifest: `[group]` `[[trigger_templates]]` (flat: name, kind, script, + kind
params); build_bundle emits them; rejected on an app node.
- Apply: group nodes reconcile trigger-template rows; tree Phase B expands each
chain trigger template into a concrete trigger per descendant app —
placeholders resolved in every spec string leaf, then the typed trigger is
rebuilt and inserted through the shared `insert_bundle_trigger_tx` (email
secrets resolved + re-sealed per recipient app). Idempotent via from_template
(one trigger per template per app; semantic-identity compare → no-op /
delete+recreate / reap). Collision with a hand-declared trigger or between
templates is a hard error. Each resolved trigger is re-validated with the
per-kind shape check (cron/queue/etc.) so a {var:}-injected bad
schedule/timeout can't slip in.
- Authz: declaring → GroupScriptsWrite; receiving → AppManageTriggers, plus
AppSecretsRead for email-trigger recipients (parity with hand-declared email).
- Plan/token/report extended for trigger templates; blast radius covers both.
- Tests: tests/templates.rs trigger fan-out (kv + {app_slug}, stable-ids,
prune-reap); resolve_placeholders_in_json unit test; schema golden reblessed.
Reviewed (no CRITICAL; isolation + per-app email-secret sealing verified
sound). Closed a HIGH validation-parity gap (re-validate resolved triggers) and
a MEDIUM authz gap (AppSecretsRead for email expansions). v1.2 Hierarchies
template work (M4) complete.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1904 lines
68 KiB
Rust
1904 lines
68 KiB
Rust
//! `TriggerRepo` — CRUD over the `triggers` parent + per-kind detail
|
|
//! tables. The admin endpoints (commit 4) sit on top of this; the
|
|
//! dispatcher (commit 5) reads `list_matching_*` to fan out events to
|
|
//! handler scripts.
|
|
|
|
use async_trait::async_trait;
|
|
use chrono::{DateTime, Utc};
|
|
use futures::stream::{self, TryStreamExt};
|
|
use picloud_shared::{
|
|
AdminUserId, AppId, DocsEventOp, FilesEventOp, KvEventOp, ScriptId, TriggerId,
|
|
};
|
|
use serde::{Deserialize, Serialize};
|
|
use sqlx::PgPool;
|
|
use uuid::Uuid;
|
|
|
|
use crate::trigger_config::BackoffShape;
|
|
|
|
#[derive(Debug, thiserror::Error)]
|
|
pub enum TriggerRepoError {
|
|
#[error("database error: {0}")]
|
|
Db(#[from] sqlx::Error),
|
|
|
|
#[error("trigger not found: {0}")]
|
|
NotFound(TriggerId),
|
|
|
|
#[error("invalid trigger payload: {0}")]
|
|
Invalid(String),
|
|
}
|
|
|
|
/// Parent-table row plus the per-kind detail merged in. Serialized
|
|
/// back to admin clients via the JSON API.
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct Trigger {
|
|
pub id: TriggerId,
|
|
pub app_id: AppId,
|
|
pub script_id: ScriptId,
|
|
/// §4.5 per-app trigger identifier; the manifest merge/upsert key.
|
|
pub name: String,
|
|
pub kind: TriggerKind,
|
|
pub enabled: bool,
|
|
pub dispatch_mode: TriggerDispatchMode,
|
|
pub retry_max_attempts: u32,
|
|
pub retry_backoff: BackoffShape,
|
|
pub retry_base_ms: u32,
|
|
pub registered_by_principal: AdminUserId,
|
|
pub created_at: DateTime<Utc>,
|
|
pub updated_at: DateTime<Utc>,
|
|
pub details: TriggerDetails,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
|
#[serde(rename_all = "snake_case")]
|
|
pub enum TriggerKind {
|
|
Kv,
|
|
Docs,
|
|
DeadLetter,
|
|
/// v1.1.4.
|
|
Cron,
|
|
/// v1.1.5.
|
|
Files,
|
|
/// v1.1.5.
|
|
Pubsub,
|
|
/// v1.1.7. Inbound email via the webhook receiver.
|
|
Email,
|
|
/// v1.1.9. Durable queue consumer — exactly one consumer per
|
|
/// `(app_id, queue_name)`.
|
|
Queue,
|
|
}
|
|
|
|
impl TriggerKind {
|
|
#[must_use]
|
|
pub const fn as_str(self) -> &'static str {
|
|
match self {
|
|
Self::Kv => "kv",
|
|
Self::Docs => "docs",
|
|
Self::DeadLetter => "dead_letter",
|
|
Self::Cron => "cron",
|
|
Self::Files => "files",
|
|
Self::Pubsub => "pubsub",
|
|
Self::Email => "email",
|
|
Self::Queue => "queue",
|
|
}
|
|
}
|
|
|
|
#[must_use]
|
|
pub fn from_wire(s: &str) -> Option<Self> {
|
|
match s {
|
|
"kv" => Some(Self::Kv),
|
|
"docs" => Some(Self::Docs),
|
|
"dead_letter" => Some(Self::DeadLetter),
|
|
"cron" => Some(Self::Cron),
|
|
"files" => Some(Self::Files),
|
|
"pubsub" => Some(Self::Pubsub),
|
|
"email" => Some(Self::Email),
|
|
"queue" => Some(Self::Queue),
|
|
_ => None,
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
|
#[serde(rename_all = "snake_case")]
|
|
pub enum TriggerDispatchMode {
|
|
Sync,
|
|
Async,
|
|
}
|
|
|
|
impl TriggerDispatchMode {
|
|
#[must_use]
|
|
pub const fn as_str(self) -> &'static str {
|
|
match self {
|
|
Self::Sync => "sync",
|
|
Self::Async => "async",
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
#[serde(tag = "kind", rename_all = "snake_case")]
|
|
pub enum TriggerDetails {
|
|
Kv {
|
|
collection_glob: String,
|
|
ops: Vec<KvEventOp>,
|
|
},
|
|
Docs {
|
|
collection_glob: String,
|
|
ops: Vec<DocsEventOp>,
|
|
},
|
|
DeadLetter {
|
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
source_filter: Option<String>,
|
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
trigger_id_filter: Option<TriggerId>,
|
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
script_id_filter: Option<ScriptId>,
|
|
},
|
|
/// v1.1.4. The 6-field cron schedule + IANA timezone the trigger
|
|
/// fires on, plus the last enqueue time (for dashboard display).
|
|
Cron {
|
|
schedule: String,
|
|
timezone: String,
|
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
last_fired_at: Option<DateTime<Utc>>,
|
|
},
|
|
/// v1.1.5. Same shape as KV/docs: a collection glob + op subset.
|
|
Files {
|
|
collection_glob: String,
|
|
ops: Vec<FilesEventOp>,
|
|
},
|
|
/// v1.1.5. A topic pattern: exact, `<prefix>.*`, or `*`.
|
|
Pubsub { topic_pattern: String },
|
|
/// v1.1.7. Inbound email. The HMAC `inbound_secret` is never
|
|
/// surfaced (it's encrypted at rest); we expose only whether one is
|
|
/// configured so the admin UI can show "signed" vs "unsigned".
|
|
Email { has_inbound_secret: bool },
|
|
/// v1.1.9. Queue consumer. The queue name is the per-queue grouping
|
|
/// key; visibility timeout is the per-trigger override; last_fired_at
|
|
/// is updated by the dispatcher when a message is claimed.
|
|
Queue {
|
|
queue_name: String,
|
|
visibility_timeout_secs: u32,
|
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
last_fired_at: Option<DateTime<Utc>>,
|
|
},
|
|
}
|
|
|
|
/// Create payload for a KV trigger. Defaults applied at the admin
|
|
/// layer (uses `TriggerConfig::from_env` to fill retry settings if
|
|
/// the request omitted them — keeps the row auditable).
|
|
#[derive(Debug, Clone)]
|
|
pub struct CreateKvTrigger {
|
|
pub script_id: ScriptId,
|
|
pub collection_glob: String,
|
|
pub ops: Vec<KvEventOp>,
|
|
pub dispatch_mode: TriggerDispatchMode,
|
|
pub retry_max_attempts: u32,
|
|
pub retry_backoff: BackoffShape,
|
|
pub retry_base_ms: u32,
|
|
pub registered_by_principal: AdminUserId,
|
|
}
|
|
|
|
/// Create payload for a docs trigger (v1.1.2). Same shape as KV with
|
|
/// `DocsEventOp` ops instead of `KvEventOp`.
|
|
#[derive(Debug, Clone)]
|
|
pub struct CreateDocsTrigger {
|
|
pub script_id: ScriptId,
|
|
pub collection_glob: String,
|
|
pub ops: Vec<DocsEventOp>,
|
|
pub dispatch_mode: TriggerDispatchMode,
|
|
pub retry_max_attempts: u32,
|
|
pub retry_backoff: BackoffShape,
|
|
pub retry_base_ms: u32,
|
|
pub registered_by_principal: AdminUserId,
|
|
}
|
|
|
|
#[derive(Debug, Clone)]
|
|
pub struct CreateDeadLetterTrigger {
|
|
pub script_id: ScriptId,
|
|
pub source_filter: Option<String>,
|
|
pub trigger_id_filter: Option<TriggerId>,
|
|
pub script_id_filter: Option<ScriptId>,
|
|
pub registered_by_principal: AdminUserId,
|
|
}
|
|
|
|
/// Create payload for a cron trigger (v1.1.4). `schedule` is a 6-field
|
|
/// cron expression and `timezone` an IANA name; both are validated
|
|
/// (by the admin endpoint and defensively by the repo) before insert.
|
|
#[derive(Debug, Clone)]
|
|
pub struct CreateCronTrigger {
|
|
pub script_id: ScriptId,
|
|
pub schedule: String,
|
|
pub timezone: String,
|
|
pub dispatch_mode: TriggerDispatchMode,
|
|
pub retry_max_attempts: u32,
|
|
pub retry_backoff: BackoffShape,
|
|
pub retry_base_ms: u32,
|
|
pub registered_by_principal: AdminUserId,
|
|
}
|
|
|
|
/// Create payload for a files trigger (v1.1.5). Same shape as KV with
|
|
/// `FilesEventOp` ops.
|
|
#[derive(Debug, Clone)]
|
|
pub struct CreateFilesTrigger {
|
|
pub script_id: ScriptId,
|
|
pub collection_glob: String,
|
|
pub ops: Vec<FilesEventOp>,
|
|
pub dispatch_mode: TriggerDispatchMode,
|
|
pub retry_max_attempts: u32,
|
|
pub retry_backoff: BackoffShape,
|
|
pub retry_base_ms: u32,
|
|
pub registered_by_principal: AdminUserId,
|
|
}
|
|
|
|
/// One match for the dispatcher's files trigger fan-out lookup
|
|
/// (v1.1.5). Same shape as `KvTriggerMatch`.
|
|
#[derive(Debug, Clone)]
|
|
pub struct FilesTriggerMatch {
|
|
pub trigger_id: TriggerId,
|
|
pub script_id: ScriptId,
|
|
pub dispatch_mode: TriggerDispatchMode,
|
|
pub retry_max_attempts: u32,
|
|
pub retry_backoff: BackoffShape,
|
|
pub retry_base_ms: u32,
|
|
pub registered_by_principal: AdminUserId,
|
|
}
|
|
|
|
/// Create payload for a pubsub trigger (v1.1.5). `topic_pattern` is
|
|
/// validated (exact / `<prefix>.*` / `*`) before insert.
|
|
#[derive(Debug, Clone)]
|
|
pub struct CreatePubsubTrigger {
|
|
pub script_id: ScriptId,
|
|
pub topic_pattern: String,
|
|
pub dispatch_mode: TriggerDispatchMode,
|
|
pub retry_max_attempts: u32,
|
|
pub retry_backoff: BackoffShape,
|
|
pub retry_base_ms: u32,
|
|
pub registered_by_principal: AdminUserId,
|
|
}
|
|
|
|
/// Create payload for an email trigger (v1.1.7). `inbound_secret_*` is
|
|
/// the already-encrypted HMAC secret (sealed by the admin layer with the
|
|
/// process master key) or `None` for an unsigned trigger.
|
|
#[derive(Debug, Clone)]
|
|
pub struct CreateEmailTrigger {
|
|
pub script_id: ScriptId,
|
|
pub inbound_secret_encrypted: Option<Vec<u8>>,
|
|
pub inbound_secret_nonce: Option<Vec<u8>>,
|
|
pub registered_by_principal: AdminUserId,
|
|
}
|
|
|
|
/// Create payload for a queue:receive trigger (v1.1.9). One consumer
|
|
/// per `(app_id, queue_name)` is enforced inside `create_queue_trigger`
|
|
/// via a pg_advisory_xact_lock + SELECT.
|
|
#[derive(Debug, Clone)]
|
|
pub struct CreateQueueTrigger {
|
|
pub script_id: ScriptId,
|
|
pub queue_name: String,
|
|
pub visibility_timeout_secs: u32,
|
|
pub dispatch_mode: TriggerDispatchMode,
|
|
pub retry_max_attempts: u32,
|
|
pub retry_backoff: BackoffShape,
|
|
pub retry_base_ms: u32,
|
|
pub registered_by_principal: AdminUserId,
|
|
}
|
|
|
|
/// One consumer the dispatcher's queue arm needs to scan messages for.
|
|
/// Pulled by `list_active_queue_consumers()` per tick.
|
|
#[derive(Debug, Clone)]
|
|
pub struct ActiveQueueConsumer {
|
|
pub trigger_id: TriggerId,
|
|
pub app_id: AppId,
|
|
pub script_id: ScriptId,
|
|
pub queue_name: String,
|
|
pub visibility_timeout_secs: u32,
|
|
pub retry_max_attempts: u32,
|
|
pub retry_backoff: BackoffShape,
|
|
pub retry_base_ms: u32,
|
|
pub registered_by_principal: AdminUserId,
|
|
}
|
|
|
|
/// What the inbound-email webhook receiver needs to verify + dispatch a
|
|
/// POST. Returned by `email_inbound_target`; `None` when the trigger
|
|
/// doesn't exist or isn't `kind = 'email'`.
|
|
#[derive(Debug, Clone)]
|
|
pub struct EmailInboundTarget {
|
|
pub app_id: AppId,
|
|
pub script_id: ScriptId,
|
|
pub enabled: bool,
|
|
pub dispatch_mode: TriggerDispatchMode,
|
|
pub registered_by_principal: AdminUserId,
|
|
/// Encrypted HMAC secret + nonce; both `None` for an unsigned
|
|
/// trigger (accepts any POST).
|
|
pub inbound_secret_encrypted: Option<Vec<u8>>,
|
|
pub inbound_secret_nonce: Option<Vec<u8>>,
|
|
}
|
|
|
|
/// One match for the dispatcher's "which KV triggers fire on this
|
|
/// event" lookup. Carries everything the dispatcher needs to construct
|
|
/// the outbox row.
|
|
#[derive(Debug, Clone)]
|
|
pub struct KvTriggerMatch {
|
|
pub trigger_id: TriggerId,
|
|
pub script_id: ScriptId,
|
|
pub dispatch_mode: TriggerDispatchMode,
|
|
pub retry_max_attempts: u32,
|
|
pub retry_backoff: BackoffShape,
|
|
pub retry_base_ms: u32,
|
|
pub registered_by_principal: AdminUserId,
|
|
}
|
|
|
|
/// One match for the dispatcher's docs trigger fan-out lookup (v1.1.2).
|
|
/// Same shape as `KvTriggerMatch`.
|
|
#[derive(Debug, Clone)]
|
|
pub struct DocsTriggerMatch {
|
|
pub trigger_id: TriggerId,
|
|
pub script_id: ScriptId,
|
|
pub dispatch_mode: TriggerDispatchMode,
|
|
pub retry_max_attempts: u32,
|
|
pub retry_backoff: BackoffShape,
|
|
pub retry_base_ms: u32,
|
|
pub registered_by_principal: AdminUserId,
|
|
}
|
|
|
|
/// One match for the dispatcher's "which dead-letter triggers fire
|
|
/// on this dead-letter row" lookup.
|
|
#[derive(Debug, Clone)]
|
|
pub struct DeadLetterTriggerMatch {
|
|
pub trigger_id: TriggerId,
|
|
pub script_id: ScriptId,
|
|
pub dispatch_mode: TriggerDispatchMode,
|
|
pub registered_by_principal: AdminUserId,
|
|
}
|
|
|
|
#[async_trait]
|
|
pub trait TriggerRepo: Send + Sync {
|
|
async fn create_kv_trigger(
|
|
&self,
|
|
app_id: AppId,
|
|
req: CreateKvTrigger,
|
|
) -> Result<Trigger, TriggerRepoError>;
|
|
|
|
/// v1.1.2.
|
|
async fn create_docs_trigger(
|
|
&self,
|
|
app_id: AppId,
|
|
req: CreateDocsTrigger,
|
|
) -> Result<Trigger, TriggerRepoError>;
|
|
|
|
async fn create_dead_letter_trigger(
|
|
&self,
|
|
app_id: AppId,
|
|
req: CreateDeadLetterTrigger,
|
|
) -> Result<Trigger, TriggerRepoError>;
|
|
|
|
/// v1.1.4. `schedule` + `timezone` are validated before insert; an
|
|
/// invalid expression or unknown IANA name returns
|
|
/// `TriggerRepoError::Invalid`.
|
|
async fn create_cron_trigger(
|
|
&self,
|
|
app_id: AppId,
|
|
req: CreateCronTrigger,
|
|
) -> Result<Trigger, TriggerRepoError>;
|
|
|
|
/// v1.1.5.
|
|
async fn create_files_trigger(
|
|
&self,
|
|
app_id: AppId,
|
|
req: CreateFilesTrigger,
|
|
) -> Result<Trigger, TriggerRepoError>;
|
|
|
|
/// v1.1.5. `topic_pattern` is validated before insert.
|
|
async fn create_pubsub_trigger(
|
|
&self,
|
|
app_id: AppId,
|
|
req: CreatePubsubTrigger,
|
|
) -> Result<Trigger, TriggerRepoError>;
|
|
|
|
/// v1.1.7. Inbound email trigger. The `inbound_secret` is stored
|
|
/// already-encrypted (the admin layer seals it).
|
|
async fn create_email_trigger(
|
|
&self,
|
|
app_id: AppId,
|
|
req: CreateEmailTrigger,
|
|
) -> Result<Trigger, TriggerRepoError>;
|
|
|
|
/// v1.1.7. The webhook receiver's hot-path lookup: resolve a
|
|
/// `kind = 'email'` trigger to its app, handler script, dispatch
|
|
/// mode, and (encrypted) HMAC secret. Returns `None` when the
|
|
/// trigger doesn't exist or isn't an email trigger.
|
|
async fn email_inbound_target(
|
|
&self,
|
|
trigger_id: TriggerId,
|
|
) -> Result<Option<EmailInboundTarget>, TriggerRepoError>;
|
|
|
|
async fn list_for_app(&self, app_id: AppId) -> Result<Vec<Trigger>, TriggerRepoError>;
|
|
|
|
async fn get(&self, id: TriggerId) -> Result<Option<Trigger>, TriggerRepoError>;
|
|
|
|
async fn delete(&self, id: TriggerId) -> Result<bool, TriggerRepoError>;
|
|
|
|
/// Dispatcher hot path: find every enabled KV trigger in `app_id`
|
|
/// whose `collection_glob` matches `collection` and whose `ops`
|
|
/// covers `op`. Glob matching done in Rust (the column is plain
|
|
/// TEXT, the matcher applies "*"/"prefix:*" semantics).
|
|
async fn list_matching_kv(
|
|
&self,
|
|
app_id: AppId,
|
|
collection: &str,
|
|
op: KvEventOp,
|
|
) -> Result<Vec<KvTriggerMatch>, TriggerRepoError>;
|
|
|
|
/// Dispatcher hot path for docs fan-out (v1.1.2). Mirrors the KV
|
|
/// fan-out logic: pull every enabled docs trigger, filter glob +
|
|
/// ops in Rust (empty ops array means "any op").
|
|
async fn list_matching_docs(
|
|
&self,
|
|
app_id: AppId,
|
|
collection: &str,
|
|
op: DocsEventOp,
|
|
) -> Result<Vec<DocsTriggerMatch>, TriggerRepoError>;
|
|
|
|
/// Dispatcher hot path for files fan-out (v1.1.5). Mirrors the KV
|
|
/// fan-out logic: pull every enabled files trigger, filter glob +
|
|
/// ops in Rust (empty ops array means "any op").
|
|
async fn list_matching_files(
|
|
&self,
|
|
app_id: AppId,
|
|
collection: &str,
|
|
op: FilesEventOp,
|
|
) -> Result<Vec<FilesTriggerMatch>, TriggerRepoError>;
|
|
|
|
/// 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".
|
|
async fn list_matching_dead_letter(
|
|
&self,
|
|
app_id: AppId,
|
|
source: &str,
|
|
trigger_id: Option<TriggerId>,
|
|
script_id: Option<ScriptId>,
|
|
) -> Result<Vec<DeadLetterTriggerMatch>, TriggerRepoError>;
|
|
|
|
/// v1.1.9. Create a queue:receive trigger. Enforces exactly one
|
|
/// consumer per `(app_id, queue_name)` via `pg_advisory_xact_lock`
|
|
/// + SELECT-then-INSERT (a partial unique index across the parent
|
|
/// app_id + detail queue_name can't be enforced in a single index).
|
|
/// Returns `TriggerRepoError::Invalid` if a consumer already exists.
|
|
async fn create_queue_trigger(
|
|
&self,
|
|
app_id: AppId,
|
|
req: CreateQueueTrigger,
|
|
) -> Result<Trigger, TriggerRepoError>;
|
|
|
|
/// v1.1.9. Dispatcher's queue arm: list every enabled queue:receive
|
|
/// trigger across all apps. The dispatcher walks these per tick and
|
|
/// attempts a claim per queue. Bounded by the number of registered
|
|
/// consumers, which is small.
|
|
async fn list_active_queue_consumers(
|
|
&self,
|
|
) -> Result<Vec<ActiveQueueConsumer>, TriggerRepoError>;
|
|
|
|
/// v1.1.9. Update `last_fired_at` on a queue trigger after a
|
|
/// successful claim. Best-effort — a failure is logged but doesn't
|
|
/// fail the dispatch.
|
|
async fn touch_queue_trigger_last_fired_at(
|
|
&self,
|
|
trigger_id: TriggerId,
|
|
at: DateTime<Utc>,
|
|
) -> Result<(), TriggerRepoError>;
|
|
}
|
|
|
|
// ----------------------------------------------------------------------------
|
|
// Postgres impl
|
|
// ----------------------------------------------------------------------------
|
|
|
|
pub struct PostgresTriggerRepo {
|
|
pool: PgPool,
|
|
}
|
|
|
|
impl PostgresTriggerRepo {
|
|
#[must_use]
|
|
pub fn new(pool: PgPool) -> Self {
|
|
Self { pool }
|
|
}
|
|
}
|
|
|
|
/// Insert a trigger (parent row + per-kind detail) within an existing
|
|
/// transaction — used by the declarative `apply` engine. Supports the
|
|
/// five settled kinds; `email`/`queue`/`dead_letter` have their own
|
|
/// create paths and are rejected here.
|
|
#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
|
|
pub(crate) async fn insert_trigger_tx(
|
|
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
|
app_id: AppId,
|
|
script_id: ScriptId,
|
|
registered_by: AdminUserId,
|
|
dispatch_mode: TriggerDispatchMode,
|
|
retry_max_attempts: u32,
|
|
retry_backoff: BackoffShape,
|
|
retry_base_ms: u32,
|
|
details: &TriggerDetails,
|
|
from_template: Option<Uuid>,
|
|
) -> Result<TriggerId, TriggerRepoError> {
|
|
let kind = match details {
|
|
TriggerDetails::Kv { .. } => "kv",
|
|
TriggerDetails::Docs { .. } => "docs",
|
|
TriggerDetails::Files { .. } => "files",
|
|
TriggerDetails::Cron { .. } => "cron",
|
|
TriggerDetails::Pubsub { .. } => "pubsub",
|
|
TriggerDetails::Queue { .. } => "queue",
|
|
TriggerDetails::DeadLetter { .. } | TriggerDetails::Email { .. } => {
|
|
return Err(TriggerRepoError::Invalid(
|
|
"trigger kind not supported by declarative apply".into(),
|
|
));
|
|
}
|
|
};
|
|
// Queue: enforce the one-consumer-per-(app_id, queue_name) invariant —
|
|
// the same advisory-lock + existence guard the interactive
|
|
// `create_queue_trigger` uses. Without this, a concurrent apply +
|
|
// interactive create on disjoint locks could double-register a queue
|
|
// consumer (there is no DB unique constraint backing the invariant).
|
|
if let TriggerDetails::Queue { queue_name, .. } = details {
|
|
sqlx::query("SELECT pg_advisory_xact_lock($1)")
|
|
.bind(advisory_lock_key(app_id, queue_name))
|
|
.execute(&mut **tx)
|
|
.await?;
|
|
let existing: Option<(Uuid,)> = sqlx::query_as(
|
|
"SELECT t.id FROM triggers t \
|
|
JOIN queue_trigger_details d ON d.trigger_id = t.id \
|
|
WHERE t.app_id = $1 AND t.kind = 'queue' AND d.queue_name = $2",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.bind(queue_name)
|
|
.fetch_optional(&mut **tx)
|
|
.await?;
|
|
if existing.is_some() {
|
|
return Err(TriggerRepoError::Invalid(format!(
|
|
"queue '{queue_name}' already has a consumer trigger; remove the existing one first"
|
|
)));
|
|
}
|
|
}
|
|
let row: (Uuid,) = sqlx::query_as(
|
|
"INSERT INTO triggers ( \
|
|
app_id, script_id, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal, from_template \
|
|
) VALUES ($1, $2, $3, TRUE, $4, $5, $6, $7, $8, $9) RETURNING id",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.bind(script_id.into_inner())
|
|
.bind(kind)
|
|
.bind(dispatch_mode.as_str())
|
|
.bind(i32::try_from(retry_max_attempts).unwrap_or(3))
|
|
.bind(retry_backoff.as_str())
|
|
.bind(i32::try_from(retry_base_ms).unwrap_or(1000))
|
|
.bind(registered_by.into_inner())
|
|
.bind(from_template)
|
|
.fetch_one(&mut **tx)
|
|
.await?;
|
|
let tid = row.0;
|
|
|
|
match details {
|
|
TriggerDetails::Kv {
|
|
collection_glob,
|
|
ops,
|
|
} => {
|
|
let ops_str: Vec<String> = ops.iter().map(|o| o.as_str().to_string()).collect();
|
|
sqlx::query(
|
|
"INSERT INTO kv_trigger_details (trigger_id, collection_glob, ops) \
|
|
VALUES ($1, $2, $3)",
|
|
)
|
|
.bind(tid)
|
|
.bind(collection_glob)
|
|
.bind(&ops_str)
|
|
.execute(&mut **tx)
|
|
.await?;
|
|
}
|
|
TriggerDetails::Docs {
|
|
collection_glob,
|
|
ops,
|
|
} => {
|
|
let ops_str: Vec<String> = ops.iter().map(|o| o.as_str().to_string()).collect();
|
|
sqlx::query(
|
|
"INSERT INTO docs_trigger_details (trigger_id, collection_glob, ops) \
|
|
VALUES ($1, $2, $3)",
|
|
)
|
|
.bind(tid)
|
|
.bind(collection_glob)
|
|
.bind(&ops_str)
|
|
.execute(&mut **tx)
|
|
.await?;
|
|
}
|
|
TriggerDetails::Files {
|
|
collection_glob,
|
|
ops,
|
|
} => {
|
|
let ops_str: Vec<String> = ops.iter().map(|o| o.as_str().to_string()).collect();
|
|
sqlx::query(
|
|
"INSERT INTO files_trigger_details (trigger_id, collection_glob, ops) \
|
|
VALUES ($1, $2, $3)",
|
|
)
|
|
.bind(tid)
|
|
.bind(collection_glob)
|
|
.bind(&ops_str)
|
|
.execute(&mut **tx)
|
|
.await?;
|
|
}
|
|
TriggerDetails::Cron {
|
|
schedule, timezone, ..
|
|
} => {
|
|
sqlx::query(
|
|
"INSERT INTO cron_trigger_details (trigger_id, schedule, timezone) \
|
|
VALUES ($1, $2, $3)",
|
|
)
|
|
.bind(tid)
|
|
.bind(schedule)
|
|
.bind(timezone)
|
|
.execute(&mut **tx)
|
|
.await?;
|
|
}
|
|
TriggerDetails::Pubsub { topic_pattern } => {
|
|
sqlx::query(
|
|
"INSERT INTO pubsub_trigger_details (trigger_id, topic_pattern) VALUES ($1, $2)",
|
|
)
|
|
.bind(tid)
|
|
.bind(topic_pattern)
|
|
.execute(&mut **tx)
|
|
.await?;
|
|
}
|
|
TriggerDetails::Queue {
|
|
queue_name,
|
|
visibility_timeout_secs,
|
|
..
|
|
} => {
|
|
sqlx::query(
|
|
"INSERT INTO queue_trigger_details \
|
|
(trigger_id, queue_name, visibility_timeout_secs) \
|
|
VALUES ($1, $2, $3)",
|
|
)
|
|
.bind(tid)
|
|
.bind(queue_name)
|
|
.bind(i32::try_from(*visibility_timeout_secs).unwrap_or(30))
|
|
.execute(&mut **tx)
|
|
.await?;
|
|
}
|
|
TriggerDetails::DeadLetter { .. } | TriggerDetails::Email { .. } => {
|
|
unreachable!("guarded above")
|
|
}
|
|
}
|
|
Ok(tid.into())
|
|
}
|
|
|
|
/// Insert an email trigger within a transaction. The inbound HMAC secret
|
|
/// is sealed by the apply engine (resolved from the app's secret store);
|
|
/// this writes the ciphertext. Parent retry settings match the
|
|
/// interactive `create_email_trigger` path (async, 3, exponential, 1000).
|
|
pub(crate) async fn insert_email_trigger_tx(
|
|
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
|
app_id: AppId,
|
|
script_id: ScriptId,
|
|
registered_by: AdminUserId,
|
|
inbound_secret_encrypted: &[u8],
|
|
inbound_secret_nonce: &[u8],
|
|
from_template: Option<Uuid>,
|
|
) -> Result<TriggerId, TriggerRepoError> {
|
|
let row: (Uuid,) = sqlx::query_as(
|
|
"INSERT INTO triggers ( \
|
|
app_id, script_id, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal, from_template \
|
|
) VALUES ($1, $2, 'email', TRUE, 'async', 3, 'exponential', 1000, $3, $4) RETURNING id",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.bind(script_id.into_inner())
|
|
.bind(registered_by.into_inner())
|
|
.bind(from_template)
|
|
.fetch_one(&mut **tx)
|
|
.await?;
|
|
sqlx::query(
|
|
"INSERT INTO email_trigger_details \
|
|
(trigger_id, inbound_secret_encrypted, inbound_secret_nonce) \
|
|
VALUES ($1, $2, $3)",
|
|
)
|
|
.bind(row.0)
|
|
.bind(inbound_secret_encrypted)
|
|
.bind(inbound_secret_nonce)
|
|
.execute(&mut **tx)
|
|
.await?;
|
|
Ok(row.0.into())
|
|
}
|
|
|
|
/// Delete a trigger by id within an existing transaction (its detail row
|
|
/// cascades via the FK). Used by `apply --prune`.
|
|
pub(crate) async fn delete_trigger_tx(
|
|
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
|
id: TriggerId,
|
|
) -> Result<(), TriggerRepoError> {
|
|
sqlx::query("DELETE FROM triggers WHERE id = $1")
|
|
.bind(id.into_inner())
|
|
.execute(&mut **tx)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
#[async_trait]
|
|
impl TriggerRepo for PostgresTriggerRepo {
|
|
async fn create_kv_trigger(
|
|
&self,
|
|
app_id: AppId,
|
|
req: CreateKvTrigger,
|
|
) -> Result<Trigger, TriggerRepoError> {
|
|
if req.collection_glob.is_empty() {
|
|
return Err(TriggerRepoError::Invalid(
|
|
"collection_glob must not be empty".into(),
|
|
));
|
|
}
|
|
let mut tx = self.pool.begin().await?;
|
|
let parent: TriggerRow = sqlx::query_as(
|
|
"INSERT INTO triggers ( \
|
|
app_id, script_id, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal \
|
|
) VALUES ($1, $2, 'kv', TRUE, $3, $4, $5, $6, $7) \
|
|
RETURNING id, app_id, script_id, name, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal, created_at, updated_at",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.bind(req.script_id.into_inner())
|
|
.bind(req.dispatch_mode.as_str())
|
|
.bind(i32::try_from(req.retry_max_attempts).unwrap_or(3))
|
|
.bind(req.retry_backoff.as_str())
|
|
.bind(i32::try_from(req.retry_base_ms).unwrap_or(1000))
|
|
.bind(req.registered_by_principal.into_inner())
|
|
.fetch_one(&mut *tx)
|
|
.await?;
|
|
|
|
let ops_str: Vec<String> = req.ops.iter().map(|o| o.as_str().to_string()).collect();
|
|
sqlx::query(
|
|
"INSERT INTO kv_trigger_details (trigger_id, collection_glob, ops) \
|
|
VALUES ($1, $2, $3)",
|
|
)
|
|
.bind(parent.id)
|
|
.bind(&req.collection_glob)
|
|
.bind(&ops_str)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
tx.commit().await?;
|
|
|
|
Ok(Trigger {
|
|
id: parent.id.into(),
|
|
app_id: parent.app_id.into(),
|
|
script_id: parent.script_id.into(),
|
|
name: parent.name.clone(),
|
|
kind: TriggerKind::Kv,
|
|
enabled: parent.enabled,
|
|
dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
|
|
retry_max_attempts: u32::try_from(parent.retry_max_attempts).unwrap_or(3),
|
|
retry_backoff: BackoffShape::from_wire(&parent.retry_backoff)
|
|
.unwrap_or(BackoffShape::Exponential),
|
|
retry_base_ms: u32::try_from(parent.retry_base_ms).unwrap_or(1000),
|
|
registered_by_principal: parent.registered_by_principal.into(),
|
|
created_at: parent.created_at,
|
|
updated_at: parent.updated_at,
|
|
details: TriggerDetails::Kv {
|
|
collection_glob: req.collection_glob,
|
|
ops: req.ops,
|
|
},
|
|
})
|
|
}
|
|
|
|
async fn create_docs_trigger(
|
|
&self,
|
|
app_id: AppId,
|
|
req: CreateDocsTrigger,
|
|
) -> Result<Trigger, TriggerRepoError> {
|
|
if req.collection_glob.is_empty() {
|
|
return Err(TriggerRepoError::Invalid(
|
|
"collection_glob must not be empty".into(),
|
|
));
|
|
}
|
|
let mut tx = self.pool.begin().await?;
|
|
let parent: TriggerRow = sqlx::query_as(
|
|
"INSERT INTO triggers ( \
|
|
app_id, script_id, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal \
|
|
) VALUES ($1, $2, 'docs', TRUE, $3, $4, $5, $6, $7) \
|
|
RETURNING id, app_id, script_id, name, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal, created_at, updated_at",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.bind(req.script_id.into_inner())
|
|
.bind(req.dispatch_mode.as_str())
|
|
.bind(i32::try_from(req.retry_max_attempts).unwrap_or(3))
|
|
.bind(req.retry_backoff.as_str())
|
|
.bind(i32::try_from(req.retry_base_ms).unwrap_or(1000))
|
|
.bind(req.registered_by_principal.into_inner())
|
|
.fetch_one(&mut *tx)
|
|
.await?;
|
|
|
|
let ops_str: Vec<String> = req.ops.iter().map(|o| o.as_str().to_string()).collect();
|
|
sqlx::query(
|
|
"INSERT INTO docs_trigger_details (trigger_id, collection_glob, ops) \
|
|
VALUES ($1, $2, $3)",
|
|
)
|
|
.bind(parent.id)
|
|
.bind(&req.collection_glob)
|
|
.bind(&ops_str)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
tx.commit().await?;
|
|
|
|
Ok(Trigger {
|
|
id: parent.id.into(),
|
|
app_id: parent.app_id.into(),
|
|
script_id: parent.script_id.into(),
|
|
name: parent.name.clone(),
|
|
kind: TriggerKind::Docs,
|
|
enabled: parent.enabled,
|
|
dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
|
|
retry_max_attempts: u32::try_from(parent.retry_max_attempts).unwrap_or(3),
|
|
retry_backoff: BackoffShape::from_wire(&parent.retry_backoff)
|
|
.unwrap_or(BackoffShape::Exponential),
|
|
retry_base_ms: u32::try_from(parent.retry_base_ms).unwrap_or(1000),
|
|
registered_by_principal: parent.registered_by_principal.into(),
|
|
created_at: parent.created_at,
|
|
updated_at: parent.updated_at,
|
|
details: TriggerDetails::Docs {
|
|
collection_glob: req.collection_glob,
|
|
ops: req.ops,
|
|
},
|
|
})
|
|
}
|
|
|
|
async fn create_dead_letter_trigger(
|
|
&self,
|
|
app_id: AppId,
|
|
req: CreateDeadLetterTrigger,
|
|
) -> Result<Trigger, TriggerRepoError> {
|
|
let mut tx = self.pool.begin().await?;
|
|
// Dead-letter triggers force max_attempts=1 (design notes §4
|
|
// recursion-stop). Backoff/base_ms irrelevant but the columns
|
|
// are NOT NULL — store sensible values.
|
|
let parent: TriggerRow = sqlx::query_as(
|
|
"INSERT INTO triggers ( \
|
|
app_id, script_id, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal \
|
|
) VALUES ($1, $2, 'dead_letter', TRUE, 'async', 1, 'constant', 0, $3) \
|
|
RETURNING id, app_id, script_id, name, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal, created_at, updated_at",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.bind(req.script_id.into_inner())
|
|
.bind(req.registered_by_principal.into_inner())
|
|
.fetch_one(&mut *tx)
|
|
.await?;
|
|
|
|
sqlx::query(
|
|
"INSERT INTO dead_letter_trigger_details \
|
|
(trigger_id, source_filter, trigger_id_filter, script_id_filter) \
|
|
VALUES ($1, $2, $3, $4)",
|
|
)
|
|
.bind(parent.id)
|
|
.bind(req.source_filter.as_deref())
|
|
.bind(req.trigger_id_filter.map(TriggerId::into_inner))
|
|
.bind(req.script_id_filter.map(ScriptId::into_inner))
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
tx.commit().await?;
|
|
|
|
Ok(Trigger {
|
|
id: parent.id.into(),
|
|
app_id: parent.app_id.into(),
|
|
script_id: parent.script_id.into(),
|
|
name: parent.name.clone(),
|
|
kind: TriggerKind::DeadLetter,
|
|
enabled: parent.enabled,
|
|
dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
|
|
retry_max_attempts: u32::try_from(parent.retry_max_attempts).unwrap_or(1),
|
|
retry_backoff: BackoffShape::from_wire(&parent.retry_backoff)
|
|
.unwrap_or(BackoffShape::Constant),
|
|
retry_base_ms: u32::try_from(parent.retry_base_ms).unwrap_or(0),
|
|
registered_by_principal: parent.registered_by_principal.into(),
|
|
created_at: parent.created_at,
|
|
updated_at: parent.updated_at,
|
|
details: TriggerDetails::DeadLetter {
|
|
source_filter: req.source_filter,
|
|
trigger_id_filter: req.trigger_id_filter,
|
|
script_id_filter: req.script_id_filter,
|
|
},
|
|
})
|
|
}
|
|
|
|
async fn create_cron_trigger(
|
|
&self,
|
|
app_id: AppId,
|
|
req: CreateCronTrigger,
|
|
) -> Result<Trigger, TriggerRepoError> {
|
|
// Defense-in-depth validation (the admin endpoint validates too).
|
|
crate::cron_scheduler::validate_schedule(&req.schedule)
|
|
.map_err(|e| TriggerRepoError::Invalid(format!("invalid cron schedule: {e}")))?;
|
|
crate::cron_scheduler::validate_timezone(&req.timezone)
|
|
.map_err(|e| TriggerRepoError::Invalid(format!("invalid timezone: {e}")))?;
|
|
|
|
let mut tx = self.pool.begin().await?;
|
|
let parent: TriggerRow = sqlx::query_as(
|
|
"INSERT INTO triggers ( \
|
|
app_id, script_id, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal \
|
|
) VALUES ($1, $2, 'cron', TRUE, $3, $4, $5, $6, $7) \
|
|
RETURNING id, app_id, script_id, name, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal, created_at, updated_at",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.bind(req.script_id.into_inner())
|
|
.bind(req.dispatch_mode.as_str())
|
|
.bind(i32::try_from(req.retry_max_attempts).unwrap_or(3))
|
|
.bind(req.retry_backoff.as_str())
|
|
.bind(i32::try_from(req.retry_base_ms).unwrap_or(1000))
|
|
.bind(req.registered_by_principal.into_inner())
|
|
.fetch_one(&mut *tx)
|
|
.await?;
|
|
|
|
sqlx::query(
|
|
"INSERT INTO cron_trigger_details (trigger_id, schedule, timezone) \
|
|
VALUES ($1, $2, $3)",
|
|
)
|
|
.bind(parent.id)
|
|
.bind(&req.schedule)
|
|
.bind(&req.timezone)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
tx.commit().await?;
|
|
|
|
Ok(Trigger {
|
|
id: parent.id.into(),
|
|
app_id: parent.app_id.into(),
|
|
script_id: parent.script_id.into(),
|
|
name: parent.name.clone(),
|
|
kind: TriggerKind::Cron,
|
|
enabled: parent.enabled,
|
|
dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
|
|
retry_max_attempts: u32::try_from(parent.retry_max_attempts).unwrap_or(3),
|
|
retry_backoff: BackoffShape::from_wire(&parent.retry_backoff)
|
|
.unwrap_or(BackoffShape::Exponential),
|
|
retry_base_ms: u32::try_from(parent.retry_base_ms).unwrap_or(1000),
|
|
registered_by_principal: parent.registered_by_principal.into(),
|
|
created_at: parent.created_at,
|
|
updated_at: parent.updated_at,
|
|
details: TriggerDetails::Cron {
|
|
schedule: req.schedule,
|
|
timezone: req.timezone,
|
|
last_fired_at: None,
|
|
},
|
|
})
|
|
}
|
|
|
|
async fn create_files_trigger(
|
|
&self,
|
|
app_id: AppId,
|
|
req: CreateFilesTrigger,
|
|
) -> Result<Trigger, TriggerRepoError> {
|
|
if req.collection_glob.is_empty() {
|
|
return Err(TriggerRepoError::Invalid(
|
|
"collection_glob must not be empty".into(),
|
|
));
|
|
}
|
|
let mut tx = self.pool.begin().await?;
|
|
let parent: TriggerRow = sqlx::query_as(
|
|
"INSERT INTO triggers ( \
|
|
app_id, script_id, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal \
|
|
) VALUES ($1, $2, 'files', TRUE, $3, $4, $5, $6, $7) \
|
|
RETURNING id, app_id, script_id, name, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal, created_at, updated_at",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.bind(req.script_id.into_inner())
|
|
.bind(req.dispatch_mode.as_str())
|
|
.bind(i32::try_from(req.retry_max_attempts).unwrap_or(3))
|
|
.bind(req.retry_backoff.as_str())
|
|
.bind(i32::try_from(req.retry_base_ms).unwrap_or(1000))
|
|
.bind(req.registered_by_principal.into_inner())
|
|
.fetch_one(&mut *tx)
|
|
.await?;
|
|
|
|
let ops_str: Vec<String> = req.ops.iter().map(|o| o.as_str().to_string()).collect();
|
|
sqlx::query(
|
|
"INSERT INTO files_trigger_details (trigger_id, collection_glob, ops) \
|
|
VALUES ($1, $2, $3)",
|
|
)
|
|
.bind(parent.id)
|
|
.bind(&req.collection_glob)
|
|
.bind(&ops_str)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
tx.commit().await?;
|
|
|
|
Ok(Trigger {
|
|
id: parent.id.into(),
|
|
app_id: parent.app_id.into(),
|
|
script_id: parent.script_id.into(),
|
|
name: parent.name.clone(),
|
|
kind: TriggerKind::Files,
|
|
enabled: parent.enabled,
|
|
dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
|
|
retry_max_attempts: u32::try_from(parent.retry_max_attempts).unwrap_or(3),
|
|
retry_backoff: BackoffShape::from_wire(&parent.retry_backoff)
|
|
.unwrap_or(BackoffShape::Exponential),
|
|
retry_base_ms: u32::try_from(parent.retry_base_ms).unwrap_or(1000),
|
|
registered_by_principal: parent.registered_by_principal.into(),
|
|
created_at: parent.created_at,
|
|
updated_at: parent.updated_at,
|
|
details: TriggerDetails::Files {
|
|
collection_glob: req.collection_glob,
|
|
ops: req.ops,
|
|
},
|
|
})
|
|
}
|
|
|
|
async fn create_pubsub_trigger(
|
|
&self,
|
|
app_id: AppId,
|
|
req: CreatePubsubTrigger,
|
|
) -> Result<Trigger, TriggerRepoError> {
|
|
// Defense-in-depth validation (the admin endpoint validates too).
|
|
picloud_shared::validate_topic_pattern(&req.topic_pattern)
|
|
.map_err(TriggerRepoError::Invalid)?;
|
|
|
|
let mut tx = self.pool.begin().await?;
|
|
let parent: TriggerRow = sqlx::query_as(
|
|
"INSERT INTO triggers ( \
|
|
app_id, script_id, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal \
|
|
) VALUES ($1, $2, 'pubsub', TRUE, $3, $4, $5, $6, $7) \
|
|
RETURNING id, app_id, script_id, name, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal, created_at, updated_at",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.bind(req.script_id.into_inner())
|
|
.bind(req.dispatch_mode.as_str())
|
|
.bind(i32::try_from(req.retry_max_attempts).unwrap_or(3))
|
|
.bind(req.retry_backoff.as_str())
|
|
.bind(i32::try_from(req.retry_base_ms).unwrap_or(1000))
|
|
.bind(req.registered_by_principal.into_inner())
|
|
.fetch_one(&mut *tx)
|
|
.await?;
|
|
|
|
sqlx::query(
|
|
"INSERT INTO pubsub_trigger_details (trigger_id, topic_pattern) VALUES ($1, $2)",
|
|
)
|
|
.bind(parent.id)
|
|
.bind(&req.topic_pattern)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
tx.commit().await?;
|
|
|
|
Ok(Trigger {
|
|
id: parent.id.into(),
|
|
app_id: parent.app_id.into(),
|
|
script_id: parent.script_id.into(),
|
|
name: parent.name.clone(),
|
|
kind: TriggerKind::Pubsub,
|
|
enabled: parent.enabled,
|
|
dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
|
|
retry_max_attempts: u32::try_from(parent.retry_max_attempts).unwrap_or(3),
|
|
retry_backoff: BackoffShape::from_wire(&parent.retry_backoff)
|
|
.unwrap_or(BackoffShape::Exponential),
|
|
retry_base_ms: u32::try_from(parent.retry_base_ms).unwrap_or(1000),
|
|
registered_by_principal: parent.registered_by_principal.into(),
|
|
created_at: parent.created_at,
|
|
updated_at: parent.updated_at,
|
|
details: TriggerDetails::Pubsub {
|
|
topic_pattern: req.topic_pattern,
|
|
},
|
|
})
|
|
}
|
|
|
|
async fn create_email_trigger(
|
|
&self,
|
|
app_id: AppId,
|
|
req: CreateEmailTrigger,
|
|
) -> Result<Trigger, TriggerRepoError> {
|
|
let has_inbound_secret = req.inbound_secret_encrypted.is_some();
|
|
let mut tx = self.pool.begin().await?;
|
|
// Inbound email is delivered async like every other fan-out
|
|
// event; the receiver enqueues an outbox row the dispatcher
|
|
// picks up. Retry settings use the standard defaults.
|
|
let parent: TriggerRow = sqlx::query_as(
|
|
"INSERT INTO triggers ( \
|
|
app_id, script_id, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal \
|
|
) VALUES ($1, $2, 'email', TRUE, 'async', 3, 'exponential', 1000, $3) \
|
|
RETURNING id, app_id, script_id, name, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal, created_at, updated_at",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.bind(req.script_id.into_inner())
|
|
.bind(req.registered_by_principal.into_inner())
|
|
.fetch_one(&mut *tx)
|
|
.await?;
|
|
|
|
sqlx::query(
|
|
"INSERT INTO email_trigger_details \
|
|
(trigger_id, inbound_secret_encrypted, inbound_secret_nonce) \
|
|
VALUES ($1, $2, $3)",
|
|
)
|
|
.bind(parent.id)
|
|
.bind(req.inbound_secret_encrypted.as_deref())
|
|
.bind(req.inbound_secret_nonce.as_deref())
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
tx.commit().await?;
|
|
|
|
Ok(Trigger {
|
|
id: parent.id.into(),
|
|
app_id: parent.app_id.into(),
|
|
script_id: parent.script_id.into(),
|
|
name: parent.name.clone(),
|
|
kind: TriggerKind::Email,
|
|
enabled: parent.enabled,
|
|
dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
|
|
retry_max_attempts: u32::try_from(parent.retry_max_attempts).unwrap_or(3),
|
|
retry_backoff: BackoffShape::from_wire(&parent.retry_backoff)
|
|
.unwrap_or(BackoffShape::Exponential),
|
|
retry_base_ms: u32::try_from(parent.retry_base_ms).unwrap_or(1000),
|
|
registered_by_principal: parent.registered_by_principal.into(),
|
|
created_at: parent.created_at,
|
|
updated_at: parent.updated_at,
|
|
details: TriggerDetails::Email { has_inbound_secret },
|
|
})
|
|
}
|
|
|
|
async fn email_inbound_target(
|
|
&self,
|
|
trigger_id: TriggerId,
|
|
) -> Result<Option<EmailInboundTarget>, TriggerRepoError> {
|
|
let row: Option<EmailInboundRow> = sqlx::query_as(
|
|
"SELECT t.app_id, t.script_id, t.enabled, t.dispatch_mode, \
|
|
t.registered_by_principal, \
|
|
d.inbound_secret_encrypted, d.inbound_secret_nonce \
|
|
FROM triggers t \
|
|
JOIN email_trigger_details d ON d.trigger_id = t.id \
|
|
WHERE t.id = $1 AND t.kind = 'email'",
|
|
)
|
|
.bind(trigger_id.into_inner())
|
|
.fetch_optional(&self.pool)
|
|
.await?;
|
|
Ok(row.map(|r| EmailInboundTarget {
|
|
app_id: r.app_id.into(),
|
|
script_id: r.script_id.into(),
|
|
enabled: r.enabled,
|
|
dispatch_mode: dispatch_from_str(&r.dispatch_mode),
|
|
registered_by_principal: r.registered_by_principal.into(),
|
|
inbound_secret_encrypted: r.inbound_secret_encrypted,
|
|
inbound_secret_nonce: r.inbound_secret_nonce,
|
|
}))
|
|
}
|
|
|
|
async fn list_for_app(&self, app_id: AppId) -> Result<Vec<Trigger>, TriggerRepoError> {
|
|
let parents: Vec<TriggerRow> = sqlx::query_as(
|
|
"SELECT id, app_id, script_id, name, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal, created_at, updated_at \
|
|
FROM triggers WHERE app_id = $1 ORDER BY created_at DESC",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.fetch_all(&self.pool)
|
|
.await?;
|
|
|
|
// F-P-008 (partial): concurrent hydration so a dashboard with N
|
|
// triggers waits for max(per-row-latency) instead of
|
|
// sum(per-row-latency). Still N+1 queries against the details
|
|
// tables — collapsing to a single per-kind LEFT JOIN is a
|
|
// larger SQL refactor tracked as v1.2.
|
|
let pool = self.pool.clone();
|
|
let out = stream::iter(parents.into_iter().map(Ok))
|
|
.map_ok(|p| {
|
|
let pool = pool.clone();
|
|
async move { hydrate_one(&pool, p).await }
|
|
})
|
|
.try_buffered(8)
|
|
.try_collect::<Vec<_>>()
|
|
.await?;
|
|
Ok(out)
|
|
}
|
|
|
|
async fn get(&self, id: TriggerId) -> Result<Option<Trigger>, TriggerRepoError> {
|
|
let parent: Option<TriggerRow> = sqlx::query_as(
|
|
"SELECT id, app_id, script_id, name, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal, created_at, updated_at \
|
|
FROM triggers WHERE id = $1",
|
|
)
|
|
.bind(id.into_inner())
|
|
.fetch_optional(&self.pool)
|
|
.await?;
|
|
match parent {
|
|
Some(p) => Ok(Some(hydrate_one(&self.pool, p).await?)),
|
|
None => Ok(None),
|
|
}
|
|
}
|
|
|
|
async fn delete(&self, id: TriggerId) -> Result<bool, TriggerRepoError> {
|
|
// ON DELETE CASCADE on the detail tables takes care of them.
|
|
let res = sqlx::query("DELETE FROM triggers WHERE id = $1")
|
|
.bind(id.into_inner())
|
|
.execute(&self.pool)
|
|
.await?;
|
|
Ok(res.rows_affected() > 0)
|
|
}
|
|
|
|
async fn list_matching_kv(
|
|
&self,
|
|
app_id: AppId,
|
|
collection: &str,
|
|
op: KvEventOp,
|
|
) -> Result<Vec<KvTriggerMatch>, 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<KvMatchRow> = sqlx::query_as(
|
|
"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",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.fetch_all(&self.pool)
|
|
.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)
|
|
}
|
|
|
|
async fn list_matching_docs(
|
|
&self,
|
|
app_id: AppId,
|
|
collection: &str,
|
|
op: DocsEventOp,
|
|
) -> Result<Vec<DocsTriggerMatch>, 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<KvMatchRow> = sqlx::query_as(
|
|
"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",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.fetch_all(&self.pool)
|
|
.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)
|
|
}
|
|
|
|
async fn list_matching_files(
|
|
&self,
|
|
app_id: AppId,
|
|
collection: &str,
|
|
op: FilesEventOp,
|
|
) -> Result<Vec<FilesTriggerMatch>, TriggerRepoError> {
|
|
// Mirrors list_matching_kv: pull every enabled files trigger,
|
|
// filter glob + ops in Rust (empty ops array means "any op").
|
|
let rows: Vec<KvMatchRow> = sqlx::query_as(
|
|
"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",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.fetch_all(&self.pool)
|
|
.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)
|
|
}
|
|
|
|
async fn list_matching_dead_letter(
|
|
&self,
|
|
app_id: AppId,
|
|
source: &str,
|
|
trigger_id: Option<TriggerId>,
|
|
script_id: Option<ScriptId>,
|
|
) -> Result<Vec<DeadLetterTriggerMatch>, TriggerRepoError> {
|
|
let rows: Vec<DlMatchRow> = sqlx::query_as(
|
|
"SELECT t.id, t.script_id, t.dispatch_mode, t.registered_by_principal, \
|
|
d.source_filter, d.trigger_id_filter, d.script_id_filter \
|
|
FROM triggers t \
|
|
JOIN dead_letter_trigger_details d ON d.trigger_id = t.id \
|
|
WHERE t.app_id = $1 AND t.kind = 'dead_letter' AND t.enabled = TRUE \
|
|
AND (d.source_filter IS NULL OR d.source_filter = $2) \
|
|
AND (d.trigger_id_filter IS NULL OR d.trigger_id_filter = $3) \
|
|
AND (d.script_id_filter IS NULL OR d.script_id_filter = $4)",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.bind(source)
|
|
.bind(trigger_id.map(TriggerId::into_inner))
|
|
.bind(script_id.map(ScriptId::into_inner))
|
|
.fetch_all(&self.pool)
|
|
.await?;
|
|
|
|
Ok(rows
|
|
.into_iter()
|
|
.map(|r| DeadLetterTriggerMatch {
|
|
trigger_id: r.id.into(),
|
|
script_id: r.script_id.into(),
|
|
dispatch_mode: dispatch_from_str(&r.dispatch_mode),
|
|
registered_by_principal: r.registered_by_principal.into(),
|
|
})
|
|
.collect())
|
|
}
|
|
|
|
async fn create_queue_trigger(
|
|
&self,
|
|
app_id: AppId,
|
|
req: CreateQueueTrigger,
|
|
) -> Result<Trigger, TriggerRepoError> {
|
|
if req.queue_name.is_empty() {
|
|
return Err(TriggerRepoError::Invalid(
|
|
"queue_name must not be empty".into(),
|
|
));
|
|
}
|
|
if !(5..=3600).contains(&req.visibility_timeout_secs) {
|
|
return Err(TriggerRepoError::Invalid(
|
|
"visibility_timeout_secs must be in [5, 3600]".into(),
|
|
));
|
|
}
|
|
|
|
let mut tx = self.pool.begin().await?;
|
|
|
|
// Per-(app_id, queue_name) advisory lock serializes concurrent
|
|
// creates so the SELECT-then-INSERT below is atomic. The lock
|
|
// is per-transaction (xact_lock) and automatically released on
|
|
// commit/rollback.
|
|
let lock_key = advisory_lock_key(app_id, &req.queue_name);
|
|
sqlx::query("SELECT pg_advisory_xact_lock($1)")
|
|
.bind(lock_key)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
// Reject duplicate consumer (one-per-queue invariant).
|
|
let existing: Option<(Uuid,)> = sqlx::query_as(
|
|
"SELECT t.id FROM triggers t \
|
|
JOIN queue_trigger_details d ON d.trigger_id = t.id \
|
|
WHERE t.app_id = $1 AND t.kind = 'queue' \
|
|
AND d.queue_name = $2",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.bind(&req.queue_name)
|
|
.fetch_optional(&mut *tx)
|
|
.await?;
|
|
if existing.is_some() {
|
|
return Err(TriggerRepoError::Invalid(format!(
|
|
"queue '{}' already has a consumer trigger; remove the existing one first",
|
|
req.queue_name
|
|
)));
|
|
}
|
|
|
|
let parent: TriggerRow = sqlx::query_as(
|
|
"INSERT INTO triggers ( \
|
|
app_id, script_id, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal \
|
|
) VALUES ($1, $2, 'queue', TRUE, $3, $4, $5, $6, $7) \
|
|
RETURNING id, app_id, script_id, name, kind, enabled, dispatch_mode, \
|
|
retry_max_attempts, retry_backoff, retry_base_ms, \
|
|
registered_by_principal, created_at, updated_at",
|
|
)
|
|
.bind(app_id.into_inner())
|
|
.bind(req.script_id.into_inner())
|
|
.bind(req.dispatch_mode.as_str())
|
|
.bind(i32::try_from(req.retry_max_attempts).unwrap_or(3))
|
|
.bind(req.retry_backoff.as_str())
|
|
.bind(i32::try_from(req.retry_base_ms).unwrap_or(1000))
|
|
.bind(req.registered_by_principal.into_inner())
|
|
.fetch_one(&mut *tx)
|
|
.await?;
|
|
|
|
sqlx::query(
|
|
"INSERT INTO queue_trigger_details \
|
|
(trigger_id, queue_name, visibility_timeout_secs) \
|
|
VALUES ($1, $2, $3)",
|
|
)
|
|
.bind(parent.id)
|
|
.bind(&req.queue_name)
|
|
.bind(i32::try_from(req.visibility_timeout_secs).unwrap_or(30))
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
|
|
tx.commit().await?;
|
|
|
|
Ok(Trigger {
|
|
id: parent.id.into(),
|
|
app_id: parent.app_id.into(),
|
|
script_id: parent.script_id.into(),
|
|
name: parent.name.clone(),
|
|
kind: TriggerKind::Queue,
|
|
enabled: parent.enabled,
|
|
dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
|
|
retry_max_attempts: u32::try_from(parent.retry_max_attempts).unwrap_or(3),
|
|
retry_backoff: BackoffShape::from_wire(&parent.retry_backoff)
|
|
.unwrap_or(BackoffShape::Exponential),
|
|
retry_base_ms: u32::try_from(parent.retry_base_ms).unwrap_or(1000),
|
|
registered_by_principal: parent.registered_by_principal.into(),
|
|
created_at: parent.created_at,
|
|
updated_at: parent.updated_at,
|
|
details: TriggerDetails::Queue {
|
|
queue_name: req.queue_name,
|
|
visibility_timeout_secs: req.visibility_timeout_secs,
|
|
last_fired_at: None,
|
|
},
|
|
})
|
|
}
|
|
|
|
async fn list_active_queue_consumers(
|
|
&self,
|
|
) -> Result<Vec<ActiveQueueConsumer>, TriggerRepoError> {
|
|
let rows: Vec<QueueConsumerRow> = sqlx::query_as(
|
|
"SELECT t.id AS trigger_id, t.app_id, t.script_id, d.queue_name, \
|
|
d.visibility_timeout_secs, \
|
|
t.retry_max_attempts, t.retry_backoff, t.retry_base_ms, \
|
|
t.registered_by_principal \
|
|
FROM triggers t \
|
|
JOIN queue_trigger_details d ON d.trigger_id = t.id \
|
|
JOIN scripts s ON s.id = t.script_id \
|
|
WHERE t.kind = 'queue' AND t.enabled = TRUE AND s.enabled = TRUE",
|
|
)
|
|
.fetch_all(&self.pool)
|
|
.await?;
|
|
Ok(rows
|
|
.into_iter()
|
|
.map(|r| ActiveQueueConsumer {
|
|
trigger_id: r.trigger_id.into(),
|
|
app_id: r.app_id.into(),
|
|
script_id: r.script_id.into(),
|
|
queue_name: r.queue_name,
|
|
visibility_timeout_secs: u32::try_from(r.visibility_timeout_secs).unwrap_or(30),
|
|
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(),
|
|
})
|
|
.collect())
|
|
}
|
|
|
|
async fn touch_queue_trigger_last_fired_at(
|
|
&self,
|
|
trigger_id: TriggerId,
|
|
at: DateTime<Utc>,
|
|
) -> Result<(), TriggerRepoError> {
|
|
sqlx::query("UPDATE queue_trigger_details SET last_fired_at = $2 WHERE trigger_id = $1")
|
|
.bind(trigger_id.into_inner())
|
|
.bind(at)
|
|
.execute(&self.pool)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
/// Stable per-(app_id, queue_name) i64 derived from the UUID bytes +
|
|
/// the queue name. Used as the key for `pg_advisory_xact_lock` so
|
|
/// concurrent `create_queue_trigger` calls for the same queue serialize.
|
|
fn advisory_lock_key(app_id: AppId, queue_name: &str) -> i64 {
|
|
use std::collections::hash_map::DefaultHasher;
|
|
use std::hash::{Hash, Hasher};
|
|
let mut h = DefaultHasher::new();
|
|
app_id.into_inner().hash(&mut h);
|
|
queue_name.hash(&mut h);
|
|
// Postgres advisory locks key on bigint; reinterpret the u64 as i64.
|
|
#[allow(clippy::cast_possible_wrap)]
|
|
let key = h.finish() as i64;
|
|
key
|
|
}
|
|
|
|
#[allow(clippy::too_many_lines)]
|
|
async fn hydrate_one(pool: &PgPool, parent: TriggerRow) -> Result<Trigger, TriggerRepoError> {
|
|
let kind = TriggerKind::from_wire(&parent.kind).ok_or_else(|| {
|
|
TriggerRepoError::Invalid(format!("unknown trigger kind {}", parent.kind))
|
|
})?;
|
|
|
|
let details = match kind {
|
|
TriggerKind::Kv => {
|
|
let row: KvDetailRow = sqlx::query_as(
|
|
"SELECT collection_glob, ops FROM kv_trigger_details WHERE trigger_id = $1",
|
|
)
|
|
.bind(parent.id)
|
|
.fetch_one(pool)
|
|
.await?;
|
|
let ops = row
|
|
.ops
|
|
.iter()
|
|
.filter_map(|s| KvEventOp::from_wire(s))
|
|
.collect();
|
|
TriggerDetails::Kv {
|
|
collection_glob: row.collection_glob,
|
|
ops,
|
|
}
|
|
}
|
|
TriggerKind::Docs => {
|
|
let row: KvDetailRow = sqlx::query_as(
|
|
"SELECT collection_glob, ops FROM docs_trigger_details WHERE trigger_id = $1",
|
|
)
|
|
.bind(parent.id)
|
|
.fetch_one(pool)
|
|
.await?;
|
|
let ops = row
|
|
.ops
|
|
.iter()
|
|
.filter_map(|s| DocsEventOp::from_wire(s))
|
|
.collect();
|
|
TriggerDetails::Docs {
|
|
collection_glob: row.collection_glob,
|
|
ops,
|
|
}
|
|
}
|
|
TriggerKind::DeadLetter => {
|
|
let row: DlDetailRow = sqlx::query_as(
|
|
"SELECT source_filter, trigger_id_filter, script_id_filter \
|
|
FROM dead_letter_trigger_details WHERE trigger_id = $1",
|
|
)
|
|
.bind(parent.id)
|
|
.fetch_one(pool)
|
|
.await?;
|
|
TriggerDetails::DeadLetter {
|
|
source_filter: row.source_filter,
|
|
trigger_id_filter: row.trigger_id_filter.map(Into::into),
|
|
script_id_filter: row.script_id_filter.map(Into::into),
|
|
}
|
|
}
|
|
TriggerKind::Cron => {
|
|
let row: CronDetailRow = sqlx::query_as(
|
|
"SELECT schedule, timezone, last_fired_at \
|
|
FROM cron_trigger_details WHERE trigger_id = $1",
|
|
)
|
|
.bind(parent.id)
|
|
.fetch_one(pool)
|
|
.await?;
|
|
TriggerDetails::Cron {
|
|
schedule: row.schedule,
|
|
timezone: row.timezone,
|
|
last_fired_at: row.last_fired_at,
|
|
}
|
|
}
|
|
TriggerKind::Files => {
|
|
let row: KvDetailRow = sqlx::query_as(
|
|
"SELECT collection_glob, ops FROM files_trigger_details WHERE trigger_id = $1",
|
|
)
|
|
.bind(parent.id)
|
|
.fetch_one(pool)
|
|
.await?;
|
|
let ops = row
|
|
.ops
|
|
.iter()
|
|
.filter_map(|s| FilesEventOp::from_wire(s))
|
|
.collect();
|
|
TriggerDetails::Files {
|
|
collection_glob: row.collection_glob,
|
|
ops,
|
|
}
|
|
}
|
|
TriggerKind::Pubsub => {
|
|
let row: PubsubDetailRow = sqlx::query_as(
|
|
"SELECT topic_pattern FROM pubsub_trigger_details WHERE trigger_id = $1",
|
|
)
|
|
.bind(parent.id)
|
|
.fetch_one(pool)
|
|
.await?;
|
|
TriggerDetails::Pubsub {
|
|
topic_pattern: row.topic_pattern,
|
|
}
|
|
}
|
|
TriggerKind::Email => {
|
|
let row: EmailDetailRow = sqlx::query_as(
|
|
"SELECT inbound_secret_encrypted FROM email_trigger_details WHERE trigger_id = $1",
|
|
)
|
|
.bind(parent.id)
|
|
.fetch_one(pool)
|
|
.await?;
|
|
TriggerDetails::Email {
|
|
has_inbound_secret: row.inbound_secret_encrypted.is_some(),
|
|
}
|
|
}
|
|
TriggerKind::Queue => {
|
|
let row: QueueDetailRow = sqlx::query_as(
|
|
"SELECT queue_name, visibility_timeout_secs, last_fired_at \
|
|
FROM queue_trigger_details WHERE trigger_id = $1",
|
|
)
|
|
.bind(parent.id)
|
|
.fetch_one(pool)
|
|
.await?;
|
|
TriggerDetails::Queue {
|
|
queue_name: row.queue_name,
|
|
visibility_timeout_secs: u32::try_from(row.visibility_timeout_secs).unwrap_or(30),
|
|
last_fired_at: row.last_fired_at,
|
|
}
|
|
}
|
|
};
|
|
|
|
Ok(Trigger {
|
|
id: parent.id.into(),
|
|
app_id: parent.app_id.into(),
|
|
script_id: parent.script_id.into(),
|
|
name: parent.name,
|
|
kind,
|
|
enabled: parent.enabled,
|
|
dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
|
|
retry_max_attempts: u32::try_from(parent.retry_max_attempts).unwrap_or(3),
|
|
retry_backoff: BackoffShape::from_wire(&parent.retry_backoff)
|
|
.unwrap_or(BackoffShape::Exponential),
|
|
retry_base_ms: u32::try_from(parent.retry_base_ms).unwrap_or(1000),
|
|
registered_by_principal: parent.registered_by_principal.into(),
|
|
created_at: parent.created_at,
|
|
updated_at: parent.updated_at,
|
|
details,
|
|
})
|
|
}
|
|
|
|
fn dispatch_from_str(s: &str) -> TriggerDispatchMode {
|
|
match s {
|
|
"sync" => TriggerDispatchMode::Sync,
|
|
_ => TriggerDispatchMode::Async,
|
|
}
|
|
}
|
|
|
|
/// Match a `collection_glob` against an actual `collection` name.
|
|
/// Supported forms (in priority order):
|
|
/// - `"*"` → matches every collection
|
|
/// - `"foo*"` → prefix match (anything starting with "foo")
|
|
/// - `"foo"` → exact match
|
|
#[must_use]
|
|
pub fn collection_matches(glob: &str, collection: &str) -> bool {
|
|
if glob == "*" {
|
|
return true;
|
|
}
|
|
if let Some(prefix) = glob.strip_suffix('*') {
|
|
return collection.starts_with(prefix);
|
|
}
|
|
glob == collection
|
|
}
|
|
|
|
#[derive(sqlx::FromRow)]
|
|
struct TriggerRow {
|
|
id: Uuid,
|
|
app_id: Uuid,
|
|
script_id: Uuid,
|
|
name: String,
|
|
kind: String,
|
|
enabled: bool,
|
|
dispatch_mode: String,
|
|
retry_max_attempts: i32,
|
|
retry_backoff: String,
|
|
retry_base_ms: i32,
|
|
registered_by_principal: Uuid,
|
|
created_at: DateTime<Utc>,
|
|
updated_at: DateTime<Utc>,
|
|
}
|
|
|
|
#[derive(sqlx::FromRow)]
|
|
struct KvDetailRow {
|
|
collection_glob: String,
|
|
ops: Vec<String>,
|
|
}
|
|
|
|
#[derive(sqlx::FromRow)]
|
|
struct CronDetailRow {
|
|
schedule: String,
|
|
timezone: String,
|
|
last_fired_at: Option<DateTime<Utc>>,
|
|
}
|
|
|
|
#[derive(sqlx::FromRow)]
|
|
struct PubsubDetailRow {
|
|
topic_pattern: String,
|
|
}
|
|
|
|
#[derive(sqlx::FromRow)]
|
|
struct EmailDetailRow {
|
|
inbound_secret_encrypted: Option<Vec<u8>>,
|
|
}
|
|
|
|
#[derive(sqlx::FromRow)]
|
|
struct QueueDetailRow {
|
|
queue_name: String,
|
|
visibility_timeout_secs: i32,
|
|
last_fired_at: Option<DateTime<Utc>>,
|
|
}
|
|
|
|
#[derive(sqlx::FromRow)]
|
|
struct QueueConsumerRow {
|
|
trigger_id: Uuid,
|
|
app_id: Uuid,
|
|
script_id: Uuid,
|
|
queue_name: String,
|
|
visibility_timeout_secs: i32,
|
|
retry_max_attempts: i32,
|
|
retry_backoff: String,
|
|
retry_base_ms: i32,
|
|
registered_by_principal: Uuid,
|
|
}
|
|
|
|
#[derive(sqlx::FromRow)]
|
|
struct EmailInboundRow {
|
|
app_id: Uuid,
|
|
script_id: Uuid,
|
|
enabled: bool,
|
|
dispatch_mode: String,
|
|
registered_by_principal: Uuid,
|
|
inbound_secret_encrypted: Option<Vec<u8>>,
|
|
inbound_secret_nonce: Option<Vec<u8>>,
|
|
}
|
|
|
|
#[derive(sqlx::FromRow)]
|
|
#[allow(clippy::struct_field_names)]
|
|
struct DlDetailRow {
|
|
source_filter: Option<String>,
|
|
trigger_id_filter: Option<Uuid>,
|
|
script_id_filter: Option<Uuid>,
|
|
}
|
|
|
|
#[derive(sqlx::FromRow)]
|
|
struct KvMatchRow {
|
|
id: Uuid,
|
|
script_id: Uuid,
|
|
dispatch_mode: String,
|
|
retry_max_attempts: i32,
|
|
retry_backoff: String,
|
|
retry_base_ms: i32,
|
|
registered_by_principal: Uuid,
|
|
collection_glob: String,
|
|
ops: Vec<String>,
|
|
}
|
|
|
|
#[derive(sqlx::FromRow)]
|
|
struct DlMatchRow {
|
|
id: Uuid,
|
|
script_id: Uuid,
|
|
dispatch_mode: String,
|
|
registered_by_principal: Uuid,
|
|
#[allow(dead_code)]
|
|
source_filter: Option<String>,
|
|
#[allow(dead_code)]
|
|
trigger_id_filter: Option<Uuid>,
|
|
#[allow(dead_code)]
|
|
script_id_filter: Option<Uuid>,
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn collection_matcher_handles_star_prefix_exact() {
|
|
assert!(collection_matches("*", "widgets"));
|
|
assert!(collection_matches("*", ""));
|
|
assert!(collection_matches("users:*", "users:1"));
|
|
assert!(collection_matches("users:*", "users:"));
|
|
assert!(!collection_matches("users:*", "orgs:1"));
|
|
assert!(collection_matches("widgets", "widgets"));
|
|
assert!(!collection_matches("widgets", "Widgets"));
|
|
}
|
|
|
|
#[test]
|
|
fn trigger_kind_queue_round_trips_through_wire() {
|
|
assert_eq!(TriggerKind::Queue.as_str(), "queue");
|
|
assert_eq!(TriggerKind::from_wire("queue"), Some(TriggerKind::Queue));
|
|
}
|
|
|
|
#[test]
|
|
fn advisory_lock_key_is_stable_per_app_queue_pair() {
|
|
let app_id_a: AppId = uuid::Uuid::nil().into();
|
|
let app_id_b: AppId = uuid::Uuid::from_u128(1).into();
|
|
// Same inputs → same key.
|
|
assert_eq!(
|
|
advisory_lock_key(app_id_a, "jobs"),
|
|
advisory_lock_key(app_id_a, "jobs")
|
|
);
|
|
// Different app → different key (extremely high probability).
|
|
assert_ne!(
|
|
advisory_lock_key(app_id_a, "jobs"),
|
|
advisory_lock_key(app_id_b, "jobs")
|
|
);
|
|
// Different queue → different key.
|
|
assert_ne!(
|
|
advisory_lock_key(app_id_a, "jobs"),
|
|
advisory_lock_key(app_id_a, "other")
|
|
);
|
|
}
|
|
}
|