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

feat: declarative project-tool foundation (pull/plan/apply/prune)

Browse Source 
Add a server-side, atomic, declarative reconcile loop for a single app —
the foundation of the project-tool design. Developers describe an app's
scripts, routes, triggers, and secret-names in `picloud.toml`, then
`pic pull / plan / apply [--prune]` to converge live state to the manifest.

Server (manager-core):
- apply_service: a pure diff engine (compute_diff) shared by plan and
  apply, plus an ApplyService that composes the existing per-repo writes
  into ONE Postgres transaction. Identity keys mirror the DB UNIQUE
  constraints (script=lower(name); route=(method,host_kind,host,
  path_kind,path); trigger=per-kind semantic tuple; secret=name).
  Apply takes a per-app advisory lock, recomputes the diff in-tx, applies
  scripts -> routes -> triggers, prunes dependents-first, commits, then
  refreshes the route table once post-commit.
- apply_api: POST /apps/{id}/plan (AppRead) and /apps/{id}/apply.
  Apply requires the per-kind write caps the bundle exercises (all three
  when --prune), plus AppSecretsRead when it binds an email trigger.
- tx-accepting repo siblings (insert/update/delete *_tx) so the existing
  create/update/delete delegate to one SQL definition each.
- email triggers reference an inbound secret by NAME; the value is
  resolved, decrypted (AAD-bound), and re-sealed server-side at apply —
  it never travels in the manifest.

CLI (picloud-cli):
- manifest.rs (picloud.toml model), client plan/apply, and the pull/plan/
  apply commands. pull rejects filesystem-unsafe script names up front.

Safety properties enforced and tested:
- idempotent: a freshly-pulled manifest re-applies as all-NoOp.
- atomic: a mid-bundle failure rolls back with nothing written.
- routes delete-before-insert so a freed binding is reusable in one apply.
- queue one-consumer invariant held inside the shared tx.
- email triggers are never pruned, and a script that still owns an
  email/dead-letter trigger can't be pruned (the FK cascade would destroy
  the sealed secret) — refused with a pointer to `pic triggers rm`.
- plan and apply agree on unset email-secret references.

No migration: the existing schema's UNIQUE constraints serve as identity
keys. Groups, env-scoping, and the `enabled` toggle are later milestones.

