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

//! TRACES: FR-EXP-6 | FR-EXP-7 | FR-NC-10
//! Placing an exported file, and the cache that makes offline unremarkable.
//!
//! [`dr_export`] turns a frame into bytes and a name and stops there, because
//! where those bytes go differs by more than a path. This is the other half:
//! it decides the destination and gets them there.
//!
//! # Everything is staged first
//!
//! An export to the server is written to a local **outbox** before any upload
//! is attempted, and the upload drains that outbox afterwards. Not a fallback
//! for the offline case — the *only* path, with offline merely meaning the
//! drain finds nothing to do.
//!
//! Doing it the other way, uploading directly and staging only on failure,
//! looks simpler and has two bad properties. The failure path is then the one
//! that is rarely exercised and always broken, and the moment the network
//! drops mid-batch some exports exist and some do not with nothing recording
//! which. Staging first means an export is *finished* the instant it is
//! written; the upload is a separate promise the app keeps later.
//!
//! # Why the outbox is not in the cache
//!
//! It lives beside the catalog, with the thumbnail shards, rather than under
//! the evictable cache. `dr_catalog::cache` draws the line already: passively
//! cached originals are a convenience and go under LRU, pinned ones are a
//! promise and never do. An export waiting to upload is a promise — the user
//! was told the export succeeded — and sweeping it away to reclaim disk would
//! destroy work that no longer exists anywhere else.
//!
//! # The batch (FR-EXP-7)
//!
//! The second half of this file is a worker that exports a whole selection.
//! It runs on a thread of its own and reaches the interface through the same
//! two mechanisms everything else here does: an `mpsc` channel drained by a
//! Slint timer, and a row in [`crate::activity`].
//!
//! **Why the worker opens its own sessions.** A [`crate::DevelopSession`] owns a GPU
//! `AdjustPass`, and the one the interface is holding is the one the canvas
//! renders from — handing it to a worker would mean the develop view could not
//! draw while a batch ran, which is the freeze this exists to remove. A
//! `GpuContext` is an `Arc` pair over a device and queue and is cheap to
//! clone, so the worker takes a clone and opens each photograph for itself.
//! The cost is a `Demosaicer` and an `AdjustPass` per image rather than one
//! for the run; against a full-resolution decode, render and encode it is
//! small, and it keeps this file out of the pipeline that `develop` owns.
//!
//! **The open image is the exception.** Its edit lives in the interface's
//! session and may not have reached a sidecar yet, so a worker that re-opened
//! the file for itself would export the *saved* version rather than the one on
//! screen. That one frame is therefore rendered by the caller and handed over
//! as [`Source::Rendered`]; everything after the render still moves off the UI
//! thread.

use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::rc::Rc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender};
use std::sync::Arc;
use std::time::Duration;

use dr_export::{Encoded, NameContext};
use dr_sync::{RemoteBackend, RemotePath};
use dr_sync_nextcloud::{AppCredentials, NextcloudBackend};
use dr_types::{ExportSettings, ExportTarget};

use crate::AppWindow;

/// Where exports wait for a server that is not there yet.
///
/// Beside the catalog, for the reason in the module docs. Per account,
/// because the destination folder is a path on one particular server and an
/// entry queued for one account is meaningless to another.
pub fn outbox_dir(server: &str, user_id: &str) -> PathBuf {
    crate::library::catalog_path(server, user_id)
        .parent()
        .map(|p| p.join("outbox"))
        .unwrap_or_else(|| std::env::temp_dir().join("darkroom-outbox"))
}

/// One export waiting to go up.
///
/// The record sits beside the bytes as `<name>.dest`, holding the remote
/// folder it belongs in. A single flat file rather than a database: the queue
/// is small, the entries are independent, and the recovery story for a
/// half-written text file is to ignore it — which is exactly what parsing it
/// does.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Pending {
    /// The staged bytes on this device.
    pub local: PathBuf,
    /// Remote folder, relative to the library root. Empty means the root.
    pub remote_dir: String,
    /// The filename to give it there.
    pub name: String,
}

impl Pending {
    /// Full remote path for this entry, under `root`.
    fn remote_path(&self, root: &str) -> RemotePath {
        let mut parts: Vec<&str> = Vec::new();
        for segment in [root, self.remote_dir.as_str()] {
            for part in segment.split('/') {
                if !part.is_empty() {
                    parts.push(part);
                }
            }
        }
        parts.push(&self.name);
        RemotePath::new(parts.join("/"))
    }

    /// The folder this entry's file belongs in, as a remote path.
    fn remote_folder(&self, root: &str) -> RemotePath {
        let mut parts: Vec<&str> = Vec::new();
        for segment in [root, self.remote_dir.as_str()] {
            for part in segment.split('/') {
                if !part.is_empty() {
                    parts.push(part);
                }
            }
        }
        RemotePath::new(parts.join("/"))
    }
}

/// Where an export was put, for the interface to report.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Placed {
    /// Written straight to a folder on this device.
    Device(PathBuf),
    /// Staged locally, awaiting upload to the named remote folder.
    Queued { local: PathBuf, remote_dir: String },
}

impl Placed {
    /// A sentence for the status line.
    pub fn describe(&self) -> String {
        match self {
            Placed::Device(path) => format!(
                "Exported {}",
                path.file_name()
                    .map(|n| n.to_string_lossy().into_owned())
                    .unwrap_or_default()
            ),
            // Named as queued rather than exported: the file is real and
            // finished, but it is not yet where the user asked for it, and
            // saying "exported to Nextcloud" before it has uploaded would be
            // a claim the app cannot keep if the disk is pulled.
            Placed::Queued { remote_dir, .. } => {
                let dir = if remote_dir.is_empty() {
                    "the library root".to_string()
                } else {
                    remote_dir.clone()
                };
                format!("Queued for {dir}")
            }
        }
    }
}

