//! Per-tick drains the watcher runs alongside ingest.
//!
//! These passes were previously inlined in `main.rs`; they exist because
//! a quick scan only walks recently-modified files, so any backlog of
//! rows missing a `content_hash` / `date_taken` / face detection
//! wouldn't otherwise drain except during the once-an-hour full scan.
//! Each function is bounded per call by a `*_PER_TICK` env-var cap.

use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use std::sync::{Arc, Mutex};

use chrono::{DateTime, Datelike, NaiveDate, TimeZone};
use lazy_static::lazy_static;
use log::{debug, info, warn};
use prometheus::IntGauge;

use crate::camera_tz;
use crate::content_hash;
use crate::database::models::PrecomputedReel;
use crate::database::{ExifDao, PrecomputedReelDao, UtcBackfillCandidate};
use crate::date_resolver;
use crate::face_watch;
use crate::faces;
use crate::file_types;
use crate::libraries;
use crate::tags;

/// Compute and persist content_hash for image_exif rows where it's NULL.
///
/// Bounded per call by `FACE_HASH_BACKFILL_MAX_PER_TICK` (default 2000)
/// so a watcher tick on a large legacy library doesn't block for hours
/// blake3-ing every photo at once. Subsequent scans pick up the rest.
/// For 50k+ libraries the dedicated `cargo run --bin backfill_hashes`
/// is still faster (it doesn't fight a watcher loop for the DAO mutex).
///
/// Drains unhashed image_exif rows by querying them directly, independent
/// of the filesystem walk. Quick scans only walk recently-modified files,
/// so a backlog of pre-existing unhashed rows never enters
/// `process_new_files`'s candidate set — left alone, it would only drain
/// on full scans (default once an hour). Calling this every tick keeps
/// the face-detection backlog moving regardless.
///
/// Returns the number of rows successfully backfilled this pass.
pub fn backfill_unhashed_backlog(
    library: &libraries::Library,
    exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
) -> usize {
    let cap: i64 = dotenv::var("FACE_HASH_BACKFILL_MAX_PER_TICK")
        .ok()
        .and_then(|s| s.parse().ok())
        .filter(|n: &i64| *n > 0)
        .unwrap_or(2000);

    // Fetch up to cap+1 rows so we can tell "more remain" without a
    // separate count query. Across libraries — there's no per-library
    // filter on get_rows_missing_hash today — but we only ever update
    // rows whose library_id matches the caller's library, so other
    // libraries' rows just get skipped here and picked up on the next
    // library's tick. Negligible cost given the cap.
    let rows: Vec<(i32, String)> = {
        let mut dao = exif_dao.lock().expect("Unable to lock ExifDao");
        dao.get_rows_missing_hash(cap + 1).unwrap_or_default()
    };
    if rows.is_empty() {
        return 0;
    }

    let more_than_cap = rows.len() as i64 > cap;
    let base_path = std::path::Path::new(&library.root_path);

    let mut backfilled = 0usize;
    let mut errors = 0usize;
    let mut skipped_other_lib = 0usize;
    for (lib_id, rel_path) in rows.iter().take(cap as usize) {
        if *lib_id != library.id {
            skipped_other_lib += 1;
            continue;
        }
        let abs = base_path.join(rel_path);
        if !abs.exists() {
            // File walked away — the watcher's reconciliation pass will
            // remove the orphan exif row eventually.
            continue;
        }
        match content_hash::compute(&abs) {
            Ok(id) => {
                let mut dao = exif_dao.lock().expect("Unable to lock ExifDao");
                if let Err(e) =
                    dao.backfill_content_hash(library.id, rel_path, &id.content_hash, id.size_bytes)
                {
                    warn!(
                        "face_watch: backfill_content_hash failed for {}: {:?}",
                        rel_path, e
                    );
                    errors += 1;
                } else {
                    backfilled += 1;
                }
            }
            Err(e) => {
                debug!(
                    "face_watch: hash compute failed for {} ({:?})",
                    abs.display(),
                    e
                );
                errors += 1;
            }
        }
    }

    if backfilled > 0 || errors > 0 || more_than_cap {
        info!(
            "face_watch: backfill pass for library '{}': hashed {} ({} error(s), {} skipped to other libraries; {} cap, more_remain={})",
            library.name, backfilled, errors, skipped_other_lib, cap, more_than_cap
        );
    }
    backfilled
}

/// `DATE_BACKFILL_MAX_PER_TICK` (default 500): the per-tick cap shared by
/// both date drains.
fn date_backfill_cap() -> i64 {
    dotenv::var("DATE_BACKFILL_MAX_PER_TICK")
        .ok()
        .and_then(|s| s.parse().ok())
        .filter(|n: &i64| *n > 0)
        .unwrap_or(500)
}

/// Drain image_exif rows whose `date_taken` was never resolved or was
/// resolved by the weakest fallback (`fs_time`). Runs the canonical-date
/// waterfall — exiftool batch (one subprocess for the whole tick's
/// rows) → filename regex → earliest_fs_time — and persists each
/// resolution with its source tag. Capped per tick by
/// `DATE_BACKFILL_MAX_PER_TICK` (default 500) so a 14k-row library
/// drains over a few quick-scan ticks without blocking the watcher.
///
/// kamadak-exif is intentionally skipped here: the row already has a
/// NULL date_taken because the ingest path's kamadak-exif call returned
/// nothing, and re-running it would just produce the same answer.
/// exiftool is the meaningful new attempt — it handles videos and
/// MakerNote-hosted dates kamadak can't reach.
pub fn backfill_missing_date_taken(
    library: &libraries::Library,
    exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
) -> usize {
    let cap = date_backfill_cap();

    let rows: Vec<(i32, String)> = {
        let mut dao = exif_dao.lock().expect("Unable to lock ExifDao");
        dao.get_rows_needing_date_backfill(library.id, cap + 1)
            .unwrap_or_default()
    };
    if rows.is_empty() {
        return 0;
    }

    let more_than_cap = rows.len() as i64 > cap;
    let base_path = std::path::Path::new(&library.root_path);

    // Build absolute paths and drop rows whose files no longer exist —
    // the missing-file scan in library_maintenance retires deleted rows
    // separately. Without this filter, NULL-date rows for missing files
    // would loop through the drain forever (no source can resolve them).
    let mut existing: Vec<(String, PathBuf)> = Vec::with_capacity(rows.len());
    for (_, rel_path) in rows.iter().take(cap as usize) {
        let abs = base_path.join(rel_path);
        if abs.exists() {
            existing.push((rel_path.clone(), abs));
        }
    }
    if existing.is_empty() {
        return 0;
    }

    // One exiftool subprocess for the whole batch; the resolver falls
    // through to filename / fs_time per file when exiftool can't supply
    // a date (or isn't installed at all).
    let paths: Vec<PathBuf> = existing.iter().map(|(_, p)| p.clone()).collect();
    let resolved = date_resolver::resolve_dates_batch(&paths, &HashMap::new());

    let mut backfilled = 0usize;
    let mut unresolved = 0usize;
    let mut by_source: HashMap<&'static str, usize> = HashMap::new();
    {
        let mut dao = exif_dao.lock().expect("Unable to lock ExifDao");
        for (rel_path, abs) in &existing {
            let Some(rd) = resolved.get(abs).copied() else {
                unresolved += 1;
                continue;
            };
            match dao.backfill_date_taken(
                library.id,
                rel_path,
                rd.timestamp,
                rd.source.as_str(),
                rd.date_utc,
                rd.camera_offset_minutes,
                rd.tz_source.map(|t| t.as_str()),
            ) {
                Ok(()) => {
                    backfilled += 1;
                    *by_source.entry(rd.source.as_str()).or_insert(0) += 1;
                }
                Err(e) => {
                    warn!(
                        "date_backfill: update failed for lib {} {}: {:?}",
                        library.id, rel_path, e
                    );
                }
            }
        }
    }

    if backfilled > 0 || unresolved > 0 || more_than_cap {
        info!(
            "date_backfill: library '{}': resolved {} ({:?}), {} unresolved, cap={}, more_remain={}",
            library.name, backfilled, by_source, unresolved, cap, more_than_cap
        );
    }
    backfilled
}

