Files
EventSnap/backend/src/handlers/admin.rs
fabi 281eb3bec7 fix(export): refuse an export that cannot fit, and stop peaking at two generations
Nothing in export.rs ever asked whether the keepsake would fit. Both archives write
their media `Compression::Stored`, so each is essentially a byte-for-byte second copy
of the originals -- Gallery.zip always, and Memories.zip for every video and every
image at or under 5 MB. On the documented CX33 (80 GB, all three volumes on one
filesystem) the upload quota's fixed point leaves ~40 GB free, and a release spawns
BOTH halves concurrently against it.

The failure is not "the export failed", it is "the deliverable is stuck":

  1. ENOSPC lands partway through a multi-GB write.
  2. The epoch has already moved, so the job row is `failed` at the CURRENT
     generation and readiness (epoch = event.export_epoch AND status = 'done') is
     false -- GET /export/zip 404s.
  3. The last good archive sits on disk, unreferenced and unreachable.
  4. POST /host/export/rebuild, the only escape, re-arms the same doomed write.

Three changes.

Reclaim before building. `prune_stale_export_files` ran only after the new archive
was written, renamed and finalised. That reads as durability but buys nothing: the
moment `invalidate_and_arm` bumps the epoch the old archive is ALREADY unreachable,
so keeping it reserves gigabytes for a download nobody can perform -- and for a
takedown it is content someone explicitly asked to have removed. Peak usage is now
one generation. Narrower than the post-finalize prune on purpose: final archives
only, never a `.tmp` or a `viewer_tmp_` dir, since a superseded worker can still be
streaming into those and at build START is far more likely to be alive.

Preflight the space. SUM(original_size_bytes) over exactly `query_uploads`'
visibility filter, +10% for ZIP overhead, multiplied by the number of armed jobs --
without that multiplier each of the two concurrent halves independently sees "it
fits" and together they don't. Runs AFTER claim_job, not before as reported: bailing
before the claim leaves the row `pending` with no worker and no error, the
spinner-forever state `mark_failed`'s status guard exists to prevent. Fails open when
the mount can't be read, exactly as the upload quota does.

Show the host the reason. /export/status returned {status, progress_pct} and nothing
else, so the host dashboard could only render "fehlgeschlagen" next to the retry
button. The message was written to the row and surfaced solely in the ADMIN job list
-- a different screen, possibly a different person. It now travels with the status,
and only on a failure, so a message left on a since-succeeded row can't appear beside
a green "ist bereit".

Tests: 10 unit (the u128 clamp caught a real bug in the first draft -- saturating_mul
then /100 turns an overflow into a number ~100x too small, the one direction that
authorises the write being guarded against; the carried-forward archive must survive
its own older epoch in the filename), 4 DB-backed (the estimate is asserted against
the row set the archive actually contains, not against a restatement of the WHERE
clause, so the two queries cannot drift), 3 e2e over the four-hop plumbing.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-07-29 19:38:26 +02:00

