cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

perf: fix issues #1,#2,#3,#13 from performance audit

Browse Source 
- #1  image_transcode: dedicated rayon pool + moka cache (no tokio blocking)
- #2  ltree materialized paths: eliminate N+1 folder/file queries
- #3  compute_content_hash: read blob_hash column instead of loading file into RAM
- #13 event delegation + DocumentFragment: ~15 delegated listeners replace ~19k per-item listeners

Backend: 82/82 tests pass, cargo check clean.
Frontend: ui.js and app.js syntax-validated via Node.js.
This commit is contained in:
Dionisio
2026-02-16 13:10:44 +01:00
parent e69de1643f
commit b7bd656a43
16 changed files with 949 additions and 905 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/infrastructure/repositories/pg/file_blob_read_repository.rs
+112 -174
View File
@@ -3,6 +3,9 @@
//! Implements `FileReadPort` using:
//! - `storage.files` table for metadata lookups
//! - `DedupPort` for reading content-addressable blobs from the filesystem
//!
//! File paths are resolved by JOINing with `storage.folders.path` (the
//! materialized path column), so no recursive CTEs or N+1 queries are needed.
use async_trait::async_trait;
use bytes::Bytes;
@@ -15,18 +18,14 @@ use crate::application::ports::dedup_ports::DedupPort;
use crate::application::ports::storage_ports::FileReadPort;
use crate::common::errors::DomainError;
use crate::domain::entities::file::File;
use crate::domain::repositories::folder_repository::FolderRepository;
use crate::domain::services::path_service::StoragePath;
use super::folder_db_repository::FolderDbRepository;
/// File read repository backed by PostgreSQL metadata + blob storage.
pub struct FileBlobReadRepository {
pool: Arc<PgPool>,
dedup: Arc<dyn DedupPort>,
folder_repo: Arc<FolderDbRepository>,
/// Lightweight cache: file_id → blob_hash.
/// Populated by `get_file()`, consumed by `get_blob_hash()`.
/// Populated by `get_file()`, consumed by `resolve_blob_hash()`.
/// Avoids an extra SQL round-trip on the hot download path.
hash_cache: std::sync::Mutex<HashMap<String, String>>,
}
@@ -35,42 +34,35 @@ impl FileBlobReadRepository {
pub fn new(
pool: Arc<PgPool>,
dedup: Arc<dyn DedupPort>,
folder_repo: Arc<FolderDbRepository>,
_folder_repo: Arc<super::folder_db_repository::FolderDbRepository>,
) -> Self {
Self {
pool,
dedup,
folder_repo,
hash_cache: std::sync::Mutex::new(HashMap::new()),
}
}
/// Build a virtual StoragePath for a file.
async fn build_file_path(
&self,
folder_id: Option<&str>,
file_name: &str,
) -> Result<StoragePath, DomainError> {
if let Some(fid) = folder_id {
let folder_path = self.folder_repo.get_folder_path(fid).await?;
Ok(folder_path.join(file_name))
} else {
Ok(StoragePath::from_string(file_name))
/// Build a `StoragePath` from the materialized folder path + file name.
fn make_file_path(folder_path: Option<&str>, file_name: &str) -> StoragePath {
match folder_path {
Some(fp) if !fp.is_empty() => StoragePath::from_string(&format!("{fp}/{file_name}")),
_ => StoragePath::from_string(file_name),
}
}
#[allow(clippy::too_many_arguments)]
async fn row_to_file(
&self,
fn row_to_file(
id: String,
name: String,
folder_id: Option<String>,
folder_path: Option<String>,
size: i64,
mime_type: String,
created_at: i64,
modified_at: i64,
) -> Result<File, DomainError> {
let storage_path = self.build_file_path(folder_id.as_deref(), &name).await?;
let storage_path = Self::make_file_path(folder_path.as_deref(), &name);
File::with_timestamps(
id,
name,
@@ -84,9 +76,9 @@ impl FileBlobReadRepository {
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("entity: {e}")))
}
/// Get the blob hash for a file.
/// Resolve the blob hash for a file (internal helper).
/// Checks the in-memory cache first (populated by `get_file`).
async fn get_blob_hash(&self, file_id: &str) -> Result<String, DomainError> {
async fn resolve_blob_hash(&self, file_id: &str) -> Result<String, DomainError> {
// Fast path: already cached from a prior get_file call
if let Some(hash) = self.hash_cache.lock().unwrap().remove(file_id) {
return Ok(hash);
@@ -109,23 +101,26 @@ impl FileReadPort for FileBlobReadRepository {
let row = sqlx::query_as::<
_,
(
String,
String,
Option<String>,
i64,
String,
i64,
i64,
String,
String, // id
String, // name
Option<String>, // folder_id
Option<String>, // folder path
i64, // size
String, // mime_type
i64, // created_at
i64, // updated_at
String, // blob_hash
),
>(
r#"
SELECT id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
blob_hash
FROM storage.files
WHERE id = $1::uuid AND NOT is_trashed
SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.id = $1::uuid AND NOT fi.is_trashed
"#,
)
.bind(id)
@@ -139,23 +134,24 @@ impl FileReadPort for FileBlobReadRepository {
self.hash_cache
.lock()
.unwrap()
.insert(id.to_string(), row.7.clone());
.insert(id.to_string(), row.8.clone());
self.row_to_file(row.0, row.1, row.2, row.3, row.4, row.5, row.6)
.await
Self::row_to_file(row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7)
}
async fn list_files(&self, folder_id: Option<&str>) -> Result<Vec<File>, DomainError> {
let rows: Vec<(String, String, Option<String>, i64, String, i64, i64)> =
let rows: Vec<(String, String, Option<String>, Option<String>, i64, String, i64, i64)> =
if let Some(fid) = folder_id {
sqlx::query_as(
r#"
SELECT id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.files
WHERE folder_id = $1::uuid AND NOT is_trashed
ORDER BY name
SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.folder_id = $1::uuid AND NOT fi.is_trashed
ORDER BY fi.name
"#,
)
.bind(fid)
@@ -164,12 +160,14 @@ impl FileReadPort for FileBlobReadRepository {
} else {
sqlx::query_as(
r#"
SELECT id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.files
WHERE folder_id IS NULL AND NOT is_trashed
ORDER BY name
SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.folder_id IS NULL AND NOT fi.is_trashed
ORDER BY fi.name
"#,
)
.fetch_all(self.pool.as_ref())
@@ -177,38 +175,19 @@ impl FileReadPort for FileBlobReadRepository {
}
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("list: {e}")))?;
// ── N+1 fix: resolve the folder path ONCE (all rows share
// the same folder_id when listing a specific folder). ──
let shared_folder_path = if let Some(fid) = folder_id {
Some(self.folder_repo.get_folder_path(fid).await?)
} else {
None
};
rows.into_iter()
.map(|(id, name, fid, fpath, size, mime, ca, ma)| {
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma)
})
.collect()
}
let mut files = Vec::with_capacity(rows.len());
for (id, name, fid, size, mime, ca, ma) in rows {
let storage_path = match &shared_folder_path {
Some(fp) => fp.join(&name),
None => StoragePath::from_string(&name),
};
let file = File::with_timestamps(
id,
name,
storage_path,
size as u64,
mime,
fid,
ca as u64,
ma as u64,
)
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("entity: {e}")))?;
files.push(file);
}
Ok(files)
async fn get_blob_hash(&self, file_id: &str) -> Result<String, DomainError> {
self.resolve_blob_hash(file_id).await
}
async fn get_file_content(&self, id: &str) -> Result<Vec<u8>, DomainError> {
let blob_hash = self.get_blob_hash(id).await?;
let blob_hash = self.resolve_blob_hash(id).await?;
self.dedup.read_blob(&blob_hash).await
}
@@ -218,7 +197,7 @@ impl FileReadPort for FileBlobReadRepository {
) -> Result<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>, DomainError> {
// True streaming: reads the blob file in 64 KB chunks.
// Memory usage is ~64 KB regardless of file size.
let blob_hash = self.get_blob_hash(id).await?;
let blob_hash = self.resolve_blob_hash(id).await?;
let stream = self.dedup.read_blob_stream(&blob_hash).await?;
Ok(Box::new(stream))
}
@@ -231,7 +210,7 @@ impl FileReadPort for FileBlobReadRepository {
) -> Result<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>, DomainError> {
// True range streaming: seeks to `start` and reads only the requested range.
// A 1 MB range on a 1 GB file uses ~64 KB of RAM.
let blob_hash = self.get_blob_hash(id).await?;
let blob_hash = self.resolve_blob_hash(id).await?;
let stream = self
.dedup
.read_blob_range_stream(&blob_hash, start, end)
@@ -242,16 +221,17 @@ impl FileReadPort for FileBlobReadRepository {
async fn get_file_mmap(&self, id: &str) -> Result<Bytes, DomainError> {
// For RPi targets, mmap is less beneficial than streaming.
// Keep as a fallback that loads content for small/medium files.
let blob_hash = self.get_blob_hash(id).await?;
let blob_hash = self.resolve_blob_hash(id).await?;
self.dedup.read_blob_bytes(&blob_hash).await
}
async fn get_file_path(&self, id: &str) -> Result<StoragePath, DomainError> {
let row = sqlx::query_as::<_, (String, Option<String>)>(
r#"
SELECT name, folder_id::text
FROM storage.files
WHERE id = $1::uuid AND NOT is_trashed
SELECT fi.name, fo.path
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.id = $1::uuid AND NOT fi.is_trashed
"#,
)
.bind(id)
@@ -260,11 +240,10 @@ impl FileReadPort for FileBlobReadRepository {
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("path: {e}")))?
.ok_or_else(|| DomainError::not_found("File", id))?;
self.build_file_path(row.1.as_deref(), &row.0).await
Ok(Self::make_file_path(row.1.as_deref(), &row.0))
}
async fn get_parent_folder_id(&self, path: &str) -> Result<String, DomainError> {
// Walk the path to find the parent folder, searching by folder names
let path = path.trim_start_matches('/').trim_end_matches('/');
let segments: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();
@@ -272,49 +251,30 @@ impl FileReadPort for FileBlobReadRepository {
return Err(DomainError::not_found("Folder", "empty path"));
}
// For path "a/b/c/file.txt", the parent folder path is "a/b/c"
// But we don't know which part is folders vs filename.
// Walk segments trying to find matching folders.
let mut current_parent: Option<String> = None;
// For a path like "Home - user/Docs/file.txt", the parent folder path
// is everything except the last segment: "Home - user/Docs"
// We try the longest folder path first.
let folder_path = segments[..segments.len() - 1].join("/");
for segment in &segments {
let row = if let Some(ref pid) = current_parent {
sqlx::query_as::<_, (String,)>(
r#"
SELECT id::text FROM storage.folders
WHERE name = $1 AND parent_id = $2::uuid AND NOT is_trashed
"#,
)
.bind(segment)
.bind(pid)
.fetch_optional(self.pool.as_ref())
.await
} else {
sqlx::query_as::<_, (String,)>(
r#"
SELECT id::text FROM storage.folders
WHERE name = $1 AND parent_id IS NULL AND NOT is_trashed
"#,
)
.bind(segment)
.fetch_optional(self.pool.as_ref())
.await
}
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("path walk: {e}")))?;
match row {
Some(r) => current_parent = Some(r.0),
None => break, // This segment is not a folder → it's the filename
}
if folder_path.is_empty() {
return Err(DomainError::not_found(
"Folder",
format!("parent for path: {path}"),
));
}
current_parent
.ok_or_else(|| DomainError::not_found("Folder", format!("parent for path: {path}")))
sqlx::query_scalar::<_, String>(
"SELECT id::text FROM storage.folders WHERE path = $1 AND NOT is_trashed",
)
.bind(&folder_path)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("parent lookup: {e}")))?
.ok_or_else(|| DomainError::not_found("Folder", format!("parent for path: {path}")))
}
/// Direct SQL lookup: split path into folder segments + filename,
/// walk the folder hierarchy, then match the file by name + folder_id.
/// O(depth) queries instead of O(total_files).
/// Direct SQL lookup using materialized folder paths.
/// O(1) query instead of O(depth) folder walk.
async fn find_file_by_path(&self, path: &str) -> Result<Option<File>, DomainError> {
let path = path.trim_start_matches('/').trim_end_matches('/');
let segments: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();
@@ -323,62 +283,40 @@ impl FileReadPort for FileBlobReadRepository {
return Ok(None);
}
// Last segment is the filename, preceding segments are folders
// Last segment is the filename, preceding segments are the folder path
let filename = segments[segments.len() - 1];
let folder_segments = &segments[..segments.len() - 1];
let folder_path = segments[..segments.len() - 1].join("/");
// Walk folder hierarchy to find parent folder_id
let mut current_parent: Option<String> = None;
for segment in folder_segments {
let row = if let Some(ref pid) = current_parent {
sqlx::query_scalar::<_, String>(
"SELECT id::text FROM storage.folders WHERE name = $1 AND parent_id = $2::uuid AND NOT is_trashed",
)
.bind(segment)
.bind(pid)
.fetch_optional(self.pool.as_ref())
.await
} else {
sqlx::query_scalar::<_, String>(
"SELECT id::text FROM storage.folders WHERE name = $1 AND parent_id IS NULL AND NOT is_trashed",
)
.bind(segment)
.fetch_optional(self.pool.as_ref())
.await
}
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("path walk: {e}")))?;
match row {
Some(id) => current_parent = Some(id),
None => return Ok(None), // Folder not found → file doesn't exist at this path
}
}
// Now find the file by name + folder_id
let row = if let Some(ref fid) = current_parent {
sqlx::query_as::<_, (String, String, Option<String>, i64, String, i64, i64)>(
let row = if folder_path.is_empty() {
// File at root level (no parent folder)
sqlx::query_as::<_, (String, String, Option<String>, Option<String>, i64, String, i64, i64)>(
r#"
SELECT id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.files
WHERE name = $1 AND folder_id = $2::uuid AND NOT is_trashed
SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.name = $1 AND fi.folder_id IS NULL AND NOT fi.is_trashed
"#,
)
.bind(filename)
.bind(fid)
.fetch_optional(self.pool.as_ref())
.await
} else {
sqlx::query_as::<_, (String, String, Option<String>, i64, String, i64, i64)>(
// File inside a folder — look up by folder path + filename
sqlx::query_as::<_, (String, String, Option<String>, Option<String>, i64, String, i64, i64)>(
r#"
SELECT id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.files
WHERE name = $1 AND folder_id IS NULL AND NOT is_trashed
SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint
FROM storage.files fi
JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fo.path = $1 AND fi.name = $2 AND NOT fi.is_trashed
"#,
)
.bind(&folder_path)
.bind(filename)
.fetch_optional(self.pool.as_ref())
.await
@@ -386,9 +324,9 @@ impl FileReadPort for FileBlobReadRepository {
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("find file: {e}")))?;
match row {
Some(r) => Ok(Some(
self.row_to_file(r.0, r.1, r.2, r.3, r.4, r.5, r.6).await?,
)),
Some(r) => Ok(Some(Self::row_to_file(
r.0, r.1, r.2, r.3, r.4, r.5, r.6, r.7,
)?)),
None => Ok(None),
}
}
src/infrastructure/repositories/pg/file_blob_write_repository.rs
+54 -34
View File
@@ -3,6 +3,9 @@
//! Implements `FileWritePort` using:
//! - `storage.files` table for metadata
//! - `DedupPort` for content-addressable blob storage on the filesystem
//!
//! File paths are resolved by querying the materialized `storage.folders.path`
//! column (O(1) per lookup), so no recursive CTEs are needed.
use async_trait::async_trait;
use sqlx::PgPool;
@@ -13,7 +16,6 @@ use crate::application::ports::dedup_ports::DedupPort;
use crate::application::ports::storage_ports::FileWritePort;
use crate::common::errors::DomainError;
use crate::domain::entities::file::File;
use crate::domain::repositories::folder_repository::FolderRepository;
use crate::domain::services::path_service::StoragePath;
use super::folder_db_repository::FolderDbRepository;
@@ -38,32 +40,49 @@ impl FileBlobWriteRepository {
}
}
/// Build a virtual StoragePath for a file from its DB metadata.
async fn build_file_path(
/// Build a `StoragePath` from the materialized folder path + file name.
fn make_file_path(folder_path: Option<&str>, file_name: &str) -> StoragePath {
match folder_path {
Some(fp) if !fp.is_empty() => StoragePath::from_string(&format!("{fp}/{file_name}")),
_ => StoragePath::from_string(file_name),
}
}
/// Look up the materialized folder path. O(1) — no recursive CTE.
async fn lookup_folder_path(
&self,
folder_id: Option<&str>,
file_name: &str,
) -> Result<StoragePath, DomainError> {
if let Some(fid) = folder_id {
let folder_path = self.folder_repo.get_folder_path(fid).await?;
Ok(folder_path.join(file_name))
} else {
Ok(StoragePath::from_string(file_name))
) -> Result<Option<String>, DomainError> {
match folder_id {
Some(fid) => {
let path: String = sqlx::query_scalar(
"SELECT path FROM storage.folders WHERE id = $1::uuid",
)
.bind(fid)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| {
DomainError::internal_error("FileBlobWrite", format!("folder path: {e}"))
})?
.ok_or_else(|| DomainError::not_found("Folder", fid))?;
Ok(Some(path))
}
None => Ok(None),
}
}
#[allow(clippy::too_many_arguments)]
async fn row_to_file(
&self,
fn row_to_file(
id: String,
name: String,
folder_id: Option<String>,
folder_path: Option<String>,
size: i64,
mime_type: String,
created_at: i64,
modified_at: i64,
) -> Result<File, DomainError> {
let storage_path = self.build_file_path(folder_id.as_deref(), &name).await?;
let storage_path = Self::make_file_path(folder_path.as_deref(), &name);
File::with_timestamps(
id,
name,
@@ -159,8 +178,8 @@ impl FileWritePort for FileBlobWriteRepository {
&blob_hash[..12]
);
self.row_to_file(row.0, name, folder_id, size, content_type, row.1, row.2)
.await
let folder_path = self.lookup_folder_path(folder_id.as_deref()).await?;
Self::row_to_file(row.0, name, folder_id, folder_path, size, content_type, row.1, row.2)
}
async fn save_file_from_temp(
@@ -232,16 +251,17 @@ impl FileWritePort for FileBlobWriteRepository {
&blob_hash[..12]
);
self.row_to_file(
let folder_path = self.lookup_folder_path(folder_id.as_deref()).await?;
Self::row_to_file(
row.0,
name,
folder_id,
folder_path,
size as i64,
content_type,
row.1,
row.2,
)
.await
}
async fn move_file(
@@ -267,8 +287,8 @@ impl FileWritePort for FileBlobWriteRepository {
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("move: {e}")))?
.ok_or_else(|| DomainError::not_found("File", file_id))?;
self.row_to_file(row.0, row.1, row.2, row.3, row.4, row.5, row.6)
.await
let folder_path = self.lookup_folder_path(row.2.as_deref()).await?;
Self::row_to_file(row.0, row.1, row.2, folder_path, row.3, row.4, row.5, row.6)
}
async fn copy_file(
@@ -350,8 +370,8 @@ impl FileWritePort for FileBlobWriteRepository {
&blob_hash[..12]
);
self.row_to_file(row.0, row.1, row.2, row.3, row.4, row.5, row.6)
.await
let folder_path = self.lookup_folder_path(row.2.as_deref()).await?;
Self::row_to_file(row.0, row.1, row.2, folder_path, row.3, row.4, row.5, row.6)
}
async fn rename_file(&self, file_id: &str, new_name: &str) -> Result<File, DomainError> {
@@ -379,8 +399,8 @@ impl FileWritePort for FileBlobWriteRepository {
})?
.ok_or_else(|| DomainError::not_found("File", file_id))?;
self.row_to_file(row.0, row.1, row.2, row.3, row.4, row.5, row.6)
.await
let folder_path = self.lookup_folder_path(row.2.as_deref()).await?;
Self::row_to_file(row.0, row.1, row.2, folder_path, row.3, row.4, row.5, row.6)
}
async fn delete_file(&self, id: &str) -> Result<(), DomainError> {
@@ -502,17 +522,17 @@ impl FileWritePort for FileBlobWriteRepository {
.await
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("deferred: {e}")))?;
let file = self
.row_to_file(
row.0.clone(),
name,
folder_id,
size as i64,
content_type,
row.1,
row.2,
)
.await?;
let folder_path = self.lookup_folder_path(folder_id.as_deref()).await?;
let file = Self::row_to_file(
row.0.clone(),
name,
folder_id,
folder_path,
size as i64,
content_type,
row.1,
row.2,
)?;
// The target_path is not meaningful for blob storage (content goes to .blobs/)
// but the WriteBehindCache API requires it. We return a synthetic path.
src/infrastructure/repositories/pg/folder_db_repository.rs
+104 -120
View File
@@ -3,6 +3,10 @@
//! Implements `FolderRepository` (and thus `FolderStoragePort`) using the
//! `storage.folders` table. Folders are purely virtual — no physical
//! directories are created on the filesystem.
//!
//! Folder paths are **materialized** in a `path TEXT` column maintained by
//! database triggers, so reading a folder's full path is always O(1) — no
//! recursive CTEs or N+1 queries.
use async_trait::async_trait;
use sqlx::PgPool;
@@ -41,43 +45,20 @@ impl FolderDbRepository {
// ── helpers ──────────────────────────────────────────────────
/// Build the full virtual path for a folder by walking up the `parent_id` chain.
async fn build_folder_path(&self, folder_id: &str) -> Result<StoragePath, DomainError> {
let rows = sqlx::query_as::<_, (String, i32)>(
r#"
WITH RECURSIVE ancestors AS (
SELECT id, name, parent_id, 0 AS depth
FROM storage.folders
WHERE id = $1::uuid
UNION ALL
SELECT f.id, f.name, f.parent_id, a.depth + 1
FROM storage.folders f
JOIN ancestors a ON f.id = a.parent_id
)
SELECT name, depth FROM ancestors ORDER BY depth DESC
"#,
)
.bind(folder_id)
.fetch_all(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("path query: {e}")))?;
let path_parts: Vec<&str> = rows.iter().map(|(name, _)| name.as_str()).collect();
let path_str = path_parts.join("/");
Ok(StoragePath::from_string(&path_str))
}
/// Convert a database row into a `Folder` domain entity.
async fn row_to_folder(
&self,
///
/// The `path` comes directly from the materialized `path` column — no
/// extra queries needed.
fn row_to_folder(
id: String,
name: String,
path: String,
parent_id: Option<String>,
user_id: Option<String>,
created_at: i64,
modified_at: i64,
) -> Result<Folder, DomainError> {
let storage_path = self.build_folder_path(&id).await?;
let storage_path = StoragePath::from_string(&path);
Folder::with_timestamps_and_owner(
id,
name,
@@ -117,11 +98,12 @@ impl FolderRepository for FolderDbRepository {
));
};
let row = sqlx::query_as::<_, (String, i64, i64)>(
let row = sqlx::query_as::<_, (String, String, i64, i64)>(
r#"
INSERT INTO storage.folders (name, parent_id, user_id)
VALUES ($1, $2::uuid, $3)
RETURNING id::text,
path,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
"#,
@@ -143,14 +125,13 @@ impl FolderRepository for FolderDbRepository {
DomainError::internal_error("FolderDb", format!("insert: {e}"))
})?;
self.row_to_folder(row.0, name, parent_id, Some(user_id), row.1, row.2)
.await
Self::row_to_folder(row.0, name, row.1, parent_id, Some(user_id), row.2, row.3)
}
async fn get_folder(&self, id: &str) -> Result<Folder, DomainError> {
let row = sqlx::query_as::<_, (String, String, Option<String>, String, i64, i64)>(
let row = sqlx::query_as::<_, (String, String, String, Option<String>, String, i64, i64)>(
r#"
SELECT id::text, name, parent_id::text, user_id,
SELECT id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
@@ -163,61 +144,41 @@ impl FolderRepository for FolderDbRepository {
.map_err(|e| DomainError::internal_error("FolderDb", format!("get: {e}")))?
.ok_or_else(|| DomainError::not_found("Folder", id))?;
self.row_to_folder(row.0, row.1, row.2, Some(row.3), row.4, row.5).await
Self::row_to_folder(row.0, row.1, row.2, row.3, Some(row.4), row.5, row.6)
}
async fn get_folder_by_path(&self, storage_path: &StoragePath) -> Result<Folder, DomainError> {
// Walk the path segments to find the folder.
let path_str = storage_path.to_string();
let segments: Vec<&str> = path_str.split('/').filter(|s| !s.is_empty()).collect();
// Strip leading '/' if present — DB stores "Home - user/Docs", not "/Home - user/Docs"
let lookup = path_str.strip_prefix('/').unwrap_or(&path_str);
if segments.is_empty() {
if lookup.is_empty() {
return Err(DomainError::not_found("Folder", "empty path"));
}
let mut current_parent: Option<String> = None;
let mut current_id = String::new();
let row = sqlx::query_as::<_, (String, String, String, Option<String>, String, i64, i64)>(
r#"
SELECT id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
WHERE path = $1 AND NOT is_trashed
"#,
)
.bind(lookup)
.fetch_optional(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("path lookup: {e}")))?
.ok_or_else(|| DomainError::not_found("Folder", lookup))?;
for segment in &segments {
let row = if let Some(ref pid) = current_parent {
sqlx::query_as::<_, (String,)>(
r#"
SELECT id::text FROM storage.folders
WHERE name = $1 AND parent_id = $2::uuid AND NOT is_trashed
"#,
)
.bind(segment)
.bind(pid)
.fetch_optional(self.pool())
.await
} else {
sqlx::query_as::<_, (String,)>(
r#"
SELECT id::text FROM storage.folders
WHERE name = $1 AND parent_id IS NULL AND NOT is_trashed
"#,
)
.bind(segment)
.fetch_optional(self.pool())
.await
}
.map_err(|e| DomainError::internal_error("FolderDb", format!("path walk: {e}")))?
.ok_or_else(|| {
DomainError::not_found("Folder", format!("segment '{segment}' in path"))
})?;
current_id = row.0;
current_parent = Some(current_id.clone());
}
self.get_folder(&current_id).await
Self::row_to_folder(row.0, row.1, row.2, row.3, Some(row.4), row.5, row.6)
}
async fn list_folders(&self, parent_id: Option<&str>) -> Result<Vec<Folder>, DomainError> {
let rows: Vec<(String, String, Option<String>, String, i64, i64)> = if let Some(pid) = parent_id {
let rows: Vec<(String, String, String, Option<String>, String, i64, i64)> = if let Some(pid) = parent_id {
sqlx::query_as(
r#"
SELECT id::text, name, parent_id::text, user_id,
SELECT id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
@@ -231,7 +192,7 @@ impl FolderRepository for FolderDbRepository {
} else {
sqlx::query_as(
r#"
SELECT id::text, name, parent_id::text, user_id,
SELECT id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
@@ -244,11 +205,11 @@ impl FolderRepository for FolderDbRepository {
}
.map_err(|e| DomainError::internal_error("FolderDb", format!("list: {e}")))?;
let mut folders = Vec::with_capacity(rows.len());
for (id, name, pid, uid, ca, ma) in rows {
folders.push(self.row_to_folder(id, name, pid, Some(uid), ca, ma).await?);
}
Ok(folders)
rows.into_iter()
.map(|(id, name, path, pid, uid, ca, ma)| {
Self::row_to_folder(id, name, path, pid, Some(uid), ca, ma)
})
.collect()
}
async fn list_folders_by_owner(
@@ -256,12 +217,10 @@ impl FolderRepository for FolderDbRepository {
parent_id: Option<&str>,
owner_id: &str,
) -> Result<Vec<Folder>, DomainError> {
let rows: Vec<(String, String, Option<String>, String, i64, i64)> = if let Some(pid) = parent_id {
// For sub-folders the owner is implicit (parent belongs to user),
// but we still filter to be safe.
let rows: Vec<(String, String, String, Option<String>, String, i64, i64)> = if let Some(pid) = parent_id {
sqlx::query_as(
r#"
SELECT id::text, name, parent_id::text, user_id,
SELECT id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
@@ -274,10 +233,9 @@ impl FolderRepository for FolderDbRepository {
.fetch_all(self.pool())
.await
} else {
// Root-level: only this user's home folders
sqlx::query_as(
r#"
SELECT id::text, name, parent_id::text, user_id,
SELECT id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
@@ -291,11 +249,11 @@ impl FolderRepository for FolderDbRepository {
}
.map_err(|e| DomainError::internal_error("FolderDb", format!("list_by_owner: {e}")))?;
let mut folders = Vec::with_capacity(rows.len());
for (id, name, pid, uid, ca, ma) in rows {
folders.push(self.row_to_folder(id, name, pid, Some(uid), ca, ma).await?);
}
Ok(folders)
rows.into_iter()
.map(|(id, name, path, pid, uid, ca, ma)| {
Self::row_to_folder(id, name, path, pid, Some(uid), ca, ma)
})
.collect()
}
async fn list_folders_paginated(
@@ -326,10 +284,10 @@ impl FolderRepository for FolderDbRepository {
None
};
let rows: Vec<(String, String, Option<String>, String, i64, i64)> = if let Some(pid) = parent_id {
let rows: Vec<(String, String, String, Option<String>, String, i64, i64)> = if let Some(pid) = parent_id {
sqlx::query_as(
r#"
SELECT id::text, name, parent_id::text, user_id,
SELECT id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
@@ -346,7 +304,7 @@ impl FolderRepository for FolderDbRepository {
} else {
sqlx::query_as(
r#"
SELECT id::text, name, parent_id::text, user_id,
SELECT id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
@@ -362,11 +320,13 @@ impl FolderRepository for FolderDbRepository {
}
.map_err(|e| DomainError::internal_error("FolderDb", format!("paginate: {e}")))?;
let mut folders = Vec::with_capacity(rows.len());
for (id, name, pid, uid, ca, ma) in rows {
folders.push(self.row_to_folder(id, name, pid, Some(uid), ca, ma).await?);
}
Ok((folders, total))
let folders: Result<Vec<Folder>, DomainError> = rows
.into_iter()
.map(|(id, name, path, pid, uid, ca, ma)| {
Self::row_to_folder(id, name, path, pid, Some(uid), ca, ma)
})
.collect();
Ok((folders?, total))
}
async fn list_folders_by_owner_paginated(
@@ -400,10 +360,10 @@ impl FolderRepository for FolderDbRepository {
None
};
let rows: Vec<(String, String, Option<String>, String, i64, i64)> = if let Some(pid) = parent_id {
let rows: Vec<(String, String, String, Option<String>, String, i64, i64)> = if let Some(pid) = parent_id {
sqlx::query_as(
r#"
SELECT id::text, name, parent_id::text, user_id,
SELECT id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
@@ -421,7 +381,7 @@ impl FolderRepository for FolderDbRepository {
} else {
sqlx::query_as(
r#"
SELECT id::text, name, parent_id::text, user_id,
SELECT id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
@@ -438,14 +398,18 @@ impl FolderRepository for FolderDbRepository {
}
.map_err(|e| DomainError::internal_error("FolderDb", format!("paginate_by_owner: {e}")))?;
let mut folders = Vec::with_capacity(rows.len());
for (id, name, pid, uid, ca, ma) in rows {
folders.push(self.row_to_folder(id, name, pid, Some(uid), ca, ma).await?);
}
Ok((folders, total))
let folders: Result<Vec<Folder>, DomainError> = rows
.into_iter()
.map(|(id, name, path, pid, uid, ca, ma)| {
Self::row_to_folder(id, name, path, pid, Some(uid), ca, ma)
})
.collect();
Ok((folders?, total))
}
async fn rename_folder(&self, id: &str, new_name: String) -> Result<Folder, DomainError> {
// The BEFORE UPDATE trigger on `name` will recompute path/lpath
// and cascade to descendants automatically.
sqlx::query(
r#"
UPDATE storage.folders
@@ -474,6 +438,8 @@ impl FolderRepository for FolderDbRepository {
id: &str,
new_parent_id: Option<&str>,
) -> Result<Folder, DomainError> {
// The BEFORE UPDATE trigger on `parent_id` will recompute path/lpath
// and cascade to descendants automatically.
sqlx::query(
r#"
UPDATE storage.folders
@@ -505,16 +471,31 @@ impl FolderRepository for FolderDbRepository {
}
async fn folder_exists(&self, storage_path: &StoragePath) -> Result<bool, DomainError> {
// Try to find by walking the path
match self.get_folder_by_path(storage_path).await {
Ok(_) => Ok(true),
Err(e) if e.to_string().contains("not found") => Ok(false),
Err(e) => Err(e),
}
let path_str = storage_path.to_string();
let lookup = path_str.strip_prefix('/').unwrap_or(&path_str);
let exists: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM storage.folders WHERE path = $1 AND NOT is_trashed)",
)
.bind(lookup)
.fetch_one(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("exists: {e}")))?;
Ok(exists)
}
async fn get_folder_path(&self, id: &str) -> Result<StoragePath, DomainError> {
self.build_folder_path(id).await
let path: String = sqlx::query_scalar(
"SELECT path FROM storage.folders WHERE id = $1::uuid",
)
.bind(id)
.fetch_optional(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("get_path: {e}")))?
.ok_or_else(|| DomainError::not_found("Folder", id))?;
Ok(StoragePath::from_string(&path))
}
// ── Trash operations ──
@@ -566,6 +547,8 @@ impl FolderRepository for FolderDbRepository {
_original_path: &str,
) -> Result<(), DomainError> {
// Atomic CTE: restore folder + all descendant files in a single statement.
// The BEFORE UPDATE trigger on parent_id will recompute path/lpath
// automatically when original_parent_id is restored.
let result = sqlx::query_scalar::<_, i64>(
r#"
WITH restore_folder AS (
@@ -632,12 +615,13 @@ impl FolderDbRepository {
user_id: &str,
name: &str,
) -> Result<Folder, DomainError> {
let row = sqlx::query_as::<_, (String, i64, i64)>(
let row = sqlx::query_as::<_, (String, String, i64, i64)>(
r#"
INSERT INTO storage.folders (name, parent_id, user_id)
VALUES ($1, NULL, $2)
ON CONFLICT DO NOTHING
RETURNING id::text,
path,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
"#,
@@ -649,12 +633,13 @@ impl FolderDbRepository {
.map_err(|e| DomainError::internal_error("FolderDb", format!("home folder: {e}")))?;
match row {
Some((id, ca, ma)) => self.row_to_folder(id, name.to_string(), None, Some(user_id.to_string()), ca, ma).await,
Some((id, path, ca, ma)) => Self::row_to_folder(id, name.to_string(), path, None, Some(user_id.to_string()), ca, ma),
None => {
// Already exists — fetch it
let existing = sqlx::query_as::<_, (String, i64, i64)>(
let existing = sqlx::query_as::<_, (String, String, i64, i64)>(
r#"
SELECT id::text,
path,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
FROM storage.folders
@@ -666,8 +651,7 @@ impl FolderDbRepository {
.fetch_one(self.pool())
.await
.map_err(|e| DomainError::internal_error("FolderDb", format!("home fetch: {e}")))?;
self.row_to_folder(existing.0, name.to_string(), None, Some(user_id.to_string()), existing.1, existing.2)
.await
Self::row_to_folder(existing.0, name.to_string(), existing.1, None, Some(user_id.to_string()), existing.2, existing.3)
}
}
}
src/infrastructure/services/dedup_service.rs
-4
View File
@@ -810,10 +810,6 @@ impl DedupPort for DedupService {
self.remove_reference(hash).await
}
fn hash_bytes(&self, content: &[u8]) -> String {
DedupService::hash_bytes(content)
}
async fn hash_file(&self, path: &Path) -> Result<String, DomainError> {
DedupService::hash_file(path)
.await
src/infrastructure/services/image_transcode_service.rs
+168 -160
View File
@@ -3,22 +3,20 @@
//! Automatically transcodes images to WebP format when the browser supports it,
//! reducing bandwidth by 30-50% compared to JPEG/PNG.
//!
//! Features:
//! - Detects browser WebP support via Accept header
//! - Caches transcoded versions to avoid re-conversion
//! Architecture:
//! - **Dedicated `rayon` thread pool** for CPU-bound transcoding (never blocks Tokio)
//! - **`moka` lock-free cache** for hot transcoded images (no write-lock on reads)
//! - Disk cache for persistence across restarts
//! - Supports JPEG, PNG, GIF → WebP conversion
//! - Configurable quality settings
//! - Falls back to original if conversion fails
//! - Falls back to original if conversion fails or result is larger
use async_trait::async_trait;
use bytes::Bytes;
use image::{DynamicImage, ImageFormat};
use lru::LruCache;
use std::num::NonZeroUsize;
use image::ImageFormat;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, OnceLock};
use tokio::fs;
use tokio::sync::RwLock;
use crate::application::ports::transcode_ports::{
ImageTranscodePort, OutputFormat as PortOutputFormat, TranscodeStatsDto,
@@ -28,11 +26,20 @@ use crate::domain::errors::{DomainError, ErrorKind};
/// Maximum file size for transcoding (5MB - larger files stream directly)
pub const MAX_TRANSCODE_SIZE: u64 = 5 * 1024 * 1024;
/// Cache key for transcoded images
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct TranscodeKey {
file_id: String,
format: OutputFormat,
/// Number of threads in the dedicated transcoding pool
const TRANSCODE_POOL_THREADS: usize = 2;
/// Dedicated rayon thread pool for CPU-bound image transcoding.
/// Isolated from Tokio's blocking pool to prevent starvation of other I/O.
fn transcode_pool() -> &'static rayon::ThreadPool {
static POOL: OnceLock<rayon::ThreadPool> = OnceLock::new();
POOL.get_or_init(|| {
rayon::ThreadPoolBuilder::new()
.num_threads(TRANSCODE_POOL_THREADS)
.thread_name(|idx| format!("transcode-{idx}"))
.build()
.expect("Failed to create transcode thread pool")
})
}
/// Supported output formats
@@ -75,7 +82,6 @@ impl BrowserCapabilities {
/// Get the best output format for this browser
pub fn best_format(&self) -> Option<OutputFormat> {
// WebP has best support currently
if self.supports_webp {
Some(OutputFormat::WebP)
} else {
@@ -84,21 +90,17 @@ impl BrowserCapabilities {
}
}
/// Image Transcoding Service
pub struct ImageTranscodeService {
/// Cache directory for transcoded images
cache_dir: PathBuf,
/// In-memory LRU cache for hot transcoded images
memory_cache: Arc<RwLock<LruCache<TranscodeKey, Bytes>>>,
/// Maximum memory cache size in bytes
max_memory_bytes: usize,
/// Current memory usage
current_memory_bytes: Arc<RwLock<usize>>,
/// Statistics
stats: Arc<RwLock<TranscodeStats>>,
/// Lock-free transcoding statistics using atomics (no RwLock needed)
#[derive(Debug, Default)]
struct AtomicTranscodeStats {
cache_hits: AtomicU64,
disk_hits: AtomicU64,
transcodes: AtomicU64,
bytes_saved: AtomicU64,
transcode_errors: AtomicU64,
}
/// Transcoding statistics
/// Snapshot of transcoding statistics
#[derive(Debug, Default, Clone)]
pub struct TranscodeStats {
pub cache_hits: u64,
@@ -108,19 +110,58 @@ pub struct TranscodeStats {
pub transcode_errors: u64,
}
impl AtomicTranscodeStats {
fn snapshot(&self) -> TranscodeStats {
TranscodeStats {
cache_hits: self.cache_hits.load(Ordering::Relaxed),
disk_hits: self.disk_hits.load(Ordering::Relaxed),
transcodes: self.transcodes.load(Ordering::Relaxed),
bytes_saved: self.bytes_saved.load(Ordering::Relaxed),
transcode_errors: self.transcode_errors.load(Ordering::Relaxed),
}
}
}
/// Image Transcoding Service
///
/// Uses a dedicated `rayon` thread pool for CPU-bound work and `moka` for
/// lock-free concurrent caching with automatic weight-based eviction.
pub struct ImageTranscodeService {
/// Cache directory for transcoded images on disk
cache_dir: PathBuf,
/// Lock-free concurrent cache (moka) — no write-lock on reads
memory_cache: moka::future::Cache<String, Bytes>,
/// Lock-free statistics
stats: Arc<AtomicTranscodeStats>,
}
impl ImageTranscodeService {
/// Create new transcoding service
///
/// - `storage_root`: base path for disk cache
/// - `max_cache_entries`: max number of transcoded images in memory
/// - `max_memory_bytes`: max total bytes for in-memory cache
pub fn new(storage_root: &Path, max_cache_entries: usize, max_memory_bytes: usize) -> Self {
let cache_dir = storage_root.join(".transcoded");
// Build moka cache with weight-based eviction (by content size)
let memory_cache = moka::future::Cache::builder()
.max_capacity(max_memory_bytes as u64)
.weigher(|_key: &String, value: &Bytes| -> u32 {
// Weight = byte size, capped to u32::MAX
value.len().min(u32::MAX as usize) as u32
})
.time_to_live(std::time::Duration::from_secs(600)) // 10 min TTL for freshness
.build();
// Ignore max_cache_entries — moka uses weight-based eviction, which is
// more accurate than entry-count limits for variable-size images.
let _ = max_cache_entries;
Self {
cache_dir,
memory_cache: Arc::new(RwLock::new(LruCache::new(
NonZeroUsize::new(max_cache_entries).unwrap_or(NonZeroUsize::new(1000).unwrap()),
))),
max_memory_bytes,
current_memory_bytes: Arc::new(RwLock::new(0)),
stats: Arc::new(RwLock::new(TranscodeStats::default())),
memory_cache,
stats: Arc::new(AtomicTranscodeStats::default()),
}
}
@@ -129,7 +170,8 @@ impl ImageTranscodeService {
fs::create_dir_all(&self.cache_dir).await?;
fs::create_dir_all(self.cache_dir.join("webp")).await?;
tracing::info!(
"🖼️ Image transcode service initialized at {:?}",
"🖼️ Image transcode service initialized (rayon pool: {} threads, cache dir: {:?})",
TRANSCODE_POOL_THREADS,
self.cache_dir
);
Ok(())
@@ -148,8 +190,8 @@ impl ImageTranscodeService {
Self::can_transcode(mime_type) && file_size <= MAX_TRANSCODE_SIZE
}
/// Get transcoded version of an image
/// Returns (content, mime_type, was_transcoded)
/// Get transcoded version of an image.
/// Returns `(content, mime_type, was_transcoded)`.
pub async fn get_transcoded(
&self,
file_id: &str,
@@ -157,34 +199,25 @@ impl ImageTranscodeService {
original_mime: &str,
target_format: OutputFormat,
) -> Result<(Bytes, String, bool), String> {
let key = TranscodeKey {
file_id: file_id.to_string(),
format: target_format,
};
let cache_key = format!("{}:{}", file_id, target_format.extension());
// Check memory cache first
{
let mut cache = self.memory_cache.write().await;
if let Some(cached) = cache.get(&key) {
let mut stats = self.stats.write().await;
stats.cache_hits += 1;
tracing::debug!("🔥 Transcode memory cache HIT: {}", file_id);
return Ok((cached.clone(), target_format.mime_type().to_string(), true));
}
// ── 1. Check moka memory cache (lock-free read) ──
if let Some(cached) = self.memory_cache.get(&cache_key).await {
self.stats.cache_hits.fetch_add(1, Ordering::Relaxed);
tracing::debug!("🔥 Transcode memory cache HIT: {}", file_id);
return Ok((cached, target_format.mime_type().to_string(), true));
}
// Check disk cache
// ── 2. Check disk cache (async fs) ──
let cache_path = self.get_cache_path(file_id, target_format);
if cache_path.exists() {
if tokio::fs::try_exists(&cache_path).await.unwrap_or(false) {
match fs::read(&cache_path).await {
Ok(data) => {
let content = Bytes::from(data);
// Store in memory cache
self.cache_in_memory(&key, content.clone()).await;
let mut stats = self.stats.write().await;
stats.disk_hits += 1;
self.memory_cache
.insert(cache_key.clone(), content.clone())
.await;
self.stats.disk_hits.fetch_add(1, Ordering::Relaxed);
tracing::debug!("💾 Transcode disk cache HIT: {}", file_id);
return Ok((content, target_format.mime_type().to_string(), true));
}
@@ -194,16 +227,27 @@ impl ImageTranscodeService {
}
}
// Need to transcode
let transcoded = self.transcode_image(original_content, original_mime, target_format)?;
let transcoded_bytes = Bytes::from(transcoded.clone());
// ── 3. Transcode on dedicated rayon pool (never blocks Tokio) ──
let content_owned = original_content.to_vec();
let mime_owned = original_mime.to_string();
// Calculate savings
let (tx, rx) = tokio::sync::oneshot::channel();
transcode_pool().spawn(move || {
let result = transcode_image_blocking(&content_owned, &mime_owned, target_format);
let _ = tx.send(result);
});
let transcoded = rx
.await
.map_err(|_| "Transcode task was cancelled".to_string())??;
let transcoded_bytes = Bytes::from(transcoded);
// ── 4. Evaluate savings ──
let original_size = original_content.len();
let transcoded_size = transcoded_bytes.len();
let saved = original_size.saturating_sub(transcoded_size);
// Only use transcoded if it's actually smaller
if transcoded_size >= original_size {
tracing::debug!(
"⚠️ Transcode not beneficial for {}: {} -> {} bytes",
@@ -218,27 +262,30 @@ impl ImageTranscodeService {
));
}
// Save to disk cache (async, don't wait)
let saved = original_size - transcoded_size;
// ── 5. Persist to disk cache (fire-and-forget) ──
let cache_path_clone = cache_path.clone();
let transcoded_clone = transcoded.clone();
let transcoded_for_disk = transcoded_bytes.clone();
tokio::spawn(async move {
if let Some(parent) = cache_path_clone.parent() {
let _ = fs::create_dir_all(parent).await;
}
if let Err(e) = fs::write(&cache_path_clone, &transcoded_clone).await {
if let Err(e) = fs::write(&cache_path_clone, &transcoded_for_disk).await {
tracing::warn!("Failed to cache transcoded image: {}", e);
}
});
// Store in memory cache
self.cache_in_memory(&key, transcoded_bytes.clone()).await;
// ── 6. Store in moka memory cache (lock-free) ──
self.memory_cache
.insert(cache_key, transcoded_bytes.clone())
.await;
// Update stats
{
let mut stats = self.stats.write().await;
stats.transcodes += 1;
stats.bytes_saved += saved as u64;
}
// ── 7. Update stats (lock-free atomics) ──
self.stats.transcodes.fetch_add(1, Ordering::Relaxed);
self.stats
.bytes_saved
.fetch_add(saved as u64, Ordering::Relaxed);
tracing::info!(
"✨ Transcoded {}: {} -> {} bytes ({:.1}% smaller)",
@@ -255,43 +302,6 @@ impl ImageTranscodeService {
))
}
/// Perform actual image transcoding
fn transcode_image(
&self,
content: &[u8],
original_mime: &str,
target_format: OutputFormat,
) -> Result<Vec<u8>, String> {
// Determine input format
let input_format = match original_mime {
"image/jpeg" | "image/jpg" => ImageFormat::Jpeg,
"image/png" => ImageFormat::Png,
"image/gif" => ImageFormat::Gif,
_ => return Err(format!("Unsupported input format: {}", original_mime)),
};
// Load image
let img = image::load_from_memory_with_format(content, input_format)
.map_err(|e| format!("Failed to decode image: {}", e))?;
// Encode to target format
match target_format {
OutputFormat::WebP => self.encode_webp(&img),
}
}
/// Encode image to WebP
fn encode_webp(&self, img: &DynamicImage) -> Result<Vec<u8>, String> {
let mut buffer = Vec::new();
let mut cursor = std::io::Cursor::new(&mut buffer);
// Use image crate's WebP encoder
img.write_to(&mut cursor, ImageFormat::WebP)
.map_err(|e| format!("Failed to encode WebP: {}", e))?;
Ok(buffer)
}
/// Get path for cached transcoded file
fn get_cache_path(&self, file_id: &str, format: OutputFormat) -> PathBuf {
self.cache_dir
@@ -299,67 +309,28 @@ impl ImageTranscodeService {
.join(format!("{}.{}", file_id, format.extension()))
}
/// Store transcoded image in memory cache
async fn cache_in_memory(&self, key: &TranscodeKey, content: Bytes) {
let size = content.len();
let mut current = self.current_memory_bytes.write().await;
// Evict if needed
while *current + size > self.max_memory_bytes {
let mut cache = self.memory_cache.write().await;
if let Some((_, evicted)) = cache.pop_lru() {
*current = current.saturating_sub(evicted.len());
} else {
break;
}
}
// Add to cache
if *current + size <= self.max_memory_bytes {
let mut cache = self.memory_cache.write().await;
cache.put(key.clone(), content);
*current += size;
}
}
/// Invalidate cached transcodes for a file
pub async fn invalidate(&self, file_id: &str) {
// Remove from memory cache
{
let mut cache = self.memory_cache.write().await;
let key = TranscodeKey {
file_id: file_id.to_string(),
format: OutputFormat::WebP,
};
if let Some(removed) = cache.pop(&key) {
let mut current = self.current_memory_bytes.write().await;
*current = current.saturating_sub(removed.len());
}
}
let cache_key = format!("{}:{}", file_id, OutputFormat::WebP.extension());
self.memory_cache.invalidate(&cache_key).await;
// Remove disk cache
let cache_path = self.get_cache_path(file_id, OutputFormat::WebP);
let _ = fs::remove_file(&cache_path).await;
}
/// Get transcoding statistics
pub async fn get_stats(&self) -> TranscodeStats {
self.stats.read().await.clone()
self.stats.snapshot()
}
/// Clear all caches
pub async fn clear_cache(&self) -> std::io::Result<()> {
// Clear memory
{
let mut cache = self.memory_cache.write().await;
cache.clear();
let mut current = self.current_memory_bytes.write().await;
*current = 0;
}
self.memory_cache.invalidate_all();
// Clear disk
if self.cache_dir.exists() {
if tokio::fs::try_exists(&self.cache_dir)
.await
.unwrap_or(false)
{
fs::remove_dir_all(&self.cache_dir).await?;
fs::create_dir_all(&self.cache_dir).await?;
fs::create_dir_all(self.cache_dir.join("webp")).await?;
@@ -369,6 +340,36 @@ impl ImageTranscodeService {
}
}
// ─── CPU-bound transcoding (runs on rayon, never on Tokio) ───────────────────
/// Perform actual image transcoding. This is a pure CPU function — safe to call
/// from `rayon::spawn` or `spawn_blocking`.
fn transcode_image_blocking(
content: &[u8],
original_mime: &str,
target_format: OutputFormat,
) -> Result<Vec<u8>, String> {
let input_format = match original_mime {
"image/jpeg" | "image/jpg" => ImageFormat::Jpeg,
"image/png" => ImageFormat::Png,
"image/gif" => ImageFormat::Gif,
_ => return Err(format!("Unsupported input format: {}", original_mime)),
};
let img = image::load_from_memory_with_format(content, input_format)
.map_err(|e| format!("Failed to decode image: {}", e))?;
match target_format {
OutputFormat::WebP => {
let mut buffer = Vec::new();
let mut cursor = std::io::Cursor::new(&mut buffer);
img.write_to(&mut cursor, ImageFormat::WebP)
.map_err(|e| format!("Failed to encode WebP: {}", e))?;
Ok(buffer)
}
}
}
// ─── Port implementation ─────────────────────────────────────────────────────
/// Convert port OutputFormat to infra OutputFormat.
@@ -481,4 +482,11 @@ mod tests {
1024 * 1024
));
}
#[test]
fn test_transcode_pool_initializes() {
// Verify the pool can be created without panic
let pool = transcode_pool();
assert_eq!(pool.current_num_threads(), TRANSCODE_POOL_THREADS);
}
}