EventSnap/backend/src/services/maintenance.rs

//! Startup recovery + periodic background hygiene.
//!
//! Two responsibilities:
//!
//! 1. **Startup sweep** — when the server boots, fix rows left in an "in-progress"
//!    state by the previous (possibly crashed) instance. Compression and export jobs
//!    each leave a status row when they begin; if the process is killed mid-run, that
//!    row stays `'processing'` / `'running'` forever, blocking re-tries and leaving
//!    users staring at a spinner. Resetting them on startup recovers gracefully.
//!
//! 2. **Periodic tasks** — pruning that should happen "every hour" rather than per
//!    request: expired sessions (otherwise the table grows unboundedly), the
//!    rate-limiter's in-memory windows (so keys for IPs that left long ago don't
//!    accumulate), and the media of soft-deleted uploads — both the ones whose compression
//!    permanently failed and the ones a guest or host deliberately removed — which are
//!    retained for a recovery window and then reclaimed.

use std::path::PathBuf;
use std::time::Duration;

use sqlx::PgPool;

use crate::services::rate_limiter::RateLimiter;
use crate::services::sse_tickets::SseTicketStore;

/// How long a permanently-failed upload's original is kept on disk before it is
/// reclaimed.
///
/// The compression worker stops deleting originals on failure — a transient error must
/// never destroy the guest's only copy of a photo they can't retake. But the row is
/// soft-deleted and the uploader's quota IS refunded, so without a sweep those bytes are
/// invisible, unowned, and free: a reproducible codec failure lets one guest accumulate
/// orphans at no personal cost, and because `active_uploaders` counts only users with
/// non-deleted uploads, dropping out of that count actually RAISES everyone's per-user
/// ceiling while the disk gets fuller.
///
/// Two weeks is comfortably longer than any single event, so an operator investigating a
/// failed upload still has the file, while the leak stays bounded.
const FAILED_ORIGINAL_RETENTION_DAYS: i64 = 14;

/// How long a DELIBERATELY deleted upload's files are kept before they are reclaimed.
///
/// The same leak, reached by the ordinary path rather than the exceptional one.
/// `soft_delete_in_event` stamps `deleted_at` and refunds `total_upload_bytes`, but nothing ever
/// removed the bytes — so the quota stopped bounding the disk. Upload 500 MB, delete, quota is back
/// to zero, upload another 500 MB: not an attack, just a guest curating their camera roll, which is
/// what people do. The host then sees guests hitting "Du hast dein Upload-Limit erreicht" while the
/// admin widget shows a disk full of files no upload row points at, and the quota message is
/// actively misleading because the space really is gone — just not to anyone the accounting can
/// name.
///
/// Much shorter than the failure window on purpose. Fourteen days outlives the whole event, so a
/// deliberate delete would never reclaim anything while it mattered. A day still gives an operator
/// a recovery window for a mis-tap.
///
/// NOTE what this does NOT do: within the window the bytes are still spent and still unaccounted,
/// so a guest deleting and re-uploading through an eight-hour event can outrun the sweep. Bounding
/// that would mean holding the quota until the file is actually reclaimed rather than refunding at
/// `deleted_at` — a deliberate trade, and the reason the low-disk warning exists.
const DELETED_UPLOAD_RETENTION_HOURS: i64 = 24;

