feat(modules): group-collection registry + group-KV storage schema (§11.6 C1)

Data layer for shared cross-app group collections (KV-only MVP), nothing wired
yet. Two migrations + two repos, modeled on the extension_points (0051) marker
and the kv_entries (0007) store:

- 0052_group_collections.sql: owner-polymorphic marker table declaring a
  collection name group-shared, with a `kind` discriminator (CHECK 'kv' for
  now; generalizes to docs/files/topics/queue later) and per-owner
  partial-unique (owner, LOWER(name), kind) indexes. CASCADE — a marker is
  config, not code.
- 0053_group_kv_entries.sql: the shared store, keyed by (group_id, collection,
  key) — NO app_id (a shared row belongs to the group). CASCADE on group delete
  (data dies with its group, like vars/secrets config; an app delete leaves it).
- group_collection_repo: list_for_owner, insert/delete_collection_tx, and the
  load-bearing resolve_owning_group — walks the reading app's chain
  (CHAIN_LEVELS_CTE) for the nearest ancestor group declaring the name
  (nearest-wins via ORDER BY depth LIMIT 1). That walk IS the isolation
  boundary; the join is on group_owner only.
- group_kv_repo: a near-clone of kv_repo keyed by group_id.

Schema snapshot re-blessed (53 migrations).

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

crates/manager-core/src/group_kv_repo.rs

