nils/papercrate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
82aa8948cfd6dfa03369da76d86dc98b3bccf98d
papercrate/backend/src/routes/documents.rs
T
nils 82aa8948cf tenants
2025-10-27 11:32:22 +01:00

2472 lines
78 KiB
Rust
Raw Blame History

use std::{
collections::{HashMap, HashSet},
time::Duration,
};
use axum::extract::{Json, Multipart, Path, Query, State};
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 serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use tracing::{debug, error, info, warn};
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;
use super::folders::gather_descendant_folder_ids;
use crate::auth::TenantScopedConn;
use crate::error::{AppError, AppResult};
use crate::jobs::{enqueue_job, JOB_ANALYZE_DOCUMENT};
use crate::models::{
Correspondent, Document, DocumentAsset, DocumentAssetObject, DocumentCorrespondent,
DocumentVersion, NewDocument, NewDocumentCorrespondent, 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,
};
use crate::state::AppState;
use crate::utils::{
db::{no_content, validate_bulk_ids, IntoJsonResponse},
http::inline_content_disposition,
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::{
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;
#[derive(Deserialize, IntoParams, ToSchema)]
#[into_params(parameter_in = Query)]
pub struct DocumentListQuery {
pub folder_id: Option<Uuid>,
#[serde(default)]
pub include_deleted: bool,
#[serde(default)]
pub include_descendants: Option<bool>,
pub query: Option<String>,
pub tags: Option<String>,
pub correspondents: Option<String>,
}
#[derive(Deserialize, IntoParams, ToSchema)]
#[into_params(parameter_in = Query)]
pub struct AssetRequestQuery {
#[serde(default)]
pub force: bool,
}
#[derive(Deserialize, IntoParams, ToSchema)]
#[into_params(parameter_in = Query)]
pub struct DocumentCheckQuery {
pub checksum: String,
}
#[derive(Serialize, ToSchema)]
pub struct DocumentCheckResponse {
pub exists: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub document_id: Option<Uuid>,
#[serde(skip_serializing_if = "Option::is_none")]
pub title: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub filename: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub version_id: Option<Uuid>,
#[serde(skip_serializing_if = "Option::is_none")]
pub version_number: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub uploaded_at: Option<String>,
}
#[derive(Serialize, ToSchema)]
pub struct TagResponse {
pub id: Uuid,
pub label: String,
pub color: Option<String>,
}
impl From<Tag> for TagResponse {
fn from(tag: Tag) -> Self {
Self {
id: tag.id,
label: tag.label,
color: tag.color,
}
}
}
#[derive(Serialize, Clone, ToSchema)]
pub struct DocumentVersionResponse {
pub id: Uuid,
pub version_number: i32,
pub s3_key: String,
pub size_bytes: i64,
pub checksum: String,
pub created_at: String,
pub metadata: Value,
#[serde(skip_serializing_if = "Option::is_none")]
pub operations_summary: Option<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<i32>,
}
#[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<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub expires_at: Option<i64>,
}
#[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<i32>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub objects: Vec<DocumentAssetObjectResponse>,
}
#[derive(Serialize, Clone, ToSchema)]
pub struct DocumentCurrentVersionResponse {
#[serde(flatten)]
pub version: DocumentVersionResponse,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub assets: Vec<DocumentAssetResponse>,
pub download_path: String,
}
#[derive(Serialize, Clone, ToSchema)]
pub struct DocumentCorrespondentResponse {
pub id: Uuid,
pub name: String,
pub role: String,
pub metadata: Value,
pub assigned_at: String,
}
#[derive(Serialize, ToSchema)]
pub struct DocumentResponse {
pub id: Uuid,
pub filename: String,
pub title: String,
pub original_name: String,
pub content_type: Option<String>,
pub folder_id: Option<Uuid>,
pub uploaded_at: String,
pub updated_at: String,
pub deleted_at: Option<String>,
pub issued_at: Option<String>,
pub metadata: Value,
pub tags: Vec<TagResponse>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub correspondents: Vec<DocumentCorrespondentResponse>,
#[serde(skip_serializing_if = "Option::is_none")]
pub current_version: Option<DocumentCurrentVersionResponse>,
}
#[derive(Serialize, ToSchema)]
pub struct DocumentDetailResponse {
pub document: DocumentResponse,
}
#[derive(Serialize, ToSchema)]
pub struct DocumentDownloadResponse {
pub url: String,
pub expires_in: u64,
pub filename: String,
pub content_type: Option<String>,
pub size_bytes: i64,
}
#[derive(Serialize, ToSchema)]
pub struct BulkReanalyzeResponse {
pub queued: usize,
}
#[derive(Deserialize, ToSchema)]
pub struct BulkMoveRequest {
pub document_ids: Vec<Uuid>,
pub folder_id: Option<Uuid>,
}
#[derive(Deserialize, ToSchema)]
pub struct UpdateDocumentRequest {
pub title: Option<String>,
}
#[derive(Serialize, ToSchema)]
pub struct BulkMoveResponse {
pub updated: usize,
}
#[derive(Deserialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum BulkTagAction {
Add,
Remove,
}
#[derive(Deserialize, ToSchema)]
pub struct BulkTagRequest {
pub document_ids: Vec<Uuid>,
pub tag_ids: Vec<Uuid>,
pub action: BulkTagAction,
}
#[derive(Serialize, ToSchema)]
pub struct BulkTagResponse {
pub added: usize,
pub removed: usize,
}
#[derive(Serialize, ToSchema)]
pub struct BulkCorrespondentResponse {
pub assigned: usize,
pub removed: usize,
}
#[derive(Deserialize, ToSchema)]
pub struct CorrespondentAssignmentInput {
pub correspondent_id: Uuid,
pub role: String,
}
#[derive(Deserialize, ToSchema)]
pub struct AssignCorrespondentsRequest {
pub assignments: Vec<CorrespondentAssignmentInput>,
#[serde(default)]
pub replace: bool,
}
#[derive(Deserialize, Copy, Clone, PartialEq, Eq, ToSchema)]
#[serde(rename_all = "lowercase")]
pub enum BulkCorrespondentAction {
Add,
Remove,
}
fn default_bulk_correspondent_action() -> BulkCorrespondentAction {
BulkCorrespondentAction::Add
}
#[derive(Deserialize, ToSchema)]
pub struct BulkCorrespondentsRequest {
pub document_ids: Vec<Uuid>,
pub assignments: Vec<CorrespondentAssignmentInput>,
#[serde(default = "default_bulk_correspondent_action")]
pub action: BulkCorrespondentAction,
}
#[derive(Deserialize, IntoParams, ToSchema)]
#[into_params(parameter_in = Query)]
pub struct CorrespondentRoleQuery {
pub role: String,
}
#[derive(Deserialize, ToSchema)]
pub struct BulkReanalyzeSelectionRequest {
pub document_ids: Vec<Uuid>,
#[serde(default = "default_true")]
pub force: bool,
}
fn default_true() -> bool {
true
}
struct UploadRequest {
bytes: Vec<u8>,
original_name: String,
content_type: Option<String>,
folder_id: Option<Uuid>,
metadata: Value,
title_override: Option<String>,
tag_ids: Vec<Uuid>,
correspondents: Vec<CorrespondentAssignmentInput>,
issued_at_override: Option<NaiveDateTime>,
skip_if_existing: bool,
}
enum UploadOutcome {
Created(DocumentDetailResponse),
Reused(DocumentDetailResponse),
Skipped { document_id: Uuid },
}
#[derive(ToSchema)]
pub struct UploadDocumentForm {
#[schema(value_type = String, format = Binary)]
pub file: String,
#[schema(nullable)]
pub folder_id: Option<Uuid>,
#[schema(nullable)]
pub metadata: Option<Value>,
#[schema(nullable)]
pub title: Option<String>,
#[schema(nullable, value_type = Vec<Uuid>)]
pub tag_ids: Option<Vec<Uuid>>,
#[schema(nullable, value_type = Vec<CorrespondentAssignmentInput>)]
pub correspondents: Option<Vec<CorrespondentAssignmentInput>>,
#[schema(nullable, example = "2024-01-01T00:00:00Z")]
pub issued_at: Option<String>,
#[schema(nullable)]
pub skip_existing: Option<bool>,
}
#[derive(Deserialize, ToSchema)]
pub struct MoveDocumentRequest {
pub folder_id: Option<Uuid>,
}
#[derive(Deserialize, ToSchema)]
pub struct AssignTagsRequest {
pub tag_ids: Vec<Uuid>,
}
#[derive(Deserialize, Default, IntoParams, ToSchema)]
#[into_params(parameter_in = Query)]
pub struct AssetObjectsQuery {
#[serde(default)]
pub start: Option<i32>,
#[serde(default)]
pub limit: Option<i32>,
}
pub async fn list_documents(
State(state): State<AppState>,
Query(params): Query<DocumentListQuery>,
TenantScopedConn {
mut conn,
tenant_id,
user_id,
..
}: TenantScopedConn,
) -> AppResult<Json<Vec<DocumentResponse>>> {
let DocumentListQuery {
folder_id,
include_deleted,
include_descendants,
query,
tags,
correspondents,
} = params;
let mut docs_query = documents::table
.filter(documents::tenant_id.eq(tenant_id))
.into_boxed();
if !include_deleted {
docs_query = docs_query.filter(documents::deleted_at.is_null());
}
let search_text = query
.as_ref()
.map(|s| s.trim())
.filter(|s| !s.is_empty())
.map(|s| s.to_owned());
let tags_param = tags
.as_ref()
.map(|s| s.trim())
.filter(|s| !s.is_empty())
.map(|s| s.to_owned());
let correspondents_param = correspondents
.as_ref()
.map(|s| s.trim())
.filter(|s| !s.is_empty())
.map(|s| s.to_owned());
let include_descendants = include_descendants.unwrap_or(true);
match (folder_id, include_descendants) {
(Some(folder_id), true) => {
let descendant_ids = gather_descendant_folder_ids(&mut conn, tenant_id, folder_id)?;
docs_query = docs_query.filter(documents::folder_id.eq_any(descendant_ids));
}
(Some(folder_id), false) => {
docs_query = docs_query.filter(documents::folder_id.eq(Some(folder_id)));
}
(None, false) => {
docs_query = docs_query.filter(documents::folder_id.is_null());
}
(None, true) => {}
}
let mut filter_ids: Option<HashSet<Uuid>> = None;
let mut quickwit_order: Option<Vec<Uuid>> = None;
if let Some(query_str) = search_text.as_ref() {
debug!(query = %query_str, "performing quickwit document search");
let endpoint = state
.config
.quickwit_endpoint
.as_ref()
.ok_or_else(|| AppError::internal("quickwit endpoint not configured"))?;
let tenant = state.tenants.get_by_id(tenant_id)?;
let index = tenant
.quickwit_index
.as_ref()
.ok_or_else(|| AppError::internal("quickwit index not configured for tenant"))?;
let ids = quickwit_search(endpoint, index, tenant_id, query_str)
.await
.map_err(|err| AppError::internal(format!("quickwit search failed: {err}")))?;
if ids.is_empty() {
return Ok(Json(vec![]));
}
quickwit_order = Some(ids.clone());
let set: HashSet<Uuid> = ids.into_iter().collect();
filter_ids = Some(match &filter_ids {
Some(existing) => existing.intersection(&set).copied().collect(),
None => set,
});
}
if let Some(tags_param) = tags_param.as_ref() {
let tag_ids: Result<Vec<Uuid>, _> = tags_param
.split(',')
.map(|s| Uuid::parse_str(s.trim()))
.collect();
if let Ok(ids) = tag_ids {
if !ids.is_empty() {
let mut doc_id_set: Option<HashSet<Uuid>> = None;
for tag_id in &ids {
let docs_for_tag: Vec<Uuid> = document_tags::table
.filter(document_tags::tag_id.eq(*tag_id))
.select(document_tags::document_id)
.load(&mut conn)?;
let docs_set: HashSet<Uuid> = docs_for_tag.into_iter().collect();
doc_id_set = Some(match doc_id_set {
Some(existing) => existing.intersection(&docs_set).cloned().collect(),
None => docs_set,
});
if let Some(ref set) = doc_id_set {
if set.is_empty() {
break;
}
}
}
let matching_doc_ids: HashSet<Uuid> = doc_id_set.unwrap_or_default();
if matching_doc_ids.is_empty() {
return Ok(Json(vec![]));
}
let new_filter = match &filter_ids {
Some(existing) => existing.intersection(&matching_doc_ids).copied().collect(),
None => matching_doc_ids.clone(),
};
filter_ids = Some(new_filter);
}
}
}
if let Some(correspondents_param) = correspondents_param.as_ref() {
let correspondent_ids: Result<Vec<Uuid>, _> = correspondents_param
.split(',')
.map(|s| Uuid::parse_str(s.trim()))
.collect();
if let Ok(ids) = correspondent_ids {
if !ids.is_empty() {
let mut doc_id_set: Option<HashSet<Uuid>> = None;
for correspondent_id in &ids {
let docs_for_correspondent: Vec<Uuid> = document_correspondents::table
.filter(document_correspondents::correspondent_id.eq(*correspondent_id))
.select(document_correspondents::document_id)
.load(&mut conn)?;
let docs_set: HashSet<Uuid> = docs_for_correspondent.into_iter().collect();
doc_id_set = Some(match doc_id_set {
Some(existing) => existing.intersection(&docs_set).cloned().collect(),
None => docs_set,
});
if let Some(ref set) = doc_id_set {
if set.is_empty() {
break;
}
}
}
let matching_doc_ids: HashSet<Uuid> = doc_id_set.unwrap_or_default();
if matching_doc_ids.is_empty() {
return Ok(Json(vec![]));
}
let new_filter = match &filter_ids {
Some(existing) => existing.intersection(&matching_doc_ids).copied().collect(),
None => matching_doc_ids.clone(),
};
filter_ids = Some(new_filter);
}
}
}
if let Some(ref set) = filter_ids {
if set.is_empty() {
return Ok(Json(vec![]));
}
let ids_vec: Vec<Uuid> = set.iter().copied().collect();
docs_query = docs_query.filter(documents::id.eq_any(ids_vec));
}
let docs: Vec<Document> = if let Some(order_ids) = quickwit_order.as_ref() {
let relevant_ids: Vec<Uuid> = if let Some(filter_set) = filter_ids.as_ref() {
order_ids
.iter()
.copied()
.filter(|id| filter_set.contains(id))
.collect()
} else {
order_ids.clone()
};
if relevant_ids.is_empty() {
return Ok(Json(vec![]));
}
let fetched: Vec<Document> = docs_query.load(&mut conn)?;
let mut by_id: HashMap<Uuid, Document> =
fetched.into_iter().map(|doc| (doc.id, doc)).collect();
let mut ordered = Vec::with_capacity(by_id.len());
for id in relevant_ids {
if let Some(doc) = by_id.remove(&id) {
ordered.push(doc);
}
}
if !by_id.is_empty() {
let mut remaining: Vec<Document> = by_id.into_values().collect();
remaining.sort_by(|a, b| b.uploaded_at.cmp(&a.uploaded_at));
ordered.extend(remaining);
}
ordered
} else {
docs_query
.order(documents::uploaded_at.desc())
.load(&mut conn)?
};
let doc_ids: Vec<Uuid> = docs.iter().map(|doc| doc.id).collect();
let tags_map = load_tags_for_documents(&mut conn, &doc_ids)?;
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 mut response = Vec::with_capacity(doc_ids.len());
for doc in docs {
let tags = tags_map.get(&doc.id).cloned();
let correspondents = correspondents_map.remove(&doc.id).unwrap_or_default();
let current_version = primary_versions.get(&doc.id).cloned();
response.push(to_document_response(
&state,
user_id,
doc,
tags,
correspondents,
current_version,
)?);
}
Ok(Json(response))
}
pub async fn check_document(
Query(query): Query<DocumentCheckQuery>,
TenantScopedConn {
mut conn,
tenant_id,
..
}: TenantScopedConn,
) -> AppResult<Json<DocumentCheckResponse>> {
let checksum_raw = query.checksum.trim();
if checksum_raw.is_empty() {
return Err(AppError::bad_request("checksum must not be empty"));
}
let checksum = checksum_raw.to_ascii_lowercase();
if !checksum.chars().all(|ch| ch.is_ascii_hexdigit()) {
return Err(AppError::bad_request(
"checksum must be a hex-encoded string",
));
}
let record: Option<(Document, DocumentVersion)> = documents::table
.inner_join(
document_versions::table.on(document_versions::id.eq(documents::current_version_id)),
)
.filter(documents::tenant_id.eq(tenant_id))
.filter(document_versions::checksum.eq(&checksum))
.select((documents::all_columns, document_versions::all_columns))
.first(&mut conn)
.optional()?;
if let Some((document, version)) = record {
Ok(Json(DocumentCheckResponse {
exists: true,
document_id: Some(document.id),
title: Some(document.title.clone()),
filename: Some(document.filename.clone()),
version_id: Some(version.id),
version_number: Some(version.version_number),
uploaded_at: Some(to_iso(document.uploaded_at)),
}))
} else {
Ok(Json(DocumentCheckResponse {
exists: false,
document_id: None,
title: None,
filename: None,
version_id: None,
version_number: None,
uploaded_at: None,
}))
}
}
pub async fn get_document(
State(state): State<AppState>,
Path(document_id): Path<Uuid>,
TenantScopedConn {
mut conn,
tenant_id,
user_id,
..
}: TenantScopedConn,
) -> AppResult<Json<DocumentDetailResponse>> {
let doc: Document = documents::table
.find(document_id)
.filter(documents::tenant_id.eq(tenant_id))
.first(&mut conn)?;
if doc.deleted_at.is_some() {
return Err(AppError::not_found());
}
let current_version: DocumentVersion = document_versions::table
.find(doc.current_version_id)
.first(&mut conn)?;
let tags_map = load_tags_for_documents(&mut conn, &[document_id])?;
let mut correspondents_map = load_correspondents_for_documents(&mut conn, &[document_id])?;
let version_id = current_version.id;
drop(conn);
let assets = load_asset_responses(&state, tenant_id, version_id).await?;
let version_response = to_version_response(current_version, true);
Ok(Json(DocumentDetailResponse {
document: to_document_response(
&state,
user_id,
doc,
tags_map.get(&document_id).cloned(),
correspondents_map.remove(&document_id).unwrap_or_default(),
Some((version_response, assets)),
)?,
}))
}
pub async fn upload_document(
State(state): State<AppState>,
TenantScopedConn {
tenant_id, user_id, ..
}: TenantScopedConn,
mut multipart: Multipart,
) -> AppResult<impl IntoResponse> {
let mut file_bytes: Option<Vec<u8>> = None;
let mut original_name: Option<String> = None;
let mut content_type: Option<String> = None;
let mut folder_id: Option<Uuid> = None;
let mut metadata: Value = Value::Object(Default::default());
let mut tag_ids: Vec<Uuid> = Vec::new();
let mut correspondents: Vec<CorrespondentAssignmentInput> = Vec::new();
let mut issued_at_override: Option<NaiveDateTime> = None;
let mut skip_if_existing = false;
let mut title_override: Option<String> = None;
while let Some(field) = multipart.next_field().await.map_err(|err| {
let msg = format!("invalid multipart data: {err}");
error!(error = %err, "invalid multipart data");
AppError::bad_request(msg)
})? {
let name = field.name().map(|n| n.to_string());
match name.as_deref() {
Some("file") => {
let file_name = field.file_name().map(|n| n.to_string());
original_name = file_name.clone();
content_type = field.content_type().map(|mime| mime.to_string());
let data = field.bytes().await.map_err(|err| {
let msg = format!("failed to read file bytes: {err}");
error!(error = %err, "failed to read file bytes");
AppError::bad_request(msg)
})?;
file_bytes = Some(data.to_vec());
}
Some("folder_id") => {
let value = field.text().await.map_err(|err| {
let msg = format!("invalid folder id: {err}");
error!(error = %err, "invalid folder id");
AppError::bad_request(msg)
})?;
if !value.trim().is_empty() {
let parsed = Uuid::parse_str(value.trim())
.map_err(|_| AppError::bad_request("folder_id must be a valid UUID"))?;
folder_id = Some(parsed);
}
}
Some("metadata") => {
let value = field.text().await.map_err(|err| {
let msg = format!("invalid metadata: {err}");
error!(error = %err, "invalid metadata payload");
AppError::bad_request(msg)
})?;
metadata = serde_json::from_str(&value).map_err(|err| {
let msg = format!("metadata must be valid JSON: {err}");
error!(error = %err, "metadata parse failure");
AppError::bad_request(msg)
})?;
}
Some("title") => {
let value = field.text().await.map_err(|err| {
let msg = format!("invalid title: {err}");
error!(error = %err, "invalid title payload");
AppError::bad_request(msg)
})?;
let trimmed = value.trim();
if !trimmed.is_empty() {
title_override = Some(trimmed.to_string());
}
}
Some("tag_ids") => {
let value = field.text().await.map_err(|err| {
let msg = format!("invalid tag_ids: {err}");
error!(error = %err, "invalid tag_ids payload");
AppError::bad_request(msg)
})?;
let parsed: Vec<String> = serde_json::from_str(&value).map_err(|err| {
let msg = format!("tag_ids must be a JSON array of UUID strings: {err}");
error!(error = %err, "invalid tag_ids json");
AppError::bad_request(msg)
})?;
let mut set = HashSet::new();
for raw in parsed {
let trimmed = raw.trim();
if trimmed.is_empty() {
continue;
}
let uuid = Uuid::parse_str(trimmed)
.map_err(|_| AppError::bad_request("tag_ids must contain valid UUIDs"))?;
set.insert(uuid);
}
tag_ids = set.into_iter().collect();
}
Some("correspondents") => {
let value = field.text().await.map_err(|err| {
let msg = format!("invalid correspondents: {err}");
error!(error = %err, "invalid correspondents payload");
AppError::bad_request(msg)
})?;
correspondents = serde_json::from_str(&value).map_err(|err| {
let msg = format!(
"correspondents must be a JSON array of {{correspondent_id, role}} objects: {err}"
);
error!(error = %err, "invalid correspondents json");
AppError::bad_request(msg)
})?;
}
Some("issued_at") => {
let value = field.text().await.map_err(|err| {
let msg = format!("invalid issued_at: {err}");
error!(error = %err, "invalid issued_at payload");
AppError::bad_request(msg)
})?;
let trimmed = value.trim();
if !trimmed.is_empty() {
let parsed = DateTime::parse_from_rfc3339(trimmed).map_err(|err| {
let msg = format!("issued_at must be an RFC3339 timestamp: {err}");
error!(error = %err, "invalid issued_at format");
AppError::bad_request(msg)
})?;
issued_at_override = Some(parsed.naive_utc());
}
}
Some("skip_existing") => {
let value = field.text().await.map_err(|err| {
let msg = format!("invalid skip_existing flag: {err}");
error!(error = %err, "invalid skip_existing payload");
AppError::bad_request(msg)
})?;
skip_if_existing = matches!(
value.trim().to_ascii_lowercase().as_str(),
"1" | "true" | "yes"
);
}
_ => {}
}
}
let file_bytes = file_bytes.ok_or_else(|| {
error!("upload rejected: missing file field");
AppError::bad_request("file field is required")
})?;
if file_bytes.is_empty() {
error!("upload rejected: empty file payload");
return Err(AppError::bad_request("file field must not be empty"));
}
let original_name = original_name.ok_or_else(|| {
error!("upload rejected: missing original filename");
AppError::bad_request("filename is required")
})?;
let original_name_for_log = original_name.clone();
let request = UploadRequest {
bytes: file_bytes,
original_name,
content_type,
folder_id,
metadata,
title_override,
tag_ids,
correspondents,
issued_at_override,
skip_if_existing,
};
let outcome = match process_upload(&state, request, tenant_id, user_id).await {
Ok(outcome) => outcome,
Err(err) => {
error!(error = ?err, original_name = %original_name_for_log, "document upload failed");
return Err(err);
}
};
let response = match outcome {
UploadOutcome::Created(detail) => {
info!(
document_id = %detail.document.id,
original_name = %detail.document.original_name,
created = true,
reused_existing = false,
"document upload succeeded",
);
(StatusCode::CREATED, Json(detail)).into_response()
}
UploadOutcome::Reused(detail) => {
info!(
document_id = %detail.document.id,
original_name = %detail.document.original_name,
created = false,
reused_existing = true,
"document upload succeeded",
);
(StatusCode::OK, Json(detail)).into_response()
}
UploadOutcome::Skipped { document_id } => {
info!(document_id = %document_id, "document upload skipped by client request");
StatusCode::NO_CONTENT.into_response()
}
};
Ok(response)
}
pub async fn request_document_assets(
Path(document_id): Path<Uuid>,
Query(query): Query<AssetRequestQuery>,
TenantScopedConn {
mut conn,
tenant_id,
..
}: TenantScopedConn,
) -> AppResult<StatusCode> {
let document: Document = documents::table
.find(document_id)
.filter(documents::tenant_id.eq(tenant_id))
.first(&mut conn)?;
if document.deleted_at.is_some() {
return Err(AppError::not_found());
}
enqueue_job(
&mut conn,
tenant_id,
JOB_ANALYZE_DOCUMENT,
json!({
"document_id": document_id,
"document_version_id": document.current_version_id,
"force": query.force,
}),
None,
)
.map_err(|err| AppError::internal(format!("failed to enqueue analyze job: {err}")))?;
Ok(StatusCode::ACCEPTED)
}
pub async fn reanalyze_selected_documents(
TenantScopedConn {
mut conn,
tenant_id,
..
}: TenantScopedConn,
Json(payload): Json<BulkReanalyzeSelectionRequest>,
) -> AppResult<(StatusCode, Json<BulkReanalyzeResponse>)> {
let BulkReanalyzeSelectionRequest {
mut document_ids,
force,
} = payload;
validate_bulk_ids(&mut document_ids, "document_ids")?;
let targets: Vec<(Uuid, Uuid)> = documents::table
.filter(documents::id.eq_any(&document_ids))
.filter(documents::deleted_at.is_null())
.filter(documents::tenant_id.eq(tenant_id))
.select((documents::id, documents::current_version_id))
.load(&mut conn)?;
if targets.len() != document_ids.len() {
return Err(AppError::bad_request(
"one or more documents do not exist or are inaccessible",
));
}
let mut queued = 0usize;
for (document_id, version_id) in targets {
enqueue_job(
&mut conn,
tenant_id,
JOB_ANALYZE_DOCUMENT,
json!({
"document_id": document_id,
"document_version_id": version_id,
"force": force,
}),
None,
)
.map_err(|err| AppError::internal(format!("failed to enqueue analyze job: {err}")))?;
queued += 1;
}
Ok((StatusCode::ACCEPTED, Json(BulkReanalyzeResponse { queued })))
}
pub async fn list_document_assets(
State(state): State<AppState>,
Path(document_id): Path<Uuid>,
TenantScopedConn {
mut conn,
tenant_id,
..
}: TenantScopedConn,
) -> AppResult<Json<Vec<DocumentAssetResponse>>> {
let document: Document = documents::table
.find(document_id)
.filter(documents::tenant_id.eq(tenant_id))
.first(&mut conn)?;
if document.deleted_at.is_some() {
return Err(AppError::not_found());
}
let version_id = document.current_version_id;
drop(conn);
let assets = load_asset_responses(&state, tenant_id, version_id).await?;
Ok(Json(assets))
}
pub async fn get_document_asset(
State(state): State<AppState>,
Path(asset_id): Path<Uuid>,
Query(query): Query<AssetObjectsQuery>,
TenantScopedConn {
mut conn,
tenant_id,
..
}: TenantScopedConn,
) -> AppResult<Json<DocumentAssetDetailResponse>> {
let asset: DocumentAsset = match document_assets::table
.find(asset_id)
.filter(document_assets::tenant_id.eq(tenant_id))
.first(&mut conn)
.optional()?
{
Some(asset) => asset,
None => return Err(AppError::not_found()),
};
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 end = start
.checked_add(limit - 1)
.ok_or_else(|| AppError::bad_request("requested range is too large"))?;
let objects: Vec<DocumentAssetObject> = 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 storage = state.storage_for_tenant(tenant_id)?;
let mut object_responses = Vec::with_capacity(objects.len());
for object in objects {
let url = storage
.presign_get_object(
&object.s3_key,
Duration::from_secs(PRESIGNED_URL_EXPIRY_SECONDS),
)
.await
.map_err(|err| AppError::internal(format!("failed to generate asset URL: {err}")))?;
object_responses.push(to_asset_object_response(
object,
Some(url),
Some(expires_at),
));
}
if object_responses.is_empty() {
return Err(AppError::not_found());
}
Ok(Json(to_asset_detail_response(asset, object_responses)))
}
pub async fn download_document(
State(state): State<AppState>,
Path(document_id): Path<Uuid>,
TenantScopedConn {
mut conn,
tenant_id,
..
}: TenantScopedConn,
) -> AppResult<Json<DocumentDownloadResponse>> {
let doc: Document = documents::table
.find(document_id)
.filter(documents::tenant_id.eq(tenant_id))
.first(&mut conn)?;
if doc.deleted_at.is_some() {
return Err(AppError::not_found());
}
let version: DocumentVersion = document_versions::table
.find(doc.current_version_id)
.first(&mut conn)?;
let storage = state.storage_for_tenant(tenant_id)?;
let presigned_url = storage
.presign_get_object(
&version.s3_key,
Duration::from_secs(PRESIGNED_URL_EXPIRY_SECONDS),
)
.await
.map_err(|err| AppError::internal(format!("failed to generate download URL: {err}")))?;
Ok(Json(DocumentDownloadResponse {
url: presigned_url,
expires_in: PRESIGNED_URL_EXPIRY_SECONDS,
filename: doc.original_name.clone(),
content_type: doc.content_type.clone(),
size_bytes: version.size_bytes,
}))
}
pub async fn download_with_token(
State(state): State<AppState>,
Path(token): Path<String>,
) -> AppResult<impl IntoResponse> {
let claims = state
.jwt
.verify_download_token(&token)
.map_err(|_| AppError::unauthorized())?;
let mut conn = state.db_for_tenant(claims.tenant_id)?;
let doc: Document = documents::table
.find(claims.doc_id)
.filter(documents::tenant_id.eq(claims.tenant_id))
.first(&mut conn)?;
if doc.deleted_at.is_some() {
return Err(AppError::not_found());
}
let version: DocumentVersion = document_versions::table
.find(doc.current_version_id)
.first(&mut conn)?;
let now = Utc::now().naive_utc();
let has_active_refresh: bool = select(exists(
refresh_dsl::refresh_tokens
.filter(refresh_dsl::user_id.eq(claims.user_id))
.filter(refresh_dsl::tenant_id.eq(claims.tenant_id))
.filter(refresh_dsl::revoked_at.is_null())
.filter(refresh_dsl::expires_at.gt(now)),
))
.get_result(&mut conn)?;
if !has_active_refresh {
return Err(AppError::unauthorized());
}
drop(conn);
let storage = state.storage_for_tenant(claims.tenant_id)?;
let presigned_url = storage
.presign_get_object(
&version.s3_key,
Duration::from_secs(PRESIGNED_URL_EXPIRY_SECONDS),
)
.await
.map_err(|err| AppError::internal(format!("failed to generate download URL: {err}")))?;
Ok(axum::response::Redirect::temporary(&presigned_url))
}
pub async fn delete_document(
Path(document_id): Path<Uuid>,
TenantScopedConn {
mut conn,
tenant_id,
..
}: TenantScopedConn,
) -> AppResult<impl IntoResponse> {
let now = Utc::now().naive_utc();
diesel::update(
documents::table
.find(document_id)
.filter(documents::tenant_id.eq(tenant_id)),
)
.set((
documents::deleted_at.eq(Some(now)),
documents::updated_at.eq(now),
))
.execute(&mut conn)?;
Ok(StatusCode::NO_CONTENT)
}
pub async fn update_document(
State(state): State<AppState>,
Path(document_id): Path<Uuid>,
TenantScopedConn {
mut conn,
tenant_id,
user_id,
..
}: TenantScopedConn,
Json(payload): Json<UpdateDocumentRequest>,
) -> AppResult<Json<DocumentDetailResponse>> {
let mut document: Document = documents::table
.find(document_id)
.filter(documents::tenant_id.eq(tenant_id))
.first(&mut conn)?;
if document.deleted_at.is_some() {
return Err(AppError::not_found());
}
let new_title = match payload.title {
Some(ref title) => {
let trimmed = title.trim();
if trimmed.is_empty() {
return Err(AppError::bad_request("title must not be empty"));
}
Some(trimmed.to_string())
}
None => None,
};
if new_title.is_none() {
return Err(AppError::bad_request("no changes provided"));
}
if let Some(title) = new_title {
let now = Utc::now().naive_utc();
let new_filename = filename_with_retained_extension(&title, &document.filename);
let target = documents::table
.find(document_id)
.filter(documents::tenant_id.eq(tenant_id));
let update_result = diesel::update(target).set((
documents::title.eq(&title),
documents::filename.eq(&new_filename),
documents::updated_at.eq(now),
));
match update_result.execute(&mut conn) {
Ok(_) => {}
Err(diesel::result::Error::DatabaseError(DatabaseErrorKind::UniqueViolation, _)) => {
return Err(AppError::conflict(
"another document in this folder already uses that filename",
)
.with_code("duplicate_filename"));
}
Err(err) => return Err(AppError::from(err)),
}
document = documents::table
.find(document_id)
.filter(documents::tenant_id.eq(tenant_id))
.first(&mut conn)?;
}
let current_version: DocumentVersion = document_versions::table
.find(document.current_version_id)
.first(&mut conn)?;
let tags_map = load_tags_for_documents(&mut conn, &[document_id])?;
let mut correspondents_map = load_correspondents_for_documents(&mut conn, &[document_id])?;
let version_id = current_version.id;
drop(conn);
let assets = load_asset_responses(&state, tenant_id, version_id).await?;
let version_response = to_version_response(current_version, true);
Ok(Json(DocumentDetailResponse {
document: to_document_response(
&state,
user_id,
document,
tags_map.get(&document_id).cloned(),
correspondents_map.remove(&document_id).unwrap_or_default(),
Some((version_response, assets)),
)?,
}))
}
pub async fn move_document(
Path(document_id): Path<Uuid>,
TenantScopedConn {
mut conn,
tenant_id,
..
}: TenantScopedConn,
Json(payload): Json<MoveDocumentRequest>,
) -> AppResult<impl IntoResponse> {
if let Some(folder_id) = payload.folder_id {
ensure_folder_exists_on_conn(&mut conn, tenant_id, folder_id)?;
}
let now = Utc::now().naive_utc();
diesel::update(
documents::table
.find(document_id)
.filter(documents::tenant_id.eq(tenant_id)),
)
.set((
documents::folder_id.eq(payload.folder_id),
documents::updated_at.eq(now),
))
.execute(&mut conn)?;
Ok(StatusCode::NO_CONTENT)
}
pub async fn bulk_move_documents(
TenantScopedConn {
mut conn,
tenant_id,
..
}: TenantScopedConn,
Json(payload): Json<BulkMoveRequest>,
) -> AppResult<(StatusCode, Json<BulkMoveResponse>)> {
let BulkMoveRequest {
mut document_ids,
folder_id,
} = payload;
if document_ids.is_empty() {
return Err(AppError::bad_request("document_ids must not be empty"));
}
document_ids.sort();
document_ids.dedup();
if let Some(target_folder) = folder_id {
ensure_folder_exists_on_conn(&mut conn, tenant_id, target_folder)?;
}
let existing: Vec<(Uuid, Option<NaiveDateTime>)> = documents::table
.filter(documents::id.eq_any(&document_ids))
.filter(documents::tenant_id.eq(tenant_id))
.select((documents::id, documents::deleted_at))
.load(&mut conn)?;
if existing.len() != document_ids.len() {
return Err(AppError::bad_request(
"one or more documents do not exist or are inaccessible",
));
}
if existing.iter().any(|(_, deleted)| deleted.is_some()) {
return Err(AppError::bad_request("cannot move deleted documents"));
}
let now = Utc::now().naive_utc();
let updated = match diesel::update(
documents::table
.filter(documents::id.eq_any(&document_ids))
.filter(documents::tenant_id.eq(tenant_id)),
)
.set((
documents::folder_id.eq(folder_id),
documents::updated_at.eq(now),
))
.execute(&mut conn)
{
Ok(value) => value,
Err(diesel::result::Error::DatabaseError(kind, info)) => {
error!(
?kind,
detail = ?info.details(),
constraint = info.constraint_name(),
tenant_id = %tenant_id,
target_folder = folder_id.map(|id| id.to_string()),
"bulk move update failed"
);
let message = info
.constraint_name()
.map(|name| format!("constraint {name} prevented moving documents"))
.unwrap_or_else(|| "unable to move documents due to a constraint".to_string());
return Err(AppError::conflict(message));
}
Err(err) => {
error!(
?err,
tenant_id = %tenant_id,
target_folder = folder_id.map(|id| id.to_string()),
"bulk move update failed"
);
return Err(AppError::from(err));
}
};
let body = BulkMoveResponse { updated };
Ok((StatusCode::OK, body.into_json()?))
}
pub async fn assign_correspondents(
Path(document_id): Path<Uuid>,
TenantScopedConn {
mut conn,
tenant_id,
user_id,
..
}: TenantScopedConn,
Json(payload): Json<AssignCorrespondentsRequest>,
) -> AppResult<StatusCode> {
if payload.assignments.is_empty() {
return Err(AppError::bad_request("assignments must not be empty"));
}
let (normalized_pairs, _correspondent_ids, roles_vec) =
normalize_correspondent_assignments(&payload.assignments)?;
let replace = payload.replace;
conn.transaction::<(), AppError, _>(|conn| {
let document: Document = documents::table
.find(document_id)
.filter(documents::tenant_id.eq(tenant_id))
.first(conn)?;
if document.deleted_at.is_some() {
return Err(AppError::not_found());
}
let mut deleted = 0;
if replace {
deleted = diesel::delete(
document_correspondents::table
.filter(document_correspondents::document_id.eq(document_id))
.filter(document_correspondents::tenant_id.eq(tenant_id))
.filter(document_correspondents::role.eq_any(&roles_vec)),
)
.execute(conn)?;
}
let inserted =
insert_document_correspondents(conn, tenant_id, &document, user_id, &normalized_pairs)?;
if replace && deleted > 0 && 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(())
})?;
no_content()
}
pub async fn bulk_assign_correspondents(
TenantScopedConn {
mut conn,
tenant_id,
user_id,
..
}: TenantScopedConn,
Json(payload): Json<BulkCorrespondentsRequest>,
) -> AppResult<(StatusCode, Json<BulkCorrespondentResponse>)> {
if payload.assignments.is_empty() {
return Err(AppError::bad_request("assignments must not be empty"));
}
let mut document_ids = payload.document_ids;
validate_bulk_ids(&mut document_ids, "document_ids")?;
let (normalized_pairs, correspondents_vec, _roles_vec) =
normalize_correspondent_assignments(&payload.assignments)?;
let action = payload.action;
let user_id_val = user_id;
let (assigned, removed) = conn.transaction::<(usize, usize), AppError, _>(|conn| {
let docs: Vec<(Uuid, Option<NaiveDateTime>)> = documents::table
.filter(documents::id.eq_any(&document_ids))
.filter(documents::tenant_id.eq(tenant_id))
.select((documents::id, documents::deleted_at))
.load(conn)?;
if docs.len() != document_ids.len() {
return Err(AppError::bad_request(
"one or more documents do not exist or are inaccessible",
));
}
if docs.iter().any(|(_, deleted)| deleted.is_some()) {
return Err(AppError::bad_request(
"cannot assign correspondents to deleted documents",
));
}
if !correspondents_vec.is_empty() {
let existing: Vec<Correspondent> = correspondents::table
.filter(correspondents::id.eq_any(&correspondents_vec))
.filter(correspondents::tenant_id.eq(tenant_id))
.load(conn)?;
if existing.len() != correspondents_vec.len() {
return Err(AppError::bad_request(
"one or more correspondents do not exist",
));
}
}
match action {
BulkCorrespondentAction::Add => {
use diesel::dsl::not;
let mut grouped_by_role: HashMap<String, Vec<Uuid>> = HashMap::new();
for (correspondent_id, role) in &normalized_pairs {
grouped_by_role
.entry(role.clone())
.or_default()
.push(*correspondent_id);
}
let mut removed = 0;
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(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());
for doc_id in &document_ids {
for (correspondent_id, role) in &normalized_pairs {
new_rows.push(NewDocumentCorrespondent {
document_id: *doc_id,
correspondent_id: *correspondent_id,
role: role.clone(),
assigned_by: Some(user_id_val),
tenant_id,
});
}
}
let assigned = if new_rows.is_empty() {
0
} else {
diesel::insert_into(document_correspondents::table)
.values(&new_rows)
.on_conflict_do_nothing()
.execute(conn)?
};
if assigned > 0 || removed > 0 {
diesel::update(
documents::table
.filter(documents::id.eq_any(&document_ids))
.filter(documents::tenant_id.eq(tenant_id)),
)
.set(documents::updated_at.eq(Utc::now().naive_utc()))
.execute(conn)?;
}
Ok((assigned, removed))
}
BulkCorrespondentAction::Remove => {
let mut removed = 0;
if !normalized_pairs.is_empty() {
let mut grouped: HashMap<String, Vec<Uuid>> = HashMap::new();
for (correspondent_id, role) in &normalized_pairs {
grouped
.entry(role.clone())
.or_default()
.push(*correspondent_id);
}
for (role, ids) in grouped {
removed += 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(role.as_str()))
.filter(document_correspondents::correspondent_id.eq_any(&ids)),
)
.execute(conn)?;
}
}
if removed > 0 {
diesel::update(
documents::table
.filter(documents::id.eq_any(&document_ids))
.filter(documents::tenant_id.eq(tenant_id)),
)
.set(documents::updated_at.eq(Utc::now().naive_utc()))
.execute(conn)?;
}
Ok((0, removed))
}
}
})?;
let body = BulkCorrespondentResponse { assigned, removed };
Ok((StatusCode::OK, body.into_json()?))
}
pub async fn remove_correspondent(
Path((document_id, correspondent_id)): Path<(Uuid, Uuid)>,
Query(query): Query<CorrespondentRoleQuery>,
TenantScopedConn {
mut conn,
tenant_id,
..
}: TenantScopedConn,
) -> AppResult<StatusCode> {
let role = normalize_role(&query.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(", ")
)));
}
let document: Document = documents::table
.find(document_id)
.filter(documents::tenant_id.eq(tenant_id))
.first(&mut conn)?;
if document.deleted_at.is_some() {
return Err(AppError::not_found());
}
let deleted = diesel::delete(
document_correspondents::table
.filter(document_correspondents::document_id.eq(document_id))
.filter(document_correspondents::tenant_id.eq(tenant_id))
.filter(document_correspondents::correspondent_id.eq(correspondent_id))
.filter(document_correspondents::role.eq(&role)),
)
.execute(&mut conn)?;
if deleted == 0 {
return Err(AppError::not_found());
}
diesel::update(
documents::table
.find(document_id)
.filter(documents::tenant_id.eq(tenant_id)),
)
.set(documents::updated_at.eq(Utc::now().naive_utc()))
.execute(&mut conn)?;
no_content()
}
pub async fn assign_tags(
Path(document_id): Path<Uuid>,
TenantScopedConn {
mut conn,
tenant_id,
user_id,
..
}: TenantScopedConn,
Json(payload): Json<AssignTagsRequest>,
) -> AppResult<impl IntoResponse> {
if payload.tag_ids.is_empty() {
return Err(AppError::bad_request("tag_ids must not be empty"));
}
let document: Document = documents::table
.find(document_id)
.filter(documents::tenant_id.eq(tenant_id))
.first(&mut conn)?;
assign_tags_internal(
&mut conn,
tenant_id,
&document,
&payload.tag_ids,
Some(user_id),
)?;
Ok(StatusCode::NO_CONTENT)
}
pub async fn bulk_update_tags(
TenantScopedConn {
mut conn,
tenant_id,
user_id,
..
}: TenantScopedConn,
Json(payload): Json<BulkTagRequest>,
) -> AppResult<(StatusCode, Json<BulkTagResponse>)> {
let BulkTagRequest {
mut document_ids,
mut tag_ids,
action,
} = payload;
validate_bulk_ids(&mut document_ids, "document_ids")?;
validate_bulk_ids(&mut tag_ids, "tag_ids")?;
let docs: Vec<(Uuid, Option<NaiveDateTime>)> = documents::table
.filter(documents::id.eq_any(&document_ids))
.filter(documents::tenant_id.eq(tenant_id))
.select((documents::id, documents::deleted_at))
.load(&mut conn)?;
if docs.len() != document_ids.len() {
return Err(AppError::bad_request(
"one or more documents do not exist or are inaccessible",
));
}
if docs.iter().any(|(_, deleted)| deleted.is_some()) {
return Err(AppError::bad_request(
"cannot assign or remove tags from deleted documents",
));
}
let existing_tags: Vec<Tag> = tags::table
.filter(tags::id.eq_any(&tag_ids))
.filter(tags::tenant_id.eq(tenant_id))
.load(&mut conn)?;
if existing_tags.len() != tag_ids.len() {
return Err(AppError::bad_request("one or more tags do not exist"));
}
let response = match action {
BulkTagAction::Add => {
let mut inserts = Vec::with_capacity(document_ids.len() * tag_ids.len());
for doc_id in &document_ids {
for tag_id in &tag_ids {
inserts.push(NewDocumentTag {
document_id: *doc_id,
tag_id: *tag_id,
assigned_by: Some(user_id),
tenant_id,
});
}
}
let added = if inserts.is_empty() {
0
} else {
diesel::insert_into(document_tags::table)
.values(&inserts)
.on_conflict_do_nothing()
.execute(&mut conn)?
};
BulkTagResponse { added, removed: 0 }
}
BulkTagAction::Remove => {
let removed = diesel::delete(
document_tags::table
.filter(document_tags::document_id.eq_any(&document_ids))
.filter(document_tags::tenant_id.eq(tenant_id))
.filter(document_tags::tag_id.eq_any(&tag_ids)),
)
.execute(&mut conn)?;
BulkTagResponse { added: 0, removed }
}
};
Ok((StatusCode::OK, response.into_json()?))
}
pub async fn remove_tag(
Path((document_id, tag_id)): Path<(Uuid, Uuid)>,
TenantScopedConn {
mut conn,
tenant_id,
..
}: TenantScopedConn,
) -> AppResult<StatusCode> {
diesel::delete(
document_tags::table
.filter(document_tags::document_id.eq(document_id))
.filter(document_tags::tenant_id.eq(tenant_id))
.filter(document_tags::tag_id.eq(tag_id)),
)
.execute(&mut conn)?;
no_content()
}
async fn process_upload(
state: &AppState,
request: UploadRequest,
tenant_id: Uuid,
user_id: Uuid,
) -> AppResult<UploadOutcome> {
let UploadRequest {
bytes,
original_name,
content_type,
folder_id,
metadata,
title_override,
tag_ids,
correspondents,
issued_at_override,
skip_if_existing,
} = request;
if let Some(folder) = folder_id {
let mut conn = state.db_for_tenant(tenant_id)?;
ensure_folder_exists_on_conn(&mut conn, tenant_id, folder)?;
}
let doc_id = Uuid::new_v4();
let version_id = Uuid::new_v4();
let version_number = 1;
let derived_title = title_override
.as_ref()
.map(|value| value.trim())
.filter(|value| !value.is_empty())
.map(|value| value.to_string())
.unwrap_or_else(|| derive_document_title(&original_name));
let stored_filename = filename_with_retained_extension(&derived_title, &original_name);
let checksum = Sha256::digest(&bytes);
let checksum_hex = hex::encode(checksum);
let size_bytes = bytes.len() as i64;
let s3_key = document_version_object_key(doc_id, version_number, version_id);
{
let mut conn = state.db_for_tenant(tenant_id)?;
let existing = documents::table
.inner_join(
document_versions::table
.on(document_versions::id.eq(documents::current_version_id)),
)
.filter(documents::tenant_id.eq(tenant_id))
.filter(document_versions::checksum.eq(&checksum_hex))
.select((documents::all_columns, document_versions::all_columns))
.first::<(Document, DocumentVersion)>(&mut conn)
.optional()?;
if let Some((mut document, version)) = existing {
if skip_if_existing {
info!(
document_id = %document.id,
checksum = %checksum_hex,
"upload skipped existing document due to skip flag",
);
return Ok(UploadOutcome::Skipped {
document_id: document.id,
});
}
if let Some(issued_at) = issued_at_override {
if document.issued_at != Some(issued_at) {
diesel::update(
documents::table
.find(document.id)
.filter(documents::tenant_id.eq(tenant_id)),
)
.set((
documents::issued_at.eq(Some(issued_at)),
documents::updated_at.eq(Utc::now().naive_utc()),
))
.execute(&mut conn)?;
document.issued_at = Some(issued_at);
}
}
assign_tags_internal(&mut conn, tenant_id, &document, &tag_ids, Some(user_id))?;
assign_correspondents_internal(
&mut conn,
tenant_id,
&document,
user_id,
&correspondents,
)?;
if document.deleted_at.is_some() {
let now = Utc::now().naive_utc();
diesel::update(documents::table.find(document.id))
.set((
documents::deleted_at.eq(None::<NaiveDateTime>),
documents::updated_at.eq(now),
))
.execute(&mut conn)?;
document.deleted_at = None;
document.updated_at = now;
}
let tags_map = load_tags_for_documents(&mut conn, &[document.id])?;
let mut correspondents_map =
load_correspondents_for_documents(&mut conn, &[document.id])?;
let tags = tags_map.get(&document.id).cloned();
let correspondents = correspondents_map.remove(&document.id).unwrap_or_default();
drop(conn);
let assets = load_asset_responses(state, tenant_id, version.id).await?;
let version_response = to_version_response(version.clone(), true);
info!(
document_id = %document.id,
checksum = %checksum_hex,
"upload deduplicated existing document"
);
return Ok(UploadOutcome::Reused(DocumentDetailResponse {
document: to_document_response(
state,
user_id,
document,
tags,
correspondents,
Some((version_response, assets)),
)?,
}));
}
}
let content_disposition = inline_content_disposition(&stored_filename);
let storage = state.storage_for_tenant(tenant_id)?;
storage
.put_object(
&s3_key,
bytes.clone(),
content_type.clone(),
content_disposition.clone(),
)
.await
.map_err(|err| {
error!(error = %err, key = %s3_key, "failed to store document");
AppError::internal(format!("failed to store document: {err}"))
})?;
let metadata_value = if metadata.is_null() {
Value::Object(Default::default())
} else {
metadata
};
let (document, version) = {
let mut conn = state.db_for_tenant(tenant_id)?;
let transaction_result = conn.transaction(|conn| {
let new_document = NewDocument {
id: doc_id,
filename: stored_filename.clone(),
original_name: original_name.clone(),
content_type: content_type.clone(),
folder_id,
current_version_id: version_id,
issued_at: issued_at_override,
title: derived_title.clone(),
metadata: metadata_value.clone(),
tenant_id,
};
diesel::insert_into(documents::table)
.values(&new_document)
.execute(conn)?;
let new_version = NewDocumentVersion {
id: version_id,
document_id: doc_id,
version_number,
s3_key: s3_key.clone(),
size_bytes,
checksum: checksum_hex.clone(),
metadata: Value::Object(Default::default()),
operations_summary: Value::Object(Default::default()),
tenant_id,
};
diesel::insert_into(document_versions::table)
.values(&new_version)
.execute(conn)?;
let document: Document = documents::table.find(doc_id).first(conn)?;
let version: DocumentVersion = document_versions::table.find(version_id).first(conn)?;
Ok::<_, diesel::result::Error>((document, version))
});
match transaction_result {
Ok(result) => result,
Err(diesel::result::Error::DatabaseError(DatabaseErrorKind::UniqueViolation, _)) => {
return Err(AppError::conflict(
"another document in this folder already uses that filename",
)
.with_code("duplicate_filename"))
}
Err(err) => return Err(AppError::from(err)),
}
};
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_correspondents_internal(&mut conn, tenant_id, &document, user_id, &correspondents)?;
let tags_map = load_tags_for_documents(&mut conn, &[doc_id])?;
let mut correspondents_map = load_correspondents_for_documents(&mut conn, &[doc_id])?;
let tags = tags_map.get(&doc_id).cloned();
let correspondents = correspondents_map.remove(&doc_id).unwrap_or_default();
drop(conn);
DocumentDetailResponse {
document: to_document_response(
state,
user_id,
document,
tags,
correspondents,
Some((to_version_response(version.clone(), true), Vec::new())),
)?,
}
};
if let Ok(mut conn) = state.db_for_tenant(tenant_id) {
if let Err(err) = enqueue_job(
&mut conn,
tenant_id,
JOB_ANALYZE_DOCUMENT,
json!({
"document_id": doc_id,
"document_version_id": version.id,
"force": false,
}),
None,
) {
warn!(document_id = %doc_id, error = %err, "failed to enqueue analyze job");
}
} else {
warn!(document_id = %doc_id, "failed to enqueue analyze job due to pool error");
}
Ok(UploadOutcome::Created(detail))
}
fn assign_tags_internal(
conn: &mut PgConnection,
tenant_id: Uuid,
document: &Document,
raw_tag_ids: &[Uuid],
assigned_by: Option<Uuid>,
) -> AppResult<usize> {
if raw_tag_ids.is_empty() {
return Ok(0);
}
let mut tag_ids: Vec<Uuid> = raw_tag_ids.iter().copied().collect();
tag_ids.sort_unstable();
tag_ids.dedup();
if tag_ids.is_empty() {
return Ok(0);
}
let existing: Vec<Uuid> = 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<NewDocumentTag> = 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<usize> {
if assignments.is_empty() {
return Ok(0);
}
let (normalized_pairs, _correspondent_ids, _roles) =
normalize_correspondent_assignments(assignments)?;
insert_document_correspondents(conn, tenant_id, document, user_id, &normalized_pairs)
}
fn insert_document_correspondents(
conn: &mut PgConnection,
tenant_id: Uuid,
document: &Document,
user_id: Uuid,
normalized_pairs: &[(Uuid, String)],
) -> AppResult<usize> {
if normalized_pairs.is_empty() {
return Ok(0);
}
let mut correspondent_ids: Vec<Uuid> = normalized_pairs.iter().map(|(id, _)| *id).collect();
correspondent_ids.sort_unstable();
correspondent_ids.dedup();
if !correspondent_ids.is_empty() {
let existing: Vec<Uuid> = correspondents::table
.filter(correspondents::id.eq_any(&correspondent_ids))
.filter(correspondents::tenant_id.eq(tenant_id))
.select(correspondents::id)
.load(conn)?;
if existing.len() != correspondent_ids.len() {
return Err(AppError::bad_request(
"one or more correspondents do not exist",
));
}
}
let new_rows: Vec<NewDocumentCorrespondent> = normalized_pairs
.iter()
.map(|(correspondent_id, role)| NewDocumentCorrespondent {
document_id: document.id,
correspondent_id: *correspondent_id,
role: role.clone(),
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<HashMap<Uuid, Vec<Tag>>> {
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<Uuid, Vec<Tag>> = 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<HashMap<Uuid, Vec<DocumentCorrespondentResponse>>> {
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::role.asc(),
document_correspondents::assigned_at.asc(),
))
.load(conn)?;
let mut map: HashMap<Uuid, Vec<DocumentCorrespondentResponse>> = HashMap::new();
for (assignment, correspondent) in rows {
map.entry(assignment.document_id)
.or_default()
.push(DocumentCorrespondentResponse {
id: correspondent.id,
name: correspondent.name,
role: assignment.role,
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<HashMap<Uuid, (DocumentVersionResponse, Vec<DocumentAssetResponse>)>> {
if documents.is_empty() {
return Ok(HashMap::new());
}
let mut doc_to_version: HashMap<Uuid, Uuid> = HashMap::with_capacity(documents.len());
let mut version_ids: Vec<Uuid> = 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<DocumentVersion> = document_versions::table
.filter(document_versions::id.eq_any(&version_ids))
.load(&mut conn)?;
let mut version_map: HashMap<Uuid, DocumentVersion> = HashMap::new();
for version in versions {
version_map.insert(version.id, version);
}
let assets: Vec<(DocumentAsset, Option<DocumentAssetObject>)> = 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<Uuid, Vec<DocumentAssetResponse>> = 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<Uuid, (DocumentVersionResponse, Vec<DocumentAssetResponse>)> =
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, false), assets));
}
}
Ok(result)
}
pub(crate) fn to_document_response(
state: &AppState,
user_id: Uuid,
doc: Document,
tags: Option<Vec<Tag>>,
correspondents: Vec<DocumentCorrespondentResponse>,
current_version: Option<(DocumentVersionResponse, Vec<DocumentAssetResponse>)>,
) -> AppResult<DocumentResponse> {
let current_version = if let Some((version, assets)) = current_version {
let download_path = build_download_path(state, &doc, user_id)?;
Some(DocumentCurrentVersionResponse {
version,
assets,
download_path,
})
} else {
None
};
Ok(DocumentResponse {
id: doc.id,
filename: doc.filename,
title: doc.title,
original_name: doc.original_name,
content_type: doc.content_type,
folder_id: doc.folder_id,
uploaded_at: to_iso(doc.uploaded_at),
updated_at: to_iso(doc.updated_at),
deleted_at: doc.deleted_at.map(to_iso),
issued_at: doc.issued_at.map(to_iso),
metadata: doc.metadata,
tags: tags
.unwrap_or_default()
.into_iter()
.map(TagResponse::from)
.collect(),
correspondents,
current_version,
})
}
async fn load_asset_responses(
state: &AppState,
tenant_id: Uuid,
version_id: Uuid,
) -> AppResult<Vec<DocumentAssetResponse>> {
let mut conn = state.db_for_tenant(tenant_id)?;
let assets: Vec<(DocumentAsset, Option<DocumentAssetObject>)> = 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<Vec<Uuid>> {
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<Value>,
}
Reference in New Issue View Git Blame Copy Permalink