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

feat(apply): owner-generic apply engine — group-node plan/apply (Phase 5 C1)

Browse Source 
Phase 5 C1. Generalize the reconcile engine from app-only to an `ApplyOwner
{ App | Group }`, so a GROUP's own config + scripts can be declaratively
reconciled — the foundation for the tree-spanning project tool.

A group node is a subset of an app node (scripts + vars only; no routes /
triggers / secrets-values), so the diff, script/route/trigger/var CRUD, prune,
idempotency, and the Phase-4 inherited-target resolution are all reused
unchanged. Only the owner-varying bits switch on `ApplyOwner`:
  * `load_current` — `list_for_group` + `VarOwner::Group` for a group (empty
    routes/triggers/secrets); `list_for_app` for an app, exactly as before.
  * script create owner (`NewScript{app_id|group_id}`), var writes (new
    `set_group_var_tx`/`delete_group_var_tx` mirroring the app variants at
    scope `*`), and the advisory lock (owner-tagged).
  * `validate_bundle_for` rejects routes/triggers on a group bundle (422);
    route-host validation, inherited-target resolution, the post-commit route
    refresh, and the unreachable-endpoint warning are app-only.

`apply`/`plan` are unchanged thin wrappers over the new `apply_owner`/
`plan_owner`. New endpoints `POST /groups/{id}/{plan,apply}` gated by
`GroupScripts{Read,Write}` / `GroupVarsWrite`. `ApplyService` gains a `groups`
repo (the tree apply in C3 needs it too).

Live-validated: group plan → apply (group-owned script + var) → idempotent
re-plan noop → routes rejected 422 → DB confirms group ownership. App apply
unchanged: 391 manager-core unit tests + the apply/prune/staleness/overlay
journeys all green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-06-25 21:46:52 +02:00
parent 99e7100f0b
commit 65049a6262
3 changed files with 360 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
crates/manager-core/src/apply_api.rs
View File

