diff --git a/backend/migrations/202511120000_collapse_asset_objects/down.sql b/backend/migrations/202511120000_collapse_asset_objects/down.sql new file mode 100644 index 0000000..6c3a430 --- /dev/null +++ b/backend/migrations/202511120000_collapse_asset_objects/down.sql @@ -0,0 +1,24 @@ +CREATE TABLE tenant.document_asset_objects ( + id UUID PRIMARY KEY, + asset_id UUID NOT NULL REFERENCES tenant.document_assets(id) ON DELETE CASCADE, + ordinal INT NOT NULL, + s3_key TEXT NOT NULL, + metadata JSONB NOT NULL DEFAULT '{}'::jsonb, + tenant_id UUID NOT NULL REFERENCES shared.tenants(id), + CONSTRAINT document_asset_objects_ordinal_positive CHECK (ordinal >= 1), + CONSTRAINT document_asset_objects_asset_ordinal_unique UNIQUE (asset_id, ordinal) +); + +CREATE INDEX idx_document_asset_objects_asset_ordinal + ON tenant.document_asset_objects(asset_id, ordinal); + +CREATE INDEX document_asset_objects_tenant_id_idx ON tenant.document_asset_objects(tenant_id); + +ALTER TABLE tenant.document_assets ADD COLUMN cardinality INT; +UPDATE tenant.document_assets SET cardinality = 1; + +INSERT INTO tenant.document_asset_objects (id, asset_id, ordinal, s3_key, metadata, tenant_id) +SELECT gen_random_uuid(), id, 1, s3_key, metadata, tenant_id +FROM tenant.document_assets; + +ALTER TABLE tenant.document_assets DROP COLUMN s3_key; diff --git a/backend/migrations/202511120000_collapse_asset_objects/up.sql b/backend/migrations/202511120000_collapse_asset_objects/up.sql new file mode 100644 index 0000000..cf26ae4 --- /dev/null +++ b/backend/migrations/202511120000_collapse_asset_objects/up.sql @@ -0,0 +1,27 @@ +-- Prevent concurrent inserts/updates during backfill. +LOCK TABLE tenant.document_asset_objects IN ACCESS EXCLUSIVE MODE; +LOCK TABLE tenant.document_assets IN ACCESS EXCLUSIVE MODE; + +ALTER TABLE tenant.document_assets ADD COLUMN s3_key TEXT; + +UPDATE tenant.document_assets AS da +SET s3_key = o.s3_key, + metadata = COALESCE(da.metadata, '{}'::jsonb) || COALESCE(o.metadata, '{}'::jsonb) +FROM tenant.document_asset_objects AS o +WHERE o.asset_id = da.id + AND o.ordinal = 1; + +DELETE FROM tenant.document_asset_objects WHERE ordinal <> 1; + +DO $$ +BEGIN + IF EXISTS (SELECT 1 FROM tenant.document_assets WHERE s3_key IS NULL) THEN + RAISE EXCEPTION 'cannot drop document_asset_objects: some assets are missing a populated ordinal 1 object (s3_key null)'; + END IF; +END +$$; + +ALTER TABLE tenant.document_assets ALTER COLUMN s3_key SET NOT NULL; +ALTER TABLE tenant.document_assets DROP COLUMN cardinality; + +DROP TABLE tenant.document_asset_objects; diff --git a/backend/src/auth/jwt.rs b/backend/src/auth/jwt.rs index 714ec10..6c4f496 100644 --- a/backend/src/auth/jwt.rs +++ b/backend/src/auth/jwt.rs @@ -115,17 +115,13 @@ impl JwtService { pub fn generate_asset_download_token( &self, asset_id: Uuid, - object_id: Uuid, user_id: Uuid, tenant_id: Uuid, ) -> Result { let now = Utc::now(); let exp = now + self.download_expiry; let claims = DownloadClaims { - subject: DownloadSubject::AssetObject { - asset_id, - object_id, - }, + subject: DownloadSubject::Asset { asset_id }, user_id, tenant_id, iss: self.issuer.clone(), @@ -216,7 +212,7 @@ pub struct Claims { #[serde(tag = "scope", rename_all = "snake_case")] pub enum DownloadSubject { Document { doc_id: Uuid, version_id: Uuid }, - AssetObject { asset_id: Uuid, object_id: Uuid }, + Asset { asset_id: Uuid }, } #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/backend/src/bin/admin.rs b/backend/src/bin/admin.rs index f513a17..a8d174c 100644 --- a/backend/src/bin/admin.rs +++ b/backend/src/bin/admin.rs @@ -19,14 +19,11 @@ use papercrate::{ jobs::{enqueue_job, JOB_ANALYZE_DOCUMENT, JOB_DELETE_TENANT}, migrations::MIGRATIONS, models::{ - DocumentAsset, DocumentAssetObject, MagicToken, MagicTokenKind, NewUser, NewUserMembership, - Tenant, TenantStatus, User, + DocumentAsset, MagicToken, MagicTokenKind, NewUser, NewUserMembership, Tenant, + TenantStatus, User, }, s3, - schema::{ - document_asset_objects, document_assets, documents, magic_tokens, tenants, - user_memberships, users, - }, + schema::{document_assets, documents, magic_tokens, tenants, user_memberships, users}, storage::{ObjectStorage, S3Storage, TenantStorage}, tenants::{apply_tenant_guc, clear_tenant_context, TenantService}, utils::{text::normalize_identifier, tracing::init_tracing}, @@ -788,29 +785,15 @@ async fn delete_assets_for_tenant( let asset_ids: Vec = assets.iter().map(|asset| asset.id).collect(); - let objects: Vec = document_asset_objects::table - .filter(document_asset_objects::tenant_id.eq(tenant.id)) - .filter(document_asset_objects::asset_id.eq_any(&asset_ids)) - .load(&mut conn) - .with_context(|| format!("failed to load asset objects for tenant {}", tenant.name))?; - - for object in &objects { - if let Err(err) = tenant_storage.delete_object(&object.s3_key).await { + for asset in &assets { + if let Err(err) = tenant_storage.delete_object(&asset.s3_key).await { eprintln!( "Failed to delete object {} (tenant {}): {err}", - object.s3_key, tenant.name + asset.s3_key, tenant.name ); } } - diesel::delete( - document_asset_objects::table - .filter(document_asset_objects::tenant_id.eq(tenant.id)) - .filter(document_asset_objects::asset_id.eq_any(&asset_ids)), - ) - .execute(&mut conn) - .with_context(|| format!("failed to remove asset objects for tenant {}", tenant.name))?; - diesel::delete( document_assets::table .filter(document_assets::tenant_id.eq(tenant.id)) diff --git a/backend/src/documents/asset.rs b/backend/src/documents/asset.rs index ab4087f..1c4c6fe 100644 --- a/backend/src/documents/asset.rs +++ b/backend/src/documents/asset.rs @@ -8,8 +8,8 @@ use utoipa::ToSchema; use uuid::Uuid; use crate::error::{AppError, AppResult}; -use crate::models::{Document, DocumentAsset, DocumentAssetObject, DocumentVersion}; -use crate::schema::{document_asset_objects, document_assets, document_versions}; +use crate::models::{Document, DocumentAsset, DocumentVersion}; +use crate::schema::{document_assets, document_versions}; use crate::state::{AppState, PgPooledConnection}; use crate::utils::{http::inline_content_disposition, time::to_iso}; @@ -22,17 +22,6 @@ pub struct DocumentAssetResponse { pub metadata: Value, #[serde(skip_serializing_if = "Option::is_none")] #[schema(nullable)] - pub cardinality: Option, -} - -#[derive(Serialize, Clone, ToSchema)] -pub struct DocumentAssetObjectResponse { - pub id: Uuid, - pub ordinal: i32, - #[schema(value_type = Object)] - pub metadata: Value, - #[serde(skip_serializing_if = "Option::is_none")] - #[schema(nullable)] pub url: Option, #[serde(skip_serializing_if = "Option::is_none")] #[schema(nullable)] @@ -49,9 +38,10 @@ pub struct DocumentAssetDetailResponse { pub created_at: String, #[serde(skip_serializing_if = "Option::is_none")] #[schema(nullable)] - pub cardinality: Option, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub objects: Vec, + pub url: Option, + #[serde(skip_serializing_if = "Option::is_none")] + #[schema(nullable)] + pub expires_at: Option, } #[derive(Serialize, Clone, ToSchema)] @@ -107,13 +97,15 @@ pub fn to_asset_summary(asset: DocumentAsset) -> DocumentAssetResponse { asset_type: asset.asset_type, mime_type: asset.mime_type, metadata: asset.metadata, - cardinality: asset.cardinality, + url: None, + expires_at: None, } } pub fn to_asset_detail_response( asset: DocumentAsset, - objects: Vec, + url: Option, + expires_at: Option, ) -> DocumentAssetDetailResponse { DocumentAssetDetailResponse { id: asset.id, @@ -121,30 +113,13 @@ pub fn to_asset_detail_response( mime_type: asset.mime_type, metadata: asset.metadata, created_at: to_iso(asset.created_at), - cardinality: asset.cardinality, - objects, - } -} - -pub fn to_asset_object_response( - object: DocumentAssetObject, - url: Option, - expires_at: Option, -) -> DocumentAssetObjectResponse { - DocumentAssetObjectResponse { - id: object.id, - ordinal: object.ordinal, - metadata: object.metadata, url, expires_at, } } -pub fn asset_object_disposition( - asset: &DocumentAsset, - object: &DocumentAssetObject, -) -> Option { - let filename = format!("{}-{}", asset.asset_type, object.ordinal); +pub fn asset_disposition(asset: &DocumentAsset) -> Option { + let filename = asset.asset_type.clone(); inline_content_disposition(&filename) } @@ -168,25 +143,13 @@ pub fn load_asset_responses_with_conn( tenant_id: Uuid, version_id: Uuid, ) -> AppResult> { - let assets: Vec<(DocumentAsset, Option)> = document_assets::table - .left_outer_join( - document_asset_objects::table.on(document_asset_objects::asset_id - .eq(document_assets::id) - .and(document_asset_objects::ordinal.eq(1))), - ) + let assets: Vec = document_assets::table .filter(document_assets::document_version_id.eq(version_id)) .filter(document_assets::tenant_id.eq(tenant_id)) .order(document_assets::created_at.asc()) - .select(( - document_assets::all_columns, - document_asset_objects::all_columns.nullable(), - )) .load(conn)?; - Ok(assets - .into_iter() - .map(|(asset, _)| to_asset_summary(asset)) - .collect()) + Ok(assets.into_iter().map(to_asset_summary).collect()) } pub fn load_primary_assets( @@ -216,25 +179,16 @@ pub fn load_primary_assets( version_map.insert(version.id, version); } - let assets: Vec<(DocumentAsset, Option)> = document_assets::table - .left_outer_join( - document_asset_objects::table.on(document_asset_objects::asset_id - .eq(document_assets::id) - .and(document_asset_objects::ordinal.eq(1))), - ) + let assets: Vec = document_assets::table .filter(document_assets::document_version_id.eq_any(&version_ids)) .order(( document_assets::document_version_id.asc(), document_assets::created_at.asc(), )) - .select(( - document_assets::all_columns, - document_asset_objects::all_columns.nullable(), - )) .load(conn)?; let mut assets_by_version: HashMap> = HashMap::new(); - for (asset, _object) in assets { + for asset in assets { let version_id = asset.document_version_id; let response = to_asset_summary(asset); assets_by_version diff --git a/backend/src/models.rs b/backend/src/models.rs index ece7530..e2483b7 100644 --- a/backend/src/models.rs +++ b/backend/src/models.rs @@ -621,7 +621,7 @@ pub struct DocumentAsset { pub mime_type: String, pub metadata: serde_json::Value, pub created_at: NaiveDateTime, - pub cardinality: Option, + pub s3_key: String, pub tenant_id: Uuid, } @@ -633,30 +633,7 @@ pub struct NewDocumentAsset { pub asset_type: String, pub mime_type: String, pub metadata: serde_json::Value, - pub cardinality: Option, - pub tenant_id: Uuid, -} - -#[derive(Debug, Clone, Queryable, Identifiable, Associations)] -#[diesel(table_name = document_asset_objects)] -#[diesel(belongs_to(DocumentAsset, foreign_key = asset_id))] -pub struct DocumentAssetObject { - pub id: Uuid, - pub asset_id: Uuid, - pub ordinal: i32, pub s3_key: String, - pub metadata: serde_json::Value, - pub tenant_id: Uuid, -} - -#[derive(Debug, Insertable)] -#[diesel(table_name = document_asset_objects)] -pub struct NewDocumentAssetObject { - pub id: Uuid, - pub asset_id: Uuid, - pub ordinal: i32, - pub s3_key: String, - pub metadata: serde_json::Value, pub tenant_id: Uuid, } diff --git a/backend/src/openapi.rs b/backend/src/openapi.rs index 7e947af..348b640 100644 --- a/backend/src/openapi.rs +++ b/backend/src/openapi.rs @@ -74,8 +74,8 @@ pub mod schemas { }; pub use crate::auth::AuthenticatedUser; pub use crate::documents::asset::{ - DocumentAssetDetailResponse, DocumentAssetObjectResponse, DocumentAssetResponse, - DocumentVersionDetailResponse, DocumentVersionResponse, + DocumentAssetDetailResponse, DocumentAssetResponse, DocumentVersionDetailResponse, + DocumentVersionResponse, }; pub use crate::documents::correspondents::DocumentCorrespondentResponse; pub use crate::error::ApiErrorResponse; @@ -85,8 +85,8 @@ pub mod schemas { UpdateCorrespondentRequest, }; pub use crate::routes::documents::{ - AssetObjectsQuery, AssetRequestQuery, DocumentCheckQuery, MoveDocumentRequest, - RestoreDocumentRequest, UploadDocumentForm, + AssetRequestQuery, DocumentCheckQuery, MoveDocumentRequest, RestoreDocumentRequest, + UploadDocumentForm, }; pub use crate::routes::folders::FolderContentsResponse; pub use crate::routes::tags::{CreateTagRequest, TagCatalogEntry, UpdateTagRequest}; diff --git a/backend/src/routes/documents.rs b/backend/src/routes/documents.rs index 20900cd..6dd1ee7 100644 --- a/backend/src/routes/documents.rs +++ b/backend/src/routes/documents.rs @@ -16,17 +16,16 @@ use uuid::Uuid; use crate::auth::{ensure_active_tenant_with_conn, jwt::DownloadSubject, TenantScopedConn}; use crate::documents::asset::{ - asset_object_disposition, DocumentAssetDetailResponse, DocumentAssetResponse, + asset_disposition, DocumentAssetDetailResponse, DocumentAssetResponse, DocumentVersionDetailResponse, DocumentVersionResponse, }; #[allow(unused_imports)] use crate::error::ApiErrorResponse; use crate::error::{AppError, AppResult}; use crate::http::responders::{accepted_json, created_json, no_content, ok_json, JsonResponse}; -use crate::models::{Document, DocumentAsset, DocumentAssetObject, DocumentVersion}; +use crate::models::{Document, DocumentAsset, DocumentVersion}; use crate::schema::{ - document_asset_objects, document_assets, document_versions, documents, - user_sessions::dsl as session_dsl, + document_assets, document_versions, documents, user_sessions::dsl as session_dsl, }; use crate::services::correspondents::{ AssignCorrespondentsRequest, BulkCorrespondentAction, BulkCorrespondentResponse, @@ -91,15 +90,6 @@ pub struct RestoreDocumentRequest { pub folder_id: Option, } -#[derive(Deserialize, Default, IntoParams, ToSchema)] -#[into_params(parameter_in = Query)] -pub struct AssetObjectsQuery { - #[serde(default)] - pub start: Option, - #[serde(default)] - pub limit: Option, -} - #[utoipa::path( get, path = "/api/documents", @@ -467,12 +457,13 @@ pub async fn list_document_assets( TenantScopedConn { mut conn, tenant_id, + user_id, .. }: TenantScopedConn, ) -> AppResult>> { let service = DocumentsService::new(&state); let assets = service - .list_document_assets(&mut conn, tenant_id, document_id) + .list_document_assets(&mut conn, tenant_id, user_id, document_id) .await?; ok_json(assets) } @@ -480,14 +471,13 @@ pub async fn list_document_assets( #[utoipa::path( get, path = "/api/assets/{asset_id}", - params(("asset_id" = Uuid, Path, description = "Asset ID"), AssetObjectsQuery), + params(("asset_id" = Uuid, Path, description = "Asset ID")), responses((status = 200, description = "Asset detail", body = DocumentAssetDetailResponse)), tag = "Assets" )] pub async fn get_document_asset( State(state): State, Path(asset_id): Path, - Query(query): Query, TenantScopedConn { conn, tenant_id, @@ -495,17 +485,9 @@ pub async fn get_document_asset( .. }: TenantScopedConn, ) -> AppResult> { - let start = query.start.unwrap_or(1); - let limit = query.limit.unwrap_or(1); - if start < 1 { - return Err(AppError::bad_request("start must be at least 1")); - } - if limit < 1 { - return Err(AppError::bad_request("limit must be at least 1")); - } let service = DocumentsService::new(&state); let detail = service - .get_document_asset(conn, tenant_id, user_id, asset_id, start, limit) + .get_document_asset(conn, tenant_id, user_id, asset_id) .await?; ok_json(detail) } @@ -637,20 +619,7 @@ pub async fn download_with_token( ) .await } - DownloadSubject::AssetObject { - asset_id, - object_id, - } => { - if !state.config.proxy_downloads { - return Err(AppError::not_found()); - } - - let object: DocumentAssetObject = document_asset_objects::table - .find(*object_id) - .filter(document_asset_objects::asset_id.eq(*asset_id)) - .filter(document_asset_objects::tenant_id.eq(claims.tenant_id)) - .first(&mut conn)?; - + DownloadSubject::Asset { asset_id } => { let asset: DocumentAsset = document_assets::table .find(*asset_id) .filter(document_assets::tenant_id.eq(claims.tenant_id)) @@ -659,14 +628,28 @@ pub async fn download_with_token( drop(conn); let storage = state.storage_for_tenant(claims.tenant_id)?; - let disposition = asset_object_disposition(&asset, &object); + let disposition = asset_disposition(&asset); + + if !state.config.proxy_downloads { + let presigned_url = storage + .presign_get_object( + &asset.s3_key, + Duration::from_secs(PRESIGNED_URL_EXPIRY_SECONDS), + disposition.as_deref(), + ) + .await + .storage_context("failed to generate download URL")?; + + return Ok(axum::response::Redirect::temporary(&presigned_url).into_response()); + } + proxy_storage_object( storage, - &object.s3_key, + &asset.s3_key, disposition.as_deref(), headers.get(header::RANGE).cloned(), Some(asset.mime_type.as_str()), - Some(object.id.to_string()), + Some(asset.id.to_string()), ) .await } @@ -1078,13 +1061,11 @@ pub async fn remove_tag( crate::routes::documents::MoveDocumentRequest, crate::routes::documents::BulkReanalyzeSelectionRequest, crate::routes::documents::BulkReanalyzeResponse, - crate::routes::documents::AssetObjectsQuery, crate::routes::documents::UploadDocumentForm, crate::documents::asset::DocumentVersionResponse, crate::documents::asset::DocumentVersionDetailResponse, crate::documents::asset::DocumentAssetResponse, crate::documents::asset::DocumentAssetDetailResponse, - crate::documents::asset::DocumentAssetObjectResponse, crate::documents::correspondents::DocumentCorrespondentResponse, crate::error::ApiErrorResponse, )) diff --git a/backend/src/routes/folders.rs b/backend/src/routes/folders.rs index 5bbd6bc..2a374d2 100644 --- a/backend/src/routes/folders.rs +++ b/backend/src/routes/folders.rs @@ -150,7 +150,7 @@ pub async fn list_folder_contents( )?; let documents = if include_documents { - service.hydrate_documents(&mut conn, user_id, documents)? + service.hydrate_documents(&mut conn, tenant_id, user_id, documents)? } else { Vec::new() }; diff --git a/backend/src/schema.rs b/backend/src/schema.rs index 4a42aeb..5d86f65 100644 --- a/backend/src/schema.rs +++ b/backend/src/schema.rs @@ -26,17 +26,6 @@ diesel::table! { } } -diesel::table! { - document_asset_objects (id) { - id -> Uuid, - asset_id -> Uuid, - ordinal -> Int4, - s3_key -> Text, - metadata -> Jsonb, - tenant_id -> Uuid, - } -} - diesel::table! { document_assets (id) { id -> Uuid, @@ -45,7 +34,7 @@ diesel::table! { mime_type -> Text, metadata -> Jsonb, created_at -> Timestamptz, - cardinality -> Nullable, + s3_key -> Text, tenant_id -> Uuid, } } @@ -294,8 +283,6 @@ diesel::table! { diesel::joinable!(correspondents -> tenants (tenant_id)); diesel::joinable!(capability_set_capabilities -> capability_sets (capability_set_id)); diesel::joinable!(capability_sets -> tenants (tenant_id)); -diesel::joinable!(document_asset_objects -> document_assets (asset_id)); -diesel::joinable!(document_asset_objects -> tenants (tenant_id)); diesel::joinable!(document_assets -> document_versions (document_version_id)); diesel::joinable!(document_assets -> tenants (tenant_id)); diesel::joinable!(document_correspondents -> correspondents (correspondent_id)); @@ -329,7 +316,6 @@ diesel::allow_tables_to_appear_in_same_query!( correspondents, capability_set_capabilities, capability_sets, - document_asset_objects, document_assets, document_correspondents, document_tags, diff --git a/backend/src/services/documents.rs b/backend/src/services/documents.rs index 312ee5b..3118d64 100644 --- a/backend/src/services/documents.rs +++ b/backend/src/services/documents.rs @@ -1,9 +1,6 @@ -use std::{ - collections::{HashMap, HashSet}, - time::Duration, -}; +use std::collections::{HashMap, HashSet}; -use chrono::{DateTime, NaiveDateTime, Utc}; +use chrono::{DateTime, Duration as ChronoDuration, NaiveDateTime, Utc}; use diesel::{ dsl::{exists, not, sql}, prelude::*, @@ -20,11 +17,9 @@ use uuid::Uuid; use crate::documents::{ asset::{ - asset_object_disposition, build_download_path, derive_document_title, - filename_with_retained_extension, load_asset_responses_with_conn, load_primary_assets, - to_asset_detail_response, to_asset_object_response, to_version_response, - DocumentAssetDetailResponse, DocumentAssetResponse, DocumentVersionDetailResponse, - DocumentVersionResponse, + build_download_path, derive_document_title, filename_with_retained_extension, + to_asset_detail_response, to_version_response, DocumentAssetDetailResponse, + DocumentAssetResponse, DocumentVersionDetailResponse, DocumentVersionResponse, }, correspondents::{ insert_document_correspondents, normalize_correspondent_ids, DocumentCorrespondentResponse, @@ -39,12 +34,10 @@ use crate::documents::{ use crate::error::{AppError, AppResult}; use crate::jobs::{enqueue_job, JobQueueError, JOB_ANALYZE_DOCUMENT, JOB_PURGE_DOCUMENT}; use crate::models::{ - Document, DocumentAsset, DocumentAssetObject, DocumentVersion, NewDocument, NewDocumentVersion, - Tag, + Document, DocumentAsset, DocumentVersion, NewDocument, NewDocumentVersion, Tag, }; use crate::schema::{ - document_asset_objects, document_assets, document_correspondents, document_tags, - document_versions, documents, folders, + document_assets, document_correspondents, document_tags, document_versions, documents, folders, }; use crate::services::{ correspondents::CorrespondentAssignmentInput, folders::gather_descendant_folder_ids, @@ -52,17 +45,11 @@ use crate::services::{ }; use crate::state::{AppState, PgPooledConnection}; use crate::utils::{ - db::validate_bulk_ids, - error::StorageResultExt, - http::inline_content_disposition, - json::{classify_nullable, NullableValue}, - setops::{intersect_option_sets, load_linked_doc_ids}, - storage_paths::document_version_object_key, - time::to_iso, + db::validate_bulk_ids, error::StorageResultExt, http::inline_content_disposition, + json::classify_nullable, json::NullableValue, setops::intersect_option_sets, + setops::load_linked_doc_ids, storage_paths::document_version_object_key, time::to_iso, }; -const PRESIGNED_URL_EXPIRY_SECONDS: u64 = 300; - #[derive(Deserialize, IntoParams, ToSchema, Clone)] #[into_params(parameter_in = Query)] pub struct DocumentListQuery { @@ -371,7 +358,7 @@ impl<'a> DocumentsService<'a> { .cloned() .unwrap_or_else(|| (Vec::new(), Vec::new())); - let assets = load_asset_responses_with_conn(conn, tenant_id, current_version.id)?; + let assets = self.load_asset_responses(conn, tenant_id, current_version.id, user_id)?; let current_version_data = Some((to_version_response(current_version), assets)); let response = @@ -601,7 +588,7 @@ impl<'a> DocumentsService<'a> { } let docs: Vec = docs_query.load(conn)?; - let mut responses = self.hydrate_documents(conn, user_id, docs)?; + let mut responses = self.hydrate_documents(conn, tenant_id, user_id, docs)?; if let Some(order) = quickwit_order { let order_map: HashMap = order @@ -618,6 +605,7 @@ impl<'a> DocumentsService<'a> { pub fn hydrate_documents( &self, conn: &mut PgPooledConnection, + tenant_id: Uuid, user_id: Uuid, docs: Vec, ) -> AppResult> { @@ -627,14 +615,57 @@ impl<'a> DocumentsService<'a> { let doc_ids: Vec = docs.iter().map(|doc| doc.id).collect(); let mut relations = load_tags_and_correspondents(conn, &doc_ids)?; - let primary_versions = load_primary_assets(conn, &docs)?; + let mut doc_to_version: HashMap = HashMap::with_capacity(doc_ids.len()); + let mut version_ids: Vec = Vec::with_capacity(doc_ids.len()); + for doc in &docs { + doc_to_version.insert(doc.id, doc.current_version_id); + version_ids.push(doc.current_version_id); + } + + version_ids.sort(); + version_ids.dedup(); + + let versions: Vec = document_versions::table + .filter(document_versions::id.eq_any(&version_ids)) + .filter(document_versions::tenant_id.eq(tenant_id)) + .load(conn)?; + + let mut version_map: HashMap = HashMap::new(); + for version in versions { + version_map.insert(version.id, version); + } + + let assets: Vec = document_assets::table + .filter(document_assets::document_version_id.eq_any(&version_ids)) + .filter(document_assets::tenant_id.eq(tenant_id)) + .order(( + document_assets::document_version_id.asc(), + document_assets::created_at.asc(), + )) + .load(conn)?; + + let mut assets_by_version: HashMap> = HashMap::new(); + for asset in assets { + let version_id = asset.document_version_id; + let response = self.asset_response(asset, tenant_id, user_id)?; + assets_by_version + .entry(version_id) + .or_default() + .push(response); + } docs.into_iter() .map(|doc| { let (tags, correspondents) = relations .remove(&doc.id) .unwrap_or_else(|| (Vec::new(), Vec::new())); - let current_version = primary_versions.get(&doc.id).cloned(); + let current_version = doc_to_version + .get(&doc.id) + .and_then(|version_id| version_map.remove(version_id)) + .map(|version| { + let assets = assets_by_version.remove(&version.id).unwrap_or_default(); + (to_version_response(version), assets) + }); self.to_document_response(user_id, doc, tags, correspondents, current_version) }) .collect() @@ -892,12 +923,24 @@ impl<'a> DocumentsService<'a> { &self, conn: &mut PgPooledConnection, tenant_id: Uuid, + user_id: Uuid, document_id: Uuid, ) -> AppResult> { let document = load_active_document(conn, tenant_id, document_id)?; let version_id = document.current_version_id; - Ok(load_asset_responses_with_conn(conn, tenant_id, version_id)?) + let assets = document_assets::table + .filter(document_assets::document_version_id.eq(version_id)) + .filter(document_assets::tenant_id.eq(tenant_id)) + .order(document_assets::created_at.asc()) + .load::(conn)?; + + let mut responses = Vec::with_capacity(assets.len()); + for asset in assets { + responses.push(self.asset_response(asset, tenant_id, user_id)?); + } + + Ok(responses) } pub async fn get_document_asset( @@ -906,8 +949,6 @@ impl<'a> DocumentsService<'a> { tenant_id: Uuid, user_id: Uuid, asset_id: Uuid, - start: i32, - limit: i32, ) -> AppResult { let asset: DocumentAsset = match document_assets::table .find(asset_id) @@ -919,77 +960,11 @@ impl<'a> DocumentsService<'a> { None => return Err(AppError::not_found()), }; - if start < 1 { - return Err(AppError::bad_request("start must be at least 1")); - } - if limit < 1 { - return Err(AppError::bad_request("limit must be at least 1")); - } - - let end = start - .checked_add(limit - 1) - .ok_or_else(|| AppError::bad_request("requested range is too large"))?; - - let objects: Vec = document_asset_objects::table - .filter(document_asset_objects::asset_id.eq(asset_id)) - .filter(document_asset_objects::tenant_id.eq(tenant_id)) - .filter(document_asset_objects::ordinal.ge(start)) - .filter(document_asset_objects::ordinal.le(end)) - .order(document_asset_objects::ordinal.asc()) - .load(&mut conn)?; - drop(conn); - let expires_at = Utc::now() - .timestamp_millis() - .checked_add((PRESIGNED_URL_EXPIRY_SECONDS as i64) * 1000) - .ok_or_else(|| AppError::internal("failed to compute expiry timestamp"))?; + let (url, expires_at) = self.asset_download_url(asset.id, tenant_id, user_id)?; - let storage = (!self.state.config.proxy_downloads) - .then(|| self.state.storage_for_tenant(tenant_id)) - .transpose()?; - - let mut object_responses = Vec::with_capacity(objects.len()); - for object in objects { - let response_disposition = asset_object_disposition(&asset, &object); - - let (url, object_expires_at) = if self.state.config.proxy_downloads { - let token = self - .state - .jwt - .generate_asset_download_token(asset.id, object.id, user_id, tenant_id) - .map_err(|err| { - error!(error = ?err, "failed to issue asset download token"); - AppError::internal("failed to issue asset download token") - })?; - (format!("/api/download/{token}"), None) - } else { - let storage = storage - .as_ref() - .expect("storage preloaded when proxy disabled"); - let url = storage - .presign_get_object( - &object.s3_key, - Duration::from_secs(PRESIGNED_URL_EXPIRY_SECONDS), - response_disposition.as_deref(), - ) - .await - .storage_context("failed to generate asset URL")?; - (url, Some(expires_at)) - }; - - object_responses.push(to_asset_object_response( - object, - Some(url), - object_expires_at, - )); - } - - if object_responses.is_empty() { - return Err(AppError::not_found()); - } - - Ok(to_asset_detail_response(asset, object_responses)) + Ok(to_asset_detail_response(asset, Some(url), Some(expires_at))) } pub fn list_document_versions( @@ -1025,7 +1000,7 @@ impl<'a> DocumentsService<'a> { .filter(document_versions::tenant_id.eq(tenant_id)) .first(conn)?; - let assets = load_asset_responses_with_conn(conn, tenant_id, version.id)?; + let assets = self.load_asset_responses(conn, tenant_id, version.id, user_id)?; let download_path = build_download_path(self.state, &document, version.id, user_id)?; let version_core = to_version_response(version); @@ -1180,7 +1155,7 @@ impl<'a> DocumentsService<'a> { let tags_and_correspondents = load_tags_and_correspondents(conn, &[document_id])?; let version_id = current_version.id; - let assets = load_asset_responses_with_conn(conn, tenant_id, version_id)?; + let assets = self.load_asset_responses(conn, tenant_id, version_id, user_id)?; let version_response = to_version_response(current_version); let (tags, correspondents) = tags_and_correspondents .get(&document_id) @@ -1493,7 +1468,7 @@ impl<'a> DocumentsService<'a> { .cloned() .unwrap_or_else(|| (Vec::new(), Vec::new())); - let assets = load_asset_responses_with_conn(conn, tenant_id, version.id)?; + let assets = self.load_asset_responses(conn, tenant_id, version.id, user_id)?; let version_response = to_version_response(version.clone()); info!( @@ -1514,4 +1489,66 @@ impl<'a> DocumentsService<'a> { Ok(Some(detail)) } + + fn asset_download_url( + &self, + asset_id: Uuid, + tenant_id: Uuid, + user_id: Uuid, + ) -> AppResult<(String, i64)> { + let token = self + .state + .jwt + .generate_asset_download_token(asset_id, user_id, tenant_id) + .map_err(|err| { + error!(error = ?err, "failed to issue asset download token"); + AppError::internal("failed to issue asset download token") + })?; + + let expires_at = Utc::now() + .checked_add_signed(ChronoDuration::minutes( + self.state.config.download_token_expiry_minutes, + )) + .ok_or_else(|| AppError::internal("failed to compute download expiry"))? + .timestamp_millis(); + + Ok((format!("/api/download/{token}"), expires_at)) + } + + fn asset_response( + &self, + asset: DocumentAsset, + tenant_id: Uuid, + user_id: Uuid, + ) -> AppResult { + let (url, expires_at) = self.asset_download_url(asset.id, tenant_id, user_id)?; + + Ok(DocumentAssetResponse { + id: asset.id, + asset_type: asset.asset_type, + mime_type: asset.mime_type, + metadata: asset.metadata, + url: Some(url), + expires_at: Some(expires_at), + }) + } + + fn load_asset_responses( + &self, + conn: &mut PgPooledConnection, + tenant_id: Uuid, + version_id: Uuid, + user_id: Uuid, + ) -> AppResult> { + let assets: Vec = document_assets::table + .filter(document_assets::document_version_id.eq(version_id)) + .filter(document_assets::tenant_id.eq(tenant_id)) + .order(document_assets::created_at.asc()) + .load(conn)?; + + assets + .into_iter() + .map(|asset| self.asset_response(asset, tenant_id, user_id)) + .collect() + } } diff --git a/backend/src/services/folders.rs b/backend/src/services/folders.rs index efbaf37..b64f798 100644 --- a/backend/src/services/folders.rs +++ b/backend/src/services/folders.rs @@ -555,10 +555,11 @@ impl<'a> FolderService<'a> { pub fn hydrate_documents( &self, conn: &mut PgPooledConnection, + tenant_id: Uuid, user_id: Uuid, docs: Vec, ) -> AppResult> { - DocumentsService::new(self.state).hydrate_documents(conn, user_id, docs) + DocumentsService::new(self.state).hydrate_documents(conn, tenant_id, user_id, docs) } } diff --git a/backend/src/test_support/mod.rs b/backend/src/test_support/mod.rs index f3b22d1..3ac95e5 100644 --- a/backend/src/test_support/mod.rs +++ b/backend/src/test_support/mod.rs @@ -877,7 +877,6 @@ async fn prepare_database(pool: &PgPool) -> Result<()> { fn truncate_all(conn: &mut PgConnection) -> Result<()> { conn.batch_execute( "TRUNCATE TABLE \ - tenant.document_asset_objects, \ tenant.document_assets, \ tenant.document_correspondents, \ tenant.correspondents, \ diff --git a/backend/src/utils/storage_paths.rs b/backend/src/utils/storage_paths.rs index 1f0a740..2db3e2c 100644 --- a/backend/src/utils/storage_paths.rs +++ b/backend/src/utils/storage_paths.rs @@ -48,8 +48,8 @@ fn document_asset_type_prefix(document_id: Uuid, version_number: i32, asset_type ) } -/// Returns the storage prefix under which the asset objects for a type/id pair live. -pub fn document_asset_object_prefix( +/// Returns the storage key for an asset (single object). +pub fn document_asset_key( document_id: Uuid, version_number: i32, asset_type: &str, @@ -62,21 +62,6 @@ pub fn document_asset_object_prefix( ) } -/// Returns the full storage key for a specific asset object (ordinal). -pub fn document_asset_object_key( - document_id: Uuid, - version_number: i32, - asset_type: &str, - asset_id: Uuid, - ordinal: i32, -) -> String { - format!( - "{}/{}", - document_asset_object_prefix(document_id, version_number, asset_type, asset_id), - ordinal - ) -} - #[cfg(test)] mod tests { use super::*; @@ -103,13 +88,8 @@ mod tests { ); assert_eq!( - document_asset_object_prefix(document_id, 3, "thumbnail", asset_id), + document_asset_key(document_id, 3, "thumbnail", asset_id), format!("documents/{document_id}/v3/assets/thumbnail/{asset_id}") ); - - assert_eq!( - document_asset_object_key(document_id, 3, "thumbnail", asset_id, 2), - format!("documents/{document_id}/v3/assets/thumbnail/{asset_id}/2") - ); } } diff --git a/backend/src/workers/common.rs b/backend/src/workers/common.rs index 4428663..fddacbb 100644 --- a/backend/src/workers/common.rs +++ b/backend/src/workers/common.rs @@ -3,8 +3,8 @@ use std::collections::HashMap; use diesel::{prelude::*, PgConnection}; use uuid::Uuid; -use crate::models::{Document, DocumentAsset, DocumentAssetObject, DocumentVersion}; -use crate::schema::{document_asset_objects, document_assets, document_versions, documents}; +use crate::models::{Document, DocumentAsset, DocumentVersion}; +use crate::schema::{document_assets, document_versions, documents}; use crate::state::AppState; pub(crate) struct LoadedDocumentVersion { @@ -41,7 +41,6 @@ pub(crate) fn load_document_version( pub struct LoadedAsset { pub asset: DocumentAsset, - pub objects: Vec, } pub(crate) fn load_version_assets( @@ -65,26 +64,9 @@ pub(crate) fn load_version_assets( .load(conn) .map_err(|err| format!("{err:?}"))?; - let asset_ids: Vec = assets.iter().map(|asset| asset.id).collect(); - - let mut object_map: HashMap> = HashMap::new(); - if !asset_ids.is_empty() { - let objects: Vec = document_asset_objects::table - .filter(document_asset_objects::asset_id.eq_any(&asset_ids)) - .filter(document_asset_objects::tenant_id.eq(tenant_id)) - .order(document_asset_objects::ordinal.asc()) - .load(conn) - .map_err(|err| format!("{err:?}"))?; - - for object in objects { - object_map.entry(object.asset_id).or_default().push(object); - } - } - let mut result = HashMap::with_capacity(assets.len()); for asset in assets { - let objects = object_map.remove(&asset.id).unwrap_or_default(); - result.insert(asset.asset_type.clone(), LoadedAsset { asset, objects }); + result.insert(asset.asset_type.clone(), LoadedAsset { asset }); } Ok(result) diff --git a/backend/src/workers/index.rs b/backend/src/workers/index.rs index 5e090ec..7a6c5af 100644 --- a/backend/src/workers/index.rs +++ b/backend/src/workers/index.rs @@ -48,12 +48,8 @@ impl Task for IndexDocumentTask { .asset(OCR_TEXT_ASSET_TYPE) .await? .ok_or_else(|| TaskError::fail("missing OCR text asset"))?; - let object = asset - .objects - .first() - .ok_or_else(|| TaskError::fail("missing OCR text object"))?; - let s3_key = object.s3_key.clone(); + let s3_key = asset.asset.s3_key.clone(); let bytes = ctx.storage().get_object(&s3_key).await.map_err(|err| { TaskError::retry( Duration::from_secs(30), diff --git a/backend/src/workers/issued_at.rs b/backend/src/workers/issued_at.rs index 0549000..238d3b4 100644 --- a/backend/src/workers/issued_at.rs +++ b/backend/src/workers/issued_at.rs @@ -430,8 +430,7 @@ async fn load_document_text(ctx: &mut DocumentVersionTaskContext) -> TaskResult< let assets = ctx.assets().await?; assets .get(OCR_TEXT_ASSET_TYPE) - .and_then(|asset| asset.objects.first()) - .map(|object| object.s3_key.clone()) + .map(|asset| asset.asset.s3_key.clone()) }; let Some(key) = object_key else { diff --git a/backend/src/workers/ocr.rs b/backend/src/workers/ocr.rs index bb4e750..dddc0f6 100644 --- a/backend/src/workers/ocr.rs +++ b/backend/src/workers/ocr.rs @@ -19,13 +19,10 @@ use uuid::Uuid; use crate::{ documents::asset::delete_asset, error::AppResult, - models::{ - Document, DocumentAsset, DocumentAssetObject, DocumentVersion, NewDocumentAsset, - NewDocumentAssetObject, - }, - schema::{document_asset_objects, document_assets}, + models::{Document, DocumentAsset, DocumentVersion, NewDocumentAsset}, + schema::document_assets, state::AppState, - utils::storage_paths::document_asset_object_prefix, + utils::storage_paths::document_asset_key, }; use super::taskflow::{ @@ -83,7 +80,7 @@ impl Task for GenerateOcrTask { remove_existing_ocr_asset(ctx, &context).await; let asset_id = Uuid::new_v4(); - let s3_key = document_asset_object_prefix( + let s3_key = document_asset_key( context.document.id, context.version.version_number, OCR_TEXT_ASSET_TYPE, @@ -123,7 +120,6 @@ struct OcrContext { document: Document, version: DocumentVersion, existing_asset: Option, - existing_objects: Vec, skip: bool, } @@ -135,16 +131,13 @@ async fn build_ocr_context( let version = ctx.version().await?.clone(); let asset = ctx.asset(OCR_TEXT_ASSET_TYPE).await?; - let (existing_asset, existing_objects) = asset - .map(|asset| (Some(asset.asset.clone()), asset.objects.clone())) - .unwrap_or((None, Vec::new())); + let existing_asset = asset.map(|asset| asset.asset.clone()); if !document_is_pdf(&document) { return Ok(OcrContext { document, version, existing_asset, - existing_objects, skip: true, }); } @@ -155,22 +148,19 @@ async fn build_ocr_context( document, version, existing_asset, - existing_objects, skip, }) } async fn remove_existing_ocr_asset(ctx: &DocumentVersionTaskContext, context: &OcrContext) { if let Some(existing_asset) = &context.existing_asset { - for object in &context.existing_objects { - if let Err(err) = ctx.storage().delete_object(&object.s3_key).await { - warn!( - job_id = %ctx.job_id(), - error = %err, - s3_key = %object.s3_key, - "failed to delete existing ocr asset object" - ); - } + if let Err(err) = ctx.storage().delete_object(&existing_asset.s3_key).await { + warn!( + job_id = %ctx.job_id(), + error = %err, + s3_key = %existing_asset.s3_key, + "failed to delete existing ocr asset object" + ); } let tenant_id = context.document.tenant_id; @@ -239,7 +229,7 @@ fn persist_ocr_metadata( asset_type: OCR_TEXT_ASSET_TYPE.to_string(), mime_type: "text/plain".to_string(), metadata, - cardinality: Some(1), + s3_key: s3_key.to_string(), tenant_id, }; @@ -253,34 +243,12 @@ fn persist_ocr_metadata( .set(( document_assets::mime_type.eq(excluded(document_assets::mime_type)), document_assets::metadata.eq(excluded(document_assets::metadata)), - document_assets::cardinality.eq(excluded(document_assets::cardinality)), + document_assets::s3_key.eq(excluded(document_assets::s3_key)), document_assets::id.eq(excluded(document_assets::id)), )) .execute(&mut conn) .map_err(|err| format!("{err:?}"))?; - diesel::delete( - document_asset_objects::table - .filter(document_asset_objects::asset_id.eq(asset_id)) - .filter(document_asset_objects::tenant_id.eq(tenant_id)), - ) - .execute(&mut conn) - .map_err(|err| format!("{err:?}"))?; - - let object = NewDocumentAssetObject { - id: Uuid::new_v4(), - asset_id, - ordinal: 1, - s3_key: s3_key.to_string(), - metadata: json!({}), - tenant_id, - }; - - diesel::insert_into(document_asset_objects::table) - .values(&object) - .execute(&mut conn) - .map_err(|err| format!("{err:?}"))?; - Ok(()) } diff --git a/backend/src/workers/purge.rs b/backend/src/workers/purge.rs index 052d3e5..4d4e305 100644 --- a/backend/src/workers/purge.rs +++ b/backend/src/workers/purge.rs @@ -11,7 +11,7 @@ use uuid::Uuid; use crate::auth::ensure_active_tenant; use crate::jobs::JOB_PURGE_DOCUMENT; use crate::models::{Document, DocumentVersion}; -use crate::schema::{document_asset_objects, document_assets, document_versions}; +use crate::schema::{document_assets, document_versions}; use crate::state::AppState; use crate::storage::TenantStorage; @@ -235,21 +235,11 @@ fn prepare_purge_context( let asset_keys = if version_ids.is_empty() { Vec::new() } else { - let asset_ids: Vec = document_assets::table + document_assets::table .filter(document_assets::document_version_id.eq_any(&version_ids)) .filter(document_assets::tenant_id.eq(tenant_id)) - .select(document_assets::id) - .load(conn)?; - - if asset_ids.is_empty() { - Vec::new() - } else { - document_asset_objects::table - .filter(document_asset_objects::asset_id.eq_any(&asset_ids)) - .filter(document_asset_objects::tenant_id.eq(tenant_id)) - .select(document_asset_objects::s3_key) - .load(conn)? - } + .select(document_assets::s3_key) + .load(conn)? }; Ok(Some(PurgeContext { diff --git a/backend/src/workers/tenants.rs b/backend/src/workers/tenants.rs index b523e6c..2a235f9 100644 --- a/backend/src/workers/tenants.rs +++ b/backend/src/workers/tenants.rs @@ -22,9 +22,8 @@ use crate::documents::search::{delete_quickwit_index, ensure_quickwit_index}; use crate::jobs::{JOB_DELETE_TENANT, JOB_PROVISION_TENANT}; use crate::models::{NewUserMembership, Tenant, TenantStatus}; use crate::schema::{ - api_tokens, correspondents, document_asset_objects, document_assets, document_correspondents, - document_tags, document_versions, documents, folders, tags, tenants, user_memberships, - user_sessions, + api_tokens, correspondents, document_assets, document_correspondents, document_tags, + document_versions, documents, folders, tags, tenants, user_memberships, user_sessions, }; use crate::state::AppState; use crate::tenants::TenantRepository; @@ -401,8 +400,8 @@ fn collect_object_keys(conn: &mut PgConnection) -> Result(conn)?; - let asset_keys = document_asset_objects::table - .select(document_asset_objects::s3_key) + let asset_keys = document_assets::table + .select(document_assets::s3_key) .load::(conn)?; Ok(TenantObjectKeys { @@ -459,10 +458,6 @@ fn delete_tenant_rows( remove_memberships: bool, ) -> Result<(), diesel::result::Error> { conn.transaction(|conn| { - diesel::delete( - document_asset_objects::table.filter(document_asset_objects::tenant_id.eq(tenant_id)), - ) - .execute(conn)?; diesel::delete(document_assets::table.filter(document_assets::tenant_id.eq(tenant_id))) .execute(conn)?; diesel::delete( diff --git a/backend/src/workers/thumbnails.rs b/backend/src/workers/thumbnails.rs index 9c6eea7..9000d3b 100644 --- a/backend/src/workers/thumbnails.rs +++ b/backend/src/workers/thumbnails.rs @@ -5,7 +5,7 @@ use chrono::Utc; use diesel::{pg::upsert::excluded, prelude::*}; use image::{GenericImageView, ImageFormat, ImageReader}; use pdfium_render::prelude::*; -use serde_json::{json, Map, Value}; +use serde_json::{Map, Value}; use tokio::task; use tracing::{info, warn}; use uuid::Uuid; @@ -13,13 +13,10 @@ use uuid::Uuid; use crate::{ documents::asset::delete_asset, error::AppResult, - models::{ - Document, DocumentAsset, DocumentAssetObject, DocumentVersion, NewDocumentAsset, - NewDocumentAssetObject, - }, - schema::{document_asset_objects, document_assets, document_versions}, + models::{Document, DocumentAsset, DocumentVersion, NewDocumentAsset}, + schema::{document_assets, document_versions}, state::AppState, - utils::storage_paths::document_asset_object_key, + utils::storage_paths::document_asset_key, }; use super::{ @@ -82,7 +79,7 @@ impl Task for GenerateThumbnailsTask { let thumbnail_asset_id = Uuid::new_v4(); - let thumbnail_objects = upload_generated_objects( + let thumbnail_persistence = upload_generated_asset( ctx, &context, THUMBNAIL_ASSET_TYPE, @@ -91,11 +88,7 @@ impl Task for GenerateThumbnailsTask { ) .await?; - let asset_persistences = vec![AssetPersistence { - asset_type: THUMBNAIL_ASSET_TYPE, - asset_id: thumbnail_asset_id, - objects: thumbnail_objects, - }]; + let asset_persistences = vec![thumbnail_persistence]; let state = ctx.state().clone(); let tenant_id = context.document.tenant_id; @@ -131,32 +124,22 @@ async fn build_thumbnail_context( document, version, existing_thumbnail: None, - existing_thumbnail_objects: Vec::new(), skip: true, tenant_id, }); } let assets = ctx.assets().await?; - let (existing_thumbnail, existing_thumbnail_objects) = assets + let existing_thumbnail = assets .get(THUMBNAIL_ASSET_TYPE) - .map(|entry| (Some(entry.asset.clone()), entry.objects.clone())) - .unwrap_or((None, Vec::new())); - let expected_cardinality = expected_asset_cardinality(&document, &version); - let thumbnail_cardinality = existing_thumbnail - .as_ref() - .and_then(|asset| asset.cardinality) - .unwrap_or_else(|| existing_thumbnail_objects.len() as i32); - let needs_regeneration = thumbnail_cardinality < expected_cardinality - || (existing_thumbnail_objects.len() as i32) < expected_cardinality; + .map(|entry| entry.asset.clone()); - let skip = existing_thumbnail.is_some() && !force && !needs_regeneration; + let skip = existing_thumbnail.is_some() && !force; Ok(ThumbnailContext { document, version, existing_thumbnail, - existing_thumbnail_objects, skip, tenant_id, }) @@ -167,25 +150,18 @@ async fn remove_existing_thumbnail_assets( context: &ThumbnailContext, ) { if let Some(existing_thumbnail) = &context.existing_thumbnail { - delete_asset_with_objects(ctx, existing_thumbnail, &context.existing_thumbnail_objects) - .await; + delete_asset_object(ctx, existing_thumbnail).await; } } -async fn delete_asset_with_objects( - ctx: &DocumentVersionTaskContext, - asset: &DocumentAsset, - objects: &[DocumentAssetObject], -) { - for object in objects { - if let Err(err) = ctx.storage().delete_object(&object.s3_key).await { - warn!( - job_id = %ctx.job_id(), - error = %err, - s3_key = %object.s3_key, - "failed to delete existing asset object" - ); - } +async fn delete_asset_object(ctx: &DocumentVersionTaskContext, asset: &DocumentAsset) { + if let Err(err) = ctx.storage().delete_object(&asset.s3_key).await { + warn!( + job_id = %ctx.job_id(), + error = %err, + s3_key = %asset.s3_key, + "failed to delete existing asset object" + ); } let tenant_id = ctx.tenant_id(); @@ -217,54 +193,45 @@ async fn delete_asset_with_objects( } } -async fn upload_generated_objects( +async fn upload_generated_asset( ctx: &DocumentVersionTaskContext, context: &ThumbnailContext, asset_type: &str, asset_id: Uuid, asset: &GeneratedAsset, -) -> TaskResult> { - let mut objects = Vec::with_capacity(asset.objects.len()); +) -> TaskResult { + let image = &asset.image; - for (index, image) in asset.objects.iter().enumerate() { - if index + 1 > i32::MAX as usize { - return Err(TaskError::fail("too many generated asset objects")); - } - let ordinal = (index + 1) as i32; - let s3_key = document_asset_object_key( - context.document.id, - context.version.version_number, - asset_type, - asset_id, - ordinal, - ); + let s3_key = document_asset_key( + context.document.id, + context.version.version_number, + asset_type, + asset_id, + ); - ctx.storage() - .put_object( - &s3_key, - image.image_bytes.clone(), - Some("image/webp".into()), - None, - ) - .await - .map_err(|err| TaskError::retry(Duration::from_secs(30), err.to_string()))?; + ctx.storage() + .put_object( + &s3_key, + image.image_bytes.clone(), + Some("image/webp".into()), + None, + ) + .await + .map_err(|err| TaskError::retry(Duration::from_secs(30), err.to_string()))?; - objects.push(AssetObjectPersistence { - ordinal, - s3_key, - width: image.width, - height: image.height, - }); - } - - Ok(objects) + Ok(AssetPersistence { + asset_type: asset_type.to_string(), + asset_id, + s3_key, + width: image.width, + height: image.height, + }) } struct ThumbnailContext { document: Document, version: DocumentVersion, existing_thumbnail: Option, - existing_thumbnail_objects: Vec, skip: bool, tenant_id: Uuid, } @@ -276,7 +243,7 @@ struct GeneratedImage { } struct GeneratedAsset { - objects: Vec, + image: GeneratedImage, } struct GeneratedAssets { @@ -284,19 +251,14 @@ struct GeneratedAssets { page_count: Option, } -struct AssetObjectPersistence { - ordinal: i32, +struct AssetPersistence { + asset_type: String, + asset_id: Uuid, s3_key: String, width: Option, height: Option, } -struct AssetPersistence { - asset_type: &'static str, - asset_id: Uuid, - objects: Vec, -} - fn generate_thumbnails(document: &Document, bytes: &[u8]) -> Result { if document_is_pdf(document) { let pdf_assets = generate_pdf_assets(bytes)?; @@ -334,9 +296,7 @@ fn generate_image_assets(bytes: &[u8]) -> Result { let thumbnail = encode_dynamic_image(thumbnail_image)?; - Ok(GeneratedAsset { - objects: vec![thumbnail], - }) + Ok(GeneratedAsset { image: thumbnail }) } struct PdfGeneratedAssets { @@ -354,6 +314,9 @@ fn generate_pdf_assets(bytes: &[u8]) -> Result { let pages = document.pages(); let total_pages = pages.len() as usize; + if total_pages == 0 { + return Err("pdf has no pages".to_string()); + } let render_config = PdfRenderConfig::new() .set_target_width(RENDER_WIDTH as i32) @@ -361,29 +324,21 @@ fn generate_pdf_assets(bytes: &[u8]) -> Result { .render_form_data(true) .rotate_if_landscape(PdfPageRenderRotation::None, true); - let mut thumbnail_objects: Vec = Vec::with_capacity(total_pages); + let first_page = pages.get(0).map_err(|err| format!("load page 0: {err}"))?; - for page_index in 0..total_pages { - let page = pages - .get(u16::try_from(page_index).map_err(|_| "page index overflow".to_string())?) - .map_err(|err| format!("load page {page_index}: {err}"))?; + let bitmap = first_page + .render_with_config(&render_config) + .map_err(|err| format!("render pdf page 0: {err}"))?; - let bitmap = page - .render_with_config(&render_config) - .map_err(|err| format!("render pdf page {page_index}: {err}"))?; + let render_buffer = bitmap.as_image().to_rgb8(); + let render_image = image::DynamicImage::ImageRgb8(render_buffer); - let render_buffer = bitmap.as_image().to_rgb8(); - let render_image = image::DynamicImage::ImageRgb8(render_buffer); - - let thumbnail_image = - if render_image.width() > THUMBNAIL_WIDTH || render_image.height() > THUMBNAIL_HEIGHT { - render_image.thumbnail(THUMBNAIL_WIDTH, THUMBNAIL_HEIGHT) - } else { - render_image.clone() - }; - - thumbnail_objects.push(encode_dynamic_image(thumbnail_image)?); - } + let thumbnail_image = + if render_image.width() > THUMBNAIL_WIDTH || render_image.height() > THUMBNAIL_HEIGHT { + render_image.thumbnail(THUMBNAIL_WIDTH, THUMBNAIL_HEIGHT) + } else { + render_image.clone() + }; let page_count: u32 = total_pages .try_into() @@ -391,7 +346,7 @@ fn generate_pdf_assets(bytes: &[u8]) -> Result { Ok(PdfGeneratedAssets { thumbnail: GeneratedAsset { - objects: thumbnail_objects, + image: encode_dynamic_image(thumbnail_image)?, }, page_count, }) @@ -421,28 +376,25 @@ fn persist_assets_metadata( .map_err(|err| format!("{err:?}"))?; for asset in assets { - if asset.objects.is_empty() { - return Err(format!( - "asset {} has no generated objects", - asset.asset_type - )); + let mut metadata_map = Map::new(); + if let Some(width) = asset.width { + metadata_map.insert("width".to_string(), Value::from(width)); } - - let object_count: i32 = asset - .objects - .len() - .try_into() - .map_err(|_| "asset contains too many objects".to_string())?; + if let Some(height) = asset.height { + metadata_map.insert("height".to_string(), Value::from(height)); + } + metadata_map.insert( + "generated_at".to_string(), + Value::from(Utc::now().to_rfc3339()), + ); let new_asset = NewDocumentAsset { id: asset.asset_id, document_version_id: version_id, - asset_type: asset.asset_type.to_string(), + asset_type: asset.asset_type.clone(), mime_type: "image/webp".to_string(), - metadata: json!({ - "generated_at": Utc::now().to_rfc3339(), - }), - cardinality: Some(object_count), + metadata: Value::Object(metadata_map), + s3_key: asset.s3_key.clone(), tenant_id, }; @@ -456,44 +408,10 @@ fn persist_assets_metadata( .set(( document_assets::mime_type.eq(excluded(document_assets::mime_type)), document_assets::metadata.eq(excluded(document_assets::metadata)), - document_assets::cardinality.eq(excluded(document_assets::cardinality)), + document_assets::s3_key.eq(excluded(document_assets::s3_key)), )) .execute(&mut conn) .map_err(|err| format!("{err:?}"))?; - - diesel::delete( - document_asset_objects::table - .filter(document_asset_objects::asset_id.eq(new_asset.id)) - .filter(document_asset_objects::tenant_id.eq(tenant_id)), - ) - .execute(&mut conn) - .map_err(|err| format!("{err:?}"))?; - - for object in &asset.objects { - let mut metadata_map = Map::new(); - if let Some(width) = object.width { - metadata_map.insert("width".to_string(), Value::from(width)); - } - if let Some(height) = object.height { - metadata_map.insert("height".to_string(), Value::from(height)); - } - - let object_metadata = Value::Object(metadata_map); - - let new_object = NewDocumentAssetObject { - id: Uuid::new_v4(), - asset_id: new_asset.id, - ordinal: object.ordinal, - s3_key: object.s3_key.clone(), - metadata: object_metadata, - tenant_id, - }; - - diesel::insert_into(document_asset_objects::table) - .values(&new_object) - .execute(&mut conn) - .map_err(|err| format!("{err:?}"))?; - } } Ok(()) @@ -557,22 +475,3 @@ fn document_is_pdf(document: &Document) -> bool { .unwrap_or(false) }) } - -fn expected_asset_cardinality(document: &Document, version: &DocumentVersion) -> i32 { - if let Value::Object(map) = &version.metadata { - if let Some(count) = map.get("page_count").and_then(|v| v.as_i64()) { - if count > 0 { - return count - .min(i64::from(i32::MAX)) - .try_into() - .unwrap_or(i32::MAX); - } - } - } - - if document_is_pdf(document) { - 1 - } else { - 1 - } -} diff --git a/backend/tests/documents_flow.rs b/backend/tests/documents_flow.rs index 61eeeaf..5735362 100644 --- a/backend/tests/documents_flow.rs +++ b/backend/tests/documents_flow.rs @@ -7,8 +7,8 @@ use serde_json::{json, Value}; use uuid::Uuid; use papercrate::jobs::{mark_job_succeeded, JOB_PURGE_DOCUMENT}; -use papercrate::models::{Job, NewDocumentAsset, NewDocumentAssetObject}; -use papercrate::schema::{document_asset_objects, document_assets}; +use papercrate::models::{Job, NewDocumentAsset}; +use papercrate::schema::document_assets; use papercrate::workers::{purge::PurgeDocumentJob, JobExecution, JobHandler}; use std::sync::Arc; #[derive(Deserialize)] @@ -66,12 +66,6 @@ struct DocumentAssetInfo { #[derive(Deserialize)] struct AssetProxyDetail { - id: Uuid, - objects: Vec, -} - -#[derive(Deserialize)] -struct AssetProxyObject { id: Uuid, url: Option, expires_at: Option, @@ -326,7 +320,7 @@ async fn asset_detail_uses_proxy_urls_when_configured() -> Result<()> { .map_err(|err| anyhow!("tenant connection: {err:?}"))?; let asset_id = Uuid::new_v4(); - let object_id = Uuid::new_v4(); + let s3_key = "objects/preview.png".to_string(); diesel::insert_into(document_assets::table) .values(&NewDocumentAsset { @@ -335,18 +329,7 @@ async fn asset_detail_uses_proxy_urls_when_configured() -> Result<()> { asset_type: "preview".to_string(), mime_type: "image/png".to_string(), metadata: json!({}), - cardinality: Some(1), - tenant_id, - }) - .execute(&mut conn)?; - - diesel::insert_into(document_asset_objects::table) - .values(&NewDocumentAssetObject { - id: object_id, - asset_id, - ordinal: 1, - s3_key: "objects/preview.png".to_string(), - metadata: json!({}), + s3_key: s3_key.clone(), tenant_id, }) .execute(&mut conn)?; @@ -360,15 +343,12 @@ async fn asset_detail_uses_proxy_urls_when_configured() -> Result<()> { let body = body_to_vec(response.into_body()).await?; let asset_detail: AssetProxyDetail = serde_json::from_slice(&body)?; assert_eq!(asset_detail.id, asset_id); - assert_eq!(asset_detail.objects.len(), 1); - let object = &asset_detail.objects[0]; - assert_eq!(object.id, object_id); - let url = object + let url = asset_detail .url .as_deref() .ok_or_else(|| anyhow!("missing url"))?; assert!(url.starts_with("/api/download/")); - assert!(object.expires_at.is_none()); + assert!(asset_detail.expires_at.is_some()); app.cleanup().await?; Ok(()) diff --git a/docker-composer.dev.yml b/docker-compose.dev.yml similarity index 100% rename from docker-composer.dev.yml rename to docker-compose.dev.yml diff --git a/k8s/papercrate/templates/migrate-job.yaml b/k8s/papercrate/templates/migrate-job.yaml index 7a100dd..be87a1c 100644 --- a/k8s/papercrate/templates/migrate-job.yaml +++ b/k8s/papercrate/templates/migrate-job.yaml @@ -65,6 +65,11 @@ spec: secretKeyRef: name: {{ $migrateSecretName }} key: {{ $migrateSecretKey }} + - name: DATABASE_URL + valueFrom: + secretKeyRef: + name: {{ $migrateSecretName }} + key: {{ $migrateSecretKey }} {{- range .Values.migrateJob.extraEnv }} - name: {{ .name }} value: {{ .value | quote }}