EventSnap/backend/src/services/export.rs

use std::path::{Path, PathBuf};
use std::time::Duration;

use anyhow::{Context, Result};
use async_zip::tokio::write::ZipFileWriter;
use async_zip::{Compression, ZipEntryBuilder};
use chrono::{DateTime, Utc};
use futures::io::{AllowStdIo, copy as fcopy};
use include_dir::{Dir, include_dir};
use serde::Serialize;
use sqlx::PgPool;
use tokio::io::AsyncWriteExt;
use tokio::sync::broadcast;
use tokio_util::compat::TokioAsyncReadCompatExt;
use uuid::Uuid;

use crate::state::SseEvent;

// ── Embedded viewer assets (pre-built SvelteKit static output) ──────────────

static VIEWER_DIR: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/static/export-viewer");

// ── Shared visibility filter ─────────────────────────────────────────────────

/// The predicate that decides what lands in a keepsake, as ONE definition.
///
/// Two queries have to agree on it: [`query_uploads`], which selects the rows the archives are
/// built from, and [`estimate_export_bytes`], which sizes them for the disk preflight. They used
/// to state it separately, and the direction of drift matters — an estimate that misses rows the
/// archive writes UNDER-reserves, which is the exact ENOSPC the preflight exists to prevent.
///
/// A `SRC:`-marked copy in the integration tests cannot catch that: drift means production moved
/// and the copy didn't, so both sides of such a test sit still and it keeps passing. Sharing the
/// fragment removes the failure by construction instead, and leaves the test doing what it is
/// actually good at — pinning the behaviour.
///
/// CONTRACT: callers must alias `upload` as `u` and join `"user"` as `usr`, and bind the event id
/// as `$1`.
macro_rules! export_visibility_where {
    () => {
        "WHERE u.event_id = $1 AND u.deleted_at IS NULL
           AND usr.uploads_hidden = FALSE AND usr.is_banned = FALSE"
    };
}

// ── DB query rows ────────────────────────────────────────────────────────────

#[derive(sqlx::FromRow)]
struct ExportUploadRow {
    id: Uuid,
    original_path: String,
    mime_type: String,
    caption: Option<String>,
    uploader_name: String,
    like_count: i64,
    created_at: DateTime<Utc>,
}

#[derive(sqlx::FromRow)]
struct ExportCommentRow {
    upload_id: Uuid,
    uploader_name: String,
    body: String,
    created_at: DateTime<Utc>,
}

// ── Viewer JSON structs (serialised to data.json) ───────────────────────────

#[derive(Serialize)]
struct ViewerData {
    event: ViewerEvent,
    posts: Vec<ViewerPost>,
}

#[derive(Serialize)]
struct ViewerEvent {
    name: String,
    exported_at: String,
    // Mirrors the live COMMENTS_ENABLED flag so the offline keepsake hides all comment
    // UI (buttons, counts, sections) when the feature was off for the event.
    comments_enabled: bool,
}

#[derive(Serialize)]
struct ViewerPost {
    id: String,
    uploader: String,
    caption: String,
    tags: Vec<String>,
    timestamp: String,
    likes: i64,
    comments: Vec<ViewerComment>,
    media: ViewerMedia,
}

#[derive(Serialize)]
struct ViewerComment {
    author: String,
    text: String,
    timestamp: String,
}

#[derive(Serialize)]
struct ViewerMedia {
    #[serde(rename = "type")]
    media_type: String,
    thumb: String,
    full: String,
}

// ── Entry point ──────────────────────────────────────────────────────────────

/// Arm both export jobs at `epoch`, unless a type is ALREADY complete at that same epoch.
///
/// Takes an executor rather than a pool so `release_gallery` can run it inside the same
/// transaction that bumps the epoch — the release must be all-or-nothing.
///
/// The `WHERE` on the upsert is the old `if ready { continue }` skip, expressed correctly. Its
/// purpose was always "startup recovery must not clobber a good half", and that intent is fine —
/// the bug was that it keyed off a SEPARATELY STORED ready flag that a stale worker could set.
/// Here the condition IS the readiness predicate itself (`done` at the current epoch), so it cannot
/// disagree with reality. On a release it is always true (the epoch just moved, so no row can be
/// done at the new epoch) and both types regenerate; on recovery it preserves a genuinely finished
/// half and re-arms only what is missing.
pub async fn enqueue_jobs_at_epoch(
    conn: &mut sqlx::PgConnection,
    event_id: Uuid,
    epoch: i64,
) -> Result<()> {
    enqueue_types_at_epoch(conn, event_id, epoch, &["zip", "html"]).await
}

/// Arm a specific subset of the export types at `epoch` (see [`enqueue_jobs_at_epoch`]).
pub async fn enqueue_types_at_epoch(
    conn: &mut sqlx::PgConnection,
    event_id: Uuid,
    epoch: i64,
    types: &[&str],
) -> Result<()> {
    for export_type in types {
        sqlx::query(
            "INSERT INTO export_job (event_id, type, status, progress_pct, epoch)
             VALUES ($1, $2::export_type, 'pending', 0, $3)
             ON CONFLICT (event_id, type) DO UPDATE
                 SET status = 'pending', progress_pct = 0, file_path = NULL,
                     error_message = NULL, completed_at = NULL,
                     epoch = EXCLUDED.epoch
             WHERE export_job.status <> 'done' OR export_job.epoch <> EXCLUDED.epoch",
        )
        .bind(event_id)
        .bind(export_type)
        .bind(epoch)
        .execute(&mut *conn)
        .await?;
    }
    Ok(())
}

/// The export artifacts a content change can invalidate.
///
/// The ZIP holds only the media originals; the HTML viewer additionally embeds captions, likes and
/// comments. So a comment moderation needs only the viewer rebuilt — rebuilding the multi-GB ZIP for
/// it would 404 the photo download for minutes to change nothing in it.
#[derive(Clone, Copy, PartialEq, Eq)]
pub enum Affects {
    /// Media changed (an upload removed/hidden) — both artifacts are stale.
    Both,
    /// Only viewer content changed (a comment) — carry the ZIP forward untouched.
    ViewerOnly,
}

/// Handle for a regeneration that a caller has ARMED inside its transaction but not yet started.
/// The workers must only be spawned AFTER that transaction commits, or they could snapshot the
/// database before the change that triggered them is visible.
pub struct PendingRegen {
    pub event_id: Uuid,
    pub event_name: String,
    pub epoch: i64,
}

/// Invalidate the current keepsake and arm a rebuild — IN THE CALLER'S TRANSACTION.
///
/// Any content removal that happens after a release (a takedown, a ban, a guest deleting their own
/// photo) must reach the keepsake: the archive is the artifact people keep forever, and "please take
/// my photo out" is the one request that most needs to be honoured there. Bumping the epoch retires
/// the current archive instantly (readiness is derived from the epoch), so the stale download stops
/// being served the moment the change commits, and a fresh worker rebuilds without the content.
///
/// Runs in the caller's tx ON PURPOSE. If the removal committed but the regeneration didn't, the
/// taken-down photo would stay downloadable forever and nothing would notice: the keepsake still
/// looks complete at the current epoch, so recovery skips it, and the host can no longer even find
/// the upload to retry. Returns `None` when the event isn't released (nothing to invalidate — the
/// export is built fresh at release time).
pub async fn invalidate_and_arm(
    conn: &mut sqlx::PgConnection,
    event_slug: &str,
    affects: Affects,
) -> Result<Option<PendingRegen>> {
    let bumped: Option<(Uuid, String, i64)> = sqlx::query_as(
        "UPDATE event SET export_epoch = export_epoch + 1
         WHERE slug = $1 AND export_released_at IS NOT NULL
         RETURNING id, name, export_epoch",
    )
    .bind(event_slug)
    .fetch_optional(&mut *conn)
    .await?;

    let Some((event_id, event_name, epoch)) = bumped else {
        return Ok(None);
    };

    // A comment-only change doesn't alter the ZIP's contents (the ZIP holds media, not comments), so
    // we'd rather carry the finished archive into the new epoch than spend minutes rebuilding it.
    //
    // But "finished" is the whole precondition, and it is NOT guaranteed: between `release_gallery`
    // and the ZIP worker completing, the row sits at `pending`/`running` — MINUTES, for a real
    // multi-GB gallery — and deleting a comment right after release is an utterly ordinary thing to
    // do. If we blindly re-armed only the viewer, the carry-forward would match nothing, the ZIP row
    // would be left stranded at the retired epoch, and NOTHING would ever re-arm it: the in-flight
    // worker finishes and writes `done` at an epoch `export_current` no longer matches, so
    // `GET /export/zip` 404s forever (short of a boot or the host finding the rebuild button).
    //
    // So the carry-forward's OWN result decides. It matched ⇒ there is a current, finished ZIP and
    // only the viewer needs rebuilding. It didn't ⇒ there is no ZIP to preserve, and the ZIP must be
    // rebuilt at the new epoch like any other invalidation. Never assume; ask the UPDATE.
    let carried = if affects == Affects::ViewerOnly {
        sqlx::query(
            "UPDATE export_job SET epoch = $2
             WHERE event_id = $1 AND type = 'zip'::export_type
               AND status = 'done' AND epoch = $2 - 1",
        )
        .bind(event_id)
        .bind(epoch)
        .execute(&mut *conn)
        .await?
        .rows_affected()
            == 1
    } else {
        false
    };

    // (`prune_stale_export_files` protects any file a current-epoch row still points at, so a
    // carried archive isn't swept for having an older epoch in its name.)
    let types: &[&str] = if carried { &["html"] } else { &["zip", "html"] };
    enqueue_types_at_epoch(&mut *conn, event_id, epoch, types).await?;

    Ok(Some(PendingRegen {
        event_id,
        event_name,
        epoch,
    }))
}

/// Startup export recovery: re-arm exports for any released event whose keepsake isn't fully
/// present at the current epoch (a crash mid-export, or a `done` row whose file has since gone
/// missing). Without this, `release_gallery` would reject a retry with "bereits freigegeben" and
/// downloads would 404 forever. Runs once at boot, after `AppState` exists.
///
/// Unlike the old version, this VERIFIES THE FILE IS ACTUALLY ON DISK. Recovery used to be purely
/// flag-driven, so a `done` row whose archive had been lost (a wiped/restored volume, or a
/// truncated write) was a permanent dead end: the download 404'd, recovery skipped the event
/// because it looked complete, and the only escape was manual DB surgery.
pub async fn recover_exports(
    pool: PgPool,
    media_path: PathBuf,
    export_path: PathBuf,
    comments_enabled: bool,
    sse_tx: broadcast::Sender<SseEvent>,
) {
    let rows = match sqlx::query_as::<_, (Uuid, String, i64)>(
        "SELECT id, name, export_epoch FROM event WHERE export_released_at IS NOT NULL",
    )
    .fetch_all(&pool)
    .await
    {
        Ok(r) => r,
        Err(e) => {
            tracing::error!("export recovery: failed to query released events: {e:#}");
            return;
        }
    };

    // Crash-orphaned temps are immortal otherwise: `prune_stale_export_files` deliberately never
    // touches `.tmp` (a superseded worker may still be streaming into one), and nothing else sweeps
    // them. A hard kill mid-export — especially one whose epoch has since moved on — strands a
    // full-gallery-sized file forever. Boot is the one moment this is unambiguously safe.
    sweep_orphan_temps(&export_path).await;

    for (event_id, event_name, epoch) in rows {
        // Retire any `done` row at the current epoch whose file is missing or empty, so the
        // upsert below re-arms it instead of leaving an undownloadable "ready" keepsake.
        if let Err(e) = invalidate_missing_files(&pool, &export_path, event_id, epoch).await {
            tracing::error!("export recovery: file verification failed for {event_id}: {e:#}");
        }

        let complete: bool = sqlx::query_scalar(
            "SELECT COUNT(*) = 2 FROM export_job
              WHERE event_id = $1 AND epoch = $2 AND status = 'done'",
        )
        .bind(event_id)
        .bind(epoch)
        .fetch_one(&pool)
        .await
        .unwrap_or(false);
        if complete {
            continue;
        }

        tracing::warn!(
            "export recovery: re-arming export jobs for event {event_id} @ epoch {epoch}"
        );
        let mut conn = match pool.acquire().await {
            Ok(c) => c,
            Err(e) => {
                tracing::error!("export recovery: cannot acquire connection for {event_id}: {e:#}");
                continue;
            }
        };
        if let Err(e) = enqueue_jobs_at_epoch(&mut conn, event_id, epoch).await {
            tracing::error!("export recovery: failed to re-arm for event {event_id}: {e:#}");
            continue;
        }
        spawn_export_jobs(
            event_id,
            event_name,
            epoch,
            comments_enabled,
            Duration::ZERO,
            pool.clone(),
            media_path.clone(),
            export_path.clone(),
            sse_tx.clone(),
        );
    }
}

/// Remove every export temp artifact at boot. Called from `recover_exports`, after the startup sweep
/// has marked all `running` jobs `failed` and before any worker is spawned — so no live worker can
/// own one of these. A worker that is re-armed will recreate its temp from scratch.
async fn sweep_orphan_temps(export_path: &Path) {
    let mut rd = match tokio::fs::read_dir(export_path).await {
        Ok(rd) => rd,
        Err(_) => return,
    };
    let mut removed = 0u32;
    while let Ok(Some(entry)) = rd.next_entry().await {
        let name = entry.file_name();
        let name = name.to_string_lossy();
        let is_temp = name.ends_with(".tmp") || name.starts_with("viewer_tmp_");
        if !is_temp {
            continue;
        }
        let is_dir = entry.file_type().await.map(|t| t.is_dir()).unwrap_or(false);
        let r = if is_dir {
            tokio::fs::remove_dir_all(entry.path()).await
        } else {
            tokio::fs::remove_file(entry.path()).await
        };
        if r.is_ok() {
            removed += 1;
        }
    }
    if removed > 0 {
        tracing::warn!("export recovery: swept {removed} orphaned export temp artifact(s)");
    }
}

