diff --git a/backend/migrations/202503240005_create_document_asset_objects/down.sql b/backend/migrations/202503240005_create_document_asset_objects/down.sql new file mode 100644 index 0000000..f4ce4ef --- /dev/null +++ b/backend/migrations/202503240005_create_document_asset_objects/down.sql @@ -0,0 +1,34 @@ +ALTER TABLE document_assets + ADD COLUMN s3_key TEXT; + +UPDATE document_assets AS da +SET s3_key = dao.s3_key +FROM document_asset_objects AS dao +WHERE dao.asset_id = da.id + AND dao.ordinal = 1 + AND da.s3_key IS NULL; + +ALTER TABLE document_assets + ALTER COLUMN s3_key SET NOT NULL; + +UPDATE document_assets AS da +SET metadata = jsonb_set(da.metadata, '{width}', dao.metadata->'width', true) +FROM document_asset_objects AS dao +WHERE dao.asset_id = da.id + AND dao.ordinal = 1 + AND dao.metadata ? 'width'; + +UPDATE document_assets AS da +SET metadata = jsonb_set(da.metadata, '{height}', dao.metadata->'height', true) +FROM document_asset_objects AS dao +WHERE dao.asset_id = da.id + AND dao.ordinal = 1 + AND dao.metadata ? 'height'; + +DROP INDEX IF EXISTS idx_document_asset_objects_asset_ordinal; +DROP TABLE IF EXISTS document_asset_objects; + +ALTER TABLE document_assets + DROP CONSTRAINT IF EXISTS document_assets_cardinality_positive; +ALTER TABLE document_assets + DROP COLUMN IF EXISTS cardinality; diff --git a/backend/migrations/202503240005_create_document_asset_objects/up.sql b/backend/migrations/202503240005_create_document_asset_objects/up.sql new file mode 100644 index 0000000..46574e5 --- /dev/null +++ b/backend/migrations/202503240005_create_document_asset_objects/up.sql @@ -0,0 +1,54 @@ +ALTER TABLE document_assets + ADD COLUMN cardinality INTEGER, + ADD CONSTRAINT document_assets_cardinality_positive CHECK (cardinality IS NULL OR cardinality >= 1); + +CREATE TABLE document_asset_objects ( + id UUID PRIMARY KEY, + asset_id UUID NOT NULL REFERENCES document_assets(id) ON DELETE CASCADE, + ordinal INTEGER NOT NULL, + s3_key TEXT NOT NULL, + metadata JSONB NOT NULL DEFAULT '{}'::jsonb, + 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 document_asset_objects (asset_id, ordinal); + +INSERT INTO document_asset_objects (id, asset_id, ordinal, s3_key, metadata) +SELECT + gen_random_uuid(), + id, + 1, + s3_key, + '{}'::jsonb +FROM document_assets; + +UPDATE document_asset_objects AS dao +SET metadata = jsonb_set(dao.metadata, '{width}', da.metadata->'width', true) +FROM document_assets AS da +WHERE dao.asset_id = da.id + AND dao.ordinal = 1 + AND da.metadata ? 'width'; + +UPDATE document_asset_objects AS dao +SET metadata = jsonb_set(dao.metadata, '{height}', da.metadata->'height', true) +FROM document_assets AS da +WHERE dao.asset_id = da.id + AND dao.ordinal = 1 + AND da.metadata ? 'height'; + +UPDATE document_assets +SET metadata = metadata - 'width' +WHERE metadata ? 'width'; + +UPDATE document_assets +SET metadata = metadata - 'height' +WHERE metadata ? 'height'; + +UPDATE document_assets +SET cardinality = 1 +WHERE cardinality IS NULL; + +ALTER TABLE document_assets + DROP COLUMN s3_key; diff --git a/backend/src/bin/maintenance.rs b/backend/src/bin/maintenance.rs index 27e51bf..f6e17f2 100644 --- a/backend/src/bin/maintenance.rs +++ b/backend/src/bin/maintenance.rs @@ -2,13 +2,14 @@ use std::env; use anyhow::{Context, Result}; use diesel::prelude::*; +use uuid::Uuid; use backend::{ config::AppConfig, db, - models::DocumentAsset, + models::{DocumentAsset, DocumentAssetObject}, s3, - schema::document_assets, + schema::{document_asset_objects, document_assets}, storage::{ObjectStorage, S3Storage}, }; @@ -57,11 +58,18 @@ async fn delete_all_assets() -> Result<()> { println!("Deleting {} assets…", assets.len()); - for asset in &assets { - if let Err(err) = storage.delete_object(&asset.s3_key).await { + let asset_ids: Vec = assets.iter().map(|asset| asset.id).collect(); + + let objects: Vec = document_asset_objects::table + .filter(document_asset_objects::asset_id.eq_any(&asset_ids)) + .load(&mut conn) + .context("failed to load document asset objects")?; + + for object in &objects { + if let Err(err) = storage.delete_object(&object.s3_key).await { eprintln!( "Failed to delete object {} from storage: {err}", - asset.s3_key + object.s3_key ); } } diff --git a/backend/src/models.rs b/backend/src/models.rs index 8473f86..afeedbb 100644 --- a/backend/src/models.rs +++ b/backend/src/models.rs @@ -109,10 +109,10 @@ pub struct DocumentAsset { pub id: Uuid, pub document_version_id: Uuid, pub asset_type: String, - pub s3_key: String, pub mime_type: String, pub metadata: serde_json::Value, pub created_at: NaiveDateTime, + pub cardinality: Option, } #[derive(Debug, Insertable)] @@ -121,9 +121,30 @@ pub struct NewDocumentAsset { pub id: Uuid, pub document_version_id: Uuid, pub asset_type: String, - pub s3_key: String, pub mime_type: String, pub metadata: serde_json::Value, + pub cardinality: Option, +} + +#[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, +} + +#[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, } #[derive(Debug, Clone, Queryable, Identifiable)] diff --git a/backend/src/routes/documents.rs b/backend/src/routes/documents.rs index 64971a6..210e80d 100644 --- a/backend/src/routes/documents.rs +++ b/backend/src/routes/documents.rs @@ -22,12 +22,13 @@ use crate::auth::AuthenticatedUser; use crate::error::{AppError, AppResult}; use crate::jobs::{enqueue_job, JOB_ANALYZE_DOCUMENT}; use crate::models::{ - Correspondent, Document, DocumentAsset, DocumentCorrespondent, DocumentVersion, NewDocument, - NewDocumentCorrespondent, NewDocumentTag, NewDocumentVersion, Tag, + Correspondent, Document, DocumentAsset, DocumentAssetObject, DocumentCorrespondent, + DocumentVersion, NewDocument, NewDocumentCorrespondent, NewDocumentTag, NewDocumentVersion, + Tag, }; use crate::schema::{ - correspondents, document_assets, document_correspondents, document_tags, document_versions, - documents, folders, refresh_tokens::dsl as refresh_dsl, tags, + correspondents, document_asset_objects, document_assets, document_correspondents, + document_tags, document_versions, documents, folders, refresh_tokens::dsl as refresh_dsl, tags, }; use crate::state::AppState; @@ -856,7 +857,14 @@ pub async fn get_document_asset( return Err(AppError::not_found()); } - let asset: DocumentAsset = document_assets::table.find(asset_id).first(&mut conn)?; + let (asset, object): (DocumentAsset, DocumentAssetObject) = document_assets::table + .inner_join( + document_asset_objects::table.on(document_asset_objects::asset_id + .eq(document_assets::id) + .and(document_asset_objects::ordinal.eq(1))), + ) + .filter(document_assets::id.eq(asset_id)) + .first(&mut conn)?; let version: DocumentVersion = document_versions::table .find(asset.document_version_id) .first(&mut conn)?; @@ -865,7 +873,8 @@ pub async fn get_document_asset( return Err(AppError::not_found()); } - let s3_key = asset.s3_key.clone(); + let s3_key = object.s3_key.clone(); + let object_metadata = object.metadata.clone(); drop(conn); let presigned_url = state @@ -874,7 +883,11 @@ pub async fn get_document_asset( .await .map_err(|err| AppError::internal(format!("failed to generate asset URL: {err}")))?; - Ok(Json(to_asset_response(asset, Some(presigned_url)))) + Ok(Json(to_asset_response( + asset, + Some(object_metadata), + Some(presigned_url), + ))) } pub async fn download_document( @@ -1784,18 +1797,27 @@ pub(crate) async fn load_primary_assets( version_map.insert(version.id, version); } - let assets: Vec = document_assets::table + 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))), + ) .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(&mut conn)?; let mut assets_by_version: HashMap> = HashMap::new(); - for asset in assets { + for (asset, object) in assets { let version_id = asset.document_version_id; - let response = to_asset_response(asset, None); + let response = to_asset_response(asset, object.map(|o| o.metadata), None); assets_by_version .entry(version_id) .or_default() @@ -1878,8 +1900,26 @@ fn to_version_response(version: DocumentVersion) -> DocumentVersionResponse { } } -fn to_asset_response(asset: DocumentAsset, url: Option) -> DocumentAssetResponse { - let metadata = asset.metadata.clone(); +fn merge_metadata(base: &Value, overlay: Option<&Value>) -> Value { + match (base, overlay) { + (Value::Object(base_obj), Some(Value::Object(overlay_obj))) => { + let mut merged = base_obj.clone(); + for (key, value) in overlay_obj { + merged.insert(key.clone(), value.clone()); + } + Value::Object(merged) + } + (_, Some(value)) => value.clone(), + (value, None) => value.clone(), + } +} + +fn to_asset_response( + asset: DocumentAsset, + object_metadata: Option, + url: Option, +) -> DocumentAssetResponse { + let metadata = merge_metadata(&asset.metadata, object_metadata.as_ref()); DocumentAssetResponse { id: asset.id, asset_type: asset.asset_type, @@ -1931,15 +1971,24 @@ async fn load_asset_responses( version_id: Uuid, ) -> AppResult> { let mut conn = state.db()?; - let assets: Vec = document_assets::table + 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))), + ) .filter(document_assets::document_version_id.eq(version_id)) .order(document_assets::created_at.asc()) + .select(( + document_assets::all_columns, + document_asset_objects::all_columns.nullable(), + )) .load(&mut conn)?; drop(conn); Ok(assets .into_iter() - .map(|asset| to_asset_response(asset, None)) + .map(|(asset, object)| to_asset_response(asset, object.map(|o| o.metadata), None)) .collect()) } diff --git a/backend/src/schema.rs b/backend/src/schema.rs index 57ff54a..85591f6 100644 --- a/backend/src/schema.rs +++ b/backend/src/schema.rs @@ -11,15 +11,25 @@ diesel::table! { } } +diesel::table! { + document_asset_objects (id) { + id -> Uuid, + asset_id -> Uuid, + ordinal -> Int4, + s3_key -> Text, + metadata -> Jsonb, + } +} + diesel::table! { document_assets (id) { id -> Uuid, document_version_id -> Uuid, asset_type -> Text, - s3_key -> Text, mime_type -> Text, metadata -> Jsonb, created_at -> Timestamptz, + cardinality -> Nullable, } } @@ -143,6 +153,7 @@ diesel::table! { } } +diesel::joinable!(document_asset_objects -> document_assets (asset_id)); diesel::joinable!(document_assets -> document_versions (document_version_id)); diesel::joinable!(document_correspondents -> correspondents (correspondent_id)); diesel::joinable!(document_correspondents -> documents (document_id)); @@ -155,6 +166,7 @@ diesel::joinable!(refresh_tokens -> users (user_id)); diesel::allow_tables_to_appear_in_same_query!( correspondents, + document_asset_objects, document_assets, document_correspondents, document_tags, diff --git a/backend/src/workers/index.rs b/backend/src/workers/index.rs index 0cac716..8a26b53 100644 --- a/backend/src/workers/index.rs +++ b/backend/src/workers/index.rs @@ -12,8 +12,8 @@ use uuid::Uuid; use crate::{ jobs::JOB_INDEX_DOCUMENT_TEXT, - models::{Document, DocumentAsset, DocumentVersion}, - schema::{document_assets, document_versions, documents}, + models::{Document, DocumentVersion}, + schema::{document_asset_objects, document_assets, document_versions, documents}, state::AppState, }; @@ -87,15 +87,15 @@ impl JobHandler for IndexDocumentTextJob { } }; - if context.text_asset.is_none() { + if context.text_s3_key.is_none() { warn!(job_id = %job.id, "missing OCR text asset; failing indexing job"); return JobExecution::Failed { error: "missing OCR text asset".into(), }; } - let asset = context.text_asset.unwrap(); - let text = match state.storage.get_object(&asset.s3_key).await { + let s3_key = context.text_s3_key.unwrap(); + let text = match state.storage.get_object(&s3_key).await { Ok(bytes) => match String::from_utf8(bytes) { Ok(text) => text, Err(err) => { @@ -169,7 +169,7 @@ impl JobHandler for IndexDocumentTextJob { struct IndexContext { document: Document, version: DocumentVersion, - text_asset: Option, + text_s3_key: Option, } fn load_context(state: Arc, payload: &IndexPayload) -> Result { @@ -189,9 +189,14 @@ fn load_context(state: Arc, payload: &IndexPayload) -> Result = document_assets::table + let text_s3_key: Option = document_asset_objects::table + .inner_join( + document_assets::table.on(document_asset_objects::asset_id.eq(document_assets::id)), + ) .filter(document_assets::document_version_id.eq(payload.document_version_id)) .filter(document_assets::asset_type.eq(OCR_TEXT_ASSET_TYPE)) + .filter(document_asset_objects::ordinal.eq(1)) + .select(document_asset_objects::s3_key) .first(&mut conn) .optional() .map_err(|err| format!("{err:?}"))?; @@ -199,6 +204,6 @@ fn load_context(state: Arc, payload: &IndexPayload) -> Result, + existing_objects: Vec, skip: bool, } @@ -218,29 +226,41 @@ fn load_ocr_context(state: Arc, payload: &OcrPayload) -> Result = document_assets::table + let existing_asset: Option = document_assets::table .filter(document_assets::document_version_id.eq(payload.document_version_id)) .filter(document_assets::asset_type.eq(OCR_TEXT_ASSET_TYPE)) .first(&mut conn) .optional() .map_err(|err| format!("{err:?}"))?; + let existing_objects: Vec = if let Some(asset) = &existing_asset { + document_asset_objects::table + .filter(document_asset_objects::asset_id.eq(asset.id)) + .order(document_asset_objects::ordinal.asc()) + .load(&mut conn) + .map_err(|err| format!("{err:?}"))? + } else { + Vec::new() + }; + let is_pdf = document_is_pdf(&document); if !is_pdf { return Ok(OcrContext { document, version, - existing_asset: existing, + existing_asset: existing_asset, + existing_objects, skip: true, }); } - let skip = existing.is_some() && !payload.force; + let skip = existing_asset.is_some() && !payload.force; Ok(OcrContext { document, version, - existing_asset: existing, + existing_asset, + existing_objects, skip, }) } @@ -371,16 +391,22 @@ fn persist_ocr_metadata( ) -> Result<(), String> { let mut conn = state.db().map_err(|err| format!("{err:?}"))?; + if let Some(existing_asset) = &context.existing_asset { + diesel::delete(document_assets::table.filter(document_assets::id.eq(existing_asset.id))) + .execute(&mut conn) + .map_err(|err| format!("{err:?}"))?; + } + let new_asset = NewDocumentAsset { id: asset_id, document_version_id: context.version.id, asset_type: OCR_TEXT_ASSET_TYPE.to_string(), - s3_key: s3_key.to_string(), mime_type: "text/plain".to_string(), metadata: json!({ "generated_at": Utc::now().to_rfc3339(), "source": source, }), + cardinality: Some(1), }; diesel::insert_into(document_assets::table) @@ -391,9 +417,41 @@ fn persist_ocr_metadata( )) .do_update() .set(( - document_assets::s3_key.eq(excluded(document_assets::s3_key)), 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)), + )) + .execute(&mut conn) + .map_err(|err| format!("{err:?}"))?; + + let existing_object_id: Option = document_asset_objects::table + .filter(document_asset_objects::asset_id.eq(asset_id)) + .filter(document_asset_objects::ordinal.eq(1)) + .select(document_asset_objects::id) + .first(&mut conn) + .optional() + .map_err(|err| format!("{err:?}"))?; + + let object_id = existing_object_id.unwrap_or_else(Uuid::new_v4); + + let new_object = NewDocumentAssetObject { + id: object_id, + asset_id, + ordinal: 1, + s3_key: s3_key.to_string(), + metadata: json!({}), + }; + + diesel::insert_into(document_asset_objects::table) + .values(&new_object) + .on_conflict(( + document_asset_objects::asset_id, + document_asset_objects::ordinal, + )) + .do_update() + .set(( + document_asset_objects::s3_key.eq(excluded(document_asset_objects::s3_key)), + document_asset_objects::metadata.eq(excluded(document_asset_objects::metadata)), )) .execute(&mut conn) .map_err(|err| format!("{err:?}"))?; diff --git a/backend/src/workers/thumbnails.rs b/backend/src/workers/thumbnails.rs index 9e98eda..04358ce 100644 --- a/backend/src/workers/thumbnails.rs +++ b/backend/src/workers/thumbnails.rs @@ -13,8 +13,11 @@ use uuid::Uuid; use crate::{ jobs::JOB_GENERATE_THUMBNAILS, - models::{Document, DocumentAsset, DocumentVersion, NewDocumentAsset}, - schema::{document_assets, document_versions, documents}, + models::{ + Document, DocumentAsset, DocumentAssetObject, DocumentVersion, NewDocumentAsset, + NewDocumentAssetObject, + }, + schema::{document_asset_objects, document_assets, document_versions, documents}, state::AppState, }; @@ -142,11 +145,33 @@ impl JobHandler for GenerateThumbnailsJob { } } - let thumbnail_asset_id = initial - .existing_thumbnail - .as_ref() - .map(|asset| asset.id) - .unwrap_or_else(Uuid::new_v4); + if initial.existing_preview.is_some() { + for object in &initial.existing_preview_objects { + if let Err(err) = state.storage.delete_object(&object.s3_key).await { + warn!( + job_id = %job.id, + error = %err, + s3_key = %object.s3_key, + "failed to delete existing preview object" + ); + } + } + } + + if initial.existing_thumbnail.is_some() { + for object in &initial.existing_thumbnail_objects { + if let Err(err) = state.storage.delete_object(&object.s3_key).await { + warn!( + job_id = %job.id, + error = %err, + s3_key = %object.s3_key, + "failed to delete existing thumbnail object" + ); + } + } + } + + let thumbnail_asset_id = Uuid::new_v4(); let thumbnail_s3_key = format!( "documents/{}/v{}/assets/{}/{}", initial.document.id, @@ -155,11 +180,7 @@ impl JobHandler for GenerateThumbnailsJob { thumbnail_asset_id ); - let preview_asset_id = initial - .existing_preview - .as_ref() - .map(|asset| asset.id) - .unwrap_or_else(Uuid::new_v4); + let preview_asset_id = Uuid::new_v4(); let preview_s3_key = format!( "documents/{}/v{}/assets/{}/{}", initial.document.id, @@ -250,7 +271,9 @@ struct ThumbnailContext { document: Document, version: DocumentVersion, existing_thumbnail: Option, + existing_thumbnail_objects: Vec, existing_preview: Option, + existing_preview_objects: Vec, skip: bool, } @@ -303,11 +326,27 @@ fn load_thumbnail_context( .map_err(|err| format!("{err:?}"))?; let mut existing_thumbnail = None; + let mut existing_thumbnail_objects: Vec = Vec::new(); let mut existing_preview = None; + let mut existing_preview_objects: Vec = Vec::new(); for asset in existing_assets { match asset.asset_type.as_str() { - THUMBNAIL_ASSET_TYPE => existing_thumbnail = Some(asset), - PREVIEW_ASSET_TYPE => existing_preview = Some(asset), + THUMBNAIL_ASSET_TYPE => { + existing_thumbnail_objects = document_asset_objects::table + .filter(document_asset_objects::asset_id.eq(asset.id)) + .order(document_asset_objects::ordinal.asc()) + .load(&mut conn) + .map_err(|err| format!("{err:?}"))?; + existing_thumbnail = Some(asset); + } + PREVIEW_ASSET_TYPE => { + existing_preview_objects = document_asset_objects::table + .filter(document_asset_objects::asset_id.eq(asset.id)) + .order(document_asset_objects::ordinal.asc()) + .load(&mut conn) + .map_err(|err| format!("{err:?}"))?; + existing_preview = Some(asset); + } _ => {} } } @@ -323,7 +362,9 @@ fn load_thumbnail_context( document, version, existing_thumbnail, + existing_thumbnail_objects, existing_preview, + existing_preview_objects, skip, }) } @@ -462,18 +503,30 @@ fn persist_assets_metadata( ) -> Result<(), String> { let mut conn = state.db().map_err(|err| format!("{err:?}"))?; + if let Some(existing_preview) = &context.existing_preview { + diesel::delete(document_assets::table.filter(document_assets::id.eq(existing_preview.id))) + .execute(&mut conn) + .map_err(|err| format!("{err:?}"))?; + } + + if let Some(existing_thumbnail) = &context.existing_thumbnail { + diesel::delete( + document_assets::table.filter(document_assets::id.eq(existing_thumbnail.id)), + ) + .execute(&mut conn) + .map_err(|err| format!("{err:?}"))?; + } + for asset in assets { let new_asset = NewDocumentAsset { id: asset.asset_id, document_version_id: context.version.id, asset_type: asset.asset_type.to_string(), - s3_key: asset.s3_key.to_string(), mime_type: "image/png".to_string(), metadata: json!({ "generated_at": Utc::now().to_rfc3339(), - "width": asset.generated.width, - "height": asset.generated.height, }), + cardinality: Some(1), }; diesel::insert_into(document_assets::table) @@ -484,9 +537,51 @@ fn persist_assets_metadata( )) .do_update() .set(( - document_assets::s3_key.eq(excluded(document_assets::s3_key)), 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)), + )) + .execute(&mut conn) + .map_err(|err| format!("{err:?}"))?; + + let existing_object_id: Option = document_asset_objects::table + .filter(document_asset_objects::asset_id.eq(asset.asset_id)) + .filter(document_asset_objects::ordinal.eq(1)) + .select(document_asset_objects::id) + .first(&mut conn) + .optional() + .map_err(|err| format!("{err:?}"))?; + + let object_id = existing_object_id.unwrap_or_else(Uuid::new_v4); + + let mut metadata_map = Map::new(); + if let Some(width) = asset.generated.width { + metadata_map.insert("width".to_string(), Value::from(width)); + } + if let Some(height) = asset.generated.height { + metadata_map.insert("height".to_string(), Value::from(height)); + } + + let object_metadata = Value::Object(metadata_map); + + let new_object = NewDocumentAssetObject { + id: object_id, + asset_id: asset.asset_id, + ordinal: 1, + s3_key: asset.s3_key.to_string(), + metadata: object_metadata, + }; + + diesel::insert_into(document_asset_objects::table) + .values(&new_object) + .on_conflict(( + document_asset_objects::asset_id, + document_asset_objects::ordinal, + )) + .do_update() + .set(( + document_asset_objects::s3_key.eq(excluded(document_asset_objects::s3_key)), + document_asset_objects::metadata.eq(excluded(document_asset_objects::metadata)), )) .execute(&mut conn) .map_err(|err| format!("{err:?}"))?;