EventSnap/backend/src/handlers/host.rs

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

use crate::auth::middleware::RequireHost;
use crate::error::AppError;
use crate::models::comment::Comment;
use crate::models::event::Event;
use crate::models::session::Session;
use crate::models::upload::Upload;
use crate::models::user::UserRole;
use crate::services::export::Affects;
use crate::state::{AppState, SseEvent};

// ── DTOs ─────────────────────────────────────────────────────────────────────

#[derive(Serialize, sqlx::FromRow)]
pub struct UserSummary {
    pub id: Uuid,
    pub display_name: String,
    pub role: String,
    pub is_banned: bool,
    pub uploads_hidden: bool,
    pub upload_count: i64,
    pub total_upload_bytes: i64,
    pub created_at: DateTime<Utc>,
}

#[derive(Serialize)]
pub struct EventStatus {
    pub name: String,
    pub is_active: bool,
    pub uploads_locked: bool,
    pub export_released: bool,
    /// Free space on the volume the keepsake is written to. `None` when the mount can't be
    /// resolved — the UI hides the widget rather than rendering a confident zero.
    pub disk_free_bytes: Option<u64>,
    /// What a full keepsake build would need right now (both halves).
    pub keepsake_required_bytes: u64,
    /// Whether the host should be warned. See [`disk_is_low`].
    pub disk_low: bool,
}

/// Absolute floor below which free space is worth surfacing regardless of gallery size — the
/// threshold the README has carried on the roadmap since v1.
const LOW_DISK_FLOOR_BYTES: u64 = 10_000_000_000;

/// Is free space low enough that the host needs to know?
///
/// Two triggers, because a fixed threshold answers the wrong question. `postgres_data`,
/// `media_data` and `exports_data` are all Docker named volumes on one filesystem, so a full disk
/// does not degrade one subsystem — it stops Postgres writing and takes the event down. That is
/// what the absolute floor is for.
///
/// The second trigger is the one that actually earns its place: the keepsake needs room for two
/// gallery-sized archives, and the only moment a host can do anything about that is BEFORE they
/// release. Warning at "you could not build the keepsake right now" turns a post-event dead end
/// into a decision someone can still make.
fn disk_is_low(free: u64, keepsake_required: u64) -> bool {
    free < LOW_DISK_FLOOR_BYTES || free < keepsake_required
}

/// Count non-banned hosts/admins in the event OTHER than `excluding` — the operators
/// who would remain if `excluding` were demoted or banned. Used to enforce the "an event
/// always keeps at least one operator" floor.
async fn remaining_operators(
    state: &AppState,
    event_id: Uuid,
    excluding: Uuid,
) -> Result<i64, AppError> {
    let count = sqlx::query_scalar::<_, i64>(
        "SELECT COUNT(*) FROM \"user\"
         WHERE event_id = $1 AND id != $2
           AND role IN ('host', 'admin') AND is_banned = FALSE",
    )
    .bind(event_id)
    .bind(excluding)
    .fetch_one(&state.pool)
    .await?;
    Ok(count)
}

#[derive(Deserialize)]
pub struct SetRoleRequest {
    pub role: String,
}

// ── Handlers ─────────────────────────────────────────────────────────────────

pub async fn get_event_status(
    State(state): State<AppState>,
    RequireHost(_auth): RequireHost,
) -> Result<Json<EventStatus>, AppError> {
    let event = Event::find_by_slug(&state.pool, &state.config.event_slug)
        .await?
        .ok_or_else(|| AppError::NotFound("Event nicht gefunden.".into()))?;

    // Measured on the EXPORT volume, not the media one: that is where the cliff is, and it is a
    // distinct mount point even when both are backed by the same filesystem. The cached reading is
    // right here — this is advisory, polled on every dashboard load, and a 15s-stale number costs
    // nothing (unlike the export preflight, which reads uncached because it is about to write).
    let free = state
        .disk_cache
        .snapshot(&state.config.export_path)
        .map(|d| d.free);
    let keepsake_required_bytes =
        crate::services::export::keepsake_space_required(&state.pool, event.id)
            .await
            .unwrap_or(0);

    Ok(Json(EventStatus {
        name: event.name,
        is_active: event.is_active,
        uploads_locked: event.uploads_locked_at.is_some(),
        export_released: event.export_released_at.is_some(),
        disk_free_bytes: free,
        keepsake_required_bytes,
        // Unknown free space is NOT low. Fails open, exactly as the upload quota and the export
        // preflight do: a scary banner on an unreadable mount would train the host to ignore it.
        disk_low: free.is_some_and(|f| disk_is_low(f, keepsake_required_bytes)),
    }))
}

