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,
}

/// 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.
///
/// IT MUST FIRE BEFORE THE UPLOAD GATE CLOSES, and that is why the reserve and the margin are
/// here. The gate in `handlers::upload` refuses at
/// `free < keepsake_required + DISK_RESERVE_BYTES + UPLOAD_GATE_HEADROOM_BYTES`;
/// warning at `free < keepsake_required` alone meant the two differed by the whole reserve, so
/// the wall was always hit FIRST. Every guest would be blocked from uploading while this
/// dashboard showed a comfortable disk and no banner at all — on the shipped 40 GB box, uploads
/// stopping with ~27 GB free and nothing on screen to explain it, with no operator present.
///
/// The 25% margin makes it a warning rather than an obituary: the host sees it while there is
/// still room to act (delete a few large videos, which refunds immediately and reopens the gate).
fn disk_is_low(free: u64, keepsake_required: u64) -> bool {
    // Mirrors the gate EXACTLY, headroom included. The gate now demands
    // `UPLOAD_GATE_HEADROOM_BYTES` more than the export preflight does, so that ordinary
    // end-of-night writes cannot flip the preflight after uploads have already stopped. Leaving
    // that term out here would shrink the warning's lead by 1.5 GB — and the whole point of this
    // function is that the banner must appear while the host can still act.
    let gate_closes_at = keepsake_required
        .saturating_add(crate::handlers::upload::DISK_RESERVE_BYTES as u64)
        .saturating_add(crate::handlers::upload::UPLOAD_GATE_HEADROOM_BYTES as u64);
    let warn_at = gate_closes_at.saturating_add(gate_closes_at / 4);
    // No separate absolute-floor clause. There used to be `free < LOW_DISK_FLOOR_BYTES ||` here,
    // and it was unreachable: `gate_closes_at` is at least DISK_RESERVE_BYTES, so `warn_at` is at
    // least 1.25x it (12.5 GB) — always above the 10 GB floor. Two tests were named after that
    // clause and neither could fail if it were deleted. Keeping dead code that tests claim to
    // cover is worse than not having it.
    free < warn_at
}

/// 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.
///
/// Takes a CONNECTION, not the pool, and every caller passes the same transaction it is about to
/// write in — after taking [`lock_operator_floor`]. Read on the pool beforehand, this count was a
/// snapshot that any concurrent operator-removing action could invalidate before the UPDATE landed:
/// an admin demoting host B while host A calls `DELETE /me` saw two independent checks each observe
/// the other still present, both commit, and the event end up with zero operators — which is not
/// recoverable from inside the app, since appointing an operator requires being one.
async fn remaining_operators(
    conn: &mut sqlx::PgConnection,
    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(conn)
    .await?;
    Ok(count)
}