lazy_static! {
    /// Rows not yet on the true-instant date contract (`date_utc IS
    /// NULL`), summed across libraries. Refreshed by the watcher every
    /// tick; cutover (#120) waits on this reaching zero.
    pub static ref DATE_BACKFILL_REMAINING: IntGauge = IntGauge::new(
        "imageserver_date_backfill_remaining",
        "Rows not yet converted to the true-instant date contract"
    )
    .unwrap();
}

/// Upper bound on rows the true-instant drain pages past in one tick,
/// as a multiple of the cap. Unconvertible rows (missing files) don't
/// spend the cap, but a library full of them must not turn one tick into
/// a full-table walk.
const UTC_BACKFILL_SCAN_FACTOR: i64 = 20;

/// One row's resolved new-contract triple, pending write.
struct Conversion {
    row: UtcBackfillCandidate,
    date_utc: i64,
    camera_offset_minutes: Option<i32>,
    tz_source: camera_tz::TzSource,
}

/// What one tick of [`backfill_utc_dates`] did for one library.
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct UtcBackfillOutcome {
    /// Rows whose new-contract triple was written this tick.
    pub converted: usize,
    /// Rows paged past because their file is missing (didn't spend cap).
    pub skipped_missing: usize,
    /// Rows the waterfall couldn't convert faithfully (left for later).
    pub unresolved: usize,
    /// Converted rows whose display-zone calendar day moved.
    pub day_changed: usize,
    /// Pre-generated day reels deleted because a day they cover moved.
    pub reels_invalidated: usize,
}

/// True-instant backfill drain (ADR-0001, ticket #118): convert rows
/// still on a legacy mixed convention (`date_utc IS NULL AND date_taken
/// IS NOT NULL`) to true instants by re-running the canonical waterfall —
/// one exiftool subprocess per batch, same call shape as the #117 audit
/// preview (`resolve_dates_batch` with empty priors). Writes only the
/// new-contract triple; the legacy `date_taken` / `date_taken_source`
/// pair is frozen. Capped per tick by `DATE_BACKFILL_MAX_PER_TICK`
/// (default 500), shared with [`backfill_missing_date_taken`].
///
/// Legacy `manual` rows (overrides set before #115) aren't re-resolved
/// from the file — that would discard the operator's pick. Their legacy
/// value is the wall clock the UI sent, so they convert the same way a
/// manual edit does today (`camera_tz::manual_date_resolution` in the
/// display zone).
///
/// `cursor` is the caller's per-library resume point (last row id
/// visited), carried across ticks and wrapped to 0 at the end of the
/// table. Rows that can't convert — missing files, no exiftool, no
/// faithful resolution — are left behind the cursor instead of being
/// re-collected first every tick, so they can never stall the rows after
/// them (the "spun on the same lowest-id rows" failure of the legacy
/// drain's first version).
///
/// Day-boundary invalidation: when a converted instant falls on a
/// different display-zone calendar day than the legacy value did (as the
/// legacy `/memories` SQL rendered it), the pre-generated `day` reels
/// covering either month-day are deleted so the nightly pre-gen
/// regenerates them. Daily conversation summaries are SMS-derived and
/// don't depend on photo dates, so they're not touched.
pub fn backfill_utc_dates(
    library: &libraries::Library,
    exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
    reel_dao: &Arc<Mutex<Box<dyn PrecomputedReelDao>>>,
    display_tz: &str,
    cursor: &mut i32,
) -> UtcBackfillOutcome {
    let cap = date_backfill_cap();
    backfill_utc_dates_capped(library, exif_dao, reel_dao, display_tz, cursor, cap)
}

fn backfill_utc_dates_capped(
    library: &libraries::Library,
    exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
    reel_dao: &Arc<Mutex<Box<dyn PrecomputedReelDao>>>,
    display_tz: &str,
    cursor: &mut i32,
    cap: i64,
) -> UtcBackfillOutcome {
    let mut outcome = UtcBackfillOutcome::default();
    let base_path = std::path::Path::new(&library.root_path);

    // The watcher only calls this for online libraries; re-check here so
    // an unmounted root can never partially convert (manual rows need no
    // file and would otherwise convert while the rest waited).
    if !base_path.is_dir() {
        return outcome;
    }

    let zone_name = camera_tz::display_zone_for(Some(display_tz));
    let Ok(zone) = zone_name.parse::<chrono_tz::Tz>() else {
        return outcome;
    };
    // Without exiftool, legacy EXIF/exiftool rows would fall through the
    // empty-prior waterfall to fs_time and land a wrong instant. Leave
    // them for a deploy that has it.
    let exiftool_ok = date_resolver::exiftool_available();

    // Page by id from the resume cursor until `cap` convertible rows are
    // collected. Rows that can't convert this tick are paged past, not
    // counted against the cap.
    let mut manual: Vec<UtcBackfillCandidate> = Vec::new();
    let mut filed: Vec<(UtcBackfillCandidate, PathBuf)> = Vec::new();
    let mut scanned = 0i64;
    'paging: loop {
        let page = {
            let mut dao = exif_dao.lock().expect("Unable to lock ExifDao");
            dao.get_rows_needing_utc_backfill(library.id, *cursor, cap)
                .unwrap_or_default()
        };
        let page_len = page.len() as i64;
        for row in page {
            *cursor = row.id;
            scanned += 1;
            match row.date_taken_source.as_deref() {
                Some("manual") => manual.push(row),
                Some("exif") | Some("exiftool") if !exiftool_ok => outcome.unresolved += 1,
                _ => {
                    let abs = base_path.join(&row.rel_path);
                    if abs.exists() {
                        filed.push((row, abs));
                    } else {
                        // The missing-file scan retires these rows.
                        outcome.skipped_missing += 1;
                    }
                }
            }
            if (manual.len() + filed.len()) as i64 >= cap {
                break 'paging;
            }
        }
        if page_len < cap {
            // End of the table: wrap so next tick revisits what was left
            // behind (and anything that became convertible since).
            *cursor = 0;
            break;
        }
        if scanned >= cap * UTC_BACKFILL_SCAN_FACTOR {
            break;
        }
    }
    let more_remain = (manual.len() + filed.len()) as i64 >= cap;

    let mut conversions: Vec<Conversion> = Vec::with_capacity(manual.len() + filed.len());
    for row in manual {
        match camera_tz::manual_date_resolution(Some(&zone_name), row.date_taken) {
            Some((utc, offset)) => conversions.push(Conversion {
                row,
                date_utc: utc,
                camera_offset_minutes: Some(offset),
                tz_source: camera_tz::TzSource::Manual,
            }),
            None => outcome.unresolved += 1,
        }
    }
    if !filed.is_empty() {
        // One exiftool subprocess for the whole batch — byte-identical
        // call shape to the #117 audit preview.
        let paths: Vec<PathBuf> = filed.iter().map(|(_, p)| p.clone()).collect();
        let resolved = date_resolver::resolve_dates_batch(&paths, &HashMap::new());
        for (row, abs) in filed {
            let legacy_had_tag = matches!(
                row.date_taken_source.as_deref(),
                Some("exif") | Some("exiftool")
            );
            match resolved.get(&abs) {
                Some(rd) if !(legacy_had_tag && rd.source == date_resolver::DateSource::FsTime) => {
                    match (rd.date_utc, rd.tz_source) {
                        (Some(utc), Some(tz)) => conversions.push(Conversion {
                            row,
                            date_utc: utc,
                            camera_offset_minutes: rd.camera_offset_minutes,
                            tz_source: tz,
                        }),
                        _ => outcome.unresolved += 1,
                    }
                }
                // No result, or a row whose legacy date came from a tag
                // now resolving to fs_time (exiftool failed on it): don't
                // bake a filesystem time in over a real capture date.
                _ => {
                    debug!(
                        "utc_backfill: no faithful resolution for lib {} {}",
                        library.id, row.rel_path
                    );
                    outcome.unresolved += 1;
                }
            }
        }
    }

    let mut affected_days: HashSet<(u32, u32)> = HashSet::new();
    {
        let mut dao = exif_dao.lock().expect("Unable to lock ExifDao");
        for Conversion {
            row,
            date_utc: utc,
            camera_offset_minutes: offset,
            tz_source,
        } in &conversions
        {
            match dao.backfill_date_utc(row.id, *utc, *offset, tz_source.as_str()) {
                Ok(true) => {
                    outcome.converted += 1;
                    let before = display_day(zone, row.date_taken);
                    let after = display_day(zone, *utc);
                    if let (Some(before), Some(after)) = (before, after)
                        && before != after
                    {
                        outcome.day_changed += 1;
                        affected_days.insert((before.month(), before.day()));
                        affected_days.insert((after.month(), after.day()));
                    }
                }
                // A manual edit / ingest raced us and already wrote it.
                Ok(false) => {}
                Err(e) => warn!(
                    "utc_backfill: update failed for lib {} {}: {:?}",
                    library.id, row.rel_path, e
                ),
            }
        }
    }

    if !affected_days.is_empty() {
        outcome.reels_invalidated = invalidate_day_reels(library, reel_dao, zone, &affected_days);
    }

    if outcome.converted > 0 || outcome.unresolved > 0 || more_remain {
        info!(
            "utc_backfill: library '{}': {:?}, cap={}, more_remain={}",
            library.name, outcome, cap, more_remain
        );
    }
    outcome
}

