DarkRoom/ui/dr-ui/src/activity.rs

//! TRACES: FR-CAT-1 | FR-NC-6 | FR-NC-6c
//! One register of the work running behind the interface.
//!
//! Every background operation in this crate has the same shape: a worker
//! thread with an `mpsc` channel, drained by a `slint::Timer` on the UI thread.
//! Each of them used to report into a window property of its own —
//! `library-thumbs-done`, `library-pin-total`, `library-syncing` — and that had
//! two consequences worth removing.
//!
//! - **Only the grid could see them.** A pin download outlives the view it was
//!   started from, so opening an image made an hour of transfers invisible: the
//!   properties are read by `LibraryGrid` and nothing else.
//! - **There was no answer to "what is this doing".** The answer was spread
//!   across eight properties that no code path ever collected, which is why the
//!   settings page could not show a list of transfers: there was nothing to
//!   list.
//!
//! So the jobs report here instead, and this publishes twice: an aggregate that
//! drives the bar across the top of the shell, and a row per job for the
//! settings page.
//!
//! # Everything here is single-threaded on purpose
//!
//! No `Arc`, no lock. The workers already hand their progress to the UI thread
//! through a channel, and the drain that reads it is the only thing that talks
//! to this register — so the shared-state problem was solved before this file
//! existed, and re-solving it with a mutex would only add a lock that is never
//! contended.
//!
//! # A job that stops reporting cannot hang the bar
//!
//! [`Activity`] is a handle whose `Drop` removes a still-running job. The drain
//! closures own their handle, so a worker that dies mid-transfer, a timer
//! replaced by a newer batch, or a window that closes all take their rows with
//! them. Without that, one lost `Finished` message would leave the bar sweeping
//! for the rest of the session — and a progress indicator that lies about
//! whether anything is happening is worse than none.

use std::cell::{Cell, RefCell};
use std::rc::{Rc, Weak};

use slint::ComponentHandle;

use crate::{ActivityRow, AppWindow};

/// How many stopped jobs to keep.
///
/// A short history rather than none: "did the sync work?" is asked *after* the
/// sync, and a list that empties the instant a job ends can only ever answer
/// questions about the present. Short, because it is a status list and not a
/// log — the ones worth keeping past this are failures, and those are held
/// until the user clears them.
const KEEP_STOPPED: usize = 8;

/// How often the register publishes into the window.
///
/// Publishing on every change would rebuild the row model once per drained
/// message, and a thumbnail batch drains a hundred in a tick. This is fast
/// enough that a bar looks live and slow enough that a burst of completions
/// costs one update rather than a hundred.
const PUBLISH_INTERVAL: std::time::Duration = std::time::Duration::from_millis(150);

/// What kind of work a job is.
///
/// Coarser than the set of functions that start jobs: the user's question is
/// "is something downloading", not which of three call sites issued the fetch.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Kind {
    /// Walking the remote library (FR-CAT-1).
    Scan,
    /// Reading capture times across the whole library.
    Index,
    /// Previews for the cells on screen.
    Thumbnails,
    /// Original files coming down — a pin, or an image being opened.
    Download,
    /// Sidecars and derived data going up.
    Upload,
    /// The two-way exchange of catalog and shards.
    Sync,
    /// Moving files to the trash, restoring them, or emptying it.
    Trash,
    /// TRACES: FR-EXP-7
    /// Rendering and writing finished files.
    Export,
    /// TRACES: FR-DEV-3g
    /// The learned demosaic and denoise working through a photograph.
    Denoise,
}