pub async fn list_users(
    State(state): State<AppState>,
    RequireHost(auth): RequireHost,
) -> Result<Json<Vec<UserSummary>>, AppError> {
    let rows = sqlx::query_as::<_, UserSummary>(
        "SELECT u.id,
                u.display_name,
                u.role::text AS role,
                u.is_banned,
                u.uploads_hidden,
                COALESCE(COUNT(up.id), 0) AS upload_count,
                u.total_upload_bytes,
                u.created_at
         FROM \"user\" u
         LEFT JOIN upload up ON up.user_id = u.id AND up.deleted_at IS NULL
         WHERE u.event_id = $1
         GROUP BY u.id
         ORDER BY u.created_at ASC",
    )
    .bind(auth.event_id)
    .fetch_all(&state.pool)
    .await?;

    Ok(Json(rows))
}

pub async fn ban_user(
    State(state): State<AppState>,
    RequireHost(auth): RequireHost,
    Path(user_id): Path<Uuid>,
) -> Result<StatusCode, AppError> {
    // The ban request carries no body — ban always hides (no per-request options).
    // Cannot ban yourself or another host/admin
    if user_id == auth.user_id {
        return Err(AppError::BadRequest(
            "Du kannst dich nicht selbst sperren.".into(),
        ));
    }
    let target = sqlx::query_as::<_, (String,)>(
        "SELECT role::text FROM \"user\" WHERE id = $1 AND event_id = $2",
    )
    .bind(user_id)
    .bind(auth.event_id)
    .fetch_optional(&state.pool)
    .await?
    .ok_or_else(|| AppError::NotFound("Benutzer nicht gefunden.".into()))?;

    if target.0 == "admin"
        || (target.0 == "host" && auth.role != crate::models::user::UserRole::Admin)
    {
        return Err(AppError::Forbidden(
            "Du kannst diesen Benutzer nicht sperren.".into(),
        ));
    }

    // Floor: never leave the event with zero operators. Banning removes the target from
    // the active-operator pool, so refuse if they're the last non-banned host/admin.
    if target.0 == "host" && remaining_operators(&state, auth.event_id, user_id).await? == 0 {
        return Err(AppError::BadRequest(
            "Der letzte Host kann nicht gesperrt werden.".into(),
        ));
    }

    // Ban ALWAYS hides: a banned user's content is "gone" everywhere. The visibility
    // views/queries now also filter on `is_banned` (defense in depth), and we set
    // `uploads_hidden` so the existing `user-hidden` live-eviction path fires too. The old
    // opt-out checkbox is gone — `hide_uploads` in the request is ignored.
    //
    // We deliberately do NOT revoke the banned user's sessions. This is a *read-only ban*
    // by design (USER_JOURNEYS §10.3): the user keeps read access to the feed and can still
    // download the released export — writes and host/admin actions are what the ban blocks
    // (enforced live on the write handlers + Require{Host,Admin}). Revoking sessions would
    // contradict that model, break the documented "banned guest can still download the
    // keepsake" flow, and be ineffective anyway (the user could just /recover a new session).
    //
    // The ban and the keepsake invalidation are ONE transaction — see `host_delete_upload`.
    let mut tx = state.pool.begin().await?;
    sqlx::query(
        "UPDATE \"user\"
         SET is_banned = TRUE, uploads_hidden = TRUE, uploads_hidden_at = NOW()
         WHERE id = $1 AND event_id = $2",
    )
    .bind(user_id)
    .bind(auth.event_id)
    .execute(&mut *tx)
    .await?;

    // A ban hides the user's uploads EVERYWHERE — and the keepsake is the place that matters most,
    // because it is the copy people keep. The export already filters `is_banned = FALSE`, so a
    // FUTURE export excludes them; without this, an ALREADY-RELEASED archive would keep serving a
    // banned user's photos forever. Same class as a takedown, so same treatment.
    let regen = crate::services::export::invalidate_and_arm(
        &mut tx,
        &state.config.event_slug,
        Affects::Both,
    )
    .await?;
    tx.commit().await?;
    if let Some(r) = regen {
        start_regen(&state, r);
    }

    // Evict their content live from every feed + the diashow so it disappears without
    // each viewer having to reload. (Their own SSE stream is separately dropped by the
    // is_banned revalidation in `sse::stream`, so they stop receiving live pushes while
    // retaining plain read access.)
    let _ = state.sse_tx.send(SseEvent::new(
        "user-hidden",
        serde_json::json!({ "user_id": user_id }).to_string(),
    ));

    tracing::info!(
        actor_user_id = %auth.user_id,
        target_user_id = %user_id,
        event_id = %auth.event_id,
        "host: ban_user"
    );

    Ok(StatusCode::NO_CONTENT)
}

