//! `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, GroupId, KvEventOp, ScriptId, ScriptOwner,
    TriggerId,
};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use uuid::Uuid;

use crate::config_resolver::CHAIN_LEVELS_CTE;
use crate::trigger_config::BackoffShape;

/// §11 tail opt-out: exclude an INHERITED (group-owned) trigger whose handler
/// script name a suppression on the firing app's chain declines. Appended to
/// each dispatch match query's WHERE clause; `$1` is the firing app_id (already
/// bound by the `CHAIN_LEVELS_CTE`). §11 tail M1: the anti-join joins the
/// `chain` CTE, so a suppression owned by the firing app OR any ANCESTOR GROUP
/// on its chain applies — a group declines a template for its whole subtree.
/// The `t.group_id IS NOT NULL` guard keeps an app's OWN trigger unsuppressable
/// (an owner can only decline what it inherits); `t.sealed = FALSE` keeps a
/// mandatory template non-declinable.
pub(crate) const TRIGGER_SUPPRESSION_ANTIJOIN: &str = " \
    AND NOT EXISTS ( \
        SELECT 1 FROM template_suppressions ts \
        JOIN scripts s ON s.id = t.script_id \
        JOIN chain sc ON (ts.app_id = sc.app_owner OR ts.group_id = sc.group_owner) \
        WHERE ts.target_kind = 'trigger' \
          AND LOWER(ts.reference) = LOWER(s.name) \
          AND t.group_id IS NOT NULL \
          AND t.sealed = FALSE)";

#[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.
// enabled + sealed + shared + materialized are distinct trigger facets, not a
// bit-packed config — a struct is the right shape.
#[allow(clippy::struct_excessive_bools)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Trigger {
    pub id: TriggerId,
    /// Polymorphic owner (§11 tail): an app owns a trigger; a group owns a
    /// trigger TEMPLATE. Exactly one is set (DB CHECK). Mirrors `Script`.
    pub app_id: Option<AppId>,
    pub group_id: Option<GroupId>,
    pub script_id: ScriptId,
    /// §4.5 per-app trigger identifier; the manifest merge/upsert key.
    pub name: String,
    pub kind: TriggerKind,
    pub enabled: bool,
    /// §11 tail: `true` for a sealed (non-suppressible) group trigger template.
    /// Always `false` for an app-owned trigger. Part of the apply diff identity
    /// so toggling it re-materializes the row.
    #[serde(default)]
    pub sealed: bool,
    /// §11.6: `true` for a shared-collection group trigger (watches a shared
    /// collection, not per-app ones). Also part of the apply diff identity.
    #[serde(default)]
    pub shared: bool,
    /// §4.5 M5: `true` for a materialized copy of a group stateful template
    /// (`materialized_from` is set). Read-only — inherited, not app-authored.
    /// Derived, not stored directly (see `TriggerRow::materialized_from`).
    #[serde(default)]
    pub materialized: 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>>,
    /// Envelope version of the sealed inbound secret (0 = legacy no-AAD, 1 =
    /// AAD-bound to the sealing owner). `0` when there is no secret.
    pub inbound_secret_version: i16,
    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,
    /// §11.6 D3: `Some(group)` when this consumer is a materialized copy of a
    /// SHARED group queue template — it drains `group_queue_messages(group,
    /// queue_name)` (competing consumers) instead of the per-app store. `None`
    /// for an ordinary per-app or non-shared-template consumer.
    pub shared_group: Option<GroupId>,
}

/// 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>>,
    /// Envelope version of the sealed secret (0 = legacy no-AAD, 1 = AAD-bound).
    pub inbound_secret_version: i16,
    /// §M5.5 / audit H-D1: the group this secret was sealed under, when this row
    /// is a materialized copy of a group EMAIL TEMPLATE (`materialized_from ->
    /// template.group_id`). `None` for a standalone per-app trigger (sealed under
    /// the app). Drives the AAD owner at open time for a v1 secret.
    pub sealing_group: Option<GroupId>,
}

