Make write_input_overwrite public
This commit makes the `write_input_overwrite` function public. This is necessary because the new `handle_analyze_required` function in `bulk.rs` needs to call it. Previously, it was only used internally by `auto_trigger.rs`.
This commit is contained in:
@@ -496,7 +496,7 @@ pub(crate) async fn read_failure_marker(case_dir: &Path) -> Option<FailureMarker
|
||||
/// Auto-trigger re-analyses may legitimately replace an existing input
|
||||
/// (e.g. a new recording arrived between two reloads while the previous
|
||||
/// job was still queued). `.tmp` + fsync + rename keeps it atomic.
|
||||
async fn write_input_overwrite(path: &Path, input: &AnalysisInput) -> std::io::Result<()> {
|
||||
pub(crate) async fn write_input_overwrite(path: &Path, input: &AnalysisInput) -> std::io::Result<()> {
|
||||
let bytes = serde_json::to_vec_pretty(input)
|
||||
.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
|
||||
let tmp = path.with_extension("json.tmp");
|
||||
|
||||
+119
-13
@@ -8,8 +8,11 @@ use doctate_common::timestamp::now_rfc3339;
|
||||
use serde::Deserialize;
|
||||
use tracing::{info, warn};
|
||||
|
||||
use crate::analyze::auto_trigger::{
|
||||
self, CaseAnalysisState, write_input_overwrite,
|
||||
};
|
||||
use crate::analyze::backend::any_backend_available;
|
||||
use crate::analyze::{ANALYSIS_INPUT_FILE, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE};
|
||||
use crate::analyze::{ANALYSIS_INPUT_FILE, AnalyzeJob, AnalyzeSender};
|
||||
use crate::auth::AuthenticatedWebUser;
|
||||
use crate::config::Config;
|
||||
use crate::csrf::{CsrfForm, HasCsrfToken};
|
||||
@@ -112,18 +115,9 @@ async fn bulk_analyze(
|
||||
skipped += 1;
|
||||
continue;
|
||||
};
|
||||
// Same order as the single-case handler: drop old document first so
|
||||
// the UI doesn't keep showing "ausgewertet" during re-analysis.
|
||||
let document_path = case_dir.join(DOCUMENT_FILE);
|
||||
match tokio::fs::remove_file(&document_path).await {
|
||||
Ok(_) => {}
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
|
||||
Err(e) => {
|
||||
warn!(slug = %slug, case_id = %case_id, error = %e, "bulk-analyze: remove old document failed");
|
||||
skipped += 1;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
// The previous document.json is left in place: the UI keeps
|
||||
// rendering it as "letztes Dokument" with a stale-banner until
|
||||
// the worker overwrites it atomically on the next successful run.
|
||||
let input = match build_analysis_input(&case_dir, "").await {
|
||||
Ok(i) => i,
|
||||
Err(_) => {
|
||||
@@ -216,3 +210,115 @@ async fn bulk_reset(
|
||||
}
|
||||
info!(slug = %slug, ok, skipped, "bulk-reset completed");
|
||||
}
|
||||
|
||||
/// CSRF-validated form payload for the user-facing "Aktualisieren"-bar.
|
||||
/// No selection list — the handler operates on every case in
|
||||
/// `Required` or `Idle` state for the authenticated user.
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub struct AnalyzeRequiredForm {
|
||||
#[serde(default)]
|
||||
csrf_token: String,
|
||||
}
|
||||
|
||||
impl HasCsrfToken for AnalyzeRequiredForm {
|
||||
fn csrf_token(&self) -> &str {
|
||||
&self.csrf_token
|
||||
}
|
||||
}
|
||||
|
||||
/// POST /cases/analyze-required — manual override for the idle-trigger
|
||||
/// wait. Walks every UUID-named, non-closed case under the user root
|
||||
/// and force-enqueues those in `CaseAnalysisState::Required` or `Idle`.
|
||||
/// `Failed`, `Queued`, and `Current` cases are skipped. Unlike the
|
||||
/// admin `/cases/bulk` analyze action, this is not gated to admins —
|
||||
/// it is the user-visible "Aktualisieren"-bar above the case list.
|
||||
pub async fn handle_analyze_required(
|
||||
user: AuthenticatedWebUser,
|
||||
State(config): State<Arc<Config>>,
|
||||
State(analyze_tx): State<AnalyzeSender>,
|
||||
State(events_tx): State<EventSender>,
|
||||
CsrfForm(_form): CsrfForm<AnalyzeRequiredForm>,
|
||||
) -> Result<Redirect, AppError> {
|
||||
let user_root = config.data_path.join(&user.slug);
|
||||
|
||||
if !any_backend_available() {
|
||||
warn!(slug = %user.slug, "analyze-required: no LLM backend available, skipping");
|
||||
return Ok(Redirect::to("/cases"));
|
||||
}
|
||||
|
||||
let now = std::time::SystemTime::now();
|
||||
let idle_threshold =
|
||||
std::time::Duration::from_secs(config.analysis_idle_threshold_secs);
|
||||
|
||||
let mut ok = 0usize;
|
||||
let mut skipped = 0usize;
|
||||
|
||||
let Ok(mut entries) = tokio::fs::read_dir(&user_root).await else {
|
||||
info!(slug = %user.slug, ok, skipped, "analyze-required: no user dir");
|
||||
return Ok(Redirect::to("/cases"));
|
||||
};
|
||||
|
||||
while let Ok(Some(entry)) = entries.next_entry().await {
|
||||
let Ok(name) = entry.file_name().into_string() else {
|
||||
continue;
|
||||
};
|
||||
if uuid::Uuid::parse_str(&name).is_err() {
|
||||
continue;
|
||||
}
|
||||
let case_dir = entry.path();
|
||||
if !case_dir.is_dir() {
|
||||
continue;
|
||||
}
|
||||
if crate::paths::is_closed(&case_dir).await {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Force-enqueue both Required and Idle: the explicit user click
|
||||
// overrides the wait-for-quiet semantics. Failed/Queued/Current
|
||||
// are intentionally left alone (Failed → manual per-backend retry
|
||||
// path on the case page; Queued → already in flight; Current →
|
||||
// nothing to do).
|
||||
match auto_trigger::evaluate_state(&case_dir, idle_threshold, now).await {
|
||||
CaseAnalysisState::Required { .. } | CaseAnalysisState::Idle { .. } => {}
|
||||
_ => {
|
||||
skipped += 1;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
let input = match build_analysis_input(&case_dir, "").await {
|
||||
Ok(i) => i,
|
||||
Err(_) => {
|
||||
warn!(slug = %user.slug, case = %case_dir.display(), "analyze-required: build_analysis_input failed");
|
||||
skipped += 1;
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let input_path = case_dir.join(ANALYSIS_INPUT_FILE);
|
||||
if let Err(e) = write_input_overwrite(&input_path, &input).await {
|
||||
warn!(slug = %user.slug, case = %case_dir.display(), error = %e, "analyze-required: write input failed");
|
||||
skipped += 1;
|
||||
continue;
|
||||
}
|
||||
if analyze_tx
|
||||
.send(AnalyzeJob {
|
||||
case_dir: case_dir.clone(),
|
||||
})
|
||||
.is_err()
|
||||
{
|
||||
warn!(slug = %user.slug, case = %case_dir.display(), "analyze-required: send failed (worker gone)");
|
||||
skipped += 1;
|
||||
continue;
|
||||
}
|
||||
events::emit(
|
||||
&events_tx,
|
||||
&user.slug,
|
||||
&name,
|
||||
CaseEventKind::AnalysisQueued,
|
||||
);
|
||||
ok += 1;
|
||||
}
|
||||
|
||||
info!(slug = %user.slug, ok, skipped, "analyze-required completed");
|
||||
Ok(Redirect::to("/cases"))
|
||||
}
|
||||
|
||||
@@ -70,6 +70,10 @@ pub fn api_router() -> Router<AppState> {
|
||||
post(case_actions::handle_purge_closed),
|
||||
)
|
||||
.route("/cases/bulk", post(bulk::handle_bulk_action))
|
||||
.route(
|
||||
"/cases/analyze-required",
|
||||
post(bulk::handle_analyze_required),
|
||||
)
|
||||
.route("/audio/{user}/{case_id}/{filename}", get(web::handle_audio))
|
||||
.route("/events", get(events::handle_events))
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user