feat(drive): stop writing in {file,folder}.user_id
plan:
```
What step 5 is actually doing
Goal: stop writing to storage.files.user_id / storage.folders.user_id on new INSERTs. This is a prerequisite for step 6 (dropping the column entirely). The column has existed since the pre-D0 single-owner era; every read that used to key on it has already been migrated to drive-membership grants over the last N days.
What has to change together for this to be safe:
┌──────────────┬─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┬───────────────────────────────────────────────────────────────────────────────────────────────────────┐
│ Piece │ What changes │ Why │
├──────────────┼─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┼───────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ Migration │ Swap storage.files uniqueness indexes from (folder_id, name, user_id) to (drive_id, folder_id, name) │ Otherwise, two new rows with user_id = NULL would both be allowed (PG treats NULLs as distinct) — │
│ (a) │ │ uniqueness silently breaks │
├──────────────┼─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┼───────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ Migration │ ALTER user_id DROP NOT NULL on both tables │ Otherwise, dropping the INSERT bind violates NOT NULL and every write 500s │
│ (b) │ │ │
├──────────────┼─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┼───────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ Migration │ Drop dead user_id-leading indexes │ Cheap cleanup — nothing scans them anymore │
│ (c) │ │ │
├──────────────┼─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┼───────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ PL/pgSQL (d) │ Rewrite storage.copy_folder_tree without user_id in the INSERT column list │ Cross-drive copy runs entirely in SQL, needs the same treatment │
├──────────────┼─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┼───────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ Rust (e) │ ~12 INSERT column-list drops across folder_db_repository, file_blob_write_repository, drive_pg_repository, dedup_service, │ Actual write path │
│ │ folder_service, load-seed.rs │ │
├──────────────┼─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┼───────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ Rust (f) │ Simplify folder_db_repository::create_folder's parent lookup to only fetch drive_id (was fetching (user_id, drive_id)) │ It's fetching a value it no longer needs │
└──────────────┴─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┴───────────────────────────────────────────────────────────────────────────────────────────────────────┘
```
This commit is contained in:
@@ -124,7 +124,10 @@ pub struct FavoriteResourceRow {
|
||||
pub size: i64,
|
||||
pub resource_created_at: DateTime<Utc>,
|
||||
pub modified_at: DateTime<Utc>,
|
||||
pub owner_id: Uuid,
|
||||
/// Post-D7: nullable on new rows (the legacy `storage.{files,folders}.user_id`
|
||||
/// column is no longer written). `is_owner` is now `false` when
|
||||
/// this is `None` — see the SQL projection in favorites repo.
|
||||
pub owner_id: Option<Uuid>,
|
||||
/// Drive that owns this row. Surfaced on the favorites listing
|
||||
/// so a UI can tell when a favorited item lives in a different
|
||||
/// drive than the user's home (post-D6 cross-drive moves +
|
||||
|
||||
@@ -205,7 +205,11 @@ pub struct FolderResourceRow {
|
||||
pub size: i64,
|
||||
pub created_at: DateTime<Utc>,
|
||||
pub modified_at: DateTime<Utc>,
|
||||
pub owner_id: Uuid,
|
||||
/// Post-D7: the legacy `user_id` column on `storage.{files,folders}`
|
||||
/// is nullable — new rows leave it NULL — so this optional. UI
|
||||
/// surfaces should prefer `created_by` / `updated_by` on the
|
||||
/// per-resource DTO instead.
|
||||
pub owner_id: Option<Uuid>,
|
||||
/// Drive that owns this row. Same column as
|
||||
/// `storage.folders.drive_id` / `storage.files.drive_id`. Surfaced
|
||||
/// on the listing so a UI can tell when a child lives in a
|
||||
|
||||
@@ -104,7 +104,10 @@ pub struct RecentResourceRow {
|
||||
pub size: i64,
|
||||
pub resource_created_at: DateTime<Utc>,
|
||||
pub modified_at: DateTime<Utc>,
|
||||
pub owner_id: Uuid,
|
||||
/// Post-D7: nullable on new rows (the legacy
|
||||
/// `storage.{files,folders}.user_id` column is no longer written).
|
||||
/// Consumers should prefer §14 provenance columns.
|
||||
pub owner_id: Option<Uuid>,
|
||||
/// Drive that owns this row. Surfaced on the recent listing
|
||||
/// so a UI can tell when a recently-accessed item lives in a
|
||||
/// different drive than the user's home (post-D6 cross-drive
|
||||
|
||||
@@ -60,7 +60,10 @@ pub struct TrashResourceRow {
|
||||
pub size: i64,
|
||||
pub resource_created_at: DateTime<Utc>,
|
||||
pub modified_at: DateTime<Utc>,
|
||||
pub owner_id: Uuid,
|
||||
/// Post-D7: nullable on new rows (the legacy `storage.{files,folders}.user_id`
|
||||
/// column is no longer written). Consumers should prefer §14
|
||||
/// provenance columns when available.
|
||||
pub owner_id: Option<Uuid>,
|
||||
/// Drive the trashed item belongs to. Surfaced verbatim on the wire
|
||||
/// (`TrashResourceItemDto.drive_id`) so the `/trash` UI can group by
|
||||
/// drive without an extra lookup per row. D2b: filtering by drive is
|
||||
|
||||
@@ -317,19 +317,22 @@ impl FileUploadService {
|
||||
/// Incremental (`+size`, O(1)) and fire-and-forget on a background task, so
|
||||
/// it adds neither latency nor a `SUM(size)` over the user's whole library
|
||||
/// to the upload path (the previous full recompute was O(N) per upload,
|
||||
/// O(N²) for a bulk upload). Keyed by the file's `owner_id`; drift — e.g.
|
||||
/// deletes, which don't decrement — is reconciled by the periodic sweep. A
|
||||
/// DTO without a resolvable owner is simply left to that sweep.
|
||||
fn maybe_update_storage_usage(&self, file: &FileDto) {
|
||||
/// O(N²) for a bulk upload). Drift — e.g. deletes, which don't decrement —
|
||||
/// is reconciled by the periodic sweep.
|
||||
///
|
||||
/// Post-D7: `file.owner_id` is now nullable and unpopulated on new
|
||||
/// rows, so the envelope owner comes from `caller_id` (the user who
|
||||
/// just did the upload). The user-side delta is guarded by
|
||||
/// `add_user_storage_usage_delta_if_personal` — it only fires when
|
||||
/// the target drive is `kind='personal'`, so a shared-drive upload
|
||||
/// still doesn't touch any user envelope.
|
||||
fn maybe_update_storage_usage(&self, file: &FileDto, caller_id: Uuid) {
|
||||
let Some(storage_service) = &self.storage_usage_service else {
|
||||
return;
|
||||
};
|
||||
let delta = file.size as i64;
|
||||
|
||||
let owner = file
|
||||
.owner_id
|
||||
.as_deref()
|
||||
.and_then(|s| Uuid::parse_str(s).ok());
|
||||
let owner = Some(caller_id);
|
||||
let folder = file
|
||||
.folder_id
|
||||
.as_deref()
|
||||
@@ -410,7 +413,7 @@ impl FileUploadUseCase for FileUploadService {
|
||||
"📡 STREAMING UPLOAD: {} ({} bytes, ID: {})",
|
||||
name, blob.size, dto.id
|
||||
);
|
||||
self.maybe_update_storage_usage(&dto);
|
||||
self.maybe_update_storage_usage(&dto, caller_id);
|
||||
if let Some(hook) = &self.file_lifecycle_hook {
|
||||
hook.on_file_created(&dto.id, &dto.content_hash, &dto.mime_type, blob.is_new_blob);
|
||||
}
|
||||
|
||||
@@ -1081,17 +1081,18 @@ mod cascade_hook_integration_tests {
|
||||
let blob_hash = blake3::hash(format!("cascade-{label}-{}", Uuid::new_v4()).as_bytes())
|
||||
.to_hex()
|
||||
.to_string();
|
||||
// Post-D7: `user_id` omitted — the column is nullable and
|
||||
// provenance flows through `created_by` / `updated_by`.
|
||||
sqlx::query_scalar(
|
||||
"INSERT INTO storage.files
|
||||
(name, user_id, drive_id, folder_id, blob_hash, size, created_by, updated_by)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $7)
|
||||
(name, drive_id, folder_id, blob_hash, size, created_by, updated_by)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $6)
|
||||
RETURNING id",
|
||||
)
|
||||
.bind(format!(
|
||||
"rust-test-cascade-{label}-{}",
|
||||
&Uuid::new_v4().to_string()[..8]
|
||||
))
|
||||
.bind(user_id)
|
||||
.bind(drive_id)
|
||||
.bind(folder_id)
|
||||
.bind(&blob_hash)
|
||||
|
||||
@@ -962,7 +962,7 @@ fn row_to_item_dto(row: TrashResourceRow) -> TrashResourceItemDto {
|
||||
name: row.name.clone(),
|
||||
path,
|
||||
parent_id: row.parent_id.map(|u| u.to_string()),
|
||||
owner_id: Some(row.owner_id.to_string()),
|
||||
owner_id: row.owner_id.map(|u| u.to_string()),
|
||||
// D2b: the trash listing query now SELECTs `drive_id` (the
|
||||
// unified view exposes it). Surfaced so per-drive grouping
|
||||
// in the `/trash` UI doesn't need an extra lookup per row.
|
||||
@@ -1013,7 +1013,7 @@ fn row_to_item_dto(row: TrashResourceRow) -> TrashResourceItemDto {
|
||||
icon_special_class: std::sync::Arc::from(icon_special_class_for(&row.name, mime)),
|
||||
category: std::sync::Arc::from(category_for(&row.name, mime)),
|
||||
size_formatted: format_file_size(size_bytes),
|
||||
owner_id: Some(row.owner_id.to_string()),
|
||||
owner_id: row.owner_id.map(|u| u.to_string()),
|
||||
sort_date: None,
|
||||
content_hash,
|
||||
etag,
|
||||
|
||||
+16
-10
@@ -526,10 +526,12 @@ async fn build_subtree(
|
||||
|
||||
// Level 0 — the subtree's "root" sits inside `mount_under`, not at
|
||||
// parent_id=NULL. drive_id is inherited from the mount point.
|
||||
// Post-D7: `user_id` omitted; `created_by` / `updated_by` bind to
|
||||
// the seed caller.
|
||||
let root: (Uuid,) = sqlx::query_as(
|
||||
"INSERT INTO storage.folders
|
||||
(name, parent_id, user_id, drive_id, created_by, updated_by)
|
||||
SELECT $1, parent.id, $2, parent.drive_id, $2, $2
|
||||
(name, parent_id, drive_id, created_by, updated_by)
|
||||
SELECT $1, parent.id, parent.drive_id, $2, $2
|
||||
FROM storage.folders parent
|
||||
WHERE parent.id = $3::uuid
|
||||
RETURNING id",
|
||||
@@ -568,13 +570,13 @@ async fn build_subtree(
|
||||
);
|
||||
|
||||
// drive_id derives from the parent folder — same pattern as
|
||||
// file_blob_write_repository's resolve_owner_and_drive helper.
|
||||
// Every parent in `current_level` already has a drive_id set,
|
||||
// so the JOIN is guaranteed to find one.
|
||||
// file_blob_write_repository's resolve_parent_drive helper.
|
||||
// Post-D7: `user_id` omitted; provenance via `created_by` /
|
||||
// `updated_by`.
|
||||
let rows: Vec<(Uuid,)> = sqlx::query_as(
|
||||
"INSERT INTO storage.folders
|
||||
(name, parent_id, user_id, drive_id, created_by, updated_by)
|
||||
SELECT f.name, f.parent_id, $1, parent.drive_id, $1, $1
|
||||
(name, parent_id, drive_id, created_by, updated_by)
|
||||
SELECT f.name, f.parent_id, parent.drive_id, $1, $1
|
||||
FROM UNNEST($2::uuid[], $3::text[]) AS f(parent_id, name)
|
||||
JOIN storage.folders parent ON parent.id = f.parent_id
|
||||
RETURNING id",
|
||||
@@ -653,13 +655,17 @@ async fn insert_files(
|
||||
|
||||
// Post-D0: storage.files.drive_id is NOT NULL — derive it from the
|
||||
// parent folder (same pattern as file_blob_write_repository's
|
||||
// INSERTs and the resolve_owner_and_drive helper). The folder's
|
||||
// INSERTs and the resolve_parent_drive helper). The folder's
|
||||
// drive_id was set during the M2 backfill or by the lifecycle hook
|
||||
// for users provisioned after D0.
|
||||
//
|
||||
// Post-D7: `user_id` omitted; `created_by` / `updated_by` bind to
|
||||
// the seed caller so provenance is preserved.
|
||||
sqlx::query(
|
||||
"INSERT INTO storage.files
|
||||
(name, folder_id, user_id, drive_id, blob_hash, size, mime_type)
|
||||
SELECT f.name, f.folder_id, $1, fo.drive_id, $2, 0, 'text/plain'
|
||||
(name, folder_id, drive_id, blob_hash, size, mime_type,
|
||||
created_by, updated_by)
|
||||
SELECT f.name, f.folder_id, fo.drive_id, $2, 0, 'text/plain', $1, $1
|
||||
FROM UNNEST($3::uuid[], $4::text[]) AS f(folder_id, name)
|
||||
JOIN storage.folders fo ON fo.id = f.folder_id",
|
||||
)
|
||||
|
||||
@@ -143,11 +143,16 @@ impl DriveRepository for DrivePgRepository {
|
||||
|
||||
// 2. Root folder. `parent_id IS NULL` makes it a root in the
|
||||
// drive; `drive_id` closes the FK in this direction.
|
||||
//
|
||||
// Post-D7: `user_id` omitted from the INSERT column list —
|
||||
// the column is nullable and no longer written to on new
|
||||
// rows. `created_by` / `updated_by` bind to the owner
|
||||
// (§14 provenance).
|
||||
let folder_id: Uuid = sqlx::query_scalar(
|
||||
r#"
|
||||
INSERT INTO storage.folders
|
||||
(name, parent_id, user_id, drive_id, created_by, updated_by)
|
||||
VALUES ('Personal', NULL, $1, $2, $1, $1)
|
||||
(name, parent_id, drive_id, created_by, updated_by)
|
||||
VALUES ('Personal', NULL, $2, $1, $1)
|
||||
RETURNING id
|
||||
"#,
|
||||
)
|
||||
@@ -244,14 +249,14 @@ impl DriveRepository for DrivePgRepository {
|
||||
.await
|
||||
.map_err(|e| Self::map_sqlx_err("create_shared_drive_atomic.drive", e))?;
|
||||
|
||||
// 2. Root folder. The folder's `user_id` carries the admin (legacy
|
||||
// column still NOT NULL during the dual-write window — D7
|
||||
// drops it once `drive_id` is the canonical ownership signal).
|
||||
// 2. Root folder. Post-D7: `user_id` omitted — the column is
|
||||
// nullable and unused on new rows. `created_by` / `updated_by`
|
||||
// bind to `granted_by` (§14 provenance).
|
||||
let folder_id: Uuid = sqlx::query_scalar(
|
||||
r#"
|
||||
INSERT INTO storage.folders
|
||||
(name, parent_id, user_id, drive_id, created_by, updated_by)
|
||||
VALUES ($1, NULL, $2, $3, $2, $2)
|
||||
(name, parent_id, drive_id, created_by, updated_by)
|
||||
VALUES ($1, NULL, $3, $2, $2)
|
||||
RETURNING id
|
||||
"#,
|
||||
)
|
||||
|
||||
@@ -577,7 +577,7 @@ LIMIT $6"
|
||||
size,
|
||||
resource_created_at: row.get("resource_created_at"),
|
||||
modified_at: row.get("modified_at"),
|
||||
owner_id: row.get("owner_id"),
|
||||
owner_id: row.try_get("owner_id").ok(),
|
||||
drive_id: row.get("drive_id"),
|
||||
blob_hash: row.try_get("blob_hash").ok(),
|
||||
is_owner: row.try_get("is_owner").unwrap_or(false),
|
||||
|
||||
@@ -127,30 +127,24 @@ impl FileBlobWriteRepository {
|
||||
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("entity: {e}")))
|
||||
}
|
||||
|
||||
/// Derive `(user_id, drive_id)` from the parent folder. Both are
|
||||
/// needed during the D0 dual-write window: `user_id` for the legacy
|
||||
/// column (dropped in D7) and `drive_id` for the new owning-drive
|
||||
/// reference.
|
||||
async fn resolve_owner_and_drive(
|
||||
&self,
|
||||
folder_id: Option<&str>,
|
||||
) -> Result<(Uuid, Uuid), DomainError> {
|
||||
/// Derive `drive_id` from the parent folder. Post-D7: only the
|
||||
/// drive is needed — the legacy `user_id` column is no longer
|
||||
/// written on new rows.
|
||||
async fn resolve_parent_drive(&self, folder_id: Option<&str>) -> Result<Uuid, DomainError> {
|
||||
match folder_id {
|
||||
Some(fid) => {
|
||||
let row: Option<(Uuid, Uuid)> = sqlx::query_as::<_, (Uuid, Uuid)>(
|
||||
"SELECT user_id, drive_id FROM storage.folders WHERE id = $1::uuid",
|
||||
)
|
||||
.bind(fid)
|
||||
.fetch_optional(self.pool.as_ref())
|
||||
.await
|
||||
.map_err(|e| {
|
||||
DomainError::internal_error("FileBlobWrite", format!("parent lookup: {e}"))
|
||||
})?;
|
||||
row.ok_or_else(|| DomainError::not_found("Folder", fid))
|
||||
}
|
||||
Some(fid) => sqlx::query_scalar::<_, Uuid>(
|
||||
"SELECT drive_id FROM storage.folders WHERE id = $1::uuid",
|
||||
)
|
||||
.bind(fid)
|
||||
.fetch_optional(self.pool.as_ref())
|
||||
.await
|
||||
.map_err(|e| {
|
||||
DomainError::internal_error("FileBlobWrite", format!("parent lookup: {e}"))
|
||||
})?
|
||||
.ok_or_else(|| DomainError::not_found("Folder", fid)),
|
||||
None => Err(DomainError::internal_error(
|
||||
"FileBlobWrite",
|
||||
"folder_id is required to determine file owner",
|
||||
"folder_id is required to determine the target drive",
|
||||
)),
|
||||
}
|
||||
}
|
||||
@@ -290,20 +284,22 @@ impl FileBlobWriteRepository {
|
||||
// attempt's error falls through untouched so the 23505 mapping holds
|
||||
// (a retried INSERT can legitimately lose to a concurrent identical
|
||||
// upload).
|
||||
// Post-D7: `user_id` omitted from the INSERT column list and the
|
||||
// parent CTE. `drive_id` alone is the inherit-from-parent axis;
|
||||
// provenance is `created_by` / `updated_by` (§14).
|
||||
let result = retry_on_deadlock("files.insert", || {
|
||||
sqlx::query_as::<_, (String, Uuid, String, i64, i64, Option<Uuid>, Option<Uuid>)>(
|
||||
sqlx::query_as::<_, (String, 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
|
||||
SELECT id, drive_id, path FROM storage.folders WHERE id = $2::uuid
|
||||
)
|
||||
INSERT INTO storage.files
|
||||
(name, folder_id, user_id, drive_id, blob_hash, size,
|
||||
(name, folder_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,
|
||||
SELECT $1, parent.id, parent.drive_id, $3, $4,
|
||||
$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,
|
||||
@@ -322,71 +318,70 @@ impl FileBlobWriteRepository {
|
||||
})
|
||||
.await;
|
||||
|
||||
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));
|
||||
let (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
|
||||
);
|
||||
}
|
||||
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")
|
||||
{
|
||||
// Idempotent re-upload: if the conflicting file already
|
||||
// holds IDENTICAL content (same folder, same name, same
|
||||
// blob hash), treat this as success and return that file
|
||||
// instead of erroring. Re-uploading a partially-uploaded
|
||||
// folder then becomes a clean no-op for everything that
|
||||
// already landed — only the genuinely missing files
|
||||
// transfer — instead of surfacing hundreds of spurious
|
||||
// "already exists" failures. The duplicate blob reference
|
||||
// taken during ingest was just released above, so the
|
||||
// existing file's own reference is the only one (correct);
|
||||
// a different-content clash still returns the conflict.
|
||||
match self.fetch_identical_file(fid, &name, blob_hash).await {
|
||||
Ok(Some(existing)) => {
|
||||
tracing::info!(
|
||||
"♻️ IDEMPOTENT UPLOAD: {} already present, identical content (hash: {})",
|
||||
name,
|
||||
&blob_hash[..12]
|
||||
);
|
||||
return Ok(existing);
|
||||
}
|
||||
Ok(None) => {} // genuine conflict (different content)
|
||||
Err(lookup_err) => {
|
||||
tracing::warn!(
|
||||
"idempotency lookup failed for {} (hash {}): {} — returning conflict",
|
||||
name,
|
||||
&blob_hash[..12],
|
||||
lookup_err
|
||||
);
|
||||
}
|
||||
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")
|
||||
{
|
||||
// Idempotent re-upload: if the conflicting file already
|
||||
// holds IDENTICAL content (same folder, same name, same
|
||||
// blob hash), treat this as success and return that file
|
||||
// instead of erroring. Re-uploading a partially-uploaded
|
||||
// folder then becomes a clean no-op for everything that
|
||||
// already landed — only the genuinely missing files
|
||||
// transfer — instead of surfacing hundreds of spurious
|
||||
// "already exists" failures. The duplicate blob reference
|
||||
// taken during ingest was just released above, so the
|
||||
// existing file's own reference is the only one (correct);
|
||||
// a different-content clash still returns the conflict.
|
||||
match self.fetch_identical_file(fid, &name, blob_hash).await {
|
||||
Ok(Some(existing)) => {
|
||||
tracing::info!(
|
||||
"♻️ IDEMPOTENT UPLOAD: {} already present, identical content (hash: {})",
|
||||
name,
|
||||
&blob_hash[..12]
|
||||
);
|
||||
return Ok(existing);
|
||||
}
|
||||
Ok(None) => {} // genuine conflict (different content)
|
||||
Err(lookup_err) => {
|
||||
tracing::warn!(
|
||||
"idempotency lookup failed for {} (hash {}): {} — returning conflict",
|
||||
name,
|
||||
&blob_hash[..12],
|
||||
lookup_err
|
||||
);
|
||||
}
|
||||
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::already_exists(
|
||||
"File",
|
||||
format!("'{name}' already exists in this folder"),
|
||||
));
|
||||
}
|
||||
};
|
||||
return Err(DomainError::internal_error(
|
||||
"FileBlobWrite",
|
||||
format!("insert: {e}"),
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
tracing::info!(
|
||||
"📡 STREAMING WRITE: {} ({} bytes, hash: {})",
|
||||
@@ -404,7 +399,7 @@ impl FileBlobWriteRepository {
|
||||
content_type,
|
||||
created_at,
|
||||
updated_at,
|
||||
Some(user_id),
|
||||
None, // Post-D7: `files.user_id` no longer written on new rows.
|
||||
blob_hash.to_string(),
|
||||
created_by,
|
||||
updated_by,
|
||||
@@ -421,11 +416,13 @@ impl FileBlobWriteRepository {
|
||||
name: &str,
|
||||
blob_hash: &str,
|
||||
) -> Result<Option<File>, DomainError> {
|
||||
// Post-D7: `f.user_id` is nullable on new rows; use
|
||||
// `Option<Uuid>` to accept NULL.
|
||||
let row = sqlx::query_as::<
|
||||
_,
|
||||
(
|
||||
String,
|
||||
Uuid,
|
||||
Option<Uuid>,
|
||||
String,
|
||||
i64,
|
||||
i64,
|
||||
@@ -482,7 +479,7 @@ impl FileBlobWriteRepository {
|
||||
mime_type,
|
||||
created_at,
|
||||
updated_at,
|
||||
Some(user_id),
|
||||
user_id,
|
||||
blob_hash.to_string(),
|
||||
created_by,
|
||||
updated_by,
|
||||
@@ -611,29 +608,27 @@ impl FileWritePort for FileBlobWriteRepository {
|
||||
>(
|
||||
r#"
|
||||
WITH src AS (
|
||||
SELECT name, folder_id, user_id, blob_hash, size, mime_type, category_order
|
||||
SELECT name, folder_id, blob_hash, size, mime_type, category_order
|
||||
FROM storage.files
|
||||
WHERE id = $1::uuid AND NOT is_trashed
|
||||
),
|
||||
-- The destination folder may differ from the source's
|
||||
-- folder (when $2 is set); derive drive_id from the
|
||||
-- DESTINATION so cross-drive copies land in the right
|
||||
-- drive. Files in personal drives only copy within the
|
||||
-- same drive today, but the join makes the migration
|
||||
-- future-proof for D2's cross-drive copy story.
|
||||
-- drive. Post-D7: `user_id` no longer projected — the
|
||||
-- column is not written on new rows.
|
||||
dest_folder AS (
|
||||
SELECT id, user_id, drive_id
|
||||
SELECT id, drive_id
|
||||
FROM storage.folders
|
||||
WHERE id = COALESCE($2::uuid,
|
||||
(SELECT folder_id FROM src))
|
||||
),
|
||||
new_file AS (
|
||||
INSERT INTO storage.files
|
||||
(name, folder_id, user_id, drive_id, blob_hash, size,
|
||||
(name, folder_id, drive_id, blob_hash, size,
|
||||
mime_type, category_order, created_by, updated_by)
|
||||
SELECT COALESCE($3::text, src.name),
|
||||
dest_folder.id,
|
||||
dest_folder.user_id,
|
||||
dest_folder.drive_id,
|
||||
src.blob_hash,
|
||||
src.size,
|
||||
@@ -837,23 +832,21 @@ impl FileWritePort for FileBlobWriteRepository {
|
||||
size: u64,
|
||||
caller_id: Uuid,
|
||||
) -> Result<(File, PathBuf), DomainError> {
|
||||
let (user_id, drive_id) = self.resolve_owner_and_drive(folder_id.as_deref()).await?;
|
||||
let drive_id = self.resolve_parent_drive(folder_id.as_deref()).await?;
|
||||
|
||||
// For deferred registration we use a placeholder hash.
|
||||
// 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.
|
||||
// Post-D7: `user_id` omitted from the INSERT column list.
|
||||
// §14: `created_by = $8 = updated_by = caller_id`.
|
||||
let row = retry_on_deadlock("files.insert_deferred", || {
|
||||
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,
|
||||
(name, folder_id, drive_id, blob_hash, size,
|
||||
mime_type, category_order, created_by, updated_by)
|
||||
VALUES ($1, $2::uuid, $3, $4, $5, $6, $7, $8, $9, $9)
|
||||
VALUES ($1, $2::uuid, $3, $4, $5, $6, $7, $8, $8)
|
||||
RETURNING id::text,
|
||||
EXTRACT(EPOCH FROM created_at)::bigint,
|
||||
EXTRACT(EPOCH FROM updated_at)::bigint,
|
||||
@@ -863,7 +856,6 @@ impl FileWritePort for FileBlobWriteRepository {
|
||||
)
|
||||
.bind(&name)
|
||||
.bind(&folder_id)
|
||||
.bind(user_id)
|
||||
.bind(drive_id)
|
||||
.bind(placeholder_hash)
|
||||
.bind(size as i64)
|
||||
@@ -885,7 +877,7 @@ impl FileWritePort for FileBlobWriteRepository {
|
||||
content_type,
|
||||
row.1,
|
||||
row.2,
|
||||
Some(user_id),
|
||||
None, // Post-D7: `files.user_id` no longer written on new rows.
|
||||
String::new(),
|
||||
row.3,
|
||||
row.4,
|
||||
|
||||
@@ -28,12 +28,16 @@ use crate::domain::services::path_service::StoragePath;
|
||||
/// `drive_id` is the post-D0 `NOT NULL` scope axis for path-based
|
||||
/// lookups. `created_by` / `updated_by` are the §14 provenance
|
||||
/// columns, nullable because the FK is `ON DELETE SET NULL`.
|
||||
///
|
||||
/// Post-D7 `user_id` is `Option<Uuid>` — the column is nullable and
|
||||
/// left NULL on new rows (see migration
|
||||
/// `20260902000000_files_folders_user_id_nullable.sql`).
|
||||
type FolderRow = (
|
||||
String,
|
||||
String,
|
||||
String,
|
||||
Option<String>,
|
||||
Uuid,
|
||||
Option<Uuid>,
|
||||
Uuid,
|
||||
i64,
|
||||
i64,
|
||||
@@ -43,13 +47,14 @@ type FolderRow = (
|
||||
);
|
||||
|
||||
/// Type alias for paginated folder rows (includes total_count as
|
||||
/// the last element after the §14 provenance columns).
|
||||
/// the last element after the §14 provenance columns). Same
|
||||
/// nullability semantics as [`FolderRow`].
|
||||
type FolderRowPaginated = (
|
||||
String,
|
||||
String,
|
||||
String,
|
||||
Option<String>,
|
||||
Uuid,
|
||||
Option<Uuid>,
|
||||
Uuid,
|
||||
i64,
|
||||
i64,
|
||||
@@ -59,7 +64,8 @@ type FolderRowPaginated = (
|
||||
i64,
|
||||
);
|
||||
|
||||
/// Type alias for folder rows with optional user_id.
|
||||
/// Type alias for folder rows with optional user_id — kept as a
|
||||
/// separate name for the search-shape SELECTs.
|
||||
/// Includes the §14 provenance columns `created_by` / `updated_by`.
|
||||
type FolderRowOptUser = (
|
||||
String,
|
||||
@@ -201,9 +207,7 @@ 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, r.8, r.9, r.10)
|
||||
})
|
||||
.map(|r| Self::row_to_folder(r.0, r.1, r.2, r.3, r.4, r.5, r.6, r.7, r.8, r.9, r.10))
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
@@ -215,13 +219,19 @@ impl FolderRepository for FolderDbRepository {
|
||||
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
|
||||
// drive beforehand (done during user registration via the
|
||||
// lifecycle hook).
|
||||
let (user_id, drive_id): (Uuid, Uuid) = if let Some(ref pid) = parent_id {
|
||||
sqlx::query_as::<_, (Uuid, Uuid)>(
|
||||
"SELECT user_id, drive_id FROM storage.folders WHERE id = $1::uuid",
|
||||
// Derive `drive_id` from the parent folder. Root-level folders
|
||||
// are reserved for the atomic drive-creation transaction in
|
||||
// `DrivePgRepository::create_personal_drive_atomic` (see
|
||||
// `docs/plan/drive.md` §3) — the no-orphan-root-folder trigger
|
||||
// enforces this at the DB level.
|
||||
//
|
||||
// Post-D7: only `drive_id` is fetched from the parent. The
|
||||
// legacy `user_id` column is no longer written to on new rows
|
||||
// (migration `20260902000000_files_folders_user_id_nullable.sql`);
|
||||
// provenance flows through `created_by` / `updated_by` (§14).
|
||||
let drive_id: Uuid = if let Some(ref pid) = parent_id {
|
||||
sqlx::query_scalar::<_, Uuid>(
|
||||
"SELECT drive_id FROM storage.folders WHERE id = $1::uuid",
|
||||
)
|
||||
.bind(pid)
|
||||
.fetch_optional(self.pool())
|
||||
@@ -238,20 +248,20 @@ impl FolderRepository for FolderDbRepository {
|
||||
));
|
||||
};
|
||||
|
||||
// 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.
|
||||
// Post-D7: no `user_id` in the INSERT column list — the column
|
||||
// is nullable and copied rows / new rows leave it NULL.
|
||||
// `created_by` / `updated_by` carry §14 provenance (both bind to
|
||||
// the caller — pre-D2 that's silently the parent's owner too,
|
||||
// but 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.
|
||||
// RETURNING 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, $5, $5)
|
||||
(name, parent_id, drive_id, created_by, updated_by)
|
||||
VALUES ($1, $2::uuid, $3, $4, $4)
|
||||
RETURNING id::text,
|
||||
path,
|
||||
EXTRACT(EPOCH FROM created_at)::bigint,
|
||||
@@ -263,7 +273,6 @@ impl FolderRepository for FolderDbRepository {
|
||||
)
|
||||
.bind(&name)
|
||||
.bind(&parent_id)
|
||||
.bind(user_id)
|
||||
.bind(drive_id)
|
||||
.bind(caller_id)
|
||||
.fetch_one(self.pool())
|
||||
@@ -281,18 +290,11 @@ impl FolderRepository for FolderDbRepository {
|
||||
})?;
|
||||
|
||||
Self::row_to_folder(
|
||||
row.0,
|
||||
name,
|
||||
row.1,
|
||||
parent_id,
|
||||
Some(user_id),
|
||||
drive_id,
|
||||
row.2,
|
||||
row.3,
|
||||
row.4,
|
||||
row.0, name, row.1, parent_id,
|
||||
None, // Post-D7: `folders.user_id` no longer written on new rows.
|
||||
drive_id, row.2, row.3, row.4,
|
||||
// Fresh from RETURNING — caller_id was bound to both columns.
|
||||
row.5,
|
||||
row.6,
|
||||
row.5, row.6,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -315,17 +317,7 @@ impl FolderRepository for FolderDbRepository {
|
||||
.ok_or_else(|| DomainError::not_found("Folder", id))?;
|
||||
|
||||
Self::row_to_folder(
|
||||
row.0,
|
||||
row.1,
|
||||
row.2,
|
||||
row.3,
|
||||
Some(row.4),
|
||||
row.5,
|
||||
row.6,
|
||||
row.7,
|
||||
row.8,
|
||||
row.9,
|
||||
row.10,
|
||||
row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7, row.8, row.9, row.10,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -368,17 +360,7 @@ impl FolderRepository for FolderDbRepository {
|
||||
.ok_or_else(|| DomainError::not_found("Folder", lookup))?;
|
||||
|
||||
Self::row_to_folder(
|
||||
row.0,
|
||||
row.1,
|
||||
row.2,
|
||||
row.3,
|
||||
Some(row.4),
|
||||
row.5,
|
||||
row.6,
|
||||
row.7,
|
||||
row.8,
|
||||
row.9,
|
||||
row.10,
|
||||
row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7, row.8, row.9, row.10,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -420,7 +402,7 @@ impl FolderRepository for FolderDbRepository {
|
||||
|
||||
rows.into_iter()
|
||||
.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)
|
||||
Self::row_to_folder(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
@@ -468,7 +450,7 @@ impl FolderRepository for FolderDbRepository {
|
||||
|
||||
rows.into_iter()
|
||||
.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)
|
||||
Self::row_to_folder(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
@@ -537,7 +519,7 @@ impl FolderRepository for FolderDbRepository {
|
||||
.into_iter()
|
||||
.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)
|
||||
Self::row_to_folder(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)
|
||||
},
|
||||
)
|
||||
.collect();
|
||||
@@ -590,7 +572,7 @@ impl FolderRepository for FolderDbRepository {
|
||||
.into_iter()
|
||||
.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)
|
||||
Self::row_to_folder(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)
|
||||
},
|
||||
)
|
||||
.collect();
|
||||
@@ -644,17 +626,7 @@ impl FolderRepository for FolderDbRepository {
|
||||
.ok_or_else(|| DomainError::not_found("Folder", id))?;
|
||||
|
||||
Self::row_to_folder(
|
||||
row.0,
|
||||
row.1,
|
||||
row.2,
|
||||
row.3,
|
||||
Some(row.4),
|
||||
row.5,
|
||||
row.6,
|
||||
row.7,
|
||||
row.8,
|
||||
row.9,
|
||||
row.10,
|
||||
row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7, row.8, row.9, row.10,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -708,17 +680,7 @@ impl FolderRepository for FolderDbRepository {
|
||||
.ok_or_else(|| DomainError::not_found("Folder", id))?;
|
||||
|
||||
Self::row_to_folder(
|
||||
row.0,
|
||||
row.1,
|
||||
row.2,
|
||||
row.3,
|
||||
Some(row.4),
|
||||
row.5,
|
||||
row.6,
|
||||
row.7,
|
||||
row.8,
|
||||
row.9,
|
||||
row.10,
|
||||
row.0, row.1, row.2, row.3, row.4, row.5, row.6, row.7, row.8, row.9, row.10,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -1305,7 +1267,7 @@ impl FolderRepository for FolderDbRepository {
|
||||
|
||||
rows.into_iter()
|
||||
.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)
|
||||
Self::row_to_folder(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
@@ -1594,7 +1556,7 @@ impl FolderDbRepository {
|
||||
i64,
|
||||
chrono::DateTime<chrono::Utc>,
|
||||
chrono::DateTime<chrono::Utc>,
|
||||
Uuid,
|
||||
Option<Uuid>, // user_id (post-D7: nullable on new rows)
|
||||
Uuid,
|
||||
Option<String>,
|
||||
String,
|
||||
|
||||
@@ -486,7 +486,7 @@ LIMIT $6"
|
||||
size,
|
||||
resource_created_at: row.get("resource_created_at"),
|
||||
modified_at: row.get("modified_at"),
|
||||
owner_id: row.get("owner_id"),
|
||||
owner_id: row.try_get("owner_id").ok(),
|
||||
drive_id: row.get("drive_id"),
|
||||
blob_hash: row.try_get("blob_hash").ok(),
|
||||
is_owner: row.try_get("is_owner").unwrap_or(false),
|
||||
|
||||
@@ -123,26 +123,64 @@ impl TrashRepository for TrashDbRepository {
|
||||
}
|
||||
|
||||
async fn get_trash_items(&self, user_id: &Uuid) -> Result<Vec<TrashedItem>> {
|
||||
let rows =
|
||||
sqlx::query_as::<_, (Uuid, String, String, Uuid, Option<DateTime<Utc>>, String)>(
|
||||
r#"
|
||||
// Post-D7: the `WHERE t.user_id = $1` filter no longer works —
|
||||
// new rows land with `user_id = NULL`, so the trash view's
|
||||
// `user_id` projection is nullable and can't be the scope
|
||||
// axis any more. Filter by drive-membership instead: any
|
||||
// trashed item in a drive the caller has any role_grant on.
|
||||
// Group memberships expand inline via
|
||||
// `storage.caller_group_ids`. Same predicate shape as
|
||||
// `list_root_folders_for_caller` / the file listings.
|
||||
//
|
||||
// Legacy method — the paginated `list_resources_paged` is
|
||||
// the modern shape and takes explicit drive_ids from the
|
||||
// service layer.
|
||||
let rows = sqlx::query_as::<
|
||||
_,
|
||||
(
|
||||
Uuid,
|
||||
String,
|
||||
String,
|
||||
Option<Uuid>,
|
||||
Option<DateTime<Utc>>,
|
||||
String,
|
||||
),
|
||||
>(
|
||||
r#"
|
||||
SELECT t.id, t.name, t.item_type, t.user_id, t.trashed_at,
|
||||
COALESCE(p.path || '/' || t.name, t.name) AS original_path
|
||||
FROM storage.trash_items t
|
||||
LEFT JOIN storage.folders p ON p.id = t.original_parent_id
|
||||
WHERE t.user_id = $1
|
||||
WHERE EXISTS (
|
||||
SELECT 1 FROM storage.role_grants g
|
||||
WHERE g.resource_type = 'drive'
|
||||
AND g.resource_id = t.drive_id
|
||||
AND (g.expires_at IS NULL OR g.expires_at > NOW())
|
||||
AND (
|
||||
(g.subject_type = 'user' AND g.subject_id = $1)
|
||||
OR (g.subject_type = 'group' AND g.subject_id IN
|
||||
(SELECT storage.caller_group_ids($1)))
|
||||
)
|
||||
)
|
||||
ORDER BY t.trashed_at DESC
|
||||
"#,
|
||||
)
|
||||
.bind(user_id)
|
||||
.fetch_all(self.pool.as_ref())
|
||||
.await
|
||||
.map_err(|e| DomainError::internal_error("TrashDb", format!("list: {e}")))?;
|
||||
)
|
||||
.bind(user_id)
|
||||
.fetch_all(self.pool.as_ref())
|
||||
.await
|
||||
.map_err(|e| DomainError::internal_error("TrashDb", format!("list: {e}")))?;
|
||||
|
||||
Ok(rows
|
||||
.into_iter()
|
||||
.map(|(id, name, item_type, uid, trashed_at, path)| {
|
||||
self.row_to_trashed_item(id, name, item_type, uid, trashed_at, path)
|
||||
self.row_to_trashed_item(
|
||||
id,
|
||||
name,
|
||||
item_type,
|
||||
uid.unwrap_or(*user_id),
|
||||
trashed_at,
|
||||
path,
|
||||
)
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
@@ -153,7 +191,23 @@ impl TrashRepository for TrashDbRepository {
|
||||
// …)` in the service callers (`restore_item`, `delete_permanently`).
|
||||
// The drive precheck in `pg_acl_engine` then resolves Owner-on-drive
|
||||
// → Delete-permission for items in shared drives.
|
||||
let row = sqlx::query_as::<_, (Uuid, String, String, Uuid, Option<DateTime<Utc>>, String)>(
|
||||
//
|
||||
// Post-D7: `t.user_id` is nullable (new rows land NULL). The
|
||||
// entity's `user_id` field is still non-optional; fall back to
|
||||
// `Uuid::nil()` when the view row is NULL. AuthZ decisions
|
||||
// don't consult this field — they've already resolved the
|
||||
// caller's role on the target's drive.
|
||||
let row = sqlx::query_as::<
|
||||
_,
|
||||
(
|
||||
Uuid,
|
||||
String,
|
||||
String,
|
||||
Option<Uuid>,
|
||||
Option<DateTime<Utc>>,
|
||||
String,
|
||||
),
|
||||
>(
|
||||
r#"
|
||||
SELECT t.id, t.name, t.item_type, t.user_id, t.trashed_at,
|
||||
COALESCE(p.path || '/' || t.name, t.name) AS original_path
|
||||
@@ -168,7 +222,14 @@ impl TrashRepository for TrashDbRepository {
|
||||
.map_err(|e| DomainError::internal_error("TrashDb", format!("get: {e}")))?;
|
||||
|
||||
Ok(row.map(|(id, name, item_type, uid, trashed_at, path)| {
|
||||
self.row_to_trashed_item(id, name, item_type, uid, trashed_at, path)
|
||||
self.row_to_trashed_item(
|
||||
id,
|
||||
name,
|
||||
item_type,
|
||||
uid.unwrap_or_else(Uuid::nil),
|
||||
trashed_at,
|
||||
path,
|
||||
)
|
||||
}))
|
||||
}
|
||||
|
||||
@@ -529,7 +590,7 @@ LIMIT $6"
|
||||
size,
|
||||
resource_created_at: row.get("resource_created_at"),
|
||||
modified_at: row.get("modified_at"),
|
||||
owner_id: row.get("owner_id"),
|
||||
owner_id: row.try_get("owner_id").ok(),
|
||||
drive_id: row.get("drive_id"),
|
||||
blob_hash: row.try_get("blob_hash").ok(),
|
||||
trashed_at,
|
||||
|
||||
@@ -1127,11 +1127,36 @@ impl DedupService {
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
/// Returns `true` if `user_id` owns at least one (even trashed) file that
|
||||
/// references the blob identified by `hash`.
|
||||
/// Returns `true` if the caller has a **writable role** on at least one
|
||||
/// drive containing a (possibly trashed) file that references the blob
|
||||
/// identified by `hash`.
|
||||
///
|
||||
/// Post-D7 (`project_d7_policy_calls` LOCKED): same
|
||||
/// drive-membership + writable-role predicate as
|
||||
/// [`claimable_chunks`] / [`pin_claimable_chunks`] — MUST stay in
|
||||
/// lockstep with them. Group memberships (direct + transitive)
|
||||
/// expand inline via `storage.caller_group_ids($2)`. Viewers /
|
||||
/// commenters are excluded — they can't legitimately upload into
|
||||
/// a drive, so they can't claim "already-uploaded" via dedup.
|
||||
pub async fn user_owns_blob_reference(&self, hash: &str, user_id: &str) -> bool {
|
||||
sqlx::query_scalar::<_, bool>(
|
||||
"SELECT EXISTS(SELECT 1 FROM storage.files WHERE blob_hash = $1 AND user_id = $2::uuid)",
|
||||
"SELECT EXISTS(
|
||||
SELECT 1
|
||||
FROM storage.files f
|
||||
WHERE f.blob_hash = $1
|
||||
AND EXISTS (
|
||||
SELECT 1 FROM storage.role_grants g
|
||||
WHERE g.resource_type = 'drive'
|
||||
AND g.resource_id = f.drive_id
|
||||
AND g.role IN ('owner', 'editor', 'contributor')
|
||||
AND (g.expires_at IS NULL OR g.expires_at > NOW())
|
||||
AND (
|
||||
(g.subject_type = 'user' AND g.subject_id = $2::uuid)
|
||||
OR (g.subject_type = 'group' AND g.subject_id IN
|
||||
(SELECT storage.caller_group_ids($2::uuid)))
|
||||
)
|
||||
)
|
||||
)",
|
||||
)
|
||||
.bind(hash)
|
||||
.bind(user_id)
|
||||
@@ -1141,13 +1166,14 @@ impl DedupService {
|
||||
}
|
||||
|
||||
/// Batch variant of [`Self::user_owns_blob_reference`]: given candidate
|
||||
/// hashes, return the subset the user already references — in ONE query
|
||||
/// (backed by `idx_files_blob_hash`). Lets a client hash a whole upload set
|
||||
/// and learn which files it can skip with a single round trip instead of
|
||||
/// one probe per file.
|
||||
/// hashes, return the subset the caller can already reference — in ONE
|
||||
/// query (backed by `idx_files_blob_hash`). Lets a client hash a whole
|
||||
/// upload set and learn which files it can skip with a single round trip
|
||||
/// instead of one probe per file.
|
||||
///
|
||||
/// User-scoped, exactly like the single check: only the caller's own blobs
|
||||
/// are returned, so it cannot probe whether *other* users hold a blob.
|
||||
/// Post-D7: same drive-membership + writable-role predicate as the
|
||||
/// single check. Anti-enumeration is preserved — only hashes present
|
||||
/// in a drive the caller can write to come back.
|
||||
pub async fn user_owned_blob_references(
|
||||
&self,
|
||||
hashes: &[String],
|
||||
@@ -1157,8 +1183,21 @@ impl DedupService {
|
||||
return Vec::new();
|
||||
}
|
||||
sqlx::query_scalar::<_, String>(
|
||||
"SELECT DISTINCT blob_hash FROM storage.files \
|
||||
WHERE blob_hash = ANY($1) AND user_id = $2::uuid",
|
||||
"SELECT DISTINCT f.blob_hash
|
||||
FROM storage.files f
|
||||
WHERE f.blob_hash = ANY($1)
|
||||
AND EXISTS (
|
||||
SELECT 1 FROM storage.role_grants g
|
||||
WHERE g.resource_type = 'drive'
|
||||
AND g.resource_id = f.drive_id
|
||||
AND g.role IN ('owner', 'editor', 'contributor')
|
||||
AND (g.expires_at IS NULL OR g.expires_at > NOW())
|
||||
AND (
|
||||
(g.subject_type = 'user' AND g.subject_id = $2::uuid)
|
||||
OR (g.subject_type = 'group' AND g.subject_id IN
|
||||
(SELECT storage.caller_group_ids($2::uuid)))
|
||||
)
|
||||
)",
|
||||
)
|
||||
.bind(hashes)
|
||||
.bind(user_id)
|
||||
@@ -3008,19 +3047,20 @@ mod rechunk_integration_tests {
|
||||
.await
|
||||
.expect("insert legacy blob row");
|
||||
|
||||
let (user_id, drive_id) = seed_user(pool).await;
|
||||
let (_user_id, drive_id) = seed_user(pool).await;
|
||||
let mut file_ids = Vec::new();
|
||||
for i in 0..n_files {
|
||||
let name = format!(
|
||||
"rust-test-rechunk-{label}-{}-{i}",
|
||||
&Uuid::new_v4().to_string()[..8]
|
||||
);
|
||||
// Post-D7: `user_id` omitted — column is nullable and unused
|
||||
// on new rows.
|
||||
let id: Uuid = sqlx::query_scalar(
|
||||
"INSERT INTO storage.files (name, user_id, drive_id, blob_hash, size)
|
||||
VALUES ($1, $2, $3, $4, $5) RETURNING id",
|
||||
"INSERT INTO storage.files (name, drive_id, blob_hash, size)
|
||||
VALUES ($1, $2, $3, $4) RETURNING id",
|
||||
)
|
||||
.bind(&name)
|
||||
.bind(user_id)
|
||||
.bind(drive_id)
|
||||
.bind(&hash)
|
||||
.bind(data.len() as i64)
|
||||
@@ -3320,22 +3360,23 @@ mod delta_upload_integration_tests {
|
||||
async fn seed_owned_content(
|
||||
svc: &DedupService,
|
||||
pool: &PgPool,
|
||||
user_id: Uuid,
|
||||
_user_id: Uuid,
|
||||
drive_id: Uuid,
|
||||
data: &[u8],
|
||||
label: &str,
|
||||
) -> (String, Vec<String>, Uuid) {
|
||||
let file_hash = blake3::hash(data).to_hex().to_string();
|
||||
|
||||
// Post-D7: `user_id` omitted — column is nullable and unused on
|
||||
// new rows.
|
||||
let file_id: Uuid = sqlx::query_scalar(
|
||||
"INSERT INTO storage.files (name, user_id, drive_id, blob_hash, size)
|
||||
VALUES ($1, $2, $3, $4, $5) RETURNING id",
|
||||
"INSERT INTO storage.files (name, drive_id, blob_hash, size)
|
||||
VALUES ($1, $2, $3, $4) RETURNING id",
|
||||
)
|
||||
.bind(format!(
|
||||
"rust-test-delta-{label}-{}",
|
||||
&Uuid::new_v4().to_string()[..8]
|
||||
))
|
||||
.bind(user_id)
|
||||
.bind(drive_id)
|
||||
.bind(&file_hash)
|
||||
.bind(data.len() as i64)
|
||||
|
||||
@@ -242,20 +242,30 @@ impl ContentIndexWorker {
|
||||
|
||||
// Authoritative state re-read: a queued 'upsert' whose row vanished
|
||||
// or got trashed in the meantime becomes a delete.
|
||||
let files: Vec<(Uuid, String, String, String, String, String, i64)> =
|
||||
if upsert_candidates.is_empty() {
|
||||
Vec::new()
|
||||
} else {
|
||||
sqlx::query_as(
|
||||
"SELECT fi.id, fi.user_id::text, fi.drive_id::text, fi.name,
|
||||
fi.blob_hash, fi.mime_type, fi.size
|
||||
FROM storage.files fi
|
||||
WHERE fi.id = ANY($1) AND NOT fi.is_trashed",
|
||||
)
|
||||
.bind(&upsert_candidates)
|
||||
.fetch_all(self.maintenance_pool.as_ref())
|
||||
.await?
|
||||
};
|
||||
//
|
||||
// Post-D7: `fi.user_id` is nullable and new rows land NULL.
|
||||
// The projected `user_id::text` therefore comes back as
|
||||
// `Option<String>`; we normalise to `""` at the tuple boundary
|
||||
// so the downstream indexing code doesn't have to change.
|
||||
// The Tantivy `user_id` field is defence-in-depth only — every
|
||||
// query is Must-scoped by `drive_id`.
|
||||
// (file_id, user_id, drive_id, name, blob_hash, mime, size).
|
||||
// `user_id` is `Option<String>` because post-D7 `storage.files.user_id`
|
||||
// is nullable — new rows land NULL.
|
||||
type FileIndexRow = (Uuid, Option<String>, String, String, String, String, i64);
|
||||
let files: Vec<FileIndexRow> = if upsert_candidates.is_empty() {
|
||||
Vec::new()
|
||||
} else {
|
||||
sqlx::query_as(
|
||||
"SELECT fi.id, fi.user_id::text, fi.drive_id::text, fi.name,
|
||||
fi.blob_hash, fi.mime_type, fi.size
|
||||
FROM storage.files fi
|
||||
WHERE fi.id = ANY($1) AND NOT fi.is_trashed",
|
||||
)
|
||||
.bind(&upsert_candidates)
|
||||
.fetch_all(self.maintenance_pool.as_ref())
|
||||
.await?
|
||||
};
|
||||
let found: HashSet<Uuid> = files.iter().map(|f| f.0).collect();
|
||||
deletes.extend(upsert_candidates.iter().filter(|id| !found.contains(id)));
|
||||
|
||||
@@ -301,7 +311,7 @@ impl ContentIndexWorker {
|
||||
.map(|t| truncate_on_char(t, PREVIEW_BYTES));
|
||||
records.push(IndexDocRecord {
|
||||
file_id: file_id.to_string(),
|
||||
user_id,
|
||||
user_id: user_id.unwrap_or_default(),
|
||||
drive_id,
|
||||
name,
|
||||
content,
|
||||
|
||||
@@ -215,7 +215,7 @@ pub async fn list_favorites_resources(
|
||||
name: row.name.clone(),
|
||||
path,
|
||||
parent_id: row.parent_id.map(|u| u.to_string()),
|
||||
owner_id: Some(row.owner_id.to_string()),
|
||||
owner_id: row.owner_id.map(|u| u.to_string()),
|
||||
drive_id: row.drive_id,
|
||||
created_at: row.resource_created_at.timestamp() as u64,
|
||||
modified_at: row.modified_at.timestamp() as u64,
|
||||
@@ -265,7 +265,7 @@ pub async fn list_favorites_resources(
|
||||
)),
|
||||
category: std::sync::Arc::from(category_for(&row.name, mime)),
|
||||
size_formatted: format_file_size(size_bytes),
|
||||
owner_id: Some(row.owner_id.to_string()),
|
||||
owner_id: row.owner_id.map(|u| u.to_string()),
|
||||
sort_date: None,
|
||||
content_hash,
|
||||
etag,
|
||||
|
||||
@@ -504,7 +504,7 @@ pub async fn list_folder_resources(
|
||||
name: row.name.clone(),
|
||||
path: String::new(), // cleared — share recipients must not see hierarchy
|
||||
parent_id: row.parent_id.map(|u| u.to_string()),
|
||||
owner_id: Some(row.owner_id.to_string()),
|
||||
owner_id: row.owner_id.map(|u| u.to_string()),
|
||||
drive_id: row.drive_id,
|
||||
created_at: row.created_at.timestamp() as u64,
|
||||
modified_at: row.modified_at.timestamp() as u64,
|
||||
@@ -553,7 +553,7 @@ pub async fn list_folder_resources(
|
||||
icon_special_class: Arc::from(icon_special_class_for(&row.name, mime)),
|
||||
category: Arc::from(category_for(&row.name, mime)),
|
||||
size_formatted: format_file_size(size_bytes),
|
||||
owner_id: Some(row.owner_id.to_string()),
|
||||
owner_id: row.owner_id.map(|u| u.to_string()),
|
||||
sort_date: None,
|
||||
content_hash,
|
||||
etag,
|
||||
|
||||
@@ -246,7 +246,7 @@ pub async fn list_recent_resources(
|
||||
name: row.name.clone(),
|
||||
path,
|
||||
parent_id: row.parent_id.map(|u| u.to_string()),
|
||||
owner_id: Some(row.owner_id.to_string()),
|
||||
owner_id: row.owner_id.map(|u| u.to_string()),
|
||||
drive_id: row.drive_id,
|
||||
created_at: row.resource_created_at.timestamp() as u64,
|
||||
modified_at: row.modified_at.timestamp() as u64,
|
||||
@@ -294,7 +294,7 @@ pub async fn list_recent_resources(
|
||||
)),
|
||||
category: std::sync::Arc::from(category_for(&row.name, mime)),
|
||||
size_formatted: format_file_size(size_bytes),
|
||||
owner_id: Some(row.owner_id.to_string()),
|
||||
owner_id: row.owner_id.map(|u| u.to_string()),
|
||||
sort_date: None,
|
||||
content_hash,
|
||||
etag,
|
||||
|
||||
Reference in New Issue
Block a user