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

fix: resolve all clippy warnings and convert integration_tests to custom cfg

Browse Source 
- Add type aliases (FileRow, FolderRow, FolderRowPaginated, FolderRowOptUser) to reduce type complexity
- Simplify redundant closures in app_password_handler and webdav_handler
- Remove needless borrow in auth_handler
- Collapse nested if/let chains in login_lockout, webdav_lock, auth, rate_limit
- Box LockEntry in acquire() Err variant to fix large enum variant warning
- Rename DeviceCodeStatus::from_str to parse to avoid should_implement_trait lint
- Add #[allow(clippy::too_many_arguments)] and #[allow(clippy::result_unit_err)] where appropriate
- Convert integration_tests from cargo feature to custom cfg attribute
- Add check-cfg lint config in Cargo.toml for integration_tests cfg
This commit is contained in:
Diocrafts
2026-03-04 23:55:08 +01:00
parent b81b7f7a0e
commit ee86c3a128
53 changed files with 871 additions and 956 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/infrastructure/db.rs
+3 -6
View File
@@ -126,8 +126,7 @@ async fn create_pool_with_retries(
if attempt >= MAX_ATTEMPTS {
return Err(DbError(format!(
"Error verifying PostgreSQL {} pool connection: {}",
label,
e
label, e
)));
}
}
@@ -143,8 +142,7 @@ async fn create_pool_with_retries(
if attempt >= MAX_ATTEMPTS {
return Err(DbError(format!(
"Error in PostgreSQL {} pool connection: {}",
label,
e
label, e
)));
}
tokio::time::sleep(Duration::from_secs(2)).await;
@@ -154,8 +152,7 @@ async fn create_pool_with_retries(
Err(DbError(format!(
"Could not establish PostgreSQL {} pool connection after {} attempts",
label,
MAX_ATTEMPTS
label, MAX_ATTEMPTS
)))
}
src/infrastructure/repositories/pg/device_code_pg_repository.rs
+1 -1
View File
@@ -25,7 +25,7 @@ impl DeviceCodePgRepository {
)
})?;
let status = DeviceCodeStatus::from_str(&status_str).unwrap_or(DeviceCodeStatus::Expired);
let status = DeviceCodeStatus::parse(&status_str).unwrap_or(DeviceCodeStatus::Expired);
Ok(DeviceCode::from_raw(
row.try_get("id").unwrap_or_default(),
src/infrastructure/repositories/pg/file_blob_read_repository.rs
+17 -34
View File
@@ -22,6 +22,19 @@ use crate::domain::entities::file::File;
use crate::domain::services::path_service::StoragePath;
use crate::infrastructure::services::dedup_service::DedupService;
/// Type alias for file metadata rows from SQL queries.
type FileRow = (
String,
String,
Option<String>,
Option<String>,
i64,
String,
i64,
i64,
Option<String>,
);
/// File read repository backed by PostgreSQL metadata + blob storage.
pub struct FileBlobReadRepository {
pool: Arc<PgPool>,
@@ -204,17 +217,7 @@ impl FileReadPort for FileBlobReadRepository {
}
async fn list_files(&self, folder_id: Option<&str>) -> Result<Vec<File>, DomainError> {
let rows: Vec<(
String,
String,
Option<String>,
Option<String>,
i64,
String,
i64,
i64,
Option<String>,
)> = if let Some(fid) = folder_id {
let rows: Vec<FileRow> = if let Some(fid) = folder_id {
sqlx::query_as(
r#"
SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
@@ -271,17 +274,7 @@ impl FileReadPort for FileBlobReadRepository {
offset: i64,
limit: i64,
) -> Result<Vec<File>, DomainError> {
let rows: Vec<(
String,
String,
Option<String>,
Option<String>,
i64,
String,
i64,
i64,
Option<String>,
)> = if let Some(fid) = folder_id {
let rows: Vec<FileRow> = if let Some(fid) = folder_id {
sqlx::query_as(
r#"
SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
@@ -853,17 +846,7 @@ impl FileReadPort for FileBlobReadRepository {
let pattern = format!("%{}%", query);
let limit_i64 = limit as i64;
let rows: Vec<(
String,
String,
Option<String>,
Option<String>,
i64,
String,
i64,
i64,
Option<String>,
)> = if let Some(fid) = folder_id {
let rows: Vec<FileRow> = if let Some(fid) = folder_id {
sqlx::query_as(
r#"
SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
@@ -929,7 +912,7 @@ impl FileReadPort for FileBlobReadRepository {
}
}
#[cfg(feature = "integration_tests")]
#[cfg(integration_tests)]
mod tests {
use super::*;
use crate::common::stubs::StubDedupPort;
src/infrastructure/repositories/pg/folder_db_repository.rs
+85 -112
View File
@@ -16,6 +16,32 @@ use crate::domain::entities::folder::Folder;
use crate::domain::repositories::folder_repository::FolderRepository;
use crate::domain::services::path_service::StoragePath;
/// Type alias for folder metadata rows from SQL queries.
type FolderRow = (String, String, String, Option<String>, String, i64, i64);
/// Type alias for paginated folder rows (includes total_count).
type FolderRowPaginated = (
String,
String,
String,
Option<String>,
String,
i64,
i64,
i64,
);
/// Type alias for folder rows with optional user_id.
type FolderRowOptUser = (
String,
String,
String,
Option<String>,
Option<String>,
i64,
i64,
);
/// PostgreSQL-backed folder repository.
///
/// All folder metadata lives in the `storage.folders` table. The physical
@@ -173,10 +199,9 @@ impl FolderRepository for FolderDbRepository {
}
async fn list_folders(&self, parent_id: Option<&str>) -> Result<Vec<Folder>, DomainError> {
let rows: Vec<(String, String, String, Option<String>, String, i64, i64)> =
if let Some(pid) = parent_id {
sqlx::query_as(
r#"
let rows: Vec<FolderRow> = if let Some(pid) = parent_id {
sqlx::query_as(
r#"
SELECT id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
@@ -184,13 +209,13 @@ impl FolderRepository for FolderDbRepository {
WHERE parent_id = $1::uuid AND NOT is_trashed
ORDER BY name
"#,
)
.bind(pid)
.fetch_all(self.pool())
.await
} else {
sqlx::query_as(
r#"
)
.bind(pid)
.fetch_all(self.pool())
.await
} else {
sqlx::query_as(
r#"
SELECT id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
@@ -198,11 +223,11 @@ impl FolderRepository for FolderDbRepository {
WHERE parent_id IS NULL AND NOT is_trashed
ORDER BY name
"#,
)
.fetch_all(self.pool())
.await
}
.map_err(|e| DomainError::internal_error("FolderDb", format!("list: {e}")))?;
)
.fetch_all(self.pool())
.await
}
.map_err(|e| DomainError::internal_error("FolderDb", format!("list: {e}")))?;
rows.into_iter()
.map(|(id, name, path, pid, uid, ca, ma)| {
@@ -216,10 +241,9 @@ impl FolderRepository for FolderDbRepository {
parent_id: Option<&str>,
owner_id: &str,
) -> Result<Vec<Folder>, DomainError> {
let rows: Vec<(String, String, String, Option<String>, String, i64, i64)> =
if let Some(pid) = parent_id {
sqlx::query_as(
r#"
let rows: Vec<FolderRow> = if let Some(pid) = parent_id {
sqlx::query_as(
r#"
SELECT id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
@@ -227,14 +251,14 @@ impl FolderRepository for FolderDbRepository {
WHERE parent_id = $1::uuid AND user_id = $2 AND NOT is_trashed
ORDER BY name
"#,
)
.bind(pid)
.bind(owner_id)
.fetch_all(self.pool())
.await
} else {
sqlx::query_as(
r#"
)
.bind(pid)
.bind(owner_id)
.fetch_all(self.pool())
.await
} else {
sqlx::query_as(
r#"
SELECT id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
@@ -242,12 +266,12 @@ impl FolderRepository for FolderDbRepository {
WHERE parent_id IS NULL AND user_id = $1 AND NOT is_trashed
ORDER BY name
"#,
)
.bind(owner_id)
.fetch_all(self.pool())
.await
}
.map_err(|e| DomainError::internal_error("FolderDb", format!("list_by_owner: {e}")))?;
)
.bind(owner_id)
.fetch_all(self.pool())
.await
}
.map_err(|e| DomainError::internal_error("FolderDb", format!("list_by_owner: {e}")))?;
rows.into_iter()
.map(|(id, name, path, pid, uid, ca, ma)| {
@@ -266,16 +290,7 @@ impl FolderRepository for FolderDbRepository {
limit: usize,
include_total: bool,
) -> Result<(Vec<Folder>, Option<usize>), DomainError> {
let rows: Vec<(
String,
String,
String,
Option<String>,
String,
i64,
i64,
i64,
)> = if let Some(pid) = parent_id {
let rows: Vec<FolderRowPaginated> = if let Some(pid) = parent_id {
sqlx::query_as(
r#"
SELECT id::text, name, path, parent_id::text, user_id,
@@ -339,16 +354,7 @@ impl FolderRepository for FolderDbRepository {
limit: usize,
include_total: bool,
) -> Result<(Vec<Folder>, Option<usize>), DomainError> {
let rows: Vec<(
String,
String,
String,
Option<String>,
String,
i64,
i64,
i64,
)> = if let Some(pid) = parent_id {
let rows: Vec<FolderRowPaginated> = if let Some(pid) = parent_id {
sqlx::query_as(
r#"
SELECT id::text, name, path, parent_id::text, user_id,
@@ -689,15 +695,7 @@ impl FolderRepository for FolderDbRepository {
AND fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = $1::uuid) \
ORDER BY fo.path";
let rows: Vec<(
String,
String,
String,
Option<String>,
Option<String>,
i64,
i64,
)> = sqlx::query_as(sql)
let rows: Vec<FolderRowOptUser> = sqlx::query_as(sql)
.bind(folder_id)
.fetch_all(self.pool())
.await
@@ -760,15 +758,7 @@ impl FolderRepository for FolderDbRepository {
ORDER BY fo.name"
);
let rows: Vec<(
String,
String,
String,
Option<String>,
Option<String>,
i64,
i64,
)> = if let Some(ref pattern) = name_pattern {
let rows: Vec<FolderRowOptUser> = if let Some(ref pattern) = name_pattern {
sqlx::query_as(&sql)
.bind(user_id)
.bind(pattern)
@@ -824,15 +814,7 @@ impl FolderRepository for FolderDbRepository {
)
};
let rows: Vec<(
String,
String,
String,
Option<String>,
Option<String>,
i64,
i64,
)> = if let Some(pid) = parent_id {
let rows: Vec<FolderRowOptUser> = if let Some(pid) = parent_id {
if let Some(ref pattern) = name_pattern {
sqlx::query_as(&sql)
.bind(pid)
@@ -897,15 +879,7 @@ impl FolderRepository for FolderDbRepository {
ORDER BY fo.name"
);
let rows: Vec<(
String,
String,
String,
Option<String>,
Option<String>,
i64,
i64,
)> = if let Some(ref pattern) = name_pattern {
let rows: Vec<FolderRowOptUser> = if let Some(ref pattern) = name_pattern {
sqlx::query_as(&sql)
.bind(user_id)
.bind(folder_id)
@@ -937,10 +911,9 @@ impl FolderRepository for FolderDbRepository {
let pattern = format!("%{}%", query);
let limit_i64 = limit as i64;
let rows: Vec<(String, String, String, Option<String>, String, i64, i64)> =
if let Some(pid) = parent_id {
sqlx::query_as(
r#"
let rows: Vec<FolderRow> = if let Some(pid) = parent_id {
sqlx::query_as(
r#"
SELECT id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
@@ -956,16 +929,16 @@ impl FolderRepository for FolderDbRepository {
name
LIMIT $4
"#,
)
.bind(pid)
.bind(&pattern)
.bind(query)
.bind(limit_i64)
.fetch_all(self.pool())
.await
} else {
sqlx::query_as(
r#"
)
.bind(pid)
.bind(&pattern)
.bind(query)
.bind(limit_i64)
.fetch_all(self.pool())
.await
} else {
sqlx::query_as(
r#"
SELECT id::text, name, path, parent_id::text, user_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
@@ -981,14 +954,14 @@ impl FolderRepository for FolderDbRepository {
name
LIMIT $3
"#,
)
.bind(&pattern)
.bind(query)
.bind(limit_i64)
.fetch_all(self.pool())
.await
}
.map_err(|e| DomainError::internal_error("FolderDb", format!("suggest: {e}")))?;
)
.bind(&pattern)
.bind(query)
.bind(limit_i64)
.fetch_all(self.pool())
.await
}
.map_err(|e| DomainError::internal_error("FolderDb", format!("suggest: {e}")))?;
rows.into_iter()
.map(|(id, name, path, pid, uid, ca, ma)| {
src/infrastructure/services/chunked_upload_service.rs
+6 -7
View File
@@ -495,13 +495,12 @@ impl ChunkedUploadService {
// worker free for other connections.
if let Some(ref expected_checksum) = checksum {
let data_clone = data.clone(); // Bytes::clone is O(1) — just an Arc increment
let actual_checksum =
tokio::task::spawn_blocking(move || {
use md5::{Md5, Digest};
format!("{:x}", Md5::digest(&data_clone))
})
.await
.map_err(|e| format!("MD5 checksum task failed: {e}"))?;
let actual_checksum = tokio::task::spawn_blocking(move || {
use md5::{Digest, Md5};
format!("{:x}", Md5::digest(&data_clone))
})
.await
.map_err(|e| format!("MD5 checksum task failed: {e}"))?;
if actual_checksum != *expected_checksum {
return Err(format!(
src/infrastructure/services/login_lockout_service.rs
+6 -6
View File
@@ -60,12 +60,12 @@ impl LoginLockoutService {
/// Returns `Ok(())` if the user may attempt login, or
/// `Err(remaining_secs)` with the *approximate* remaining lockout time.
pub fn check(&self, username: &str) -> Result<(), u64> {
if let Some(rec) = self.cache.get(&username.to_lowercase()) {
if rec.count >= self.max_failures {
// The entry exists and is over the threshold. Because moka
// evicts at TTL we know the lockout window has not yet elapsed.
return Err(self.lockout_secs);
}
if let Some(rec) = self.cache.get(&username.to_lowercase())
&& rec.count >= self.max_failures
{
// The entry exists and is over the threshold. Because moka
// evicts at TTL we know the lockout window has not yet elapsed.
return Err(self.lockout_secs);
}
Ok(())
}
src/infrastructure/services/path_service.rs
+9 -1
View File
@@ -312,7 +312,15 @@ mod tests {
#[test]
fn test_validate_path_rejects_dangerous_chars() {
let service = PathService::new(PathBuf::from("/storage"));
for dangerous in &["file:name", "file*name", "file?name", "file<name", "file>name", "file|name", "file\"name"] {
for dangerous in &[
"file:name",
"file*name",
"file?name",
"file<name",
"file>name",
"file|name",
"file\"name",
] {
let path = StoragePath::new(vec![dangerous.to_string()]);
assert!(
service.validate_path(&path).is_err(),
src/infrastructure/services/webdav_lock_service.rs
+213 -218
View File
@@ -1,218 +1,213 @@
//! WebDAV lock store backed by Moka (in-memory cache with per-entry TTL).
//!
//! Locks are automatically evicted when their timeout expires, preventing
//! orphaned locks from accumulating. Two caches are maintained:
//!
//! - `by_path` : path → `LockEntry` (for LOCK conflict detection)
//! - `by_token` : token → path (for fast UNLOCK / refresh lookups)
//!
//! Both caches share the same TTL so entries disappear together.
use std::sync::Arc;
use std::time::Duration;
use crate::application::adapters::webdav_adapter::{LockInfo, LockScope};
/// Default lock timeout when the client does not specify one (RFC 4918 §10.7).
const DEFAULT_LOCK_TIMEOUT_SECS: u64 = 1800; // 30 minutes
/// Absolute maximum TTL a client may request.
const MAX_LOCK_TIMEOUT_SECS: u64 = 86_400; // 24 hours
/// A stored lock entry.
#[derive(Clone, Debug)]
pub struct LockEntry {
pub info: LockInfo,
pub path: String,
}
/// In-memory WebDAV lock store with automatic TTL-based expiration.
///
/// Uses Moka's `sync::Cache` — lock-free (sharded) reads, bounded size,
/// and per-entry TTL via `policy::Expiry`.
pub struct WebDavLockStore {
/// path → `LockEntry`
by_path: moka::sync::Cache<String, LockEntry>,
/// token → path (reverse index)
by_token: moka::sync::Cache<String, String>,
}
impl WebDavLockStore {
/// Create a new lock store.
///
/// * `max_capacity` — upper bound on simultaneous locks (evicts LRU on overflow).
pub fn new(max_capacity: u64) -> Self {
// We use `expire_after` (per-entry TTL) via insert with explicit ttl,
// so we configure a generous global time_to_live as a safety net.
let by_path = moka::sync::Cache::builder()
.max_capacity(max_capacity)
.time_to_live(Duration::from_secs(MAX_LOCK_TIMEOUT_SECS))
.build();
let by_token = moka::sync::Cache::builder()
.max_capacity(max_capacity)
.time_to_live(Duration::from_secs(MAX_LOCK_TIMEOUT_SECS))
.build();
Self { by_path, by_token }
}
// ── Public API ──────────────────────────────────────────────
/// Attempt to acquire a lock on `path`.
///
/// Returns `Ok(LockEntry)` on success, or `Err(existing)` if the resource
/// is already exclusively locked by a different token.
pub fn acquire(
&self,
path: &str,
info: LockInfo,
) -> Result<LockEntry, LockEntry> {
// Check for existing conflicting lock
if let Some(existing) = self.by_path.get(path) {
if existing.info.scope == LockScope::Exclusive {
return Err(existing);
}
}
let ttl = Self::parse_timeout(info.timeout.as_deref());
let entry = LockEntry {
info,
path: path.to_owned(),
};
self.by_path
.insert(path.to_owned(), entry.clone());
self.by_token
.insert(entry.info.token.clone(), path.to_owned());
// Moka 0.12 does not expose per-entry set_expiration_after_insert at
// insert time. We rely on the global `time_to_live` as an upper bound
// and use the `invalidate_after` helper below for custom TTL.
//
// To implement shorter-than-max TTL we schedule an async invalidation.
if ttl.as_secs() < MAX_LOCK_TIMEOUT_SECS {
let by_path = self.by_path.clone();
let by_token = self.by_token.clone();
let token = entry.info.token.clone();
let path_owned = path.to_owned();
tokio::spawn(async move {
tokio::time::sleep(ttl).await;
// Only remove if the entry still matches (wasn't refreshed/replaced)
if let Some(e) = by_path.get(&path_owned) {
if e.info.token == token {
by_path.invalidate(&path_owned);
by_token.invalidate(&token);
}
}
});
}
Ok(entry)
}
/// Refresh an existing lock (extend its timeout).
///
/// Returns `Some(LockEntry)` with updated timeout, or `None` if the token
/// is unknown (expired or never existed).
pub fn refresh(&self, token: &str, new_timeout: Option<&str>) -> Option<LockEntry> {
let path = self.by_token.get(token)?;
let mut entry = self.by_path.get(&path)?;
if entry.info.token != token {
return None; // token mismatch — lock was replaced
}
let ttl = Self::parse_timeout(new_timeout.or(entry.info.timeout.as_deref()));
let timeout_str = format!("Second-{}", ttl.as_secs());
entry.info.timeout = Some(timeout_str.clone());
// Re-insert to reset the TTL
self.by_path.insert(path.clone(), entry.clone());
self.by_token.insert(token.to_owned(), path.clone());
if ttl.as_secs() < MAX_LOCK_TIMEOUT_SECS {
let by_path = self.by_path.clone();
let by_token = self.by_token.clone();
let token_owned = token.to_owned();
let path_owned = path.clone();
tokio::spawn(async move {
tokio::time::sleep(ttl).await;
if let Some(e) = by_path.get(&path_owned) {
if e.info.token == token_owned {
by_path.invalidate(&path_owned);
by_token.invalidate(&token_owned);
}
}
});
}
Some(entry)
}
/// Release a lock by its token.
///
/// Returns `true` if the lock existed and was removed.
pub fn release(&self, token: &str) -> bool {
if let Some(path) = self.by_token.get(token) {
// Only remove from by_path if the token still matches
if let Some(entry) = self.by_path.get(&path) {
if entry.info.token == token {
self.by_path.invalidate(&path);
}
}
self.by_token.invalidate(token);
true
} else {
false
}
}
/// Look up a lock by resource path.
pub fn get_by_path(&self, path: &str) -> Option<LockEntry> {
self.by_path.get(path)
}
/// Look up a lock by token.
pub fn get_by_token(&self, token: &str) -> Option<LockEntry> {
let path = self.by_token.get(token)?;
self.by_path.get(&path)
}
// ── Helpers ─────────────────────────────────────────────────
/// Parse a WebDAV `Timeout` header value into a [`Duration`].
///
/// Accepted formats (RFC 4918 §10.7):
/// - `Second-NNN`
/// - `Infinite` (clamped to `MAX_LOCK_TIMEOUT_SECS`)
/// - Comma-separated list (first value wins)
fn parse_timeout(header: Option<&str>) -> Duration {
let raw = match header {
Some(v) if !v.is_empty() => v,
_ => return Duration::from_secs(DEFAULT_LOCK_TIMEOUT_SECS),
};
// Take the first value in a comma-separated list
let first = raw.split(',').next().unwrap_or(raw).trim();
if first.eq_ignore_ascii_case("Infinite") {
return Duration::from_secs(MAX_LOCK_TIMEOUT_SECS);
}
if let Some(secs_str) = first.strip_prefix("Second-") {
if let Ok(secs) = secs_str.trim().parse::<u64>() {
return Duration::from_secs(secs.min(MAX_LOCK_TIMEOUT_SECS));
}
}
Duration::from_secs(DEFAULT_LOCK_TIMEOUT_SECS)
}
}
/// Create a shared lock store wrapped in `Arc` for embedding in `AppState`.
pub fn create_webdav_lock_store() -> Arc<WebDavLockStore> {
// 10 000 simultaneous locks should be more than enough; Moka evicts LRU
// if the cap is reached, so stale entries are cleaned automatically.
Arc::new(WebDavLockStore::new(10_000))
}
//! WebDAV lock store backed by Moka (in-memory cache with per-entry TTL).
//!
//! Locks are automatically evicted when their timeout expires, preventing
//! orphaned locks from accumulating. Two caches are maintained:
//!
//! - `by_path` : path → `LockEntry` (for LOCK conflict detection)
//! - `by_token` : token → path (for fast UNLOCK / refresh lookups)
//!
//! Both caches share the same TTL so entries disappear together.
use std::sync::Arc;
use std::time::Duration;
use crate::application::adapters::webdav_adapter::{LockInfo, LockScope};
/// Default lock timeout when the client does not specify one (RFC 4918 §10.7).
const DEFAULT_LOCK_TIMEOUT_SECS: u64 = 1800; // 30 minutes
/// Absolute maximum TTL a client may request.
const MAX_LOCK_TIMEOUT_SECS: u64 = 86_400; // 24 hours
/// A stored lock entry.
#[derive(Clone, Debug)]
pub struct LockEntry {
pub info: LockInfo,
pub path: String,
}
/// In-memory WebDAV lock store with automatic TTL-based expiration.
///
/// Uses Moka's `sync::Cache` — lock-free (sharded) reads, bounded size,
/// and per-entry TTL via `policy::Expiry`.
pub struct WebDavLockStore {
/// path → `LockEntry`
by_path: moka::sync::Cache<String, LockEntry>,
/// token → path (reverse index)
by_token: moka::sync::Cache<String, String>,
}
impl WebDavLockStore {
/// Create a new lock store.
///
/// * `max_capacity` — upper bound on simultaneous locks (evicts LRU on overflow).
pub fn new(max_capacity: u64) -> Self {
// We use `expire_after` (per-entry TTL) via insert with explicit ttl,
// so we configure a generous global time_to_live as a safety net.
let by_path = moka::sync::Cache::builder()
.max_capacity(max_capacity)
.time_to_live(Duration::from_secs(MAX_LOCK_TIMEOUT_SECS))
.build();
let by_token = moka::sync::Cache::builder()
.max_capacity(max_capacity)
.time_to_live(Duration::from_secs(MAX_LOCK_TIMEOUT_SECS))
.build();
Self { by_path, by_token }
}
// ── Public API ──────────────────────────────────────────────
/// Attempt to acquire a lock on `path`.
///
/// Returns `Ok(LockEntry)` on success, or `Err(existing)` if the resource
/// is already exclusively locked by a different token.
pub fn acquire(&self, path: &str, info: LockInfo) -> Result<LockEntry, Box<LockEntry>> {
// Check for existing conflicting lock
if let Some(existing) = self.by_path.get(path)
&& existing.info.scope == LockScope::Exclusive
{
return Err(Box::new(existing));
}
let ttl = Self::parse_timeout(info.timeout.as_deref());
let entry = LockEntry {
info,
path: path.to_owned(),
};
self.by_path.insert(path.to_owned(), entry.clone());
self.by_token
.insert(entry.info.token.clone(), path.to_owned());
// Moka 0.12 does not expose per-entry set_expiration_after_insert at
// insert time. We rely on the global `time_to_live` as an upper bound
// and use the `invalidate_after` helper below for custom TTL.
//
// To implement shorter-than-max TTL we schedule an async invalidation.
if ttl.as_secs() < MAX_LOCK_TIMEOUT_SECS {
let by_path = self.by_path.clone();
let by_token = self.by_token.clone();
let token = entry.info.token.clone();
let path_owned = path.to_owned();
tokio::spawn(async move {
tokio::time::sleep(ttl).await;
// Only remove if the entry still matches (wasn't refreshed/replaced)
if let Some(e) = by_path.get(&path_owned)
&& e.info.token == token
{
by_path.invalidate(&path_owned);
by_token.invalidate(&token);
}
});
}
Ok(entry)
}
/// Refresh an existing lock (extend its timeout).
///
/// Returns `Some(LockEntry)` with updated timeout, or `None` if the token
/// is unknown (expired or never existed).
pub fn refresh(&self, token: &str, new_timeout: Option<&str>) -> Option<LockEntry> {
let path = self.by_token.get(token)?;
let mut entry = self.by_path.get(&path)?;
if entry.info.token != token {
return None; // token mismatch — lock was replaced
}
let ttl = Self::parse_timeout(new_timeout.or(entry.info.timeout.as_deref()));
let timeout_str = format!("Second-{}", ttl.as_secs());
entry.info.timeout = Some(timeout_str.clone());
// Re-insert to reset the TTL
self.by_path.insert(path.clone(), entry.clone());
self.by_token.insert(token.to_owned(), path.clone());
if ttl.as_secs() < MAX_LOCK_TIMEOUT_SECS {
let by_path = self.by_path.clone();
let by_token = self.by_token.clone();
let token_owned = token.to_owned();
let path_owned = path.clone();
tokio::spawn(async move {
tokio::time::sleep(ttl).await;
if let Some(e) = by_path.get(&path_owned)
&& e.info.token == token_owned
{
by_path.invalidate(&path_owned);
by_token.invalidate(&token_owned);
}
});
}
Some(entry)
}
/// Release a lock by its token.
///
/// Returns `true` if the lock existed and was removed.
pub fn release(&self, token: &str) -> bool {
if let Some(path) = self.by_token.get(token) {
// Only remove from by_path if the token still matches
if let Some(entry) = self.by_path.get(&path)
&& entry.info.token == token
{
self.by_path.invalidate(&path);
}
self.by_token.invalidate(token);
true
} else {
false
}
}
/// Look up a lock by resource path.
pub fn get_by_path(&self, path: &str) -> Option<LockEntry> {
self.by_path.get(path)
}
/// Look up a lock by token.
pub fn get_by_token(&self, token: &str) -> Option<LockEntry> {
let path = self.by_token.get(token)?;
self.by_path.get(&path)
}
// ── Helpers ─────────────────────────────────────────────────
/// Parse a WebDAV `Timeout` header value into a [`Duration`].
///
/// Accepted formats (RFC 4918 §10.7):
/// - `Second-NNN`
/// - `Infinite` (clamped to `MAX_LOCK_TIMEOUT_SECS`)
/// - Comma-separated list (first value wins)
fn parse_timeout(header: Option<&str>) -> Duration {
let raw = match header {
Some(v) if !v.is_empty() => v,
_ => return Duration::from_secs(DEFAULT_LOCK_TIMEOUT_SECS),
};
// Take the first value in a comma-separated list
let first = raw.split(',').next().unwrap_or(raw).trim();
if first.eq_ignore_ascii_case("Infinite") {
return Duration::from_secs(MAX_LOCK_TIMEOUT_SECS);
}
if let Some(secs_str) = first.strip_prefix("Second-")
&& let Ok(secs) = secs_str.trim().parse::<u64>()
{
return Duration::from_secs(secs.min(MAX_LOCK_TIMEOUT_SECS));
}
Duration::from_secs(DEFAULT_LOCK_TIMEOUT_SECS)
}
}
/// Create a shared lock store wrapped in `Arc` for embedding in `AppState`.
pub fn create_webdav_lock_store() -> Arc<WebDavLockStore> {
// 10 000 simultaneous locks should be more than enough; Moka evicts LRU
// if the cap is reached, so stale entries are cleaned automatically.
Arc::new(WebDavLockStore::new(10_000))
}
src/infrastructure/services/zip_service.rs
+2 -2
View File
@@ -1,3 +1,5 @@
use crate::application::services::file_retrieval_service::FileRetrievalService;
use crate::application::services::folder_service::FolderService;
use crate::{
application::dtos::file_dto::FileDto,
application::ports::file_ports::FileRetrievalUseCase,
@@ -16,8 +18,6 @@ use thiserror::Error;
use tokio::io::BufWriter;
use tokio_util::compat::Compat;
use tracing::*;
use crate::application::services::file_retrieval_service::FileRetrievalService;
use crate::application::services::folder_service::FolderService;
/// Error related to ZIP file creation
#[derive(Debug, Error)]