use std::sync::Arc; use std::time::Duration; use chrono::Duration as ChronoDuration; use kvitt_server::tasks::scheduler::{ GENERATION_TIME, PRUNE_TIME, generation_job, next_run, prune_job, run_daily, }; use kvitt_server::time::{FixedClock, format_ts, parse_ts}; use tokio::sync::mpsc; use crate::common::*; use crate::m4; fn ts(text: &str) -> chrono::DateTime { parse_ts(text).unwrap() } #[test] fn next_06_00_is_local_time_across_both_dst_changes() { // Summer time (CEST, UTC+2): 06:00 is 04:00Z. assert_eq!( next_run(ts("2026-10-17T03:59:59.999Z"), GENERATION_TIME).unwrap(), ts("2026-10-17T04:00:00.000Z") ); // Strictly after: at 06:00 the next run is tomorrow. assert_eq!( next_run(ts("2026-10-17T04:00:00.000Z"), GENERATION_TIME).unwrap(), ts("2026-10-18T04:00:00.000Z") ); // Autumn: summer time ends 25 Oct 2026 at 03:00 CEST; 06:00 that day is CET, 05:00Z (a 25-hour day). assert_eq!( next_run(ts("2026-10-24T04:00:00.000Z"), GENERATION_TIME).unwrap(), ts("2026-10-25T05:00:00.000Z") ); assert_eq!( next_run(ts("2026-10-25T05:00:00.000Z"), GENERATION_TIME).unwrap(), ts("2026-10-26T05:00:00.000Z") ); // Spring: summer time starts 28 Mar 2027 at 02:00 CET; 06:00 that day is CEST, 04:00Z (a 23-hour day). assert_eq!( next_run(ts("2027-03-27T05:00:00.000Z"), GENERATION_TIME).unwrap(), ts("2027-03-28T04:00:00.000Z") ); // Pruning at 04:00 on both DST days. assert_eq!( next_run(ts("2026-10-24T02:00:00.000Z"), PRUNE_TIME).unwrap(), ts("2026-10-25T03:00:00.000Z") ); assert_eq!( next_run(ts("2027-03-27T03:00:00.000Z"), PRUNE_TIME).unwrap(), ts("2027-03-28T02:00:00.000Z") ); } /// Starts `run_daily` at 06:00 on `clock` with a 2 ms poll; every run sends the clock's time. fn daily(clock: &FixedClock, run_at_start: bool) -> mpsc::UnboundedReceiver { let (tx, rx) = mpsc::unbounded_channel(); let clock2 = clock.clone(); let shared: Arc = Arc::new(clock.clone()); tokio::spawn(async move { run_daily( shared, GENERATION_TIME, run_at_start, Duration::from_millis(2), || { let tx = tx.clone(); let now = kvitt_server::time::Clock::now_ts(&clock2); async move { let _ = tx.send(now); } }, ) .await; }); rx } async fn next(rx: &mut mpsc::UnboundedReceiver) -> Option { tokio::time::timeout(Duration::from_millis(200), rx.recv()) .await .ok() .flatten() } #[tokio::test] async fn runs_at_start_then_at_06_00_and_survives_clock_jumps() { let clock = FixedClock::new(ts("2026-10-17T03:00:00.000Z")); // 05:00 CEST let mut rx = daily(&clock, true); assert_eq!( next(&mut rx).await.as_deref(), Some("2026-10-17T03:00:00.000Z"), "run at start" ); assert_eq!(next(&mut rx).await, None, "not before 06:00"); clock.set(ts("2026-10-17T03:59:59.999Z")); assert_eq!(next(&mut rx).await, None); clock.set(ts("2026-10-17T04:00:00.000Z")); assert_eq!( next(&mut rx).await.as_deref(), Some("2026-10-17T04:00:00.000Z") ); assert_eq!(next(&mut rx).await, None, "once per day"); // Forward jump over three days (downtime, NTP after boot): one run, catch-up is the job's business. clock.set(ts("2026-10-20T12:00:00.000Z")); assert_eq!( next(&mut rx).await.as_deref(), Some("2026-10-20T12:00:00.000Z") ); assert_eq!(next(&mut rx).await, None); // Backward jump by a week: the next run is re-planned for the new 06:00, not the old one. clock.set(ts("2026-10-13T03:00:00.000Z")); assert_eq!(next(&mut rx).await, None); clock.set(ts("2026-10-13T04:00:00.000Z")); assert_eq!( next(&mut rx).await.as_deref(), Some("2026-10-13T04:00:00.000Z") ); } #[tokio::test] async fn without_run_at_start_the_first_run_is_the_next_local_time() { let clock = FixedClock::new(ts("2026-10-17T05:00:00.000Z")); let mut rx = daily(&clock, false); assert_eq!(next(&mut rx).await, None); clock.set(ts("2026-10-18T04:00:00.000Z")); assert_eq!( next(&mut rx).await.as_deref(), Some("2026-10-18T04:00:00.000Z") ); } async fn insert_template( app: &TestApp, id: &str, day: i64, start: &str, last: Option<&str>, paused: bool, deleted: bool, ) { let version = app.next_version().await; sqlx::query( "INSERT INTO recurring_templates (id, account_id, category_id, amount_ore, description, notes, payer_user_id, \ split_mode, day_of_month, start_period, end_period, paused, last_generated_period, created_by_user_id, \ updated_by_user_id, created_at, updated_at, deleted_at, server_version) \ VALUES (?, ?, ?, 1250000, 'Hyra', '', ?, 'default', ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?)", ) .bind(id) .bind(ACCOUNT_1) .bind(HOME) .bind(ADMIN_ID) .bind(day) .bind(start) .bind(i64::from(paused)) .bind(last) .bind(ADMIN_ID) .bind(ADMIN_ID) .bind("2026-06-01T00:00:00.000Z") .bind("2026-06-01T00:00:00.000Z") .bind(deleted.then_some("2026-06-02T00:00:00.000Z")) .bind(version) .execute(&app.state.db) .await .unwrap(); } async fn periods(app: &TestApp, template: &str) -> Vec { sqlx::query_scalar("SELECT recurring_period FROM expenses WHERE recurring_template_id = ? ORDER BY recurring_period") .bind(template) .fetch_all(&app.state.db) .await .unwrap() } #[tokio::test] async fn start_runs_generation_catching_up_after_downtime_and_skips_paused_and_deleted() { let mut app = m4::app().await; let active = m4::new_id(); let paused = m4::new_id(); let deleted = m4::new_id(); // Last generated July; the server was down since. Now is 17 October 14:00. insert_template(&app, &active, 15, "2026-06", Some("2026-07"), false, false).await; insert_template(&app, &paused, 1, "2026-06", None, true, false).await; insert_template(&app, &deleted, 1, "2026-06", None, false, true).await; // Missing credentials in the test config: push disabled, the scheduler still runs. let state = kvitt_server::tasks::start(app.state.clone()).await.unwrap(); app.state = state; let deadline = std::time::Instant::now() + Duration::from_secs(5); while periods(&app, &active).await.len() < 3 && std::time::Instant::now() < deadline { tokio::time::sleep(Duration::from_millis(10)).await; } assert_eq!( periods(&app, &active).await, ["2026-08", "2026-09", "2026-10"] ); assert!(periods(&app, &paused).await.is_empty()); assert!(periods(&app, &deleted).await.is_empty()); let audit: i64 = app .scalar_i64("SELECT count(*) FROM audit_log WHERE source = 'recurring'") .await; assert_eq!(audit, 3); } #[tokio::test] async fn the_06_00_tick_generates_the_period_due_that_morning() { let app = m4::app().await; let id = m4::new_id(); insert_template(&app, &id, 15, "2026-11", None, false, false).await; // 15 November 2026, 05:59 CET: not yet. app.set_now("2026-11-15T04:59:00.000Z"); generation_job(&app.state).await; assert!(periods(&app, &id).await.is_empty()); let state = app.state.clone(); let shared = state.clock.clone(); tokio::spawn(async move { run_daily( shared, GENERATION_TIME, false, Duration::from_millis(2), || generation_job(&state), ) .await; }); tokio::time::sleep(Duration::from_millis(30)).await; assert!(periods(&app, &id).await.is_empty()); app.set_now("2026-11-15T05:00:00.000Z"); let deadline = std::time::Instant::now() + Duration::from_secs(5); while periods(&app, &id).await.is_empty() && std::time::Instant::now() < deadline { tokio::time::sleep(Duration::from_millis(5)).await; } assert_eq!(periods(&app, &id).await, ["2026-11"]); let date: String = sqlx::query_scalar("SELECT expense_date FROM expenses") .fetch_one(&app.state.db) .await .unwrap(); assert_eq!(date, "2026-11-15"); } async fn insert_mutation(app: &TestApp, received_at: &str) -> String { let id = m4::new_id(); sqlx::query( "INSERT INTO sync_mutations (mutation_id, device_id, entity_type, entity_id, result, received_at) \ VALUES (?, ?, 'expense', ?, '{}', ?)", ) .bind(&id) .bind(ADMIN_DEVICE_ID) .bind(m4::new_id()) .bind(received_at) .execute(&app.state.db) .await .unwrap(); id } #[tokio::test] async fn pruning_deletes_rows_older_than_90_days_and_keeps_the_rest() { let app = m4::app().await; let now = ts(NOW); insert_mutation( &app, &format_ts(now - ChronoDuration::days(90) - ChronoDuration::milliseconds(1)), ) .await; insert_mutation(&app, "2026-01-01T00:00:00.000Z").await; let boundary = insert_mutation(&app, &format_ts(now - ChronoDuration::days(90))).await; let recent = insert_mutation(&app, &format_ts(now - ChronoDuration::days(89))).await; prune_job(&app.state).await; let left: Vec = sqlx::query_scalar("SELECT mutation_id FROM sync_mutations ORDER BY received_at") .fetch_all(&app.state.db) .await .unwrap(); assert_eq!(left, [boundary, recent]); } #[tokio::test] async fn pruning_runs_at_04_00_local() { let app = m4::app().await; // 03:59 CEST on 18 October. app.set_now("2026-10-18T01:59:00.000Z"); insert_mutation(&app, "2026-07-01T00:00:00.000Z").await; let state = app.state.clone(); let shared = state.clock.clone(); tokio::spawn(async move { run_daily(shared, PRUNE_TIME, false, Duration::from_millis(2), || { prune_job(&state) }) .await; }); tokio::time::sleep(Duration::from_millis(30)).await; assert_eq!(m4::count(&app, "sync_mutations").await, 1); app.set_now("2026-10-18T02:00:00.000Z"); let deadline = std::time::Instant::now() + Duration::from_secs(5); while m4::count(&app, "sync_mutations").await > 0 && std::time::Instant::now() < deadline { tokio::time::sleep(Duration::from_millis(5)).await; } assert_eq!(m4::count(&app, "sync_mutations").await, 0); }