backend: correspondents

backend/src/routes/documents.rs

@@ -1,4 +1,8 @@
-use std::{collections::HashMap, path::Path as FsPath, time::Duration};
+use std::{
+    collections::{HashMap, HashSet},
+    path::Path as FsPath,
+    time::Duration,
+};
 
 use axum::extract::{Json, Multipart, Path, Query, State};
 use axum::http::StatusCode;
@@ -16,15 +20,25 @@ use crate::auth::AuthenticatedUser;
 use crate::error::{AppError, AppResult};
 use crate::jobs::{enqueue_job, JOB_ANALYZE_DOCUMENT};
 use crate::models::{
-    Document, DocumentAsset, DocumentVersion, NewDocument, NewDocumentTag, NewDocumentVersion, Tag,
+    Correspondent, Document, DocumentAsset, DocumentCorrespondent, DocumentVersion, NewDocument,
+    NewDocumentCorrespondent, NewDocumentTag, NewDocumentVersion, Tag,
 };
 use crate::schema::{
-    document_assets, document_tags, document_versions, documents, folders,
-    refresh_tokens::dsl as refresh_dsl, tags,
+    correspondents, document_assets, document_correspondents, document_tags, document_versions,
+    documents, folders, refresh_tokens::dsl as refresh_dsl, tags,
 };
 use crate::state::AppState;
 
 const PRESIGNED_URL_EXPIRY_SECONDS: u64 = 300;
+pub const CORRESPONDENT_ROLES: &[&str] = &["sender", "receiver", "other"];
+
+fn normalize_role(value: &str) -> String {
+    value.trim().to_lowercase()
+}
+
+fn is_valid_correspondent_role(role: &str) -> bool {
+    CORRESPONDENT_ROLES.iter().any(|allowed| *allowed == role)
+}
 
 fn inline_content_disposition(filename: &str) -> Option<String> {
     if filename.is_empty() {
@@ -108,6 +122,15 @@ pub struct DocumentCurrentVersionResponse {
     pub download_path: String,
 }
 
+#[derive(Serialize, Clone)]
+pub struct DocumentCorrespondentResponse {
+    pub id: Uuid,
+    pub name: String,
+    pub role: String,
+    pub metadata: Value,
+    pub assigned_at: String,
+}
+
 #[derive(Serialize)]
 pub struct DocumentResponse {
     pub id: Uuid,
@@ -122,6 +145,8 @@ pub struct DocumentResponse {
     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>,
 }
@@ -180,6 +205,24 @@ pub struct BulkTagResponse {
     pub removed: usize,
 }
 
+#[derive(Deserialize)]
+pub struct CorrespondentAssignmentInput {
+    pub correspondent_id: Uuid,
+    pub role: String,
+}
+
+#[derive(Deserialize)]
+pub struct AssignCorrespondentsRequest {
+    pub assignments: Vec<CorrespondentAssignmentInput>,
+    #[serde(default)]
+    pub replace: bool,
+}
+
+#[derive(Deserialize)]
+pub struct CorrespondentRoleQuery {
+    pub role: String,
+}
+
 #[derive(Deserialize)]
 pub struct BulkReanalyzeSelectionRequest {
     pub document_ids: Vec<Uuid>,
@@ -242,6 +285,7 @@ pub async fn list_documents(
 
     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, &docs).await?;
@@ -249,12 +293,14 @@ pub async fn list_documents(
     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.user_id,
             doc,
             tags,
+            correspondents,
             current_version,
         )?);
     }
@@ -279,6 +325,7 @@ pub async fn get_document(
         .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);
 
@@ -291,6 +338,7 @@ pub async fn get_document(
             user.user_id,
             doc,
             tags_map.get(&document_id).cloned(),
+            correspondents_map.remove(&document_id).unwrap_or_default(),
             Some((version_response, assets)),
         )?,
     }))
@@ -689,6 +737,7 @@ pub async fn update_document(
         .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);
 
@@ -701,6 +750,7 @@ pub async fn update_document(
             user.user_id,
             document,
             tags_map.get(&document_id).cloned(),
+            correspondents_map.remove(&document_id).unwrap_or_default(),
             Some((version_response, assets)),
         )?,
     }))
@@ -775,6 +825,157 @@ pub async fn bulk_move_documents(
     Ok((StatusCode::OK, Json(BulkMoveResponse { updated })))
 }
 
