//! Generic background-worker runtime.
//!
//! Every recv-loop worker in the app (analysis, edit, forge, import, export,
//! search, loose-files) routes its command loop through [`spawn_worker`]. The
//! factory owns the command channel's `Receiver` and the `recv` loop, and wraps
//! each command's handler in `catch_unwind` so a panic becomes a caller-defined
//! event instead of silently killing the thread, which would stop every future
//! command of that kind with no UI signal.
//!
//! Every recv-loop worker routes through here, so the `catch_unwind` lives in one
//! place instead of being copy-pasted (and forgotten) per worker, the
//! historically-recurring "added a worker, forgot the guard" drift (the
//! `loose_files_worker` regression). This is enforced by convention, not by the
//! compiler: nothing *prevents* a new worker from hand-rolling its own
//! `mpsc::channel()` + `rx.recv()` loop. The rule is: any worker that loops over
//! commands MUST be spawned via [`spawn_worker`]; one-shot job threads (which run
//! a single task and exit, e.g. the classifier job and the preview decode) carry
//! their own `catch_unwind` and are a different shape. A new bare recv loop is a
//! review smell, not a build error.
//!
//! ## Cancellation
//!
//! Every worker gets a shared cancel flag for free (in [`WorkerCtx`]). It is set
//! to `true` automatically when the [`WorkerHandle`] is dropped, so an in-flight
//! long operation aborts promptly on shutdown instead of blocking the join.
//! Workers that support an explicit Cancel command wrap the handle in a thin
//! newtype whose `send` calls [`WorkerHandle::request_cancel`] before enqueuing,
//! so the flag is set synchronously by the GUI thread (the queued command itself
//! becomes a no-op). Long operations reset the flag at their start via
//! [`WorkerCtx::reset_cancel`] so a stale cancel does not abort the next command.

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, mpsc};
use std::thread;
use std::time::Duration;

/// Bound on the per-worker event channel. The channel is the one place an
/// unbounded queue could balloon, the analysis batch fans out over rayon and
/// emits a per-sample event, which on a 100k-sample import outruns a slow GUI
/// drain. A bounded channel applies backpressure (the producer waits) instead of
/// growing without limit; the cap is generous so a normally-draining GUI never
/// blocks a worker.
const EVENT_CHANNEL_CAP: usize = 4096;

/// Sink handed to a worker step so it can emit zero or more events per command.
///
/// Cloneable and `Sync` (it wraps an `mpsc::SyncSender`), so a step may share it
/// across rayon worker threads (e.g. the analysis batch fan-out). [`emit`](Self::emit)
/// applies bounded backpressure but never blocks past cancellation.
pub struct EventSink<Ev> {
    tx: mpsc::SyncSender<Ev>,
    cancel: Arc<AtomicBool>,
}

impl<Ev> Clone for EventSink<Ev> {
    fn clone(&self) -> Self {
        Self {
            tx: self.tx.clone(),
            cancel: Arc::clone(&self.cancel),
        }
    }
}

impl<Ev> EventSink<Ev> {
    /// Emit one event back to the handle.
    ///
    /// The event channel is bounded, so this applies backpressure when the GUI
    /// hasn't drained yet: it retries rather than dropping the event or growing
    /// the queue. It must never block *past cancellation*, though, a worker stuck
    /// here while the handle is being dropped would hang the join on quit, so once
    /// the cancel flag is set it drops the event and returns. Also returns if the
    /// receiver is gone (handle dropped).
    pub fn emit(&self, ev: Ev) {
        let mut pending = ev;
        loop {
            match self.tx.try_send(pending) {
                Ok(()) => return,
                Err(mpsc::TrySendError::Full(returned)) => {
                    if self.cancel.load(Ordering::Acquire) {
                        return; // shutting down, drop the event rather than hang
                    }
                    pending = returned;
                    thread::sleep(Duration::from_millis(1));
                }
                Err(mpsc::TrySendError::Disconnected(_)) => return,
            }
        }
    }

    /// Create a sink paired with a receiver. For tests that exercise functions
    /// taking an `&EventSink` directly, outside a spawned worker. The test must
    /// stay under [`EVENT_CHANNEL_CAP`] unsent events or drain, since the standalone
    /// sink has no cancel flag to release a full-channel wait.
    pub fn channel() -> (Self, mpsc::Receiver<Ev>) {
        let (tx, rx) = mpsc::sync_channel(EVENT_CHANNEL_CAP);
        (
            Self {
                tx,
                cancel: Arc::new(AtomicBool::new(false)),
            },
            rx,
        )
    }
}