/// Reset rows left in flight by a previous crashed instance. Run once on startup,
/// before the HTTP server starts taking requests, so users never observe the
/// half-state.
pub async fn startup_recovery(pool: &PgPool) {
    // Uploads whose preview generation was interrupted. Marking them 'failed' is
    // safer than re-queueing — the original file is still on disk, the user can
    // delete + re-upload if they care, and we avoid double-processing risk.
    match sqlx::query(
        "UPDATE upload SET compression_status = 'failed'
         WHERE compression_status = 'processing'",
    )
    .execute(pool)
    .await
    {
        Ok(r) if r.rows_affected() > 0 => {
            tracing::warn!(
                "startup recovery: reset {} stuck upload(s) from 'processing' to 'failed'",
                r.rows_affected()
            );
        }
        Ok(_) => {}
        Err(e) => tracing::error!("startup recovery: failed to sweep uploads: {e:#}"),
    }

    // Export jobs interrupted mid-run are marked 'failed' here so they aren't left
    // 'running' forever. The host CANNOT re-trigger a released export (release_gallery
    // rejects an already-released event), so `export::recover_exports` re-spawns these
    // failed-but-released jobs from `main` once `AppState` exists (it needs the media
    // paths + SSE sender this fn doesn't have).
    //
    // SINGLE-INSTANCE ASSUMPTION: this sweep is unscoped — it reaps every `running` row, not just
    // this process's. That is correct for the supported deployment (one app container; see
    // docker-compose.yml), where no other process can own a job at boot. If EventSnap is ever run
    // with more than one replica, a booting replica would mark a peer's live workers `failed`.
    // This is NOT merely wasteful — do not assume the epoch model contains it. Recovery re-arms the
    // job at the SAME epoch, so the reaped-but-still-alive worker and the fresh one would BOTH hold
    // that epoch; the guards carry no owner/lease token, so they share the same artifact paths and
    // either can win the other's `finalize_job`. That can corrupt or delete the keepsake. Running
    // more than one replica therefore requires an owner/lease/heartbeat on export_job first. That
    // machinery is not worth building for a single-container app — so this is a documented
    // CONSTRAINT, not a hidden assumption: do not scale this service horizontally as-is.
    match sqlx::query(
        "UPDATE export_job
         SET status = 'failed',
             error_message = COALESCE(error_message, 'Server-Neustart während des Exports')
         WHERE status = 'running'",
    )
    .execute(pool)
    .await
    {
        Ok(r) if r.rows_affected() > 0 => {
            tracing::warn!(
                "startup recovery: reset {} stuck export job(s) from 'running' to 'failed'",
                r.rows_affected()
            );
        }
        Ok(_) => {}
        Err(e) => tracing::error!("startup recovery: failed to sweep export_job: {e:#}"),
    }
}

/// How long a file in `originals/` may exist without a database row before it is treated as
/// abandoned.
///
/// This window is the ONLY thing making the sweep safe, because the upload handler renames the
/// temp file into its final path BEFORE committing the row: for a short moment a perfectly
/// healthy upload legitimately looks exactly like an orphan. Six hours is far beyond any live
/// request (a 576 MiB body over a bad venue uplink is minutes, and the request itself is bounded
/// by the reverse proxy) while still reclaiming the leak inside a single event.
///
/// DO NOT SHORTEN THIS to make a test faster — a value below the longest possible in-flight
/// upload deletes photos out from under the request that is committing them.
const ORPHAN_UPLOAD_RETENTION_HOURS: u64 = 6;

/// Spawns a background task that periodically:
/// - deletes session rows whose `expires_at` is more than a day in the past
/// - prunes the in-memory rate-limiter HashMap of empty windows
/// - drops expired SSE tickets (30s TTL but the map keeps the slot until pruned)
///
/// Cadence is 1h — fine for both jobs at our scale.
pub fn spawn_periodic_tasks(
    pool: PgPool,
    rate_limiter: RateLimiter,
    sse_tickets: SseTicketStore,
    media_path: PathBuf,
) {
    tokio::spawn(async move {
        let mut tick = tokio::time::interval(Duration::from_secs(3600));
        // Fire the first tick immediately, then hourly.
        tick.tick().await;
        loop {
            tick.tick().await;
            cleanup_sessions(&pool).await;
            cleanup_deleted_media(&pool, &media_path).await;
            sweep_orphan_originals(&pool, &media_path).await;
            rate_limiter.prune();
            sse_tickets.prune();
        }
    });
}

