//! 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, 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, 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, GroupKvRepoError>; async fn has( &self, group_id: GroupId, collection: &str, key: &str, ) -> Result; async fn list( &self, group_id: GroupId, collection: &str, cursor: Option<&str>, limit: u32, ) -> Result; /// §11.6 quota: total key count across all of the group's shared-KV /// collections. Default `Ok(0)` so non-Postgres impls skip the quota check. async fn count_rows(&self, group_id: GroupId) -> Result { let _ = group_id; Ok(0) } } 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, 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, GroupKvRepoError> { let row: Option<(Option,)> = 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, 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 { 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 { 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 = 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 }) } async fn count_rows(&self, group_id: GroupId) -> Result { let (n,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM group_kv_entries WHERE group_id = $1") .bind(group_id.into_inner()) .fetch_one(&self.pool) .await?; Ok(u64::try_from(n).unwrap_or(0)) } } fn encode_cursor(last_key: &str) -> String { URL_SAFE_NO_PAD.encode(last_key.as_bytes()) } fn decode_cursor(cursor: &str) -> Result { let bytes = URL_SAFE_NO_PAD .decode(cursor) .map_err(|_| GroupKvRepoError::InvalidCursor)?; String::from_utf8(bytes).map_err(|_| GroupKvRepoError::InvalidCursor) }