/// Calendar day of a unix instant in the display zone. The legacy value
/// goes through here too: the legacy `/memories` SQL rendered
/// `date_taken` as an instant in the client's zone, so that is the day
/// the row's derived day data was built from.
fn display_day(zone: chrono_tz::Tz, epoch: i64) -> Option<NaiveDate> {
    zone.timestamp_opt(epoch, 0)
        .single()
        .map(|dt| dt.date_naive())
}

/// The month-day an "on this day" reel covers: the date it was generated
/// on, in the zone offset it was generated with. The pre-gen fallback
/// `reels::fixed_tz_offset` (no prefs row, no env override) stores
/// `local_minus_utc()` — seconds, not minutes — so an out-of-range offset
/// falls back to the display zone.
fn reel_covered_day(reel: &PrecomputedReel, zone: chrono_tz::Tz) -> Option<NaiveDate> {
    if reel.tz_offset_minutes.abs() <= 14 * 60 {
        DateTime::from_timestamp(reel.generated_at + reel.tz_offset_minutes as i64 * 60, 0)
            .map(|dt| dt.date_naive())
    } else {
        display_day(zone, reel.generated_at)
    }
}

/// Delete the pre-generated `day` reels (this library's and the
/// all-libraries one) covering any of `affected_days`, plus their files.
/// With the ledger row gone the nightly pre-gen's freshness gate fails
/// and it renders a new one on its normal schedule. Returns rows deleted.
fn invalidate_day_reels(
    library: &libraries::Library,
    reel_dao: &Arc<Mutex<Box<dyn PrecomputedReelDao>>>,
    zone: chrono_tz::Tz,
    affected_days: &HashSet<(u32, u32)>,
) -> usize {
    let keys = vec!["all".to_string(), library.id.to_string()];
    let mut dao = reel_dao.lock().expect("Unable to lock PrecomputedReelDao");
    let reels = match dao.list_for_span("day", &keys) {
        Ok(r) => r,
        Err(e) => {
            warn!("utc_backfill: listing day reels failed: {:?}", e);
            return 0;
        }
    };
    let stale: Vec<&PrecomputedReel> = reels
        .iter()
        .filter(|r| {
            reel_covered_day(r, zone).is_some_and(|d| affected_days.contains(&(d.month(), d.day())))
        })
        .collect();
    if stale.is_empty() {
        return 0;
    }
    let ids: Vec<i32> = stale.iter().map(|r| r.id).collect();
    match dao.delete_ids(&ids) {
        Ok(n) => {
            // Unlink too, so the on-demand cache can't serve the stale
            // render under the same key.
            for r in &stale {
                let mp4 = std::path::Path::new(&r.output_path);
                let _ = std::fs::remove_file(mp4);
                let _ = std::fs::remove_file(mp4.with_extension("json"));
            }
            info!(
                "utc_backfill: invalidated {} day reel(s) for library '{}' (re-dated days {:?})",
                n, library.name, affected_days
            );
            n
        }
        Err(e) => {
            warn!("utc_backfill: deleting stale day reels failed: {:?}", e);
            0
        }
    }
}

/// Refresh the progress surface for one online library: store its
/// remaining count in `remaining` and set [`DATE_BACKFILL_REMAINING`] to
/// the sum. Offline libraries keep their last-known count (the watcher
/// doesn't call this for them), so a flapping share doesn't make the
/// gauge dip.
pub fn refresh_utc_backfill_progress(
    library: &libraries::Library,
    exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
    remaining: &mut HashMap<i32, i64>,
) {
    let count = {
        let mut dao = exif_dao.lock().expect("Unable to lock ExifDao");
        dao.count_rows_needing_utc_backfill(library.id)
    };
    if let Ok(n) = count {
        remaining.insert(library.id, n);
        DATE_BACKFILL_REMAINING.set(remaining.values().sum());
    }
}

/// Per-tick face-detection drain. Pulls a capped batch of hashed-but-
/// unscanned image_exif rows directly via the FaceDao anti-join and
/// hands them to the existing detection pass. Runs on every tick (not
/// just full scans) so the backlog moves at quick-scan cadence.
/// Per-tick CLIP encoding drain. Mirrors `process_face_backlog`: pull
/// up to `CLIP_BACKLOG_MAX_PER_TICK` candidates with a known
/// `content_hash` but no `clip_embedding`, hand them to
/// `clip_watch::run_clip_encoding_pass` for parallel fan-out, and let
/// that module write the result back via `backfill_clip_embedding`.
///
/// Idempotent — a row stays in the candidate set until its embedding
/// lands, so a transient failure (Apollo unreachable, CUDA OOM) just
/// defers to the next tick. Permanent failures (un-decodable bytes)
/// retry every tick at this point; future Branch may add a status
/// column like face_detections has.
pub fn process_clip_backlog(
    library: &libraries::Library,
    clip_client: &crate::ai::clip_client::ClipClient,
    exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
    excluded_dirs: &[String],
) {
    if !clip_client.is_enabled() {
        return;
    }
    let cap: i64 = dotenv::var("CLIP_BACKLOG_MAX_PER_TICK")
        .ok()
        .and_then(|s| s.parse().ok())
        .filter(|n: &i64| *n > 0)
        .unwrap_or(32);

    let rows: Vec<(String, String)> = {
        let mut dao = exif_dao.lock().expect("exif dao");
        match dao.list_clip_unencoded_candidates(library.id, cap) {
            Ok(r) => r,
            Err(e) => {
                warn!(
                    "clip_watch: list_clip_unencoded_candidates failed for library '{}': {:?}",
                    library.name, e
                );
                return;
            }
        }
    };
    if rows.is_empty() {
        return;
    }

    info!(
        "clip_watch: backlog drain — encoding {} candidate(s) for library '{}' (cap={})",
        rows.len(),
        library.name,
        cap
    );

    let candidates: Vec<crate::clip_watch::ClipCandidate> = rows
        .into_iter()
        .map(
            |(rel_path, content_hash)| crate::clip_watch::ClipCandidate {
                rel_path,
                content_hash,
            },
        )
        .collect();

    crate::clip_watch::run_clip_encoding_pass(
        library,
        excluded_dirs,
        clip_client,
        Arc::clone(exif_dao),
        candidates,
    );
}

