From ee554e7f38557ac3f8de3efe4ca0c55a306a83b0 Mon Sep 17 00:00:00 2001 From: fabi Date: Wed, 15 Jul 2026 19:52:17 +0200 Subject: [PATCH] style(backend): rustfmt the whole tree; gate cargo fmt --check in CI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The backend had never been run through rustfmt. Doing it in one mechanical pass (134 files) so no future functional diff is buried under formatting churn, then gating `cargo fmt --check` in checks.yml so it stays clean. Formatting only — no logic, SQL, or behaviour changed. Verified after the reformat: cargo test 56 passed, clippy --all-targets -D warnings clean, cargo fmt --check clean. This is the deferred cleanup noted when CI's Format step was first left out. Co-Authored-By: Claude Opus 4.8 --- .github/workflows/checks.yml | 6 +- backend/src/auth/handlers.rs | 44 +++++------ backend/src/auth/middleware.rs | 4 +- backend/src/config.rs | 23 +++--- backend/src/db.rs | 2 +- backend/src/error.rs | 3 +- backend/src/handlers/admin.rs | 36 ++++----- backend/src/handlers/feed.rs | 34 ++++---- backend/src/handlers/host.rs | 58 +++++++++----- backend/src/handlers/me.rs | 19 +++-- backend/src/handlers/public.rs | 2 +- backend/src/handlers/social.rs | 2 +- backend/src/handlers/sse.rs | 11 ++- backend/src/handlers/upload.rs | 73 +++++++++++------ backend/src/main.rs | 80 ++++++++++++++----- backend/src/models/comment.rs | 10 +-- backend/src/models/event.rs | 12 ++- backend/src/models/hashtag.rs | 10 ++- backend/src/models/session.rs | 5 +- backend/src/models/user.rs | 18 +++-- backend/src/services/compression.rs | 57 +++++++------- backend/src/services/config.rs | 50 ++++++++---- backend/src/services/disk.rs | 10 +-- backend/src/services/export.rs | 105 +++++++++++++++++++------ backend/src/services/maintenance.rs | 6 +- backend/src/services/rate_limiter.rs | 22 +++++- backend/src/services/sse_tickets.rs | 12 ++- backend/tests/export_epoch.rs | 112 ++++++++++++++++++++++----- backend/tests/upload_concurrency.rs | 69 ++++++++++++----- 29 files changed, 589 insertions(+), 306 deletions(-) diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml index 4d38500..e98703e 100644 --- a/.github/workflows/checks.yml +++ b/.github/workflows/checks.yml @@ -66,9 +66,9 @@ jobs: working-directory: ./backend run: cargo clippy --all-targets -- -D warnings - # NOTE: `cargo fmt --check` is deliberately NOT gated. The tree has never been rustfmt'd, so - # turning it on means a 112-file mechanical reformat that would bury every real diff under it. - # Formatting is not a correctness gate; do that cleanup on its own, then add the check here. + - name: Format + working-directory: ./backend + run: cargo fmt --check frontend: name: Frontend — vitest + svelte-check diff --git a/backend/src/auth/handlers.rs b/backend/src/auth/handlers.rs index 8e4e13a..f5ed12a 100644 --- a/backend/src/auth/handlers.rs +++ b/backend/src/auth/handlers.rs @@ -1,8 +1,8 @@ use std::time::Duration; +use axum::Json; use axum::extract::State; use axum::http::{HeaderMap, StatusCode}; -use axum::Json; use chrono::Utc; use rand::Rng; use serde::{Deserialize, Serialize}; @@ -39,8 +39,11 @@ pub async fn join( let ip = client_ip(&headers, "unknown"); let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await; let join_rate_on = config::get_bool(&state.config_cache, "join_rate_enabled", true).await; - if rate_limits_on && join_rate_on - && !state.rate_limiter.check(format!("join:{ip}"), 5, Duration::from_secs(60)) + if rate_limits_on + && join_rate_on + && !state + .rate_limiter + .check(format!("join:{ip}"), 5, Duration::from_secs(60)) { return Err(AppError::TooManyRequests( "Zu viele Anfragen. Bitte warte kurz und versuche es erneut.".into(), @@ -80,8 +83,7 @@ pub async fn join( // Generate a 4-digit PIN let pin: String = format!("{:04}", rand::rng().random_range(0..10000u32)); - let pin_hash = - bcrypt::hash(&pin, 12).map_err(|e| AppError::Internal(anyhow::anyhow!(e)))?; + let pin_hash = bcrypt::hash(&pin, 12).map_err(|e| AppError::Internal(anyhow::anyhow!(e)))?; // The pre-check above is racy: two simultaneous joins with the same name can both // pass it, and the DB's unique index then rejects the loser. Map that unique @@ -178,8 +180,7 @@ pub async fn recover( .await? .ok_or_else(|| AppError::NotFound("Event nicht gefunden.".into()))?; - let users = - User::find_by_event_and_name(&state.pool, event.id, display_name).await?; + let users = User::find_by_event_and_name(&state.pool, event.id, display_name).await?; if users.is_empty() { // No user with this name. Run a throwaway bcrypt verify so this branch takes @@ -208,8 +209,7 @@ pub async fn recover( User::reset_pin_attempts(&state.pool, user.id).await?; } - let pin_matches = bcrypt::verify(&body.pin, &user.recovery_pin_hash) - .unwrap_or(false); + let pin_matches = bcrypt::verify(&body.pin, &user.recovery_pin_hash).unwrap_or(false); if pin_matches { // Reset failed attempts on success @@ -289,13 +289,13 @@ pub async fn admin_login( // honest typos. let ip = client_ip(&headers, "unknown"); let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await; - let admin_rate_on = config::get_bool(&state.config_cache, "admin_login_rate_enabled", true).await; - if rate_limits_on && admin_rate_on - && !state.rate_limiter.check( - format!("admin_login:{ip}"), - 5, - Duration::from_secs(60), - ) + let admin_rate_on = + config::get_bool(&state.config_cache, "admin_login_rate_enabled", true).await; + if rate_limits_on + && admin_rate_on + && !state + .rate_limiter + .check(format!("admin_login:{ip}"), 5, Duration::from_secs(60)) { return Err(AppError::TooManyRequests( "Zu viele Anmeldeversuche. Bitte warte kurz und versuche es erneut.".into(), @@ -303,8 +303,7 @@ pub async fn admin_login( )); } - let valid = bcrypt::verify(&body.password, &state.config.admin_password_hash) - .unwrap_or(false); + let valid = bcrypt::verify(&body.password, &state.config.admin_password_hash).unwrap_or(false); if !valid { tracing::warn!(ip = %ip, "admin_login: wrong password"); @@ -330,8 +329,8 @@ pub async fn admin_login( let dummy_pin: String = (0..32) .map(|_| rand::rng().random_range(b'a'..=b'z') as char) .collect(); - let dummy_hash = bcrypt::hash(&dummy_pin, 4) - .map_err(|e| AppError::Internal(anyhow::anyhow!(e)))?; + let dummy_hash = + bcrypt::hash(&dummy_pin, 4).map_err(|e| AppError::Internal(anyhow::anyhow!(e)))?; let user = User::create(&state.pool, event.id, admin_name, &dummy_hash).await?; sqlx::query("UPDATE \"user\" SET role = 'admin' WHERE id = $1") .bind(user.id) @@ -364,10 +363,7 @@ pub async fn admin_login( })) } -pub async fn logout( - State(state): State, - auth: AuthUser, -) -> Result { +pub async fn logout(State(state): State, auth: AuthUser) -> Result { Session::delete_by_token_hash(&state.pool, &auth.token_hash).await?; Ok(StatusCode::NO_CONTENT) } diff --git a/backend/src/auth/middleware.rs b/backend/src/auth/middleware.rs index 1afad40..182ea94 100644 --- a/backend/src/auth/middleware.rs +++ b/backend/src/auth/middleware.rs @@ -54,7 +54,9 @@ impl FromRequestParts for AuthUser { let user = Session::find_user_by_token_hash(&state.pool, &token_hash) .await .map_err(|e| AppError::Internal(e.into()))? - .ok_or_else(|| AppError::Unauthorized("Sitzung nicht gefunden oder abgelaufen.".into()))?; + .ok_or_else(|| { + AppError::Unauthorized("Sitzung nicht gefunden oder abgelaufen.".into()) + })?; // Touch last_seen_at AND slide the session's expiry forward (fire-and-forget), so // an active client's session renews instead of hitting the fixed 30-day cliff. diff --git a/backend/src/config.rs b/backend/src/config.rs index 3cffe02..0eb4529 100644 --- a/backend/src/config.rs +++ b/backend/src/config.rs @@ -1,6 +1,6 @@ use std::path::PathBuf; -use anyhow::{anyhow, Context, Result}; +use anyhow::{Context, Result, anyhow}; /// Well-known dev JWT secret shipped in `.env.example`. If APP_ENV=production /// we refuse to start with this value; otherwise we warn loudly. @@ -67,8 +67,7 @@ pub struct AppConfig { impl AppConfig { pub fn from_env() -> Result { - let app_env = - std::env::var("APP_ENV").unwrap_or_else(|_| "development".to_string()); + let app_env = std::env::var("APP_ENV").unwrap_or_else(|_| "development".to_string()); let is_prod = app_env.eq_ignore_ascii_case("production"); let jwt_secret = std::env::var("JWT_SECRET").context("JWT_SECRET must be set")?; @@ -77,18 +76,15 @@ impl AppConfig { validate_secrets(is_prod, &jwt_secret, &admin_password_hash)?; Ok(Self { - database_url: std::env::var("DATABASE_URL") - .context("DATABASE_URL must be set")?, + database_url: std::env::var("DATABASE_URL").context("DATABASE_URL must be set")?, jwt_secret, session_expiry_days: std::env::var("SESSION_EXPIRY_DAYS") .unwrap_or_else(|_| "30".to_string()) .parse() .context("SESSION_EXPIRY_DAYS must be a number")?, admin_password_hash, - event_name: std::env::var("EVENT_NAME") - .unwrap_or_else(|_| "EventSnap".to_string()), - event_slug: std::env::var("EVENT_SLUG") - .context("EVENT_SLUG must be set")?, + event_name: std::env::var("EVENT_NAME").unwrap_or_else(|_| "EventSnap".to_string()), + event_slug: std::env::var("EVENT_SLUG").context("EVENT_SLUG must be set")?, media_path: PathBuf::from( std::env::var("MEDIA_PATH").unwrap_or_else(|_| "/media".to_string()), ), @@ -120,7 +116,10 @@ mod tests { // The exact string shipped in `.env` — >32 chars, so it must be caught by // the substring guard, not the length check. let err = validate_secrets(true, "change_me_to_a_random_64_byte_hex_string", REAL_HASH); - assert!(err.is_err(), "placeholder JWT_SECRET must be rejected in prod"); + assert!( + err.is_err(), + "placeholder JWT_SECRET must be rejected in prod" + ); } #[test] @@ -154,7 +153,9 @@ mod tests { fn placeholder_detection_is_case_insensitive() { // looks_placeholder lowercases before matching — an upper/mixed-case // placeholder must still be rejected in prod. - assert!(validate_secrets(true, "CHANGE_ME_TO_A_RANDOM_64_BYTE_HEX_STRING", REAL_HASH).is_err()); + assert!( + validate_secrets(true, "CHANGE_ME_TO_A_RANDOM_64_BYTE_HEX_STRING", REAL_HASH).is_err() + ); assert!(validate_secrets(true, REAL_SECRET, "$2Y$12$PLACEHOLDER_replace_me").is_err()); } diff --git a/backend/src/db.rs b/backend/src/db.rs index 736b663..bc7d3c0 100644 --- a/backend/src/db.rs +++ b/backend/src/db.rs @@ -1,6 +1,6 @@ use anyhow::{Context, Result}; -use sqlx::postgres::PgPoolOptions; use sqlx::PgPool; +use sqlx::postgres::PgPoolOptions; const DEFAULT_MAX_CONNECTIONS: u32 = 10; diff --git a/backend/src/error.rs b/backend/src/error.rs index 353aa36..30fa2ad 100644 --- a/backend/src/error.rs +++ b/backend/src/error.rs @@ -78,7 +78,8 @@ impl IntoResponse for AppError { if let Some(secs) = retry_after_secs && let Ok(val) = axum::http::HeaderValue::from_str(&secs.to_string()) { - resp.headers_mut().insert(axum::http::header::RETRY_AFTER, val); + resp.headers_mut() + .insert(axum::http::header::RETRY_AFTER, val); } resp } diff --git a/backend/src/handlers/admin.rs b/backend/src/handlers/admin.rs index 20dfc39..700184d 100644 --- a/backend/src/handlers/admin.rs +++ b/backend/src/handlers/admin.rs @@ -1,9 +1,9 @@ use std::collections::HashMap; use std::time::Duration; +use axum::Json; use axum::extract::{Query, State}; use axum::http::{HeaderMap, StatusCode}; -use axum::Json; use serde::{Deserialize, Serialize}; use crate::auth::middleware::RequireAdmin; @@ -45,19 +45,17 @@ pub async fn get_stats( .await? .ok_or_else(|| AppError::NotFound("Event nicht gefunden.".into()))?; - let (user_count,): (i64,) = - sqlx::query_as("SELECT COUNT(*) FROM \"user\" WHERE event_id = $1") + let (user_count,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM \"user\" WHERE event_id = $1") + .bind(event.id) + .fetch_one(&state.pool) + .await?; + + let (upload_count,): (i64,) = + sqlx::query_as("SELECT COUNT(*) FROM upload WHERE event_id = $1 AND deleted_at IS NULL") .bind(event.id) .fetch_one(&state.pool) .await?; - let (upload_count,): (i64,) = sqlx::query_as( - "SELECT COUNT(*) FROM upload WHERE event_id = $1 AND deleted_at IS NULL", - ) - .bind(event.id) - .fetch_one(&state.pool) - .await?; - let (comment_count,): (i64,) = sqlx::query_as( "SELECT COUNT(*) FROM comment c JOIN upload u ON u.id = c.upload_id @@ -90,10 +88,9 @@ pub async fn get_config( State(state): State, RequireAdmin(_auth): RequireAdmin, ) -> Result>, AppError> { - let rows: Vec<(String, String)> = - sqlx::query_as("SELECT key, value FROM config ORDER BY key") - .fetch_all(&state.pool) - .await?; + let rows: Vec<(String, String)> = sqlx::query_as("SELECT key, value FROM config ORDER BY key") + .fetch_all(&state.pool) + .await?; Ok(Json(rows.into_iter().collect())) } @@ -159,7 +156,7 @@ pub async fn patch_config( None => { return Err(AppError::BadRequest(format!( "Ungültiger Wert für {key}: muss eine Zahl sein." - ))) + ))); } }; if integer_only && n.fract() != 0.0 { @@ -297,7 +294,8 @@ pub async fn download_zip( authenticate_download_ticket(&state, &q.ticket).await?; enforce_export_rate(&state, &headers).await?; - let path = resolve_export_file(&state, "zip", "Der ZIP-Export ist noch nicht verfügbar.").await?; + let path = + resolve_export_file(&state, "zip", "Der ZIP-Export ist noch nicht verfügbar.").await?; serve_file(path, "Gallery.zip", "application/zip").await } @@ -361,7 +359,7 @@ async fn serve_file( content_type: &str, ) -> Result { use axum::body::Body; - use axum::http::{header, Response, StatusCode}; + use axum::http::{Response, StatusCode, header}; use tokio_util::io::ReaderStream; let file = tokio::fs::File::open(&path) @@ -418,9 +416,7 @@ pub async fn export_status( let job_status = |type_name: &str| { jobs.iter() .find(|(t, _, _)| t == type_name) - .map(|(_, status, pct)| { - serde_json::json!({ "status": status, "progress_pct": pct }) - }) + .map(|(_, status, pct)| serde_json::json!({ "status": status, "progress_pct": pct })) .unwrap_or_else(|| serde_json::json!({ "status": "locked", "progress_pct": 0 })) }; diff --git a/backend/src/handlers/feed.rs b/backend/src/handlers/feed.rs index fb65476..889e071 100644 --- a/backend/src/handlers/feed.rs +++ b/backend/src/handlers/feed.rs @@ -1,8 +1,8 @@ use std::time::Duration; +use axum::Json; use axum::extract::{Query, State}; use axum::http::HeaderMap; -use axum::Json; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use uuid::Uuid; @@ -130,7 +130,11 @@ pub async fn feed( let has_more = rows.len() as i64 > limit; let rows: Vec = rows.into_iter().take(limit as usize).collect(); - let next_cursor = if has_more { rows.last().map(|r| r.id) } else { None }; + let next_cursor = if has_more { + rows.last().map(|r| r.id) + } else { + None + }; // Batch check which uploads the current user has liked let upload_ids: Vec = rows.iter().map(|r| r.id).collect(); @@ -327,12 +331,11 @@ pub async fn hashtags( State(state): State, auth: AuthUser, ) -> Result>, AppError> { - let rows: Vec<(String, i64)> = sqlx::query_as( - "SELECT tag, upload_count FROM v_hashtag_counts WHERE event_id = $1", - ) - .bind(auth.event_id) - .fetch_all(&state.pool) - .await?; + let rows: Vec<(String, i64)> = + sqlx::query_as("SELECT tag, upload_count FROM v_hashtag_counts WHERE event_id = $1") + .bind(auth.event_id) + .fetch_all(&state.pool) + .await?; Ok(Json( rows.into_iter() @@ -362,14 +365,13 @@ async fn get_liked_set( if upload_ids.is_empty() { return std::collections::HashSet::new(); } - let rows: Vec<(Uuid,)> = sqlx::query_as( - "SELECT upload_id FROM \"like\" WHERE user_id = $1 AND upload_id = ANY($2)", - ) - .bind(user_id) - .bind(upload_ids) - .fetch_all(pool) - .await - .unwrap_or_default(); + let rows: Vec<(Uuid,)> = + sqlx::query_as("SELECT upload_id FROM \"like\" WHERE user_id = $1 AND upload_id = ANY($2)") + .bind(user_id) + .bind(upload_ids) + .fetch_all(pool) + .await + .unwrap_or_default(); rows.into_iter().map(|r| r.0).collect() } diff --git a/backend/src/handlers/host.rs b/backend/src/handlers/host.rs index f5f9d54..fcdc48a 100644 --- a/backend/src/handlers/host.rs +++ b/backend/src/handlers/host.rs @@ -1,6 +1,6 @@ +use axum::Json; use axum::extract::{Path, State}; use axum::http::StatusCode; -use axum::Json; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use uuid::Uuid; @@ -57,7 +57,6 @@ async fn remaining_operators( Ok(count) } - #[derive(Deserialize)] pub struct SetRoleRequest { pub role: String, @@ -115,7 +114,9 @@ pub async fn ban_user( // 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())); + 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", @@ -126,8 +127,12 @@ pub async fn ban_user( .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())); + if target.0 == "admin" + || (target.0 == "host" && auth.role != crate::models::user::UserRole::Admin) + { + return Err(AppError::Forbidden( + "Du kannst diesen Benutzer nicht sperren.".into(), + )); } // Floor: never leave the event with zero operators. Banning removes the target from @@ -166,9 +171,12 @@ pub async fn ban_user( // 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?; + 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); @@ -237,9 +245,12 @@ pub async fn unban_user( // 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?; + 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); @@ -266,9 +277,12 @@ pub async fn rebuild_export( RequireHost(auth): RequireHost, ) -> Result { let mut tx = state.pool.begin().await?; - let regen = - crate::services::export::invalidate_and_arm(&mut tx, &state.config.event_slug, Affects::Both) - .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 { @@ -316,7 +330,7 @@ pub async fn set_role( _ => { return Err(AppError::BadRequest( "Ungültige Rolle. Erlaubt: guest, host.".into(), - )) + )); } }; @@ -413,13 +427,12 @@ pub async fn reset_user_pin( _ => { 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 = - bcrypt::hash(&pin, 12).map_err(|e| AppError::Internal(anyhow::anyhow!(e)))?; + let pin_hash = bcrypt::hash(&pin, 12).map_err(|e| AppError::Internal(anyhow::anyhow!(e)))?; sqlx::query( "UPDATE \"user\" @@ -564,9 +577,12 @@ pub async fn host_delete_upload( 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?; + 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( diff --git a/backend/src/handlers/me.rs b/backend/src/handlers/me.rs index 7cc0ed2..76c4757 100644 --- a/backend/src/handlers/me.rs +++ b/backend/src/handlers/me.rs @@ -7,8 +7,8 @@ //! account page loads this once on mount instead of issuing several round trips. //! - `GET /api/v1/me/quota` — live per-user storage quota estimate. -use axum::extract::State; use axum::Json; +use axum::extract::State; use serde::Serialize; use crate::auth::middleware::AuthUser; @@ -73,12 +73,19 @@ pub async fn get_context( let privacy_note = config::get_str(&state.config_cache, "privacy_note", "").await; let quota_enabled = config::get_bool(&state.config_cache, "quota_enabled", true).await; - let storage_quota_enabled = config::get_bool(&state.config_cache, "storage_quota_enabled", true).await; + let storage_quota_enabled = + config::get_bool(&state.config_cache, "storage_quota_enabled", true).await; - let event = crate::models::event::Event::find_by_slug(&state.pool, &state.config.event_slug) - .await?; - let uploads_locked = event.as_ref().map(|e| e.uploads_locked_at.is_some()).unwrap_or(false); - let gallery_released = event.as_ref().map(|e| e.export_released_at.is_some()).unwrap_or(false); + let event = + crate::models::event::Event::find_by_slug(&state.pool, &state.config.event_slug).await?; + let uploads_locked = event + .as_ref() + .map(|e| e.uploads_locked_at.is_some()) + .unwrap_or(false); + let gallery_released = event + .as_ref() + .map(|e| e.export_released_at.is_some()) + .unwrap_or(false); Ok(Json(MeContextDto { user_id: user.id, diff --git a/backend/src/handlers/public.rs b/backend/src/handlers/public.rs index 1131ad7..a609049 100644 --- a/backend/src/handlers/public.rs +++ b/backend/src/handlers/public.rs @@ -1,7 +1,7 @@ //! Unauthenticated, read-only endpoints safe to expose before a user has joined. -use axum::extract::State; use axum::Json; +use axum::extract::State; use serde::Serialize; use crate::state::AppState; diff --git a/backend/src/handlers/social.rs b/backend/src/handlers/social.rs index 5304033..d43f79b 100644 --- a/backend/src/handlers/social.rs +++ b/backend/src/handlers/social.rs @@ -1,6 +1,6 @@ +use axum::Json; use axum::extract::{Path, Query, State}; use axum::http::StatusCode; -use axum::Json; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use uuid::Uuid; diff --git a/backend/src/handlers/sse.rs b/backend/src/handlers/sse.rs index 31b71f0..3fae809 100644 --- a/backend/src/handlers/sse.rs +++ b/backend/src/handlers/sse.rs @@ -1,14 +1,14 @@ use std::convert::Infallible; use std::time::Duration; +use axum::Json; use axum::extract::{Query, State}; use axum::response::sse::{Event, KeepAlive, Sse}; -use axum::Json; use futures::stream::Stream; use serde::{Deserialize, Serialize}; -use tokio_stream::wrappers::errors::BroadcastStreamRecvError; -use tokio_stream::wrappers::BroadcastStream; use tokio_stream::StreamExt; +use tokio_stream::wrappers::BroadcastStream; +use tokio_stream::wrappers::errors::BroadcastStreamRecvError; use crate::auth::middleware::AuthUser; use crate::error::AppError; @@ -42,7 +42,10 @@ pub async fn issue_ticket( let server_time = sqlx::query_scalar("SELECT NOW()") .fetch_one(&state.pool) .await?; - Ok(Json(StreamTicketResponse { ticket, server_time })) + Ok(Json(StreamTicketResponse { + ticket, + server_time, + })) } /// SSE stream endpoint. Authenticates via a single-use ticket (see diff --git a/backend/src/handlers/upload.rs b/backend/src/handlers/upload.rs index e0fdbd2..aabf8a6 100644 --- a/backend/src/handlers/upload.rs +++ b/backend/src/handlers/upload.rs @@ -1,8 +1,8 @@ use std::time::Duration; +use axum::Json; use axum::extract::{Multipart, Path, State}; use axum::http::StatusCode; -use axum::Json; use chrono::{DateTime, Utc}; use serde::Deserialize; use uuid::Uuid; @@ -47,7 +47,8 @@ pub async fn upload( let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await; let upload_rate_on = config::get_bool(&state.config_cache, "upload_rate_enabled", true).await; if rate_limits_on && upload_rate_on { - let upload_rate = config::get_i64(&state.config_cache, "upload_rate_per_hour", 10).await as usize; + let upload_rate = + config::get_i64(&state.config_cache, "upload_rate_per_hour", 10).await as usize; if let Err(retry_after_secs) = state.rate_limiter.check_with_retry( format!("upload:{}", auth.user_id), upload_rate, @@ -101,7 +102,10 @@ pub async fn upload( // On success the temp file is renamed into place under its detected extension. let upload_id = Uuid::new_v4(); let event_slug = &state.config.event_slug; - let originals_dir = state.config.media_path.join(format!("originals/{event_slug}")); + let originals_dir = state + .config + .media_path + .join(format!("originals/{event_slug}")); let temp_abs = originals_dir.join(format!("{upload_id}.tmp")); let mut streamed: Option<(i64, Vec)> = None; // (size, head bytes for sniffing) @@ -139,12 +143,20 @@ pub async fn upload( streamed = Some(stream_field_to_file(field, &temp_abs, cap_bytes).await?); } "caption" => { - caption = - Some(field.text().await.map_err(|e| AppError::BadRequest(e.to_string()))?); + caption = Some( + field + .text() + .await + .map_err(|e| AppError::BadRequest(e.to_string()))?, + ); } "hashtags" => { - hashtags_csv = - Some(field.text().await.map_err(|e| AppError::BadRequest(e.to_string()))?); + hashtags_csv = Some( + field + .text() + .await + .map_err(|e| AppError::BadRequest(e.to_string()))?, + ); } _ => {} } @@ -225,7 +237,8 @@ pub async fn upload( // number of active uploaders. Gated by master + per-area toggles so the admin can // disable it on trusted instances. let quota_on = config::get_bool(&state.config_cache, "quota_enabled", true).await; - let storage_quota_on = config::get_bool(&state.config_cache, "storage_quota_enabled", true).await; + let storage_quota_on = + config::get_bool(&state.config_cache, "storage_quota_enabled", true).await; // When quota is enforced, this holds the byte ceiling so the increment UPDATE below can // enforce it atomically (`WHERE total + size <= limit`). Without that guard, two // concurrent uploads from the same user (e.g. phone + laptop) both pass this stale @@ -321,11 +334,13 @@ pub async fn upload( .execute(&mut *tx) .await? } else { - sqlx::query("UPDATE \"user\" SET total_upload_bytes = total_upload_bytes + $2 WHERE id = $1") - .bind(auth.user_id) - .bind(size) - .execute(&mut *tx) - .await? + sqlx::query( + "UPDATE \"user\" SET total_upload_bytes = total_upload_bytes + $2 WHERE id = $1", + ) + .bind(auth.user_id) + .bind(size) + .execute(&mut *tx) + .await? }; if inc.rows_affected() == 0 { return Err(AppError::QuotaExceeded( @@ -592,15 +607,15 @@ fn quota_limit_bytes(free_disk: i64, tolerance: f64, active_uploaders: i64) -> i /// check (upload handler) or hide the UI (quota endpoint). pub async fn compute_storage_quota(state: &AppState) -> QuotaEstimate { let quota_on = config::get_bool(&state.config_cache, "quota_enabled", true).await; - let storage_quota_on = config::get_bool(&state.config_cache, "storage_quota_enabled", true).await; + let storage_quota_on = + config::get_bool(&state.config_cache, "storage_quota_enabled", true).await; let tolerance = config::get_f64(&state.config_cache, "quota_tolerance", 0.75).await; - let (active_count,): (i64,) = sqlx::query_as( - "SELECT COUNT(DISTINCT user_id) FROM upload WHERE deleted_at IS NULL", - ) - .fetch_one(&state.pool) - .await - .unwrap_or((0,)); + let (active_count,): (i64,) = + sqlx::query_as("SELECT COUNT(DISTINCT user_id) FROM upload WHERE deleted_at IS NULL") + .fetch_one(&state.pool) + .await + .unwrap_or((0,)); let active = active_count.max(1); // Cached disk reading. `None` means we couldn't resolve the media filesystem. @@ -644,7 +659,7 @@ async fn stream_media_file( cache_control: &str, ) -> Result { use axum::body::Body; - use axum::http::{header, Response, StatusCode}; + use axum::http::{Response, StatusCode, header}; use tokio_util::io::ReaderStream; if !absolute.exists() { @@ -723,7 +738,13 @@ pub async fn get_preview( .preview_path .ok_or_else(|| AppError::NotFound("Vorschau nicht verfügbar.".into()))?; let absolute = state.config.media_path.join(&rel); - stream_media_file(&absolute, "image/jpeg".to_string(), "inline", "private, max-age=300").await + stream_media_file( + &absolute, + "image/jpeg".to_string(), + "inline", + "private, max-age=300", + ) + .await } /// Streaming access to an upload's **thumbnail** (video poster). Gated identically to @@ -739,7 +760,13 @@ pub async fn get_thumbnail( .thumbnail_path .ok_or_else(|| AppError::NotFound("Thumbnail nicht verfügbar.".into()))?; let absolute = state.config.media_path.join(&rel); - stream_media_file(&absolute, "image/jpeg".to_string(), "inline", "private, max-age=300").await + stream_media_file( + &absolute, + "image/jpeg".to_string(), + "inline", + "private, max-age=300", + ) + .await } #[cfg(test)] diff --git a/backend/src/main.rs b/backend/src/main.rs index 819e758..92d47e0 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -1,7 +1,7 @@ use anyhow::Result; +use axum::Router; use axum::extract::DefaultBodyLimit; use axum::routing::{delete, get, patch, post}; -use axum::Router; use tower_http::services::ServeDir; use tower_http::trace::TraceLayer; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; @@ -27,9 +27,10 @@ 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::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| "eventsnap_backend=debug,tower_http=debug".into()), + ) .with(tracing_subscriber::fmt::layer()) .init(); @@ -73,7 +74,10 @@ async fn main() -> Result<()> { .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/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. @@ -83,8 +87,10 @@ async fn main() -> Result<()> { // 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", + 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), @@ -112,27 +118,51 @@ async fn main() -> Result<()> { .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}/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)) + .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/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)) + .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/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}/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), @@ -145,11 +175,20 @@ async fn main() -> Result<()> { "/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)) + .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/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 @@ -158,7 +197,10 @@ async fn main() -> Result<()> { "/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)); + .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 diff --git a/backend/src/models/comment.rs b/backend/src/models/comment.rs index b7de8ff..7dd9282 100644 --- a/backend/src/models/comment.rs +++ b/backend/src/models/comment.rs @@ -83,12 +83,10 @@ impl Comment { } pub async fn find_by_id(pool: &PgPool, id: Uuid) -> Result, sqlx::Error> { - sqlx::query_as::<_, Self>( - "SELECT * FROM comment WHERE id = $1 AND deleted_at IS NULL", - ) - .bind(id) - .fetch_optional(pool) - .await + sqlx::query_as::<_, Self>("SELECT * FROM comment WHERE id = $1 AND deleted_at IS NULL") + .bind(id) + .fetch_optional(pool) + .await } /// Event-scoped soft delete. Returns `false` if the comment doesn't exist or belongs to a diff --git a/backend/src/models/event.rs b/backend/src/models/event.rs index a70476a..df19d34 100644 --- a/backend/src/models/event.rs +++ b/backend/src/models/event.rs @@ -33,13 +33,11 @@ impl Event { } pub async fn create(pool: &PgPool, slug: &str, name: &str) -> Result { - sqlx::query_as::<_, Self>( - "INSERT INTO event (slug, name) VALUES ($1, $2) RETURNING *", - ) - .bind(slug) - .bind(name) - .fetch_one(pool) - .await + sqlx::query_as::<_, Self>("INSERT INTO event (slug, name) VALUES ($1, $2) RETURNING *") + .bind(slug) + .bind(name) + .fetch_one(pool) + .await } pub async fn find_or_create( diff --git a/backend/src/models/hashtag.rs b/backend/src/models/hashtag.rs index f7f20a2..e9dde24 100644 --- a/backend/src/models/hashtag.rs +++ b/backend/src/models/hashtag.rs @@ -122,14 +122,20 @@ mod tests { assert_eq!(extract_hashtags(&format!("#{ok}")), vec![ok.clone()]); // 41+ chars → dropped entirely (not truncated). let too_long = "a".repeat(41); - assert_eq!(extract_hashtags(&format!("#{too_long}")), Vec::::new()); + assert_eq!( + extract_hashtags(&format!("#{too_long}")), + Vec::::new() + ); } #[test] fn duplicate_tags_are_returned_verbatim_not_deduplicated() { // Dedup is the DB's job (Hashtag::upsert ON CONFLICT); extraction returns each // occurrence so callers can count/link them independently. Case folds to lower. - assert_eq!(extract_hashtags("#fun #Fun #fun!"), vec!["fun", "fun", "fun"]); + assert_eq!( + extract_hashtags("#fun #Fun #fun!"), + vec!["fun", "fun", "fun"] + ); } #[test] diff --git a/backend/src/models/session.rs b/backend/src/models/session.rs index 42df793..e8e4af8 100644 --- a/backend/src/models/session.rs +++ b/backend/src/models/session.rs @@ -90,10 +90,7 @@ impl Session { Ok(()) } - pub async fn delete_by_token_hash( - pool: &PgPool, - token_hash: &str, - ) -> Result<(), sqlx::Error> { + pub async fn delete_by_token_hash(pool: &PgPool, token_hash: &str) -> Result<(), sqlx::Error> { sqlx::query("DELETE FROM session WHERE token_hash = $1") .bind(token_hash) .execute(pool) diff --git a/backend/src/models/user.rs b/backend/src/models/user.rs index 9416bde..8b1a7ae 100644 --- a/backend/src/models/user.rs +++ b/backend/src/models/user.rs @@ -109,14 +109,16 @@ impl User { Ok(row.0) } - pub async fn lock_pin(pool: &PgPool, id: Uuid, until: DateTime) -> Result<(), sqlx::Error> { - sqlx::query( - "UPDATE \"user\" SET pin_locked_until = $2 WHERE id = $1", - ) - .bind(id) - .bind(until) - .execute(pool) - .await?; + pub async fn lock_pin( + pool: &PgPool, + id: Uuid, + until: DateTime, + ) -> Result<(), sqlx::Error> { + sqlx::query("UPDATE \"user\" SET pin_locked_until = $2 WHERE id = $1") + .bind(id) + .bind(until) + .execute(pool) + .await?; Ok(()) } diff --git a/backend/src/services/compression.rs b/backend/src/services/compression.rs index 29fcb1b..ef09712 100644 --- a/backend/src/services/compression.rs +++ b/backend/src/services/compression.rs @@ -1,10 +1,10 @@ use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; use anyhow::{Context, Result}; use sqlx::PgPool; -use tokio::sync::{broadcast, Semaphore}; +use tokio::sync::{Semaphore, broadcast}; use uuid::Uuid; use crate::models::upload::Upload; @@ -23,7 +23,12 @@ pub struct CompressionWorker { } impl CompressionWorker { - pub fn new(pool: PgPool, media_path: PathBuf, concurrency: usize, sse_tx: broadcast::Sender) -> Self { + pub fn new( + pool: PgPool, + media_path: PathBuf, + concurrency: usize, + sse_tx: broadcast::Sender, + ) -> Self { Self { semaphore: Arc::new(Semaphore::new(concurrency)), pool, @@ -55,7 +60,10 @@ impl CompressionWorker { if worker.generation.load(Ordering::SeqCst) != born_at { return; } - match worker.do_process(upload_id, &original_path, &mime_type).await { + match worker + .do_process(upload_id, &original_path, &mime_type) + .await + { Ok(_) => { tracing::info!("compression completed for upload {upload_id}"); let _ = worker.sse_tx.send(SseEvent { @@ -81,7 +89,8 @@ impl CompressionWorker { } let _ = worker.sse_tx.send(SseEvent { event_type: "upload-error".to_string(), - data: serde_json::json!({ "upload_id": upload_id, "error": e.to_string() }).to_string(), + data: serde_json::json!({ "upload_id": upload_id, "error": e.to_string() }) + .to_string(), }); let _ = worker.sse_tx.send(SseEvent { event_type: "upload-deleted".to_string(), @@ -103,7 +112,9 @@ impl CompressionWorker { let original = self.media_path.join(original_path); if mime_type.starts_with("image/") { - let preview_rel = self.generate_image_preview(upload_id, &original, mime_type).await?; + let preview_rel = self + .generate_image_preview(upload_id, &original, mime_type) + .await?; Upload::set_preview_path(&self.pool, upload_id, &preview_rel).await?; tracing::info!("preview generated for upload {upload_id}"); } else if mime_type.starts_with("video/") { @@ -151,7 +162,8 @@ impl CompressionWorker { // Resize to max 800px wide, preserving aspect ratio let preview = img.resize(800, 800, image::imageops::FilterType::Lanczos3); - preview.save_with_format(&preview_path_clone, image::ImageFormat::Jpeg) + preview + .save_with_format(&preview_path_clone, image::ImageFormat::Jpeg) .context("failed to save preview")?; // If the original is PNG, try lossless compression in-place @@ -174,11 +186,7 @@ impl CompressionWorker { Ok(format!("previews/{preview_filename}")) } - async fn generate_video_thumbnail( - &self, - upload_id: Uuid, - original: &Path, - ) -> Result { + async fn generate_video_thumbnail(&self, upload_id: Uuid, original: &Path) -> Result { let thumbs_dir = self.media_path.join("thumbnails"); tokio::fs::create_dir_all(&thumbs_dir).await?; @@ -208,18 +216,14 @@ impl CompressionWorker { .spawn() .context("failed to spawn ffmpeg")?; - let status = match tokio::time::timeout( - std::time::Duration::from_secs(120), - child.wait(), - ) - .await - { - Ok(res) => res.context("ffmpeg wait failed")?, - Err(_) => { - let _ = child.kill().await; - anyhow::bail!("ffmpeg timeout after 120s"); - } - }; + let status = + match tokio::time::timeout(std::time::Duration::from_secs(120), child.wait()).await { + Ok(res) => res.context("ffmpeg wait failed")?, + Err(_) => { + let _ = child.kill().await; + anyhow::bail!("ffmpeg timeout after 120s"); + } + }; if !status.success() { // Best-effort: drain stderr for the log. @@ -228,10 +232,7 @@ impl CompressionWorker { use tokio::io::AsyncReadExt; let _ = handle.read_to_end(&mut stderr).await; } - anyhow::bail!( - "ffmpeg failed: {}", - String::from_utf8_lossy(&stderr) - ); + anyhow::bail!("ffmpeg failed: {}", String::from_utf8_lossy(&stderr)); } Ok(format!("thumbnails/{thumb_filename}")) diff --git a/backend/src/services/config.rs b/backend/src/services/config.rs index 76a8e79..1894459 100644 --- a/backend/src/services/config.rs +++ b/backend/src/services/config.rs @@ -81,17 +81,18 @@ impl ConfigCache { } // Cache miss or stale — reload the entire table in one query. - let rows: Vec<(String, String)> = - match sqlx::query_as::<_, (String, String)>("SELECT key, value FROM config") - .fetch_all(&self.pool) - .await - { - Ok(rows) => rows, - Err(e) => { - tracing::warn!(error = ?e, "config reload failed; using defaults for this read"); - return None; - } - }; + let rows: Vec<(String, String)> = match sqlx::query_as::<_, (String, String)>( + "SELECT key, value FROM config", + ) + .fetch_all(&self.pool) + .await + { + Ok(rows) => rows, + Err(e) => { + tracing::warn!(error = ?e, "config reload failed; using defaults for this read"); + return None; + } + }; let values: HashMap = rows.into_iter().collect(); let result = values.get(key).cloned(); @@ -104,26 +105,43 @@ impl ConfigCache { } pub async fn get_str(cache: &ConfigCache, key: &str, default: &str) -> String { - cache.get_raw(key).await.unwrap_or_else(|| default.to_string()) + cache + .get_raw(key) + .await + .unwrap_or_else(|| default.to_string()) } pub async fn get_i64(cache: &ConfigCache, key: &str, default: i64) -> i64 { - cache.get_raw(key).await.and_then(|v| v.parse().ok()).unwrap_or(default) + cache + .get_raw(key) + .await + .and_then(|v| v.parse().ok()) + .unwrap_or(default) } pub async fn get_usize(cache: &ConfigCache, key: &str, default: usize) -> usize { - cache.get_raw(key).await.and_then(|v| v.parse().ok()).unwrap_or(default) + cache + .get_raw(key) + .await + .and_then(|v| v.parse().ok()) + .unwrap_or(default) } pub async fn get_f64(cache: &ConfigCache, key: &str, default: f64) -> f64 { - cache.get_raw(key).await.and_then(|v| v.parse().ok()).unwrap_or(default) + cache + .get_raw(key) + .await + .and_then(|v| v.parse().ok()) + .unwrap_or(default) } /// Parses common truthy spellings used by both the migration seeds and the admin form. /// Accepts `true/false`, `1/0`, `yes/no`, `on/off` — case-insensitive. Anything else /// returns `default`. pub async fn get_bool(cache: &ConfigCache, key: &str, default: bool) -> bool { - let Some(raw) = cache.get_raw(key).await else { return default }; + let Some(raw) = cache.get_raw(key).await else { + return default; + }; match raw.trim().to_ascii_lowercase().as_str() { "true" | "1" | "yes" | "on" => true, "false" | "0" | "no" | "off" => false, diff --git a/backend/src/services/disk.rs b/backend/src/services/disk.rs index d465eec..171ef4c 100644 --- a/backend/src/services/disk.rs +++ b/backend/src/services/disk.rs @@ -123,20 +123,14 @@ mod tests { #[test] fn picks_longest_matching_mount() { // Both "/" and "/media" prefix the path; the dedicated volume must win. - let mounts = vec![ - ("/".to_string(), 100, 40), - ("/media".to_string(), 200, 150), - ]; + let mounts = vec![("/".to_string(), 100, 40), ("/media".to_string(), 200, 150)]; let d = select_disk(&mounts, "/media/originals/x.jpg").unwrap(); assert_eq!((d.total, d.free), (200, 150)); } #[test] fn falls_back_to_root_when_no_specific_mount_matches() { - let mounts = vec![ - ("/".to_string(), 100, 40), - ("/media".to_string(), 200, 150), - ]; + let mounts = vec![("/".to_string(), 100, 40), ("/media".to_string(), 200, 150)]; // "/var/lib" is only prefixed by "/". let d = select_disk(&mounts, "/var/lib/data").unwrap(); assert_eq!((d.total, d.free), (100, 40)); diff --git a/backend/src/services/export.rs b/backend/src/services/export.rs index 57f3763..b22541e 100644 --- a/backend/src/services/export.rs +++ b/backend/src/services/export.rs @@ -5,12 +5,12 @@ use anyhow::{Context, Result}; use async_zip::tokio::write::ZipFileWriter; use async_zip::{Compression, ZipEntryBuilder}; use chrono::{DateTime, Utc}; -use futures::io::{copy as fcopy, AllowStdIo}; -use include_dir::{include_dir, Dir}; +use futures::io::{AllowStdIo, copy as fcopy}; +use include_dir::{Dir, include_dir}; use serde::Serialize; use sqlx::PgPool; -use tokio::sync::broadcast; use tokio::io::AsyncWriteExt; +use tokio::sync::broadcast; use tokio_util::compat::TokioAsyncReadCompatExt; use uuid::Uuid; @@ -218,7 +218,11 @@ pub async fn invalidate_and_arm( let types: &[&str] = if carried { &["html"] } else { &["zip", "html"] }; enqueue_types_at_epoch(&mut *conn, event_id, epoch, types).await?; - Ok(Some(PendingRegen { event_id, event_name, epoch })) + Ok(Some(PendingRegen { + event_id, + event_name, + epoch, + })) } /// Startup export recovery: re-arm exports for any released event whose keepsake isn't fully @@ -275,7 +279,9 @@ pub async fn recover_exports( continue; } - tracing::warn!("export recovery: re-arming export jobs for event {event_id} @ epoch {epoch}"); + tracing::warn!( + "export recovery: re-arming export jobs for event {event_id} @ epoch {epoch}" + ); let mut conn = match pool.acquire().await { Ok(c) => c, Err(e) => { @@ -416,7 +422,9 @@ pub fn spawn_export_jobs( if !delay.is_zero() { tokio::time::sleep(delay).await; } - if let Err(e) = run_zip_export(event_id, epoch, &pool, &media_path, &export_path, &sse_tx).await { + if let Err(e) = + run_zip_export(event_id, epoch, &pool, &media_path, &export_path, &sse_tx).await + { tracing::error!("ZIP export failed for event {event_id} @ epoch {epoch}: {e:#}"); mark_failed(&pool, event_id, "zip", epoch, &e.to_string()).await; } @@ -427,9 +435,16 @@ pub fn spawn_export_jobs( if !delay.is_zero() { tokio::time::sleep(delay).await; } - if let Err(e) = - run_html_export(event_id, epoch, &event_name2, &pool2, &media_path2, &export_path2, &sse_tx2) - .await + if let Err(e) = run_html_export( + event_id, + epoch, + &event_name2, + &pool2, + &media_path2, + &export_path2, + &sse_tx2, + ) + .await { tracing::error!("HTML export failed for event {event_id} @ epoch {epoch}: {e:#}"); mark_failed(&pool2, event_id, "html", epoch, &e.to_string()).await; @@ -459,8 +474,9 @@ async fn run_zip_export( // here so a failing export can't leak them (which is what fills the disk in the first place). let res = run_zip_export_inner(epoch, event_id, pool, media_path, export_path, sse_tx).await; if res.is_err() { - let _ = tokio::fs::remove_file(export_path.join(gen_name(event_id, "Gallery", epoch, ".tmp"))) - .await; + let _ = + tokio::fs::remove_file(export_path.join(gen_name(event_id, "Gallery", epoch, ".tmp"))) + .await; } abandon_if_superseded("ZIP", event_id, epoch, res) } @@ -518,7 +534,11 @@ async fn run_zip_export_inner( let ext = ext_from_path(&row.original_path); let date = row.created_at.format("%Y-%m-%d_%H-%M").to_string(); let name_safe = sanitize_name(&row.uploader_name); - let folder = if row.mime_type.starts_with("video/") { "Videos" } else { "Photos" }; + let folder = if row.mime_type.starts_with("video/") { + "Videos" + } else { + "Photos" + }; let entry_name = format!("{folder}/{date}_{name_safe}_{}.{ext}", row.id); let builder = ZipEntryBuilder::new(entry_name.into(), Compression::Stored); @@ -579,7 +599,9 @@ async fn run_zip_export_inner( // IS the publish, atomically. A worker at a dead epoch simply writes a row nobody can see. if !finalize_job(pool, event_id, "zip", epoch, &format!("exports/{out_name}")).await { let _ = tokio::fs::remove_file(&out_path).await; - tracing::info!("ZIP export for event {event_id} superseded (epoch {epoch} retired); discarded"); + tracing::info!( + "ZIP export for event {event_id} superseded (epoch {epoch} retired); discarded" + ); return Ok(()); } @@ -627,15 +649,25 @@ async fn run_html_export( return Ok(()); } - let res = - run_html_export_inner(epoch, event_id, event_name, pool, media_path, export_path, sse_tx).await; + let res = run_html_export_inner( + epoch, + event_id, + event_name, + pool, + media_path, + export_path, + sse_tx, + ) + .await; if res.is_err() { // Clean up this generation's temp artifacts so a failing (or abandoned) export can't leak // them — the leak is what fills the disk, which is what corrupts the next archive. let _ = - tokio::fs::remove_file(export_path.join(gen_name(event_id, "Memories", epoch, ".tmp"))).await; - let _ = tokio::fs::remove_dir_all(export_path.join(format!("viewer_tmp_{event_id}_{epoch}"))) - .await; + tokio::fs::remove_file(export_path.join(gen_name(event_id, "Memories", epoch, ".tmp"))) + .await; + let _ = + tokio::fs::remove_dir_all(export_path.join(format!("viewer_tmp_{event_id}_{epoch}"))) + .await; } abandon_if_superseded("HTML", event_id, epoch, res) } @@ -724,7 +756,10 @@ async fn run_html_export_inner( match ffmpeg_result { Ok(output) if output.status.success() => {} _ => { - tracing::warn!("ffmpeg thumbnail failed for upload {}, skipping thumb", row.id); + tracing::warn!( + "ffmpeg thumbnail failed for upload {}, skipping thumb", + row.id + ); // Missing thumb entry — viewer handles missing thumbs gracefully. } } @@ -904,7 +939,10 @@ async fn run_html_export_inner( let src_file = match tokio::fs::File::open(path).await { Ok(f) => f, Err(e) => { - tracing::warn!("HTML export: skipping media {name} — cannot read {}: {e}", path.display()); + tracing::warn!( + "HTML export: skipping media {name} — cannot read {}: {e}", + path.display() + ); continue; } }; @@ -939,9 +977,19 @@ async fn run_html_export_inner( // Epoch-guarded finalize — writing `done` at a live epoch IS the publish (readiness is derived // from it), so there is no separate ready flag to flip. If our epoch was retired, we lost: // discard the stale archive. - if !finalize_job(pool, event_id, "html", epoch, &format!("exports/{out_name}")).await { + if !finalize_job( + pool, + event_id, + "html", + epoch, + &format!("exports/{out_name}"), + ) + .await + { let _ = tokio::fs::remove_file(&out_path).await; - tracing::info!("HTML export for event {event_id} superseded (epoch {epoch} retired); discarded"); + tracing::info!( + "HTML export for event {event_id} superseded (epoch {epoch} retired); discarded" + ); return Ok(()); } @@ -1077,7 +1125,10 @@ async fn finalize_job( /// `Gallery..zip`, `Gallery.zip..tmp`, `viewer_tmp__`). Returns None if the /// name doesn't fit the shape, so unrelated files are left untouched. fn parse_gen_seq(name: &str, prefix: &str, suffix: &str) -> Option { - name.strip_prefix(prefix)?.strip_suffix(suffix)?.parse::().ok() + name.strip_prefix(prefix)? + .strip_suffix(suffix)? + .parse::() + .ok() } /// Best-effort removal of stale per-generation export artifacts for one export type. Deletes @@ -1276,7 +1327,13 @@ fn ext_from_path(path: &str) -> &str { fn sanitize_name(name: &str) -> String { name.chars() - .map(|c| if c.is_alphanumeric() || c == '-' { c } else { '_' }) + .map(|c| { + if c.is_alphanumeric() || c == '-' { + c + } else { + '_' + } + }) .collect() } diff --git a/backend/src/services/maintenance.rs b/backend/src/services/maintenance.rs index d4846db..0460d59 100644 --- a/backend/src/services/maintenance.rs +++ b/backend/src/services/maintenance.rs @@ -87,11 +87,7 @@ pub async fn startup_recovery(pool: &PgPool) { /// - 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, -) { +pub fn spawn_periodic_tasks(pool: PgPool, rate_limiter: RateLimiter, sse_tickets: SseTicketStore) { tokio::spawn(async move { let mut tick = tokio::time::interval(Duration::from_secs(3600)); // Fire the first tick immediately, then hourly. diff --git a/backend/src/services/rate_limiter.rs b/backend/src/services/rate_limiter.rs index d2501f0..bc2f247 100644 --- a/backend/src/services/rate_limiter.rs +++ b/backend/src/services/rate_limiter.rs @@ -24,7 +24,12 @@ impl RateLimiter { /// Returns `Ok(())` if allowed, `Err(retry_after_secs)` if rate-limited. /// `retry_after_secs` is how long until the oldest slot in the window expires. - pub fn check_with_retry(&self, key: impl Into, max: usize, window: Duration) -> Result<(), u64> { + pub fn check_with_retry( + &self, + key: impl Into, + max: usize, + window: Duration, + ) -> Result<(), u64> { let now = Instant::now(); let key = key.into(); let mut map = self.windows.lock().unwrap(); @@ -120,7 +125,10 @@ mod tests { assert!(rl.check("k", 1, w)); assert!(!rl.check("k", 1, w)); std::thread::sleep(Duration::from_millis(55)); - assert!(rl.check("k", 1, w), "the slot should expire once the window passes"); + assert!( + rl.check("k", 1, w), + "the slot should expire once the window passes" + ); } /// `retry_after` is not a "some number in range" — it is the time until the oldest slot @@ -174,7 +182,10 @@ mod tests { let retry = rl.check_with_retry("k", 1, w).unwrap_err(); // The sub-second remainder truncates to 0; clients must never be told "retry in 0s" // (that's a busy-loop). The `.max(1)` floor is what prevents it. - assert_eq!(retry, 1, "a sub-second remainder must floor to 1, got {retry}"); + assert_eq!( + retry, 1, + "a sub-second remainder must floor to 1, got {retry}" + ); } #[test] @@ -251,7 +262,10 @@ mod tests { fn client_ip_ignores_spoofed_leftmost_entry() { // A client prepending a fake IP to dodge throttles must not win. let mut h = HeaderMap::new(); - h.insert("x-forwarded-for", "1.2.3.4, 9.9.9.9, 203.0.113.7".parse().unwrap()); + h.insert( + "x-forwarded-for", + "1.2.3.4, 9.9.9.9, 203.0.113.7".parse().unwrap(), + ); assert_eq!(client_ip(&h, "fallback"), "203.0.113.7"); } diff --git a/backend/src/services/sse_tickets.rs b/backend/src/services/sse_tickets.rs index 7476ba0..535b27c 100644 --- a/backend/src/services/sse_tickets.rs +++ b/backend/src/services/sse_tickets.rs @@ -90,7 +90,11 @@ mod tests { let ticket = store.issue("hash-1".into()); assert_eq!(store.consume(&ticket).as_deref(), Some("hash-1")); // Single-use: a replay of the same ticket is rejected. - assert_eq!(store.consume(&ticket), None, "a consumed ticket must not be reusable"); + assert_eq!( + store.consume(&ticket), + None, + "a consumed ticket must not be reusable" + ); } #[test] @@ -131,6 +135,10 @@ mod tests { .expect("host uptime should exceed the ticket TTL"), }, ); - assert_eq!(store.consume(&stale), None, "an expired ticket must not authenticate"); + assert_eq!( + store.consume(&stale), + None, + "an expired ticket must not authenticate" + ); } } diff --git a/backend/tests/export_epoch.rs b/backend/tests/export_epoch.rs index 3b492e5..fb6b649 100644 --- a/backend/tests/export_epoch.rs +++ b/backend/tests/export_epoch.rs @@ -33,18 +33,34 @@ use sqlx::PgPool; #[sqlx::test] async fn release_returns_post_increment_epoch(pool: PgPool) { let event_id = seed_event(&pool, "wedding").await; - assert_eq!(event_epoch(&pool, event_id).await, 0, "a fresh event starts at epoch 0"); + assert_eq!( + event_epoch(&pool, event_id).await, + 0, + "a fresh event starts at epoch 0" + ); - let released = release_gallery(&pool, "wedding").await.expect("release claims the event"); - assert_eq!(released, 1, "RETURNING must give the epoch AFTER the +1, not before"); - assert_eq!(event_epoch(&pool, event_id).await, released, "worker's epoch == event's epoch"); + let released = release_gallery(&pool, "wedding") + .await + .expect("release claims the event"); + assert_eq!( + released, 1, + "RETURNING must give the epoch AFTER the +1, not before" + ); + assert_eq!( + event_epoch(&pool, event_id).await, + released, + "worker's epoch == event's epoch" + ); // The jobs armed by the release carry exactly that epoch — this is what makes the worker's // guarded writes match. for t in ["zip", "html"] { let (status, epoch, _) = job_row(&pool, event_id, t).await.expect("job armed"); assert_eq!(status, "pending"); - assert_eq!(epoch, released, "{t} job must be armed at the epoch the worker was born with"); + assert_eq!( + epoch, released, + "{t} job must be armed at the epoch the worker was born with" + ); } } @@ -61,15 +77,27 @@ async fn epoch_is_strictly_monotonic_across_reopen(pool: PgPool) { assert_eq!(release_gallery(&pool, "wedding").await, Some(1)); // A duplicate release is a no-op (`WHERE export_released_at IS NULL`) and must NOT bump. - assert_eq!(release_gallery(&pool, "wedding").await, None, "already released"); - assert_eq!(event_epoch(&pool, event_id).await, 1, "a rejected release must not move the epoch"); + assert_eq!( + release_gallery(&pool, "wedding").await, + None, + "already released" + ); + assert_eq!( + event_epoch(&pool, event_id).await, + 1, + "a rejected release must not move the epoch" + ); // Reopen retires the generation with ONE write. assert_eq!(open_event(&pool, "wedding").await, 1); assert_eq!(event_epoch(&pool, event_id).await, 2, "reopen bumps"); // And re-releasing bumps again — never back to 1. - assert_eq!(release_gallery(&pool, "wedding").await, Some(3), "re-release bumps again"); + assert_eq!( + release_gallery(&pool, "wedding").await, + Some(3), + "re-release bumps again" + ); assert_eq!(event_epoch(&pool, event_id).await, 3); } @@ -91,11 +119,19 @@ async fn retired_epoch_worker_writes_are_no_ops(pool: PgPool) { let old_epoch = release_gallery(&pool, "wedding").await.unwrap(); // Worker A is born at epoch 1 and claims the ZIP. - assert!(claim_job(&pool, event_id, "zip", old_epoch).await, "worker A wins its claim"); - assert!(update_progress(&pool, event_id, "zip", old_epoch, 40).await, "still live at 40%"); + assert!( + claim_job(&pool, event_id, "zip", old_epoch).await, + "worker A wins its claim" + ); + assert!( + update_progress(&pool, event_id, "zip", old_epoch, 40).await, + "still live at 40%" + ); // ── A takedown lands mid-export: `invalidate_and_arm(Affects::Both)` bumps and re-arms. ── - let (_, _, new_epoch) = bump_epoch(&pool, "wedding").await.expect("bump on a released event"); + let (_, _, new_epoch) = bump_epoch(&pool, "wedding") + .await + .expect("bump on a released event"); assert_eq!(new_epoch, old_epoch + 1); let mut conn = pool.acquire().await.unwrap(); enqueue_types_at_epoch(&mut conn, event_id, new_epoch, &["zip", "html"]).await; @@ -116,7 +152,10 @@ async fn retired_epoch_worker_writes_are_no_ops(pool: PgPool) { // fresh worker. (If worker A had won, this row would read `done` at epoch 1.) let (status, epoch, file_path) = job_row(&pool, event_id, "zip").await.unwrap(); assert_eq!((status.as_str(), epoch), ("pending", new_epoch)); - assert_eq!(file_path, None, "the loser's file_path must never be recorded"); + assert_eq!( + file_path, None, + "the loser's file_path must never be recorded" + ); // And nothing is downloadable — not the stale archive, not anything. assert_eq!(downloadable(&pool, event_id, "zip").await, None); @@ -177,13 +216,25 @@ async fn export_current_is_exactly_the_invariant(pool: PgPool) { // (name, released?, status, job epoch offset from the event epoch, expected visible) let cases: &[(&str, bool, &str, i64, bool)] = &[ ("released + done + current epoch", true, "done", 0, true), - ("NOT released (done, epoch matches)", false, "done", 0, false), + ( + "NOT released (done, epoch matches)", + false, + "done", + 0, + false, + ), ("NOT done: pending", true, "pending", 0, false), ("NOT done: running", true, "running", 0, false), ("NOT done: failed", true, "failed", 0, false), ("stale epoch (done, released)", true, "done", -1, false), ("future epoch (done, released)", true, "done", 1, false), - ("migration-014 retired sentinel epoch -1", true, "done", -2, false), + ( + "migration-014 retired sentinel epoch -1", + true, + "done", + -2, + false, + ), ]; for (i, (name, released, status, offset, expect_visible)) in cases.iter().enumerate() { @@ -209,7 +260,11 @@ async fn export_current_is_exactly_the_invariant(pool: PgPool) { .await .expect("clear armed jobs"); - let job_epoch = if *offset == -2 { -1 } else { event_epoch_now + offset }; + let job_epoch = if *offset == -2 { + -1 + } else { + event_epoch_now + offset + }; sqlx::query( "INSERT INTO export_job (event_id, type, status, progress_pct, epoch, file_path) VALUES ($1, 'zip', $2::export_status, 100, $3, 'exports/Gallery.zip')", @@ -251,12 +306,17 @@ async fn viewer_only_carries_a_done_zip_forward(pool: PgPool) { assert!(claim_job(&pool, event_id, t, e1).await); assert!(finalize_job(&pool, event_id, t, e1, &format!("exports/{t}.{e1}.zip")).await); } - let zip_file = downloadable(&pool, event_id, "zip").await.expect("zip is live"); + let zip_file = downloadable(&pool, event_id, "zip") + .await + .expect("zip is live"); // ── A comment is moderated: invalidate_and_arm(Affects::ViewerOnly). ── let (_, _, e2) = bump_epoch(&pool, "wedding").await.unwrap(); let carried = carry_zip_forward(&pool, event_id, e2).await; - assert!(carried, "a `done` ZIP at epoch-1 MUST be carried forward (rows_affected == 1)"); + assert!( + carried, + "a `done` ZIP at epoch-1 MUST be carried forward (rows_affected == 1)" + ); // Only the viewer is re-armed… let mut conn = pool.acquire().await.unwrap(); @@ -265,7 +325,11 @@ async fn viewer_only_carries_a_done_zip_forward(pool: PgPool) { // …and the ZIP is STILL DOWNLOADABLE, at the new epoch, pointing at the same, unrenamed file. let (status, epoch, _) = job_row(&pool, event_id, "zip").await.unwrap(); - assert_eq!((status.as_str(), epoch), ("done", e2), "the ZIP row rode the epoch bump"); + assert_eq!( + (status.as_str(), epoch), + ("done", e2), + "the ZIP row rode the epoch bump" + ); assert_eq!( downloadable(&pool, event_id, "zip").await, Some(zip_file), @@ -354,8 +418,16 @@ async fn enqueue_preserves_a_done_half_at_the_current_epoch(pool: PgPool) { // The good half survives untouched… let (status, epoch, file_path) = job_row(&pool, event_id, "zip").await.unwrap(); - assert_eq!((status.as_str(), epoch), ("done", e1), "a done half at the live epoch is preserved"); - assert_eq!(file_path.as_deref(), Some("exports/Gallery.1.zip"), "file_path not nulled"); + assert_eq!( + (status.as_str(), epoch), + ("done", e1), + "a done half at the live epoch is preserved" + ); + assert_eq!( + file_path.as_deref(), + Some("exports/Gallery.1.zip"), + "file_path not nulled" + ); assert!(downloadable(&pool, event_id, "zip").await.is_some()); // …and only the missing half is re-armed. diff --git a/backend/tests/upload_concurrency.rs b/backend/tests/upload_concurrency.rs index 29cc182..5ec69d0 100644 --- a/backend/tests/upload_concurrency.rs +++ b/backend/tests/upload_concurrency.rs @@ -17,12 +17,7 @@ use uuid::Uuid; /// SRC: `handlers/upload.rs:313-322` — the guarded quota increment, verbatim. /// Returns `rows_affected()`; the handler aborts the whole upload tx when this is 0. -async fn quota_inc( - exec: impl sqlx::PgExecutor<'_>, - user_id: Uuid, - size: i64, - limit: i64, -) -> u64 { +async fn quota_inc(exec: impl sqlx::PgExecutor<'_>, user_id: Uuid, size: i64, limit: i64) -> u64 { sqlx::query( "UPDATE \"user\" SET total_upload_bytes = total_upload_bytes + $2 WHERE id = $1 AND total_upload_bytes + $2 <= $3", @@ -69,7 +64,11 @@ async fn quota_two_attempts_from_one_stale_snapshot_cannot_both_commit(pool: PgP // Each upload, judged against that snapshot alone, fits: 0 + 60 <= 100. Twice. assert!(snapshot + SIZE <= LIMIT); - assert_eq!(quota_inc(&pool, user_id, SIZE, LIMIT).await, 1, "the first upload commits"); + assert_eq!( + quota_inc(&pool, user_id, SIZE, LIMIT).await, + 1, + "the first upload commits" + ); assert_eq!( quota_inc(&pool, user_id, SIZE, LIMIT).await, 0, @@ -77,7 +76,11 @@ async fn quota_two_attempts_from_one_stale_snapshot_cannot_both_commit(pool: PgP (60 + 60 = 120 > 100). rows_affected() == 0 is what makes the handler abort." ); - assert_eq!(total_bytes(&pool, user_id).await, SIZE, "never 120 — the quota held"); + assert_eq!( + total_bytes(&pool, user_id).await, + SIZE, + "never 120 — the quota held" + ); } /// The same, but genuinely CONCURRENT: two transactions that both read `total = 0`, then both try to @@ -126,9 +129,17 @@ async fn quota_guard_is_atomic_under_concurrent_transactions(pool: PgPool) { // And an upload that legitimately fits in what's left still succeeds — the guard rejects // overruns, not everything. - assert_eq!(quota_inc(&pool, user_id, 40, LIMIT).await, 1, "0 + 60 + 40 == 100, exactly at the limit"); + assert_eq!( + quota_inc(&pool, user_id, 40, LIMIT).await, + 1, + "0 + 60 + 40 == 100, exactly at the limit" + ); assert_eq!(total_bytes(&pool, user_id).await, LIMIT); - assert_eq!(quota_inc(&pool, user_id, 1, LIMIT).await, 0, "and one byte more is refused"); + assert_eq!( + quota_inc(&pool, user_id, 1, LIMIT).await, + 0, + "and one byte more is refused" + ); } // ───────────────────────────────────────────────────────────────────────────── @@ -140,11 +151,13 @@ async fn lock_and_read_event( tx: &mut sqlx::PgConnection, event_id: Uuid, ) -> (Option>, Option>) { - sqlx::query_as("SELECT uploads_locked_at, export_released_at FROM event WHERE id = $1 FOR SHARE") - .bind(event_id) - .fetch_one(tx) - .await - .expect("FOR SHARE re-check") + sqlx::query_as( + "SELECT uploads_locked_at, export_released_at FROM event WHERE id = $1 FOR SHARE", + ) + .bind(event_id) + .fetch_one(tx) + .await + .expect("FOR SHARE re-check") } /// THE GUARD AGAINST SILENT, PERMANENT PHOTO LOSS. @@ -167,7 +180,10 @@ async fn for_share_upload_lock_serializes_against_release(pool: PgPool) { // ── The guest's upload transaction takes the share lock. ── let mut upload_tx = pool.begin().await.unwrap(); let (locked, released) = lock_and_read_event(&mut upload_tx, event_id).await; - assert!(locked.is_none() && released.is_none(), "uploads are open, so we proceed to commit"); + assert!( + locked.is_none() && released.is_none(), + "uploads are open, so we proceed to commit" + ); // ── Concurrently, the host hits "Galerie freigeben". ── let release_done = Arc::new(AtomicBool::new(false)); @@ -230,7 +246,11 @@ async fn for_share_upload_lock_serializes_against_release(pool: PgPool) { .fetch_all(&pool) .await .unwrap(); - assert_eq!(snapshot, vec![upload_id], "the released keepsake CONTAINS the in-flight photo"); + assert_eq!( + snapshot, + vec![upload_id], + "the released keepsake CONTAINS the in-flight photo" + ); } /// The other side of the same lock: once the release has COMMITTED, the next upload's `FOR SHARE` @@ -255,14 +275,23 @@ async fn upload_after_release_commits_sees_the_lock_and_is_rejected(pool: PgPool // After it, the identical re-check sees the release and the handler bails out. let mut tx = pool.begin().await.unwrap(); let (locked, released) = lock_and_read_event(&mut tx, event_id).await; - assert!(released.is_some(), "the FOR SHARE re-read MUST observe the committed release"); - assert!(locked.is_some(), "release locks uploads in the same statement (release ⇒ lock)"); + assert!( + released.is_some(), + "the FOR SHARE re-read MUST observe the committed release" + ); + assert!( + locked.is_some(), + "release locks uploads in the same statement (release ⇒ lock)" + ); tx.rollback().await.unwrap(); // And a reopen makes it uploadable again — the rejection was reversible, not terminal. assert_eq!(open_event(&pool, "wedding").await, 1); let mut tx = pool.begin().await.unwrap(); let (locked, released) = lock_and_read_event(&mut tx, event_id).await; - assert!(locked.is_none() && released.is_none(), "the guest can resume their upload"); + assert!( + locked.is_none() && released.is_none(), + "the guest can resume their upload" + ); tx.rollback().await.unwrap(); }