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::{copy as fcopy, AllowStdIo}; use include_dir::{include_dir, Dir}; use serde::Serialize; use sqlx::PgPool; use tokio::sync::broadcast; use tokio::io::AsyncWriteExt; 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, uploader_name: String, like_count: i64, created_at: DateTime, } #[derive(sqlx::FromRow)] struct ExportCommentRow { upload_id: Uuid, uploader_name: String, body: String, created_at: DateTime, } // ── Viewer JSON structs (serialised to data.json) ─────────────────────────── #[derive(Serialize)] struct ViewerData { event: ViewerEvent, posts: Vec, } #[derive(Serialize)] struct ViewerEvent { name: String, exported_at: String, } #[derive(Serialize)] struct ViewerPost { id: String, uploader: String, caption: String, tags: Vec, timestamp: String, likes: i64, comments: Vec, 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> { 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, sse_tx: broadcast::Sender, ) { 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, 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)> = 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, delay: Duration, pool: PgPool, media_path: PathBuf, export_path: PathBuf, sse_tx: broadcast::Sender, ) { 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, &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, ) -> 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(()); } // 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::().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, ) -> 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, } } } async fn run_html_export( event_id: Uuid, epoch: i64, event_name: &str, pool: &PgPool, media_path: &Path, export_path: &Path, sse_tx: &broadcast::Sender, ) -> 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(()); } let res = run_html_export_inner(epoch, event_id, event_name, 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) } async fn run_html_export_inner( epoch: i64, event_id: Uuid, event_name: &str, pool: &PgPool, media_path: &Path, export_path: &Path, sse_tx: &broadcast::Sender, ) -> 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 = 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 genuinely can: the // compression worker hard-deletes an original when its transcode fails, and it can still be // running when the gallery is released. 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<()> { let img = image::open(&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<()> { let img = image::open(&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 = 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 = 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(), }, posts: viewer_posts, }; let data_json = serde_json::to_string_pretty(&viewer_data).context("failed to serialize data.json")?; 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 embedded viewer assets (index.html, _app/*, etc.) write_dir_to_zip(&VIEWER_DIR, &mut zip).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 (the // compression worker deletes originals on transcode failure) 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> { 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> { 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> { 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 { 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 `` (e.g. /// `Gallery..zip`, `Gallery.zip..tmp`, `viewer_tmp__`). 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 { name.strip_prefix(prefix)?.strip_suffix(suffix)?.parse::().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 (`..zip`), its temp (`.zip..tmp`), and — for html — /// the `viewer_tmp__` 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, ) { // Files still referenced by a live (current-epoch) job row are OFF LIMITS regardless of the // epoch in their name. A ViewerOnly regeneration carries the finished ZIP forward by re-stamping // its row to the new epoch WITHOUT renaming the file — so `Gallery...zip` is still // the served archive, and deleting it by filename-epoch would 404 the download. let protected: Vec = 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(); // 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.. / Memories.. 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, ) { 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(), }); } } } /// Recursively write all files from an embedded `include_dir::Dir` into a ZIP. async fn write_dir_to_zip( dir: &include_dir::Dir<'_>, zip: &mut ZipFileWriter, ) -> Result<()> { for file in dir.files() { let path = file.path().to_string_lossy().to_string(); let contents = file.contents(); 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(contents)); fcopy(&mut cursor, &mut entry).await?; entry.close().await?; } for sub_dir in dir.dirs() { Box::pin(write_dir_to_zip(sub_dir, zip)).await?; } Ok(()) } 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";