/// Write an encoded export to wherever the settings say it goes.
///
/// `destination` is a filesystem path for [`ExportTarget::Device`] and a
/// remote folder for [`ExportTarget::Remote`] — the widening `ExportSettings`
/// documents, resolved here because this is the layer that knows what a path
/// means on this platform.
pub fn place(
    encoded: &Encoded,
    target: ExportTarget,
    destination: &str,
    outbox: &Path,
) -> Result<Placed, String> {
    match target {
        ExportTarget::Device => {
            if destination.trim().is_empty() {
                return Err("No export folder is set. Choose one in Settings.".into());
            }
            let dir = PathBuf::from(destination);
            std::fs::create_dir_all(&dir).map_err(|e| format!("{}: {e}", dir.display()))?;
            let path = dir.join(&encoded.name);
            std::fs::write(&path, &encoded.bytes)
                .map_err(|e| format!("{}: {e}", path.display()))?;
            Ok(Placed::Device(path))
        }
        ExportTarget::Remote => {
            let local = stage(encoded, destination, outbox)?;
            Ok(Placed::Queued {
                local,
                remote_dir: destination.to_string(),
            })
        }
    }
}

/// Write bytes and their destination record into the outbox.
fn stage(encoded: &Encoded, remote_dir: &str, outbox: &Path) -> Result<PathBuf, String> {
    std::fs::create_dir_all(outbox).map_err(|e| format!("{}: {e}", outbox.display()))?;

    // The staged name is the export's name, deduplicated against the outbox
    // rather than against the server: two exports queued before either has
    // uploaded would otherwise overwrite each other here, and the second one
    // would silently replace the first before anybody saw it.
    let mut candidate = outbox.join(&encoded.name);
    let mut n = 1;
    while candidate.exists() {
        let stem = Path::new(&encoded.name)
            .file_stem()
            .map(|s| s.to_string_lossy().into_owned())
            .unwrap_or_else(|| "export".into());
        let ext = Path::new(&encoded.name)
            .extension()
            .map(|s| s.to_string_lossy().into_owned())
            .unwrap_or_default();
        candidate = outbox.join(format!("{stem}-{n}.{ext}"));
        n += 1;
        if n > 10_000 {
            return Err("the outbox is full of files by this name".into());
        }
    }

    // Bytes first, then the record. The order matters on a process that may
    // be killed between the two: an orphan payload with no record is ignored
    // by the drain and swept later, where a record naming bytes that were
    // never written would be a permanent failure retried forever.
    std::fs::write(&candidate, &encoded.bytes)
        .map_err(|e| format!("{}: {e}", candidate.display()))?;

    let record = candidate.with_extension(format!(
        "{}.dest",
        candidate
            .extension()
            .map(|s| s.to_string_lossy().into_owned())
            .unwrap_or_default()
    ));
    // The remote folder and the intended name, one per line. Not JSON: two
    // strings do not need a parser, and a format a human can repair by hand
    // is worth something for a queue holding the only copy of someone's work.
    std::fs::write(&record, format!("{remote_dir}\n{}\n", encoded.name))
        .map_err(|e| format!("{}: {e}", record.display()))?;

    Ok(candidate)
}

/// Everything currently waiting in the outbox.
///
/// A payload with no record is skipped rather than guessed at — see the write
/// order in [`stage`].
pub fn pending(outbox: &Path) -> Vec<Pending> {
    let Ok(entries) = std::fs::read_dir(outbox) else {
        return Vec::new();
    };

    let mut out = Vec::new();
    for entry in entries.flatten() {
        let path = entry.path();
        if path.extension().and_then(|e| e.to_str()) != Some("dest") {
            continue;
        }
        // `photo.jpg.dest` describes `photo.jpg`.
        let local = path.with_extension("");
        if !local.exists() {
            continue;
        }
        let Ok(text) = std::fs::read_to_string(&path) else {
            continue;
        };
        let mut lines = text.lines();
        let remote_dir = lines.next().unwrap_or("").to_string();
        let name = lines.next().unwrap_or("").to_string();
        if name.is_empty() {
            continue;
        }
        out.push(Pending {
            local,
            remote_dir,
            name,
        });
    }
    // Stable order so a drain is reproducible and a stuck entry is obvious
    // rather than appearing to move around the queue.
    out.sort_by(|a, b| a.local.cmp(&b.local));
    out
}

/// How many exports are waiting. For the interface to show, and cheap enough
/// to call on a redraw.
pub fn pending_count(outbox: &Path) -> usize {
    pending(outbox).len()
}

/// Remove an entry and its record, once it is safely on the server.
fn clear(entry: &Pending) {
    let record = PathBuf::from(format!("{}.dest", entry.local.display()));
    let _ = std::fs::remove_file(&entry.local);
    let _ = std::fs::remove_file(&record);
}

/// Progress from the upload worker.
#[derive(Debug)]
pub enum UploadMessage {
    Status(String),
    /// Uploaded, still pending, and the first error if there was one.
    Finished {
        uploaded: usize,
        remaining: usize,
        error: Option<String>,
    },
}

/// Drain the outbox to the server.
///
/// Its own thread with its own runtime, like every other network path here —
/// the Slint loop must never block (NFR-P9).
///
/// A failure leaves the entry in place and stops the run. Continuing past a
/// network error would burn the whole queue against a server that is not
/// answering, and the next pass costs nothing.
pub fn spawn_upload(
    creds: AppCredentials,
    user_id: String,
    root: String,
    outbox: PathBuf,
) -> std::sync::mpsc::Receiver<UploadMessage> {
    let (tx, rx) = std::sync::mpsc::channel();

    std::thread::spawn(move || {
        let rt = match crate::net_runtime::build() {
            Ok(rt) => rt,
            Err(e) => {
                let _ = tx.send(UploadMessage::Finished {
                    uploaded: 0,
                    remaining: pending_count(&outbox),
                    error: Some(e.to_string()),
                });
                return;
            }
        };

        rt.block_on(async {
            let backend = match NextcloudBackend::new(&creds, &user_id) {
                Ok(b) => b,
                Err(e) => {
                    let _ = tx.send(UploadMessage::Finished {
                        uploaded: 0,
                        remaining: pending_count(&outbox),
                        error: Some(e.to_string()),
                    });
                    return;
                }
            };

            let queue = pending(&outbox);
            let total = queue.len();
            let mut uploaded = 0;
            let mut error = None;

            for (i, entry) in queue.iter().enumerate() {
                let _ = tx.send(UploadMessage::Status(format!(
                    "uploading {} ({}/{total})",
                    entry.name,
                    i + 1
                )));

                let Ok(bytes) = std::fs::read(&entry.local) else {
                    // The payload vanished under us. Drop the record too;
                    // retrying forever against a file that is gone helps
                    // nobody.
                    clear(entry);
                    continue;
                };

                // The folder may not exist — this is the first export into it
                // — and `create_dir` treats "already there" as success, so it
                // is unconditional rather than guarded by a check that would
                // cost a request every time.
                if let Err(e) = backend.create_dir(&entry.remote_folder(&root)).await {
                    error = Some(e.to_string());
                    break;
                }

                match backend.put(&entry.remote_path(&root), bytes, None).await {
                    Ok(_) => {
                        clear(entry);
                        uploaded += 1;
                    }
                    Err(e) => {
                        error = Some(e.to_string());
                        break;
                    }
                }
            }

            let _ = tx.send(UploadMessage::Finished {
                uploaded,
                remaining: pending_count(&outbox),
                error,
            });
        });
    });

    rx
}