/// Serialise every action that can remove an operator from an event.
///
/// The same key `me::delete_account` takes — namespace 4242, `hashtext(event_id)` — and it MUST
/// stay identical, or the two families of caller lock against nothing. An advisory lock is used
/// rather than a row lock because it is a separate lock space and so cannot join the
/// `event`/`user` row-lock graph that moderation traffic already traverses in both directions;
/// it is released automatically when the transaction ends.
///
/// **Call this FIRST in the transaction, before taking any row lock.** Being a separate lock space
/// means it cannot form a cycle *with itself*, not that ordering is free: all three callers go on
/// to lock `user` and `event` rows, so a caller that took those rows first and reached for this
/// lock afterwards would deadlock against one that did it the other way round. Postgres would
/// break the tie by killing one transaction with a 500. Every caller acquires it first; keep it
/// that way.
pub(crate) async fn lock_operator_floor(
    conn: &mut sqlx::PgConnection,
    event_id: Uuid,
) -> Result<(), AppError> {
    sqlx::query("SELECT pg_advisory_xact_lock(4242, hashtext($1::text))")
        .bind(event_id)
        .execute(conn)
        .await?;
    Ok(())
}

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

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

    // 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.
    //
    // INSIDE the transaction and behind the operator lock — see `remaining_operators`. Checked on
    // the pool beforehand, this raced `set_role` and `DELETE /me` into an event with no operator.
    if target.0 == "host" {
        lock_operator_floor(&mut tx, auth.event_id).await?;
        if remaining_operators(&mut tx, auth.event_id, user_id).await? == 0 {
            return Err(AppError::BadRequest(
                "Der letzte Host kann nicht gesperrt werden.".into(),
            ));
        }
    }

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

    crate::services::audit::record(
        &state.pool,
        auth.event_id,
        auth.user_id,
        None,
        auth.role.clone(),
        "ban_user",
        Some(user_id),
        None,
        None,
    )
    .await;

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

    // The exact mirror of `ban_user`'s `user-hidden`, and it was missing entirely: every open
    // feed and the unattended projector kept the guest evicted until somebody reloaded the page
    // by hand. Meanwhile the host's own confirm copy promises the photos "come back to the
    // gallery, die Diashow und den Export" — so the one surface that would have shown the host
    // their action had worked showed the opposite.
    //
    // Also the signal a banned guest's upload queue waits on: their queued photos parked with
    // the blob intact rather than being purged (see `AppError::UserBanned`), and this is what
    // releases them.
    let _ = state.sse_tx.send(SseEvent::new(
        "user-shown",
        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: unban_user"
    );

    crate::services::audit::record(
        &state.pool,
        auth.event_id,
        auth.user_id,
        None,
        auth.role.clone(),
        "unban_user",
        Some(user_id),
        None,
        None,
    )
    .await;

    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.
    //
    // The check and the UPDATE are ONE transaction, behind the operator lock — see
    // `remaining_operators`. Split apart on the pool, this raced `ban_user` and `DELETE /me`.
    let mut tx = state.pool.begin().await?;
    if new_role == "guest" && target.0 == "host" {
        lock_operator_floor(&mut tx, auth.event_id).await?;
        if remaining_operators(&mut tx, 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(&mut *tx)
        .await?;
    tx.commit().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"
    );

    crate::services::audit::record(
        &state.pool,
        auth.event_id,
        auth.user_id,
        None,
        auth.role.clone(),
        "set_role",
        Some(user_id),
        None,
        None,
    )
    .await;
    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"
    );

    crate::services::audit::record(
        &state.pool,
        auth.event_id,
        auth.user_id,
        None,
        auth.role.clone(),
        "reset_pin",
        Some(user_id),
        None,
        None,
    )
    .await;

    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.
        //
        // Measured from the START of the burst, not from this request — a fixed per-request delay
        // meant a steady stream of invalidations faster than one per 20s deferred the build
        // forever, leaving the keepsake permanently 404 and the UI stuck on "Wird vorbereitet…".
        // See export::regen_delay_for.
        crate::services::export::regen_delay_for(regen.event_id),
        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?;
    // `by_host: true` — the takedown holds the uploader's idempotency key so a late retry from
    // their queue cannot resurrect the photo. See migration 031.
    let deleted = Upload::soft_delete_in_event(&mut tx, upload_id, auth.event_id, true).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"
    );

    crate::services::audit::record(
        &state.pool,
        auth.event_id,
        auth.user_id,
        None,
        auth.role.clone(),
        "delete_upload",
        Some(upload_id),
        None,
        None,
    )
    .await;

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

    crate::services::audit::record(
        &state.pool,
        auth.event_id,
        auth.user_id,
        None,
        auth.role.clone(),
        "delete_comment",
        Some(comment_id),
        None,
        None,
    )
    .await;
    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", "{}"));
    }

    // Was logged NOWHERE at all before this — not even a tracing line. A host reading
    // the record the morning after had no way to see when uploads were locked or the
    // gallery released, which are the two actions that change what every guest can do.
    crate::services::audit::record(
        &state.pool,
        auth.event_id,
        auth.user_id,
        None,
        auth.role.clone(),
        "lock_uploads",
        None,
        None,
        None,
    )
    .await;
    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", "{}"));
    }

    // Was logged NOWHERE at all before this — not even a tracing line. A host reading
    // the record the morning after had no way to see when uploads were locked or the
    // gallery released, which are the two actions that change what every guest can do.
    crate::services::audit::record(
        &state.pool,
        auth.event_id,
        auth.user_id,
        None,
        auth.role.clone(),
        "unlock_uploads",
        None,
        None,
        None,
    )
    .await;
    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.
    //
    // SPAWNED IMMEDIATELY AFTER THE COMMIT, BEFORE ANY OTHER `.await`. Every `invalidate_and_arm`
    // call site does this; `me::delete_account` carries the same note. The audit write below used
    // to sit here, and it is two pool round-trips that can each wait up to the 5 s acquire timeout
    // — right at the moment `event-closed` has just fanned out to ~100 phones whose queues all hit
    // the API at once, so the pool is as contended as it ever gets. Drop the handler future during
    // that suspension (the host's phone sleeps, the tab closes, Caddy times the request out) and
    // the task never spawns: the event is released, uploads are locked, both `export_job` rows sit
    // `pending` at the live epoch, and no worker exists. `/export/*` 404s, the page sits on "Wird
    // vorbereitet…", `recover_exports` only runs at boot, and `release_gallery` refuses a retry
    // because the gallery is already released.
    //
    // This is the one path that arms the FIRST build of the keepsake, so it is the worst possible
    // place to reintroduce that window.
    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(),
    );

    // Was logged NOWHERE at all before this — not even a tracing line. A host reading
    // the record the morning after had no way to see when uploads were locked or the
    // gallery released, which are the two actions that change what every guest can do.
    //
    // Last, deliberately: it is best-effort by design (it swallows its own errors), so nothing
    // downstream may depend on it having completed.
    crate::services::audit::record(
        &state.pool,
        auth.event_id,
        auth.user_id,
        None,
        auth.role.clone(),
        "release_gallery",
        None,
        None,
        None,
    )
    .await;

    Ok(StatusCode::NO_CONTENT)
}

