//! Thumbnail engine (M2): worker pool, disk cache, cancellation, timing logs.
//!
//! ```text
//! UI ──submit──> dispatcher ──round-robin──> N workers ──results──> UI
//! worker:
//!   1. disk cache hit?  thumbs/{sha1(path|mtime|size)}.jpg  → decode, done
//!   2. else: shrink-decode → orient → fit → encode → cache (atomic rename)
//!   3. generation check → deliver (id, rgba8) — the UI thread builds the
//!      slint::Image (Image is !Send, pixel bytes are not)
//! ```
//!
//! Cancellation is a process-wide generation counter (`AtomicU64`): the UI
//! bumps it whenever the list is reset (new scan); workers that finish a job
//! from a stale generation keep the disk-cache write (content-keyed, still
//! valid) but drop the UI delivery. The queue is unbounded on purpose —
//! backlog is visible in the timing logs (acceptance: it must not grow while
//! scrolling).
//!
//! Queue ordering is PREPEND (not FIFO): `submit` puts each new batch at the
//! FRONT of the queue. The most recently requested content is what the user
//! is looking at (or scrolling back to); letting a backlog of stale far-away
//! cold decodes sit ahead of it starved fast disk-cache-hit re-submissions —
//! the scroll-back-to-top shimmer bug. Regression: `scroll_back_cache_hit_not_starved`.

use std::collections::VecDeque;
use std::fs;
use std::ops::Range;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::mpsc;
use std::sync::{Arc, Condvar, Mutex};
use std::time::Instant;

use sha1::{Digest, Sha1};

use crate::db;
use crate::images;

/// Grid thumbnails: ~400 px (longest side). Cells render ~350–420 px physical
/// (3 columns) on a phone. Was 500: each newly-painted cell's 700 KB texture
/// upload stalls the render frame 25–40 ms on the phone (measured 2026-10-28:
/// the 25–40 ms stall bucket tracks cell-entry rate and vanishes with 1×1
/// test images). 400 cuts the upload ~36% at a 1.04× upscale — visually
/// identical in a 3-column grid. The target is part of the cache key, so a
/// future different-size target coexists in the same dir.
pub const GRID_TARGET: u32 = 400;
const JPEG_QUALITY: u8 = 85;

/// Per-worker queue cap. A burst (one tick's submissions) is never bigger
/// than the viewport (~45 cells), so fresh viewport work ALWAYS survives the
/// cap; overflow is dropped from the TAIL (oldest transit work). 96 ≈ two
/// screens per worker of headroom.
pub const MAX_WORKER_QUEUE: usize = 96;

#[derive(Debug, Clone)]
pub struct Job {
    /// Row index in the UI model at submit time.
    pub id: usize,
    pub path: String,
    pub mtime: i64,
    pub size: i64,
    /// Generation the job belongs to (stale jobs are not delivered).
    pub gen: u64,
}

/// One worker's job queue: Mutex<VecDeque> + Condvar, NOT mpsc — mpsc is
/// FIFO-only, but the point of this design is that a new batch (the user's
/// current intent) must be placed AHEAD of older queued work. The
/// dispatcher prepends each burst in order; the worker pops the front.
struct WorkerQueue {
    q: Mutex<VecDeque<Job>>,
    cv: Condvar,
}

impl WorkerQueue {
    fn new() -> Self {
        Self { q: Mutex::new(VecDeque::new()), cv: Condvar::new() }
    }
}

#[derive(Debug)]
pub struct JobResult {
    pub id: usize,
    pub gen: u64,
    pub path: String,
    pub ok: bool,
    /// The job was DROPPED before being decoded (worker-queue overflow:
    /// it was stale transit work behind newer viewport work). The UI clears
    /// its in-flight marker; the cell is NOT failed — a re-entry re-submits
    /// it (disk-cache hit if it was ever decoded, otherwise a fresh decode).
    pub skipped: bool,
    /// RGBA8 (A=255), orientation applied, fitted to the engine's target
    /// size; empty when `!ok`/`skipped`. Expanded on the worker because
    /// Slint's Skia backend can't take RGB888 (it would convert
    /// byte-by-byte on the UI thread during the render frame).
    ///
    /// Delivered as PLAIN BYTES (no GPU handles): the UI thread uploads
    /// them to Slint's own wgpu device at drain time (gputex) — a texture
    /// must be created on the device of the world that renders it, and
    /// only the render thread knows which that is across window
    /// recreations.
    pub rgb: Vec<u8>,
    pub width: u32,
    pub height: u32,
    pub from_cache: bool,
    pub elapsed_ms: u128,
}