// ---------------------------------------------------------------------------
// Batch export (FR-EXP-7)
// ---------------------------------------------------------------------------

/// TRACES: NFR-ARCH-3
/// The stop flag a running batch reads.
///
/// One bit rather than a channel. Cancellation is read thousands of times more
/// often than it is written, and a message would only be seen if the worker
/// happened to be listening at the moment the button was pressed rather than at
/// the next point it is safe to stop.
///
/// `Relaxed` throughout: the flag guards no other data, so there is nothing for
/// an acquire/release pair to publish. The only ordering that matters is that
/// the write eventually becomes visible, which every ordering guarantees.
#[derive(Clone, Default)]
pub struct Cancel(Arc<AtomicBool>);

impl Cancel {
    pub fn cancel(&self) {
        self.0.store(true, Ordering::Relaxed);
    }

    pub fn is_cancelled(&self) -> bool {
        self.0.load(Ordering::Relaxed)
    }
}

/// How long the worker blocks on another worker before rereading the flag.
///
/// TRACES: NFR-ARCH-3
/// This is the cancellation bound for everything the batch spends its time
/// *waiting* on — a download of tens of megabytes would otherwise hold a
/// cancelled batch open until the transfer finished. It is not the bound for
/// the frame being rendered and encoded when the button is pressed: that has no
/// interior stopping point, so the true worst case is one image, and on a 24 MP
/// frame that is well past NFR-ARCH-3's 100 ms target. Closing that gap needs
/// the render itself to become interruptible (NFR-ARCH-2's scheduler), not a
/// finer poll here.
const CANCEL_POLL: Duration = Duration::from_millis(100);

/// How often the drain looks at what the worker has said.
const DRAIN_INTERVAL: Duration = Duration::from_millis(120);

/// Where one image's pixels come from.
pub enum Source {
    /// A photograph in the library, fetched and rendered by the worker.
    ///
    /// Carries its own cache context because that is assembled from the catalog
    /// and the session, and a worker thread can reach neither.
    Library {
        path: String,
        cache: Option<crate::library::CacheContext>,
    },
    /// A frame the caller has already rendered — the image open in develop.
    ///
    /// See the module docs for why this one cannot be left to the worker.
    Rendered {
        stem: String,
        frame: dr_export::Frame,
    },
}

impl Source {
    /// What to call this image in a progress row.
    fn describe(&self) -> String {
        match self {
            Source::Library { path, .. } => path.rsplit('/').next().unwrap_or(path).to_string(),
            Source::Rendered { stem, .. } => stem.clone(),
        }
    }
}

/// Everything a batch needs, assembled on the UI thread.
///
/// Assembled there and not in the worker because most of it is only reachable
/// from a `RefCell` the interface owns: the session, the settings record, the
/// catalog. Passing the finished answers means the worker borrows nothing.
pub struct BatchRequest {
    pub sources: Vec<Source>,
    /// Credentials for the account the library is open on. `None` where no
    /// library is open, which is fine for a [`Source::Rendered`] and fatal for
    /// anything that has to be fetched.
    pub creds: Option<(AppCredentials, String)>,
    pub settings: ExportSettings,
    pub outbox: PathBuf,
    pub sidecar_cache: PathBuf,
    pub offline: bool,
    /// `None` on a build with no adapter, where a library image cannot be
    /// rendered at all — reported per image rather than refused up front, so
    /// the reason lands in the same place every other failure does.
    pub gpu: Option<dr_gpu::GpuContext>,
}

/// TRACES: NFR-ARCH-4
/// Why one image of a batch produced no file.
///
/// One variant per stage. "Export failed" repeated forty times is not a report
/// anybody can act on: a destination folder that cannot be written is one fix,
/// a body rawler cannot decode is another, and a server that went away is a
/// third — and only the first is worth stopping the batch to correct.
#[derive(Debug, thiserror::Error)]
pub enum ItemError {
    #[error("could not be fetched: {0}")]
    Fetch(String),

    #[error("could not be opened for editing: {0}")]
    Open(String),

    #[error("could not be rendered: {0}")]
    Render(String),

    /// The name was taken and the collision policy is Skip. Not really a
    /// failure — the user asked for exactly this — but it is still an image
    /// that produced no file, and a batch reporting "40 exported" when it wrote
    /// 31 would be lying.
    #[error("a file of that name is already there, and the collision setting is Skip")]
    NameTaken,

