feat(v1.1.1-triggers): trigger CRUD admin endpoints

`/api/v1/admin/apps/{id}/triggers/*` — separate POST endpoints per
kind (kv / dead_letter) so each request validates against the
correct shape. List and DELETE work across both kinds.

Gated on `Capability::AppManageTriggers(app_id)`, which maps onto
`Scope::AppAdmin` (no new scope variants — seven-scope commitment
held) and is granted at the per-app `AppAdmin` role.

Request payloads accept `dispatch_mode` (defaults to `async`) and
retry-override fields. Omitted retry fields fall back to
`TriggerConfig::from_env`, which the binary plumbs into
`TriggersState` so the row is auditable from itself (no lazy
resolution at dispatch time). `registered_by_principal` is taken
from the authenticated principal — design notes §4: "a trigger
execution runs as the principal that registered the trigger".

DELETE loads the trigger first and 404s if its `app_id` doesn't
match the path — prevents a caller with rights on app A from
deleting a trigger via app B's path (bound-key safety net).

In-memory tests cover: app-not-found, member-without-role 403,
default-fallback for retry settings when request omits them,
empty-glob rejection, cross-app delete is treated as not-found.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/manager-core/src/lib.rs

@@ -35,6 +35,7 @@ pub mod sandbox;
 pub mod scheduler;
 pub mod trigger_config;
 pub mod trigger_repo;