@@ -0,0 +1,222 @@
+//! Low-level Postgres CRUD over `group_kv_entries` (§11.6). A near-clone of
+//! [`crate::kv_repo`] keyed by the owning `group_id` instead of `app_id` —
+//! authorization, group resolution, event policy, and empty-collection
+//! validation live one layer up in `GroupKvServiceImpl`.
+
+use async_trait::async_trait;
+use base64::engine::general_purpose::URL_SAFE_NO_PAD;
+use base64::Engine as _;
+use picloud_shared::{GroupId, KvListPage};
+use sqlx::PgPool;
+
+#[derive(Debug, thiserror::Error)]
+pub enum GroupKvRepoError {
+    #[error("database error: {0}")]
+    Db(#[from] sqlx::Error),
+
+    #[error("invalid pagination cursor")]
+    InvalidCursor,
+}
+
+/// Repo surface. The trait is exposed so tests can substitute an in-memory
+/// backing without spinning up Postgres.
+#[async_trait]
+pub trait GroupKvRepo: Send + Sync {
+    async fn get(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        key: &str,
+    ) -> Result<Option<serde_json::Value>, GroupKvRepoError>;
+
+    /// Upserts the row. Returns the previous value (if any) so callers can
+    /// determine whether this was an `insert` or an `update`.
+    async fn set(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        key: &str,
+        value: serde_json::Value,
+    ) -> Result<Option<serde_json::Value>, GroupKvRepoError>;
+
+    /// Returns the deleted value if present, `None` if the row didn't exist.
+    async fn delete(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        key: &str,
+    ) -> Result<Option<serde_json::Value>, GroupKvRepoError>;
+
+    async fn has(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        key: &str,
+    ) -> Result<bool, GroupKvRepoError>;
+
+    async fn list(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        cursor: Option<&str>,
+        limit: u32,
+    ) -> Result<KvListPage, GroupKvRepoError>;
+}
+
+pub struct PostgresGroupKvRepo {
+    pool: PgPool,
+}
+
+impl PostgresGroupKvRepo {
+    #[must_use]
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+/// Hard ceiling on `list` page size (mirrors `kv_repo`).
+const KV_LIST_MAX_LIMIT: u32 = 1_000;
+const KV_LIST_DEFAULT_LIMIT: u32 = 100;
+
+#[async_trait]
+impl GroupKvRepo for PostgresGroupKvRepo {
+    async fn get(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        key: &str,
+    ) -> Result<Option<serde_json::Value>, GroupKvRepoError> {
+        let row: Option<(serde_json::Value,)> = sqlx::query_as(
+            "SELECT value FROM group_kv_entries \
+             WHERE group_id = $1 AND collection = $2 AND key = $3",
+        )
+        .bind(group_id.into_inner())
+        .bind(collection)
+        .bind(key)
+        .fetch_optional(&self.pool)
+        .await?;
+        Ok(row.map(|(v,)| v))
+    }
+
+    async fn set(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        key: &str,
+        value: serde_json::Value,
+    ) -> Result<Option<serde_json::Value>, GroupKvRepoError> {
+        let row: Option<(Option<serde_json::Value>,)> = sqlx::query_as(
+            "WITH prev AS (\
+                SELECT value FROM group_kv_entries \
+                WHERE group_id = $1 AND collection = $2 AND key = $3\
+             ), \
+             upserted AS (\
+                INSERT INTO group_kv_entries (group_id, collection, key, value) \
+                VALUES ($1, $2, $3, $4) \
+                ON CONFLICT (group_id, collection, key) DO UPDATE \
+                  SET value = EXCLUDED.value, updated_at = NOW() \
+                RETURNING 1\
+             ) \
+             SELECT (SELECT value FROM prev) FROM upserted",
+        )
+        .bind(group_id.into_inner())
+        .bind(collection)
+        .bind(key)
+        .bind(value)
+        .fetch_optional(&self.pool)
+        .await?;
+        Ok(row.and_then(|(v,)| v))
+    }
+
+    async fn delete(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        key: &str,
+    ) -> Result<Option<serde_json::Value>, GroupKvRepoError> {
+        let row: Option<(serde_json::Value,)> = sqlx::query_as(
+            "DELETE FROM group_kv_entries \
+             WHERE group_id = $1 AND collection = $2 AND key = $3 \
+             RETURNING value",
+        )
+        .bind(group_id.into_inner())
+        .bind(collection)
+        .bind(key)
+        .fetch_optional(&self.pool)
+        .await?;
+        Ok(row.map(|(v,)| v))
+    }
+
+    async fn has(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        key: &str,
+    ) -> Result<bool, GroupKvRepoError> {
+        let row: Option<(i64,)> = sqlx::query_as(
+            "SELECT 1 FROM group_kv_entries \
+             WHERE group_id = $1 AND collection = $2 AND key = $3",
+        )
+        .bind(group_id.into_inner())
+        .bind(collection)
+        .bind(key)
+        .fetch_optional(&self.pool)
+        .await?;
+        Ok(row.is_some())
+    }
+
+    async fn list(
+        &self,
+        group_id: GroupId,
+        collection: &str,
+        cursor: Option<&str>,
+        limit: u32,
+    ) -> Result<KvListPage, GroupKvRepoError> {
+        let limit = if limit == 0 {
+            KV_LIST_DEFAULT_LIMIT
+        } else {
+            limit.min(KV_LIST_MAX_LIMIT)
+        };
+
+        let last_key = match cursor {
+            Some(c) => Some(decode_cursor(c)?),
+            None => None,
+        };
+
+        let take = i64::from(limit) + 1;
+        let rows: Vec<(String,)> = sqlx::query_as(
+            "SELECT key FROM group_kv_entries \
+             WHERE group_id = $1 AND collection = $2 \
+               AND ($3::text IS NULL OR key > $3) \
+             ORDER BY key ASC \
+             LIMIT $4",
+        )
+        .bind(group_id.into_inner())
+        .bind(collection)
+        .bind(last_key.as_deref())
+        .bind(take)
+        .fetch_all(&self.pool)
+        .await?;
+
+        let mut keys: Vec<String> = rows.into_iter().map(|(k,)| k).collect();
+        let next_cursor = if keys.len() > limit as usize {
+            keys.truncate(limit as usize);
+            keys.last().map(|k| encode_cursor(k))
+        } else {
+            None
+        };
+
+        Ok(KvListPage { keys, next_cursor })
+    }
+}
+
+fn encode_cursor(last_key: &str) -> String {
+    URL_SAFE_NO_PAD.encode(last_key.as_bytes())
+}
+
+fn decode_cursor(cursor: &str) -> Result<String, GroupKvRepoError> {
+    let bytes = URL_SAFE_NO_PAD
+        .decode(cursor)
+        .map_err(|_| GroupKvRepoError::InvalidCursor)?;
+    String::from_utf8(bytes).map_err(|_| GroupKvRepoError::InvalidCursor)
+}