feat(modules): GroupDocsService + docs::shared_collection handle (§11.6 docs C3)
The runtime for shared group docs collections, mirroring the group-KV vertical with the docs surface (filter parsing, JSON-object validation, value-size cap). - shared: GroupDocsService trait + GroupDocsError (+ CollectionNotShared) + NoopGroupDocsService; group_docs field on Services + with_group_docs(). - manager-core: GroupDocsServiceImpl — owning_group resolves kind='docs' from cx.app_id's chain (CollectionNotShared off-chain); reads-open (script_gate GroupDocsRead) / writes-authed (script_gate_require_principal GroupDocsWrite); reuses parse_filter + docs value-size cap; no events. Unit tests cover off-chain CollectionNotShared, anon-read-OK/anon-write-Forbidden, non-object data rejected. - executor-core: GroupDocsHandle + docs::shared_collection constructor + the create/get/find/find_one/update/delete/list methods (dispatch by receiver type), reusing doc_to_map/parse_doc_id. - picloud: wire PostgresGroupDocsRepo + GroupDocsServiceImpl. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
483
crates/manager-core/src/group_docs_service.rs
Normal file
483
crates/manager-core/src/group_docs_service.rs
Normal file
@@ -0,0 +1,483 @@
|
||||
//! `GroupDocsServiceImpl` — wires `GroupDocsRepo` + the group-collection
|
||||
//! registry underneath the `picloud_shared::GroupDocsService` trait that scripts
|
||||
//! reach via the `docs::shared_collection("name")` Rhai handle (§11.6).
|
||||
//!
|
||||
//! Combines the group-KV service pattern (owner resolution from `cx.app_id`,
|
||||
//! reads-open / writes-authed authz) with the docs surface (filter parsing,
|
||||
//! JSON-object validation, value-size cap). No event emission in the MVP (the
|
||||
//! "group trigger has no app to watch" deferral).
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use picloud_shared::{
|
||||
DocId, DocRow, DocsListPage, GroupDocsError, GroupDocsService, GroupId, SdkCallCx,
|
||||
};
|
||||
|
||||
use crate::authz::{self, AuthzRepo, Capability};
|
||||
use crate::docs_filter::{parse_filter, FilterParseError};
|
||||
use crate::docs_service::docs_max_value_bytes_from_env;
|
||||
use crate::group_collection_repo::GroupCollectionResolver;
|
||||
use crate::group_docs_repo::{GroupDocsRepo, GroupDocsRepoError};
|
||||
|
||||
/// The registry `kind` this service resolves.
|
||||
const KIND_DOCS: &str = "docs";
|
||||
|
||||
pub struct GroupDocsServiceImpl {
|
||||
repo: Arc<dyn GroupDocsRepo>,
|
||||
resolver: Arc<dyn GroupCollectionResolver>,
|
||||
authz: Arc<dyn AuthzRepo>,
|
||||
max_value_bytes: usize,
|
||||
}
|
||||
|
||||
impl GroupDocsServiceImpl {
|
||||
#[must_use]
|
||||
pub fn new(
|
||||
repo: Arc<dyn GroupDocsRepo>,
|
||||
resolver: Arc<dyn GroupCollectionResolver>,
|
||||
authz: Arc<dyn AuthzRepo>,
|
||||
) -> Self {
|
||||
Self::with_max_value_bytes(repo, resolver, authz, docs_max_value_bytes_from_env())
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub fn with_max_value_bytes(
|
||||
repo: Arc<dyn GroupDocsRepo>,
|
||||
resolver: Arc<dyn GroupCollectionResolver>,
|
||||
authz: Arc<dyn AuthzRepo>,
|
||||
max_value_bytes: usize,
|
||||
) -> Self {
|
||||
Self {
|
||||
repo,
|
||||
resolver,
|
||||
authz,
|
||||
max_value_bytes,
|
||||
}
|
||||
}
|
||||
|
||||
/// The structural boundary: resolve the collection to its owning group
|
||||
/// (kind=`docs`) on the calling app's chain. `CollectionNotShared` when no
|
||||
/// ancestor group declares it.
|
||||
async fn owning_group(
|
||||
&self,
|
||||
cx: &SdkCallCx,
|
||||
collection: &str,
|
||||
) -> Result<GroupId, GroupDocsError> {
|
||||
if collection.is_empty() {
|
||||
return Err(GroupDocsError::InvalidCollection);
|
||||
}
|
||||
self.resolver
|
||||
.resolve_owning_group(cx.app_id, collection, KIND_DOCS)
|
||||
.await
|
||||
.map_err(|e| GroupDocsError::Backend(e.to_string()))?
|
||||
.ok_or_else(|| GroupDocsError::CollectionNotShared(collection.to_string()))
|
||||
}
|
||||
|
||||
fn check_data_size(&self, data: &serde_json::Value) -> Result<(), GroupDocsError> {
|
||||
let encoded_len = serde_json::to_vec(data)
|
||||
.map(|v| v.len())
|
||||
.map_err(|e| GroupDocsError::Backend(format!("encode doc data: {e}")))?;
|
||||
if encoded_len > self.max_value_bytes {
|
||||
return Err(GroupDocsError::ValueTooLarge {
|
||||
limit: self.max_value_bytes,
|
||||
actual: encoded_len,
|
||||
});
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn check_read(&self, cx: &SdkCallCx, group_id: GroupId) -> Result<(), GroupDocsError> {
|
||||
authz::script_gate(
|
||||
&*self.authz,
|
||||
cx,
|
||||
Capability::GroupDocsRead(group_id),
|
||||
|| GroupDocsError::Forbidden,
|
||||
GroupDocsError::Backend,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn check_write(&self, cx: &SdkCallCx, group_id: GroupId) -> Result<(), GroupDocsError> {
|
||||
// Fails closed for an anonymous principal — shared mutation always needs
|
||||
// an authenticated editor+ on the owning group.
|
||||
authz::script_gate_require_principal(
|
||||
&*self.authz,
|
||||
cx,
|
||||
Capability::GroupDocsWrite(group_id),
|
||||
|| GroupDocsError::Forbidden,
|
||||
GroupDocsError::Backend,
|
||||
)
|
||||
.await
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_data(data: &serde_json::Value) -> Result<(), GroupDocsError> {
|
||||
if !data.is_object() {
|
||||
return Err(GroupDocsError::InvalidData);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
impl From<GroupDocsRepoError> for GroupDocsError {
|
||||
fn from(e: GroupDocsRepoError) -> Self {
|
||||
Self::Backend(e.to_string())
|
||||
}
|
||||
}
|
||||
|
||||
impl From<FilterParseError> for GroupDocsError {
|
||||
fn from(e: FilterParseError) -> Self {
|
||||
match e {
|
||||
FilterParseError::InvalidFilter(s) => Self::InvalidFilter(s),
|
||||
FilterParseError::UnsupportedOperator(s) => Self::UnsupportedOperator(s),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl GroupDocsService for GroupDocsServiceImpl {
|
||||
async fn create(
|
||||
&self,
|
||||
cx: &SdkCallCx,
|
||||
collection: &str,
|
||||
data: serde_json::Value,
|
||||
) -> Result<DocId, GroupDocsError> {
|
||||
let group_id = self.owning_group(cx, collection).await?;
|
||||
validate_data(&data)?;
|
||||
self.check_data_size(&data)?;
|
||||
self.check_write(cx, group_id).await?;
|
||||
Ok(self.repo.create(group_id, collection, data).await?.id)
|
||||
}
|
||||
|
||||
async fn get(
|
||||
&self,
|
||||
cx: &SdkCallCx,
|
||||
collection: &str,
|
||||
id: DocId,
|
||||
) -> Result<Option<DocRow>, GroupDocsError> {
|
||||
let group_id = self.owning_group(cx, collection).await?;
|
||||
self.check_read(cx, group_id).await?;
|
||||
Ok(self.repo.get(group_id, collection, id).await?)
|
||||
}
|
||||
|
||||
async fn find(
|
||||
&self,
|
||||
cx: &SdkCallCx,
|
||||
collection: &str,
|
||||
filter: serde_json::Value,
|
||||
) -> Result<Vec<DocRow>, GroupDocsError> {
|
||||
let group_id = self.owning_group(cx, collection).await?;
|
||||
self.check_read(cx, group_id).await?;
|
||||
let parsed = parse_filter(&filter)?;
|
||||
Ok(self.repo.find(group_id, collection, &parsed).await?)
|
||||
}
|
||||
|
||||
async fn find_one(
|
||||
&self,
|
||||
cx: &SdkCallCx,
|
||||
collection: &str,
|
||||
filter: serde_json::Value,
|
||||
) -> Result<Option<DocRow>, GroupDocsError> {
|
||||
let group_id = self.owning_group(cx, collection).await?;
|
||||
self.check_read(cx, group_id).await?;
|
||||
let mut parsed = parse_filter(&filter)?;
|
||||
if parsed.limit.is_none() {
|
||||
parsed.limit = Some(1);
|
||||
}
|
||||
let rows = self.repo.find(group_id, collection, &parsed).await?;
|
||||
Ok(rows.into_iter().next())
|
||||
}
|
||||
|
||||
async fn update(
|
||||
&self,
|
||||
cx: &SdkCallCx,
|
||||
collection: &str,
|
||||
id: DocId,
|
||||
data: serde_json::Value,
|
||||
) -> Result<(), GroupDocsError> {
|
||||
let group_id = self.owning_group(cx, collection).await?;
|
||||
validate_data(&data)?;
|
||||
self.check_data_size(&data)?;
|
||||
self.check_write(cx, group_id).await?;
|
||||
match self.repo.update(group_id, collection, id, data).await? {
|
||||
Some(_) => Ok(()),
|
||||
None => Err(GroupDocsError::NotFound),
|
||||
}
|
||||
}
|
||||
|
||||
async fn delete(
|
||||
&self,
|
||||
cx: &SdkCallCx,
|
||||
collection: &str,
|
||||
id: DocId,
|
||||
) -> Result<bool, GroupDocsError> {
|
||||
let group_id = self.owning_group(cx, collection).await?;
|
||||
self.check_write(cx, group_id).await?;
|
||||
Ok(self.repo.delete(group_id, collection, id).await?.is_some())
|
||||
}
|
||||
|
||||
async fn list(
|
||||
&self,
|
||||
cx: &SdkCallCx,
|
||||
collection: &str,
|
||||
cursor: Option<&str>,
|
||||
limit: u32,
|
||||
) -> Result<DocsListPage, GroupDocsError> {
|
||||
let group_id = self.owning_group(cx, collection).await?;
|
||||
self.check_read(cx, group_id).await?;
|
||||
Ok(self.repo.list(group_id, collection, cursor, limit).await?)
|
||||
}
|
||||
}
|
||||
|
||||
// ----------------------------------------------------------------------------
|
||||
// Tests — in-memory repo + a fake resolver so unit tests don't need Postgres.
|
||||
// ----------------------------------------------------------------------------
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::authz::{AuthzError, AuthzRepo};
|
||||
use crate::docs_filter::DocsFilter;
|
||||
use chrono::Utc;
|
||||
use picloud_shared::{
|
||||
AdminUserId, AppId, AppRole, ExecutionId, InstanceRole, Principal, RequestId, ScriptId,
|
||||
UserId,
|
||||
};
|
||||
use std::collections::HashMap;
|
||||
use tokio::sync::Mutex;
|
||||
use uuid::Uuid;
|
||||
|
||||
#[derive(Default)]
|
||||
struct InMemoryGroupDocsRepo {
|
||||
// (group, collection) -> id -> data
|
||||
data: Mutex<HashMap<(GroupId, String), HashMap<Uuid, serde_json::Value>>>,
|
||||
}
|
||||
|
||||
fn row(id: Uuid, data: serde_json::Value) -> DocRow {
|
||||
DocRow {
|
||||
id,
|
||||
data,
|
||||
created_at: Utc::now(),
|
||||
updated_at: Utc::now(),
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl GroupDocsRepo for InMemoryGroupDocsRepo {
|
||||
async fn create(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
data: serde_json::Value,
|
||||
) -> Result<DocRow, GroupDocsRepoError> {
|
||||
let id = Uuid::new_v4();
|
||||
self.data
|
||||
.lock()
|
||||
.await
|
||||
.entry((group_id, collection.to_string()))
|
||||
.or_default()
|
||||
.insert(id, data.clone());
|
||||
Ok(row(id, data))
|
||||
}
|
||||
|
||||
async fn get(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
id: DocId,
|
||||
) -> Result<Option<DocRow>, GroupDocsRepoError> {
|
||||
Ok(self
|
||||
.data
|
||||
.lock()
|
||||
.await
|
||||
.get(&(group_id, collection.to_string()))
|
||||
.and_then(|m| m.get(&id).cloned())
|
||||
.map(|d| row(id, d)))
|
||||
}
|
||||
|
||||
// The fake ignores the filter (filter SQL is covered by docs_repo tests
|
||||
// + the journey); returns all docs in the collection.
|
||||
async fn find(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
_filter: &DocsFilter,
|
||||
) -> Result<Vec<DocRow>, GroupDocsRepoError> {
|
||||
Ok(self
|
||||
.data
|
||||
.lock()
|
||||
.await
|
||||
.get(&(group_id, collection.to_string()))
|
||||
.map(|m| m.iter().map(|(id, d)| row(*id, d.clone())).collect())
|
||||
.unwrap_or_default())
|
||||
}
|
||||
|
||||
async fn update(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
id: DocId,
|
||||
data: serde_json::Value,
|
||||
) -> Result<Option<serde_json::Value>, GroupDocsRepoError> {
|
||||
let mut guard = self.data.lock().await;
|
||||
let coll = guard.entry((group_id, collection.to_string())).or_default();
|
||||
Ok(coll.insert(id, data))
|
||||
}
|
||||
|
||||
async fn delete(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
id: DocId,
|
||||
) -> Result<Option<serde_json::Value>, GroupDocsRepoError> {
|
||||
Ok(self
|
||||
.data
|
||||
.lock()
|
||||
.await
|
||||
.get_mut(&(group_id, collection.to_string()))
|
||||
.and_then(|m| m.remove(&id)))
|
||||
}
|
||||
|
||||
async fn list(
|
||||
&self,
|
||||
_group_id: GroupId,
|
||||
_collection: &str,
|
||||
_cursor: Option<&str>,
|
||||
_limit: u32,
|
||||
) -> Result<DocsListPage, GroupDocsRepoError> {
|
||||
Ok(DocsListPage {
|
||||
docs: vec![],
|
||||
next_cursor: None,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct FakeResolver {
|
||||
map: HashMap<(AppId, String), GroupId>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl GroupCollectionResolver for FakeResolver {
|
||||
async fn resolve_owning_group(
|
||||
&self,
|
||||
app_id: AppId,
|
||||
name: &str,
|
||||
_kind: &str,
|
||||
) -> Result<Option<GroupId>, sqlx::Error> {
|
||||
Ok(self.map.get(&(app_id, name.to_lowercase())).copied())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct DenyingAuthzRepo;
|
||||
|
||||
#[async_trait]
|
||||
impl AuthzRepo for DenyingAuthzRepo {
|
||||
async fn membership(
|
||||
&self,
|
||||
_user_id: UserId,
|
||||
_app_id: AppId,
|
||||
) -> Result<Option<AppRole>, AuthzError> {
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
|
||||
fn cx_with(app_id: AppId, principal: Option<Principal>) -> SdkCallCx {
|
||||
SdkCallCx {
|
||||
app_id,
|
||||
script_id: ScriptId::new(),
|
||||
principal,
|
||||
execution_id: ExecutionId::new(),
|
||||
request_id: RequestId::new(),
|
||||
trigger_depth: 0,
|
||||
root_execution_id: ExecutionId::new(),
|
||||
is_dead_letter_handler: false,
|
||||
event: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn owner() -> Principal {
|
||||
Principal {
|
||||
user_id: AdminUserId::new(),
|
||||
instance_role: InstanceRole::Owner,
|
||||
scopes: None,
|
||||
app_binding: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn svc(resolver: FakeResolver) -> GroupDocsServiceImpl {
|
||||
GroupDocsServiceImpl::new(
|
||||
Arc::new(InMemoryGroupDocsRepo::default()),
|
||||
Arc::new(resolver),
|
||||
Arc::new(DenyingAuthzRepo),
|
||||
)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn unrelated_app_gets_collection_not_shared() {
|
||||
let app_a = AppId::new();
|
||||
let app_b = AppId::new();
|
||||
let group = GroupId::new();
|
||||
let mut resolver = FakeResolver::default();
|
||||
resolver.map.insert((app_a, "articles".into()), group);
|
||||
let docs = svc(resolver);
|
||||
|
||||
let cx_a = cx_with(app_a, Some(owner()));
|
||||
let id = docs
|
||||
.create(&cx_a, "articles", serde_json::json!({"t": "hi"}))
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(docs.get(&cx_a, "articles", id).await.unwrap().is_some());
|
||||
|
||||
let cx_b = cx_with(app_b, Some(owner()));
|
||||
let err = docs
|
||||
.find(&cx_b, "articles", serde_json::json!({}))
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(matches!(err, GroupDocsError::CollectionNotShared(c) if c == "articles"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn reads_open_writes_require_auth() {
|
||||
let app = AppId::new();
|
||||
let group = GroupId::new();
|
||||
let mut resolver = FakeResolver::default();
|
||||
resolver.map.insert((app, "articles".into()), group);
|
||||
let docs = svc(resolver);
|
||||
|
||||
let owner_cx = cx_with(app, Some(owner()));
|
||||
docs.create(&owner_cx, "articles", serde_json::json!({"t": "hi"}))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Anonymous READ (find) is allowed.
|
||||
let anon = cx_with(app, None);
|
||||
let hits = docs
|
||||
.find(&anon, "articles", serde_json::json!({}))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(hits.len(), 1);
|
||||
|
||||
// Anonymous WRITE (create) fails closed.
|
||||
let err = docs
|
||||
.create(&anon, "articles", serde_json::json!({"t": "x"}))
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(matches!(err, GroupDocsError::Forbidden));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn non_object_data_rejected() {
|
||||
let app = AppId::new();
|
||||
let group = GroupId::new();
|
||||
let mut resolver = FakeResolver::default();
|
||||
resolver.map.insert((app, "articles".into()), group);
|
||||
let docs = svc(resolver);
|
||||
let cx = cx_with(app, Some(owner()));
|
||||
let err = docs
|
||||
.create(&cx, "articles", serde_json::json!("not-an-object"))
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(matches!(err, GroupDocsError::InvalidData));
|
||||
}
|
||||
}
|
||||
@@ -52,6 +52,7 @@ pub mod files_sweep;
|
||||
pub mod gc;
|
||||
pub mod group_collection_repo;
|
||||
pub mod group_docs_repo;
|
||||
pub mod group_docs_service;
|
||||
pub mod group_kv_repo;
|
||||
pub mod group_kv_service;
|
||||
pub mod group_members_repo;
|
||||
@@ -182,6 +183,7 @@ pub use files_sweep::{spawn_files_orphan_sweep, sweep_orphan_tmp_files, SweepSta
|
||||
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_docs_service::GroupDocsServiceImpl;
|
||||
pub use group_kv_repo::{GroupKvRepo, GroupKvRepoError, PostgresGroupKvRepo};
|
||||
pub use group_kv_service::GroupKvServiceImpl;
|
||||
pub use group_members_repo::{
|
||||
|
||||
Reference in New Issue
Block a user