use std::time::Duration;

use axum::Json;
use axum::extract::{Multipart, Path, State};
use axum::http::StatusCode;
use chrono::{DateTime, Utc};
use serde::Deserialize;
use uuid::Uuid;

use crate::auth::middleware::AuthUser;
use crate::error::AppError;
use crate::models::hashtag::{self, Hashtag};
use crate::models::upload::{Upload, UploadDto};
use crate::models::user::User;
use crate::services::config;
use crate::state::AppState;

const MAX_CAPTION_LENGTH: usize = 2000;

/// Byte ceiling for the caption field, enforced WHILE reading it.
///
/// `Field::text()` buffers the entire field before returning, and this is the one route whose
/// `DefaultBodyLimit` is raised to 576 MiB (main.rs) — so `caption=<576 MiB of text>` allocated
/// 576 MiB of heap per concurrent request inside a 1 GiB container, and the
/// `MAX_CAPTION_LENGTH` check only ran afterwards, on a string that had already been built.
/// 4 bytes per code point is the worst case for UTF-8, so this can never reject a caption the
/// character limit would have accepted.
const MAX_CAPTION_BYTES: usize = MAX_CAPTION_LENGTH * 4;

/// Byte ceiling for the raw hashtag CSV. Generous next to what the tag caps below allow.
const MAX_HASHTAGS_BYTES: usize = 4 * 1024;

/// Hashtags stored per upload. The CSV was never length-checked at all and was split into an
/// unbounded `Vec`, then upserted TAG BY TAG inside the commit transaction — which holds a
/// `FOR SHARE` lock on the event row, so one request could stall every other upload behind
/// tens of thousands of round trips.
const MAX_HASHTAGS_PER_UPLOAD: usize = 30;
/// Characters per stored tag. `extract_hashtags` already self-bounds at 40; this covers the CSV
/// path, which had no bound of its own.
const MAX_HASHTAG_LENGTH: usize = 50;

/// Read a multipart text field, refusing it the moment it exceeds `max_bytes`.
///
/// The point is to fail DURING the read rather than after it — `Field::text()` cannot, because
/// it has already allocated the whole thing by the time it returns.
async fn read_text_field_bounded(
    mut field: axum::extract::multipart::Field<'_>,
    max_bytes: usize,
) -> Result<String, AppError> {
    let mut buf: Vec<u8> = Vec::new();
    while let Some(chunk) = field
        .chunk()
        .await
        .map_err(|e| AppError::BadRequest(e.to_string()))?
    {
        if buf.len() + chunk.len() > max_bytes {
            return Err(AppError::BadRequest("Eingabe ist zu lang.".to_string()));
        }
        buf.extend_from_slice(&chunk);
    }
    String::from_utf8(buf).map_err(|_| AppError::BadRequest("Ungültige Zeichenkodierung.".into()))
}

/// Normalise, dedupe and CAP the tags for one upload.
///
/// Extracted as a pure function so the caps are testable without standing up multipart, and
/// shared by the upload and edit paths — which previously disagreed: upload lowercased and
/// stripped `#`, while edit upserted raw strings, so `#Party` via edit and `party` via upload
/// became two different hashtag rows.
///
/// Truncates rather than rejecting. `extract_hashtags` legitimately derives tags from a
/// 2000-character caption, and 400-ing a guest for writing an enthusiastic caption would be a
/// worse outcome than silently keeping the first 30.
fn normalize_tags(caption_tags: Vec<String>, csv: Option<&str>) -> Vec<String> {
    let mut tags = caption_tags;
    if let Some(csv) = csv {
        for tag in csv.split(',') {
            let t = tag.trim().trim_start_matches('#').to_lowercase();
            if !t.is_empty() {
                tags.push(t);
            }
        }
    }
    tags.sort();
    tags.dedup();
    tags.retain(|t| t.chars().count() <= MAX_HASHTAG_LENGTH);
    tags.truncate(MAX_HASHTAGS_PER_UPLOAD);
    tags
}

/// Owns the bytes an in-flight upload has written to disk, and deletes them unless the
/// request reaches the point where a database row takes ownership.
///
/// Reclaim used to be a dozen explicit `remove_file` calls on the handler's return paths.
/// That covers every way the handler can FINISH, and none of the ways it can simply STOP:
/// when a client disconnects mid-body — a phone leaving wifi, iOS killing a backgrounded
/// PWA, the user hitting back — axum drops the handler future at a `.await` inside
/// `field.chunk()`, and no return path runs at all. The partial file then survives forever:
/// it has no upload row, so `cleanup_deleted_media` (which is row-driven) can never see it,
/// and no sweeper existed for the originals directory. Those bytes are also invisible to the
/// quota while still consuming the free disk that `quota_limit_bytes` divides among guests.
///
/// A drop guard is the only construct that survives cancellation, because dropping the future
/// is exactly what runs it.
struct TempFileGuard {
    /// `None` once disarmed — a row now owns these bytes.
    path: Option<std::path::PathBuf>,
}

impl TempFileGuard {
    fn new(path: std::path::PathBuf) -> Self {
        Self { path: Some(path) }
    }

    /// Follow the bytes to their new location after a rename.
    ///
    /// NOT `disarm`. Between the rename and the commit the file exists under its FINAL name
    /// with still no row pointing at it, so that window needs guarding just as much as the
    /// `.tmp` did — arguably more, since a leftover final-named original looks legitimate.
    fn retarget(&mut self, path: std::path::PathBuf) {
        self.path = Some(path);
    }

    /// Hand ownership to the committed row. Only correct after `tx.commit()` succeeds.
    fn disarm(&mut self) {
        self.path = None;
    }
}

impl Drop for TempFileGuard {
    fn drop(&mut self) {
        let Some(path) = self.path.take() else {
            return;
        };
        // std::fs, not tokio::fs: `Drop` cannot await, and a runtime-dependent unlink is not
        // guaranteed a live runtime here (shutdown drops in-flight tasks).
        match std::fs::remove_file(&path) {
            Ok(()) => tracing::debug!(path = %path.display(), "reclaimed an abandoned upload"),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => {
                tracing::warn!(error = ?e, path = %path.display(), "failed to reclaim an abandoned upload")
            }
        }
    }
}

/// Allowlist of accepted media types, keyed by the MIME that `infer` derives from
/// the file's magic bytes. The detected MIME (not the client-declared one) is what
/// we trust, store, and hand to the compression pipeline — so a text-based payload
/// (SVG/HTML/JS) can never be stored or served on-origin. Each entry maps to the
/// server-controlled file extension we persist the original under.
///
/// HEIC/HEIF are deliberately excluded: the preview pipeline (`image` crate, and
/// the bundled ffmpeg 6.1) cannot decode them, so accepting them would store files
/// that never get a thumbnail. iOS Safari already transcodes HEIC→JPEG when a photo
/// is selected via a file input, so this rejects only the rare HEIC-preserving
/// upload path — with a clear error rather than a silently broken post.
const ALLOWED_MEDIA: &[(&str, &str)] = &[
    ("image/jpeg", "jpg"),
    ("image/png", "png"),
    ("image/webp", "webp"),
    ("image/gif", "gif"),
    ("video/mp4", "mp4"),
    ("video/quicktime", "mov"),
    ("video/webm", "webm"),
];