/// Reset any `done` job at the current epoch whose archive is absent or zero-length. Guards
/// against DB/disk divergence (lost volume, truncated write) that recovery would otherwise
/// mistake for a finished keepsake.
async fn invalidate_missing_files(
    pool: &PgPool,
    export_path: &Path,
    event_id: Uuid,
    epoch: i64,
) -> Result<()> {
    let done: Vec<(String, Option<String>)> = sqlx::query_as(
        "SELECT type::text, file_path FROM export_job
          WHERE event_id = $1 AND epoch = $2 AND status = 'done'",
    )
    .bind(event_id)
    .bind(epoch)
    .fetch_all(pool)
    .await?;

    for (export_type, file_path) in done {
        let intact = match file_path.as_deref().and_then(|p| Path::new(p).file_name()) {
            Some(name) => tokio::fs::metadata(export_path.join(name))
                .await
                .map(|m| m.is_file() && m.len() > 0)
                .unwrap_or(false),
            None => false,
        };
        if !intact {
            tracing::warn!(
                "export recovery: {export_type} export for event {event_id} is marked done but its \
                 file is missing/empty — re-arming it"
            );
            sqlx::query(
                "UPDATE export_job SET status = 'failed', file_path = NULL,
                        error_message = 'Exportdatei fehlt — wird neu erzeugt'
                  WHERE event_id = $1 AND type = $2::export_type AND epoch = $3",
            )
            .bind(event_id)
            .bind(&export_type)
            .bind(epoch)
            .execute(pool)
            .await?;
        }
    }
    Ok(())
}

/// How long a regeneration waits before claiming. A takedown pass ("remove these five photos") is a
/// burst of independent requests, each of which retires the previous generation. Without a delay,
/// each one immediately spawns two full exports — and a superseded worker is INERT, not STOPPED, so
/// it still runs every ffmpeg spawn and image resize and writes an entire archive before discovering
/// it lost. Five deletes would leave ten workers alive, each holding a full-gallery-sized temp file,
/// inside a 1 GB container.
///
/// Sleeping before `claim_job` collapses the burst for free: a worker whose epoch is retired during
/// the delay fails its claim and does ZERO work. Release and boot-recovery pass ZERO — those must
/// start immediately.
pub const REGEN_DEBOUNCE: Duration = Duration::from_secs(20);

/// Tracks when the CURRENT un-served burst of invalidations began, per event.
///
/// The starvation this fixes (B7): every invalidation bumps `export_epoch` immediately and arms a
/// worker that sleeps `REGEN_DEBOUNCE` before claiming — and any further bump inside that window
/// retires it. Nothing rate-limits epoch bumps; only the HTTP requests are limited, at 120/min for
/// comment deletion and 30/min for caption edits. So one ordinary guest deleting a comment per
/// second, or flipping a caption A/B/A, keeps `GET /export/html` at 404 for the rest of the event
/// while the UI shows "Wird vorbereitet…" forever. **A host moderating faster than one action per
/// 20 seconds produces the same result by accident** — which is what a takedown pass looks like.
///
/// The fix is to measure the debounce from the FIRST request in a burst rather than the latest, so
/// the wait is bounded no matter how long the burst runs: coalescing still collapses a rapid pass
/// into one build, but a build always starts within `REGEN_DEBOUNCE` of the burst beginning.
///
/// In-memory on purpose. It is a scheduling hint, not state: losing it on restart is harmless
/// because `recover_exports` re-arms anything unfinished at boot anyway, and the worst case of a
/// stale entry is one build starting immediately instead of debounced.
type BurstStarts = std::collections::HashMap<Uuid, std::time::Instant>;
static REGEN_BURST_START: std::sync::LazyLock<std::sync::Mutex<BurstStarts>> =
    std::sync::LazyLock::new(|| std::sync::Mutex::new(BurstStarts::new()));

/// How long to defer the next regen worker for `event_id`, and record that a burst is running.
///
/// Returns `REGEN_DEBOUNCE` for the first invalidation of a burst and progressively less for
/// each one after it, reaching zero once the burst has been going for a full debounce window.
pub fn regen_delay_for(event_id: Uuid) -> Duration {
    let mut map = match REGEN_BURST_START.lock() {
        Ok(m) => m,
        // A poisoned mutex must not take the keepsake down: fall back to the plain debounce,
        // which is the pre-existing behaviour.
        Err(e) => e.into_inner(),
    };
    let started = *map.entry(event_id).or_insert_with(std::time::Instant::now);
    REGEN_DEBOUNCE.saturating_sub(started.elapsed())
}

/// A regen worker has begun (or the event settled), so the next invalidation starts a fresh burst.
pub fn clear_regen_burst(event_id: Uuid) {
    if let Ok(mut map) = REGEN_BURST_START.lock() {
        map.remove(&event_id);
    }
}

// Export worker entry point: every argument is state the spawned worker is BORN with (notably
// `epoch`). Bundling them into a struct would be a pure-refactor risk on the epoch logic for no
// gain, so the arity stands.
#[allow(clippy::too_many_arguments)]
pub fn spawn_export_jobs(
    event_id: Uuid,
    event_name: String,
    epoch: i64,
    comments_enabled: bool,
    delay: Duration,
    pool: PgPool,
    media_path: PathBuf,
    export_path: PathBuf,
    sse_tx: broadcast::Sender<SseEvent>,
) {
    let pool2 = pool.clone();
    let media_path2 = media_path.clone();
    let export_path2 = export_path.clone();
    let sse_tx2 = sse_tx.clone();
    let event_name2 = event_name.clone();

    tokio::spawn(async move {
        // The worker is BORN with its epoch — it never learns it from the DB. A worker whose epoch
        // has been retired is inert by construction: every write it makes is `epoch`-guarded on the
        // row it is updating, so it simply matches nothing. No cross-table guard, no race.
        if !delay.is_zero() {
            tokio::time::sleep(delay).await;
        }
        // The burst has now been served: whatever arrives next starts a fresh debounce window.
        clear_regen_burst(event_id);
        // Run the body in its OWN task so a panic surfaces as a `JoinError` here rather than
        // unwinding past `mark_failed` — see `supervise_export`.
        let (p, m, x, s) = (
            pool.clone(),
            media_path.clone(),
            export_path.clone(),
            sse_tx.clone(),
        );
        let inner =
            tokio::spawn(async move { run_zip_export(event_id, epoch, &p, &m, &x, &s).await });
        supervise_export(inner, &pool, event_id, "zip", epoch).await;
        maybe_broadcast_complete(&pool, event_id, &sse_tx).await;
    });

    tokio::spawn(async move {
        if !delay.is_zero() {
            tokio::time::sleep(delay).await;
        }
        let (p, m, x, s) = (
            pool2.clone(),
            media_path2.clone(),
            export_path2.clone(),
            sse_tx2.clone(),
        );
        let inner = tokio::spawn(async move {
            run_html_export(
                event_id,
                epoch,
                &event_name2,
                comments_enabled,
                &p,
                &m,
                &x,
                &s,
            )
            .await
        });
        supervise_export(inner, &pool2, event_id, "html", epoch).await;
        maybe_broadcast_complete(&pool2, event_id, &sse_tx2).await;
    });
}

/// Await an export worker and make sure the job row NEVER stays at `running`.
///
/// Both workers were bare `tokio::spawn`s whose `Err` path was handled correctly — but a PANIC
/// unwound straight past `mark_failed` and `maybe_broadcast_complete`, leaving the row at whatever
/// `progress_pct` it had reached. The UI renders that as "Wird erstellt (77%)" with the download
/// disabled, permanently: nothing sweeps `running` rows, and `recover_exports` only runs at boot.
/// So the single most likely cause of a stuck keepsake was also the one the error handling missed.
///
/// Running the body as a child task turns that panic into a `JoinError` we can act on. The
/// maintenance loop already supervises itself this way; this extends the same pattern to the two
/// workers that actually produce the thing guests came for.
async fn supervise_export(
    handle: tokio::task::JoinHandle<Result<()>>,
    pool: &PgPool,
    event_id: Uuid,
    export_type: &str,
    epoch: i64,
) {
    match handle.await {
        Ok(Ok(())) => {}
        Ok(Err(e)) => {
            tracing::error!(
                "{export_type} export failed for event {event_id} @ epoch {epoch}: {e:#}"
            );
            mark_failed(pool, event_id, export_type, epoch, &e.to_string()).await;
        }
        Err(join_err) => {
            // Panicked, or cancelled at shutdown. Either way the row must not be left claiming to
            // be in progress — a failed job the host can retry beats one that lies forever.
            tracing::error!(
                panicked = join_err.is_panic(),
                "{export_type} export task for event {event_id} @ epoch {epoch} died without \
                 reporting: {join_err}"
            );
            mark_failed(
                pool,
                event_id,
                export_type,
                epoch,
                "Interner Fehler beim Erstellen des Keepsakes. Bitte über \"Neu erzeugen\" \
                 erneut versuchen.",
            )
            .await;
        }
    }
}

/// Preflight that will sacrifice the previous generation rather than deadlock, and the order is
/// the whole point.
///
/// Deferring the prune (so a failed rebuild can never leave the event with no archive at all) has
/// a cost the first version of this did not follow through on: at rebuild time the previous
/// generation is still on disk and still counted against free space, so the preflight demands room
/// for BOTH. That halves the gallery size a rebuild can survive relative to the size the upload
/// gate allows — and every path that bumps the epoch (a guest deleting their own photo, a caption
/// edit, a ban, "Neu erzeugen") retires the current keepsake IMMEDIATELY on commit. So above that
/// threshold the download 404s, the rebuild is refused, and the only thing that could free the
/// space is the prune that now only runs on success. Permanently stuck, unreachable from any
/// handler.
///
/// So: try to build while preserving the old generation. If that genuinely does not fit, the old
/// generation is the one thing we can reclaim — sacrifice it and try once more. A keepsake that
/// exists beats one we preserved but can never replace.
///
/// But ONLY when the sacrifice is sufficient: we measure what pruning would free and compare it to
/// the shortfall first. Pruning and hoping meant a failed second check left nothing on disk at all
/// — no old archive and no new one — which is worse than either outcome this function chooses
/// between. When the old archive cannot buy us a rebuild, it stays.
///
/// SHARED by both halves deliberately. This started as two copies and one of them (HTML) silently
/// kept the single-phase form, so the ZIP archive rebuilt and the viewer stayed permanently stuck
/// — the exact deadlock above, on half the product. `prefix` is the only thing that differs, and
/// it must be the caller's OWN prefix: pruning the other half's archives from here would reclaim
/// space a sibling worker is about to need, on its behalf, without its knowledge.
async fn ensure_export_space_reclaiming(
    pool: &PgPool,
    event_id: Uuid,
    export_path: &Path,
    prefix: &str,
    epoch: i64,
) -> Result<()> {
    if ensure_export_space(pool, event_id, export_path)
        .await
        .is_ok()
    {
        return Ok(());
    }

    // LOOK BEFORE YOU DESTROY.
    //
    // This used to prune unconditionally and then re-check. When the re-check ALSO failed, the old
    // keepsake was already gone and the rebuild returned `Err` — leaving **no archive on disk at
    // all**, which is strictly worse than the deadlock the reclaim exists to avoid. It needs no
    // host action to reach: any guest deleting their own photo arms this path, and the doc comment
    // above asserts the opposite invariant.
    //
    // So sacrifice the old generation only when doing so is actually sufficient. When it is not,
    // that old keepsake is the only one anybody will ever have — keep it, and report the failure
    // the same way the non-reclaiming check does. `rebuild_export` is the host's retry once they
    // have freed space.
    let deficit = match export_space_deficit(pool, event_id, export_path).await? {
        // Raced back into having room (a concurrent prune, a guest deleting an upload). Nothing
        // to reclaim and nothing to fail.
        None => return Ok(()),
        Some(d) => d,
    };
    let reclaimable =
        reclaimable_superseded_bytes(pool, export_path, prefix, event_id, epoch).await;

    // PRUNE EVEN WHEN IT IS NOT ENOUGH ON ITS OWN. This used to refuse unless
    // `reclaimable >= deficit`, on the premise that "deleting it would leave no archive at all" —
    // and that premise does not survive contact with `prune_superseded_archives`, which can only
    // ever touch generations `n < keep_seq` that `protected_files` does not name. Those are exactly
    // the archives no handler can serve: a download resolves through `export_current`, which
    // requires `j.epoch = e.export_epoch`, and the epoch only ever increments. The bytes this
    // branch was protecting were already unreachable from every route, and the next successful
    // build deletes them anyway.
    //
    // What the refusal did cost was the only in-app way out of the deadlock this function exists to
    // break. `reclaimable` is scoped to the caller's OWN prefix — one old archive, ~1.1x the
    // gallery — while `deficit` is sized for both halves plus the 10 GB reserve. So on a tight
    // disk each worker independently measures its own share as insufficient and neither prunes,
    // while the two shares are JOINTLY sufficient. Every "Neu erzeugen" reruns the identical
    // arithmetic and refuses identically: permanently stuck, with dead archives on the volume that
    // nothing will reclaim and nothing can serve.
    //
    // So reclaim what we can and let the re-check below decide. If it still does not fit, we fail
    // exactly as the plain preflight would — but the sibling worker's prune has now freed its share
    // too, and the host's retry converges instead of looping.
    if reclaimable < deficit {
        tracing::warn!(
            deficit,
            reclaimable,
            "pruning the previous {prefix} keepsake will not free the full shortfall on its own; \
             reclaiming anyway — it is already unservable, and the sibling half frees the rest"
        );
    }

    tracing::warn!(
        deficit,
        reclaimable,
        "not enough room to rebuild {prefix} alongside the previous keepsake; reclaiming it first"
    );
    prune_superseded_archives(pool, export_path, prefix, event_id, epoch).await;
    ensure_export_space(pool, event_id, export_path).await
}