pub fn process_face_backlog(
    library: &libraries::Library,
    face_client: &crate::ai::face_client::FaceClient,
    face_dao: &Arc<Mutex<Box<dyn faces::FaceDao>>>,
    tag_dao: &Arc<Mutex<Box<dyn tags::TagDao>>>,
    excluded_dirs: &[String],
) {
    let cap: i64 = dotenv::var("FACE_BACKLOG_MAX_PER_TICK")
        .ok()
        .and_then(|s| s.parse().ok())
        .filter(|n: &i64| *n > 0)
        .unwrap_or(64);

    let rows: Vec<(String, String)> = {
        let mut dao = face_dao.lock().expect("face dao");
        match dao.list_unscanned_candidates(library.id, cap) {
            Ok(r) => r,
            Err(e) => {
                warn!(
                    "face_watch: list_unscanned_candidates failed for library '{}': {:?}",
                    library.name, e
                );
                return;
            }
        }
    };
    if rows.is_empty() {
        return;
    }

    info!(
        "face_watch: backlog drain — running detection on {} candidate(s) for library '{}' (cap={})",
        rows.len(),
        library.name,
        cap
    );

    let candidates: Vec<face_watch::FaceCandidate> = rows
        .into_iter()
        .map(|(rel_path, content_hash)| face_watch::FaceCandidate {
            rel_path,
            content_hash,
        })
        .collect();

    face_watch::run_face_detection_pass(
        library,
        excluded_dirs,
        face_client,
        Arc::clone(face_dao),
        Arc::clone(tag_dao),
        candidates,
    );
}

/// Compute content_hash for any image rows the walker just touched
/// whose stored EXIF row is still hash-less. Called from
/// `process_new_files` so freshly-ingested files don't have to wait for
/// the next standalone `backfill_unhashed_backlog` tick before face
/// detection can key on their bytes.
///
/// Cap is on **successes only**. An earlier version counted errors too,
/// so a pocket of chronically-unhashable files at the front of the
/// table (vanished mid-scan, permission denied, etc.) burned the budget
/// every tick and the rest of the backlog never advanced.
pub fn backfill_missing_content_hashes(
    files: &[(PathBuf, String)],
    library: &libraries::Library,
    exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
) {
    let image_paths: Vec<String> = files
        .iter()
        .filter(|(p, _)| !file_types::is_video_file(p))
        .map(|(_, rel)| rel.clone())
        .collect();
    if image_paths.is_empty() {
        return;
    }

    let exif_records = {
        let mut dao = exif_dao.lock().expect("Unable to lock ExifDao");
        dao.get_exif_batch(Some(library.id), &image_paths)
            .unwrap_or_default()
    };
    // Cheap lookup back from rel_path → absolute file_path so
    // content_hash::compute can read the bytes.
    let path_by_rel: HashMap<String, &PathBuf> =
        files.iter().map(|(p, rel)| (rel.clone(), p)).collect();

    let cap: usize = dotenv::var("FACE_HASH_BACKFILL_MAX_PER_TICK")
        .ok()
        .and_then(|s| s.parse().ok())
        .filter(|n: &usize| *n > 0)
        .unwrap_or(2000);

    // Count the unhashed backlog up front so we can surface "still needs
    // backfill: N" in the log — without it, a face-scan that's stuck at
    // 44% looks stalled when really it's chipping through hashes.
    let unhashed_total = exif_records
        .iter()
        .filter(|r| r.content_hash.is_none())
        .count();

    let mut backfilled = 0usize;
    let mut errors = 0usize;
    for record in &exif_records {
        if backfilled >= cap {
            break;
        }
        if record.content_hash.is_some() {
            continue;
        }
        let Some(file_path) = path_by_rel.get(&record.file_path) else {
            // Walked file went missing between the directory scan and now;
            // next tick will retry naturally.
            continue;
        };
        match content_hash::compute(file_path) {
            Ok(id) => {
                let mut dao = exif_dao.lock().expect("Unable to lock ExifDao");
                if let Err(e) = dao.backfill_content_hash(
                    library.id,
                    &record.file_path,
                    &id.content_hash,
                    id.size_bytes,
                ) {
                    warn!(
                        "face_watch: backfill_content_hash failed for {}: {:?}",
                        record.file_path, e
                    );
                    errors += 1;
                } else {
                    backfilled += 1;
                }
            }
            Err(e) => {
                debug!(
                    "face_watch: hash compute failed for {} ({:?})",
                    file_path.display(),
                    e
                );
                errors += 1;
            }
        }
    }
    // Always log when there's an unhashed backlog so an operator
    // looking at "scan stuck at 44%" can see backfill is running and
    // how much remains. Quiet only when there's nothing to do.
    if unhashed_total > 0 || backfilled > 0 || errors > 0 {
        let remaining = unhashed_total.saturating_sub(backfilled);
        info!(
            "face_watch: backfilled {}/{} content_hash for library '{}' ({} error(s); {} still need backfill; cap={})",
            backfilled, unhashed_total, library.name, errors, remaining, cap
        );
    }
}

/// Build the face-detection candidate list for a scan tick.
///
/// Returns `(rel_path, content_hash)` for every image file that has a
/// content_hash recorded in image_exif but no row in face_detections
/// yet. Re-querying image_exif here picks up rows the EXIF write loop
/// just inserted alongside any pre-existing rows the watcher walked
/// over — covers both new uploads and the initial backlog scan.
pub fn build_face_candidates(
    library: &libraries::Library,
    files: &[(PathBuf, String)],
    exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
    face_dao: &Arc<Mutex<Box<dyn faces::FaceDao>>>,
) -> Vec<face_watch::FaceCandidate> {
    // Restrict to image files; videos aren't face-scanned in v1 (kamadak
    // doesn't even register them in image_exif).
    let image_paths: Vec<String> = files
        .iter()
        .filter(|(p, _)| !file_types::is_video_file(p))
        .map(|(_, rel)| rel.clone())
        .collect();
    if image_paths.is_empty() {
        return Vec::new();
    }

    let exif_records = {
        let mut dao = exif_dao.lock().expect("Unable to lock ExifDao");
        dao.get_exif_batch(Some(library.id), &image_paths)
            .unwrap_or_default()
    };
    // rel_path → content_hash (only rows with a hash; without one we have
    // nothing to key face data against).
    let mut hash_by_path: HashMap<String, String> = HashMap::with_capacity(exif_records.len());
    for record in exif_records {
        if let Some(h) = record.content_hash {
            hash_by_path.insert(record.file_path, h);
        }
    }

    let mut candidates = Vec::new();
    let mut dao = face_dao.lock().expect("face dao");
    for rel_path in image_paths {
        let Some(hash) = hash_by_path.get(&rel_path) else {
            continue;
        };
        match dao.already_scanned(hash) {
            Ok(true) => continue,
            Ok(false) => candidates.push(face_watch::FaceCandidate {
                rel_path,
                content_hash: hash.clone(),
            }),
            Err(e) => {
                warn!("face_watch: already_scanned errored for {}: {:?}", hash, e);
            }
        }
    }
    candidates
}

#[cfg(test)]
mod tests {
    use super::*;

    use std::fs;
    use std::sync::{Arc, Mutex};

    use diesel::prelude::*;
    use tempfile::TempDir;

    use crate::database::models::{InsertImageExif, InsertLibrary};
    use crate::database::test::in_memory_db_connection;
    use crate::database::{ExifDao, SqliteExifDao, schema};
    use crate::faces::{FaceDao, SqliteFaceDao};
    use crate::libraries::Library;

    /// Everything `setup` hands back to a test: tempdir, library, shared
    /// connection, and the two DAOs. Aliased to keep clippy's
    /// type-complexity lint satisfied.
    type SetupFixture = (
        TempDir,
        Library,
        Arc<Mutex<diesel::SqliteConnection>>,
        Arc<Mutex<Box<dyn ExifDao>>>,
        Arc<Mutex<Box<dyn FaceDao>>>,
    );