pub struct Engine {
    /// Job queue: Mutex<VecDeque> + Condvar (not mpsc — mpsc is FIFO-only).
    /// NEW submissions are PREPENDED: the most recently requested content is
    /// what the user is looking at (or scrolling back to); stale far-away
    /// prefetch must not starve it. Concretely fixes the scroll-back-to-top
    /// case where fast disk-cache-hit jobs waited seconds behind a backlog of
    /// slow cold decodes (regression: `scroll_back_cache_hit_not_starved`).
    jobs: Arc<Mutex<VecDeque<Job>>>,
    job_cv: Arc<Condvar>,
    closed: Arc<AtomicBool>,
    /// Only the UI thread drains; the Mutex just makes Engine Sync so it can
    /// live in a process-wide OnceLock.
    rx: Mutex<mpsc::Receiver<JobResult>>,
    gen: Arc<AtomicU64>,
    cache_dir: PathBuf,
    in_flight: Arc<AtomicUsize>,
    done: Arc<AtomicU64>,
    hits: Arc<AtomicU64>,
    errs: Arc<AtomicU64>,
    target: u32,
}

impl Drop for Engine {
    fn drop(&mut self) {
        self.closed.store(true, Ordering::SeqCst);
        self.job_cv.notify_all();
    }
}

/// Cache file for a job: `thumbs/{sha1(path|mtime|size|target)}.jpg` — the
/// target is in the key so different render sizes coexist (M2 list 64 px,
/// M3 grid 500 px).
pub fn cache_path(cache_dir: &Path, path: &str, mtime: i64, size: i64, target: u32) -> PathBuf {
    let mut h = Sha1::new();
    h.update(path.as_bytes());
    h.update(b"|");
    h.update(mtime.to_le_bytes());
    h.update(b"|");
    h.update(size.to_le_bytes());
    h.update(b"|");
    h.update(target.to_le_bytes());
    let hex: String = h.finalize().iter().map(|b| format!("{b:02x}")).collect();
    cache_dir.join(format!("{hex}.jpg"))
}

impl Engine {
    /// Spawn the dispatcher + `workers` worker threads. All jobs render at
    /// `target` px (longest side). `max_worker_queue` bounds each worker's
    /// pending queue (stale transit work beyond the cap is dropped with a
    /// `skipped` result); pass `usize::MAX` for uncapped (tests).
    pub fn new(cache_dir: PathBuf, workers: usize, target: u32, max_worker_queue: usize) -> Result<Self, String> {
        fs::create_dir_all(&cache_dir).map_err(|e| format!("mkdir {}: {e}", cache_dir.display()))?;
        let jobs = Arc::new(Mutex::new(VecDeque::<Job>::new()));
        let job_cv = Arc::new(Condvar::new());
        let closed = Arc::new(AtomicBool::new(false));
        let (res_tx, res_rx) = mpsc::channel::<JobResult>();
        let gen = Arc::new(AtomicU64::new(1));
        let in_flight = Arc::new(AtomicUsize::new(0));
        let stats = Stats {
            done: Arc::new(AtomicU64::new(0)),
            hits: Arc::new(AtomicU64::new(0)),
            errs: Arc::new(AtomicU64::new(0)),
        };

        let mut wqs: Vec<Arc<WorkerQueue>> = Vec::with_capacity(workers);
        for i in 0..workers {
            let wq = Arc::new(WorkerQueue::new());
            let (wq_t, res_tx, gen, in_flight, closed_w) =
                (wq.clone(), res_tx.clone(), gen.clone(), in_flight.clone(), closed.clone());
            let (done, hits, errs) = (stats.done.clone(), stats.hits.clone(), stats.errs.clone());
            let cache_dir = cache_dir.clone();
            std::thread::Builder::new()
                .name(format!("thumb-{i}"))
                .spawn(move || {
                    // Jank fix: decode bursts must not preempt the UI thread
                    // (same process priority by default) or starve the vsync
                    // waker. +2 keeps them ahead of background apps.
                    crate::prio::set_self_nice(2);
                    worker_loop(
                        &wq_t, res_tx, gen, in_flight, done, hits, errs, cache_dir, target, closed_w,
                    )
                })
                .map_err(|e| e.to_string())?;
            wqs.push(wq);
        }

        // Dispatcher: drains the central queue as priority-ordered bursts
        // and PREPENDS each burst to the worker queues (the cap drops the
        // oldest tail work with `skipped` results). Exits when the engine is
        // dropped and the queue has drained.
        let (d_jobs, d_cv, d_closed) = (jobs.clone(), job_cv.clone(), closed.clone());
        let (d_res, d_inflight) = (res_tx.clone(), in_flight.clone());
        std::thread::Builder::new()
            .name("thumb-dispatch".into())
            .spawn(move || {
                crate::prio::set_self_nice(2);
                dispatch_loop(d_jobs, d_cv, d_closed, wqs, d_res, d_inflight, max_worker_queue)
            })
            .map_err(|e| e.to_string())?;
        drop(res_tx);

        Ok(Self {
            jobs,
            job_cv,
            closed,
            rx: Mutex::new(res_rx),
            gen,
            cache_dir,
            in_flight,
            done: stats.done,
            hits: stats.hits,
            errs: stats.errs,
            target,
        })
    }

