PiCloud/crates/manager-core/src/trigger_repo.rs

//! `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 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,
    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,
}

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",
        }
    }

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

/// 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,
}

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

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

// ----------------------------------------------------------------------------
// Postgres impl
// ----------------------------------------------------------------------------

pub struct PostgresTriggerRepo {
    pool: PgPool,
}

impl PostgresTriggerRepo {
    #[must_use]
    pub fn new(pool: PgPool) -> Self {
        Self { pool }
    }
}

#[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, 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(),
            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, 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(),
            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, 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(),
            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, 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(),
            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, 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(),
            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, 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(),
            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 list_for_app(&self, app_id: AppId) -> Result<Vec<Trigger>, TriggerRepoError> {
        let parents: Vec<TriggerRow> = sqlx::query_as(
            "SELECT id, app_id, script_id, 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?;

        let mut out = Vec::with_capacity(parents.len());
        for p in parents {
            out.push(hydrate_one(&self.pool, p).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, 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())
    }
}

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

    Ok(Trigger {
        id: parent.id.into(),
        app_id: parent.app_id.into(),
        script_id: parent.script_id.into(),
        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,
    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)]
#[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"));
    }
}