feat(vars): admin CRUD API + pic vars CLI

Completes the vars half of Phase 3 end-to-end:
- vars_repo: VarOwner{Group|App} upsert/delete/list (owner-kind-specific
  SQL; ON CONFLICT restates the partial-index predicate).
- vars_api: GET/PUT/DELETE under /apps/{id}/vars and /groups/{id}/vars,
  resolve-then-require gated on App/GroupVars{Read,Write}, secrets-style
  error mapping + key/env-scope validation.
- pic vars ls/set/rm (--group|--app, --env, --json, --tombstone).
- journey test: a group var is inherited by a descendant app's
  vars::get(), and an app-level value overrides it (proximity) — green.

386 manager-core lib tests + the vars journey pass; clippy clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/manager-core/src/vars_repo.rs

@@ -0,0 +1,210 @@
+//! Low-level Postgres CRUD over `vars` — the write/admin side of the
+//! Phase-3 config layer (the read/resolution side lives in
+//! `config_resolver`). Storage-only: it upserts and lists an owner's OWN
+//! rows. Authorization, env-scope validation, and value encoding live one
+//! layer up in `vars_api`.
+//!
+//! A var is owned by exactly one group OR one app (the migration's
+//! `vars_owner_exactly_one` CHECK). Because the owner is split across two
+//! nullable columns (`group_id`, `app_id`), the upsert writes
+//! owner-kind-specific SQL with the matching partial-unique conflict
+//! target.
+
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use picloud_shared::{AppId, GroupId};
+use serde_json::Value as JsonValue;
+use sqlx::PgPool;
+
+#[derive(Debug, thiserror::Error)]
+pub enum VarsRepoError {
+    #[error("database error: {0}")]
+    Db(#[from] sqlx::Error),
+}
+
+/// Which side of the polymorphic owner a var hangs off. The repo chooses
+/// the conflict target (group vs app partial-unique index) from this.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum VarOwner {
+    Group(GroupId),
+    App(AppId),
+}
+
+/// One of an owner's OWN var rows (NOT a resolved/inherited value). Backs
+/// the admin list surface.
+#[derive(Debug, Clone)]
+pub struct VarRow {
+    pub environment_scope: String,
+    pub key: String,
+    pub value: JsonValue,
+    pub is_tombstone: bool,
+    pub updated_at: DateTime<Utc>,
+}
+
+/// Repo surface. A trait so service/handler tests can substitute an
+/// in-memory backing without Postgres.
+#[async_trait]
+pub trait VarsRepo: Send + Sync {
+    /// Upsert one (owner, env_scope, key) row.
+    async fn set(
+        &self,
+        owner: VarOwner,
+        env_scope: &str,
+        key: &str,
+        value: &JsonValue,
+        is_tombstone: bool,
+    ) -> Result<(), VarsRepoError>;
+
+    /// Delete one (owner, env_scope, key) row; returns whether a row was
+    /// present.
+    async fn delete(
+        &self,
+        owner: VarOwner,
+        env_scope: &str,
+        key: &str,
+    ) -> Result<bool, VarsRepoError>;
+
+    /// The owner's OWN rows only (NOT resolved/inherited), ordered by
+    /// (key, environment_scope).
+    async fn list_for_owner(&self, owner: VarOwner) -> Result<Vec<VarRow>, VarsRepoError>;
+}
+
+pub struct PostgresVarsRepo {
+    pool: PgPool,
+}
+
+impl PostgresVarsRepo {
+    #[must_use]
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl VarsRepo for PostgresVarsRepo {
+    async fn set(
+        &self,
+        owner: VarOwner,
+        env_scope: &str,
+        key: &str,
+        value: &JsonValue,
+        is_tombstone: bool,
+    ) -> Result<(), VarsRepoError> {
+        // Owner-kind-specific SQL: only one of the two nullable owner
+        // columns is written, and the conflict target is the matching
+        // partial-unique index.
+        match owner {
+            VarOwner::Group(g) => {
+                sqlx::query(
+                    // The conflict target is a PARTIAL unique index, so the
+                    // index predicate (`WHERE group_id IS NOT NULL`) must be
+                    // restated for Postgres to infer the arbiter.
+                    "INSERT INTO vars (group_id, environment_scope, key, value, is_tombstone) \
+                     VALUES ($1, $2, $3, $4, $5) \
+                     ON CONFLICT (group_id, environment_scope, key) \
+                       WHERE group_id IS NOT NULL DO UPDATE \
+                       SET value = EXCLUDED.value, \
+                           is_tombstone = EXCLUDED.is_tombstone, \
+                           updated_at = NOW()",
+                )
+                .bind(g.into_inner())
+                .bind(env_scope)
+                .bind(key)
+                .bind(value)
+                .bind(is_tombstone)
+                .execute(&self.pool)
+                .await?;
+            }
+            VarOwner::App(a) => {
+                sqlx::query(
+                    // Partial-index conflict target — restate the predicate.
+                    "INSERT INTO vars (app_id, environment_scope, key, value, is_tombstone) \
+                     VALUES ($1, $2, $3, $4, $5) \
+                     ON CONFLICT (app_id, environment_scope, key) \
+                       WHERE app_id IS NOT NULL DO UPDATE \
+                       SET value = EXCLUDED.value, \
+                           is_tombstone = EXCLUDED.is_tombstone, \
+                           updated_at = NOW()",
+                )
+                .bind(a.into_inner())
+                .bind(env_scope)
+                .bind(key)
+                .bind(value)
+                .bind(is_tombstone)
+                .execute(&self.pool)
+                .await?;
+            }
+        }
+        Ok(())
+    }
+
+    async fn delete(
+        &self,
+        owner: VarOwner,
+        env_scope: &str,
+        key: &str,
+    ) -> Result<bool, VarsRepoError> {
+        let res = match owner {
+            VarOwner::Group(g) => {
+                sqlx::query(
+                    "DELETE FROM vars \
+                     WHERE group_id = $1 AND environment_scope = $2 AND key = $3",
+                )
+                .bind(g.into_inner())
+                .bind(env_scope)
+                .bind(key)
+                .execute(&self.pool)
+                .await?
+            }
+            VarOwner::App(a) => {
+                sqlx::query(
+                    "DELETE FROM vars \
+                     WHERE app_id = $1 AND environment_scope = $2 AND key = $3",
+                )
+                .bind(a.into_inner())
+                .bind(env_scope)
+                .bind(key)
+                .execute(&self.pool)
+                .await?
+            }
+        };
+        Ok(res.rows_affected() > 0)
+    }
+
+    async fn list_for_owner(&self, owner: VarOwner) -> Result<Vec<VarRow>, VarsRepoError> {
+        let rows: Vec<(String, String, JsonValue, bool, DateTime<Utc>)> = match owner {
+            VarOwner::Group(g) => {
+                sqlx::query_as(
+                    "SELECT environment_scope, key, value, is_tombstone, updated_at \
+                 FROM vars WHERE group_id = $1 \
+                 ORDER BY key ASC, environment_scope ASC",
+                )
+                .bind(g.into_inner())
+                .fetch_all(&self.pool)
+                .await?
+            }
+            VarOwner::App(a) => {
+                sqlx::query_as(
+                    "SELECT environment_scope, key, value, is_tombstone, updated_at \
+                 FROM vars WHERE app_id = $1 \
+                 ORDER BY key ASC, environment_scope ASC",
+                )
+                .bind(a.into_inner())
+                .fetch_all(&self.pool)
+                .await?
+            }
+        };
+        Ok(rows
+            .into_iter()
+            .map(
+                |(environment_scope, key, value, is_tombstone, updated_at)| VarRow {
+                    environment_scope,
+                    key,
+                    value,
+                    is_tombstone,
+                    updated_at,
+                },
+            )
+            .collect())
+    }
+}