/// Reclaim the media of soft-deleted uploads once they are past their retention window.
///
/// ONLY ever touches rows with `deleted_at IS NOT NULL`, so it can never reach a live upload. Two
/// classes, two windows, because the two deletes mean different things:
///
/// - a compression failure the guest didn't ask for and may want investigated —
///   [`FAILED_ORIGINAL_RETENTION_DAYS`];
/// - a deliberate removal by the guest or the host — [`DELETED_UPLOAD_RETENTION_HOURS`].
///
/// ALL FOUR paths are reclaimed, not just the original. The previous version cleared
/// `original_path` alone, which was right for its only case (a failed compression produces no
/// derivatives) but wrong the moment the sweep reaches a successfully processed upload: preview,
/// display and thumbnail are each a separate file on disk, none of them counted in
/// `original_size_bytes`, and nothing else ever removed them.
///
/// Every column is cleared in the same pass, which makes the sweep idempotent and stops a later run
/// re-reporting files that are already gone. The ROW is kept: it is the audit trail, it costs a few
/// hundred bytes, and `backfill_stale_derivatives` is guarded on `deleted_at IS NULL` so a nulled
/// `preview_path` can never make it regenerate what was just reclaimed.
async fn cleanup_deleted_media(pool: &PgPool, media_path: &std::path::Path) {
    type Row = (
        uuid::Uuid,
        String,
        Option<String>,
        Option<String>,
        Option<String>,
    );
    let rows = sqlx::query_as::<_, Row>(
        "SELECT id, original_path, preview_path, display_path, thumbnail_path FROM upload
         WHERE deleted_at IS NOT NULL
           AND CASE WHEN compression_status = 'failed'
                    THEN deleted_at < NOW() - ($1 || ' days')::interval
                    ELSE deleted_at < NOW() - ($2 || ' hours')::interval
               END
           AND (original_path <> '' OR preview_path IS NOT NULL
                OR display_path IS NOT NULL OR thumbnail_path IS NOT NULL)",
    )
    .bind(FAILED_ORIGINAL_RETENTION_DAYS.to_string())
    .bind(DELETED_UPLOAD_RETENTION_HOURS.to_string())
    .fetch_all(pool)
    .await;

    let rows = match rows {
        Ok(r) => r,
        Err(e) => {
            tracing::warn!(error = ?e, "deleted-media sweep query failed");
            return;
        }
    };
    if rows.is_empty() {
        return;
    }

    let mut reclaimed = 0usize;
    for (id, original, preview, display, thumbnail) in rows {
        let paths: Vec<String> = std::iter::once(original)
            .filter(|p| !p.is_empty())
            .chain([preview, display, thumbnail].into_iter().flatten())
            .collect();

        // All-or-nothing per row: the columns are only cleared once every file for that upload is
        // gone. Clearing after a partial success would strand the survivors with nothing pointing
        // at them — the same unowned-bytes state this sweep exists to drain.
        let mut all_gone = true;
        for rel in &paths {
            let absolute = media_path.join(rel);
            match tokio::fs::remove_file(&absolute).await {
                Ok(()) => reclaimed += 1,
                // Already gone (manual cleanup, restored backup) — still counts as reclaimed for
                // the purpose of clearing the columns, or the row is re-selected every hour forever.
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                Err(e) => {
                    tracing::warn!(error = ?e, %id, path = %absolute.display(),
                        "could not reclaim deleted media; leaving the row for the next sweep");
                    all_gone = false;
                }
            }
        }
        if !all_gone {
            continue;
        }

        if let Err(e) = sqlx::query(
            "UPDATE upload SET original_path = '', preview_path = NULL,
                    display_path = NULL, thumbnail_path = NULL
             WHERE id = $1",
        )
        .bind(id)
        .execute(pool)
        .await
        {
            tracing::warn!(error = ?e, %id, "reclaimed the files but could not clear the paths");
        }
    }

    if reclaimed > 0 {
        tracing::info!(
            "reclaimed {reclaimed} file(s) from soft-deleted uploads (deliberate deletes after \
             {DELETED_UPLOAD_RETENTION_HOURS}h, compression failures after \
             {FAILED_ORIGINAL_RETENTION_DAYS}d)"
        );
    }
}

async fn cleanup_sessions(pool: &PgPool) {
    match sqlx::query("DELETE FROM session WHERE expires_at < NOW() - INTERVAL '1 day'")
        .execute(pool)
        .await
    {
        Ok(r) if r.rows_affected() > 0 => {
            tracing::info!("cleaned up {} expired session(s)", r.rows_affected());
        }
        Ok(_) => {}
        Err(e) => tracing::warn!("session cleanup failed: {e:#}"),
    }
}