impl Kind {
    /// Whether this job moves bytes over the network.
    ///
    /// FR-NC-6c: the user's question on a slow or metered connection is about
    /// *transfers* specifically, so the row says which jobs are ones.
    pub fn is_transfer(self) -> bool {
        match self {
            Kind::Scan
            | Kind::Thumbnails
            | Kind::Download
            | Kind::Upload
            | Kind::Sync
            | Kind::Trash
            // A batch export fetches every original it does not already have
            // cached, which on a selection of three hundred RAW files is the
            // largest transfer this application ever starts. It moves nothing
            // at all when the whole selection is already on disk, but the coarse
            // answer has to be the cautious one: the question this flag answers
            // is asked by somebody on a metered connection.
            | Kind::Export => true,
            // Reads headers over the network today, but it is bounded by the
            // catalog rather than by anything the user asked to move, and it
            // stores nothing. Calling it a transfer would put an hours-long
            // background sweep in the same sentence as a download they are
            // waiting on.
            Kind::Index => false,
            // Local work on a file already open.
            Kind::Denoise => false,
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum State {
    Running,
    Done,
    Failed,
}

/// One job's row in the register.
#[derive(Debug, Clone)]
struct Record {
    id: u64,
    kind: Kind,
    title: String,
    detail: String,
    done: usize,
    /// Zero means "no denominator" — a directory walk cannot know its extent,
    /// and a percentage invented for it is worse than admitting as much.
    total: usize,
    state: State,
}

impl Record {
    fn fraction(&self) -> f32 {
        if self.total == 0 {
            return 0.0;
        }
        (self.done as f32 / self.total as f32).clamp(0.0, 1.0)
    }

    fn row(&self) -> ActivityRow {
        ActivityRow {
            title: self.title.clone().into(),
            detail: self.detail.clone().into(),
            fraction: self.fraction(),
            determinate: self.total > 0,
            running: self.state == State::Running,
            failed: self.state == State::Failed,
            transfer: self.kind.is_transfer(),
        }
    }
}

/// The aggregate the shell's bar is drawn from.
#[derive(Debug, Clone, PartialEq)]
pub struct Summary {
    pub running: usize,
    /// Stopped jobs still listed — recent successes and unread failures.
    pub kept: usize,
    /// `None` where nothing running knows its own extent.
    pub fraction: Option<f32>,
}

/// Every background job the application is running.
pub struct ActivityLog {
    records: RefCell<Vec<Record>>,
    next_id: Cell<u64>,
    /// Where to publish. Absent in tests, which is what makes the whole
    /// register testable without a window.
    window: RefCell<Option<slint::Weak<AppWindow>>>,
    pump: RefCell<Option<slint::Timer>>,
    dirty: Cell<bool>,
}

impl Default for ActivityLog {
    /// An empty register attached to nothing.
    ///
    /// Publishing is a no-op until [`ActivityLog::attach`] gives it a window,
    /// which is what lets a controller hold one in a test with no interface
    /// running — and what makes this file's own tests possible.
    fn default() -> Self {
        Self {
            records: RefCell::new(Vec::new()),
            next_id: Cell::new(1),
            window: RefCell::new(None),
            pump: RefCell::new(None),
            dirty: Cell::new(false),
        }
    }
}

impl ActivityLog {
    pub fn new() -> Rc<Self> {
        Rc::new(Self::default())
    }

    /// Start publishing into `window`.
    ///
    /// The timer holds a *weak* reference back: an `Rc` here would be a cycle
    /// through the closure, and the register would outlive the window it is
    /// describing.
    pub fn attach(self: &Rc<Self>, window: &AppWindow) {
        *self.window.borrow_mut() = Some(window.as_weak());

        let weak = Rc::downgrade(self);
        let timer = slint::Timer::default();
        timer.start(slint::TimerMode::Repeated, PUBLISH_INTERVAL, move || {
            let Some(log) = Weak::upgrade(&weak) else {
                return;
            };
            if log.dirty.replace(false) {
                log.publish();
            }
        });
        *self.pump.borrow_mut() = Some(timer);

        // Once immediately, so an empty register starts out saying so rather
        // than leaving whatever the properties defaulted to.
        self.publish();
    }

    /// Register a job and hand back the handle that reports on it.
    pub fn begin(self: &Rc<Self>, kind: Kind, title: impl Into<String>) -> Activity {
        let id = self.next_id.get();
        self.next_id.set(id + 1);

        self.records.borrow_mut().push(Record {
            id,
            kind,
            title: title.into(),
            detail: String::new(),
            done: 0,
            total: 0,
            state: State::Running,
        });
        self.touch();

        Activity {
            log: self.clone(),
            id,
        }
    }

    /// Drop every stopped job, keeping what is still running.
    pub fn clear_finished(&self) {
        self.records
            .borrow_mut()
            .retain(|r| r.state == State::Running);
        self.touch();
    }

    pub fn summary(&self) -> Summary {
        let records = self.records.borrow();
        let running: Vec<&Record> = records
            .iter()
            .filter(|r| r.state == State::Running)
            .collect();

        // Summed across jobs rather than averaged: two downloads of wildly
        // different size are one wait as far as the person watching is
        // concerned, and averaging their percentages would have the bar jump
        // backwards whenever a small job joined a large one.
        let (done, total) = running
            .iter()
            .filter(|r| r.total > 0)
            .fold((0usize, 0usize), |(d, t), r| (d + r.done, t + r.total));

        Summary {
            running: running.len(),
            kept: records.len() - running.len(),
            fraction: (total > 0).then(|| (done as f32 / total as f32).clamp(0.0, 1.0)),
        }
    }

    /// The rows the settings page lists, running jobs first.
    ///
    /// Stable within each group — insertion order — so a list being watched
    /// does not reshuffle itself under the reader every time a count changes.
    fn rows(&self) -> Vec<ActivityRow> {
        let records = self.records.borrow();
        records
            .iter()
            .filter(|r| r.state == State::Running)
            .chain(records.iter().filter(|r| r.state != State::Running))
            .map(Record::row)
            .collect()
    }

    fn publish(&self) {
        let Some(window) = self.window.borrow().as_ref().and_then(slint::Weak::upgrade) else {
            return;
        };

        // Built before anything is written, so no property setter can run with
        // `records` borrowed.
        let rows = self.rows();
        let summary = self.summary();

        window.set_activity_rows(slint::ModelRc::new(slint::VecModel::from(rows)));
        window.set_activity_busy(summary.running > 0);
        window.set_activity_running(summary.running as i32);
        window.set_activity_kept(summary.kept as i32);
        window.set_activity_determinate(summary.fraction.is_some());
        window.set_activity_fraction(summary.fraction.unwrap_or(0.0));
    }

    fn touch(&self) {
        self.dirty.set(true);
    }

    /// Edit a *running* job.
    ///
    /// A stopped one is left alone: a drain loop reads whatever the channel
    /// still holds after the message that ended the job, and a queued progress
    /// update arriving second would otherwise overwrite the failure the user
    /// needs to read with a count that no longer means anything.
    fn with(&self, id: u64, f: impl FnOnce(&mut Record)) {
        if let Some(record) = self
            .records
            .borrow_mut()
            .iter_mut()
            .find(|r| r.id == id && r.state == State::Running)
        {
            f(record);
        }
        self.touch();
    }

    /// Stop a job, keeping it in the list.
    ///
    /// First outcome wins, for the reason given on [`ActivityLog::with`]: a
    /// worker that reports a failure and then closes its channel has failed,
    /// and the closing must not relabel it as finished.
    fn stop(&self, id: u64, state: State, detail: String) {
        {
            let mut records = self.records.borrow_mut();
            let Some(record) = records
                .iter_mut()
                .find(|r| r.id == id && r.state == State::Running)
            else {
                return;
            };
            record.state = state;
            record.detail = detail;
            // A finished job reads as complete whatever it counted: a scan that
            // ends having found forty of an unknown number is done, not 40%.
            if state == State::Done && record.total > 0 {
                record.done = record.total;
            }
            trim(&mut records);
        }
        self.touch();
    }

    /// Take a running job out of the register entirely.
    ///
    /// Returns whether there was one to remove, which is how [`Activity::drop`]
    /// tells "the drain forgot about this" from "it stopped properly".
    fn remove(&self, id: u64) -> bool {
        let removed = {
            let mut records = self.records.borrow_mut();
            let before = records.len();
            records.retain(|r| !(r.id == id && r.state == State::Running));
            records.len() != before
        };
        self.touch();
        removed
    }
}

/// Drop the oldest stopped jobs past [`KEEP_STOPPED`].
///
/// Failures are exempt: a transfer that failed while the user was elsewhere is
/// the single most useful thing this list holds, and letting eight successful
/// thumbnail batches push it out would lose exactly the row worth keeping.
fn trim(records: &mut Vec<Record>) {
    let mut excess = records
        .iter()
        .filter(|r| r.state == State::Done)
        .count()
        .saturating_sub(KEEP_STOPPED);

    records.retain(|r| {
        if excess > 0 && r.state == State::Done {
            excess -= 1;
            return false;
        }
        true
    });
}

/// Bytes as a figure to put beside a transfer.
///
/// Three scales rather than one: a sidecar is a few kilobytes and a RAW file is
/// tens of megabytes, and a single unit makes one of them read as "0.0" or as
/// six digits. One decimal at most — this is a status line, not a measurement.
pub fn describe_bytes(bytes: u64) -> String {
    const KB: f64 = 1024.0;
    const MB: f64 = 1024.0 * KB;
    const GB: f64 = 1024.0 * MB;

    let b = bytes as f64;
    if b >= GB {
        format!("{:.1} GB", b / GB)
    } else if b >= MB {
        format!("{:.1} MB", b / MB)
    } else {
        format!("{:.0} kB", (b / KB).ceil())
    }
}

/// TRACES: FR-NC-6a
/// A download someone is watching, as the line under its bar and how full the
/// bar is: `received` of `total` bytes, and a fraction in 0..1 — or below
/// zero when there is no total, which the bar draws as indeterminate rather
/// than as a position it would have to invent.
pub fn describe_download(received: u64, total: Option<u64>) -> (String, f32) {
    match total.filter(|&t| t > 0) {
        // Nothing yet: the size alone says what the wait is for, where
        // "0 kB of 38.0 MB" would read as a transfer that has stalled.
        Some(t) if received == 0 => (format!("Downloading {}", describe_bytes(t)), 0.0),
        // A file that grew since the scan measured it can overrun the
        // catalog's length; the bar stops full rather than past its end.
        Some(t) => (
            format!(
                "Downloading — {} of {}",
                describe_bytes(received),
                describe_bytes(t.max(received))
            ),
            (received as f64 / t as f64).min(1.0) as f32,
        ),
        None if received > 0 => (format!("Downloading — {}", describe_bytes(received)), -1.0),
        None => ("Downloading…".to_string(), -1.0),
    }
}

/// A running job, held by whatever is reporting on it.
///
/// Every method is idempotent and every one is a no-op once the job has
/// stopped, because the drains that call them are loops over a channel that may
/// deliver a late message after the one that ended the job.
pub struct Activity {
    log: Rc<ActivityLog>,
    id: u64,
}

impl Activity {
    /// What the job is currently doing, in the job's own words.
    pub fn detail(&self, detail: impl Into<String>) {
        let detail = detail.into();
        self.log.with(self.id, |r| r.detail = detail);
    }

    /// How much work there is. Zero leaves the job indeterminate.
    pub fn total(&self, total: usize) {
        self.log.with(self.id, |r| r.total = total);
    }

    /// More work turned up mid-flight — a batch that grew once the plan came
    /// back. Added rather than replaced, so the count already served stays
    /// meaningful.
    pub fn add_total(&self, extra: usize) {
        self.log.with(self.id, |r| r.total += extra);
    }

    /// One unit of work finished.
    pub fn advance(&self) {
        self.log.with(self.id, |r| r.done += 1);
    }

    /// Set both counts at once, for workers that report a running total.
    pub fn progress(&self, done: usize, total: usize) {
        self.log.with(self.id, |r| {
            r.done = done;
            r.total = total;
        });
    }

    /// The job finished. `note` is what the list shows afterwards.
    pub fn finish(&self, note: impl Into<String>) {
        self.log.stop(self.id, State::Done, note.into());
    }

    /// The job finished and is not worth remembering.
    ///
    /// For routine work the user never asked for by name: scrolling the grid
    /// starts a thumbnail batch every second or so, and keeping those would
    /// push a failed transfer out of the list within moments of it happening.
    /// A failure is still kept — only [`Activity::finish`]'s history is
    /// skipped.
    pub fn finish_quietly(&self) {
        self.log.remove(self.id);
    }

    /// The job stopped without doing what it set out to do.
    ///
    /// Kept in the list until the user clears it: this is the row they came to
    /// the settings page to find.
    pub fn fail(&self, message: impl Into<String>) {
        self.log.stop(self.id, State::Failed, message.into());
    }
}

impl Drop for Activity {
    /// A job whose handle goes away while it is still running is removed, not
    /// marked failed. Nobody is coming back to report on it, and a row frozen
    /// at "downloading, 12 of 900" would sit in the list claiming to be live
    /// for the rest of the session.
    fn drop(&mut self) {
        if self.log.remove(self.id) {
            log::debug!("a background job ended without saying so");
        }
    }
}

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

    #[test]
    fn a_download_reads_as_what_it_knows() {
        let mb = 1024 * 1024;
        assert_eq!(
            describe_download(0, None),
            ("Downloading…".to_string(), -1.0)
        );
        assert_eq!(
            describe_download(0, Some(38 * mb)),
            ("Downloading 38.0 MB".to_string(), 0.0)
        );
        let (text, fraction) = describe_download(19 * mb, Some(38 * mb));
        assert_eq!(text, "Downloading — 19.0 MB of 38.0 MB");
        assert_eq!(fraction, 0.5);
        let (text, fraction) = describe_download(12 * mb, None);
        assert_eq!(text, "Downloading — 12.0 MB");
        assert!(fraction < 0.0, "no total, no position");
        let (text, fraction) = describe_download(40 * mb, Some(38 * mb));
        assert_eq!(text, "Downloading — 40.0 MB of 40.0 MB");
        assert_eq!(fraction, 1.0, "an overrun stops full");
    }

    #[test]
    fn a_running_job_makes_the_bar_busy() {
        let log = ActivityLog::new();
        assert_eq!(log.summary().running, 0);

        let scan = log.begin(Kind::Scan, "Scanning");
        assert_eq!(log.summary().running, 1);
        // A scan has no denominator, so the bar must not claim a position.
        assert_eq!(log.summary().fraction, None);

        scan.finish("up to date");
        assert_eq!(log.summary().running, 0);
    }

    #[test]
    fn progress_is_summed_across_jobs() {
        let log = ActivityLog::new();
        let a = log.begin(Kind::Thumbnails, "Thumbnails");
        let b = log.begin(Kind::Download, "Downloading");
        a.progress(1, 10);
        b.progress(4, 10);

        // 5 of 20, not the average of 10% and 40%.
        assert_eq!(log.summary().fraction, Some(0.25));
    }

    #[test]
    fn a_job_without_a_denominator_does_not_dilute_one_that_has_it() {
        let log = ActivityLog::new();
        let scan = log.begin(Kind::Scan, "Scanning");
        let fetch = log.begin(Kind::Download, "Downloading");
        fetch.progress(3, 4);
        scan.detail("120 folders");

        assert_eq!(log.summary().fraction, Some(0.75));
    }

    #[test]
    fn a_dropped_handle_takes_its_row_with_it() {
        let log = ActivityLog::new();
        {
            let _thumbs = log.begin(Kind::Thumbnails, "Thumbnails");
            assert_eq!(log.summary().running, 1);
        }
        // The worker died, or a newer batch replaced the timer holding this.
        // Either way the bar must not sweep for ever.
        assert_eq!(log.summary().running, 0);
        assert_eq!(log.summary().kept, 0);
    }

    #[test]
    fn a_finished_job_is_kept_and_reads_as_complete() {
        let log = ActivityLog::new();
        let sweep = log.begin(Kind::Index, "Indexing");
        sweep.progress(40, 100);
        // Ended early with everything it was going to do done.
        sweep.finish("4000 dated");
        drop(sweep);

        let summary = log.summary();
        assert_eq!(summary.running, 0);
        assert_eq!(summary.kept, 1, "a finished job stays in the list");

        let rows = log.rows();
        assert_eq!(rows[0].fraction, 1.0);
        assert!(!rows[0].running);
        assert!(!rows[0].failed);
    }

    #[test]
    fn a_failure_survives_a_run_of_successes() {
        let log = ActivityLog::new();
        log.begin(Kind::Download, "Downloading")
            .fail("host is down");
        for i in 0..KEEP_STOPPED * 2 {
            log.begin(Kind::Thumbnails, format!("Batch {i}"))
                .finish("done");
        }

        let rows = log.rows();
        assert!(
            rows.iter().any(|r| r.failed),
            "the one row worth keeping was pushed out by routine successes"
        );
        assert!(rows.len() <= KEEP_STOPPED + 1, "history is bounded");
    }

    #[test]
    fn clearing_leaves_running_jobs_alone() {
        let log = ActivityLog::new();
        let scan = log.begin(Kind::Scan, "Scanning");
        log.begin(Kind::Sync, "Syncing").fail("timed out");

        log.clear_finished();
        assert_eq!(log.summary().kept, 0);
        assert_eq!(log.summary().running, 1);
        drop(scan);
    }

    #[test]
    fn running_jobs_are_listed_first() {
        let log = ActivityLog::new();
        log.begin(Kind::Sync, "Syncing").finish("nothing to do");
        let scan = log.begin(Kind::Scan, "Scanning");

        let rows = log.rows();
        assert_eq!(rows[0].title, "Scanning");
        assert!(rows[0].running);
        drop(scan);
    }

    #[test]
    fn a_late_message_cannot_restart_a_stopped_job() {
        let log = ActivityLog::new();
        let fetch = log.begin(Kind::Download, "Downloading");
        fetch.fail("connection reset");
        // The drain reads one more queued message before it notices.
        fetch.advance();

        assert_eq!(log.summary().running, 0);
        assert!(log.rows()[0].failed);
    }
}