/// Largest share of an event's media an archive may silently omit and still publish.
///
/// Not zero, deliberately. A skip is *expected* in ordinary operation — a guest deleting their own
/// photo mid-build, or the hourly reclaim collecting an original past its retention window — and
/// the writers go out of their way to degrade one entry rather than fail the whole keepsake,
/// because a released event cannot be rebuilt by the host without reopening uploads. Making any
/// skip fatal would turn a tolerated one-photo gap into total loss of the keepsake, which is the
/// regression the TOCTOU comments in each writer warn about.
///
/// What must never happen is publishing an archive that is missing *most* of the event. 10% of a
/// ~100-photo wedding is ~10 photos: far above the one-or-two a live delete explains, far below
/// the "media_path is wrong so nothing opened" catastrophe.
const MAX_SKIPPED_FRACTION: f64 = 0.10;

/// Skips tolerated regardless of how small the event is.
///
/// A pure fraction inverts this check on a small gallery: at nine uploads one skip is 11%, so the
/// single most ordinary event there is — a guest deleting their own photo while the archive builds
/// — failed the whole keepsake. That is exactly the "one-photo gap becomes total loss" outcome
/// [`MAX_SKIPPED_FRACTION`]'s own comment says it exists to avoid, and it is worst early on, when
/// a host testing a release has only a handful of photos.
///
/// Two is the number of concurrent live deletes worth absorbing; beyond that on a tiny gallery the
/// `written == 0` guard still catches the misconfiguration case, which is the one that matters.
const MIN_TOLERATED_SKIPS: usize = 2;

/// Decide whether an archive that skipped some media may still be published.
///
/// Rejects two cases:
///   * `written == 0` with media expected — the misconfiguration case. This is what produced a
///     few-hundred-byte ZIP containing zero photos that passed every automated check, was
///     advertised green, and was handed to every guest.
///   * more than [`MAX_SKIPPED_FRACTION`] omitted — enough missing that the archive misrepresents
///     the event even though it is structurally valid.
///
/// A tolerated partial still logs at `error` level with the exact counts, so the gap is visible in
/// the record rather than buried in per-entry warnings nobody aggregates.
fn check_export_completeness(
    kind: &str,
    event_id: Uuid,
    expected: usize,
    written: usize,
    skipped: usize,
) -> Result<()> {
    if skipped == 0 {
        return Ok(());
    }
    if written == 0 && expected > 0 {
        anyhow::bail!(
            "{kind}-Export enthält keine einzige Datei ({expected} erwartet, alle \
             übersprungen). Das deutet auf einen falschen MEDIA_PATH oder fehlende \
             Zugriffsrechte hin — das Keepsake wurde NICHT veröffentlicht."
        );
    }
    // A PROPORTIONAL skip is loud, but it is NOT fatal, and the difference is the whole point.
    //
    // This used to `bail!` once skips passed `max(2, 10% of expected)`. That inverts
    // MAX_SKIPPED_FRACTION's own justification two doc comments up — "making any skip fatal would
    // turn a tolerated one-photo gap into total loss of the keepsake" — because the refusal is
    // DETERMINISTIC ACROSS RETRIES. The files that could not be read are still unreadable when the
    // host taps "Neu erzeugen", and by then the gallery is released and the uploads cannot be
    // collected again. So on a 30-photo event, 4 unreadable originals stopped publishing the other
    // 26 — not once, but forever. Refusing to hand over an incomplete keepsake is only defensible
    // if something better can still arrive; here nothing can.
    //
    // `written == 0` above stays fatal, and it is the case that actually mattered: a wrong
    // MEDIA_PATH produced a few-hundred-byte archive with zero photos that passed every automated
    // check and was advertised as ready. That one is a misconfiguration the host CAN fix and retry.
    //
    // Everything short of that publishes, with the counts at `error` level so the gap is in the
    // record rather than buried in per-entry warnings. A partial keepsake is what the guests get to
    // keep; an aborted one is nothing at all.
    let budget = MIN_TOLERATED_SKIPS.max((expected as f64 * MAX_SKIPPED_FRACTION).floor() as usize);
    let level_note = if expected > 0 && skipped > budget {
        "MATERIALLY INCOMPLETE — published anyway because a refused rebuild is not recoverable"
    } else {
        "published with missing media"
    };
    tracing::error!(
        %event_id, kind, expected, written, skipped, budget,
        "{level_note} — {skipped} of {expected} entries could not be read"
    );
    Ok(())
}

/// Take the process-wide heavy-image permit if this file is big enough to need it.
///
/// Mirrors the compression worker's gate exactly (a header probe, no pixels decoded), so the two
/// producers of heavy image work agree on what "heavy" means and serialise against each other
/// rather than each against itself.
async fn heavy_permit_for(path: &Path) -> Option<tokio::sync::SemaphorePermit<'static>> {
    let estimate = crate::services::imaging::estimated_processing_peak_bytes(path, 2048);
    match estimate {
        Some(bytes) if bytes > crate::services::imaging::HEAVY_IMAGE_BYTES => {
            crate::services::imaging::HEAVY_IMAGE_PERMITS
                .acquire()
                .await
                .ok()
        }
        _ => None,
    }
}

// ── ZIP export ───────────────────────────────────────────────────────────────

async fn run_zip_export(
    event_id: Uuid,
    epoch: i64,
    pool: &PgPool,
    media_path: &Path,
    export_path: &Path,
    sse_tx: &broadcast::Sender<SseEvent>,
) -> Result<()> {
    if !claim_job(pool, event_id, "zip", epoch).await? {
        // Either another worker owns this generation, or our epoch has been retired by a
        // reopen/re-release. Both mean: not ours. Bail without touching anything.
        return Ok(());
    }

    // AFTER the claim, not before. A preflight that bailed before claiming would leave the row
    // `pending` with no worker and no error — the spinner-forever state `mark_failed`'s status
    // guard was widened to prevent. Failing here goes through the caller's `mark_failed`, so the
    // host gets the reason.
    ensure_export_space_reclaiming(pool, event_id, export_path, "Gallery", epoch).await?;

    // On error, mark THIS generation failed — a no-op if we've since been superseded (the
    // caller in `spawn_export_jobs` does it, epoch-guarded). Temp artifacts are cleaned up
    // here so a failing export can't leak them (which is what fills the disk in the first place).
    let res = run_zip_export_inner(epoch, event_id, pool, media_path, export_path, sse_tx).await;
    if res.is_err() {
        let _ =
            tokio::fs::remove_file(export_path.join(gen_name(event_id, "Gallery", epoch, ".tmp")))
                .await;
    }

    // Reclaim the PREVIOUS generation only once this one has actually landed.
    //
    // This used to run before the preflight, reasoning that the superseded archive is already
    // unreachable and its space is usually exactly what the rebuild needs. That is true about
    // REACHABILITY and false about RECOVERABILITY: an epoch is a database value that can be
    // rolled back, deleted bytes cannot. Any rebuild that then failed — ENOSPC mid-write, an
    // OOM, a hung ffmpeg, a host tapping "Neu erzeugen" on a bad day — left the event with NO
    // archive at all, which is the one outcome the whole product exists to prevent, at the one
    // moment nobody is watching.
    //
    // The cost of deferring is that a rebuild now needs room for both generations at once, and
    // `ensure_export_space` above no longer gets to count the old archive's bytes as available.
    // That is the correct trade: it converts "silently destroyed the only copy" into "refused
    // to start, and said why".
    if res.is_ok() {
        prune_superseded_archives(pool, export_path, "Gallery", event_id, epoch).await;
    }

    abandon_if_superseded("ZIP", event_id, epoch, res)
}

/// A worker that discovers mid-export that its generation was retired has not FAILED — it simply
/// lost. Swallow the sentinel so we don't log an error or (pointlessly) try to mark a row we no
/// longer own as failed. Its temp artifacts were already removed by the caller.
fn abandon_if_superseded(kind: &str, event_id: Uuid, epoch: i64, res: Result<()>) -> Result<()> {
    match res {
        Err(e) if e.downcast_ref::<Superseded>().is_some() => {
            tracing::info!(
                "{kind} export for event {event_id} superseded mid-export (epoch {epoch} retired); abandoned"
            );
            Ok(())
        }
        other => other,
    }
}

/// Per-generation, per-EVENT artifact name. The event id matters: all events share one exports
/// volume, and a name keyed only by generation would let two events collide on the same path (and
/// let one event's prune delete another's live keepsake). The viewer temp dir was already
/// event-scoped; the archives were not.
fn gen_name(event_id: Uuid, prefix: &str, epoch: i64, suffix: &str) -> String {
    format!("{prefix}.{event_id}.{epoch}{suffix}")
}

async fn run_zip_export_inner(
    epoch: i64,
    event_id: Uuid,
    pool: &PgPool,
    media_path: &Path,
    export_path: &Path,
    sse_tx: &broadcast::Sender<SseEvent>,
) -> Result<()> {
    let uploads = query_uploads(pool, event_id).await?;
    let total = uploads.len().max(1) as f32;

    // Written OUTSIDE media_path: the public /media ServeDir must never reach these.
    let exports_dir = export_path.to_path_buf();
    tokio::fs::create_dir_all(&exports_dir).await?;

    // Per-generation paths: a superseded worker (older epoch) and the fresh worker never share a
    // file on disk, so neither can truncate or interleave the other's bytes.
    let tmp_path = exports_dir.join(gen_name(event_id, "Gallery", epoch, ".tmp"));
    let out_name = gen_name(event_id, "Gallery", epoch, ".zip");
    let out_path = exports_dir.join(&out_name);

    // Skips are TOLERATED but no longer SILENT — see `check_export_completeness`. All three
    // writers continued past an unreadable source with only a `warn!`, nothing counted them,
    // nothing reached `error_message`, and the job finalized at `progress_pct = 100,
    // status = 'done'`. So a wrong `media_path` after a compose edit produced a few-hundred-byte
    // ZIP with zero media that passed the liveness check, was advertised green, and was handed to
    // every guest as their keepsake.
    let mut written = 0usize;
    let mut skipped = 0usize;

    {
        let file = tokio::fs::File::create(&tmp_path).await?;
        let mut zip = ZipFileWriter::with_tokio(file);

        for (i, row) in uploads.iter().enumerate() {
            let src = media_path.join(&row.original_path);
            let ext = ext_from_path(&row.original_path);
            let date = row.created_at.format("%Y-%m-%d_%H-%M").to_string();
            let name_safe = sanitize_name(&row.uploader_name);
            let folder = if row.mime_type.starts_with("video/") {
                "Videos"
            } else {
                "Photos"
            };
            let entry_name = format!("{folder}/{date}_{name_safe}_{}.{ext}", row.id);

            let builder = keepsake_entry(entry_name, Compression::Stored);

            // Open BEFORE writing the entry header. A missing source is skipped (the media file
            // was deleted, or its processing failed) — but it must be skipped without aborting the
            // whole export. The old code did `if !src.exists() { continue }` and then `open(..)?`,
            // a TOCTOU: a file vanishing in between turned a tolerated gap into a hard error that
            // failed the entire keepsake, permanently (the event is already released, so the host
            // cannot retry). Opening first collapses the check and the use into one operation.
            let src_file = match tokio::fs::File::open(&src).await {
                Ok(f) => f,
                Err(e) => {
                    tracing::warn!(
                        "ZIP export: skipping upload {} — cannot read {}: {e}",
                        row.id,
                        src.display()
                    );
                    skipped += 1;
                    continue;
                }
            };
            written += 1;

            let mut entry = zip.write_entry_stream(builder).await?;
            let mut f = src_file.compat();
            fcopy(&mut f, &mut entry).await?;
            entry.close().await?;

            let pct = ((i + 1) as f32 / total * 100.0) as i16;
            // Also our liveness check: if we've been retired, stop NOW rather than grinding through
            // the rest of the gallery to build an archive we would immediately delete.
            if !update_progress(pool, event_id, "zip", epoch, pct.min(99)).await {
                return Err(Superseded.into());
            }
        }

        // FLUSH AND FSYNC BEFORE THE RENAME.
        //
        // `async_zip`'s `close()` writes the central directory and then hands back the inner
        // writer — it never flushes it. And `tokio::fs::File` is write-behind: dropping it awaits
        // nothing and SILENTLY DISCARDS any error from the last write. So on a full disk the final
        // chunk (the one carrying the end-of-central-directory record) could fail, the error would
        // vanish, and we would rename a truncated archive into place, mark it done, and then prune
        // the last good generation. `sync_all` both surfaces that error and makes the bytes durable
        // before the DB is told the archive exists.
        let mut file = zip.close().await?.into_inner();
        file.flush().await?;
        file.sync_all().await?;
    }

    // Refuse to publish an archive that lost media it was supposed to contain. Checked BEFORE the
    // rename so a rejected build never reaches a servable path.
    check_export_completeness("ZIP", event_id, uploads.len(), written, skipped)
        .inspect_err(|_| tracing::error!("ZIP export for event {event_id}: refusing to publish"))?;

    tokio::fs::rename(&tmp_path, &out_path).await?;

    // Commit ONLY if our generation is still current. `finalize_job` is guarded on `epoch` — a
    // predicate on the very row it updates, so Postgres re-evaluates it correctly even when the
    // statement blocks behind a concurrent reopen. If we lost, our archive is stale: discard it.
    //
    // Note what ISN'T here any more: the ready-flag flip. Readiness is derived from
    // (released AND job.epoch = event.epoch AND status = 'done'), so writing `done` at a live epoch
    // IS the publish, atomically. A worker at a dead epoch simply writes a row nobody can see.
    if !finalize_job(pool, event_id, "zip", epoch, &format!("exports/{out_name}")).await? {
        let _ = tokio::fs::remove_file(&out_path).await;
        tracing::info!(
            "ZIP export for event {event_id} superseded (epoch {epoch} retired); discarded"
        );
        // Err, NOT Ok. `abandon_if_superseded` swallows this sentinel for the caller, so the
        // outcome is unchanged — but `run_*_export`'s deferred prune keys off `res.is_ok()`,
        // and a worker that LOST the epoch race reporting success made it reclaim generations
        // older than its own RETIRED epoch. At that moment the winning generation is still
        // `pending` with no file, so nothing protected the last good archive and it was deleted
        // with no replacement on disk — the precise outcome deferring the prune exists to stop.
        return Err(Superseded.into());
    }

    prune_stale_export_files(pool, &exports_dir, "Gallery", event_id, epoch).await;

    let _ = sse_tx.send(SseEvent {
        event_type: "export-progress".to_string(),
        data: serde_json::json!({ "type": "zip", "progress_pct": 100 }).to_string(),
    });

    tracing::info!("ZIP export complete for event {event_id}");
    Ok(())
}

