455 lines
15 KiB
Rust
455 lines
15 KiB
Rust
use std::path::{Path, PathBuf};
|
|
use std::sync::Arc;
|
|
use std::time::SystemTime;
|
|
|
|
use askama::Template;
|
|
use axum::extract::{Path as AxumPath, State};
|
|
use axum::response::{Html, Redirect};
|
|
use time::format_description::well_known::Rfc3339;
|
|
use time::OffsetDateTime;
|
|
use tokio::io::AsyncWriteExt;
|
|
use tracing::{info, warn};
|
|
|
|
use crate::analyze::{AnalysisInput, AnalyzeJob, AnalyzeSender, RecordingInput};
|
|
use crate::auth::AuthenticatedWebUser;
|
|
use crate::config::Config;
|
|
use crate::error::AppError;
|
|
use crate::paths::{read_delete_marker, write_delete_marker, DeleteMarker, DELETE_MARKER};
|
|
use crate::routes::user_web::locate_case;
|
|
|
|
#[derive(Template)]
|
|
#[template(path = "document.html")]
|
|
struct DocumentTemplate {
|
|
case_id: String,
|
|
version: u32,
|
|
content: String,
|
|
}
|
|
|
|
/// POST /web/cases/{case_id}/analyze
|
|
///
|
|
/// Trigger an LLM analysis for the case. If no document exists yet, writes
|
|
/// `analysis_input_v1.json`; otherwise writes `analysis_input_v{N+1}.json`
|
|
/// where N is the current latest document version. The worker picks up the
|
|
/// new input and produces `document_v{version}.md`. Available to all logged-in
|
|
/// users (no admin check). Race protection via `create_new` on the input file.
|
|
pub async fn handle_analyze_case(
|
|
user: AuthenticatedWebUser,
|
|
State(config): State<Arc<Config>>,
|
|
State(analyze_tx): State<AnalyzeSender>,
|
|
AxumPath(case_id): AxumPath<String>,
|
|
) -> Result<Redirect, AppError> {
|
|
uuid::Uuid::parse_str(&case_id)
|
|
.map_err(|_| AppError::BadRequest("Invalid case_id".into()))?;
|
|
|
|
if !config.llm_configured() {
|
|
return Err(AppError::ServiceUnavailable(
|
|
"LLM-Analyse nicht konfiguriert".into(),
|
|
));
|
|
}
|
|
|
|
let user_root = config.data_path.join(&user.slug);
|
|
let case_dir = match locate_case(&user_root, &case_id).await {
|
|
Some(p) => p,
|
|
None => {
|
|
warn!(slug = %user.slug, case_id = %case_id, "analyze: case not found");
|
|
return Err(AppError::NotFound("Case not found".into()));
|
|
}
|
|
};
|
|
|
|
let version = find_latest_document(&case_dir)
|
|
.await
|
|
.map(|(v, _)| v + 1)
|
|
.unwrap_or(1);
|
|
|
|
let input = build_analysis_input(&case_dir, version).await?;
|
|
let input_path = case_dir.join(format!("analysis_input_v{version}.json"));
|
|
write_input_create_new(&input_path, &input).await?;
|
|
|
|
// Unbounded send: only fails if the worker has gone away (programmer error
|
|
// or shutdown race). Log but don't fail the request — the file is already
|
|
// on disk and recovery will pick it up on next startup.
|
|
if analyze_tx
|
|
.send(AnalyzeJob {
|
|
case_dir: case_dir.clone(),
|
|
version,
|
|
})
|
|
.is_err()
|
|
{
|
|
warn!(case_id = %case_id, "analyze send failed (worker gone)");
|
|
}
|
|
|
|
info!(
|
|
slug = %user.slug,
|
|
case_id = %case_id,
|
|
version,
|
|
recordings = input.recordings.len(),
|
|
"analyze requested; analysis enqueued"
|
|
);
|
|
|
|
Ok(Redirect::to("/web/cases"))
|
|
}
|
|
|
|
/// GET /web/cases/{case_id}/document
|
|
///
|
|
/// Render the latest `document_v{N}.md` for the case. The handler scans the
|
|
/// case directory and picks the highest N — no symlink, no separate pointer.
|
|
pub async fn handle_document_view(
|
|
user: AuthenticatedWebUser,
|
|
State(config): State<Arc<Config>>,
|
|
AxumPath(case_id): AxumPath<String>,
|
|
) -> Result<Html<String>, AppError> {
|
|
uuid::Uuid::parse_str(&case_id)
|
|
.map_err(|_| AppError::BadRequest("Invalid case_id".into()))?;
|
|
|
|
let user_root = config.data_path.join(&user.slug);
|
|
let case_dir = match locate_case(&user_root, &case_id).await {
|
|
Some(p) => p,
|
|
None => {
|
|
warn!(slug = %user.slug, case_id = %case_id, "document view: case not found");
|
|
return Err(AppError::NotFound("Case not found".into()));
|
|
}
|
|
};
|
|
|
|
let (version, content) = find_latest_document(&case_dir)
|
|
.await
|
|
.ok_or_else(|| AppError::NotFound("Noch kein Dokument erzeugt".into()))?;
|
|
|
|
DocumentTemplate {
|
|
case_id,
|
|
version,
|
|
content,
|
|
}
|
|
.render()
|
|
.map(Html)
|
|
.map_err(|e| AppError::Internal(format!("Template render failed: {e}")))
|
|
}
|
|
|
|
/// Build an `AnalysisInput` for the given case directory at the given version.
|
|
/// Collects all `.m4a` (excluding `.m4a.failed`), reads the matching
|
|
/// `.transcript.txt` for each, skips blank transcripts, and computes
|
|
/// `last_recording_mtime` as the max mtime over ALL `.m4a` (including blank
|
|
/// ones — a late blank addendum still counts as activity).
|
|
pub(crate) async fn build_analysis_input(
|
|
case_dir: &Path,
|
|
version: u32,
|
|
) -> Result<AnalysisInput, AppError> {
|
|
let m4as = collect_m4as(case_dir).await?;
|
|
if m4as.is_empty() {
|
|
return Err(AppError::BadRequest("Keine Aufnahmen im Fall".into()));
|
|
}
|
|
|
|
let mut recordings: Vec<RecordingInput> = Vec::new();
|
|
let mut last_mtime: SystemTime = SystemTime::UNIX_EPOCH;
|
|
for (m4a_path, mtime) in &m4as {
|
|
if *mtime > last_mtime {
|
|
last_mtime = *mtime;
|
|
}
|
|
|
|
let transcript_path = m4a_path.with_extension("transcript.txt");
|
|
let text = tokio::fs::read_to_string(&transcript_path)
|
|
.await
|
|
.map_err(|_| {
|
|
AppError::BadRequest("Nicht alle Aufnahmen sind transkribiert".into())
|
|
})?;
|
|
|
|
if text.trim().is_empty() {
|
|
continue;
|
|
}
|
|
|
|
let stem = m4a_path.file_stem().and_then(|s| s.to_str()).unwrap_or("");
|
|
recordings.push(RecordingInput {
|
|
recorded_at: filename_stem_to_recorded_at(stem),
|
|
text,
|
|
});
|
|
}
|
|
|
|
// Silent-only case: `recordings` may be empty here. The analyze worker
|
|
// handles that by writing a stub document instead of calling the LLM.
|
|
let last_recording_mtime = OffsetDateTime::from(last_mtime)
|
|
.format(&Rfc3339)
|
|
.map_err(|e| AppError::Internal(format!("format mtime: {e}")))?;
|
|
|
|
Ok(AnalysisInput {
|
|
version,
|
|
last_recording_mtime,
|
|
recordings,
|
|
})
|
|
}
|
|
|
|
/// Return `(path, mtime)` for every `.m4a` file directly under `case_dir`,
|
|
/// sorted by filename (chronological, since filenames are UTC timestamps).
|
|
/// `.m4a.failed` entries are skipped.
|
|
async fn collect_m4as(case_dir: &Path) -> Result<Vec<(PathBuf, SystemTime)>, AppError> {
|
|
let mut out = Vec::new();
|
|
let mut entries = tokio::fs::read_dir(case_dir).await?;
|
|
while let Some(entry) = entries.next_entry().await? {
|
|
let filename = match entry.file_name().into_string() {
|
|
Ok(s) => s,
|
|
Err(_) => continue,
|
|
};
|
|
if !filename.ends_with(".m4a") {
|
|
continue;
|
|
}
|
|
let path = entry.path();
|
|
let mtime = entry.metadata().await?.modified()?;
|
|
out.push((path, mtime));
|
|
}
|
|
out.sort_by(|a, b| a.0.cmp(&b.0));
|
|
Ok(out)
|
|
}
|
|
|
|
/// Serialize `input` and write it to `path` using `create_new` to atomically
|
|
/// reserve the destination. Returns `Conflict` if another close already won
|
|
/// the race.
|
|
///
|
|
/// Trade-off: if the process crashes between `open` and `write_all`/`sync_all`,
|
|
/// a zero-byte or partially-written JSON remains. The worker will fail to
|
|
/// deserialize it and log an error — manual cleanup required. For MVP the
|
|
/// simpler check-and-create is acceptable.
|
|
pub(crate) async fn write_input_create_new(
|
|
path: &Path,
|
|
input: &AnalysisInput,
|
|
) -> Result<(), AppError> {
|
|
let bytes = serde_json::to_vec_pretty(input)
|
|
.map_err(|e| AppError::Internal(format!("serialize input: {e}")))?;
|
|
|
|
let mut f = match tokio::fs::OpenOptions::new()
|
|
.write(true)
|
|
.create_new(true)
|
|
.open(path)
|
|
.await
|
|
{
|
|
Ok(f) => f,
|
|
Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
|
|
return Err(AppError::Conflict(
|
|
"Analyse läuft bereits für diesen Fall".into(),
|
|
));
|
|
}
|
|
Err(e) => return Err(AppError::Internal(e.to_string())),
|
|
};
|
|
f.write_all(&bytes).await?;
|
|
f.sync_all().await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Reverse the filesystem-safe timestamp mapping applied in `upload.rs`:
|
|
/// `:` in the time portion was replaced with `-`. Split at `T`, undo hyphens
|
|
/// only in the time part to avoid touching date hyphens.
|
|
fn filename_stem_to_recorded_at(stem: &str) -> String {
|
|
match stem.split_once('T') {
|
|
Some((date, time)) => format!("{}T{}", date, time.replace('-', ":")),
|
|
None => stem.to_string(),
|
|
}
|
|
}
|
|
|
|
/// Scan `case_dir` for `document_v{N}.md`, pick the highest N, return its
|
|
/// content together with N. Returns `None` if no document exists.
|
|
pub(crate) async fn find_latest_document(case_dir: &Path) -> Option<(u32, String)> {
|
|
let mut highest: Option<u32> = None;
|
|
let mut entries = tokio::fs::read_dir(case_dir).await.ok()?;
|
|
while let Ok(Some(entry)) = entries.next_entry().await {
|
|
let name = entry.file_name();
|
|
let Some(name_str) = name.to_str() else { continue };
|
|
let Some(version) = parse_document_version(name_str) else {
|
|
continue;
|
|
};
|
|
highest = match highest {
|
|
Some(h) if h >= version => Some(h),
|
|
_ => Some(version),
|
|
};
|
|
}
|
|
let v = highest?;
|
|
let content = tokio::fs::read_to_string(case_dir.join(format!("document_v{v}.md")))
|
|
.await
|
|
.ok()?;
|
|
Some((v, content))
|
|
}
|
|
|
|
fn parse_document_version(filename: &str) -> Option<u32> {
|
|
let rest = filename.strip_prefix("document_v")?;
|
|
let num = rest.strip_suffix(".md")?;
|
|
if num.is_empty() || (num.starts_with('0') && num.len() > 1) {
|
|
return None;
|
|
}
|
|
num.parse::<u32>().ok()
|
|
}
|
|
|
|
/// POST /web/cases/{case_id}/delete
|
|
///
|
|
/// Soft-delete: writes a `.deleted` JSON marker into the case directory.
|
|
/// The case immediately disappears from listings and detail views (404).
|
|
/// Reversible via `handle_undo_delete` while the marker still has the
|
|
/// most recent `deleted_at` timestamp among the user's deleted cases.
|
|
pub async fn handle_delete_case(
|
|
user: AuthenticatedWebUser,
|
|
State(config): State<Arc<Config>>,
|
|
AxumPath(case_id): AxumPath<String>,
|
|
) -> Result<Redirect, AppError> {
|
|
uuid::Uuid::parse_str(&case_id)
|
|
.map_err(|_| AppError::BadRequest("Invalid case_id".into()))?;
|
|
|
|
let user_root = config.data_path.join(&user.slug);
|
|
let case_dir = match locate_case(&user_root, &case_id).await {
|
|
Some(p) => p,
|
|
None => {
|
|
warn!(slug = %user.slug, case_id = %case_id, "delete: case not found");
|
|
return Err(AppError::NotFound("Case not found".into()));
|
|
}
|
|
};
|
|
|
|
let marker = DeleteMarker {
|
|
batch: uuid::Uuid::new_v4(),
|
|
deleted_at: now_rfc3339()?,
|
|
};
|
|
write_delete_marker(&case_dir, &marker).await?;
|
|
|
|
info!(
|
|
slug = %user.slug,
|
|
case_id = %case_id,
|
|
batch = %marker.batch,
|
|
"case soft-deleted"
|
|
);
|
|
|
|
Ok(Redirect::to("/web/cases"))
|
|
}
|
|
|
|
/// POST /web/cases/undo-delete
|
|
///
|
|
/// Restore every case in the most recent delete batch by removing its
|
|
/// `.deleted` marker. "Most recent" = highest `deleted_at` across all
|
|
/// markers under the user's data directory; ties broken by `batch`-UUID
|
|
/// lexicographic order. No-op if no markers exist.
|
|
pub async fn handle_undo_delete(
|
|
user: AuthenticatedWebUser,
|
|
State(config): State<Arc<Config>>,
|
|
) -> Result<Redirect, AppError> {
|
|
let user_root = config.data_path.join(&user.slug);
|
|
|
|
let latest = latest_delete_batch(&user_root).await;
|
|
let Some((batch, _)) = latest else {
|
|
info!(slug = %user.slug, "undo-delete: nothing to undo");
|
|
return Ok(Redirect::to("/web/cases"));
|
|
};
|
|
|
|
let restored = restore_batch(&user_root, batch).await;
|
|
info!(slug = %user.slug, batch = %batch, restored, "undo-delete completed");
|
|
|
|
Ok(Redirect::to("/web/cases"))
|
|
}
|
|
|
|
/// Summarize the most recent delete batch under `user_root`: returns
|
|
/// `(batch_uuid, count)` where count is the number of cases sharing that
|
|
/// batch. `None` if no `.deleted` markers exist. Single directory scan.
|
|
pub(crate) async fn summarize_latest_batch(user_root: &Path) -> Option<(uuid::Uuid, usize)> {
|
|
let (batch, _) = latest_delete_batch(user_root).await?;
|
|
let mut count = 0usize;
|
|
let mut entries = tokio::fs::read_dir(user_root).await.ok()?;
|
|
while let Ok(Some(entry)) = entries.next_entry().await {
|
|
let case_path = entry.path();
|
|
if !case_path.is_dir() {
|
|
continue;
|
|
}
|
|
if let Some(marker) = read_delete_marker(&case_path).await
|
|
&& marker.batch == batch
|
|
{
|
|
count += 1;
|
|
}
|
|
}
|
|
Some((batch, count))
|
|
}
|
|
|
|
/// Scan the user's data dir, return the `(batch, deleted_at)` of the
|
|
/// most-recently-deleted case across all markers. `None` if no markers.
|
|
async fn latest_delete_batch(user_root: &Path) -> Option<(uuid::Uuid, String)> {
|
|
let mut entries = tokio::fs::read_dir(user_root).await.ok()?;
|
|
let mut best: Option<(uuid::Uuid, String)> = None;
|
|
while let Ok(Some(entry)) = entries.next_entry().await {
|
|
if !entry.path().is_dir() {
|
|
continue;
|
|
}
|
|
let Some(marker) = read_delete_marker(&entry.path()).await else {
|
|
continue;
|
|
};
|
|
let candidate = (marker.batch, marker.deleted_at);
|
|
best = match best {
|
|
None => Some(candidate),
|
|
// Order: deleted_at desc, then batch desc as tie-breaker.
|
|
Some(ref cur)
|
|
if candidate.1 > cur.1
|
|
|| (candidate.1 == cur.1 && candidate.0 > cur.0) =>
|
|
{
|
|
Some(candidate)
|
|
}
|
|
other => other,
|
|
};
|
|
}
|
|
best
|
|
}
|
|
|
|
/// Remove `.deleted` markers from every case whose marker carries `batch`.
|
|
/// Returns the number of restored cases.
|
|
async fn restore_batch(user_root: &Path, batch: uuid::Uuid) -> usize {
|
|
let mut count = 0;
|
|
let mut entries = match tokio::fs::read_dir(user_root).await {
|
|
Ok(r) => r,
|
|
Err(_) => return 0,
|
|
};
|
|
while let Ok(Some(entry)) = entries.next_entry().await {
|
|
let case_path = entry.path();
|
|
if !case_path.is_dir() {
|
|
continue;
|
|
}
|
|
let Some(marker) = read_delete_marker(&case_path).await else {
|
|
continue;
|
|
};
|
|
if marker.batch != batch {
|
|
continue;
|
|
}
|
|
if let Err(e) = tokio::fs::remove_file(case_path.join(DELETE_MARKER)).await {
|
|
warn!(case_dir = ?case_path, error = %e, "failed to remove .deleted marker");
|
|
continue;
|
|
}
|
|
count += 1;
|
|
}
|
|
count
|
|
}
|
|
|
|
fn now_rfc3339() -> Result<String, AppError> {
|
|
OffsetDateTime::now_utc()
|
|
.format(&Rfc3339)
|
|
.map_err(|e| AppError::Internal(format!("format timestamp: {e}")))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn filename_stem_roundtrip() {
|
|
assert_eq!(
|
|
filename_stem_to_recorded_at("2026-04-15T10-32-00Z"),
|
|
"2026-04-15T10:32:00Z"
|
|
);
|
|
// Date-only hyphens are preserved.
|
|
assert_eq!(
|
|
filename_stem_to_recorded_at("2026-04-15T10-32-00.123Z"),
|
|
"2026-04-15T10:32:00.123Z"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn parses_valid_document_names() {
|
|
assert_eq!(parse_document_version("document_v1.md"), Some(1));
|
|
assert_eq!(parse_document_version("document_v42.md"), Some(42));
|
|
}
|
|
|
|
#[test]
|
|
fn rejects_invalid_document_names() {
|
|
assert_eq!(parse_document_version("document_v01.md"), None);
|
|
assert_eq!(parse_document_version("document_v.md"), None);
|
|
assert_eq!(parse_document_version("document_vx.md"), None);
|
|
assert_eq!(parse_document_version("analysis_input_v1.json"), None);
|
|
assert_eq!(parse_document_version(""), None);
|
|
}
|
|
}
|