//! `FilesRepo` — the metadata row (Postgres) + blob bytes (filesystem) //! storage layer for the v1.1.5 `files::*` SDK. //! //! Unlike KV/docs, this repo owns BOTH halves of a file: the `files` //! row (metadata + SHA-256 checksum) and the bytes on disk at //! `/files////`. //! It owns both because the write must be atomic across them — a crash //! mid-write must never leave a readable half-written file. //! //! ## Atomic write protocol (`create` / `update`) //! 1. Validate (collection path-safety; caps live one layer up). //! 2. `create_dir_all` the shard dir with `0o700`. //! 3. SHA-256 the in-memory bytes (single pass) while writing to //! `.tmp.`. //! 4. `fsync` the temp file. //! 5. `rename` temp → final (atomic on POSIX). //! 6. `fsync` the parent dir (so the rename is durable). //! 7. INSERT / UPDATE the DB row. //! //! A crash between 1–5 leaves an orphan `*.tmp.*` (never read). A crash //! between 5–7 leaves a file with no row — never reachable via the SDK //! (reads start from the row). Both are reclaimed by a future orphan //! sweep (deferred to v1.1.6+; see HANDBACK §7). //! //! ## Atomic delete protocol //! 1. SELECT + DELETE the row inside one transaction; commit. //! 2. `unlink` the file (outside the tx). A failure here leaves an //! orphan; a failure before the commit changes nothing. //! //! ## Checksum-on-read //! `get` reads the file, hashes it, and compares against the stored //! checksum — returning `FilesError::Corrupted` (and logging the path //! at error level) on a mismatch. It never auto-deletes; the operator //! decides what to do with a metadata-vs-bytes divergence. use std::env; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; use async_trait::async_trait; use base64::engine::general_purpose::URL_SAFE_NO_PAD; use base64::Engine as _; use chrono::{DateTime, Utc}; use picloud_shared::{AppId, FileMeta, FileUpdate, FilesListPage, NewFile}; use sha2::{Digest, Sha256}; use sqlx::PgPool; use uuid::Uuid; /// 100 MB default per-file cap. pub const DEFAULT_MAX_FILE_SIZE_BYTES: usize = 100 * 1024 * 1024; /// Default filesystem root (relative to the process CWD). pub const DEFAULT_FILES_ROOT: &str = "./data"; const FILES_LIST_MAX_LIMIT: u32 = 1_000; const FILES_LIST_DEFAULT_LIMIT: u32 = 100; /// Monotonic counter feeding unique temp-file suffixes (combined with /// the pid). Avoids `rand` in the storage layer per the brief. static TMP_COUNTER: AtomicU64 = AtomicU64::new(0); #[derive(Debug, thiserror::Error)] pub enum FilesRepoError { #[error("database error: {0}")] Db(#[from] sqlx::Error), #[error("filesystem error: {0}")] Io(String), #[error("invalid collection name: {0}")] InvalidCollection(String), /// The bytes on disk no longer match the stored checksum (or are /// missing entirely while the row persists). #[error("file content corrupted (checksum mismatch)")] Corrupted, #[error("invalid pagination cursor")] InvalidCursor, } /// Outbound-files tunables. Env-overridable following the same pattern /// as `HttpConfig::from_env`. #[derive(Debug, Clone)] pub struct FilesConfig { pub root: PathBuf, pub max_file_size_bytes: usize, } impl FilesConfig { #[must_use] pub fn conservative() -> Self { Self { root: PathBuf::from(DEFAULT_FILES_ROOT), max_file_size_bytes: DEFAULT_MAX_FILE_SIZE_BYTES, } } #[must_use] pub fn from_env() -> Self { let mut c = Self::conservative(); if let Ok(v) = env::var("PICLOUD_FILES_ROOT") { if !v.trim().is_empty() { c.root = PathBuf::from(v); } } if let Ok(v) = env::var("PICLOUD_FILES_MAX_FILE_SIZE_BYTES") { match v.parse::() { Ok(n) => c.max_file_size_bytes = n, Err(e) => { tracing::warn!(error = %e, "ignoring invalid PICLOUD_FILES_MAX_FILE_SIZE_BYTES"); } } } c } } impl Default for FilesConfig { fn default() -> Self { Self::conservative() } } /// The new+prior metadata returned from a successful `update`, so the /// service can emit a `ServiceEvent` with the change-data-capture /// surface (`old_payload`). #[derive(Debug, Clone)] pub struct FileUpdated { pub new: FileMeta, pub prev: FileMeta, } #[async_trait] pub trait FilesRepo: Send + Sync { async fn create( &self, app_id: AppId, collection: &str, new: NewFile, ) -> Result; async fn head( &self, app_id: AppId, collection: &str, id: Uuid, ) -> Result, FilesRepoError>; /// Reads + checksum-verifies the bytes. `Ok(None)` when no row /// exists; `Err(Corrupted)` when the row exists but the bytes are /// missing or mismatched. async fn get( &self, app_id: AppId, collection: &str, id: Uuid, ) -> Result>, FilesRepoError>; /// `Ok(None)` when no row exists (the SDK turns this into /// `FilesError::NotFound`). async fn update( &self, app_id: AppId, collection: &str, id: Uuid, upd: FileUpdate, ) -> Result, FilesRepoError>; /// Returns the deleted row's metadata if present, `None` otherwise. async fn delete( &self, app_id: AppId, collection: &str, id: Uuid, ) -> Result, FilesRepoError>; async fn list( &self, app_id: AppId, collection: &str, cursor: Option<&str>, limit: u32, ) -> Result; } /// Filesystem-bytes + Postgres-metadata repo. pub struct FsFilesRepo { pool: PgPool, config: FilesConfig, } impl FsFilesRepo { #[must_use] pub fn new(pool: PgPool, config: FilesConfig) -> Self { Self { pool, config } } /// Defensive path-component guard. The service already validates the /// collection at the SDK boundary; this is belt-and-suspenders so a /// future caller can't smuggle a traversal sequence onto disk. fn guard_collection(collection: &str) -> Result<(), FilesRepoError> { if collection.is_empty() || collection.contains('/') || collection.contains('\\') || collection.contains("..") || collection.contains('\0') { return Err(FilesRepoError::InvalidCollection(collection.to_string())); } Ok(()) } fn final_path(&self, app_id: AppId, collection: &str, id: Uuid) -> PathBuf { final_path_at(&self.config.root, app_id, collection, id) } fn write_atomic( &self, app_id: AppId, collection: &str, id: Uuid, bytes: &[u8], ) -> Result { write_atomic_at(&self.config.root, app_id, collection, id, bytes) } } fn shard_dir_at(root: &Path, app_id: AppId, collection: &str, id_str: &str) -> PathBuf { root.join("files") .join(app_id.into_inner().to_string()) .join(collection) .join(&id_str[..2]) } fn final_path_at(root: &Path, app_id: AppId, collection: &str, id: Uuid) -> PathBuf { let id_str = id.to_string(); shard_dir_at(root, app_id, collection, &id_str).join(&id_str) } /// Steps 2–6 of the atomic-write protocol. Returns the lowercase hex /// SHA-256 of the bytes (computed in a single pass over the in-memory /// buffer — the file is never re-read). Free function so the fs /// mechanics are unit-testable without a Postgres pool. fn write_atomic_at( root: &Path, app_id: AppId, collection: &str, id: Uuid, bytes: &[u8], ) -> Result { use std::io::Write as _; let id_str = id.to_string(); let dir = shard_dir_at(root, app_id, collection, &id_str); create_dir_all_secure(&dir)?; // Single-pass checksum over the in-memory buffer. let mut hasher = Sha256::new(); hasher.update(bytes); let checksum = hex_lower(&hasher.finalize()); let seq = TMP_COUNTER.fetch_add(1, Ordering::Relaxed); let tmp = dir.join(format!("{id_str}.tmp.{}-{seq}", std::process::id())); let final_path = dir.join(&id_str); { let mut f = std::fs::File::create(&tmp).map_err(io_err)?; f.write_all(bytes).map_err(io_err)?; f.sync_all().map_err(io_err)?; // fsync temp } std::fs::rename(&tmp, &final_path).map_err(io_err)?; // atomic // fsync the parent dir so the rename is durable. if let Ok(dirf) = std::fs::File::open(&dir) { let _ = dirf.sync_all(); } Ok(checksum) } /// Read + checksum-verify the bytes at the given path-set. Free /// function mirror of the `get` read path. Returns `Corrupted` when the /// bytes are missing or don't match `expected_checksum`. fn read_verify_at( root: &Path, app_id: AppId, collection: &str, id: Uuid, expected_checksum: &str, ) -> Result, FilesRepoError> { let path = final_path_at(root, app_id, collection, id); let bytes = match std::fs::read(&path) { Ok(b) => b, Err(e) => { tracing::error!( path = %path.display(), error = %e, "files: row exists but bytes are unreadable — treating as corrupted" ); return Err(FilesRepoError::Corrupted); } }; let mut hasher = Sha256::new(); hasher.update(&bytes); let actual = hex_lower(&hasher.finalize()); if actual != expected_checksum { tracing::error!( path = %path.display(), expected = %expected_checksum, actual = %actual, "files: checksum mismatch on read — content corrupted" ); return Err(FilesRepoError::Corrupted); } Ok(bytes) } #[async_trait] impl FilesRepo for FsFilesRepo { async fn create( &self, app_id: AppId, 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(app_id, collection, id, &new.data)?; let row: FileRow = sqlx::query_as( "INSERT INTO files \ (app_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(app_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, app_id: AppId, collection: &str, id: Uuid, ) -> Result, FilesRepoError> { 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 files WHERE app_id = $1 AND collection = $2 AND id = $3", ) .bind(app_id.into_inner()) .bind(collection) .bind(id) .fetch_optional(&self.pool) .await?; Ok(row.map(FileRow::into_meta)) } async fn get( &self, app_id: AppId, collection: &str, id: Uuid, ) -> Result>, FilesRepoError> { Self::guard_collection(collection)?; let row: Option<(String,)> = sqlx::query_as( "SELECT checksum_sha256 FROM files \ WHERE app_id = $1 AND collection = $2 AND id = $3", ) .bind(app_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.config.root, app_id, collection, id, &stored_checksum)?; Ok(Some(bytes)) } async fn update( &self, app_id: AppId, collection: &str, id: Uuid, upd: FileUpdate, ) -> Result, FilesRepoError> { Self::guard_collection(collection)?; // Read the prior row first (existence check + CDC surface). let Some(prev) = self.head(app_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(app_id, collection, id, &upd.data)?; let row: FileRow = sqlx::query_as( "UPDATE files SET \ name = COALESCE($4, name), \ content_type = COALESCE($5, content_type), \ size_bytes = $6, \ checksum_sha256 = $7, \ updated_at = NOW() \ WHERE app_id = $1 AND collection = $2 AND id = $3 \ RETURNING id, collection, name, content_type, size_bytes, \ checksum_sha256, created_at, updated_at", ) .bind(app_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, app_id: AppId, collection: &str, id: Uuid, ) -> Result, FilesRepoError> { Self::guard_collection(collection)?; // SELECT + DELETE in one tx; unlink afterwards (outside the tx). 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 files WHERE app_id = $1 AND collection = $2 AND id = $3 \ FOR UPDATE", ) .bind(app_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 files WHERE app_id = $1 AND collection = $2 AND id = $3") .bind(app_id.into_inner()) .bind(collection) .bind(id) .execute(&mut *tx) .await?; tx.commit().await?; // Row is gone; unlink the bytes. A failure here leaves an orphan // file (reclaimed by a future sweep) — not fatal. let path = self.final_path(app_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, "files: unlink after delete failed (orphan)"); } } Ok(Some(row.into_meta())) } async fn list( &self, app_id: AppId, 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 files \ WHERE app_id = $1 AND collection = $2 \ AND ($3::uuid IS NULL OR id > $3) \ ORDER BY id ASC \ LIMIT $4", ) .bind(app_id.into_inner()) .bind(collection) .bind(last_id) .bind(take) .fetch_all(&self.pool) .await?; let mut files: Vec = rows.into_iter().map(FileRow::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 }) } } // ---------------------------------------------------------------------------- // Helpers // ---------------------------------------------------------------------------- fn io_err(e: std::io::Error) -> FilesRepoError { FilesRepoError::Io(e.to_string()) } /// `create_dir_all` with `0o700` on the created tree (Unix). On other /// platforms it falls back to the default permissions. fn create_dir_all_secure(dir: &Path) -> Result<(), FilesRepoError> { #[cfg(unix)] { use std::os::unix::fs::DirBuilderExt as _; std::fs::DirBuilder::new() .recursive(true) .mode(0o700) .create(dir) .map_err(io_err) } #[cfg(not(unix))] { std::fs::create_dir_all(dir).map_err(io_err) } } fn hex_lower(bytes: &[u8]) -> String { let mut s = String::with_capacity(bytes.len() * 2); for b in bytes { use std::fmt::Write as _; let _ = write!(s, "{b:02x}"); } s } fn encode_cursor(last_id: Uuid) -> String { URL_SAFE_NO_PAD.encode(last_id.to_string().as_bytes()) } fn decode_cursor(cursor: &str) -> Result { let bytes = URL_SAFE_NO_PAD .decode(cursor) .map_err(|_| FilesRepoError::InvalidCursor)?; let s = String::from_utf8(bytes).map_err(|_| FilesRepoError::InvalidCursor)?; Uuid::parse_str(&s).map_err(|_| FilesRepoError::InvalidCursor) } #[derive(sqlx::FromRow)] struct FileRow { id: Uuid, collection: String, name: String, content_type: String, size_bytes: i64, checksum_sha256: String, created_at: DateTime, updated_at: DateTime, } impl FileRow { 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, } } } #[cfg(test)] mod tests { use super::*; #[test] fn hex_lower_matches_known_sha256_vector() { // SHA-256("abc") — NIST known-answer vector. let mut h = Sha256::new(); h.update(b"abc"); assert_eq!( hex_lower(&h.finalize()), "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad" ); } #[test] fn hex_lower_of_empty_is_known_vector() { let mut h = Sha256::new(); h.update(b""); assert_eq!( hex_lower(&h.finalize()), "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" ); } #[test] fn cursor_round_trips() { let id = Uuid::new_v4(); let enc = encode_cursor(id); assert_eq!(decode_cursor(&enc).unwrap(), id); assert!(matches!( decode_cursor("!!not-base64!!"), Err(FilesRepoError::InvalidCursor) )); } #[test] fn guard_collection_rejects_traversal() { assert!(FsFilesRepo::guard_collection("avatars").is_ok()); assert!(FsFilesRepo::guard_collection("a/b").is_err()); assert!(FsFilesRepo::guard_collection("..").is_err()); assert!(FsFilesRepo::guard_collection("a..b").is_err()); assert!(FsFilesRepo::guard_collection("").is_err()); assert!(FsFilesRepo::guard_collection("a\0b").is_err()); } #[test] fn config_from_env_defaults_are_conservative() { let c = FilesConfig::conservative(); assert_eq!(c.max_file_size_bytes, DEFAULT_MAX_FILE_SIZE_BYTES); assert_eq!(c.root, PathBuf::from(DEFAULT_FILES_ROOT)); } // ------------------------------------------------------------------ // Tempdir-backed filesystem mechanics — exercise the atomic write, // single-pass checksum, and checksum-on-read tamper detection // without needing a Postgres pool. // ------------------------------------------------------------------ use picloud_shared::AppId; /// Process-unique scratch dir under the system temp dir. Cleaned up /// by each test via `remove_dir_all`. fn unique_tmp_root() -> PathBuf { let seq = TMP_COUNTER.fetch_add(1, Ordering::Relaxed); let dir = std::env::temp_dir().join(format!("picloud-files-test-{}-{seq}", std::process::id())); std::fs::create_dir_all(&dir).unwrap(); dir } #[test] fn write_atomic_then_read_verify_round_trips() { let root = unique_tmp_root(); let app = AppId::new(); let id = Uuid::new_v4(); let bytes = b"hello picloud files".to_vec(); let checksum = write_atomic_at(&root, app, "avatars", id, &bytes).unwrap(); // Single-pass checksum matches an independent hash of the bytes. let mut h = Sha256::new(); h.update(&bytes); assert_eq!(checksum, hex_lower(&h.finalize())); let read = read_verify_at(&root, app, "avatars", id, &checksum).unwrap(); assert_eq!(read, bytes); std::fs::remove_dir_all(&root).ok(); } #[test] fn read_verify_detects_tampering_as_corrupted() { let root = unique_tmp_root(); let app = AppId::new(); let id = Uuid::new_v4(); let checksum = write_atomic_at(&root, app, "c", id, b"original").unwrap(); // Mutate the bytes behind the repo's back. let path = final_path_at(&root, app, "c", id); std::fs::write(&path, b"tampered").unwrap(); let err = read_verify_at(&root, app, "c", id, &checksum).unwrap_err(); assert!(matches!(err, FilesRepoError::Corrupted)); std::fs::remove_dir_all(&root).ok(); } #[test] fn read_verify_missing_bytes_is_corrupted() { let root = unique_tmp_root(); let app = AppId::new(); let id = Uuid::new_v4(); // No write — the file never existed. let err = read_verify_at(&root, app, "c", id, "deadbeef").unwrap_err(); assert!(matches!(err, FilesRepoError::Corrupted)); std::fs::remove_dir_all(&root).ok(); } #[test] fn atomic_write_leaves_no_tmp_file_after_success() { let root = unique_tmp_root(); let app = AppId::new(); let id = Uuid::new_v4(); write_atomic_at(&root, app, "c", id, b"data").unwrap(); let id_str = id.to_string(); let dir = shard_dir_at(&root, app, "c", &id_str); let entries: Vec<_> = std::fs::read_dir(&dir) .unwrap() .filter_map(Result::ok) .map(|e| e.file_name().to_string_lossy().into_owned()) .collect(); // Exactly the final file is visible — no `*.tmp.*` orphan. assert_eq!(entries, vec![id_str]); assert!(!entries.iter().any(|n| n.contains(".tmp."))); std::fs::remove_dir_all(&root).ok(); } #[test] fn id_shard_uses_first_two_chars() { let root = PathBuf::from("/tmp/x"); let app = AppId::new(); let id = Uuid::new_v4(); let id_str = id.to_string(); let path = final_path_at(&root, app, "col", id); let shard = &id_str[..2]; assert!(path .to_string_lossy() .contains(&format!("/col/{shard}/{id_str}"))); } #[cfg(unix)] #[test] fn shard_tree_created_with_0700() { use std::os::unix::fs::PermissionsExt as _; let root = unique_tmp_root(); let app = AppId::new(); let id = Uuid::new_v4(); write_atomic_at(&root, app, "c", id, b"data").unwrap(); let id_str = id.to_string(); let dir = shard_dir_at(&root, app, "c", &id_str); let mode = std::fs::metadata(&dir).unwrap().permissions().mode(); assert_eq!(mode & 0o777, 0o700, "shard dir should be 0o700"); std::fs::remove_dir_all(&root).ok(); } }