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");

// ── 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);

// 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;
        }
        if let Err(e) =
            run_zip_export(event_id, epoch, &pool, &media_path, &export_path, &sse_tx).await
        {
            tracing::error!("ZIP export failed for event {event_id} @ epoch {epoch}: {e:#}");
            mark_failed(&pool, event_id, "zip", epoch, &e.to_string()).await;
        }
        maybe_broadcast_complete(&pool, event_id, &sse_tx).await;
    });

    tokio::spawn(async move {
        if !delay.is_zero() {
            tokio::time::sleep(delay).await;
        }
        if let Err(e) = run_html_export(
            event_id,
            epoch,
            &event_name2,
            comments_enabled,
            &pool2,
            &media_path2,
            &export_path2,
            &sse_tx2,
        )
        .await
        {
            tracing::error!("HTML export failed for event {event_id} @ epoch {epoch}: {e:#}");
            mark_failed(&pool2, event_id, "html", epoch, &e.to_string()).await;
        }
        maybe_broadcast_complete(&pool2, event_id, &sse_tx2).await;
    });
}

// ── 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(());
    }

    // Reclaim BEFORE measuring: the superseded archive is already unreachable, and the space it
    // holds is very often exactly the space this rebuild needs.
    prune_superseded_archives(pool, export_path, "Gallery", event_id, epoch).await;
    // 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(pool, event_id, export_path).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;
    }
    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);

    {
        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 = ZipEntryBuilder::new(entry_name.into(), 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()
                    );
                    continue;
                }
            };

            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?;
    }

    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"
        );
        return Ok(());
    }

    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: reclaim the superseded generation first, then refuse at the door rather
    // than ENOSPC mid-write.
    prune_superseded_archives(pool, export_path, "Memories", event_id, epoch).await;
    ensure_export_space(pool, event_id, export_path).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;
    }
    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.
    let mut media_manifest: Vec<(String, MediaSource)> = Vec::new();

    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()
                );
                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}");

            // Video thumbnail via ffmpeg
            let thumb_path = media_tmp.join(&thumb);
            let ffmpeg_result = tokio::process::Command::new("ffmpeg")
                .args([
                    "-i",
                    src.to_str().unwrap_or_default(),
                    "-vframes",
                    "1",
                    "-ss",
                    "00:00:01",
                    "-vf",
                    "scale=400:-1",
                    "-y",
                    thumb_path.to_str().unwrap_or_default(),
                ])
                .output()
                .await;

            match ffmpeg_result {
                Ok(output) if output.status.success() => {}
                _ => {
                    tracing::warn!(
                        "ffmpeg thumbnail failed for upload {}, skipping thumb",
                        row.id
                    );
                    // Missing thumb entry — viewer handles missing thumbs gracefully.
                }
            }

            // 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).
            (thumb, 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();

            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?;

            if let Err(e) = thumb_result {
                tracing::warn!("thumbnail generation failed for upload {}: {e:#}", row.id);
            }

            // 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();

                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?;

                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, full, full_source)
        };

        // Register this post's two media entries. Thumbnails always come from temp
        // (they're freshly generated); the full variant's source was decided above.
        media_manifest.push((
            thumb_name.clone(),
            MediaSource::Temp(media_tmp.join(&thumb_name)),
        ));
        media_manifest.push((full_name.clone(), full_source));

        // 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()
                },
                thumb: format!("media/{thumb_name}"),
                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 = ZipEntryBuilder::new("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 = ZipEntryBuilder::new("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;

        for (name, source) 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()
                    );
                    continue;
                }
            };

            let entry_name = format!("media/{name}");
            let builder = ZipEntryBuilder::new(entry_name.into(), 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;
            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?;
    }

    // 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"
        );
        return Ok(());
    }

    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>(
        "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
         WHERE u.event_id = $1 AND u.deleted_at IS NULL
           AND usr.uploads_hidden = FALSE AND usr.is_banned = FALSE
         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.
async fn finalize_job(
    pool: &PgPool,
    event_id: Uuid,
    export_type: &str,
    epoch: i64,
    file_path: &str,
) -> bool {
    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
    .map(|r| r.rows_affected() > 0)
    .unwrap_or(false)
}

/// 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 BEFORE this generation starts writing its own.
///
/// Peak disk usage used to be two full generations, because the only prune ran after the new archive
/// was written, renamed and finalised. That ordering reads as durability ("don't delete the good
/// keepsake before the replacement is safe") but it buys nothing: readiness is derived from
/// `job.epoch = event.export_epoch AND status = 'done'`, so the moment `invalidate_and_arm` bumps
/// the epoch the old archive is ALREADY unreachable — `GET /export/zip` 404s whether the file is on
/// disk or not. Keeping it only reserves gigabytes for a download nobody can perform, and for
/// `Affects::Both` (a takedown) it is content someone has explicitly asked to have removed. So a
/// rebuild reclaims first and peaks at one generation.
///
/// 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.
///
/// Matches [`query_uploads`]' visibility filter exactly, so hidden/banned uploads aren't counted.
pub async fn estimate_export_bytes(pool: &PgPool, event_id: Uuid) -> Result<u64> {
    let (bytes,): (i64,) = sqlx::query_as(
        "SELECT COALESCE(SUM(u.original_size_bytes), 0)::bigint
           FROM upload u
           JOIN \"user\" usr ON usr.id = u.user_id
          WHERE u.event_id = $1 AND u.deleted_at IS NULL
            AND usr.uploads_hidden = FALSE AND usr.is_banned = FALSE",
    )
    .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.
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(());
    };

    if free < needed {
        let gb = |b: u64| b as f64 / 1_000_000_000.0;
        tracing::error!(
            needed,
            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, frei sind {:.1} GB. \
             Bitte Speicher freigeben und das Keepsake anschließend neu erstellen.",
            gb(needed),
            gb(free)
        );
    }
    Ok(())
}

/// 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.)
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 `</` so a caption containing `</script>` can't break out of the tag.
            let safe = data_json.replace("</", "<\\/");
            // 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 = ZipEntryBuilder::new(path.into(), 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 = ZipEntryBuilder::new(path.into(), 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_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, &[]));
        }
    }

    #[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);
    }
}