// ── HTML viewer export ──────────────────────────────────────────────────────

/// Where a media entry's bytes come from at ZIP-writing time. Derived variants
/// (thumbnails, compressed images) are staged under the temp dir; original-fidelity
/// variants (videos, small images) are streamed straight from the source on disk so
/// the export never transiently doubles disk usage by copying large files to temp.
enum MediaSource {
    Temp(PathBuf),
    Original(PathBuf),
}

impl MediaSource {
    fn path(&self) -> &Path {
        match self {
            MediaSource::Temp(p) | MediaSource::Original(p) => p,
        }
    }
}

#[allow(clippy::too_many_arguments)]
async fn run_html_export(
    event_id: Uuid,
    epoch: i64,
    event_name: &str,
    comments_enabled: bool,
    pool: &PgPool,
    media_path: &Path,
    export_path: &Path,
    sse_tx: &broadcast::Sender<SseEvent>,
) -> Result<()> {
    if !claim_job(pool, event_id, "html", epoch).await? {
        // Another worker owns this generation, or our epoch has been retired. Not ours.
        return Ok(());
    }

    // See run_zip_export: refuse at the door rather than ENOSPC mid-write, and reclaim the
    // superseded generation only AFTER this one lands — unless it is the only way to land at all.
    ensure_export_space_reclaiming(pool, event_id, export_path, "Memories", epoch).await?;

    let res = run_html_export_inner(
        epoch,
        event_id,
        event_name,
        comments_enabled,
        pool,
        media_path,
        export_path,
        sse_tx,
    )
    .await;
    if res.is_err() {
        // Clean up this generation's temp artifacts so a failing (or abandoned) export can't leak
        // them — the leak is what fills the disk, which is what corrupts the next archive.
        let _ =
            tokio::fs::remove_file(export_path.join(gen_name(event_id, "Memories", epoch, ".tmp")))
                .await;
        let _ =
            tokio::fs::remove_dir_all(export_path.join(format!("viewer_tmp_{event_id}_{epoch}")))
                .await;
    }

    // Only once the new keepsake exists — see the reasoning in run_zip_export. A failed rebuild
    // must never be the reason the previous one is gone.
    if res.is_ok() {
        prune_superseded_archives(pool, export_path, "Memories", event_id, epoch).await;
    }

    abandon_if_superseded("HTML", event_id, epoch, res)
}

#[allow(clippy::too_many_arguments)]
async fn run_html_export_inner(
    epoch: i64,
    event_id: Uuid,
    event_name: &str,
    comments_enabled: bool,
    pool: &PgPool,
    media_path: &Path,
    export_path: &Path,
    sse_tx: &broadcast::Sender<SseEvent>,
) -> Result<()> {
    // 1. Query data
    let uploads = query_uploads(pool, event_id).await?;
    let comments = query_comments(pool, event_id).await?;
    let hashtags_per_upload = query_hashtags(pool, event_id).await?;
    let total = uploads.len().max(1) as f32;

    let _ = update_progress(pool, event_id, "html", epoch, 5).await;

    // Written OUTSIDE media_path: the public /media ServeDir must never reach these.
    let exports_dir = export_path.to_path_buf();
    tokio::fs::create_dir_all(&exports_dir).await?;

    // 2. Create temp directory for media processing (per-generation — see run_zip_export).
    let tmp_dir = exports_dir.join(format!("viewer_tmp_{event_id}_{epoch}"));
    let media_tmp = tmp_dir.join("media");
    tokio::fs::create_dir_all(&media_tmp).await?;

    // 3. Process media and build post data
    let mut viewer_posts: Vec<ViewerPost> = Vec::new();
    // (zip entry name under media/, where its bytes come from). Built here, streamed
    // into the ZIP in step 5 — so we also know the exact file count without a rescan.
    // The bool is "this entry is the FULL variant", i.e. the one `data.json` advertises as the
    // photo itself. It exists so `check_export_completeness` can count photos rather than files —
    // see the call site. Exactly one full entry is pushed per upload that survives the stat.
    let mut media_manifest: Vec<(String, MediaSource, bool)> = Vec::new();
    // Uploads dropped at the stat below never enter `media_manifest`, so without counting them
    // here a wholly-unreadable media directory yields an EMPTY manifest — expected 0, skipped 0 —
    // and the completeness check downstream would wave it through as a legitimately empty event.
    // These are the same loss as a skip at write time and are checked as one.
    let mut upload_skipped = 0usize;

    for (i, row) in uploads.iter().enumerate() {
        let src = media_path.join(&row.original_path);
        // Stat ONCE, up front, and skip this upload if the source is gone. The old code probed with
        // `exists()` here and then did `metadata(&src).await?` further down — a TOCTOU whose `?`
        // aborted the ENTIRE keepsake if the file vanished in between. It can still happen: the
        // compression worker no longer deletes originals on failure, but the hourly sweep reclaims
        // them once past the retention window, and an owner or host delete can land mid-export. A
        // missing source must degrade one entry, never the whole archive (which, once released, the
        // host cannot rebuild without reopening uploads).
        let src_meta = match tokio::fs::metadata(&src).await {
            Ok(m) => m,
            Err(e) => {
                tracing::warn!(
                    "HTML export: skipping upload {} — cannot stat {}: {e}",
                    row.id,
                    src.display()
                );
                upload_skipped += 1;
                continue;
            }
        };

        let is_video = row.mime_type.starts_with("video/");
        let id_str = row.id.to_string();

        // Generate thumbnail and full variant. `full_source` says where the full-res
        // bytes come from at ZIP time — for videos and small images that's the original
        // on disk (streamed directly, never copied to temp).
        let (thumb_name, full_name, full_source) = if is_video {
            let thumb = format!("{id_str}_thumb.jpg");
            let full_ext = ext_from_path(&row.original_path);
            let full = format!("{id_str}.{full_ext}");

            // Poster frame via the shared helper, which owns the seek order, the 120s timeout
            // (this call site had NONE — a hung ffmpeg would strand the export at `running`
            // forever) and the artifact check.
            let thumb_path = media_tmp.join(&thumb);
            let produced =
                match crate::services::video::extract_poster_frame(&src, &thumb_path, 400).await {
                    Ok(produced) => produced,
                    Err(e) => {
                        tracing::warn!("poster extraction errored for upload {}: {e:#}", row.id);
                        false
                    }
                };
            if !produced {
                tracing::info!(
                    upload_id = %row.id,
                    "no poster frame for this video; exporting it without one"
                );
            }

            // Stream the video full-res straight from the original at ZIP time — no
            // copy to temp (that used to transiently double disk usage per video).
            (
                produced.then(|| thumb.clone()),
                full,
                MediaSource::Original(src.clone()),
            )
        } else {
            let thumb = format!("{id_str}_thumb.jpg");
            let ext = ext_from_path(&row.original_path);
            let full = format!("{id_str}_full.{ext}");

            // Image thumbnail: resize to 400px wide
            let src_clone = src.clone();
            let thumb_path = media_tmp.join(&thumb);
            let thumb_path_clone = thumb_path.clone();

            // Same process-wide memory permit the compression worker takes. Without it, a
            // release fired while the last phone photos were still compressing put an export
            // decode and a heavy compression job in the same 1 GiB cgroup — and the OOM kill
            // marks the export failed, which `recover_exports` then re-spawns into the same
            // conditions on the next boot.
            let _heavy = heavy_permit_for(&src).await;
            let thumb_result = tokio::task::spawn_blocking(move || -> Result<()> {
                // `decode_oriented`, not `image::open`: the latter ignores the EXIF
                // orientation tag AND applies no decode limits. Using it here is why every
                // portrait photo came out sideways in the keepsake's HTML viewer grid — the
                // re-encode below drops the tag, so the viewer cannot recover it.
                let img = crate::services::imaging::decode_oriented(&src_clone)
                    .context("failed to open image for thumbnail")?;
                let resized = img.resize(400, 400, image::imageops::FilterType::Lanczos3);
                resized
                    .save_with_format(&thumb_path_clone, image::ImageFormat::Jpeg)
                    .context("failed to save thumbnail")?;
                Ok(())
            })
            .await
            // NOT `?`. The `?` here was on the JoinError, not on the closure's Result — so a
            // decoder PANIC (the `image` crate can panic on malformed input, and a resize can
            // abort on allocation) propagated out and failed the ENTIRE keepsake, where the very
            // same file merely failing returns `Err` and costs one tile. Worse, it was
            // deterministic: "Neu erzeugen" reads the same poison file and dies the same way. That
            // is the failure shape the completeness guard was reversed to eliminate, arriving
            // through the other door.
            .unwrap_or_else(|e| Err(anyhow::anyhow!("thumbnail task panicked: {e}")));

            // Same dangling-reference hazard as the video branch: a failure here left `thumb`
            // pointing at a file the ZIP writer would then skip, so `data.json` advertised an
            // entry the archive didn't contain. An undecodable image is rarer than a sub-second
            // clip, but the broken tile is identical.
            let thumb_ok = match thumb_result {
                Ok(()) => true,
                Err(e) => {
                    tracing::warn!("thumbnail generation failed for upload {}: {e:#}", row.id);
                    false
                }
            };

            // Full variant: compress to temp if >5MB, otherwise stream the original
            // as-is (no temp copy). `src_meta` was stat'd once at the top of the loop.
            let full_source = if src_meta.len() > 5_000_000 {
                let src_clone = src.clone();
                let full_path = media_tmp.join(&full);
                let full_path_clone = full_path.clone();

                // See the thumbnail above. This branch is the more expensive of the two: it
                // only runs for originals over 5 MB, i.e. exactly the giants.
                let _heavy = heavy_permit_for(&src).await;
                let compress_result = tokio::task::spawn_blocking(move || -> Result<()> {
                    // Same reason as the thumbnail above. This branch only runs for originals
                    // over 5 MB, which is why the viewer's full image looked correct for small
                    // photos and sideways for large ones — an inconsistency that reads as a
                    // viewer bug rather than an export one.
                    let img = crate::services::imaging::decode_oriented(&src_clone)
                        .context("failed to open image for compression")?;
                    let resized = img.resize(2000, 2000, image::imageops::FilterType::Lanczos3);
                    resized
                        .save_with_format(&full_path_clone, image::ImageFormat::Jpeg)
                        .context("failed to save compressed full image")?;
                    Ok(())
                })
                .await
                // See the thumbnail branch: a panic here must cost this one full variant (the
                // original is then streamed as-is below), never the whole keepsake.
                .unwrap_or_else(|e| Err(anyhow::anyhow!("full-image task panicked: {e}")));

                match compress_result {
                    Ok(()) => MediaSource::Temp(full_path),
                    Err(e) => {
                        tracing::warn!(
                            "compression failed for upload {}, using original: {e:#}",
                            row.id
                        );
                        MediaSource::Original(src.clone())
                    }
                }
            } else {
                MediaSource::Original(src.clone())
            };

            (thumb_ok.then_some(thumb), full, full_source)
        };

        // Register this post's media entries. The thumbnail is registered ONLY when one was
        // actually produced: pushing a manifest entry for a file that doesn't exist made the ZIP
        // writer skip it silently while `data.json` still advertised it — the viewer then drew a
        // broken image tile for an entry the archive never contained.
        if let Some(name) = &thumb_name {
            media_manifest.push((name.clone(), MediaSource::Temp(media_tmp.join(name)), false));
        }
        media_manifest.push((full_name.clone(), full_source, true));

        // Build comments for this upload
        let post_comments: Vec<ViewerComment> = comments
            .iter()
            .filter(|c| c.upload_id == row.id)
            .map(|c| ViewerComment {
                author: c.uploader_name.clone(),
                text: c.body.clone(),
                timestamp: c.created_at.to_rfc3339(),
            })
            .collect();

        // Build tags for this upload
        let tags: Vec<String> = hashtags_per_upload
            .iter()
            .filter(|(uid, _)| *uid == row.id)
            .map(|(_, tag)| tag.clone())
            .collect();

        viewer_posts.push(ViewerPost {
            id: id_str,
            uploader: row.uploader_name.clone(),
            caption: row.caption.clone().unwrap_or_default(),
            tags,
            timestamp: row.created_at.to_rfc3339(),
            likes: row.like_count,
            comments: post_comments,
            media: ViewerMedia {
                media_type: if is_video {
                    "video".to_string()
                } else {
                    "image".to_string()
                },
                // Empty when there is no poster. The viewer already guards on this
                // (`{#if post.media.thumb}` → a video tile with a play glyph, or the placeholder
                // icon for an image), so telling it the truth is the entire fix — no schema
                // change, no viewer rebuild. What was broken was the backend always claiming a
                // thumbnail existed.
                thumb: thumb_name
                    .as_ref()
                    .map(|n| format!("media/{n}"))
                    .unwrap_or_default(),
                full: format!("media/{full_name}"),
            },
        });

        let pct = 10 + ((i + 1) as f32 / total * 60.0) as i16;
        if !update_progress(pool, event_id, "html", epoch, pct.min(69)).await {
            return Err(Superseded.into());
        }
    }

    // 4. Build data.json
    let viewer_data = ViewerData {
        event: ViewerEvent {
            name: event_name.to_string(),
            exported_at: Utc::now().to_rfc3339(),
            comments_enabled,
        },
        posts: viewer_posts,
    };
    let data_json =
        serde_json::to_string_pretty(&viewer_data).context("failed to serialize data.json")?;

    // Match the live app's colour theme in the offline keepsake.
    let (theme_primary, theme_accent) = resolve_theme_seeds(pool).await;
    let theme_css = theme_override_css(&theme_primary, &theme_accent);

    let _ = update_progress(pool, event_id, "html", epoch, 72).await;

    // 5. Create ZIP (per-generation paths — see run_zip_export)
    let tmp_path = exports_dir.join(gen_name(event_id, "Memories", epoch, ".tmp"));
    let out_name = gen_name(event_id, "Memories", epoch, ".zip");
    let out_path = exports_dir.join(&out_name);

    {
        let file = tokio::fs::File::create(&tmp_path).await?;
        let mut zip = ZipFileWriter::with_tokio(file);

        // Write the embedded single-file viewer, injecting the export data as a
        // `window.__EXPORT_DATA__` global into index.html. Guests double-click
        // index.html (file://), where a cross-origin fetch() of a sibling file is
        // blocked — so the data must be inlined rather than fetched from data.json.
        write_viewer_with_data(&VIEWER_DIR, &mut zip, &data_json, theme_css.as_deref()).await?;

        let _ = update_progress(pool, event_id, "html", epoch, 75).await;

        // Write data.json
        {
            let builder = keepsake_entry("data.json".into(), Compression::Deflate);
            let mut entry = zip.write_entry_stream(builder).await?;
            let mut cursor = AllowStdIo::new(std::io::Cursor::new(data_json.as_bytes()));
            fcopy(&mut cursor, &mut entry).await?;
            entry.close().await?;
        }

        // Write README.txt
        {
            let builder = keepsake_entry("README.txt".into(), Compression::Deflate);
            let mut entry = zip.write_entry_stream(builder).await?;
            let mut cursor = AllowStdIo::new(std::io::Cursor::new(README_TEXT.as_bytes()));
            fcopy(&mut cursor, &mut entry).await?;
            entry.close().await?;
        }

        let _ = update_progress(pool, event_id, "html", epoch, 78).await;

        // Write media files from the manifest built in step 3. Thumbnails and derived
        // image variants stream from temp; video and small-image full variants stream
        // straight from the original on disk. Sources that don't exist (e.g. a thumb
        // whose ffmpeg step failed) are skipped — the viewer tolerates gaps.
        let file_total = media_manifest.len().max(1) as f32;
        let mut files_written = 0u32;
        // Photos, not files — the number `check_export_completeness` is actually about.
        //
        // `files_written` counts MANIFEST ROWS, and there are up to two per upload: a thumbnail and
        // a full variant. Thumbnails are 400px JPEGs this export GENERATES ITSELF into its own temp
        // dir, so they are no evidence that any original was captured. Counting them meant the one
        // remaining fatal case — nothing at all was written — could not fire while thumbnails kept
        // succeeding: if the media volume became unreadable after the stat pass, every original
        // open failed and every thumb open succeeded, and a keepsake with 100 thumbnails and ZERO
        // full-resolution photos published green, `done` at the live epoch, with the download
        // button lit. Boot recovery skips a `done` job, so nothing would ever have rebuilt it.
        let mut full_written = 0usize;
        // See `check_export_completeness`: a viewer that tolerates gaps must still not publish an
        // archive with no media in it at all.
        let mut media_skipped = 0usize;

        for (name, source, is_full) in &media_manifest {
            let path = source.path();
            // Open-first: a source that disappeared between the manifest being built and now (a
            // delete, or the hourly sweep reclaiming a long-failed original) must skip this entry,
            // not fail the whole viewer. Opening collapses the check and the use into one operation.
            let src_file = match tokio::fs::File::open(path).await {
                Ok(f) => f,
                Err(e) => {
                    tracing::warn!(
                        "HTML export: skipping media {name} — cannot read {}: {e}",
                        path.display()
                    );
                    media_skipped += 1;
                    continue;
                }
            };

            let entry_name = format!("media/{name}");
            let builder = keepsake_entry(entry_name, Compression::Stored);
            let mut zip_entry = zip.write_entry_stream(builder).await?;
            let mut f = src_file.compat();
            fcopy(&mut f, &mut zip_entry).await?;
            zip_entry.close().await?;

            files_written += 1;
            if *is_full {
                full_written += 1;
            }
            let pct = 78 + (files_written as f32 / file_total * 20.0) as i16;
            if !update_progress(pool, event_id, "html", epoch, pct.min(98)).await {
                return Err(Superseded.into());
            }
        }

        // Flush + fsync before the rename — see run_zip_export for why dropping the file here
        // would silently discard a failed final write (and hand us a truncated keepsake).
        let mut file = zip.close().await?.into_inner();
        file.flush().await?;
        file.sync_all().await?;

        // Before the rename, so a rejected viewer never reaches a servable path. Inside this
        // scope because `media_manifest` and the counters are scoped here.
        // Measured against the full upload set, not the manifest: an upload dropped at the stat
        // and one dropped at the write are the same loss to the guest looking for their photo.
        check_export_completeness(
            "HTML",
            event_id,
            uploads.len(),
            full_written,
            upload_skipped + media_skipped,
        )
        .inspect_err(|_| {
            tracing::error!("HTML export for event {event_id}: refusing to publish")
        })?;
    }

    // 6. Finalise
    tokio::fs::rename(&tmp_path, &out_path).await?;

    // Clean up temp directory
    let _ = tokio::fs::remove_dir_all(&tmp_dir).await;

    // Epoch-guarded finalize — writing `done` at a live epoch IS the publish (readiness is derived
    // from it), so there is no separate ready flag to flip. If our epoch was retired, we lost:
    // discard the stale archive.
    if !finalize_job(
        pool,
        event_id,
        "html",
        epoch,
        &format!("exports/{out_name}"),
    )
    .await?
    {
        let _ = tokio::fs::remove_file(&out_path).await;
        tracing::info!(
            "HTML export for event {event_id} superseded (epoch {epoch} retired); discarded"
        );
        // Err, NOT Ok. `abandon_if_superseded` swallows this sentinel for the caller, so the
        // outcome is unchanged — but `run_*_export`'s deferred prune keys off `res.is_ok()`,
        // and a worker that LOST the epoch race reporting success made it reclaim generations
        // older than its own RETIRED epoch. At that moment the winning generation is still
        // `pending` with no file, so nothing protected the last good archive and it was deleted
        // with no replacement on disk — the precise outcome deferring the prune exists to stop.
        return Err(Superseded.into());
    }

    prune_stale_export_files(pool, &exports_dir, "Memories", event_id, epoch).await;

    let _ = sse_tx.send(SseEvent {
        event_type: "export-progress".to_string(),
        data: serde_json::json!({ "type": "html", "progress_pct": 100 }).to_string(),
    });

    tracing::info!("HTML viewer export complete for event {event_id}");
    Ok(())
}