525 lines
21 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
use std::collections::HashMap;
use std::time::Duration;
use axum::Json;
use axum::extract::{Query, State};
use axum::http::StatusCode;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::auth::middleware::RequireAdmin;
use crate::error::AppError;
use crate::services::config;
use crate::state::AppState;
// ── DTOs ─────────────────────────────────────────────────────────────────────
#[derive(Serialize)]
pub struct StatsDto {
pub user_count: i64,
pub upload_count: i64,
pub comment_count: i64,
pub disk_total_bytes: u64,
pub disk_used_bytes: u64,
pub disk_free_bytes: u64,
}
#[derive(Serialize, sqlx::FromRow)]
pub struct ExportJobDto {
pub id: uuid::Uuid,
pub r#type: String,
pub status: String,
pub progress_pct: i16,
pub error_message: Option<String>,
pub created_at: chrono::DateTime<chrono::Utc>,
pub completed_at: Option<chrono::DateTime<chrono::Utc>>,
}
// ── Handlers ─────────────────────────────────────────────────────────────────
pub async fn get_stats(
State(state): State<AppState>,
RequireAdmin(_auth): RequireAdmin,
) -> Result<Json<StatsDto>, AppError> {
let event = crate::models::event::Event::find_by_slug(&state.pool, &state.config.event_slug)
.await?
.ok_or_else(|| AppError::NotFound("Event nicht gefunden.".into()))?;
let (user_count,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM \"user\" WHERE event_id = $1")
.bind(event.id)
.fetch_one(&state.pool)
.await?;
let (upload_count,): (i64,) =
sqlx::query_as("SELECT COUNT(*) FROM upload WHERE event_id = $1 AND deleted_at IS NULL")
.bind(event.id)
.fetch_one(&state.pool)
.await?;
let (comment_count,): (i64,) = sqlx::query_as(
"SELECT COUNT(*) FROM comment c
JOIN upload u ON u.id = c.upload_id
WHERE u.event_id = $1 AND c.deleted_at IS NULL",
)
.bind(event.id)
.fetch_one(&state.pool)
.await?;
// Disk usage from the shared cache (unknown mount → zeros, same as before).
let (disk_total, disk_free) = state
.disk_cache
.snapshot(&state.config.media_path)
.map(|d| (d.total, d.free))
.unwrap_or((0, 0));
let disk_used = disk_total.saturating_sub(disk_free);
Ok(Json(StatsDto {
user_count,
upload_count,
comment_count,
disk_total_bytes: disk_total,
disk_used_bytes: disk_used,
disk_free_bytes: disk_free,
}))
}
pub async fn get_config(
State(state): State<AppState>,
RequireAdmin(_auth): RequireAdmin,
) -> Result<Json<HashMap<String, String>>, AppError> {
let rows: Vec<(String, String)> = sqlx::query_as("SELECT key, value FROM config ORDER BY key")
.fetch_all(&state.pool)
.await?;
Ok(Json(rows.into_iter().collect()))
}
/// Documents the wire shape of `PATCH /admin/config` (a flat `{key: value}` object).
/// `patch_config` extracts the `HashMap` directly rather than going through this newtype, so it is
/// never constructed in Rust — it stays as the serde-derived description of the request body.
#[allow(dead_code)]
#[derive(Deserialize)]
pub struct PatchConfigRequest(pub HashMap<String, String>);
pub async fn patch_config(
State(state): State<AppState>,
RequireAdmin(_auth): RequireAdmin,
Json(body): Json<HashMap<String, String>>,
) -> Result<StatusCode, AppError> {
// Numeric keys validated as f64; boolean keys validated as truthy strings; the
// privacy note is free text. Splitting these explicitly is verbose but makes the
// failure mode for typos obvious (`Unbekannter Schlüssel: ...`).
// (key, integer_only, min, max). Ranges reject values that `parse::<f64>` would
// accept but that silently revert to the hardcoded default at read time
// (get_usize/get_i64 can't parse negatives/NaN/fractionals). `compression_concurrency`
// is intentionally absent — it's read once at boot, so a live edit was a no-op.
const NUMERIC_SPECS: &[(&str, bool, f64, f64)] = &[
("max_image_size_mb", true, 1.0, 1024.0),
("max_video_size_mb", true, 1.0, 10240.0),
("upload_rate_per_hour", true, 1.0, 100_000.0),
("feed_rate_per_min", true, 1.0, 100_000.0),
("export_rate_per_day", true, 1.0, 100_000.0),
// Loose per-IP ceiling on /join. The real anti-spam bucket is per (ip, name); this
// only bounds raw volume from one source, so it must stay well above the size of a
// party arriving at once (see migration 017).
("join_ip_rate_per_min", true, 1.0, 100_000.0),
// Same shape for /recover: the per-(ip, name) bucket is the anti-guessing control,
// this only bounds a name-cycling flood in front of a cost-12 bcrypt (migration 019).
("recover_ip_rate_per_min", true, 1.0, 100_000.0),
("quota_tolerance", false, 0.0, 1.0),
("estimated_guest_count", true, 1.0, 1_000_000.0),
];
const BOOL_KEYS: &[&str] = &[
"rate_limits_enabled",
"upload_rate_enabled",
"feed_rate_enabled",
"export_rate_enabled",
"join_rate_enabled",
// These two per-area rate toggles are HONOURED by their handlers (auth/handlers.rs reads
// `admin_login_rate_enabled` and `recover_rate_enabled`, both defaulting true) but were
// missing from this allowlist — so the switch existed in code and could never be flipped.
"admin_login_rate_enabled",
"recover_rate_enabled",
"quota_enabled",
"storage_quota_enabled",
"upload_count_quota_enabled",
];
const TEXT_KEYS: &[&str] = &[
"privacy_note",
"theme_preset",
"theme_primary",
"theme_accent",
];
const PRIVACY_NOTE_MAX_LEN: usize = 16 * 1024; // 16 KiB free text is plenty
// Preset ids the frontend knows how to render (mirror of PRESETS in
// frontend/src/lib/theme/palette.ts). "custom" means "use the theme_primary/accent
// seeds verbatim". Kept in sync by hand — a new preset must be added in both places.
const THEME_PRESETS: &[&str] = &[
"champagne-gold",
"rose",
"sage",
"dusk-blue",
"classic-silver",
"custom",
];
let mut privacy_note_changed = false;
let mut theme_changed = false;
// Validate every key first so a bad value in the batch can't leave a partial
// update behind — validation must fully precede any write.
for (key, value) in &body {
let key_str = key.as_str();
if let Some(&(_, integer_only, min, max)) =
NUMERIC_SPECS.iter().find(|(k, ..)| *k == key_str)
{
let n = value.trim().parse::<f64>().ok().filter(|n| n.is_finite());
let n = match n {
Some(n) => n,
None => {
return Err(AppError::BadRequest(format!(
"Ungültiger Wert für {key}: muss eine Zahl sein."
)));
}
};
if integer_only && n.fract() != 0.0 {
return Err(AppError::BadRequest(format!(
"Ungültiger Wert für {key}: muss eine ganze Zahl sein."
)));
}
if n < min || n > max {
return Err(AppError::BadRequest(format!(
"Wert für {key} liegt außerhalb des zulässigen Bereichs ({min}{max})."
)));
}
} else if BOOL_KEYS.contains(&key_str) {
match value.trim().to_ascii_lowercase().as_str() {
"true" | "false" | "1" | "0" | "yes" | "no" | "on" | "off" => {}
_ => {
return Err(AppError::BadRequest(format!(
"Ungültiger Wert für {key}: muss true oder false sein."
)));
}
}
} else if TEXT_KEYS.contains(&key_str) {
// Count characters, not bytes — the message says "Zeichen" and a
// multi-byte grapheme shouldn't count against the limit multiple times.
if value.chars().count() > PRIVACY_NOTE_MAX_LEN {
return Err(AppError::BadRequest(format!(
"Wert für {key} ist zu lang (max. {PRIVACY_NOTE_MAX_LEN} Zeichen)."
)));
}
match key_str {
"privacy_note" => privacy_note_changed = true,
"theme_preset" => {
if !THEME_PRESETS.contains(&value.trim()) {
return Err(AppError::BadRequest(format!("Ungültiges Theme: {value}.")));
}
theme_changed = true;
}
"theme_primary" | "theme_accent" => {
if !is_hex_color(value.trim()) {
return Err(AppError::BadRequest(format!(
"Ungültige Farbe für {key}: muss #rrggbb sein."
)));
}
theme_changed = true;
}
_ => {}
}
} else {
return Err(AppError::BadRequest(format!(
"Unbekannter Konfigurationsschlüssel: {key}"
)));
}
}
// Apply all writes in one transaction — the batch is all-or-nothing.
let mut tx = state.pool.begin().await?;
for (key, value) in &body {
sqlx::query(
"INSERT INTO config (key, value, updated_at) VALUES ($1, $2, NOW())
ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = NOW()",
)
.bind(key)
.bind(value)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
// The config cache must reflect this write on the very next read (tests PATCH then
// immediately assert the new value takes effect). Invalidate synchronously here —
// the TTL is only a backstop and must not be relied on for correctness.
state.config_cache.invalidate();
// Notify all clients that a publicly-readable config value changed so their stores
// (e.g. the privacy note in My Account) refresh without a manual reload.
if privacy_note_changed || theme_changed {
let mut keys: Vec<&str> = Vec::new();
if privacy_note_changed {
keys.push("privacy_note");
}
if theme_changed {
keys.push("theme");
}
let _ = state.sse_tx.send(crate::state::SseEvent::new(
"event-updated",
serde_json::json!({ "keys": keys }).to_string(),
));
}
Ok(StatusCode::NO_CONTENT)
}
/// A strict `#rrggbb` hex-colour check (6 hex digits, leading `#`). Deliberately not
/// accepting shorthand/`#rgba` so the value is safe to drop straight into CSS.
fn is_hex_color(s: &str) -> bool {
let bytes = s.as_bytes();
bytes.len() == 7 && bytes[0] == b'#' && bytes[1..].iter().all(|b| b.is_ascii_hexdigit())
}
pub async fn get_export_jobs(
State(state): State<AppState>,
RequireAdmin(_auth): RequireAdmin,
) -> Result<Json<Vec<ExportJobDto>>, AppError> {
let event = crate::models::event::Event::find_by_slug(&state.pool, &state.config.event_slug)
.await?
.ok_or_else(|| AppError::NotFound("Event nicht gefunden.".into()))?;
let jobs = sqlx::query_as::<_, ExportJobDto>(
"SELECT id, type::text, status::text, progress_pct, error_message, created_at, completed_at
FROM export_job
WHERE event_id = $1
ORDER BY created_at DESC",
)
.bind(event.id)
.fetch_all(&state.pool)
.await?;
Ok(Json(jobs))
}
// ── Export download endpoints (authenticated guests) ─────────────────────────
#[derive(Deserialize)]
pub struct DownloadQuery {
pub ticket: String,
}
/// Mint a short-lived ticket for a browser-driven export download. The download
/// is a top-level navigation so the multi-GB ZIP streams straight to disk instead
/// of being buffered in memory by `fetch()` + `blob()` — but a navigation can't
/// carry an `Authorization` header, so the client exchanges its Bearer token for
/// an opaque ticket here, then hits `/export/zip?ticket=...`. Reuses the same
/// single-use, 30s-TTL store as the SSE stream.
pub async fn export_ticket(
State(state): State<AppState>,
auth: crate::auth::middleware::AuthUser,
) -> Json<serde_json::Value> {
// NOTE: intentionally NOT gated on `is_banned`. A banned user keeps *read* access
// by design (USER_JOURNEYS §10.3, FEATURES: "Can still download the export once
// released — Spec design choice"). The export is read-only, so it stays available
// to them, consistent with the read-only-ban model.
let ticket = state.sse_tickets.issue(auth.token_hash);
Json(serde_json::json!({ "ticket": ticket }))
}
/// Validate a download ticket (single-use) and confirm its session still exists.
/// Resolve a single-use download ticket to the user who minted it. The caller needs the
/// id to key the export rate limit per-user (see `enforce_export_rate`).
async fn authenticate_download_ticket(state: &AppState, ticket: &str) -> Result<Uuid, AppError> {
let token_hash = state
.sse_tickets
.consume(ticket)
.ok_or_else(|| AppError::Unauthorized("Ticket ungültig oder abgelaufen.".into()))?;
let session = crate::models::session::Session::find_by_token_hash(&state.pool, &token_hash)
.await
.map_err(|e| AppError::Internal(e.into()))?
.ok_or_else(|| AppError::Unauthorized("Sitzung nicht gefunden.".into()))?;
Ok(session.user_id)
}
pub async fn download_zip(
State(state): State<AppState>,
Query(q): Query<DownloadQuery>,
) -> Result<axum::response::Response, AppError> {
let user_id = authenticate_download_ticket(&state, &q.ticket).await?;
enforce_export_rate(&state, user_id).await?;
let path =
resolve_export_file(&state, "zip", "Der ZIP-Export ist noch nicht verfügbar.").await?;
serve_file(path, "Gallery.zip", "application/zip").await
}
/// Resolve the on-disk path of the CURRENT export generation — readiness check and path lookup in
/// ONE read, through the `export_current` view (migration 014).
///
/// This used to be two statements: the caller checked `event.export_zip_ready`, then this fetched
/// `export_job.file_path`. A reopen landing between them served a keepsake that should have 404'd —
/// the same stale-keepsake class, leaking through the read path, and it existed only because
/// readiness was a stored copy rather than a derivation. `export_current` gives both facts from one
/// consistent snapshot: it yields a row only when the event is released AND the job is `done` at the
/// event's current epoch, so "is it ready" and "which file" can no longer disagree.
async fn resolve_export_file(
state: &AppState,
export_type: &str,
not_ready_msg: &str,
) -> Result<std::path::PathBuf, AppError> {
let file_path: Option<(Option<String>,)> = sqlx::query_as(
"SELECT c.file_path FROM export_current c
JOIN event e ON e.id = c.event_id
WHERE e.slug = $1 AND c.type = $2::export_type AND c.status = 'done'",
)
.bind(&state.config.event_slug)
.bind(export_type)
.fetch_optional(&state.pool)
.await
.map_err(|e| AppError::Internal(e.into()))?;
let Some((Some(rel),)) = file_path else {
return Err(AppError::NotFound(not_ready_msg.into()));
};
// `file_path` is stored as `exports/<name>`; the base dir is already `export_path`, so
// join only the file name (defends against any absolute/`..` content too).
let name = std::path::Path::new(&rel)
.file_name()
.ok_or_else(|| AppError::NotFound("Exportdatei nicht gefunden.".into()))?;
let path = state.config.export_path.join(name);
if !path.exists() {
return Err(AppError::NotFound("Exportdatei nicht gefunden.".into()));
}
Ok(path)
}
pub async fn download_html(
State(state): State<AppState>,
Query(q): Query<DownloadQuery>,
) -> Result<axum::response::Response, AppError> {
let user_id = authenticate_download_ticket(&state, &q.ticket).await?;
enforce_export_rate(&state, user_id).await?;
let path =
resolve_export_file(&state, "html", "Der HTML-Export ist noch nicht verfügbar.").await?;
serve_file(path, "Memories.zip", "application/zip").await
}
async fn serve_file(
path: std::path::PathBuf,
filename: &str,
content_type: &str,
) -> Result<axum::response::Response, AppError> {
use axum::body::Body;
use axum::http::{Response, StatusCode, header};
use tokio_util::io::ReaderStream;
let file = tokio::fs::File::open(&path)
.await
.map_err(|e| AppError::Internal(e.into()))?;
let metadata = file
.metadata()
.await
.map_err(|e| AppError::Internal(e.into()))?;
let stream = ReaderStream::new(file);
let disposition = format!("attachment; filename=\"{filename}\"");
let response = Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, content_type)
.header(header::CONTENT_DISPOSITION, disposition)
.header(header::CONTENT_LENGTH, metadata.len())
.body(Body::from_stream(stream))
.map_err(|e| AppError::Internal(e.into()))?;
Ok(response)
}
/// Also expose export status to all authenticated users (guests need it for the export page)
pub async fn export_status(
State(state): State<AppState>,
_auth: crate::auth::middleware::AuthUser,
) -> Result<Json<serde_json::Value>, AppError> {
let event = crate::models::event::Event::find_by_slug(&state.pool, &state.config.event_slug)
.await?
.ok_or_else(|| AppError::NotFound("Event nicht gefunden.".into()))?;
let released = event.export_released_at.is_some();
// ONE statement: the epoch comparison happens inside the query, against a single snapshot.
// Binding the epoch read by a previous statement would let a release/regeneration commit in
// between and yield `{released: true, zip: locked, html: locked}` — a state that never existed.
//
// Only jobs at the CURRENT epoch are reported. A row left behind by a retired generation (a
// worker superseded mid-run) is meaningless — surfacing its frozen `running`/77% would show a
// progress bar that never moves for a keepsake nobody is building. It reads as "locked" (no
// current job), which is exactly what it is.
// `error_message` is carried here, not just on the admin dashboard's job list. The host is the
// one who releases the keepsake and the one who owns the "Erneut versuchen" button, but this
// endpoint used to hand them a bare `failed` — so a fully actionable reason (notably the disk
// preflight's "needs X GB, Y GB free") was written to the row and then shown to nobody who
// could act on it. An admin-only diagnostic is not a diagnostic for the person on the spot.
let jobs: Vec<(String, String, i16, Option<String>)> = sqlx::query_as(
"SELECT j.type::text, j.status::text, j.progress_pct, j.error_message
FROM export_job j
JOIN event e ON e.id = j.event_id
WHERE e.id = $1 AND j.epoch = e.export_epoch",
)
.bind(event.id)
.fetch_all(&state.pool)
.await?;
let job_status = |type_name: &str| {
jobs.iter()
.find(|(t, _, _, _)| t == type_name)
.map(|(_, status, pct, err)| {
serde_json::json!({
"status": status,
"progress_pct": pct,
// Only on a failure. A stale message left on a row that has since been re-armed
// would otherwise show an error next to a running progress bar.
"error_message": if status == "failed" { err.clone() } else { None },
})
})
.unwrap_or_else(|| {
serde_json::json!({
"status": "locked", "progress_pct": 0, "error_message": null,
})
})
};
Ok(Json(serde_json::json!({
"released": released,
"zip": job_status("zip"),
"html": job_status("html"),
})))
}
/// Centralised guard for the export rate limit. Same pattern as upload/feed: master
/// switch + per-endpoint switch + numeric value, all stored in `config` and read on
/// each request.
async fn enforce_export_rate(state: &AppState, user_id: Uuid) -> Result<(), AppError> {
let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await;
let export_rate_on = config::get_bool(&state.config_cache, "export_rate_enabled", true).await;
if !(rate_limits_on && export_rate_on) {
return Ok(());
}
let limit = config::get_usize(&state.config_cache, "export_rate_per_day", 3).await;
// Keyed per-user. This was the worst of the IP-keyed limiters: 3 downloads per DAY
// shared across every guest behind the venue's public IP, so the fourth person to
// fetch their keepsake was locked out until the next day.
if let Err(retry_after_secs) = state.rate_limiter.check_with_retry(
format!("export:{user_id}"),
limit,
Duration::from_secs(86400),
) {
return Err(AppError::TooManyRequests(
"Zu viele Anfragen. Bitte warte kurz und versuche es erneut.".into(),
Some(retry_after_secs),
));
}
Ok(())
}