pub async fn upload(
    State(state): State<AppState>,
    auth: AuthUser,
    mut multipart: Multipart,
) -> Result<(StatusCode, Json<UploadDto>), AppError> {
    // Rate limit: N uploads per hour per user. Gated by master + per-endpoint toggles.
    let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await;
    let upload_rate_on = config::get_bool(&state.config_cache, "upload_rate_enabled", true).await;
    if rate_limits_on && upload_rate_on {
        let upload_rate =
            config::get_i64(&state.config_cache, "upload_rate_per_hour", 100).await as usize;
        if let Err(retry_after_secs) = state.rate_limiter.check_with_retry(
            format!("upload:{}", auth.user_id),
            upload_rate,
            Duration::from_secs(3600),
        ) {
            drain_multipart(multipart).await;
            return Err(AppError::TooManyRequests(
                "Du hast dein Upload-Limit für diese Stunde erreicht.".into(),
                Some(retry_after_secs),
            ));
        }
    }

    // Check if user is banned
    let user = User::find_by_id(&state.pool, auth.user_id)
        .await?
        .ok_or_else(|| AppError::NotFound("Benutzer nicht gefunden.".into()))?;
    if user.is_banned {
        drain_multipart(multipart).await;
        // `UserBanned`, not `Forbidden`: a ban is reversible, so the client must KEEP the queued
        // blob and park it until `user-shown` arrives. Under the generic `forbidden` code it
        // purged the photo from IndexedDB and moved the row to `blocked`, which has no retry
        // button — so an unban restored everything except whatever was in flight.
        return Err(AppError::UserBanned("Du bist gesperrt.".into()));
    }

    // Check if uploads are locked
    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()))?;
    // RELEASE IS CHECKED FIRST, AND THE ORDER IS THE WHOLE POINT.
    //
    // `release ⇒ lock`, so a released gallery satisfies BOTH conditions. Testing the lock first
    // made this branch unreachable: every post-release upload — the overwhelmingly common case,
    // since release is the end-of-event action every guest's queue runs into — answered
    // `uploads_locked`, and the `GalleryReleased` arm below was dead code that read as if it
    // worked. The commit-time re-check further down splits the two correctly, so the two paths
    // also disagreed about the same event state depending on where the upload was intercepted.
    //
    // The codes are not interchangeable to the client (see upload-queue.ts): `uploads_locked`
    // charges an attempt and re-pushes the whole photo on the backoff ladder, and tells the guest
    // to find it via the camera button. `gallery_released` PARKS it — no attempt charged, no
    // re-push — and says the photo is safe but needs the hosts to reopen the gallery. Against an
    // answer that cannot change on its own, the first is a cellular data leak with a misleading
    // message attached.
    //
    // Both keep the blob; both are cleared by `event-opened`. Only the retry behaviour differs.
    if event.export_released_at.is_some() {
        drain_multipart(multipart).await;
        return Err(AppError::GalleryReleased(
            "Die Galerie ist abgeschlossen — es können keine neuen Fotos mehr hinzugefügt werden."
                .into(),
        ));
    }
    if event.uploads_locked_at.is_some() {
        drain_multipart(multipart).await;
        // A PLAIN lock (the host paused uploads mid-event) is the reversible-and-likely-soon case,
        // so auto-retry is right here: the client keeps the blob and resumes on `event-opened`.
        return Err(AppError::UploadsLocked("Uploads sind gesperrt.".into()));
    }

    // Read config limits from DB
    let max_image_mb: i64 = config::get_i64(&state.config_cache, "max_image_size_mb", 20).await;
    let max_video_mb: i64 = config::get_i64(&state.config_cache, "max_video_size_mb", 500).await;

    // The uploaded file is streamed straight to a temp file on disk (never buffered
    // whole in memory — a 500 MB video used to cost 500 MB of RAM per concurrent
    // upload). We only keep the first ≤512 bytes in memory for magic-byte sniffing.
    // On success the temp file is renamed into place under its detected extension.
    let upload_id = Uuid::new_v4();
    let event_slug = &state.config.event_slug;
    let originals_dir = state
        .config
        .media_path
        .join(format!("originals/{event_slug}"));
    let temp_abs = originals_dir.join(format!("{upload_id}.tmp"));
    // Armed before anything can create the file, so there is no window in which bytes exist
    // unowned. From here on, EVERY exit — return, `?`, panic, or the future being dropped
    // mid-body by a client disconnect — reclaims them, and the explicit `remove_file` calls
    // that used to be sprinkled over the return paths are gone. One owner, one rule.
    let mut file_guard = TempFileGuard::new(temp_abs.clone());

    let mut streamed: Option<(i64, Vec<u8>)> = None; // (size, head bytes for sniffing)
    let mut caption: Option<String> = None;
    let mut hashtags_csv: Option<String> = None;
    // The client's idempotency key. Optional: an older client, or any other caller, simply
    // doesn't send one and gets the previous behaviour.
    let mut client_upload_id: Option<Uuid> = None;
    // Admission reservation for this body's temp bytes. Declared out here so it lives until the
    // handler returns — the temp file exists for that whole span, and releasing early would let
    // the next upload reserve space this one is still occupying. Dropping it is the release, so
    // every exit path (success, error, client disconnect) returns the budget automatically.
    let mut _admission: Option<tokio::sync::OwnedSemaphorePermit> = None;

    // The multipart read is wrapped so the field loop can use `?` freely; reclaiming the temp
    // file on failure is `file_guard`'s job, not this block's.
    let parse_result: Result<(), AppError> = async {
        while let Some(field) = multipart
            .next_field()
            .await
            .map_err(|e| AppError::BadRequest(e.to_string()))?
        {
            let name = field.name().unwrap_or_default().to_string();
            match name.as_str() {
                "file" => {
                    // The client-declared Content-Type does NOT determine the stored
                    // MIME/extension — those come from the file's magic bytes below. The
                    // declared type only picks the streaming cap so an oversized body is
                    // aborted early; a mislabelled type only makes the cap *stricter*
                    // (safe), and the authoritative per-class check still runs on the
                    // detected type.
                    let declared = field.content_type().unwrap_or("").to_string();
                    let cap_bytes = if declared.starts_with("video/") {
                        (max_video_mb * 1024 * 1024) as usize
                    } else if declared.starts_with("image/") {
                        (max_image_mb * 1024 * 1024) as usize
                    } else {
                        (max_image_mb.max(max_video_mb) * 1024 * 1024) as usize
                    };
                    // ADMISSION BEFORE THE FIRST BYTE TOUCHES DISK. The headroom gate below can
                    // only refuse to COMMIT an upload — by the time it runs, the body has already
                    // been streamed to its temp file. Nothing else bounds how many bodies stream
                    // at once (axum imposes no limit, the tower stack is just TraceLayer, Caddy
                    // passes through), so ~100 guests tapping "upload all" after the ceremony put
                    // 10-20 GB of `.tmp` on a 40 GB volume that the gate cannot see, eating the
                    // reserve that keeps Postgres able to write WAL. The permit is held until the
                    // handler returns, which is exactly as long as the temp file can exist.
                    _admission = Some(state.upload_admission.reserve(cap_bytes).await.ok_or_else(
                        || {
                            AppError::ServiceUnavailable(
                                "Gerade laden sehr viele Gäste hoch. Dein Foto bleibt in der \
                                     Warteschlange und wird gleich automatisch gesendet."
                                    .into(),
                                Some(30),
                            )
                        },
                    )?);
                    tokio::fs::create_dir_all(&originals_dir)
                        .await
                        .map_err(|e| AppError::Internal(e.into()))?;
                    streamed = Some(stream_field_to_file(field, &temp_abs, cap_bytes).await?);
                }
                "caption" => {
                    caption = Some(read_text_field_bounded(field, MAX_CAPTION_BYTES).await?);
                }
                "hashtags" => {
                    hashtags_csv = Some(read_text_field_bounded(field, MAX_HASHTAGS_BYTES).await?);
                }
                "client_upload_id" => {
                    // BOUNDED, like every other text field here. This used `Field::text()`,
                    // which buffers without any ceiling of its own: axum builds its multipart
                    // reader with no `SizeLimit`, so the only bound was this route's 576 MiB
                    // body limit — and `text()` then decodes that into a second full String.
                    // One request from any authenticated guest, declaring `client_upload_id`
                    // and sending 576 MiB of padding, peaks well past the app container's 1 GB
                    // and gets it OOM-killed: every SSE stream dropped, every in-flight upload's
                    // temp file stranded, `restart: unless-stopped` cycling it. `caption` and
                    // `hashtags` were bounded by the helper for exactly this reason; this field
                    // arrived later (migration 022) and missed it.
                    //
                    // 64 bytes fits a hyphenated UUID (36) with room to spare.
                    let raw = read_text_field_bounded(field, 64).await?;
                    // A malformed key is not worth rejecting an upload over — the photo is the
                    // thing the guest cares about. Drop the key and lose only the retry
                    // protection, which is exactly where we were before it existed.
                    client_upload_id = Uuid::parse_str(raw.trim()).ok();
                }
                _ => {}
            }
        }
        Ok(())
    }
    .await;

    parse_result?;

    // Idempotency, fast path: this key already has a live upload, so the previous attempt DID
    // succeed and only its response was lost. Replay that response instead of storing the photo
    // a second time and charging the guest's quota twice.
    //
    // The body has necessarily already been streamed to disk — the key arrives as a multipart
    // field, so it cannot be known before the body is read. Re-sending the bytes is the client's
    // cost and it has already been paid by the time we get here; what has to be prevented is a
    // second ROW, and that is what this does. The concurrent case (two retries in flight at once)
    // is caught by the unique index inside the transaction below.
    //
    // The temp file needs no explicit cleanup here: `file_guard` is armed and reclaims it when
    // this early return drops it.
    if let Some(cid) = client_upload_id
        && let Some(existing) = Upload::find_by_client_upload_id(&state.pool, auth.user_id, cid)
            .await
            .map_err(AppError::from)?
    {
        tracing::info!(
            client_upload_id = %cid, upload_id = %existing.id,
            "duplicate upload suppressed; replaying the original response"
        );
        let dto = replay_upload_dto(&state, &existing, &user.display_name).await;
        return Ok((StatusCode::OK, Json(dto)));
    }

    // From here on the temp file may exist. Every exit reclaims it via `file_guard` — see
    // TempFileGuard for why the explicit per-branch cleanup this replaced was not enough.
    let (size, head) = match streamed {
        Some(s) => s,
        None => return Err(AppError::BadRequest("Keine Datei hochgeladen.".into())),
    };

    // Validate caption length. Counted in chars (code points) to match the
    // "Zeichen" wording in the error message — `.len()` would be bytes and
    // reject perfectly valid German/emoji captions early.
    if let Some(ref cap) = caption
        && cap.chars().count() > MAX_CAPTION_LENGTH
    {
        return Err(AppError::BadRequest(format!(
            "Beschreibung ist zu lang. Maximum: {} Zeichen.",
            MAX_CAPTION_LENGTH
        )));
    }

    // Determine the file type from its magic bytes and require it to be on the
    // allowlist. `infer` returns None for text-based payloads (SVG/HTML/JS), so
    // those are rejected outright — closing the stored-XSS vector. Both the MIME
    // we persist and the on-disk extension come from the detected type, never from
    // client-supplied values.
    let kind = match infer::get(&head) {
        Some(k) => k,
        None => {
            return Err(AppError::BadRequest(
                "Dateityp nicht erkannt oder nicht unterstützt.".into(),
            ));
        }
    };
    let (mime, ext) = match ALLOWED_MEDIA
        .iter()
        .find(|(allowed, _)| *allowed == kind.mime_type())
        .map(|(m, e)| ((*m).to_string(), *e))
    {
        Some(v) => v,
        None => {
            return Err(AppError::BadRequest(format!(
                "Dateityp wird nicht unterstützt: {}.",
                kind.mime_type()
            )));
        }
    };

    // Validate file size against the authoritative per-detected-class limit.
    let max_bytes = if mime.starts_with("video/") {
        max_video_mb * 1024 * 1024
    } else {
        max_image_mb * 1024 * 1024
    };
    if size > max_bytes {
        return Err(AppError::BadRequest(format!(
            "Datei ist zu groß. Maximum: {} MB.",
            max_bytes / (1024 * 1024)
        )));
    }

    // Images only: refuse anything the compression worker could never decode, reading just
    // the header. Without this the upload is accepted with a 201 and then silently
    // soft-deleted minutes later when the worker gives up — the guest sees the photo
    // vanish with, at best, a vague "could not be processed". Rejecting here gives them a
    // reason at the door that they can act on, and it uses the SAME budget the worker
    // enforces, so admission and processing cannot disagree.
    //
    // Both probes open the file and run the codec's header parse — synchronous filesystem and
    // CPU work. They ran inline on the async task, which on this 2-vCPU box means tokio has
    // exactly two worker threads and every upload stalled half the runtime's request-serving
    // capacity. Everything else in the app that blocks (image encode, bcrypt) is already on the
    // blocking pool; this was the one that wasn't.
    if mime.starts_with("image/") {
        let probe_path = temp_abs.clone();
        let probe = tokio::task::spawn_blocking(move || {
            let over = crate::services::imaging::exceeds_decode_budget(&probe_path);
            // Only pay for the second header read when it will actually be shown to the guest.
            let mp = over
                .then(|| crate::services::imaging::megapixels(&probe_path))
                .flatten();
            (over, mp)
        })
        .await;
        // A join error is the blocking pool panicking or shutting down. That says nothing about
        // the image, so admit it and let the compression worker be the judge rather than
        // rejecting a photo for an infrastructure reason.
        let (over_budget, mp) = probe.unwrap_or_else(|e| {
            tracing::warn!(error = ?e, "decode-budget probe failed to run; admitting the upload");
            (false, None)
        });
        if over_budget {
            tracing::info!(
                %mime, megapixels = ?mp,
                "rejecting an image that exceeds the decode budget at admission"
            );
            let _ = tokio::fs::remove_file(&temp_abs).await;
            let detail = mp.map_or(String::new(), |mp| format!(" (ca. {mp:.0} Megapixel)"));
            return Err(AppError::BadRequest(format!(
                "Bild hat zu viele Bildpunkte{detail} und kann nicht verarbeitet werden. \
                 Bitte verkleinere es und lade es erneut hoch."
            )));
        }
    }

    // Per-user storage quota — dynamic formula based on available disk space and the
    // number of active uploaders. Gated by master + per-area toggles so the admin can
    // disable it on trusted instances.
    let quota_on = config::get_bool(&state.config_cache, "quota_enabled", true).await;
    let storage_quota_on =
        config::get_bool(&state.config_cache, "storage_quota_enabled", true).await;
    // GLOBAL DISK GATE, checked before the per-user ceiling and independent of every quota
    // toggle. The per-user quota is a fairness mechanism, not a disk guarantee — and since it
    // carries a floor (MIN_QUOTA_LIMIT_BYTES) so a guest's allowance stops shrinking as the
    // party fills up, the aggregate ceiling it used to imply is gone entirely. Something has to
    // own "do not fill the volume", because `postgres_data`, `media_data` and `exports_data`
    // share one filesystem: the end state is not a degraded feature, it is Postgres unable to
    // write WAL and the whole event down with nobody watching.
    //
    // WHAT IS RESERVED IS NOT A CONSTANT. A flat reserve answers "can Postgres still write",
    // which is necessary and not sufficient: the keepsake needs room for BOTH halves at once —
    // `required_free_bytes` is `media × 1.1 × 2`, since the ZIP and the HTML viewer are each
    // gallery-sized. On the 40 GB box this runs on, a flat 10 GB reserve let uploads continue to
    // roughly 25 GB of media while the release needed `2.2 × 25 + 10` = 65 GB free. Every upload
    // in that band succeeded and then the archive could never be built — the product's entire
    // promise, failing silently at the end of the night with nobody there to notice.
    //
    // So the gate enforces the invariant that actually matters: never accept an upload that
    // would make the keepsake unbuildable. It shares `required_free_bytes` with the export
    // preflight so the two cannot drift into disagreeing about the same question.
    //
    // Deliberately NOT gated behind `quota_enabled`. That switch exists so an operator can stop
    // rationing space between guests; it was never meant to authorise running the disk to zero,
    // and an operator flipping it at 23:00 to unblock a guest should not silently disarm the
    // last thing standing between the party and a dead database.
    // `disk_cache`, not the uncached `disk::free_bytes`. That function deliberately bypasses the
    // cache for the EXPORT PREFLIGHT, where a sibling worker can move free space by tens of GB
    // inside the TTL and a stale reading would authorise the write that fills the disk. This is
    // the opposite situation: the busiest write path in the app, on a 2 vCPU box, where an
    // uncached read means `sysinfo::Disks::new_with_refreshed_list()` — a synchronous scan of
    // every mount, on the async runtime — for every single photo. The quota check immediately
    // below already accepts the same 15s staleness for the same question.
    if let Some(disk) = state.disk_cache.snapshot(&state.config.media_path) {
        // NOTE: `free` already excludes this upload. The body was streamed to its temp file
        // during multipart parsing, well above, so the bytes are on disk before this runs —
        // subtracting `size` here again would refuse a full file-size early.
        //
        // `media_total` is the opposite: it is a sum of committed DB rows, and this upload's row
        // does not exist yet, so the prospective total does need `+ size`.
        let free = disk.free as i64;
        let media_after = state
            .media_total
            .get(&state.pool, &state.config.event_slug)
            .await
            .saturating_add(size);
        let keepsake_needs =
            crate::services::export::required_free_bytes(media_after.max(0) as u64, 2) as i64;
        let required = keepsake_needs.saturating_add(DISK_RESERVE_BYTES);
        if free < required {
            tracing::error!(
                free_bytes = free,
                upload_size = size,
                media_after,
                keepsake_needs,
                reserve = DISK_RESERVE_BYTES,
                "refusing upload: it would leave too little room to build the keepsake"
            );
            return Err(AppError::QuotaExceeded(
                "Der Speicher des Events ist fast voll — damit die Galerie am Ende noch als \
                 Download erstellt werden kann, sind neue Uploads jetzt gesperrt. Bitte sag \
                 einem Host Bescheid."
                    .into(),
            ));
        }
    }
    // Failing OPEN when the disk can't be read is deliberate and matches the per-user quota
    // below: refusing every upload because a `statfs` failed would be a worse outage than the
    // one being guarded against.

    // When quota is enforced, this holds the byte ceiling so the increment UPDATE below can
    // enforce it atomically (`WHERE total + size <= limit`). Without that guard, two
    // concurrent uploads from the same user (e.g. phone + laptop) both pass this stale
    // pre-check and both increment, blowing past the quota. The pre-check stays as a
    // fast path that avoids the disk write when the user is already clearly over.
    let mut quota_limit: Option<i64> = None;
    if quota_on && storage_quota_on {
        let estimate = compute_storage_quota(&state).await;
        if let Some(limit) = estimate.limit_bytes {
            quota_limit = Some(limit);
            let prospective_total = user.total_upload_bytes.saturating_add(size);
            if prospective_total > limit {
                return Err(AppError::QuotaExceeded(
                    // Name the remedy, because the guest cannot see the number. Every quota
                    // display is staff-gated by design, so a guest hitting this had no idea what
                    // the limit was, how close they were, or what to do — and the one sentence
                    // that tells them ("delete older posts") lived inside the staff-only block.
                    "Du hast dein Upload-Limit für dieses Event erreicht. Lösche ältere eigene \
                 Beiträge, um wieder Platz zu schaffen."
                        .into(),
                ));
            }
        }
    }

    // All checks passed — atomically move the temp file to its final, extension-correct
    // path (same directory, so the rename is cheap and atomic).
    let relative_path = format!("originals/{event_slug}/{upload_id}.{ext}");
    let absolute_path = state.config.media_path.join(&relative_path);
    tokio::fs::rename(&temp_abs, &absolute_path)
        .await
        .map_err(|e| AppError::Internal(e.into()))?;
    // THERE MUST BE NO `.await` BETWEEN THE RENAME AND THIS LINE. Both statements resolve on
    // the same poll, so the future cannot be dropped between them and the guard is never
    // pointing at a path that no longer holds the bytes. If the rename fails the guard still
    // owns `temp_abs`, which is why retargeting comes after it rather than before.
    file_guard.retarget(absolute_path.clone());

    // Process hashtags from caption and explicit CSV, capped — see `normalize_tags`.
    let tags = normalize_tags(
        caption
            .as_deref()
            .map(hashtag::extract_hashtags)
            .unwrap_or_default(),
        hashtags_csv.as_deref(),
    );

    // Quota accounting, the upload row, and its hashtag links must be atomic: a
    // crash between the bytes increment and the insert would permanently charge
    // bytes with no row to reclaim them (silent quota erosion / spurious lockout).
    let tx_result: Result<Upload, AppError> = async {
        let mut tx = state.pool.begin().await?;

        // RE-CHECK THE LOCK, UNDER A ROW LOCK, INSIDE THE COMMIT TX.
        //
        // The pre-flight check at the top of this handler ran BEFORE we streamed the body — which
        // for a 500 MB video is minutes. Trusting it here is a TOCTOU that silently loses photos
        // from the keepsake, and it is the real cause of the "stale keepsake" bug that survived
        // three rounds of fixes inside the export state machine:
        //
        //   1. guest starts a big upload; the lock check passes (event open)
        //   2. host releases the gallery → uploads lock, export workers snapshot the uploads table
        //   3. this upload commits AFTER that snapshot → it shows up in the live feed but is
        //      MISSING from the downloaded keepsake, permanently (nothing ever regenerates it)
        //
        // `FOR SHARE` conflicts with the `UPDATE event` in `release_gallery`, which serializes us
        // against it. Either we take the lock first — and release (hence the export snapshot) is
        // strictly ordered after our commit, so the snapshot CONTAINS this upload — or release
        // commits first and we observe the lock here and reject. Either way the keepsake is
        // complete. `UploadsLocked` (not Forbidden) is reversible: the client keeps the blob and
        // resumes it when the host reopens.
        let (locked_at, released_at): (Option<DateTime<Utc>>, Option<DateTime<Utc>>) =
            sqlx::query_as(
                "SELECT uploads_locked_at, export_released_at FROM event WHERE id = $1 FOR SHARE",
            )
            .bind(auth.event_id)
            .fetch_one(&mut *tx)
            .await?;
        // Same order as the fast-path check above, and for the same reason: `release ⇒ lock`, so
        // testing the lock first would collapse a release into `uploads_locked` and set the client
        // auto-retrying a photo that can never be accepted until a host reopens the gallery. A
        // guest who lost the race with `release_gallery` must get `gallery_released` so the queue
        // parks it instead.
        if released_at.is_some() {
            return Err(AppError::GalleryReleased(
                "Die Galerie ist abgeschlossen — es können keine neuen Fotos mehr hinzugefügt \
                 werden."
                    .into(),
            ));
        }
        if locked_at.is_some() {
            return Err(AppError::UploadsLocked("Uploads sind gesperrt.".into()));
        }

        // Increment the user's byte total. When a quota is in force, guard it atomically
        // (`total + size <= limit`) so two concurrent uploads can't both slip past the
        // stale pre-check — the loser's UPDATE matches 0 rows and we abort with the same
        // terminal quota error (the tx rolls back on drop; the on-disk file is cleaned by
        // the error path below).
        let inc = if let Some(limit) = quota_limit {
            sqlx::query(
                "UPDATE \"user\" SET total_upload_bytes = total_upload_bytes + $2
                 WHERE id = $1 AND total_upload_bytes + $2 <= $3",
            )
            .bind(auth.user_id)
            .bind(size)
            .bind(limit)
            .execute(&mut *tx)
            .await?
        } else {
            sqlx::query(
                "UPDATE \"user\" SET total_upload_bytes = total_upload_bytes + $2 WHERE id = $1",
            )
            .bind(auth.user_id)
            .bind(size)
            .execute(&mut *tx)
            .await?
        };
        if inc.rows_affected() == 0 {
            return Err(AppError::QuotaExceeded(
                // Name the remedy, because the guest cannot see the number. Every quota
                // display is staff-gated by design, so a guest hitting this had no idea what
                // the limit was, how close they were, or what to do — and the one sentence
                // that tells them ("delete older posts") lived inside the staff-only block.
                "Du hast dein Upload-Limit für dieses Event erreicht. Lösche ältere eigene \
                 Beiträge, um wieder Platz zu schaffen."
                    .into(),
            ));
        }
        // `None` means a concurrent request already stored this key. The transaction — quota
        // increment included — is abandoned by returning here, and the caller replays the winning
        // row. This is the narrow race the fast path above cannot see: two retries of the same
        // photo in flight at the same moment.
        let Some(upload) = Upload::create(
            &mut *tx,
            auth.event_id,
            auth.user_id,
            &relative_path,
            &mime,
            size,
            caption.as_deref(),
            client_upload_id,
        )
        .await?
        else {
            return Err(AppError::Conflict(DUPLICATE_UPLOAD_MARKER.into()));
        };
        for tag in &tags {
            let h = Hashtag::upsert(&mut *tx, auth.event_id, tag).await?;
            Hashtag::link_to_upload(&mut *tx, upload.id, h.id).await?;
        }

        // Hand the bytes to the row BEFORE committing, not after.
        //
        // `tx.commit().await` is a suspension point, and a COMMIT already written to the
        // socket is applied by Postgres whether or not this future lives to read the reply.
        // Disarming afterwards left a real window: the guest walks out of range mid-commit,
        // axum drops the future, Postgres commits the row anyway, and `Drop` deletes the file
        // that freshly committed row points at. The result is invisible to every repair path
        // — the row is live so the deleted-media sweep skips it, the file is gone so the
        // orphan sweep skips it — and it is missing from the keepsake with nothing in the log
        // naming it as loss.
        //
        // Disarming first cannot fix the cancellation (nothing in-process can), but it moves
        // the failure to the recoverable side: if we are dropped mid-commit the bytes leak,
        // and leaked bytes under a final name are exactly what the orphan sweeper reclaims.
        // A committed row whose file we deleted is unrecoverable. Prefer the leak.
        file_guard.disarm();
        if let Err(e) = tx.commit().await {
            // Deliberately do NOT re-arm the guard here.
            //
            // A `commit()` that returns `Err` is INDETERMINATE, not "definitely rolled back".
            // sqlx writes `COMMIT` to the socket and awaits the reply; if the connection dies
            // after Postgres flushed the WAL record but before that reply arrives (a db
            // restart, a killed backend, a network blip), the row is durably committed and we
            // are told it failed. Re-arming would then delete the file a live row points at —
            // the exact unrecoverable case the comment above says to avoid, just reached
            // through the error path instead of the cancellation path.
            //
            // It is worse than it sounds, because the client retries: the idempotency fast
            // path finds the committed row, answers 200, and the phone purges the only other
            // copy of the photo. So we prefer the leak in both directions. If the commit
            // genuinely did not apply, `sweep_orphan_originals` reclaims the bytes on its next
            // pass (it deletes files with no DB row, which is precisely this case).
            tracing::error!(
                error = ?e,
                path = %absolute_path.display(),
                "upload commit returned an error; leaving the file in place because the commit \
                 may still have applied — the orphan sweeper reclaims it if it did not"
            );
            return Err(e.into());
        }
        Ok(upload)
    }
    .await;

    // The file is already on disk at `absolute_path`, and `file_guard` was retargeted to it
    // above — so every path out of here that is NOT a successful commit leaves the guard armed
    // and reclaims the bytes on the way out. That covers the concurrent-duplicate loser below
    // as well as the plain error case, and unlike the explicit `remove_file` calls it replaces,
    // it also covers axum dropping this future instead of returning.
    //
    // The successful-commit case disarmed the guard inside the block, immediately before
    // `tx.commit()` — see the comment there for why it cannot be done out here.
    let upload = match tx_result {
        Ok(u) => u,
        // The concurrent duplicate resolved inside the transaction. The winner's row is committed;
        // answer with it so both retries of the same photo get the same successful reply. The
        // loser's bytes are reclaimed by the guard when this return drops it.
        Err(AppError::Conflict(ref marker)) if marker == DUPLICATE_UPLOAD_MARKER => {
            let existing = match client_upload_id {
                Some(cid) => Upload::find_by_client_upload_id(&state.pool, auth.user_id, cid)
                    .await
                    .map_err(AppError::from)?,
                None => None,
            };
            // If the winning row has vanished between the conflict and this lookup (deleted in
            // the intervening milliseconds), there is nothing to replay — report the conflict.
            let existing = existing.ok_or_else(|| {
                AppError::Conflict("Dieser Upload wurde bereits verarbeitet.".into())
            })?;
            tracing::info!(
                upload_id = %existing.id,
                "concurrent duplicate upload resolved; replaying the stored row"
            );
            let dto = replay_upload_dto(&state, &existing, &user.display_name).await;
            return Ok((StatusCode::OK, Json(dto)));
        }
        Err(e) => return Err(e),
    };

    // Spawn compression task
    state
        .compression
        .process(upload.id, relative_path, mime.clone());

    // Broadcast SSE event
    let dto = UploadDto {
        id: upload.id,
        user_id: auth.user_id,
        uploader_name: user.display_name,
        preview_url: None,
        thumbnail_url: None,
        mime_type: mime,
        caption,
        hashtags: tags,
        like_count: 0,
        comment_count: 0,
        liked_by_me: false,
        created_at: upload.created_at,
    };

    let _ = state.sse_tx.send(crate::state::SseEvent::new(
        "new-upload",
        serde_json::to_string(&dto).unwrap_or_default(),
    ));

    Ok((StatusCode::CREATED, Json(dto)))
}

