Merge pull request #207 from jaredwolff/fix/nextcloud-sync-conflict

Fix Nextcloud sync conflict caused by static UUID ETags
This commit is contained in:
Dionisio Pozo
2026-03-15 20:28:39 +01:00
committed by GitHub
19 changed files with 206 additions and 82 deletions
Generated
+2 -2
View File
@@ -3051,9 +3051,9 @@ dependencies = [
[[package]] [[package]]
name = "quinn-proto" name = "quinn-proto"
version = "0.11.13" version = "0.11.14"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f1906b49b0c3bc04b5fe5d86a77925ae6524a19b816ae38ce1e426255f1d8a31" checksum = "434b42fec591c96ef50e21e886936e66d3cc3f737104fdb9b737c40ffb94c098"
dependencies = [ dependencies = [
"bytes", "bytes",
"getrandom 0.3.4", "getrandom 0.3.4",
+7
View File
@@ -56,6 +56,11 @@ pub struct FileDto {
/// Only populated by the /api/photos endpoint. /// Only populated by the /api/photos endpoint.
#[serde(skip_serializing_if = "Option::is_none")] #[serde(skip_serializing_if = "Option::is_none")]
pub sort_date: Option<u64>, pub sort_date: Option<u64>,
/// Content-addressable ETag (= blob_hash). Changes on every content write.
/// Used for WebDAV/Nextcloud ETag headers. Omitted from REST API JSON.
#[serde(skip)]
pub etag: String,
} }
impl From<File> for FileDto { impl From<File> for FileDto {
@@ -85,6 +90,7 @@ impl From<File> for FileDto {
size_formatted, size_formatted,
owner_id: parts.owner_id.map(|u| u.to_string()), owner_id: parts.owner_id.map(|u| u.to_string()),
sort_date: None, sort_date: None,
etag: parts.etag,
} }
} }
} }
@@ -124,6 +130,7 @@ impl FileDto {
category: Arc::from("Document"), category: Arc::from("Document"),
size_formatted: "0 Bytes".to_string(), size_formatted: "0 Bytes".to_string(),
owner_id: None, owner_id: None,
etag: String::new(),
sort_date: None, sort_date: None,
} }
} }
+4 -2
View File
@@ -69,7 +69,8 @@ pub trait FileUploadUseCase: Send + Sync + 'static {
path: &str, path: &str,
content: &[u8], content: &[u8],
content_type: &str, content_type: &str,
) -> Result<(), DomainError>; modified_at: Option<i64>,
) -> Result<FileDto, DomainError>;
/// Streaming update — spools body to a temp file with incremental hash, /// Streaming update — spools body to a temp file with incremental hash,
/// then atomically replaces the file content via dedup store. /// then atomically replaces the file content via dedup store.
@@ -83,7 +84,8 @@ pub trait FileUploadUseCase: Send + Sync + 'static {
size: u64, size: u64,
content_type: &str, content_type: &str,
pre_computed_hash: Option<String>, pre_computed_hash: Option<String>,
) -> Result<(), DomainError>; modified_at: Option<i64>,
) -> Result<FileDto, DomainError>;
} }
// ───────────────────────────────────────────────────── // ─────────────────────────────────────────────────────
+2 -1
View File
@@ -291,7 +291,8 @@ pub trait FileWritePort: Send + Sync + 'static {
size: u64, size: u64,
content_type: Option<String>, content_type: Option<String>,
pre_computed_hash: Option<String>, pre_computed_hash: Option<String>,
) -> Result<(), DomainError>; modified_at: Option<i64>,
) -> Result<String, DomainError>;
/// Registers file metadata WITHOUT writing content to disk (write-behind). /// Registers file metadata WITHOUT writing content to disk (write-behind).
/// ///
@@ -227,7 +227,8 @@ impl FileUploadUseCase for FileUploadService {
path: &str, path: &str,
content: &[u8], content: &[u8],
content_type: &str, content_type: &str,
) -> Result<(), DomainError> { modified_at: Option<i64>,
) -> Result<FileDto, DomainError> {
// Spool to temp file + hash // Spool to temp file + hash
let temp = tempfile::NamedTempFile::new() let temp = tempfile::NamedTempFile::new()
.map_err(|e| DomainError::internal_error("FileUpload", format!("temp file: {e}")))?; .map_err(|e| DomainError::internal_error("FileUpload", format!("temp file: {e}")))?;
@@ -242,6 +243,7 @@ impl FileUploadUseCase for FileUploadService {
content.len() as u64, content.len() as u64,
content_type, content_type,
Some(hash), Some(hash),
modified_at,
) )
.await .await
} }
@@ -260,21 +262,26 @@ impl FileUploadUseCase for FileUploadService {
size: u64, size: u64,
content_type: &str, content_type: &str,
pre_computed_hash: Option<String>, pre_computed_hash: Option<String>,
) -> Result<(), DomainError> { modified_at: Option<i64>,
) -> Result<FileDto, DomainError> {
// Try to find the existing file first // Try to find the existing file first
if let Some(file_read) = &self.file_read if let Some(file_read) = &self.file_read
&& let Some(file) = file_read.find_file_by_path(path).await? && let Some(file) = file_read.find_file_by_path(path).await?
{ {
let file_id = file.id().to_string();
self.file_write self.file_write
.update_file_content_from_temp( .update_file_content_from_temp(
file.id(), &file_id,
temp_path, temp_path,
size, size,
Some(content_type.to_string()), Some(content_type.to_string()),
pre_computed_hash, pre_computed_hash,
modified_at,
) )
.await?; .await?;
return Ok(()); // Re-read to get fresh DTO with updated etag and timestamps.
let updated = file_read.get_file(&file_id).await?;
return Ok(FileDto::from(updated));
} }
// File doesn't exist — create it via streaming upload // File doesn't exist — create it via streaming upload
@@ -297,7 +304,8 @@ impl FileUploadUseCase for FileUploadService {
None None
}; };
self.file_write let created = self
.file_write
.save_file_from_temp( .save_file_from_temp(
filename.to_string(), filename.to_string(),
parent_id, parent_id,
@@ -307,6 +315,6 @@ impl FileUploadUseCase for FileUploadService {
pre_computed_hash, pre_computed_hash,
) )
.await?; .await?;
Ok(()) Ok(FileDto::from(created))
} }
} }
@@ -204,8 +204,9 @@ impl FileWritePort for MockFileWritePort {
_size: u64, _size: u64,
_content_type: Option<String>, _content_type: Option<String>,
_pre_computed_hash: Option<String>, _pre_computed_hash: Option<String>,
) -> Result<(), DomainError> { _modified_at: Option<i64>,
Ok(()) ) -> Result<String, DomainError> {
Ok(String::new())
} }
async fn register_file_deferred( async fn register_file_deferred(
@@ -556,8 +556,9 @@ impl FileWritePort for MockFileRepository {
_size: u64, _size: u64,
_content_type: Option<String>, _content_type: Option<String>,
_pre_computed_hash: Option<String>, _pre_computed_hash: Option<String>,
) -> std::result::Result<(), DomainError> { _modified_at: Option<i64>,
Ok(()) ) -> std::result::Result<String, DomainError> {
Ok(String::new())
} }
async fn register_file_deferred( async fn register_file_deferred(
+9 -6
View File
@@ -185,8 +185,9 @@ impl FileWritePort for StubFileWritePort {
_size: u64, _size: u64,
_content_type: Option<String>, _content_type: Option<String>,
_pre_computed_hash: Option<String>, _pre_computed_hash: Option<String>,
) -> Result<(), DomainError> { _modified_at: Option<i64>,
Ok(()) ) -> Result<String, DomainError> {
Ok(String::new())
} }
async fn register_file_deferred( async fn register_file_deferred(
@@ -477,8 +478,9 @@ impl FileUploadUseCase for StubFileUploadUseCase {
_path: &str, _path: &str,
_content: &[u8], _content: &[u8],
_content_type: &str, _content_type: &str,
) -> Result<(), DomainError> { _modified_at: Option<i64>,
Ok(()) ) -> Result<FileDto, DomainError> {
Ok(FileDto::default())
} }
async fn update_file_streaming( async fn update_file_streaming(
@@ -488,8 +490,9 @@ impl FileUploadUseCase for StubFileUploadUseCase {
_size: u64, _size: u64,
_content_type: &str, _content_type: &str,
_pre_computed_hash: Option<String>, _pre_computed_hash: Option<String>,
) -> Result<(), DomainError> { _modified_at: Option<i64>,
Ok(()) ) -> Result<FileDto, DomainError> {
Ok(FileDto::default())
} }
} }
+1 -1
View File
@@ -243,7 +243,7 @@ impl Calendar {
pub fn update_color(&mut self, color: Option<String>) -> Result<()> { pub fn update_color(&mut self, color: Option<String>) -> Result<()> {
// Validate color format if provided // Validate color format if provided
if let Some(color_str) = &color { if let Some(color_str) = &color {
Self::validate_color(&color_str)?; Self::validate_color(color_str)?;
} }
self.color = color; self.color = color;
+44
View File
@@ -21,6 +21,7 @@ pub struct FileParts {
pub created_at: u64, pub created_at: u64,
pub modified_at: u64, pub modified_at: u64,
pub owner_id: Option<Uuid>, pub owner_id: Option<Uuid>,
pub etag: String,
} }
/** /**
@@ -64,6 +65,9 @@ pub struct File {
/// Owner user ID (from storage.files.user_id) /// Owner user ID (from storage.files.user_id)
owner_id: Option<Uuid>, owner_id: Option<Uuid>,
/// Content-addressable ETag (= blob_hash). Changes on every content write.
etag: String,
} }
// We no longer need this module, now we use a String directly // We no longer need this module, now we use a String directly
@@ -81,6 +85,7 @@ impl Default for File {
created_at: 0, created_at: 0,
modified_at: 0, modified_at: 0,
owner_id: None, owner_id: None,
etag: String::new(),
} }
} }
} }
@@ -119,6 +124,7 @@ impl File {
created_at: now, created_at: now,
modified_at: now, modified_at: now,
owner_id: None, owner_id: None,
etag: String::new(),
}) })
} }
@@ -150,6 +156,7 @@ impl File {
created_at, created_at,
modified_at, modified_at,
owner_id: None, owner_id: None,
etag: String::new(),
}) })
} }
@@ -164,6 +171,33 @@ impl File {
created_at: u64, created_at: u64,
modified_at: u64, modified_at: u64,
owner_id: Option<Uuid>, owner_id: Option<Uuid>,
) -> FileResult<Self> {
Self::with_timestamps_and_etag(
id,
name,
storage_path,
size,
mime_type,
folder_id,
created_at,
modified_at,
owner_id,
String::new(),
)
}
#[allow(clippy::too_many_arguments)]
pub fn with_timestamps_and_etag(
id: String,
name: String,
storage_path: StoragePath,
size: u64,
mime_type: String,
folder_id: Option<String>,
created_at: u64,
modified_at: u64,
owner_id: Option<Uuid>,
etag: String,
) -> FileResult<Self> { ) -> FileResult<Self> {
// Validate file name // Validate file name
if name.is_empty() || name.contains('/') || name.contains('\\') { if name.is_empty() || name.contains('/') || name.contains('\\') {
@@ -184,6 +218,7 @@ impl File {
created_at, created_at,
modified_at, modified_at,
owner_id, owner_id,
etag,
}) })
} }
@@ -203,9 +238,14 @@ impl File {
created_at: self.created_at, created_at: self.created_at,
modified_at: self.modified_at, modified_at: self.modified_at,
owner_id: self.owner_id, owner_id: self.owner_id,
etag: self.etag,
} }
} }
pub fn etag(&self) -> &str {
&self.etag
}
// Getters // Getters
pub fn id(&self) -> &str { pub fn id(&self) -> &str {
&self.id &self.id
@@ -273,6 +313,7 @@ impl File {
created_at, created_at,
modified_at, modified_at,
owner_id: None, owner_id: None,
etag: String::new(),
} }
} }
@@ -311,6 +352,7 @@ impl File {
created_at: self.created_at, created_at: self.created_at,
modified_at: now, modified_at: now,
owner_id: self.owner_id, owner_id: self.owner_id,
etag: self.etag.clone(),
}) })
} }
@@ -345,6 +387,7 @@ impl File {
created_at: self.created_at, created_at: self.created_at,
modified_at: now, modified_at: now,
owner_id: self.owner_id, owner_id: self.owner_id,
etag: self.etag.clone(),
}) })
} }
@@ -366,6 +409,7 @@ impl File {
created_at: self.created_at, created_at: self.created_at,
modified_at: now, modified_at: now,
owner_id: self.owner_id, owner_id: self.owner_id,
etag: self.etag.clone(),
} }
} }
} }
@@ -17,6 +17,7 @@ type MediaFileRow = (
String, // mime_type String, // mime_type
i64, // created_at i64, // created_at
i64, // updated_at i64, // updated_at
String, // blob_hash
Option<Uuid>, // user_id Option<Uuid>, // user_id
i64, // sort_date i64, // sort_date
); );
@@ -38,6 +39,7 @@ use crate::infrastructure::services::dedup_service::DedupService;
use uuid::Uuid; use uuid::Uuid;
/// Type alias for file metadata rows from SQL queries. /// Type alias for file metadata rows from SQL queries.
/// Fields: id, name, folder_id, folder_path, size, mime_type, created_at, updated_at, blob_hash, user_id
type FileRow = ( type FileRow = (
String, String,
String, String,
@@ -47,6 +49,7 @@ type FileRow = (
String, String,
i64, i64,
i64, i64,
String,
Option<Uuid>, Option<Uuid>,
); );
@@ -114,10 +117,11 @@ impl FileBlobReadRepository {
mime_type: String, mime_type: String,
created_at: i64, created_at: i64,
modified_at: i64, modified_at: i64,
etag: String,
owner_id: Option<Uuid>, owner_id: Option<Uuid>,
) -> Result<File, DomainError> { ) -> Result<File, DomainError> {
let storage_path = Self::make_file_path(folder_path.as_deref(), &name); let storage_path = Self::make_file_path(folder_path.as_deref(), &name);
File::with_timestamps( File::with_timestamps_and_etag(
id, id,
name, name,
storage_path, storage_path,
@@ -127,6 +131,7 @@ impl FileBlobReadRepository {
created_at as u64, created_at as u64,
modified_at as u64, modified_at as u64,
owner_id, owner_id,
etag,
) )
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("entity: {e}"))) .map_err(|e| DomainError::internal_error("FileBlobRead", format!("entity: {e}")))
} }
@@ -183,6 +188,7 @@ impl FileBlobReadRepository {
fi.size, fi.mime_type, fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint, EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint, EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id, fi.user_id,
EXTRACT(EPOCH FROM fi.media_sort_date)::bigint AS sort_date EXTRACT(EPOCH FROM fi.media_sort_date)::bigint AS sort_date
FROM storage.files fi FROM storage.files fi
@@ -206,9 +212,9 @@ impl FileBlobReadRepository {
let mut files = Vec::with_capacity(rows.len()); let mut files = Vec::with_capacity(rows.len());
let mut sort_dates = Vec::with_capacity(rows.len()); let mut sort_dates = Vec::with_capacity(rows.len());
for (id, name, fid, fpath, size, mime, ca, ma, uid, sd) in rows { for (id, name, fid, fpath, size, mime, ca, ma, etag, uid, sd) in rows {
files.push(Self::row_to_file( files.push(Self::row_to_file(
id, name, fid, fpath, size, mime, ca, ma, uid, id, name, fid, fpath, size, mime, ca, ma, etag, uid,
)?); )?);
sort_dates.push(sd); sort_dates.push(sd);
} }
@@ -257,7 +263,7 @@ impl FileReadPort for FileBlobReadRepository {
self.hash_cache.insert(id.to_string(), row.8.clone()); self.hash_cache.insert(id.to_string(), row.8.clone());
Self::row_to_file( Self::row_to_file(
row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7, row.9, row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7, row.8, row.9,
) )
} }
@@ -302,7 +308,7 @@ impl FileReadPort for FileBlobReadRepository {
self.hash_cache.insert(id.to_string(), row.8.clone()); self.hash_cache.insert(id.to_string(), row.8.clone());
Self::row_to_file( Self::row_to_file(
row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7, row.9, row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7, row.8, row.9,
) )
} }
@@ -315,6 +321,7 @@ impl FileReadPort for FileBlobReadRepository {
fi.size, fi.mime_type, fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint, EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint, EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id fi.user_id
FROM storage.files fi FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
@@ -332,6 +339,7 @@ impl FileReadPort for FileBlobReadRepository {
fi.size, fi.mime_type, fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint, EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint, EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id fi.user_id
FROM storage.files fi FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
@@ -345,8 +353,8 @@ impl FileReadPort for FileBlobReadRepository {
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("list: {e}")))?; .map_err(|e| DomainError::internal_error("FileBlobRead", format!("list: {e}")))?;
rows.into_iter() rows.into_iter()
.map(|(id, name, fid, fpath, size, mime, ca, ma, uid)| { .map(|(id, name, fid, fpath, size, mime, ca, ma, etag, uid)| {
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, uid) Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, etag, uid)
}) })
.collect() .collect()
} }
@@ -365,6 +373,7 @@ impl FileReadPort for FileBlobReadRepository {
fi.size, fi.mime_type, fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint, EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint, EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id fi.user_id
FROM storage.files fi FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
@@ -384,6 +393,7 @@ impl FileReadPort for FileBlobReadRepository {
fi.size, fi.mime_type, fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint, EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint, EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id fi.user_id
FROM storage.files fi FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
@@ -399,8 +409,8 @@ impl FileReadPort for FileBlobReadRepository {
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("list_for_owner: {e}")))?; .map_err(|e| DomainError::internal_error("FileBlobRead", format!("list_for_owner: {e}")))?;
rows.into_iter() rows.into_iter()
.map(|(id, name, fid, fpath, size, mime, ca, ma, uid)| { .map(|(id, name, fid, fpath, size, mime, ca, ma, etag, uid)| {
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, uid) Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, etag, uid)
}) })
.collect() .collect()
} }
@@ -427,6 +437,7 @@ impl FileReadPort for FileBlobReadRepository {
fi.size, fi.mime_type, fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint, EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint, EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id fi.user_id
FROM storage.files fi FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
@@ -447,6 +458,7 @@ impl FileReadPort for FileBlobReadRepository {
fi.size, fi.mime_type, fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint, EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint, EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id fi.user_id
FROM storage.files fi FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
@@ -463,8 +475,8 @@ impl FileReadPort for FileBlobReadRepository {
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("list_batch: {e}")))?; .map_err(|e| DomainError::internal_error("FileBlobRead", format!("list_batch: {e}")))?;
rows.into_iter() rows.into_iter()
.map(|(id, name, fid, fpath, size, mime, ca, ma, uid)| { .map(|(id, name, fid, fpath, size, mime, ca, ma, etag, uid)| {
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, uid) Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, etag, uid)
}) })
.collect() .collect()
} }
@@ -485,6 +497,7 @@ impl FileReadPort for FileBlobReadRepository {
fi.size, fi.mime_type, fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint, EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint, EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id fi.user_id
FROM storage.files fi FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
@@ -507,6 +520,7 @@ impl FileReadPort for FileBlobReadRepository {
fi.size, fi.mime_type, fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint, EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint, EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id fi.user_id
FROM storage.files fi FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
@@ -527,8 +541,8 @@ impl FileReadPort for FileBlobReadRepository {
})?; })?;
rows.into_iter() rows.into_iter()
.map(|(id, name, fid, fpath, size, mime, ca, ma, uid)| { .map(|(id, name, fid, fpath, size, mime, ca, ma, etag, uid)| {
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, uid) Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, etag, uid)
}) })
.collect() .collect()
} }
@@ -645,6 +659,7 @@ impl FileReadPort for FileBlobReadRepository {
String, String,
i64, i64,
i64, i64,
String,
Option<Uuid>, Option<Uuid>,
), ),
>( >(
@@ -653,6 +668,7 @@ impl FileReadPort for FileBlobReadRepository {
fi.size, fi.mime_type, fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint, EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint, EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id fi.user_id
FROM storage.files fi FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
@@ -675,6 +691,7 @@ impl FileReadPort for FileBlobReadRepository {
String, String,
i64, i64,
i64, i64,
String,
Option<Uuid>, Option<Uuid>,
), ),
>( >(
@@ -683,6 +700,7 @@ impl FileReadPort for FileBlobReadRepository {
fi.size, fi.mime_type, fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint, EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint, EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id fi.user_id
FROM storage.files fi FROM storage.files fi
JOIN storage.folders fo ON fo.id = fi.folder_id JOIN storage.folders fo ON fo.id = fi.folder_id
@@ -698,7 +716,7 @@ impl FileReadPort for FileBlobReadRepository {
match row { match row {
Some(r) => Ok(Some(Self::row_to_file( Some(r) => Ok(Some(Self::row_to_file(
r.0, r.1, r.2, r.3, r.4, r.5, r.6, r.7, r.8, r.0, r.1, r.2, r.3, r.4, r.5, r.6, r.7, r.8, r.9,
)?)), )?)),
None => Ok(None), None => Ok(None),
} }
@@ -718,13 +736,14 @@ impl FileReadPort for FileBlobReadRepository {
let stream = async_stream::try_stream! { let stream = async_stream::try_stream! {
let mut row_stream = sqlx::query_as::<_, ( let mut row_stream = sqlx::query_as::<_, (
String, String, Option<String>, Option<String>, String, String, Option<String>, Option<String>,
i64, String, i64, i64, Option<Uuid>, i64, String, i64, i64, String, Option<Uuid>,
)>( )>(
r#" r#"
SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path, SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
fi.size, fi.mime_type, fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint, EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint, EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id fi.user_id
FROM storage.files fi FROM storage.files fi
JOIN storage.folders fo ON fo.id = fi.folder_id JOIN storage.folders fo ON fo.id = fi.folder_id
@@ -739,9 +758,9 @@ impl FileReadPort for FileBlobReadRepository {
while let Some(row) = row_stream.try_next().await.map_err(|e| { while let Some(row) = row_stream.try_next().await.map_err(|e| {
DomainError::internal_error("FileBlobRead", format!("subtree stream: {e}")) DomainError::internal_error("FileBlobRead", format!("subtree stream: {e}"))
})? { })? {
let (id, name, fid, fpath, size, mime, ca, ma, uid) = row; let (id, name, fid, fpath, size, mime, ca, ma, etag, uid) = row;
let file = FileBlobReadRepository::row_to_file( let file = FileBlobReadRepository::row_to_file(
id, name, fid, fpath, size, mime, ca, ma, uid, id, name, fid, fpath, size, mime, ca, ma, etag, uid,
)?; )?;
yield file; yield file;
} }
@@ -803,6 +822,7 @@ impl FileReadPort for FileBlobReadRepository {
fi.size, fi.mime_type, \ fi.size, fi.mime_type, \
EXTRACT(EPOCH FROM fi.created_at)::bigint, \ EXTRACT(EPOCH FROM fi.created_at)::bigint, \
EXTRACT(EPOCH FROM fi.updated_at)::bigint, \ EXTRACT(EPOCH FROM fi.updated_at)::bigint, \
fi.blob_hash, \
fi.user_id, \ fi.user_id, \
COUNT(*) OVER() AS total_count \ COUNT(*) OVER() AS total_count \
FROM storage.files fi \ FROM storage.files fi \
@@ -824,6 +844,7 @@ impl FileReadPort for FileBlobReadRepository {
String, String,
i64, i64,
i64, i64,
String,
Option<Uuid>, Option<Uuid>,
i64, i64,
), ),
@@ -847,13 +868,15 @@ impl FileReadPort for FileBlobReadRepository {
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("search: {e}")))?; .map_err(|e| DomainError::internal_error("FileBlobRead", format!("search: {e}")))?;
// total_count is the same in every row; 0 when result set is empty. // total_count is the same in every row; 0 when result set is empty.
let total_count = rows.first().map_or(0, |r| r.9) as usize; let total_count = rows.first().map_or(0, |r| r.10) as usize;
let files = rows let files = rows
.into_iter() .into_iter()
.map(|(id, name, fid, fpath, size, mime, ca, ma, uid, _total)| { .map(
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, uid) |(id, name, fid, fpath, size, mime, ca, ma, etag, uid, _total)| {
}) Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, etag, uid)
},
)
.collect::<Result<Vec<_>, _>>() .collect::<Result<Vec<_>, _>>()
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("mapping: {e}")))?; .map_err(|e| DomainError::internal_error("FileBlobRead", format!("mapping: {e}")))?;
@@ -964,6 +987,7 @@ impl FileReadPort for FileBlobReadRepository {
fi.size, fi.mime_type, \ fi.size, fi.mime_type, \
EXTRACT(EPOCH FROM fi.created_at)::bigint, \ EXTRACT(EPOCH FROM fi.created_at)::bigint, \
EXTRACT(EPOCH FROM fi.updated_at)::bigint, \ EXTRACT(EPOCH FROM fi.updated_at)::bigint, \
fi.blob_hash, \
fi.user_id, \ fi.user_id, \
COUNT(*) OVER() AS total_count \ COUNT(*) OVER() AS total_count \
FROM storage.files fi \ FROM storage.files fi \
@@ -985,6 +1009,7 @@ impl FileReadPort for FileBlobReadRepository {
String, String,
i64, i64,
i64, i64,
String,
Option<Uuid>, Option<Uuid>,
i64, i64,
), ),
@@ -1029,13 +1054,15 @@ impl FileReadPort for FileBlobReadRepository {
DomainError::internal_error("FileBlobRead", format!("subtree search: {e}")) DomainError::internal_error("FileBlobRead", format!("subtree search: {e}"))
})?; })?;
let total_count = rows.first().map_or(0, |r| r.9) as usize; let total_count = rows.first().map_or(0, |r| r.10) as usize;
let files = rows let files = rows
.into_iter() .into_iter()
.map(|(id, name, fid, fpath, size, mime, ca, ma, uid, _total)| { .map(
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, uid) |(id, name, fid, fpath, size, mime, ca, ma, etag, uid, _total)| {
}) Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, etag, uid)
},
)
.collect::<Result<Vec<_>, _>>() .collect::<Result<Vec<_>, _>>()
.map_err(|e| { .map_err(|e| {
DomainError::internal_error("FileBlobRead", format!("subtree mapping: {e}")) DomainError::internal_error("FileBlobRead", format!("subtree mapping: {e}"))
@@ -1074,6 +1101,7 @@ impl FileReadPort for FileBlobReadRepository {
fi.size, fi.mime_type, fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint, EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint, EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id fi.user_id
FROM storage.files fi FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
@@ -1102,6 +1130,7 @@ impl FileReadPort for FileBlobReadRepository {
fi.size, fi.mime_type, fi.size, fi.mime_type,
EXTRACT(EPOCH FROM fi.created_at)::bigint, EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint, EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id fi.user_id
FROM storage.files fi FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
@@ -1126,8 +1155,8 @@ impl FileReadPort for FileBlobReadRepository {
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("suggest: {e}")))?; .map_err(|e| DomainError::internal_error("FileBlobRead", format!("suggest: {e}")))?;
rows.into_iter() rows.into_iter()
.map(|(id, name, fid, fpath, size, mime, ca, ma, uid)| { .map(|(id, name, fid, fpath, size, mime, ca, ma, etag, uid)| {
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, uid) Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, etag, uid)
}) })
.collect() .collect()
} }
@@ -100,9 +100,10 @@ impl FileBlobWriteRepository {
created_at: i64, created_at: i64,
modified_at: i64, modified_at: i64,
owner_id: Option<Uuid>, owner_id: Option<Uuid>,
etag: String,
) -> Result<File, DomainError> { ) -> Result<File, DomainError> {
let storage_path = Self::make_file_path(folder_path.as_deref(), &name); let storage_path = Self::make_file_path(folder_path.as_deref(), &name);
File::with_timestamps( File::with_timestamps_and_etag(
id, id,
name, name,
storage_path, storage_path,
@@ -112,6 +113,7 @@ impl FileBlobWriteRepository {
created_at as u64, created_at as u64,
modified_at as u64, modified_at as u64,
owner_id, owner_id,
etag,
) )
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("entity: {e}"))) .map_err(|e| DomainError::internal_error("FileBlobWrite", format!("entity: {e}")))
} }
@@ -132,12 +134,16 @@ impl FileBlobWriteRepository {
/// Uses a CTE to capture the old hash before updating so the old blob /// Uses a CTE to capture the old hash before updating so the old blob
/// reference can be decremented afterwards. Compensates on failure by /// reference can be decremented afterwards. Compensates on failure by
/// removing the new blob reference. /// removing the new blob reference.
///
/// `modified_at`: if `Some`, sets `updated_at` to that Unix timestamp;
/// if `None`, uses `NOW()` (server time). Returns the new hash on success.
async fn swap_blob_hash( async fn swap_blob_hash(
&self, &self,
file_id: &str, file_id: &str,
new_hash: &str, new_hash: &str,
new_size: i64, new_size: i64,
) -> Result<(), DomainError> { modified_at: Option<i64>,
) -> Result<String, DomainError> {
// Atomic CTE: capture old hash then update in one round-trip, no TOCTOU. // Atomic CTE: capture old hash then update in one round-trip, no TOCTOU.
let old_hash = match sqlx::query_scalar::<_, String>( let old_hash = match sqlx::query_scalar::<_, String>(
r#" r#"
@@ -145,7 +151,8 @@ impl FileBlobWriteRepository {
SELECT id, blob_hash FROM storage.files WHERE id = $3::uuid FOR UPDATE SELECT id, blob_hash FROM storage.files WHERE id = $3::uuid FOR UPDATE
) )
UPDATE storage.files f UPDATE storage.files f
SET blob_hash = $1, size = $2, updated_at = NOW() SET blob_hash = $1, size = $2,
updated_at = COALESCE(to_timestamp($4), NOW())
FROM old FROM old
WHERE f.id = old.id WHERE f.id = old.id
RETURNING old.blob_hash RETURNING old.blob_hash
@@ -154,6 +161,7 @@ impl FileBlobWriteRepository {
.bind(new_hash) .bind(new_hash)
.bind(new_size) .bind(new_size)
.bind(file_id) .bind(file_id)
.bind(modified_at.map(|t| t as f64))
.fetch_optional(self.pool.as_ref()) .fetch_optional(self.pool.as_ref())
.await .await
{ {
@@ -192,7 +200,7 @@ impl FileBlobWriteRepository {
); );
} }
Ok(()) Ok(new_hash.to_string())
} }
} }
@@ -277,6 +285,7 @@ impl FileWritePort for FileBlobWriteRepository {
row.1, row.1,
row.2, row.2,
Some(user_id), Some(user_id),
blob_hash.clone(),
) )
} }
@@ -314,6 +323,7 @@ impl FileWritePort for FileBlobWriteRepository {
row.5, row.5,
row.6, row.6,
None, None,
String::new(),
) )
} }
@@ -407,6 +417,7 @@ impl FileWritePort for FileBlobWriteRepository {
row.5, row.5,
row.6, row.6,
None, None,
row.7,
) )
} }
@@ -446,6 +457,7 @@ impl FileWritePort for FileBlobWriteRepository {
row.5, row.5,
row.6, row.6,
None, None,
String::new(),
) )
} }
@@ -474,7 +486,8 @@ impl FileWritePort for FileBlobWriteRepository {
size: u64, size: u64,
content_type: Option<String>, content_type: Option<String>,
pre_computed_hash: Option<String>, pre_computed_hash: Option<String>,
) -> Result<(), DomainError> { modified_at: Option<i64>,
) -> Result<String, DomainError> {
// Streaming: pass pre-computed hash so dedup skips re-reading the file. // Streaming: pass pre-computed hash so dedup skips re-reading the file.
let dedup_result = self let dedup_result = self
.dedup .dedup
@@ -482,7 +495,8 @@ impl FileWritePort for FileBlobWriteRepository {
.await?; .await?;
let new_hash = dedup_result.hash().to_string(); let new_hash = dedup_result.hash().to_string();
self.swap_blob_hash(file_id, &new_hash, size as i64).await self.swap_blob_hash(file_id, &new_hash, size as i64, modified_at)
.await
} }
async fn register_file_deferred( async fn register_file_deferred(
@@ -528,6 +542,7 @@ impl FileWritePort for FileBlobWriteRepository {
row.1, row.1,
row.2, row.2,
Some(user_id), Some(user_id),
String::new(),
)?; )?;
// The target_path is not meaningful for blob storage (content goes to .blobs/) // The target_path is not meaningful for blob storage (content goes to .blobs/)
@@ -175,6 +175,7 @@ impl PathResolverService {
size_formatted: format_file_size(sz), size_formatted: format_file_size(sz),
owner_id: uid, owner_id: uid,
sort_date: None, sort_date: None,
etag: String::new(),
})) }))
} }
} }
@@ -928,6 +928,7 @@ async fn handle_put(
total_bytes as u64, total_bytes as u64,
&content_type, &content_type,
Some(hash), Some(hash),
None,
) )
.await; .await;
@@ -280,6 +280,7 @@ async fn put_file(
total_bytes, total_bytes,
&content_type, &content_type,
Some(hash), Some(hash),
None,
) )
.await; .await;
+2 -2
View File
@@ -505,8 +505,8 @@ fn capabilities_payload(state: &AppState, ocs_version: u8) -> serde_json::Value
"chunking": "1.0" "chunking": "1.0"
}, },
"checksums": { "checksums": {
"preferredUploadType": "SHA1", "preferredUploadType": "",
"supportedTypes": ["SHA1", "MD5"] "supportedTypes": []
}, },
"files_sharing": { "files_sharing": {
"api_enabled": false, "api_enabled": false,
@@ -266,6 +266,7 @@ fn file_dto_from_search(fr: &crate::application::dtos::search_dto::SearchFileRes
size_formatted: format_file_size(fr.size), size_formatted: format_file_size(fr.size),
owner_id: None, owner_id: None,
sort_date: None, sort_date: None,
etag: String::new(),
} }
} }
+23 -8
View File
@@ -116,6 +116,12 @@ async fn handle_assemble(
.ok_or_else(|| AppError::bad_request("Missing Destination header"))? .ok_or_else(|| AppError::bad_request("Missing Destination header"))?
.to_string(); .to_string();
let oc_mtime = req
.headers()
.get("x-oc-mtime")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.parse::<i64>().ok());
let dest_subpath = extract_files_subpath(&destination, &user.username) let dest_subpath = extract_files_subpath(&destination, &user.username)
.ok_or_else(|| AppError::bad_request("Invalid Destination URL"))?; .ok_or_else(|| AppError::bad_request("Invalid Destination URL"))?;
@@ -144,11 +150,19 @@ async fn handle_assemble(
// Check if file exists (update vs create). // Check if file exists (update vs create).
let existing = file_service.get_file_by_path(&internal_path).await; let existing = file_service.get_file_by_path(&internal_path).await;
if existing.is_ok() { let etag: Option<String> = if existing.is_ok() {
upload_service let dto = upload_service
.update_file_streaming(&internal_path, &temp_path, size, &content_type, None) .update_file_streaming(
&internal_path,
&temp_path,
size,
&content_type,
None,
oc_mtime,
)
.await .await
.map_err(|e| AppError::internal_error(format!("Failed to update file: {}", e)))?; .map_err(|e| AppError::internal_error(format!("Failed to update file: {}", e)))?;
Some(dto.etag)
} else { } else {
// For new files we still need to read the temp file since create_file takes &[u8]. // For new files we still need to read the temp file since create_file takes &[u8].
let assembled = tokio::fs::read(&temp_path).await.map_err(|e| { let assembled = tokio::fs::read(&temp_path).await.map_err(|e| {
@@ -166,11 +180,12 @@ async fn handle_assemble(
); );
let parent_internal = parent_internal.trim_end_matches('/'); let parent_internal = parent_internal.trim_end_matches('/');
upload_service let dto = upload_service
.create_file(parent_internal, filename, &assembled, &content_type) .create_file(parent_internal, filename, &assembled, &content_type)
.await .await
.map_err(|e| AppError::internal_error(format!("Failed to create file: {}", e)))?; .map_err(|e| AppError::internal_error(format!("Failed to create file: {}", e)))?;
} Some(dto.etag)
};
// Clean up temp file (session cleanup below removes the directory anyway). // Clean up temp file (session cleanup below removes the directory anyway).
let _ = tokio::fs::remove_file(&temp_path).await; let _ = tokio::fs::remove_file(&temp_path).await;
@@ -178,11 +193,11 @@ async fn handle_assemble(
// Cleanup session. // Cleanup session.
let _ = nc.chunked_uploads.cleanup(&user.username, upload_id).await; let _ = nc.chunked_uploads.cleanup(&user.username, upload_id).await;
// Return etag if we can fetch the file. if let Some(tag) = etag {
if let Ok(file) = file_service.get_file_by_path(&internal_path).await {
return Ok(Response::builder() return Ok(Response::builder()
.status(StatusCode::CREATED) .status(StatusCode::CREATED)
.header(header::ETAG, format!("\"{}\"", file.id)) .header(header::ETAG, format!("\"{}\"", tag))
.header("oc-etag", format!("\"{}\"", tag))
.body(Body::empty()) .body(Body::empty())
.unwrap()); .unwrap());
} }
+11 -17
View File
@@ -526,7 +526,7 @@ async fn handle_put(
.unwrap_or("application/octet-stream") .unwrap_or("application/octet-stream")
.to_string(); .to_string();
let _oc_mtime = req let oc_mtime = req
.headers() .headers()
.get("x-oc-mtime") .get("x-oc-mtime")
.and_then(|v| v.to_str().ok()) .and_then(|v| v.to_str().ok())
@@ -545,24 +545,16 @@ async fn handle_put(
let existing = file_service.get_file_by_path(&internal_path).await; let existing = file_service.get_file_by_path(&internal_path).await;
if existing.is_ok() { if existing.is_ok() {
// Update existing file. // Update existing file — returns FileDto with fresh content-hash etag.
upload_service let updated = upload_service
.update_file(&internal_path, &body_bytes, &content_type) .update_file(&internal_path, &body_bytes, &content_type, oc_mtime)
.await .await
.map_err(|e| AppError::internal_error(format!("Failed to update file: {}", e)))?; .map_err(|e| AppError::internal_error(format!("Failed to update file: {}", e)))?;
// Re-fetch for etag.
if let Ok(updated) = file_service.get_file_by_path(&internal_path).await {
let builder = Response::builder()
.status(StatusCode::NO_CONTENT)
.header(header::ETAG, format!("\"{}\"", updated.id))
.header("oc-etag", format!("\"{}\"", updated.id));
return Ok(builder.body(Body::empty()).unwrap());
}
return Ok(Response::builder() return Ok(Response::builder()
.status(StatusCode::NO_CONTENT) .status(StatusCode::NO_CONTENT)
.header(header::ETAG, format!("\"{}\"", updated.etag))
.header("oc-etag", format!("\"{}\"", updated.etag))
.body(Body::empty()) .body(Body::empty())
.unwrap()); .unwrap());
} }
@@ -582,8 +574,8 @@ async fn handle_put(
let builder = Response::builder() let builder = Response::builder()
.status(StatusCode::CREATED) .status(StatusCode::CREATED)
.header(header::ETAG, format!("\"{}\"", file_dto.id)) .header(header::ETAG, format!("\"{}\"", file_dto.etag))
.header("oc-etag", format!("\"{}\"", file_dto.id)); .header("oc-etag", format!("\"{}\"", file_dto.etag));
Ok(builder.body(Body::empty()).unwrap()) Ok(builder.body(Body::empty()).unwrap())
} }
@@ -1126,7 +1118,7 @@ pub fn write_file_response<W: std::io::Write>(
.unwrap_or_else(Utc::now); .unwrap_or_else(Utc::now);
write_text_element(xml, "d:getlastmodified", &modified_at.to_rfc2822())?; write_text_element(xml, "d:getlastmodified", &modified_at.to_rfc2822())?;
write_text_element(xml, "d:getetag", &format!("\"{}\"", file.id))?; write_text_element(xml, "d:getetag", &format!("\"{}\"", file.etag))?;
write_text_element(xml, "d:creationdate", &created_at.to_rfc3339())?; write_text_element(xml, "d:creationdate", &created_at.to_rfc3339())?;
// Nextcloud/ownCloud properties // Nextcloud/ownCloud properties
@@ -1166,6 +1158,8 @@ pub fn write_file_response<W: std::io::Write>(
write_text_element(xml, "nc:is-encrypted", "0")?; write_text_element(xml, "nc:is-encrypted", "0")?;
write_text_element(xml, "nc:mount-type", "")?; write_text_element(xml, "nc:mount-type", "")?;
write_text_element(xml, "nc:creation_time", &file.created_at.to_string())?;
write_text_element(xml, "nc:upload_time", &file.modified_at.to_string())?;
xml.write_event(Event::End(BytesEnd::new("d:prop"))) xml.write_event(Event::End(BytesEnd::new("d:prop")))
.xml_err()?; .xml_err()?;