    pub fn target(&self) -> u32 {
        self.target
    }

    pub fn cache_dir(&self) -> &Path {
        &self.cache_dir
    }

    pub fn generation(&self) -> u64 {
        self.gen.load(Ordering::Relaxed)
    }

    /// Invalidate all in-flight/pending work (list reset).
    pub fn bump_generation(&self) -> u64 {
        self.gen.fetch_add(1, Ordering::Relaxed) + 1
    }

    /// Submit jobs, PREPENDED to the queue: fresh submissions express the
    /// user's current (or return) intent and must be decoded before stale
    /// far-away prefetch. Within a batch the original order is preserved.
    pub fn submit(&self, jobs: &[Job]) {
        if jobs.is_empty() {
            return;
        }
        {
            let mut dq = self.jobs.lock().unwrap();
            for j in jobs.iter().rev() {
                dq.push_front(j.clone());
            }
        }
        self.in_flight.fetch_add(jobs.len(), Ordering::Relaxed);
        self.job_cv.notify_one();
    }

    /// Drop QUEUED (not yet started) jobs outside `range`, returning the
    /// number dropped. Queued work for cells the user has scrolled away from
    /// is low value: if they come back, the disk cache serves it in ~10 ms.
    /// This bounds the queue to the retention band, so viewport work never
    /// waits behind stale transit work (in-flight jobs finish; their
    /// out-of-viewport results are dropped at consumption).
    pub fn prune_outside(&self, range: &Range<usize>) -> Vec<Job> {
        let mut dq = self.jobs.lock().unwrap();
        let mut dropped: Vec<Job> = Vec::new();
        dq.retain(|j| {
            if range.contains(&j.id) {
                true
            } else {
                dropped.push(j.clone());
                false
            }
        });
        if !dropped.is_empty() {
            self.in_flight.fetch_sub(dropped.len(), Ordering::Relaxed);
        }
        dropped
    }

    /// Non-blocking drain (UI thread). Stale-generation results are dropped.
    pub fn drain(&self) -> Vec<JobResult> {
        let mut out = Vec::new();
        let gen = self.generation();
        let rx = self.rx.lock().unwrap();
        while let Ok(r) = rx.try_recv() {
            if r.gen == gen {
                out.push(r);
            }
        }
        out
    }

    /// "queued" = accepted but not yet finished.
    pub fn queued(&self) -> usize {
        self.in_flight.load(Ordering::Relaxed)
    }

    pub fn log_summary(&self) -> String {
        format!(
            "thumbs: {} done ({} cache hit, {} failed), {} queued",
            self.done.load(Ordering::Relaxed),
            self.hits.load(Ordering::Relaxed),
            self.errs.load(Ordering::Relaxed),
            self.queued(),
        )
    }
}