#[derive(Deserialize)]
pub struct EditUploadRequest {
    pub caption: Option<String>,
    pub hashtags: Option<Vec<String>>,
}

pub async fn edit_upload(
    State(state): State<AppState>,
    auth: AuthUser,
    Path(upload_id): Path<Uuid>,
    Json(body): Json<EditUploadRequest>,
) -> Result<StatusCode, AppError> {
    // Banned users keep read access but cannot mutate (USER_JOURNEYS §10).
    if auth.is_banned {
        return Err(AppError::Forbidden("Du bist gesperrt.".into()));
    }
    let upload = Upload::find_by_id_and_event(&state.pool, upload_id, auth.event_id)
        .await?
        .ok_or_else(|| AppError::NotFound("Upload nicht gefunden.".into()))?;

    if upload.user_id != auth.user_id {
        return Err(AppError::Forbidden("Nur eigene Uploads bearbeiten.".into()));
    }

    // This endpoint had no rate limit of any kind, while every other mutating route has one.
    let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await;
    let edit_rate_on =
        config::get_bool(&state.config_cache, "upload_edit_rate_enabled", true).await;
    if rate_limits_on && edit_rate_on {
        let edit_rate =
            config::get_i64(&state.config_cache, "upload_edit_rate_per_min", 30).await as usize;
        if let Err(retry_after_secs) = state.rate_limiter.check_with_retry(
            format!("upload_edit:{}", auth.user_id),
            edit_rate,
            Duration::from_secs(60),
        ) {
            return Err(AppError::TooManyRequests(
                "Zu viele Änderungen. Bitte warte kurz.".into(),
                Some(retry_after_secs),
            ));
        }
    }

    // Validate to the same limits as the upload path. This route had none at all, so a caption
    // rejected at upload could be set here instead, and the tags went in raw — meaning `#Party`
    // via edit and `party` via upload became two different hashtag rows.
    if let Some(ref caption) = body.caption
        && caption.chars().count() > MAX_CAPTION_LENGTH
    {
        return Err(AppError::BadRequest(format!(
            "Beschreibung ist zu lang. Maximum: {MAX_CAPTION_LENGTH} Zeichen."
        )));
    }
    let normalized_tags = body
        .hashtags
        .as_ref()
        .map(|tags| normalize_tags(tags.clone(), None));

    // A PATCH that changes nothing must not retire the keepsake generation.
    //
    // `invalidate_and_arm` below ran unconditionally, outside both `if let Some(...)` guards, so
    // `PATCH {}` — which any authenticated guest can send in a loop against their own upload —
    // bumped export_epoch and armed a fresh pair of full-gallery export workers every time.
    // REGEN_DEBOUNCE bounds the rate of that, not the total work, so the keepsake could be kept
    // permanently un-downloadable.
    //
    // The hashtag half of that guard was wrong, and the comment here used to defend it: it said
    // re-sending an identical list "is not a free loop". It is exactly a free loop. `PATCH
    // {"hashtags": []}` carries no photo, no bytes and no client-side cost, yet it made
    // `normalized_tags` `Some`, sailed past the caption-only check, and bumped `export_epoch` on
    // every request — retiring the HTML keepsake instantly. REGEN_DEBOUNCE (20s) throttles when a
    // rebuild may START, not the epoch bump, so at the 30/min this endpoint allows no rebuild ever
    // gets a quiet window to finish in and `GET /export/html` 404s for the whole event. The ZIP is
    // carried forward, so this denied exactly half the product.
    //
    // So compare properly. One indexed lookup against `upload_hashtag` is cheap next to the
    // full-gallery rebuild a false positive arms.
    let caption_changed = match (&body.caption, &upload.caption) {
        (Some(new), existing) => Some(new.as_str()) != existing.as_deref(),
        (None, _) => false,
    };
    let tags_changed = match &normalized_tags {
        None => false,
        Some(incoming) => {
            // Compare on the same normalised form `upsert` keys on, so "#Party", "party" and
            // " #PARTY " are all the same tag and none of them counts as an edit.
            let mut want: Vec<String> = incoming
                .iter()
                .map(|t| t.trim().trim_start_matches('#').to_lowercase())
                .collect();
            want.sort();
            want.dedup();
            let have = Hashtag::normalized_for_upload(&state.pool, upload_id).await?;
            want != have
        }
    };
    if !caption_changed && !tags_changed {
        return Ok(StatusCode::OK);
    }

    // Caption update + hashtag wipe-then-relink in one transaction, so a crash
    // mid-relink can't leave the upload with its hashtags stripped.
    //
    // Editing is intentionally allowed while uploads are locked or the gallery is released — like
    // comments and likes, the lock freezes *new uploads* only (USER_JOURNEYS §9.3). But a caption
    // is embedded in the HTML viewer keepsake (the ZIP holds media only — see export.rs), so an
    // edit AFTER release must regenerate the viewer, or the downloadable keepsake keeps showing the
    // old caption forever while the live feed shows the new one. Same atomicity as delete_upload:
    // the edit and its invalidation share one tx so a dropped handler can't leave them disagreeing.
    // `Affects::ViewerOnly` carries the finished ZIP forward (the media didn't change); when the
    // gallery isn't released, `invalidate_and_arm` returns None and this is a no-op.
    let mut tx = state.pool.begin().await?;
    if let Some(ref caption) = body.caption {
        Upload::update_caption(&mut *tx, upload_id, Some(caption)).await?;
    }
    if let Some(ref hashtags) = normalized_tags {
        Hashtag::unlink_all_from_upload(&mut *tx, upload_id).await?;
        // Sort + dedup before upserting, exactly as the upload path does. `Hashtag::upsert`
        // takes row locks, so two transactions touching the same two tags in OPPOSITE order
        // deadlock; Postgres aborts one after ~1s and the guest gets a 500. Here the order is
        // whatever the client sent, so it is genuinely attacker-free but genuinely unordered.
        // Sort on the NORMALISED form — that is the key `upsert` actually locks on.
        let mut tags: Vec<&String> = hashtags.iter().collect();
        tags.sort_by_key(|t| t.trim().trim_start_matches('#').to_lowercase());
        tags.dedup_by_key(|t| t.trim().trim_start_matches('#').to_lowercase());
        for tag in tags {
            let h = Hashtag::upsert(&mut *tx, auth.event_id, tag).await?;
            Hashtag::link_to_upload(&mut *tx, upload_id, h.id).await?;
        }
    }
    let regen = crate::services::export::invalidate_and_arm(
        &mut tx,
        &state.config.event_slug,
        crate::services::export::Affects::ViewerOnly,
    )
    .await?;
    tx.commit().await?;
    if let Some(r) = regen {
        crate::handlers::host::start_regen(&state, r);
    }

    Ok(StatusCode::OK)
}

