PiCloud/crates/manager-core/src/group_kv_repo.rs

//! 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>;

    /// §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<u64, GroupKvRepoError> {
        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<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 })
    }

    async fn count_rows(&self, group_id: GroupId) -> Result<u64, GroupKvRepoError> {
        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<String, GroupKvRepoError> {
    let bytes = URL_SAFE_NO_PAD
        .decode(cursor)
        .map_err(|_| GroupKvRepoError::InvalidCursor)?;
    String::from_utf8(bytes).map_err(|_| GroupKvRepoError::InvalidCursor)
}