diff --git a/backend/src/documents/asset.rs b/backend/src/documents/asset.rs new file mode 100644 index 0000000..8054213 --- /dev/null +++ b/backend/src/documents/asset.rs @@ -0,0 +1,263 @@ +use std::collections::HashMap; +use std::path::Path as FsPath; + +use diesel::prelude::*; +use serde::Serialize; +use serde_json::Value; +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::state::AppState; +use crate::utils::time::to_iso; + +#[derive(Serialize, Clone, ToSchema)] +pub struct DocumentAssetResponse { + pub id: Uuid, + pub asset_type: String, + pub mime_type: String, + pub metadata: Value, + #[serde(skip_serializing_if = "Option::is_none")] + pub cardinality: Option, +} + +#[derive(Serialize, Clone, ToSchema)] +pub struct DocumentAssetObjectResponse { + pub id: Uuid, + pub ordinal: i32, + pub metadata: Value, + #[serde(skip_serializing_if = "Option::is_none")] + pub url: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub expires_at: Option, +} + +#[derive(Serialize, ToSchema)] +pub struct DocumentAssetDetailResponse { + pub id: Uuid, + pub asset_type: String, + pub mime_type: String, + pub metadata: Value, + pub created_at: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub cardinality: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub objects: Vec, +} + +#[derive(Serialize, Clone, ToSchema)] +pub struct DocumentVersionResponse { + pub id: Uuid, + pub version_number: i32, + pub size_bytes: i64, + pub checksum: String, + pub created_at: String, + pub metadata: Value, +} + +#[derive(Serialize, Clone, ToSchema)] +pub struct DocumentVersionDetailResponse { + #[serde(flatten)] + pub version: DocumentVersionResponse, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub assets: Vec, + pub download_path: String, +} + +pub fn build_download_path( + state: &AppState, + document: &Document, + user_id: Uuid, +) -> AppResult { + state + .jwt + .generate_download_token(document.id, user_id, document.tenant_id) + .map(|token| format!("/download/{token}")) + .map_err(|err| AppError::internal(format!("failed to generate download token: {err}"))) +} + +pub fn to_version_response(version: DocumentVersion) -> DocumentVersionResponse { + DocumentVersionResponse { + id: version.id, + version_number: version.version_number, + size_bytes: version.size_bytes, + checksum: version.checksum, + created_at: to_iso(version.created_at), + metadata: version.metadata, + } +} + +pub fn to_asset_summary(asset: DocumentAsset) -> DocumentAssetResponse { + DocumentAssetResponse { + id: asset.id, + asset_type: asset.asset_type, + mime_type: asset.mime_type, + metadata: asset.metadata, + cardinality: asset.cardinality, + } +} + +pub fn to_asset_detail_response( + asset: DocumentAsset, + objects: Vec, +) -> DocumentAssetDetailResponse { + DocumentAssetDetailResponse { + id: asset.id, + asset_type: asset.asset_type, + 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 async fn load_asset_responses( + state: &AppState, + tenant_id: Uuid, + version_id: Uuid, +) -> AppResult> { + let mut conn = state.db_for_tenant(tenant_id)?; + 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)) + .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(&mut conn)?; + drop(conn); + + Ok(assets + .into_iter() + .map(|(asset, _)| to_asset_summary(asset)) + .collect()) +} + +pub fn load_primary_assets( + state: &AppState, + tenant_id: Uuid, + documents: &[Document], +) -> AppResult)>> { + if documents.is_empty() { + return Ok(HashMap::new()); + } + + let mut doc_to_version: HashMap = HashMap::with_capacity(documents.len()); + let mut version_ids: Vec = Vec::with_capacity(documents.len()); + for doc in documents { + doc_to_version.insert(doc.id, doc.current_version_id); + version_ids.push(doc.current_version_id); + } + + version_ids.sort(); + version_ids.dedup(); + + let mut conn = state.db_for_tenant(tenant_id)?; + let versions: Vec = document_versions::table + .filter(document_versions::id.eq_any(&version_ids)) + .load(&mut conn)?; + + let mut version_map: HashMap = HashMap::new(); + for version in versions { + 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))), + ) + .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)?; + + drop(conn); + + let mut assets_by_version: HashMap> = HashMap::new(); + for (asset, _object) in assets { + let version_id = asset.document_version_id; + let response = to_asset_summary(asset); + assets_by_version + .entry(version_id) + .or_default() + .push(response); + } + + let mut result: HashMap)> = + HashMap::with_capacity(doc_to_version.len()); + for (doc_id, version_id) in doc_to_version { + if let Some(version) = version_map.remove(&version_id) { + let assets = assets_by_version.remove(&version_id).unwrap_or_default(); + result.insert(doc_id, (to_version_response(version), assets)); + } + } + + Ok(result) +} + +pub fn derive_document_title(original: &str) -> String { + let trimmed = original.trim(); + if trimmed.is_empty() { + return "Document".to_string(); + } + + let stem = FsPath::new(trimmed) + .file_stem() + .and_then(|s| s.to_str()) + .map(|s| s.trim()) + .filter(|s| !s.is_empty()) + .map(|s| s.to_string()); + + stem.unwrap_or_else(|| trimmed.to_string()) +} + +pub fn filename_with_retained_extension(title: &str, current_filename: &str) -> String { + let extension = FsPath::new(current_filename) + .extension() + .and_then(|ext| ext.to_str()); + + if let Some(ext) = extension { + if title + .rsplit_once('.') + .map(|(_, existing_ext)| existing_ext.eq_ignore_ascii_case(ext)) + .unwrap_or(false) + { + title.to_string() + } else { + format!("{title}.{ext}") + } + } else { + title.to_string() + } +} diff --git a/backend/src/documents/correspondents.rs b/backend/src/documents/correspondents.rs new file mode 100644 index 0000000..93ee604 --- /dev/null +++ b/backend/src/documents/correspondents.rs @@ -0,0 +1,120 @@ +use std::collections::HashMap; + +use chrono::Utc; +use diesel::prelude::*; +use serde::Serialize; +use serde_json::Value; +use utoipa::ToSchema; +use uuid::Uuid; + +use crate::error::{AppError, AppResult}; +use crate::models::{Correspondent, DocumentCorrespondent, NewDocumentCorrespondent}; +use crate::schema::{correspondents, document_correspondents, documents}; +use crate::utils::time::to_iso; + +#[derive(Serialize, Clone, ToSchema)] +pub struct DocumentCorrespondentResponse { + pub id: Uuid, + pub name: String, + pub metadata: Value, + pub assigned_at: String, +} + +pub fn normalize_correspondent_ids(ids: &[Uuid]) -> AppResult> { + let mut unique: Vec = ids.iter().copied().collect(); + unique.sort_unstable(); + unique.dedup(); + + if unique.is_empty() { + return Err(AppError::bad_request( + "assignments must contain at least one correspondent", + )); + } + + Ok(unique) +} + +pub fn insert_document_correspondents( + conn: &mut PgConnection, + tenant_id: Uuid, + document_id: Uuid, + user_id: Uuid, + correspondent_ids: &[Uuid], +) -> AppResult { + let ids = normalize_correspondent_ids(correspondent_ids)?; + + let existing: Vec = correspondents::table + .filter(correspondents::id.eq_any(&ids)) + .filter(correspondents::tenant_id.eq(tenant_id)) + .select(correspondents::id) + .load(conn)?; + + if existing.len() != ids.len() { + return Err(AppError::bad_request( + "one or more correspondents do not exist", + )); + } + + let new_rows: Vec = ids + .into_iter() + .map(|correspondent_id| NewDocumentCorrespondent { + document_id, + correspondent_id, + assigned_by: Some(user_id), + tenant_id, + }) + .collect(); + + if new_rows.is_empty() { + return Ok(0); + } + + let inserted = diesel::insert_into(document_correspondents::table) + .values(&new_rows) + .on_conflict_do_nothing() + .execute(conn)?; + + if inserted > 0 { + diesel::update( + documents::table + .find(document_id) + .filter(documents::tenant_id.eq(tenant_id)), + ) + .set(documents::updated_at.eq(Utc::now().naive_utc())) + .execute(conn)?; + } + + Ok(inserted) +} + +pub fn load_correspondents_for_documents( + conn: &mut PgConnection, + document_ids: &[Uuid], +) -> AppResult>> { + if document_ids.is_empty() { + return Ok(HashMap::new()); + } + + let rows: Vec<(DocumentCorrespondent, Correspondent)> = document_correspondents::table + .inner_join(correspondents::table) + .filter(document_correspondents::document_id.eq_any(document_ids)) + .order(( + document_correspondents::document_id.asc(), + document_correspondents::assigned_at.asc(), + )) + .load(conn)?; + + let mut map: HashMap> = HashMap::new(); + for (assignment, correspondent) in rows { + map.entry(assignment.document_id) + .or_default() + .push(DocumentCorrespondentResponse { + id: correspondent.id, + name: correspondent.name, + metadata: correspondent.metadata, + assigned_at: to_iso(assignment.assigned_at), + }); + } + + Ok(map) +} diff --git a/backend/src/documents/folders.rs b/backend/src/documents/folders.rs new file mode 100644 index 0000000..ee25e0c --- /dev/null +++ b/backend/src/documents/folders.rs @@ -0,0 +1,21 @@ +use diesel::dsl::exists; +use diesel::prelude::*; +use uuid::Uuid; + +use crate::error::AppResult; +use crate::schema::folders; +use crate::utils::validation::ensure_exists; + +pub fn ensure_folder_exists_on_conn( + conn: &mut PgConnection, + tenant_id: Uuid, + folder_id: Uuid, +) -> AppResult<()> { + let exists: bool = diesel::select(exists( + folders::table + .filter(folders::id.eq(folder_id)) + .filter(folders::tenant_id.eq(tenant_id)), + )) + .get_result(conn)?; + ensure_exists(exists, "folder") +} diff --git a/backend/src/documents/metadata.rs b/backend/src/documents/metadata.rs new file mode 100644 index 0000000..88e3400 --- /dev/null +++ b/backend/src/documents/metadata.rs @@ -0,0 +1,52 @@ +use serde_json::{map::Entry, Map, Value}; + +use crate::error::{AppError, AppResult}; + +pub fn merge_document_metadata(existing: Value, updates: Value) -> AppResult { + let mut base = match existing { + Value::Object(map) => map, + Value::Null => Map::new(), + _ => { + return Err(AppError::bad_request( + "existing metadata is not an object; set replace=true to overwrite", + )); + } + }; + + let incoming = match updates { + Value::Object(map) => map, + _ => { + return Err(AppError::bad_request( + "metadata value must be a JSON object when replace is false", + )); + } + }; + + merge_metadata_maps(&mut base, incoming); + Ok(Value::Object(base)) +} + +fn merge_metadata_maps(target: &mut Map, updates: Map) { + for (key, value) in updates { + match target.entry(key) { + Entry::Occupied(mut entry) => { + let existing = entry.get_mut(); + match value { + Value::Object(update_map) => { + if let Value::Object(existing_map) = existing { + merge_metadata_maps(existing_map, update_map); + } else { + *existing = Value::Object(update_map); + } + } + other => { + *existing = other; + } + } + } + Entry::Vacant(entry) => { + entry.insert(value); + } + } + } +} diff --git a/backend/src/documents/mod.rs b/backend/src/documents/mod.rs new file mode 100644 index 0000000..de99228 --- /dev/null +++ b/backend/src/documents/mod.rs @@ -0,0 +1,6 @@ +pub mod asset; +pub mod correspondents; +pub mod folders; +pub mod metadata; +pub mod search; +pub mod tags; diff --git a/backend/src/routes/documents/search_utils.rs b/backend/src/documents/search.rs similarity index 50% rename from backend/src/routes/documents/search_utils.rs rename to backend/src/documents/search.rs index 67178d8..fa89231 100644 --- a/backend/src/routes/documents/search_utils.rs +++ b/backend/src/documents/search.rs @@ -1,6 +1,14 @@ -use serde_json::Value; +use std::collections::HashSet; + +use anyhow::{anyhow, Result}; +use reqwest::Client; +use serde::Deserialize; +use serde_json::{json, Value}; +use tracing::{debug, error}; use uuid::Uuid; +pub const QUICKWIT_MAX_HITS: usize = 200; + pub fn build_quickwit_query(input: &str) -> Option { let tokens: Vec = input .split_whitespace() @@ -38,6 +46,67 @@ pub fn escape_quickwit_token(token: &str) -> String { escaped } +pub async fn quickwit_search( + endpoint: &str, + index: &str, + tenant_id: Uuid, + query: &str, +) -> Result> { + let tenant_clause = format!("tenant_id:{}", tenant_id); + let quickwit_query = match build_quickwit_query(query) { + Some(q) => { + debug!(%query, quickwit_query = %q, "built quickwit search query"); + format!("{} AND ({})", tenant_clause, q) + } + None => { + debug!(%query, "quickwit search skipped because query produced no tokens"); + return Ok(vec![]); + } + }; + + let client = Client::new(); + let url = format!("{}/api/v1/{}/search", endpoint.trim_end_matches('/'), index); + + let payload = json!({ + "query": quickwit_query, + "max_hits": QUICKWIT_MAX_HITS, + }); + + debug!(%url, payload = %payload, "sending quickwit search request"); + let response = client.post(url).json(&payload).send().await?; + if !response.status().is_success() { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + error!(%status, body = %body, "quickwit search request failed"); + return Err(anyhow!( + "quickwit search failed with status {status}: {body}" + )); + } + + let data: QuickwitSearchResponse = response.json().await?; + debug!("quickwit search response parsed successfully"); + let QuickwitSearchResponse { hits } = data; + + let total_hits = hits.len(); + let mut seen = HashSet::new(); + let mut doc_ids = Vec::with_capacity(total_hits); + + for hit in hits { + if let Some(doc_id) = extract_document_id(&hit) { + if seen.insert(doc_id) { + doc_ids.push(doc_id); + } + } + } + + debug!( + total_hits = total_hits, + unique_ids = doc_ids.len(), + "quickwit search completed" + ); + Ok(doc_ids) +} + pub fn extract_document_id(hit: &Value) -> Option { for key in ["_source", "source", "fields", "stored_fields"] { if let Some(value) = hit.get(key) { @@ -89,3 +158,9 @@ pub fn parse_uuid_value(value: &Value) -> Option { None } + +#[derive(Deserialize)] +struct QuickwitSearchResponse { + #[serde(default)] + hits: Vec, +} diff --git a/backend/src/documents/tags.rs b/backend/src/documents/tags.rs new file mode 100644 index 0000000..2a35890 --- /dev/null +++ b/backend/src/documents/tags.rs @@ -0,0 +1,80 @@ +use std::collections::HashMap; + +use diesel::prelude::*; +use uuid::Uuid; + +use crate::error::{AppError, AppResult}; +use crate::models::{Document, NewDocumentTag, Tag}; +use crate::schema::{document_tags, tags}; + +pub fn assign_tags( + conn: &mut PgConnection, + tenant_id: Uuid, + document: &Document, + raw_tag_ids: &[Uuid], + assigned_by: Option, +) -> AppResult { + if raw_tag_ids.is_empty() { + return Ok(0); + } + + let mut tag_ids: Vec = raw_tag_ids.iter().copied().collect(); + tag_ids.sort_unstable(); + tag_ids.dedup(); + + if tag_ids.is_empty() { + return Ok(0); + } + + let existing: Vec = tags::table + .filter(tags::id.eq_any(&tag_ids)) + .filter(tags::tenant_id.eq(tenant_id)) + .select(tags::id) + .load(conn)?; + + if existing.len() != tag_ids.len() { + return Err(AppError::bad_request("one or more tags do not exist")); + } + + let new_tags: Vec = tag_ids + .into_iter() + .map(|tag_id| NewDocumentTag { + document_id: document.id, + tag_id, + assigned_by, + tenant_id, + }) + .collect(); + + if new_tags.is_empty() { + return Ok(0); + } + + let inserted = diesel::insert_into(document_tags::table) + .values(&new_tags) + .on_conflict_do_nothing() + .execute(conn)?; + + Ok(inserted) +} + +pub fn load_tags_for_documents( + conn: &mut PgConnection, + document_ids: &[Uuid], +) -> AppResult>> { + if document_ids.is_empty() { + return Ok(HashMap::new()); + } + + let rows: Vec<(Uuid, Tag)> = document_tags::table + .inner_join(tags::table) + .filter(document_tags::document_id.eq_any(document_ids)) + .select((document_tags::document_id, tags::all_columns)) + .load(conn)?; + + let mut map: HashMap> = HashMap::new(); + for (doc_id, tag) in rows { + map.entry(doc_id).or_default().push(tag); + } + Ok(map) +} diff --git a/backend/src/lib.rs b/backend/src/lib.rs index 50b333f..5af55c1 100644 --- a/backend/src/lib.rs +++ b/backend/src/lib.rs @@ -1,6 +1,7 @@ pub mod auth; pub mod config; pub mod db; +pub mod documents; pub mod error; pub mod jobs; pub mod models; diff --git a/backend/src/routes/documents.rs b/backend/src/routes/documents.rs index 81bf059..8a45214 100644 --- a/backend/src/routes/documents.rs +++ b/backend/src/routes/documents.rs @@ -8,10 +8,9 @@ use axum::http::StatusCode; use axum::response::IntoResponse; use chrono::{DateTime, NaiveDateTime, Utc}; use diesel::dsl::exists; -use diesel::{prelude::*, result::DatabaseErrorKind, select, PgConnection}; -use reqwest::Client; +use diesel::{prelude::*, result::DatabaseErrorKind, select}; use serde::{Deserialize, Serialize}; -use serde_json::{json, map::Entry, Map, Value}; +use serde_json::{json, Value}; use sha2::{Digest, Sha256}; use tracing::{debug, error, info, warn}; use utoipa::{IntoParams, ToSchema}; @@ -19,16 +18,31 @@ use uuid::Uuid; use super::folders::gather_descendant_folder_ids; use crate::auth::TenantScopedConn; +use crate::documents::{ + asset::{ + build_download_path, derive_document_title, filename_with_retained_extension, + load_asset_responses, load_primary_assets, to_asset_detail_response, + to_asset_object_response, to_version_response, DocumentAssetDetailResponse, + DocumentAssetResponse, DocumentVersionDetailResponse, DocumentVersionResponse, + }, + correspondents::{ + insert_document_correspondents, load_correspondents_for_documents, + normalize_correspondent_ids, DocumentCorrespondentResponse, + }, + folders::ensure_folder_exists_on_conn, + metadata::merge_document_metadata, + search::quickwit_search, + tags::{assign_tags as assign_tags_to_document, load_tags_for_documents}, +}; use crate::error::{AppError, AppResult}; use crate::jobs::{enqueue_job, JOB_ANALYZE_DOCUMENT, JOB_INDEX_DOCUMENT_TEXT}; use crate::models::{ - Correspondent, Document, DocumentAsset, DocumentAssetObject, DocumentCorrespondent, - DocumentVersion, NewDocument, NewDocumentCorrespondent, NewDocumentTag, NewDocumentVersion, - Tag, + Document, DocumentAsset, DocumentAssetObject, DocumentVersion, NewDocument, NewDocumentTag, + NewDocumentVersion, Tag, }; use crate::schema::{ - correspondents, document_asset_objects, document_assets, document_correspondents, - document_tags, document_versions, documents, folders, refresh_tokens::dsl as refresh_dsl, tags, + document_asset_objects, document_assets, document_correspondents, document_tags, + document_versions, documents, folders, refresh_tokens::dsl as refresh_dsl, tags, }; use crate::state::AppState; use crate::utils::{ @@ -37,22 +51,9 @@ use crate::utils::{ json::{classify_nullable, NullableValue}, storage_paths::document_version_object_key, time::to_iso, - validation::ensure_exists, }; -mod asset_utils; -mod correspondent_utils; -mod search_utils; - -use asset_utils::{ - build_download_path, derive_document_title, filename_with_retained_extension, - to_asset_detail_response, to_asset_object_response, to_asset_summary, to_version_response, -}; -use correspondent_utils::normalize_correspondent_assignments; -use search_utils::{build_quickwit_query, extract_document_id}; - const PRESIGNED_URL_EXPIRY_SECONDS: u64 = 300; -const QUICKWIT_MAX_HITS: usize = 200; #[derive(Deserialize, IntoParams, ToSchema)] #[into_params(parameter_in = Query)] @@ -126,67 +127,6 @@ impl From for TagResponse { } } -#[derive(Serialize, Clone, ToSchema)] -pub struct DocumentVersionResponse { - pub id: Uuid, - pub version_number: i32, - pub size_bytes: i64, - pub checksum: String, - pub created_at: String, - pub metadata: Value, -} - -#[derive(Serialize, Clone, ToSchema)] -pub struct DocumentAssetResponse { - pub id: Uuid, - pub asset_type: String, - pub mime_type: String, - pub metadata: Value, - #[serde(skip_serializing_if = "Option::is_none")] - pub cardinality: Option, -} - -#[derive(Serialize, Clone, ToSchema)] -pub struct DocumentAssetObjectResponse { - pub id: Uuid, - pub ordinal: i32, - pub metadata: Value, - #[serde(skip_serializing_if = "Option::is_none")] - pub url: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub expires_at: Option, -} - -#[derive(Serialize, ToSchema)] -pub struct DocumentAssetDetailResponse { - pub id: Uuid, - pub asset_type: String, - pub mime_type: String, - pub metadata: Value, - pub created_at: String, - #[serde(skip_serializing_if = "Option::is_none")] - pub cardinality: Option, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub objects: Vec, -} - -#[derive(Serialize, Clone, ToSchema)] -pub struct DocumentVersionDetailResponse { - #[serde(flatten)] - pub version: DocumentVersionResponse, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub assets: Vec, - pub download_path: String, -} - -#[derive(Serialize, Clone, ToSchema)] -pub struct DocumentCorrespondentResponse { - pub id: Uuid, - pub name: String, - pub metadata: Value, - pub assigned_at: String, -} - #[derive(Serialize, ToSchema)] pub struct DocumentResponse { pub id: Uuid, @@ -626,7 +566,7 @@ pub async fn list_documents( let mut correspondents_map = load_correspondents_for_documents(&mut conn, &doc_ids)?; drop(conn); - let primary_versions = load_primary_assets(&state, tenant_id, &docs).await?; + let primary_versions = load_primary_assets(&state, tenant_id, &docs)?; let mut response = Vec::with_capacity(doc_ids.len()); for doc in docs { let tags = tags_map.get(&doc.id).cloned(); @@ -1487,55 +1427,6 @@ pub async fn restore_document( Ok(StatusCode::NO_CONTENT) } -fn merge_document_metadata(existing: Value, updates: Value) -> AppResult { - let mut base = match existing { - Value::Object(map) => map, - Value::Null => Map::new(), - _ => { - return Err(AppError::bad_request( - "existing metadata is not an object; set replace=true to overwrite", - )); - } - }; - - let incoming = match updates { - Value::Object(map) => map, - _ => { - return Err(AppError::bad_request( - "metadata value must be a JSON object when replace is false", - )); - } - }; - - merge_metadata_maps(&mut base, incoming); - Ok(Value::Object(base)) -} - -fn merge_metadata_maps(target: &mut Map, updates: Map) { - for (key, value) in updates { - match target.entry(key) { - Entry::Occupied(mut entry) => { - let existing = entry.get_mut(); - match value { - Value::Object(update_map) => { - if let Value::Object(existing_map) = existing { - merge_metadata_maps(existing_map, update_map); - } else { - *existing = Value::Object(update_map); - } - } - other => { - *existing = other; - } - } - } - Entry::Vacant(entry) => { - entry.insert(value); - } - } - } -} - pub async fn move_document( Path(document_id): Path, TenantScopedConn { @@ -1661,7 +1552,12 @@ pub async fn assign_correspondents( return Err(AppError::bad_request("assignments must not be empty")); } - let correspondent_ids = normalize_correspondent_assignments(&payload.assignments)?; + let raw_correspondent_ids: Vec = payload + .assignments + .iter() + .map(|assignment| assignment.correspondent_id) + .collect(); + let correspondent_ids = normalize_correspondent_ids(&raw_correspondent_ids)?; let replace = payload.replace; conn.transaction::<(), AppError, _>(|conn| { @@ -1739,7 +1635,12 @@ pub async fn bulk_assign_correspondents( let mut document_ids = payload.document_ids; validate_bulk_ids(&mut document_ids, "document_ids")?; - let correspondent_ids = normalize_correspondent_assignments(&payload.assignments)?; + let raw_correspondent_ids: Vec = payload + .assignments + .iter() + .map(|assignment| assignment.correspondent_id) + .collect(); + let correspondent_ids = normalize_correspondent_ids(&raw_correspondent_ids)?; let action = payload.action; let user_id_val = user_id; let (assigned, removed) = conn.transaction::<(usize, usize), AppError, _>(|conn| { @@ -1761,18 +1662,6 @@ pub async fn bulk_assign_correspondents( )); } - if !correspondent_ids.is_empty() { - let existing: Vec = correspondents::table - .filter(correspondents::id.eq_any(&correspondent_ids)) - .filter(correspondents::tenant_id.eq(tenant_id)) - .load(conn)?; - if existing.len() != correspondent_ids.len() { - return Err(AppError::bad_request( - "one or more correspondents do not exist", - )); - } - } - match action { BulkCorrespondentAction::Add => { let mut assigned_total = 0; @@ -1880,7 +1769,7 @@ pub async fn assign_tags( .filter(documents::tenant_id.eq(tenant_id)) .first(&mut conn)?; - assign_tags_internal( + assign_tags_to_document( &mut conn, tenant_id, &document, @@ -2077,15 +1966,22 @@ async fn process_upload( } } - assign_tags_internal(&mut conn, tenant_id, &document, &tag_ids, Some(user_id))?; + assign_tags_to_document(&mut conn, tenant_id, &document, &tag_ids, Some(user_id))?; - assign_correspondents_internal( - &mut conn, - tenant_id, - &document, - user_id, - &correspondents, - )?; + if !correspondents.is_empty() { + let raw_ids: Vec = correspondents + .iter() + .map(|assignment| assignment.correspondent_id) + .collect(); + let correspondent_ids = normalize_correspondent_ids(&raw_ids)?; + insert_document_correspondents( + &mut conn, + tenant_id, + document.id, + user_id, + &correspondent_ids, + )?; + } if document.deleted_at.is_some() { let now = Utc::now().naive_utc(); @@ -2205,9 +2101,22 @@ async fn process_upload( let detail = { let mut conn = state.db_for_tenant(tenant_id)?; - assign_tags_internal(&mut conn, tenant_id, &document, &tag_ids, Some(user_id))?; + assign_tags_to_document(&mut conn, tenant_id, &document, &tag_ids, Some(user_id))?; - assign_correspondents_internal(&mut conn, tenant_id, &document, user_id, &correspondents)?; + if !correspondents.is_empty() { + let raw_ids: Vec = correspondents + .iter() + .map(|assignment| assignment.correspondent_id) + .collect(); + let correspondent_ids = normalize_correspondent_ids(&raw_ids)?; + insert_document_correspondents( + &mut conn, + tenant_id, + document.id, + user_id, + &correspondent_ids, + )?; + } let tags_map = load_tags_for_documents(&mut conn, &[doc_id])?; let mut correspondents_map = load_correspondents_for_documents(&mut conn, &[doc_id])?; @@ -2248,271 +2157,6 @@ async fn process_upload( Ok(UploadOutcome::Created(detail)) } -fn assign_tags_internal( - conn: &mut PgConnection, - tenant_id: Uuid, - document: &Document, - raw_tag_ids: &[Uuid], - assigned_by: Option, -) -> AppResult { - if raw_tag_ids.is_empty() { - return Ok(0); - } - - let mut tag_ids: Vec = raw_tag_ids.iter().copied().collect(); - tag_ids.sort_unstable(); - tag_ids.dedup(); - - if tag_ids.is_empty() { - return Ok(0); - } - - let existing: Vec = tags::table - .filter(tags::id.eq_any(&tag_ids)) - .filter(tags::tenant_id.eq(tenant_id)) - .select(tags::id) - .load(conn)?; - - if existing.len() != tag_ids.len() { - return Err(AppError::bad_request("one or more tags do not exist")); - } - - let new_tags: Vec = tag_ids - .into_iter() - .map(|tag_id| NewDocumentTag { - document_id: document.id, - tag_id, - assigned_by, - tenant_id, - }) - .collect(); - - if new_tags.is_empty() { - return Ok(0); - } - - let inserted = diesel::insert_into(document_tags::table) - .values(&new_tags) - .on_conflict_do_nothing() - .execute(conn)?; - - Ok(inserted) -} - -fn assign_correspondents_internal( - conn: &mut PgConnection, - tenant_id: Uuid, - document: &Document, - user_id: Uuid, - assignments: &[CorrespondentAssignmentInput], -) -> AppResult { - if assignments.is_empty() { - return Ok(0); - } - - let ids = normalize_correspondent_assignments(assignments)?; - - insert_document_correspondents(conn, tenant_id, document.id, user_id, &ids) -} - -fn insert_document_correspondents( - conn: &mut PgConnection, - tenant_id: Uuid, - document_id: Uuid, - user_id: Uuid, - correspondent_ids: &[Uuid], -) -> AppResult { - if correspondent_ids.is_empty() { - return Ok(0); - } - - let mut unique_ids: Vec = correspondent_ids.to_vec(); - unique_ids.sort_unstable(); - unique_ids.dedup(); - - if !unique_ids.is_empty() { - let existing: Vec = correspondents::table - .filter(correspondents::id.eq_any(&unique_ids)) - .filter(correspondents::tenant_id.eq(tenant_id)) - .select(correspondents::id) - .load(conn)?; - - if existing.len() != unique_ids.len() { - return Err(AppError::bad_request( - "one or more correspondents do not exist", - )); - } - } - - let new_rows: Vec = unique_ids - .into_iter() - .map(|correspondent_id| NewDocumentCorrespondent { - document_id, - correspondent_id, - assigned_by: Some(user_id), - tenant_id, - }) - .collect(); - - if new_rows.is_empty() { - return Ok(0); - } - - let inserted = diesel::insert_into(document_correspondents::table) - .values(&new_rows) - .on_conflict_do_nothing() - .execute(conn)?; - - if inserted > 0 { - diesel::update( - documents::table - .find(document_id) - .filter(documents::tenant_id.eq(tenant_id)), - ) - .set(documents::updated_at.eq(Utc::now().naive_utc())) - .execute(conn)?; - } - - Ok(inserted) -} - -fn ensure_folder_exists_on_conn( - conn: &mut PgConnection, - tenant_id: Uuid, - folder_id: Uuid, -) -> AppResult<()> { - let exists: bool = diesel::select(exists( - folders::table - .filter(folders::id.eq(folder_id)) - .filter(folders::tenant_id.eq(tenant_id)), - )) - .get_result(conn)?; - ensure_exists(exists, "folder") -} - -pub(crate) fn load_tags_for_documents( - conn: &mut PgConnection, - document_ids: &[Uuid], -) -> AppResult>> { - if document_ids.is_empty() { - return Ok(HashMap::new()); - } - - let rows: Vec<(Uuid, Tag)> = document_tags::table - .inner_join(tags::table) - .filter(document_tags::document_id.eq_any(document_ids)) - .select((document_tags::document_id, tags::all_columns)) - .load(conn)?; - - let mut map: HashMap> = HashMap::new(); - for (doc_id, tag) in rows { - map.entry(doc_id).or_default().push(tag); - } - Ok(map) -} - -pub(crate) fn load_correspondents_for_documents( - conn: &mut PgConnection, - document_ids: &[Uuid], -) -> AppResult>> { - if document_ids.is_empty() { - return Ok(HashMap::new()); - } - - let rows: Vec<(DocumentCorrespondent, Correspondent)> = document_correspondents::table - .inner_join(correspondents::table) - .filter(document_correspondents::document_id.eq_any(document_ids)) - .order(( - document_correspondents::document_id.asc(), - document_correspondents::assigned_at.asc(), - )) - .load(conn)?; - - let mut map: HashMap> = HashMap::new(); - for (assignment, correspondent) in rows { - map.entry(assignment.document_id) - .or_default() - .push(DocumentCorrespondentResponse { - id: correspondent.id, - name: correspondent.name, - metadata: correspondent.metadata, - assigned_at: to_iso(assignment.assigned_at), - }); - } - - Ok(map) -} - -pub(crate) async fn load_primary_assets( - state: &AppState, - tenant_id: Uuid, - documents: &[Document], -) -> AppResult)>> { - if documents.is_empty() { - return Ok(HashMap::new()); - } - - let mut doc_to_version: HashMap = HashMap::with_capacity(documents.len()); - let mut version_ids: Vec = Vec::with_capacity(documents.len()); - for doc in documents { - doc_to_version.insert(doc.id, doc.current_version_id); - version_ids.push(doc.current_version_id); - } - - version_ids.sort(); - version_ids.dedup(); - - let mut conn = state.db_for_tenant(tenant_id)?; - let versions: Vec = document_versions::table - .filter(document_versions::id.eq_any(&version_ids)) - .load(&mut conn)?; - - let mut version_map: HashMap = HashMap::new(); - for version in versions { - 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))), - ) - .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, _object) in assets { - let version_id = asset.document_version_id; - let response = to_asset_summary(asset); - assets_by_version - .entry(version_id) - .or_default() - .push(response); - } - - drop(conn); - - let mut result: HashMap)> = - HashMap::with_capacity(doc_to_version.len()); - for (doc_id, version_id) in doc_to_version { - if let Some(version) = version_map.remove(&version_id) { - let assets = assets_by_version.remove(&version_id).unwrap_or_default(); - result.insert(doc_id, (to_version_response(version), assets)); - } - } - - Ok(result) -} - pub(crate) fn to_document_response( state: &AppState, user_id: Uuid, @@ -2553,97 +2197,3 @@ pub(crate) fn to_document_response( current_version, }) } - -async fn load_asset_responses( - state: &AppState, - tenant_id: Uuid, - version_id: Uuid, -) -> AppResult> { - let mut conn = state.db_for_tenant(tenant_id)?; - 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)) - .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(&mut conn)?; - drop(conn); - - Ok(assets - .into_iter() - .map(|(asset, _object)| to_asset_summary(asset)) - .collect()) -} - -async fn quickwit_search( - endpoint: &str, - index: &str, - tenant_id: Uuid, - query: &str, -) -> anyhow::Result> { - let tenant_clause = format!("tenant_id:{}", tenant_id); - let quickwit_query = match build_quickwit_query(query) { - Some(q) => { - debug!(%query, quickwit_query = %q, "built quickwit search query"); - format!("{} AND ({})", tenant_clause, q) - } - None => { - debug!(%query, "quickwit search skipped because query produced no tokens"); - return Ok(vec![]); - } - }; - - let client = Client::new(); - let url = format!("{}/api/v1/{}/search", endpoint.trim_end_matches('/'), index); - - let payload = json!({ - "query": quickwit_query, - "max_hits": QUICKWIT_MAX_HITS, - }); - - debug!(%url, payload = %payload, "sending quickwit search request"); - let response = client.post(url).json(&payload).send().await?; - if !response.status().is_success() { - let status = response.status(); - let body = response.text().await.unwrap_or_default(); - error!(%status, body = %body, "quickwit search request failed"); - return Err(anyhow::anyhow!( - "quickwit search failed with status {status}: {body}" - )); - } - - let data: QuickwitSearchResponse = response.json().await?; - debug!("quickwit search response parsed successfully"); - let QuickwitSearchResponse { hits } = data; - let mut seen = HashSet::new(); - let mut doc_ids = Vec::new(); - let total_hits = hits.len(); - - for hit in hits { - if let Some(doc_id) = extract_document_id(&hit) { - if seen.insert(doc_id) { - doc_ids.push(doc_id); - } - } - } - - debug!( - total_hits = total_hits, - unique_ids = doc_ids.len(), - "quickwit search completed" - ); - Ok(doc_ids) -} - -#[derive(Deserialize)] -struct QuickwitSearchResponse { - #[serde(default)] - hits: Vec, -} diff --git a/backend/src/routes/documents/asset_utils.rs b/backend/src/routes/documents/asset_utils.rs deleted file mode 100644 index cb6b018..0000000 --- a/backend/src/routes/documents/asset_utils.rs +++ /dev/null @@ -1,111 +0,0 @@ -use std::path::Path as FsPath; - -use uuid::Uuid; - -use crate::error::{AppError, AppResult}; -use crate::models::{Document, DocumentAsset, DocumentAssetObject, DocumentVersion}; -use crate::state::AppState; -use crate::utils::time::to_iso; - -use super::{ - DocumentAssetDetailResponse, DocumentAssetObjectResponse, DocumentAssetResponse, - DocumentVersionResponse, -}; - -pub fn build_download_path( - state: &AppState, - document: &Document, - user_id: Uuid, -) -> AppResult { - state - .jwt - .generate_download_token(document.id, user_id, document.tenant_id) - .map(|token| format!("/download/{token}")) - .map_err(|err| AppError::internal(format!("failed to generate download token: {err}"))) -} - -pub fn to_version_response(version: DocumentVersion) -> DocumentVersionResponse { - DocumentVersionResponse { - id: version.id, - version_number: version.version_number, - size_bytes: version.size_bytes, - checksum: version.checksum, - created_at: to_iso(version.created_at), - metadata: version.metadata, - } -} - -pub fn to_asset_summary(asset: DocumentAsset) -> DocumentAssetResponse { - DocumentAssetResponse { - id: asset.id, - asset_type: asset.asset_type, - mime_type: asset.mime_type, - metadata: asset.metadata, - cardinality: asset.cardinality, - } -} - -pub fn to_asset_detail_response( - asset: DocumentAsset, - objects: Vec, -) -> DocumentAssetDetailResponse { - DocumentAssetDetailResponse { - id: asset.id, - asset_type: asset.asset_type, - 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 derive_document_title(original: &str) -> String { - let trimmed = original.trim(); - if trimmed.is_empty() { - return "Document".to_string(); - } - - let stem = FsPath::new(trimmed) - .file_stem() - .and_then(|s| s.to_str()) - .map(|s| s.trim()) - .filter(|s| !s.is_empty()) - .map(|s| s.to_string()); - - stem.unwrap_or_else(|| trimmed.to_string()) -} - -pub fn filename_with_retained_extension(title: &str, current_filename: &str) -> String { - let extension = FsPath::new(current_filename) - .extension() - .and_then(|ext| ext.to_str()); - - if let Some(ext) = extension { - if title - .rsplit_once('.') - .map(|(_, existing_ext)| existing_ext.eq_ignore_ascii_case(ext)) - .unwrap_or(false) - { - title.to_string() - } else { - format!("{title}.{ext}") - } - } else { - title.to_string() - } -} diff --git a/backend/src/routes/documents/correspondent_utils.rs b/backend/src/routes/documents/correspondent_utils.rs deleted file mode 100644 index 93f5615..0000000 --- a/backend/src/routes/documents/correspondent_utils.rs +++ /dev/null @@ -1,28 +0,0 @@ -use std::collections::HashSet; - -use uuid::Uuid; - -use crate::error::{AppError, AppResult}; - -use super::CorrespondentAssignmentInput; - -pub fn normalize_correspondent_assignments( - assignments: &[CorrespondentAssignmentInput], -) -> AppResult> { - let mut unique_ids = HashSet::new(); - let mut result = Vec::new(); - - for assignment in assignments { - if unique_ids.insert(assignment.correspondent_id) { - result.push(assignment.correspondent_id); - } - } - - if result.is_empty() { - return Err(AppError::bad_request( - "assignments must contain at least one correspondent", - )); - } - - Ok(result) -} diff --git a/backend/src/routes/folders.rs b/backend/src/routes/folders.rs index 688ebff..6ee3225 100644 --- a/backend/src/routes/folders.rs +++ b/backend/src/routes/folders.rs @@ -15,10 +15,12 @@ use crate::{ error::{AppError, AppResult}, }; -use super::documents::{ - load_correspondents_for_documents, load_primary_assets, load_tags_for_documents, - to_document_response, DocumentResponse, +use crate::documents::{ + asset::load_primary_assets, + correspondents::load_correspondents_for_documents, + tags::load_tags_for_documents, }; +use super::documents::{to_document_response, DocumentResponse}; use crate::utils::{ json::{classify_nullable, NullableValue}, time::to_iso, @@ -309,7 +311,7 @@ pub async fn list_folder_contents( let mut correspondents_map = load_correspondents_for_documents(&mut conn, &doc_ids)?; drop(conn); - let primary_versions = load_primary_assets(&state, tenant_id, &docs).await?; + let primary_versions = load_primary_assets(&state, tenant_id, &docs)?; let mut documents = Vec::with_capacity(doc_ids.len()); for doc in docs {