pub async fn delete_upload(
    State(state): State<AppState>,
    auth: AuthUser,
    Path(upload_id): Path<Uuid>,
) -> Result<StatusCode, AppError> {
    // Banned users keep read access but cannot mutate (USER_JOURNEYS §10).
    if auth.is_banned {
        return Err(AppError::Forbidden("Du bist gesperrt.".into()));
    }
    let upload = Upload::find_by_id_and_event(&state.pool, upload_id, auth.event_id)
        .await?
        .ok_or_else(|| AppError::NotFound("Upload nicht gefunden.".into()))?;

    if upload.user_id != auth.user_id {
        return Err(AppError::Forbidden("Nur eigene Uploads löschen.".into()));
    }

    // Atomic with the keepsake invalidation: a guest removing their own photo must have it removed
    // from the downloadable archive too, and a half-applied delete would leave it there forever.
    let mut tx = state.pool.begin().await?;
    Upload::soft_delete_in_event(&mut tx, upload_id, auth.event_id).await?;
    let regen = crate::services::export::invalidate_and_arm(
        &mut tx,
        &state.config.event_slug,
        crate::services::export::Affects::Both,
    )
    .await?;
    tx.commit().await?;
    if let Some(r) = regen {
        crate::handlers::host::start_regen(&state, r);
    }

    // Evict the card live on every other feed + the projector diashow — otherwise
    // a self-deleted post lingers until each viewer manually reloads. Same event
    // the host-delete path already emits and the frontend already handles.
    let _ = state.sse_tx.send(crate::state::SseEvent::new(
        "upload-deleted",
        serde_json::json!({ "upload_id": upload_id }).to_string(),
    ));

    Ok(StatusCode::NO_CONTENT)
}

