nils/papercrate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor

Browse Source
This commit is contained in:
nils
2025-10-28 02:05:38 +01:00
parent 1e177580c9
commit eb93a2131f
12 changed files with 691 additions and 660 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/routes/documents.rs
+66 -516
View File
@@ -8,10 +8,9 @@ use axum::http::StatusCode;
use axum::response::IntoResponse;
use chrono::{DateTime, NaiveDateTime, Utc};
use diesel::dsl::exists;
use diesel::{prelude::*, result::DatabaseErrorKind, select, PgConnection};
use reqwest::Client;
use diesel::{prelude::*, result::DatabaseErrorKind, select};
use serde::{Deserialize, Serialize};
use serde_json::{json, map::Entry, Map, Value};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use tracing::{debug, error, info, warn};
use utoipa::{IntoParams, ToSchema};
@@ -19,16 +18,31 @@ use uuid::Uuid;
use super::folders::gather_descendant_folder_ids;
use crate::auth::TenantScopedConn;
use crate::documents::{
asset::{
build_download_path, derive_document_title, filename_with_retained_extension,
load_asset_responses, load_primary_assets, to_asset_detail_response,
to_asset_object_response, to_version_response, DocumentAssetDetailResponse,
DocumentAssetResponse, DocumentVersionDetailResponse, DocumentVersionResponse,
},
correspondents::{
insert_document_correspondents, load_correspondents_for_documents,
normalize_correspondent_ids, DocumentCorrespondentResponse,
},
folders::ensure_folder_exists_on_conn,
metadata::merge_document_metadata,
search::quickwit_search,
tags::{assign_tags as assign_tags_to_document, load_tags_for_documents},
};
use crate::error::{AppError, AppResult};
use crate::jobs::{enqueue_job, JOB_ANALYZE_DOCUMENT, JOB_INDEX_DOCUMENT_TEXT};
use crate::models::{
Correspondent, Document, DocumentAsset, DocumentAssetObject, DocumentCorrespondent,
DocumentVersion, NewDocument, NewDocumentCorrespondent, NewDocumentTag, NewDocumentVersion,
Tag,
Document, DocumentAsset, DocumentAssetObject, DocumentVersion, NewDocument, NewDocumentTag,
NewDocumentVersion, Tag,
};
use crate::schema::{
correspondents, document_asset_objects, document_assets, document_correspondents,
document_tags, document_versions, documents, folders, refresh_tokens::dsl as refresh_dsl, tags,
document_asset_objects, document_assets, document_correspondents, document_tags,
document_versions, documents, folders, refresh_tokens::dsl as refresh_dsl, tags,
};
use crate::state::AppState;
use crate::utils::{
@@ -37,22 +51,9 @@ use crate::utils::{
json::{classify_nullable, NullableValue},
storage_paths::document_version_object_key,
time::to_iso,
validation::ensure_exists,
};
mod asset_utils;
mod correspondent_utils;
mod search_utils;
use asset_utils::{
build_download_path, derive_document_title, filename_with_retained_extension,
to_asset_detail_response, to_asset_object_response, to_asset_summary, to_version_response,
};
use correspondent_utils::normalize_correspondent_assignments;
use search_utils::{build_quickwit_query, extract_document_id};
const PRESIGNED_URL_EXPIRY_SECONDS: u64 = 300;
const QUICKWIT_MAX_HITS: usize = 200;
#[derive(Deserialize, IntoParams, ToSchema)]
#[into_params(parameter_in = Query)]
@@ -126,67 +127,6 @@ impl From<Tag> for TagResponse {
}
}
#[derive(Serialize, Clone, ToSchema)]
pub struct DocumentVersionResponse {
pub id: Uuid,
pub version_number: i32,
pub size_bytes: i64,
pub checksum: String,
pub created_at: String,
pub metadata: Value,
}
#[derive(Serialize, Clone, ToSchema)]
pub struct DocumentAssetResponse {
pub id: Uuid,
pub asset_type: String,
pub mime_type: String,
pub metadata: Value,
#[serde(skip_serializing_if = "Option::is_none")]
pub cardinality: Option<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 DocumentVersionDetailResponse {
#[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 metadata: Value,
pub assigned_at: String,
}
#[derive(Serialize, ToSchema)]
pub struct DocumentResponse {
pub id: Uuid,
@@ -626,7 +566,7 @@ pub async fn list_documents(
let mut correspondents_map = load_correspondents_for_documents(&mut conn, &doc_ids)?;
drop(conn);
let primary_versions = load_primary_assets(&state, tenant_id, &docs).await?;
let primary_versions = load_primary_assets(&state, tenant_id, &docs)?;
let mut response = Vec::with_capacity(doc_ids.len());
for doc in docs {
let tags = tags_map.get(&doc.id).cloned();
@@ -1487,55 +1427,6 @@ pub async fn restore_document(
Ok(StatusCode::NO_CONTENT)
}
fn merge_document_metadata(existing: Value, updates: Value) -> AppResult<Value> {
let mut base = match existing {
Value::Object(map) => map,
Value::Null => Map::new(),
_ => {
return Err(AppError::bad_request(
"existing metadata is not an object; set replace=true to overwrite",
));
}
};
let incoming = match updates {
Value::Object(map) => map,
_ => {
return Err(AppError::bad_request(
"metadata value must be a JSON object when replace is false",
));
}
};
merge_metadata_maps(&mut base, incoming);
Ok(Value::Object(base))
}
fn merge_metadata_maps(target: &mut Map<String, Value>, updates: Map<String, Value>) {
for (key, value) in updates {
match target.entry(key) {
Entry::Occupied(mut entry) => {
let existing = entry.get_mut();
match value {
Value::Object(update_map) => {
if let Value::Object(existing_map) = existing {
merge_metadata_maps(existing_map, update_map);
} else {
*existing = Value::Object(update_map);
}
}
other => {
*existing = other;
}
}
}
Entry::Vacant(entry) => {
entry.insert(value);
}
}
}
}
pub async fn move_document(
Path(document_id): Path<Uuid>,
TenantScopedConn {
@@ -1661,7 +1552,12 @@ pub async fn assign_correspondents(
return Err(AppError::bad_request("assignments must not be empty"));
}
let correspondent_ids = normalize_correspondent_assignments(&payload.assignments)?;
let raw_correspondent_ids: Vec<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| {
@@ -1739,7 +1635,12 @@ pub async fn bulk_assign_correspondents(
let mut document_ids = payload.document_ids;
validate_bulk_ids(&mut document_ids, "document_ids")?;
let correspondent_ids = normalize_correspondent_assignments(&payload.assignments)?;
let raw_correspondent_ids: Vec<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| {
@@ -1761,18 +1662,6 @@ pub async fn bulk_assign_correspondents(
));
}
if !correspondent_ids.is_empty() {
let existing: Vec<Correspondent> = correspondents::table
.filter(correspondents::id.eq_any(&correspondent_ids))
.filter(correspondents::tenant_id.eq(tenant_id))
.load(conn)?;
if existing.len() != correspondent_ids.len() {
return Err(AppError::bad_request(
"one or more correspondents do not exist",
));
}
}
match action {
BulkCorrespondentAction::Add => {
let mut assigned_total = 0;
@@ -1880,7 +1769,7 @@ pub async fn assign_tags(
.filter(documents::tenant_id.eq(tenant_id))
.first(&mut conn)?;
assign_tags_internal(
assign_tags_to_document(
&mut conn,
tenant_id,
&document,
@@ -2077,15 +1966,22 @@ async fn process_upload(
}
}
assign_tags_internal(&mut conn, tenant_id, &document, &tag_ids, Some(user_id))?;
assign_tags_to_document(&mut conn, tenant_id, &document, &tag_ids, Some(user_id))?;
assign_correspondents_internal(
&mut conn,
tenant_id,
&document,
user_id,
&correspondents,
)?;
if !correspondents.is_empty() {
let raw_ids: Vec<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();
@@ -2205,9 +2101,22 @@ async fn process_upload(
let detail = {
let mut conn = state.db_for_tenant(tenant_id)?;
assign_tags_internal(&mut conn, tenant_id, &document, &tag_ids, Some(user_id))?;
assign_tags_to_document(&mut conn, tenant_id, &document, &tag_ids, Some(user_id))?;
assign_correspondents_internal(&mut conn, tenant_id, &document, user_id, &correspondents)?;
if !correspondents.is_empty() {
let raw_ids: Vec<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])?;
@@ -2248,271 +2157,6 @@ async fn process_upload(
Ok(UploadOutcome::Created(detail))
}
fn assign_tags_internal(
conn: &mut PgConnection,
tenant_id: Uuid,
document: &Document,
raw_tag_ids: &[Uuid],
assigned_by: Option<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 ids = normalize_correspondent_assignments(assignments)?;
insert_document_correspondents(conn, tenant_id, document.id, user_id, &ids)
}
fn insert_document_correspondents(
conn: &mut PgConnection,
tenant_id: Uuid,
document_id: Uuid,
user_id: Uuid,
correspondent_ids: &[Uuid],
) -> AppResult<usize> {
if correspondent_ids.is_empty() {
return Ok(0);
}
let mut unique_ids: Vec<Uuid> = correspondent_ids.to_vec();
unique_ids.sort_unstable();
unique_ids.dedup();
if !unique_ids.is_empty() {
let existing: Vec<Uuid> = correspondents::table
.filter(correspondents::id.eq_any(&unique_ids))
.filter(correspondents::tenant_id.eq(tenant_id))
.select(correspondents::id)
.load(conn)?;
if existing.len() != unique_ids.len() {
return Err(AppError::bad_request(
"one or more correspondents do not exist",
));
}
}
let new_rows: Vec<NewDocumentCorrespondent> = unique_ids
.into_iter()
.map(|correspondent_id| NewDocumentCorrespondent {
document_id,
correspondent_id,
assigned_by: Some(user_id),
tenant_id,
})
.collect();
if new_rows.is_empty() {
return Ok(0);
}
let inserted = diesel::insert_into(document_correspondents::table)
.values(&new_rows)
.on_conflict_do_nothing()
.execute(conn)?;
if inserted > 0 {
diesel::update(
documents::table
.find(document_id)
.filter(documents::tenant_id.eq(tenant_id)),
)
.set(documents::updated_at.eq(Utc::now().naive_utc()))
.execute(conn)?;
}
Ok(inserted)
}
fn ensure_folder_exists_on_conn(
conn: &mut PgConnection,
tenant_id: Uuid,
folder_id: Uuid,
) -> AppResult<()> {
let exists: bool = diesel::select(exists(
folders::table
.filter(folders::id.eq(folder_id))
.filter(folders::tenant_id.eq(tenant_id)),
))
.get_result(conn)?;
ensure_exists(exists, "folder")
}
pub(crate) fn load_tags_for_documents(
conn: &mut PgConnection,
document_ids: &[Uuid],
) -> AppResult<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::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,
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), assets));
}
}
Ok(result)
}
pub(crate) fn to_document_response(
state: &AppState,
user_id: Uuid,
@@ -2553,97 +2197,3 @@ pub(crate) fn to_document_response(
current_version,
})
}
async fn load_asset_responses(
state: &AppState,
tenant_id: Uuid,
version_id: Uuid,
) -> AppResult<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>,
}