2207 lines
70 KiB
Rust
2207 lines
70 KiB
Rust
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};
|
|
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::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::{
|
|
Document, DocumentAsset, DocumentAssetObject, DocumentVersion, NewDocument, NewDocumentTag,
|
|
NewDocumentVersion, Tag,
|
|
};
|
|
use crate::schema::{
|
|
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},
|
|
error::StorageResultExt,
|
|
http::inline_content_disposition,
|
|
json::{classify_nullable, NullableValue},
|
|
storage_paths::document_version_object_key,
|
|
time::to_iso,
|
|
};
|
|
|
|
const PRESIGNED_URL_EXPIRY_SECONDS: u64 = 300;
|
|
|
|
#[derive(Deserialize, IntoParams, ToSchema)]
|
|
#[into_params(parameter_in = Query)]
|
|
pub struct DocumentListQuery {
|
|
pub folder_id: Option<Uuid>,
|
|
#[serde(default)]
|
|
pub include_descendants: Option<bool>,
|
|
pub query: Option<String>,
|
|
pub tags: Option<String>,
|
|
pub correspondents: Option<String>,
|
|
#[serde(default = "default_document_status_filter")]
|
|
pub status: DocumentStatusFilter,
|
|
}
|
|
|
|
fn default_document_status_filter() -> DocumentStatusFilter {
|
|
DocumentStatusFilter::Active
|
|
}
|
|
|
|
#[derive(Clone, Copy, Deserialize, Serialize, ToSchema)]
|
|
#[serde(rename_all = "lowercase")]
|
|
pub enum DocumentStatusFilter {
|
|
Active,
|
|
Deleted,
|
|
All,
|
|
}
|
|
|
|
#[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 created_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, 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 created_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<DocumentVersionDetailResponse>,
|
|
}
|
|
#[derive(Serialize, ToSchema)]
|
|
pub struct DocumentDetailResponse {
|
|
pub document: DocumentResponse,
|
|
}
|
|
|
|
#[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 DocumentMetadataUpdate {
|
|
pub value: Value,
|
|
#[serde(default)]
|
|
#[schema(default = false)]
|
|
pub replace: bool,
|
|
}
|
|
|
|
#[derive(Deserialize, ToSchema)]
|
|
pub struct UpdateDocumentRequest {
|
|
#[serde(default)]
|
|
pub title: Option<String>,
|
|
#[serde(default)]
|
|
#[schema(nullable, value_type = Option<String>)]
|
|
pub issued_at: Option<Value>,
|
|
#[serde(default)]
|
|
#[schema(nullable)]
|
|
pub metadata: Option<DocumentMetadataUpdate>,
|
|
}
|
|
|
|
#[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,
|
|
}
|
|
|
|
#[derive(Deserialize, ToSchema)]
|
|
pub struct AssignCorrespondentsRequest {
|
|
pub assignments: Vec<CorrespondentAssignmentInput>,
|
|
#[serde(default)]
|
|
pub replace: bool,
|
|
}
|
|
|
|
#[derive(Default, AsChangeset)]
|
|
#[diesel(table_name = documents)]
|
|
struct DocumentUpdateChangeset {
|
|
title: Option<String>,
|
|
filename: Option<String>,
|
|
issued_at: Option<Option<NaiveDateTime>>,
|
|
metadata: Option<Value>,
|
|
updated_at: Option<NaiveDateTime>,
|
|
}
|
|
|
|
#[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, 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 RestoreDocumentRequest {
|
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
|
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_descendants,
|
|
query,
|
|
tags,
|
|
correspondents,
|
|
status,
|
|
} = params;
|
|
|
|
let mut docs_query = documents::table
|
|
.filter(documents::tenant_id.eq(tenant_id))
|
|
.into_boxed();
|
|
|
|
match status {
|
|
DocumentStatusFilter::Active => {
|
|
docs_query = docs_query.filter(documents::deleted_at.is_null());
|
|
}
|
|
DocumentStatusFilter::Deleted => {
|
|
docs_query = docs_query.filter(documents::deleted_at.is_not_null());
|
|
}
|
|
DocumentStatusFilter::All => {}
|
|
}
|
|
|
|
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| {
|
|
error!(error = ?err, "quickwit search failed");
|
|
AppError::internal("quickwit search failed")
|
|
})?;
|
|
|
|
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.created_at.cmp(&a.created_at));
|
|
ordered.extend(remaining);
|
|
}
|
|
|
|
ordered
|
|
} else {
|
|
docs_query
|
|
.order(documents::created_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)?;
|
|
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),
|
|
created_at: Some(to_iso(document.created_at)),
|
|
}))
|
|
} else {
|
|
Ok(Json(DocumentCheckResponse {
|
|
exists: false,
|
|
document_id: None,
|
|
title: None,
|
|
filename: None,
|
|
version_id: None,
|
|
version_number: None,
|
|
created_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);
|
|
|
|
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}} 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| {
|
|
error!(error = ?err, "failed to enqueue analyze job");
|
|
AppError::internal("failed to enqueue analyze job")
|
|
})?;
|
|
|
|
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| {
|
|
error!(error = ?err, "failed to enqueue analyze job");
|
|
AppError::internal("failed to enqueue analyze job")
|
|
})?;
|
|
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
|
|
.storage_context("failed to generate asset URL")?;
|
|
|
|
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 list_document_versions(
|
|
Path(document_id): Path<Uuid>,
|
|
TenantScopedConn {
|
|
mut conn,
|
|
tenant_id,
|
|
..
|
|
}: TenantScopedConn,
|
|
) -> AppResult<Json<Vec<DocumentVersionResponse>>> {
|
|
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 versions: Vec<DocumentVersion> = document_versions::table
|
|
.filter(document_versions::document_id.eq(document_id))
|
|
.filter(document_versions::tenant_id.eq(tenant_id))
|
|
.order(document_versions::version_number.asc())
|
|
.load(&mut conn)?;
|
|
|
|
let versions: Vec<DocumentVersionResponse> =
|
|
versions.into_iter().map(to_version_response).collect();
|
|
|
|
Ok(Json(versions))
|
|
}
|
|
|
|
pub async fn get_document_version(
|
|
State(state): State<AppState>,
|
|
Path((document_id, version_id)): Path<(Uuid, Uuid)>,
|
|
TenantScopedConn {
|
|
mut conn,
|
|
tenant_id,
|
|
user_id,
|
|
..
|
|
}: TenantScopedConn,
|
|
) -> AppResult<Json<DocumentVersionDetailResponse>> {
|
|
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: DocumentVersion = document_versions::table
|
|
.find(version_id)
|
|
.filter(document_versions::document_id.eq(document_id))
|
|
.filter(document_versions::tenant_id.eq(tenant_id))
|
|
.first(&mut conn)?;
|
|
|
|
drop(conn);
|
|
|
|
let assets = load_asset_responses(&state, tenant_id, version.id).await?;
|
|
let download_path = build_download_path(&state, &document, user_id)?;
|
|
let version_core = to_version_response(version);
|
|
|
|
Ok(Json(DocumentVersionDetailResponse {
|
|
version: version_core,
|
|
assets,
|
|
download_path,
|
|
}))
|
|
}
|
|
|
|
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
|
|
.storage_context("failed to generate download URL")?;
|
|
|
|
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<Value>,
|
|
) -> 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 payload_obj = payload
|
|
.as_object()
|
|
.ok_or_else(|| AppError::bad_request("request body must be a JSON object"))?;
|
|
|
|
let title = match payload_obj.get("title") {
|
|
None | Some(Value::Null) => None,
|
|
Some(Value::String(value)) => Some(value.clone()),
|
|
Some(_) => return Err(AppError::bad_request("title must be a string")),
|
|
};
|
|
|
|
let issued_at_class =
|
|
classify_nullable(payload_obj.get("issued_at")).map_err(AppError::bad_request)?;
|
|
|
|
let metadata = match payload_obj.get("metadata") {
|
|
None | Some(Value::Null) => None,
|
|
Some(value) => Some(
|
|
serde_json::from_value::<DocumentMetadataUpdate>(value.clone())
|
|
.map_err(|err| AppError::bad_request(format!("invalid metadata payload: {err}")))?,
|
|
),
|
|
};
|
|
|
|
let mut changes = DocumentUpdateChangeset::default();
|
|
let mut has_changes = false;
|
|
let mut title_changed = false;
|
|
|
|
if let Some(ref candidate) = title {
|
|
let trimmed = candidate.trim();
|
|
if trimmed.is_empty() {
|
|
return Err(AppError::bad_request("title must not be empty"));
|
|
}
|
|
if trimmed != document.title {
|
|
let new_title = trimmed.to_string();
|
|
let new_filename = filename_with_retained_extension(&new_title, &document.filename);
|
|
changes.title = Some(new_title);
|
|
if new_filename != document.filename {
|
|
changes.filename = Some(new_filename);
|
|
}
|
|
has_changes = true;
|
|
title_changed = true;
|
|
}
|
|
}
|
|
|
|
match issued_at_class {
|
|
NullableValue::Omitted => {}
|
|
NullableValue::Null => {
|
|
if document.issued_at.is_some() {
|
|
changes.issued_at = Some(None);
|
|
has_changes = true;
|
|
}
|
|
}
|
|
NullableValue::String(raw) => {
|
|
let trimmed = raw.trim();
|
|
if trimmed.is_empty() {
|
|
return Err(AppError::bad_request("issued_at must not be empty"));
|
|
}
|
|
let parsed = DateTime::parse_from_rfc3339(trimmed).map_err(|err| {
|
|
let msg = format!("issued_at must be an RFC3339 timestamp: {err}");
|
|
AppError::bad_request(msg)
|
|
})?;
|
|
let normalized = Some(parsed.naive_utc());
|
|
if document.issued_at != normalized {
|
|
changes.issued_at = Some(normalized);
|
|
has_changes = true;
|
|
}
|
|
}
|
|
}
|
|
|
|
if let Some(metadata_update) = metadata {
|
|
let next_metadata = if metadata_update.replace {
|
|
metadata_update.value
|
|
} else {
|
|
merge_document_metadata(document.metadata.clone(), metadata_update.value)?
|
|
};
|
|
|
|
if document.metadata != next_metadata {
|
|
changes.metadata = Some(next_metadata);
|
|
has_changes = true;
|
|
}
|
|
}
|
|
|
|
if !has_changes {
|
|
return Err(AppError::bad_request("no changes provided"));
|
|
}
|
|
|
|
let now = Utc::now().naive_utc();
|
|
changes.updated_at = Some(now);
|
|
|
|
let target = documents::table
|
|
.find(document_id)
|
|
.filter(documents::tenant_id.eq(tenant_id));
|
|
|
|
let update_result = diesel::update(target).set(&changes);
|
|
|
|
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)?;
|
|
|
|
if title_changed {
|
|
if let Err(err) = enqueue_job(
|
|
&mut conn,
|
|
tenant_id,
|
|
JOB_INDEX_DOCUMENT_TEXT,
|
|
json!({
|
|
"document_id": document.id,
|
|
"document_version_id": current_version.id,
|
|
}),
|
|
None,
|
|
) {
|
|
warn!(
|
|
document_id = %document.id,
|
|
version_id = %current_version.id,
|
|
error = %err,
|
|
"failed to enqueue reindex job after title change"
|
|
);
|
|
}
|
|
}
|
|
|
|
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);
|
|
|
|
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 restore_document(
|
|
Path(document_id): Path<Uuid>,
|
|
TenantScopedConn {
|
|
mut conn,
|
|
tenant_id,
|
|
..
|
|
}: TenantScopedConn,
|
|
Json(payload): Json<RestoreDocumentRequest>,
|
|
) -> AppResult<StatusCode> {
|
|
let mut document: Document = documents::table
|
|
.find(document_id)
|
|
.filter(documents::tenant_id.eq(tenant_id))
|
|
.first(&mut conn)?;
|
|
|
|
if document.deleted_at.is_none() {
|
|
return Ok(StatusCode::NO_CONTENT);
|
|
}
|
|
|
|
if let Some(folder_id) = payload.folder_id {
|
|
ensure_folder_exists_on_conn(&mut conn, tenant_id, folder_id)?;
|
|
document.folder_id = Some(folder_id);
|
|
} else if let Some(folder_id) = document.folder_id {
|
|
let exists: bool = diesel::select(exists(
|
|
folders::table
|
|
.filter(folders::id.eq(folder_id))
|
|
.filter(folders::tenant_id.eq(tenant_id)),
|
|
))
|
|
.get_result(&mut conn)?;
|
|
|
|
if !exists {
|
|
document.folder_id = None;
|
|
}
|
|
}
|
|
|
|
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::<Option<NaiveDateTime>>(None),
|
|
documents::folder_id.eq(document.folder_id),
|
|
documents::updated_at.eq(now),
|
|
))
|
|
.execute(&mut conn)?;
|
|
|
|
Ok(StatusCode::NO_CONTENT)
|
|
}
|
|
|
|
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 raw_correspondent_ids: Vec<Uuid> = 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| {
|
|
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 updated = false;
|
|
if replace {
|
|
use diesel::dsl::not;
|
|
|
|
let base = document_correspondents::table
|
|
.filter(document_correspondents::document_id.eq(document_id))
|
|
.filter(document_correspondents::tenant_id.eq(tenant_id));
|
|
|
|
let removed = if correspondent_ids.is_empty() {
|
|
diesel::delete(base).execute(conn)?
|
|
} else {
|
|
diesel::delete(base.filter(not(
|
|
document_correspondents::correspondent_id.eq_any(&correspondent_ids),
|
|
)))
|
|
.execute(conn)?
|
|
};
|
|
|
|
if removed > 0 {
|
|
updated = true;
|
|
}
|
|
}
|
|
|
|
let inserted = insert_document_correspondents(
|
|
conn,
|
|
tenant_id,
|
|
document.id,
|
|
user_id,
|
|
&correspondent_ids,
|
|
)?;
|
|
|
|
if inserted > 0 {
|
|
updated = true;
|
|
}
|
|
|
|
if updated && 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 raw_correspondent_ids: Vec<Uuid> = 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| {
|
|
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",
|
|
));
|
|
}
|
|
|
|
match action {
|
|
BulkCorrespondentAction::Add => {
|
|
let mut assigned_total = 0;
|
|
for (doc_id, _) in &docs {
|
|
assigned_total += insert_document_correspondents(
|
|
conn,
|
|
tenant_id,
|
|
*doc_id,
|
|
user_id_val,
|
|
&correspondent_ids,
|
|
)?;
|
|
}
|
|
|
|
Ok((assigned_total, 0))
|
|
}
|
|
BulkCorrespondentAction::Remove => {
|
|
if correspondent_ids.is_empty() {
|
|
return Ok((0, 0));
|
|
}
|
|
|
|
let 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::correspondent_id.eq_any(&correspondent_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)>,
|
|
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());
|
|
}
|
|
|
|
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)),
|
|
)
|
|
.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_to_document(
|
|
&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_to_document(&mut conn, tenant_id, &document, &tag_ids, Some(user_id))?;
|
|
|
|
if !correspondents.is_empty() {
|
|
let raw_ids: Vec<Uuid> = 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();
|
|
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());
|
|
|
|
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
|
|
.storage_context("failed to store document")?;
|
|
|
|
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()),
|
|
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_to_document(&mut conn, tenant_id, &document, &tag_ids, Some(user_id))?;
|
|
|
|
if !correspondents.is_empty() {
|
|
let raw_ids: Vec<Uuid> = 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])?;
|
|
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()), 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))
|
|
}
|
|
|
|
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(DocumentVersionDetailResponse {
|
|
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,
|
|
created_at: to_iso(doc.created_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,
|
|
})
|
|
}
|