pub async fn unban_user(
    State(state): State<AppState>,
    RequireHost(auth): RequireHost,
    Path(user_id): Path<Uuid>,
) -> Result<StatusCode, AppError> {
    // Mirror the ban guard: a host may only lift bans on guests, never on hosts or
    // admins. Without this a host could override an admin's ban of another host,
    // which is asymmetric with `ban_user` and lets a host escalate a peer back in.
    let target = sqlx::query_as::<_, (String,)>(
        "SELECT role::text FROM \"user\" WHERE id = $1 AND event_id = $2",
    )
    .bind(user_id)
    .bind(auth.event_id)
    .fetch_optional(&state.pool)
    .await?
    .ok_or_else(|| AppError::NotFound("Benutzer nicht gefunden.".into()))?;

    if target.0 == "admin"
        || (target.0 == "host" && auth.role != crate::models::user::UserRole::Admin)
    {
        return Err(AppError::Forbidden(
            "Du kannst diesen Benutzer nicht entsperren.".into(),
        ));
    }

    // Unban restores visibility too: ban set `uploads_hidden = TRUE`, so clearing only
    // `is_banned` would leave their content invisible. Clear all three (the timestamp too,
    // so a future ban stamps a fresh `uploads_hidden_at` the reconnect delta will replay).
    let mut tx = state.pool.begin().await?;
    let result = sqlx::query(
        "UPDATE \"user\"
         SET is_banned = FALSE, uploads_hidden = FALSE, uploads_hidden_at = NULL
         WHERE id = $1 AND event_id = $2",
    )
    .bind(user_id)
    .bind(auth.event_id)
    .execute(&mut *tx)
    .await?;
    if result.rows_affected() == 0 {
        return Err(AppError::NotFound("Benutzer nicht gefunden.".into()));
    }

    // The mirror of the ban case: an unban RESTORES their uploads to the export query, so an
    // already-released keepsake is now missing content it should contain. Rebuild it.
    let regen = crate::services::export::invalidate_and_arm(
        &mut tx,
        &state.config.event_slug,
        Affects::Both,
    )
    .await?;
    tx.commit().await?;
    if let Some(r) = regen {
        start_regen(&state, r);
    }

    tracing::info!(
        actor_user_id = %auth.user_id,
        target_user_id = %user_id,
        event_id = %auth.event_id,
        "host: unban_user"
    );
    Ok(StatusCode::NO_CONTENT)
}

