feat(modules): group-docs schema + kind-generalized registry (§11.6 docs C1)
Data layer for extending shared group collections from KV to docs. KV behavior
unchanged (callers pass kind="kv").
- 0054_group_docs.sql: widen the group_collections kind CHECK to ('kv','docs')
and add the group-keyed group_docs store (clone of docs/0013, keyed by
group_id, CASCADE on group delete) + its (group_id,collection) and data GIN
indexes.
- group_collection_repo: thread a `kind` parameter through list_for_owner,
resolve_owning_group, insert/delete_collection_tx; add list_all_for_owner ->
(name,kind) for the kind-aware diff (D4). Relocate the injectable
GroupCollectionResolver trait + Postgres impl here (now shared by kv+docs;
resolve takes kind) — was in group_kv_service.
- group_docs_repo: a near-clone of docs_repo keyed by group_id; find reuses the
generalized build_find_query.
- docs_repo: generalize build_find_query(table, owner_col, owner_id, …) — both
are compile-time literals (no injection); app docs passes ("docs","app_id"),
group docs ("group_docs","group_id"). row_to_doc/build_find_query are now
pub(crate) for reuse.
Schema snapshot re-blessed (55 migrations).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -850,6 +850,7 @@ impl ApplyService {
|
||||
&mut *tx,
|
||||
owner.as_script_owner(),
|
||||
&ch.key,
|
||||
"kv",
|
||||
)
|
||||
.await
|
||||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||||
@@ -956,6 +957,7 @@ impl ApplyService {
|
||||
&mut *tx,
|
||||
owner.as_script_owner(),
|
||||
&ch.key,
|
||||
"kv",
|
||||
)
|
||||
.await
|
||||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||||
@@ -1800,7 +1802,7 @@ impl ApplyService {
|
||||
owner: ApplyOwner,
|
||||
) -> Result<Vec<CollectionInfo>, ApplyError> {
|
||||
let names =
|
||||
crate::group_collection_repo::list_for_owner(&self.pool, owner.as_script_owner())
|
||||
crate::group_collection_repo::list_for_owner(&self.pool, owner.as_script_owner(), "kv")
|
||||
.await
|
||||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||||
Ok(names
|
||||
@@ -2076,7 +2078,7 @@ impl ApplyService {
|
||||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||||
// Shared group-collection markers declared directly at this node (§11.6).
|
||||
let collection_names =
|
||||
crate::group_collection_repo::list_for_owner(&self.pool, owner.as_script_owner())
|
||||
crate::group_collection_repo::list_for_owner(&self.pool, owner.as_script_owner(), "kv")
|
||||
.await
|
||||
.map_err(|e| ApplyError::Backend(e.to_string()))?;
|
||||
Ok(CurrentState {
|
||||
|
||||
@@ -167,7 +167,7 @@ impl DocsRepo for PostgresDocsRepo {
|
||||
collection: &str,
|
||||
filter: &DocsFilter,
|
||||
) -> Result<Vec<DocRow>, DocsRepoError> {
|
||||
let mut qb = build_find_query(app_id, collection, filter);
|
||||
let mut qb = build_find_query("docs", "app_id", app_id.into_inner(), collection, filter);
|
||||
let rows = qb.build().fetch_all(&self.pool).await?;
|
||||
rows.into_iter().map(row_to_doc).collect()
|
||||
}
|
||||
@@ -281,7 +281,7 @@ impl DocsRepo for PostgresDocsRepo {
|
||||
}
|
||||
}
|
||||
|
||||
fn row_to_doc(row: PgRow) -> Result<DocRow, DocsRepoError> {
|
||||
pub(crate) fn row_to_doc(row: PgRow) -> Result<DocRow, DocsRepoError> {
|
||||
Ok(DocRow {
|
||||
id: row.try_get("id")?,
|
||||
data: row.try_get("data")?,
|
||||
@@ -316,14 +316,22 @@ fn decode_cursor(cursor: &str) -> Result<Uuid, DocsRepoError> {
|
||||
// **No user input ever lands in the SQL text unparameterized.**
|
||||
// ----------------------------------------------------------------------------
|
||||
|
||||
fn build_find_query<'a>(
|
||||
app_id: AppId,
|
||||
/// Build the `find` query for a docs store. `table` and `owner_col` are
|
||||
/// compile-time literals (`"docs"`/`"app_id"` for app docs, `"group_docs"`/
|
||||
/// `"group_id"` for §11.6 group-shared docs) — never user input, so
|
||||
/// interpolating them is injection-safe. `pub(crate)` so the group-docs repo
|
||||
/// reuses this single source for the (security-sensitive) query SQL.
|
||||
pub(crate) fn build_find_query<'a>(
|
||||
table: &'static str,
|
||||
owner_col: &'static str,
|
||||
owner_id: uuid::Uuid,
|
||||
collection: &'a str,
|
||||
filter: &'a DocsFilter,
|
||||
) -> QueryBuilder<'a, Postgres> {
|
||||
let mut qb =
|
||||
QueryBuilder::new("SELECT id, data, created_at, updated_at FROM docs WHERE app_id = ");
|
||||
qb.push_bind(app_id.into_inner());
|
||||
let mut qb = QueryBuilder::new(format!(
|
||||
"SELECT id, data, created_at, updated_at FROM {table} WHERE {owner_col} = "
|
||||
));
|
||||
qb.push_bind(owner_id);
|
||||
qb.push(" AND collection = ");
|
||||
qb.push_bind(collection);
|
||||
|
||||
@@ -448,7 +456,13 @@ mod sql_shape_tests {
|
||||
|
||||
fn sql_for(filter_json: serde_json::Value) -> String {
|
||||
let filter = parse_filter(&filter_json).unwrap();
|
||||
let qb = build_find_query(AppId::new(), "users", &filter);
|
||||
let qb = build_find_query(
|
||||
"docs",
|
||||
"app_id",
|
||||
AppId::new().into_inner(),
|
||||
"users",
|
||||
&filter,
|
||||
);
|
||||
qb.sql().to_string()
|
||||
}
|
||||
|
||||
|
||||
@@ -1,30 +1,31 @@
|
||||
//! Group-collection markers (§11.6) — the `group_collections` table (0052).
|
||||
//!
|
||||
//! A marker `(owner, name, kind)` declares that a collection name is
|
||||
//! group-shared (the data lives in a per-kind store; `group_kv_entries` for
|
||||
//! `kind='kv'`). This module holds the read + transactional-write helpers and
|
||||
//! the runtime resolver; it mirrors the var/secret/extension-point tx-function
|
||||
//! style (free functions over a `&PgPool` / `&mut Transaction`), keyed by the
|
||||
//! shared [`ScriptOwner`].
|
||||
//! group-shared. `kind` selects the storage backend: `'kv'` →
|
||||
//! `group_kv_entries` (0053), `'docs'` → `group_docs` (0054). This module holds
|
||||
//! the read + transactional-write helpers, the runtime resolver, and the
|
||||
//! injectable [`GroupCollectionResolver`] trait the per-kind services consume.
|
||||
//!
|
||||
//! The load-bearing function is [`resolve_owning_group`]: it walks the reading
|
||||
//! app's ancestor chain ([`CHAIN_LEVELS_CTE`]) for the nearest group declaring
|
||||
//! the collection. That walk **is** the isolation boundary — a foreign app's
|
||||
//! chain never contains the owning group, so the name does not resolve.
|
||||
//! the collection of the requested `kind`. That walk **is** the isolation
|
||||
//! boundary — a foreign app's chain never contains the owning group, so the
|
||||
//! name does not resolve.
|
||||
|
||||
use async_trait::async_trait;
|
||||
use picloud_shared::{AppId, GroupId, ScriptOwner};
|
||||
use sqlx::{PgPool, Postgres, Transaction};
|
||||
|
||||
use crate::config_resolver::CHAIN_LEVELS_CTE;
|
||||
|
||||
/// The only storage kind that ships in the MVP. The `kind` column generalizes
|
||||
/// to docs/files/topics/queue later; every query pins this value for now.
|
||||
const KIND_KV: &str = "kv";
|
||||
|
||||
/// List the collection names declared **directly at** `owner` (not inherited),
|
||||
/// case-insensitively sorted. Used by `load_current` (apply diff) and the
|
||||
/// read-only `collections ls`. MVP is `kind='kv'` only.
|
||||
pub async fn list_for_owner(pool: &PgPool, owner: ScriptOwner) -> Result<Vec<String>, sqlx::Error> {
|
||||
/// List the collection names of `kind` declared **directly at** `owner` (not
|
||||
/// inherited), case-insensitively sorted. Used by `load_current` (apply diff)
|
||||
/// and the read-only `collections ls`.
|
||||
pub async fn list_for_owner(
|
||||
pool: &PgPool,
|
||||
owner: ScriptOwner,
|
||||
kind: &str,
|
||||
) -> Result<Vec<String>, sqlx::Error> {
|
||||
let rows: Vec<(String,)> = match owner {
|
||||
ScriptOwner::App(a) => {
|
||||
sqlx::query_as(
|
||||
@@ -32,7 +33,7 @@ pub async fn list_for_owner(pool: &PgPool, owner: ScriptOwner) -> Result<Vec<Str
|
||||
WHERE app_id = $1 AND kind = $2 ORDER BY LOWER(name)",
|
||||
)
|
||||
.bind(a.into_inner())
|
||||
.bind(KIND_KV)
|
||||
.bind(kind)
|
||||
.fetch_all(pool)
|
||||
.await?
|
||||
}
|
||||
@@ -42,7 +43,7 @@ pub async fn list_for_owner(pool: &PgPool, owner: ScriptOwner) -> Result<Vec<Str
|
||||
WHERE group_id = $1 AND kind = $2 ORDER BY LOWER(name)",
|
||||
)
|
||||
.bind(g.into_inner())
|
||||
.bind(KIND_KV)
|
||||
.bind(kind)
|
||||
.fetch_all(pool)
|
||||
.await?
|
||||
}
|
||||
@@ -50,11 +51,40 @@ pub async fn list_for_owner(pool: &PgPool, owner: ScriptOwner) -> Result<Vec<Str
|
||||
Ok(rows.into_iter().map(|(n,)| n).collect())
|
||||
}
|
||||
|
||||
/// Resolve the group that OWNS the shared collection `name` for a reading app:
|
||||
/// the nearest ancestor group on the app's chain that declares it (`kind='kv'`).
|
||||
/// Returns `None` when no group on the chain shares that name — the structural
|
||||
/// "not shared with you" boundary. Nearest-wins (CoW shadowing) is enforced by
|
||||
/// `ORDER BY depth ASC LIMIT 1` and is security-relevant.
|
||||
/// List **all** shared-collection declarations at `owner` as `(name, kind)`
|
||||
/// pairs, sorted. Used by the kind-aware apply diff and `collections ls`.
|
||||
pub async fn list_all_for_owner(
|
||||
pool: &PgPool,
|
||||
owner: ScriptOwner,
|
||||
) -> Result<Vec<(String, String)>, sqlx::Error> {
|
||||
let rows: Vec<(String, String)> = match owner {
|
||||
ScriptOwner::App(a) => {
|
||||
sqlx::query_as(
|
||||
"SELECT name, kind FROM group_collections \
|
||||
WHERE app_id = $1 ORDER BY kind, LOWER(name)",
|
||||
)
|
||||
.bind(a.into_inner())
|
||||
.fetch_all(pool)
|
||||
.await?
|
||||
}
|
||||
ScriptOwner::Group(g) => {
|
||||
sqlx::query_as(
|
||||
"SELECT name, kind FROM group_collections \
|
||||
WHERE group_id = $1 ORDER BY kind, LOWER(name)",
|
||||
)
|
||||
.bind(g.into_inner())
|
||||
.fetch_all(pool)
|
||||
.await?
|
||||
}
|
||||
};
|
||||
Ok(rows)
|
||||
}
|
||||
|
||||
/// Resolve the group that OWNS the shared collection `(name, kind)` for a
|
||||
/// reading app: the nearest ancestor group on the app's chain that declares it.
|
||||
/// Returns `None` when no group on the chain shares that `(name, kind)` — the
|
||||
/// structural "not shared with you" boundary. Nearest-wins (CoW shadowing) is
|
||||
/// enforced by `ORDER BY depth ASC LIMIT 1` and is security-relevant.
|
||||
///
|
||||
/// The join is on `group_owner` only: an app-declared marker (the degenerate
|
||||
/// case) never makes a collection visible to *other* apps — sharing is a group
|
||||
@@ -63,6 +93,7 @@ pub async fn resolve_owning_group(
|
||||
pool: &PgPool,
|
||||
app_id: AppId,
|
||||
name: &str,
|
||||
kind: &str,
|
||||
) -> Result<Option<GroupId>, sqlx::Error> {
|
||||
let row: Option<(uuid::Uuid,)> = sqlx::query_as(&format!(
|
||||
"{CHAIN_LEVELS_CTE} \
|
||||
@@ -73,19 +104,20 @@ pub async fn resolve_owning_group(
|
||||
))
|
||||
.bind(app_id.into_inner())
|
||||
.bind(name)
|
||||
.bind(KIND_KV)
|
||||
.bind(kind)
|
||||
.fetch_optional(pool)
|
||||
.await?;
|
||||
Ok(row.map(|(id,)| GroupId::from(id)))
|
||||
}
|
||||
|
||||
/// Insert a collection marker at `owner`, in the apply transaction. Idempotent:
|
||||
/// a re-apply of an already-declared name is a no-op (`ON CONFLICT DO NOTHING`),
|
||||
/// so the marker survives without a spurious version bump.
|
||||
/// Insert a `(name, kind)` marker at `owner`, in the apply transaction.
|
||||
/// Idempotent: a re-apply of an already-declared marker is a no-op
|
||||
/// (`ON CONFLICT DO NOTHING`), so it survives without a spurious version bump.
|
||||
pub async fn insert_collection_tx(
|
||||
tx: &mut Transaction<'_, Postgres>,
|
||||
owner: ScriptOwner,
|
||||
name: &str,
|
||||
kind: &str,
|
||||
) -> Result<(), sqlx::Error> {
|
||||
match owner {
|
||||
ScriptOwner::App(a) => {
|
||||
@@ -95,7 +127,7 @@ pub async fn insert_collection_tx(
|
||||
)
|
||||
.bind(a.into_inner())
|
||||
.bind(name)
|
||||
.bind(KIND_KV)
|
||||
.bind(kind)
|
||||
.execute(&mut **tx)
|
||||
.await?;
|
||||
}
|
||||
@@ -106,7 +138,7 @@ pub async fn insert_collection_tx(
|
||||
)
|
||||
.bind(g.into_inner())
|
||||
.bind(name)
|
||||
.bind(KIND_KV)
|
||||
.bind(kind)
|
||||
.execute(&mut **tx)
|
||||
.await?;
|
||||
}
|
||||
@@ -114,14 +146,15 @@ pub async fn insert_collection_tx(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Delete a collection marker at `owner` (case-insensitive), in the apply
|
||||
/// transaction. Used by `--prune` when the manifest stops declaring a name.
|
||||
/// The `group_kv_entries` data is NOT dropped here — pruning a marker hides the
|
||||
/// store but leaves the data until the owning group is deleted.
|
||||
/// Delete a `(name, kind)` marker at `owner` (case-insensitive), in the apply
|
||||
/// transaction. Used by `--prune`. The storage data is NOT dropped here —
|
||||
/// pruning a marker hides the store but leaves the data until the owning group
|
||||
/// is deleted.
|
||||
pub async fn delete_collection_tx(
|
||||
tx: &mut Transaction<'_, Postgres>,
|
||||
owner: ScriptOwner,
|
||||
name: &str,
|
||||
kind: &str,
|
||||
) -> Result<(), sqlx::Error> {
|
||||
match owner {
|
||||
ScriptOwner::App(a) => {
|
||||
@@ -131,7 +164,7 @@ pub async fn delete_collection_tx(
|
||||
)
|
||||
.bind(a.into_inner())
|
||||
.bind(name)
|
||||
.bind(KIND_KV)
|
||||
.bind(kind)
|
||||
.execute(&mut **tx)
|
||||
.await?;
|
||||
}
|
||||
@@ -142,10 +175,48 @@ pub async fn delete_collection_tx(
|
||||
)
|
||||
.bind(g.into_inner())
|
||||
.bind(name)
|
||||
.bind(KIND_KV)
|
||||
.bind(kind)
|
||||
.execute(&mut **tx)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Resolves a shared-collection name+kind to its owning group for a calling app
|
||||
/// (nearest ancestor group on the app's chain that declares it). Behind a trait
|
||||
/// so the per-kind services (`GroupKvServiceImpl`, `GroupDocsServiceImpl`) can
|
||||
/// inject a fake in unit tests without Postgres.
|
||||
#[async_trait]
|
||||
pub trait GroupCollectionResolver: Send + Sync {
|
||||
async fn resolve_owning_group(
|
||||
&self,
|
||||
app_id: AppId,
|
||||
name: &str,
|
||||
kind: &str,
|
||||
) -> Result<Option<GroupId>, sqlx::Error>;
|
||||
}
|
||||
|
||||
/// Postgres-backed resolver — delegates to [`resolve_owning_group`].
|
||||
pub struct PostgresGroupCollectionResolver {
|
||||
pool: PgPool,
|
||||
}
|
||||
|
||||
impl PostgresGroupCollectionResolver {
|
||||
#[must_use]
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl GroupCollectionResolver for PostgresGroupCollectionResolver {
|
||||
async fn resolve_owning_group(
|
||||
&self,
|
||||
app_id: AppId,
|
||||
name: &str,
|
||||
kind: &str,
|
||||
) -> Result<Option<GroupId>, sqlx::Error> {
|
||||
resolve_owning_group(&self.pool, app_id, name, kind).await
|
||||
}
|
||||
}
|
||||
|
||||
284
crates/manager-core/src/group_docs_repo.rs
Normal file
284
crates/manager-core/src/group_docs_repo.rs
Normal file
@@ -0,0 +1,284 @@
|
||||
//! Low-level Postgres CRUD over `group_docs` (§11.6). A near-clone of
|
||||
//! [`crate::docs_repo`] keyed by the owning `group_id` instead of `app_id`.
|
||||
//! The `find` query is built by the shared [`crate::docs_repo::build_find_query`]
|
||||
//! (parameterized on table + owner column), so the security-sensitive filter
|
||||
//! SQL has a single source. Authorization, group resolution, value validation,
|
||||
//! and event policy live one layer up in `GroupDocsServiceImpl`.
|
||||
|
||||
use async_trait::async_trait;
|
||||
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
|
||||
use base64::Engine as _;
|
||||
use chrono::{DateTime, Utc};
|
||||
use picloud_shared::{DocId, DocRow, DocsListPage, GroupId};
|
||||
use serde_json::Value;
|
||||
use sqlx::PgPool;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::docs_filter::DocsFilter;
|
||||
use crate::docs_repo::{build_find_query, row_to_doc, DocsRepoError};
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum GroupDocsRepoError {
|
||||
#[error("database error: {0}")]
|
||||
Db(#[from] sqlx::Error),
|
||||
|
||||
#[error("invalid pagination cursor")]
|
||||
InvalidCursor,
|
||||
}
|
||||
|
||||
impl From<DocsRepoError> for GroupDocsRepoError {
|
||||
fn from(e: DocsRepoError) -> Self {
|
||||
match e {
|
||||
DocsRepoError::Db(e) => Self::Db(e),
|
||||
DocsRepoError::InvalidCursor => Self::InvalidCursor,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
pub trait GroupDocsRepo: Send + Sync {
|
||||
async fn create(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
data: Value,
|
||||
) -> Result<DocRow, GroupDocsRepoError>;
|
||||
|
||||
async fn get(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
id: DocId,
|
||||
) -> Result<Option<DocRow>, GroupDocsRepoError>;
|
||||
|
||||
async fn find(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
filter: &DocsFilter,
|
||||
) -> Result<Vec<DocRow>, GroupDocsRepoError>;
|
||||
|
||||
/// Returns the previous data (for the would-be event), `None` if missing.
|
||||
async fn update(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
id: DocId,
|
||||
data: Value,
|
||||
) -> Result<Option<Value>, GroupDocsRepoError>;
|
||||
|
||||
async fn delete(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
id: DocId,
|
||||
) -> Result<Option<Value>, GroupDocsRepoError>;
|
||||
|
||||
async fn list(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
cursor: Option<&str>,
|
||||
limit: u32,
|
||||
) -> Result<DocsListPage, GroupDocsRepoError>;
|
||||
}
|
||||
|
||||
pub struct PostgresGroupDocsRepo {
|
||||
pool: PgPool,
|
||||
}
|
||||
|
||||
impl PostgresGroupDocsRepo {
|
||||
#[must_use]
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
}
|
||||
|
||||
const DOCS_LIST_MAX_LIMIT: u32 = 1_000;
|
||||
const DOCS_LIST_DEFAULT_LIMIT: u32 = 100;
|
||||
|
||||
#[async_trait]
|
||||
impl GroupDocsRepo for PostgresGroupDocsRepo {
|
||||
async fn create(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
data: Value,
|
||||
) -> Result<DocRow, GroupDocsRepoError> {
|
||||
let id = Uuid::new_v4();
|
||||
let row: (DateTime<Utc>, DateTime<Utc>) = sqlx::query_as(
|
||||
"INSERT INTO group_docs (group_id, collection, id, data) \
|
||||
VALUES ($1, $2, $3, $4) \
|
||||
RETURNING created_at, updated_at",
|
||||
)
|
||||
.bind(group_id.into_inner())
|
||||
.bind(collection)
|
||||
.bind(id)
|
||||
.bind(&data)
|
||||
.fetch_one(&self.pool)
|
||||
.await?;
|
||||
Ok(DocRow {
|
||||
id,
|
||||
data,
|
||||
created_at: row.0,
|
||||
updated_at: row.1,
|
||||
})
|
||||
}
|
||||
|
||||
async fn get(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
id: DocId,
|
||||
) -> Result<Option<DocRow>, GroupDocsRepoError> {
|
||||
let row: Option<(Value, DateTime<Utc>, DateTime<Utc>)> = sqlx::query_as(
|
||||
"SELECT data, created_at, updated_at FROM group_docs \
|
||||
WHERE group_id = $1 AND collection = $2 AND id = $3",
|
||||
)
|
||||
.bind(group_id.into_inner())
|
||||
.bind(collection)
|
||||
.bind(id)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?;
|
||||
Ok(row.map(|(data, created_at, updated_at)| DocRow {
|
||||
id,
|
||||
data,
|
||||
created_at,
|
||||
updated_at,
|
||||
}))
|
||||
}
|
||||
|
||||
async fn find(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
filter: &DocsFilter,
|
||||
) -> Result<Vec<DocRow>, GroupDocsRepoError> {
|
||||
let mut qb = build_find_query(
|
||||
"group_docs",
|
||||
"group_id",
|
||||
group_id.into_inner(),
|
||||
collection,
|
||||
filter,
|
||||
);
|
||||
let rows = qb.build().fetch_all(&self.pool).await?;
|
||||
rows.into_iter()
|
||||
.map(row_to_doc)
|
||||
.collect::<Result<Vec<_>, _>>()
|
||||
.map_err(Into::into)
|
||||
}
|
||||
|
||||
async fn update(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
id: DocId,
|
||||
data: Value,
|
||||
) -> Result<Option<Value>, GroupDocsRepoError> {
|
||||
let row: Option<(Option<Value>,)> = sqlx::query_as(
|
||||
"WITH prev AS ( \
|
||||
SELECT data FROM group_docs \
|
||||
WHERE group_id = $1 AND collection = $2 AND id = $3 \
|
||||
), \
|
||||
updated AS ( \
|
||||
UPDATE group_docs SET data = $4, updated_at = NOW() \
|
||||
WHERE group_id = $1 AND collection = $2 AND id = $3 \
|
||||
RETURNING 1 \
|
||||
) \
|
||||
SELECT (SELECT data FROM prev) FROM updated",
|
||||
)
|
||||
.bind(group_id.into_inner())
|
||||
.bind(collection)
|
||||
.bind(id)
|
||||
.bind(&data)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?;
|
||||
Ok(row.and_then(|(v,)| v))
|
||||
}
|
||||
|
||||
async fn delete(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
id: DocId,
|
||||
) -> Result<Option<Value>, GroupDocsRepoError> {
|
||||
let row: Option<(Value,)> = sqlx::query_as(
|
||||
"DELETE FROM group_docs \
|
||||
WHERE group_id = $1 AND collection = $2 AND id = $3 \
|
||||
RETURNING data",
|
||||
)
|
||||
.bind(group_id.into_inner())
|
||||
.bind(collection)
|
||||
.bind(id)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?;
|
||||
Ok(row.map(|(v,)| v))
|
||||
}
|
||||
|
||||
async fn list(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
cursor: Option<&str>,
|
||||
limit: u32,
|
||||
) -> Result<DocsListPage, GroupDocsRepoError> {
|
||||
let limit = if limit == 0 {
|
||||
DOCS_LIST_DEFAULT_LIMIT
|
||||
} else {
|
||||
limit.min(DOCS_LIST_MAX_LIMIT)
|
||||
};
|
||||
|
||||
let last_id = match cursor {
|
||||
Some(c) => Some(decode_cursor(c)?),
|
||||
None => None,
|
||||
};
|
||||
|
||||
let take = i64::from(limit) + 1;
|
||||
let rows: Vec<(Uuid, Value, DateTime<Utc>, DateTime<Utc>)> = sqlx::query_as(
|
||||
"SELECT id, data, created_at, updated_at FROM group_docs \
|
||||
WHERE group_id = $1 AND collection = $2 \
|
||||
AND ($3::uuid IS NULL OR id > $3) \
|
||||
ORDER BY id ASC \
|
||||
LIMIT $4",
|
||||
)
|
||||
.bind(group_id.into_inner())
|
||||
.bind(collection)
|
||||
.bind(last_id)
|
||||
.bind(take)
|
||||
.fetch_all(&self.pool)
|
||||
.await?;
|
||||
|
||||
let mut docs: Vec<DocRow> = rows
|
||||
.into_iter()
|
||||
.map(|(id, data, created_at, updated_at)| DocRow {
|
||||
id,
|
||||
data,
|
||||
created_at,
|
||||
updated_at,
|
||||
})
|
||||
.collect();
|
||||
let next_cursor = if docs.len() > limit as usize {
|
||||
docs.truncate(limit as usize);
|
||||
docs.last().map(|d| encode_cursor(&d.id))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
Ok(DocsListPage { docs, next_cursor })
|
||||
}
|
||||
}
|
||||
|
||||
fn encode_cursor(last_id: &Uuid) -> String {
|
||||
URL_SAFE_NO_PAD.encode(last_id.as_bytes())
|
||||
}
|
||||
|
||||
fn decode_cursor(cursor: &str) -> Result<Uuid, GroupDocsRepoError> {
|
||||
let bytes = URL_SAFE_NO_PAD
|
||||
.decode(cursor)
|
||||
.map_err(|_| GroupDocsRepoError::InvalidCursor)?;
|
||||
let arr: [u8; 16] = bytes
|
||||
.as_slice()
|
||||
.try_into()
|
||||
.map_err(|_| GroupDocsRepoError::InvalidCursor)?;
|
||||
Ok(Uuid::from_bytes(arr))
|
||||
}
|
||||
@@ -22,48 +22,15 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use picloud_shared::{AppId, GroupId, GroupKvError, GroupKvService, KvListPage, SdkCallCx};
|
||||
use sqlx::PgPool;
|
||||
use picloud_shared::{GroupId, GroupKvError, GroupKvService, KvListPage, SdkCallCx};
|
||||
|
||||
use crate::authz::{self, AuthzRepo, Capability};
|
||||
use crate::group_collection_repo;
|
||||
use crate::group_collection_repo::GroupCollectionResolver;
|
||||
use crate::group_kv_repo::{GroupKvRepo, GroupKvRepoError};
|
||||
use crate::kv_service::kv_max_value_bytes_from_env;
|
||||
|
||||
/// Resolves a shared-collection name to its owning group for a calling app
|
||||
/// (nearest ancestor group on the app's chain that declares it). Behind a
|
||||
/// trait so unit tests can inject a fake without Postgres.
|
||||
#[async_trait]
|
||||
pub trait GroupCollectionResolver: Send + Sync {
|
||||
async fn resolve_owning_group(
|
||||
&self,
|
||||
app_id: AppId,
|
||||
name: &str,
|
||||
) -> Result<Option<GroupId>, sqlx::Error>;
|
||||
}
|
||||
|
||||
/// Postgres-backed resolver — delegates to [`group_collection_repo`].
|
||||
pub struct PostgresGroupCollectionResolver {
|
||||
pool: PgPool,
|
||||
}
|
||||
|
||||
impl PostgresGroupCollectionResolver {
|
||||
#[must_use]
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl GroupCollectionResolver for PostgresGroupCollectionResolver {
|
||||
async fn resolve_owning_group(
|
||||
&self,
|
||||
app_id: AppId,
|
||||
name: &str,
|
||||
) -> Result<Option<GroupId>, sqlx::Error> {
|
||||
group_collection_repo::resolve_owning_group(&self.pool, app_id, name).await
|
||||
}
|
||||
}
|
||||
/// The registry `kind` this service resolves.
|
||||
const KIND_KV: &str = "kv";
|
||||
|
||||
pub struct GroupKvServiceImpl {
|
||||
repo: Arc<dyn GroupKvRepo>,
|
||||
@@ -109,7 +76,7 @@ impl GroupKvServiceImpl {
|
||||
return Err(GroupKvError::InvalidCollection);
|
||||
}
|
||||
self.resolver
|
||||
.resolve_owning_group(cx.app_id, collection)
|
||||
.resolve_owning_group(cx.app_id, collection, KIND_KV)
|
||||
.await
|
||||
.map_err(|e| GroupKvError::Backend(e.to_string()))?
|
||||
.ok_or_else(|| GroupKvError::CollectionNotShared(collection.to_string()))
|
||||
@@ -220,7 +187,8 @@ mod tests {
|
||||
use super::*;
|
||||
use crate::authz::{AuthzError, AuthzRepo};
|
||||
use picloud_shared::{
|
||||
AdminUserId, AppRole, ExecutionId, InstanceRole, Principal, RequestId, ScriptId, UserId,
|
||||
AdminUserId, AppId, AppRole, ExecutionId, InstanceRole, Principal, RequestId, ScriptId,
|
||||
UserId,
|
||||
};
|
||||
use std::collections::{BTreeMap, HashMap};
|
||||
use tokio::sync::Mutex;
|
||||
@@ -321,6 +289,7 @@ mod tests {
|
||||
&self,
|
||||
app_id: AppId,
|
||||
name: &str,
|
||||
_kind: &str,
|
||||
) -> Result<Option<GroupId>, sqlx::Error> {
|
||||
Ok(self.map.get(&(app_id, name.to_lowercase())).copied())
|
||||
}
|
||||
|
||||
@@ -51,6 +51,7 @@ pub mod files_service;
|
||||
pub mod files_sweep;
|
||||
pub mod gc;
|
||||
pub mod group_collection_repo;
|
||||
pub mod group_docs_repo;
|
||||
pub mod group_kv_repo;
|
||||
pub mod group_kv_service;
|
||||
pub mod group_members_repo;
|
||||
@@ -179,10 +180,10 @@ pub use files_repo::{FilesConfig, FilesRepo, FilesRepoError, FsFilesRepo};
|
||||
pub use files_service::FilesServiceImpl;
|
||||
pub use files_sweep::{spawn_files_orphan_sweep, sweep_orphan_tmp_files, SweepStats};
|
||||
pub use gc::{spawn_abandoned_gc, spawn_app_user_token_gc, spawn_dead_letter_gc};
|
||||
pub use group_collection_repo::{GroupCollectionResolver, PostgresGroupCollectionResolver};
|
||||
pub use group_docs_repo::{GroupDocsRepo, GroupDocsRepoError, PostgresGroupDocsRepo};
|
||||
pub use group_kv_repo::{GroupKvRepo, GroupKvRepoError, PostgresGroupKvRepo};
|
||||
pub use group_kv_service::{
|
||||
GroupCollectionResolver, GroupKvServiceImpl, PostgresGroupCollectionResolver,
|
||||
};
|
||||
pub use group_kv_service::GroupKvServiceImpl;
|
||||
pub use group_members_repo::{
|
||||
GroupMembersRepository, GroupMembersRepositoryError, GroupMembershipDetail, GroupMembershipRow,
|
||||
PostgresGroupMembersRepository,
|
||||
|
||||
Reference in New Issue
Block a user