/// Context passed to each worker step: an [`EventSink`] plus the shared cancel
/// flag. Both are `Sync`, so `&WorkerCtx` can be shared into rayon closures.
pub struct WorkerCtx<Ev> {
    sink: EventSink<Ev>,
    cancel: Arc<AtomicBool>,
}

impl<Ev> WorkerCtx<Ev> {
    /// Emit one event.
    pub fn emit(&self, ev: Ev) {
        self.sink.emit(ev);
    }

    /// The event sink, for handing to code that emits many events.
    pub fn sink(&self) -> &EventSink<Ev> {
        &self.sink
    }

    /// Whether cancellation has been requested for the current operation.
    pub fn is_cancelled(&self) -> bool {
        self.cancel.load(Ordering::Acquire)
    }

    /// The raw cancel flag, for passing to processing functions that take
    /// `&AtomicBool` (e.g. `chop_to_temp`, `process_edit`).
    pub fn cancel_flag(&self) -> &AtomicBool {
        &self.cancel
    }

    /// Clear the cancel flag at the start of a fresh operation so a stale cancel
    /// from a previous command does not abort this one.
    pub fn reset_cancel(&self) {
        self.cancel.store(false, Ordering::Release);
    }
}

/// Handle to a background worker. Sends commands, polls events; `Drop` requests
/// cancellation, closes the command channel (ending the worker's recv loop), and
/// joins the thread.
///
/// Workers expose a thin newtype around this rather than the raw handle, so
/// worker-specific `send` semantics (mapping a Cancel command to
/// [`request_cancel`](Self::request_cancel)) live at the edge, not in the loop.
pub struct WorkerHandle<Cmd, Ev> {
    cmd_tx: Option<mpsc::Sender<Cmd>>,
    event_rx: Mutex<mpsc::Receiver<Ev>>,
    cancel: Arc<AtomicBool>,
    thread: Option<thread::JoinHandle<()>>,
}

impl<Cmd, Ev> WorkerHandle<Cmd, Ev> {
    /// Poll for the next event without blocking.
    pub fn try_recv(&self) -> Option<Ev> {
        self.event_rx.lock().ok()?.try_recv().ok()
    }

    /// Enqueue a command for the worker. Returns `false` if the worker is no
    /// longer alive, its receive loop has ended (e.g. `init` failed, or the
    /// thread panicked out past the guard), so the command was dropped. Callers
    /// that set a "busy" flag on dispatch must check this: setting busy for a
    /// command that never reached a worker means no terminal event will ever
    /// clear the flag.
    #[must_use]
    pub fn send(&self, cmd: Cmd) -> bool {
        match &self.cmd_tx {
            Some(tx) => tx.send(cmd).is_ok(),
            None => false,
        }
    }

    /// Request cancellation of the in-flight operation synchronously (on the
    /// caller's thread). Newtype handles call this from `send` for their Cancel
    /// command so the flag is set before the running operation next checks it.
    pub fn request_cancel(&self) {
        self.cancel.store(true, Ordering::Release);
    }
}

impl<Cmd, Ev> Drop for WorkerHandle<Cmd, Ev> {
    fn drop(&mut self) {
        // Abort any in-flight long operation, then close the channel so the
        // recv loop ends, then join.
        self.cancel.store(true, Ordering::Release);
        self.cmd_tx = None;
        if let Some(handle) = self.thread.take() {
            let _ = handle.join();
        }
    }
}