/// Force a keepsake rebuild. The ESCAPE HATCH.
///
/// Without this, a failed or stranded export is terminal at runtime: `release_gallery` refuses an
/// already-released event ("bereits freigegeben"), `recover_exports` only runs at boot, and there is
/// no other retry path — so the host's only options were restarting the container or reopening the
/// event (which unlocks uploads to every guest and discards the release). This is also the recovery
/// path for a keepsake that went stale for any reason we haven't thought of.
pub async fn rebuild_export(
    State(state): State<AppState>,
    RequireHost(auth): RequireHost,
) -> Result<StatusCode, AppError> {
    let mut tx = state.pool.begin().await?;
    let regen = crate::services::export::invalidate_and_arm(
        &mut tx,
        &state.config.event_slug,
        Affects::Both,
    )
    .await?;
    tx.commit().await?;

    let Some(r) = regen else {
        return Err(AppError::BadRequest(
            "Die Galerie ist nicht freigegeben — es gibt nichts neu zu erzeugen.".into(),
        ));
    };

    // No debounce: this is an explicit, deliberate host action, not a burst.
    for export_type in ["zip", "html"] {
        let _ = state.sse_tx.send(SseEvent::new(
            "export-progress",
            serde_json::json!({ "type": export_type, "progress_pct": 0 }).to_string(),
        ));
    }
    crate::services::export::spawn_export_jobs(
        r.event_id,
        r.event_name,
        r.epoch,
        state.config.comments_enabled,
        std::time::Duration::ZERO,
        state.pool.clone(),
        state.config.media_path.clone(),
        state.config.export_path.clone(),
        state.sse_tx.clone(),
    );

    tracing::info!(actor_user_id = %auth.user_id, "host: rebuild_export");
    Ok(StatusCode::NO_CONTENT)
}

pub async fn set_role(
    State(state): State<AppState>,
    RequireHost(auth): RequireHost,
    Path(user_id): Path<Uuid>,
    Json(body): Json<SetRoleRequest>,
) -> Result<StatusCode, AppError> {
    if user_id == auth.user_id {
        return Err(AppError::BadRequest(
            "Du kannst deine eigene Rolle nicht ändern.".into(),
        ));
    }
    let new_role = match body.role.as_str() {
        "guest" => "guest",
        "host" => "host",
        _ => {
            return Err(AppError::BadRequest(
                "Ungültige Rolle. Erlaubt: guest, host.".into(),
            ));
        }
    };

    // Look up the current role so we can apply the host-vs-admin guard. A plain host may
    // promote/demote GUESTS only; it may not change any host's or admin's role (see the
    // guard below — this closes the demote-a-peer-host→ban/PIN-reset takeover chain, F1).
    // Only an admin may change a host's role. Admins may do anything except change
    // themselves (blocked above).
    let target = sqlx::query_as::<_, (String,)>(
        "SELECT role::text FROM \"user\" WHERE id = $1 AND event_id = $2",
    )
    .bind(user_id)
    .bind(auth.event_id)
    .fetch_optional(&state.pool)
    .await?
    .ok_or_else(|| AppError::NotFound("Benutzer nicht gefunden.".into()))?;

    // Admins are untouchable by hosts. A plain host also may not demote another
    // *host*: without this guard a host could demote a peer host to guest and then
    // ban / PIN-reset (→ account-takeover via /recover) them — the ban/pin-reset peer
    // guards key off the target's *current* role, so a prior demotion would launder
    // past them. Only an admin may change a host's role.
    if target.0 == "admin" || (target.0 == "host" && auth.role != UserRole::Admin) {
        return Err(AppError::Forbidden(
            "Du darfst die Rolle dieses Benutzers nicht ändern.".into(),
        ));
    }

    // Floor: demoting the last non-banned host/admin to guest would leave the event with
    // no operator. Refuse.
    if new_role == "guest"
        && target.0 == "host"
        && remaining_operators(&state, auth.event_id, user_id).await? == 0
    {
        return Err(AppError::BadRequest(
            "Der letzte Host kann nicht zum Gast gemacht werden.".into(),
        ));
    }

    sqlx::query("UPDATE \"user\" SET role = $2::user_role WHERE id = $1 AND event_id = $3")
        .bind(user_id)
        .bind(new_role)
        .bind(auth.event_id)
        .execute(&state.pool)
        .await?;
    tracing::info!(
        actor_user_id = %auth.user_id,
        target_user_id = %user_id,
        event_id = %auth.event_id,
        old_role = %target.0,
        new_role,
        "host: set_role"
    );
    Ok(StatusCode::NO_CONTENT)
}

#[derive(Serialize)]
pub struct PinResetResponse {
    /// Plaintext PIN — shown to the operator **once**. Never persisted client-side.
    pub pin: String,
}

