feat(modules): GroupFilesService + files::shared_collection handle (§11.6 files C3)
- shared/group_files: GroupFilesService trait (mirror FilesService, no group id arg — owner resolved from cx.app_id), GroupFilesError (clone of FilesError + CollectionNotShared) with From<FilesError>, Noop. - services: group_files field + with_group_files setter (default Noop). - group_files_service: GroupFilesServiceImpl — owning_group via the registry resolver (kind=files) → CollectionNotShared; reads open (script_gate), writes fail closed (script_gate_require_principal); reuses validate_files_collection + the NewFile/FileUpdate validators + sanitize_stored_content_type + the per-file size cap; no events. - sdk/files: GroupFilesHandle + files::shared_collection(name) + the create/head/get/update/delete/list group methods (Blob in/out). - picloud: construct FsGroupFilesRepo (sharing the files root + size cap) + GroupFilesServiceImpl, .with_group_files(...). Unit tests: off-chain → CollectionNotShared; anon read OK / anon write Forbidden; oversize → TooLarge. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
@@ -255,12 +255,7 @@ pub(crate) fn shard_dir_at(
|
||||
.join(&id_str[..2])
|
||||
}
|
||||
|
||||
pub(crate) fn final_path_at(
|
||||
root: &Path,
|
||||
owner_rel: &Path,
|
||||
collection: &str,
|
||||
id: Uuid,
|
||||
) -> PathBuf {
|
||||
pub(crate) fn final_path_at(root: &Path, owner_rel: &Path, collection: &str, id: Uuid) -> PathBuf {
|
||||
let id_str = id.to_string();
|
||||
shard_dir_at(root, owner_rel, collection, &id_str).join(&id_str)
|
||||
}
|
||||
|
||||
492
crates/manager-core/src/group_files_service.rs
Normal file
492
crates/manager-core/src/group_files_service.rs
Normal file
@@ -0,0 +1,492 @@
|
||||
//! `GroupFilesServiceImpl` — wires `GroupFilesRepo` + the group-collection
|
||||
//! registry underneath the `picloud_shared::GroupFilesService` trait that scripts
|
||||
//! reach via the `files::shared_collection("name")` Rhai handle (§11.6).
|
||||
//!
|
||||
//! Combines the group-KV/docs service pattern (owner resolution from `cx.app_id`,
|
||||
//! reads-open / writes-authed authz) with the files surface (collection
|
||||
//! path-validation, field + size-cap validation, content-type sanitization). 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::{
|
||||
sanitize_stored_content_type, validate_files_collection, FileMeta, FileUpdate, FilesListPage,
|
||||
GroupFilesError, GroupFilesService, GroupId, NewFile, SdkCallCx,
|
||||
};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::authz::{self, AuthzRepo, Capability};
|
||||
use crate::files_repo::FileUpdated;
|
||||
use crate::group_collection_repo::GroupCollectionResolver;
|
||||
use crate::group_files_repo::{GroupFilesRepo, GroupFilesRepoError};
|
||||
|
||||
/// The registry `kind` this service resolves.
|
||||
const KIND_FILES: &str = "files";
|
||||
|
||||
pub struct GroupFilesServiceImpl {
|
||||
repo: Arc<dyn GroupFilesRepo>,
|
||||
resolver: Arc<dyn GroupCollectionResolver>,
|
||||
authz: Arc<dyn AuthzRepo>,
|
||||
max_file_size_bytes: usize,
|
||||
}
|
||||
|
||||
impl GroupFilesServiceImpl {
|
||||
#[must_use]
|
||||
pub fn new(
|
||||
repo: Arc<dyn GroupFilesRepo>,
|
||||
resolver: Arc<dyn GroupCollectionResolver>,
|
||||
authz: Arc<dyn AuthzRepo>,
|
||||
max_file_size_bytes: usize,
|
||||
) -> Self {
|
||||
Self {
|
||||
repo,
|
||||
resolver,
|
||||
authz,
|
||||
max_file_size_bytes,
|
||||
}
|
||||
}
|
||||
|
||||
/// The structural boundary: resolve the collection to its owning group
|
||||
/// (kind=`files`) on the calling app's chain. `CollectionNotShared` when no
|
||||
/// ancestor group declares it.
|
||||
async fn owning_group(
|
||||
&self,
|
||||
cx: &SdkCallCx,
|
||||
collection: &str,
|
||||
) -> Result<GroupId, GroupFilesError> {
|
||||
self.resolver
|
||||
.resolve_owning_group(cx.app_id, collection, KIND_FILES)
|
||||
.await
|
||||
.map_err(|e| GroupFilesError::Backend(e.to_string()))?
|
||||
.ok_or_else(|| GroupFilesError::CollectionNotShared(collection.to_string()))
|
||||
}
|
||||
|
||||
async fn check_read(&self, cx: &SdkCallCx, group_id: GroupId) -> Result<(), GroupFilesError> {
|
||||
authz::script_gate(
|
||||
&*self.authz,
|
||||
cx,
|
||||
Capability::GroupFilesRead(group_id),
|
||||
|| GroupFilesError::Forbidden,
|
||||
GroupFilesError::Backend,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn check_write(&self, cx: &SdkCallCx, group_id: GroupId) -> Result<(), GroupFilesError> {
|
||||
// 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::GroupFilesWrite(group_id),
|
||||
|| GroupFilesError::Forbidden,
|
||||
GroupFilesError::Backend,
|
||||
)
|
||||
.await
|
||||
}
|
||||
}
|
||||
|
||||
/// Invalid UUIDs aren't an error shape the SDK exposes — for reads/deletes they
|
||||
/// simply mean "no such file" (mirrors `files_service::parse_id`).
|
||||
fn parse_id(id: &str) -> Option<Uuid> {
|
||||
Uuid::parse_str(id).ok()
|
||||
}
|
||||
|
||||
impl From<GroupFilesRepoError> for GroupFilesError {
|
||||
fn from(e: GroupFilesRepoError) -> Self {
|
||||
match e {
|
||||
GroupFilesRepoError::Corrupted => Self::Corrupted,
|
||||
GroupFilesRepoError::InvalidCollection(c) => Self::InvalidCollection(c),
|
||||
other => Self::Backend(other.to_string()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl GroupFilesService for GroupFilesServiceImpl {
|
||||
async fn create(
|
||||
&self,
|
||||
cx: &SdkCallCx,
|
||||
collection: &str,
|
||||
mut new: NewFile,
|
||||
) -> Result<Uuid, GroupFilesError> {
|
||||
validate_files_collection(collection)?;
|
||||
let group_id = self.owning_group(cx, collection).await?;
|
||||
new.validate(self.max_file_size_bytes)?;
|
||||
// Coerce dangerous render types to application/octet-stream after the
|
||||
// shape checks pass (same as app files, audit 2026-06-11 C-2).
|
||||
new.content_type = sanitize_stored_content_type(&new.content_type);
|
||||
self.check_write(cx, group_id).await?;
|
||||
Ok(self.repo.create(group_id, collection, new).await?.id)
|
||||
}
|
||||
|
||||
async fn head(
|
||||
&self,
|
||||
cx: &SdkCallCx,
|
||||
collection: &str,
|
||||
id: &str,
|
||||
) -> Result<Option<FileMeta>, GroupFilesError> {
|
||||
validate_files_collection(collection)?;
|
||||
let group_id = self.owning_group(cx, collection).await?;
|
||||
self.check_read(cx, group_id).await?;
|
||||
let Some(uuid) = parse_id(id) else {
|
||||
return Ok(None);
|
||||
};
|
||||
Ok(self.repo.head(group_id, collection, uuid).await?)
|
||||
}
|
||||
|
||||
async fn get(
|
||||
&self,
|
||||
cx: &SdkCallCx,
|
||||
collection: &str,
|
||||
id: &str,
|
||||
) -> Result<Option<Vec<u8>>, GroupFilesError> {
|
||||
validate_files_collection(collection)?;
|
||||
let group_id = self.owning_group(cx, collection).await?;
|
||||
self.check_read(cx, group_id).await?;
|
||||
let Some(uuid) = parse_id(id) else {
|
||||
return Ok(None);
|
||||
};
|
||||
Ok(self.repo.get(group_id, collection, uuid).await?)
|
||||
}
|
||||
|
||||
async fn update(
|
||||
&self,
|
||||
cx: &SdkCallCx,
|
||||
collection: &str,
|
||||
id: &str,
|
||||
mut upd: FileUpdate,
|
||||
) -> Result<(), GroupFilesError> {
|
||||
validate_files_collection(collection)?;
|
||||
let group_id = self.owning_group(cx, collection).await?;
|
||||
upd.validate(self.max_file_size_bytes)?;
|
||||
if let Some(ct) = upd.content_type.as_deref() {
|
||||
upd.content_type = Some(sanitize_stored_content_type(ct));
|
||||
}
|
||||
self.check_write(cx, group_id).await?;
|
||||
let Some(uuid) = parse_id(id) else {
|
||||
return Err(GroupFilesError::NotFound);
|
||||
};
|
||||
match self.repo.update(group_id, collection, uuid, upd).await? {
|
||||
Some(FileUpdated { .. }) => Ok(()),
|
||||
None => Err(GroupFilesError::NotFound),
|
||||
}
|
||||
}
|
||||
|
||||
async fn delete(
|
||||
&self,
|
||||
cx: &SdkCallCx,
|
||||
collection: &str,
|
||||
id: &str,
|
||||
) -> Result<bool, GroupFilesError> {
|
||||
validate_files_collection(collection)?;
|
||||
let group_id = self.owning_group(cx, collection).await?;
|
||||
self.check_write(cx, group_id).await?;
|
||||
let Some(uuid) = parse_id(id) else {
|
||||
return Ok(false);
|
||||
};
|
||||
Ok(self
|
||||
.repo
|
||||
.delete(group_id, collection, uuid)
|
||||
.await?
|
||||
.is_some())
|
||||
}
|
||||
|
||||
async fn list(
|
||||
&self,
|
||||
cx: &SdkCallCx,
|
||||
collection: &str,
|
||||
cursor: Option<&str>,
|
||||
limit: u32,
|
||||
) -> Result<FilesListPage, GroupFilesError> {
|
||||
validate_files_collection(collection)?;
|
||||
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 or
|
||||
// a filesystem. The on-disk atomic-write/checksum mechanics are covered by the
|
||||
// tempdir tests in `files_repo.rs`.
|
||||
// ----------------------------------------------------------------------------
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::authz::{AuthzError, AuthzRepo};
|
||||
use crate::group_files_repo::GroupFilesRepoError;
|
||||
use chrono::Utc;
|
||||
use picloud_shared::{
|
||||
AdminUserId, AppId, AppRole, ExecutionId, InstanceRole, Principal, RequestId, ScriptId,
|
||||
UserId,
|
||||
};
|
||||
use std::collections::HashMap;
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
#[derive(Default)]
|
||||
struct InMemoryGroupFilesRepo {
|
||||
#[allow(clippy::type_complexity)]
|
||||
data: Mutex<HashMap<(GroupId, String, Uuid), (FileMeta, Vec<u8>)>>,
|
||||
}
|
||||
|
||||
fn meta(id: Uuid, collection: &str, new: &NewFile) -> FileMeta {
|
||||
FileMeta {
|
||||
id,
|
||||
collection: collection.to_string(),
|
||||
name: new.name.clone(),
|
||||
content_type: new.content_type.clone(),
|
||||
size: new.data.len() as u64,
|
||||
checksum: String::new(),
|
||||
created_at: Utc::now(),
|
||||
updated_at: Utc::now(),
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl GroupFilesRepo for InMemoryGroupFilesRepo {
|
||||
async fn create(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
new: NewFile,
|
||||
) -> Result<FileMeta, GroupFilesRepoError> {
|
||||
let id = Uuid::new_v4();
|
||||
let m = meta(id, collection, &new);
|
||||
self.data.lock().await.insert(
|
||||
(group_id, collection.to_string(), id),
|
||||
(m.clone(), new.data),
|
||||
);
|
||||
Ok(m)
|
||||
}
|
||||
|
||||
async fn head(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
id: Uuid,
|
||||
) -> Result<Option<FileMeta>, GroupFilesRepoError> {
|
||||
Ok(self
|
||||
.data
|
||||
.lock()
|
||||
.await
|
||||
.get(&(group_id, collection.to_string(), id))
|
||||
.map(|(m, _)| m.clone()))
|
||||
}
|
||||
|
||||
async fn get(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
id: Uuid,
|
||||
) -> Result<Option<Vec<u8>>, GroupFilesRepoError> {
|
||||
Ok(self
|
||||
.data
|
||||
.lock()
|
||||
.await
|
||||
.get(&(group_id, collection.to_string(), id))
|
||||
.map(|(_, b)| b.clone()))
|
||||
}
|
||||
|
||||
async fn update(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
id: Uuid,
|
||||
upd: FileUpdate,
|
||||
) -> Result<Option<FileUpdated>, GroupFilesRepoError> {
|
||||
let mut data = self.data.lock().await;
|
||||
let key = (group_id, collection.to_string(), id);
|
||||
let Some((prev, _)) = data.get(&key).cloned() else {
|
||||
return Ok(None);
|
||||
};
|
||||
let mut m = prev.clone();
|
||||
m.size = upd.data.len() as u64;
|
||||
if let Some(n) = &upd.name {
|
||||
m.name = n.clone();
|
||||
}
|
||||
data.insert(key, (m.clone(), upd.data));
|
||||
Ok(Some(FileUpdated { new: m, prev }))
|
||||
}
|
||||
|
||||
async fn delete(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
id: Uuid,
|
||||
) -> Result<Option<FileMeta>, GroupFilesRepoError> {
|
||||
Ok(self
|
||||
.data
|
||||
.lock()
|
||||
.await
|
||||
.remove(&(group_id, collection.to_string(), id))
|
||||
.map(|(m, _)| m))
|
||||
}
|
||||
|
||||
async fn list(
|
||||
&self,
|
||||
group_id: GroupId,
|
||||
collection: &str,
|
||||
_cursor: Option<&str>,
|
||||
_limit: u32,
|
||||
) -> Result<FilesListPage, GroupFilesRepoError> {
|
||||
let files = self
|
||||
.data
|
||||
.lock()
|
||||
.await
|
||||
.iter()
|
||||
.filter(|((g, c, _), _)| *g == group_id && c == collection)
|
||||
.map(|(_, (m, _))| m.clone())
|
||||
.collect();
|
||||
Ok(FilesListPage {
|
||||
files,
|
||||
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) -> GroupFilesServiceImpl {
|
||||
GroupFilesServiceImpl::new(
|
||||
Arc::new(InMemoryGroupFilesRepo::default()),
|
||||
Arc::new(resolver),
|
||||
Arc::new(DenyingAuthzRepo),
|
||||
10 * 1024 * 1024,
|
||||
)
|
||||
}
|
||||
|
||||
fn new_file(name: &str, data: &[u8]) -> NewFile {
|
||||
NewFile {
|
||||
name: name.to_string(),
|
||||
content_type: "application/octet-stream".to_string(),
|
||||
data: data.to_vec(),
|
||||
}
|
||||
}
|
||||
|
||||
#[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, "assets".into()), group);
|
||||
let files = svc(resolver);
|
||||
|
||||
let cx_a = cx_with(app_a, Some(owner()));
|
||||
let id = files
|
||||
.create(&cx_a, "assets", new_file("a.txt", b"hi"))
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(files
|
||||
.get(&cx_a, "assets", &id.to_string())
|
||||
.await
|
||||
.unwrap()
|
||||
.is_some());
|
||||
|
||||
let cx_b = cx_with(app_b, Some(owner()));
|
||||
let err = files.list(&cx_b, "assets", None, 100).await.unwrap_err();
|
||||
assert!(matches!(err, GroupFilesError::CollectionNotShared(c) if c == "assets"));
|
||||
}
|
||||
|
||||
#[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, "assets".into()), group);
|
||||
let files = svc(resolver);
|
||||
|
||||
let owner_cx = cx_with(app, Some(owner()));
|
||||
let id = files
|
||||
.create(&owner_cx, "assets", new_file("a.txt", b"hi"))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Anonymous READ (get) is allowed.
|
||||
let anon = cx_with(app, None);
|
||||
let bytes = files.get(&anon, "assets", &id.to_string()).await.unwrap();
|
||||
assert_eq!(bytes, Some(b"hi".to_vec()));
|
||||
|
||||
// Anonymous WRITE (create) fails closed.
|
||||
let err = files
|
||||
.create(&anon, "assets", new_file("x.txt", b"x"))
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(matches!(err, GroupFilesError::Forbidden));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn oversize_blob_rejected() {
|
||||
let app = AppId::new();
|
||||
let group = GroupId::new();
|
||||
let mut resolver = FakeResolver::default();
|
||||
resolver.map.insert((app, "assets".into()), group);
|
||||
let files = GroupFilesServiceImpl::new(
|
||||
Arc::new(InMemoryGroupFilesRepo::default()),
|
||||
Arc::new(resolver),
|
||||
Arc::new(DenyingAuthzRepo),
|
||||
8, // tiny cap
|
||||
);
|
||||
let cx = cx_with(app, Some(owner()));
|
||||
let err = files
|
||||
.create(&cx, "assets", new_file("big", b"123456789"))
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(matches!(err, GroupFilesError::TooLarge { limit: 8, .. }));
|
||||
}
|
||||
}
|
||||
@@ -54,6 +54,7 @@ pub mod group_collection_repo;
|
||||
pub mod group_docs_repo;
|
||||
pub mod group_docs_service;
|
||||
pub mod group_files_repo;
|
||||
pub mod group_files_service;
|
||||
pub mod group_kv_repo;
|
||||
pub mod group_kv_service;
|
||||
pub mod group_members_repo;
|
||||
@@ -186,6 +187,7 @@ pub use group_collection_repo::{GroupCollectionResolver, PostgresGroupCollection
|
||||
pub use group_docs_repo::{GroupDocsRepo, GroupDocsRepoError, PostgresGroupDocsRepo};
|
||||
pub use group_docs_service::GroupDocsServiceImpl;
|
||||
pub use group_files_repo::{FsGroupFilesRepo, GroupFilesRepo, GroupFilesRepoError};
|
||||
pub use group_files_service::GroupFilesServiceImpl;
|
||||
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