feat(group-blobs): read-only operator admin API for shared collections (M4.1+M4.2)
group_blobs_api mirrors the per-app kv_api/files_api for groups: GET
/groups/{id}/kv[/{c}/{k}], /docs[/{c}/{id}], /files[/{c}/{id}] — list +
fetch/download a group's §11.6 shared KV/docs/files. Authz GroupKvRead/
GroupDocsRead/GroupFilesRead (viewer tier, reads-open trust model). Read-only
by design (writes stay script-only). The host hoists the three group repos so
the SDK services and this router share one instance.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
313
crates/manager-core/src/group_blobs_api.rs
Normal file
313
crates/manager-core/src/group_blobs_api.rs
Normal file
@@ -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<dyn GroupKvRepo>,
|
||||
pub docs: Arc<dyn GroupDocsRepo>,
|
||||
pub files: Arc<dyn GroupFilesRepo>,
|
||||
pub groups: Arc<dyn GroupRepository>,
|
||||
pub authz: Arc<dyn AuthzRepo>,
|
||||
}
|
||||
|
||||
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<String>,
|
||||
#[serde(default)]
|
||||
pub limit: Option<u32>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize)]
|
||||
struct ListKeysResponse {
|
||||
keys: Vec<String>,
|
||||
next_cursor: Option<String>,
|
||||
}
|
||||
|
||||
async fn list_kv(
|
||||
State(s): State<GroupBlobsState>,
|
||||
Extension(principal): Extension<Principal>,
|
||||
Path(ident): Path<String>,
|
||||
Query(q): Query<ListQuery>,
|
||||
) -> Result<Json<ListKeysResponse>, 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<GroupBlobsState>,
|
||||
Extension(principal): Extension<Principal>,
|
||||
Path((ident, collection, key)): Path<(String, String, String)>,
|
||||
) -> Result<Json<serde_json::Value>, 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<DocEntry>,
|
||||
next_cursor: Option<String>,
|
||||
}
|
||||
|
||||
async fn list_docs(
|
||||
State(s): State<GroupBlobsState>,
|
||||
Extension(principal): Extension<Principal>,
|
||||
Path(ident): Path<String>,
|
||||
Query(q): Query<ListQuery>,
|
||||
) -> Result<Json<ListDocsResponse>, 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<GroupBlobsState>,
|
||||
Extension(principal): Extension<Principal>,
|
||||
Path((ident, collection, doc_id)): Path<(String, String, String)>,
|
||||
) -> Result<Json<serde_json::Value>, 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::<Uuid>()
|
||||
.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<GroupBlobsState>,
|
||||
Extension(principal): Extension<Principal>,
|
||||
Path(ident): Path<String>,
|
||||
Query(q): Query<ListQuery>,
|
||||
) -> Result<Json<serde_json::Value>, 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<GroupBlobsState>,
|
||||
Extension(principal): Extension<Principal>,
|
||||
Path((ident, collection, file_id)): Path<(String, String, String)>,
|
||||
) -> Result<Response, GroupBlobsError> {
|
||||
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::<Uuid>()
|
||||
.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<GroupId, GroupBlobsError> {
|
||||
let found = if let Ok(uuid) = ident.parse::<Uuid>() {
|
||||
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<AuthzDenied> 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()
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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<dyn picloud_shared::GroupKvService> = 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<dyn picloud_shared::GroupDocsService> = 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(
|
||||
// `<root>/files/groups/<group_id>/...`, 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<dyn picloud_shared::GroupFilesService> = 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));
|
||||
|
||||
Reference in New Issue
Block a user