    /// Build a tempdir-backed library + DAOs sharing a single in-memory
    /// SQLite connection (so cross-table joins like
    /// `list_unscanned_candidates` see consistent state).
    fn setup() -> SetupFixture {
        let tmp = TempDir::new().expect("tempdir");
        let mut conn = in_memory_db_connection();
        // Migration seeds library id=1 with a placeholder root; rewrite it
        // to point at the tempdir so `<root>/<rel_path>` resolves to real
        // files this test creates.
        diesel::update(schema::libraries::table.filter(schema::libraries::id.eq(1)))
            .set(schema::libraries::root_path.eq(tmp.path().to_string_lossy().to_string()))
            .execute(&mut conn)
            .expect("rewrite library 1 root");
        // Add a second library so cross-library skip cases have somewhere
        // to put their rows.
        diesel::insert_into(schema::libraries::table)
            .values(InsertLibrary {
                name: "other",
                root_path: "/tmp/other-test-lib",
                created_at: 0,
                enabled: true,
                excluded_dirs: None,
            })
            .execute(&mut conn)
            .expect("seed second library");

        let library = Library {
            id: 1,
            name: "main".to_string(),
            root_path: tmp.path().to_string_lossy().to_string(),
            enabled: true,
            excluded_dirs: Vec::new(),
        };
        let shared = Arc::new(Mutex::new(conn));
        let exif_dao: Arc<Mutex<Box<dyn ExifDao>>> = Arc::new(Mutex::new(Box::new(
            SqliteExifDao::from_shared(Arc::clone(&shared)),
        )));
        let face_dao: Arc<Mutex<Box<dyn FaceDao>>> = Arc::new(Mutex::new(Box::new(
            SqliteFaceDao::from_connection(Arc::clone(&shared)),
        )));
        (tmp, library, shared, exif_dao, face_dao)
    }

    fn insert_exif(
        exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
        lib_id: i32,
        rel: &str,
        content_hash: Option<&str>,
    ) {
        let mut dao = exif_dao.lock().unwrap();
        dao.store_exif(InsertImageExif {
            library_id: lib_id,
            file_path: rel.to_string(),
            camera_make: None,
            camera_model: None,
            lens_model: None,
            width: None,
            height: None,
            orientation: None,
            gps_latitude: None,
            gps_longitude: None,
            gps_altitude: None,
            focal_length: None,
            aperture: None,
            shutter_speed: None,
            iso: None,
            date_taken: None,
            created_time: 0,
            last_modified: 0,
            content_hash: content_hash.map(|s| s.to_string()),
            size_bytes: None,
            phash_64: None,
            dhash_64: None,
            date_taken_source: None,
            date_utc: None,
            camera_offset_minutes: None,
            date_taken_tz_source: None,
        })
        .expect("insert");
    }

    fn write_image(root: &std::path::Path, rel: &str, bytes: &[u8]) {
        let abs = root.join(rel);
        if let Some(parent) = abs.parent() {
            fs::create_dir_all(parent).expect("mkdir");
        }
        fs::write(abs, bytes).expect("write file");
    }

    #[test]
    fn backfill_unhashed_backlog_hashes_missing_rows_in_this_library() {
        let (tmp, library, _conn, exif_dao, _face_dao) = setup();
        write_image(tmp.path(), "a.jpg", b"alpha-bytes");
        write_image(tmp.path(), "b.jpg", b"bravo-bytes");
        insert_exif(&exif_dao, 1, "a.jpg", None);
        insert_exif(&exif_dao, 1, "b.jpg", None);

        let backfilled = backfill_unhashed_backlog(&library, &exif_dao);
        assert_eq!(backfilled, 2);

        let mut dao = exif_dao.lock().unwrap();
        let rows = dao
            .get_exif_batch(Some(1), &["a.jpg".to_string(), "b.jpg".to_string()])
            .unwrap();
        assert_eq!(rows.len(), 2);
        for r in rows {
            assert!(
                r.content_hash.is_some(),
                "row {} should have a hash",
                r.file_path
            );
        }
    }

    #[test]
    fn backfill_unhashed_backlog_skips_other_libraries_and_missing_files() {
        let (tmp, library, _conn, exif_dao, _face_dao) = setup();
        write_image(tmp.path(), "exists.jpg", b"hello");
        // Row for this library whose file is missing on disk:
        insert_exif(&exif_dao, 1, "ghost.jpg", None);
        insert_exif(&exif_dao, 1, "exists.jpg", None);
        // Row in the other library — must be skipped (different lib_id).
        insert_exif(&exif_dao, 2, "other.jpg", None);

        let backfilled = backfill_unhashed_backlog(&library, &exif_dao);
        assert_eq!(backfilled, 1, "only the existing in-library file hashes");

        let mut dao = exif_dao.lock().unwrap();
        let other = dao
            .get_exif_batch(Some(2), &["other.jpg".to_string()])
            .unwrap();
        assert_eq!(other.len(), 1);
        assert!(
            other[0].content_hash.is_none(),
            "other-library row must remain unhashed"
        );
        let ghost = dao
            .get_exif_batch(Some(1), &["ghost.jpg".to_string()])
            .unwrap();
        assert_eq!(ghost.len(), 1);
        assert!(
            ghost[0].content_hash.is_none(),
            "missing-on-disk row stays unhashed (reconciliation removes it later)"
        );
    }

    #[test]
    fn backfill_unhashed_backlog_respects_per_tick_cap() {
        // Env-var-driven cap; the function reads it on every call, so we
        // can set it just for this test and unset before returning.
        // Serial guard: tests in the same binary may share env, but each
        // backfill call re-reads — and we only care that the cap shape
        // (success count <= cap, more_remain logged) holds.
        unsafe {
            std::env::set_var("FACE_HASH_BACKFILL_MAX_PER_TICK", "2");
        }
        let (tmp, library, _conn, exif_dao, _face_dao) = setup();
        for i in 0..5 {
            let rel = format!("img_{}.jpg", i);
            write_image(tmp.path(), &rel, format!("bytes-{}", i).as_bytes());
            insert_exif(&exif_dao, 1, &rel, None);
        }

        let backfilled = backfill_unhashed_backlog(&library, &exif_dao);
        assert_eq!(backfilled, 2, "cap=2 must bound the per-tick successes");
        unsafe {
            std::env::remove_var("FACE_HASH_BACKFILL_MAX_PER_TICK");
        }
    }

    #[test]
    fn backfill_missing_content_hashes_skips_videos_and_hashed_rows() {
        let (tmp, library, _conn, exif_dao, _face_dao) = setup();
        // Two image rows (one already hashed, one not), one video.
        write_image(tmp.path(), "fresh.jpg", b"fresh-pixels");
        write_image(tmp.path(), "already.jpg", b"already-pixels");
        write_image(tmp.path(), "clip.mp4", b"video-bytes");
        insert_exif(&exif_dao, 1, "fresh.jpg", None);
        insert_exif(&exif_dao, 1, "already.jpg", Some("pre-existing-hash"));
        insert_exif(&exif_dao, 1, "clip.mp4", None);

        let files: Vec<(PathBuf, String)> = vec![
            (tmp.path().join("fresh.jpg"), "fresh.jpg".to_string()),
            (tmp.path().join("already.jpg"), "already.jpg".to_string()),
            (tmp.path().join("clip.mp4"), "clip.mp4".to_string()),
        ];
        backfill_missing_content_hashes(&files, &library, &exif_dao);

        let mut dao = exif_dao.lock().unwrap();
        let rows = dao
            .get_exif_batch(
                Some(1),
                &[
                    "fresh.jpg".to_string(),
                    "already.jpg".to_string(),
                    "clip.mp4".to_string(),
                ],
            )
            .unwrap();
        let by_path: HashMap<String, Option<String>> = rows
            .into_iter()
            .map(|r| (r.file_path, r.content_hash))
            .collect();
        assert!(
            by_path["fresh.jpg"].is_some(),
            "fresh image must get a hash"
        );
        assert_eq!(
            by_path["already.jpg"].as_deref(),
            Some("pre-existing-hash"),
            "already-hashed image left untouched"
        );
        assert!(
            by_path["clip.mp4"].is_none(),
            "video skipped (not face-scanned, no hash needed via this path)"
        );
    }

