diff --git a/crates/manager-core/src/group_blobs_api.rs b/crates/manager-core/src/group_blobs_api.rs new file mode 100644 index 0000000..ff149e3 --- /dev/null +++ b/crates/manager-core/src/group_blobs_api.rs @@ -0,0 +1,313 @@ +//! `/api/v1/admin/groups/{id}/{kv,docs,files}*` — read-only operator inspection +//! of a group's §11.6 SHARED collections (M4). Mirrors the per-app `kv_api` / +//! `files_api` admin surface for groups so `pic {kv,docs,files} ls --group` (and +//! a future dashboard tab) can browse shared data without a script. +//! +//! **Read-only by design** — shared writes go through the SDK +//! (`kv::shared_collection(...)` etc.), which run the reads-open / writes-authed +//! authz + fire `shared = true` triggers; an admin write would bypass both. The +//! deferral (operator write/delete on shared blobs) stands. +//! +//! Capabilities: `GroupKvRead` / `GroupDocsRead` / `GroupFilesRead`, resolved +//! against the group loaded from the path (the same tier the SDK read path uses). + +use std::sync::Arc; + +use axum::extract::{Path, Query, State}; +use axum::http::header::{CONTENT_LENGTH, CONTENT_TYPE}; +use axum::response::{IntoResponse, Json, Response}; +use axum::routing::get; +use axum::{Extension, Router}; +use picloud_shared::{GroupId, Principal}; +use serde::{Deserialize, Serialize}; +use serde_json::json; +use uuid::Uuid; + +use crate::authz::{require, AuthzDenied, AuthzRepo, Capability}; +use crate::group_docs_repo::GroupDocsRepo; +use crate::group_files_repo::GroupFilesRepo; +use crate::group_kv_repo::GroupKvRepo; +use crate::group_repo::GroupRepository; + +#[derive(Clone)] +pub struct GroupBlobsState { + pub kv: Arc, + pub docs: Arc, + pub files: Arc, + pub groups: Arc, + pub authz: Arc, +} + +pub fn group_blobs_router(state: GroupBlobsState) -> Router { + Router::new() + .route("/groups/{id}/kv", get(list_kv)) + .route("/groups/{id}/kv/{collection}/{key}", get(get_kv)) + .route("/groups/{id}/docs", get(list_docs)) + .route("/groups/{id}/docs/{collection}/{doc_id}", get(get_doc)) + .route("/groups/{id}/files", get(list_files)) + .route("/groups/{id}/files/{collection}/{file_id}", get(get_file)) + .with_state(state) +} + +#[derive(Debug, Deserialize)] +pub struct ListQuery { + pub collection: String, + #[serde(default)] + pub cursor: Option, + #[serde(default)] + pub limit: Option, +} + +#[derive(Debug, Serialize)] +struct ListKeysResponse { + keys: Vec, + next_cursor: Option, +} + +async fn list_kv( + State(s): State, + Extension(principal): Extension, + Path(ident): Path, + Query(q): Query, +) -> Result, GroupBlobsError> { + let group_id = resolve_group(&*s.groups, &ident).await?; + require( + s.authz.as_ref(), + &principal, + Capability::GroupKvRead(group_id), + ) + .await?; + let page = + s.kv.list( + group_id, + &q.collection, + q.cursor.as_deref(), + q.limit.unwrap_or(0), + ) + .await + .map_err(|e| GroupBlobsError::Backend(e.to_string()))?; + Ok(Json(ListKeysResponse { + keys: page.keys, + next_cursor: page.next_cursor, + })) +} + +async fn get_kv( + State(s): State, + Extension(principal): Extension, + Path((ident, collection, key)): Path<(String, String, String)>, +) -> Result, GroupBlobsError> { + let group_id = resolve_group(&*s.groups, &ident).await?; + require( + s.authz.as_ref(), + &principal, + Capability::GroupKvRead(group_id), + ) + .await?; + let value = + s.kv.get(group_id, &collection, &key) + .await + .map_err(|e| GroupBlobsError::Backend(e.to_string()))? + .ok_or(GroupBlobsError::NotFound)?; + Ok(Json(json!({ "value": value }))) +} + +#[derive(Debug, Serialize)] +struct DocEntry { + id: String, + data: serde_json::Value, +} + +#[derive(Debug, Serialize)] +struct ListDocsResponse { + docs: Vec, + next_cursor: Option, +} + +async fn list_docs( + State(s): State, + Extension(principal): Extension, + Path(ident): Path, + Query(q): Query, +) -> Result, GroupBlobsError> { + let group_id = resolve_group(&*s.groups, &ident).await?; + require( + s.authz.as_ref(), + &principal, + Capability::GroupDocsRead(group_id), + ) + .await?; + let page = s + .docs + .list( + group_id, + &q.collection, + q.cursor.as_deref(), + q.limit.unwrap_or(0), + ) + .await + .map_err(|e| GroupBlobsError::Backend(e.to_string()))?; + Ok(Json(ListDocsResponse { + docs: page + .docs + .into_iter() + .map(|d| DocEntry { + id: d.id.to_string(), + data: d.data, + }) + .collect(), + next_cursor: page.next_cursor, + })) +} + +async fn get_doc( + State(s): State, + Extension(principal): Extension, + Path((ident, collection, doc_id)): Path<(String, String, String)>, +) -> Result, GroupBlobsError> { + let group_id = resolve_group(&*s.groups, &ident).await?; + require( + s.authz.as_ref(), + &principal, + Capability::GroupDocsRead(group_id), + ) + .await?; + let id = doc_id + .parse::() + .map_err(|_| GroupBlobsError::NotFound)?; + let row = s + .docs + .get(group_id, &collection, id) + .await + .map_err(|e| GroupBlobsError::Backend(e.to_string()))? + .ok_or(GroupBlobsError::NotFound)?; + Ok(Json(json!({ "id": row.id.to_string(), "data": row.data }))) +} + +async fn list_files( + State(s): State, + Extension(principal): Extension, + Path(ident): Path, + Query(q): Query, +) -> Result, GroupBlobsError> { + let group_id = resolve_group(&*s.groups, &ident).await?; + require( + s.authz.as_ref(), + &principal, + Capability::GroupFilesRead(group_id), + ) + .await?; + let page = s + .files + .list( + group_id, + &q.collection, + q.cursor.as_deref(), + q.limit.unwrap_or(0), + ) + .await + .map_err(|e| GroupBlobsError::Backend(e.to_string()))?; + // FileMeta is Serialize; return the metadata list + cursor. + Ok(Json(json!({ + "files": page.files, + "next_cursor": page.next_cursor, + }))) +} + +async fn get_file( + State(s): State, + Extension(principal): Extension, + Path((ident, collection, file_id)): Path<(String, String, String)>, +) -> Result { + let group_id = resolve_group(&*s.groups, &ident).await?; + require( + s.authz.as_ref(), + &principal, + Capability::GroupFilesRead(group_id), + ) + .await?; + let id = file_id + .parse::() + .map_err(|_| GroupBlobsError::NotFound)?; + let meta = s + .files + .head(group_id, &collection, id) + .await + .map_err(|e| GroupBlobsError::Backend(e.to_string()))? + .ok_or(GroupBlobsError::NotFound)?; + let bytes = s + .files + .get(group_id, &collection, id) + .await + .map_err(|e| GroupBlobsError::Backend(e.to_string()))? + .ok_or(GroupBlobsError::NotFound)?; + Ok(( + [ + (CONTENT_TYPE, meta.content_type), + (CONTENT_LENGTH, bytes.len().to_string()), + ], + bytes, + ) + .into_response()) +} + +async fn resolve_group( + groups: &dyn GroupRepository, + ident: &str, +) -> Result { + let found = if let Ok(uuid) = ident.parse::() { + groups + .get_by_id(uuid.into()) + .await + .map_err(|e| GroupBlobsError::Backend(e.to_string()))? + } else { + groups + .get_by_slug(ident) + .await + .map_err(|e| GroupBlobsError::Backend(e.to_string()))? + }; + found.map(|g| g.id).ok_or(GroupBlobsError::GroupNotFound) +} + +#[derive(Debug, thiserror::Error)] +pub enum GroupBlobsError { + #[error("group not found")] + GroupNotFound, + #[error("not found")] + NotFound, + #[error("forbidden")] + Forbidden, + #[error("authorization repo error: {0}")] + AuthzRepo(String), + #[error("backend: {0}")] + Backend(String), +} + +impl From for GroupBlobsError { + fn from(d: AuthzDenied) -> Self { + match d { + AuthzDenied::Denied => Self::Forbidden, + AuthzDenied::Repo(e) => Self::AuthzRepo(e.to_string()), + } + } +} + +impl IntoResponse for GroupBlobsError { + fn into_response(self) -> Response { + use axum::http::StatusCode; + let (status, body) = match &self { + Self::GroupNotFound | Self::NotFound => { + (StatusCode::NOT_FOUND, json!({ "error": self.to_string() })) + } + Self::Forbidden => (StatusCode::FORBIDDEN, json!({ "error": self.to_string() })), + Self::AuthzRepo(e) | Self::Backend(e) => { + tracing::error!(error = %e, "group blobs admin error"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + json!({ "error": "internal error" }), + ) + } + }; + (status, Json(body)).into_response() + } +} diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index dda3c23..c0a0c2f 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -50,6 +50,7 @@ pub mod files_repo; pub mod files_service; pub mod files_sweep; pub mod gc; +pub mod group_blobs_api; pub mod group_collection_repo; pub mod group_docs_repo; pub mod group_docs_service; diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index f651193..2940e4c 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -161,9 +161,13 @@ pub async fn build_app( )); // §11.6 shared group collections (KV): a separate store keyed by the owning // group, resolved from cx.app_id's chain. Reuses the KV value-size cap. + // §11.6 shared-collection repos hoisted so both the SDK services AND the M4 + // read-only operator admin API (`group_blobs_api`) share one instance. + let group_kv_repo = Arc::new(PostgresGroupKvRepo::new(pool.clone())); + let group_docs_repo = Arc::new(PostgresGroupDocsRepo::new(pool.clone())); let group_kv: Arc = Arc::new( GroupKvServiceImpl::with_max_value_bytes( - Arc::new(PostgresGroupKvRepo::new(pool.clone())), + group_kv_repo.clone(), Arc::new(PostgresGroupCollectionResolver::new(pool.clone())), authz.clone(), picloud_manager_core::kv_service::kv_max_value_bytes_from_env(), @@ -174,7 +178,7 @@ pub async fn build_app( // §11.6 shared group collections (docs): group-keyed group_docs store. let group_docs: Arc = Arc::new( GroupDocsServiceImpl::with_max_value_bytes( - Arc::new(PostgresGroupDocsRepo::new(pool.clone())), + group_docs_repo.clone(), Arc::new(PostgresGroupCollectionResolver::new(pool.clone())), authz.clone(), picloud_manager_core::docs_service::docs_max_value_bytes_from_env(), @@ -223,9 +227,10 @@ pub async fn build_app( // `/files/groups//...`, resolved from cx.app_id's chain. // Shares the per-file size cap; fires `shared = true` triggers under the // writer app (§11.6 shared-collection triggers). + let group_files_repo = Arc::new(FsGroupFilesRepo::new(pool.clone(), files_root.clone())); let group_files: Arc = Arc::new( GroupFilesServiceImpl::new( - Arc::new(FsGroupFilesRepo::new(pool.clone(), files_root.clone())), + group_files_repo.clone(), Arc::new(PostgresGroupCollectionResolver::new(pool.clone())), authz.clone(), files_max_size, @@ -683,6 +688,16 @@ pub async fn build_app( apps: apps_repo.clone(), authz: authz.clone(), })) + // §11.6 M4: read-only operator inspection of a group's shared collections. + .merge(picloud_manager_core::group_blobs_api::group_blobs_router( + picloud_manager_core::group_blobs_api::GroupBlobsState { + kv: group_kv_repo.clone(), + docs: group_docs_repo.clone(), + files: group_files_repo.clone(), + groups: groups_repo.clone(), + authz: authz.clone(), + }, + )) .merge(topics_router(topics_state)) .merge(secrets_router(secrets_state)) .merge(dead_letters_router(dead_letters_state));