diff --git a/backend/src/config.rs b/backend/src/config.rs index 4e63ed9..37c73f7 100644 --- a/backend/src/config.rs +++ b/backend/src/config.rs @@ -133,8 +133,12 @@ impl AppConfig { fn redact_database_url(raw: &str) -> String { match Url::parse(raw) { Ok(mut parsed) => { - let _ = parsed.set_password(Some("*****")); - parsed.to_string() + if parsed.password().is_some() { + let _ = parsed.set_password(Some("*****")); + parsed.to_string() + } else { + raw.to_string() + } } Err(_) => "***".to_string(), } diff --git a/backend/src/routes/documents.rs b/backend/src/routes/documents.rs index 5ff178a..356854c 100644 --- a/backend/src/routes/documents.rs +++ b/backend/src/routes/documents.rs @@ -1,6 +1,5 @@ use std::{ collections::{HashMap, HashSet}, - path::Path as FsPath, time::Duration, }; @@ -39,17 +38,22 @@ use crate::utils::{ 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::{ + is_valid_correspondent_role, normalize_correspondent_assignments, normalize_role, + CORRESPONDENT_ROLES, +}; +use search_utils::{build_quickwit_query, extract_document_id}; + const PRESIGNED_URL_EXPIRY_SECONDS: u64 = 300; const QUICKWIT_MAX_HITS: usize = 200; -pub const CORRESPONDENT_ROLES: &[&str] = &["sender", "receiver", "other"]; - -fn normalize_role(value: &str) -> String { - value.trim().to_lowercase() -} - -fn is_valid_correspondent_role(role: &str) -> bool { - CORRESPONDENT_ROLES.iter().any(|allowed| *allowed == role) -} #[derive(Deserialize)] pub struct DocumentListQuery { @@ -263,50 +267,6 @@ pub struct BulkCorrespondentsRequest { pub action: BulkCorrespondentAction, } -fn normalize_correspondent_assignments( - assignments: &[CorrespondentAssignmentInput], -) -> AppResult<(Vec<(Uuid, String)>, Vec, Vec)> { - let mut unique_pairs: HashSet<(Uuid, String)> = HashSet::new(); - let mut normalized_pairs: Vec<(Uuid, String)> = Vec::new(); - let mut role_set: HashSet = HashSet::new(); - let mut correspondent_ids: HashSet = HashSet::new(); - - for assignment in assignments { - let role = normalize_role(&assignment.role); - if role.is_empty() { - return Err(AppError::bad_request("role must not be empty")); - } - if !is_valid_correspondent_role(&role) { - return Err(AppError::bad_request(format!( - "invalid correspondent role '{role}'. Allowed roles: {}", - CORRESPONDENT_ROLES.join(", ") - ))); - } - - if !unique_pairs.insert((assignment.correspondent_id, role.clone())) { - continue; - } - - normalized_pairs.push((assignment.correspondent_id, role.clone())); - role_set.insert(role); - correspondent_ids.insert(assignment.correspondent_id); - } - - if normalized_pairs.is_empty() { - return Err(AppError::bad_request( - "assignments must contain at least one unique correspondent/role pair", - )); - } - - let mut correspondents_vec: Vec = correspondent_ids.into_iter().collect(); - correspondents_vec.sort(); - - let mut roles_vec: Vec = role_set.into_iter().collect(); - roles_vec.sort(); - - Ok((normalized_pairs, correspondents_vec, roles_vec)) -} - #[derive(Deserialize)] pub struct CorrespondentRoleQuery { pub role: String, @@ -1330,7 +1290,7 @@ pub async fn bulk_assign_correspondents( let mut document_ids = payload.document_ids; validate_bulk_ids(&mut document_ids, "document_ids")?; - let (normalized_pairs, correspondents_vec, roles_vec) = + let (normalized_pairs, correspondents_vec, _roles_vec) = normalize_correspondent_assignments(&payload.assignments)?; let action = payload.action; let user_id_val = user_id; @@ -1367,15 +1327,33 @@ pub async fn bulk_assign_correspondents( match action { BulkCorrespondentAction::Add => { + use diesel::dsl::not; + + let mut grouped_by_role: HashMap> = HashMap::new(); + for (correspondent_id, role) in &normalized_pairs { + grouped_by_role + .entry(role.clone()) + .or_default() + .push(*correspondent_id); + } + let mut removed = 0; - if !roles_vec.is_empty() { - removed = diesel::delete( + for (role, ids) in grouped_by_role.iter() { + if ids.is_empty() { + continue; + } + let maintained_ids = ids.clone(); + let deleted = diesel::delete( document_correspondents::table .filter(document_correspondents::document_id.eq_any(&document_ids)) .filter(document_correspondents::tenant_id.eq(tenant_id)) - .filter(document_correspondents::role.eq_any(&roles_vec)), + .filter(document_correspondents::role.eq(role.as_str())) + .filter(not( + document_correspondents::correspondent_id.eq_any(maintained_ids) + )), ) .execute(conn)?; + removed += deleted; } let mut new_rows = Vec::with_capacity(document_ids.len() * normalized_pairs.len()); @@ -2021,109 +1999,6 @@ pub(crate) fn to_document_response( }) } -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}"))) -} - -fn to_version_response( - version: DocumentVersion, - include_operations_summary: bool, -) -> DocumentVersionResponse { - DocumentVersionResponse { - id: version.id, - version_number: version.version_number, - s3_key: version.s3_key, - size_bytes: version.size_bytes, - checksum: version.checksum, - created_at: to_iso(version.created_at), - metadata: version.metadata, - operations_summary: if include_operations_summary { - Some(version.operations_summary) - } else { - None - }, - } -} - -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, - } -} - -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, - } -} - -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, - } -} - -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()) -} - -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() - } -} - async fn load_asset_responses( state: &AppState, tenant_id: Uuid, @@ -2212,97 +2087,8 @@ async fn quickwit_search( Ok(doc_ids) } -fn build_quickwit_query(input: &str) -> Option { - let tokens: Vec = input - .split_whitespace() - .filter(|token| !token.is_empty()) - .map(|token| { - let normalized = token.to_lowercase(); - escape_quickwit_token(&normalized) - }) - .collect(); - - if tokens.is_empty() { - return None; - } - - let parts: Vec = tokens - .into_iter() - .map(|token| format!("(title:{token} OR text:{token})")) - .collect(); - - Some(parts.join(" AND ")) -} - -fn escape_quickwit_token(token: &str) -> String { - let mut escaped = String::with_capacity(token.len()); - for ch in token.chars() { - match ch { - '+' | '-' | '&' | '|' | '!' | '(' | ')' | '{' | '}' | '[' | ']' | '^' | '"' | '~' - | '*' | '?' | ':' | '\\' | '/' => { - escaped.push('\\'); - escaped.push(ch); - } - _ => escaped.push(ch), - } - } - escaped -} - #[derive(Deserialize)] struct QuickwitSearchResponse { #[serde(default)] hits: Vec, } - -fn extract_document_id(hit: &Value) -> Option { - for key in ["_source", "source", "fields", "stored_fields"] { - if let Some(value) = hit.get(key) { - if let Some(uuid) = extract_uuid_from_value(value) { - return Some(uuid); - } - } - } - - if let Some(value) = hit.get("document_id") { - if let Some(uuid) = extract_uuid_from_value(value) { - return Some(uuid); - } - } - - None -} - -fn extract_uuid_from_value(value: &Value) -> Option { - if let Some(obj) = value.as_object() { - if let Some(inner) = obj.get("document_id") { - return parse_uuid_value(inner); - } - } - - if let Some(arr) = value.as_array() { - for item in arr { - if let Some(uuid) = extract_uuid_from_value(item) { - return Some(uuid); - } - } - } - - parse_uuid_value(value) -} - -fn parse_uuid_value(value: &Value) -> Option { - if let Some(s) = value.as_str() { - return Uuid::parse_str(s).ok(); - } - - if let Some(arr) = value.as_array() { - for item in arr { - if let Some(uuid) = parse_uuid_value(item) { - return Some(uuid); - } - } - } - - None -} diff --git a/backend/src/routes/documents/asset_utils.rs b/backend/src/routes/documents/asset_utils.rs new file mode 100644 index 0000000..57d48ac --- /dev/null +++ b/backend/src/routes/documents/asset_utils.rs @@ -0,0 +1,120 @@ +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, + include_operations_summary: bool, +) -> DocumentVersionResponse { + DocumentVersionResponse { + id: version.id, + version_number: version.version_number, + s3_key: version.s3_key, + size_bytes: version.size_bytes, + checksum: version.checksum, + created_at: to_iso(version.created_at), + metadata: version.metadata, + operations_summary: if include_operations_summary { + Some(version.operations_summary) + } else { + None + }, + } +} + +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 new file mode 100644 index 0000000..2d3a7d5 --- /dev/null +++ b/backend/src/routes/documents/correspondent_utils.rs @@ -0,0 +1,61 @@ +use std::collections::HashSet; + +use uuid::Uuid; + +use crate::error::{AppError, AppResult}; + +use super::CorrespondentAssignmentInput; + +pub const CORRESPONDENT_ROLES: &[&str] = &["sender", "receiver", "other"]; + +pub fn normalize_role(value: &str) -> String { + value.trim().to_lowercase() +} + +pub fn is_valid_correspondent_role(role: &str) -> bool { + CORRESPONDENT_ROLES.iter().any(|allowed| *allowed == role) +} + +pub fn normalize_correspondent_assignments( + assignments: &[CorrespondentAssignmentInput], +) -> AppResult<(Vec<(Uuid, String)>, Vec, Vec)> { + let mut unique_pairs: HashSet<(Uuid, String)> = HashSet::new(); + let mut normalized_pairs: Vec<(Uuid, String)> = Vec::new(); + let mut role_set: HashSet = HashSet::new(); + let mut correspondent_ids: HashSet = HashSet::new(); + + for assignment in assignments { + let role = normalize_role(&assignment.role); + if role.is_empty() { + return Err(AppError::bad_request("role must not be empty")); + } + if !is_valid_correspondent_role(&role) { + return Err(AppError::bad_request(format!( + "invalid correspondent role '{role}'. Allowed roles: {}", + CORRESPONDENT_ROLES.join(", ") + ))); + } + + if !unique_pairs.insert((assignment.correspondent_id, role.clone())) { + continue; + } + + normalized_pairs.push((assignment.correspondent_id, role.clone())); + role_set.insert(role); + correspondent_ids.insert(assignment.correspondent_id); + } + + if normalized_pairs.is_empty() { + return Err(AppError::bad_request( + "assignments must contain at least one unique correspondent/role pair", + )); + } + + let mut correspondents_vec: Vec = correspondent_ids.into_iter().collect(); + correspondents_vec.sort(); + + let mut roles_vec: Vec = role_set.into_iter().collect(); + roles_vec.sort(); + + Ok((normalized_pairs, correspondents_vec, roles_vec)) +} diff --git a/backend/src/routes/documents/search_utils.rs b/backend/src/routes/documents/search_utils.rs new file mode 100644 index 0000000..67178d8 --- /dev/null +++ b/backend/src/routes/documents/search_utils.rs @@ -0,0 +1,91 @@ +use serde_json::Value; +use uuid::Uuid; + +pub fn build_quickwit_query(input: &str) -> Option { + let tokens: Vec = input + .split_whitespace() + .filter(|token| !token.is_empty()) + .map(|token| { + let normalized = token.to_lowercase(); + escape_quickwit_token(&normalized) + }) + .collect(); + + if tokens.is_empty() { + return None; + } + + let parts: Vec = tokens + .into_iter() + .map(|token| format!("(title:{token} OR text:{token})")) + .collect(); + + Some(parts.join(" AND ")) +} + +pub fn escape_quickwit_token(token: &str) -> String { + let mut escaped = String::with_capacity(token.len()); + for ch in token.chars() { + match ch { + '+' | '-' | '&' | '|' | '!' | '(' | ')' | '{' | '}' | '[' | ']' | '^' | '"' | '~' + | '*' | '?' | ':' | '\\' | '/' => { + escaped.push('\\'); + escaped.push(ch); + } + _ => escaped.push(ch), + } + } + escaped +} + +pub fn extract_document_id(hit: &Value) -> Option { + for key in ["_source", "source", "fields", "stored_fields"] { + if let Some(value) = hit.get(key) { + if let Some(uuid) = extract_uuid_from_value(value) { + return Some(uuid); + } + } + } + + if let Some(value) = hit.get("document_id") { + if let Some(uuid) = extract_uuid_from_value(value) { + return Some(uuid); + } + } + + None +} + +pub fn extract_uuid_from_value(value: &Value) -> Option { + if let Some(obj) = value.as_object() { + if let Some(inner) = obj.get("document_id") { + return parse_uuid_value(inner); + } + } + + if let Some(arr) = value.as_array() { + for item in arr { + if let Some(uuid) = extract_uuid_from_value(item) { + return Some(uuid); + } + } + } + + parse_uuid_value(value) +} + +pub fn parse_uuid_value(value: &Value) -> Option { + if let Some(s) = value.as_str() { + return Uuid::parse_str(s).ok(); + } + + if let Some(arr) = value.as_array() { + for item in arr { + if let Some(uuid) = parse_uuid_value(item) { + return Some(uuid); + } + } + } + + None +} diff --git a/backend/tests/common/mod.rs b/backend/tests/common/mod.rs index 8d87919..fffa711 100644 --- a/backend/tests/common/mod.rs +++ b/backend/tests/common/mod.rs @@ -11,12 +11,13 @@ use axum::Router; use backend::auth::jwt::JwtService; use backend::config::AppConfig; use backend::db::{self, PgPool}; -use backend::models::{Job, NewUser, NewUserMembership}; +use backend::models::{Job, NewUser, NewUserMembership, Tenant}; use backend::routes; use backend::state::AppState; use backend::storage::ObjectStorage; use diesel::connection::SimpleConnection; use diesel::prelude::*; +use diesel::OptionalExtension; use diesel::PgConnection; use diesel_migrations::{embed_migrations, EmbeddedMigrations, MigrationHarness}; use http_body_util::BodyExt; @@ -151,11 +152,15 @@ impl TestApp { let state = AppState::new(pool.clone(), config, storage_for_state, jwt); let router = routes::create_router(state.clone()); - Ok(Self { + let app = Self { state, router, storage, - }) + }; + + app.ensure_default_tenant().await?; + + Ok(app) } pub async fn cleanup(&self) -> Result<()> { @@ -168,7 +173,10 @@ impl TestApp { Ok(()) }) .await - .context("cleanup task panicked")? + .context("cleanup task panicked")?; + + self.ensure_default_tenant().await?; + Ok(()) } #[allow(dead_code)] @@ -176,6 +184,19 @@ impl TestApp { self.storage.clone() } + pub async fn storage_key_for(&self, key: &str) -> Result { + let tenant = self + .state + .tenants + .get_by_slug(&self.state.config.default_tenant_slug) + .map_err(|err| anyhow!("default tenant not found: {:?}", err))?; + let root = tenant + .storage_root + .clone() + .ok_or_else(|| anyhow!("default tenant missing storage root"))?; + Ok(format!("{}{}", root, key)) + } + pub async fn insert_user(&self, username: &str, password: &str, role: &str) -> Result { let username = username.to_string(); let password = password.to_string(); @@ -184,7 +205,7 @@ impl TestApp { .state .tenants .tenant_id_for_slug(&self.state.config.default_tenant_slug) - .context("default tenant not found")?; + .map_err(|err| anyhow!("default tenant not found: {:?}", err))?; self.with_conn(move |conn| { let password_hash = hash_password(&password)?; let user = NewUser { @@ -213,6 +234,60 @@ impl TestApp { .await } + async fn ensure_default_tenant(&self) -> Result { + let slug_value = self.state.config.default_tenant_slug.clone(); + let quickwit_enabled = self.state.config.quickwit_endpoint.is_some(); + self.with_conn(move |conn| { + use backend::schema::tenants::dsl as tenants_dsl; + + let existing = tenants_dsl::tenants + .filter(tenants_dsl::slug.eq(&slug_value)) + .first::(conn) + .optional() + .context("failed to load default tenant")?; + + let tenant_id = if let Some(current) = existing { + let desired_root = current + .storage_root + .clone() + .filter(|root| root.ends_with('/')) + .unwrap_or_else(|| format!("test-tenants/{}/", current.id)); + + if current.storage_root.as_deref() != Some(desired_root.as_str()) { + diesel::update(tenants_dsl::tenants.filter(tenants_dsl::id.eq(current.id))) + .set(tenants_dsl::storage_root.eq(Some(desired_root))) + .execute(conn) + .context("failed to update default tenant storage root")?; + } + + current.id + } else { + let new_id = Uuid::new_v4(); + let root = format!("test-tenants/{}/", new_id); + let quickwit_value = if quickwit_enabled { + Some(format!("documents-{}", new_id)) + } else { + None + }; + + diesel::insert_into(tenants_dsl::tenants) + .values(( + tenants_dsl::id.eq(new_id), + tenants_dsl::slug.eq(&slug_value), + tenants_dsl::storage_root.eq(Some(root)), + tenants_dsl::quickwit_index.eq(quickwit_value), + )) + .execute(conn) + .context("failed to insert default tenant")?; + + new_id + }; + + Ok(tenant_id) + }) + .await + } + pub async fn login_token(&self, username: &str, password: &str) -> Result { #[derive(Serialize)] struct LoginPayload<'a> { @@ -490,7 +565,22 @@ async fn prepare_database(pool: &PgPool) -> Result<()> { fn truncate_all(conn: &mut PgConnection) -> Result<()> { conn.batch_execute( - "TRUNCATE TABLE document_tags, document_versions, documents, folders, tags, users RESTART IDENTITY CASCADE;", + "TRUNCATE TABLE \ + document_asset_objects, \ + document_assets, \ + document_correspondents, \ + correspondents, \ + document_tags, \ + document_versions, \ + documents, \ + folders, \ + jobs, \ + refresh_tokens, \ + tags, \ + user_memberships, \ + users, \ + tenants \ + RESTART IDENTITY CASCADE;", ) .context("failed to truncate tables")?; Ok(()) diff --git a/backend/tests/documents_flow.rs b/backend/tests/documents_flow.rs index 40e6a60..e401793 100644 --- a/backend/tests/documents_flow.rs +++ b/backend/tests/documents_flow.rs @@ -56,12 +56,12 @@ struct DocumentDownload { filename: String, } -#[derive(Deserialize)] - #[derive(Deserialize)] struct BulkReanalyze { queued: usize, } + +#[derive(Deserialize)] struct BulkMoveResult { updated: usize, } @@ -186,9 +186,10 @@ async fn upload_and_list_document() -> Result<()> { assert_eq!(current_version.size_bytes, file_bytes.len() as i64); assert!(current_version.assets.is_empty()); + let storage_key = app.storage_key_for(¤t_version.s3_key).await?; let stored = app .storage() - .get(¤t_version.s3_key) + .get(&storage_key) .await .expect("object stored"); assert_eq!(stored.bytes, file_bytes); @@ -318,7 +319,6 @@ async fn duplicate_and_restore_document() -> Result<()> { Ok(()) } - #[tokio::test] async fn bulk_move_documents_to_folder() -> Result<()> { let _lock = acquire_db_lock().await;