diff --git a/backend/src/bin/admin.rs b/backend/src/bin/admin.rs index 95b557c..63b2062 100644 --- a/backend/src/bin/admin.rs +++ b/backend/src/bin/admin.rs @@ -15,9 +15,12 @@ use uuid::Uuid; use backend::{ config::AppConfig, db::{self, PgPool}, + jobs::{enqueue_job, JOB_ANALYZE_DOCUMENT}, models::{DocumentAsset, DocumentAssetObject, NewUser, NewUserMembership, Tenant, User}, s3, - schema::{document_asset_objects, document_assets, tenants, user_memberships, users}, + schema::{ + document_asset_objects, document_assets, documents, tenants, user_memberships, users, + }, storage::{ObjectStorage, S3Storage, TenantStorage}, utils::tracing::init_tracing, }; @@ -83,6 +86,9 @@ enum Command { username: String, slug: String, }, + ReanalyzeDocuments { + slug: String, + }, ListTenants, DeleteAssets(String), QuickwitCreate(String), @@ -100,6 +106,7 @@ impl Command { delete-tenant \n\ add-user-to-tenant [role]\n\ remove-user-from-tenant \n\ + reanalyze-documents \n\ list-tenants\n\ delete-assets \n\ quickwit-create-index \n\ @@ -138,6 +145,9 @@ impl Command { username: args.next().ok_or_else(|| anyhow!("username required"))?, slug: args.next().ok_or_else(|| anyhow!("tenant slug required"))?, }), + Some("reanalyze-documents") => Ok(Self::ReanalyzeDocuments { + slug: args.next().ok_or_else(|| anyhow!("tenant slug required"))?, + }), Some("list-tenants") => Ok(Self::ListTenants), Some("delete-assets") => Ok(Self::DeleteAssets( args.next().ok_or_else(|| anyhow!("tenant slug required"))?, @@ -179,6 +189,7 @@ async fn main() -> Result<()> { Command::RemoveUserFromTenant { username, slug } => { remove_user_from_tenant(&pool, &username, &slug)? } + Command::ReanalyzeDocuments { slug } => reanalyze_documents(&pool, &slug)?, Command::ListTenants => list_tenants(&pool)?, Command::DeleteAssets(slug) => delete_assets_for_tenant(&config, &pool, &slug).await?, Command::QuickwitCreate(slug) => { @@ -436,6 +447,50 @@ fn remove_user_from_tenant(pool: &PgPool, username: &str, slug: &str) -> Result< Ok(()) } +fn reanalyze_documents(pool: &PgPool, slug: &str) -> Result<()> { + let mut conn = pool.get().context("failed to get database connection")?; + + let tenant: Tenant = tenants::table + .filter(tenants::slug.eq(slug)) + .first(&mut conn) + .optional()? + .ok_or_else(|| anyhow!("tenant '{}' not found", slug))?; + + let targets: Vec<(Uuid, Uuid)> = documents::table + .filter(documents::tenant_id.eq(tenant.id)) + .filter(documents::deleted_at.is_null()) + .select((documents::id, documents::current_version_id)) + .load(&mut conn)?; + + if targets.is_empty() { + println!("tenant '{}' has no active documents", slug); + return Ok(()); + } + + let mut queued = 0usize; + for (document_id, version_id) in targets { + enqueue_job( + &mut conn, + tenant.id, + JOB_ANALYZE_DOCUMENT, + serde_json::json!({ + "document_id": document_id, + "document_version_id": version_id, + "force": true, + }), + None, + ) + .map_err(|err| anyhow!("failed to enqueue analyze job: {}", err))?; + queued += 1; + } + + println!( + "queued {} documents for re-analysis in tenant '{}'", + queued, slug + ); + Ok(()) +} + fn list_tenants(pool: &PgPool) -> Result<()> { let mut conn = pool.get().context("failed to get database connection")?; let tenants: Vec = tenants::table diff --git a/backend/src/routes/documents.rs b/backend/src/routes/documents.rs index 8863f5f..5ff178a 100644 --- a/backend/src/routes/documents.rs +++ b/backend/src/routes/documents.rs @@ -789,39 +789,6 @@ pub async fn request_document_assets( Ok(StatusCode::ACCEPTED) } -pub async fn reanalyze_all_documents( - TenantScopedConn { - mut conn, - tenant_id, - .. - }: TenantScopedConn, -) -> AppResult<(StatusCode, Json)> { - let targets: Vec<(Uuid, Uuid)> = documents::table - .filter(documents::deleted_at.is_null()) - .filter(documents::tenant_id.eq(tenant_id)) - .select((documents::id, documents::current_version_id)) - .load(&mut conn)?; - - 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": true, - }), - None, - ) - .map_err(|err| AppError::internal(format!("failed to enqueue analyze job: {err}")))?; - queued += 1; - } - - Ok((StatusCode::ACCEPTED, Json(BulkReanalyzeResponse { queued }))) -} - pub async fn reanalyze_selected_documents( TenantScopedConn { mut conn, diff --git a/backend/src/routes/mod.rs b/backend/src/routes/mod.rs index f794f5d..707d696 100644 --- a/backend/src/routes/mod.rs +++ b/backend/src/routes/mod.rs @@ -58,7 +58,6 @@ pub fn create_router(state: AppState) -> Router<()> { "/", get(documents::list_documents).post(documents::upload_document), ) - .route("/reanalyze", post(documents::reanalyze_all_documents)) .route("/bulk/move", post(documents::bulk_move_documents)) .route("/bulk/tags", post(documents::bulk_update_tags)) .route( diff --git a/backend/tests/documents_flow.rs b/backend/tests/documents_flow.rs index 7a3ad5a..40e6a60 100644 --- a/backend/tests/documents_flow.rs +++ b/backend/tests/documents_flow.rs @@ -56,12 +56,12 @@ struct DocumentDownload { filename: String, } +#[derive(Deserialize)] + #[derive(Deserialize)] struct BulkReanalyze { queued: usize, } - -#[derive(Deserialize)] struct BulkMoveResult { updated: usize, } @@ -318,97 +318,6 @@ async fn duplicate_and_restore_document() -> Result<()> { Ok(()) } -#[tokio::test] -async fn bulk_reanalyze_documents() -> Result<()> { - let _lock = acquire_db_lock().await; - let app = TestApp::new().await?; - - let password = "bulkpass"; - app.insert_user("alex", password, "admin").await?; - let token = app.login_token("alex", password).await?; - - app.clear_jobs().await?; - - let first_bytes = b"first doc"; - let first = app - .upload_document( - "/api/documents", - "first.txt", - "text/plain", - first_bytes, - None, - &token, - ) - .await?; - assert_eq!(first.status(), StatusCode::CREATED); - let first_body = body_to_vec(first.into_body()).await?; - let first_detail: DocumentDetail = serde_json::from_slice(&first_body)?; - - let second_bytes = b"second doc"; - let second = app - .upload_document( - "/api/documents", - "second.txt", - "text/plain", - second_bytes, - None, - &token, - ) - .await?; - assert_eq!(second.status(), StatusCode::CREATED); - let second_body = body_to_vec(second.into_body()).await?; - let second_detail: DocumentDetail = serde_json::from_slice(&second_body)?; - - app.clear_jobs().await?; - - let response = app - .post_json( - "/api/documents/reanalyze", - &serde_json::json!({}), - Some(&token), - ) - .await?; - assert_eq!(response.status(), StatusCode::ACCEPTED); - let body = body_to_vec(response.into_body()).await?; - let bulk: BulkReanalyze = serde_json::from_slice(&body)?; - assert_eq!(bulk.queued, 2); - - let jobs = app.jobs_by_type("analyze-document").await?; - assert_eq!(jobs.len(), 2); - let mut payload_docs = Vec::new(); - for job in jobs { - let payload: AnalyzeJobPayload = serde_json::from_value(job.payload)?; - assert!(payload.force); - payload_docs.push((payload.document_id, payload.document_version_id)); - } - - let mut expected = vec![ - ( - first_detail.document.id, - first_detail - .document - .current_version - .as_ref() - .expect("first current version") - .id, - ), - ( - second_detail.document.id, - second_detail - .document - .current_version - .as_ref() - .expect("second current version") - .id, - ), - ]; - payload_docs.sort(); - expected.sort(); - assert_eq!(payload_docs, expected); - - app.cleanup().await?; - Ok(()) -} #[tokio::test] async fn bulk_move_documents_to_folder() -> Result<()> { diff --git a/docs/api.txt b/docs/api.txt index 563e2cf..21084fb 100644 --- a/docs/api.txt +++ b/docs/api.txt @@ -18,7 +18,6 @@ Documents --------- - GET /api/documents - List or search documents. Optional filters: `folder_id` (defaults to root when omitted), `include_deleted`, `include_descendants` (defaults to true when a `folder_id` is provided and no other override is supplied), `query` (Quickwit full-text), `tags` (comma-separated tag UUIDs), and `correspondents` (comma-separated correspondent UUIDs). Each entry includes tags, correspondent assignments, and current version info. - POST /api/documents - Upload a document via multipart form-data (`file`, optional metadata/folder fields). -- POST /api/documents/reanalyze - Queue re-analysis for every non-deleted document. - POST /api/documents/bulk/move - Move multiple documents to a target folder. - POST /api/documents/bulk/tags - Add or remove tags across multiple documents. - POST /api/documents/bulk/correspondents - Bulk correspondent actions. Default `action=add` replaces existing assignments for the provided roles before adding the supplied correspondents; `action=remove` drops the specified correspondent/role pairs. diff --git a/frontend/src/index.jsx b/frontend/src/index.jsx index 8ae6011..8f878da 100644 --- a/frontend/src/index.jsx +++ b/frontend/src/index.jsx @@ -4070,26 +4070,6 @@ const AppLayout = () => { } }, [appDispatch, setStatusMessage]); - const handleBulkReanalyze = useCallback(async () => { - try { - setLoading(true); - const { data } = await api.post('/documents/reanalyze'); - const total = data?.queued ?? 0; - const suffix = total === 1 ? '' : 's'; - setStatusMessage( - `Queued re-analysis for ${total} document${suffix}.`, - 'success', - ); - } catch (error) { - const message = - error.response?.data?.error || 'Failed to queue document re-analysis.'; - notifyApiError(error, message); - } finally { - setLoading(false); - } - }, [notifyApiError, setStatusMessage]); - - const folderClickHandlers = { onToggle: async (folderId) => { const node = folderNodes.get(folderId); @@ -4600,7 +4580,6 @@ const AppLayout = () => { status, setStatusMessage, dropOverlayState, - handleBulkReanalyze, handleLogout, sidebarProps, tags, @@ -4638,7 +4617,6 @@ const AppLayout = () => { status, setStatusMessage, dropOverlayState, - handleBulkReanalyze, handleLogout, sidebarProps, tags, @@ -4757,13 +4735,6 @@ const AppLayout = () => { )}
-