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

feat(drive): complete updated_by created_by

Browse Source
This commit is contained in:
Edouard Vanbelle
2026-06-19 10:51:13 +02:00
parent 06116dc6e7
commit e7f4826778
34 changed files with 987 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/infrastructure/repositories/pg/file_blob_read_repository.rs
+112 -48
View File
@@ -19,6 +19,8 @@ type MediaFileRow = (
i64, // updated_at
String, // blob_hash
Option<Uuid>, // user_id
Option<Uuid>, // created_by (§14 provenance)
Option<Uuid>, // updated_by (§14 provenance)
i64, // sort_date
Option<i32>, // width
Option<i32>, // height
@@ -42,7 +44,9 @@ use crate::infrastructure::services::dedup_service::DedupService;
use uuid::Uuid;
/// 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
/// Fields: id, name, folder_id, folder_path, size, mime_type,
/// created_at, updated_at, blob_hash, user_id, created_by, updated_by.
/// `created_by` / `updated_by` are the §14 provenance columns.
type FileRow = (
String,
String,
@@ -54,6 +58,8 @@ type FileRow = (
i64,
String,
Option<Uuid>,
Option<Uuid>,
Option<Uuid>,
);
/// Append the optional type/date/size filters from `criteria` to
@@ -228,7 +234,8 @@ impl FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint, \
EXTRACT(EPOCH FROM fi.updated_at)::bigint, \
fi.blob_hash, \
fi.user_id \
fi.user_id, \
fi.created_by, fi.updated_by \
FROM storage.files fi \
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id \
WHERE {where_clause}"
@@ -248,8 +255,10 @@ impl FileBlobReadRepository {
rows.into_iter()
.map(
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)| {
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub)| {
Self::row_to_file(
id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub,
)
},
)
.collect::<Result<Vec<_>, _>>()
@@ -277,7 +286,8 @@ impl FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint, \
EXTRACT(EPOCH FROM fi.updated_at)::bigint, \
fi.blob_hash, \
fi.user_id \
fi.user_id, \
fi.created_by, fi.updated_by \
FROM storage.files fi \
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id \
WHERE fi.id = ANY($1) AND NOT fi.is_trashed",
@@ -291,8 +301,10 @@ impl FileBlobReadRepository {
rows.into_iter()
.map(
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)| {
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub)| {
Self::row_to_file(
id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub,
)
},
)
.collect::<Result<Vec<_>, _>>()
@@ -357,9 +369,11 @@ impl FileBlobReadRepository {
modified_at: i64,
blob_hash: String,
owner_id: Option<Uuid>,
created_by: Option<Uuid>,
updated_by: Option<Uuid>,
) -> Result<File, DomainError> {
let storage_path = Self::make_file_path(folder_path.as_deref(), &name);
File::with_timestamps_and_blob_hash(
File::with_timestamps_blob_hash_and_provenance(
id,
name,
storage_path,
@@ -370,6 +384,8 @@ impl FileBlobReadRepository {
modified_at as u64,
owner_id,
blob_hash,
created_by,
updated_by,
)
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("entity: {e}")))
}
@@ -428,6 +444,7 @@ impl FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id,
fi.created_by, fi.updated_by,
EXTRACT(EPOCH FROM fi.media_sort_date)::bigint AS sort_date,
fm.width, fm.height
FROM storage.files fi
@@ -453,9 +470,9 @@ impl FileBlobReadRepository {
let mut sort_dates = Vec::with_capacity(rows.len());
let mut dims = Vec::with_capacity(rows.len());
for (id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, sd, w, h) in rows {
for (id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub, sd, w, h) in rows {
files.push(Self::row_to_file(
id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid,
id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub,
)?);
sort_dates.push(sd);
dims.push((w, h));
@@ -530,6 +547,8 @@ impl FileReadPort for FileBlobReadRepository {
i64, // updated_at
String, // blob_hash
Option<Uuid>, // user_id (owner)
Option<Uuid>, // created_by (§14)
Option<Uuid>, // updated_by (§14)
),
>(
r#"
@@ -538,7 +557,8 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id
fi.user_id,
fi.created_by, fi.updated_by
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.id = $1::uuid AND NOT fi.is_trashed
@@ -555,7 +575,7 @@ impl FileReadPort for FileBlobReadRepository {
self.hash_cache.insert(id.to_string(), row.8.clone());
Self::row_to_file(
row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7, row.8, row.9,
row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7, row.8, row.9, row.10, row.11,
)
}
@@ -576,6 +596,8 @@ impl FileReadPort for FileBlobReadRepository {
i64,
String,
Option<Uuid>,
Option<Uuid>, // created_by (§14)
Option<Uuid>, // updated_by (§14)
),
>(
r#"
@@ -584,7 +606,8 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id
fi.user_id,
fi.created_by, fi.updated_by
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.id = $1::uuid
@@ -598,7 +621,7 @@ impl FileReadPort for FileBlobReadRepository {
self.hash_cache.insert(id.to_string(), row.8.clone());
Self::row_to_file(
row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7, row.8, row.9,
row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7, row.8, row.9, row.10, row.11,
)
}
@@ -616,6 +639,8 @@ impl FileReadPort for FileBlobReadRepository {
i64, // updated_at
String, // blob_hash
Option<Uuid>, // user_id (owner)
Option<Uuid>, // created_by (§14)
Option<Uuid>, // updated_by (§14)
),
>(
r#"
@@ -624,7 +649,8 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id
fi.user_id,
fi.created_by, fi.updated_by
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.id = $1::uuid
@@ -643,7 +669,7 @@ impl FileReadPort for FileBlobReadRepository {
self.hash_cache.insert(id.to_string(), row.8.clone());
Self::row_to_file(
row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7, row.8, row.9,
row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7, row.8, row.9, row.10, row.11,
)
}
@@ -657,7 +683,8 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id
fi.user_id,
fi.created_by, fi.updated_by
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.folder_id = $1::uuid AND NOT fi.is_trashed
@@ -675,7 +702,8 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id
fi.user_id,
fi.created_by, fi.updated_by
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.folder_id IS NULL AND NOT fi.is_trashed
@@ -689,8 +717,10 @@ impl FileReadPort for FileBlobReadRepository {
rows.into_iter()
.map(
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)| {
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub)| {
Self::row_to_file(
id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub,
)
},
)
.collect()
@@ -711,7 +741,8 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id
fi.user_id,
fi.created_by, fi.updated_by
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.folder_id = $1::uuid AND NOT fi.is_trashed
@@ -731,7 +762,8 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id
fi.user_id,
fi.created_by, fi.updated_by
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.folder_id IS NULL AND NOT fi.is_trashed
@@ -747,8 +779,10 @@ impl FileReadPort for FileBlobReadRepository {
rows.into_iter()
.map(
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)| {
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub)| {
Self::row_to_file(
id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub,
)
},
)
.collect()
@@ -777,7 +811,8 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id
fi.user_id,
fi.created_by, fi.updated_by
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.folder_id = $1::uuid AND NOT fi.is_trashed
@@ -798,7 +833,8 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id
fi.user_id,
fi.created_by, fi.updated_by
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.folder_id IS NULL AND NOT fi.is_trashed
@@ -815,8 +851,10 @@ impl FileReadPort for FileBlobReadRepository {
rows.into_iter()
.map(
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)| {
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub)| {
Self::row_to_file(
id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub,
)
},
)
.collect()
@@ -839,7 +877,8 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id
fi.user_id,
fi.created_by, fi.updated_by
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.folder_id = $1::uuid AND NOT fi.is_trashed
@@ -862,7 +901,8 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id
fi.user_id,
fi.created_by, fi.updated_by
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.folder_id IS NULL AND NOT fi.is_trashed
@@ -883,8 +923,10 @@ impl FileReadPort for FileBlobReadRepository {
rows.into_iter()
.map(
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)| {
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub)| {
Self::row_to_file(
id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub,
)
},
)
.collect()
@@ -1035,6 +1077,8 @@ impl FileReadPort for FileBlobReadRepository {
i64,
String,
Option<Uuid>,
Option<Uuid>, // created_by (§14)
Option<Uuid>, // updated_by (§14)
),
>(
r#"
@@ -1043,7 +1087,8 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id
fi.user_id,
fi.created_by, fi.updated_by
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.name = $1 AND fi.folder_id IS NULL
@@ -1072,6 +1117,8 @@ impl FileReadPort for FileBlobReadRepository {
i64,
String,
Option<Uuid>,
Option<Uuid>, // created_by (§14)
Option<Uuid>, // updated_by (§14)
),
>(
r#"
@@ -1080,7 +1127,8 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id
fi.user_id,
fi.created_by, fi.updated_by
FROM storage.files fi
JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fo.path = $1 AND fi.name = $2
@@ -1097,7 +1145,7 @@ impl FileReadPort for FileBlobReadRepository {
match row {
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.9,
r.0, r.1, r.2, r.3, r.4, r.5, r.6, r.7, r.8, r.9, r.10, r.11,
)?)),
None => Ok(None),
}
@@ -1118,6 +1166,7 @@ impl FileReadPort for FileBlobReadRepository {
let mut row_stream = sqlx::query_as::<_, (
String, String, Option<String>, Option<String>,
i64, String, i64, i64, String, Option<Uuid>,
Option<Uuid>, Option<Uuid>,
)>(
r#"
SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
@@ -1125,7 +1174,8 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id
fi.user_id,
fi.created_by, fi.updated_by
FROM storage.files fi
JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = $1::uuid)
@@ -1139,9 +1189,9 @@ impl FileReadPort for FileBlobReadRepository {
while let Some(row) = row_stream.try_next().await.map_err(|e| {
DomainError::internal_error("FileBlobRead", format!("subtree stream: {e}"))
})? {
let (id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid) = row;
let (id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub) = row;
let file = FileBlobReadRepository::row_to_file(
id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid,
id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub,
)?;
yield file;
}
@@ -1205,6 +1255,7 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.updated_at)::bigint, \
fi.blob_hash, \
fi.user_id, \
fi.created_by, fi.updated_by, \
COUNT(*) OVER() AS total_count \
FROM storage.files fi \
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id \
@@ -1227,6 +1278,8 @@ impl FileReadPort for FileBlobReadRepository {
i64,
String,
Option<Uuid>,
Option<Uuid>, // created_by (§14)
Option<Uuid>, // updated_by (§14)
i64,
),
>(&sql)
@@ -1249,13 +1302,15 @@ impl FileReadPort for FileBlobReadRepository {
.map_err(|e| DomainError::internal_error("FileBlobRead", format!("search: {e}")))?;
// total_count is the same in every row; 0 when result set is empty.
let total_count = rows.first().map_or(0, |r| r.10) as usize;
let total_count = rows.first().map_or(0, |r| r.12) as usize;
let files = rows
.into_iter()
.map(
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, _total)| {
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub, _total)| {
Self::row_to_file(
id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub,
)
},
)
.collect::<Result<Vec<_>, _>>()
@@ -1331,6 +1386,7 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.updated_at)::bigint, \
fi.blob_hash, \
fi.user_id, \
fi.created_by, fi.updated_by, \
COUNT(*) OVER() AS total_count \
FROM storage.files fi \
JOIN storage.folders fo ON fo.id = fi.folder_id \
@@ -1353,6 +1409,8 @@ impl FileReadPort for FileBlobReadRepository {
i64,
String,
Option<Uuid>,
Option<Uuid>, // created_by (§14)
Option<Uuid>, // updated_by (§14)
i64,
),
>(&sql)
@@ -1373,13 +1431,15 @@ impl FileReadPort for FileBlobReadRepository {
DomainError::internal_error("FileBlobRead", format!("subtree search: {e}"))
})?;
let total_count = rows.first().map_or(0, |r| r.10) as usize;
let total_count = rows.first().map_or(0, |r| r.12) as usize;
let files = rows
.into_iter()
.map(
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, _total)| {
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub, _total)| {
Self::row_to_file(
id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub,
)
},
)
.collect::<Result<Vec<_>, _>>()
@@ -1421,7 +1481,8 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id
fi.user_id,
fi.created_by, fi.updated_by
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.folder_id = $1::uuid
@@ -1450,7 +1511,8 @@ impl FileReadPort for FileBlobReadRepository {
EXTRACT(EPOCH FROM fi.created_at)::bigint,
EXTRACT(EPOCH FROM fi.updated_at)::bigint,
fi.blob_hash,
fi.user_id
fi.user_id,
fi.created_by, fi.updated_by
FROM storage.files fi
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id
WHERE fi.folder_id IS NULL
@@ -1475,8 +1537,10 @@ impl FileReadPort for FileBlobReadRepository {
rows.into_iter()
.map(
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)| {
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub)| {
Self::row_to_file(
id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid, cb, ub,
)
},
)
.collect()
src/infrastructure/repositories/pg/file_blob_write_repository.rs
+158 -55
View File
@@ -115,9 +115,11 @@ impl FileBlobWriteRepository {
modified_at: i64,
owner_id: Option<Uuid>,
blob_hash: String,
created_by: Option<Uuid>,
updated_by: Option<Uuid>,
) -> Result<File, DomainError> {
let storage_path = Self::make_file_path(folder_path.as_deref(), &name);
File::with_timestamps_and_blob_hash(
File::with_timestamps_blob_hash_and_provenance(
id,
name,
storage_path,
@@ -128,6 +130,8 @@ impl FileBlobWriteRepository {
modified_at as u64,
owner_id,
blob_hash,
created_by,
updated_by,
)
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("entity: {e}")))
}
@@ -171,12 +175,18 @@ impl FileBlobWriteRepository {
/// `(new_hash, updated_at_epoch)` on success — the effective timestamp
/// is returned so callers can rebuild the fresh entity without
/// re-reading the row.
///
/// §14: `updated_by = $5` (caller_id). The caller mutated this
/// row — not the row's owner. D2 shared drives let non-owners
/// overwrite content; the previous `updated_by = f.user_id` would
/// have silently recorded the wrong principal.
async fn swap_blob_hash(
&self,
file_id: &str,
new_hash: &str,
new_size: i64,
modified_at: Option<i64>,
caller_id: Uuid,
) -> Result<(String, i64), DomainError> {
// Atomic CTE: capture old hash then update in one round-trip, no TOCTOU.
// Deadlock victims (40P01) retry before the compensation below runs —
@@ -190,7 +200,7 @@ impl FileBlobWriteRepository {
UPDATE storage.files f
SET blob_hash = $1, size = $2,
updated_at = COALESCE(to_timestamp($4), NOW()),
updated_by = f.user_id
updated_by = $5
FROM old
WHERE f.id = old.id
RETURNING old.blob_hash, EXTRACT(EPOCH FROM f.updated_at)::bigint
@@ -200,6 +210,7 @@ impl FileBlobWriteRepository {
.bind(new_size)
.bind(file_id)
.bind(modified_at.map(|t| t as f64))
.bind(caller_id)
.fetch_optional(self.pool.as_ref())
})
.await
@@ -245,6 +256,12 @@ impl FileBlobWriteRepository {
/// Register a file row pointing at a blob already stored in the chunk
/// store (the upload-ingest layer streamed the content in). Consumes the
/// caller's blob reference: any failure releases it before returning.
///
/// §14: `created_by = $7 = updated_by = caller_id` — authorship
/// belongs to the principal performing the upload, not to the parent
/// folder's owner. In D2 shared drives a non-owner member can upload
/// into a folder Alice owns; binding `parent.user_id` would have
/// silently recorded Alice as the author.
async fn save_file_with_blob_impl(
&self,
name: String,
@@ -252,6 +269,7 @@ impl FileBlobWriteRepository {
content_type: String,
blob_hash: &str,
size: u64,
caller_id: Uuid,
) -> Result<File, DomainError> {
// Root files have no parent folder to derive an owner from — keep the
// previous resolve_user_id(None) contract (release the ref, error out).
@@ -282,7 +300,7 @@ impl FileBlobWriteRepository {
// (a retried INSERT can legitimately lose to a concurrent identical
// upload).
let result = retry_on_deadlock("files.insert", || {
sqlx::query_as::<_, (String, Uuid, String, i64, i64)>(
sqlx::query_as::<_, (String, Uuid, String, i64, i64, Option<Uuid>, Option<Uuid>)>(
r#"
WITH parent AS (
SELECT id, user_id, drive_id, path FROM storage.folders WHERE id = $2::uuid
@@ -291,13 +309,15 @@ impl FileBlobWriteRepository {
(name, folder_id, user_id, drive_id, blob_hash, size,
mime_type, category_order, created_by, updated_by)
SELECT $1, parent.id, parent.user_id, parent.drive_id, $3, $4,
$5, $6, parent.user_id, parent.user_id
$5, $6, $7, $7
FROM parent
RETURNING id::text,
user_id,
(SELECT path FROM parent),
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
EXTRACT(EPOCH FROM updated_at)::bigint,
created_by,
updated_by
"#,
)
.bind(&name)
@@ -306,44 +326,46 @@ impl FileBlobWriteRepository {
.bind(size as i64)
.bind(&content_type)
.bind(category_order_for(&name, &content_type))
.bind(caller_id)
.fetch_optional(self.pool.as_ref())
})
.await;
let (id, user_id, folder_path, created_at, updated_at) = match result {
Ok(Some(row)) => row,
Ok(None) => {
if let Err(rollback_err) = self.dedup.remove_reference(blob_hash).await {
tracing::error!(
"Blob orphaned after missing parent folder — hash: {}, err: {}",
&blob_hash[..12],
rollback_err
);
let (id, user_id, folder_path, created_at, updated_at, created_by, updated_by) =
match result {
Ok(Some(row)) => row,
Ok(None) => {
if let Err(rollback_err) = self.dedup.remove_reference(blob_hash).await {
tracing::error!(
"Blob orphaned after missing parent folder — hash: {}, err: {}",
&blob_hash[..12],
rollback_err
);
}
return Err(DomainError::not_found("Folder", fid));
}
return Err(DomainError::not_found("Folder", fid));
}
Err(e) => {
if let Err(rollback_err) = self.dedup.remove_reference(blob_hash).await {
tracing::error!(
"Blob orphaned after failed INSERT — hash: {}, err: {}",
&blob_hash[..12],
rollback_err
);
}
if let sqlx::Error::Database(ref db_err) = e
&& db_err.code().as_deref() == Some("23505")
{
return Err(DomainError::already_exists(
"File",
format!("'{name}' already exists in this folder"),
Err(e) => {
if let Err(rollback_err) = self.dedup.remove_reference(blob_hash).await {
tracing::error!(
"Blob orphaned after failed INSERT — hash: {}, err: {}",
&blob_hash[..12],
rollback_err
);
}
if let sqlx::Error::Database(ref db_err) = e
&& db_err.code().as_deref() == Some("23505")
{
return Err(DomainError::already_exists(
"File",
format!("'{name}' already exists in this folder"),
));
}
return Err(DomainError::internal_error(
"FileBlobWrite",
format!("insert: {e}"),
));
}
return Err(DomainError::internal_error(
"FileBlobWrite",
format!("insert: {e}"),
));
}
};
};
tracing::info!(
"📡 STREAMING WRITE: {} ({} bytes, hash: {})",
@@ -363,6 +385,8 @@ impl FileBlobWriteRepository {
updated_at,
Some(user_id),
blob_hash.to_string(),
created_by,
updated_by,
)
}
}
@@ -375,8 +399,9 @@ impl FileWritePort for FileBlobWriteRepository {
content_type: String,
blob_hash: &str,
size: u64,
caller_id: Uuid,
) -> Result<File, DomainError> {
self.save_file_with_blob_impl(name, folder_id, content_type, blob_hash, size)
self.save_file_with_blob_impl(name, folder_id, content_type, blob_hash, size, caller_id)
.await
}
@@ -384,9 +409,30 @@ impl FileWritePort for FileBlobWriteRepository {
&self,
file_id: &str,
target_folder_id: Option<String>,
caller_id: Uuid,
) -> Result<File, DomainError> {
// If moving to a different folder, get the new user_id (must be same user)
let row = sqlx::query_as::<_, (String, String, Option<String>, i64, String, i64, i64)>(
// If moving to a different folder, get the new user_id (must be same user).
//
// §14: `updated_by = $3` (caller_id) — the caller mutated this
// row. The previous COALESCE derived authorship from the
// destination folder's owner, which is wrong: dest's user_id
// has no claim to authorship of the file's content. D2 shared
// drives surface this most starkly (Alice moves Bob's file
// into Charlie's drive — `updated_by` must be Alice).
let row = sqlx::query_as::<
_,
(
String,
String,
Option<String>,
i64,
String,
i64,
i64,
Option<Uuid>,
Option<Uuid>,
),
>(
r#"
WITH dest AS (
SELECT user_id, drive_id FROM storage.folders WHERE id = $1::uuid
@@ -396,15 +442,17 @@ impl FileWritePort for FileBlobWriteRepository {
user_id = COALESCE((SELECT user_id FROM dest), f.user_id),
drive_id = COALESCE((SELECT drive_id FROM dest), f.drive_id),
updated_at = NOW(),
updated_by = COALESCE((SELECT user_id FROM dest), f.user_id)
updated_by = $3
WHERE f.id = $2::uuid AND NOT f.is_trashed
RETURNING f.id::text, f.name, f.folder_id::text, f.size, f.mime_type,
EXTRACT(EPOCH FROM f.created_at)::bigint,
EXTRACT(EPOCH FROM f.updated_at)::bigint
EXTRACT(EPOCH FROM f.updated_at)::bigint,
f.created_by, f.updated_by
"#,
)
.bind(&target_folder_id)
.bind(file_id)
.bind(caller_id)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("move: {e}")))?
@@ -422,6 +470,8 @@ impl FileWritePort for FileBlobWriteRepository {
row.6,
None,
String::new(),
row.7,
row.8,
)
}
@@ -430,9 +480,16 @@ impl FileWritePort for FileBlobWriteRepository {
file_id: &str,
target_folder_id: Option<String>,
new_name: Option<&str>,
caller_id: Uuid,
) -> Result<File, DomainError> {
// Atomic CTE: read source file → insert new row with same blob_hash → increment ref_count.
// Single round-trip; blob content is NOT copied (dedup makes this zero-copy).
//
// §14: `created_by = $4 = updated_by = caller_id` — the caller
// authored this copy. The previous binding used
// `dest_folder.user_id` which silently recorded the destination
// folder's owner as the author when Adam copied a file into
// Alice's folder.
let target_fid = target_folder_id.clone();
let rename_to = new_name.map(|s| s.to_string());
@@ -448,6 +505,8 @@ impl FileWritePort for FileBlobWriteRepository {
i64,
i64,
String,
Option<Uuid>,
Option<Uuid>,
),
>(
r#"
@@ -480,13 +539,15 @@ impl FileWritePort for FileBlobWriteRepository {
src.size,
src.mime_type,
src.category_order,
dest_folder.user_id,
dest_folder.user_id
$4,
$4
FROM src, dest_folder
RETURNING id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
blob_hash
blob_hash,
created_by,
updated_by
)
SELECT * FROM new_file
"#,
@@ -494,6 +555,7 @@ impl FileWritePort for FileBlobWriteRepository {
.bind(file_id)
.bind(&target_fid)
.bind(&rename_to)
.bind(caller_id)
.fetch_optional(self.pool.as_ref())
})
.await
@@ -539,22 +601,45 @@ impl FileWritePort for FileBlobWriteRepository {
row.6,
None,
row.7,
row.8,
row.9,
)
}
async fn rename_file(&self, file_id: &str, new_name: &str) -> Result<File, DomainError> {
let row = sqlx::query_as::<_, (String, String, Option<String>, i64, String, i64, i64)>(
async fn rename_file(
&self,
file_id: &str,
new_name: &str,
caller_id: Uuid,
) -> Result<File, DomainError> {
// §14: `updated_by = $3` (caller_id), see move_file.
let row = sqlx::query_as::<
_,
(
String,
String,
Option<String>,
i64,
String,
i64,
i64,
Option<Uuid>,
Option<Uuid>,
),
>(
r#"
UPDATE storage.files
SET name = $1, updated_at = NOW(), updated_by = user_id
SET name = $1, updated_at = NOW(), updated_by = $3
WHERE id = $2::uuid AND NOT is_trashed
RETURNING id::text, name, folder_id::text, size, mime_type,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
EXTRACT(EPOCH FROM updated_at)::bigint,
created_by, updated_by
"#,
)
.bind(new_name)
.bind(file_id)
.bind(caller_id)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| {
@@ -579,6 +664,8 @@ impl FileWritePort for FileBlobWriteRepository {
row.6,
None,
String::new(),
row.7,
row.8,
)
}
@@ -608,12 +695,13 @@ impl FileWritePort for FileBlobWriteRepository {
blob_hash: &str,
size: u64,
modified_at: Option<i64>,
caller_id: Uuid,
) -> Result<(String, i64), DomainError> {
// The content was already ingested into the chunk store by the
// upload-ingest layer; swap_blob_hash consumes its reference and
// releases it on failure.
let swapped = self
.swap_blob_hash(file_id, blob_hash, size as i64, modified_at)
.swap_blob_hash(file_id, blob_hash, size as i64, modified_at, caller_id)
.await?;
// The file now maps to a different blob — drop the read-side cache
// entry so streaming downloads cannot serve the previous content
@@ -628,6 +716,7 @@ impl FileWritePort for FileBlobWriteRepository {
folder_id: Option<String>,
content_type: String,
size: u64,
caller_id: Uuid,
) -> Result<(File, PathBuf), DomainError> {
let (user_id, drive_id) = self.resolve_owner_and_drive(folder_id.as_deref()).await?;
@@ -635,16 +724,22 @@ impl FileWritePort for FileBlobWriteRepository {
// The write-behind cache will call update_file_content later.
let placeholder_hash = "0000000000000000000000000000000000000000000000000000000000000000";
// §14: `created_by = $9 = updated_by = caller_id`. The legacy
// `user_id` column (dropped in D7) stays bound to the parent
// folder's owner; only the two provenance columns flip to the
// caller — see save_file_with_blob_impl.
let row = retry_on_deadlock("files.insert_deferred", || {
sqlx::query_as::<_, (String, i64, i64)>(
sqlx::query_as::<_, (String, i64, i64, Option<Uuid>, Option<Uuid>)>(
r#"
INSERT INTO storage.files
(name, folder_id, user_id, drive_id, blob_hash, size,
mime_type, category_order, created_by, updated_by)
VALUES ($1, $2::uuid, $3, $4, $5, $6, $7, $8, $3, $3)
VALUES ($1, $2::uuid, $3, $4, $5, $6, $7, $8, $9, $9)
RETURNING id::text,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint
EXTRACT(EPOCH FROM updated_at)::bigint,
created_by,
updated_by
"#,
)
.bind(&name)
@@ -655,6 +750,7 @@ impl FileWritePort for FileBlobWriteRepository {
.bind(size as i64)
.bind(&content_type)
.bind(category_order_for(&name, &content_type))
.bind(caller_id)
.fetch_one(self.pool.as_ref())
})
.await
@@ -672,6 +768,8 @@ impl FileWritePort for FileBlobWriteRepository {
row.2,
Some(user_id),
String::new(),
row.3,
row.4,
)?;
// The target_path is not meaningful for blob storage (content goes to .blobs/)
@@ -683,7 +781,8 @@ impl FileWritePort for FileBlobWriteRepository {
// ── Trash operations ──
async fn move_to_trash(&self, file_id: &str) -> Result<(), DomainError> {
async fn move_to_trash(&self, file_id: &str, caller_id: Uuid) -> Result<(), DomainError> {
// §14: `updated_by = $2` (caller_id), see move_file.
let result = sqlx::query(
r#"
UPDATE storage.files
@@ -691,11 +790,12 @@ impl FileWritePort for FileBlobWriteRepository {
trashed_at = NOW(),
original_folder_id = folder_id,
updated_at = NOW(),
updated_by = user_id
updated_by = $2
WHERE id = $1::uuid AND NOT is_trashed
"#,
)
.bind(file_id)
.bind(caller_id)
.execute(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("trash: {e}")))?;
@@ -710,7 +810,9 @@ impl FileWritePort for FileBlobWriteRepository {
&self,
file_id: &str,
_original_path: &str,
caller_id: Uuid,
) -> Result<(), DomainError> {
// §14: `updated_by = $2` (caller_id), see move_file.
let result = sqlx::query(
r#"
UPDATE storage.files
@@ -719,11 +821,12 @@ impl FileWritePort for FileBlobWriteRepository {
folder_id = COALESCE(original_folder_id, folder_id),
original_folder_id = NULL,
updated_at = NOW(),
updated_by = user_id
updated_by = $2
WHERE id = $1::uuid AND is_trashed
"#,
)
.bind(file_id)
.bind(caller_id)
.execute(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("restore: {e}")))?;
src/infrastructure/repositories/pg/folder_db_repository.rs
+164 -69
View File
@@ -22,11 +22,12 @@ use crate::domain::services::path_service::StoragePath;
/// Type alias for folder metadata rows from SQL queries.
/// Tuple order: id, name, path, parent_id, user_id, drive_id,
/// created_at, modified_at, tree_modified_at. The trailing
/// `tree_modified_at` feeds [`Folder::etag`] — every SELECT here
/// must include `EXTRACT(EPOCH FROM tree_modified_at)::bigint`.
/// created_at, modified_at, tree_modified_at, created_by, updated_by.
/// The trailing `tree_modified_at` feeds [`Folder::etag`] — every
/// SELECT here must include `EXTRACT(EPOCH FROM tree_modified_at)::bigint`.
/// `drive_id` is the post-D0 `NOT NULL` scope axis for path-based
/// lookups.
/// lookups. `created_by` / `updated_by` are the §14 provenance
/// columns, nullable because the FK is `ON DELETE SET NULL`.
type FolderRow = (
String,
String,
@@ -37,10 +38,12 @@ type FolderRow = (
i64,
i64,
i64,
Option<Uuid>,
Option<Uuid>,
);
/// Type alias for paginated folder rows (includes total_count as
/// the last element after `tree_modified_at`).
/// the last element after the §14 provenance columns).
type FolderRowPaginated = (
String,
String,
@@ -51,10 +54,13 @@ type FolderRowPaginated = (
i64,
i64,
i64,
Option<Uuid>,
Option<Uuid>,
i64,
);
/// Type alias for folder rows with optional user_id.
/// Includes the §14 provenance columns `created_by` / `updated_by`.
type FolderRowOptUser = (
String,
String,
@@ -65,6 +71,8 @@ type FolderRowOptUser = (
i64,
i64,
i64,
Option<Uuid>,
Option<Uuid>,
);
/// PostgreSQL-backed folder repository.
@@ -98,7 +106,9 @@ impl FolderDbRepository {
/// Convert a database row into a `Folder` domain entity.
///
/// The `path` comes directly from the materialized `path` column — no
/// extra queries needed.
/// extra queries needed. `created_by` / `updated_by` carry the
/// §14 provenance signal through the entity layer; both are
/// `Option<Uuid>` because the FK is `ON DELETE SET NULL`.
#[allow(clippy::too_many_arguments)]
fn row_to_folder(
id: String,
@@ -110,9 +120,11 @@ impl FolderDbRepository {
created_at: i64,
modified_at: i64,
tree_modified_at: i64,
created_by: Option<Uuid>,
updated_by: Option<Uuid>,
) -> Result<Folder, DomainError> {
let storage_path = StoragePath::from_string(&path);
Folder::with_timestamps_and_tree(
Folder::with_timestamps_tree_and_provenance(
id,
name,
storage_path,
@@ -122,6 +134,8 @@ impl FolderDbRepository {
created_at as u64,
modified_at as u64,
tree_modified_at as u64,
created_by,
updated_by,
)
.map_err(|e| DomainError::internal_error("FolderDb", format!("entity: {e}")))
}
@@ -139,10 +153,11 @@ impl FolderDbRepository {
let rows = sqlx::query_as::<_, FolderRow>(
r#"
SELECT id::text, name, path, parent_id::text, user_id,
SELECT id::text, name, path, parent_id::text, user_id, drive_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by, updated_by
FROM storage.folders
WHERE id = ANY($1) AND NOT is_trashed
"#,
@@ -153,7 +168,9 @@ impl FolderDbRepository {
.map_err(|e| DomainError::internal_error("FolderDb", format!("get_folders_by_ids: {e}")))?;
rows.into_iter()
.map(|r| Self::row_to_folder(r.0, r.1, r.2, r.3, Some(r.4), r.5, r.6, r.7))
.map(|r| {
Self::row_to_folder(r.0, r.1, r.2, r.3, Some(r.4), r.5, r.6, r.7, r.8, r.9, r.10)
})
.collect()
}
}
@@ -163,6 +180,7 @@ impl FolderRepository for FolderDbRepository {
&self,
name: String,
parent_id: Option<String>,
caller_id: Uuid,
) -> Result<Folder, DomainError> {
// Derive (user_id, drive_id) from parent folder in one round-trip.
// Root-level folders require the caller to have set up the home
@@ -184,28 +202,34 @@ impl FolderRepository for FolderDbRepository {
));
};
// D0 dual-write: drive_id alongside user_id (drops in D7), plus
// provenance columns created_by/updated_by. The repo derives
// created_by from user_id because the parent's owner is the
// creator on personal drives (the only kind that exists in D0).
// D2 plumbs the real caller_id when shared drives let other
// members write into a drive they don't own.
let row = sqlx::query_as::<_, (String, String, i64, i64, i64)>(
// D0 dual-write: drive_id alongside user_id (drops in D7); plus
// §14 provenance — `created_by` / `updated_by` bind to the caller
// ($5), NOT to the parent folder's `user_id`. Pre-D2 they're
// silently equivalent (only the parent's owner can write); the
// distinction matters once shared drives let an Editor mutate
// a folder owned by someone else.
//
// RETURNING also surfaces the two provenance columns so the
// built entity / DTO carries fresh values without a re-read.
let row = sqlx::query_as::<_, (String, String, i64, i64, i64, Option<Uuid>, Option<Uuid>)>(
r#"
INSERT INTO storage.folders
(name, parent_id, user_id, drive_id, created_by, updated_by)
VALUES ($1, $2::uuid, $3, $4, $3, $3)
VALUES ($1, $2::uuid, $3, $4, $5, $5)
RETURNING id::text,
path,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by,
updated_by
"#,
)
.bind(&name)
.bind(&parent_id)
.bind(user_id)
.bind(drive_id)
.bind(caller_id)
.fetch_one(self.pool())
.await
.map_err(|e| {
@@ -230,6 +254,9 @@ impl FolderRepository for FolderDbRepository {
row.2,
row.3,
row.4,
// Fresh from RETURNING — caller_id was bound to both columns.
row.5,
row.6,
)
}
@@ -239,7 +266,8 @@ impl FolderRepository for FolderDbRepository {
SELECT id::text, name, path, parent_id::text, user_id, drive_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by, updated_by
FROM storage.folders
WHERE id = $1::uuid AND NOT is_trashed
"#,
@@ -260,6 +288,8 @@ impl FolderRepository for FolderDbRepository {
row.6,
row.7,
row.8,
row.9,
row.10,
)
}
@@ -288,7 +318,8 @@ impl FolderRepository for FolderDbRepository {
SELECT id::text, name, path, parent_id::text, user_id, drive_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by, updated_by
FROM storage.folders
WHERE path = $1 AND drive_id = $2 AND NOT is_trashed
"#,
@@ -310,6 +341,8 @@ impl FolderRepository for FolderDbRepository {
row.6,
row.7,
row.8,
row.9,
row.10,
)
}
@@ -321,7 +354,8 @@ impl FolderRepository for FolderDbRepository {
SELECT id::text, name, path, parent_id::text, user_id, drive_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by, updated_by
FROM storage.folders
WHERE parent_id = $1::uuid AND NOT is_trashed
ORDER BY name
@@ -336,7 +370,8 @@ impl FolderRepository for FolderDbRepository {
SELECT id::text, name, path, parent_id::text, user_id, drive_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by, updated_by
FROM storage.folders
WHERE parent_id IS NULL AND NOT is_trashed
ORDER BY name
@@ -348,8 +383,8 @@ impl FolderRepository for FolderDbRepository {
.map_err(|e| DomainError::internal_error("FolderDb", format!("list: {e}")))?;
rows.into_iter()
.map(|(id, name, path, pid, uid, did, ca, ma, tma)| {
Self::row_to_folder(id, name, path, pid, Some(uid), did, ca, ma, tma)
.map(|(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)| {
Self::row_to_folder(id, name, path, pid, Some(uid), did, ca, ma, tma, cb, ub)
})
.collect()
}
@@ -366,7 +401,8 @@ impl FolderRepository for FolderDbRepository {
SELECT id::text, name, path, parent_id::text, user_id, drive_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by, updated_by
FROM storage.folders
WHERE parent_id = $1::uuid AND user_id = $2 AND NOT is_trashed
ORDER BY name
@@ -382,7 +418,8 @@ impl FolderRepository for FolderDbRepository {
SELECT id::text, name, path, parent_id::text, user_id, drive_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by, updated_by
FROM storage.folders
WHERE parent_id IS NULL AND user_id = $1 AND NOT is_trashed
ORDER BY name
@@ -395,8 +432,8 @@ impl FolderRepository for FolderDbRepository {
.map_err(|e| DomainError::internal_error("FolderDb", format!("list_by_owner: {e}")))?;
rows.into_iter()
.map(|(id, name, path, pid, uid, did, ca, ma, tma)| {
Self::row_to_folder(id, name, path, pid, Some(uid), did, ca, ma, tma)
.map(|(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)| {
Self::row_to_folder(id, name, path, pid, Some(uid), did, ca, ma, tma, cb, ub)
})
.collect()
}
@@ -419,6 +456,7 @@ impl FolderRepository for FolderDbRepository {
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by, updated_by,
COUNT(*) OVER() AS total_count
FROM storage.folders
WHERE parent_id = $1::uuid AND NOT is_trashed
@@ -438,6 +476,7 @@ impl FolderRepository for FolderDbRepository {
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by, updated_by,
COUNT(*) OVER() AS total_count
FROM storage.folders
WHERE parent_id IS NULL AND NOT is_trashed
@@ -454,16 +493,18 @@ impl FolderRepository for FolderDbRepository {
// total_count is identical in every row; 0 when the result set is empty.
let total = if include_total {
Some(rows.first().map_or(0, |r| r.9) as usize)
Some(rows.first().map_or(0, |r| r.11) as usize)
} else {
None
};
let folders: Result<Vec<Folder>, DomainError> = rows
.into_iter()
.map(|(id, name, path, pid, uid, did, ca, ma, tma, _total)| {
Self::row_to_folder(id, name, path, pid, Some(uid), did, ca, ma, tma)
})
.map(
|(id, name, path, pid, uid, did, ca, ma, tma, cb, ub, _total)| {
Self::row_to_folder(id, name, path, pid, Some(uid), did, ca, ma, tma, cb, ub)
},
)
.collect();
Ok((folders?, total))
}
@@ -486,6 +527,7 @@ impl FolderRepository for FolderDbRepository {
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by, updated_by,
COUNT(*) OVER() AS total_count
FROM storage.folders
WHERE parent_id = $1::uuid AND user_id = $2 AND NOT is_trashed
@@ -506,6 +548,7 @@ impl FolderRepository for FolderDbRepository {
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by, updated_by,
COUNT(*) OVER() AS total_count
FROM storage.folders
WHERE parent_id IS NULL AND user_id = $1 AND NOT is_trashed
@@ -522,41 +565,55 @@ impl FolderRepository for FolderDbRepository {
.map_err(|e| DomainError::internal_error("FolderDb", format!("paginate_by_owner: {e}")))?;
let total = if include_total {
Some(rows.first().map_or(0, |r| r.9) as usize)
Some(rows.first().map_or(0, |r| r.11) as usize)
} else {
None
};
let folders: Result<Vec<Folder>, DomainError> = rows
.into_iter()
.map(|(id, name, path, pid, uid, did, ca, ma, tma, _total)| {
Self::row_to_folder(id, name, path, pid, Some(uid), did, ca, ma, tma)
})
.map(
|(id, name, path, pid, uid, did, ca, ma, tma, cb, ub, _total)| {
Self::row_to_folder(id, name, path, pid, Some(uid), did, ca, ma, tma, cb, ub)
},
)
.collect();
Ok((folders?, total))
}
async fn rename_folder(&self, id: &str, new_name: String) -> Result<Folder, DomainError> {
async fn rename_folder(
&self,
id: &str,
new_name: String,
caller_id: Uuid,
) -> Result<Folder, DomainError> {
// The BEFORE UPDATE trigger recomputes path/lpath for this row;
// the AFTER UPDATE cascade trigger then batch-updates all
// descendants in a single UPDATE using the GiST lpath index.
// That multi-row rewrite can deadlock against the tree-ETag
// flusher's id-ordered ancestor bump — retry instead of failing
// the user's operation (40P01 only; 23505 still maps below).
//
// §14: `updated_by = $3` (caller_id) — the caller mutated this
// row, not the row's owner. In D2 a shared-drive member can
// rename a row they don't own; the previous `updated_by = user_id`
// would have silently recorded the wrong principal.
let row = retry_on_deadlock("folders.rename", || {
sqlx::query_as::<_, FolderRow>(
r#"
UPDATE storage.folders
SET name = $1, updated_at = NOW(), updated_by = user_id
SET name = $1, updated_at = NOW(), updated_by = $3
WHERE id = $2::uuid AND NOT is_trashed
RETURNING id::text, name, path, parent_id::text, user_id, drive_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by, updated_by
"#,
)
.bind(&new_name)
.bind(id)
.bind(caller_id)
.fetch_optional(self.pool())
})
.await
@@ -580,6 +637,8 @@ impl FolderRepository for FolderDbRepository {
row.6,
row.7,
row.8,
row.9,
row.10,
)
}
@@ -587,25 +646,30 @@ impl FolderRepository for FolderDbRepository {
&self,
id: &str,
new_parent_id: Option<&str>,
caller_id: Uuid,
) -> Result<Folder, DomainError> {
// The BEFORE UPDATE trigger recomputes path/lpath for this row;
// the AFTER UPDATE cascade trigger then batch-updates all
// descendants in a single UPDATE using the GiST lpath index.
// Retried on deadlock vs the tree-ETag flusher (see rename_folder).
//
// §14: `updated_by = $3` (caller_id), see rename_folder.
let row = retry_on_deadlock("folders.move", || {
sqlx::query_as::<_, FolderRow>(
r#"
UPDATE storage.folders
SET parent_id = $1::uuid, updated_at = NOW(), updated_by = user_id
SET parent_id = $1::uuid, updated_at = NOW(), updated_by = $3
WHERE id = $2::uuid AND NOT is_trashed
RETURNING id::text, name, path, parent_id::text, user_id, drive_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by, updated_by
"#,
)
.bind(new_parent_id)
.bind(id)
.bind(caller_id)
.fetch_optional(self.pool())
})
.await
@@ -622,6 +686,8 @@ impl FolderRepository for FolderDbRepository {
row.6,
row.7,
row.8,
row.9,
row.10,
)
}
@@ -697,7 +763,7 @@ impl FolderRepository for FolderDbRepository {
// ── Trash operations ──
async fn move_to_trash(&self, folder_id: &str) -> Result<(), DomainError> {
async fn move_to_trash(&self, folder_id: &str, caller_id: Uuid) -> Result<(), DomainError> {
// Soft-delete the whole subtree in one statement: the root flips
// `is_trashed` and records `original_parent_id` so restore knows
// where to put it back; every descendant (folder or file) that
@@ -712,6 +778,10 @@ impl FolderRepository for FolderDbRepository {
// `/g9-tree/file.txt` still resolved 207 even though the parent
// collection was gone) — a class of data-integrity drift that
// confused desktop-sync tree walks.
//
// §14: all three CTE branches stamp `updated_by = $2`
// (caller_id). The cascade is "the caller trashed this
// subtree", not "each owner trashed their own row".
let result = retry_on_deadlock("folders.trash", || {
sqlx::query_scalar::<_, i64>(
r#"
@@ -721,7 +791,7 @@ impl FolderRepository for FolderDbRepository {
trashed_at = NOW(),
original_parent_id = parent_id,
updated_at = NOW(),
updated_by = user_id
updated_by = $2
WHERE id = $1::uuid AND NOT is_trashed
RETURNING id, lpath
),
@@ -730,7 +800,7 @@ impl FolderRepository for FolderDbRepository {
SET is_trashed = TRUE,
trashed_at = NOW(),
updated_at = NOW(),
updated_by = f.user_id
updated_by = $2
FROM trash_root tr
WHERE f.lpath <@ tr.lpath
AND f.id != tr.id
@@ -742,7 +812,7 @@ impl FolderRepository for FolderDbRepository {
SET is_trashed = TRUE,
trashed_at = NOW(),
updated_at = NOW(),
updated_by = fi.user_id
updated_by = $2
FROM trash_root tr
JOIN storage.folders f ON f.lpath <@ tr.lpath
WHERE fi.folder_id = f.id
@@ -753,6 +823,7 @@ impl FolderRepository for FolderDbRepository {
"#,
)
.bind(folder_id)
.bind(caller_id)
.fetch_one(self.pool())
})
.await
@@ -769,6 +840,7 @@ impl FolderRepository for FolderDbRepository {
&self,
folder_id: &str,
_original_path: &str,
caller_id: Uuid,
) -> Result<(), DomainError> {
// Inverse of the cascade in `move_to_trash`: restore the root
// (BEFORE UPDATE trigger recomputes path/lpath via the parent_id
@@ -778,6 +850,10 @@ impl FolderRepository for FolderDbRepository {
// *before* this folder went to trash have `original_*` set, so
// they correctly stay in trash and continue to show up as
// top-level trash entries via `storage.trash_items`.
//
// §14: all three CTE branches stamp `updated_by = $2`
// (caller_id). Restoration is "the caller restored this
// subtree", regardless of who originally owned each row.
let result = retry_on_deadlock("folders.restore", || {
sqlx::query_scalar::<_, i64>(
r#"
@@ -788,7 +864,7 @@ impl FolderRepository for FolderDbRepository {
parent_id = COALESCE(original_parent_id, parent_id),
original_parent_id = NULL,
updated_at = NOW(),
updated_by = user_id
updated_by = $2
WHERE id = $1::uuid AND is_trashed
RETURNING id, lpath
),
@@ -797,7 +873,7 @@ impl FolderRepository for FolderDbRepository {
SET is_trashed = FALSE,
trashed_at = NULL,
updated_at = NOW(),
updated_by = f.user_id
updated_by = $2
FROM restore_root rr
WHERE f.lpath <@ rr.lpath
AND f.id != rr.id
@@ -810,7 +886,7 @@ impl FolderRepository for FolderDbRepository {
SET is_trashed = FALSE,
trashed_at = NULL,
updated_at = NOW(),
updated_by = fi.user_id
updated_by = $2
FROM restore_root rr
JOIN storage.folders f ON f.lpath <@ rr.lpath
WHERE fi.folder_id = f.id
@@ -822,6 +898,7 @@ impl FolderRepository for FolderDbRepository {
"#,
)
.bind(folder_id)
.bind(caller_id)
.fetch_one(self.pool())
})
.await
@@ -911,16 +988,25 @@ impl FolderRepository for FolderDbRepository {
ca,
ma,
tma,
// INSERT stamped both provenance columns from user_id
// (D0 dual-write); D2 will plumb the real caller_id.
Some(user_id),
Some(user_id),
),
None => {
// Already exists — fetch it
let existing = sqlx::query_as::<_, (String, String, i64, i64, i64)>(
// Already exists — fetch it. SELECT also pulls the §14
// provenance columns so the entity layer reflects DB truth.
let existing = sqlx::query_as::<
_,
(String, String, i64, i64, i64, Option<Uuid>, Option<Uuid>),
>(
r#"
SELECT id::text,
path,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by, updated_by
FROM storage.folders
WHERE name = $1 AND user_id = $2 AND parent_id IS NULL
"#,
@@ -940,6 +1026,8 @@ impl FolderRepository for FolderDbRepository {
existing.2,
existing.3,
existing.4,
existing.5,
existing.6,
)
}
}
@@ -955,7 +1043,8 @@ impl FolderRepository for FolderDbRepository {
fo.user_id, fo.drive_id, \
EXTRACT(EPOCH FROM fo.created_at)::bigint, \
EXTRACT(EPOCH FROM fo.updated_at)::bigint, \
EXTRACT(EPOCH FROM fo.tree_modified_at)::bigint \
EXTRACT(EPOCH FROM fo.tree_modified_at)::bigint, \
fo.created_by, fo.updated_by \
FROM storage.folders fo \
WHERE fo.is_trashed = false \
AND fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = $1::uuid) \
@@ -970,8 +1059,8 @@ impl FolderRepository for FolderDbRepository {
})?;
rows.into_iter()
.map(|(id, name, path, pid, uid, did, ca, ma, tma)| {
Self::row_to_folder(id, name, path, pid, uid, did, ca, ma, tma)
.map(|(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)| {
Self::row_to_folder(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)
})
.collect()
}
@@ -1018,7 +1107,8 @@ impl FolderRepository for FolderDbRepository {
fo.user_id, fo.drive_id, \
EXTRACT(EPOCH FROM fo.created_at)::bigint, \
EXTRACT(EPOCH FROM fo.updated_at)::bigint, \
EXTRACT(EPOCH FROM fo.tree_modified_at)::bigint \
EXTRACT(EPOCH FROM fo.tree_modified_at)::bigint, \
fo.created_by, fo.updated_by \
FROM storage.folders fo \
WHERE fo.user_id = $1 \
AND fo.is_trashed = false \
@@ -1042,8 +1132,8 @@ impl FolderRepository for FolderDbRepository {
return rows
.into_iter()
.map(|(id, name, path, pid, uid, did, ca, ma, tma)| {
Self::row_to_folder(id, name, path, pid, uid, did, ca, ma, tma)
.map(|(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)| {
Self::row_to_folder(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)
})
.collect();
}
@@ -1055,7 +1145,8 @@ impl FolderRepository for FolderDbRepository {
fo.user_id, fo.drive_id, \
EXTRACT(EPOCH FROM fo.created_at)::bigint, \
EXTRACT(EPOCH FROM fo.updated_at)::bigint, \
EXTRACT(EPOCH FROM fo.tree_modified_at)::bigint \
EXTRACT(EPOCH FROM fo.tree_modified_at)::bigint, \
fo.created_by, fo.updated_by \
FROM storage.folders fo \
WHERE fo.parent_id = $1::uuid \
AND fo.user_id = $2 \
@@ -1074,7 +1165,8 @@ impl FolderRepository for FolderDbRepository {
fo.user_id, fo.drive_id, \
EXTRACT(EPOCH FROM fo.created_at)::bigint, \
EXTRACT(EPOCH FROM fo.updated_at)::bigint, \
EXTRACT(EPOCH FROM fo.tree_modified_at)::bigint \
EXTRACT(EPOCH FROM fo.tree_modified_at)::bigint, \
fo.created_by, fo.updated_by \
FROM storage.folders fo \
WHERE fo.parent_id IS NULL \
AND fo.user_id = $1 \
@@ -1114,8 +1206,8 @@ impl FolderRepository for FolderDbRepository {
.map_err(|e| DomainError::internal_error("FolderDb", format!("search_folders: {e}")))?;
rows.into_iter()
.map(|(id, name, path, pid, uid, did, ca, ma, tma)| {
Self::row_to_folder(id, name, path, pid, uid, did, ca, ma, tma)
.map(|(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)| {
Self::row_to_folder(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)
})
.collect()
}
@@ -1143,7 +1235,8 @@ impl FolderRepository for FolderDbRepository {
fo.user_id, fo.drive_id, \
EXTRACT(EPOCH FROM fo.created_at)::bigint, \
EXTRACT(EPOCH FROM fo.updated_at)::bigint, \
EXTRACT(EPOCH FROM fo.tree_modified_at)::bigint \
EXTRACT(EPOCH FROM fo.tree_modified_at)::bigint, \
fo.created_by, fo.updated_by \
FROM storage.folders fo \
WHERE fo.user_id = $1 \
AND fo.is_trashed = false \
@@ -1170,8 +1263,8 @@ impl FolderRepository for FolderDbRepository {
.map_err(|e| DomainError::internal_error("FolderDb", format!("descendant search: {e}")))?;
rows.into_iter()
.map(|(id, name, path, pid, uid, did, ca, ma, tma)| {
Self::row_to_folder(id, name, path, pid, uid, did, ca, ma, tma)
.map(|(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)| {
Self::row_to_folder(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)
})
.collect()
}
@@ -1192,7 +1285,8 @@ impl FolderRepository for FolderDbRepository {
SELECT id::text, name, path, parent_id::text, user_id, drive_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by, updated_by
FROM storage.folders
WHERE parent_id = $1::uuid
AND NOT is_trashed
@@ -1218,7 +1312,8 @@ impl FolderRepository for FolderDbRepository {
SELECT id::text, name, path, parent_id::text, user_id, drive_id,
EXTRACT(EPOCH FROM created_at)::bigint,
EXTRACT(EPOCH FROM updated_at)::bigint,
EXTRACT(EPOCH FROM tree_modified_at)::bigint
EXTRACT(EPOCH FROM tree_modified_at)::bigint,
created_by, updated_by
FROM storage.folders
WHERE parent_id IS NULL
AND NOT is_trashed
@@ -1241,8 +1336,8 @@ impl FolderRepository for FolderDbRepository {
.map_err(|e| DomainError::internal_error("FolderDb", format!("suggest: {e}")))?;
rows.into_iter()
.map(|(id, name, path, pid, uid, did, ca, ma, tma)| {
Self::row_to_folder(id, name, path, pid, Some(uid), did, ca, ma, tma)
.map(|(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)| {
Self::row_to_folder(id, name, path, pid, Some(uid), did, ca, ma, tma, cb, ub)
})
.collect()
}
src/infrastructure/services/path_resolver_service.rs
+9
View File
@@ -162,6 +162,12 @@ impl PathResolverService {
icon_class: Arc::from("fas fa-folder"),
icon_special_class: Arc::from("folder-icon"),
category: Arc::from("Folder"),
// §14 provenance not selected by this resolver path —
// it's used for existence/type discrimination, not
// detailed DTO emission. Callers that need provenance
// reload through the repo.
created_by: None,
updated_by: None,
})),
_ => {
let mime = mime_type.unwrap_or_else(|| "application/octet-stream".to_string());
@@ -188,6 +194,9 @@ impl PathResolverService {
sort_date: None,
content_hash: String::new(),
etag: String::new(),
// §14 provenance not selected by this resolver path
created_by: None,
updated_by: None,
}))
}
}