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

feat(v1.1.2-docs): triggers framework + dispatcher + emitter extended for docs

Browse Source 
The docs trigger kind hangs off the same Layout-E shape that v1.1.1
established for KV: a parent triggers row + a docs_trigger_details
row (collection_glob TEXT + ops TEXT[]) with the empty-array =
any-op semantic preserved.

- trigger_repo.rs adds TriggerKind::Docs + TriggerDetails::Docs +
  CreateDocsTrigger + DocsTriggerMatch + PostgresTriggerRepo
  implementations of create_docs_trigger and list_matching_docs.
  list_matching_docs mirrors KV's Rust-side filter (does NOT push
  ops membership into SQL — that would exclude empty-ops rows).
- outbox_repo.rs adds OutboxSourceKind::Docs to the enum + wire form.
- dispatcher.rs's generic Kv | DeadLetter routing arm extends to
  Kv | DeadLetter | Docs. No kind-specific logic needed — the
  resolve_trigger + build_exec_request path is already abstract.
- outbox_event_emitter.rs gains a "docs" arm in the emit match plus
  emit_docs which builds TriggerEvent::Docs (carrying data +
  prev_data) and fans out across matching triggers.
- triggers_api.rs adds CreateDocsTriggerRequest + create_docs_trigger
  + the POST /api/v1/admin/apps/{id}/triggers/docs route, all
  guarded by Capability::AppManageTriggers (same as KV).

3 new triggers_api unit tests covering happy path, empty-glob
rejection, and capability denial. All existing trigger-related
tests still pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-06-02 19:55:27 +02:00
parent 06678f4496
commit ef5930910b
5 changed files with 425 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -5,7 +5,7 @@
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use picloud_shared::{AdminUserId, AppId, KvEventOp, ScriptId, TriggerId};
use picloud_shared::{AdminUserId, AppId, DocsEventOp, KvEventOp, ScriptId, TriggerId};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use uuid::Uuid;
@@ -47,6 +47,7 @@ pub struct Trigger {
#[serde(rename_all = "snake_case")]
pub enum TriggerKind {
Kv,
Docs,
DeadLetter,
}
@@ -55,6 +56,7 @@ impl TriggerKind {
pub const fn as_str(self) -> &'static str {
match self {
Self::Kv => "kv",
Self::Docs => "docs",
Self::DeadLetter => "dead_letter",
}
}
@@ -63,6 +65,7 @@ impl TriggerKind {
pub fn from_wire(s: &str) -> Option<Self> {
match s {
"kv" => Some(Self::Kv),
"docs" => Some(Self::Docs),
"dead_letter" => Some(Self::DeadLetter),
_ => None,
}
@@ -93,6 +96,10 @@ pub enum TriggerDetails {
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>,
@@ -118,6 +125,20 @@ pub struct CreateKvTrigger {
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,
@@ -141,6 +162,19 @@ pub struct KvTriggerMatch {
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)]
@@ -159,6 +193,13 @@ pub trait TriggerRepo: Send + Sync {
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,
@@ -182,6 +223,16 @@ pub trait TriggerRepo: Send + Sync {
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 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".
@@ -276,6 +327,71 @@ impl TriggerRepo for PostgresTriggerRepo {
})
}
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,
@@ -427,6 +543,54 @@ impl TriggerRepo for PostgresTriggerRepo {
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_dead_letter(
&self,
app_id: AppId,
@@ -486,6 +650,23 @@ async fn hydrate_one(pool: &PgPool, parent: TriggerRow) -> Result<Trigger, Trigg
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 \