/// Number of leading bytes retained in memory for magic-byte (`infer`) sniffing. Every
/// allowed type's signature sits well within this; 512 is comfortably generous.
const HEAD_SNIFF_BYTES: usize = 512;

/// Stream a multipart field straight to `dest`, aborting with a 400 the moment it
/// exceeds `max_bytes`. Only the first [`HEAD_SNIFF_BYTES`] bytes are kept in memory
/// (for type detection); the rest goes chunk-by-chunk to disk, so peak memory is a
/// single chunk rather than the whole file. Returns `(total_size, head_bytes)`. On any
/// error the partial temp file is removed so no stray `.tmp` is left behind.
async fn stream_field_to_file(
    mut field: axum::extract::multipart::Field<'_>,
    dest: &std::path::Path,
    max_bytes: usize,
) -> Result<(i64, Vec<u8>), AppError> {
    use tokio::io::AsyncWriteExt;

    let mut file = tokio::fs::File::create(dest)
        .await
        .map_err(|e| AppError::Internal(e.into()))?;
    let mut total: usize = 0;
    let mut head: Vec<u8> = Vec::with_capacity(HEAD_SNIFF_BYTES);

    loop {
        let chunk = match field.chunk().await {
            Ok(Some(c)) => c,
            Ok(None) => break,
            Err(e) => {
                let _ = file.shutdown().await;
                let _ = tokio::fs::remove_file(dest).await;
                return Err(AppError::BadRequest(format!(
                    "Datei konnte nicht gelesen werden: {e}"
                )));
            }
        };

        total = total.saturating_add(chunk.len());
        if total > max_bytes {
            let _ = file.shutdown().await;
            let _ = tokio::fs::remove_file(dest).await;
            return Err(AppError::BadRequest(format!(
                "Datei ist zu groß. Maximum: {} MB.",
                max_bytes / (1024 * 1024)
            )));
        }

        if head.len() < HEAD_SNIFF_BYTES {
            let need = HEAD_SNIFF_BYTES - head.len();
            head.extend_from_slice(&chunk[..need.min(chunk.len())]);
        }

        if let Err(e) = file.write_all(&chunk).await {
            let _ = tokio::fs::remove_file(dest).await;
            return Err(AppError::Internal(e.into()));
        }
    }

    if let Err(e) = file.flush().await {
        let _ = tokio::fs::remove_file(dest).await;
        return Err(AppError::Internal(e.into()));
    }

    Ok((total as i64, head))
}

/// Sentinel for the duplicate detected INSIDE the commit transaction. It never reaches a client:
/// the caller intercepts this exact `Conflict` and answers with the stored row. A marker rather
/// than a new `AppError` variant because the condition is local to this one handler and returning
/// early is the only way to abandon the transaction from inside the async block.
const DUPLICATE_UPLOAD_MARKER: &str = "__duplicate_client_upload_id__";