/// Spawn a background worker.
///
/// - `init` runs once on the worker thread to build per-worker state (open a DB
///   or store, etc.). If it returns `Err`, `on_init_err` maps the error to an
///   event, that event is emitted, and the thread exits.
/// - `step` handles one command, using the [`WorkerCtx`] to emit any number of
///   events. Every `step` call is wrapped in `catch_unwind`; a panic is mapped
///   to a terminal event via `on_panic(&state)` (so the UI un-wedges for the
///   operation that was in flight) and the loop continues (the worker survives).
///   `on_panic` receives `&State` so it can carry context a step recorded before
///   panicking (e.g. the hash of the edit being processed).
///
/// For workers with no persistent state, use `State = ()` and an infallible
/// `init` returning `Ok::<(), std::convert::Infallible>(())` with
/// `on_init_err = |e| match e {}`.
pub fn spawn_worker<Cmd, Ev, State, InitErr>(
    name: &str,
    init: impl FnOnce() -> Result<State, InitErr> + Send + 'static,
    on_init_err: impl FnOnce(InitErr) -> Ev + Send + 'static,
    on_panic: impl Fn(&State) -> Ev + Send + 'static,
    mut step: impl FnMut(&mut State, Cmd, &WorkerCtx<Ev>) + Send + 'static,
) -> std::io::Result<WorkerHandle<Cmd, Ev>>
where
    Cmd: Send + 'static,
    Ev: Send + 'static,
    State: 'static,
    InitErr: 'static,
{
    // Commands stay unbounded (low-volume, one per user action, blocking the GUI
    // on a command send would be worse than a tiny queue). Only events are bounded.
    let (cmd_tx, cmd_rx) = mpsc::channel::<Cmd>();
    let (event_tx, event_rx) = mpsc::sync_channel::<Ev>(EVENT_CHANNEL_CAP);
    let cancel = Arc::new(AtomicBool::new(false));
    let cancel_worker = Arc::clone(&cancel);

    #[allow(clippy::disallowed_methods)] // sanctioned spawn point (per-step catch_unwind)
    let thread = thread::Builder::new()
        .name(name.to_string())
        .spawn(move || {
            let ctx = WorkerCtx {
                sink: EventSink {
                    tx: event_tx,
                    cancel: Arc::clone(&cancel_worker),
                },
                cancel: cancel_worker,
            };
            let mut state = match init() {
                Ok(s) => s,
                Err(e) => {
                    ctx.emit(on_init_err(e));
                    return;
                }
            };
            // Owned, guarded recv loop, the only consumer of `cmd_rx`, which
            // never leaves this function. A panic in `step` becomes an event,
            // not a dead thread.
            while let Ok(cmd) = cmd_rx.recv() {
                let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                    step(&mut state, cmd, &ctx);
                }));
                if result.is_err() {
                    ctx.emit(on_panic(&state));
                }
            }
        })?;

    Ok(WorkerHandle {
        cmd_tx: Some(cmd_tx),
        event_rx: Mutex::new(event_rx),
        cancel,
        thread: Some(thread),
    })
}

/// Handle to a one-shot job thread (see [`spawn_job`]). Poll [`try_recv`](Self::try_recv)
/// for the single result event; `Drop` joins the thread.
pub struct JobHandle<Ev> {
    event_rx: Mutex<mpsc::Receiver<Ev>>,
    thread: Option<thread::JoinHandle<()>>,
}

impl<Ev> JobHandle<Ev> {
    /// Poll for the job's result without blocking.
    pub fn try_recv(&self) -> Option<Ev> {
        self.event_rx.lock().ok()?.try_recv().ok()
    }
}

impl<Ev> Drop for JobHandle<Ev> {
    fn drop(&mut self) {
        if let Some(handle) = self.thread.take() {
            let _ = handle.join();
        }
    }
}

/// Spawn a one-shot job thread: `body` runs once, its return value is sent back
/// as the single result event, and the thread exits.
///
/// This is the one-shot counterpart to [`spawn_worker`] (which owns a command
/// recv loop). A panic in `body` is caught and mapped to `on_panic()`, the same
/// isolation guarantee, so the GUI's `try_recv` always resolves a terminal event
/// instead of hanging on a silently-dead thread. Use this for any
/// run-once-then-exit thread (classifier jobs, one-shot library passes); use
/// `spawn_worker` for a long-lived command loop, and [`spawn_detached`] for a
/// thread that restores its own state via Drop and reports no event. Bare
/// `thread::spawn` / `Builder::spawn` is denied by clippy (`disallowed-methods`):
/// every worker thread must route through one of these three panic-isolated
/// seals, so a new recv/stream loop cannot reintroduce the silent-stuck-state
/// failure that recurred across three ultra-fuzz runs.
pub fn spawn_job<Ev, F>(
    name: &str,
    on_panic: impl FnOnce() -> Ev + Send + 'static,
    body: F,
) -> std::io::Result<JobHandle<Ev>>
where
    Ev: Send + 'static,
    F: FnOnce() -> Ev + Send + 'static,
{
    let (tx, rx) = mpsc::channel::<Ev>();
    #[allow(clippy::disallowed_methods)] // sanctioned spawn point (panic-isolated)
    let thread = thread::Builder::new()
        .name(name.to_string())
        .spawn(move || {
            let ev = std::panic::catch_unwind(std::panic::AssertUnwindSafe(body))
                .unwrap_or_else(|_| on_panic());
            let _ = tx.send(ev);
        })?;
    Ok(JobHandle {
        event_rx: Mutex::new(rx),
        thread: Some(thread),
    })
}

