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

reanalyze removal

Browse Source
This commit is contained in:
nils
2025-10-24 01:17:39 +02:00
parent b80ec5c6ac
commit 72df3dec3b
6 changed files with 58 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/bin/admin.rs
+56 -1
View File
@@ -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 <slug>\n\
add-user-to-tenant <username> <slug> [role]\n\
remove-user-from-tenant <username> <slug>\n\
reanalyze-documents <slug>\n\
list-tenants\n\
delete-assets <slug>\n\
quickwit-create-index <slug>\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<Tenant> = tenants::table
backend/src/routes/documents.rs
-33
View File
@@ -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<BulkReanalyzeResponse>)> {
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,
backend/src/routes/mod.rs
-1
View File
@@ -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(
backend/tests/documents_flow.rs
+2 -93
View File
@@ -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<()> {