Round 1 stopped the compression worker deleting an upload's original on failure — a transient ENOSPC or a codec panic must never destroy the only copy of a photo a guest cannot retake. But it left `Upload::soft_delete`'s quota refund in place, so the bytes stayed on disk while the uploader was charged nothing for them. That is worse than it first looks. The row is soft-deleted, so the file is invisible and unowned; a guest hitting a reproducible codec failure can accumulate orphans indefinitely at zero personal cost. And `active_uploaders` counts only users with non-deleted uploads, so dropping out of that count RAISES everyone's per-user ceiling — the leak loosens the very quota meant to contain it. Keep the refund: the uploader didn't cause the failure and shouldn't silently lose quota to it. Bound the leak instead, with an hourly sweep alongside the existing session cleanup in `spawn_periodic_tasks`, reclaiming failed originals older than 14 days — comfortably longer than any single event, so an operator investigating a failed upload still has the file. The selection predicate is the entire safety argument, so it is deliberately narrow: `compression_status = 'failed'` AND soft-deleted AND past the window AND `original_path <> ''`. That is exactly the state the give-up path leaves behind, and it cannot reach a live upload, an owner-deleted one, or a failure still inside its recovery window. `original_path` is cleared after a successful reclaim, which makes the sweep idempotent — otherwise a row whose file is already gone is re-selected on every tick forever. The row itself is kept as the audit trail. Tests reproduce the selection verbatim (same pattern as upload_concurrency) and assert it against five near-misses that must survive, both sides of the retention boundary, and the idempotence property. Also fixes two comments in export.rs still claiming "the compression worker hard-deletes an original when its transcode fails" — no longer true, and the defensive handling they justify is now justified by this sweep and by ordinary deletes instead. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
330 lines
13 KiB
Rust
330 lines
13 KiB
Rust
use anyhow::Result;
|
|
use axum::Router;
|
|
use axum::extract::DefaultBodyLimit;
|
|
use axum::routing::{delete, get, patch, post};
|
|
use tower_http::trace::TraceLayer;
|
|
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
|
|
|
|
mod auth;
|
|
mod config;
|
|
mod db;
|
|
mod error;
|
|
mod handlers;
|
|
mod models;
|
|
mod services;
|
|
mod state;
|
|
|
|
use config::AppConfig;
|
|
use state::AppState;
|
|
|
|
/// Hard HTTP body cap for the upload endpoint (576 MiB). Backstop against
|
|
/// memory-exhaustion; precise per-class size limits are enforced in the handler.
|
|
const MAX_UPLOAD_BYTES: usize = 576 * 1024 * 1024;
|
|
|
|
#[tokio::main]
|
|
async fn main() -> Result<()> {
|
|
dotenvy::dotenv().ok();
|
|
|
|
tracing_subscriber::registry()
|
|
.with(
|
|
tracing_subscriber::EnvFilter::try_from_default_env()
|
|
.unwrap_or_else(|_| "eventsnap_backend=debug,tower_http=debug".into()),
|
|
)
|
|
.with(tracing_subscriber::fmt::layer())
|
|
.init();
|
|
|
|
let config = AppConfig::from_env()?;
|
|
let pool = db::create_pool(&config.database_url).await?;
|
|
|
|
// Reset any rows left mid-flight by a previous (possibly crashed) instance —
|
|
// stuck `compression_status='processing'` uploads and `status='running'` export
|
|
// jobs. Must run before the server starts taking requests so clients never see
|
|
// the half-state.
|
|
services::maintenance::startup_recovery(&pool).await;
|
|
|
|
let state = AppState::new(pool.clone(), config.clone());
|
|
|
|
// Regenerate image derivatives an older pipeline produced: the big-screen display for
|
|
// uploads processed before it existed (v0.17.x), and anything predating the current
|
|
// DERIVATIVES_REV (rev 1 applies the EXIF orientation, without which every portrait
|
|
// phone photo is stored sideways). Fire-and-forget behind the compression semaphore;
|
|
// originals are never touched, so a failure just retries on the next start.
|
|
state.compression.backfill_stale_derivatives().await;
|
|
|
|
// Re-spawn exports for events that were released but whose keepsake never finished
|
|
// (crash mid-export). Needs the media/export paths + SSE sender, so it runs here
|
|
// rather than inside `startup_recovery`. Fire-and-forget: the workers run in the
|
|
// background; the HTTP server can start accepting requests meanwhile.
|
|
services::export::recover_exports(
|
|
pool.clone(),
|
|
config.media_path.clone(),
|
|
config.export_path.clone(),
|
|
config.comments_enabled,
|
|
state.sse_tx.clone(),
|
|
)
|
|
.await;
|
|
|
|
// Hourly background hygiene: prune expired sessions, evict cold rate-limiter
|
|
// keys. Keeps the DB and process from growing unboundedly over multi-day events.
|
|
services::maintenance::spawn_periodic_tasks(
|
|
pool,
|
|
state.rate_limiter.clone(),
|
|
state.sse_tickets.clone(),
|
|
config.media_path.clone(),
|
|
);
|
|
|
|
// Ensure media directories exist
|
|
tokio::fs::create_dir_all(&config.media_path).await.ok();
|
|
|
|
let api = Router::new()
|
|
// Auth
|
|
.route("/api/v1/event", get(handlers::public::get_public_event))
|
|
.route("/api/v1/join", post(auth::handlers::join))
|
|
.route("/api/v1/recover", post(auth::handlers::recover))
|
|
// Forgotten-PIN escape hatch: ask a host to reset it (unauthenticated, throttled).
|
|
.route(
|
|
"/api/v1/recover/request",
|
|
post(auth::handlers::request_pin_reset),
|
|
)
|
|
.route("/api/v1/admin/login", post(auth::handlers::admin_login))
|
|
.route("/api/v1/session", delete(auth::handlers::logout))
|
|
// "Sign out everywhere" — revoke all of the caller's sessions.
|
|
.route("/api/v1/sessions", delete(auth::handlers::logout_all))
|
|
// Upload — HTTP-level body cap as an OOM backstop. The handler still enforces
|
|
// the precise per-class limits from DB config (max_image/video_size_mb); this
|
|
// layer just stops a multi-GB body from being buffered into memory before that
|
|
// check runs. Sized generously above the default 500 MB video limit + multipart
|
|
// overhead — if an admin raises max_video_size_mb above this, bump MAX_UPLOAD_BYTES.
|
|
.route(
|
|
"/api/v1/upload",
|
|
post(handlers::upload::upload).route_layer(DefaultBodyLimit::max(MAX_UPLOAD_BYTES)),
|
|
)
|
|
.route(
|
|
"/api/v1/upload/{id}",
|
|
patch(handlers::upload::edit_upload).delete(handlers::upload::delete_upload),
|
|
)
|
|
.route(
|
|
"/api/v1/upload/{id}/original",
|
|
get(handlers::upload::get_original),
|
|
)
|
|
// Preview/thumbnail variants are gated the same way as originals (visibility
|
|
// check + direct /media block below) so moderation actually revokes access to
|
|
// the displayed images, not just the full-res download.
|
|
.route(
|
|
"/api/v1/upload/{id}/preview",
|
|
get(handlers::upload::get_preview),
|
|
)
|
|
.route(
|
|
"/api/v1/upload/{id}/display",
|
|
get(handlers::upload::get_display),
|
|
)
|
|
.route(
|
|
"/api/v1/upload/{id}/thumbnail",
|
|
get(handlers::upload::get_thumbnail),
|
|
)
|
|
// Current-user endpoints (live quota estimate, profile + privacy note bundle)
|
|
.route("/api/v1/me/context", get(handlers::me::get_context))
|
|
.route("/api/v1/me/quota", get(handlers::me::get_quota))
|
|
// Feed
|
|
.route("/api/v1/feed", get(handlers::feed::feed))
|
|
.route("/api/v1/feed/delta", get(handlers::feed::feed_delta))
|
|
.route("/api/v1/hashtags", get(handlers::feed::hashtags))
|
|
// Social
|
|
.route(
|
|
"/api/v1/upload/{id}/like",
|
|
post(handlers::social::toggle_like),
|
|
)
|
|
.route(
|
|
"/api/v1/upload/{id}/comments",
|
|
get(handlers::social::list_comments).post(handlers::social::add_comment),
|
|
)
|
|
.route(
|
|
"/api/v1/comment/{id}",
|
|
delete(handlers::social::delete_comment),
|
|
)
|
|
// SSE
|
|
.route("/api/v1/stream", get(handlers::sse::stream))
|
|
.route("/api/v1/stream/ticket", post(handlers::sse::issue_ticket))
|
|
// Host Dashboard
|
|
.route("/api/v1/host/event", get(handlers::host::get_event_status))
|
|
.route(
|
|
"/api/v1/host/event/close",
|
|
post(handlers::host::close_event),
|
|
)
|
|
.route("/api/v1/host/event/open", post(handlers::host::open_event))
|
|
.route(
|
|
"/api/v1/host/gallery/release",
|
|
post(handlers::host::release_gallery),
|
|
)
|
|
// Escape hatch: force a keepsake rebuild. Without it a failed export is terminal at runtime
|
|
// (release_gallery refuses an already-released event; recovery only runs at boot).
|
|
.route(
|
|
"/api/v1/host/export/rebuild",
|
|
post(handlers::host::rebuild_export),
|
|
)
|
|
.route("/api/v1/host/users", get(handlers::host::list_users))
|
|
.route(
|
|
"/api/v1/host/users/{id}/ban",
|
|
post(handlers::host::ban_user),
|
|
)
|
|
.route(
|
|
"/api/v1/host/users/{id}/unban",
|
|
post(handlers::host::unban_user),
|
|
)
|
|
.route(
|
|
"/api/v1/host/users/{id}/role",
|
|
patch(handlers::host::set_role),
|
|
)
|
|
.route(
|
|
"/api/v1/host/users/{id}/pin-reset",
|
|
post(handlers::host::reset_user_pin),
|
|
)
|
|
.route(
|
|
"/api/v1/host/pin-reset-requests",
|
|
get(handlers::host::list_pin_reset_requests),
|
|
)
|
|
.route(
|
|
"/api/v1/host/pin-reset-requests/{id}",
|
|
delete(handlers::host::dismiss_pin_reset_request),
|
|
)
|
|
.route(
|
|
"/api/v1/host/upload/{id}",
|
|
delete(handlers::host::host_delete_upload),
|
|
)
|
|
.route(
|
|
"/api/v1/host/comment/{id}",
|
|
delete(handlers::host::host_delete_comment),
|
|
)
|
|
// Export (all authenticated users)
|
|
.route("/api/v1/export/status", get(handlers::admin::export_status))
|
|
.route(
|
|
"/api/v1/export/ticket",
|
|
post(handlers::admin::export_ticket),
|
|
)
|
|
.route("/api/v1/export/zip", get(handlers::admin::download_zip))
|
|
.route("/api/v1/export/html", get(handlers::admin::download_html))
|
|
// Admin Dashboard
|
|
.route("/api/v1/admin/stats", get(handlers::admin::get_stats))
|
|
.route(
|
|
"/api/v1/admin/config",
|
|
get(handlers::admin::get_config).patch(handlers::admin::patch_config),
|
|
)
|
|
.route(
|
|
"/api/v1/admin/export/jobs",
|
|
get(handlers::admin::get_export_jobs),
|
|
);
|
|
|
|
// Test-only route: a hard reset for the Playwright E2E harness. The handler
|
|
// is compiled in always, but the route is only attached when
|
|
// `EVENTSNAP_TEST_MODE=1`. In production the call returns 404 — the route
|
|
// simply isn't there.
|
|
let api = if handlers::test_admin::is_test_mode() {
|
|
tracing::warn!(
|
|
"EVENTSNAP_TEST_MODE=1 — registering /api/v1/admin/__truncate. \
|
|
DO NOT enable this in production."
|
|
);
|
|
api.route(
|
|
"/api/v1/admin/__truncate",
|
|
post(handlers::test_admin::truncate_all),
|
|
)
|
|
} else {
|
|
api
|
|
};
|
|
|
|
// NOTE: media is deliberately NOT served over HTTP.
|
|
//
|
|
// Files live under `media_path` so the compression worker and the export job can read
|
|
// them off disk, but nothing may pull them straight from `/media/**` — that bypasses
|
|
// the visibility checks (soft-delete + ban-hide) that make a host takedown stick.
|
|
// Every legitimate fetch goes through `/api/v1/upload/{id}/{original,preview,display,
|
|
// thumbnail}`, which filter via `find_visible_media`; those are the only media URLs the
|
|
// backend ever emits (see `handlers::feed`).
|
|
//
|
|
// This used to be a `ServeDir` on `/media` with four `nest_service` blockers on the
|
|
// subtrees above it. That was bypassable: axum routes on the RAW path while `ServeDir`
|
|
// percent-decodes afterwards, so `/media/%70reviews/{id}.jpg` missed every blocker,
|
|
// fell through to the `ServeDir`, and was decoded back to `previews/` on disk — serving
|
|
// a taken-down photo to anyone, unauthenticated. Any single escaped byte worked, in all
|
|
// four subtrees. Deleting the route removes the vector outright rather than racing the
|
|
// decoder; `/media/**` now 404s regardless of encoding.
|
|
let router = Router::new()
|
|
.route("/health", get(|| async { "ok" }))
|
|
.merge(api)
|
|
.layer(TraceLayer::new_for_http())
|
|
.with_state(state);
|
|
|
|
let listener = tokio::net::TcpListener::bind(("0.0.0.0", config.app_port)).await?;
|
|
tracing::info!("listening on {}", listener.local_addr()?);
|
|
// `into_make_service_with_connect_info` is required by the pre-auth handlers, which
|
|
// extract `ConnectInfo<SocketAddr>` to use the peer address as the rate-limit key when
|
|
// X-Forwarded-For is absent. Without it those extractors fail at runtime.
|
|
axum::serve(
|
|
listener,
|
|
router.into_make_service_with_connect_info::<std::net::SocketAddr>(),
|
|
)
|
|
.with_graceful_shutdown(shutdown_signal())
|
|
.await?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Hard cap on how long we wait for in-flight connections to drain after a shutdown
|
|
/// signal. Uploads (streamed to disk) finish in well under this; the cap exists because
|
|
/// long-lived SSE streams never end on their own and would otherwise keep the graceful
|
|
/// drain — and thus the process — pending until the orchestrator force-kills it.
|
|
const SHUTDOWN_GRACE: std::time::Duration = std::time::Duration::from_secs(10);
|
|
|
|
/// Resolves on SIGINT (Ctrl-C) or SIGTERM (container stop / deploy). Letting
|
|
/// `axum::serve` drain in-flight requests on this signal means a redeploy no longer
|
|
/// truncates uploads mid-flight; any background compression/export half-states that a
|
|
/// hard kill would leave are already reconciled by `startup_recovery` on the next boot.
|
|
///
|
|
/// Once the signal fires we also arm a detached backstop that force-exits after
|
|
/// [`SHUTDOWN_GRACE`]. Without it, open SSE streams (which have no natural end) would
|
|
/// hold the graceful drain open indefinitely; the backstop bounds shutdown regardless
|
|
/// of the orchestrator's own kill timeout. If the drain completes first, `main` returns
|
|
/// and the process exits before the timer ever fires.
|
|
async fn shutdown_signal() {
|
|
let ctrl_c = async {
|
|
let _ = tokio::signal::ctrl_c().await;
|
|
};
|
|
|
|
#[cfg(unix)]
|
|
let terminate = async {
|
|
match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
|
|
Ok(mut sig) => {
|
|
sig.recv().await;
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(error = ?e, "failed to install SIGTERM handler");
|
|
// Never resolve — fall back to ctrl_c only.
|
|
std::future::pending::<()>().await;
|
|
}
|
|
}
|
|
};
|
|
|
|
#[cfg(not(unix))]
|
|
let terminate = std::future::pending::<()>();
|
|
|
|
tokio::select! {
|
|
_ = ctrl_c => {}
|
|
_ = terminate => {}
|
|
}
|
|
|
|
tracing::info!(
|
|
"shutdown signal received, draining in-flight requests (max {}s)",
|
|
SHUTDOWN_GRACE.as_secs()
|
|
);
|
|
|
|
// Backstop: if the graceful drain is still blocked after the grace window (almost
|
|
// always because SSE streams are still open), exit anyway so deploys aren't stalled.
|
|
tokio::spawn(async {
|
|
tokio::time::sleep(SHUTDOWN_GRACE).await;
|
|
tracing::warn!(
|
|
"graceful drain exceeded {}s (likely open SSE streams); forcing exit",
|
|
SHUTDOWN_GRACE.as_secs()
|
|
);
|
|
std::process::exit(0);
|
|
});
|
|
}
|