// ── DB helpers ───────────────────────────────────────────────────────────────

async fn query_uploads(pool: &PgPool, event_id: Uuid) -> Result<Vec<ExportUploadRow>> {
    Ok(sqlx::query_as::<_, ExportUploadRow>(concat!(
        "SELECT u.id, u.original_path, u.mime_type, u.caption,
                usr.display_name AS uploader_name,
                COUNT(DISTINCT l.user_id) AS like_count,
                u.created_at
         FROM upload u
         JOIN \"user\" usr ON usr.id = u.user_id
         LEFT JOIN \"like\" l ON l.upload_id = u.id
         ",
        export_visibility_where!(),
        "
         GROUP BY u.id, usr.display_name
         ORDER BY u.created_at ASC",
    ))
    .bind(event_id)
    .fetch_all(pool)
    .await?)
}

async fn query_comments(pool: &PgPool, event_id: Uuid) -> Result<Vec<ExportCommentRow>> {
    Ok(sqlx::query_as::<_, ExportCommentRow>(
        "SELECT c.upload_id, usr.display_name AS uploader_name, c.body, c.created_at
         FROM comment c
         JOIN \"user\" usr ON usr.id = c.user_id
         JOIN upload u ON u.id = c.upload_id
         WHERE u.event_id = $1 AND c.deleted_at IS NULL AND u.deleted_at IS NULL
           AND usr.uploads_hidden = FALSE AND usr.is_banned = FALSE
         ORDER BY c.created_at ASC",
    )
    .bind(event_id)
    .fetch_all(pool)
    .await?)
}

