date_backfill: per-tick drain for unresolved date_taken rows

Adds two ExifDao methods (`get_rows_needing_date_backfill` /
`backfill_date_taken`) and a `backfill_missing_date_taken` watcher pass
that runs on every tick alongside `backfill_unhashed_backlog`.

The drain queries the partial index for rows where `date_taken IS NULL`
or `date_taken_source = 'fs_time'`, batches up to
`DATE_BACKFILL_MAX_PER_TICK` paths (default 500), and feeds them through
`date_resolver::resolve_dates_batch` — a single exiftool subprocess
covers the whole tick. Rows that newly resolve to `exiftool` /
`filename` / `fs_time` get persisted via `backfill_date_taken` (touches
only `date_taken` + `date_taken_source` so EXIF / hash / perceptual
columns survive).

`filename`-sourced rows are intentionally not re-resolved — the regex
is authoritative when it matches and re-running exiftool wouldn't
change the answer. Files that have disappeared from disk are skipped
so a ghost row doesn't loop through the drain forever; the
missing-file scan in `library_maintenance` retires those separately.

Comes with two DAO unit tests (eligibility filter + column-isolation).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/main.rs

@@ -2125,6 +2125,15 @@ fn watch_files(
                     );
                 }
 
+                // Date-taken backfill: drain rows whose canonical date is
+                // either unresolved or only fs_time-sourced. Independent
+                // of face detection — runs even on deploys that don't
+                // configure Apollo, since `/memories` depends on it.
+                {
+                    let context = opentelemetry::Context::new();
+                    backfill_missing_date_taken(&context, lib, &exif_dao);
+                }
+
                 if is_full_scan {
                     info!(
                         "Running full scan for library '{}' (scan #{})",
@@ -2706,6 +2715,103 @@ fn backfill_unhashed_backlog(
     backfilled
 }
 
+/// 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.
+fn backfill_missing_date_taken(
+    context: &opentelemetry::Context,
+    library: &libraries::Library,
+    exif_dao: &Arc<Mutex<Box<dyn ExifDao>>>,
+) -> usize {
+    let cap: i64 = dotenv::var("DATE_BACKFILL_MAX_PER_TICK")
+        .ok()
+        .and_then(|s| s.parse().ok())
+        .filter(|n: &i64| *n > 0)
+        .unwrap_or(500);
+
+    let rows: Vec<(i32, String)> = {
+        let mut dao = exif_dao.lock().expect("Unable to lock ExifDao");
+        dao.get_rows_needing_date_backfill(context, 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() as usize);
+    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(
+                context,
+                library.id,
+                rel_path,
+                rd.timestamp,
+                rd.source.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
+}
+
 /// 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