@@ -11,21 +11,24 @@ use axum::{
routing::post,
Extension, Json, Router,
};
use picloud_shared::{AppId, Principal};
use picloud_shared::{AppId, GroupId, Principal};
use serde::Deserialize;
use serde_json::json;
use crate::app_repo::AppRepository;
use crate::apply_service::{
ApplyError, ApplyReport, ApplyService, Bundle, BundleTrigger, PlanResult,
ApplyError, ApplyOwner, ApplyReport, ApplyService, Bundle, BundleTrigger, PlanResult,
};
use crate::authz::{require, AuthzDenied, Capability};
use crate::group_repo::GroupRepository;
/// Build the apply/plan router. Mounted under `/api/v1/admin`.
pub fn apply_router(service: ApplyService) -> Router {
Router::new()
.route("/apps/{id}/plan", post(plan_handler))
.route("/apps/{id}/apply", post(apply_handler))
.route("/groups/{id}/plan", post(group_plan_handler))
.route("/groups/{id}/apply", post(group_apply_handler))
.with_state(service)
}
@@ -135,6 +138,91 @@ async fn plan_handler(
Ok(Json(plan))
}
// ----------------------------------------------------------------------------
// Group node apply (Phase 5): a group declares only scripts + vars (+ secret
// names). The bundle reuses the app `Bundle` shape with empty routes/triggers;
// the service rejects routes/triggers on a group node.
// ----------------------------------------------------------------------------
async fn group_plan_handler(
State(svc): State<ApplyService>,
Extension(principal): Extension<Principal>,
Path(id_or_slug): Path<String>,
Json(bundle): Json<Bundle>,
) -> Result<Json<PlanResult>, ApplyError> {
let group_id = resolve_group_id(svc.groups.as_ref(), &id_or_slug).await?;
// Plan discloses the group's script + var + secret names; viewer-tier reads.
require(
svc.authz.as_ref(),
&principal,
Capability::GroupScriptsRead(group_id),
)
.await
.map_err(map_authz)?;
let plan = svc.plan_owner(ApplyOwner::Group(group_id), &bundle).await?;
Ok(Json(plan))
}
async fn group_apply_handler(
State(svc): State<ApplyService>,
Extension(principal): Extension<Principal>,
Path(id_or_slug): Path<String>,
Json(req): Json<ApplyRequest>,
) -> Result<Json<ApplyReport>, ApplyError> {
let group_id = resolve_group_id(svc.groups.as_ref(), &id_or_slug).await?;
// Write caps per touched kind (+ all kinds on prune). Group scripts are
// editor-tier (`GroupScriptsWrite`), group vars likewise (`GroupVarsWrite`).
if req.prune || !req.bundle.scripts.is_empty() {
require(
svc.authz.as_ref(),
&principal,
Capability::GroupScriptsWrite(group_id),
)
.await
.map_err(map_authz)?;
}
if req.prune || !req.bundle.vars.is_empty() {
require(
svc.authz.as_ref(),
&principal,
Capability::GroupVarsWrite(group_id),
)
.await
.map_err(map_authz)?;
}
let report = svc
.apply_owner(
ApplyOwner::Group(group_id),
&req.bundle,
req.prune,
principal.user_id,
req.expected_token.as_deref(),
)
.await?;
Ok(Json(report))
}
/// Resolve a slug-or-id path param to a `GroupId`, mapping miss → 404.
async fn resolve_group_id(
groups: &dyn GroupRepository,
ident: &str,
) -> Result<GroupId, ApplyError> {
let found = if let Ok(uuid) = ident.parse::<uuid::Uuid>() {
groups
.get_by_id(uuid.into())
.await
.map_err(|e| ApplyError::Backend(e.to_string()))?
} else {
groups
.get_by_slug(ident)
.await
.map_err(|e| ApplyError::Backend(e.to_string()))?
};
found
.map(|g| g.id)
.ok_or_else(|| ApplyError::AppNotFound(format!("group {ident}")))
}
/// Resolve a slug-or-id path param to an `AppId`, mapping miss → 404.
/// Mirrors the `triggers_api` helper of the same shape.
async fn resolve_app_id(apps: &dyn AppRepository, ident: &str) -> Result<AppId, ApplyError> {

322
crates/manager-core/src/apply_service.rs
View File

@@ -31,8 +31,9 @@ use std::sync::Arc;
use picloud_orchestrator_core::routing::{pattern, RouteTable};
use picloud_shared::{
AdminUserId, AppId, DispatchMode, DocsEventOp, FilesEventOp, HostKind, KvEventOp, MasterKey,
PathKind, Route, Script, ScriptId, ScriptKind, ScriptSandbox, ScriptValidator, TriggerId,
AdminUserId, AppId, DispatchMode, DocsEventOp, FilesEventOp, GroupId, HostKind, KvEventOp,
MasterKey, PathKind, Route, Script, ScriptId, ScriptKind, ScriptSandbox, ScriptValidator,
TriggerId,
};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
@@ -399,6 +400,8 @@ pub struct ApplyService {
/// 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>,
/// App domain claims — validates a route's host belongs to the app.
pub domains: Arc<dyn AppDomainRepository>,
pub authz: Arc<dyn AuthzRepo>,
@@ -419,27 +422,79 @@ impl ApplyService {
/// # Errors
/// `Invalid` for a malformed bundle; `Backend` for repo failures.
pub async fn plan(&self, app_id: AppId, bundle: &Bundle) -> Result<PlanResult, ApplyError> {
let inherited = self.resolve_inherited_targets(app_id, bundle).await?;
self.validate_bundle(bundle, &keys_set(&inherited))?;
self.validate_route_hosts(app_id, bundle).await?;
let current = self.load_current(app_id).await?;
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))?;
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(&current).await?;
validate_email_secrets_present(bundle, &current.secret_names)?;
Ok(PlanResult {
plan: compute_diff_with_names(&current, bundle, &current_names),
// Fingerprint of the live state this plan was computed against, so
// `apply` can refuse if the app changed underneath it (§4.2).
// `apply` can refuse if the node changed underneath it (§4.2).
state_token: state_token_with_names(&current, &current_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.
#[allow(clippy::too_many_lines)]
pub async fn apply(
&self,
app_id: AppId,
@@ -448,9 +503,35 @@ impl ApplyService {
actor: AdminUserId,
expected_token: Option<&str>,
) -> Result<ApplyReport, ApplyError> {
let inherited = self.resolve_inherited_targets(app_id, bundle).await?;
self.validate_bundle(bundle, &keys_set(&inherited))?;
self.validate_route_hosts(app_id, bundle).await?;
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))?;
if let ApplyOwner::App(app_id) = owner {
self.validate_route_hosts(app_id, bundle).await?;
}
let mut tx = self
.pool
@@ -466,12 +547,12 @@ impl ApplyService {
// 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(app_id))
.bind(apply_lock_key(owner))
.execute(&mut *tx)
.await
.map_err(|e| ApplyError::Backend(e.to_string()))?;
let current = self.load_current(app_id).await?;
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(&current).await?;
@@ -497,9 +578,14 @@ impl ApplyService {
// 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));
report
.warnings
.extend(unreachable_endpoint_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));
}
let bundle_scripts: HashMap<String, &BundleScript> = bundle
.scripts
@@ -524,9 +610,10 @@ impl ApplyService {
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: Some(app_id),
group_id: None,
app_id: script_app_id,
group_id: script_group_id,
name: bs.name.clone(),
description: bs.description.clone(),
source: bs.source.clone(),
@@ -638,6 +725,9 @@ impl ApplyService {
};
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;
@@ -655,6 +745,8 @@ impl ApplyService {
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 {
@@ -688,11 +780,15 @@ impl ApplyService {
for ch in &plan.vars {
match ch.op {
Op::Create => {
set_app_var_tx(&mut tx, app_id, &ch.key, &bundle.vars[&ch.key]).await?;
owner
.set_var_tx(&mut tx, &ch.key, &bundle.vars[&ch.key])
.await?;
report.vars_created += 1;
}
Op::Update => {
set_app_var_tx(&mut tx, app_id, &ch.key, &bundle.vars[&ch.key]).await?;
owner
.set_var_tx(&mut tx, &ch.key, &bundle.vars[&ch.key])
.await?;
report.vars_updated += 1;
}
Op::NoOp | Op::Delete => {}
@@ -770,7 +866,7 @@ impl ApplyService {
// pruned): a live app var the manifest dropped is deleted.
for ch in &plan.vars {
if ch.op == Op::Delete {
delete_app_var_tx(&mut tx, app_id, &ch.key).await?;
owner.delete_var_tx(&mut tx, &ch.key).await?;
report.vars_deleted += 1;
}
}
@@ -782,12 +878,15 @@ impl ApplyService {
// 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.
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());
// 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)
}
@@ -1146,30 +1245,44 @@ impl ApplyService {
Ok(())
}
async fn load_current(&self, app_id: AppId) -> Result<CurrentState, ApplyError> {
let scripts = self
.scripts
.list_for_app(app_id)
.await
.map_err(|e| ApplyError::Backend(e.to_string()))?;
let routes = self
.routes
.list_for_app(app_id)
.await
.map_err(|e| ApplyError::Backend(e.to_string()))?;
let triggers = self
.triggers
.list_for_app(app_id)
.await
.map_err(|e| ApplyError::Backend(e.to_string()))?;
let secret_names = self.all_secret_names(app_id).await?;
// App's OWN vars, narrowed to scope `*` and real values: the manifest
// reconciles app-own env-agnostic config, not inherited group vars
// (those aren't in this owner's rows at all) nor tombstones (an
// imperative suppress the manifest can't express).
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(VarOwner::App(app_id))
.list_for_owner(var_owner)
.await
.map_err(|e| ApplyError::Backend(e.to_string()))?
.into_iter()
@@ -1350,6 +1463,99 @@ async fn delete_app_var_tx(
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
@@ -2118,13 +2324,23 @@ fn state_token_with_names(
format!("{h:016x}")
}
/// Per-app advisory lock key, namespaced so it can't collide with the
/// queue-trigger lock space.
fn apply_lock_key(app_id: AppId) -> i64 {
/// 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);
app_id.into_inner().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())
}

1
crates/picloud/src/lib.rs
View File

@@ -473,6 +473,7 @@ pub async fn build_app(
secrets: secrets_repo.clone(),
vars: Arc::new(PostgresVarsRepo::new(pool.clone())),
apps: apps_repo.clone(),
groups: groups_repo.clone(),
domains: domains_repo.clone(),
authz: authz.clone(),
validator: engine.clone(),