async fn query_hashtags(pool: &PgPool, event_id: Uuid) -> Result<Vec<(Uuid, String)>> {
    let rows: Vec<(Uuid, String)> = sqlx::query_as(
        "SELECT uh.upload_id, h.tag
         FROM upload_hashtag uh
         JOIN hashtag h ON h.id = uh.hashtag_id
         JOIN upload u ON u.id = uh.upload_id
         WHERE h.event_id = $1 AND u.deleted_at IS NULL",
    )
    .bind(event_id)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// Claim the pending job for `(event, type)` AT OUR EPOCH. `true` only if we won it.
///
/// Every predicate here is on the row being updated — no cross-table `EXISTS`. That is deliberate
/// and load-bearing. Under READ COMMITTED, when an UPDATE blocks on a row lock and the blocker
/// commits, Postgres re-evaluates the WHERE against the *updated target row* but answers subqueries
/// against OTHER tables from the statement's ORIGINAL snapshot. The previous guard
/// (`EXISTS (SELECT 1 FROM event WHERE ... export_released_at IS NOT NULL)`) was therefore unsound:
/// a claim blocking behind a concurrent reopen could see the pre-reopen event snapshot, pass the
/// check, and return the post-bump seq — handing the worker a LIVE generation on an event that was
/// already reopened. A same-row `epoch = $3` predicate is re-evaluated correctly by EPQ.
///
/// NOTE what this does and does NOT guarantee. It compares our birth epoch to the JOB ROW's epoch —
/// not to `event.export_epoch`. `open_event` bumps the event's epoch and writes nothing to
/// `export_job` (that is the point of the design: one write retires everything). So after a reopen
/// the row is still `pending` at our epoch and this claim SUCCEEDS: the worker will build an archive
/// nobody can ever see, because retirement is enforced at READ time (`export_current` requires
/// `j.epoch = e.export_epoch`), not at write time. That is wasted work, not incorrectness — and the
/// `update_progress` liveness check bails such a worker out early. Do not "optimise" this into a
/// cross-table check: that is exactly the unsound guard we removed.
///
/// Errors are distinguished from a lost claim: silently treating a pool timeout as "someone else
/// owns it" left the row `pending` at 0% with no live worker and no error — a spinner forever.
async fn claim_job(pool: &PgPool, event_id: Uuid, export_type: &str, epoch: i64) -> Result<bool> {
    let r = sqlx::query(
        "UPDATE export_job SET status = 'running'
         WHERE event_id = $1 AND type = $2::export_type
           AND epoch = $3 AND status = 'pending'",
    )
    .bind(event_id)
    .bind(export_type)
    .bind(epoch)
    .execute(pool)
    .await
    .context("claiming export job")?;
    Ok(r.rows_affected() > 0)
}

/// Guarded finalize: mark this export `done` and record its file path ONLY if our generation is
/// still current (`epoch` matches and we still hold it `running`).
///
/// This IS the publish step. Readiness is derived from `(released AND epoch = event.export_epoch
/// AND status = 'done')`, so a single row write makes the keepsake downloadable — there is no
/// second flag to flip and therefore no window between "done" and "visible". A worker whose epoch
/// was retired matches nothing here and must discard its output.
///
/// ERRORS ARE DISTINGUISHED FROM A LOST RACE, for the same reason `claim_job` distinguishes them —
/// and the consequence here is strictly worse. This used to end `.unwrap_or(false)`, collapsing a
/// pool timeout into "we were superseded". At that point the archive is already built, fsynced and
/// renamed into place, so the caller went on to DELETE the finished multi-GB file and return the
/// `Superseded` sentinel — which `abandon_if_superseded` swallows into `Ok(())`, so
/// `spawn_export_jobs` never called `mark_failed` either. The row stayed `running` at 99% at the
/// LIVE epoch, which the host dashboard renders as "Wird erstellt (99 %)" with the download
/// disabled, forever: no sweep re-examines `running` rows, and `recover_exports` runs only at boot.
///
/// A pool timeout is not exotic here. `max_connections` is 10, `acquire_timeout` 5s, and this fires
/// at the end of a full-gallery export while ~100 guests are uploading. Returning `Err` instead
/// leaves the finished archive on disk and lets the caller's `mark_failed` record a real reason.
async fn finalize_job(
    pool: &PgPool,
    event_id: Uuid,
    export_type: &str,
    epoch: i64,
    file_path: &str,
) -> Result<bool> {
    let r = sqlx::query(
        "UPDATE export_job
         SET status = 'done', progress_pct = 100, file_path = $3, completed_at = NOW()
         WHERE event_id = $1 AND type = $2::export_type
           AND epoch = $4 AND status = 'running'",
    )
    .bind(event_id)
    .bind(export_type)
    .bind(file_path)
    .bind(epoch)
    .execute(pool)
    .await
    .context("finalizing export job")?;
    Ok(r.rows_affected() > 0)
}

/// Parse the trailing generation number out of `<prefix><n><suffix>` (e.g.
/// `Gallery.<n>.zip`, `Gallery.zip.<n>.tmp`, `viewer_tmp_<event>_<n>`). Returns None if the
/// name doesn't fit the shape, so unrelated files are left untouched.
fn parse_gen_seq(name: &str, prefix: &str, suffix: &str) -> Option<i64> {
    name.strip_prefix(prefix)?
        .strip_suffix(suffix)?
        .parse::<i64>()
        .ok()
}

/// Filenames a live (current-epoch, `done`) job row still points at — OFF LIMITS to every prune,
/// regardless of the epoch encoded in the name.
///
/// A ViewerOnly regeneration carries the finished ZIP forward by re-stamping its row to the new
/// epoch WITHOUT renaming the file, so `Gallery.<event>.<older>.zip` is still the served archive and
/// deleting it by filename-epoch would 404 the download.
async fn protected_files(pool: &PgPool, event_id: Uuid) -> Vec<String> {
    sqlx::query_scalar::<_, String>(
        "SELECT split_part(j.file_path, '/', -1) FROM export_job j
          JOIN event e ON e.id = j.event_id
         WHERE j.event_id = $1 AND j.epoch = e.export_epoch
           AND j.status = 'done' AND j.file_path IS NOT NULL",
    )
    .bind(event_id)
    .fetch_all(pool)
    .await
    .unwrap_or_default()
}

/// Reclaim superseded FINAL archives.
///
/// CALLED AFTER A SUCCESSFUL BUILD, not before one. This doc used to argue the opposite at
/// length — that since readiness is derived from `job.epoch = event.export_epoch`, a superseded
/// archive is already unreachable and keeping it "buys nothing". That reasoning is right about
/// REACHABILITY and wrong about RECOVERABILITY: an epoch is a database value that can be rolled
/// back, deleted bytes cannot. Pruning first meant any rebuild that then failed — ENOSPC, an OOM,
/// a hung ffmpeg — left the event with NO archive at all, which is the one outcome the product
/// exists to prevent, at the one moment nobody is watching.
///
/// The single exception is phase 2 of `ensure_export_space_reclaiming`, where the previous
/// generation's bytes are the only way the rebuild can fit at all. Both call sites carry the full
/// reasoning; do not "restore" a pre-build prune on the strength of this function's convenience.
///
/// Narrower than [`prune_stale_export_files`] on purpose: FINAL archives only. Those are inert — a
/// superseded worker either already renamed its file (and will delete it itself when its guarded
/// `finalize_job` fails) or never will. `.tmp` files and `viewer_tmp_` staging dirs are NOT touched
/// here: a superseded worker can still be streaming into those, and at build START it is far more
/// likely to be alive than at finalize time.
async fn prune_superseded_archives(
    pool: &PgPool,
    exports_dir: &Path,
    prefix: &str,
    event_id: Uuid,
    keep_seq: i64,
) {
    let protected = protected_files(pool, event_id).await;
    let final_prefix = format!("{prefix}.{event_id}.");
    let mut rd = match tokio::fs::read_dir(exports_dir).await {
        Ok(rd) => rd,
        Err(_) => return,
    };
    let mut reclaimed = 0u64;
    while let Ok(Some(entry)) = rd.next_entry().await {
        let name = entry.file_name();
        let name = name.to_string_lossy();
        if !is_superseded_archive(&name, &final_prefix, keep_seq, &protected) {
            continue;
        }
        let len = entry.metadata().await.map(|m| m.len()).unwrap_or(0);
        if tokio::fs::remove_file(entry.path()).await.is_ok() {
            reclaimed += len;
        }
    }
    if reclaimed > 0 {
        tracing::info!(
            "reclaimed {reclaimed} bytes of superseded {prefix} archives before rebuilding \
             event {event_id} @ epoch {keep_seq}"
        );
    }
}

/// Is `name` a FINAL archive of a strictly-older generation that no live row points at?
///
/// Pure so the two dangerous cases can be pinned without a filesystem: the carried-forward archive
/// (protected despite an older epoch in its name) and the in-flight `.tmp` (never matched at all).
fn is_superseded_archive(
    name: &str,
    final_prefix: &str,
    keep_seq: i64,
    protected: &[String],
) -> bool {
    if protected.iter().any(|p| p == name) {
        return false;
    }
    parse_gen_seq(name, final_prefix, ".zip").is_some_and(|n| n < keep_seq)
}

/// Bytes the media in this event's keepsake will occupy, as an UPPER BOUND per archive.
///
/// Both archives write their media entries `Compression::Stored`, so an archive is essentially a
/// byte-for-byte second copy of the originals: `Gallery.zip` always, and `Memories.zip` for every
/// video ([`MediaSource::Original`]) and every image at or under 5 MB. Images over 5 MB are
/// downscaled to 2000px first, which only makes this estimate more conservative — the direction we
/// want, since being wrong low means ENOSPC halfway through.
///
/// Shares [`query_uploads`]' visibility filter via [`export_visibility_where`], so hidden/banned
/// uploads can't be counted here but skipped there (or the reverse, which under-reserves).
pub async fn estimate_export_bytes(pool: &PgPool, event_id: Uuid) -> Result<u64> {
    let (bytes,): (i64,) = sqlx::query_as(concat!(
        "SELECT COALESCE(SUM(u.original_size_bytes), 0)::bigint
           FROM upload u
           JOIN \"user\" usr ON usr.id = u.user_id
           ",
        export_visibility_where!(),
    ))
    .bind(event_id)
    .fetch_one(pool)
    .await
    .context("estimating the export size")?;
    Ok(bytes.max(0) as u64)
}

/// Headroom multiplier over the raw media sum: ZIP central directory, per-entry headers, the
/// embedded viewer, and the HTML export's temp staging.
const EXPORT_SIZE_OVERHEAD_PCT: u64 = 110;

/// Bytes this export must have available, given the raw media sum and how many jobs are competing.
///
/// `armed` is the multiplier that keeps two concurrent workers honest. `spawn_export_jobs` starts
/// the ZIP and the HTML halves at the same instant and both are gallery-sized, so a worker that
/// reserved only for itself would see "it fits", its sibling would independently see the same, and
/// together they would not fit — which is precisely the ENOSPC this preflight exists to prevent.
/// Computed in `u128` and clamped, NOT with `saturating_mul`: saturating first and then dividing by
/// 100 quietly turns an overflow into a number ~100x too small, which is the one direction that
/// matters here — an under-estimate authorises the very write the preflight exists to refuse.
pub(crate) fn required_free_bytes(media_bytes: u64, armed: i64) -> u64 {
    let needed = media_bytes as u128 * EXPORT_SIZE_OVERHEAD_PCT as u128 / 100
        * armed.max(1).min(i64::from(u32::MAX)) as u128;
    needed.min(u64::MAX as u128) as u64
}

/// Free bytes a full keepsake build would need RIGHT NOW, both halves included.
///
/// The same arithmetic the preflight uses, exposed so the host dashboard can warn BEFORE the
/// release rather than reporting a failure after it. The preflight can only ever say "this didn't
/// fit"; at that point the gallery is full, the event is over, and the remedies (ask guests to stop
/// uploading, grow the volume) are all much harder. Hard-codes both halves because that is what a
/// release arms.
pub async fn keepsake_space_required(pool: &PgPool, event_id: Uuid) -> Result<u64> {
    Ok(required_free_bytes(
        estimate_export_bytes(pool, event_id).await?,
        2,
    ))
}

/// Refuse to start an export that cannot fit, with a reason the host can act on.
///
/// Without this the failure mode is ENOSPC halfway through a multi-GB write, and the wreckage
/// outlives it: the epoch has already moved, so the job row is `failed` at the CURRENT generation
/// and `GET /export/zip` 404s, while the last good archive sits on disk unreferenced. The host's
/// only escape (`POST /host/export/rebuild`) needs the very space that isn't there. Failing at the
/// door instead leaves the disk untouched and puts a number in front of the operator.
///
/// Both halves are spawned concurrently and both are gallery-sized, so a worker must reserve for
/// its live sibling too — otherwise each independently sees "it fits", and together they don't.
async fn ensure_export_space(pool: &PgPool, event_id: Uuid, export_path: &Path) -> Result<()> {
    let media_bytes = estimate_export_bytes(pool, event_id).await?;

    // Every job armed at any epoch for this event that hasn't finished is competing for this disk.
    let (armed,): (i64,) = sqlx::query_as(
        "SELECT COUNT(*) FROM export_job
          WHERE event_id = $1 AND status IN ('pending', 'running')",
    )
    .bind(event_id)
    .fetch_one(pool)
    .await
    .context("counting armed export jobs")?;
    let needed = required_free_bytes(media_bytes, armed);

    // `None` = the mount couldn't be resolved. Fail OPEN, exactly as the upload quota does: refusing
    // to build the keepsake because we can't read a number would be a worse failure than trying.
    let Some(free) = crate::services::disk::free_bytes(export_path) else {
        tracing::warn!("export preflight: disk snapshot unavailable; proceeding without the check");
        return Ok(());
    };

    // The archive may not consume the last byte of the volume. `needed` alone authorised an
    // export sized at exactly `free`: it would pass the check, run for half an hour, and land
    // the box at zero — at which point Postgres cannot write WAL and the event is over, with
    // the keepsake still unfinished. `postgres_data`, `media_data` and `exports_data` share one
    // filesystem, so "enough room for the archive" was never the same question as "enough room
    // for the archive AND a working database".
    //
    // Same reserve the upload path and the host dashboard's low-disk banner use, so all three
    // agree on what "full" means.
    let required = needed.saturating_add(crate::handlers::upload::DISK_RESERVE_BYTES as u64);

    if free < required {
        let gb = |b: u64| b as f64 / 1_000_000_000.0;
        tracing::error!(
            needed,
            required,
            free,
            armed,
            "export preflight: not enough free space to build the keepsake for event {event_id}"
        );
        anyhow::bail!(
            "Nicht genug Speicherplatz für das Keepsake: benötigt ca. {:.1} GB (plus {:.0} GB \
             Reserve), frei sind {:.1} GB. Bitte Speicher freigeben und das Keepsake \
             anschließend neu erstellen.",
            gb(needed),
            gb(crate::handlers::upload::DISK_RESERVE_BYTES as u64),
            gb(free)
        );
    }
    Ok(())
}

/// How many bytes short of buildable we are, or `None` when there is already room.
///
/// Split out of `ensure_export_space` so the reclaiming wrapper can ask "would pruning be
/// enough?" BEFORE it destroys anything. Mirrors that function's arithmetic exactly; the one
/// deliberate difference is that an unresolvable mount reports `None` (fail open) rather than a
/// deficit, so a missing disk reading can never be the reason we delete the only archive.
async fn export_space_deficit(
    pool: &PgPool,
    event_id: Uuid,
    export_path: &Path,
) -> Result<Option<u64>> {
    let media_bytes = estimate_export_bytes(pool, event_id).await?;
    let (armed,): (i64,) = sqlx::query_as(
        "SELECT COUNT(*) FROM export_job
          WHERE event_id = $1 AND status IN ('pending', 'running')",
    )
    .bind(event_id)
    .fetch_one(pool)
    .await
    .context("counting armed export jobs")?;
    let required = required_free_bytes(media_bytes, armed)
        .saturating_add(crate::handlers::upload::DISK_RESERVE_BYTES as u64);

    let Some(free) = crate::services::disk::free_bytes(export_path) else {
        return Ok(None);
    };
    Ok((free < required).then(|| required - free))
}

/// Sum what `prune_superseded_archives` WOULD reclaim, without deleting anything.
///
/// Deliberately shares `is_superseded_archive` and `protected_files` with the real prune, so the
/// estimate cannot drift from what actually gets removed. An unreadable directory reports 0,
/// which makes the caller refuse to prune — the safe direction.
async fn reclaimable_superseded_bytes(
    pool: &PgPool,
    exports_dir: &Path,
    prefix: &str,
    event_id: Uuid,
    keep_seq: i64,
) -> u64 {
    let protected = protected_files(pool, event_id).await;
    let final_prefix = format!("{prefix}.{event_id}.");
    let Ok(mut rd) = tokio::fs::read_dir(exports_dir).await else {
        return 0;
    };
    let mut total = 0u64;
    while let Ok(Some(entry)) = rd.next_entry().await {
        let name = entry.file_name();
        let name = name.to_string_lossy();
        if !is_superseded_archive(&name, &final_prefix, keep_seq, &protected) {
            continue;
        }
        total = total.saturating_add(entry.metadata().await.map(|m| m.len()).unwrap_or(0));
    }
    total
}

/// Best-effort removal of stale per-generation export artifacts for one export type. Deletes
/// ONLY strictly-older generations (`n < keep_seq`) — never `keep_seq`'s own current file,
/// and never a NEWER generation that a concurrent re-release may already be producing (that
/// "delete all but mine" would let a lagging older winner nuke a newer live file). Covers the
/// finished archive (`<prefix>.<n>.zip`), its temp (`<prefix>.zip.<n>.tmp`), and — for html —
/// the `viewer_tmp_<event>_<n>` staging dir, so crash-orphaned per-generation temps don't
/// accumulate. Runs after a worker wins its finalize; older generations can never be served
/// again (their `file_path` was nulled by the re-release that superseded them). Tolerates
/// races and IO errors — purely disk hygiene.
async fn prune_stale_export_files(
    pool: &PgPool,
    exports_dir: &Path,
    prefix: &str,
    event_id: Uuid,
    keep_seq: i64,
) {
    let protected = protected_files(pool, event_id).await;

    // EVERY shape is event-scoped. All events share one exports volume, so a name keyed only by
    // generation would let event A's prune delete event B's live keepsake (and let two events
    // collide on the same archive path). `viewer_tmp_` was already scoped; the archives were not.
    let final_prefix = format!("{prefix}.{event_id}."); // Gallery.<event>. / Memories.<event>.
    let viewer_prefix = format!("viewer_tmp_{event_id}_");
    let mut rd = match tokio::fs::read_dir(exports_dir).await {
        Ok(rd) => rd,
        Err(_) => return,
    };
    while let Ok(Some(entry)) = rd.next_entry().await {
        let name = entry.file_name();
        let name = name.to_string_lossy();
        // Try each artifact shape; a strictly-older generation in any of them marks it for
        // deletion. Only the FINAL archive and the viewer staging dir are swept here — never a
        // `.tmp`, which may belong to a superseded worker that is still streaming into it. Deleting
        // that out from under it made its rename fail with a confusing hard error; it cleans up its
        // own temp on the way out now.
        let seq = parse_gen_seq(&name, &final_prefix, ".zip").or_else(|| {
            if prefix == "Memories" {
                parse_gen_seq(&name, &viewer_prefix, "")
            } else {
                None
            }
        });
        if protected.iter().any(|p| p == name.as_ref()) {
            continue;
        }
        if seq.is_some_and(|n| n < keep_seq) {
            let path = entry.path();
            let is_dir = entry.file_type().await.map(|t| t.is_dir()).unwrap_or(false);
            let _ = if is_dir {
                tokio::fs::remove_dir_all(&path).await
            } else {
                tokio::fs::remove_file(&path).await
            };
        }
    }
}

/// Mark this worker's export `failed` — ONLY for the generation it owns (`epoch` matches). A
/// superseded worker's failure is a no-op, so it can't clobber the fresh generation's row.
///
/// The status guard admits `pending` as well as `running`: if `claim_job` itself ERRORS (pool
/// timeout, connection reset) the row is still `pending`, and a guard of `status = 'running'` would
/// match nothing — leaving the job sitting at 0% with no worker and no error message, a spinner
/// forever. The epoch guard is what keeps this safe; the status guard is only there to avoid
/// stomping a `done` row.
async fn mark_failed(pool: &PgPool, event_id: Uuid, export_type: &str, epoch: i64, msg: &str) {
    let _ = sqlx::query(
        "UPDATE export_job SET status = 'failed', error_message = $3
         WHERE event_id = $1 AND type = $2::export_type
           AND epoch = $4 AND status IN ('running', 'pending')",
    )
    .bind(event_id)
    .bind(export_type)
    .bind(msg)
    .bind(epoch)
    .execute(pool)
    .await;
}

/// Update the progress bar for THIS generation only, and report whether we're STILL THE LIVE
/// GENERATION.
///
/// The WHERE clause (`epoch = ours AND status = 'running'`) is exactly the liveness predicate, so
/// this write already tells us for free whether we've been superseded — it just used to throw the
/// answer away. Returning it lets the file loops bail the instant a reopen/re-release/takedown
/// retires us, instead of grinding through every remaining image and writing a whole archive we
/// will then delete. `false` = we lost (or the row is gone); a DB error is reported as still-live,
/// because a transient blip must not abandon a legitimate export.
async fn update_progress(
    pool: &PgPool,
    event_id: Uuid,
    export_type: &str,
    epoch: i64,
    pct: i16,
) -> bool {
    match sqlx::query(
        "UPDATE export_job SET progress_pct = $3
         WHERE event_id = $1 AND type = $2::export_type
           AND epoch = $4 AND status = 'running'",
    )
    .bind(event_id)
    .bind(export_type)
    .bind(pct)
    .bind(epoch)
    .execute(pool)
    .await
    {
        Ok(r) => r.rows_affected() > 0,
        // Don't abandon a good export over a transient pool hiccup — the epoch-guarded finalize is
        // still the authority, so at worst we do some extra work.
        Err(e) => {
            tracing::warn!("progress update failed for {export_type} @ epoch {epoch}: {e:#}");
            true
        }
    }
}

/// Sentinel error for "our generation was retired mid-export". The caller discards its output and
/// returns Ok — this is a normal, expected outcome, not a failure worth marking on the job row.
#[derive(Debug)]
struct Superseded;

impl std::fmt::Display for Superseded {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "export generation superseded")
    }
}