    #[error(transparent)]
    Encode(#[from] dr_export::ExportError),

    #[error("could not be written: {0}")]
    Place(String),
}

/// What the worker says as it goes.
#[derive(Debug)]
pub enum BatchMessage {
    /// Work has begun on one image; `done` counts the ones behind it.
    Started {
        done: usize,
        total: usize,
        name: String,
    },
    /// One image is finished, for better or worse.
    Item {
        name: String,
        outcome: Result<Placed, ItemError>,
    },
    Finished {
        exported: usize,
        failed: usize,
        cancelled: bool,
    },
}

/// Export a selection on a thread of its own.
pub fn spawn_batch(request: BatchRequest, cancel: Cancel) -> Receiver<BatchMessage> {
    let (tx, rx) = std::sync::mpsc::channel();
    std::thread::spawn(move || run(request, &cancel, &tx));
    rx
}

/// The batch itself, one image at a time.
///
/// Sequential rather than parallel, which FR-EXP-7's "uses all available cores"
/// does not yet get. One reason and one excuse: the reason is that a
/// full-resolution frame is tens of megabytes and four in flight is a
/// straightforward way to exhaust a tablet; the excuse is that the GPU is
/// shared with the interface, and the render is where the time goes.
fn run(mut request: BatchRequest, cancel: &Cancel, tx: &Sender<BatchMessage>) {
    let sources = std::mem::take(&mut request.sources);
    let total = sources.len();

    // Names this run has already written. See [`resolve_batch_name`] for why
    // the destination alone is not enough to keep two exports apart.
    let mut issued: HashSet<String> = HashSet::new();
    let (mut exported, mut failed) = (0usize, 0usize);

    for (i, source) in sources.into_iter().enumerate() {
        if cancel.is_cancelled() {
            break;
        }

        let name = source.describe();
        let _ = tx.send(BatchMessage::Started {
            done: i,
            total,
            name: name.clone(),
        });

        // `None` is cancellation mid-image, which is not an outcome for this
        // photograph: nothing failed, the user simply stopped asking.
        let Some(outcome) = export_one(&request, source, i as u32 + 1, &mut issued, cancel) else {
            break;
        };

        // Counted, reported, and then on to the next one. A failure here must
        // not end the run — the whole point of exporting three hundred frames
        // unattended is that the one unreadable file is a line in a report
        // rather than an evening lost (FR-EXP-7).
        match &outcome {
            Ok(_) => exported += 1,
            Err(_) => failed += 1,
        }
        let _ = tx.send(BatchMessage::Item { name, outcome });
    }

    let _ = tx.send(BatchMessage::Finished {
        exported,
        failed,
        cancelled: cancel.is_cancelled(),
    });
}

/// One image, start to finish. `None` where the run was cancelled part way.
fn export_one(
    request: &BatchRequest,
    source: Source,
    sequence: u32,
    issued: &mut HashSet<String>,
    cancel: &Cancel,
) -> Option<Result<Placed, ItemError>> {
    let (stem, date, frame) = match source {
        Source::Rendered { stem, frame } => (stem, String::new(), frame),
        Source::Library { path, cache } => {
            match render_from_library(request, &path, cache, cancel)? {
                Ok(rendered) => rendered,
                Err(e) => return Some(Err(e)),
            }
        }
    };

    Some(place_frame(request, &stem, &date, sequence, &frame, issued))
}

/// Fetch a photograph, apply its stored edit, and render it at full size.
fn render_from_library(
    request: &BatchRequest,
    path: &str,
    cache: Option<crate::library::CacheContext>,
    cancel: &Cancel,
) -> Option<Result<(String, String, dr_export::Frame), ItemError>> {
    let Some((creds, user_id)) = request.creds.clone() else {
        return Some(Err(ItemError::Fetch("no library is open".into())));
    };
    let Some(gpu) = request.gpu.as_ref() else {
        return Some(Err(ItemError::Open(
            "this build found no GPU adapter".into(),
        )));
    };

    // Both started before either is waited on, exactly as opening an image in
    // develop does: the sidecar is three orders of magnitude smaller than the
    // RAW, so it costs nothing to have in hand by the time there is a session
    // to apply it to.
    let sidecar_rx = crate::library::spawn_sidecar_fetch(
        creds.clone(),
        user_id.clone(),
        path.to_string(),
        request.sidecar_cache.clone(),
        request.offline,
    );
    let bytes_rx = crate::library::spawn_full_fetch(creds, user_id, path.to_string(), cache);

    let bytes = match wait_for(&bytes_rx, cancel) {
        Waited::Got(Ok(bytes)) => bytes,
        Waited::Got(Err(e)) => return Some(Err(ItemError::Fetch(e.message))),
        Waited::Cancelled => return None,
        Waited::Silent => {
            return Some(Err(ItemError::Fetch(
                "the download ended without answering".into(),
            )))
        }
    };

    let sidecar = match wait_for(&sidecar_rx, cancel) {
        Waited::Got(sidecar) => sidecar,
        Waited::Cancelled => return None,
        // An unedited photograph has no sidecar and a fetch that died looks
        // identical from here. Exporting at defaults is what opening it would
        // do, and refusing the image over a missing edit it may never have had
        // would fail the common case.
        Waited::Silent => None,
    };

    let meta = dr_decode::metadata(&bytes).unwrap_or_default();
    let orientation = meta.orientation.unwrap_or_default();
    // `{date}` is the capture date, not today's: a template naming exports by
    // when the shutter fired is the reason the token exists.
    let date = meta
        .captured_at
        .map(crate::library_ui::format_date)
        .unwrap_or_default();

    let mut session = match crate::open_session(gpu, &bytes, orientation) {
        Ok(session) => session,
        Err(e) => return Some(Err(ItemError::Open(e))),
    };

    // TRACES: FR-CAT-8
    // The stored edit. FR-EXP-9 is about resolution; this is the other half of
    // exporting what the user actually has, and a batch that skipped it would
    // write out three hundred unedited frames without saying so.
    if let Some(version) = sidecar
        .as_ref()
        .and_then(dr_pipeline::Sidecar::default_version)
    {
        session.apply_version(version);
    }

    // The last cheap place to stop. Everything past here is a full-resolution
    // render and an encode with no interior stopping point — see [`CANCEL_POLL`].
    if cancel.is_cancelled() {
        return None;
    }

    let frame = match session.render_for_export(request.settings.colour_space) {
        Ok(frame) => frame,
        Err(e) => return Some(Err(ItemError::Render(e))),
    };

    let stem = Path::new(path)
        .file_stem()
        .map(|s| s.to_string_lossy().into_owned())
        .unwrap_or_else(|| "export".into());

    Some(Ok((stem, date, frame)))
}

/// Name, encode and write one rendered frame.
///
/// The tail every source shares, whoever rendered it.
fn place_frame(
    request: &BatchRequest,
    stem: &str,
    date: &str,
    sequence: u32,
    frame: &dr_export::Frame,
    issued: &mut HashSet<String>,
) -> Result<Placed, ItemError> {
    // The size is resolved before the name because `{dimensions}` is one of the
    // tokens a template can carry.
    let (width, height) = dr_export::target_size(
        frame.width,
        frame.height,
        request.settings.sizing,
        request.settings.allow_upscaling,
    );

    let ctx = NameContext {
        source_stem: stem,
        sequence,
        date,
        width,
        height,
        preset: "",
    };

    let name = resolve_batch_name(&request.settings, &ctx, issued).ok_or(ItemError::NameTaken)?;
    let encoded = dr_export::export(frame, &request.settings, name)?;

    place(
        &encoded,
        request.settings.target,
        &request.settings.destination,
        &request.outbox,
    )
    .map_err(ItemError::Place)
}

/// The name this export takes, avoiding both what was in the destination and
/// what this run has already written.
///
/// Those are two different collisions and they deserve two different answers.
/// [`dr_types::CollisionPolicy`] is the user's answer to "a file of this name
/// was already there", and Overwrite is a perfectly reasonable one. It is not
/// an answer to "the frame I exported four seconds ago was also called this":
/// two photographs of the same stem in different folders are ordinary in a
/// library, and a batch that quietly handed back fewer files than images —
/// having destroyed its own output — is not something anybody asked for. So a
/// name this run issued is always stepped past, whatever the policy says about
/// the folder.
fn resolve_batch_name(
    settings: &ExportSettings,
    ctx: &NameContext<'_>,
    issued: &mut HashSet<String>,
) -> Option<String> {
    let dir = PathBuf::from(&settings.destination);
    let taken = |name: &str| -> bool {
        match settings.target {
            dr_types::ExportTarget::Device => dir.join(name).exists(),
            // A queued export cannot see the server, and may never be able to.
            // Names are kept apart in the outbox instead — see [`stage`].
            dr_types::ExportTarget::Remote => false,
        }
    };

    let name = dr_export::resolve_name(
        &settings.filename_template,
        ctx,
        settings.format,
        settings.collision,
        &|name| taken(name) || issued.contains(name),
    )?;

    // Only reachable under Overwrite, which hands back the taken name by
    // design. Skip and Increment have already been through the closure above.
    let name = if issued.contains(&name) {
        step_past(&name, &|candidate| {
            taken(candidate) || issued.contains(candidate)
        })?
    } else {
        name
    };

    issued.insert(name.clone());
    Some(name)
}

/// `photo.jpg` → `photo-1.jpg`, and on until the name is free.
fn step_past(name: &str, taken: &dyn Fn(&str) -> bool) -> Option<String> {
    let path = Path::new(name);
    let stem = path
        .file_stem()
        .map(|s| s.to_string_lossy().into_owned())
        .unwrap_or_else(|| "export".into());
    let ext = path
        .extension()
        .map(|s| s.to_string_lossy().into_owned())
        .unwrap_or_default();

    // Bounded for the same reason `resolve_name`'s own search is: a destination
    // that reports every name as taken has to fail rather than spin.
    (1..10_000)
        .map(|n| {
            if ext.is_empty() {
                format!("{stem}-{n}")
            } else {
                format!("{stem}-{n}.{ext}")
            }
        })
        .find(|candidate| !taken(candidate))
}

/// What waiting on another worker produced.
///
/// Three answers rather than an `Option`, because "the user cancelled" and "the
/// worker died without a word" have the same shape and want opposite treatment:
/// one ends the batch quietly, the other is a failure belonging to one image.
enum Waited<T> {
    Got(T),
    Cancelled,
    Silent,
}

/// Block on another worker's answer without going deaf to cancellation.
fn wait_for<T>(rx: &Receiver<T>, cancel: &Cancel) -> Waited<T> {
    loop {
        match rx.recv_timeout(CANCEL_POLL) {
            Ok(value) => return Waited::Got(value),
            Err(RecvTimeoutError::Timeout) => {
                if cancel.is_cancelled() {
                    return Waited::Cancelled;
                }
            }
            Err(RecvTimeoutError::Disconnected) => return Waited::Silent,
        }
    }
}

/// Which status line a batch writes its commentary to.
///
/// One drain serves both entry points rather than two near-identical timers, so
/// it has to be told where the words go: develop has its own line beside the
/// export button, and the grid has the header's.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Reporting {
    Develop,
    Library,
}