/// Generate a fresh PIN for another user, returning the plaintext exactly once.
///
/// Authorisation:
/// - Host caller → may reset **guest** PINs only.
/// - Admin caller → may reset **guest** and **host** PINs (never another admin).
/// - Target ≠ caller.
pub async fn reset_user_pin(
    State(state): State<AppState>,
    RequireHost(auth): RequireHost,
    Path(user_id): Path<Uuid>,
) -> Result<Json<PinResetResponse>, AppError> {
    use rand::Rng;

    if user_id == auth.user_id {
        return Err(AppError::BadRequest(
            "Du kannst deine eigene PIN nicht über diese Funktion zurücksetzen.".into(),
        ));
    }

    let target = sqlx::query_as::<_, (String,)>(
        "SELECT role::text FROM \"user\" WHERE id = $1 AND event_id = $2",
    )
    .bind(user_id)
    .bind(auth.event_id)
    .fetch_optional(&state.pool)
    .await?
    .ok_or_else(|| AppError::NotFound("Benutzer nicht gefunden.".into()))?;

    match (auth.role.clone(), target.0.as_str()) {
        (UserRole::Admin, "guest" | "host") => {}
        (UserRole::Host, "guest") => {}
        _ => {
            return Err(AppError::Forbidden(
                "Du darfst die PIN dieses Benutzers nicht zurücksetzen.".into(),
            ));
        }
    }

    let pin: String = format!("{:04}", rand::rng().random_range(0..10000u32));
    let pin_hash = crate::auth::handlers::hash_password(pin.clone(), 12).await?;

    sqlx::query(
        "UPDATE \"user\"
         SET recovery_pin_hash = $1,
             failed_pin_attempts = 0,
             pin_locked_until = NULL
         WHERE id = $2 AND event_id = $3",
    )
    .bind(&pin_hash)
    .bind(user_id)
    .bind(auth.event_id)
    .execute(&state.pool)
    .await?;

    // A PIN reset means the old credential is compromised/forgotten — revoke every
    // existing session so old devices must re-authenticate with the new PIN. This is a
    // security-relevant revoke: if it fails, the old sessions stay valid (sessions are
    // token- not PIN-bound), so surface the error in logs rather than swallowing it
    // silently while reporting success to the host.
    if let Err(e) = Session::delete_all_for_user(&state.pool, user_id).await {
        tracing::error!(error = ?e, user_id = %user_id, "PIN reset: failed to revoke sessions");
    }

    // Resolve any pending in-app "I forgot my PIN" request for this user.
    let _ = sqlx::query("DELETE FROM pin_reset_request WHERE user_id = $1")
        .bind(user_id)
        .execute(&state.pool)
        .await;

    // Notify the *recipient* device(s) if they happen to be online so they can clear
    // their cached local PIN. They'll save the new one on the next /recover.
    let _ = state.sse_tx.send(SseEvent::new(
        "pin-reset",
        serde_json::json!({ "user_id": user_id }).to_string(),
    ));

    tracing::info!(
        actor_user_id = %auth.user_id,
        target_user_id = %user_id,
        event_id = %auth.event_id,
        "host: reset_user_pin"
    );

    Ok(Json(PinResetResponse { pin }))
}

#[derive(Serialize, sqlx::FromRow)]
pub struct PinResetRequestSummary {
    pub id: Uuid,
    pub user_id: Uuid,
    pub display_name: String,
    pub created_at: DateTime<Utc>,
}

/// List pending in-app PIN-reset requests so a host can action them (via the existing
/// `reset_user_pin`, which also clears the request).
pub async fn list_pin_reset_requests(
    State(state): State<AppState>,
    RequireHost(auth): RequireHost,
) -> Result<Json<Vec<PinResetRequestSummary>>, AppError> {
    let rows = sqlx::query_as::<_, PinResetRequestSummary>(
        "SELECT r.id, r.user_id, u.display_name, r.created_at
         FROM pin_reset_request r
         JOIN \"user\" u ON u.id = r.user_id
         WHERE r.event_id = $1
         ORDER BY r.created_at ASC",
    )
    .bind(auth.event_id)
    .fetch_all(&state.pool)
    .await?;
    Ok(Json(rows))
}