+pub async fn assign_correspondents(
+    State(state): State<AppState>,
+    Path(document_id): Path<Uuid>,
+    user: AuthenticatedUser,
+    Json(payload): Json<AssignCorrespondentsRequest>,
+) -> AppResult<impl IntoResponse> {
+    if payload.assignments.is_empty() {
+        return Err(AppError::bad_request("assignments must not be empty"));
+    }
+
+    let mut unique_pairs: HashSet<(Uuid, String)> = HashSet::new();
+    let mut normalized_pairs: Vec<(Uuid, String)> = Vec::new();
+    let mut role_set: HashSet<String> = HashSet::new();
+    let mut correspondent_ids: HashSet<Uuid> = HashSet::new();
+
+    for assignment in &payload.assignments {
+        let role = normalize_role(&assignment.role);
+        if role.is_empty() {
+            return Err(AppError::bad_request("role must not be empty"));
+        }
+        if !is_valid_correspondent_role(&role) {
+            return Err(AppError::bad_request(format!(
+                "invalid correspondent role '{role}'. Allowed roles: {}",
+                CORRESPONDENT_ROLES.join(", ")
+            )));
+        }
+
+        if !unique_pairs.insert((assignment.correspondent_id, role.clone())) {
+            continue;
+        }
+
+        normalized_pairs.push((assignment.correspondent_id, role.clone()));
+        role_set.insert(role);
+        correspondent_ids.insert(assignment.correspondent_id);
+    }
+
+    if normalized_pairs.is_empty() {
+        return Err(AppError::bad_request(
+            "assignments must contain at least one unique correspondent/role pair",
+        ));
+    }
+
+    let correspondents_vec: Vec<Uuid> = correspondent_ids.into_iter().collect();
+    let roles_vec: Vec<String> = role_set.into_iter().collect();
+    let replace = payload.replace;
+    let user_id = user.user_id;
+
+    let mut conn = state.db()?;
+    conn.transaction::<(), AppError, _>(|conn| {
+        let document: Document = documents::table.find(document_id).first(conn)?;
+        if document.deleted_at.is_some() {
+            return Err(AppError::not_found());
+        }
+
+        if !correspondents_vec.is_empty() {
+            let existing: Vec<Correspondent> = correspondents::table
+                .filter(correspondents::id.eq_any(&correspondents_vec))
+                .load(conn)?;
+            if existing.len() != correspondents_vec.len() {
+                return Err(AppError::bad_request(
+                    "one or more correspondents do not exist",
+                ));
+            }
+        }
+
+        let mut changed = false;
+        if replace {
+            let deleted = diesel::delete(
+                document_correspondents::table
+                    .filter(document_correspondents::document_id.eq(document_id))
+                    .filter(document_correspondents::role.eq_any(&roles_vec)),
+            )
+            .execute(conn)?;
+            if deleted > 0 {
+                changed = true;
+            }
+        }
+
+        let new_rows: Vec<NewDocumentCorrespondent> = normalized_pairs
+            .iter()
+            .map(|(correspondent_id, role)| NewDocumentCorrespondent {
+                document_id,
+                correspondent_id: *correspondent_id,
+                role: role.clone(),
+                assigned_by: Some(user_id),
+            })
+            .collect();
+
+        if !new_rows.is_empty() {
+            let inserted = diesel::insert_into(document_correspondents::table)
+                .values(&new_rows)
+                .on_conflict_do_nothing()
+                .execute(conn)?;
+            if inserted > 0 {
+                changed = true;
+            }
+        }
+
+        if changed {
+            diesel::update(documents::table.find(document_id))
+                .set(documents::updated_at.eq(Utc::now().naive_utc()))
+                .execute(conn)?;
+        }
+
+        Ok(())
+    })?;
+
+    Ok(StatusCode::NO_CONTENT)
+}
+
+pub async fn remove_correspondent(
+    State(state): State<AppState>,
+    Path((document_id, correspondent_id)): Path<(Uuid, Uuid)>,
+    Query(query): Query<CorrespondentRoleQuery>,
+) -> AppResult<impl IntoResponse> {
+    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 mut conn = state.db()?;
+    let document: Document = documents::table.find(document_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::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))
+        .set(documents::updated_at.eq(Utc::now().naive_utc()))
+        .execute(&mut conn)?;
+
+    Ok(StatusCode::NO_CONTENT)
+}
+
 pub async fn assign_tags(
     State(state): State<AppState>,
     Path(document_id): Path<Uuid>,
@@ -975,7 +1176,10 @@ async fn process_upload(
             }
 
             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, version.id).await?;
             let version_response = to_version_response(version.clone());
@@ -993,6 +1197,7 @@ async fn process_upload(
                         user_id,
                         document,
                         tags,
+                        correspondents,
                         Some((version_response, assets)),
                     )?,
                 },
@@ -1068,6 +1273,7 @@ async fn process_upload(
             user_id,
             document,
             None,
+            Vec::new(),
             Some((to_version_response(version.clone()), Vec::new())),
         )?,
     };
@@ -1126,6 +1332,40 @@ pub(crate) fn load_tags_for_documents(
     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,
     documents: &[Document],
@@ -1191,6 +1431,7 @@ pub(crate) fn to_document_response(
     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 {
@@ -1221,6 +1462,7 @@ pub(crate) fn to_document_response(
             .into_iter()
             .map(TagResponse::from)
             .collect(),
+        correspondents,
         current_version,
     })
 }