/// TRACES: FR-EXP-7 | NFR-P9
/// Drain a batch's progress on the UI thread.
///
/// `slot` holds the timer, and dropping what was there before is what stops a
/// superseded batch's drain — and, through [`crate::activity::Activity`]'s
/// `Drop`, takes its row with it.
pub fn drain_batch(
    weak: slint::Weak<AppWindow>,
    activity: &Rc<crate::activity::ActivityLog>,
    slot: &Rc<std::cell::RefCell<Option<slint::Timer>>>,
    rx: Receiver<BatchMessage>,
    total: usize,
    reporting: Reporting,
    // Run when the batch finishes having written something. A queued export
    // is finished on disk but not where the user asked for it, and waiting
    // for the next sync pass to notice reads — correctly — as an export that
    // did not upload.
    on_exported: impl Fn() + 'static,
) {
    let job = activity.begin(
        crate::activity::Kind::Export,
        if total == 1 {
            "Exporting".to_string()
        } else {
            format!("Exporting {total} images")
        },
    );
    job.total(total);

    // The first failure, kept for the summary. Only the first: a run where
    // every image failed for the same reason should say that reason once, and
    // the rest are in the log.
    let mut first_failure: Option<String> = None;

    let timer = slint::Timer::default();
    let held = slot.clone();

    timer.start(slint::TimerMode::Repeated, DRAIN_INTERVAL, move || {
        let Some(w) = weak.upgrade() else { return };

        loop {
            let message = match rx.try_recv() {
                Ok(m) => m,
                Err(std::sync::mpsc::TryRecvError::Empty) => return,
                Err(std::sync::mpsc::TryRecvError::Disconnected) => {
                    // A worker that died without a word must not leave the
                    // button saying "Cancel export" for the rest of the session.
                    job.fail("ended unexpectedly");
                    settle(&w, reporting, "Export ended unexpectedly");
                    stop_timer(&held);
                    return;
                }
            };

            match message {
                BatchMessage::Started { done, total, name } => {
                    job.progress(done, total);
                    job.detail(name.clone());
                    report(
                        &w,
                        reporting,
                        &format!("Exporting {name} ({}/{total})", done + 1),
                    );
                }
                BatchMessage::Item { name, outcome } => match outcome {
                    Ok(placed) => log::info!("{name}: {}", placed.describe()),
                    Err(e) => {
                        // Every failure is logged, not only the first: the
                        // summary is a sentence and this is the record of which
                        // photographs it is about (NFR-ARCH-4).
                        log::warn!("exporting {name}: {e}");
                        first_failure.get_or_insert_with(|| format!("{name}: {e}"));
                    }
                },
                BatchMessage::Finished {
                    exported,
                    failed,
                    cancelled,
                } => {
                    let (text, is_failure) =
                        summarise(exported, failed, cancelled, first_failure.as_deref());
                    if is_failure {
                        job.fail(text.clone());
                    } else {
                        job.finish(text.clone());
                    }
                    settle(&w, reporting, &text);
                    if exported > 0 {
                        on_exported();
                    }
                    stop_timer(&held);
                    return;
                }
            }
        }
    });

    *slot.borrow_mut() = Some(timer);
}

