c8acac1d20
Lift the Phase-4-lite restrictions now that import resolution is lexical: - `create_group_script` (group_scripts_api): allow `kind = module` and group scripts with `import`s; record the import edges. Module-shape and sandbox-ceiling validation stay. - The declarative apply path already reconciles group modules generically (`BundleScript.kind` + `owner.script_owner()` flow through `reconcile_node_tx`); no change needed beyond stale-comment fixes. Adds the §5.5 dangling-import plan check (single-node plan/apply): `check_imports_resolve` rejects an `import "<m>"` that resolves to neither a bundle-local module nor a module reachable lexically up the node's chain — caught at plan time instead of as a runtime 404. Roots resolution at the node's owner (an app node reaches ancestor group modules; a group node walks its own ancestry). ApplyService gains an injected `ModuleSource`. The tree path opts out (a node may import a module created by another node in the same uncommitted tx); the runtime check is the backstop there. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
3525 lines
130 KiB
Rust
3525 lines
130 KiB
Rust
//! Declarative reconcile engine for a single app.
|
||
//!
|
||
//! Takes a desired-state [`Bundle`] (an app's scripts, routes, triggers,
|
||
//! and the names of its secrets) and either diffs it against the live
|
||
//! database ([`ApplyService::plan`], read-only) or reconciles it in one
|
||
//! transaction ([`ApplyService::apply`]).
|
||
//!
|
||
//! Identity keys match the DB UNIQUE constraints so the diff is stable:
|
||
//! scripts by `lower(name)`, routes by the `(method, host, path)` tuple,
|
||
//! triggers by their per-kind semantic tuple, secrets by name.
|
||
//!
|
||
//! **Trigger-identity limitation:** a trigger's identity *is* its semantic
|
||
//! definition, so the diff only ever Creates or Deletes (a change is
|
||
//! delete-old + create-new). For queue the identity is `queue|{queue_name}`
|
||
//! (script-independent — it encodes the one-consumer-per-queue invariant)
|
||
//! and for email it is `email|{script}`. Consequence: rebinding a queue to
|
||
//! a different script, changing its visibility timeout, or rotating an email
|
||
//! trigger's secret/reference diffs as a **NoOp and is not applied**. To
|
||
//! change those, recreate the trigger (drop it from the manifest, apply
|
||
//! `--prune`, then re-add it).
|
||
//!
|
||
//! Other known limitations vs. the interactive API: a script **rename**
|
||
//! diffs as delete+create, so under `--prune` it loses the old script's id,
|
||
//! version counter, and execution logs; an `endpoint`→`module` kind flip is
|
||
//! not guarded against *live* (un-pruned) routes/triggers still bound to the
|
||
//! script; and cross-pattern route conflict-vs-live is not checked (only
|
||
//! identical tuples collide). See `validate_bundle`.
|
||
|
||
use std::collections::{BTreeSet, HashMap, HashSet};
|
||
use std::sync::Arc;
|
||
|
||
use picloud_orchestrator_core::routing::{pattern, RouteTable};
|
||
use picloud_shared::{
|
||
AdminUserId, AppId, DispatchMode, DocsEventOp, FilesEventOp, GroupId, HostKind, KvEventOp,
|
||
MasterKey, PathKind, Route, Script, ScriptId, ScriptKind, ScriptSandbox, ScriptValidator,
|
||
TriggerId,
|
||
};
|
||
use serde::{Deserialize, Serialize};
|
||
use sqlx::PgPool;
|
||
|
||
use crate::app_domain_repo::AppDomainRepository;
|
||
use crate::app_repo::AppRepository;
|
||
use crate::authz::AuthzRepo;
|
||
use crate::repo::{
|
||
delete_script_tx, insert_script_tx, update_script_tx, NewScript, ScriptPatch, ScriptRepository,
|
||
ScriptRepositoryError,
|
||
};
|
||
use crate::route_repo::{delete_route_tx, insert_route_tx, NewRoute, RouteRepository};
|
||
use crate::sandbox::SandboxCeiling;
|
||
use crate::secrets_repo::SecretsRepo;
|
||
use crate::trigger_config::TriggerConfig;
|
||
use crate::trigger_repo::{
|
||
delete_trigger_tx, insert_email_trigger_tx, insert_trigger_tx, Trigger, TriggerDetails,
|
||
TriggerDispatchMode, TriggerRepo, TriggerRepoError,
|
||
};
|
||
use crate::vars_repo::{VarOwner, VarsRepo};
|
||
|
||
/// Reserved path prefixes that user routes may not claim (mirrors the
|
||
/// route-create rejection — see CLAUDE.md "Reserved path prefixes").
|
||
const RESERVED_PATH_PREFIXES: &[&str] = &["/api/", "/admin/"];
|
||
const RESERVED_PATH_EXACT: &[&str] = &["/healthz", "/version"];
|
||
|
||
// ----------------------------------------------------------------------------
|
||
// Wire types — desired state (request) and the computed plan (response)
|
||
// ----------------------------------------------------------------------------
|
||
|
||
/// Desired state for one app, sent by `pic plan` / `pic apply`.
|
||
#[derive(Debug, Clone, Deserialize)]
|
||
pub struct Bundle {
|
||
#[serde(default)]
|
||
pub scripts: Vec<BundleScript>,
|
||
#[serde(default)]
|
||
pub routes: Vec<BundleRoute>,
|
||
#[serde(default)]
|
||
pub triggers: Vec<BundleTrigger>,
|
||
/// Declared secret *names* only; values are pushed out-of-band.
|
||
#[serde(default)]
|
||
pub secrets: Vec<String>,
|
||
/// Declared app-owned config `vars` — key → JSON value. Unlike secrets,
|
||
/// var values are non-secret and live inline in the manifest. Written at
|
||
/// app scope `*` (an app *is* an environment, §2 — per-env scoping is a
|
||
/// group concern, deferred to the nested-manifest model). Tombstones are
|
||
/// not expressible here; use `pic vars set --tombstone`.
|
||
#[serde(default)]
|
||
pub vars: std::collections::BTreeMap<String, serde_json::Value>,
|
||
}
|
||
|
||
#[derive(Debug, Clone, Deserialize)]
|
||
pub struct BundleScript {
|
||
pub name: String,
|
||
pub source: String,
|
||
#[serde(default)]
|
||
pub kind: ScriptKind,
|
||
#[serde(default)]
|
||
pub description: Option<String>,
|
||
#[serde(default)]
|
||
pub timeout_seconds: Option<i32>,
|
||
#[serde(default)]
|
||
pub memory_limit_mb: Option<i32>,
|
||
#[serde(default)]
|
||
pub sandbox: Option<ScriptSandbox>,
|
||
/// Three-state lifecycle (§4.3); omitted ⇒ active. Declarative (not
|
||
/// leave-as-is): a script absent the key is `enabled = true`.
|
||
#[serde(default = "picloud_shared::default_true")]
|
||
pub enabled: bool,
|
||
}
|
||
|
||
#[derive(Debug, Clone, Deserialize)]
|
||
pub struct BundleRoute {
|
||
/// Name of the script this route binds to.
|
||
pub script: String,
|
||
#[serde(default)]
|
||
pub method: Option<String>,
|
||
pub host_kind: HostKind,
|
||
#[serde(default)]
|
||
pub host: String,
|
||
#[serde(default)]
|
||
pub host_param_name: Option<String>,
|
||
pub path_kind: PathKind,
|
||
pub path: String,
|
||
#[serde(default)]
|
||
pub dispatch_mode: DispatchMode,
|
||
/// Three-state lifecycle (§4.3); omitted ⇒ active.
|
||
#[serde(default = "picloud_shared::default_true")]
|
||
pub enabled: bool,
|
||
}
|
||
|
||
/// Desired trigger, tagged by kind on the wire (`{ "kind": "cron", … }`).
|
||
#[derive(Debug, Clone, Deserialize)]
|
||
#[serde(tag = "kind", rename_all = "snake_case")]
|
||
pub enum BundleTrigger {
|
||
Kv {
|
||
script: String,
|
||
collection_glob: String,
|
||
#[serde(default)]
|
||
ops: Vec<KvEventOp>,
|
||
#[serde(default)]
|
||
dispatch_mode: Option<TriggerDispatchMode>,
|
||
#[serde(default)]
|
||
retry_max_attempts: Option<u32>,
|
||
},
|
||
Docs {
|
||
script: String,
|
||
collection_glob: String,
|
||
#[serde(default)]
|
||
ops: Vec<DocsEventOp>,
|
||
#[serde(default)]
|
||
dispatch_mode: Option<TriggerDispatchMode>,
|
||
#[serde(default)]
|
||
retry_max_attempts: Option<u32>,
|
||
},
|
||
Files {
|
||
script: String,
|
||
collection_glob: String,
|
||
#[serde(default)]
|
||
ops: Vec<FilesEventOp>,
|
||
#[serde(default)]
|
||
dispatch_mode: Option<TriggerDispatchMode>,
|
||
#[serde(default)]
|
||
retry_max_attempts: Option<u32>,
|
||
},
|
||
Cron {
|
||
script: String,
|
||
schedule: String,
|
||
#[serde(default = "default_timezone")]
|
||
timezone: String,
|
||
#[serde(default)]
|
||
dispatch_mode: Option<TriggerDispatchMode>,
|
||
#[serde(default)]
|
||
retry_max_attempts: Option<u32>,
|
||
},
|
||
Pubsub {
|
||
script: String,
|
||
topic_pattern: String,
|
||
#[serde(default)]
|
||
dispatch_mode: Option<TriggerDispatchMode>,
|
||
#[serde(default)]
|
||
retry_max_attempts: Option<u32>,
|
||
},
|
||
Email {
|
||
script: String,
|
||
/// Name of the secret (set via `pic secret set`) holding the
|
||
/// inbound HMAC value; resolved + sealed at apply time. Never the
|
||
/// value itself.
|
||
inbound_secret_ref: String,
|
||
#[serde(default)]
|
||
dispatch_mode: Option<TriggerDispatchMode>,
|
||
#[serde(default)]
|
||
retry_max_attempts: Option<u32>,
|
||
},
|
||
Queue {
|
||
script: String,
|
||
queue_name: String,
|
||
#[serde(default)]
|
||
visibility_timeout_secs: Option<u32>,
|
||
#[serde(default)]
|
||
dispatch_mode: Option<TriggerDispatchMode>,
|
||
#[serde(default)]
|
||
retry_max_attempts: Option<u32>,
|
||
},
|
||
}
|
||
|
||
fn default_timezone() -> String {
|
||
"UTC".to_string()
|
||
}
|
||
|
||
impl BundleTrigger {
|
||
/// The script name this trigger fires.
|
||
#[must_use]
|
||
pub fn script(&self) -> &str {
|
||
match self {
|
||
Self::Kv { script, .. }
|
||
| Self::Docs { script, .. }
|
||
| Self::Files { script, .. }
|
||
| Self::Cron { script, .. }
|
||
| Self::Pubsub { script, .. }
|
||
| Self::Email { script, .. }
|
||
| Self::Queue { script, .. } => script,
|
||
}
|
||
}
|
||
|
||
/// Whether this trigger is an email trigger, which resolves and decrypts
|
||
/// a stored secret by reference at apply time (gating an extra capability).
|
||
#[must_use]
|
||
pub fn is_email(&self) -> bool {
|
||
matches!(self, Self::Email { .. })
|
||
}
|
||
|
||
/// Stable semantic identity (the per-kind tuple), used for the diff.
|
||
#[must_use]
|
||
pub fn identity(&self) -> String {
|
||
match self {
|
||
Self::Kv {
|
||
script,
|
||
collection_glob,
|
||
ops,
|
||
..
|
||
} => format!(
|
||
"kv|{script}|{collection_glob}|{}",
|
||
sorted_csv(ops.iter().copied().map(KvEventOp::as_str))
|
||
),
|
||
Self::Docs {
|
||
script,
|
||
collection_glob,
|
||
ops,
|
||
..
|
||
} => format!(
|
||
"docs|{script}|{collection_glob}|{}",
|
||
sorted_csv(ops.iter().copied().map(DocsEventOp::as_str))
|
||
),
|
||
Self::Files {
|
||
script,
|
||
collection_glob,
|
||
ops,
|
||
..
|
||
} => format!(
|
||
"files|{script}|{collection_glob}|{}",
|
||
sorted_csv(ops.iter().copied().map(FilesEventOp::as_str))
|
||
),
|
||
Self::Cron {
|
||
script,
|
||
schedule,
|
||
timezone,
|
||
..
|
||
} => format!("cron|{script}|{schedule}|{timezone}"),
|
||
Self::Pubsub {
|
||
script,
|
||
topic_pattern,
|
||
..
|
||
} => format!("pubsub|{script}|{topic_pattern}"),
|
||
Self::Email { script, .. } => format!("email|{script}"),
|
||
Self::Queue { queue_name, .. } => format!("queue|{queue_name}"),
|
||
}
|
||
}
|
||
|
||
#[must_use]
|
||
pub fn kind_str(&self) -> &'static str {
|
||
match self {
|
||
Self::Kv { .. } => "kv",
|
||
Self::Docs { .. } => "docs",
|
||
Self::Files { .. } => "files",
|
||
Self::Cron { .. } => "cron",
|
||
Self::Pubsub { .. } => "pubsub",
|
||
Self::Email { .. } => "email",
|
||
Self::Queue { .. } => "queue",
|
||
}
|
||
}
|
||
}
|
||
|
||
/// One change in the plan. `key` is the human-renderable identity.
|
||
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
|
||
pub struct ResourceChange {
|
||
pub op: Op,
|
||
pub key: String,
|
||
/// Optional one-line note (e.g. why an update, or "value must be set").
|
||
#[serde(skip_serializing_if = "Option::is_none")]
|
||
pub detail: Option<String>,
|
||
}
|
||
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
|
||
#[serde(rename_all = "lowercase")]
|
||
pub enum Op {
|
||
Create,
|
||
Update,
|
||
NoOp,
|
||
Delete,
|
||
}
|
||
|
||
/// The computed diff, grouped by resource kind.
|
||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
|
||
pub struct Plan {
|
||
pub scripts: Vec<ResourceChange>,
|
||
pub routes: Vec<ResourceChange>,
|
||
pub triggers: Vec<ResourceChange>,
|
||
pub secrets: Vec<ResourceChange>,
|
||
#[serde(default)]
|
||
pub vars: Vec<ResourceChange>,
|
||
}
|
||
|
||
impl Plan {
|
||
/// True when every resource is a no-op (nothing to apply).
|
||
#[must_use]
|
||
pub fn is_noop(&self) -> bool {
|
||
self.scripts
|
||
.iter()
|
||
.chain(&self.routes)
|
||
.chain(&self.triggers)
|
||
.chain(&self.secrets)
|
||
.chain(&self.vars)
|
||
.all(|c| c.op == Op::NoOp)
|
||
}
|
||
}
|
||
|
||
/// What `plan` returns: the diff plus a fingerprint of the live state it was
|
||
/// computed against. The token is flattened onto the plan JSON, so the wire
|
||
/// shape stays `{ scripts, routes, triggers, secrets, state_token }`.
|
||
#[derive(Debug, Clone, Serialize)]
|
||
pub struct PlanResult {
|
||
#[serde(flatten)]
|
||
pub plan: Plan,
|
||
pub state_token: String,
|
||
}
|
||
|
||
// ----------------------------------------------------------------------------
|
||
// Tree apply (Phase 5): a whole project subtree applied in one transaction.
|
||
// ----------------------------------------------------------------------------
|
||
|
||
/// Whether a tree node is an app or a group.
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||
#[serde(rename_all = "lowercase")]
|
||
pub enum NodeKind {
|
||
App,
|
||
Group,
|
||
}
|
||
|
||
/// One node of a project tree: its kind, slug (must pre-exist on the server),
|
||
/// and its desired-state bundle (a group's bundle has no routes/triggers).
|
||
/// Request-only (deserialized from the CLI's tree bundle).
|
||
#[derive(Debug, Clone, Deserialize)]
|
||
pub struct TreeNode {
|
||
pub kind: NodeKind,
|
||
pub slug: String,
|
||
pub bundle: Bundle,
|
||
}
|
||
|
||
/// A whole project subtree submitted to `apply_tree`.
|
||
#[derive(Debug, Clone, Default, Deserialize)]
|
||
pub struct TreeBundle {
|
||
#[serde(default)]
|
||
pub nodes: Vec<TreeNode>,
|
||
}
|
||
|
||
/// One node's diff within a tree plan.
|
||
#[derive(Debug, Clone, Serialize)]
|
||
pub struct NodePlan {
|
||
pub kind: NodeKind,
|
||
pub slug: String,
|
||
#[serde(flatten)]
|
||
pub plan: Plan,
|
||
}
|
||
|
||
/// What `plan_tree` returns: a per-node diff plus ONE combined bound-plan token
|
||
/// (fingerprint of every node's content + every in-scope group's structure
|
||
/// version, so a reparent or content edit between plan and apply is caught).
|
||
#[derive(Debug, Clone, Serialize)]
|
||
pub struct TreePlanResult {
|
||
pub nodes: Vec<NodePlan>,
|
||
pub state_token: String,
|
||
}
|
||
|
||
// ----------------------------------------------------------------------------
|
||
// Current state snapshot
|
||
// ----------------------------------------------------------------------------
|
||
|
||
/// The app's live state, loaded once for a diff.
|
||
#[derive(Debug, Default)]
|
||
pub struct CurrentState {
|
||
pub scripts: Vec<Script>,
|
||
pub routes: Vec<Route>,
|
||
pub triggers: Vec<Trigger>,
|
||
pub secret_names: Vec<String>,
|
||
/// App's OWN `vars` at scope `*` (key, value), non-tombstone — the set
|
||
/// the manifest reconciles. Inherited group vars and tombstones are
|
||
/// excluded (the manifest manages only the app's own real values).
|
||
pub vars: Vec<(String, serde_json::Value)>,
|
||
}
|
||
|
||
// ----------------------------------------------------------------------------
|
||
// Errors
|
||
// ----------------------------------------------------------------------------
|
||
|
||
#[derive(Debug, thiserror::Error)]
|
||
pub enum ApplyError {
|
||
#[error("app not found: {0}")]
|
||
AppNotFound(String),
|
||
#[error("invalid manifest: {0}")]
|
||
Invalid(String),
|
||
#[error(
|
||
"live state changed since `pic plan` (someone edited this app's scripts, \
|
||
routes, triggers, or secrets); re-run `pic plan` to review, then apply — \
|
||
or `pic apply --force` to skip the check"
|
||
)]
|
||
StateMoved,
|
||
#[error("forbidden")]
|
||
Forbidden,
|
||
#[error("authorization repo error: {0}")]
|
||
AuthzRepo(String),
|
||
#[error("backend: {0}")]
|
||
Backend(String),
|
||
}
|
||
|
||
// ----------------------------------------------------------------------------
|
||
// Service
|
||
// ----------------------------------------------------------------------------
|
||
|
||
/// Reconcile engine. Holds trait-object repos for the read/diff path; the
|
||
/// transactional write path (`apply`) reuses the same handles plus `pool`.
|
||
#[derive(Clone)]
|
||
pub struct ApplyService {
|
||
/// Pool for the transactional write path (`apply`).
|
||
pub pool: PgPool,
|
||
pub scripts: Arc<dyn ScriptRepository>,
|
||
pub routes: Arc<dyn RouteRepository>,
|
||
pub triggers: Arc<dyn TriggerRepo>,
|
||
pub secrets: Arc<dyn SecretsRepo>,
|
||
/// App-owned `vars` reconciliation (read/diff path; the transactional
|
||
/// write goes through `set_app_var_tx` / `delete_app_var_tx`).
|
||
pub vars: Arc<dyn VarsRepo>,
|
||
pub apps: Arc<dyn AppRepository>,
|
||
/// Group lookups (Phase 5): resolve a group slug→id for group/tree apply.
|
||
pub groups: Arc<dyn crate::group_repo::GroupRepository>,
|
||
/// Module resolver (Phase 4b): lexical, origin-aware module lookup used by
|
||
/// the single-node dangling-import plan check (§5.5).
|
||
pub modules: Arc<dyn picloud_shared::ModuleSource>,
|
||
/// App domain claims — validates a route's host belongs to the app.
|
||
pub domains: Arc<dyn AppDomainRepository>,
|
||
pub authz: Arc<dyn AuthzRepo>,
|
||
pub validator: Arc<dyn ScriptValidator>,
|
||
pub sandbox_ceiling: SandboxCeiling,
|
||
/// Default retry/dispatch settings filled when a bundle trigger omits them.
|
||
pub trigger_config: TriggerConfig,
|
||
/// Shared route snapshot, rebuilt once after a successful apply.
|
||
pub route_table: Arc<RouteTable>,
|
||
/// Master key — seals email triggers' resolved inbound secrets.
|
||
pub master_key: MasterKey,
|
||
}
|
||
|
||
impl ApplyService {
|
||
/// Compute the diff between `bundle` and app `app_id`'s live state.
|
||
/// Read-only: validates the bundle, loads current state, diffs.
|
||
///
|
||
/// # Errors
|
||
/// `Invalid` for a malformed bundle; `Backend` for repo failures.
|
||
pub async fn plan(&self, app_id: AppId, bundle: &Bundle) -> Result<PlanResult, ApplyError> {
|
||
self.plan_owner(ApplyOwner::App(app_id), bundle).await
|
||
}
|
||
|
||
/// Owner-generic plan (Phase 5): diff `bundle` against an app OR group node.
|
||
/// A group node has only scripts + vars — routes/triggers are rejected, and
|
||
/// the app-only inherited-target + route-host validation is skipped.
|
||
///
|
||
/// # Errors
|
||
/// `Invalid` for a malformed bundle; `Backend` for repo failures.
|
||
pub async fn plan_owner(
|
||
&self,
|
||
owner: ApplyOwner,
|
||
bundle: &Bundle,
|
||
) -> Result<PlanResult, ApplyError> {
|
||
let inherited = self.resolve_inherited_targets_for(owner, bundle).await?;
|
||
self.validate_bundle_for(owner, bundle, &keys_set(&inherited))?;
|
||
self.check_imports_resolve(owner, bundle).await?;
|
||
if let ApplyOwner::App(app_id) = owner {
|
||
self.validate_route_hosts(app_id, bundle).await?;
|
||
}
|
||
let current = self.load_current(owner).await?;
|
||
let current_names = self.resolve_current_target_names(¤t).await?;
|
||
validate_email_secrets_present(bundle, ¤t.secret_names)?;
|
||
Ok(PlanResult {
|
||
plan: compute_diff_with_names(¤t, bundle, ¤t_names),
|
||
// Fingerprint of the live state this plan was computed against, so
|
||
// `apply` can refuse if the node changed underneath it (§4.2).
|
||
state_token: state_token_with_names(¤t, ¤t_names),
|
||
})
|
||
}
|
||
|
||
/// Inherited group-script targets for an app node; a group node has no
|
||
/// routes/triggers to bind, so there is nothing to inherit (empty map).
|
||
async fn resolve_inherited_targets_for(
|
||
&self,
|
||
owner: ApplyOwner,
|
||
bundle: &Bundle,
|
||
) -> Result<HashMap<String, ScriptId>, ApplyError> {
|
||
match owner {
|
||
ApplyOwner::App(app_id) => self.resolve_inherited_targets(app_id, bundle).await,
|
||
ApplyOwner::Group(_) => Ok(HashMap::new()),
|
||
}
|
||
}
|
||
|
||
/// `validate_bundle`, plus a group-node guard: a group owns only scripts +
|
||
/// vars (+ declared secret names), so a `[group]` manifest carrying routes
|
||
/// or triggers is a 422 — those are app concerns.
|
||
fn validate_bundle_for(
|
||
&self,
|
||
owner: ApplyOwner,
|
||
bundle: &Bundle,
|
||
inherited_endpoints: &HashSet<String>,
|
||
) -> Result<(), ApplyError> {
|
||
if matches!(owner, ApplyOwner::Group(_)) {
|
||
if !bundle.routes.is_empty() {
|
||
return Err(ApplyError::Invalid(
|
||
"a group manifest cannot declare routes — routes bind to an app".into(),
|
||
));
|
||
}
|
||
if !bundle.triggers.is_empty() {
|
||
return Err(ApplyError::Invalid(
|
||
"a group manifest cannot declare triggers — triggers belong to an app".into(),
|
||
));
|
||
}
|
||
}
|
||
self.validate_bundle(bundle, inherited_endpoints)
|
||
}
|
||
|
||
/// Reconcile app `app_id` to `bundle` in a single transaction. Creates
|
||
/// and updates are always applied; deletions of resources absent from the
|
||
/// bundle are executed only when `prune` is set (secrets are never pruned).
|
||
///
|
||
/// # Errors
|
||
/// `Invalid` for a bad bundle; `Backend` for repo/transaction failures.
|
||
/// Reconcile ONE node's scripts/routes/triggers/vars inside an existing
|
||
/// transaction (Phase 5). Returns the node's final `name -> script_id` map
|
||
/// (current + created), which the tree apply records so a later app node
|
||
/// can bind a route to a group script created earlier in the SAME tx. The
|
||
/// single-node `apply_owner` and the multi-node tree apply both call this.
|
||
#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
|
||
async fn reconcile_node_tx(
|
||
&self,
|
||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||
owner: ApplyOwner,
|
||
bundle: &Bundle,
|
||
plan: &Plan,
|
||
current: &CurrentState,
|
||
current_names: &HashMap<ScriptId, String>,
|
||
inherited: &HashMap<String, ScriptId>,
|
||
prune: bool,
|
||
actor: AdminUserId,
|
||
report: &mut ApplyReport,
|
||
) -> Result<HashMap<String, ScriptId>, ApplyError> {
|
||
let bundle_scripts: HashMap<String, &BundleScript> = bundle
|
||
.scripts
|
||
.iter()
|
||
.map(|s| (s.name.to_lowercase(), s))
|
||
.collect();
|
||
let current_scripts: HashMap<String, &Script> = current
|
||
.scripts
|
||
.iter()
|
||
.map(|s| (s.name.to_lowercase(), s))
|
||
.collect();
|
||
// name -> id for resolving route/trigger targets (existing + created).
|
||
let mut name_to_id: HashMap<String, ScriptId> = current
|
||
.scripts
|
||
.iter()
|
||
.map(|s| (s.name.to_lowercase(), s.id))
|
||
.collect();
|
||
|
||
// 1. Scripts — create + update first so routes/triggers resolve.
|
||
for ch in &plan.scripts {
|
||
let key = ch.key.to_lowercase();
|
||
match ch.op {
|
||
Op::Create => {
|
||
let bs = bundle_scripts[&key];
|
||
let (script_app_id, script_group_id) = owner.script_owner();
|
||
let new = NewScript {
|
||
app_id: script_app_id,
|
||
group_id: script_group_id,
|
||
name: bs.name.clone(),
|
||
description: bs.description.clone(),
|
||
source: bs.source.clone(),
|
||
kind: bs.kind,
|
||
timeout_seconds: bs.timeout_seconds,
|
||
memory_limit_mb: bs.memory_limit_mb,
|
||
sandbox: bs.sandbox,
|
||
enabled: bs.enabled,
|
||
imports: self.script_imports(bs)?,
|
||
};
|
||
let created = insert_script_tx(&mut *tx, &new).await.map_err(map_repo)?;
|
||
name_to_id.insert(created.name.to_lowercase(), created.id);
|
||
report.scripts_created += 1;
|
||
}
|
||
Op::Update => {
|
||
let bs = bundle_scripts[&key];
|
||
let cur = current_scripts[&key];
|
||
let patch = ScriptPatch {
|
||
name: None,
|
||
// Sparse: `None` (omitted in the manifest) leaves the
|
||
// stored description untouched; `Some(text)` sets it.
|
||
// Mirrors `script_update_reason`'s leave-as-is rule.
|
||
description: bs.description.clone().map(Some),
|
||
source: Some(bs.source.clone()),
|
||
timeout_seconds: bs.timeout_seconds,
|
||
memory_limit_mb: bs.memory_limit_mb,
|
||
sandbox: bs.sandbox,
|
||
kind: Some(bs.kind),
|
||
// Declarative: always reconcile to the manifest's value.
|
||
enabled: Some(bs.enabled),
|
||
imports: Some(self.script_imports(bs)?),
|
||
};
|
||
update_script_tx(&mut *tx, cur.id, &patch)
|
||
.await
|
||
.map_err(map_repo)?;
|
||
report.scripts_updated += 1;
|
||
}
|
||
Op::NoOp | Op::Delete => {}
|
||
}
|
||
}
|
||
|
||
// Phase 4: route/trigger targets the manifest doesn't define as app-own
|
||
// scripts may bind to inherited group endpoints. `or_insert` keeps
|
||
// app-own precedence (those names aren't in `inherited` by
|
||
// construction, but be defensive about CoW).
|
||
for (name, id) in inherited {
|
||
name_to_id.entry(name.clone()).or_insert(*id);
|
||
}
|
||
|
||
// 2. Routes — identity is the (method, host, path) tuple; an
|
||
// "update" (rebind / dispatch change) is delete + insert.
|
||
let current_routes: HashMap<String, &Route> = current
|
||
.routes
|
||
.iter()
|
||
.map(|r| {
|
||
(
|
||
route_key(
|
||
r.method.as_deref(),
|
||
r.host_kind,
|
||
&r.host,
|
||
r.path_kind,
|
||
&r.path,
|
||
),
|
||
r,
|
||
)
|
||
})
|
||
.collect();
|
||
let bundle_routes: HashMap<String, &BundleRoute> = bundle
|
||
.routes
|
||
.iter()
|
||
.map(|r| {
|
||
(
|
||
route_key(
|
||
r.method.as_deref(),
|
||
r.host_kind,
|
||
&r.host,
|
||
r.path_kind,
|
||
&r.path,
|
||
),
|
||
r,
|
||
)
|
||
})
|
||
.collect();
|
||
// 2. Routes — delete before insert so a binding freed by an update's
|
||
// delete-half or a prune removal can be reused by a create in the
|
||
// SAME apply without tripping the per-app UNIQUE binding index.
|
||
// Phase A — deletes: update replacements always; removals when pruning.
|
||
for ch in &plan.routes {
|
||
match ch.op {
|
||
Op::Update => {
|
||
let cur = current_routes[&ch.key];
|
||
delete_route_tx(&mut *tx, cur.id).await.map_err(map_repo)?;
|
||
}
|
||
Op::Delete if prune => {
|
||
if let Some(cur) = current_routes.get(&ch.key) {
|
||
delete_route_tx(&mut *tx, cur.id).await.map_err(map_repo)?;
|
||
report.routes_deleted += 1;
|
||
}
|
||
}
|
||
_ => {}
|
||
}
|
||
}
|
||
// Phase B — inserts: creates and update replacements.
|
||
for ch in &plan.routes {
|
||
let is_update = match ch.op {
|
||
Op::Create => false,
|
||
Op::Update => true,
|
||
_ => continue,
|
||
};
|
||
let br = bundle_routes[&ch.key];
|
||
let sid = resolve_script(&name_to_id, &br.script)?;
|
||
// Routes only exist on app nodes (a group bundle has none, so this
|
||
// loop is empty for a group — validated above).
|
||
let app_id = owner.app_id().expect("routes only exist on app nodes");
|
||
insert_bundle_route(&mut *tx, app_id, sid, br).await?;
|
||
if is_update {
|
||
report.routes_updated += 1;
|
||
} else {
|
||
report.routes_created += 1;
|
||
}
|
||
}
|
||
|
||
// 3. Triggers — semantic identity captures the whole definition,
|
||
// so the only op is Create (a change is delete-old + create-new;
|
||
// the stale one is removed by prune). Additive: create only.
|
||
let bundle_triggers: HashMap<String, &BundleTrigger> =
|
||
bundle.triggers.iter().map(|t| (t.identity(), t)).collect();
|
||
for ch in &plan.triggers {
|
||
if ch.op != Op::Create {
|
||
continue;
|
||
}
|
||
// Triggers only exist on app nodes (a group bundle has none).
|
||
let app_id = owner.app_id().expect("triggers only exist on app nodes");
|
||
let bt = bundle_triggers[&ch.key];
|
||
let sid = resolve_script(&name_to_id, bt.script())?;
|
||
if let BundleTrigger::Email {
|
||
inbound_secret_ref, ..
|
||
} = bt
|
||
{
|
||
// Resolve the referenced secret, decrypt it, and re-seal it
|
||
// into the email trigger's detail row — the value never
|
||
// travels in the manifest.
|
||
let (ct, nonce) = self.resolve_and_seal(app_id, inbound_secret_ref).await?;
|
||
insert_email_trigger_tx(&mut *tx, app_id, sid, actor, &ct, &nonce)
|
||
.await
|
||
.map_err(map_trig)?;
|
||
} else {
|
||
let (dispatch, retry_max, backoff, base) = self.trigger_settings(bt);
|
||
let details = bundle_trigger_details(
|
||
bt,
|
||
self.trigger_config.queue_default_visibility_timeout_secs,
|
||
);
|
||
insert_trigger_tx(
|
||
&mut *tx, app_id, sid, actor, dispatch, retry_max, backoff, base, &details,
|
||
)
|
||
.await
|
||
.map_err(map_trig)?;
|
||
}
|
||
report.triggers_created += 1;
|
||
}
|
||
|
||
// 3b. Vars — upsert every Create/Update at app scope `*`, in-tx so a
|
||
// later failure rolls them back with everything else.
|
||
for ch in &plan.vars {
|
||
match ch.op {
|
||
Op::Create => {
|
||
owner
|
||
.set_var_tx(&mut *tx, &ch.key, &bundle.vars[&ch.key])
|
||
.await?;
|
||
report.vars_created += 1;
|
||
}
|
||
Op::Update => {
|
||
owner
|
||
.set_var_tx(&mut *tx, &ch.key, &bundle.vars[&ch.key])
|
||
.await?;
|
||
report.vars_updated += 1;
|
||
}
|
||
Op::NoOp | Op::Delete => {}
|
||
}
|
||
}
|
||
|
||
// 4. Prune (only with --prune): delete stale triggers, then scripts
|
||
// (route removals already happened in the route delete-pass above).
|
||
// Secret pruning is deliberately deferred (destructive + irreversible):
|
||
// stale secrets are surfaced in the plan but never deleted here.
|
||
if prune {
|
||
// Include inherited group-script bindings (Phase 4) so a trigger
|
||
// bound to a group script resolves its identity and is pruned when
|
||
// dropped from the manifest — not silently orphaned.
|
||
let id_to_name: HashMap<ScriptId, String> = current
|
||
.scripts
|
||
.iter()
|
||
.map(|s| (s.id, s.name.clone()))
|
||
.chain(current_names.iter().map(|(id, n)| (*id, n.clone())))
|
||
.collect();
|
||
let trigger_id_by_identity: HashMap<String, TriggerId> = current
|
||
.triggers
|
||
.iter()
|
||
.filter_map(|t| current_trigger_identity(t, &id_to_name).map(|i| (i, t.id)))
|
||
.collect();
|
||
for ch in &plan.triggers {
|
||
if ch.op == Op::Delete {
|
||
if let Some(id) = trigger_id_by_identity.get(&ch.key) {
|
||
delete_trigger_tx(&mut *tx, *id).await.map_err(map_trig)?;
|
||
report.triggers_deleted += 1;
|
||
}
|
||
}
|
||
}
|
||
|
||
// Scripts owning a trigger the manifest can't represent (email,
|
||
// dead_letter) must not be silently pruned: deleting the script
|
||
// would cascade-destroy that trigger (and its sealed secret) via
|
||
// the `triggers.script_id ON DELETE CASCADE` FK, defeating the
|
||
// "never prune email triggers" guarantee. Refuse and point the
|
||
// operator at the explicit removal path.
|
||
let protected: HashSet<ScriptId> = current
|
||
.triggers
|
||
.iter()
|
||
.filter(|t| {
|
||
matches!(
|
||
t.details,
|
||
TriggerDetails::Email { .. } | TriggerDetails::DeadLetter { .. }
|
||
)
|
||
})
|
||
.map(|t| t.script_id)
|
||
.collect();
|
||
let script_id_by_name: HashMap<String, ScriptId> = current
|
||
.scripts
|
||
.iter()
|
||
.map(|s| (s.name.to_lowercase(), s.id))
|
||
.collect();
|
||
for ch in &plan.scripts {
|
||
if ch.op == Op::Delete {
|
||
if let Some(id) = script_id_by_name.get(&ch.key.to_lowercase()) {
|
||
if protected.contains(id) {
|
||
return Err(ApplyError::Invalid(format!(
|
||
"cannot prune script `{}`: it still has an email or \
|
||
dead-letter trigger that the manifest can't represent; \
|
||
remove the trigger with `pic triggers rm` first",
|
||
ch.key
|
||
)));
|
||
}
|
||
delete_script_tx(&mut *tx, *id).await.map_err(map_repo)?;
|
||
report.scripts_deleted += 1;
|
||
}
|
||
}
|
||
}
|
||
|
||
// Vars are prunable config (unlike secrets, which are never
|
||
// pruned): a live app var the manifest dropped is deleted.
|
||
for ch in &plan.vars {
|
||
if ch.op == Op::Delete {
|
||
owner.delete_var_tx(&mut *tx, &ch.key).await?;
|
||
report.vars_deleted += 1;
|
||
}
|
||
}
|
||
}
|
||
Ok(name_to_id)
|
||
}
|
||
|
||
pub async fn apply(
|
||
&self,
|
||
app_id: AppId,
|
||
bundle: &Bundle,
|
||
prune: bool,
|
||
actor: AdminUserId,
|
||
expected_token: Option<&str>,
|
||
) -> Result<ApplyReport, ApplyError> {
|
||
self.apply_owner(
|
||
ApplyOwner::App(app_id),
|
||
bundle,
|
||
prune,
|
||
actor,
|
||
expected_token,
|
||
)
|
||
.await
|
||
}
|
||
|
||
/// Owner-generic apply (Phase 5): reconcile an app OR group node to `bundle`
|
||
/// in a single transaction. A group node reconciles only its scripts + vars.
|
||
///
|
||
/// # Errors
|
||
/// `Invalid` for a bad bundle; `Backend` for repo/transaction failures.
|
||
#[allow(clippy::too_many_lines)]
|
||
pub async fn apply_owner(
|
||
&self,
|
||
owner: ApplyOwner,
|
||
bundle: &Bundle,
|
||
prune: bool,
|
||
actor: AdminUserId,
|
||
expected_token: Option<&str>,
|
||
) -> Result<ApplyReport, ApplyError> {
|
||
let inherited = self.resolve_inherited_targets_for(owner, bundle).await?;
|
||
self.validate_bundle_for(owner, bundle, &keys_set(&inherited))?;
|
||
self.check_imports_resolve(owner, bundle).await?;
|
||
if let ApplyOwner::App(app_id) = owner {
|
||
self.validate_route_hosts(app_id, bundle).await?;
|
||
}
|
||
|
||
let mut tx = self
|
||
.pool
|
||
.begin()
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||
// Serialize concurrent applies to the same app. NOTE: this lock
|
||
// serializes apply-vs-apply only — `load_current` below reads on the
|
||
// pool (a separate connection), so an *interactive* admin write
|
||
// committing between the read and our writes can surface as a
|
||
// unique-constraint conflict (clean tx rollback, no corruption) or a
|
||
// stale no-op delete. Reading current state through `&mut *tx` is the
|
||
// proper fix and is left as a follow-up. (The queue one-consumer
|
||
// invariant is closed independently in `insert_trigger_tx`.)
|
||
sqlx::query("SELECT pg_advisory_xact_lock($1)")
|
||
.bind(apply_lock_key(owner))
|
||
.execute(&mut *tx)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||
|
||
let current = self.load_current(owner).await?;
|
||
// Phase 4: names of group scripts the current routes/triggers bind to,
|
||
// so the token / diff / prune resolve inherited bindings (see method).
|
||
let current_names = self.resolve_current_target_names(¤t).await?;
|
||
// Bound-plan check (§4.2): if the caller passed the token from a prior
|
||
// `pic plan`, refuse when the live state changed since. Computed from
|
||
// the same `load_current` snapshot the diff uses, before any mutation
|
||
// (so a stale apply rolls back nothing). NOTE: like the diff, this read
|
||
// is on the pool, not `&mut *tx` (see the lock comment above), so it
|
||
// catches drift between plan and apply but shares that same narrow
|
||
// apply-vs-interactive-write window — it is not a substitute for the
|
||
// tx-scoped read the follow-up will add.
|
||
if let Some(expected) = expected_token {
|
||
if state_token_with_names(¤t, ¤t_names) != expected {
|
||
return Err(ApplyError::StateMoved);
|
||
}
|
||
}
|
||
// Surface a missing email-secret reference here so `plan` and `apply`
|
||
// agree, rather than only failing deep in `resolve_and_seal` below.
|
||
validate_email_secrets_present(bundle, ¤t.secret_names)?;
|
||
let plan = compute_diff_with_names(¤t, bundle, ¤t_names);
|
||
let mut report = ApplyReport::default();
|
||
// §4.7 warning: an enabled binding pointing at a disabled script is
|
||
// deployed-but-unreachable. Not an error (it's valid desired state),
|
||
// but surfaced so the operator isn't surprised by a silent 404.
|
||
report.warnings.extend(disabled_target_warnings(bundle));
|
||
// A group endpoint is reached by *inheritance* (a descendant app binds
|
||
// it by name), never by its own route, so the "no route/trigger"
|
||
// warning is noise for every group script — emit it for apps only.
|
||
if matches!(owner, ApplyOwner::App(_)) {
|
||
report
|
||
.warnings
|
||
.extend(unreachable_endpoint_warnings(bundle));
|
||
}
|
||
|
||
self.reconcile_node_tx(
|
||
&mut tx,
|
||
owner,
|
||
bundle,
|
||
&plan,
|
||
¤t,
|
||
¤t_names,
|
||
&inherited,
|
||
prune,
|
||
actor,
|
||
&mut report,
|
||
)
|
||
.await?;
|
||
|
||
tx.commit()
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||
|
||
// Post-commit: rebuild the orchestrator's route snapshot once. A
|
||
// failure here doesn't undo the committed DB write — the table
|
||
// self-heals on the next route write or restart. A group node touches
|
||
// no routes, so there's nothing to refresh.
|
||
if matches!(owner, ApplyOwner::App(_)) {
|
||
if let Err(e) = self.refresh_route_table().await {
|
||
tracing::warn!(error = %e, "apply: route table refresh failed");
|
||
report
|
||
.warnings
|
||
.push("route table refresh failed; it will self-heal".into());
|
||
}
|
||
}
|
||
Ok(report)
|
||
}
|
||
|
||
// ------------------------------------------------------------------------
|
||
// Tree apply (Phase 5): a project subtree reconciled in ONE transaction.
|
||
// ------------------------------------------------------------------------
|
||
|
||
/// Diff a whole project subtree, read-only. Returns a per-node plan + one
|
||
/// combined bound-plan token.
|
||
///
|
||
/// # Errors
|
||
/// `Invalid` for a malformed/unknown bundle; `Backend` for repo failures.
|
||
pub async fn plan_tree(&self, bundle: &TreeBundle) -> Result<TreePlanResult, ApplyError> {
|
||
let (prepared, token) = self.prepare_tree(bundle).await?;
|
||
let nodes = prepared
|
||
.into_iter()
|
||
.map(|p| NodePlan {
|
||
kind: p.node.kind,
|
||
slug: p.node.slug.clone(),
|
||
plan: p.plan,
|
||
})
|
||
.collect();
|
||
Ok(TreePlanResult {
|
||
nodes,
|
||
state_token: token,
|
||
})
|
||
}
|
||
|
||
/// Reconcile a whole project subtree in a single transaction (all-or-
|
||
/// nothing). Groups reconcile before apps, so an app route/trigger can bind
|
||
/// a group script created earlier in the SAME transaction (resolved via the
|
||
/// in-memory `group_script_index`, since uncommitted rows aren't visible to
|
||
/// the pool-based inherited resolver).
|
||
///
|
||
/// # Errors
|
||
/// `Invalid` for a bad bundle; `StateMoved` if the bound token is stale;
|
||
/// `Backend` for repo/transaction failures.
|
||
pub async fn apply_tree(
|
||
&self,
|
||
bundle: &TreeBundle,
|
||
prune: bool,
|
||
actor: AdminUserId,
|
||
expected_token: Option<&str>,
|
||
) -> Result<ApplyReport, ApplyError> {
|
||
let (prepared, token) = self.prepare_tree(bundle).await?;
|
||
if let Some(expected) = expected_token {
|
||
if token != expected {
|
||
return Err(ApplyError::StateMoved);
|
||
}
|
||
}
|
||
|
||
let mut tx = self
|
||
.pool
|
||
.begin()
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||
// Lock every node's apply key, in sorted order, so concurrent applies
|
||
// touching any shared node serialize and the ordering can't deadlock.
|
||
let mut lock_keys: Vec<i64> = prepared.iter().map(|p| apply_lock_key(p.owner)).collect();
|
||
lock_keys.sort_unstable();
|
||
lock_keys.dedup();
|
||
for k in lock_keys {
|
||
sqlx::query("SELECT pg_advisory_xact_lock($1)")
|
||
.bind(k)
|
||
.execute(&mut *tx)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||
}
|
||
|
||
let mut report = ApplyReport::default();
|
||
let mut group_script_index: HashMap<GroupId, HashMap<String, ScriptId>> = HashMap::new();
|
||
let mut any_app = false;
|
||
|
||
// Phase A — groups first: reconcile each and record its post-create
|
||
// `name -> id` so descendant app nodes can resolve inherited bindings.
|
||
for p in &prepared {
|
||
if let ApplyOwner::Group(gid) = p.owner {
|
||
report
|
||
.warnings
|
||
.extend(disabled_target_warnings(&p.node.bundle));
|
||
let name_to_id = self
|
||
.reconcile_node_tx(
|
||
&mut tx,
|
||
p.owner,
|
||
&p.node.bundle,
|
||
&p.plan,
|
||
&p.current,
|
||
&p.current_names,
|
||
&HashMap::new(),
|
||
prune,
|
||
actor,
|
||
&mut report,
|
||
)
|
||
.await?;
|
||
group_script_index.insert(gid, name_to_id);
|
||
}
|
||
}
|
||
|
||
// Phase B — apps: resolve inherited targets against in-tree groups (the
|
||
// ids just created) and pre-existing ancestors, then reconcile.
|
||
for p in &prepared {
|
||
if let ApplyOwner::App(_) = p.owner {
|
||
any_app = true;
|
||
report
|
||
.warnings
|
||
.extend(disabled_target_warnings(&p.node.bundle));
|
||
report
|
||
.warnings
|
||
.extend(unreachable_endpoint_warnings(&p.node.bundle));
|
||
let inherited = self
|
||
.tree_inherited_for_app(&p.app_chain, &p.node.bundle, &group_script_index)
|
||
.await?;
|
||
self.reconcile_node_tx(
|
||
&mut tx,
|
||
p.owner,
|
||
&p.node.bundle,
|
||
&p.plan,
|
||
&p.current,
|
||
&p.current_names,
|
||
&inherited,
|
||
prune,
|
||
actor,
|
||
&mut report,
|
||
)
|
||
.await?;
|
||
}
|
||
}
|
||
|
||
tx.commit()
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||
// One post-commit route refresh covers every app node touched.
|
||
if any_app {
|
||
if let Err(e) = self.refresh_route_table().await {
|
||
tracing::warn!(error = %e, "apply_tree: route table refresh failed");
|
||
report
|
||
.warnings
|
||
.push("route table refresh failed; it will self-heal".into());
|
||
}
|
||
}
|
||
Ok(report)
|
||
}
|
||
|
||
/// Resolve nodes to owners, validate each (apps' route/trigger targets may
|
||
/// bind to in-tree ancestor group scripts), compute each node's plan, and
|
||
/// fold a combined bound-plan token. Shared by `plan_tree` / `apply_tree`.
|
||
#[allow(clippy::too_many_lines)]
|
||
async fn prepare_tree<'a>(
|
||
&self,
|
||
bundle: &'a TreeBundle,
|
||
) -> Result<(Vec<PreparedNode<'a>>, String), ApplyError> {
|
||
if bundle.nodes.is_empty() {
|
||
return Err(ApplyError::Invalid("project tree has no nodes".into()));
|
||
}
|
||
// Register in-tree group nodes first: their (resolved) ids and endpoint
|
||
// script names, which an app node's binding validation may reference.
|
||
let mut group_id_by_slug: HashMap<String, GroupId> = HashMap::new();
|
||
let mut group_script_names: HashMap<GroupId, HashSet<String>> = HashMap::new();
|
||
let mut seen_slugs: HashSet<String> = HashSet::new();
|
||
for n in &bundle.nodes {
|
||
let key = format!("{:?}:{}", n.kind, n.slug.to_lowercase());
|
||
if !seen_slugs.insert(key) {
|
||
return Err(ApplyError::Invalid(format!(
|
||
"project tree lists node `{}` more than once",
|
||
n.slug
|
||
)));
|
||
}
|
||
if n.kind == NodeKind::Group {
|
||
let gid = self.resolve_group(&n.slug).await?;
|
||
group_id_by_slug.insert(n.slug.to_lowercase(), gid);
|
||
group_script_names.insert(
|
||
gid,
|
||
n.bundle
|
||
.scripts
|
||
.iter()
|
||
.filter(|s| s.kind == ScriptKind::Endpoint)
|
||
.map(|s| s.name.to_lowercase())
|
||
.collect(),
|
||
);
|
||
}
|
||
}
|
||
|
||
let mut prepared = Vec::with_capacity(bundle.nodes.len());
|
||
let mut token_parts: Vec<String> = Vec::new();
|
||
let mut versioned_groups: BTreeSet<GroupId> = BTreeSet::new();
|
||
|
||
for n in &bundle.nodes {
|
||
let (owner, app_chain) = match n.kind {
|
||
NodeKind::Group => {
|
||
let gid = group_id_by_slug[&n.slug.to_lowercase()];
|
||
versioned_groups.insert(gid);
|
||
(ApplyOwner::Group(gid), Vec::new())
|
||
}
|
||
NodeKind::App => {
|
||
let app = self.resolve_app(&n.slug).await?;
|
||
let chain = self
|
||
.groups
|
||
.ancestors(app.group_id)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||
for g in &chain {
|
||
versioned_groups.insert(g.id);
|
||
}
|
||
(
|
||
ApplyOwner::App(app.id),
|
||
chain.iter().map(|g| g.id).collect::<Vec<_>>(),
|
||
)
|
||
}
|
||
};
|
||
|
||
// Inherited endpoint NAMES the bindings may reference: pre-existing
|
||
// (pool resolver) plus in-tree ancestor group scripts.
|
||
let inherited_names: HashSet<String> = match owner {
|
||
ApplyOwner::App(app_id) => {
|
||
let mut names =
|
||
keys_set(&self.resolve_inherited_targets(app_id, &n.bundle).await?);
|
||
for gid in &app_chain {
|
||
if let Some(s) = group_script_names.get(gid) {
|
||
names.extend(s.iter().cloned());
|
||
}
|
||
}
|
||
names
|
||
}
|
||
ApplyOwner::Group(_) => HashSet::new(),
|
||
};
|
||
self.validate_bundle_for(owner, &n.bundle, &inherited_names)?;
|
||
if let ApplyOwner::App(app_id) = owner {
|
||
self.validate_route_hosts(app_id, &n.bundle).await?;
|
||
}
|
||
|
||
let current = self.load_current(owner).await?;
|
||
let current_names = self.resolve_current_target_names(¤t).await?;
|
||
validate_email_secrets_present(&n.bundle, ¤t.secret_names)?;
|
||
let plan = compute_diff_with_names(¤t, &n.bundle, ¤t_names);
|
||
token_parts.push(format!(
|
||
"n|{}|{}",
|
||
n.slug.to_lowercase(),
|
||
state_token_with_names(¤t, ¤t_names)
|
||
));
|
||
prepared.push(PreparedNode {
|
||
owner,
|
||
node: n,
|
||
current,
|
||
current_names,
|
||
plan,
|
||
app_chain,
|
||
});
|
||
}
|
||
|
||
// Fold in every in-scope group's structure version, so a reparent or a
|
||
// new app under the subtree between plan and apply trips StateMoved.
|
||
for gid in &versioned_groups {
|
||
if let Some(g) = self
|
||
.groups
|
||
.get_by_id(*gid)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?
|
||
{
|
||
token_parts.push(format!("gs|{gid}|{}", g.structure_version));
|
||
}
|
||
}
|
||
token_parts.sort_unstable();
|
||
Ok((prepared, combine_tokens(&token_parts)))
|
||
}
|
||
|
||
/// Resolve an app node's inherited route/trigger targets to script ids,
|
||
/// nearest-ancestor-wins, considering BOTH in-tree group nodes (their
|
||
/// in-tx `name -> id`, incl. just-created scripts) and pre-existing
|
||
/// out-of-tree ancestor groups (their committed rows).
|
||
async fn tree_inherited_for_app(
|
||
&self,
|
||
app_chain: &[GroupId],
|
||
bundle: &Bundle,
|
||
index: &HashMap<GroupId, HashMap<String, ScriptId>>,
|
||
) -> Result<HashMap<String, ScriptId>, ApplyError> {
|
||
let own: HashSet<String> = bundle
|
||
.scripts
|
||
.iter()
|
||
.map(|s| s.name.to_lowercase())
|
||
.collect();
|
||
let mut referenced: HashSet<String> = HashSet::new();
|
||
for r in &bundle.routes {
|
||
referenced.insert(r.script.to_lowercase());
|
||
}
|
||
for t in &bundle.triggers {
|
||
referenced.insert(t.script().to_lowercase());
|
||
}
|
||
referenced.retain(|n| !own.contains(n));
|
||
if referenced.is_empty() {
|
||
return Ok(HashMap::new());
|
||
}
|
||
// Per-ancestor name→id maps, nearest-first.
|
||
let mut chain_maps: Vec<HashMap<String, ScriptId>> = Vec::with_capacity(app_chain.len());
|
||
for gid in app_chain {
|
||
if let Some(idx) = index.get(gid) {
|
||
chain_maps.push(idx.clone());
|
||
} else {
|
||
let m = self
|
||
.scripts
|
||
.list_for_group(*gid)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?
|
||
.into_iter()
|
||
.filter(|s| s.kind == ScriptKind::Endpoint)
|
||
.map(|s| (s.name.to_lowercase(), s.id))
|
||
.collect();
|
||
chain_maps.push(m);
|
||
}
|
||
}
|
||
let mut out = HashMap::new();
|
||
for name in referenced {
|
||
for m in &chain_maps {
|
||
if let Some(id) = m.get(&name) {
|
||
out.insert(name.clone(), *id);
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
Ok(out)
|
||
}
|
||
|
||
async fn resolve_app(&self, ident: &str) -> Result<picloud_shared::App, ApplyError> {
|
||
crate::app_repo::resolve_app(self.apps.as_ref(), ident)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?
|
||
.map(|l| l.app)
|
||
.ok_or_else(|| ApplyError::AppNotFound(ident.to_string()))
|
||
}
|
||
|
||
async fn resolve_group(&self, ident: &str) -> Result<GroupId, ApplyError> {
|
||
let found = if let Ok(uuid) = ident.parse::<uuid::Uuid>() {
|
||
self.groups.get_by_id(uuid.into()).await
|
||
} else {
|
||
self.groups.get_by_slug(ident).await
|
||
};
|
||
found
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?
|
||
.map(|g| g.id)
|
||
.ok_or_else(|| ApplyError::AppNotFound(format!("group {ident}")))
|
||
}
|
||
|
||
fn script_imports(&self, bs: &BundleScript) -> Result<Vec<String>, ApplyError> {
|
||
let res = if bs.kind == ScriptKind::Module {
|
||
self.validator.validate_module(&bs.source)
|
||
} else {
|
||
self.validator.validate(&bs.source)
|
||
};
|
||
res.map(|v| v.imports)
|
||
.map_err(|e| ApplyError::Invalid(format!("script `{}`: {e}", bs.name)))
|
||
}
|
||
|
||
fn trigger_settings(
|
||
&self,
|
||
bt: &BundleTrigger,
|
||
) -> (
|
||
TriggerDispatchMode,
|
||
u32,
|
||
crate::trigger_config::BackoffShape,
|
||
u32,
|
||
) {
|
||
let (dispatch, retry_max) = match bt {
|
||
BundleTrigger::Kv {
|
||
dispatch_mode,
|
||
retry_max_attempts,
|
||
..
|
||
}
|
||
| BundleTrigger::Docs {
|
||
dispatch_mode,
|
||
retry_max_attempts,
|
||
..
|
||
}
|
||
| BundleTrigger::Files {
|
||
dispatch_mode,
|
||
retry_max_attempts,
|
||
..
|
||
}
|
||
| BundleTrigger::Cron {
|
||
dispatch_mode,
|
||
retry_max_attempts,
|
||
..
|
||
}
|
||
| BundleTrigger::Pubsub {
|
||
dispatch_mode,
|
||
retry_max_attempts,
|
||
..
|
||
}
|
||
| BundleTrigger::Email {
|
||
dispatch_mode,
|
||
retry_max_attempts,
|
||
..
|
||
}
|
||
| BundleTrigger::Queue {
|
||
dispatch_mode,
|
||
retry_max_attempts,
|
||
..
|
||
} => (
|
||
dispatch_mode.unwrap_or(TriggerDispatchMode::Async),
|
||
retry_max_attempts.unwrap_or(self.trigger_config.retry_max_attempts),
|
||
),
|
||
};
|
||
(
|
||
dispatch,
|
||
retry_max,
|
||
self.trigger_config.retry_backoff,
|
||
self.trigger_config.retry_base_ms,
|
||
)
|
||
}
|
||
|
||
async fn refresh_route_table(&self) -> Result<(), ApplyError> {
|
||
let rows = self
|
||
.routes
|
||
.list_all()
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||
let compiled = crate::route_admin::compile_routes(&rows);
|
||
self.route_table.replace_all(compiled);
|
||
Ok(())
|
||
}
|
||
|
||
/// Resolve a referenced secret to plaintext, then re-seal it for an
|
||
/// email trigger's inbound-secret column. The value never appears in
|
||
/// the manifest — only the secret's name does.
|
||
async fn resolve_and_seal(
|
||
&self,
|
||
app_id: AppId,
|
||
name: &str,
|
||
) -> Result<(Vec<u8>, Vec<u8>), ApplyError> {
|
||
let stored = self
|
||
.secrets
|
||
.get(crate::secrets_service::SecretOwner::App(app_id), "*", name)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?
|
||
.ok_or_else(|| {
|
||
ApplyError::Invalid(format!(
|
||
"email trigger references secret `{name}`, which is not set \
|
||
(push it with `pic secret set {name}`)"
|
||
))
|
||
})?;
|
||
let plaintext = crate::secrets_service::open(
|
||
&self.master_key,
|
||
crate::secrets_service::SecretOwner::App(app_id),
|
||
name,
|
||
&stored,
|
||
)
|
||
.map_err(|e| ApplyError::Invalid(format!("could not read secret `{name}`: {e}")))?;
|
||
// The inbound HMAC must be a non-empty string — an empty/whitespace
|
||
// key is forgeable (preserves the spirit of audit 2026-06-11 H-B2).
|
||
match plaintext.as_str() {
|
||
Some(s) if !s.trim().is_empty() => {}
|
||
_ => {
|
||
return Err(ApplyError::Invalid(format!(
|
||
"secret `{name}` must be a non-empty string to sign an email trigger"
|
||
)))
|
||
}
|
||
}
|
||
let (ct, nonce) = crate::secrets_service::seal_legacy(
|
||
&self.master_key,
|
||
&plaintext,
|
||
crate::secrets_service::DEFAULT_SECRET_MAX_VALUE_BYTES,
|
||
)
|
||
.map_err(|e| ApplyError::Invalid(format!("could not seal secret `{name}`: {e}")))?;
|
||
Ok((ct, nonce.to_vec()))
|
||
}
|
||
|
||
/// Validate each bundle route's host against the app's domain claims —
|
||
/// the cross-state check the interactive route API enforces. Async, so
|
||
/// it runs alongside `plan`/`apply`, not in the sync `validate_bundle`.
|
||
async fn validate_route_hosts(&self, app_id: AppId, bundle: &Bundle) -> Result<(), ApplyError> {
|
||
for r in &bundle.routes {
|
||
crate::route_admin::validate_route_host_against_app(
|
||
self.domains.as_ref(),
|
||
app_id,
|
||
r.host_kind,
|
||
&r.host,
|
||
)
|
||
.await
|
||
.map_err(|e| ApplyError::Invalid(format!("route host claim: {e}")))?;
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// Bundle-local validation: catches the errors visible without the
|
||
/// live state (syntax, ceilings, dangling/wrong-kind targets, intra-
|
||
/// bundle duplicates, reserved paths, host/path structural validity).
|
||
/// Host-claim validation runs separately in `validate_route_hosts`.
|
||
///
|
||
/// NOTE (known gap vs. the interactive route API): cross-pattern route
|
||
/// conflict-vs-live (e.g. a param route overlapping a live exact route)
|
||
/// is NOT checked — the DB unique index catches only identical tuples,
|
||
/// so overlapping patterns are accepted.
|
||
#[allow(clippy::too_many_lines)]
|
||
/// Phase 4: resolve route/trigger target names that the manifest does NOT
|
||
/// declare as app-own scripts to inherited group-owned endpoint scripts on
|
||
/// the app's chain. Returns `name(lowercased) → script_id` for the ones
|
||
/// that resolve to a group endpoint. App-own names (in `bundle.scripts`)
|
||
/// are skipped — they bind through the normal `name_to_id` path and take
|
||
/// precedence (CoW). Resolved before the apply transaction; the group
|
||
/// scripts pre-exist (committed), so a pool read is correct and stable.
|
||
async fn resolve_inherited_targets(
|
||
&self,
|
||
app_id: AppId,
|
||
bundle: &Bundle,
|
||
) -> Result<HashMap<String, ScriptId>, ApplyError> {
|
||
let own: HashSet<String> = bundle
|
||
.scripts
|
||
.iter()
|
||
.map(|s| s.name.to_lowercase())
|
||
.collect();
|
||
let mut referenced: HashSet<String> = HashSet::new();
|
||
for r in &bundle.routes {
|
||
referenced.insert(r.script.to_lowercase());
|
||
}
|
||
for t in &bundle.triggers {
|
||
referenced.insert(t.script().to_lowercase());
|
||
}
|
||
let mut out = HashMap::new();
|
||
for name in referenced {
|
||
if own.contains(&name) {
|
||
continue;
|
||
}
|
||
if let Some(s) = self
|
||
.scripts
|
||
.get_by_name_inherited(app_id, &name)
|
||
.await
|
||
.map_err(map_repo)?
|
||
{
|
||
// Only genuinely-inherited group endpoints qualify as
|
||
// route/trigger targets. A depth-0 app-own match is already
|
||
// covered by `name_to_id`; skip it so the manifest keeps
|
||
// precedence. Group modules (Phase 4b) exist but are never
|
||
// bindable targets — the kind guard excludes them.
|
||
if s.group_id.is_some() && s.kind == ScriptKind::Endpoint {
|
||
out.insert(name, s.id);
|
||
}
|
||
}
|
||
}
|
||
Ok(out)
|
||
}
|
||
|
||
/// Phase 4: resolve the names of group-owned scripts the app's CURRENT
|
||
/// routes/triggers are bound to, `script_id → name`. These ids aren't in
|
||
/// `current.scripts` (app-own only), so this is what lets the diff /
|
||
/// state-token / prune resolve an existing inherited binding back to its
|
||
/// name — independent of whether the bundle still references it (so a
|
||
/// dropped inherited binding still diffs to a Delete, and an inherited-bound
|
||
/// trigger still moves the state token).
|
||
async fn resolve_current_target_names(
|
||
&self,
|
||
current: &CurrentState,
|
||
) -> Result<HashMap<ScriptId, String>, ApplyError> {
|
||
let own: HashSet<ScriptId> = current.scripts.iter().map(|s| s.id).collect();
|
||
let mut ids: HashSet<ScriptId> = HashSet::new();
|
||
for r in ¤t.routes {
|
||
if !own.contains(&r.script_id) {
|
||
ids.insert(r.script_id);
|
||
}
|
||
}
|
||
for t in ¤t.triggers {
|
||
if !own.contains(&t.script_id) {
|
||
ids.insert(t.script_id);
|
||
}
|
||
}
|
||
let mut out = HashMap::new();
|
||
for id in ids {
|
||
if let Some(s) = self.scripts.get(id).await.map_err(map_repo)? {
|
||
out.insert(id, s.name);
|
||
}
|
||
}
|
||
Ok(out)
|
||
}
|
||
|
||
/// §5.5 dangling-import plan check (Phase 4b, single-node path). Every
|
||
/// `import "<m>"` in a bundle script must resolve to a module declared in
|
||
/// the same bundle OR reachable lexically up the node's chain — otherwise
|
||
/// the apply would write a script that 404s its import at runtime. Lexical:
|
||
/// the node's owner IS the defining node for its own scripts, so resolution
|
||
/// roots at `owner` (an app node reaches ancestor group modules; a group
|
||
/// node walks its own ancestry, sealed from apps below).
|
||
///
|
||
/// Not run on the tree path: a tree node may import a module declared by
|
||
/// another node created in the same transaction (not yet committed), so the
|
||
/// in-tree set would have to be threaded through; the runtime check is the
|
||
/// backstop there.
|
||
async fn check_imports_resolve(
|
||
&self,
|
||
owner: ApplyOwner,
|
||
bundle: &Bundle,
|
||
) -> Result<(), ApplyError> {
|
||
// Modules this bundle declares (lowercased) satisfy their own imports.
|
||
let local: HashSet<String> = bundle
|
||
.scripts
|
||
.iter()
|
||
.filter(|s| s.kind == ScriptKind::Module)
|
||
.map(|s| s.name.to_lowercase())
|
||
.collect();
|
||
let origin = match owner {
|
||
ApplyOwner::App(a) => picloud_shared::ScriptOwner::App(a),
|
||
ApplyOwner::Group(g) => picloud_shared::ScriptOwner::Group(g),
|
||
};
|
||
// Resolve each distinct imported name once.
|
||
let mut checked: HashSet<String> = HashSet::new();
|
||
for s in &bundle.scripts {
|
||
for imp in self.script_imports(s)? {
|
||
let key = imp.to_lowercase();
|
||
if local.contains(&key) || !checked.insert(key) {
|
||
continue;
|
||
}
|
||
let found = self
|
||
.modules
|
||
.resolve(origin, &imp)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||
if found.is_none() {
|
||
return Err(ApplyError::Invalid(format!(
|
||
"script `{}` imports unknown module `{imp}` \
|
||
(not declared here and not inherited from an ancestor)",
|
||
s.name
|
||
)));
|
||
}
|
||
}
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// `inherited_endpoints` (lowercased names) are group-owned endpoint
|
||
/// scripts reachable from the app's chain that a route/trigger may bind to
|
||
/// even though the manifest doesn't declare them (Phase 4). They count as
|
||
/// known endpoints for the binding checks below.
|
||
fn validate_bundle(
|
||
&self,
|
||
bundle: &Bundle,
|
||
inherited_endpoints: &HashSet<String>,
|
||
) -> Result<(), ApplyError> {
|
||
// Scripts: unique names; each source compiles for its kind.
|
||
let mut names: HashSet<String> = HashSet::new();
|
||
for s in &bundle.scripts {
|
||
if !names.insert(s.name.to_lowercase()) {
|
||
return Err(ApplyError::Invalid(format!(
|
||
"duplicate script name `{}`",
|
||
s.name
|
||
)));
|
||
}
|
||
let res = if s.kind == ScriptKind::Module {
|
||
// Parity with the interactive create path (`api.rs`): a module
|
||
// may not shadow a built-in SDK namespace, or `import "kv"`
|
||
// could resolve to a user module instead of the real bridge.
|
||
if crate::api::RESERVED_MODULE_NAMES.contains(&s.name.as_str()) {
|
||
return Err(ApplyError::Invalid(format!(
|
||
"script `{}`: reserved module name (shadows a built-in SDK namespace)",
|
||
s.name
|
||
)));
|
||
}
|
||
self.validator.validate_module(&s.source)
|
||
} else {
|
||
self.validator.validate(&s.source)
|
||
};
|
||
res.map_err(|e| ApplyError::Invalid(format!("script `{}`: {e}", s.name)))?;
|
||
if let Some(sb) = &s.sandbox {
|
||
self.sandbox_ceiling
|
||
.check(sb)
|
||
.map_err(|e| ApplyError::Invalid(format!("script `{}`: {e}", s.name)))?;
|
||
}
|
||
}
|
||
let endpoints: HashSet<&str> = bundle
|
||
.scripts
|
||
.iter()
|
||
.filter(|s| s.kind == ScriptKind::Endpoint)
|
||
.map(|s| s.name.as_str())
|
||
.collect();
|
||
let all_names: HashSet<&str> = bundle.scripts.iter().map(|s| s.name.as_str()).collect();
|
||
|
||
// Routes: reserved paths; bind to an endpoint script; unique tuples.
|
||
let mut route_keys: HashSet<String> = HashSet::new();
|
||
for r in &bundle.routes {
|
||
reject_reserved_path(&r.path)?;
|
||
// Structural validity (parity with the interactive route API):
|
||
// a route whose path/host doesn't parse would otherwise be
|
||
// written and then silently dropped by `compile_routes`.
|
||
pattern::parse_path(r.path_kind, &r.path)
|
||
.map_err(|e| ApplyError::Invalid(format!("route path `{}`: {e}", r.path)))?;
|
||
pattern::parse_host(r.host_kind, &r.host, r.host_param_name.as_deref())
|
||
.map_err(|e| ApplyError::Invalid(format!("route host `{}`: {e}", r.host)))?;
|
||
let inherited = inherited_endpoints.contains(&r.script.to_lowercase());
|
||
if !all_names.contains(r.script.as_str()) && !inherited {
|
||
return Err(ApplyError::Invalid(format!(
|
||
"route {} binds to unknown script `{}`",
|
||
route_key(
|
||
r.method.as_deref(),
|
||
r.host_kind,
|
||
&r.host,
|
||
r.path_kind,
|
||
&r.path
|
||
),
|
||
r.script
|
||
)));
|
||
}
|
||
// Inherited targets are resolved as endpoints by construction
|
||
// (the inherited-name resolver excludes group modules); only
|
||
// check kind for app-own names.
|
||
if !inherited && !endpoints.contains(r.script.as_str()) {
|
||
return Err(ApplyError::Invalid(format!(
|
||
"route binds to `{}` which is a module, not an endpoint",
|
||
r.script
|
||
)));
|
||
}
|
||
let key = route_key(
|
||
r.method.as_deref(),
|
||
r.host_kind,
|
||
&r.host,
|
||
r.path_kind,
|
||
&r.path,
|
||
);
|
||
if !route_keys.insert(key.clone()) {
|
||
return Err(ApplyError::Invalid(format!("duplicate route `{key}`")));
|
||
}
|
||
}
|
||
|
||
// Triggers: bind to an endpoint script; unique semantic identity.
|
||
let mut trig_keys: HashSet<String> = HashSet::new();
|
||
for t in &bundle.triggers {
|
||
let inherited = inherited_endpoints.contains(&t.script().to_lowercase());
|
||
if !all_names.contains(t.script()) && !inherited {
|
||
return Err(ApplyError::Invalid(format!(
|
||
"{} trigger binds to unknown script `{}`",
|
||
t.kind_str(),
|
||
t.script()
|
||
)));
|
||
}
|
||
if !inherited && !endpoints.contains(t.script()) {
|
||
return Err(ApplyError::Invalid(format!(
|
||
"{} trigger binds to `{}` which is a module, not an endpoint",
|
||
t.kind_str(),
|
||
t.script()
|
||
)));
|
||
}
|
||
validate_trigger_shape(t)?;
|
||
let id = t.identity();
|
||
if !trig_keys.insert(id.clone()) {
|
||
return Err(ApplyError::Invalid(format!("duplicate trigger `{id}`")));
|
||
}
|
||
}
|
||
|
||
// Vars: kebab keys (same rule the vars admin API enforces), so a
|
||
// manifest can't write a key the interactive surface would reject.
|
||
for key in bundle.vars.keys() {
|
||
validate_var_key(key)?;
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
async fn load_current(&self, owner: ApplyOwner) -> Result<CurrentState, ApplyError> {
|
||
// A group node has only scripts + vars; routes / triggers / secrets are
|
||
// app-only and stay empty (the bundle for a group declares none).
|
||
let (scripts, routes, triggers, secret_names, var_owner) = match owner {
|
||
ApplyOwner::App(app_id) => (
|
||
self.scripts
|
||
.list_for_app(app_id)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?,
|
||
self.routes
|
||
.list_for_app(app_id)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?,
|
||
self.triggers
|
||
.list_for_app(app_id)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?,
|
||
self.all_secret_names(app_id).await?,
|
||
VarOwner::App(app_id),
|
||
),
|
||
ApplyOwner::Group(group_id) => (
|
||
self.scripts
|
||
.list_for_group(group_id)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?,
|
||
Vec::new(),
|
||
Vec::new(),
|
||
Vec::new(),
|
||
VarOwner::Group(group_id),
|
||
),
|
||
};
|
||
// The owner's OWN vars, narrowed to scope `*` and real values: the
|
||
// manifest reconciles env-agnostic config, not inherited vars (not in
|
||
// this owner's rows) nor tombstones (an imperative suppress the manifest
|
||
// can't express).
|
||
let vars = self
|
||
.vars
|
||
.list_for_owner(var_owner)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?
|
||
.into_iter()
|
||
.filter(|r| r.environment_scope == "*" && !r.is_tombstone)
|
||
.map(|r| (r.key, r.value))
|
||
.collect();
|
||
Ok(CurrentState {
|
||
scripts,
|
||
routes,
|
||
triggers,
|
||
secret_names,
|
||
vars,
|
||
})
|
||
}
|
||
|
||
async fn all_secret_names(&self, app_id: AppId) -> Result<Vec<String>, ApplyError> {
|
||
let mut names = Vec::new();
|
||
let mut cursor: Option<String> = None;
|
||
loop {
|
||
let page = self
|
||
.secrets
|
||
.list_names(
|
||
crate::secrets_service::SecretOwner::App(app_id),
|
||
cursor.as_deref(),
|
||
200,
|
||
)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||
names.extend(page.names);
|
||
match page.next_cursor {
|
||
Some(c) => cursor = Some(c),
|
||
None => break,
|
||
}
|
||
}
|
||
Ok(names)
|
||
}
|
||
}
|
||
|
||
// ----------------------------------------------------------------------------
|
||
// Diff (pure)
|
||
// ----------------------------------------------------------------------------
|
||
|
||
/// Diff desired `bundle` against `current`. Pure — unit-tested directly.
|
||
#[must_use]
|
||
pub fn compute_diff(current: &CurrentState, bundle: &Bundle) -> Plan {
|
||
compute_diff_with_names(current, bundle, &HashMap::new())
|
||
}
|
||
|
||
/// `compute_diff` with `current_names` — the names of group-owned (inherited)
|
||
/// scripts the app's CURRENT routes/triggers are bound to, `script_id → name`
|
||
/// (Phase 4). Those ids aren't in `current.scripts` (app-own only), so without
|
||
/// them a current binding to a group script resolves to no name: it reads as a
|
||
/// perpetual rebind, and a binding dropped from the manifest never diffs to a
|
||
/// Delete (so `--prune` can't remove it). The public `compute_diff` passes an
|
||
/// empty map (no inheritance).
|
||
fn compute_diff_with_names(
|
||
current: &CurrentState,
|
||
bundle: &Bundle,
|
||
current_names: &HashMap<ScriptId, String>,
|
||
) -> Plan {
|
||
let script_name_by_id: HashMap<ScriptId, String> = current
|
||
.scripts
|
||
.iter()
|
||
.map(|s| (s.id, s.name.clone()))
|
||
// Inherited group bindings' ids → names. Disjoint from app-own ids by
|
||
// construction, so order doesn't matter.
|
||
.chain(current_names.iter().map(|(id, n)| (*id, n.clone())))
|
||
.collect();
|
||
|
||
Plan {
|
||
scripts: diff_scripts(current, bundle),
|
||
routes: diff_routes(current, bundle, &script_name_by_id),
|
||
triggers: diff_triggers(current, bundle, &script_name_by_id),
|
||
secrets: diff_secrets(current, bundle),
|
||
vars: diff_vars(current, bundle),
|
||
}
|
||
}
|
||
|
||
/// Diff app-owned vars by key. Unlike secrets (whose values live out-of-band),
|
||
/// a var's value IS in the manifest, so equality is value-sensitive: a changed
|
||
/// value is an `Update`. Live vars absent from the manifest are `Delete`
|
||
/// (applied only under `--prune`).
|
||
fn diff_vars(current: &CurrentState, bundle: &Bundle) -> Vec<ResourceChange> {
|
||
let live: HashMap<&str, &serde_json::Value> =
|
||
current.vars.iter().map(|(k, v)| (k.as_str(), v)).collect();
|
||
|
||
let mut out = Vec::new();
|
||
for (key, value) in &bundle.vars {
|
||
match live.get(key.as_str()) {
|
||
Some(cur) if *cur == value => out.push(ResourceChange {
|
||
op: Op::NoOp,
|
||
key: key.clone(),
|
||
detail: None,
|
||
}),
|
||
Some(_) => out.push(ResourceChange {
|
||
op: Op::Update,
|
||
key: key.clone(),
|
||
detail: Some("value changed".into()),
|
||
}),
|
||
None => out.push(ResourceChange {
|
||
op: Op::Create,
|
||
key: key.clone(),
|
||
detail: None,
|
||
}),
|
||
}
|
||
}
|
||
for (key, _) in ¤t.vars {
|
||
if !bundle.vars.contains_key(key) {
|
||
out.push(ResourceChange {
|
||
op: Op::Delete,
|
||
key: key.clone(),
|
||
detail: Some("on server, not declared".into()),
|
||
});
|
||
}
|
||
}
|
||
out
|
||
}
|
||
|
||
/// Kebab var key (`^[a-z0-9][a-z0-9-]{0,127}$`) — mirrors `vars_api`'s
|
||
/// `validate_key` so the manifest and the interactive surface agree.
|
||
fn validate_var_key(key: &str) -> Result<(), ApplyError> {
|
||
let ok = !key.is_empty()
|
||
&& key.len() <= 128
|
||
&& key
|
||
.chars()
|
||
.next()
|
||
.is_some_and(|c| c.is_ascii_lowercase() || c.is_ascii_digit())
|
||
&& key
|
||
.chars()
|
||
.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-');
|
||
if ok {
|
||
Ok(())
|
||
} else {
|
||
Err(ApplyError::Invalid(format!(
|
||
"var key `{key}` must be 1–128 chars of lowercase letters, digits, and hyphens, \
|
||
starting with a letter or digit"
|
||
)))
|
||
}
|
||
}
|
||
|
||
/// Upsert one app-owned var at scope `*`, in the apply transaction. Mirrors
|
||
/// `PostgresVarsRepo::set` for `VarOwner::App` (the partial-index predicate is
|
||
/// restated in the conflict target) but runs against `&mut tx` so it commits
|
||
/// atomically with the rest of the apply. Clears any tombstone.
|
||
async fn set_app_var_tx(
|
||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||
app_id: AppId,
|
||
key: &str,
|
||
value: &serde_json::Value,
|
||
) -> Result<(), ApplyError> {
|
||
sqlx::query(
|
||
"INSERT INTO vars (app_id, environment_scope, key, value, is_tombstone) \
|
||
VALUES ($1, '*', $2, $3, FALSE) \
|
||
ON CONFLICT (app_id, environment_scope, key) WHERE app_id IS NOT NULL DO UPDATE \
|
||
SET value = EXCLUDED.value, is_tombstone = FALSE, updated_at = NOW()",
|
||
)
|
||
.bind(app_id.into_inner())
|
||
.bind(key)
|
||
.bind(value)
|
||
.execute(&mut **tx)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||
Ok(())
|
||
}
|
||
|
||
/// Delete one app-owned var at scope `*`, in the apply transaction.
|
||
async fn delete_app_var_tx(
|
||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||
app_id: AppId,
|
||
key: &str,
|
||
) -> Result<(), ApplyError> {
|
||
sqlx::query("DELETE FROM vars WHERE app_id = $1 AND environment_scope = '*' AND key = $2")
|
||
.bind(app_id.into_inner())
|
||
.bind(key)
|
||
.execute(&mut **tx)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||
Ok(())
|
||
}
|
||
|
||
/// Upsert one group-owned var at scope `*`, in the apply transaction (Phase 5).
|
||
/// Mirrors `set_app_var_tx` for `VarOwner::Group`. Env-scoped group vars
|
||
/// (`@staging`) stay imperative for now — a group manifest declares only
|
||
/// env-agnostic (`*`) group config.
|
||
async fn set_group_var_tx(
|
||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||
group_id: GroupId,
|
||
key: &str,
|
||
value: &serde_json::Value,
|
||
) -> Result<(), ApplyError> {
|
||
sqlx::query(
|
||
"INSERT INTO vars (group_id, environment_scope, key, value, is_tombstone) \
|
||
VALUES ($1, '*', $2, $3, FALSE) \
|
||
ON CONFLICT (group_id, environment_scope, key) WHERE group_id IS NOT NULL DO UPDATE \
|
||
SET value = EXCLUDED.value, is_tombstone = FALSE, updated_at = NOW()",
|
||
)
|
||
.bind(group_id.into_inner())
|
||
.bind(key)
|
||
.bind(value)
|
||
.execute(&mut **tx)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||
Ok(())
|
||
}
|
||
|
||
/// Delete one group-owned var at scope `*`, in the apply transaction (Phase 5).
|
||
async fn delete_group_var_tx(
|
||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||
group_id: GroupId,
|
||
key: &str,
|
||
) -> Result<(), ApplyError> {
|
||
sqlx::query("DELETE FROM vars WHERE group_id = $1 AND environment_scope = '*' AND key = $2")
|
||
.bind(group_id.into_inner())
|
||
.bind(key)
|
||
.execute(&mut **tx)
|
||
.await
|
||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||
Ok(())
|
||
}
|
||
|
||
/// Owner of an apply node (Phase 5): an app or a group. The apply engine is
|
||
/// owner-generic — only script-create ownership, var writes, the current-state
|
||
/// load, and the advisory lock differ; the diff and all other CRUD are shared.
|
||
/// A group node has only scripts + vars (no routes / triggers / secrets).
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||
pub enum ApplyOwner {
|
||
App(AppId),
|
||
Group(GroupId),
|
||
}
|
||
|
||
impl ApplyOwner {
|
||
/// `(app_id, group_id)` for a `NewScript` — exactly one is `Some`.
|
||
fn script_owner(self) -> (Option<AppId>, Option<GroupId>) {
|
||
match self {
|
||
Self::App(a) => (Some(a), None),
|
||
Self::Group(g) => (None, Some(g)),
|
||
}
|
||
}
|
||
|
||
/// The app id, when this is an app node. Routes/triggers/email-secrets only
|
||
/// exist on app nodes, so their reconcile paths call this with an
|
||
/// `expect` — guarded by validation that rejects them on a group bundle.
|
||
fn app_id(self) -> Option<AppId> {
|
||
match self {
|
||
Self::App(a) => Some(a),
|
||
Self::Group(_) => None,
|
||
}
|
||
}
|
||
|
||
async fn set_var_tx(
|
||
self,
|
||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||
key: &str,
|
||
value: &serde_json::Value,
|
||
) -> Result<(), ApplyError> {
|
||
match self {
|
||
Self::App(a) => set_app_var_tx(tx, a, key, value).await,
|
||
Self::Group(g) => set_group_var_tx(tx, g, key, value).await,
|
||
}
|
||
}
|
||
|
||
async fn delete_var_tx(
|
||
self,
|
||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||
key: &str,
|
||
) -> Result<(), ApplyError> {
|
||
match self {
|
||
Self::App(a) => delete_app_var_tx(tx, a, key).await,
|
||
Self::Group(g) => delete_group_var_tx(tx, g, key).await,
|
||
}
|
||
}
|
||
}
|
||
|
||
fn diff_scripts(current: &CurrentState, bundle: &Bundle) -> Vec<ResourceChange> {
|
||
let by_name: HashMap<String, &Script> = current
|
||
.scripts
|
||
.iter()
|
||
.map(|s| (s.name.to_lowercase(), s))
|
||
.collect();
|
||
let desired: HashSet<String> = bundle
|
||
.scripts
|
||
.iter()
|
||
.map(|s| s.name.to_lowercase())
|
||
.collect();
|
||
|
||
let mut out = Vec::new();
|
||
for s in &bundle.scripts {
|
||
match by_name.get(&s.name.to_lowercase()) {
|
||
None => out.push(ResourceChange {
|
||
op: Op::Create,
|
||
key: s.name.clone(),
|
||
detail: None,
|
||
}),
|
||
Some(cur) => {
|
||
if let Some(reason) = script_update_reason(cur, s) {
|
||
out.push(ResourceChange {
|
||
op: Op::Update,
|
||
key: s.name.clone(),
|
||
detail: Some(reason),
|
||
});
|
||
} else {
|
||
out.push(ResourceChange {
|
||
op: Op::NoOp,
|
||
key: s.name.clone(),
|
||
detail: None,
|
||
});
|
||
}
|
||
}
|
||
}
|
||
}
|
||
for s in ¤t.scripts {
|
||
if !desired.contains(&s.name.to_lowercase()) {
|
||
out.push(ResourceChange {
|
||
op: Op::Delete,
|
||
key: s.name.clone(),
|
||
detail: None,
|
||
});
|
||
}
|
||
}
|
||
out
|
||
}
|
||
|
||
/// Returns `Some(reason)` if the desired script differs from current. A
|
||
/// `None` optional field means "unspecified — leave as-is", so it never
|
||
/// forces an update.
|
||
fn script_update_reason(cur: &Script, desired: &BundleScript) -> Option<String> {
|
||
if cur.source != desired.source {
|
||
return Some("source changed".into());
|
||
}
|
||
if cur.kind != desired.kind {
|
||
return Some("kind changed".into());
|
||
}
|
||
// Declarative (default true): always reconcile to the manifest's value.
|
||
if cur.enabled != desired.enabled {
|
||
return Some(if desired.enabled {
|
||
"enabled".into()
|
||
} else {
|
||
"disabled".into()
|
||
});
|
||
}
|
||
// Sparse like the other optional fields: an omitted (`None`) description
|
||
// means "leave as-is", so only an explicitly-set value that differs
|
||
// forces an update. (Clearing a description is done in the dashboard.)
|
||
if let Some(d) = &desired.description {
|
||
if cur.description.as_ref() != Some(d) {
|
||
return Some("description changed".into());
|
||
}
|
||
}
|
||
if let Some(t) = desired.timeout_seconds {
|
||
if i64::from(cur.timeout_seconds) != i64::from(t) {
|
||
return Some("timeout changed".into());
|
||
}
|
||
}
|
||
if let Some(m) = desired.memory_limit_mb {
|
||
if i64::from(cur.memory_limit_mb) != i64::from(m) {
|
||
return Some("memory changed".into());
|
||
}
|
||
}
|
||
if let Some(sb) = desired.sandbox {
|
||
if cur.sandbox != sb {
|
||
return Some("sandbox changed".into());
|
||
}
|
||
}
|
||
None
|
||
}
|
||
|
||
fn diff_routes(
|
||
current: &CurrentState,
|
||
bundle: &Bundle,
|
||
name_by_id: &HashMap<ScriptId, String>,
|
||
) -> Vec<ResourceChange> {
|
||
let by_key: HashMap<String, &Route> = current
|
||
.routes
|
||
.iter()
|
||
.map(|r| {
|
||
(
|
||
route_key(
|
||
r.method.as_deref(),
|
||
r.host_kind,
|
||
&r.host,
|
||
r.path_kind,
|
||
&r.path,
|
||
),
|
||
r,
|
||
)
|
||
})
|
||
.collect();
|
||
let desired: HashSet<String> = bundle
|
||
.routes
|
||
.iter()
|
||
.map(|r| {
|
||
route_key(
|
||
r.method.as_deref(),
|
||
r.host_kind,
|
||
&r.host,
|
||
r.path_kind,
|
||
&r.path,
|
||
)
|
||
})
|
||
.collect();
|
||
|
||
let mut out = Vec::new();
|
||
for r in &bundle.routes {
|
||
let key = route_key(
|
||
r.method.as_deref(),
|
||
r.host_kind,
|
||
&r.host,
|
||
r.path_kind,
|
||
&r.path,
|
||
);
|
||
match by_key.get(&key) {
|
||
None => out.push(ResourceChange {
|
||
op: Op::Create,
|
||
key,
|
||
detail: Some(format!("→ {}", r.script)),
|
||
}),
|
||
Some(cur) => {
|
||
let cur_script = name_by_id.get(&cur.script_id).map(String::as_str);
|
||
if cur_script != Some(r.script.as_str())
|
||
|| cur.dispatch_mode != r.dispatch_mode
|
||
|| cur.host_param_name != r.host_param_name
|
||
|| cur.enabled != r.enabled
|
||
{
|
||
out.push(ResourceChange {
|
||
op: Op::Update,
|
||
key,
|
||
detail: Some(format!("→ {}", r.script)),
|
||
});
|
||
} else {
|
||
out.push(ResourceChange {
|
||
op: Op::NoOp,
|
||
key,
|
||
detail: None,
|
||
});
|
||
}
|
||
}
|
||
}
|
||
}
|
||
for r in ¤t.routes {
|
||
let key = route_key(
|
||
r.method.as_deref(),
|
||
r.host_kind,
|
||
&r.host,
|
||
r.path_kind,
|
||
&r.path,
|
||
);
|
||
if !desired.contains(&key) {
|
||
out.push(ResourceChange {
|
||
op: Op::Delete,
|
||
key,
|
||
detail: None,
|
||
});
|
||
}
|
||
}
|
||
out
|
||
}
|
||
|
||
fn diff_triggers(
|
||
current: &CurrentState,
|
||
bundle: &Bundle,
|
||
name_by_id: &HashMap<ScriptId, String>,
|
||
) -> Vec<ResourceChange> {
|
||
let by_id: HashMap<String, &Trigger> = current
|
||
.triggers
|
||
.iter()
|
||
.filter_map(|t| current_trigger_identity(t, name_by_id).map(|id| (id, t)))
|
||
.collect();
|
||
let desired: HashSet<String> = bundle
|
||
.triggers
|
||
.iter()
|
||
.map(BundleTrigger::identity)
|
||
.collect();
|
||
|
||
let mut out = Vec::new();
|
||
for t in &bundle.triggers {
|
||
let id = t.identity();
|
||
// Semantic identity captures the whole definition for these kinds,
|
||
// so a match is a no-op (changing a field changes the identity).
|
||
if by_id.contains_key(&id) {
|
||
out.push(ResourceChange {
|
||
op: Op::NoOp,
|
||
key: id,
|
||
detail: None,
|
||
});
|
||
} else {
|
||
out.push(ResourceChange {
|
||
op: Op::Create,
|
||
key: id,
|
||
detail: None,
|
||
});
|
||
}
|
||
}
|
||
for t in ¤t.triggers {
|
||
// Never prune email triggers: `pull` can't represent them (the
|
||
// server stores the sealed secret, not the `inbound_secret_ref`), so
|
||
// their absence from the manifest is ambiguous, not a delete intent.
|
||
// Manage email triggers with `pic triggers rm`.
|
||
if matches!(t.details, TriggerDetails::Email { .. }) {
|
||
continue;
|
||
}
|
||
if let Some(id) = current_trigger_identity(t, name_by_id) {
|
||
if !desired.contains(&id) {
|
||
out.push(ResourceChange {
|
||
op: Op::Delete,
|
||
key: id,
|
||
detail: None,
|
||
});
|
||
}
|
||
}
|
||
}
|
||
out
|
||
}
|
||
|
||
fn diff_secrets(current: &CurrentState, bundle: &Bundle) -> Vec<ResourceChange> {
|
||
let live: HashSet<&str> = current.secret_names.iter().map(String::as_str).collect();
|
||
let declared: HashSet<&str> = bundle.secrets.iter().map(String::as_str).collect();
|
||
|
||
let mut out = Vec::new();
|
||
for name in &bundle.secrets {
|
||
if live.contains(name.as_str()) {
|
||
out.push(ResourceChange {
|
||
op: Op::NoOp,
|
||
key: name.clone(),
|
||
detail: None,
|
||
});
|
||
} else {
|
||
out.push(ResourceChange {
|
||
op: Op::Create,
|
||
key: name.clone(),
|
||
detail: Some("declared but unset — push with `pic secret set`".into()),
|
||
});
|
||
}
|
||
}
|
||
for name in ¤t.secret_names {
|
||
if !declared.contains(name.as_str()) {
|
||
out.push(ResourceChange {
|
||
op: Op::Delete,
|
||
key: name.clone(),
|
||
detail: Some("on server, not declared".into()),
|
||
});
|
||
}
|
||
}
|
||
out
|
||
}
|
||
|
||
/// Reject any email trigger whose referenced secret isn't set, so `plan` and
|
||
/// `apply` give the same answer (apply also fails in `resolve_and_seal`, but
|
||
/// only after taking the lock — surfacing it here keeps plan honest).
|
||
///
|
||
/// This checks the secret *reference exists*. `resolve_and_seal` additionally
|
||
/// requires the resolved value to be a non-empty string; `plan` deliberately
|
||
/// does not decrypt secrets, so a reference to a set-but-empty secret passes
|
||
/// `plan` and is caught at `apply` — the one residual plan/apply divergence,
|
||
/// and a deliberate one (plan stays read-only and crypto-free).
|
||
fn validate_email_secrets_present(
|
||
bundle: &Bundle,
|
||
secret_names: &[String],
|
||
) -> Result<(), ApplyError> {
|
||
for t in &bundle.triggers {
|
||
if let BundleTrigger::Email {
|
||
script,
|
||
inbound_secret_ref,
|
||
..
|
||
} = t
|
||
{
|
||
if !secret_names.iter().any(|n| n == inbound_secret_ref) {
|
||
return Err(ApplyError::Invalid(format!(
|
||
"email trigger on `{script}` references secret \
|
||
`{inbound_secret_ref}`, which is not set; push it with \
|
||
`pic secret set` first"
|
||
)));
|
||
}
|
||
}
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// Semantic identity for a live trigger. Returns `None` only for
|
||
/// `dead_letter` (not representable in the manifest), so it's ignored by the
|
||
/// diff entirely. `email` returns `Some` so a declared email trigger matches
|
||
/// as NoOp, but `diff_triggers` separately refuses to emit a Delete for it.
|
||
fn current_trigger_identity(t: &Trigger, name_by_id: &HashMap<ScriptId, String>) -> Option<String> {
|
||
let script = name_by_id.get(&t.script_id)?;
|
||
match &t.details {
|
||
TriggerDetails::Kv {
|
||
collection_glob,
|
||
ops,
|
||
} => Some(format!(
|
||
"kv|{script}|{collection_glob}|{}",
|
||
sorted_csv(ops.iter().copied().map(KvEventOp::as_str))
|
||
)),
|
||
TriggerDetails::Docs {
|
||
collection_glob,
|
||
ops,
|
||
} => Some(format!(
|
||
"docs|{script}|{collection_glob}|{}",
|
||
sorted_csv(ops.iter().copied().map(DocsEventOp::as_str))
|
||
)),
|
||
TriggerDetails::Files {
|
||
collection_glob,
|
||
ops,
|
||
} => Some(format!(
|
||
"files|{script}|{collection_glob}|{}",
|
||
sorted_csv(ops.iter().copied().map(FilesEventOp::as_str))
|
||
)),
|
||
TriggerDetails::Cron {
|
||
schedule, timezone, ..
|
||
} => Some(format!("cron|{script}|{schedule}|{timezone}")),
|
||
TriggerDetails::Pubsub { topic_pattern } => {
|
||
Some(format!("pubsub|{script}|{topic_pattern}"))
|
||
}
|
||
TriggerDetails::Email { .. } => Some(format!("email|{script}")),
|
||
TriggerDetails::Queue { queue_name, .. } => Some(format!("queue|{queue_name}")),
|
||
TriggerDetails::DeadLetter { .. } => None,
|
||
}
|
||
}
|
||
|
||
fn route_key(
|
||
method: Option<&str>,
|
||
host_kind: HostKind,
|
||
host: &str,
|
||
path_kind: PathKind,
|
||
path: &str,
|
||
) -> String {
|
||
let m = method.map_or_else(|| "ANY".to_string(), str::to_uppercase);
|
||
format!(
|
||
"{m} {} {} {} {path}",
|
||
host_kind_str(host_kind),
|
||
host,
|
||
path_kind_str(path_kind)
|
||
)
|
||
}
|
||
|
||
fn host_kind_str(k: HostKind) -> &'static str {
|
||
match k {
|
||
HostKind::Any => "any",
|
||
HostKind::Strict => "strict",
|
||
HostKind::Wildcard => "wildcard",
|
||
}
|
||
}
|
||
|
||
fn path_kind_str(k: PathKind) -> &'static str {
|
||
match k {
|
||
PathKind::Exact => "exact",
|
||
PathKind::Prefix => "prefix",
|
||
PathKind::Param => "param",
|
||
}
|
||
}
|
||
|
||
/// Sort + comma-join op strings so order is insignificant for identity.
|
||
fn sorted_csv<'a>(ops: impl Iterator<Item = &'a str>) -> String {
|
||
let set: BTreeSet<&str> = ops.collect();
|
||
set.into_iter().collect::<Vec<_>>().join(",")
|
||
}
|
||
|
||
fn reject_reserved_path(path: &str) -> Result<(), ApplyError> {
|
||
if RESERVED_PATH_EXACT.contains(&path)
|
||
|| RESERVED_PATH_PREFIXES.iter().any(|p| path.starts_with(p))
|
||
{
|
||
return Err(ApplyError::Invalid(format!("path `{path}` is reserved")));
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// §4.7 reachability warning: an *enabled* endpoint script with no route and
|
||
/// no trigger has no event surface — it's only reachable via the
|
||
/// execute-by-id bypass / `invoke()`. Modules are exempt (never invoked
|
||
/// directly); disabled endpoints are intentionally inert, so skip them.
|
||
fn unreachable_endpoint_warnings(bundle: &Bundle) -> Vec<String> {
|
||
let bound: HashSet<&str> = bundle
|
||
.routes
|
||
.iter()
|
||
.map(|r| r.script.as_str())
|
||
.chain(bundle.triggers.iter().map(BundleTrigger::script))
|
||
.collect();
|
||
bundle
|
||
.scripts
|
||
.iter()
|
||
.filter(|s| s.kind == ScriptKind::Endpoint && s.enabled && !bound.contains(s.name.as_str()))
|
||
.map(|s| {
|
||
format!(
|
||
"endpoint `{}` has no route or trigger — only reachable via the \
|
||
execute-by-id bypass / invoke()",
|
||
s.name
|
||
)
|
||
})
|
||
.collect()
|
||
}
|
||
|
||
/// §4.7 reachability warnings: an *enabled* route or trigger bound to a
|
||
/// script the manifest marks *disabled* is deployed but unreachable (the
|
||
/// route 404s, the trigger won't fire). Valid desired state, so a warning —
|
||
/// not an error — keyed on the bundle alone.
|
||
fn disabled_target_warnings(bundle: &Bundle) -> Vec<String> {
|
||
let disabled: HashSet<&str> = bundle
|
||
.scripts
|
||
.iter()
|
||
.filter(|s| !s.enabled)
|
||
.map(|s| s.name.as_str())
|
||
.collect();
|
||
if disabled.is_empty() {
|
||
return Vec::new();
|
||
}
|
||
let mut out = Vec::new();
|
||
for r in &bundle.routes {
|
||
if r.enabled && disabled.contains(r.script.as_str()) {
|
||
out.push(format!(
|
||
"route `{}` is enabled but its script `{}` is disabled — it will 404",
|
||
route_key(
|
||
r.method.as_deref(),
|
||
r.host_kind,
|
||
&r.host,
|
||
r.path_kind,
|
||
&r.path
|
||
),
|
||
r.script
|
||
));
|
||
}
|
||
}
|
||
for t in &bundle.triggers {
|
||
if disabled.contains(t.script()) {
|
||
out.push(format!(
|
||
"{} trigger targets disabled script `{}` — it will not fire",
|
||
t.kind_str(),
|
||
t.script()
|
||
));
|
||
}
|
||
}
|
||
out
|
||
}
|
||
|
||
/// Per-kind structural validation for one desired trigger — kept at parity
|
||
/// with the interactive trigger API so `apply` can't write a trigger the
|
||
/// dashboard would have rejected. Pure (no live state), so it's unit-tested
|
||
/// directly. The script-binding checks live in `validate_bundle`.
|
||
fn validate_trigger_shape(t: &BundleTrigger) -> Result<(), ApplyError> {
|
||
match t {
|
||
BundleTrigger::Email {
|
||
inbound_secret_ref, ..
|
||
} if inbound_secret_ref.trim().is_empty() => {
|
||
return Err(ApplyError::Invalid(format!(
|
||
"email trigger on `{}` needs inbound_secret_ref (a secret name)",
|
||
t.script()
|
||
)));
|
||
}
|
||
BundleTrigger::Queue {
|
||
queue_name,
|
||
visibility_timeout_secs,
|
||
..
|
||
} => {
|
||
if queue_name.trim().is_empty() {
|
||
return Err(ApplyError::Invalid("queue trigger needs queue_name".into()));
|
||
}
|
||
if let Some(v) = visibility_timeout_secs {
|
||
// Parity with the interactive API: the floor is the dispatcher
|
||
// tick/reclaim-cadence minimum (`triggers_api`), and the ceiling
|
||
// matches `trigger_repo`'s queue check. Apply previously allowed
|
||
// a [5, 29] floor the dashboard rejects.
|
||
let min = crate::triggers_api::MIN_QUEUE_VISIBILITY_TIMEOUT_SECS;
|
||
if !(min..=3600).contains(v) {
|
||
return Err(ApplyError::Invalid(format!(
|
||
"queue `{queue_name}`: visibility_timeout_secs must be in [{min}, 3600]"
|
||
)));
|
||
}
|
||
}
|
||
}
|
||
BundleTrigger::Cron {
|
||
schedule, timezone, ..
|
||
} => {
|
||
crate::cron_scheduler::validate_schedule(schedule).map_err(|e| {
|
||
ApplyError::Invalid(format!(
|
||
"cron trigger on `{}`: invalid schedule: {e}",
|
||
t.script()
|
||
))
|
||
})?;
|
||
crate::cron_scheduler::validate_timezone(timezone).map_err(|e| {
|
||
ApplyError::Invalid(format!(
|
||
"cron trigger on `{}`: invalid timezone: {e}",
|
||
t.script()
|
||
))
|
||
})?;
|
||
}
|
||
// Parity with the interactive trigger API, which rejects an empty
|
||
// collection_glob (`create_{kv,docs,files}_trigger`).
|
||
BundleTrigger::Kv {
|
||
collection_glob, ..
|
||
}
|
||
| BundleTrigger::Docs {
|
||
collection_glob, ..
|
||
}
|
||
| BundleTrigger::Files {
|
||
collection_glob, ..
|
||
} if collection_glob.trim().is_empty() => {
|
||
return Err(ApplyError::Invalid(format!(
|
||
"{} trigger on `{}` needs a non-empty collection_glob",
|
||
t.kind_str(),
|
||
t.script()
|
||
)));
|
||
}
|
||
// Parity with `create_pubsub_trigger`'s topic-pattern check — otherwise
|
||
// a malformed pattern is written and silently never matches at dispatch.
|
||
BundleTrigger::Pubsub { topic_pattern, .. } => {
|
||
picloud_shared::validate_topic_pattern(topic_pattern).map_err(|e| {
|
||
ApplyError::Invalid(format!(
|
||
"pubsub trigger on `{}`: invalid topic_pattern: {e}",
|
||
t.script()
|
||
))
|
||
})?;
|
||
}
|
||
_ => {}
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// Summary of what an `apply` changed.
|
||
#[derive(Debug, Default, Serialize)]
|
||
pub struct ApplyReport {
|
||
pub scripts_created: u32,
|
||
pub scripts_updated: u32,
|
||
pub scripts_deleted: u32,
|
||
pub routes_created: u32,
|
||
pub routes_updated: u32,
|
||
pub routes_deleted: u32,
|
||
pub triggers_created: u32,
|
||
pub triggers_deleted: u32,
|
||
#[serde(default)]
|
||
pub vars_created: u32,
|
||
#[serde(default)]
|
||
pub vars_updated: u32,
|
||
#[serde(default)]
|
||
pub vars_deleted: u32,
|
||
#[serde(skip_serializing_if = "Vec::is_empty")]
|
||
pub warnings: Vec<String>,
|
||
}
|
||
|
||
/// The key set of a name→id map, for passing the inherited target *names*
|
||
/// into `validate_bundle` without leaking the ids.
|
||
fn keys_set(m: &HashMap<String, ScriptId>) -> HashSet<String> {
|
||
m.keys().cloned().collect()
|
||
}
|
||
|
||
/// A tree node resolved to its owner + loaded current state + computed plan,
|
||
/// ready to reconcile. `app_chain` is the ordered (nearest-first) ancestor
|
||
/// group ids for an app node (empty for a group node).
|
||
struct PreparedNode<'a> {
|
||
owner: ApplyOwner,
|
||
node: &'a TreeNode,
|
||
current: CurrentState,
|
||
current_names: HashMap<ScriptId, String>,
|
||
plan: Plan,
|
||
app_chain: Vec<GroupId>,
|
||
}
|
||
|
||
/// FNV-1a over a set of already-sorted per-resource token parts — the same
|
||
/// hashing `state_token` uses, lifted to combine many nodes into one tree
|
||
/// token. Order-independent because the caller sorts `parts`.
|
||
fn combine_tokens(parts: &[String]) -> String {
|
||
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
|
||
for p in parts {
|
||
for b in p.as_bytes() {
|
||
h ^= u64::from(*b);
|
||
h = h.wrapping_mul(0x0000_0100_0000_01b3);
|
||
}
|
||
h ^= u64::from(b'\n');
|
||
h = h.wrapping_mul(0x0000_0100_0000_01b3);
|
||
}
|
||
format!("{h:016x}")
|
||
}
|
||
|
||
fn resolve_script(
|
||
name_to_id: &HashMap<String, ScriptId>,
|
||
name: &str,
|
||
) -> Result<ScriptId, ApplyError> {
|
||
name_to_id
|
||
.get(&name.to_lowercase())
|
||
.copied()
|
||
.ok_or_else(|| ApplyError::Invalid(format!("binding references unknown script `{name}`")))
|
||
}
|
||
|
||
async fn insert_bundle_route(
|
||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||
app_id: AppId,
|
||
script_id: ScriptId,
|
||
br: &BundleRoute,
|
||
) -> Result<(), ApplyError> {
|
||
let new = NewRoute {
|
||
app_id,
|
||
script_id,
|
||
host_kind: br.host_kind,
|
||
host: br.host.clone(),
|
||
host_param_name: br.host_param_name.clone(),
|
||
path_kind: br.path_kind,
|
||
path: br.path.clone(),
|
||
method: br.method.clone(),
|
||
dispatch_mode: br.dispatch_mode,
|
||
enabled: br.enabled,
|
||
};
|
||
insert_route_tx(tx, &new).await.map_err(map_repo)?;
|
||
Ok(())
|
||
}
|
||
|
||
fn bundle_trigger_details(bt: &BundleTrigger, default_visibility: u32) -> TriggerDetails {
|
||
match bt {
|
||
BundleTrigger::Kv {
|
||
collection_glob,
|
||
ops,
|
||
..
|
||
} => TriggerDetails::Kv {
|
||
collection_glob: collection_glob.clone(),
|
||
ops: ops.clone(),
|
||
},
|
||
BundleTrigger::Docs {
|
||
collection_glob,
|
||
ops,
|
||
..
|
||
} => TriggerDetails::Docs {
|
||
collection_glob: collection_glob.clone(),
|
||
ops: ops.clone(),
|
||
},
|
||
BundleTrigger::Files {
|
||
collection_glob,
|
||
ops,
|
||
..
|
||
} => TriggerDetails::Files {
|
||
collection_glob: collection_glob.clone(),
|
||
ops: ops.clone(),
|
||
},
|
||
BundleTrigger::Cron {
|
||
schedule, timezone, ..
|
||
} => TriggerDetails::Cron {
|
||
schedule: schedule.clone(),
|
||
timezone: timezone.clone(),
|
||
last_fired_at: None,
|
||
},
|
||
BundleTrigger::Pubsub { topic_pattern, .. } => TriggerDetails::Pubsub {
|
||
topic_pattern: topic_pattern.clone(),
|
||
},
|
||
BundleTrigger::Queue {
|
||
queue_name,
|
||
visibility_timeout_secs,
|
||
..
|
||
} => TriggerDetails::Queue {
|
||
queue_name: queue_name.clone(),
|
||
visibility_timeout_secs: visibility_timeout_secs.unwrap_or(default_visibility),
|
||
last_fired_at: None,
|
||
},
|
||
BundleTrigger::Email { .. } => unreachable!("email handled separately"),
|
||
}
|
||
}
|
||
|
||
fn map_repo(e: ScriptRepositoryError) -> ApplyError {
|
||
match e {
|
||
ScriptRepositoryError::Conflict(m) => ApplyError::Invalid(m),
|
||
ScriptRepositoryError::NotFound(_) => {
|
||
ApplyError::Invalid("a referenced resource was not found".into())
|
||
}
|
||
ScriptRepositoryError::Db(e) => ApplyError::Backend(e.to_string()),
|
||
}
|
||
}
|
||
|
||
fn map_trig(e: TriggerRepoError) -> ApplyError {
|
||
match e {
|
||
TriggerRepoError::Invalid(m) => ApplyError::Invalid(m),
|
||
TriggerRepoError::NotFound(_) => ApplyError::Invalid("trigger not found".into()),
|
||
TriggerRepoError::Db(e) => ApplyError::Backend(e.to_string()),
|
||
}
|
||
}
|
||
|
||
/// A stable fingerprint of an app's live state, covering exactly what the
|
||
/// reconcile diff keys on: script identity + write-counter (`version` bumps on
|
||
/// any script edit), route identity + binding/attrs, trigger semantic identity
|
||
/// (via `current_trigger_identity`, so dead-letter triggers — which the diff
|
||
/// ignores — are excluded), and secret names. Any out-of-band change to those
|
||
/// flips the token, so `plan`-then-`apply` can detect the app moving underneath.
|
||
///
|
||
/// Deliberately mirrors the diff's inputs so a mismatch means the *reviewed
|
||
/// plan* would now differ — not merely that some unrelated row changed.
|
||
///
|
||
/// Uses FNV-1a (deterministic across process restarts, unlike `DefaultHasher`'s
|
||
/// per-process seed) so a token stored by `pic plan` still matches on a later
|
||
/// `pic apply`. A hash collision can only ever yield a false "unchanged", which
|
||
/// is the same risk class as not checking at all — never a false refusal.
|
||
#[must_use]
|
||
pub fn state_token(current: &CurrentState) -> String {
|
||
state_token_with_names(current, &HashMap::new())
|
||
}
|
||
|
||
/// `state_token` with `current_names` (group-bound binding ids → names, Phase
|
||
/// 4) so a trigger bound to an inherited group script contributes to the
|
||
/// fingerprint. Without it such a trigger is silently excluded and the
|
||
/// bound-plan (StateMoved) check has a false-negative for that binding class.
|
||
fn state_token_with_names(
|
||
current: &CurrentState,
|
||
current_names: &HashMap<ScriptId, String>,
|
||
) -> String {
|
||
let mut parts: Vec<String> = Vec::with_capacity(
|
||
current.scripts.len()
|
||
+ current.routes.len()
|
||
+ current.triggers.len()
|
||
+ current.secret_names.len()
|
||
+ current.vars.len(),
|
||
);
|
||
for s in ¤t.scripts {
|
||
parts.push(format!(
|
||
"s|{}|{}|{}",
|
||
s.name.to_lowercase(),
|
||
s.version,
|
||
s.enabled
|
||
));
|
||
}
|
||
for r in ¤t.routes {
|
||
parts.push(format!(
|
||
"r|{}|{:?}|{:?}|{}|{}",
|
||
route_key(
|
||
r.method.as_deref(),
|
||
r.host_kind,
|
||
&r.host,
|
||
r.path_kind,
|
||
&r.path
|
||
),
|
||
r.script_id,
|
||
r.dispatch_mode,
|
||
r.host_param_name.as_deref().unwrap_or(""),
|
||
r.enabled,
|
||
));
|
||
}
|
||
// Mirror exactly what the trigger diff keys on: `current_trigger_identity`
|
||
// excludes dead-letter triggers (the diff ignores them) and keys on the
|
||
// semantic identity — NOT raw id/enabled. Hashing dead-letter rows or the
|
||
// (currently inert) `enabled` flag would force a false refusal over a
|
||
// change the manifest can't represent.
|
||
let script_name_by_id: HashMap<ScriptId, String> = current
|
||
.scripts
|
||
.iter()
|
||
.map(|s| (s.id, s.name.clone()))
|
||
.chain(current_names.iter().map(|(id, n)| (*id, n.clone())))
|
||
.collect();
|
||
for t in ¤t.triggers {
|
||
if let Some(id) = current_trigger_identity(t, &script_name_by_id) {
|
||
parts.push(format!("t|{id}"));
|
||
}
|
||
}
|
||
for n in ¤t.secret_names {
|
||
parts.push(format!("k|{n}"));
|
||
}
|
||
// Vars are value-sensitive (the value lives in the manifest), so the token
|
||
// moves when a value changes — the bound-plan check then catches an edit
|
||
// between plan and apply. `to_string` on a serde_json::Value is stable
|
||
// (serde_json sorts object keys by default).
|
||
for (key, value) in ¤t.vars {
|
||
parts.push(format!(
|
||
"v|{key}|{}",
|
||
serde_json::to_string(value).unwrap_or_default()
|
||
));
|
||
}
|
||
// Order-independent: sort the per-resource tokens before hashing.
|
||
parts.sort_unstable();
|
||
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
|
||
for p in &parts {
|
||
for b in p.as_bytes() {
|
||
h ^= u64::from(*b);
|
||
h = h.wrapping_mul(0x0000_0100_0000_01b3);
|
||
}
|
||
h ^= u64::from(b'\n');
|
||
h = h.wrapping_mul(0x0000_0100_0000_01b3);
|
||
}
|
||
format!("{h:016x}")
|
||
}
|
||
|
||
/// Per-node advisory lock key, namespaced so it can't collide with the
|
||
/// queue-trigger lock space. App and group ids live in disjoint UUID spaces and
|
||
/// are tagged by owner kind, so an app and a group never share a key.
|
||
fn apply_lock_key(owner: ApplyOwner) -> i64 {
|
||
use std::hash::{Hash, Hasher};
|
||
let mut h = std::collections::hash_map::DefaultHasher::new();
|
||
"picloud-apply".hash(&mut h);
|
||
match owner {
|
||
ApplyOwner::App(a) => {
|
||
"app".hash(&mut h);
|
||
a.into_inner().hash(&mut h);
|
||
}
|
||
ApplyOwner::Group(g) => {
|
||
"group".hash(&mut h);
|
||
g.into_inner().hash(&mut h);
|
||
}
|
||
}
|
||
i64::from_ne_bytes(h.finish().to_ne_bytes())
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
use chrono::Utc;
|
||
|
||
fn script(name: &str, source: &str) -> Script {
|
||
Script {
|
||
id: ScriptId::from(uuid::Uuid::new_v4()),
|
||
app_id: Some(AppId::from(uuid::Uuid::nil())),
|
||
group_id: None,
|
||
name: name.to_string(),
|
||
description: None,
|
||
version: 1,
|
||
source: source.to_string(),
|
||
kind: ScriptKind::Endpoint,
|
||
timeout_seconds: 30,
|
||
sandbox: ScriptSandbox::empty(),
|
||
memory_limit_mb: 256,
|
||
enabled: true,
|
||
created_at: Utc::now(),
|
||
updated_at: Utc::now(),
|
||
}
|
||
}
|
||
|
||
fn bundle_script(name: &str, source: &str) -> BundleScript {
|
||
BundleScript {
|
||
name: name.to_string(),
|
||
source: source.to_string(),
|
||
kind: ScriptKind::Endpoint,
|
||
description: None,
|
||
timeout_seconds: None,
|
||
memory_limit_mb: None,
|
||
sandbox: None,
|
||
enabled: true,
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn empty_app_is_all_create() {
|
||
let current = CurrentState::default();
|
||
let bundle = Bundle {
|
||
scripts: vec![bundle_script("a", "1")],
|
||
routes: vec![],
|
||
triggers: vec![],
|
||
secrets: vec!["S".into()],
|
||
vars: std::collections::BTreeMap::new(),
|
||
};
|
||
let plan = compute_diff(¤t, &bundle);
|
||
assert_eq!(plan.scripts.len(), 1);
|
||
assert_eq!(plan.scripts[0].op, Op::Create);
|
||
assert_eq!(plan.secrets[0].op, Op::Create); // declared-but-unset
|
||
assert!(!plan.is_noop());
|
||
}
|
||
|
||
#[test]
|
||
fn identical_is_all_noop() {
|
||
let s = script("a", "let x = 1;");
|
||
let current = CurrentState {
|
||
scripts: vec![s.clone()],
|
||
secret_names: vec!["S".into()],
|
||
..CurrentState::default()
|
||
};
|
||
let bundle = Bundle {
|
||
scripts: vec![bundle_script("a", "let x = 1;")],
|
||
routes: vec![],
|
||
triggers: vec![],
|
||
secrets: vec!["S".into()],
|
||
vars: std::collections::BTreeMap::new(),
|
||
};
|
||
let plan = compute_diff(¤t, &bundle);
|
||
assert!(plan.is_noop(), "expected all no-op, got {plan:?}");
|
||
}
|
||
|
||
#[test]
|
||
fn changed_source_is_update() {
|
||
let current = CurrentState {
|
||
scripts: vec![script("a", "old")],
|
||
..CurrentState::default()
|
||
};
|
||
let bundle = Bundle {
|
||
scripts: vec![bundle_script("a", "new")],
|
||
routes: vec![],
|
||
triggers: vec![],
|
||
secrets: vec![],
|
||
vars: std::collections::BTreeMap::new(),
|
||
};
|
||
let plan = compute_diff(¤t, &bundle);
|
||
assert_eq!(plan.scripts[0].op, Op::Update);
|
||
}
|
||
|
||
#[test]
|
||
fn missing_from_bundle_is_delete() {
|
||
let current = CurrentState {
|
||
scripts: vec![script("gone", "x")],
|
||
secret_names: vec!["OLD".into()],
|
||
..CurrentState::default()
|
||
};
|
||
let bundle = Bundle {
|
||
scripts: vec![],
|
||
routes: vec![],
|
||
triggers: vec![],
|
||
secrets: vec![],
|
||
vars: std::collections::BTreeMap::new(),
|
||
};
|
||
let plan = compute_diff(¤t, &bundle);
|
||
assert_eq!(plan.scripts[0].op, Op::Delete);
|
||
assert_eq!(plan.secrets[0].op, Op::Delete);
|
||
}
|
||
|
||
#[test]
|
||
fn route_rebind_is_update() {
|
||
let s = script("a", "x");
|
||
let sid = s.id;
|
||
let route = Route {
|
||
id: uuid::Uuid::new_v4(),
|
||
app_id: AppId::from(uuid::Uuid::nil()),
|
||
script_id: sid,
|
||
host_kind: HostKind::Any,
|
||
host: String::new(),
|
||
host_param_name: None,
|
||
path_kind: PathKind::Exact,
|
||
path: "/p".into(),
|
||
method: Some("POST".into()),
|
||
dispatch_mode: DispatchMode::Sync,
|
||
enabled: true,
|
||
created_at: Utc::now(),
|
||
};
|
||
let current = CurrentState {
|
||
scripts: vec![s, script("b", "y")],
|
||
routes: vec![route],
|
||
..CurrentState::default()
|
||
};
|
||
// Same tuple, bound to a different script → update.
|
||
let bundle = Bundle {
|
||
scripts: vec![],
|
||
routes: vec![BundleRoute {
|
||
script: "b".into(),
|
||
method: Some("POST".into()),
|
||
host_kind: HostKind::Any,
|
||
host: String::new(),
|
||
host_param_name: None,
|
||
path_kind: PathKind::Exact,
|
||
path: "/p".into(),
|
||
dispatch_mode: DispatchMode::Sync,
|
||
enabled: true,
|
||
}],
|
||
triggers: vec![],
|
||
secrets: vec![],
|
||
vars: std::collections::BTreeMap::new(),
|
||
};
|
||
let plan = compute_diff(¤t, &bundle);
|
||
assert_eq!(plan.routes[0].op, Op::Update);
|
||
}
|
||
|
||
fn empty_bundle() -> Bundle {
|
||
Bundle {
|
||
scripts: vec![],
|
||
routes: vec![],
|
||
triggers: vec![],
|
||
secrets: vec![],
|
||
vars: std::collections::BTreeMap::new(),
|
||
}
|
||
}
|
||
|
||
fn trig(script_id: ScriptId, details: TriggerDetails) -> Trigger {
|
||
use crate::trigger_config::BackoffShape;
|
||
use crate::trigger_repo::TriggerKind;
|
||
let kind = match &details {
|
||
TriggerDetails::Cron { .. } => TriggerKind::Cron,
|
||
TriggerDetails::DeadLetter { .. } => TriggerKind::DeadLetter,
|
||
TriggerDetails::Queue { .. } => TriggerKind::Queue,
|
||
// Other kinds aren't distinguished by these tests (the diff keys
|
||
// on `details`, not `kind`).
|
||
_ => TriggerKind::Kv,
|
||
};
|
||
Trigger {
|
||
id: TriggerId::from(uuid::Uuid::new_v4()),
|
||
app_id: AppId::from(uuid::Uuid::nil()),
|
||
script_id,
|
||
name: "t".into(),
|
||
kind,
|
||
enabled: true,
|
||
dispatch_mode: TriggerDispatchMode::Async,
|
||
retry_max_attempts: 3,
|
||
retry_backoff: BackoffShape::Exponential,
|
||
retry_base_ms: 1000,
|
||
registered_by_principal: AdminUserId::from(uuid::Uuid::nil()),
|
||
created_at: Utc::now(),
|
||
updated_at: Utc::now(),
|
||
details,
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn script_update_only_on_real_change() {
|
||
// `script("a", ..)` defaults to timeout 30, mem 256, empty sandbox.
|
||
let current = CurrentState {
|
||
scripts: vec![script("a", "let x = 1;")],
|
||
..CurrentState::default()
|
||
};
|
||
// Same source, optional fields unset → NoOp ("None = leave as-is").
|
||
let mut b = empty_bundle();
|
||
b.scripts = vec![bundle_script("a", "let x = 1;")];
|
||
assert_eq!(compute_diff(¤t, &b).scripts[0].op, Op::NoOp);
|
||
// Explicit timeout that MATCHES current → still NoOp.
|
||
let mut same = empty_bundle();
|
||
let mut s = bundle_script("a", "let x = 1;");
|
||
s.timeout_seconds = Some(30);
|
||
same.scripts = vec![s];
|
||
assert_eq!(compute_diff(¤t, &same).scripts[0].op, Op::NoOp);
|
||
// Explicit timeout that DIFFERS → Update.
|
||
let mut chg = empty_bundle();
|
||
let mut s2 = bundle_script("a", "let x = 1;");
|
||
s2.timeout_seconds = Some(99);
|
||
chg.scripts = vec![s2];
|
||
assert_eq!(compute_diff(¤t, &chg).scripts[0].op, Op::Update);
|
||
// kind change → Update.
|
||
let mut chg2 = empty_bundle();
|
||
let mut s3 = bundle_script("a", "let x = 1;");
|
||
s3.kind = ScriptKind::Module;
|
||
chg2.scripts = vec![s3];
|
||
assert_eq!(compute_diff(¤t, &chg2).scripts[0].op, Op::Update);
|
||
}
|
||
|
||
#[test]
|
||
fn description_is_sparse() {
|
||
// An omitted (`None`) description must mean "leave as-is", matching
|
||
// the other optional fields — not "clear it".
|
||
let mut cur = script("a", "let x = 1;");
|
||
cur.description = Some("kept".into());
|
||
let current = CurrentState {
|
||
scripts: vec![cur],
|
||
..CurrentState::default()
|
||
};
|
||
// Omitted → NoOp (leave the stored description alone).
|
||
let mut omit = empty_bundle();
|
||
omit.scripts = vec![bundle_script("a", "let x = 1;")];
|
||
assert_eq!(compute_diff(¤t, &omit).scripts[0].op, Op::NoOp);
|
||
// Same explicit value → NoOp.
|
||
let mut same = empty_bundle();
|
||
let mut s = bundle_script("a", "let x = 1;");
|
||
s.description = Some("kept".into());
|
||
same.scripts = vec![s];
|
||
assert_eq!(compute_diff(¤t, &same).scripts[0].op, Op::NoOp);
|
||
// Different explicit value → Update.
|
||
let mut chg = empty_bundle();
|
||
let mut s2 = bundle_script("a", "let x = 1;");
|
||
s2.description = Some("changed".into());
|
||
chg.scripts = vec![s2];
|
||
assert_eq!(compute_diff(¤t, &chg).scripts[0].op, Op::Update);
|
||
}
|
||
|
||
#[test]
|
||
fn trigger_shape_validation_matches_interactive_api() {
|
||
// Empty collection_glob is rejected (kv/docs/files).
|
||
assert!(validate_trigger_shape(&BundleTrigger::Kv {
|
||
script: "h".into(),
|
||
collection_glob: " ".into(),
|
||
ops: vec![],
|
||
dispatch_mode: None,
|
||
retry_max_attempts: None,
|
||
})
|
||
.is_err());
|
||
// A non-empty glob passes.
|
||
assert!(validate_trigger_shape(&BundleTrigger::Docs {
|
||
script: "h".into(),
|
||
collection_glob: "users".into(),
|
||
ops: vec![],
|
||
dispatch_mode: None,
|
||
retry_max_attempts: None,
|
||
})
|
||
.is_ok());
|
||
// A malformed pubsub topic_pattern (mid-pattern wildcard) is rejected;
|
||
// a valid one passes.
|
||
assert!(validate_trigger_shape(&BundleTrigger::Pubsub {
|
||
script: "h".into(),
|
||
topic_pattern: "user.*.created".into(),
|
||
dispatch_mode: None,
|
||
retry_max_attempts: None,
|
||
})
|
||
.is_err());
|
||
assert!(validate_trigger_shape(&BundleTrigger::Pubsub {
|
||
script: "h".into(),
|
||
topic_pattern: "orders.created".into(),
|
||
dispatch_mode: None,
|
||
retry_max_attempts: None,
|
||
})
|
||
.is_ok());
|
||
// Queue visibility floor matches the interactive API (>= 30): a value
|
||
// below the floor is rejected; the floor itself and `None` are ok.
|
||
let queue = |vis| BundleTrigger::Queue {
|
||
script: "h".into(),
|
||
queue_name: "jobs".into(),
|
||
visibility_timeout_secs: vis,
|
||
dispatch_mode: None,
|
||
retry_max_attempts: None,
|
||
};
|
||
assert!(validate_trigger_shape(&queue(Some(10))).is_err());
|
||
assert!(validate_trigger_shape(&queue(Some(
|
||
crate::triggers_api::MIN_QUEUE_VISIBILITY_TIMEOUT_SECS
|
||
)))
|
||
.is_ok());
|
||
assert!(validate_trigger_shape(&queue(None)).is_ok());
|
||
// Empty email inbound_secret_ref is rejected.
|
||
assert!(validate_trigger_shape(&BundleTrigger::Email {
|
||
script: "h".into(),
|
||
inbound_secret_ref: " ".into(),
|
||
dispatch_mode: None,
|
||
retry_max_attempts: None,
|
||
})
|
||
.is_err());
|
||
}
|
||
|
||
#[test]
|
||
fn state_token_is_stable_order_independent_and_sensitive() {
|
||
let a = script("a", "x"); // version 1
|
||
let b = script("b", "y");
|
||
let base = CurrentState {
|
||
scripts: vec![a.clone(), b.clone()],
|
||
secret_names: vec!["S".into()],
|
||
..CurrentState::default()
|
||
};
|
||
// Deterministic + order-independent.
|
||
let reordered = CurrentState {
|
||
scripts: vec![b.clone(), a.clone()],
|
||
secret_names: vec!["S".into()],
|
||
..CurrentState::default()
|
||
};
|
||
assert_eq!(state_token(&base), state_token(&reordered));
|
||
// A script edit bumps `version`, which must flip the token even if the
|
||
// source-by-name set is otherwise unchanged (out-of-band redeploy).
|
||
let mut a2 = a.clone();
|
||
a2.version += 1;
|
||
let bumped = CurrentState {
|
||
scripts: vec![a2, b.clone()],
|
||
secret_names: vec!["S".into()],
|
||
..CurrentState::default()
|
||
};
|
||
assert_ne!(state_token(&base), state_token(&bumped));
|
||
// Adding a secret name flips it too.
|
||
let with_secret = CurrentState {
|
||
scripts: vec![a, b],
|
||
secret_names: vec!["S".into(), "T".into()],
|
||
..CurrentState::default()
|
||
};
|
||
assert_ne!(state_token(&base), state_token(&with_secret));
|
||
}
|
||
|
||
#[test]
|
||
fn diff_vars_create_update_noop_delete() {
|
||
let current = CurrentState {
|
||
vars: vec![
|
||
("region".into(), serde_json::json!("eu")),
|
||
("keep".into(), serde_json::json!(1)),
|
||
("stale".into(), serde_json::json!(true)),
|
||
],
|
||
..CurrentState::default()
|
||
};
|
||
let mut vars = std::collections::BTreeMap::new();
|
||
vars.insert("region".to_string(), serde_json::json!("us")); // value changed
|
||
vars.insert("keep".to_string(), serde_json::json!(1)); // identical
|
||
vars.insert("new".to_string(), serde_json::json!("x")); // not live
|
||
// "stale" is live but undeclared → Delete.
|
||
let bundle = Bundle {
|
||
vars,
|
||
..empty_bundle()
|
||
};
|
||
let plan = compute_diff(¤t, &bundle);
|
||
let op = |k: &str| plan.vars.iter().find(|c| c.key == k).unwrap().op;
|
||
assert_eq!(op("region"), Op::Update);
|
||
assert_eq!(op("keep"), Op::NoOp);
|
||
assert_eq!(op("new"), Op::Create);
|
||
assert_eq!(op("stale"), Op::Delete);
|
||
assert!(!plan.is_noop());
|
||
}
|
||
|
||
#[test]
|
||
fn state_token_is_sensitive_to_var_values_and_order_independent() {
|
||
let base = CurrentState {
|
||
vars: vec![("region".into(), serde_json::json!("eu"))],
|
||
..CurrentState::default()
|
||
};
|
||
// A value change must flip the token (covers the bound-plan check for
|
||
// an out-of-band `pic vars set` between plan and apply).
|
||
let changed = CurrentState {
|
||
vars: vec![("region".into(), serde_json::json!("us"))],
|
||
..CurrentState::default()
|
||
};
|
||
assert_ne!(state_token(&base), state_token(&changed));
|
||
// Order-independent across multiple vars.
|
||
let ab = CurrentState {
|
||
vars: vec![
|
||
("a".into(), serde_json::json!(1)),
|
||
("b".into(), serde_json::json!(2)),
|
||
],
|
||
..CurrentState::default()
|
||
};
|
||
let ba = CurrentState {
|
||
vars: vec![
|
||
("b".into(), serde_json::json!(2)),
|
||
("a".into(), serde_json::json!(1)),
|
||
],
|
||
..CurrentState::default()
|
||
};
|
||
assert_eq!(state_token(&ab), state_token(&ba));
|
||
}
|
||
|
||
#[test]
|
||
fn state_token_ignores_dead_letter_triggers() {
|
||
// The diff ignores dead-letter triggers (not manifest-representable),
|
||
// so the token must too — otherwise an out-of-band dead-letter change
|
||
// would force a spurious `--force`.
|
||
let s = script("h", "x");
|
||
let sid = s.id;
|
||
let base = CurrentState {
|
||
scripts: vec![s.clone()],
|
||
..CurrentState::default()
|
||
};
|
||
let with_dl = CurrentState {
|
||
scripts: vec![s],
|
||
triggers: vec![trig(
|
||
sid,
|
||
TriggerDetails::DeadLetter {
|
||
source_filter: None,
|
||
trigger_id_filter: None,
|
||
script_id_filter: None,
|
||
},
|
||
)],
|
||
..CurrentState::default()
|
||
};
|
||
assert_eq!(
|
||
state_token(&base),
|
||
state_token(&with_dl),
|
||
"dead-letter triggers must not affect the token"
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn enabled_is_declarative_and_diffed() {
|
||
// A live-active script the manifest marks disabled → Update; the token
|
||
// is sensitive to the flag so an out-of-band toggle is detectable.
|
||
let cur = script("a", "let x = 1;"); // enabled: true
|
||
let base = CurrentState {
|
||
scripts: vec![cur.clone()],
|
||
..CurrentState::default()
|
||
};
|
||
let mut disable = empty_bundle();
|
||
let mut bs = bundle_script("a", "let x = 1;");
|
||
bs.enabled = false;
|
||
disable.scripts = vec![bs];
|
||
assert_eq!(compute_diff(&base, &disable).scripts[0].op, Op::Update);
|
||
|
||
let mut flipped = cur;
|
||
flipped.enabled = false;
|
||
let toggled = CurrentState {
|
||
scripts: vec![flipped],
|
||
..CurrentState::default()
|
||
};
|
||
assert_ne!(
|
||
state_token(&base),
|
||
state_token(&toggled),
|
||
"token must flip on an enabled change"
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn warns_on_enabled_binding_to_disabled_script() {
|
||
let mut b = empty_bundle();
|
||
let mut s = bundle_script("h", "x");
|
||
s.enabled = false;
|
||
b.scripts = vec![s];
|
||
b.routes = vec![BundleRoute {
|
||
script: "h".into(),
|
||
method: None,
|
||
host_kind: HostKind::Any,
|
||
host: String::new(),
|
||
host_param_name: None,
|
||
path_kind: PathKind::Exact,
|
||
path: "/h".into(),
|
||
dispatch_mode: DispatchMode::Sync,
|
||
enabled: true,
|
||
}];
|
||
let w = disabled_target_warnings(&b);
|
||
assert!(
|
||
w.iter().any(|m| m.contains("will 404")),
|
||
"expected a 404 reachability warning: {w:?}"
|
||
);
|
||
// No warning when the script is active.
|
||
b.scripts[0].enabled = true;
|
||
assert!(disabled_target_warnings(&b).is_empty());
|
||
}
|
||
|
||
#[test]
|
||
fn warns_on_unreachable_endpoint() {
|
||
// §4.7: an enabled endpoint with no route and no trigger is flagged.
|
||
let mut b = empty_bundle();
|
||
b.scripts = vec![bundle_script("orphan", "x")];
|
||
let w = unreachable_endpoint_warnings(&b);
|
||
assert!(
|
||
w.iter().any(|m| m.contains("no route or trigger")),
|
||
"expected an unreachable-endpoint warning: {w:?}"
|
||
);
|
||
// Bound by a route → no warning.
|
||
b.routes = vec![BundleRoute {
|
||
script: "orphan".into(),
|
||
method: None,
|
||
host_kind: HostKind::Any,
|
||
host: String::new(),
|
||
host_param_name: None,
|
||
path_kind: PathKind::Exact,
|
||
path: "/o".into(),
|
||
dispatch_mode: DispatchMode::Sync,
|
||
enabled: true,
|
||
}];
|
||
assert!(unreachable_endpoint_warnings(&b).is_empty());
|
||
// A module is exempt even with no binding.
|
||
let mut m = empty_bundle();
|
||
let mut lib = bundle_script("lib", "x");
|
||
lib.kind = ScriptKind::Module;
|
||
m.scripts = vec![lib];
|
||
assert!(unreachable_endpoint_warnings(&m).is_empty());
|
||
}
|
||
|
||
#[test]
|
||
fn trigger_diff_create_noop_delete() {
|
||
let s = script("h", "x");
|
||
let sid = s.id;
|
||
let cron = |sched: &str| TriggerDetails::Cron {
|
||
schedule: sched.into(),
|
||
timezone: "UTC".into(),
|
||
last_fired_at: None,
|
||
};
|
||
let current = CurrentState {
|
||
scripts: vec![s],
|
||
triggers: vec![trig(sid, cron("0 0 * * * *"))],
|
||
..CurrentState::default()
|
||
};
|
||
let bundle_cron = |sched: &str| BundleTrigger::Cron {
|
||
script: "h".into(),
|
||
schedule: sched.into(),
|
||
timezone: "UTC".into(),
|
||
dispatch_mode: None,
|
||
retry_max_attempts: None,
|
||
};
|
||
// Same schedule → NoOp.
|
||
let mut same = empty_bundle();
|
||
same.triggers = vec![bundle_cron("0 0 * * * *")];
|
||
let p = compute_diff(¤t, &same);
|
||
assert!(p.triggers.iter().all(|c| c.op == Op::NoOp), "{p:?}");
|
||
// Removed → Delete.
|
||
let p = compute_diff(¤t, &empty_bundle());
|
||
assert_eq!(p.triggers.len(), 1);
|
||
assert_eq!(p.triggers[0].op, Op::Delete);
|
||
// Different schedule → Create (new) + Delete (old).
|
||
let mut chg = empty_bundle();
|
||
chg.triggers = vec![bundle_cron("0 5 * * * *")];
|
||
let p = compute_diff(¤t, &chg);
|
||
assert!(p.triggers.iter().any(|c| c.op == Op::Create), "{p:?}");
|
||
assert!(p.triggers.iter().any(|c| c.op == Op::Delete), "{p:?}");
|
||
}
|
||
|
||
#[test]
|
||
fn kv_ops_order_insensitive() {
|
||
let s = script("h", "x");
|
||
let sid = s.id;
|
||
let current = CurrentState {
|
||
scripts: vec![s],
|
||
triggers: vec![trig(
|
||
sid,
|
||
TriggerDetails::Kv {
|
||
collection_glob: "users".into(),
|
||
ops: vec![KvEventOp::Update, KvEventOp::Insert],
|
||
},
|
||
)],
|
||
..CurrentState::default()
|
||
};
|
||
let mut b = empty_bundle();
|
||
b.triggers = vec![BundleTrigger::Kv {
|
||
script: "h".into(),
|
||
collection_glob: "users".into(),
|
||
ops: vec![KvEventOp::Insert, KvEventOp::Update], // reversed
|
||
dispatch_mode: None,
|
||
retry_max_attempts: None,
|
||
}];
|
||
let p = compute_diff(¤t, &b);
|
||
assert!(
|
||
p.triggers.iter().all(|c| c.op == Op::NoOp),
|
||
"ops order must not matter: {p:?}"
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn dead_letter_trigger_is_ignored_by_diff() {
|
||
// Dead-letter triggers can't be expressed in the manifest, so the
|
||
// diff must neither match nor prune them.
|
||
let s = script("h", "x");
|
||
let sid = s.id;
|
||
let current = CurrentState {
|
||
scripts: vec![s],
|
||
triggers: vec![trig(
|
||
sid,
|
||
TriggerDetails::DeadLetter {
|
||
source_filter: None,
|
||
trigger_id_filter: None,
|
||
script_id_filter: None,
|
||
},
|
||
)],
|
||
..CurrentState::default()
|
||
};
|
||
let p = compute_diff(¤t, &empty_bundle());
|
||
assert!(p.triggers.is_empty(), "dead-letter must be ignored: {p:?}");
|
||
}
|
||
|
||
#[test]
|
||
fn email_trigger_is_never_pruned() {
|
||
// `pull` can't represent email triggers (the server keeps the sealed
|
||
// secret, not the `inbound_secret_ref`), so their absence from a
|
||
// bundle is ambiguous — the diff must not emit a Delete for them.
|
||
let s = script("h", "x");
|
||
let sid = s.id;
|
||
let current = CurrentState {
|
||
scripts: vec![s],
|
||
triggers: vec![trig(
|
||
sid,
|
||
TriggerDetails::Email {
|
||
has_inbound_secret: true,
|
||
},
|
||
)],
|
||
..CurrentState::default()
|
||
};
|
||
let p = compute_diff(¤t, &empty_bundle());
|
||
assert!(
|
||
!p.triggers.iter().any(|c| c.op == Op::Delete),
|
||
"email trigger must not be pruned: {p:?}"
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn queue_rebind_is_noop_documented_limitation() {
|
||
// Queue identity is `queue|{queue_name}` (script-independent), so a
|
||
// rebind to a different script diffs as NoOp — a known limitation.
|
||
let a = script("a", "x");
|
||
let aid = a.id;
|
||
let current = CurrentState {
|
||
scripts: vec![a, script("b", "y")],
|
||
triggers: vec![trig(
|
||
aid,
|
||
TriggerDetails::Queue {
|
||
queue_name: "jobs".into(),
|
||
visibility_timeout_secs: 30,
|
||
last_fired_at: None,
|
||
},
|
||
)],
|
||
..CurrentState::default()
|
||
};
|
||
let mut b = empty_bundle();
|
||
b.triggers = vec![BundleTrigger::Queue {
|
||
script: "b".into(), // rebind a→b
|
||
queue_name: "jobs".into(),
|
||
visibility_timeout_secs: None,
|
||
dispatch_mode: None,
|
||
retry_max_attempts: None,
|
||
}];
|
||
let p = compute_diff(¤t, &b);
|
||
assert!(
|
||
p.triggers.iter().all(|c| c.op == Op::NoOp),
|
||
"queue identity is queue_name-only: {p:?}"
|
||
);
|
||
}
|
||
}
|