feat: lazy retention sweep on /web/cases
New module retention::sweep_user_retention is invoked at the top of handle_my_cases, using the current user's retention policy. Per case: step 1 auto-closes when the newest .m4a mtime is older than auto_close_days; step 2 auto-purges when closed_at is older than auto_delete_days. Order matters - a case auto-closed in the same sweep has closed_at ~= now and naturally survives the purge check, preserving the full grace period after a vacation. Race guard: purge re-reads the close marker immediately before remove_dir_all so a concurrent upload that reopens the case cancels the purge. Every automatic action emits an info! line with reason and threshold; failures warn but never abort the sweep. Seven integration tests cover both threshold paths, the disable-via-0 escape hatch, the newest-recording rule, the vacation scenario, and the open-case skip. Tests age the comparison objects (mtime via filetime, closed_at as handwritten RFC3339) instead of faking "now" - no clock abstraction needed.
This commit is contained in:
@@ -8,6 +8,7 @@ pub mod gazetteer;
|
||||
pub mod magic_link;
|
||||
pub mod models;
|
||||
pub mod paths;
|
||||
pub mod retention;
|
||||
pub mod routes;
|
||||
pub mod transcribe;
|
||||
pub mod web_session;
|
||||
|
||||
@@ -0,0 +1,194 @@
|
||||
//! Lazy retention sweep: runs at the top of `scan_user_cases` for the
|
||||
//! signed-in user. Auto-closes stale cases and auto-purges old closed
|
||||
//! cases — no background scheduler, no cron. The natural "someone
|
||||
//! opened the case list" trigger is sufficient for doctate's single-
|
||||
//! digit-user scale.
|
||||
//!
|
||||
//! Logging is deliberate per `feedback_log_autonomous_server_actions`:
|
||||
//! every automatic action — and every race-guarded skip — emits an
|
||||
//! `info!` line with the reason and the applied thresholds.
|
||||
//!
|
||||
//! Sweep order per case: auto-close first, auto-purge second. A case
|
||||
//! that was just auto-closed has `closed_at ≈ now` and naturally
|
||||
//! survives the purge check — giving the user the full `auto_delete_days`
|
||||
//! grace period even after a long absence.
|
||||
|
||||
use std::path::Path;
|
||||
|
||||
use time::OffsetDateTime;
|
||||
use time::format_description::well_known::Rfc3339;
|
||||
use tokio::fs;
|
||||
use tracing::{info, warn};
|
||||
|
||||
use doctate_common::timestamp::now_rfc3339;
|
||||
|
||||
use crate::config::RetentionUserSettings;
|
||||
use crate::events::{self, CaseEventKind, EventSender};
|
||||
use crate::paths::{CloseMarker, is_closed, read_close_marker, write_close_marker};
|
||||
|
||||
/// Scan `user_root` once, applying auto-close and auto-purge under the
|
||||
/// user's retention policy. Idempotent and best-effort: every I/O
|
||||
/// failure is logged but never blocks the remaining cases or the
|
||||
/// enclosing list view.
|
||||
pub async fn sweep_user_retention(
|
||||
user_root: &Path,
|
||||
slug: &str,
|
||||
settings: &RetentionUserSettings,
|
||||
events_tx: &EventSender,
|
||||
) {
|
||||
if settings.auto_close_days == 0 && settings.auto_delete_days == 0 {
|
||||
return;
|
||||
}
|
||||
|
||||
let now = OffsetDateTime::now_utc();
|
||||
|
||||
let mut entries = match fs::read_dir(user_root).await {
|
||||
Ok(r) => r,
|
||||
// User dir may not exist yet (new user, no uploads). Nothing to sweep.
|
||||
Err(_) => return,
|
||||
};
|
||||
|
||||
while let Ok(Some(entry)) = entries.next_entry().await {
|
||||
let case_path = entry.path();
|
||||
if !case_path.is_dir() {
|
||||
continue;
|
||||
}
|
||||
|
||||
if settings.auto_close_days > 0 && !is_closed(&case_path).await {
|
||||
try_auto_close(&case_path, slug, settings.auto_close_days, now, events_tx).await;
|
||||
}
|
||||
if settings.auto_delete_days > 0 && is_closed(&case_path).await {
|
||||
try_auto_purge(&case_path, slug, settings.auto_delete_days, now, events_tx).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn try_auto_close(
|
||||
case_path: &Path,
|
||||
slug: &str,
|
||||
threshold_days: u32,
|
||||
now: OffsetDateTime,
|
||||
events_tx: &EventSender,
|
||||
) {
|
||||
let Some(latest_m4a_mtime) = newest_m4a_mtime(case_path).await else {
|
||||
// No .m4a yet → not a candidate.
|
||||
return;
|
||||
};
|
||||
let age_days = (now - latest_m4a_mtime).whole_days();
|
||||
if age_days <= threshold_days as i64 {
|
||||
return;
|
||||
}
|
||||
|
||||
let marker = CloseMarker {
|
||||
closed_at: now_rfc3339(),
|
||||
};
|
||||
let case_id = events::case_id_of(case_path);
|
||||
match write_close_marker(case_path, &marker).await {
|
||||
Ok(_) => {
|
||||
info!(
|
||||
slug = %slug,
|
||||
case_id = %case_id,
|
||||
last_recording_age_days = age_days,
|
||||
threshold_days = threshold_days,
|
||||
"auto-closed case: last recording older than threshold"
|
||||
);
|
||||
events::emit(events_tx, slug, case_id, CaseEventKind::CaseClosed);
|
||||
}
|
||||
Err(e) => {
|
||||
warn!(
|
||||
slug = %slug,
|
||||
case_id = %case_id,
|
||||
error = %e,
|
||||
"auto-close failed to write marker"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn try_auto_purge(
|
||||
case_path: &Path,
|
||||
slug: &str,
|
||||
threshold_days: u32,
|
||||
now: OffsetDateTime,
|
||||
events_tx: &EventSender,
|
||||
) {
|
||||
let Some(marker) = read_close_marker(case_path).await else {
|
||||
// Malformed marker → do not purge. Admin can manually fix.
|
||||
return;
|
||||
};
|
||||
let Ok(closed_at) = OffsetDateTime::parse(&marker.closed_at, &Rfc3339) else {
|
||||
warn!(
|
||||
slug = %slug,
|
||||
case_dir = ?case_path,
|
||||
closed_at = %marker.closed_at,
|
||||
"auto-purge: cannot parse closed_at; skipping"
|
||||
);
|
||||
return;
|
||||
};
|
||||
let age_days = (now - closed_at).whole_days();
|
||||
if age_days <= threshold_days as i64 {
|
||||
return;
|
||||
}
|
||||
|
||||
// Race-guard: re-read the marker immediately before `remove_dir_all`.
|
||||
// If a concurrent upload has cleared the marker, the case has been
|
||||
// reopened and must not be purged.
|
||||
if read_close_marker(case_path).await.is_none() {
|
||||
info!(
|
||||
slug = %slug,
|
||||
case_dir = ?case_path,
|
||||
reason = "reopened during purge",
|
||||
"auto-purge skipped"
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
let case_id = events::case_id_of(case_path);
|
||||
match fs::remove_dir_all(case_path).await {
|
||||
Ok(_) => {
|
||||
info!(
|
||||
slug = %slug,
|
||||
case_id = %case_id,
|
||||
closed_days_ago = age_days,
|
||||
threshold_days = threshold_days,
|
||||
"auto-purged case: closed marker older than threshold"
|
||||
);
|
||||
events::emit(events_tx, slug, case_id, CaseEventKind::CasePurged);
|
||||
}
|
||||
Err(e) => {
|
||||
warn!(
|
||||
slug = %slug,
|
||||
case_id = %case_id,
|
||||
error = %e,
|
||||
"auto-purge failed to remove dir"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Newest `.m4a` mtime in `case_dir`, as a UTC `OffsetDateTime`.
|
||||
/// `None` when no `.m4a` is present. `.m4a.failed` counts too — a
|
||||
/// recording that failed ffmpeg is still user activity.
|
||||
async fn newest_m4a_mtime(case_dir: &Path) -> Option<OffsetDateTime> {
|
||||
let mut entries = fs::read_dir(case_dir).await.ok()?;
|
||||
let mut newest: Option<OffsetDateTime> = None;
|
||||
while let Ok(Some(entry)) = entries.next_entry().await {
|
||||
let name = entry.file_name();
|
||||
let Some(s) = name.to_str() else { continue };
|
||||
if !s.ends_with(".m4a") && !s.ends_with(".m4a.failed") {
|
||||
continue;
|
||||
}
|
||||
let Ok(meta) = entry.metadata().await else {
|
||||
continue;
|
||||
};
|
||||
let Ok(mtime) = meta.modified() else {
|
||||
continue;
|
||||
};
|
||||
let ts = OffsetDateTime::from(mtime);
|
||||
newest = Some(match newest {
|
||||
Some(cur) if cur > ts => cur,
|
||||
_ => ts,
|
||||
});
|
||||
}
|
||||
newest
|
||||
}
|
||||
@@ -312,6 +312,15 @@ pub async fn handle_my_cases(
|
||||
auto_trigger::try_enqueue_all_for_user(&user_root, &pipeline.analyze_tx, &config, &events_tx)
|
||||
.await;
|
||||
|
||||
// Lazy retention sweep: at most one disk scan per /web/cases visit,
|
||||
// reusing the user's retention policy from users.toml. Runs before
|
||||
// the listing scan so any auto-close/auto-purge actions are
|
||||
// reflected in the same response.
|
||||
if let Some(u) = config.users.iter().find(|u| u.slug == user.slug) {
|
||||
crate::retention::sweep_user_retention(&user_root, &user.slug, &u.retention, &events_tx)
|
||||
.await;
|
||||
}
|
||||
|
||||
let busy = pipeline.analyze_busy.0.load(Ordering::Acquire);
|
||||
let cases = scan_user_cases(&config, &user.slug, busy).await;
|
||||
let total = cases.len();
|
||||
|
||||
@@ -0,0 +1,265 @@
|
||||
//! Lazy retention sweep: GET /web/cases must auto-close stale cases
|
||||
//! and auto-purge old closed cases, driven by per-user retention
|
||||
//! settings from users.toml. Tests age the *comparison objects*
|
||||
//! (m4a mtime, closed_at string) instead of faking "now" — gives
|
||||
//! realistic behavioural coverage without a clock abstraction.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, SystemTime};
|
||||
|
||||
use axum::body::Body;
|
||||
use axum::http::{Request, StatusCode, header};
|
||||
use filetime::FileTime;
|
||||
use tower::util::ServiceExt;
|
||||
|
||||
use doctate_server::config::{Config, RetentionUserSettings, User};
|
||||
|
||||
fn unique_tmp(label: &str) -> PathBuf {
|
||||
std::env::temp_dir().join(format!(
|
||||
"doctate-retention-{label}-{}-{}",
|
||||
std::process::id(),
|
||||
uuid::Uuid::new_v4()
|
||||
))
|
||||
}
|
||||
|
||||
fn user_with_retention(slug: &str, auto_close_days: u32, auto_delete_days: u32) -> User {
|
||||
User {
|
||||
slug: slug.into(),
|
||||
api_key: format!("key-{slug}"),
|
||||
web_password: bcrypt::hash("s", 4).unwrap(),
|
||||
role: "doctor".into(),
|
||||
whisper: Default::default(),
|
||||
retention: RetentionUserSettings {
|
||||
auto_close_days,
|
||||
auto_delete_days,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
fn config_with(data_path: PathBuf, auto_close_days: u32, auto_delete_days: u32) -> Arc<Config> {
|
||||
let users = vec![user_with_retention(
|
||||
"dr_a",
|
||||
auto_close_days,
|
||||
auto_delete_days,
|
||||
)];
|
||||
let api_keys: HashMap<String, String> = users
|
||||
.iter()
|
||||
.map(|u| (u.api_key.clone(), u.slug.clone()))
|
||||
.collect();
|
||||
Arc::new(Config {
|
||||
data_path,
|
||||
users,
|
||||
api_keys,
|
||||
..Config::test_default()
|
||||
})
|
||||
}
|
||||
|
||||
fn seed_case(data_path: &Path, slug: &str, case_id: &str) -> PathBuf {
|
||||
let dir = data_path.join(slug).join(case_id);
|
||||
std::fs::create_dir_all(&dir).unwrap();
|
||||
dir
|
||||
}
|
||||
|
||||
/// Create an m4a with a specific mtime (used to simulate "recorded N
|
||||
/// days ago"). The content is meaningless; only the mtime matters.
|
||||
fn seed_recording_with_age(case_dir: &Path, ts_hms: &str, age: Duration) {
|
||||
let filename = format!("2026-04-01T{ts_hms}Z.m4a");
|
||||
let path = case_dir.join(&filename);
|
||||
std::fs::write(&path, b"audio").unwrap();
|
||||
let target = SystemTime::now() - age;
|
||||
filetime::set_file_mtime(&path, FileTime::from_system_time(target)).unwrap();
|
||||
}
|
||||
|
||||
/// Write a `.closed` marker with a specific `closed_at` string (age
|
||||
/// is simulated by handing in a historical RFC3339 stamp).
|
||||
fn write_closed_marker(case_dir: &Path, closed_at: &str) {
|
||||
let json = format!(r#"{{"closed_at":"{closed_at}"}}"#);
|
||||
std::fs::write(case_dir.join(".closed"), json).unwrap();
|
||||
}
|
||||
|
||||
fn past_rfc3339(age: Duration) -> String {
|
||||
use time::format_description::well_known::Rfc3339;
|
||||
let t = time::OffsetDateTime::now_utc() - time::Duration::seconds(age.as_secs() as i64);
|
||||
t.format(&Rfc3339).unwrap()
|
||||
}
|
||||
|
||||
async fn login(app: axum::Router, slug: &str) -> String {
|
||||
let body = format!("slug={slug}&password=s");
|
||||
let resp = app
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.method("POST")
|
||||
.uri("/web/login")
|
||||
.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
|
||||
.body(Body::from(body))
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
for v in resp.headers().get_all(header::SET_COOKIE).iter() {
|
||||
let s = v.to_str().unwrap();
|
||||
if let Some(pair) = s.split(';').next()
|
||||
&& pair.starts_with("session=")
|
||||
{
|
||||
return pair.to_string();
|
||||
}
|
||||
}
|
||||
panic!("no session cookie after login");
|
||||
}
|
||||
|
||||
/// Trigger the sweep via its natural path: GET /web/cases.
|
||||
async fn trigger_sweep(app: &axum::Router, cookie: &str) {
|
||||
let resp = app
|
||||
.clone()
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.uri("/web/cases")
|
||||
.header(header::COOKIE, cookie)
|
||||
.body(Body::empty())
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::OK);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn auto_close_fires_when_last_recording_too_old() {
|
||||
let config = config_with(unique_tmp("close-old"), 7, 0);
|
||||
let case_id = "11111111-1111-1111-1111-111111111111";
|
||||
let case_dir = seed_case(&config.data_path, "dr_a", case_id);
|
||||
seed_recording_with_age(&case_dir, "10-00-00", Duration::from_secs(8 * 86400));
|
||||
|
||||
let app = doctate_server::create_router(config);
|
||||
let cookie = login(app.clone(), "dr_a").await;
|
||||
|
||||
assert!(!case_dir.join(".closed").exists());
|
||||
trigger_sweep(&app, &cookie).await;
|
||||
assert!(
|
||||
case_dir.join(".closed").exists(),
|
||||
"stale case must be auto-closed"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn auto_close_disabled_when_days_zero() {
|
||||
// auto_close_days = 0 → sweep never closes for age; even a very
|
||||
// old recording must stay open.
|
||||
let config = config_with(unique_tmp("close-zero"), 0, 0);
|
||||
let case_id = "11111111-1111-1111-1111-111111111111";
|
||||
let case_dir = seed_case(&config.data_path, "dr_a", case_id);
|
||||
seed_recording_with_age(&case_dir, "10-00-00", Duration::from_secs(365 * 86400));
|
||||
|
||||
let app = doctate_server::create_router(config);
|
||||
let cookie = login(app.clone(), "dr_a").await;
|
||||
|
||||
trigger_sweep(&app, &cookie).await;
|
||||
assert!(
|
||||
!case_dir.join(".closed").exists(),
|
||||
"case must stay open when auto_close_days = 0"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn auto_close_respects_newest_recording() {
|
||||
// Case has one very old .m4a and one fresh .m4a — must stay open.
|
||||
let config = config_with(unique_tmp("close-newest"), 7, 0);
|
||||
let case_id = "11111111-1111-1111-1111-111111111111";
|
||||
let case_dir = seed_case(&config.data_path, "dr_a", case_id);
|
||||
seed_recording_with_age(&case_dir, "10-00-00", Duration::from_secs(30 * 86400));
|
||||
seed_recording_with_age(&case_dir, "11-00-00", Duration::from_secs(60));
|
||||
|
||||
let app = doctate_server::create_router(config);
|
||||
let cookie = login(app.clone(), "dr_a").await;
|
||||
|
||||
trigger_sweep(&app, &cookie).await;
|
||||
assert!(
|
||||
!case_dir.join(".closed").exists(),
|
||||
"recent addendum must keep the case open"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn auto_purge_removes_old_closed_case() {
|
||||
let config = config_with(unique_tmp("purge-old"), 0, 30);
|
||||
let case_id = "11111111-1111-1111-1111-111111111111";
|
||||
let case_dir = seed_case(&config.data_path, "dr_a", case_id);
|
||||
seed_recording_with_age(&case_dir, "10-00-00", Duration::from_secs(60));
|
||||
// closed_at = 31 days ago → must be purged.
|
||||
write_closed_marker(&case_dir, &past_rfc3339(Duration::from_secs(31 * 86400)));
|
||||
|
||||
let app = doctate_server::create_router(config);
|
||||
let cookie = login(app.clone(), "dr_a").await;
|
||||
|
||||
trigger_sweep(&app, &cookie).await;
|
||||
assert!(
|
||||
!case_dir.exists(),
|
||||
"old closed case must be purged (directory gone)"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn auto_purge_disabled_when_days_zero() {
|
||||
// Admin-test guarantee: auto_delete_days = 0 means closed cases
|
||||
// are retained indefinitely.
|
||||
let config = config_with(unique_tmp("purge-zero"), 0, 0);
|
||||
let case_id = "11111111-1111-1111-1111-111111111111";
|
||||
let case_dir = seed_case(&config.data_path, "dr_a", case_id);
|
||||
seed_recording_with_age(&case_dir, "10-00-00", Duration::from_secs(60));
|
||||
write_closed_marker(&case_dir, &past_rfc3339(Duration::from_secs(365 * 86400)));
|
||||
|
||||
let app = doctate_server::create_router(config);
|
||||
let cookie = login(app.clone(), "dr_a").await;
|
||||
|
||||
trigger_sweep(&app, &cookie).await;
|
||||
assert!(
|
||||
case_dir.exists(),
|
||||
"closed case must survive when auto_delete_days = 0"
|
||||
);
|
||||
assert!(case_dir.join(".closed").exists());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fresh_auto_close_is_not_immediately_purged() {
|
||||
// Vacation scenario: user returns after 40 days offline.
|
||||
// auto_close_days = 7, auto_delete_days = 30.
|
||||
// The sweep auto-closes (step 1, closed_at ≈ now), but the purge
|
||||
// check (step 2) sees closed_at ≈ now and skips — the user gets
|
||||
// the full 30-day grace period, nothing disappears silently.
|
||||
let config = config_with(unique_tmp("vacation"), 7, 30);
|
||||
let case_id = "11111111-1111-1111-1111-111111111111";
|
||||
let case_dir = seed_case(&config.data_path, "dr_a", case_id);
|
||||
seed_recording_with_age(&case_dir, "10-00-00", Duration::from_secs(40 * 86400));
|
||||
|
||||
let app = doctate_server::create_router(config);
|
||||
let cookie = login(app.clone(), "dr_a").await;
|
||||
|
||||
trigger_sweep(&app, &cookie).await;
|
||||
assert!(
|
||||
case_dir.exists(),
|
||||
"vacation: case directory must NOT be purged in the same sweep that auto-closed it"
|
||||
);
|
||||
assert!(
|
||||
case_dir.join(".closed").exists(),
|
||||
"vacation: case must be auto-closed (but retained)"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn auto_purge_skips_open_cases() {
|
||||
let config = config_with(unique_tmp("purge-skip-open"), 0, 30);
|
||||
let case_id = "11111111-1111-1111-1111-111111111111";
|
||||
let case_dir = seed_case(&config.data_path, "dr_a", case_id);
|
||||
seed_recording_with_age(&case_dir, "10-00-00", Duration::from_secs(60));
|
||||
|
||||
let app = doctate_server::create_router(config);
|
||||
let cookie = login(app.clone(), "dr_a").await;
|
||||
|
||||
trigger_sweep(&app, &cookie).await;
|
||||
assert!(
|
||||
case_dir.exists(),
|
||||
"open case without .closed marker must never be purged"
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user