/// Dismiss a PIN-reset request without resetting (e.g. the host couldn't verify the
/// requester's identity).
pub async fn dismiss_pin_reset_request(
    State(state): State<AppState>,
    RequireHost(auth): RequireHost,
    Path(id): Path<Uuid>,
) -> Result<StatusCode, AppError> {
    sqlx::query("DELETE FROM pin_reset_request WHERE id = $1 AND event_id = $2")
        .bind(id)
        .bind(auth.event_id)
        .execute(&state.pool)
        .await?;
    Ok(StatusCode::NO_CONTENT)
}

/// Content changed AFTER the gallery was released — regenerate the keepsake.
///
/// Deleting a photo used to remove it from the live feed but leave it in the already-generated
/// archive FOREVER: the old `if ready { continue }` skip guaranteed the export was never rebuilt.
/// For a takedown ("please remove my photo") that is the one place it most needs to disappear.
///
/// Bumping the epoch (while staying released) retires the current generation, so the stale archive
/// stops being downloadable the instant the delete commits, and a fresh worker rebuilds it without
/// the removed content. The download 404s in the meantime, which is the correct answer — serving
/// the old archive would serve the deleted photo.
/// Start the workers for a regeneration that was armed inside a (now-committed) transaction, and
/// tell every client the current keepsake just became undownloadable.
///
/// The SSE matters: bumping the epoch retires the archive INSTANTLY, so `/export/zip` starts 404ing
/// the moment the change commits. Without a nudge, a guest sitting on `/export` keeps rendering an
/// enabled "download" button for the whole rebuild. Both the nav badge and the export page already
/// refetch `/export/status` on `export-progress`, so a 0% tick is the cheapest correct signal.
pub fn start_regen(state: &AppState, regen: crate::services::export::PendingRegen) {
    for export_type in ["zip", "html"] {
        let _ = state.sse_tx.send(SseEvent::new(
            "export-progress",
            serde_json::json!({ "type": export_type, "progress_pct": 0 }).to_string(),
        ));
    }
    crate::services::export::spawn_export_jobs(
        regen.event_id,
        regen.event_name,
        regen.epoch,
        state.config.comments_enabled,
        // Debounced: a takedown pass is a burst, and each request retires the last generation. The
        // delay lets superseded workers fail their claim and do zero work instead of each building
        // a full archive. See export::REGEN_DEBOUNCE.
        crate::services::export::REGEN_DEBOUNCE,
        state.pool.clone(),
        state.config.media_path.clone(),
        state.config.export_path.clone(),
        state.sse_tx.clone(),
    );
}

pub async fn host_delete_upload(
    State(state): State<AppState>,
    RequireHost(auth): RequireHost,
    Path(upload_id): Path<Uuid>,
) -> Result<StatusCode, AppError> {
    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()))?;

    // The delete and the keepsake invalidation are ONE transaction: if the delete committed and the
    // invalidation didn't, the taken-down photo would stay downloadable forever and nothing would
    // notice (the keepsake still looks complete, and the host can no longer find the upload to retry).
    let mut tx = state.pool.begin().await?;
    let deleted = Upload::soft_delete_in_event(&mut tx, upload_id, auth.event_id).await?;
    if !deleted {
        return Err(AppError::NotFound("Upload nicht gefunden.".into()));
    }
    let regen = crate::services::export::invalidate_and_arm(
        &mut tx,
        &state.config.event_slug,
        Affects::Both,
    )
    .await?;
    tx.commit().await?;

    let _ = state.sse_tx.send(SseEvent::new(
        "upload-deleted",
        serde_json::json!({ "upload_id": upload.id }).to_string(),
    ));
    if let Some(r) = regen {
        start_regen(&state, r);
    }

    tracing::info!(
        actor_user_id = %auth.user_id,
        event_id = %auth.event_id,
        upload_id = %upload.id,
        "host: host_delete_upload"
    );

    Ok(StatusCode::NO_CONTENT)
}