/// How a finished batch reads, and whether it counts as a failure.
///
/// A cancelled run is never a failure however little it exported: the user
/// stopped it, and a red row telling them so is the application arguing. A run
/// with even one failure is, and stays in the list until it is cleared — that
/// is the row somebody came looking for.
fn summarise(
    exported: usize,
    failed: usize,
    cancelled: bool,
    first_failure: Option<&str>,
) -> (String, bool) {
    if cancelled {
        return (format!("Cancelled after {exported}"), false);
    }
    if failed == 0 {
        return (format!("Exported {exported}"), false);
    }
    let detail = first_failure.unwrap_or("see the log");
    (
        format!("Exported {exported}, {failed} failed — {detail}"),
        true,
    )
}

/// Say what the batch is doing on the line the caller came from.
fn report(window: &AppWindow, reporting: Reporting, text: &str) {
    match reporting {
        Reporting::Develop => window.set_export_status(text.into()),
        Reporting::Library => window.set_library_status(text.into()),
    }
}

/// Report, and put the buttons back.
fn settle(window: &AppWindow, reporting: Reporting, text: &str) {
    report(window, reporting, text);
    match reporting {
        Reporting::Develop => window.set_export_busy(false),
        Reporting::Library => window.set_library_exporting(false),
    }
}