+pub mod triggers_api;
 
 pub use abandoned_repo::{
     AbandonedRepo, AbandonedRepoError, NewAbandonedExecution, PostgresAbandonedRepo,
@@ -95,3 +96,4 @@ pub use trigger_repo::{
     KvTriggerMatch, PostgresTriggerRepo, Trigger, TriggerDetails, TriggerDispatchMode, TriggerKind,
     TriggerRepo, TriggerRepoError,
 };
+pub use triggers_api::{triggers_router, TriggersApiError, TriggersState};

crates/manager-core/src/triggers_api.rs

@@ -0,0 +1,748 @@
+//! `/api/v1/admin/apps/{id}/triggers/*` — trigger CRUD admin endpoints.
+//!
+//! Per design notes §2, two kinds ship in v1.1.1: `kv` (with
+//! collection_glob + ops) and `dead_letter` (with optional source /
+//! trigger_id / script_id filters). Separate endpoints per kind keep
+//! validation clean.
+//!
+//! Every endpoint is guarded by `Capability::AppManageTriggers(app_id)`
+//! evaluated after the resource lookup so the capability binds to the
+//! resource's actual `app_id` (mirrors `apps_api`).
+
+use std::sync::Arc;
+
+use axum::extract::{Path, State};
+use axum::http::StatusCode;
+use axum::response::{IntoResponse, Json, Response};
+use axum::routing::{delete, get, post};
+use axum::{Extension, Router};
+use picloud_shared::{AppId, KvEventOp, Principal, ScriptId, TriggerId};
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+
+use crate::app_repo::AppRepository;
+use crate::authz::{require, AuthzDenied, AuthzError, AuthzRepo, Capability};
+use crate::trigger_config::{BackoffShape, TriggerConfig};
+use crate::trigger_repo::{
+    CreateDeadLetterTrigger, CreateKvTrigger, Trigger, TriggerDispatchMode, TriggerRepo,
+    TriggerRepoError,
+};
+
+#[derive(Clone)]
+pub struct TriggersState {
+    pub triggers: Arc<dyn TriggerRepo>,
+    pub apps: Arc<dyn AppRepository>,
+    pub authz: Arc<dyn AuthzRepo>,
+    /// Defaults applied to created triggers when the request omits
+    /// retry settings. Kept on the state struct so tests can swap
+    /// in a stricter / looser config without env tinkering.
+    pub config: TriggerConfig,
+}
+
+pub fn triggers_router(state: TriggersState) -> Router {
+    Router::new()
+        .route(
+            "/apps/{app_id}/triggers",
+            get(list_triggers).delete(noop_405),
+        )
+        .route("/apps/{app_id}/triggers/kv", post(create_kv_trigger))
+        .route(
+            "/apps/{app_id}/triggers/dead_letter",
+            post(create_dl_trigger),
+        )
+        .route(
+            "/apps/{app_id}/triggers/{trigger_id}",
+            delete(delete_trigger),
+        )
+        .with_state(state)
+}
+
+async fn noop_405() -> StatusCode {
+    StatusCode::METHOD_NOT_ALLOWED
+}
+
+// ----------------------------------------------------------------------------
+// DTOs
+// ----------------------------------------------------------------------------
+
+#[derive(Debug, Deserialize)]
+pub struct CreateKvTriggerRequest {
+    pub script_id: ScriptId,
+    pub collection_glob: String,
+    /// Subset of `{insert, update, delete}`. Empty array means "any
+    /// op" (the trigger fires on every mutation in matching
+    /// collections).
+    #[serde(default)]
+    pub ops: Vec<KvEventOp>,
+    #[serde(default = "default_dispatch")]
+    pub dispatch_mode: TriggerDispatchMode,
+    /// Overrides for the platform retry defaults. Omitted fields fall
+    /// back to `TriggerConfig` (env-overridable) at write time.
+    #[serde(default)]
+    pub retry_max_attempts: Option<u32>,
+    #[serde(default)]
+    pub retry_backoff: Option<BackoffShape>,
+    #[serde(default)]
+    pub retry_base_ms: Option<u32>,
+}
+
+const fn default_dispatch() -> TriggerDispatchMode {
+    TriggerDispatchMode::Async
+}
+
+#[derive(Debug, Deserialize)]
+pub struct CreateDeadLetterTriggerRequest {
+    pub script_id: ScriptId,
+    #[serde(default)]
+    pub source_filter: Option<String>,
+    #[serde(default)]
+    pub trigger_id_filter: Option<TriggerId>,
+    #[serde(default)]
+    pub script_id_filter: Option<ScriptId>,
+}
+
+#[derive(Debug, Serialize)]
+pub struct TriggerListResponse {
+    pub triggers: Vec<Trigger>,
+}
+
+// ----------------------------------------------------------------------------
+// Handlers
+// ----------------------------------------------------------------------------
+
+async fn list_triggers(
+    State(s): State<TriggersState>,
+    Extension(principal): Extension<Principal>,
+    Path(app_id): Path<AppId>,
+) -> Result<Json<TriggerListResponse>, TriggersApiError> {
+    ensure_app_exists(&*s.apps, app_id).await?;
+    require(
+        s.authz.as_ref(),
+        &principal,
+        Capability::AppManageTriggers(app_id),
+    )
+    .await?;
+    let triggers = s.triggers.list_for_app(app_id).await?;
+    Ok(Json(TriggerListResponse { triggers }))
+}
+
+async fn create_kv_trigger(
+    State(s): State<TriggersState>,
+    Extension(principal): Extension<Principal>,
+    Path(app_id): Path<AppId>,
+    Json(input): Json<CreateKvTriggerRequest>,
+) -> Result<(StatusCode, Json<Trigger>), TriggersApiError> {
+    ensure_app_exists(&*s.apps, app_id).await?;
+    require(
+        s.authz.as_ref(),
+        &principal,
+        Capability::AppManageTriggers(app_id),
+    )
+    .await?;
+
+    if input.collection_glob.trim().is_empty() {
+        return Err(TriggersApiError::Invalid(
+            "collection_glob must not be empty".into(),
+        ));
+    }
+
+    let req = CreateKvTrigger {
+        script_id: input.script_id,
+        collection_glob: input.collection_glob,
+        ops: input.ops,
+        dispatch_mode: input.dispatch_mode,
+        retry_max_attempts: input
+            .retry_max_attempts
+            .unwrap_or(s.config.retry_max_attempts),
+        retry_backoff: input.retry_backoff.unwrap_or(s.config.retry_backoff),
+        retry_base_ms: input.retry_base_ms.unwrap_or(s.config.retry_base_ms),
+        registered_by_principal: principal.user_id,
+    };
+    let created = s.triggers.create_kv_trigger(app_id, req).await?;
+    Ok((StatusCode::CREATED, Json(created)))
+}
+
+async fn create_dl_trigger(
+    State(s): State<TriggersState>,
+    Extension(principal): Extension<Principal>,
+    Path(app_id): Path<AppId>,
+    Json(input): Json<CreateDeadLetterTriggerRequest>,
+) -> Result<(StatusCode, Json<Trigger>), TriggersApiError> {
+    ensure_app_exists(&*s.apps, app_id).await?;
+    require(
+        s.authz.as_ref(),
+        &principal,
+        Capability::AppManageTriggers(app_id),
+    )
+    .await?;
+    let req = CreateDeadLetterTrigger {
+        script_id: input.script_id,
+        source_filter: input.source_filter,
+        trigger_id_filter: input.trigger_id_filter,
+        script_id_filter: input.script_id_filter,
+        registered_by_principal: principal.user_id,
+    };
+    let created = s.triggers.create_dead_letter_trigger(app_id, req).await?;
+    Ok((StatusCode::CREATED, Json(created)))
+}
+
+async fn delete_trigger(
+    State(s): State<TriggersState>,
+    Extension(principal): Extension<Principal>,
+    Path((app_id, trigger_id)): Path<(AppId, TriggerId)>,
+) -> Result<StatusCode, TriggersApiError> {
+    ensure_app_exists(&*s.apps, app_id).await?;
+    // Load the trigger so we can confirm it belongs to the right
+    // app; this prevents a caller from deleting a trigger by id alone
+    // when their capability is bound to a different app.
+    let trigger = s
+        .triggers
+        .get(trigger_id)
+        .await?
+        .ok_or(TriggersApiError::NotFound(trigger_id))?;
+    if trigger.app_id != app_id {
+        return Err(TriggersApiError::NotFound(trigger_id));
+    }
+    require(
+        s.authz.as_ref(),
+        &principal,
+        Capability::AppManageTriggers(app_id),
+    )
+    .await?;
+    if !s.triggers.delete(trigger_id).await? {
+        return Err(TriggersApiError::NotFound(trigger_id));
+    }
+    Ok(StatusCode::NO_CONTENT)
+}
+
+async fn ensure_app_exists(
+    apps: &dyn AppRepository,
+    app_id: AppId,
+) -> Result<(), TriggersApiError> {
+    apps.get_by_id(app_id)
+        .await
+        .map_err(|e| TriggersApiError::Backend(e.to_string()))?
+        .ok_or_else(|| TriggersApiError::AppNotFound(app_id.to_string()))?;
+    Ok(())
+}
+
+// ----------------------------------------------------------------------------
+// Errors
+// ----------------------------------------------------------------------------
+
+#[derive(Debug, thiserror::Error)]
+pub enum TriggersApiError {
+    #[error("app not found: {0}")]
+    AppNotFound(String),
+
+    #[error("trigger not found: {0}")]
+    NotFound(TriggerId),
+
+    #[error("invalid trigger: {0}")]
+    Invalid(String),
+
+    #[error("forbidden")]
+    Forbidden,
+
+    #[error("authorization repo error: {0}")]
+    AuthzRepo(String),
+
+    #[error("trigger backend: {0}")]
+    Backend(String),
+}
+
+impl From<AuthzDenied> for TriggersApiError {
+    fn from(d: AuthzDenied) -> Self {
+        match d {
+            AuthzDenied::Denied => Self::Forbidden,
+            AuthzDenied::Repo(e) => Self::AuthzRepo(e.to_string()),
+        }
+    }
+}
+
+impl From<AuthzError> for TriggersApiError {
+    fn from(e: AuthzError) -> Self {
+        Self::AuthzRepo(e.to_string())
+    }
+}
+
+impl From<TriggerRepoError> for TriggersApiError {
+    fn from(e: TriggerRepoError) -> Self {
+        match e {
+            TriggerRepoError::NotFound(id) => Self::NotFound(id),
+            TriggerRepoError::Invalid(s) => Self::Invalid(s),
+            TriggerRepoError::Db(e) => Self::Backend(e.to_string()),
+        }
+    }
+}
+
+impl IntoResponse for TriggersApiError {
+    fn into_response(self) -> Response {
+        let (status, body) = match &self {
+            Self::AppNotFound(_) | Self::NotFound(_) => {
+                (StatusCode::NOT_FOUND, json!({ "error": self.to_string() }))
+            }
+            Self::Invalid(_) => (
+                StatusCode::UNPROCESSABLE_ENTITY,
+                json!({ "error": self.to_string() }),
+            ),
+            Self::Forbidden => (StatusCode::FORBIDDEN, json!({ "error": self.to_string() })),
+            Self::AuthzRepo(e) => {
+                tracing::error!(error = %e, "triggers authz repo error");
+                (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    json!({ "error": "internal error" }),
+                )
+            }
+            Self::Backend(e) => {
+                tracing::error!(error = %e, "triggers api backend error");
+                (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    json!({ "error": "internal error" }),
+                )
+            }
+        };
+        (status, Json(body)).into_response()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    //! In-memory tests for the trigger admin path. The Axum routing
+    //! / extractor surface is exercised by integration tests (which
+    //! need a real Postgres for the trigger repo); these tests cover
+    //! the handlers' invariant logic — capability enforcement, app
+    //! validation, default fallback for retry settings.
+
+    use super::*;
+    use crate::app_repo::{AppLookup, AppRepository};
+    use crate::trigger_repo::{
+        DeadLetterTriggerMatch, KvTriggerMatch, Trigger, TriggerDetails, TriggerRepo,
+        TriggerRepoError,
+    };
+    use async_trait::async_trait;
+    use chrono::Utc;
+    use picloud_shared::{AdminUserId, App, AppRole, KvEventOp, ScriptId, TriggerId, UserId};
+    use std::collections::HashMap;
+    use tokio::sync::Mutex;
+
+    #[derive(Default)]
+    struct InMemoryTriggerRepo {
+        inner: Mutex<HashMap<TriggerId, Trigger>>,
+    }
+
+    #[async_trait]
+    impl TriggerRepo for InMemoryTriggerRepo {
+        async fn create_kv_trigger(
+            &self,
+            app_id: AppId,
+            req: CreateKvTrigger,
+        ) -> Result<Trigger, TriggerRepoError> {
+            let now = Utc::now();
+            let id = TriggerId::new();
+            let trigger = Trigger {
+                id,
+                app_id,
+                script_id: req.script_id,
+                kind: crate::trigger_repo::TriggerKind::Kv,
+                enabled: true,
+                dispatch_mode: req.dispatch_mode,
+                retry_max_attempts: req.retry_max_attempts,
+                retry_backoff: req.retry_backoff,
+                retry_base_ms: req.retry_base_ms,
+                registered_by_principal: req.registered_by_principal,
+                created_at: now,
+                updated_at: now,
+                details: TriggerDetails::Kv {
+                    collection_glob: req.collection_glob,
+                    ops: req.ops,
+                },
+            };
+            self.inner.lock().await.insert(id, trigger.clone());
+            Ok(trigger)
+        }
+        async fn create_dead_letter_trigger(
+            &self,
+            app_id: AppId,
+            req: CreateDeadLetterTrigger,
+        ) -> Result<Trigger, TriggerRepoError> {
+            let now = Utc::now();
+            let id = TriggerId::new();
+            let trigger = Trigger {
+                id,
+                app_id,
+                script_id: req.script_id,
+                kind: crate::trigger_repo::TriggerKind::DeadLetter,
+                enabled: true,
+                dispatch_mode: TriggerDispatchMode::Async,
+                retry_max_attempts: 1,
+                retry_backoff: BackoffShape::Constant,
+                retry_base_ms: 0,
+                registered_by_principal: req.registered_by_principal,
+                created_at: now,
+                updated_at: now,
+                details: TriggerDetails::DeadLetter {
+                    source_filter: req.source_filter,
+                    trigger_id_filter: req.trigger_id_filter,
+                    script_id_filter: req.script_id_filter,
+                },
+            };
+            self.inner.lock().await.insert(id, trigger.clone());
+            Ok(trigger)
+        }
+        async fn list_for_app(&self, app_id: AppId) -> Result<Vec<Trigger>, TriggerRepoError> {
+            Ok(self
+                .inner
+                .lock()
+                .await
+                .values()
+                .filter(|t| t.app_id == app_id)
+                .cloned()
+                .collect())
+        }
+        async fn get(&self, id: TriggerId) -> Result<Option<Trigger>, TriggerRepoError> {
+            Ok(self.inner.lock().await.get(&id).cloned())
+        }
+        async fn delete(&self, id: TriggerId) -> Result<bool, TriggerRepoError> {
+            Ok(self.inner.lock().await.remove(&id).is_some())
+        }
+        async fn list_matching_kv(
+            &self,
+            _app_id: AppId,
+            _collection: &str,
+            _op: KvEventOp,
+        ) -> Result<Vec<KvTriggerMatch>, TriggerRepoError> {
+            Ok(vec![])
+        }
+        async fn list_matching_dead_letter(
+            &self,
+            _app_id: AppId,
+            _source: &str,
+            _trigger_id: Option<TriggerId>,
+            _script_id: Option<ScriptId>,
+        ) -> Result<Vec<DeadLetterTriggerMatch>, TriggerRepoError> {
+            Ok(vec![])
+        }
+    }
+
+    struct InMemoryAppRepo {
+        existing: Mutex<HashMap<AppId, App>>,
+    }
+
+    impl InMemoryAppRepo {
+        fn with(app_id: AppId) -> Arc<Self> {
+            let now = Utc::now();
+            let mut existing = HashMap::new();
+            existing.insert(
+                app_id,
+                App {
+                    id: app_id,
+                    slug: "test".into(),
+                    name: "test".into(),
+                    description: None,
+                    created_at: now,
+                    updated_at: now,
+                },
+            );
+            Arc::new(Self {
+                existing: Mutex::new(existing),
+            })
+        }
+    }
+
+    #[async_trait]
+    impl AppRepository for InMemoryAppRepo {
+        async fn create(
+            &self,
+            _slug: &str,
+            _name: &str,
+            _description: Option<&str>,
+        ) -> Result<App, crate::repo::ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn create_with_takeover(
+            &self,
+            _slug: &str,
+            _name: &str,
+            _description: Option<&str>,
+        ) -> Result<App, crate::repo::ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn slug_in_history(
+            &self,
+            _slug: &str,
+        ) -> Result<Option<App>, crate::repo::ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn list(&self) -> Result<Vec<App>, crate::repo::ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn list_for_user(
+            &self,
+            _user_id: AdminUserId,
+        ) -> Result<Vec<App>, crate::repo::ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn get_by_id(
+            &self,
+            id: AppId,
+        ) -> Result<Option<App>, crate::repo::ScriptRepositoryError> {
+            Ok(self.existing.lock().await.get(&id).cloned())
+        }
+        async fn get_by_slug(
+            &self,
+            _slug: &str,
+        ) -> Result<Option<App>, crate::repo::ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn get_by_slug_or_history(
+            &self,
+            _slug: &str,
+        ) -> Result<Option<AppLookup>, crate::repo::ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn update(
+            &self,
+            _id: AppId,
+            _name: Option<&str>,
+            _description: Option<Option<&str>>,
+        ) -> Result<App, crate::repo::ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn rename_slug(
+            &self,
+            _id: AppId,
+            _new_slug: &str,
+            _take_over_history: bool,
+        ) -> Result<App, crate::repo::ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn delete(&self, _id: AppId) -> Result<(), crate::repo::ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn delete_cascade(
+            &self,
+            _id: AppId,
+        ) -> Result<(), crate::repo::ScriptRepositoryError> {
+            unimplemented!()
+        }
+        async fn count_scripts_in_app(
+            &self,
+            _id: AppId,
+        ) -> Result<i64, crate::repo::ScriptRepositoryError> {
+            unimplemented!()
+        }
+    }
+
+    struct AlwaysAllowAuthzRepo;
+    #[async_trait]
+    impl AuthzRepo for AlwaysAllowAuthzRepo {
+        async fn membership(
+            &self,
+            _user_id: UserId,
+            _app_id: AppId,
+        ) -> Result<Option<AppRole>, AuthzError> {
+            Ok(Some(AppRole::AppAdmin))
+        }
+    }
+
+    struct AlwaysDenyAuthzRepo;
+    #[async_trait]
+    impl AuthzRepo for AlwaysDenyAuthzRepo {
+        async fn membership(
+            &self,
+            _user_id: UserId,
+            _app_id: AppId,
+        ) -> Result<Option<AppRole>, AuthzError> {
+            Ok(None)
+        }
+    }
+
+    fn member_principal() -> Principal {
+        Principal {
+            user_id: AdminUserId::new(),
+            instance_role: picloud_shared::InstanceRole::Member,
+            scopes: None,
+            app_binding: None,
+        }
+    }
+
+    fn state_with(authz: Arc<dyn AuthzRepo>, app_id: AppId) -> TriggersState {
+        TriggersState {
+            triggers: Arc::new(InMemoryTriggerRepo::default()),
+            apps: InMemoryAppRepo::with(app_id),
+            authz,
+            config: TriggerConfig::conservative(),
+        }
+    }
+
+    #[tokio::test]
+    async fn unknown_app_returns_404() {
+        let state = state_with(Arc::new(AlwaysAllowAuthzRepo), AppId::new());
+        let res = create_kv_trigger(
+            State(state),
+            Extension(member_principal()),
+            Path(AppId::new()), // a different (non-existent) app
+            Json(CreateKvTriggerRequest {
+                script_id: ScriptId::new(),
+                collection_glob: "*".into(),
+                ops: vec![],
+                dispatch_mode: TriggerDispatchMode::Async,
+                retry_max_attempts: None,
+                retry_backoff: None,
+                retry_base_ms: None,
+            }),
+        )
+        .await;
+        let err = res.expect_err("missing app should error");
+        assert!(matches!(err, TriggersApiError::AppNotFound(_)));
+    }
+
+    #[tokio::test]
+    async fn member_without_role_is_forbidden() {
+        let app_id = AppId::new();
+        let state = state_with(Arc::new(AlwaysDenyAuthzRepo), app_id);
+        let res = create_kv_trigger(
+            State(state),
+            Extension(member_principal()),
+            Path(app_id),
+            Json(CreateKvTriggerRequest {
+                script_id: ScriptId::new(),
+                collection_glob: "*".into(),
+                ops: vec![],
+                dispatch_mode: TriggerDispatchMode::Async,
+                retry_max_attempts: None,
+                retry_backoff: None,
+                retry_base_ms: None,
+            }),
+        )
+        .await;
+        let err = res.expect_err("member without role should be forbidden");
+        assert!(matches!(err, TriggersApiError::Forbidden));
+    }
+
+    #[tokio::test]
+    async fn kv_trigger_uses_env_defaults_when_omitted() {
+        let app_id = AppId::new();
+        let mut state = state_with(Arc::new(AlwaysAllowAuthzRepo), app_id);
+        // Tweak the config so we can detect that defaults were used.
+        state.config.retry_max_attempts = 7;
+        state.config.retry_base_ms = 12_345;
+        let (status, Json(trigger)) = create_kv_trigger(
+            State(state),
+            Extension(member_principal()),
+            Path(app_id),
+            Json(CreateKvTriggerRequest {
+                script_id: ScriptId::new(),
+                collection_glob: "widgets".into(),
+                ops: vec![KvEventOp::Insert],
+                dispatch_mode: TriggerDispatchMode::Async,
+                retry_max_attempts: None,
+                retry_backoff: None,
+                retry_base_ms: None,
+            }),
+        )
+        .await
+        .unwrap();
+        assert_eq!(status, StatusCode::CREATED);
+        assert_eq!(trigger.retry_max_attempts, 7);
+        assert_eq!(trigger.retry_base_ms, 12_345);
+    }
+
+    #[tokio::test]
+    async fn empty_collection_glob_rejected() {
+        let app_id = AppId::new();
+        let state = state_with(Arc::new(AlwaysAllowAuthzRepo), app_id);
+        let res = create_kv_trigger(
+            State(state),
+            Extension(member_principal()),
+            Path(app_id),
+            Json(CreateKvTriggerRequest {
+                script_id: ScriptId::new(),
+                collection_glob: "   ".into(),
+                ops: vec![],
+                dispatch_mode: TriggerDispatchMode::Async,
+                retry_max_attempts: None,
+                retry_backoff: None,
+                retry_base_ms: None,
+            }),
+        )
+        .await;
+        let err = res.expect_err("empty glob should reject");
+        assert!(matches!(err, TriggersApiError::Invalid(_)));
+    }
+
+    #[tokio::test]
+    async fn delete_rejects_cross_app_trigger_id() {
+        let app_a = AppId::new();
+        let app_b = AppId::new();
+        let state = state_with(Arc::new(AlwaysAllowAuthzRepo), app_a);
+        // Inject the app_b row into the in-memory apps repo too so
+        // the path-existence check succeeds against app_a.
+        // Insert a trigger that belongs to app_a.
+        let trigger = state
+            .triggers
+            .create_kv_trigger(
+                app_a,
+                CreateKvTrigger {
+                    script_id: ScriptId::new(),
+                    collection_glob: "*".into(),
+                    ops: vec![],
+                    dispatch_mode: TriggerDispatchMode::Async,
+                    retry_max_attempts: 3,
+                    retry_backoff: BackoffShape::Exponential,
+                    retry_base_ms: 1000,
+                    registered_by_principal: AdminUserId::new(),
+                },
+            )
+            .await
+            .unwrap();
+        let _ = app_b;
+
+        // Attempt to delete via app_b's path — should 404.
+        // First, give the in-memory app repo a record for app_b.
+        // (Otherwise we'd 404 on app-existence before reaching the
+        // cross-app check.)
+        let state = TriggersState {
+            apps: {
+                let now = Utc::now();
+                let mut existing = HashMap::new();
+                existing.insert(
+                    app_a,
+                    App {
+                        id: app_a,
+                        slug: "a".into(),
+                        name: "a".into(),
+                        description: None,
+                        created_at: now,
+                        updated_at: now,
+                    },
+                );
+                existing.insert(
+                    app_b,
+                    App {
+                        id: app_b,
+                        slug: "b".into(),
+                        name: "b".into(),
+                        description: None,
+                        created_at: now,
+                        updated_at: now,
+                    },
+                );
+                Arc::new(InMemoryAppRepo {
+                    existing: Mutex::new(existing),
+                })
+            },
+            ..state
+        };
+
+        let res = delete_trigger(
+            State(state),
+            Extension(member_principal()),
+            Path((app_b, trigger.id)),
+        )
+        .await;
+        let err = res.expect_err("cross-app delete should 404");
+        assert!(matches!(err, TriggersApiError::NotFound(_)));
+    }
+}

crates/picloud/src/lib.rs

@@ -12,13 +12,14 @@ use picloud_executor_core::{Engine, Limits};
 use picloud_manager_core::{
     admin_router, admins_router, api_keys_router, app_members_router, apps_api, apps_router,
     attach_principal_if_present, auth_router, compile_routes, migrations, require_authenticated,
-    route_admin_router, AdminSessionRepository, AdminState, AdminUserRepository, AdminsState,
+    route_admin_router, triggers_router, AdminSessionRepository, AdminState, AdminUserRepository,
-    ApiKeyRepository, ApiKeysState, AppDomainRepository, AppMembersRepository, AppMembersState,
+    AdminsState, ApiKeyRepository, ApiKeysState, AppDomainRepository, AppMembersRepository,
-    AppRepository, AppsState, AuthState, AuthzRepo, KvServiceImpl, PostgresAdminSessionRepository,
+    AppMembersState, AppRepository, AppsState, AuthState, AuthzRepo, KvServiceImpl,
-    PostgresAdminUserRepository, PostgresApiKeyRepository, PostgresAppDomainRepository,
+    PostgresAdminSessionRepository, PostgresAdminUserRepository, PostgresApiKeyRepository,
-    PostgresAppMembersRepository, PostgresAppRepository, PostgresExecutionLogRepository,
+    PostgresAppDomainRepository, PostgresAppMembersRepository, PostgresAppRepository,
-    PostgresExecutionLogSink, PostgresKvRepo, PostgresRouteRepository, PostgresScriptRepository,
+    PostgresExecutionLogRepository, PostgresExecutionLogSink, PostgresKvRepo,
-    RepoResolver, RouteAdminState, RouteRepository, SandboxCeiling,
+    PostgresRouteRepository, PostgresScriptRepository, PostgresTriggerRepo, RepoResolver,
+    RouteAdminState, RouteRepository, SandboxCeiling, TriggerConfig, TriggerRepo, TriggersState,
 };
 use picloud_orchestrator_core::routing::{AppDomainTable, RouteTable};
 use picloud_orchestrator_core::{
@@ -109,6 +110,12 @@ pub async fn build_app(pool: PgPool, auth: AuthDeps) -> anyhow::Result<Router> {
     let services = Services::new(kv, Arc::new(NoopDeadLetterService), events);
     let engine = Arc::new(Engine::new(Limits::default(), services));
 
+    // Trigger repo + config, shared between the admin endpoint and the
+    // dispatcher (commit 5). Read defaults from env so operators can
+    // tune retry / depth without rebuilding the binary.
+    let trigger_repo: Arc<dyn TriggerRepo> = Arc::new(PostgresTriggerRepo::new(pool));
+    let trigger_config = TriggerConfig::from_env();
+
     // Compile the routes table once at startup; admin writes refresh it.
     let route_table = Arc::new(RouteTable::new());
     let initial = route_repo.list_all().await?;
@@ -163,6 +170,12 @@ pub async fn build_app(pool: PgPool, auth: AuthDeps) -> anyhow::Result<Router> {
         app_domains: app_domain_table.clone(),
         routes: route_table,
     };
+    let triggers_state = TriggersState {
+        triggers: trigger_repo,
+        apps: apps_repo.clone(),
+        authz: authz.clone(),
+        config: trigger_config,
+    };
     let apps_state = AppsState {
         apps: apps_repo,
         domains: domains_repo,
@@ -204,6 +217,7 @@ pub async fn build_app(pool: PgPool, auth: AuthDeps) -> anyhow::Result<Router> {
         .merge(apps_router(apps_state))
         .merge(app_members_router(app_members_state))
         .merge(api_keys_router(api_keys_state))
+        .merge(triggers_router(triggers_state))
         .layer(from_fn_with_state(
             auth_state.clone(),
             require_authenticated,