    #[test]
    fn build_face_candidates_filters_videos_unhashed_and_already_scanned() {
        let (tmp, library, _conn, exif_dao, face_dao) = setup();

        // Seed image_exif with: hashed unscanned, hashed scanned, unhashed,
        // and a video. Files don't need to exist on disk — the function
        // doesn't read them, only the DB rows.
        insert_exif(&exif_dao, 1, "fresh.jpg", Some("hash-fresh"));
        insert_exif(&exif_dao, 1, "scanned.jpg", Some("hash-scanned"));
        insert_exif(&exif_dao, 1, "unhashed.jpg", None);
        insert_exif(&exif_dao, 1, "clip.mp4", Some("hash-video"));
        // Mark `scanned.jpg`'s hash as already detected.
        {
            let mut dao = face_dao.lock().unwrap();
            dao.mark_status(1, "hash-scanned", "scanned.jpg", "no_faces", "test")
                .expect("mark scanned");
        }

        let files: Vec<(PathBuf, String)> = vec![
            (tmp.path().join("fresh.jpg"), "fresh.jpg".to_string()),
            (tmp.path().join("scanned.jpg"), "scanned.jpg".to_string()),
            (tmp.path().join("unhashed.jpg"), "unhashed.jpg".to_string()),
            (tmp.path().join("clip.mp4"), "clip.mp4".to_string()),
        ];
        let candidates = build_face_candidates(&library, &files, &exif_dao, &face_dao);

        assert_eq!(
            candidates.len(),
            1,
            "exactly fresh.jpg should be a candidate"
        );
        assert_eq!(candidates[0].rel_path, "fresh.jpg");
        assert_eq!(candidates[0].content_hash, "hash-fresh");
    }

    // --- True-instant backfill drain (ADR-0001, ticket #118) -------------

    use crate::database::SqlitePrecomputedReelDao;
    use crate::database::models::{ImageExif, InsertablePrecomputedReel};
    use chrono::NaiveDateTime;

    const NY: &str = "America/New_York";

    fn utc(y: i32, mo: u32, d: u32, h: u32, mi: u32, s: u32) -> i64 {
        NaiveDate::from_ymd_opt(y, mo, d)
            .unwrap()
            .and_hms_opt(h, mi, s)
            .unwrap()
            .and_utc()
            .timestamp()
    }

    fn reel_dao_for(
        conn: &Arc<Mutex<diesel::SqliteConnection>>,
    ) -> Arc<Mutex<Box<dyn PrecomputedReelDao>>> {
        Arc::new(Mutex::new(Box::new(
            SqlitePrecomputedReelDao::from_connection(Arc::clone(conn)),
        )))
    }

    /// Seed a row on the legacy convention: legacy pair set, new-contract
    /// triple NULL. Returns the row id.
    fn insert_legacy(
        exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
        lib_id: i32,
        rel: &str,
        date_taken: Option<i64>,
        source: Option<&str>,
    ) -> i32 {
        let mut dao = exif_dao.lock().unwrap();
        dao.store_exif(InsertImageExif {
            library_id: lib_id,
            file_path: rel.to_string(),
            camera_make: None,
            camera_model: None,
            lens_model: None,
            width: None,
            height: None,
            orientation: None,
            gps_latitude: None,
            gps_longitude: None,
            gps_altitude: None,
            focal_length: None,
            aperture: None,
            shutter_speed: None,
            iso: None,
            date_taken,
            created_time: 0,
            last_modified: 0,
            content_hash: None,
            size_bytes: None,
            phash_64: None,
            dhash_64: None,
            date_taken_source: source.map(str::to_string),
            date_utc: None,
            camera_offset_minutes: None,
            date_taken_tz_source: None,
        })
        .expect("insert")
        .id
    }

    fn row(conn: &Arc<Mutex<diesel::SqliteConnection>>, id: i32) -> ImageExif {
        let mut c = conn.lock().unwrap();
        schema::image_exif::table
            .find(id)
            .first::<ImageExif>(&mut *c)
            .expect("row")
    }

    fn remaining(exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>, lib_id: i32) -> i64 {
        exif_dao
            .lock()
            .unwrap()
            .count_rows_needing_utc_backfill(lib_id)
            .unwrap()
    }

    fn tools_available() -> bool {
        let ok = |bin: &str, arg: &str| {
            std::process::Command::new(bin)
                .arg(arg)
                .stdout(std::process::Stdio::null())
                .stderr(std::process::Stdio::null())
                .status()
                .map(|s| s.success())
                .unwrap_or(false)
        };
        ok("ffmpeg", "-version") && ok("exiftool", "-ver")
    }

    fn run_ok(cmd: &mut std::process::Command) {
        let out = cmd.output().expect("spawn fixture tool");
        assert!(
            out.status.success(),
            "fixture command failed: {}",
            String::from_utf8_lossy(&out.stderr)
        );
    }

    /// Tiny H.264 MP4 whose container `creation_time` exiftool reads as
    /// QuickTime / Track* groups (→ `utc_direct`).
    fn write_container_video(root: &std::path::Path, rel: &str, creation_time: &str) {
        run_ok(
            std::process::Command::new("ffmpeg")
                .args([
                    "-y",
                    "-loglevel",
                    "error",
                    "-f",
                    "lavfi",
                    "-i",
                    "color=c=black:s=32x32:d=0.4",
                    "-r",
                    "10",
                    "-c:v",
                    "libx264",
                    "-pix_fmt",
                    "yuv420p",
                    "-metadata",
                ])
                .arg(format!("creation_time={creation_time}"))
                .arg(root.join(rel)),
        );
    }

    /// Tiny JPEG carrying `DateTimeOriginal` + `OffsetTimeOriginal`.
    fn write_exif_jpeg(root: &std::path::Path, rel: &str, dto: &str, offset: &str) {
        let abs = root.join(rel);
        run_ok(
            std::process::Command::new("ffmpeg")
                .args([
                    "-y",
                    "-loglevel",
                    "error",
                    "-f",
                    "lavfi",
                    "-i",
                    "color=c=gray:s=16x16",
                    "-frames:v",
                    "1",
                ])
                .arg(&abs),
        );
        run_ok(
            std::process::Command::new("exiftool")
                .arg("-q")
                .arg("-overwrite_original")
                .arg(format!("-DateTimeOriginal={dto}"))
                .arg(format!("-OffsetTimeOriginal={offset}"))
                .arg(&abs),
        );
    }