fn stop_timer(slot: &Rc<std::cell::RefCell<Option<slint::Timer>>>) {
    if let Some(timer) = slot.borrow().as_ref() {
        timer.stop();
    }
}

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

    fn encoded(name: &str, bytes: &[u8]) -> Encoded {
        Encoded {
            name: name.to_string(),
            bytes: bytes.to_vec(),
            width: 4,
            height: 4,
        }
    }

    fn tmp() -> PathBuf {
        let dir = std::env::temp_dir().join(format!(
            "dr-outbox-test-{}-{:?}",
            std::process::id(),
            std::thread::current().id()
        ));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        dir
    }

    #[test]
    fn a_device_export_writes_the_file() {
        let dir = tmp();
        let target = dir.join("exports");
        let placed = place(
            &encoded("a.jpg", b"hello"),
            ExportTarget::Device,
            target.to_str().unwrap(),
            &dir.join("outbox"),
        )
        .unwrap();

        assert_eq!(placed, Placed::Device(target.join("a.jpg")));
        assert_eq!(std::fs::read(target.join("a.jpg")).unwrap(), b"hello");
    }

    #[test]
    fn a_device_export_creates_a_folder_that_is_not_there() {
        // Exporting into a folder the user typed but has not made is the
        // common case, not an error.
        let dir = tmp();
        let target = dir.join("deep/nested/exports");
        assert!(place(
            &encoded("a.jpg", b"x"),
            ExportTarget::Device,
            target.to_str().unwrap(),
            &dir,
        )
        .is_ok());
        assert!(target.join("a.jpg").exists());
    }

    #[test]
    fn a_device_export_with_no_folder_says_so() {
        // Rather than writing to the process's working directory, which is
        // wherever the app happened to be launched from.
        let dir = tmp();
        let err = place(&encoded("a.jpg", b"x"), ExportTarget::Device, "  ", &dir).unwrap_err();
        assert!(err.contains("Settings"), "unhelpful message: {err}");
    }

    #[test]
    fn a_remote_export_is_staged_rather_than_sent() {
        // The property the offline story rests on: the export is complete on
        // disk before any network call is attempted.
        let dir = tmp();
        let outbox = dir.join("outbox");
        let placed = place(
            &encoded("a.jpg", b"hello"),
            ExportTarget::Remote,
            "Exports/2026",
            &outbox,
        )
        .unwrap();

        match placed {
            Placed::Queued { local, remote_dir } => {
                assert_eq!(std::fs::read(&local).unwrap(), b"hello");
                assert_eq!(remote_dir, "Exports/2026");
            }
            other => panic!("expected a queued export, got {other:?}"),
        }
    }

    #[test]
    fn a_staged_export_is_found_again_with_its_destination() {
        // What survives a process death: the drain has to be able to
        // reconstruct where a file was going from the disk alone.
        let dir = tmp();
        let outbox = dir.join("outbox");
        place(
            &encoded("a.jpg", b"one"),
            ExportTarget::Remote,
            "Exports",
            &outbox,
        )
        .unwrap();

        let queue = pending(&outbox);
        assert_eq!(queue.len(), 1);
        assert_eq!(queue[0].remote_dir, "Exports");
        assert_eq!(queue[0].name, "a.jpg");
    }

    #[test]
    fn two_exports_of_the_same_name_both_survive_the_outbox() {
        // Both were asked for and neither has uploaded, so the second must
        // not overwrite the first while it waits.
        let dir = tmp();
        let outbox = dir.join("outbox");
        place(
            &encoded("a.jpg", b"one"),
            ExportTarget::Remote,
            "E",
            &outbox,
        )
        .unwrap();
        place(
            &encoded("a.jpg", b"two"),
            ExportTarget::Remote,
            "E",
            &outbox,
        )
        .unwrap();

        let queue = pending(&outbox);
        assert_eq!(queue.len(), 2);
        // Both still claim the name they should arrive under; only the local
        // staging name differs.
        assert!(queue.iter().all(|p| p.name == "a.jpg"));
        assert_ne!(queue[0].local, queue[1].local);
    }

    #[test]
    fn a_payload_with_no_record_is_ignored() {
        // The window a kill between the two writes leaves behind. It must not
        // become an upload to nowhere.
        let dir = tmp();
        let outbox = dir.join("outbox");
        std::fs::create_dir_all(&outbox).unwrap();
        std::fs::write(outbox.join("orphan.jpg"), b"x").unwrap();
        assert!(pending(&outbox).is_empty());
    }

    #[test]
    fn a_record_with_no_payload_is_ignored() {
        let dir = tmp();
        let outbox = dir.join("outbox");
        std::fs::create_dir_all(&outbox).unwrap();
        std::fs::write(outbox.join("ghost.jpg.dest"), "E\nghost.jpg\n").unwrap();
        assert!(pending(&outbox).is_empty());
    }

    #[test]
    fn an_empty_outbox_is_not_an_error() {
        // Called on every sync pass, including before anything is exported
        // and on a device where the directory has never been created.
        assert!(pending(Path::new("/nonexistent/darkroom/outbox")).is_empty());
        assert_eq!(pending_count(Path::new("/nonexistent/darkroom/outbox")), 0);
    }

    #[test]
    fn the_remote_path_joins_root_folder_and_name() {
        let entry = Pending {
            local: PathBuf::from("/tmp/a.jpg"),
            remote_dir: "Exports/2026".into(),
            name: "a.jpg".into(),
        };
        assert_eq!(
            entry.remote_path("Photos").as_str(),
            "Photos/Exports/2026/a.jpg"
        );
        assert_eq!(
            entry.remote_folder("Photos").as_str(),
            "Photos/Exports/2026"
        );
    }

    #[test]
    fn an_empty_folder_exports_to_the_library_root() {
        // "Ask each time" is not set here — an empty remote folder means the
        // root, and it must not produce a double slash the server rejects.
        let entry = Pending {
            local: PathBuf::from("/tmp/a.jpg"),
            remote_dir: String::new(),
            name: "a.jpg".into(),
        };
        assert_eq!(entry.remote_path("Photos").as_str(), "Photos/a.jpg");
    }

    #[test]
    fn stray_slashes_do_not_produce_an_unusable_path() {
        // The folder comes from a picker or a text field, and either can hand
        // over a leading or trailing slash.
        let entry = Pending {
            local: PathBuf::from("/tmp/a.jpg"),
            remote_dir: "/Exports/".into(),
            name: "a.jpg".into(),
        };
        assert_eq!(
            entry.remote_path("/Photos/").as_str(),
            "Photos/Exports/a.jpg"
        );
    }

    #[test]
    fn the_status_line_never_claims_an_upload_that_has_not_happened() {
        // A queued export is real and finished, but it is not on the server,
        // and saying so before it is would be a promise the app cannot keep.
        let queued = Placed::Queued {
            local: PathBuf::from("/tmp/a.jpg"),
            remote_dir: "Exports".into(),
        };
        let text = queued.describe();
        assert!(text.contains("Queued"), "{text}");
        assert!(!text.contains("Exported"), "{text}");

        assert!(Placed::Device(PathBuf::from("/tmp/a.jpg"))
            .describe()
            .contains("Exported"));
    }

    // -----------------------------------------------------------------------
    // The batch (FR-EXP-7)
    // -----------------------------------------------------------------------

    /// Settings that write PNGs into `dir`, so a test can look at what landed.
    fn to_folder(dir: &Path) -> ExportSettings {
        ExportSettings {
            format: dr_types::ExportFormat::Png,
            target: ExportTarget::Device,
            destination: dir.to_string_lossy().into_owned(),
            ..Default::default()
        }
    }

    fn request(settings: ExportSettings, sources: Vec<Source>) -> BatchRequest {
        BatchRequest {
            sources,
            creds: None,
            settings,
            outbox: std::env::temp_dir().join("dr-batch-test-outbox"),
            sidecar_cache: std::env::temp_dir().join("dr-batch-test-sidecars"),
            offline: true,
            gpu: None,
        }
    }

    /// A frame of flat pixels — enough for the encoder, cheap for a test.
    fn frame(width: u32, height: u32) -> dr_export::Frame {
        dr_export::Frame::new(width, height, vec![128; (width * height * 4) as usize])
            .expect("well-formed")
    }

    fn drive(request: BatchRequest, cancel: Cancel) -> Vec<BatchMessage> {
        let (tx, rx) = std::sync::mpsc::channel();
        run(request, &cancel, &tx);
        drop(tx);
        rx.into_iter().collect()
    }

    fn finished(messages: &[BatchMessage]) -> (usize, usize, bool) {
        messages
            .iter()
            .find_map(|m| match m {
                BatchMessage::Finished {
                    exported,
                    failed,
                    cancelled,
                } => Some((*exported, *failed, *cancelled)),
                _ => None,
            })
            .expect("a batch always says it has finished")
    }

    #[test]
    fn a_rendered_frame_reaches_the_destination_folder() {
        // The whole tail of the batch — name, size, sharpen, encode, write —
        // with no GPU and no network, which is what makes it testable at all.
        let dir = tmp();
        let out = dir.join("exports");
        let messages = drive(
            request(
                to_folder(&out),
                vec![Source::Rendered {
                    stem: "IMG_0001".into(),
                    frame: frame(16, 12),
                }],
            ),
            Cancel::default(),
        );

        assert_eq!(finished(&messages), (1, 0, false));
        assert!(out.join("IMG_0001.png").exists());
    }

    #[test]
    fn two_images_of_one_name_both_survive_the_batch() {
        // Two folders in a library holding an IMG_0001 each is ordinary, and a
        // batch that wrote one file for two photographs would destroy work
        // without saying anything. Overwrite is set deliberately: it is the
        // user's answer about the *folder*, not about this run's own output.
        let dir = tmp();
        let out = dir.join("exports");
        let mut settings = to_folder(&out);
        settings.collision = dr_types::CollisionPolicy::Overwrite;

        let messages = drive(
            request(
                settings,
                vec![
                    Source::Rendered {
                        stem: "IMG_0001".into(),
                        frame: frame(16, 12),
                    },
                    Source::Rendered {
                        stem: "IMG_0001".into(),
                        frame: frame(16, 12),
                    },
                ],
            ),
            Cancel::default(),
        );

        assert_eq!(finished(&messages), (2, 0, false));
        assert!(out.join("IMG_0001.png").exists());
        assert!(out.join("IMG_0001-1.png").exists());
    }

    #[test]
    fn a_name_that_was_already_there_still_obeys_the_collision_setting() {
        // The other half of the rule above: a file that existed *before* the
        // batch is exactly what the policy is about, and Overwrite must still
        // mean overwrite or the setting would do nothing.
        let dir = tmp();
        let out = dir.join("exports");
        std::fs::create_dir_all(&out).expect("temp dir");
        std::fs::write(out.join("IMG_0001.png"), b"older").expect("seed");

        let mut settings = to_folder(&out);
        settings.collision = dr_types::CollisionPolicy::Overwrite;

        let mut issued = HashSet::new();
        let name = resolve_batch_name(
            &settings,
            &NameContext {
                source_stem: "IMG_0001",
                sequence: 1,
                ..Default::default()
            },
            &mut issued,
        );
        assert_eq!(name.as_deref(), Some("IMG_0001.png"));
    }

    #[test]
    fn a_skipped_name_is_reported_rather_than_silently_dropped() {
        // Skip is a legitimate answer, but the image still produced no file —
        // and a batch claiming to have exported it would be lying.
        let dir = tmp();
        let out = dir.join("exports");
        std::fs::create_dir_all(&out).expect("temp dir");
        std::fs::write(out.join("IMG_0001.png"), b"older").expect("seed");

        let mut settings = to_folder(&out);
        settings.collision = dr_types::CollisionPolicy::Skip;

        let messages = drive(
            request(
                settings,
                vec![Source::Rendered {
                    stem: "IMG_0001".into(),
                    frame: frame(8, 8),
                }],
            ),
            Cancel::default(),
        );

        assert_eq!(finished(&messages), (0, 1, false));
        assert!(matches!(
            messages.iter().find_map(|m| match m {
                BatchMessage::Item { outcome, .. } => Some(outcome),
                _ => None,
            }),
            Some(Err(ItemError::NameTaken))
        ));
        assert_eq!(std::fs::read(out.join("IMG_0001.png")).unwrap(), b"older");
    }

    #[test]
    fn one_failure_does_not_abandon_the_rest_of_the_batch() {
        // FR-EXP-7's central promise. The first source cannot be fetched — no
        // library is open — and the two after it must still be attempted and
        // still be counted.
        let dir = tmp();
        let out = dir.join("exports");
        let messages = drive(
            request(
                to_folder(&out),
                vec![
                    Source::Library {
                        path: "Photos/broken.CR2".into(),
                        cache: None,
                    },
                    Source::Rendered {
                        stem: "good-a".into(),
                        frame: frame(8, 8),
                    },
                    Source::Rendered {
                        stem: "good-b".into(),
                        frame: frame(8, 8),
                    },
                ],
            ),
            Cancel::default(),
        );

        assert_eq!(finished(&messages), (2, 1, false));
        assert!(out.join("good-a.png").exists());
        assert!(out.join("good-b.png").exists());
    }

    #[test]
    fn every_image_gets_its_own_outcome() {
        // The report is per image, not one verdict for the run: two failures
        // for two different reasons have to arrive as two messages, or a
        // three-hundred-frame batch is unreportable (NFR-ARCH-4).
        let dir = tmp();
        let messages = drive(
            request(
                to_folder(&dir.join("exports")),
                vec![
                    Source::Library {
                        path: "Photos/a.CR2".into(),
                        cache: None,
                    },
                    Source::Rendered {
                        stem: "b".into(),
                        frame: frame(8, 8),
                    },
                ],
            ),
            Cancel::default(),
        );

        let outcomes: Vec<&Result<Placed, ItemError>> = messages
            .iter()
            .filter_map(|m| match m {
                BatchMessage::Item { outcome, .. } => Some(outcome),
                _ => None,
            })
            .collect();
        assert_eq!(outcomes.len(), 2);
        assert!(outcomes[0].is_err());
        assert!(outcomes[1].is_ok());
    }

    #[test]
    fn a_cancelled_batch_stops_and_writes_nothing_more() {
        // TRACES: NFR-ARCH-3
        // Cancelled before it began, which is the strongest form of the
        // property: not one file, and the run still reports itself finished
        // rather than leaving the interface waiting for a message.
        let dir = tmp();
        let out = dir.join("exports");
        let cancel = Cancel::default();
        cancel.cancel();

        let messages = drive(
            request(
                to_folder(&out),
                vec![Source::Rendered {
                    stem: "IMG_0001".into(),
                    frame: frame(8, 8),
                }],
            ),
            cancel,
        );

        assert_eq!(finished(&messages), (0, 0, true));
        assert!(!out.join("IMG_0001.png").exists());
    }

    #[test]
    fn a_cancelled_wait_gives_up_instead_of_blocking_for_ever() {
        // TRACES: NFR-ARCH-3
        // The bound on cancelling a batch that is waiting on a download. With
        // a plain `recv` this test would hang, which is precisely the bug.
        let (tx, rx) = std::sync::mpsc::channel::<u8>();
        let cancel = Cancel::default();
        cancel.cancel();

        assert!(matches!(wait_for(&rx, &cancel), Waited::Cancelled));
        drop(tx);
    }

    #[test]
    fn a_worker_that_dies_is_not_mistaken_for_a_cancellation() {
        // The two look identical from a channel and mean opposite things: one
        // ends the batch, the other fails one image and moves on.
        let (tx, rx) = std::sync::mpsc::channel::<u8>();
        drop(tx);
        assert!(matches!(wait_for(&rx, &Cancel::default()), Waited::Silent));
    }

    #[test]
    fn a_cancelled_run_is_not_reported_as_a_failure() {
        // The user stopped it. A red row saying so is the application arguing
        // with something it was told to do.
        let (text, failed) = summarise(5, 0, true, None);
        assert!(!failed, "{text}");
        assert!(text.contains('5'), "{text}");
    }

    #[test]
    fn a_run_with_a_failure_keeps_its_row_and_names_one() {
        // This is the row somebody comes to the activity list to find, so it
        // has to survive being trimmed — and it has to say which photograph.
        let (text, failed) = summarise(38, 2, false, Some("IMG_0007.CR2: could not be rendered"));
        assert!(failed);
        assert!(text.contains("IMG_0007.CR2"), "{text}");
        assert!(text.contains("38"), "{text}");
    }
}