/// 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>;

    /// Group-owned trigger TEMPLATES (§11 tail) declared directly at `group_id`.
    /// Used by the declarative apply diff for a `[group]` node. Defaults to
    /// empty so non-Postgres impls (tests) need not provide it.
    async fn list_for_group(&self, _group_id: GroupId) -> Result<Vec<Trigger>, TriggerRepoError> {
        Ok(Vec::new())
    }

    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>;

    /// §11.6 shared-collection fan-out: enabled `shared = true` triggers on the
    /// OWNING group (the group that declares the shared collection), glob-matched
    /// in Rust. The dispatch runs under the writing app; these methods key on the
    /// owning group, not the writer's chain. Default empty so non-Postgres impls
    /// degrade to "no shared triggers".
    async fn list_matching_shared_kv(
        &self,
        owning_group: GroupId,
        collection: &str,
        op: KvEventOp,
    ) -> Result<Vec<KvTriggerMatch>, TriggerRepoError> {
        let _ = (owning_group, collection, op);
        Ok(Vec::new())
    }
    async fn list_matching_shared_docs(
        &self,
        owning_group: GroupId,
        collection: &str,
        op: DocsEventOp,
    ) -> Result<Vec<DocsTriggerMatch>, TriggerRepoError> {
        let _ = (owning_group, collection, op);
        Ok(Vec::new())
    }
    async fn list_matching_shared_files(
        &self,
        owning_group: GroupId,
        collection: &str,
        op: FilesEventOp,
    ) -> Result<Vec<FilesTriggerMatch>, TriggerRepoError> {
        let _ = (owning_group, collection, op);
        Ok(Vec::new())
    }

    /// Dispatcher hot path for dead-letter fan-out. Filters: source
    /// (or any-source), originating trigger_id (or any), originating
    /// script_id (or any). Each filter is "match OR is_null".
    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 }
    }

    /// §11.6 shared-collection match: enabled `shared = true` triggers on the
    /// OWNING group, glob + op filtered in Rust (like the per-app path). `kind`
    /// and `detail_table` are hard-coded literals from the three call sites (no
    /// injection). No chain / no suppression anti-join — a shared trigger is
    /// declared on the group that owns the collection and fires under the writer.
    async fn shared_match_rows(
        &self,
        kind: &str,
        detail_table: &str,
        owning_group: GroupId,
        collection: &str,
        op_str: &str,
    ) -> Result<Vec<KvMatchRow>, TriggerRepoError> {
        let rows: Vec<KvMatchRow> = sqlx::query_as(&format!(
            "SELECT t.id, t.script_id, t.dispatch_mode, \
                    t.retry_max_attempts, t.retry_backoff, t.retry_base_ms, \
                    t.registered_by_principal, \
                    d.collection_glob, d.ops \
             FROM triggers t \
             JOIN {detail_table} d ON d.trigger_id = t.id \
             WHERE t.kind = '{kind}' AND t.enabled = TRUE \
               AND t.shared = TRUE AND t.group_id = $1"
        ))
        .bind(owning_group.into_inner())
        .fetch_all(&self.pool)
        .await?;
        Ok(rows
            .into_iter()
            .filter(|r| collection_matches(&r.collection_glob, collection))
            .filter(|r| r.ops.is_empty() || r.ops.iter().any(|o| o == op_str))
            .collect())
    }
}

/// Insert a trigger (parent row + per-kind detail) within an existing
/// 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>,
    owner: ScriptOwner,
    script_id: ScriptId,
    registered_by: AdminUserId,
    dispatch_mode: TriggerDispatchMode,
    retry_max_attempts: u32,
    retry_backoff: BackoffShape,
    retry_base_ms: u32,
    // §11 tail: `true` for a sealed (non-suppressible) group trigger template.
    // Always `false` for an app-owned trigger.
    sealed: bool,
    // §11.6: `true` for a shared-collection group trigger (watches a shared
    // collection, not per-app ones). Always `false` for an app-owned trigger.
    shared: bool,
    details: &TriggerDetails,
) -> 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(),
            ));
        }
    };
    // §11 tail: a trigger is owned by an app XOR a group (the latter a
    // template, unioned into descendant apps via the chain CTE at dispatch).
    let (owner_app_id, owner_group_id) = match owner {
        ScriptOwner::App(a) => (Some(a.into_inner()), None),
        ScriptOwner::Group(g) => (None, Some(g.into_inner())),
    };
    // 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).
    // Queue is app-only (rejected on a group), so this never runs for a
    // group owner.
    if let (TriggerDetails::Queue { queue_name, .. }, ScriptOwner::App(app_id)) = (details, owner) {
        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, group_id, script_id, kind, enabled, dispatch_mode, \
            retry_max_attempts, retry_backoff, retry_base_ms, \
            registered_by_principal, sealed, shared \
         ) VALUES ($1, $2, $3, $4, TRUE, $5, $6, $7, $8, $9, $10, $11) RETURNING id",
    )
    .bind(owner_app_id)
    .bind(owner_group_id)
    .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(sealed)
    .bind(shared)
    .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 owner's secret store);