pub async fn host_delete_comment(
    State(state): State<AppState>,
    RequireHost(auth): RequireHost,
    Path(comment_id): Path<Uuid>,
) -> Result<StatusCode, AppError> {
    let mut tx = state.pool.begin().await?;
    let deleted = Comment::soft_delete_in_event(&mut tx, comment_id, auth.event_id).await?;
    if !deleted {
        return Err(AppError::NotFound("Kommentar nicht gefunden.".into()));
    }
    // Only the HTML viewer embeds comments — the ZIP is media-only, so it is carried forward rather
    // than rebuilt. Otherwise moderating one comment would 404 the photo download for minutes to
    // change nothing in it.
    let regen = crate::services::export::invalidate_and_arm(
        &mut tx,
        &state.config.event_slug,
        Affects::ViewerOnly,
    )
    .await?;
    tx.commit().await?;

    let _ = state.sse_tx.send(SseEvent::new(
        "comment-deleted",
        serde_json::json!({ "comment_id": comment_id }).to_string(),
    ));
    if let Some(r) = regen {
        start_regen(&state, r);
    }
    tracing::info!(
        actor_user_id = %auth.user_id,
        event_id = %auth.event_id,
        comment_id = %comment_id,
        "host: host_delete_comment"
    );
    Ok(StatusCode::NO_CONTENT)
}

pub async fn close_event(
    State(state): State<AppState>,
    RequireHost(_auth): RequireHost,
) -> Result<StatusCode, AppError> {
    let result = sqlx::query(
        "UPDATE event SET uploads_locked_at = NOW() WHERE slug = $1 AND uploads_locked_at IS NULL",
    )
    .bind(&state.config.event_slug)
    .execute(&state.pool)
    .await?;

    // Only broadcast when this call actually flipped the lock — closing an
    // already-closed event is a no-op and shouldn't spam listeners.
    if result.rows_affected() > 0 {
        let _ = state.sse_tx.send(SseEvent::new("event-closed", "{}"));
    }

    Ok(StatusCode::NO_CONTENT)
}

pub async fn open_event(
    State(state): State<AppState>,
    RequireHost(_auth): RequireHost,
) -> Result<StatusCode, AppError> {
    // Reopening invalidates any prior release: the keepsake was snapshotted at release time, so
    // allowing new uploads afterwards would silently diverge the live feed from the frozen export.
    //
    // ONE statement retires the entire export generation. Bumping `export_epoch` in the same write
    // that clears `export_released_at` instantly invalidates every in-flight worker, every `done`
    // row and all readiness — because readiness is DERIVED from this epoch (migration 014), not
    // stored. There is no export_job row to touch and nothing to keep in sync, so this needs no
    // transaction. Any worker still streaming holds the old epoch and is now inert: its
    // epoch-guarded finalize matches nothing, and it discards its own output.
    let result = sqlx::query(
        "UPDATE event
         SET uploads_locked_at  = NULL,
             export_released_at = NULL,
             export_epoch       = export_epoch + 1
         WHERE slug = $1 AND (uploads_locked_at IS NOT NULL OR export_released_at IS NOT NULL)",
    )
    .bind(&state.config.event_slug)
    .execute(&state.pool)
    .await?;

    if result.rows_affected() > 0 {
        let _ = state.sse_tx.send(SseEvent::new("event-opened", "{}"));
    }

    Ok(StatusCode::NO_CONTENT)
}