/// Reclaim files in `originals/` that no upload row references.
///
/// The backstop behind [`TempFileGuard`](crate::handlers::upload). The guard covers the
/// process that is running; this covers the process that was killed — a SIGKILL, an OOM, or a
/// power cut leaves whatever bytes had been written with no `Drop` to reclaim them, and those
/// files are then permanently invisible: they have no row, so `cleanup_deleted_media` (which is
/// row-driven) can never see them, and they are not counted against any quota while still
/// consuming the free disk that `compute_storage_quota` divides among guests. On a single box
/// where all three volumes share a filesystem, that ends with Postgres unable to write WAL.
///
/// Two classes:
/// - `*.tmp` — an upload that never got as far as being renamed. Always safe past the window.
/// - everything else — a final-named original whose commit never happened.
async fn sweep_orphan_originals(pool: &PgPool, media_path: &std::path::Path) {
    let originals = media_path.join("originals");
    let cutoff = Duration::from_secs(ORPHAN_UPLOAD_RETENTION_HOURS * 3600);

    // originals/{event_slug}/{uuid}.{ext} — one level of per-event directories.
    let mut event_dirs = match tokio::fs::read_dir(&originals).await {
        Ok(rd) => rd,
        // Nothing uploaded yet; the directory is created lazily by the upload handler.
        Err(_) => return,
    };

    let mut candidates: Vec<(String, std::path::PathBuf)> = Vec::new();
    let mut temps_removed = 0u32;

    while let Ok(Some(event_dir)) = event_dirs.next_entry().await {
        if !event_dir
            .file_type()
            .await
            .map(|t| t.is_dir())
            .unwrap_or(false)
        {
            continue;
        }
        let slug = event_dir.file_name().to_string_lossy().to_string();
        let Ok(mut files) = tokio::fs::read_dir(event_dir.path()).await else {
            continue;
        };
        while let Ok(Some(entry)) = files.next_entry().await {
            let Ok(meta) = entry.metadata().await else {
                continue;
            };
            if !meta.is_file() {
                continue;
            }
            // Too young to judge: an upload committing RIGHT NOW is indistinguishable from an
            // orphan, because the rename precedes the commit.
            let recent = meta
                .modified()
                .ok()
                .and_then(|m| m.elapsed().ok())
                .is_none_or(|age| age < cutoff);
            if recent {
                continue;
            }

            let name = entry.file_name().to_string_lossy().to_string();
            if name.ends_with(".tmp") {
                // A `.tmp` never has a row by construction — no DB check needed.
                if tokio::fs::remove_file(entry.path()).await.is_ok() {
                    temps_removed += 1;
                }
                continue;
            }
            candidates.push((format!("originals/{slug}/{name}"), entry.path()));
        }
    }

    if temps_removed > 0 {
        tracing::warn!(
            "reclaimed {temps_removed} abandoned upload temp file(s) older than \
             {ORPHAN_UPLOAD_RETENTION_HOURS}h"
        );
    }
    if candidates.is_empty() {
        return;
    }

    // One query per batch, not one per file: a backlog of thousands of orphans must not turn
    // into thousands of round trips on an hourly timer.
    let mut orphans_removed = 0u32;
    for chunk in candidates.chunks(500) {
        let paths: Vec<String> = chunk.iter().map(|(rel, _)| rel.clone()).collect();
        // NO `deleted_at IS NULL` FILTER HERE. A soft-deleted row still points at its file
        // during its retention window, and reclaiming that file is `cleanup_deleted_media`'s
        // job — filtering here would race the two sweeps and destroy the exact files the
        // recovery window exists to preserve.
        let unreferenced: Result<Vec<(String,)>, _> = sqlx::query_as(
            "SELECT p FROM unnest($1::text[]) AS p
              WHERE NOT EXISTS (SELECT 1 FROM upload u WHERE u.original_path = p)",
        )
        .bind(&paths)
        .fetch_all(pool)
        .await;
        let unreferenced = match unreferenced {
            Ok(rows) => rows,
            Err(e) => {
                tracing::warn!(error = ?e, "orphan-original sweep query failed");
                return;
            }
        };
        for (rel,) in unreferenced {
            if let Some((_, abs)) = chunk.iter().find(|(r, _)| *r == rel)
                && tokio::fs::remove_file(abs).await.is_ok()
            {
                orphans_removed += 1;
            }
        }
    }

    if orphans_removed > 0 {
        tracing::warn!(
            "reclaimed {orphans_removed} original(s) with no upload row, older than \
             {ORPHAN_UPLOAD_RETENTION_HOURS}h"
        );
    }
}