impl std::error::Error for Superseded {}

/// Broadcast `export-available` once BOTH halves are downloadable. Reads the derived predicate
/// (a `done` row at the event's current epoch, on a released event) rather than a stored flag, so
/// it can never advertise a keepsake that a reopen has already invalidated.
async fn maybe_broadcast_complete(
    pool: &PgPool,
    event_id: Uuid,
    sse_tx: &broadcast::Sender<SseEvent>,
) {
    let complete: bool = sqlx::query_scalar(
        "SELECT COUNT(*) = 2 FROM export_current
          WHERE event_id = $1 AND status = 'done'",
    )
    .bind(event_id)
    .fetch_one(pool)
    .await
    .unwrap_or(false);

    {
        if complete {
            let _ = sse_tx.send(SseEvent {
                event_type: "export-available".to_string(),
                data: serde_json::json!({ "types": ["zip", "html"] }).to_string(),
            });
        }
    }
}

/// Write the embedded viewer into the ZIP, injecting the export data as a
/// `window.__EXPORT_DATA__` global into `index.html`. The keepsake is opened by
/// double-clicking `index.html` (file://), where browsers block a cross-origin
/// `fetch()` of a sibling `data.json` — so the data is inlined into the page.
/// (`data.json` is still written separately for the http-served case.)
/// Permissions stamped on every entry in both archives: `rw-r--r--`.
///
/// `ZipEntryBuilder::new` leaves the external file attribute at zero, and the host compatibility
/// defaults to Unix — so every entry was written with a stored mode of **0000**. Windows Explorer
/// ignores Unix modes and was fine, which is exactly why this survived: on Linux and macOS
/// `unzip` faithfully applies what the archive asks for, and the guest gets a directory of files
/// none of which they can open. `?---------` on every line of `unzip -Z`.
///
/// That is the keepsake — the artifact the whole event exists to produce — arriving unreadable,
/// after distribution, with no server-side symptom at all.
/// `S_IFREG | 0644`. The type bits are included because the mode is written whole into the high
/// half of the external file attribute: without them extractors see a file of type "unknown"
/// (`unzip -Z` renders `?rw-r--r--`), which works but is not what the archive means to say.
const KEEPSAKE_ENTRY_MODE: u16 = 0o100_644;

/// Build a ZIP entry for the keepsake. ALL entries in both archives go through here so the mode
/// can't be forgotten at one of the six call sites.
fn keepsake_entry(name: String, compression: Compression) -> ZipEntryBuilder {
    ZipEntryBuilder::new(name.into(), compression).unix_permissions(KEEPSAKE_ENTRY_MODE)
}

/// Escape a JSON payload for inlining inside a `<script>` element.
///
/// See the call site in [`write_viewer_with_data`] for why this is every `<` and not just `</`.
/// Kept separate so the property that matters — no `<` survives, and the value still decodes to
/// the original — can be asserted without building a ZIP.
fn escape_json_for_script(data_json: &str) -> String {
    data_json.replace('<', "\\u003c")
}

async fn write_viewer_with_data(
    dir: &include_dir::Dir<'_>,
    zip: &mut ZipFileWriter<tokio::fs::File>,
    data_json: &str,
    theme_css: Option<&str>,
) -> Result<()> {
    for file in dir.files() {
        let path = file.path().to_string_lossy().to_string();
        if path == "index.html" {
            let html = std::str::from_utf8(file.contents())
                .context("export-viewer index.html is not valid UTF-8")?;
            // Escape EVERY `<`, not just `</`.
            //
            // `</` -> `<\/` stops the obvious break-out (`</script><img onerror=…>`) and is inert
            // against XSS. It does not stop the caption steering the HTML TOKENIZER. A caption
            // containing `<!--<script` with no later `-->` puts the parser into
            // script-data-double-escaped state; from there the template's own `</script>` only
            // steps back to script-data-escaped instead of closing the element, and the rest of the
            // document — including the viewer bundle — is swallowed as script data. Nothing
            // executes and nothing leaks; `window.__EXPORT_DATA__` is simply never assigned and the
            // keepsake opens blank.
            //
            // That failure is silent and POST-DISTRIBUTION: the export succeeds, the ZIP is
            // well-formed, the job writes `done`, /export/status is green, and the host hands out a
            // file that only fails when a guest double-clicks index.html — in every copy, with no
            // way to fix it after the fact. Reachable from any guest-authored caption or comment,
            // since both are embedded in the viewer.
            //
            // `<` never appears in JSON structural syntax — only inside string values — so a global
            // replace is sound, and `<` is valid in both JSON and a JS string literal. One
            // rule covers `</script`, `<!--` and `<script` together, which is the point: the
            // previous escape was named for the single case it did handle.
            //
            // NOTE this is deliberately only for the INLINED copy. `data.json` is written
            // separately, in no HTML context, and must stay literal.
            let safe = escape_json_for_script(data_json);
            // Match the live app's colour theme: inject the same `:root:root{…}` override
            // the app builds at runtime so a rose/sage/custom event exports a rose/sage
            // keepsake (not the embedded default gold). The CSS is generated purely from
            // hex seeds (theme_override_css), so there's nothing to escape.
            let mut head_inject = String::new();
            if let Some(css) = theme_css {
                head_inject.push_str(&format!("<style id=\"es-theme\">{css}</style>"));
            }
            head_inject.push_str(&format!("<script>window.__EXPORT_DATA__={safe};</script>"));
            let injected = match html.find("</head>") {
                Some(idx) => format!("{}{}{}", &html[..idx], head_inject, &html[idx..]),
                None => format!("{head_inject}{html}"),
            };
            let builder = keepsake_entry(path, Compression::Deflate);
            let mut entry = zip.write_entry_stream(builder).await?;
            let mut cursor = AllowStdIo::new(std::io::Cursor::new(injected.as_bytes()));
            fcopy(&mut cursor, &mut entry).await?;
            entry.close().await?;
        } else {
            let builder = keepsake_entry(path, Compression::Deflate);
            let mut entry = zip.write_entry_stream(builder).await?;
            let mut cursor = AllowStdIo::new(std::io::Cursor::new(file.contents()));
            fcopy(&mut cursor, &mut entry).await?;
            entry.close().await?;
        }
    }
    for sub_dir in dir.dirs() {
        Box::pin(write_viewer_with_data(sub_dir, zip, data_json, theme_css)).await?;
    }
    Ok(())
}

/// Default champagne-gold seed — matches the hand-tuned ramp already embedded in the
/// viewer, so a default-themed event injects no override.
const KEEPSAKE_DEFAULT_SEED: &str = "#8a6a2b";

// stop → color-mix instruction. MIRRORS `LADDER` in frontend/src/lib/theme/palette.ts —
// keep the two in sync so an exported keepsake matches the live app pixel-for-pixel.
const THEME_LADDER: &[(u16, Option<&str>)] = &[
    (50, Some("white 90%")),
    (100, Some("white 80%")),
    (200, Some("white 62%")),
    (300, Some("white 42%")),
    (400, Some("white 22%")),
    (500, Some("white 9%")),
    (600, None),
    (700, Some("black 15%")),
    (800, Some("black 30%")),
    (900, Some("black 45%")),
    (950, Some("black 63%")),
];

fn theme_stop_value(seed: &str, mix: Option<&str>) -> String {
    match mix {
        Some(m) => format!("color-mix(in oklab, {seed}, {m})"),
        None => seed.to_string(),
    }
}

fn theme_ramp(families: &[&str], seed: &str) -> String {
    let mut out = String::new();
    for (stop, mix) in THEME_LADDER {
        let val = theme_stop_value(seed, *mix);
        for fam in families {
            out.push_str(&format!("--color-{fam}-{stop}:{val};"));
        }
    }
    out
}

fn is_hex_color(s: &str) -> bool {
    let b = s.as_bytes();
    b.len() == 7 && b[0] == b'#' && b[1..].iter().all(|c| c.is_ascii_hexdigit())
}

/// Build the keepsake's `:root:root{…}` colour override from two seed colours, or None
/// for the default gold (viewer already carries that ramp) or an invalid seed (fall back
/// to the embedded default rather than emit unsafe CSS). MIRRORS `buildPaletteCss` in
/// frontend/src/lib/theme/palette.ts.
fn theme_override_css(primary: &str, accent: &str) -> Option<String> {
    if !is_hex_color(primary) || !is_hex_color(accent) {
        return None;
    }
    if primary.eq_ignore_ascii_case(KEEPSAKE_DEFAULT_SEED)
        && accent.eq_ignore_ascii_case(KEEPSAKE_DEFAULT_SEED)
    {
        return None;
    }
    let accent500 = theme_stop_value(accent, Some("white 9%"));
    let mut css = String::from(":root:root{");
    css.push_str(&theme_ramp(&["blue", "primary"], primary));
    css.push_str(&theme_ramp(&["purple"], accent));
    css.push_str(&format!(
        "--color-violet-500:{accent500};--color-violet-600:{accent};"
    ));
    css.push_str(&format!(
        "--color-accent-500:{accent500};--color-accent-600:{accent};"
    ));
    css.push('}');
    Some(css)
}