pub async fn release_gallery(
    State(state): State<AppState>,
    RequireHost(_auth): RequireHost,
) -> Result<StatusCode, AppError> {
    // The release claim, the epoch bump, the upload lock and BOTH job rows are written in ONE
    // transaction. Two reasons, both of which were live bugs:
    //
    //  1. Cancellation. This handler used to commit `export_released_at` and only THEN await the
    //     enqueue. Axum drops the handler future when the client disconnects (closed tab, proxy
    //     timeout), which left the event released and uploads locked with ZERO export_job rows and
    //     no workers: downloads 404 forever and the host cannot retry, because release_gallery
    //     rejects an already-released event ("bereits freigegeben"). Only a restart escaped it.
    //     Now nothing is committed until every row is written, and the workers are spawned AFTER
    //     the commit (a detached `tokio::spawn` survives cancellation).
    //  2. Atomicity vs. a concurrent reopen. Bumping the epoch in the same statement that sets
    //     `export_released_at` means no worker and no reader can ever observe "released again but
    //     the generation hasn't moved on yet" — the window every previous fix kept leaving open.
    //
    // Release also locks uploads in the same statement (release ⇒ lock), so the export snapshot is
    // taken against a frozen upload set. `COALESCE` preserves an earlier explicit lock time.
    let mut tx = state.pool.begin().await?;

    let claimed: Option<(Uuid, String, i64)> = sqlx::query_as(
        "UPDATE event
         SET export_released_at = NOW(),
             uploads_locked_at  = COALESCE(uploads_locked_at, NOW()),
             export_epoch       = export_epoch + 1
         WHERE slug = $1 AND export_released_at IS NULL
         RETURNING id, name, export_epoch",
    )
    .bind(&state.config.event_slug)
    .fetch_optional(&mut *tx)
    .await?;

    let Some((event_id, event_name, epoch)) = claimed else {
        // Distinguish "no such event" from "already released" for a clean error.
        let exists = Event::find_by_slug(&state.pool, &state.config.event_slug)
            .await?
            .is_some();
        return Err(if exists {
            AppError::BadRequest("Galerie wurde bereits freigegeben.".into())
        } else {
            AppError::NotFound("Event nicht gefunden.".into())
        });
    };

    // Arm both types at THIS epoch. No "skip the type that's already ready" check any more: the
    // epoch bump above retired every prior generation, so there is nothing to preserve and nothing
    // to be fooled by. (That skip was how a stale ready flag used to suppress regeneration.)
    crate::services::export::enqueue_jobs_at_epoch(&mut tx, event_id, epoch).await?;

    tx.commit().await?;

    // Release locks uploads too — tell any open composer to flip to the locked UI live rather than
    // discovering it via a rejected upload.
    let _ = state.sse_tx.send(SseEvent::new("event-closed", "{}"));

    // Detached — survives this handler being cancelled.
    crate::services::export::spawn_export_jobs(
        event_id,
        event_name,
        epoch,
        state.config.comments_enabled,
        std::time::Duration::ZERO,
        state.pool.clone(),
        state.config.media_path.clone(),
        state.config.export_path.clone(),
        state.sse_tx.clone(),
    );

    Ok(StatusCode::NO_CONTENT)
}

#[cfg(test)]
mod tests {
    use super::{LOW_DISK_FLOOR_BYTES, disk_is_low};

    const GB: u64 = 1_000_000_000;

    #[test]
    fn a_healthy_disk_with_room_for_the_keepsake_is_not_low() {
        assert!(!disk_is_low(40 * GB, 25 * GB));
    }

    #[test]
    fn the_absolute_floor_fires_even_when_the_gallery_is_tiny() {
        // All three volumes share one filesystem, so running out doesn't degrade one subsystem —
        // Postgres stops being able to write and the event goes down. A 1 GB gallery would clear
        // the keepsake test comfortably; the floor is what catches this.
        assert!(disk_is_low(5 * GB, GB));
        assert!(disk_is_low(LOW_DISK_FLOOR_BYTES - 1, 0));
        assert!(!disk_is_low(LOW_DISK_FLOOR_BYTES, 0));
    }

    #[test]
    fn plenty_of_space_is_still_low_when_the_keepsake_would_not_fit() {
        // THE case the fixed threshold misses, and the one that matters: 30 GB free is nowhere near
        // any floor, but a 30 GB gallery needs room for TWO archives. The host can act on this
        // before releasing; after releasing, they cannot.
        assert!(disk_is_low(30 * GB, 66 * GB));
    }

    #[test]
    fn the_keepsake_trigger_is_exact_at_the_boundary() {
        assert!(!disk_is_low(66 * GB, 66 * GB), "exactly enough is enough");
        assert!(disk_is_low(66 * GB - 1, 66 * GB));
    }

    #[test]
    fn an_empty_gallery_needs_nothing_and_only_the_floor_applies() {
        assert!(!disk_is_low(11 * GB, 0));
        assert!(disk_is_low(9 * GB, 0));
    }
}