#[derive(Clone)]
struct Stats {
    done: Arc<AtomicU64>,
    hits: Arc<AtomicU64>,
    errs: Arc<AtomicU64>,
}

/// The dispatcher (one thread): drains the central queue as a single
/// burst — the deque is already in priority order (newest batch first, and
/// center-first within a batch) — splits it round-robin over the workers,
/// and PREPENDS each worker's share. Overflow beyond `max_worker_queue`
/// is dropped from the TAIL (the oldest transit work) with a `skipped`
/// result so the UI can clear its in-flight marker.
///
/// This is what makes "the viewport is first" true at EVERY level of the
/// pipeline: no matter how much stale transit work is queued — or in
/// flight — a fresh submission (the current viewport) is placed ahead of
/// it. A burst never exceeds the viewport (~45) < the cap (96), so fresh
/// viewport work is never itself dropped.
fn dispatch_loop(
    jobs: Arc<Mutex<VecDeque<Job>>>,
    cv: Arc<Condvar>,
    closed: Arc<AtomicBool>,
    wqs: Vec<Arc<WorkerQueue>>,
    res_tx: mpsc::Sender<JobResult>,
    in_flight: Arc<AtomicUsize>,
    max_worker_queue: usize,
) {
    loop {
        // Drain the whole central deque as one burst (priority order).
        let burst: Vec<Job> = {
            let mut dq = jobs.lock().unwrap();
            while dq.is_empty() {
                if closed.load(Ordering::SeqCst) {
                    return; // engine dropped, queue drained
                }
                dq = cv.wait(dq).unwrap();
            }
            dq.drain(..).collect()
        };

        // Round-robin split, preserving per-worker order.
        let n = wqs.len();
        let mut per: Vec<Vec<Job>> = vec![Vec::new(); n];
        for (i, j) in burst.into_iter().enumerate() {
            per[i % n].push(j);
        }

        for w in 0..n {
            if per[w].is_empty() {
                continue;
            }
            let dropped: Vec<Job> = {
                let mut g = wqs[w].q.lock().unwrap();
                let mut dropped = Vec::new();
                // Cap: prepend the burst; drop overflow from the tail
                // (oldest first: existing tail, then the burst's own tail).
                let overflow = (per[w].len() + g.len()).saturating_sub(max_worker_queue);
                for _ in 0..overflow {
                    match g.pop_back() {
                        Some(j) => dropped.push(j),
                        None => break,
                    }
                }
                while per[w].len().saturating_sub(g.len()) > max_worker_queue {
                    match per[w].pop() {
                        Some(j) => dropped.push(j),
                        None => break,
                    }
                }
                // Prepend in order (last-first): the front stays the newest.
                for j in per[w].drain(..).rev() {
                    g.push_front(j);
                }
                dropped
            };
            wqs[w].cv.notify_one();
            for j in dropped {
                in_flight.fetch_sub(1, Ordering::Relaxed);
                let _ = res_tx.send(JobResult {
                    id: j.id,
                    gen: j.gen,
                    path: j.path,
                    ok: false,
                    skipped: true,
                    rgb: Vec::new(),
                    width: 0,
                    height: 0,
                    from_cache: false,
                    elapsed_ms: 0,
                });
            }
        }
    }
}