/// this writes the ciphertext. Parent retry settings match the
/// interactive `create_email_trigger` path (async, 3, exponential, 1000).
///
/// §4.5 M5: `owner` is app-XOR-group. A `ScriptOwner::Group` writes an email
/// trigger TEMPLATE (group_id set, app_id NULL) that is never dispatched
/// directly (`email_inbound_target` filters `app_id IS NOT NULL`) — the
/// materializer copies it into a per-descendant-app row.
pub(crate) async fn insert_email_trigger_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    owner: ScriptOwner,
    script_id: ScriptId,
    registered_by: AdminUserId,
    inbound_secret_encrypted: &[u8],
    inbound_secret_nonce: &[u8],
    inbound_secret_version: i16,
) -> Result<TriggerId, TriggerRepoError> {
    let (owner_app_id, owner_group_id) = match owner {
        ScriptOwner::App(a) => (Some(a.into_inner()), None),
        ScriptOwner::Group(g) => (None, Some(g.into_inner())),
    };
    let row: (Uuid,) = sqlx::query_as(
        "INSERT INTO triggers ( \
            app_id, group_id, script_id, kind, enabled, dispatch_mode, \
            retry_max_attempts, retry_backoff, retry_base_ms, \
            registered_by_principal \
         ) VALUES ($1, $2, $3, 'email', TRUE, 'async', 3, 'exponential', 1000, $4) RETURNING id",
    )
    .bind(owner_app_id)
    .bind(owner_group_id)
    .bind(script_id.into_inner())
    .bind(registered_by.into_inner())
    .fetch_one(&mut **tx)
    .await?;
    sqlx::query(
        "INSERT INTO email_trigger_details \
            (trigger_id, inbound_secret_encrypted, inbound_secret_nonce, inbound_secret_version) \
         VALUES ($1, $2, $3, $4)",
    )
    .bind(row.0)
    .bind(inbound_secret_encrypted)
    .bind(inbound_secret_nonce)
    .bind(inbound_secret_version)
    .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.map(Into::into),
            group_id: parent.group_id.map(Into::into),
            script_id: parent.script_id.into(),
            name: parent.name.clone(),
            kind: TriggerKind::Kv,
            enabled: parent.enabled,
            sealed: parent.sealed,
            shared: parent.shared,
            materialized: parent.materialized_from.is_some(),
            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.map(Into::into),
            group_id: parent.group_id.map(Into::into),
            script_id: parent.script_id.into(),
            name: parent.name.clone(),
            kind: TriggerKind::Docs,
            enabled: parent.enabled,
            sealed: parent.sealed,
            shared: parent.shared,
            materialized: parent.materialized_from.is_some(),
            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.map(Into::into),
            group_id: parent.group_id.map(Into::into),
            script_id: parent.script_id.into(),
            name: parent.name.clone(),
            kind: TriggerKind::DeadLetter,
            enabled: parent.enabled,
            sealed: parent.sealed,
            shared: parent.shared,
            materialized: parent.materialized_from.is_some(),
            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.map(Into::into),
            group_id: parent.group_id.map(Into::into),
            script_id: parent.script_id.into(),
            name: parent.name.clone(),
            kind: TriggerKind::Cron,
            enabled: parent.enabled,
            sealed: parent.sealed,
            shared: parent.shared,
            materialized: parent.materialized_from.is_some(),
            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.map(Into::into),
            group_id: parent.group_id.map(Into::into),
            script_id: parent.script_id.into(),
            name: parent.name.clone(),
            kind: TriggerKind::Files,
            enabled: parent.enabled,
            sealed: parent.sealed,
            shared: parent.shared,
            materialized: parent.materialized_from.is_some(),
            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.map(Into::into),
            group_id: parent.group_id.map(Into::into),
            script_id: parent.script_id.into(),
            name: parent.name.clone(),
            kind: TriggerKind::Pubsub,
            enabled: parent.enabled,
            sealed: parent.sealed,
            shared: parent.shared,
            materialized: parent.materialized_from.is_some(),
            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, inbound_secret_version) \
             VALUES ($1, $2, $3, $4)",
        )
        .bind(parent.id)
        .bind(req.inbound_secret_encrypted.as_deref())
        .bind(req.inbound_secret_nonce.as_deref())
        .bind(req.inbound_secret_version)
        .execute(&mut *tx)
        .await?;

        tx.commit().await?;

        Ok(Trigger {
            id: parent.id.into(),
            app_id: parent.app_id.map(Into::into),
            group_id: parent.group_id.map(Into::into),
            script_id: parent.script_id.into(),
            name: parent.name.clone(),
            kind: TriggerKind::Email,
            enabled: parent.enabled,
            sealed: parent.sealed,
            shared: parent.shared,
            materialized: parent.materialized_from.is_some(),
            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(
            // §4.5 M5: `t.app_id IS NOT NULL` excludes a group email TEMPLATE
            // row — only its materialized per-app copies are directly invocable
            // (mirrors the cron-scheduler / queue-consumer guards). A template
            // has no app to run under, so hitting its webhook URL must 404.
            // §M5.5 / audit H-D1: `tmpl.group_id` (via `materialized_from`) is the
            // SEALING owner for a materialized copy of a group email template — the
            // AAD owner needed to open a v1 secret. NULL for a standalone app
            // trigger (sealed under the app).
            "SELECT t.app_id, t.script_id, t.enabled, t.dispatch_mode, \
                    t.registered_by_principal, \
                    d.inbound_secret_encrypted, d.inbound_secret_nonce, \
                    d.inbound_secret_version, tmpl.group_id AS sealing_group \
             FROM triggers t \
             JOIN email_trigger_details d ON d.trigger_id = t.id \
             LEFT JOIN triggers tmpl ON tmpl.id = t.materialized_from \
             WHERE t.id = $1 AND t.kind = 'email' AND t.app_id IS NOT NULL",
        )
        .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,
            inbound_secret_version: r.inbound_secret_version,
            sealing_group: r.sealing_group.map(Into::into),
        }))
    }

    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, sealed, shared, materialized_from, \
                    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 list_for_group(&self, group_id: GroupId) -> Result<Vec<Trigger>, TriggerRepoError> {
        let parents: Vec<TriggerRow> = sqlx::query_as(
            "SELECT id, app_id, group_id, script_id, name, kind, enabled, sealed, shared, dispatch_mode, \
                    retry_max_attempts, retry_backoff, retry_base_ms, \
                    registered_by_principal, created_at, updated_at \
             FROM triggers WHERE group_id = $1 ORDER BY created_at DESC",
        )
        .bind(group_id.into_inner())
        .fetch_all(&self.pool)
        .await?;
        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> {
        // §11 tail: `get` can fetch ANY trigger by id, including a group-owned
        // TEMPLATE, so it must SELECT `group_id` (unlike `list_for_app`, whose
        // rows are all app-owned) — else a template would hydrate with both
        // owners None, breaking the exactly-one invariant in memory.
        let parent: Option<TriggerRow> = sqlx::query_as(
            "SELECT id, app_id, group_id, script_id, name, kind, enabled, sealed, shared, 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(&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)
        .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(&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)
        .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(&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)
        .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_shared_kv(
        &self,
        owning_group: GroupId,
        collection: &str,
        op: KvEventOp,
    ) -> Result<Vec<KvTriggerMatch>, TriggerRepoError> {
        let rows = self
            .shared_match_rows(
                "kv",
                "kv_trigger_details",
                owning_group,
                collection,
                op.as_str(),
            )
            .await?;
        Ok(rows.into_iter().map(KvMatchRow::into_kv).collect())
    }

    async fn list_matching_shared_docs(
        &self,
        owning_group: GroupId,
        collection: &str,
        op: DocsEventOp,
    ) -> Result<Vec<DocsTriggerMatch>, TriggerRepoError> {
        let rows = self
            .shared_match_rows(
                "docs",
                "docs_trigger_details",
                owning_group,
                collection,
                op.as_str(),
            )
            .await?;
        Ok(rows.into_iter().map(KvMatchRow::into_docs).collect())
    }

    async fn list_matching_shared_files(
        &self,
        owning_group: GroupId,
        collection: &str,
        op: FilesEventOp,
    ) -> Result<Vec<FilesTriggerMatch>, TriggerRepoError> {
        let rows = self
            .shared_match_rows(
                "files",
                "files_trigger_details",
                owning_group,
                collection,
                op.as_str(),
            )
            .await?;
        Ok(rows.into_iter().map(KvMatchRow::into_files).collect())
    }

    async fn list_matching_dead_letter(
        &self,
        app_id: AppId,
        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.map(Into::into),
            group_id: parent.group_id.map(Into::into),
            script_id: parent.script_id.into(),
            name: parent.name.clone(),
            kind: TriggerKind::Queue,
            enabled: parent.enabled,
            sealed: parent.sealed,
            shared: parent.shared,
            materialized: parent.materialized_from.is_some(),
            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(
            // §11.6 D3: LEFT JOIN the SOURCE template of a materialized copy; if
            // that template is a SHARED group queue, `shared_group` is its
            // owning group and this consumer drains the group store instead of
            // the per-app one.
            "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, tmpl.group_id AS shared_group \
             FROM triggers t \
             JOIN queue_trigger_details d ON d.trigger_id = t.id \
             JOIN scripts s ON s.id = t.script_id \
             LEFT JOIN triggers tmpl \
                    ON tmpl.id = t.materialized_from AND tmpl.shared = TRUE \
             WHERE t.kind = 'queue' AND t.enabled = TRUE AND s.enabled = TRUE \
               AND t.app_id IS NOT NULL",
        )
        .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(),
                shared_group: r.shared_group.map(Into::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.map(Into::into),
        group_id: parent.group_id.map(Into::into),
        script_id: parent.script_id.into(),
        name: parent.name,
        kind,
        enabled: parent.enabled,
        sealed: parent.sealed,
        shared: parent.shared,
        materialized: parent.materialized_from.is_some(),
        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: Option<Uuid>,
    // `default` so the interactive-create RETURNING clauses (which don't list
    // group_id — they only ever build app-owned rows) still hydrate; the
    // group/app list queries SELECT it explicitly.
    #[sqlx(default)]
    group_id: Option<Uuid>,
    script_id: Uuid,
    name: String,
    kind: String,
    enabled: bool,
    // `default` (§11 tail) so the interactive-create RETURNING clauses (which
    // build app-owned rows, never sealed) still hydrate; the group/app list
    // queries that must see a true value SELECT it explicitly.
    #[sqlx(default)]
    sealed: bool,
    // §11.6: same `default` treatment as `sealed`.
    #[sqlx(default)]
    shared: bool,
    // §4.5 M5: set on a materialized copy of a group stateful template; NULL on
    // a hand-authored trigger. `default` like the flags above — only the app/
    // group list queries SELECT it.
    #[sqlx(default)]
    materialized_from: Option<Uuid>,
    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,
    // §11.6 D3: the owning group of a SHARED queue template this consumer was
    // materialized from; NULL for a per-app / non-shared consumer.
    shared_group: Option<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>>,
    inbound_secret_version: i16,
    sealing_group: Option<Uuid>,
}

#[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>,
}

impl KvMatchRow {
    fn into_kv(self) -> KvTriggerMatch {
        KvTriggerMatch {
            trigger_id: self.id.into(),
            script_id: self.script_id.into(),
            dispatch_mode: dispatch_from_str(&self.dispatch_mode),
            retry_max_attempts: u32::try_from(self.retry_max_attempts).unwrap_or(3),
            retry_backoff: BackoffShape::from_wire(&self.retry_backoff)
                .unwrap_or(BackoffShape::Exponential),
            retry_base_ms: u32::try_from(self.retry_base_ms).unwrap_or(1000),
            registered_by_principal: self.registered_by_principal.into(),
        }
    }

    fn into_docs(self) -> DocsTriggerMatch {
        let m = self.into_kv();
        DocsTriggerMatch {
            trigger_id: m.trigger_id,
            script_id: m.script_id,
            dispatch_mode: m.dispatch_mode,
            retry_max_attempts: m.retry_max_attempts,
            retry_backoff: m.retry_backoff,
            retry_base_ms: m.retry_base_ms,
            registered_by_principal: m.registered_by_principal,
        }
    }

    fn into_files(self) -> FilesTriggerMatch {
        let m = self.into_kv();
        FilesTriggerMatch {
            trigger_id: m.trigger_id,
            script_id: m.script_id,
            dispatch_mode: m.dispatch_mode,
            retry_max_attempts: m.retry_max_attempts,
            retry_backoff: m.retry_backoff,
            retry_base_ms: m.retry_base_ms,
            registered_by_principal: m.registered_by_principal,
        }
    }
}

#[derive(sqlx::FromRow)]
struct DlMatchRow {
    id: Uuid,
    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")
        );
    }
}