Tested: manager-core lib (360) + CLI bins (27) + 8 project-tool journeys
(pull/plan/apply/prune/email+queue), all green; clippy -D warnings clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-06-20 21:52:21 +02:00
parent c600177fd6
commit 3b650a2b14
21 changed files with 4055 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -502,6 +502,220 @@ impl PostgresTriggerRepo {
}
}
/// Insert a trigger (parent row + per-kind detail) within an existing
/// transaction — used by the declarative `apply` engine. Supports the
/// five settled kinds; `email`/`queue`/`dead_letter` have their own
/// create paths and are rejected here.
#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
pub(crate) async fn insert_trigger_tx(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
app_id: AppId,
script_id: ScriptId,
registered_by: AdminUserId,
dispatch_mode: TriggerDispatchMode,
retry_max_attempts: u32,
retry_backoff: BackoffShape,
retry_base_ms: u32,
details: &TriggerDetails,
) -> Result<TriggerId, TriggerRepoError> {
let kind = match details {
TriggerDetails::Kv { .. } => "kv",
TriggerDetails::Docs { .. } => "docs",
TriggerDetails::Files { .. } => "files",
TriggerDetails::Cron { .. } => "cron",
TriggerDetails::Pubsub { .. } => "pubsub",
TriggerDetails::Queue { .. } => "queue",
TriggerDetails::DeadLetter { .. } | TriggerDetails::Email { .. } => {
return Err(TriggerRepoError::Invalid(
"trigger kind not supported by declarative apply".into(),
));
}
};
// 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 {
sqlx::query("SELECT pg_advisory_xact_lock($1)")
.bind(advisory_lock_key(app_id, queue_name))
.execute(&mut **tx)
.await?;
let existing: Option<(Uuid,)> = sqlx::query_as(
"SELECT t.id FROM triggers t \
JOIN queue_trigger_details d ON d.trigger_id = t.id \
WHERE t.app_id = $1 AND t.kind = 'queue' AND d.queue_name = $2",
)
.bind(app_id.into_inner())
.bind(queue_name)
.fetch_optional(&mut **tx)
.await?;
if existing.is_some() {
return Err(TriggerRepoError::Invalid(format!(
"queue '{queue_name}' already has a consumer trigger; remove the existing one first"
)));
}
}
let row: (Uuid,) = sqlx::query_as(
"INSERT INTO triggers ( \
app_id, 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",
)
.bind(app_id.into_inner())
.bind(script_id.into_inner())
.bind(kind)
.bind(dispatch_mode.as_str())
.bind(i32::try_from(retry_max_attempts).unwrap_or(3))
.bind(retry_backoff.as_str())
.bind(i32::try_from(retry_base_ms).unwrap_or(1000))
.bind(registered_by.into_inner())
.fetch_one(&mut **tx)
.await?;
let tid = row.0;
match details {
TriggerDetails::Kv {
collection_glob,
ops,
} => {
let ops_str: Vec<String> = ops.iter().map(|o| o.as_str().to_string()).collect();
sqlx::query(
"INSERT INTO kv_trigger_details (trigger_id, collection_glob, ops) \
VALUES ($1, $2, $3)",
)
.bind(tid)
.bind(collection_glob)
.bind(&ops_str)
.execute(&mut **tx)
.await?;
}
TriggerDetails::Docs {
collection_glob,
ops,
} => {
let ops_str: Vec<String> = ops.iter().map(|o| o.as_str().to_string()).collect();
sqlx::query(
"INSERT INTO docs_trigger_details (trigger_id, collection_glob, ops) \
VALUES ($1, $2, $3)",
)
.bind(tid)
.bind(collection_glob)
.bind(&ops_str)
.execute(&mut **tx)
.await?;
}
TriggerDetails::Files {
collection_glob,
ops,
} => {
let ops_str: Vec<String> = ops.iter().map(|o| o.as_str().to_string()).collect();
sqlx::query(
"INSERT INTO files_trigger_details (trigger_id, collection_glob, ops) \
VALUES ($1, $2, $3)",
)
.bind(tid)
.bind(collection_glob)
.bind(&ops_str)
.execute(&mut **tx)
.await?;
}
TriggerDetails::Cron {
schedule, timezone, ..
} => {
sqlx::query(
"INSERT INTO cron_trigger_details (trigger_id, schedule, timezone) \
VALUES ($1, $2, $3)",
)
.bind(tid)
.bind(schedule)
.bind(timezone)
.execute(&mut **tx)
.await?;
}
TriggerDetails::Pubsub { topic_pattern } => {
sqlx::query(
"INSERT INTO pubsub_trigger_details (trigger_id, topic_pattern) VALUES ($1, $2)",
)
.bind(tid)
.bind(topic_pattern)
.execute(&mut **tx)
.await?;
}
TriggerDetails::Queue {
queue_name,
visibility_timeout_secs,
..
} => {
sqlx::query(
"INSERT INTO queue_trigger_details \
(trigger_id, queue_name, visibility_timeout_secs) \
VALUES ($1, $2, $3)",
)
.bind(tid)
.bind(queue_name)
.bind(i32::try_from(*visibility_timeout_secs).unwrap_or(30))
.execute(&mut **tx)
.await?;
}
TriggerDetails::DeadLetter { .. } | TriggerDetails::Email { .. } => {
unreachable!("guarded above")
}
}
Ok(tid.into())
}
/// Insert an email trigger within a transaction. The inbound HMAC secret
/// is sealed by the apply engine (resolved from the app's secret store);
/// this writes the ciphertext. Parent retry settings match the
/// interactive `create_email_trigger` path (async, 3, exponential, 1000).
pub(crate) async fn insert_email_trigger_tx(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
app_id: AppId,
script_id: ScriptId,
registered_by: AdminUserId,
inbound_secret_encrypted: &[u8],
inbound_secret_nonce: &[u8],
) -> Result<TriggerId, TriggerRepoError> {
let row: (Uuid,) = sqlx::query_as(
"INSERT INTO triggers ( \
app_id, script_id, kind, enabled, dispatch_mode, \
retry_max_attempts, retry_backoff, retry_base_ms, \
registered_by_principal \
) VALUES ($1, $2, 'email', TRUE, 'async', 3, 'exponential', 1000, $3) RETURNING id",
)
.bind(app_id.into_inner())
.bind(script_id.into_inner())
.bind(registered_by.into_inner())
.fetch_one(&mut **tx)
.await?;
sqlx::query(
"INSERT INTO email_trigger_details \
(trigger_id, inbound_secret_encrypted, inbound_secret_nonce) \
VALUES ($1, $2, $3)",
)
.bind(row.0)
.bind(inbound_secret_encrypted)
.bind(inbound_secret_nonce)
.execute(&mut **tx)
.await?;
Ok(row.0.into())
}
/// Delete a trigger by id within an existing transaction (its detail row
/// cascades via the FK). Used by `apply --prune`.
pub(crate) async fn delete_trigger_tx(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
id: TriggerId,
) -> Result<(), TriggerRepoError> {
sqlx::query("DELETE FROM triggers WHERE id = $1")
.bind(id.into_inner())
.execute(&mut **tx)
.await?;
Ok(())
}
#[async_trait]
impl TriggerRepo for PostgresTriggerRepo {
async fn create_kv_trigger(