fn worker_loop(
    wq: &Arc<WorkerQueue>,
    res_tx: mpsc::Sender<JobResult>,
    gen: Arc<AtomicU64>,
    in_flight: Arc<AtomicUsize>,
    done: Arc<AtomicU64>,
    hits: Arc<AtomicU64>,
    errs: Arc<AtomicU64>,
    cache_dir: PathBuf,
    target: u32,
    closed: Arc<AtomicBool>,
) {
    // One WAL connection per worker; the scan thread keeps its own.
    let mut conn = match db::open(&db::db_path()) {
        Ok(c) => Some(c),
        Err(e) => {
            log::warn!("gallery: thumb worker db error: {e}");
            None
        }
    };

    loop {
        let job = {
            let mut g = wq.q.lock().unwrap();
            while g.is_empty() {
                if closed.load(Ordering::SeqCst) {
                    return; // engine dropped, queue drained
                }
                g = wq.cv.wait(g).unwrap();
            }
            g.pop_front().unwrap()
        };
        let t0 = Instant::now();
        let (ok, thumb, from_cache) = process(&job, &cache_dir, target, &hits, &errs);
        let elapsed = t0.elapsed().as_millis();
        // Jank-hunt (`perf` feature): per-job decode cost, correlated
        // against FR gap spikes. See docs/BUGS/scroll-jank-120hz.md.
        #[cfg(feature = "perf")]
        {
            log::info!(
                "DECODE {}ms {} {}",
                elapsed,
                if from_cache { "hit" } else { "miss" },
                job.path
            );
        }

        if let Some(ref mut c) = conn {
            if let Some(ref t) = thumb {
                if let Err(e) = db::set_thumb(
                    c,
                    &job.path,
                    t.src_width as i64,
                    t.src_height as i64,
                    t.orientation as i64,
                    cache_path(&cache_dir, &job.path, job.mtime, job.size, target)
                        .to_str()
                        .unwrap_or(""),
                ) {
                    log::warn!("gallery: db set_thumb failed for {}: {e}", job.path);
                }
            }
        }

        in_flight.fetch_sub(1, Ordering::Relaxed);
        let n = done.fetch_add(1, Ordering::Relaxed) + 1;
        if n % 256 == 0 {
            log::info!(
                "gallery: thumbs {} ({} hit, {} err, {} queued, last {} ms)",
                n,
                hits.load(Ordering::Relaxed),
                errs.load(Ordering::Relaxed),
                in_flight.load(Ordering::Relaxed),
                elapsed,
            );
        }

        // Stale generation: cache write stays (content-keyed), skip delivery.
        if gen.load(Ordering::Relaxed) != job.gen {
            continue;
        }
        let (rgb, width, height) = match thumb {
            Some(t) => (t.rgb, t.width, t.height),
            None => (Vec::new(), 0, 0),
        };
        // Expand RGB8 -> RGBA8 HERE (worker thread), not in Slint's render
        // pass — see JobResult docs. Cost: ~0.3 ms per 400 px thumb.
        let rgba = if ok { crate::images::to_rgba8(&rgb) } else { rgb };
        // Delivered as plain bytes: the UI thread uploads to Slint's
        // current wgpu device at drain time (gputex::upload_rgba). No
        // worker ever touches a wgpu device — that was the source of the
        // stale-texture panics across window recreations (2026-10-28).
        let _ = res_tx.send(JobResult {
            id: job.id,
            gen: job.gen,
            path: job.path,
            ok,
            skipped: false,
            rgb: rgba,
            width,
            height,
            from_cache,
            elapsed_ms: elapsed,
        });
    }
}