/// Rebuild the response for an upload that already exists, so a retry is answered exactly as the
/// original was.
///
/// Reads the live state rather than assuming a fresh row: by the time a retry arrives — a
/// reconnect can be minutes later — the derivatives may have been generated and the photo may
/// already have been liked, and a response claiming otherwise would be wrong in a way the client
/// has no way to detect.
///
/// Every read here fails soft. This is the success path of an upload that is already safely
/// stored; degrading to a sparser response is fine, failing the request is not.
async fn replay_upload_dto(state: &AppState, upload: &Upload, uploader_name: &str) -> UploadDto {
    let hashtags: Vec<String> = sqlx::query_scalar(
        "SELECT h.tag FROM upload_hashtag uh
         JOIN hashtag h ON h.id = uh.hashtag_id
         WHERE uh.upload_id = $1
         ORDER BY h.tag",
    )
    .bind(upload.id)
    .fetch_all(&state.pool)
    .await
    .unwrap_or_default();

    let counts: Option<(i64, i64, bool)> = sqlx::query_as(
        "SELECT v.like_count, v.comment_count,
                EXISTS (SELECT 1 FROM \"like\" l WHERE l.upload_id = v.id AND l.user_id = $2)
         FROM v_feed v WHERE v.id = $1",
    )
    .bind(upload.id)
    .bind(upload.user_id)
    .fetch_optional(&state.pool)
    .await
    .ok()
    .flatten();
    let (like_count, comment_count, liked_by_me) = counts.unwrap_or((0, 0, false));

    UploadDto {
        id: upload.id,
        user_id: upload.user_id,
        uploader_name: uploader_name.to_string(),
        preview_url: upload
            .preview_path
            .as_ref()
            .map(|_| format!("/api/v1/upload/{}/preview", upload.id)),
        thumbnail_url: upload
            .thumbnail_path
            .as_ref()
            .map(|_| format!("/api/v1/upload/{}/thumbnail", upload.id)),
        mime_type: upload.mime_type.clone(),
        caption: upload.caption.clone(),
        hashtags,
        like_count,
        comment_count,
        liked_by_me,
        created_at: upload.created_at,
    }
}

/// Drain a multipart body so the HTTP connection stays clean when returning an early error.
/// Without draining, the client may still be sending the body after we've sent our response,
/// which can corrupt the keep-alive connection for subsequent requests.
async fn drain_multipart(mut mp: Multipart) {
    while let Ok(Some(mut field)) = mp.next_field().await {
        while field.chunk().await.ok().flatten().is_some() {}
    }
}

/// Snapshot of the dynamic per-user quota used both by the upload pre-check and the
/// `GET /me/quota` endpoint. `limit_bytes = None` means quota enforcement is currently
/// off (the frontend hides the widget in that case).
pub struct QuotaEstimate {
    pub limit_bytes: Option<i64>,
    pub active_uploaders: i64,
    pub free_disk_bytes: i64,
    /// The tolerance factor the limit above was computed with. Carried on the snapshot so the
    /// number is self-describing; no caller reads it back today.
    #[allow(dead_code)]
    pub tolerance: f64,
}

/// The smallest per-user ceiling this formula is ever allowed to produce.
///
/// Without a floor the quota is not a limit, it is a moving target: the numerator (free disk)
/// only falls and the denominator (uploaders who have posted) only rises, so the ceiling
/// decreases monotonically across the event. A guest comfortably under it at 20:00 is over it
/// at 22:00 having done nothing, and because a delete refunds the quota but does not free the
/// bytes for 24h, the remedy the error message names ("delete older posts") cannot move it
/// back either.
///
/// 500 MB is chosen to clear `max_video_size_mb` (500, seeded in 005) — below that the ceiling
/// could refuse a single legal video outright, which is the worst version of this: the guest
/// pushes 500 MB across cellular and is rejected on arrival, every time, with no way to comply.
///
/// This deliberately trades the quota's disk guarantee for a usability floor. The disk is now
/// bounded by the low-disk warning and the export preflight rather than by this formula alone —
/// see the reserve check in `ensure_export_space`.
const MIN_QUOTA_LIMIT_BYTES: i64 = 500 * 1024 * 1024;

/// Free space on the media volume that uploads may never consume, whatever any quota says.
///
/// The host dashboard's low-disk banner is DERIVED from this (`handlers::host::disk_is_low`)
/// rather than equal to it: the banner fires at 1.25x the gate's closing point, deliberately, so
/// the host is warned while there is still room to act instead of at the same instant guests hit
/// the wall. Shared expression, offset threshold — do not "restore" them to one number. 10 GB
/// is chosen to leave Postgres, its WAL and a rotation of container logs comfortable room on
/// the shared filesystem long after new uploads have been refused.
pub const DISK_RESERVE_BYTES: i64 = 10_000_000_000;

/// Pure per-user quota formula: `max(floor((free_disk * tolerance) / divisor), MIN)`.
///
/// `divisor` is the LARGER of the observed uploader count and the operator's
/// `estimated_guest_count`, so the ceiling settles at its final value early instead of sliding
/// down all evening as guests arrive. (Before this, `estimated_guest_count` was seeded and
/// validated in the admin whitelist but read by no code at all — an operator who set it
/// expecting a stable divisor changed nothing.) It also blunts the abuse case, where the
/// divisor was attacker-controlled: ~1000 throwaway accounts drove every real guest's ceiling
/// to ~52 MB.
///
/// Extracted from `compute_storage_quota` so it's unit-testable without a DB or disk.
fn quota_limit_bytes(free_disk: i64, tolerance: f64, active_uploaders: i64, expected: i64) -> i64 {
    let divisor = active_uploaders.max(expected).max(1);
    let budget = (free_disk as f64 * tolerance).max(0.0);
    let computed = (budget / divisor as f64).floor() as i64;
    // The floor may never exceed what the disk can actually back. Raising a ceiling the volume
    // cannot honour would hand out an allowance on a full disk — turning the quota from a
    // usability floor into a way to finish filling the filesystem that Postgres writes WAL to.
    // Only raise to the floor when the disk can back a floor-sized allowance for real.
    //
    // The earlier form was `computed.max(MIN.min(budget))`, which inverts exactly where it
    // matters: `budget` is the WHOLE disk's share, not one user's, so once budget < 500 MiB the
    // "floor" became the entire remaining budget and every uploader was authorised all of it —
    // 400 MB free with 3 uploaders promised 300 MB each. Below the floor, fall through to the
    // divided value, which is the only number that still shares the space out.
    if budget < MIN_QUOTA_LIMIT_BYTES as f64 {
        computed
    } else {
        computed.max(MIN_QUOTA_LIMIT_BYTES)
    }
}

/// Computes the per-user storage quota using
/// `max(floor((free_disk * tolerance) / max(active_uploaders, estimated_guest_count, 1)), 500 MiB)`
/// — see [`quota_limit_bytes`] for the floor's exact conditions. Returns `limit_bytes =
/// None` whenever the storage quota is currently disabled — callers should skip the
/// check (upload handler) or hide the UI (quota endpoint).
pub async fn compute_storage_quota(state: &AppState) -> QuotaEstimate {
    let quota_on = config::get_bool(&state.config_cache, "quota_enabled", true).await;
    let storage_quota_on =
        config::get_bool(&state.config_cache, "storage_quota_enabled", true).await;
    let tolerance = config::get_f64(&state.config_cache, "quota_tolerance", 0.75).await;

    // Scoped to THIS event (H12). Without the filter, reusing the install for a second event
    // carried the first one's uploaders forward permanently: event one's 30 photographers stayed
    // in event two's quota divisor, silently shrinking every new guest's ceiling for a party they
    // had nothing to do with. There is no reset path anywhere in the code or the runbook, so the
    // only fix would have been hand-written SQL.
    let (active_count,): (i64,) = sqlx::query_as(
        "SELECT COUNT(DISTINCT up.user_id) FROM upload up
           JOIN event e ON e.id = up.event_id
          WHERE up.deleted_at IS NULL AND e.slug = $1",
    )
    .bind(&state.config.event_slug)
    .fetch_one(&state.pool)
    .await
    .unwrap_or((0,));
    let active = active_count.max(1);
    // The operator's expected headcount, used as a FLOOR on the divisor so the ceiling doesn't
    // slide down as guests arrive — see `quota_limit_bytes`. Admin-editable at runtime.
    let expected_guests = config::get_i64(&state.config_cache, "estimated_guest_count", 100).await;

    // Cached disk reading. `None` means we couldn't resolve the media filesystem.
    let disk = state.disk_cache.snapshot(&state.config.media_path);
    let free_disk = disk.map(|d| d.free as i64).unwrap_or(0);

    let limit_bytes = if quota_on && storage_quota_on {
        match disk {
            Some(d) => Some(quota_limit_bytes(
                d.free as i64,
                tolerance,
                active,
                expected_guests,
            )),
            // Fail OPEN, not closed: if the disk can't be read we don't know the real
            // free space, and enforcing a 0-byte limit would reject every upload with a
            // spurious "quota reached". Skip enforcement this round and warn instead.
            None => {
                tracing::warn!(
                    "disk snapshot unavailable; skipping storage-quota enforcement this round"
                );
                None
            }
        }
    } else {
        None
    };

    QuotaEstimate {
        limit_bytes,
        active_uploaders: active,
        free_disk_bytes: free_disk,
        tolerance,
    }
}

/// Outcome of parsing a `Range` request header against a known file length.
#[derive(Debug, PartialEq, Eq)]
pub(crate) enum RangeSpec {
    /// No `Range` header, or one we deliberately don't honour (multi-range, non-`bytes`
    /// unit, malformed). RFC 9110 lets a server ignore a Range it can't process and reply
    /// 200 with the full body, which is what every one of these cases does.
    Full,
    /// A single satisfiable range, resolved to inclusive absolute offsets.
    Partial { start: u64, end: u64 },
    /// Syntactically valid but starts beyond EOF — must be answered 416, not 200, or a
    /// player can loop re-requesting it.
    Unsatisfiable,
}

/// Parse a single-range `bytes=` header against `len`.
///
/// Deliberately supports only the three forms a media element actually sends —
/// `bytes=N-`, `bytes=N-M`, `bytes=-S` (suffix) — and treats everything else as `Full`.
/// Multi-range responses need `multipart/byteranges`, which no `<video>` requires.
pub(crate) fn parse_range(header: Option<&str>, len: u64) -> RangeSpec {
    let Some(raw) = header else {
        return RangeSpec::Full;
    };
    let Some(spec) = raw.trim().strip_prefix("bytes=") else {
        return RangeSpec::Full;
    };
    // Multi-range → fall back to the whole body rather than lie about the content.
    if spec.contains(',') {
        return RangeSpec::Full;
    }
    let Some((from, to)) = spec.split_once('-') else {
        return RangeSpec::Full;
    };
    let (from, to) = (from.trim(), to.trim());

    // A zero-length file can satisfy no range at all.
    if len == 0 {
        return if from.is_empty() && to.is_empty() {
            RangeSpec::Full
        } else {
            RangeSpec::Unsatisfiable
        };
    }

    let (start, end) = if from.is_empty() {
        // Suffix form: the last `to` bytes.
        let Ok(suffix) = to.parse::<u64>() else {
            return RangeSpec::Full;
        };
        if suffix == 0 {
            return RangeSpec::Unsatisfiable;
        }
        (len.saturating_sub(suffix), len - 1)
    } else {
        let Ok(start) = from.parse::<u64>() else {
            return RangeSpec::Full;
        };
        let end = if to.is_empty() {
            len - 1
        } else {
            match to.parse::<u64>() {
                // An end past EOF is clamped, not rejected (RFC 9110 §14.1.1).
                Ok(end) => end.min(len - 1),
                Err(_) => return RangeSpec::Full,
            }
        };
        (start, end)
    };

    if start >= len || start > end {
        RangeSpec::Unsatisfiable
    } else {
        RangeSpec::Partial { start, end }
    }
}

/// Stream a media file from disk into an HTTP response with a fixed set of security
/// headers. Every media response (original, preview, display, thumbnail) goes through here
/// so they consistently carry `X-Content-Type-Options: nosniff` (defense-in-depth against
/// content-type confusion, even if the edge proxy is bypassed) plus an explicit
/// `Content-Disposition` and `Cache-Control`.
///
/// Honours a single `Range`. This is not an optimisation: iOS Safari opens every `<video>`
/// with a `Range: bytes=0-1` probe and abandons the load unless it gets a `206` with a
/// `Content-Range`. Without this, video is unplayable on the app's primary platform no
/// matter what `src` the element is given. `Accept-Ranges: bytes` is advertised on every
/// response so clients know seeking is available before they ask.
async fn stream_media_file(
    req_headers: &axum::http::HeaderMap,
    absolute: &std::path::Path,
    content_type: String,
    disposition: &str,
    cache_control: &str,
) -> Result<axum::response::Response, AppError> {
    use axum::body::Body;
    use axum::http::{Response, StatusCode, header};
    use tokio::io::{AsyncReadExt, AsyncSeekExt};
    use tokio_util::io::ReaderStream;

    if !absolute.exists() {
        return Err(AppError::NotFound("Datei nicht gefunden.".into()));
    }

    let mut file = tokio::fs::File::open(absolute)
        .await
        .map_err(|e| AppError::Internal(e.into()))?;
    let len = file
        .metadata()
        .await
        .map_err(|e| AppError::Internal(e.into()))?
        .len();

    let range = parse_range(
        req_headers.get(header::RANGE).and_then(|v| v.to_str().ok()),
        len,
    );

    let base = |status: StatusCode| {
        Response::builder()
            .status(status)
            .header(header::CONTENT_TYPE, content_type.clone())
            .header(header::CONTENT_DISPOSITION, disposition)
            .header(header::CACHE_CONTROL, cache_control)
            .header(header::X_CONTENT_TYPE_OPTIONS, "nosniff")
            .header(header::ACCEPT_RANGES, "bytes")
    };

    match range {
        RangeSpec::Full => base(StatusCode::OK)
            .header(header::CONTENT_LENGTH, len)
            .body(Body::from_stream(ReaderStream::new(file)))
            .map_err(|e| AppError::Internal(e.into())),

        RangeSpec::Partial { start, end } => {
            file.seek(std::io::SeekFrom::Start(start))
                .await
                .map_err(|e| AppError::Internal(e.into()))?;
            let span = end - start + 1;
            base(StatusCode::PARTIAL_CONTENT)
                .header(header::CONTENT_LENGTH, span)
                .header(header::CONTENT_RANGE, format!("bytes {start}-{end}/{len}"))
                .body(Body::from_stream(ReaderStream::new(file.take(span))))
                .map_err(|e| AppError::Internal(e.into()))
        }

        RangeSpec::Unsatisfiable => base(StatusCode::RANGE_NOT_SATISFIABLE)
            .header(header::CONTENT_RANGE, format!("bytes */{len}"))
            .body(Body::empty())
            .map_err(|e| AppError::Internal(e.into())),
    }
}

/// Streaming download of the original file behind an upload. Used by:
///   - the per-post "Original anzeigen" context action (`window.open`)
///   - `<img src>` / `<video src>` in the feed, lightbox, and diashow when the user is in
///     Data Mode = Original
///
/// **Auth model:** the route is intentionally unauthenticated so it works from
/// `<img src>` / `window.open`. The URL contains the upload's unguessable UUID. Unlike
/// raw `/media` files, this alias is the *only* way to fetch an original: direct
/// `/media/originals/**` access is blocked in the router, and this handler filters out
/// soft-deleted and ban-hidden uploads (via `find_visible_media`) so moderation actually
/// removes access to content. Preview and thumbnail variants are gated the same way (see
/// [`get_preview`] / [`get_thumbnail`]).
/// NO per-IP rate limit on this route, deliberately — a 600/min ceiling was added here and had to
/// come back out.
///
/// The reasoning that put it in was that `/original` serves "100 guests occasionally tapping
/// 'Original anzeigen'", so a venue-wide 10/s could only ever catch a scraper. That is not what
/// this route is. `pickMediaUrl` (frontend/src/lib/data-mode-store.ts) resolves to
/// `preview_url ?? thumbnail_url ?? /original`, and a freshly committed upload has BOTH derivatives
/// null until the compression worker reaches it — at `COMPRESSION_WORKER_CONCURRENCY=2` that is
/// minutes during a post-ceremony burst. So `/original` IS the feed's hot path for exactly the
/// newest photos, in a newest-first grid, at the busiest moment; `VirtualFeed.svelte` says as much
/// where it explains its broken-tile retry.
///
/// With every guest behind one NAT address the bucket is venue-wide: ~6 new photos fanned out by
/// `upload-new` to ~100 open feeds exhausts 600 on its own, and then every original fetch from
/// anyone at the party 429s for the rest of the window. The tiles' own 4-second retry uses a fresh
/// `?r=` nonce, so the clients then hold the bucket saturated themselves. The whole venue watches
/// the newest photos render as broken tiles, and the projector starts skipping slides.
///
/// A per-IP bucket cannot separate "one scraper" from "the entire party" when they share an
/// address, and these four media routes are unauthenticated by design (an `<img>` cannot send a
/// bearer token), so there is no per-user key to move to. Bandwidth abuse belongs at the proxy,
/// where per-connection limits still work; the certain harm here outweighed the speculative
/// protection.
pub async fn get_original(
    State(state): State<AppState>,
    headers: axum::http::HeaderMap,
    Path(upload_id): Path<Uuid>,
) -> Result<axum::response::Response, AppError> {
    let media = Upload::find_visible_media(&state.pool, upload_id)
        .await?
        .ok_or_else(|| AppError::NotFound("Upload nicht gefunden.".into()))?;

    let absolute = state.config.media_path.join(&media.original_path);
    let filename = absolute
        .file_name()
        .and_then(|n| n.to_str())
        .unwrap_or("original");
    // `inline`, not `attachment`. This route is the only source of playable video bytes
    // (there is no video derivative), and an attachment disposition is hostile to a
    // `<video>` element — Safari in particular. It also matches what the UI promises:
    // the action is labelled "Original anzeigen", i.e. view, not download.
    let disposition = format!("inline; filename=\"{filename}\"");

    // Full-res original: never cache at the edge, so a takedown revokes access promptly.
    // Range requests still work under no-store; the client simply re-fetches each range.
    stream_media_file(
        &headers,
        &absolute,
        media.mime_type,
        &disposition,
        "no-store",
    )
    .await
}

/// Streaming access to an upload's compressed **preview** image. Gated exactly like
/// [`get_original`]: `find_visible_media` drops soft-deleted / ban-hidden uploads, so a
/// deleted or moderated post's preview 404s here even though the file may still be on
/// disk. Direct `/media/previews/**` is blocked in the router, making this the only path
/// to a preview.
///
/// Served inline (it's an `<img src>`) and privately cacheable for a short window. The
/// window is intentionally short (5 min) so a moderated image stops being served to a
/// direct-URL holder promptly; the live feed already evicts the card instantly via the
/// `upload-deleted` / `user-hidden` SSE events, so this only bounds the raw-URL edge case.
pub async fn get_preview(
    State(state): State<AppState>,
    headers: axum::http::HeaderMap,
    Path(upload_id): Path<Uuid>,
) -> Result<axum::response::Response, AppError> {
    let media = Upload::find_visible_media(&state.pool, upload_id)
        .await?
        .ok_or_else(|| AppError::NotFound("Upload nicht gefunden.".into()))?;
    let rel = media
        .preview_path
        .ok_or_else(|| AppError::NotFound("Vorschau nicht verfügbar.".into()))?;
    let absolute = state.config.media_path.join(&rel);
    stream_media_file(
        &headers,
        &absolute,
        "image/jpeg".to_string(),
        "inline",
        "private, max-age=300",
    )
    .await
}

/// Streaming access to an upload's big-screen **display** derivative (~2048px), used by the
/// diashow. Gated identically to [`get_preview`]. 404s when the derivative doesn't exist yet
/// (still compressing, or an old upload the backfill hasn't reached) — the diashow then falls
/// back to the original.
pub async fn get_display(
    State(state): State<AppState>,
    headers: axum::http::HeaderMap,
    Path(upload_id): Path<Uuid>,
) -> Result<axum::response::Response, AppError> {
    let media = Upload::find_visible_media(&state.pool, upload_id)
        .await?
        .ok_or_else(|| AppError::NotFound("Upload nicht gefunden.".into()))?;
    let rel = media
        .display_path
        .ok_or_else(|| AppError::NotFound("Anzeige nicht verfügbar.".into()))?;
    let absolute = state.config.media_path.join(&rel);
    stream_media_file(
        &headers,
        &absolute,
        "image/jpeg".to_string(),
        "inline",
        "private, max-age=300",
    )
    .await
}

/// Streaming access to an upload's **thumbnail** (video poster). Gated identically to
/// [`get_preview`].
pub async fn get_thumbnail(
    State(state): State<AppState>,
    headers: axum::http::HeaderMap,
    Path(upload_id): Path<Uuid>,
) -> Result<axum::response::Response, AppError> {
    let media = Upload::find_visible_media(&state.pool, upload_id)
        .await?
        .ok_or_else(|| AppError::NotFound("Upload nicht gefunden.".into()))?;
    let rel = media
        .thumbnail_path
        .ok_or_else(|| AppError::NotFound("Thumbnail nicht verfügbar.".into()))?;
    let absolute = state.config.media_path.join(&rel);
    stream_media_file(
        &headers,
        &absolute,
        "image/jpeg".to_string(),
        "inline",
        "private, max-age=300",
    )
    .await
}

#[cfg(test)]
mod tests {
    use super::{
        DISK_RESERVE_BYTES, MIN_QUOTA_LIMIT_BYTES, RangeSpec, parse_range, quota_limit_bytes,
    };

    // `Range` handling exists because iOS Safari probes every `<video>` with
    // `Range: bytes=0-1` and abandons the load without a 206. These pin the forms a
    // media element actually sends, plus the edges that decide 206 vs 200 vs 416.
    #[test]
    fn no_range_header_is_a_full_response() {
        assert_eq!(parse_range(None, 100), RangeSpec::Full);
    }

    #[test]
    fn open_ended_range_runs_to_eof() {
        assert_eq!(
            parse_range(Some("bytes=10-"), 100),
            RangeSpec::Partial { start: 10, end: 99 }
        );
    }

    #[test]
    fn closed_range_is_inclusive_on_both_ends() {
        // The iOS probe. Two bytes, 0 and 1 — an exclusive end would return one.
        assert_eq!(
            parse_range(Some("bytes=0-1"), 100),
            RangeSpec::Partial { start: 0, end: 1 }
        );
    }

    #[test]
    fn suffix_range_returns_the_last_n_bytes() {
        assert_eq!(
            parse_range(Some("bytes=-20"), 100),
            RangeSpec::Partial { start: 80, end: 99 }
        );
    }

    #[test]
    fn suffix_longer_than_the_file_clamps_to_the_whole_file() {
        assert_eq!(
            parse_range(Some("bytes=-500"), 100),
            RangeSpec::Partial { start: 0, end: 99 }
        );
    }

    #[test]
    fn end_past_eof_is_clamped_not_rejected() {
        // RFC 9110 §14.1.1 — players routinely ask for more than is there.
        assert_eq!(
            parse_range(Some("bytes=90-999"), 100),
            RangeSpec::Partial { start: 90, end: 99 }
        );
    }

    #[test]
    fn start_past_eof_is_416_not_a_full_body() {
        // Answering 200 here makes a player re-request forever.
        assert_eq!(
            parse_range(Some("bytes=100-"), 100),
            RangeSpec::Unsatisfiable
        );
        assert_eq!(
            parse_range(Some("bytes=200-300"), 100),
            RangeSpec::Unsatisfiable
        );
    }

    #[test]
    fn inverted_range_is_unsatisfiable() {
        assert_eq!(
            parse_range(Some("bytes=50-10"), 100),
            RangeSpec::Unsatisfiable
        );
    }

    #[test]
    fn unsupported_or_malformed_forms_fall_back_to_the_full_body() {
        // Ignoring a Range we can't process and sending 200 is explicitly allowed, and
        // safer than guessing. Multi-range would need multipart/byteranges, which no
        // <video> asks for.
        for header in [
            "bytes=0-10,20-30", // multi-range
            "items=0-10",       // non-bytes unit
            "bytes=abc-def",    // garbage
            "bytes=",           // empty spec
            "nonsense",         // no unit at all
        ] {
            assert_eq!(parse_range(Some(header), 100), RangeSpec::Full, "{header}");
        }
    }

    #[test]
    fn last_byte_is_reachable() {
        assert_eq!(
            parse_range(Some("bytes=99-99"), 100),
            RangeSpec::Partial { start: 99, end: 99 }
        );
    }

    #[test]
    fn empty_file_satisfies_no_range() {
        assert_eq!(parse_range(Some("bytes=0-"), 0), RangeSpec::Unsatisfiable);
        assert_eq!(parse_range(None, 0), RangeSpec::Full);
    }

    const GB: i64 = 1024 * 1024 * 1024;

    #[test]
    fn divides_free_space_by_uploaders_with_tolerance() {
        // 100 GB * 0.75 / 3 uploaders, well above the floor so the formula shows through.
        assert_eq!(quota_limit_bytes(100 * GB, 0.75, 3, 1), 26_843_545_600);
    }

    #[test]
    fn floors_fractional_results() {
        // 100 GB * 0.75 / 7 = 11_504_376_685.71… → truncated, not rounded.
        assert_eq!(quota_limit_bytes(100 * GB, 0.75, 7, 1), 11_504_376_685);
    }

    #[test]
    fn divisor_below_one_is_clamped_to_one() {
        // Guards against divide-by-zero when no one has uploaded yet.
        assert_eq!(quota_limit_bytes(10 * GB, 1.0, 0, 0), 10 * GB);
        assert_eq!(quota_limit_bytes(10 * GB, 1.0, -5, 0), 10 * GB);
    }

    /// The property the floor exists for: a guest's ceiling must not keep shrinking as more
    /// guests arrive. Same disk, 10 uploaders vs 1000 — the second must not be starved.
    #[test]
    fn the_ceiling_stops_falling_once_it_reaches_the_floor() {
        let ten = quota_limit_bytes(70 * GB, 0.75, 10, 1);
        let thousand = quota_limit_bytes(70 * GB, 0.75, 1000, 1);
        assert!(ten > MIN_QUOTA_LIMIT_BYTES, "10 uploaders should be roomy");
        assert_eq!(
            thousand, MIN_QUOTA_LIMIT_BYTES,
            "1000 uploaders (or 1000 fake accounts) must not drive the ceiling below the floor"
        );
        assert!(
            thousand >= 500 * 1024 * 1024,
            "the floor must still clear a single max-size video"
        );
    }

    /// `estimated_guest_count` is a FLOOR on the divisor, so the ceiling settles early instead
    /// of sliding down all evening as guests arrive.
    #[test]
    fn expected_headcount_holds_the_divisor_steady_while_guests_arrive() {
        let early = quota_limit_bytes(70 * GB, 0.75, 5, 100);
        let late = quota_limit_bytes(70 * GB, 0.75, 100, 100);
        assert_eq!(
            early, late,
            "the 5th guest and the 100th must see the same ceiling"
        );
    }

    /// THE INVARIANT THE UPLOAD GATE EXISTS FOR: if an upload is accepted, the keepsake must
    /// still be buildable afterwards.
    ///
    /// An earlier version of this test computed the gate's threshold and the preflight's
    /// threshold with the SAME expression and then asserted one against the other inside an
    /// `if` on that expression — a tautology that could not fail and would not have noticed a
    /// term being added to `ensure_export_space`. What actually binds the two together is that
    /// both call `required_free_bytes`, so what is worth pinning is the SHAPE of that function
    /// and the ceiling it produces on the real volume.
    ///
    /// Models the shipped box: 40 GB, ~5 GB consumed by OS, images and Postgres.
    #[test]
    fn the_gate_ceiling_keeps_both_keepsake_halves_and_the_reserve_affordable() {
        const USABLE: i64 = 35 * GB;
        let reserve = DISK_RESERVE_BYTES;

        // Walk the gallery upward and find the last size the gate would accept.
        let mut ceiling = 0i64;
        let step = 250 * 1024 * 1024;
        let mut media = 0i64;
        while media < USABLE {
            media += step;
            let free = USABLE - media;
            if free
                >= crate::services::export::required_free_bytes(media as u64, 2) as i64 + reserve
            {
                ceiling = media;
            }
        }

        // At the ceiling, BOTH archives and the reserve must genuinely fit in what is left.
        let free_at_ceiling = USABLE - ceiling;
        let both_halves = crate::services::export::required_free_bytes(ceiling as u64, 2) as i64;
        assert!(
            free_at_ceiling >= both_halves + reserve,
            "at the ceiling the keepsake ({both_halves}) + reserve ({reserve}) must fit in \
             {free_at_ceiling}"
        );

        // And one byte more must NOT fit — i.e. the ceiling is where the gate actually closes,
        // not somewhere short of it.
        let over = ceiling + step;
        let free_over = USABLE - over;
        assert!(
            free_over
                < crate::services::export::required_free_bytes(over as u64, 2) as i64 + reserve,
            "the gate should already be closed one step past the ceiling"
        );

        // Independently: `required_free_bytes` must charge for TWO gallery-sized archives.
        // If someone changes `armed` or the overhead, this is the line that notices.
        let one = crate::services::export::required_free_bytes(ceiling as u64, 1) as i64;
        assert_eq!(both_halves, one * 2, "a release arms both halves");

        // Pinned loosely so a deliberate change to the overhead or the reserve fails loudly
        // rather than silently moving the cliff.
        assert!(
            (6 * GB..=9 * GB).contains(&ceiling),
            "expected a gallery ceiling near 7.8 GiB on a 35 GB volume, got {ceiling} bytes"
        );
    }

    /// The gate must be the binding constraint, not the per-user floor. With 100 guests each
    /// allowed 500 MB, the per-user quota alone would authorise ~50 GB on a 40 GB disk.
    #[test]
    fn the_global_gate_binds_before_the_per_user_floor_can_overfill_the_disk() {
        const USABLE: i64 = 35 * GB;
        const GUESTS: i64 = 100;

        // Premise (constant, so `const`-asserted rather than pretending to be a test):
        // the per-user floor alone over-commits the volume, so something else must bind.
        const _: () = assert!(MIN_QUOTA_LIMIT_BYTES * 100 > 35 * GB);

        // The property. The previous version of this test asserted ONLY the premise above and
        // never touched the gate, so no change to `quota_limit_bytes`, `required_free_bytes` or
        // DISK_RESERVE_BYTES could have failed it.
        //
        // Take a moment where the volume holds 10 GB of media. Check both controls against the
        // SAME state and assert they disagree in the required direction: the per-user quota still
        // says yes, and the global gate already says no.
        let media: i64 = 10 * GB;
        let free = USABLE - media;

        // Per-user: a guest who has uploaded nothing is still granted a full floor-sized
        // allowance, because the formula divides free space and then applies the floor.
        let per_user = quota_limit_bytes(free, 0.75, GUESTS, GUESTS);
        assert!(
            per_user >= MIN_QUOTA_LIMIT_BYTES,
            "the per-user quota is expected to still be permissive here, got {per_user}"
        );

        // Global: the keepsake needs both halves plus the reserve, and they no longer fit.
        let required = crate::services::export::required_free_bytes(media as u64, 2) as i64
            + DISK_RESERVE_BYTES;
        assert!(
            free < required,
            "the global gate must already be closed at {media} bytes of media: free {free} \
             vs required {required}"
        );
    }

    /// The floor must never write a cheque the volume cannot cash — otherwise a full disk still
    /// hands out a 500 MB allowance and the filesystem Postgres needs fills up — and, the subtler
    /// half, it must never hand EVERY uploader the whole remaining budget.
    #[test]
    fn the_floor_never_exceeds_what_the_disk_can_back() {
        assert_eq!(quota_limit_bytes(0, 0.75, 3, 1), 0, "no disk, no allowance");

        // 400 MB free x 0.75 = 300 MB of budget, below the 500 MiB floor. The floor must NOT
        // apply: with 3 uploaders the answer is the divided share, not the whole budget.
        let tight = quota_limit_bytes(400 * 1024 * 1024, 0.75, 3, 1);
        assert_eq!(tight, 100 * 1024 * 1024, "a scarce budget is still divided");
        assert!(tight < MIN_QUOTA_LIMIT_BYTES);

        // The promise across all uploaders must stay inside the budget.
        assert!(
            tight * 3 <= (400.0 * 1024.0 * 1024.0 * 0.75) as i64,
            "the sum of per-user allowances must not exceed the disk's share"
        );
    }

    #[test]
    fn full_tolerance_is_identity_for_a_single_uploader() {
        assert_eq!(quota_limit_bytes(50 * GB, 1.0, 1, 1), 50 * GB);
    }

    /// The guard is the only reclaim mechanism that survives a client disconnect, so its three
    /// states have to be exactly right — a wrong `disarm` leaks bytes forever, a wrong `Drop`
    /// deletes a committed guest's photo.
    mod temp_file_guard {
        use super::super::TempFileGuard;

        fn scratch(name: &str) -> std::path::PathBuf {
            let dir = std::env::temp_dir().join(format!("es-guard-{}", std::process::id()));
            std::fs::create_dir_all(&dir).unwrap();
            let p = dir.join(name);
            std::fs::write(&p, b"bytes").unwrap();
            p
        }

        #[test]
        fn dropping_an_armed_guard_reclaims_the_file() {
            let p = scratch("armed.tmp");
            drop(TempFileGuard::new(p.clone()));
            assert!(
                !p.exists(),
                "an abandoned upload must not survive the request"
            );
        }

        #[test]
        fn a_disarmed_guard_leaves_the_file_alone() {
            let p = scratch("committed.jpg");
            let mut g = TempFileGuard::new(p.clone());
            g.disarm();
            drop(g);
            assert!(
                p.exists(),
                "a committed upload's bytes must never be deleted"
            );
        }

        #[test]
        fn retarget_follows_the_rename_and_forgets_the_old_path() {
            let old = scratch("old.tmp");
            let new = scratch("new.jpg");
            let mut g = TempFileGuard::new(old.clone());
            // The rename already moved the bytes; only the new path is at risk now.
            std::fs::remove_file(&old).unwrap();
            g.retarget(new.clone());
            drop(g);
            assert!(
                !new.exists(),
                "the final-named original is orphaned too until the row commits"
            );
        }

        #[test]
        fn a_guard_whose_file_is_already_gone_is_harmless() {
            let p = scratch("vanished.tmp");
            std::fs::remove_file(&p).unwrap();
            drop(TempFileGuard::new(p)); // must not panic
        }
    }

    /// The CSV hashtag path had no length check at all and was upserted tag-by-tag INSIDE the
    /// commit transaction — which holds a FOR SHARE lock on the event row, so one request could
    /// stall every other upload behind tens of thousands of round trips.
    mod hashtag_caps {
        use super::super::{MAX_HASHTAG_LENGTH, MAX_HASHTAGS_PER_UPLOAD, normalize_tags};

        #[test]
        fn a_huge_csv_is_capped_not_upserted_in_full() {
            let csv = (0..10_000)
                .map(|i| format!("tag{i}"))
                .collect::<Vec<_>>()
                .join(",");
            let tags = normalize_tags(vec![], Some(&csv));
            assert_eq!(tags.len(), MAX_HASHTAGS_PER_UPLOAD);
        }

        #[test]
        fn an_overlong_tag_is_dropped_rather_than_stored() {
            let long = "a".repeat(MAX_HASHTAG_LENGTH + 1);
            let tags = normalize_tags(vec![], Some(&format!("ok,{long}")));
            assert_eq!(tags, vec!["ok"]);
        }

        #[test]
        fn tags_are_normalised_and_deduped_across_both_sources() {
            // The upload path lowercased and stripped `#` while the edit path did not, so
            // `#Party` and `party` became two different hashtag rows. One helper, one rule.
            let tags = normalize_tags(vec!["party".into()], Some("#Party, PARTY ,tanz"));
            assert_eq!(tags, vec!["party", "tanz"]);
        }

        #[test]
        fn truncation_keeps_a_stable_prefix_not_an_arbitrary_one() {
            // Sorted before truncation, so the same input always yields the same tags —
            // otherwise an edit could silently shuffle which 30 survived.
            let csv = "zulu,alpha,mike,bravo";
            assert_eq!(
                normalize_tags(vec![], Some(csv)),
                normalize_tags(vec![], Some("bravo,mike,alpha,zulu"))
            );
        }

        #[test]
        fn an_empty_or_absent_csv_yields_nothing() {
            assert!(normalize_tags(vec![], None).is_empty());
            assert!(normalize_tags(vec![], Some(",, ,")).is_empty());
        }
    }
}