/// Read the active theme seeds from the runtime `config` table (set by the admin UI),
/// falling back to the default gold. NOTE: an env-only default (THEME_PRIMARY set but
/// never saved via the admin UI) isn't stored in this table, so such a keepsake would
/// use gold; admin-set themes — the normal path — match the app exactly.
async fn resolve_theme_seeds(pool: &PgPool) -> (String, String) {
    async fn read(pool: &PgPool, key: &str) -> String {
        sqlx::query_scalar::<_, String>("SELECT value FROM config WHERE key = $1")
            .bind(key)
            .fetch_optional(pool)
            .await
            .ok()
            .flatten()
            .unwrap_or_else(|| KEEPSAKE_DEFAULT_SEED.to_string())
    }
    (
        read(pool, "theme_primary").await,
        read(pool, "theme_accent").await,
    )
}

fn ext_from_path(path: &str) -> &str {
    path.rsplit('.').next().unwrap_or("bin")
}

fn sanitize_name(name: &str) -> String {
    name.chars()
        .map(|c| {
            if c.is_alphanumeric() || c == '-' {
                c
            } else {
                '_'
            }
        })
        .collect()
}

// ── Static content ───────────────────────────────────────────────────────────

const README_TEXT: &str = "EventSnap Offline-Galerie\n\
\n\
So geht's:\n\
1. Entpacke diese ZIP-Datei\n\
   (Windows: Rechtsklick > \"Alle extrahieren\"; Mac: Doppelklick;\n\
    Handy: Dateimanager-App verwenden).\n\
2. Öffne \"index.html\" im Browser\n\
   (z. B. Chrome, Safari oder Firefox).\n\
3. Stöbere durch alle Fotos und Videos.\n\
   Du kannst zwischen Listen- und Rasteransicht wechseln,\n\
   nach Hashtags filtern und nach Nutzern suchen.\n\
4. Eine Internetverbindung ist nicht nötig.\n\
   Alles ist lokal auf deinem Gerät gespeichert.\n\
\n\
Viel Freude mit den Erinnerungen!\n";

// ── Tests ────────────────────────────────────────────────────────────────────

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

    const EVT: &str = "11111111-1111-1111-1111-111111111111";

    fn gallery_prefix() -> String {
        format!("Gallery.{EVT}.")
    }

    #[test]
    fn a_burst_of_invalidations_still_starts_a_build_within_one_window() {
        // B7's invariant, and the reason the burst start is tracked at all. A per-request delay
        // meant every invalidation restarted the wait, so a host moderating faster than one action
        // per 20 seconds deferred the build forever — `/export/html` 404s and the UI sits on
        // "Wird vorbereitet…" for the rest of the event.
        //
        // What must hold is that the delay is measured from the FIRST request in the burst, so the
        // wait never grows: request #50 of a takedown pass is scheduled no later than request #1
        // was. Coalescing survives (later requests still collapse into one build); starvation does
        // not.
        let event = Uuid::new_v4();
        clear_regen_burst(event);

        let first = regen_delay_for(event);
        assert!(
            first <= REGEN_DEBOUNCE,
            "the first invalidation of a burst waits at most one debounce window"
        );

        let mut previous = first;
        for _ in 0..50 {
            let next = regen_delay_for(event);
            assert!(
                next <= previous,
                "a later invalidation must never push the build further out than an earlier one"
            );
            previous = next;
        }

        // And the burst is a scheduling window, not a permanent state: once a worker has run,
        // the next invalidation is a fresh burst that gets the full coalescing delay again.
        clear_regen_burst(event);
        assert!(
            regen_delay_for(event) >= previous,
            "clearing the burst starts a fresh debounce window"
        );
        clear_regen_burst(event);
    }

    #[test]
    fn one_events_burst_never_schedules_another_events_rebuild() {
        // The map is keyed per event because the delay is a property of THAT event's moderation
        // pass. A shared window would let a busy event drag an idle one's keepsake along with it —
        // or worse, let an idle event's stale entry start a busy one's build immediately, which is
        // the stampede the debounce exists to prevent.
        let busy = Uuid::new_v4();
        let quiet = Uuid::new_v4();
        clear_regen_burst(busy);
        clear_regen_burst(quiet);

        for _ in 0..10 {
            regen_delay_for(busy);
        }
        clear_regen_burst(busy);

        assert!(
            regen_delay_for(quiet) <= REGEN_DEBOUNCE,
            "an untouched event gets its own full window"
        );
        clear_regen_burst(quiet);
    }

    #[test]
    fn a_strictly_older_archive_is_reclaimed() {
        // The whole point: at the start of a rebuild at epoch 5, generation 4's archive is dead
        // weight — readiness is derived from `epoch = event.export_epoch`, so it is already
        // unreachable — and its bytes are very often exactly the bytes the rebuild needs.
        assert!(is_superseded_archive(
            &format!("Gallery.{EVT}.4.zip"),
            &gallery_prefix(),
            5,
            &[]
        ));
    }

    #[test]
    fn our_own_and_newer_generations_are_never_touched() {
        // `keep_seq` is OUR generation; a NEWER one belongs to a re-release that has already
        // superseded us, and deleting it would let a lagging worker nuke a live keepsake.
        for seq in [5, 6] {
            assert!(
                !is_superseded_archive(
                    &format!("Gallery.{EVT}.{seq}.zip"),
                    &gallery_prefix(),
                    5,
                    &[]
                ),
                "generation {seq} must survive a prune keeping 5"
            );
        }
    }

    #[test]
    fn a_carried_forward_archive_survives_despite_an_older_epoch_in_its_name() {
        // THE dangerous case. A ViewerOnly regeneration (a moderated comment) re-stamps the
        // finished ZIP's row to the new epoch WITHOUT renaming the file, so the SERVED archive
        // legitimately carries an older generation number. Pruning it by filename-epoch would 404
        // the photo download to change nothing in it. The protected set is what stops that, and
        // moving the prune to build-start makes this case reachable far more often.
        let carried = format!("Gallery.{EVT}.4.zip");
        assert!(!is_superseded_archive(
            &carried,
            &gallery_prefix(),
            5,
            std::slice::from_ref(&carried)
        ));
    }

    #[test]
    fn temps_and_staging_dirs_are_out_of_scope_for_the_early_prune() {
        // A superseded worker may still be streaming into these, and at build START it is much
        // more likely to be alive than at finalize time. Only inert FINAL archives are reclaimed
        // here; `prune_stale_export_files` still handles the rest after we win.
        for name in [
            format!("Gallery.{EVT}.4.tmp"),
            format!("viewer_tmp_{EVT}_4"),
            format!("Memories.{EVT}.4.zip"),
        ] {
            assert!(
                !is_superseded_archive(&name, &gallery_prefix(), 5, &[]),
                "{name} must not be reclaimed by the pre-build prune"
            );
        }
    }

    #[test]
    fn another_events_archive_is_never_reclaimed() {
        // All events share one exports volume, so the prefix carries the event id.
        let other = "22222222-2222-2222-2222-222222222222";
        assert!(!is_superseded_archive(
            &format!("Gallery.{other}.4.zip"),
            &gallery_prefix(),
            5,
            &[]
        ));
    }

    #[test]
    fn unrelated_files_are_left_alone() {
        for name in ["Gallery.zip", "notes.txt", "Gallery..4.zip"] {
            assert!(!is_superseded_archive(name, &gallery_prefix(), 5, &[]));
        }
    }

    /// Every `<` is escaped, whatever it is part of.
    ///
    /// PREVENTS the regression to `</` -> `<\\/`, which is named for the one case it handles.
    /// `<!--<script` with no later `-->` drives the HTML tokenizer into
    /// script-data-double-escaped state, where the template's own `</script>` no longer closes
    /// the element — the viewer bundle is swallowed as script data, `__EXPORT_DATA__` is never
    /// assigned, and the keepsake opens blank in every copy the host has already handed out.
    #[test]
    fn no_left_angle_bracket_survives_inlining() {
        for payload in [
            r#"{"caption":"<!--<script"}"#,
            r#"{"caption":"</script><img src=x onerror=alert(1)>"}"#,
            r#"{"caption":"<!--"}"#,
            r#"{"caption":"<script>"}"#,
            r#"{"caption":"a < b"}"#,
        ] {
            let escaped = escape_json_for_script(payload);
            assert!(
                !escaped.contains('<'),
                "a surviving `<` can still steer the tokenizer: {escaped}"
            );
        }
    }

    /// The escape must not change what the viewer READS — it is a transport encoding, not a
    /// sanitiser. A caption is guest-authored text that has to render back exactly.
    #[test]
    fn the_payload_still_decodes_to_the_original_value() {
        // `<` appears only inside JSON string values, never in structural syntax, so a global
        // replace is sound — this is the assertion that says so.
        for caption in [
            "<!--<script",
            "</script><img src=x onerror=alert(1)>",
            "a < b und c > d",
            "ganz normale Bildunterschrift",
            "Herz <3",
        ] {
            let json = serde_json::json!({ "posts": [{ "caption": caption }] }).to_string();
            let escaped = escape_json_for_script(&json);
            let back: serde_json::Value =
                serde_json::from_str(&escaped).expect("the escaped form must still be valid JSON");
            assert_eq!(
                back["posts"][0]["caption"].as_str(),
                Some(caption),
                "the caption must survive the round trip unchanged"
            );
        }
    }

    /// Nothing else in the document is touched.
    #[test]
    fn a_payload_with_no_angle_brackets_is_unchanged() {
        let json = r#"{"posts":[{"caption":"schönes Foto"}]}"#;
        assert_eq!(escape_json_for_script(json), json);
    }

    #[test]
    fn a_lone_armed_job_reserves_for_one_archive() {
        // A ViewerOnly regeneration re-arms only the HTML half — reserving for two would refuse
        // a rebuild that fits perfectly well.
        assert_eq!(required_free_bytes(1_000, 1), 1_100);
    }

    #[test]
    fn two_concurrent_halves_reserve_for_both() {
        // The bug this exists to prevent: each worker independently sees "it fits", and together
        // they don't. Both halves are gallery-sized, so the reservation must be for the pair.
        assert_eq!(required_free_bytes(1_000, 2), 2_200);
    }

    #[test]
    fn a_zero_count_still_reserves_for_one() {
        // Defensive: a racing status transition must never yield a zero requirement, which would
        // wave through an export of any size onto a full disk.
        assert_eq!(required_free_bytes(1_000, 0), 1_100);
    }

    #[test]
    fn an_empty_gallery_needs_nothing() {
        assert_eq!(required_free_bytes(0, 2), 0);
    }

    #[test]
    fn a_pathological_size_saturates_instead_of_wrapping() {
        // u64 overflow would wrap to a TINY requirement and authorise the exact write we're
        // guarding against — the failure mode must be "refuse", never "wrap and allow".
        assert_eq!(required_free_bytes(u64::MAX, 2), u64::MAX);
    }

    /// The completeness gate decides whether a partially-readable gallery still ships. Both
    /// directions are dangerous: too strict and one live delete costs the guests every photo
    /// (a released event cannot be rebuilt without reopening uploads); too loose and a wrong
    /// MEDIA_PATH ships an archive that silently misrepresents the whole party.
    #[test]
    fn a_clean_export_and_a_totally_empty_one_are_unambiguous() {
        // Nothing skipped is always fine, at any size.
        assert!(check_export_completeness("zip", Uuid::nil(), 0, 0, 0).is_ok());
        assert!(check_export_completeness("zip", Uuid::nil(), 100, 100, 0).is_ok());

        // Everything skipped is the misconfiguration case — never publish it, at any size.
        assert!(check_export_completeness("zip", Uuid::nil(), 100, 0, 100).is_err());
        assert!(check_export_completeness("zip", Uuid::nil(), 3, 0, 3).is_err());
        assert!(check_export_completeness("zip", Uuid::nil(), 1, 0, 1).is_err());
    }

    #[test]
    fn a_small_gallery_survives_an_ordinary_live_delete() {
        // These are the regressions a pure 10% fraction caused: on a small gallery a single
        // guest deleting their own photo mid-build is >10%, so the whole keepsake failed.
        assert!(check_export_completeness("zip", Uuid::nil(), 9, 8, 1).is_ok());
        assert!(check_export_completeness("zip", Uuid::nil(), 3, 2, 1).is_ok());
        assert!(check_export_completeness("zip", Uuid::nil(), 4, 2, 2).is_ok());
        // Half a tiny gallery is published too, and logged as materially incomplete. Refusing it
        // would hand the guests nothing at all rather than the three photos that DID survive, and
        // the rebuild that refusal invites reads the same unreadable files again.
        assert!(check_export_completeness("zip", Uuid::nil(), 6, 3, 3).is_ok());
    }

    #[test]
    fn a_materially_incomplete_archive_is_still_published_but_shouted_about() {
        // This used to assert `is_err()` past the 10% tolerance, and that was the wrong trade.
        // The refusal is deterministic across retries — the unreadable files are still unreadable
        // when the host taps "Neu erzeugen", and the gallery is already released so the photos
        // cannot be collected again. So refusing did not buy a better archive later; it converted
        // "90 of 100 photos" into "no keepsake, ever". Past the tolerance we publish and log at
        // `error` with the counts.
        assert!(check_export_completeness("zip", Uuid::nil(), 100, 90, 10).is_ok());
        assert!(check_export_completeness("zip", Uuid::nil(), 100, 89, 11).is_ok());
        assert!(check_export_completeness("zip", Uuid::nil(), 300, 200, 100).is_ok());
    }

    #[test]
    fn an_archive_with_nothing_in_it_is_still_refused() {
        // The one fatal case, and the only one the host can actually act on: a wrong MEDIA_PATH
        // produced a few-hundred-byte ZIP with zero photos that passed every automated check and
        // was advertised as ready. That IS recoverable — fix the path, rebuild — so refusing to
        // publish it is the correct answer, unlike the partial case above.
        assert!(check_export_completeness("zip", Uuid::nil(), 100, 0, 100).is_err());
        // ...but an event that genuinely has no media is not an error.
        assert!(check_export_completeness("zip", Uuid::nil(), 0, 0, 0).is_ok());
    }
}