fn process(
    job: &Job,
    cache_dir: &Path,
    target: u32,
    hits: &Arc<AtomicU64>,
    errs: &Arc<AtomicU64>,
) -> (bool, Option<images::Thumb>, bool) {
    let cache = cache_path(cache_dir, &job.path, job.mtime, job.size, target);
    if cache.exists() {
        if let Ok(t) = images::decode_jpeg(&cache) {
            hits.fetch_add(1, Ordering::Relaxed);
            return (true, Some(t), true);
        }
        // Corrupt cache entry: regenerate.
        let _ = fs::remove_file(&cache);
    }

    match images::make_thumb(Path::new(&job.path), target) {
        Ok(t) => {
            match images::encode_jpeg(&t.rgb, t.width, t.height, JPEG_QUALITY) {
                Ok(bytes) => {
                    let tmp = cache.with_extension("tmp");
                    if fs::write(&tmp, &bytes).is_ok() {
                        if let Err(e) = fs::rename(&tmp, &cache) {
                            log::warn!("gallery: cache rename failed: {e}");
                            let _ = fs::remove_file(&tmp);
                        }
                    }
                }
                Err(e) => log::warn!("gallery: cache encode failed for {}: {e}", job.path),
            }
            (true, Some(t), false)
        }
        Err(e) => {
            errs.fetch_add(1, Ordering::Relaxed);
            log::warn!("gallery: thumb failed for {}: {e}", job.path);
            (false, None, false)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::images;
    use crate::scan::Entry;

    fn make_jpeg(w: u32, h: u32, r: u8) -> Vec<u8> {
        let mut rgb = Vec::with_capacity(3 * w as usize * h as usize);
        for _ in 0..w * h as u32 {
            rgb.extend_from_slice(&[r, 120, 200]);
        }
        images::encode_jpeg(&rgb, w, h, 90).unwrap()
    }

    /// Synthetic 5k-set engine test (HANDOFF §8/§9 M2): cold fill, timing
    /// logs, instant warm refill from the disk cache, generation cancellation.
    #[test]
    fn engine_5k_cold_warm_cancel() {
        const N: usize = 5000;
        let dir = tempfile::tempdir().unwrap();
        let cache = dir.path().join("thumbs");
        // Keep the worker threads' DB writes out of the dev database.
        // (Only this test touches the DB path; the others pass explicit tempdirs.)
        std::env::set_var("GALLERY_DB_PATH", dir.path().join("gallery.db"));

        // 1. Generate the synthetic set (varied sizes, a few "large").
        let entries = (0..N).map(|i| {
            let big = i % 500 == 0; // every 500th is a "large" photo
            let (w, h) = if big {
                (4000u32, 3000u32)
            } else {
                (600u32 + (i % 8) as u32 * 200, 400u32 + (i % 6) as u32 * 150)
            };
            let p = dir.path().join(format!("p{i:05}.jpg"));
            std::fs::write(&p, make_jpeg(w, h, (i % 250) as u8 + 2)).unwrap();
            Entry { path: p.to_str().unwrap().into(), mtime: i as i64, size: std::fs::metadata(&p).unwrap().len() as i64 }
        }).collect::<Vec<_>>();

        // 2. Cold run.
        let eng = Engine::new(cache.clone(), 4, GRID_TARGET, usize::MAX).unwrap();
        let gen = eng.generation();
        let t0 = Instant::now();
        eng.submit(&entries.iter().enumerate().map(|(i, e)| Job {
            id: i, path: e.path.clone(), mtime: e.mtime, size: e.size, gen,
        }).collect::<Vec<_>>());

        let mut first_48 = None;
        let mut got = 0usize;
        let mut fails = 0;
        let mut cache_hits = 0;
        while got < N {
            std::thread::sleep(std::time::Duration::from_millis(5));
            for r in eng.drain() {
                got += 1;
                if r.from_cache { cache_hits += 1; }
                if !r.ok { fails += 1; }
                if first_48.is_none() && got >= 48 { first_48 = Some(t0.elapsed()); }
            }
        }
        let cold_total = t0.elapsed();
        assert_eq!(fails, 0, "all 5k synthetic thumbnails must decode");
        assert_eq!(cache_hits, 0);
        assert_eq!(eng.queued(), 0);
        assert!(first_48.is_some());
        eprintln!("gallery test: cold  {N} thumbs in {cold_total:?}, first 48 in {first_48:?}");
        assert_eq!(std::fs::read_dir(&cache).unwrap().count(), N, "cache holds all entries");

        // 3. Warm run: everything must come from the disk cache.
        let eng2 = Engine::new(cache.clone(), 4, GRID_TARGET, usize::MAX).unwrap();
        let gen2 = eng2.generation();
        let t0 = Instant::now();
        eng2.submit(&entries.iter().enumerate().map(|(i, e)| Job {
            id: i, path: e.path.clone(), mtime: e.mtime, size: e.size, gen: gen2,
        }).collect::<Vec<_>>());
        let mut got = 0usize;
        let mut cache_hits = 0;
        while got < N {
            std::thread::sleep(std::time::Duration::from_millis(5));
            for r in eng2.drain() {
                got += 1;
                assert!(r.ok);
                if r.from_cache { cache_hits += 1; }
            }
        }
        let warm_total = t0.elapsed();
        assert_eq!(cache_hits, N, "warm run is 100% cache hits");
        eprintln!("gallery test: warm  {N} thumbs in {warm_total:?} (100% cache)");
        assert!(warm_total.as_secs_f64() < cold_total.as_secs_f64());

        // 4. Cancellation: submit, bump generation before draining -> dropped.
        let eng3 = Engine::new(cache.clone(), 4, GRID_TARGET, usize::MAX).unwrap();
        let stale = eng3.generation();
        let p = &entries[0];
        // Use a fresh path (mtime bump) so it is a cache miss and takes time.
        let fresh = dir.path().join("fresh.jpg");
        std::fs::write(&fresh, make_jpeg(3000, 2000, 99)).unwrap();
        let sz = std::fs::metadata(&fresh).unwrap().len() as i64;
        eng3.submit(&[Job { id: 0, path: fresh.to_str().unwrap().into(), mtime: 1, size: sz, gen: stale }]);
        std::thread::sleep(std::time::Duration::from_millis(20));
        let new_gen = eng3.bump_generation();
        assert!(new_gen != stale);
        // Drain until the (now stale) result is gone: nothing must arrive.
        std::thread::sleep(std::time::Duration::from_millis(2000));
        assert!(eng3.drain().is_empty(), "stale-generation results are dropped");
        assert_eq!(eng3.queued(), 0);
        let _ = p;
    }

    /// Scroll-back bug (regression): a user scrolls down (queue fills with
    /// slow cold decodes of far cells), then back to the top. The top cells'
    /// jobs are fast disk-cache HITS but, on a FIFO queue, they wait behind
    /// the entire cold backlog — the user sees shimmer for seconds.
    /// Re-submitted (recently requested) work must be decoded first.
    #[test]
    fn scroll_back_cache_hit_not_starved() {
        let dir = tempfile::tempdir().unwrap();
        let cache = dir.path().join("thumbs");
        std::env::set_var("GALLERY_DB_PATH", dir.path().join("gallery.db"));

        // Detailed (noise) JPEGs: non-trivial cold decodes, still small files.
        let mk = |name: &str| -> Entry {
            let (w, h) = (2048u32, 1536u32);
            let mut rgb = Vec::with_capacity(3 * (w * h) as usize);
            let mut x: u64 = 0x9e3779b97f4a7c15;
            for _ in 0..(w * h) {
                x = x
                    .wrapping_mul(6364136223846793005)
                    .wrapping_add(1442695040888963407);
                rgb.extend_from_slice(&[(x >> 33) as u8, (x >> 17) as u8, (x >> 1) as u8]);
            }
            let p = dir.path().join(name);
            std::fs::write(&p, images::encode_jpeg(&rgb, w, h, 80).unwrap()).unwrap();
            Entry {
                path: p.to_str().unwrap().into(),
                mtime: 1,
                size: std::fs::metadata(&p).unwrap().len() as i64,
            }
        };
        let top: Vec<Entry> = (0..60).map(|i| mk(&format!("top{i:03}.jpg"))).collect();
        let far: Vec<Entry> = (0..180).map(|i| mk(&format!("far{i:03}.jpg"))).collect();

        let eng = Engine::new(cache.clone(), 4, GRID_TARGET, 96).unwrap();
        let gen = eng.generation();
        let jobs = |es: &[Entry], base: usize| {
            es.iter()
                .enumerate()
                .map(|(i, e)| Job {
                    id: base + i,
                    path: e.path.clone(),
                    mtime: e.mtime,
                    size: e.size,
                    gen,
                })
                .collect::<Vec<_>>()
        };

        // Warm the top 60 into the disk cache (they were painted earlier).
        eng.submit(&jobs(&top, 0));
        let mut got = 0;
        while got < 60 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            got += eng.drain().len();
        }

        // "Scroll down": 180 cold far jobs. "Scroll back up": 60 warm tops.
        let t0 = Instant::now();
        eng.submit(&jobs(&far, 1000));
        eng.submit(&jobs(&top, 0));
        let mut tops = 0;
        let mut total = 0;
        let mut t_tops = None;
        while total < 240 {
            std::thread::sleep(std::time::Duration::from_millis(2));
            for r in eng.drain() {
                total += 1;
                if r.id < 60 && t_tops.is_none() {
                    tops += 1;
                    if tops == 60 {
                        t_tops = Some(t0.elapsed());
                    }
                }
            }
        }
        let t_tops = t_tops.unwrap();
        eprintln!(
            "scroll-back: all 60 cache-hit tops in {t_tops:?} behind a 180-cold-job backlog (total {:?})",
            t0.elapsed()
        );
        assert!(
            t_tops.as_secs_f64() < 0.75,
            "cache-hit re-entry waited {t_tops:?} behind the cold backlog — FIFO starvation"
        );
    }

    /// Worker-level priority (regression): the old design put every batch
    /// into the workers' FIFO mpsc, so a fresh viewport batch had to wait
    /// behind ALL stale transit work already queued in a worker. A new
    /// batch must now be placed AHEAD of it: with one worker, a fresh job
    /// submitted while a slow batch is queued must finish within the first
    /// two results (never after the whole old queue).
    #[test]
    fn fresh_batch_preempts_stale_worker_queue() {
        let dir = tempfile::tempdir().unwrap();
        let cache = dir.path().join("thumbs");
        std::env::set_var("GALLERY_DB_PATH", dir.path().join("gallery.db"));

        // Detailed noise JPEGs with real decode cost.
        let mk = |name: &str| -> Entry {
            let (w, h) = (1600u32, 1200u32);
            let mut rgb = Vec::with_capacity(3 * (w * h) as usize);
            let mut x: u64 = 0x243f6a8885a308d3;
            for _ in 0..(w * h) {
                x = x
                    .wrapping_mul(6364136223846793005)
                    .wrapping_add(1442695040888963407);
                rgb.extend_from_slice(&[(x >> 33) as u8, (x >> 17) as u8, (x >> 1) as u8]);
            }
            let p = dir.path().join(name);
            std::fs::write(&p, images::encode_jpeg(&rgb, w, h, 80).unwrap()).unwrap();
            Entry {
                path: p.to_str().unwrap().into(),
                mtime: 1,
                size: std::fs::metadata(&p).unwrap().len() as i64,
            }
        };
        let stale: Vec<Entry> = (0..10).map(|i| mk(&format!("stale{i}.jpg"))).collect();
        let fresh = mk("fresh_viewport.jpg");

        // ONE worker: no round-robin rescue — the ordering is purely the
        // queue's.
        let eng = Engine::new(cache, 1, GRID_TARGET, 96).unwrap();
        let gen = eng.generation();
        let stale_jobs: Vec<Job> = stale
            .iter()
            .enumerate()
            .map(|(i, e)| Job { id: i, path: e.path.clone(), mtime: e.mtime, size: e.size, gen })
            .collect();
        eng.submit(&stale_jobs);

        // Wait until the worker has started (first result = stale job 0
        // finished; the rest sit in the worker queue).
        let mut got = 0;
        while got < 1 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            got += eng.drain().len();
        }

        // Fresh viewport submission: must jump AHEAD of the 9 queued stale
        // jobs. It may finish just before/after the one stale job already
        // in flight (job 1) — but it can NEVER be behind jobs 2..9.
        let fresh_job = Job {
            id: 100,
            path: fresh.path.clone(),
            mtime: fresh.mtime,
            size: fresh.size,
            gen,
        };
        eng.submit(&[fresh_job.clone()]);
        let mut seen: Vec<usize> = Vec::new();
        while seen.iter().position(|id| *id == 100).is_none() {
            std::thread::sleep(std::time::Duration::from_millis(5));
            for r in eng.drain() {
                seen.push(r.id);
            }
            assert!(seen.len() < 8, "fresh job arrived behind the stale queue: {seen:?}");
        }
        let pos = seen.iter().position(|id| *id == 100).unwrap();
        // At most the in-flight stale job (id 1) may precede it.
        assert!(pos <= 1, "fresh job arrived in position {pos}: {seen:?}");
    }

    #[test]
    fn cache_key_stable_and_distinct() {
        let d = Path::new("/tmp/x");
        let a = cache_path(d, "/a.jpg", 1, 2, 64);
        let b = cache_path(d, "/a.jpg", 1, 2, 64);
        let c = cache_path(d, "/a.jpg", 3, 2, 64);
        let e = cache_path(d, "/a.jpg", 1, 2, 500); // different render size
        assert_eq!(a, b);
        assert_ne!(a, c);
        assert_ne!(a, e);
        assert_eq!(a.file_name().unwrap().to_str().unwrap().len(), 40 + 4); // sha1 hex + .jpg
    }
}