//! Low-level metadata (Postgres `group_files`) + blob bytes (filesystem) storage //! for §11.6 group-shared FILES collections. A near-clone of [`crate::files_repo`] //! keyed by the owning `group_id` instead of `app_id`. //! //! The security-sensitive disk mechanics — the atomic write+checksum protocol and //! checksum-on-read — are **not** duplicated: they come from the owner-relative //! free functions in [`crate::files_repo`] (`write_atomic_at` / `read_verify_at` / //! `final_path_at`), called with this repo's owner sub-path. Group blobs shard at //! `/files/groups////` — a `groups/` infix //! disjoint from the per-app `files//...` subtree, so the orphan sweeper //! ([crate::files_sweep]) covers both with one walk. Authorization, group //! resolution, value validation, and content-type sanitization live one layer up //! in `GroupFilesServiceImpl`. use std::path::{Path, PathBuf}; use async_trait::async_trait; use chrono::{DateTime, Utc}; use picloud_shared::{FileMeta, FileUpdate, FilesListPage, GroupId, NewFile}; use sqlx::PgPool; use uuid::Uuid; use crate::files_repo::{ decode_cursor, encode_cursor, final_path_at, read_verify_at, write_atomic_at, FileUpdated, FilesRepoError, }; const FILES_LIST_MAX_LIMIT: u32 = 1_000; const FILES_LIST_DEFAULT_LIMIT: u32 = 100; #[derive(Debug, thiserror::Error)] pub enum GroupFilesRepoError { #[error("database error: {0}")] Db(#[from] sqlx::Error), #[error("filesystem error: {0}")] Io(String), #[error("invalid collection name: {0}")] InvalidCollection(String), /// Bytes on disk no longer match the stored checksum (or are missing). #[error("file content corrupted (checksum mismatch)")] Corrupted, #[error("invalid pagination cursor")] InvalidCursor, } impl From for GroupFilesRepoError { fn from(e: FilesRepoError) -> Self { match e { FilesRepoError::Db(e) => Self::Db(e), FilesRepoError::Io(s) => Self::Io(s), FilesRepoError::InvalidCollection(c) => Self::InvalidCollection(c), FilesRepoError::Corrupted => Self::Corrupted, FilesRepoError::InvalidCursor => Self::InvalidCursor, } } } /// The owner-relative subdirectory of a **group**'s shared blobs under /// `/files/`: `groups/`. A UUID app-dir can never equal the /// literal `groups`, so app and group blob trees can't collide. pub(crate) fn group_owner_dir(group_id: GroupId) -> PathBuf { Path::new("groups").join(group_id.into_inner().to_string()) } #[async_trait] pub trait GroupFilesRepo: Send + Sync { async fn create( &self, group_id: GroupId, collection: &str, new: NewFile, ) -> Result; async fn head( &self, group_id: GroupId, collection: &str, id: Uuid, ) -> Result, GroupFilesRepoError>; /// Reads + checksum-verifies the bytes. `Ok(None)` when no row exists; /// `Err(Corrupted)` when the row exists but the bytes are missing/mismatched. async fn get( &self, group_id: GroupId, collection: &str, id: Uuid, ) -> Result>, GroupFilesRepoError>; /// `Ok(None)` when no row exists (the service maps that to `NotFound`). async fn update( &self, group_id: GroupId, collection: &str, id: Uuid, upd: FileUpdate, ) -> Result, GroupFilesRepoError>; async fn delete( &self, group_id: GroupId, collection: &str, id: Uuid, ) -> Result, GroupFilesRepoError>; async fn list( &self, group_id: GroupId, collection: &str, cursor: Option<&str>, limit: u32, ) -> Result; } /// Filesystem-bytes + Postgres-metadata repo for group-shared files. pub struct FsGroupFilesRepo { pool: PgPool, root: PathBuf, } impl FsGroupFilesRepo { #[must_use] pub fn new(pool: PgPool, root: PathBuf) -> Self { Self { pool, root } } /// Belt-and-suspenders path guard (the service validates at the SDK /// boundary). Mirrors `FsFilesRepo::guard_collection`. fn guard_collection(collection: &str) -> Result<(), GroupFilesRepoError> { if collection.is_empty() || collection.contains('/') || collection.contains('\\') || collection.contains("..") || collection.contains('\0') { return Err(GroupFilesRepoError::InvalidCollection( collection.to_string(), )); } Ok(()) } fn write_atomic( &self, group_id: GroupId, collection: &str, id: Uuid, bytes: &[u8], ) -> Result { Ok(write_atomic_at( &self.root, &group_owner_dir(group_id), collection, id, bytes, )?) } } #[async_trait] impl GroupFilesRepo for FsGroupFilesRepo { async fn create( &self, group_id: GroupId, collection: &str, new: NewFile, ) -> Result { Self::guard_collection(collection)?; let id = Uuid::new_v4(); let size = i64::try_from(new.data.len()).unwrap_or(i64::MAX); let checksum = self.write_atomic(group_id, collection, id, &new.data)?; let row: GroupFileRow = sqlx::query_as( "INSERT INTO group_files \ (group_id, collection, id, name, content_type, size_bytes, checksum_sha256) \ VALUES ($1, $2, $3, $4, $5, $6, $7) \ RETURNING id, collection, name, content_type, size_bytes, \ checksum_sha256, created_at, updated_at", ) .bind(group_id.into_inner()) .bind(collection) .bind(id) .bind(&new.name) .bind(&new.content_type) .bind(size) .bind(&checksum) .fetch_one(&self.pool) .await?; Ok(row.into_meta()) } async fn head( &self, group_id: GroupId, collection: &str, id: Uuid, ) -> Result, GroupFilesRepoError> { Self::guard_collection(collection)?; let row: Option = sqlx::query_as( "SELECT id, collection, name, content_type, size_bytes, \ checksum_sha256, created_at, updated_at \ FROM group_files 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(GroupFileRow::into_meta)) } async fn get( &self, group_id: GroupId, collection: &str, id: Uuid, ) -> Result>, GroupFilesRepoError> { Self::guard_collection(collection)?; let row: Option<(String,)> = sqlx::query_as( "SELECT checksum_sha256 FROM group_files \ WHERE group_id = $1 AND collection = $2 AND id = $3", ) .bind(group_id.into_inner()) .bind(collection) .bind(id) .fetch_optional(&self.pool) .await?; let Some((stored_checksum,)) = row else { return Ok(None); }; let bytes = read_verify_at( &self.root, &group_owner_dir(group_id), collection, id, &stored_checksum, )?; Ok(Some(bytes)) } async fn update( &self, group_id: GroupId, collection: &str, id: Uuid, upd: FileUpdate, ) -> Result, GroupFilesRepoError> { Self::guard_collection(collection)?; let Some(prev) = self.head(group_id, collection, id).await? else { return Ok(None); }; let size = i64::try_from(upd.data.len()).unwrap_or(i64::MAX); let checksum = self.write_atomic(group_id, collection, id, &upd.data)?; let row: GroupFileRow = sqlx::query_as( "UPDATE group_files SET \ name = COALESCE($4, name), \ content_type = COALESCE($5, content_type), \ size_bytes = $6, \ checksum_sha256 = $7, \ updated_at = NOW() \ WHERE group_id = $1 AND collection = $2 AND id = $3 \ RETURNING id, collection, name, content_type, size_bytes, \ checksum_sha256, created_at, updated_at", ) .bind(group_id.into_inner()) .bind(collection) .bind(id) .bind(upd.name.as_deref()) .bind(upd.content_type.as_deref()) .bind(size) .bind(&checksum) .fetch_one(&self.pool) .await?; Ok(Some(FileUpdated { new: row.into_meta(), prev, })) } async fn delete( &self, group_id: GroupId, collection: &str, id: Uuid, ) -> Result, GroupFilesRepoError> { Self::guard_collection(collection)?; let mut tx = self.pool.begin().await?; let row: Option = sqlx::query_as( "SELECT id, collection, name, content_type, size_bytes, \ checksum_sha256, created_at, updated_at \ FROM group_files WHERE group_id = $1 AND collection = $2 AND id = $3 \ FOR UPDATE", ) .bind(group_id.into_inner()) .bind(collection) .bind(id) .fetch_optional(&mut *tx) .await?; let Some(row) = row else { tx.rollback().await?; return Ok(None); }; sqlx::query("DELETE FROM group_files WHERE group_id = $1 AND collection = $2 AND id = $3") .bind(group_id.into_inner()) .bind(collection) .bind(id) .execute(&mut *tx) .await?; tx.commit().await?; // Row gone; unlink the bytes. A failure here leaves an orphan file // (reclaimed by the sweep) — not fatal. let path = final_path_at(&self.root, &group_owner_dir(group_id), collection, id); if let Err(e) = std::fs::remove_file(&path) { if e.kind() != std::io::ErrorKind::NotFound { tracing::warn!(path = %path.display(), error = %e, "group files: unlink after delete failed (orphan)"); } } Ok(Some(row.into_meta())) } async fn list( &self, group_id: GroupId, collection: &str, cursor: Option<&str>, limit: u32, ) -> Result { Self::guard_collection(collection)?; let limit = if limit == 0 { FILES_LIST_DEFAULT_LIMIT } else { limit.min(FILES_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 = sqlx::query_as( "SELECT id, collection, name, content_type, size_bytes, \ checksum_sha256, created_at, updated_at \ FROM group_files \ 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 files: Vec = rows.into_iter().map(GroupFileRow::into_meta).collect(); let next_cursor = if files.len() > limit as usize { files.truncate(limit as usize); files.last().map(|m| encode_cursor(m.id)) } else { None }; Ok(FilesListPage { files, next_cursor }) } } #[derive(sqlx::FromRow)] struct GroupFileRow { id: Uuid, collection: String, name: String, content_type: String, size_bytes: i64, checksum_sha256: String, created_at: DateTime, updated_at: DateTime, } impl GroupFileRow { fn into_meta(self) -> FileMeta { FileMeta { id: self.id, collection: self.collection, name: self.name, content_type: self.content_type, size: u64::try_from(self.size_bytes).unwrap_or(0), checksum: self.checksum_sha256, created_at: self.created_at, updated_at: self.updated_at, } } }