/// Spawn a detached thread whose `body` restores its own invariants on exit
/// (typically via an RAII guard whose `Drop` resets shared state) and does NOT
/// report a terminal event.
///
/// `body` runs inside `catch_unwind`, so a panic unwinds it, running its
/// destructors, which is where the state restoration happens, and is then
/// logged instead of silently killing the thread with state left wedged. This is
/// the sanctioned spawn for the recoverable detached threads that recover by
/// replacement rather than by emitting an event (the streaming preview decode,
/// whose `StreamingLease::drop` clears the `streaming` flag on every exit path
/// including panic). `spawn_job`/`spawn_worker` cover everything that reports a
/// result. A bare `thread::spawn` is denied by clippy precisely so this seal
/// (Drop + catch_unwind) can't be bypassed by a new recv/stream loop, the exact
/// regression that recurred across three ultra-fuzz runs.
pub fn spawn_detached(name: &str, body: impl FnOnce() + Send + 'static) -> std::io::Result<()> {
    let label = name.to_string();
    #[allow(clippy::disallowed_methods)] // sanctioned spawn point (panic-isolated)
    thread::Builder::new().name(label.clone()).spawn(move || {
        if std::panic::catch_unwind(std::panic::AssertUnwindSafe(body)).is_err() {
            tracing::error!(thread = %label, "detached thread panicked; state restored via Drop");
        }
    })?;
    Ok(())
}

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

    fn poll_job<Ev>(handle: &JobHandle<Ev>) -> Ev {
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
        loop {
            if let Some(ev) = handle.try_recv() {
                return ev;
            }
            assert!(std::time::Instant::now() < deadline, "job never emitted");
            std::thread::sleep(std::time::Duration::from_millis(1));
        }
    }

    #[test]
    fn spawn_job_delivers_result() {
        let handle = spawn_job("test-job-ok", || 0u32, || 42u32).unwrap();
        assert_eq!(poll_job(&handle), 42);
    }

    #[test]
    fn spawn_job_maps_panic_to_event() {
        // A panicking job body must surface the on_panic event, not hang try_recv.
        let handle = spawn_job("test-job-panic", || "panicked", || panic!("boom")).unwrap();
        assert_eq!(poll_job(&handle), "panicked");
    }

    fn recv_one<Cmd, Ev>(handle: &WorkerHandle<Cmd, Ev>) -> Ev {
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
        while std::time::Instant::now() < deadline {
            if let Some(ev) = handle.try_recv() {
                return ev;
            }
            std::thread::sleep(std::time::Duration::from_millis(1));
        }
        panic!("worker produced no event");
    }

    #[test]
    fn step_panic_becomes_event_and_worker_survives() {
        // A command that panics must not kill the worker: the next command still
        // gets processed. This is the invariant the loose_files regression broke.
        let handle = spawn_worker::<u32, &'static str, (), Infallible>(
            "test-panic",
            || Ok(()),
            |e| match e {},
            |_state| "panicked",
            |_state, cmd: u32, ctx| {
                assert!(cmd != 0, "boom");
                ctx.emit("ok");
            },
        )
        .unwrap();

        assert!(handle.send(0)); // panics (delivered, then caught)
        assert_eq!(recv_one(&handle), "panicked");
        assert!(handle.send(1)); // worker must still be alive
        assert_eq!(recv_one(&handle), "ok");
        drop(handle);
    }

    #[test]
    fn init_failure_emits_event_and_exits() {
        let handle = spawn_worker::<u32, &'static str, (), &'static str>(
            "test-init-fail",
            || Err("no db"),
            |e| e,
            |_state| "panicked",
            |_state, _cmd, _ctx| {},
        )
        .unwrap();
        assert_eq!(recv_one(&handle), "no db");
        drop(handle);
    }

    #[test]
    fn drop_sets_cancel_flag() {
        // Drop must request cancellation so an in-flight op aborts promptly.
        let observed = Arc::new(AtomicBool::new(false));
        let observed_clone = Arc::clone(&observed);
        let handle = spawn_worker::<u32, (), (), Infallible>(
            "test-cancel",
            || Ok(()),
            |e| match e {},
            |_state| (),
            move |_state, _cmd: u32, ctx| {
                // Spin until cancelled, then record that we saw it.
                while !ctx.is_cancelled() {
                    std::thread::sleep(std::time::Duration::from_millis(1));
                }
                observed_clone.store(true, Ordering::Release);
            },
        )
        .unwrap();
        let _ = handle.send(1);
        std::thread::sleep(std::time::Duration::from_millis(10));
        drop(handle); // sets cancel, the step exits, join completes
        assert!(observed.load(Ordering::Acquire));
    }
}