Fix Nextcloud sync conflict by using content-hash ETags
The Nextcloud Android client compares ETags before and after upload to verify its write landed. OxiCloud was returning the stable file UUID as the ETag, which never changed on content updates, causing false SYNC_CONFLICT errors on every upload. Five fixes applied: 1. Thread blob_hash (SHA-256) through File entity, FileDto, all read/write queries, and all WebDAV/PROPFIND responses as the ETag — changes on every content update, no DB migration needed. 2. Honor X-OC-Mtime header: parse the client-supplied mtime and use it for updated_at via COALESCE(to_timestamp($n), NOW()). 3. Disable phantom checksum capability (preferredUploadType/supportedTypes) that the server never actually implemented, stopping retry loops. 4. Add nc:creation_time and nc:upload_time to PROPFIND responses. 5. Return oc-etag header in chunked upload MOVE (assemble) responses. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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>;
|
||||||
}
|
}
|
||||||
|
|
||||||
// ─────────────────────────────────────────────────────
|
// ─────────────────────────────────────────────────────
|
||||||
|
|||||||
@@ -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
@@ -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())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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;
|
||||||
|
|||||||
@@ -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;
|
||||||
|
|
||||||
|
|||||||
@@ -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(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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());
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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()?;
|
||||||
|
|||||||
Reference in New Issue
Block a user