    #[test]
    fn utc_backfill_converts_one_row_per_source_and_is_idempotent() {
        if !tools_available() {
            eprintln!("skipping: ffmpeg/exiftool not on PATH");
            return;
        }
        let (tmp, library, conn, exif_dao, _face_dao) = setup();
        let reel_dao = reel_dao_for(&conn);

        // exif: camera wall 22:30 at +02:00 → 20:30Z. Legacy = wall-as-UTC.
        write_exif_jpeg(tmp.path(), "IMG_0001.jpg", "2024:07:04 22:30:00", "+02:00");
        let exif_legacy = utc(2024, 7, 4, 22, 30, 0);
        let exif_id = insert_legacy(
            &exif_dao,
            1,
            "IMG_0001.jpg",
            Some(exif_legacy),
            Some("exif"),
        );

        // exiftool/container: creation_time is UTC. Legacy = the old
        // server-local interpretation (EDT, +4h) — the "videos off by 4–5
        // hours" row this drain exists to fix.
        write_container_video(tmp.path(), "clip.mp4", "2017-06-11T15:58:23Z");
        let video_legacy = utc(2017, 6, 11, 19, 58, 23);
        let video_id = insert_legacy(
            &exif_dao,
            1,
            "clip.mp4",
            Some(video_legacy),
            Some("exiftool"),
        );

        // filename (wall pattern): legacy wall-as-UTC; no evidence → server zone.
        let fname = "Screenshot_2014-06-01-20-44-50.png";
        write_image(tmp.path(), fname, b"not-a-png");
        let fname_wall: NaiveDateTime = NaiveDate::from_ymd_opt(2014, 6, 1)
            .unwrap()
            .and_hms_opt(20, 44, 50)
            .unwrap();
        let fname_legacy = fname_wall.and_utc().timestamp();
        let fname_id = insert_legacy(&exif_dao, 1, fname, Some(fname_legacy), Some("filename"));

        // fs_time: no date anywhere in the file; legacy is already a true instant.
        write_image(tmp.path(), "scan.jpg", b"no-exif-here");
        let meta = fs::metadata(tmp.path().join("scan.jpg")).unwrap();
        let fs_instant: chrono::DateTime<chrono::Utc> =
            crate::utils::earliest_fs_time(&meta).unwrap().into();
        let fs_instant = fs_instant.timestamp();
        let fs_id = insert_legacy(&exif_dao, 1, "scan.jpg", Some(fs_instant), Some("fs_time"));

        // manual (pre-#115 override): the operator's wall in the display
        // zone — never re-resolved from the file.
        write_image(tmp.path(), "edited.jpg", b"no-exif-here-either");
        let manual_legacy = utc(2020, 1, 1, 12, 0, 0);
        let manual_id = insert_legacy(
            &exif_dao,
            1,
            "edited.jpg",
            Some(manual_legacy),
            Some("manual"),
        );

        assert_eq!(remaining(&exif_dao, 1), 5);
        let out = backfill_utc_dates_capped(&library, &exif_dao, &reel_dao, NY, &mut 0, 500);
        assert_eq!(out.converted, 5, "{out:?}");
        assert_eq!(out.unresolved, 0, "{out:?}");

        let legacy = |r: &ImageExif| (r.date_taken, r.date_taken_source.clone());

        let r = row(&conn, exif_id);
        assert_eq!(r.date_utc, Some(utc(2024, 7, 4, 20, 30, 0)));
        assert_eq!(r.camera_offset_minutes, Some(120));
        assert_eq!(r.date_taken_tz_source.as_deref(), Some("offset_tag"));
        assert_eq!(legacy(&r), (Some(exif_legacy), Some("exif".into())));

        let r = row(&conn, video_id);
        assert_eq!(r.date_utc, Some(utc(2017, 6, 11, 15, 58, 23)));
        assert_eq!(r.camera_offset_minutes, None);
        assert_eq!(r.date_taken_tz_source.as_deref(), Some("utc_direct"));
        assert_eq!(legacy(&r), (Some(video_legacy), Some("exiftool".into())));

        let r = row(&conn, fname_id);
        assert_eq!(r.date_taken_tz_source.as_deref(), Some("assumed_server"));
        if let Some(zone) = crate::camera_tz::server_zone() {
            let off = crate::camera_tz::offset_minutes_for_zone(zone, fname_wall);
            assert_eq!(r.camera_offset_minutes, Some(off));
            assert_eq!(
                r.date_utc,
                Some(crate::camera_tz::wall_to_instant(fname_wall, off))
            );
        }
        assert_eq!(legacy(&r), (Some(fname_legacy), Some("filename".into())));

        let r = row(&conn, fs_id);
        assert_eq!(r.date_utc, Some(fs_instant));
        assert_eq!(r.camera_offset_minutes, None);
        assert_eq!(r.date_taken_tz_source.as_deref(), Some("utc_direct"));
        assert_eq!(legacy(&r), (Some(fs_instant), Some("fs_time".into())));

        let r = row(&conn, manual_id);
        // 2020-01-01 12:00 EST (-300) → 17:00Z.
        assert_eq!(r.date_utc, Some(utc(2020, 1, 1, 17, 0, 0)));
        assert_eq!(r.camera_offset_minutes, Some(-300));
        assert_eq!(r.date_taken_tz_source.as_deref(), Some("manual"));
        assert_eq!(legacy(&r), (Some(manual_legacy), Some("manual".into())));

        // Second pass: nothing eligible, nothing changes, progress is 0.
        let ids = [exif_id, video_id, fname_id, fs_id, manual_id];
        let snapshot = |conn: &Arc<Mutex<diesel::SqliteConnection>>| {
            ids.iter()
                .map(|id| {
                    let r = row(conn, *id);
                    (
                        r.date_taken,
                        r.date_taken_source,
                        r.date_utc,
                        r.camera_offset_minutes,
                        r.date_taken_tz_source,
                    )
                })
                .collect::<Vec<_>>()
        };
        let before = snapshot(&conn);
        let again = backfill_utc_dates_capped(&library, &exif_dao, &reel_dao, NY, &mut 0, 500);
        assert_eq!(again, UtcBackfillOutcome::default());
        assert_eq!(snapshot(&conn), before);
        let mut progress = HashMap::new();
        refresh_utc_backfill_progress(&library, &exif_dao, &mut progress);
        assert_eq!(progress.get(&1), Some(&0));
    }

    #[test]
    fn utc_backfill_leaves_null_date_rows_and_other_libraries_alone() {
        let (tmp, library, conn, exif_dao, _face_dao) = setup();
        let reel_dao = reel_dao_for(&conn);
        write_image(tmp.path(), "1500000000.jpg", b"x");
        write_image(tmp.path(), "undated.jpg", b"y");
        let ts = 1_500_000_000;
        let converted = insert_legacy(&exif_dao, 1, "1500000000.jpg", Some(ts), Some("filename"));
        // Both NULL → the legacy drain's job (its writer fills the triple).
        let undated = insert_legacy(&exif_dao, 1, "undated.jpg", None, None);
        let other = insert_legacy(&exif_dao, 2, "1500000000.jpg", Some(ts), Some("filename"));

        let out = backfill_utc_dates_capped(&library, &exif_dao, &reel_dao, NY, &mut 0, 500);
        assert_eq!(out.converted, 1);
        assert_eq!(row(&conn, converted).date_utc, Some(ts));
        assert_eq!(row(&conn, undated).date_utc, None);
        assert_eq!(row(&conn, other).date_utc, None);
        // Progress counts every row not yet on the new contract.
        assert_eq!(remaining(&exif_dao, 1), 1);
    }

    #[test]
    fn utc_backfill_respects_cap_and_missing_files_dont_spend_it() {
        let (tmp, library, conn, exif_dao, _face_dao) = setup();
        let reel_dao = reel_dao_for(&conn);
        // Missing files at the lowest ids, ahead of the present ones.
        // Timestamp filenames keep the fixture hermetic (never hit exiftool).
        let mut missing = Vec::new();
        for i in 0..3 {
            let ts = 1_400_000_000 + i;
            let rel = format!("{ts}.jpg");
            missing.push(insert_legacy(
                &exif_dao,
                1,
                &rel,
                Some(ts),
                Some("filename"),
            ));
        }
        for i in 0..5 {
            let ts = 1_500_000_000 + i;
            let rel = format!("{ts}.jpg");
            write_image(tmp.path(), &rel, b"z");
            insert_legacy(&exif_dao, 1, &rel, Some(ts), Some("filename"));
        }
        assert_eq!(remaining(&exif_dao, 1), 8);

        let mut progress = HashMap::new();
        let mut seen = Vec::new();
        // Cursor carried across ticks, as the watcher does.
        let mut cursor = 0;
        for _ in 0..4 {
            let out = backfill_utc_dates_capped(&library, &exif_dao, &reel_dao, NY, &mut cursor, 2);
            refresh_utc_backfill_progress(&library, &exif_dao, &mut progress);
            seen.push((out.converted, progress[&1]));
        }
        // Cap holds (2 per tick) and the 3 missing rows never spend it.
        assert_eq!(seen, vec![(2, 6), (2, 4), (1, 3), (0, 3)]);

        // The missing-file scan retires the rest → progress reaches 0.
        {
            let mut c = conn.lock().unwrap();
            diesel::delete(
                schema::image_exif::table.filter(schema::image_exif::id.eq_any(&missing)),
            )
            .execute(&mut *c)
            .unwrap();
        }
        refresh_utc_backfill_progress(&library, &exif_dao, &mut progress);
        assert_eq!(progress[&1], 0);
    }

