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:#}"),
    }
}

/// 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;
            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:#}"),
    }
}