fabi/PiCloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(modules): author group trigger templates (event kinds) (§11 tail T2)

Browse Source 
A [group] node may now declare EVENT trigger templates (kv/docs/files/
pubsub) that bind a group-owned handler. cron/queue/email stay app-only
(rejected on a group at both the manifest and the reconcile layer).

- Trigger is now owner-polymorphic (app_id: Option, group_id: Option),
  mirroring Script's Phase-4 reshape; TriggerRow carries group_id
  (#[sqlx(default)] so interactive RETURNING clauses still hydrate).
- insert_trigger_tx takes a ScriptOwner (App XOR Group) and writes the
  group_id column; the queue advisory-lock branch is app-only.
- TriggerRepo::list_for_group loads a group's templates; load_current's
  Group arm uses it so the diff is NoOp on re-apply and prune-able.
- apply_service reconcile drops the "triggers are app-only" assumption;
  validate_bundle_for rejects non-event kinds on a group. The semantic-
  identity diff is already per-owner.
- CLI manifest allows event-kind [[triggers]] on [group], rejects
  cron/queue/email there.

Templates don't fire yet — the live dispatch union lands in T3. All
trigger/apply/manifest unit tests green; clippy -D clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-06-30 20:38:28 +02:00
parent 7f51087d5d
commit 547004e32d
4 changed files with 145 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
crates/manager-core/src/trigger_repo.rs
View File

@@ -7,7 +7,8 @@ use async_trait::async_trait;
use chrono::{DateTime, Utc};
use futures::stream::{self, TryStreamExt};
use picloud_shared::{
AdminUserId, AppId, DocsEventOp, FilesEventOp, KvEventOp, ScriptId, TriggerId,
AdminUserId, AppId, DocsEventOp, FilesEventOp, GroupId, KvEventOp, ScriptId, ScriptOwner,
TriggerId,
};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
@@ -32,7 +33,10 @@ pub enum TriggerRepoError {
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Trigger {
pub id: TriggerId,
pub app_id: AppId,
/// 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,
@@ -414,6 +418,13 @@ pub trait TriggerRepo: Send + Sync {
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>;
@@ -511,7 +522,7 @@ impl PostgresTriggerRepo {
#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
pub(crate) async fn insert_trigger_tx(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
app_id: AppId,
owner: ScriptOwner,
script_id: ScriptId,
registered_by: AdminUserId,
dispatch_mode: TriggerDispatchMode,
@@ -533,12 +544,20 @@ pub(crate) async fn insert_trigger_tx(
));
}
};
// §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).
if let TriggerDetails::Queue { queue_name, .. } = details {
// 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)
@@ -560,12 +579,13 @@ pub(crate) async fn insert_trigger_tx(
}
let row: (Uuid,) = sqlx::query_as(
"INSERT INTO triggers ( \
app_id, script_id, kind, enabled, dispatch_mode, \
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, TRUE, $4, $5, $6, $7, $8) RETURNING id",
) VALUES ($1, $2, $3, $4, TRUE, $5, $6, $7, $8, $9) RETURNING id",
)
.bind(app_id.into_inner())
.bind(owner_app_id)
.bind(owner_group_id)
.bind(script_id.into_inner())
.bind(kind)
.bind(dispatch_mode.as_str())
@@ -766,7 +786,8 @@ impl TriggerRepo for PostgresTriggerRepo {
Ok(Trigger {
id: parent.id.into(),
app_id: parent.app_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,
@@ -832,7 +853,8 @@ impl TriggerRepo for PostgresTriggerRepo {
Ok(Trigger {
id: parent.id.into(),
app_id: parent.app_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,
@@ -893,7 +915,8 @@ impl TriggerRepo for PostgresTriggerRepo {
Ok(Trigger {
id: parent.id.into(),
app_id: parent.app_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,
@@ -960,7 +983,8 @@ impl TriggerRepo for PostgresTriggerRepo {
Ok(Trigger {
id: parent.id.into(),
app_id: parent.app_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,
@@ -1027,7 +1051,8 @@ impl TriggerRepo for PostgresTriggerRepo {
Ok(Trigger {
id: parent.id.into(),
app_id: parent.app_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,
@@ -1089,7 +1114,8 @@ impl TriggerRepo for PostgresTriggerRepo {
Ok(Trigger {
id: parent.id.into(),
app_id: parent.app_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,
@@ -1149,7 +1175,8 @@ impl TriggerRepo for PostgresTriggerRepo {
Ok(Trigger {
id: parent.id.into(),
app_id: parent.app_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,
@@ -1220,6 +1247,28 @@ impl TriggerRepo for PostgresTriggerRepo {
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, 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> {
let parent: Option<TriggerRow> = sqlx::query_as(
"SELECT id, app_id, script_id, name, kind, enabled, dispatch_mode, \
@@ -1501,7 +1550,8 @@ impl TriggerRepo for PostgresTriggerRepo {
Ok(Trigger {
id: parent.id.into(),
app_id: parent.app_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,
@@ -1709,7 +1759,8 @@ async fn hydrate_one(pool: &PgPool, parent: TriggerRow) -> Result<Trigger, Trigg
Ok(Trigger {
id: parent.id.into(),
app_id: parent.app_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,
@@ -1752,7 +1803,12 @@ pub fn collection_matches(glob: &str, collection: &str) -> bool {
#[derive(sqlx::FromRow)]
struct TriggerRow {
id: Uuid,
app_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,