    #[test]
    fn utc_backfill_unresolvable_rows_cannot_stall_the_rows_behind_them() {
        let (tmp, library, conn, exif_dao, _face_dao) = setup();
        let reel_dao = reel_dao_for(&conn);
        // Present files whose legacy date came from EXIF but that carry no
        // date anywhere now: the waterfall would only find fs_time, which
        // the drain refuses to write over a tag date. More of them than
        // the cap, all at the lowest ids.
        for i in 0..4 {
            let rel = format!("bad{i}.jpg");
            write_image(tmp.path(), &rel, b"no-exif");
            insert_legacy(&exif_dao, 1, &rel, Some(1_000_000_000), Some("exif"));
        }
        let mut good = Vec::new();
        for i in 0..2 {
            let ts = 1_500_000_000 + i;
            let rel = format!("{ts}.jpg");
            write_image(tmp.path(), &rel, b"z");
            good.push(insert_legacy(
                &exif_dao,
                1,
                &rel,
                Some(ts),
                Some("filename"),
            ));
        }

        let mut cursor = 0;
        let mut converted = 0;
        for _ in 0..4 {
            let out = backfill_utc_dates_capped(&library, &exif_dao, &reel_dao, NY, &mut cursor, 2);
            converted += out.converted;
        }
        assert_eq!(converted, 2, "the rows behind the stuck ones still drain");
        for id in good {
            assert!(row(&conn, id).date_utc.is_some());
        }
        // The unresolvable rows stay pending (visible in progress), untouched.
        assert_eq!(remaining(&exif_dao, 1), 4);
    }

    #[test]
    fn utc_backfill_skips_unavailable_library_then_resumes() {
        let (tmp, library, conn, exif_dao, _face_dao) = setup();
        let reel_dao = reel_dao_for(&conn);
        write_image(tmp.path(), "1500000000.jpg", b"z");
        let filed = insert_legacy(
            &exif_dao,
            1,
            "1500000000.jpg",
            Some(1_500_000_000),
            Some("filename"),
        );
        // A manual row needs no file — it must still wait for the mount.
        let manual = insert_legacy(
            &exif_dao,
            1,
            "edited.jpg",
            Some(utc(2020, 1, 1, 12, 0, 0)),
            Some("manual"),
        );

        let unmounted = Library {
            root_path: tmp.path().join("unmounted").to_string_lossy().to_string(),
            ..library.clone()
        };
        // The watcher gate sees it stale…
        let health_map = libraries::new_health_map(std::slice::from_ref(&unmounted));
        assert!(!libraries::refresh_health(&health_map, &unmounted, true).is_online());
        // …and the drain itself converts nothing even if called.
        let out = backfill_utc_dates_capped(&unmounted, &exif_dao, &reel_dao, NY, &mut 0, 500);
        assert_eq!(out, UtcBackfillOutcome::default());
        assert_eq!(row(&conn, filed).date_utc, None);
        assert_eq!(row(&conn, manual).date_utc, None);

        // Share returns → picked up next tick.
        assert!(libraries::refresh_health(&health_map, &library, true).is_online());
        let out = backfill_utc_dates_capped(&library, &exif_dao, &reel_dao, NY, &mut 0, 500);
        assert_eq!(out.converted, 2);
        assert_eq!(remaining(&exif_dao, 1), 0);
    }

    /// Seed a pre-generated reel covering `month`-`day` (generated 03:00
    /// EDT that day, as the nightly pre-gen does). Returns (id, mp4 path).
    fn seed_reel(
        reel_dao: &Arc<Mutex<Box<dyn PrecomputedReelDao>>>,
        dir: &std::path::Path,
        span: &str,
        library_key: &str,
        month: u32,
        day: u32,
    ) -> (i32, PathBuf) {
        let mp4 = dir.join(format!("reel-{span}-{library_key}-{month}-{day}.mp4"));
        fs::write(&mp4, b"mp4").unwrap();
        let id = reel_dao
            .lock()
            .unwrap()
            .record_reel(&InsertablePrecomputedReel {
                span: span.to_string(),
                library_key: library_key.to_string(),
                cache_key: format!("{span}-{library_key}-{month}-{day}"),
                output_path: mp4.to_string_lossy().to_string(),
                title: "t".to_string(),
                media_count: 1,
                render_version: 1,
                tz_offset_minutes: -240,
                voice: None,
                generated_at: utc(2025, month, day, 7, 0, 0),
            })
            .unwrap();
        (id, mp4)
    }

    fn reel_ids(conn: &Arc<Mutex<diesel::SqliteConnection>>) -> Vec<i32> {
        let mut c = conn.lock().unwrap();
        schema::precomputed_reels::table
            .select(schema::precomputed_reels::id)
            .order(schema::precomputed_reels::id.asc())
            .load::<i32>(&mut *c)
            .unwrap()
    }

    #[test]
    fn utc_backfill_invalidates_exactly_the_moved_days_reels() {
        let (tmp, library, conn, exif_dao, _face_dao) = setup();
        let reel_dao = reel_dao_for(&conn);
        let reels_dir = TempDir::new().unwrap();

        // Manual wall 2017-06-12 02:00 in NY. The legacy value, read as an
        // instant in NY by the old /memories SQL, showed 06-11 22:00; the
        // true instant is 06-12 06:00Z = 06-12 02:00 EDT → the day moves
        // 06-11 → 06-12.
        let moved = insert_legacy(
            &exif_dao,
            1,
            "moved.jpg",
            Some(utc(2017, 6, 12, 2, 0, 0)),
            Some("manual"),
        );
        // Unchanged: a timestamp-filename row's legacy value is already the instant.
        write_image(tmp.path(), "1400000000.jpg", b"z");
        insert_legacy(
            &exif_dao,
            1,
            "1400000000.jpg",
            Some(1_400_000_000),
            Some("filename"),
        );

        let (_, mp4_11) = seed_reel(&reel_dao, reels_dir.path(), "day", "1", 6, 11);
        let (_, mp4_12) = seed_reel(&reel_dao, reels_dir.path(), "day", "all", 6, 12);
        let (other_lib, _) = seed_reel(&reel_dao, reels_dir.path(), "day", "2", 6, 12);
        let (other_day, mp4_0301) = seed_reel(&reel_dao, reels_dir.path(), "day", "1", 3, 1);
        let (week, _) = seed_reel(&reel_dao, reels_dir.path(), "week", "1", 6, 12);

        let out = backfill_utc_dates_capped(&library, &exif_dao, &reel_dao, NY, &mut 0, 500);
        assert_eq!(out.converted, 2);
        assert_eq!(out.day_changed, 1, "{out:?}");
        assert_eq!(out.reels_invalidated, 2, "{out:?}");
        assert_eq!(row(&conn, moved).date_utc, Some(utc(2017, 6, 12, 6, 0, 0)));

        // Exactly the moved days' day reels for this library (+ "all") are gone.
        assert_eq!(reel_ids(&conn), vec![other_lib, other_day, week]);
        assert!(
            !mp4_11.exists() && !mp4_12.exists(),
            "stale renders unlinked"
        );
        assert!(mp4_0301.exists());

        // With the ledger row gone the nightly pre-gen's freshness gate
        // (`latest_for` must match the new cache key) can't skip → it
        // regenerates on its normal schedule; no new scheduler involved.
        let latest = reel_dao.lock().unwrap().latest_for("day", "all").unwrap();
        assert!(latest.is_none());
    }

    #[test]
    fn utc_backfill_utc_midnight_straddle_marks_nothing() {
        let (_tmp, library, conn, exif_dao, _face_dao) = setup();
        let reel_dao = reel_dao_for(&conn);
        let reels_dir = TempDir::new().unwrap();

        // Manual wall 2017-06-11 23:30 in NY. Legacy instant 06-11 23:30Z
        // (NY 19:30); true instant 06-12 03:30Z (NY 23:30). The UTC date
        // flips 06-11 → 06-12, but the display-zone day stays 06-11.
        let straddle = insert_legacy(
            &exif_dao,
            1,
            "late.jpg",
            Some(utc(2017, 6, 11, 23, 30, 0)),
            Some("manual"),
        );
        seed_reel(&reel_dao, reels_dir.path(), "day", "1", 6, 11);
        seed_reel(&reel_dao, reels_dir.path(), "day", "all", 6, 12);

        let out = backfill_utc_dates_capped(&library, &exif_dao, &reel_dao, NY, &mut 0, 500);
        assert_eq!(out.converted, 1);
        assert_eq!(out.day_changed, 0);
        assert_eq!(out.reels_invalidated, 0);
        assert_eq!(
            row(&conn, straddle).date_utc,
            Some(utc(2017, 6, 12, 3, 30, 0))
        );
        assert_eq!(reel_ids(&conn).len(), 2);
    }
}