#[cfg(test)]
mod tests {
    use super::disk_is_low;
    use crate::handlers::upload::{DISK_RESERVE_BYTES, UPLOAD_GATE_HEADROOM_BYTES};
    use crate::services::export::required_free_bytes;

    const GB: u64 = 1_000_000_000;

    #[test]
    fn a_healthy_disk_with_room_for_the_keepsake_is_not_low() {
        // Room for the keepsake AND the reserve the upload gate holds back, with margin.
        assert!(!disk_is_low(60 * GB, 25 * GB));
    }

    /// Renamed from `the_absolute_floor_fires_...`: there is no separate floor clause any more
    /// (see `disk_is_low`). What still has to hold is the behaviour the floor was there FOR — a
    /// nearly-empty disk is low even when the gallery is small enough that the keepsake term
    /// alone would clear it, because all three volumes share one filesystem and Postgres needs
    /// room to write.
    #[test]
    fn a_nearly_empty_disk_is_low_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(9 * GB, 0));
        assert!(!disk_is_low(20 * GB, 0), "a roomy empty disk is not low");
    }

    /// THE PROPERTY THIS EXISTS FOR: the host must be warned BEFORE guests are blocked.
    ///
    /// `handlers::upload` refuses at `free < keepsake_required + DISK_RESERVE_BYTES`. If the
    /// banner fires only at or below that, the host's first signal is 100 guests being unable
    /// to upload while the dashboard shows a comfortable disk — with nobody on site to ask.
    #[test]
    fn the_banner_always_fires_before_the_upload_gate_closes() {
        // Asserting `disk_is_low(gate_closes_at, required)` is what this used to do, and it was a
        // tautology: `disk_is_low` recomputes the same `gate_closes_at` internally and compares
        // against `gate + gate/4`, so the assertion reduced to `G < G + G/4` — true for every G,
        // for any margin, even a margin of zero. It could not detect the banner being moved to
        // exactly the gate, which is the regression it is named for.
        //
        // So pin the GAP instead: find the free-space level at which the banner starts, and
        // require it to be strictly above the level at which the gate closes, by a usable amount.
        for media_gb in [0u64, 1, 4, 8, 16, 32] {
            let required = required_free_bytes(media_gb * GB, 2);
            let gate_closes_at =
                required + DISK_RESERVE_BYTES as u64 + UPLOAD_GATE_HEADROOM_BYTES as u64;

            // Just above the gate: guests can still upload, and the host must already be warned.
            assert!(
                disk_is_low(gate_closes_at + 1, required),
                "at media={media_gb}GB the banner is not yet showing while the gate still allows uploads"
            );

            // The warning must lead by a margin the host can act inside, not by one byte.
            let mut warn_starts_at = gate_closes_at;
            while disk_is_low(warn_starts_at, required) {
                warn_starts_at += GB / 10;
            }
            assert!(
                warn_starts_at >= gate_closes_at + gate_closes_at / 5,
                "at media={media_gb}GB the banner leads the gate by only                  {} bytes",
                warn_starts_at - gate_closes_at
            );
        }
    }

    /// The invariant the headroom exists for: uploads must stop while the keepsake can STILL be
    /// built, with room to spare — not at the exact instant the preflight reaches its own limit.
    ///
    /// Both thresholds used to be `required_free_bytes(media, 2) + DISK_RESERVE_BYTES`, identically.
    /// So the moment the gate refused its first upload, the export preflight was already sitting on
    /// its limit, and every byte written afterwards (WAL, container logs, the compression backlog
    /// draining at exactly that hour) pushed it under. The release would then COMMIT — event closed,
    /// uploads locked, epoch bumped, `event-closed` fanned out to every phone — and only then would
    /// both workers bail, with no second release possible.
    #[test]
    fn the_upload_gate_closes_before_the_export_preflight_would_refuse() {
        for media_gb in [0u64, 1, 4, 8, 16, 32] {
            let required = required_free_bytes(media_gb * GB, 2);

            // `services::export::preflight` bails below this.
            let preflight_refuses_below = required + DISK_RESERVE_BYTES as u64;
            // `handlers::upload` refuses below this.
            let gate_refuses_below = preflight_refuses_below + UPLOAD_GATE_HEADROOM_BYTES as u64;

            assert!(
                gate_refuses_below > preflight_refuses_below,
                "at media={media_gb}GB the gate and the preflight share a threshold, so the \
                 keepsake's fate rests on whatever is written after uploads stop"
            );

            // At the instant the last upload is refused, the preflight must still pass with the
            // whole headroom to spare — that is the slack the night's remaining writes consume.
            let free_when_gate_closes = gate_refuses_below;
            assert!(
                free_when_gate_closes
                    >= preflight_refuses_below + UPLOAD_GATE_HEADROOM_BYTES as u64,
                "at media={media_gb}GB there is no slack between the gate closing and the \
                 preflight failing"
            );
        }
    }

    #[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() {
        // The boundary is the UPLOAD GATE's threshold plus a 25% margin, not the bare keepsake
        // size — see `disk_is_low`. Warning at the bare size fired only after the gate had
        // already blocked every guest.
        let required = 20 * GB;
        let gate = required + DISK_RESERVE_BYTES as u64 + UPLOAD_GATE_HEADROOM_BYTES as u64;
        let warn_at = gate + gate / 4;
        assert!(!disk_is_low(warn_at, required), "exactly enough is enough");
        assert!(disk_is_low(warn_at - 1, required));
    }

    #[test]
    fn an_empty_gallery_still_reserves_room_for_postgres() {
        // With no gallery the keepsake term is 0, so the warn threshold collapses to
        // 1.25 x (DISK_RESERVE_BYTES + UPLOAD_GATE_HEADROOM_BYTES) = 1.25 x 11.5 GB = 14.375 GB,
        // which dominates the 10 GB absolute floor.
        assert!(!disk_is_low(15 * GB, 0));
        assert!(disk_is_low(9 * GB, 0));
    }
}