diff --git a/backend/src/auth/handlers.rs b/backend/src/auth/handlers.rs index c46cce4..561e7df 100644 --- a/backend/src/auth/handlers.rs +++ b/backend/src/auth/handlers.rs @@ -37,8 +37,8 @@ pub async fn join( Json(body): Json, ) -> Result<(StatusCode, Json), AppError> { let ip = client_ip(&headers, "unknown"); - let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await; - let join_rate_on = config::get_bool(&state.pool, "join_rate_enabled", true).await; + 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)) { @@ -134,8 +134,8 @@ pub async fn recover( // every 15 minutes, indefinitely. 5 attempts per 15 minutes per (IP, name) // softens that into a real cost. let ip = client_ip(&headers, "unknown"); - let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await; - let recover_rate_on = config::get_bool(&state.pool, "recover_rate_enabled", true).await; + let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await; + let recover_rate_on = config::get_bool(&state.config_cache, "recover_rate_enabled", true).await; if rate_limits_on && recover_rate_on { let name_key = display_name.to_lowercase(); if !state.rate_limiter.check( @@ -260,8 +260,8 @@ pub async fn admin_login( // a long-running guess campaign. 5 attempts / minute / IP is plenty for // honest typos. let ip = client_ip(&headers, "unknown"); - let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await; - let admin_rate_on = config::get_bool(&state.pool, "admin_login_rate_enabled", true).await; + 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}"), diff --git a/backend/src/auth/middleware.rs b/backend/src/auth/middleware.rs index d7051ae..46e8f00 100644 --- a/backend/src/auth/middleware.rs +++ b/backend/src/auth/middleware.rs @@ -37,34 +37,32 @@ impl FromRequestParts for AuthUser { .strip_prefix("Bearer ") .ok_or_else(|| AppError::Unauthorized("Ungültiges Token-Format.".into()))?; - let claims = jwt::verify_token(token, &state.config.jwt_secret) + // Verify the JWT (signature + expiry). We deliberately do NOT trust its role/ban + // claims — the live user row below is authoritative — so the decoded claims + // themselves aren't needed beyond this check. + jwt::verify_token(token, &state.config.jwt_secret) .map_err(|_| AppError::Unauthorized("Token ungültig oder abgelaufen.".into()))?; let token_hash = jwt::hash_token(token); - let session = Session::find_by_token_hash(&state.pool, &token_hash) + // Single round-trip: resolve the session token to its *live* user row. A + // role/ban stored in the token would survive a demote/ban for the full session + // lifetime (up to 30d), so we always re-read the user (a demoted host loses host + // powers immediately). We do NOT reject banned users here — they retain read + // access by design; writes and host/admin actions enforce the ban downstream. + 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()))?; - // Trust *live* DB state, not the JWT: a role/ban stored in the token would - // survive a demote/ban for the full session lifetime (up to 30d). Re-read - // the user row so a demoted host loses host powers immediately. We do NOT - // reject banned users here — they retain read access by design; writes and - // host/admin actions enforce the ban downstream. - let user = crate::models::user::User::find_by_id(&state.pool, claims.sub) - .await - .map_err(|e| AppError::Internal(e.into()))? - .ok_or_else(|| AppError::Unauthorized("Benutzer nicht gefunden.".into()))?; - // Update last_seen_at in the background (fire-and-forget). Failures are // non-fatal but worth surfacing — silent swallowing hides DB connection // pressure that would otherwise be the first symptom of a real problem. let pool = state.pool.clone(); - let session_id = session.id; + let touch_hash = token_hash.clone(); tokio::spawn(async move { - if let Err(e) = Session::touch(&pool, session_id).await { - tracing::warn!(error = ?e, session_id = %session_id, "session touch failed"); + if let Err(e) = Session::touch_by_token_hash(&pool, &touch_hash).await { + tracing::warn!(error = ?e, "session touch failed"); } }); diff --git a/backend/src/handlers/admin.rs b/backend/src/handlers/admin.rs index a2d4098..48edcb7 100644 --- a/backend/src/handlers/admin.rs +++ b/backend/src/handlers/admin.rs @@ -5,7 +5,6 @@ use axum::extract::{Query, State}; use axum::http::{HeaderMap, StatusCode}; use axum::Json; use serde::{Deserialize, Serialize}; -use sysinfo::System; use crate::auth::middleware::RequireAdmin; use crate::error::AppError; @@ -68,23 +67,12 @@ pub async fn get_stats( .fetch_one(&state.pool) .await?; - // Disk usage via sysinfo - let mut sys = System::new(); - sys.refresh_all(); - - let media_path = state.config.media_path.to_string_lossy().to_string(); - let (disk_total, disk_free) = sysinfo::Disks::new_with_refreshed_list() - .iter() - .find(|d| media_path.starts_with(d.mount_point().to_string_lossy().as_ref())) - .map(|d| (d.total_space(), d.available_space())) - .unwrap_or_else(|| { - // Fall back to the root disk - sysinfo::Disks::new_with_refreshed_list() - .iter() - .find(|d| d.mount_point().to_string_lossy() == "/") - .map(|d| (d.total_space(), d.available_space())) - .unwrap_or((0, 0)) - }); + // Disk usage from the shared cache (unknown mount → zeros, same as before). + let (disk_total, disk_free) = state + .disk_cache + .snapshot(&state.config.media_path) + .map(|d| (d.total, d.free)) + .unwrap_or((0, 0)); let disk_used = disk_total.saturating_sub(disk_free); @@ -216,6 +204,11 @@ pub async fn patch_config( } tx.commit().await?; + // The config cache must reflect this write on the very next read (tests PATCH then + // immediately assert the new value takes effect). Invalidate synchronously here — + // the TTL is only a backstop and must not be relied on for correctness. + state.config_cache.invalidate(); + // Notify all clients that a publicly-readable config value changed so their stores // (e.g. the privacy note in My Account) refresh without a manual reload. if privacy_note_changed { @@ -408,13 +401,13 @@ pub async fn export_status( /// switch + per-endpoint switch + numeric value, all stored in `config` and read on /// each request. async fn enforce_export_rate(state: &AppState, headers: &HeaderMap) -> Result<(), AppError> { - let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await; - let export_rate_on = config::get_bool(&state.pool, "export_rate_enabled", true).await; + let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await; + let export_rate_on = config::get_bool(&state.config_cache, "export_rate_enabled", true).await; if !(rate_limits_on && export_rate_on) { return Ok(()); } let ip = client_ip(headers, "unknown"); - let limit = config::get_usize(&state.pool, "export_rate_per_day", 3).await; + let limit = config::get_usize(&state.config_cache, "export_rate_per_day", 3).await; if !state .rate_limiter .check(format!("export:{ip}"), limit, Duration::from_secs(86400)) diff --git a/backend/src/handlers/feed.rs b/backend/src/handlers/feed.rs index 29e46e3..9e40fd4 100644 --- a/backend/src/handlers/feed.rs +++ b/backend/src/handlers/feed.rs @@ -62,10 +62,10 @@ pub async fn feed( Query(q): Query, ) -> Result, AppError> { let ip = client_ip(&headers, "unknown"); - let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await; - let feed_rate_on = config::get_bool(&state.pool, "feed_rate_enabled", true).await; + let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await; + let feed_rate_on = config::get_bool(&state.config_cache, "feed_rate_enabled", true).await; if rate_limits_on && feed_rate_on { - let rate_limit = config::get_usize(&state.pool, "feed_rate_per_min", 60).await; + let rate_limit = config::get_usize(&state.config_cache, "feed_rate_per_min", 60).await; if !state .rate_limiter .check(format!("feed:{ip}"), rate_limit, Duration::from_secs(60)) diff --git a/backend/src/handlers/me.rs b/backend/src/handlers/me.rs index 71c1df1..735293b 100644 --- a/backend/src/handlers/me.rs +++ b/backend/src/handlers/me.rs @@ -65,9 +65,9 @@ pub async fn get_context( .await? .ok_or_else(|| AppError::NotFound("Benutzer nicht gefunden.".into()))?; - let privacy_note = config::get_str(&state.pool, "privacy_note", "").await; - let quota_enabled = config::get_bool(&state.pool, "quota_enabled", true).await; - let storage_quota_enabled = config::get_bool(&state.pool, "storage_quota_enabled", true).await; + 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; Ok(Json(MeContextDto { user_id: user.id, diff --git a/backend/src/handlers/sse.rs b/backend/src/handlers/sse.rs index 394017b..7f0138f 100644 --- a/backend/src/handlers/sse.rs +++ b/backend/src/handlers/sse.rs @@ -6,6 +6,7 @@ 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; @@ -62,7 +63,14 @@ pub async fn stream( Ok(sse_event) => Some(Ok(Event::default() .event(sse_event.event_type) .data(sse_event.data))), - Err(_) => None, + // A consumer that falls behind the broadcast buffer would otherwise silently + // lose events, leaving its feed permanently stale. Instead of dropping the + // gap, tell the client to resync — the frontend responds by running a full + // feed-delta fetch (which reconciles both new uploads and deletions). + Err(BroadcastStreamRecvError::Lagged(n)) => { + tracing::warn!("SSE consumer lagged, dropped {n} event(s); emitting resync"); + Some(Ok(Event::default().event("resync").data(n.to_string()))) + } }); Ok(Sse::new(stream).keep_alive( diff --git a/backend/src/handlers/test_admin.rs b/backend/src/handlers/test_admin.rs index 0d6b2e6..ba330a5 100644 --- a/backend/src/handlers/test_admin.rs +++ b/backend/src/handlers/test_admin.rs @@ -80,6 +80,11 @@ pub async fn truncate_all( // counters don't leak into the next one. state.rate_limiter.clear(); + // The reseed above wrote the `config` table directly (bypassing patch_config), so + // the cache must be invalidated too — otherwise the first request after a truncate + // could serve the previous test's toggles. + state.config_cache.invalidate(); + Ok(StatusCode::NO_CONTENT) } diff --git a/backend/src/handlers/upload.rs b/backend/src/handlers/upload.rs index 96ad78b..3f08565 100644 --- a/backend/src/handlers/upload.rs +++ b/backend/src/handlers/upload.rs @@ -43,10 +43,10 @@ pub async fn upload( mut multipart: Multipart, ) -> Result<(StatusCode, Json), AppError> { // Rate limit: N uploads per hour per user. Gated by master + per-endpoint toggles. - let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await; - let upload_rate_on = config::get_bool(&state.pool, "upload_rate_enabled", true).await; + 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.pool, "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, @@ -79,57 +79,85 @@ pub async fn upload( } // Read config limits from DB - let max_image_mb: i64 = config::get_i64(&state.pool, "max_image_size_mb", 20).await; - let max_video_mb: i64 = config::get_i64(&state.pool, "max_video_size_mb", 500).await; + let max_image_mb: i64 = config::get_i64(&state.config_cache, "max_image_size_mb", 20).await; + let max_video_mb: i64 = config::get_i64(&state.config_cache, "max_video_size_mb", 500).await; - let mut file_data: Option> = None; + // The uploaded file is streamed straight to a temp file on disk (never buffered + // whole in memory — a 500 MB video used to cost 500 MB of RAM per concurrent + // upload). We only keep the first ≤512 bytes in memory for magic-byte sniffing. + // 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 temp_abs = originals_dir.join(format!("{upload_id}.tmp")); + + let mut streamed: Option<(i64, Vec)> = None; // (size, head bytes for sniffing) let mut caption: Option = None; let mut hashtags_csv: Option = None; - while let Some(field) = multipart.next_field().await.map_err(|e| AppError::BadRequest(e.to_string()))? { - let name = field.name().unwrap_or_default().to_string(); - match name.as_str() { - "file" => { - // Note: the client-declared filename and Content-Type do NOT determine - // the stored MIME/extension — those come from the file's magic bytes - // below. The declared type is used only to pick a memory cap so an - // oversized body can't be fully buffered before the size check. A - // mislabelled type only makes the cap *stricter* (safe); the - // authoritative per-class check still runs on the detected type. - let declared = field.content_type().unwrap_or("").to_string(); - let cap_bytes = if declared.starts_with("video/") { - (max_video_mb * 1024 * 1024) as usize - } else if declared.starts_with("image/") { - (max_image_mb * 1024 * 1024) as usize - } else { - (max_image_mb.max(max_video_mb) * 1024 * 1024) as usize - }; - file_data = Some(read_field_capped(field, cap_bytes).await?); + // Wrap the multipart read so any error after the temp file is created still cleans + // it up (a mid-stream parse failure must not leave a stray `.tmp` on disk). + let parse_result: Result<(), AppError> = async { + while let Some(field) = multipart + .next_field() + .await + .map_err(|e| AppError::BadRequest(e.to_string()))? + { + let name = field.name().unwrap_or_default().to_string(); + match name.as_str() { + "file" => { + // The client-declared Content-Type does NOT determine the stored + // MIME/extension — those come from the file's magic bytes below. The + // declared type only picks the streaming cap so an oversized body is + // aborted early; a mislabelled type only makes the cap *stricter* + // (safe), and the authoritative per-class check still runs on the + // detected type. + let declared = field.content_type().unwrap_or("").to_string(); + let cap_bytes = if declared.starts_with("video/") { + (max_video_mb * 1024 * 1024) as usize + } else if declared.starts_with("image/") { + (max_image_mb * 1024 * 1024) as usize + } else { + (max_image_mb.max(max_video_mb) * 1024 * 1024) as usize + }; + tokio::fs::create_dir_all(&originals_dir) + .await + .map_err(|e| AppError::Internal(e.into()))?; + 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()))?); + } + "hashtags" => { + hashtags_csv = + Some(field.text().await.map_err(|e| AppError::BadRequest(e.to_string()))?); + } + _ => {} } - "caption" => { - 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()))?, - ); - } - _ => {} } + Ok(()) + } + .await; + + if let Err(e) = parse_result { + let _ = tokio::fs::remove_file(&temp_abs).await; + return Err(e); } - let data = file_data.ok_or_else(|| AppError::BadRequest("Keine Datei hochgeladen.".into()))?; - let size = data.len() as i64; + // From here on the temp file may exist; every validation failure removes it before + // returning so a rejected upload never leaves bytes behind. + let (size, head) = match streamed { + Some(s) => s, + None => return Err(AppError::BadRequest("Keine Datei hochgeladen.".into())), + }; // Validate caption length. Counted in chars (code points) to match the // "Zeichen" wording in the error message — `.len()` would be bytes and // reject perfectly valid German/emoji captions early. if let Some(ref cap) = caption { if cap.chars().count() > MAX_CAPTION_LENGTH { + let _ = tokio::fs::remove_file(&temp_abs).await; return Err(AppError::BadRequest(format!( "Beschreibung ist zu lang. Maximum: {} Zeichen.", MAX_CAPTION_LENGTH @@ -142,26 +170,38 @@ pub async fn upload( // those are rejected outright — closing the stored-XSS vector. Both the MIME // we persist and the on-disk extension come from the detected type, never from // client-supplied values. - let kind = infer::get(&data) - .ok_or_else(|| AppError::BadRequest("Dateityp nicht erkannt oder nicht unterstützt.".into()))?; - let (mime, ext) = ALLOWED_MEDIA + let kind = match infer::get(&head) { + Some(k) => k, + None => { + let _ = tokio::fs::remove_file(&temp_abs).await; + return Err(AppError::BadRequest( + "Dateityp nicht erkannt oder nicht unterstützt.".into(), + )); + } + }; + let (mime, ext) = match ALLOWED_MEDIA .iter() .find(|(allowed, _)| *allowed == kind.mime_type()) .map(|(m, e)| ((*m).to_string(), *e)) - .ok_or_else(|| { - AppError::BadRequest(format!( + { + Some(v) => v, + None => { + let _ = tokio::fs::remove_file(&temp_abs).await; + return Err(AppError::BadRequest(format!( "Dateityp wird nicht unterstützt: {}.", kind.mime_type() - )) - })?; + ))); + } + }; - // Validate file size + // Validate file size against the authoritative per-detected-class limit. let max_bytes = if mime.starts_with("video/") { max_video_mb * 1024 * 1024 } else { max_image_mb * 1024 * 1024 }; if size > max_bytes { + let _ = tokio::fs::remove_file(&temp_abs).await; return Err(AppError::BadRequest(format!( "Datei ist zu groß. Maximum: {} MB.", max_bytes / (1024 * 1024) @@ -171,13 +211,14 @@ pub async fn upload( // Per-user storage quota — dynamic formula based on available disk space and the // 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.pool, "quota_enabled", true).await; - let storage_quota_on = config::get_bool(&state.pool, "storage_quota_enabled", true).await; + 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; if quota_on && storage_quota_on { let estimate = compute_storage_quota(&state).await; if let Some(limit) = estimate.limit_bytes { let prospective_total = user.total_upload_bytes.saturating_add(size); if prospective_total > limit { + let _ = tokio::fs::remove_file(&temp_abs).await; return Err(AppError::TooManyRequests( "Du hast dein Upload-Limit für dieses Event erreicht.".into(), None, @@ -186,16 +227,13 @@ pub async fn upload( } } - let upload_id = Uuid::new_v4(); - let event_slug = &state.config.event_slug; + // All checks passed — atomically move the temp file to its final, extension-correct + // path (same directory, so the rename is cheap and atomic). let relative_path = format!("originals/{event_slug}/{upload_id}.{ext}"); let absolute_path = state.config.media_path.join(&relative_path); - - // Ensure directory exists and write file - if let Some(parent) = absolute_path.parent() { - tokio::fs::create_dir_all(parent).await.map_err(|e| AppError::Internal(e.into()))?; - } - tokio::fs::write(&absolute_path, &data).await.map_err(|e| AppError::Internal(e.into()))?; + tokio::fs::rename(&temp_abs, &absolute_path) + .await + .map_err(|e| AppError::Internal(e.into()))?; // Process hashtags from caption and explicit CSV let mut tags: Vec = Vec::new(); @@ -216,27 +254,41 @@ pub async fn upload( // Quota accounting, the upload row, and its hashtag links must be atomic: a // crash between the bytes increment and the insert would permanently charge // bytes with no row to reclaim them (silent quota erosion / spurious lockout). - let mut tx = state.pool.begin().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) + let tx_result: Result = async { + let mut tx = state.pool.begin().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?; + let upload = Upload::create( + &mut *tx, + auth.event_id, + auth.user_id, + &relative_path, + &mime, + size, + caption.as_deref(), + ) .await?; - let upload = Upload::create( - &mut *tx, - auth.event_id, - auth.user_id, - &relative_path, - &mime, - size, - caption.as_deref(), - ) - .await?; - for tag in &tags { - let h = Hashtag::upsert(&mut *tx, auth.event_id, tag).await?; - Hashtag::link_to_upload(&mut *tx, upload.id, h.id).await?; + for tag in &tags { + let h = Hashtag::upsert(&mut *tx, auth.event_id, tag).await?; + Hashtag::link_to_upload(&mut *tx, upload.id, h.id).await?; + } + tx.commit().await?; + Ok(upload) } - tx.commit().await?; + .await; + + // The file is already on disk at `absolute_path`. If the transaction failed, no DB + // row will ever reference it, so remove it now rather than orphan bytes on disk. + let upload = match tx_result { + Ok(u) => u, + Err(e) => { + let _ = tokio::fs::remove_file(&absolute_path).await; + return Err(e); + } + }; // Spawn compression task state @@ -339,30 +391,68 @@ pub async fn delete_upload( Ok(StatusCode::NO_CONTENT) } -/// Read a multipart field into memory, aborting with a 400 the moment it exceeds -/// `max_bytes`. Without this the whole field is buffered (up to the HTTP body cap) -/// before the post-read size check runs, so a request claiming to be a tiny image -/// could still force hundreds of MB of allocation. Streaming with an early abort -/// bounds peak memory to roughly the applicable per-class limit. -async fn read_field_capped( +/// Number of leading bytes retained in memory for magic-byte (`infer`) sniffing. Every +/// allowed type's signature sits well within this; 512 is comfortably generous. +const HEAD_SNIFF_BYTES: usize = 512; + +/// Stream a multipart field straight to `dest`, aborting with a 400 the moment it +/// exceeds `max_bytes`. Only the first [`HEAD_SNIFF_BYTES`] bytes are kept in memory +/// (for type detection); the rest goes chunk-by-chunk to disk, so peak memory is a +/// single chunk rather than the whole file. Returns `(total_size, head_bytes)`. On any +/// error the partial temp file is removed so no stray `.tmp` is left behind. +async fn stream_field_to_file( mut field: axum::extract::multipart::Field<'_>, + dest: &std::path::Path, max_bytes: usize, -) -> Result, AppError> { - let mut buf: Vec = Vec::new(); - while let Some(chunk) = field - .chunk() +) -> Result<(i64, Vec), AppError> { + use tokio::io::AsyncWriteExt; + + let mut file = tokio::fs::File::create(dest) .await - .map_err(|e| AppError::BadRequest(format!("Datei konnte nicht gelesen werden: {e}")))? - { - if buf.len().saturating_add(chunk.len()) > max_bytes { + .map_err(|e| AppError::Internal(e.into()))?; + let mut total: usize = 0; + let mut head: Vec = Vec::with_capacity(HEAD_SNIFF_BYTES); + + loop { + let chunk = match field.chunk().await { + Ok(Some(c)) => c, + Ok(None) => break, + Err(e) => { + let _ = file.shutdown().await; + let _ = tokio::fs::remove_file(dest).await; + return Err(AppError::BadRequest(format!( + "Datei konnte nicht gelesen werden: {e}" + ))); + } + }; + + total = total.saturating_add(chunk.len()); + if total > max_bytes { + let _ = file.shutdown().await; + let _ = tokio::fs::remove_file(dest).await; return Err(AppError::BadRequest(format!( "Datei ist zu groß. Maximum: {} MB.", max_bytes / (1024 * 1024) ))); } - buf.extend_from_slice(&chunk); + + if head.len() < HEAD_SNIFF_BYTES { + let need = HEAD_SNIFF_BYTES - head.len(); + head.extend_from_slice(&chunk[..need.min(chunk.len())]); + } + + if let Err(e) = file.write_all(&chunk).await { + let _ = tokio::fs::remove_file(dest).await; + return Err(AppError::Internal(e.into())); + } } - Ok(buf) + + if let Err(e) = file.flush().await { + let _ = tokio::fs::remove_file(dest).await; + return Err(AppError::Internal(e.into())); + } + + Ok((total as i64, head)) } /// Drain a multipart body so the HTTP connection stays clean when returning an early error. @@ -396,9 +486,9 @@ fn quota_limit_bytes(free_disk: i64, tolerance: f64, active_uploaders: i64) -> i /// None` whenever the storage quota is currently disabled — callers should skip the /// 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.pool, "quota_enabled", true).await; - let storage_quota_on = config::get_bool(&state.pool, "storage_quota_enabled", true).await; - let tolerance = config::get_f64(&state.pool, "quota_tolerance", 0.75).await; + 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 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", @@ -408,21 +498,23 @@ pub async fn compute_storage_quota(state: &AppState) -> QuotaEstimate { .unwrap_or((0,)); let active = active_count.max(1); - let media_path = state.config.media_path.to_string_lossy().to_string(); - let free_disk = sysinfo::Disks::new_with_refreshed_list() - .iter() - .find(|d| media_path.starts_with(d.mount_point().to_string_lossy().as_ref())) - .map(|d| d.available_space()) - .unwrap_or_else(|| { - sysinfo::Disks::new_with_refreshed_list() - .iter() - .find(|d| d.mount_point().to_string_lossy() == "/") - .map(|d| d.available_space()) - .unwrap_or(0) - }) as i64; + // Cached disk reading. `None` means we couldn't resolve the media filesystem. + let disk = state.disk_cache.snapshot(&state.config.media_path); + let free_disk = disk.map(|d| d.free as i64).unwrap_or(0); let limit_bytes = if quota_on && storage_quota_on { - Some(quota_limit_bytes(free_disk, tolerance, active)) + match disk { + Some(d) => Some(quota_limit_bytes(d.free as i64, tolerance, active)), + // Fail OPEN, not closed: if the disk can't be read we don't know the real + // free space, and enforcing a 0-byte limit would reject every upload with a + // spurious "quota reached". Skip enforcement this round and warn instead. + None => { + tracing::warn!( + "disk snapshot unavailable; skipping storage-quota enforcement this round" + ); + None + } + } } else { None }; diff --git a/backend/src/main.rs b/backend/src/main.rs index d44c3c2..a0480df 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -161,7 +161,69 @@ async fn main() -> Result<()> { let listener = tokio::net::TcpListener::bind(("0.0.0.0", config.app_port)).await?; tracing::info!("listening on {}", listener.local_addr()?); - axum::serve(listener, router).await?; + axum::serve(listener, router) + .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); + }); +} diff --git a/backend/src/models/session.rs b/backend/src/models/session.rs index 280048f..a0c8bec 100644 --- a/backend/src/models/session.rs +++ b/backend/src/models/session.rs @@ -43,9 +43,33 @@ impl Session { .await } - pub async fn touch(pool: &PgPool, id: Uuid) -> Result<(), sqlx::Error> { - sqlx::query("UPDATE session SET last_seen_at = NOW() WHERE id = $1") - .bind(id) + /// Resolve a session token straight to its live user row in one round-trip. + /// + /// The auth extractor runs on every authenticated request and used to do two + /// sequential queries (session lookup, then user lookup); this collapses them into + /// a single `session JOIN "user"`. The `expires_at` guard mirrors + /// [`Self::find_by_token_hash`], and the user row is read live (role/ban are never + /// trusted from the JWT). Returns `None` when the session is missing/expired. + pub async fn find_user_by_token_hash( + pool: &PgPool, + token_hash: &str, + ) -> Result, sqlx::Error> { + sqlx::query_as::<_, crate::models::user::User>( + "SELECT u.* + FROM session s + JOIN \"user\" u ON u.id = s.user_id + WHERE s.token_hash = $1 AND s.expires_at > NOW()", + ) + .bind(token_hash) + .fetch_optional(pool) + .await + } + + /// Update `last_seen_at`, keyed by the token hash so the auth extractor can touch a + /// session without a separate query to fetch its id first. + pub async fn touch_by_token_hash(pool: &PgPool, token_hash: &str) -> Result<(), sqlx::Error> { + sqlx::query("UPDATE session SET last_seen_at = NOW() WHERE token_hash = $1") + .bind(token_hash) .execute(pool) .await?; Ok(()) diff --git a/backend/src/services/config.rs b/backend/src/services/config.rs index 2ef893a..76a8e79 100644 --- a/backend/src/services/config.rs +++ b/backend/src/services/config.rs @@ -1,46 +1,129 @@ -//! Reads of the runtime-tunable `config` table. +//! Reads of the runtime-tunable `config` table, fronted by an in-memory cache. //! //! Each handler used to keep a small local copy of these helpers; consolidating them //! here means one place to add a parser, one place to mock for tests, and one place to //! find when a key changes. New keys do not require code changes — they're picked up -//! the next time someone calls `get_*`. +//! the next time the cache reloads. +//! +//! ## Why a cache +//! +//! The `config` table is effectively static during an event, yet it was the busiest +//! query in the system: every request re-read each key with its own `SELECT` +//! (an upload touched it ~8 times). Against the small connection pool that was the +//! throughput ceiling. [`ConfigCache`] loads the whole table once and serves reads +//! from memory. +//! +//! ## Consistency contract +//! +//! Correctness comes from **synchronous invalidation on every write**, not from the +//! TTL. The two runtime write paths — the admin `PATCH /admin/config` handler and the +//! test-mode truncate/reseed — both call [`ConfigCache::invalidate`] after committing, +//! so the *next* read reloads from the DB and sees the new value immediately. The +//! [`RELOAD_TTL`] is only a safety net for out-of-band changes (e.g. a migration or a +//! manual DB edit); it is deliberately short but never the primary mechanism. //! //! Values are read with a default fallback so the app still starts if a key is missing //! (e.g. during a migration window). Production seeds keys via migrations 005 and 009. +use std::collections::HashMap; +use std::sync::{Arc, RwLock}; +use std::time::{Duration, Instant}; + use sqlx::PgPool; -async fn fetch_raw(pool: &PgPool, key: &str) -> Option { - sqlx::query_as::<_, (String,)>("SELECT value FROM config WHERE key = $1") - .bind(key) - .fetch_optional(pool) - .await - .ok() - .flatten() - .map(|(v,)| v) +/// How long a loaded snapshot is trusted before the next read reloads it. This is a +/// backstop for out-of-band DB changes only — every in-process write invalidates the +/// cache synchronously, so tests that PATCH-then-assert never depend on this expiring. +const RELOAD_TTL: Duration = Duration::from_secs(30); + +struct Snapshot { + values: HashMap, + loaded_at: Instant, } -pub async fn get_str(pool: &PgPool, key: &str, default: &str) -> String { - fetch_raw(pool, key).await.unwrap_or_else(|| default.to_string()) +/// In-memory cache of the entire `config` table. Cheap to `clone` (shares the pool and +/// the `Arc`), so it lives in `AppState` and every handler reads through it. +#[derive(Clone)] +pub struct ConfigCache { + pool: PgPool, + inner: Arc>>, } -pub async fn get_i64(pool: &PgPool, key: &str, default: i64) -> i64 { - fetch_raw(pool, key).await.and_then(|v| v.parse().ok()).unwrap_or(default) +impl ConfigCache { + pub fn new(pool: PgPool) -> Self { + Self { + pool, + inner: Arc::new(RwLock::new(None)), + } + } + + /// Drop the cached snapshot so the next read reloads the whole table from the DB. + /// Call this after any write to the `config` table (admin PATCH, test reseed). + pub fn invalidate(&self) { + *self.inner.write().unwrap() = None; + } + + /// Return the fresh snapshot if one is loaded and still within [`RELOAD_TTL`]. + fn fresh_snapshot(&self) -> Option> { + let guard = self.inner.read().unwrap(); + match guard.as_ref() { + Some(snap) if snap.loaded_at.elapsed() < RELOAD_TTL => Some(snap.values.clone()), + _ => None, + } + } + + /// Read one key, loading the whole table into the cache on a miss/expiry. On a DB + /// error we return `None` (callers fall back to their default) without poisoning + /// the cache. + async fn get_raw(&self, key: &str) -> Option { + if let Some(values) = self.fresh_snapshot() { + return values.get(key).cloned(); + } + + // 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 values: HashMap = rows.into_iter().collect(); + let result = values.get(key).cloned(); + *self.inner.write().unwrap() = Some(Snapshot { + values, + loaded_at: Instant::now(), + }); + result + } } -pub async fn get_usize(pool: &PgPool, key: &str, default: usize) -> usize { - fetch_raw(pool, key).await.and_then(|v| v.parse().ok()).unwrap_or(default) +pub async fn get_str(cache: &ConfigCache, key: &str, default: &str) -> String { + cache.get_raw(key).await.unwrap_or_else(|| default.to_string()) } -pub async fn get_f64(pool: &PgPool, key: &str, default: f64) -> f64 { - fetch_raw(pool, key).await.and_then(|v| v.parse().ok()).unwrap_or(default) +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) +} + +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) +} + +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) } /// 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(pool: &PgPool, key: &str, default: bool) -> bool { - let Some(raw) = fetch_raw(pool, key).await else { return default }; +pub async fn get_bool(cache: &ConfigCache, key: &str, default: bool) -> bool { + 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 new file mode 100644 index 0000000..a1783c1 --- /dev/null +++ b/backend/src/services/disk.rs @@ -0,0 +1,145 @@ +//! Cached view of the filesystem backing the media directory. +//! +//! Free/total disk space is needed on two hot paths — the per-user storage quota +//! (checked on every upload *and* every `GET /me/quota` poll) and the admin stats +//! endpoint. Reading it means `sysinfo::Disks::new_with_refreshed_list()`, which stats +//! every mounted filesystem; doing that per request is wasteful for a number that +//! barely moves. [`DiskCache`] refreshes it at most once per [`TTL`] and serves the +//! rest from memory. + +use std::path::Path; +use std::sync::{Arc, RwLock}; +use std::time::{Duration, Instant}; + +/// How long a disk reading is trusted before the next call re-stats the filesystem. +const TTL: Duration = Duration::from_secs(15); + +#[derive(Clone, Copy)] +pub struct DiskInfo { + pub total: u64, + pub free: u64, +} + +/// Cheap-to-clone cache of the media filesystem's total/free bytes. Lives in +/// `AppState`. +#[derive(Clone)] +pub struct DiskCache { + inner: Arc>>, +} + +impl DiskCache { + pub fn new() -> Self { + Self { + inner: Arc::new(RwLock::new(None)), + } + } + + /// Cached `(total, free)` bytes for the filesystem that holds `media_path`. + /// + /// Returns `None` when the mount can't be resolved — callers MUST treat that as + /// "unknown", never "zero free". (The quota path in particular fails *open* on + /// `None`: enforcing a 0-byte limit would lock every user out of uploading.) + pub fn snapshot(&self, media_path: &Path) -> Option { + if let Some((info, at)) = *self.inner.read().unwrap() { + if at.elapsed() < TTL { + return Some(info); + } + } + let info = read_disk_for_path(media_path)?; + *self.inner.write().unwrap() = Some((info, Instant::now())); + Some(info) + } +} + +impl Default for DiskCache { + fn default() -> Self { + Self::new() + } +} + +/// Resolve the filesystem backing `media_path` and read its total/free bytes. +/// +/// Snapshots the mount table via `sysinfo`, then delegates the selection to the pure +/// [`select_disk`] so the (fiddly, edge-case-prone) matching logic is unit-testable +/// without touching the real filesystem. +fn read_disk_for_path(media_path: &Path) -> Option { + let disks = sysinfo::Disks::new_with_refreshed_list(); + let mounts: Vec<(String, u64, u64)> = disks + .iter() + .map(|d| { + ( + d.mount_point().to_string_lossy().to_string(), + d.total_space(), + d.available_space(), + ) + }) + .collect(); + select_disk(&mounts, &media_path.to_string_lossy()) +} + +/// Pick the filesystem for `media_path` from a `(mount_point, total, free)` table. +/// +/// Chooses the **longest** mount point that is a prefix of `media_path` (the most +/// specific filesystem) rather than the first match — otherwise the root `/` mount, +/// which prefixes every absolute path, could shadow a dedicated `/media` volume. Falls +/// back to `/` when nothing prefixes the path (e.g. a relative media path), and to +/// `None` when even that is absent — the caller treats `None` as "unknown" and fails +/// open on quota. +fn select_disk(mounts: &[(String, u64, u64)], media_path: &str) -> Option { + mounts + .iter() + .filter(|(mp, _, _)| media_path.starts_with(mp.as_str())) + .max_by_key(|(mp, _, _)| mp.len()) + .or_else(|| mounts.iter().find(|(mp, _, _)| mp == "/")) + .map(|(_, total, free)| DiskInfo { + total: *total, + free: *free, + }) +} + +#[cfg(test)] +mod tests { + use super::select_disk; + + #[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 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), + ]; + // "/var/lib" is only prefixed by "/". + let d = select_disk(&mounts, "/var/lib/data").unwrap(); + assert_eq!((d.total, d.free), (100, 40)); + } + + #[test] + fn relative_path_uses_root_fallback() { + let mounts = vec![("/".to_string(), 100, 40)]; + // A relative path prefixes nothing, so the explicit "/" fallback applies. + let d = select_disk(&mounts, "media/originals").unwrap(); + assert_eq!((d.total, d.free), (100, 40)); + } + + #[test] + fn none_when_no_mount_matches_and_no_root() { + // No "/" present and nothing prefixes the relative path → unknown (fail-open). + let mounts = vec![("/data".to_string(), 100, 40)]; + assert!(select_disk(&mounts, "relative/path").is_none()); + } + + #[test] + fn none_on_empty_mount_table() { + assert!(select_disk(&[], "/media/x").is_none()); + } +} diff --git a/backend/src/services/export.rs b/backend/src/services/export.rs index ad908de..d1c48ec 100644 --- a/backend/src/services/export.rs +++ b/backend/src/services/export.rs @@ -192,6 +192,23 @@ async fn run_zip_export( // ── HTML viewer export ────────────────────────────────────────────────────── +/// Where a media entry's bytes come from at ZIP-writing time. Derived variants +/// (thumbnails, compressed images) are staged under the temp dir; original-fidelity +/// variants (videos, small images) are streamed straight from the source on disk so +/// the export never transiently doubles disk usage by copying large files to temp. +enum MediaSource { + Temp(PathBuf), + Original(PathBuf), +} + +impl MediaSource { + fn path(&self) -> &Path { + match self { + MediaSource::Temp(p) | MediaSource::Original(p) => p, + } + } +} + async fn run_html_export( event_id: Uuid, event_name: &str, @@ -221,6 +238,9 @@ async fn run_html_export( // 3. Process media and build post data let mut viewer_posts: Vec = Vec::new(); + // (zip entry name under media/, where its bytes come from). Built here, streamed + // into the ZIP in step 5 — so we also know the exact file count without a rescan. + let mut media_manifest: Vec<(String, MediaSource)> = Vec::new(); for (i, row) in uploads.iter().enumerate() { let src = media_path.join(&row.original_path); @@ -231,8 +251,10 @@ async fn run_html_export( let is_video = row.mime_type.starts_with("video/"); let id_str = row.id.to_string(); - // Generate thumbnail and full variant - let (thumb_name, full_name) = if is_video { + // Generate thumbnail and full variant. `full_source` says where the full-res + // bytes come from at ZIP time — for videos and small images that's the original + // on disk (streamed directly, never copied to temp). + let (thumb_name, full_name, full_source) = if is_video { let thumb = format!("{id_str}_thumb.jpg"); let full_ext = ext_from_path(&row.original_path); let full = format!("{id_str}.{full_ext}"); @@ -259,14 +281,13 @@ async fn run_html_export( Ok(output) if output.status.success() => {} _ => { tracing::warn!("ffmpeg thumbnail failed for upload {}, skipping thumb", row.id); - // Create empty thumb entry — viewer handles missing thumbs gracefully + // Missing thumb entry — viewer handles missing thumbs gracefully. } } - // Copy video as-is - tokio::fs::copy(&src, media_tmp.join(&full)).await?; - - (thumb, full) + // Stream the video full-res straight from the original at ZIP time — no + // copy to temp (that used to transiently double disk usage per video). + (thumb, full, MediaSource::Original(src.clone())) } else { let thumb = format!("{id_str}_thumb.jpg"); let ext = ext_from_path(&row.original_path); @@ -291,13 +312,12 @@ async fn run_html_export( tracing::warn!("thumbnail generation failed for upload {}: {e:#}", row.id); } - // Full variant: compress if >5MB, otherwise copy original + // Full variant: compress to temp if >5MB, otherwise stream the original + // as-is (no temp copy). let src_meta = tokio::fs::metadata(&src).await?; - let full_path = media_tmp.join(&full); - - if src_meta.len() > 5_000_000 { - // Resize to max 2000px + let full_source = if src_meta.len() > 5_000_000 { let src_clone = src.clone(); + let full_path = media_tmp.join(&full); let full_path_clone = full_path.clone(); let compress_result = tokio::task::spawn_blocking(move || -> Result<()> { @@ -311,17 +331,31 @@ async fn run_html_export( }) .await?; - if let Err(e) = compress_result { - tracing::warn!("compression failed for upload {}, copying original: {e:#}", row.id); - tokio::fs::copy(&src, &full_path).await?; + match compress_result { + Ok(()) => MediaSource::Temp(full_path), + Err(e) => { + tracing::warn!( + "compression failed for upload {}, using original: {e:#}", + row.id + ); + MediaSource::Original(src.clone()) + } } } else { - tokio::fs::copy(&src, &full_path).await?; - } + MediaSource::Original(src.clone()) + }; - (thumb, full) + (thumb, full, full_source) }; + // Register this post's two media entries. Thumbnails always come from temp + // (they're freshly generated); the full variant's source was decided above. + media_manifest.push(( + thumb_name.clone(), + MediaSource::Temp(media_tmp.join(&thumb_name)), + )); + media_manifest.push((full_name.clone(), full_source)); + // Build comments for this upload let post_comments: Vec = comments .iter() @@ -409,27 +443,23 @@ async fn run_html_export( update_progress(pool, event_id, "html", 78).await; - // Write media files from temp directory - let mut media_entries = tokio::fs::read_dir(&media_tmp).await?; - let mut file_count = 0u32; + // Write media files from the manifest built in step 3. Thumbnails and derived + // image variants stream from temp; video and small-image full variants stream + // straight from the original on disk. Sources that don't exist (e.g. a thumb + // whose ffmpeg step failed) are skipped — the viewer tolerates gaps. + let file_total = media_manifest.len().max(1) as f32; let mut files_written = 0u32; - // Count files first - { - let mut counter = tokio::fs::read_dir(&media_tmp).await?; - while counter.next_entry().await?.is_some() { - file_count += 1; + for (name, source) in &media_manifest { + let path = source.path(); + if !path.exists() { + continue; } - } - let file_total = file_count.max(1) as f32; - - while let Some(dir_entry) = media_entries.next_entry().await? { - let filename = dir_entry.file_name(); - let entry_name = format!("media/{}", filename.to_string_lossy()); + let entry_name = format!("media/{name}"); let builder = ZipEntryBuilder::new(entry_name.into(), Compression::Stored); let mut zip_entry = zip.write_entry_stream(builder).await?; - let mut f = tokio::fs::File::open(dir_entry.path()).await?.compat(); + let mut f = tokio::fs::File::open(path).await?.compat(); fcopy(&mut f, &mut zip_entry).await?; zip_entry.close().await?; diff --git a/backend/src/services/mod.rs b/backend/src/services/mod.rs index 66f838a..8f7eae2 100644 --- a/backend/src/services/mod.rs +++ b/backend/src/services/mod.rs @@ -1,5 +1,6 @@ pub mod compression; pub mod config; +pub mod disk; pub mod export; pub mod jobs; pub mod maintenance; diff --git a/backend/src/state.rs b/backend/src/state.rs index 7743bf4..917e387 100644 --- a/backend/src/state.rs +++ b/backend/src/state.rs @@ -3,6 +3,8 @@ use tokio::sync::broadcast; use crate::config::AppConfig; use crate::services::compression::CompressionWorker; +use crate::services::config::ConfigCache; +use crate::services::disk::DiskCache; use crate::services::rate_limiter::RateLimiter; use crate::services::sse_tickets::SseTicketStore; @@ -31,6 +33,11 @@ pub struct AppState { pub compression: CompressionWorker, pub rate_limiter: RateLimiter, pub sse_tickets: SseTicketStore, + /// In-memory cache in front of the `config` table. Reads go through here; the + /// admin PATCH handler and the test reseed invalidate it after committing. + pub config_cache: ConfigCache, + /// Cached total/free bytes for the media filesystem (quota + admin stats). + pub disk_cache: DiskCache, } impl AppState { @@ -42,6 +49,7 @@ impl AppState { config.compression_concurrency, sse_tx.clone(), ); + let config_cache = ConfigCache::new(pool.clone()); Self { pool, config, @@ -49,6 +57,8 @@ impl AppState { compression, rate_limiter: RateLimiter::new(), sse_tickets: SseTicketStore::new(), + config_cache, + disk_cache: DiskCache::new(), } } } diff --git a/e2e/specs/06-export/export-video.spec.ts b/e2e/specs/06-export/export-video.spec.ts new file mode 100644 index 0000000..615cb93 --- /dev/null +++ b/e2e/specs/06-export/export-video.spec.ts @@ -0,0 +1,64 @@ +/** + * Covers the HTML export's VIDEO path (perf/stability fix P4). The export used to + * copy every original — including full-size videos — into a temp dir and then re-read + * them into the ZIP (transient 2× disk). It now streams video originals straight from + * disk into the archive. The image-only export specs never exercised that branch, so + * this drives a real video upload → export → and proves the video entry lands in + * Memories.zip (i.e. the streamed-from-original path works and the video isn't dropped). + * + * The fixture clip is <1s, so ffmpeg extracts no thumbnail frame — but exits 0, so the + * compression worker keeps the upload (it isn't auto-cleaned). That's the intended + * shape here: the full video is exported even when its thumbnail is absent. + */ +import { test, expect } from '../../fixtures/test'; +import { uploadRaw } from '../../helpers/upload-client'; +import { readFileSync } from 'node:fs'; +import { join } from 'node:path'; + +const BASE = process.env.E2E_FRONTEND_URL ?? 'http://localhost:3101'; + +test.describe('Export — video streaming (P4)', () => { + test('a real video upload is streamed into Memories.zip (not dropped)', async ({ host }) => { + test.setTimeout(60_000); + const bearer = { Authorization: `Bearer ${host.jwt}` }; + + // Seed a real video upload owned by the host. + const mp4 = readFileSync(join(process.cwd(), 'fixtures', 'media', 'sample.mp4')); + const up = await uploadRaw(host.jwt, mp4, { filename: 'clip.mp4', contentType: 'video/mp4' }); + expect(up.status).toBe(201); + const { id } = await up.json(); + + // Release the gallery → spawns the real zip + html export jobs. + const rel = await fetch(`${BASE}/api/v1/host/gallery/release`, { method: 'POST', headers: bearer }); + expect(rel.status).toBe(204); + + // Wait for the HTML (Memories.zip) job — that's the one whose media staging P4 changed. + await expect + .poll( + async () => { + const res = await fetch(`${BASE}/api/v1/export/status`, { headers: bearer }); + return (await res.json()).html?.status; + }, + { timeout: 45_000, intervals: [500] } + ) + .toBe('done'); + + // Download Memories.zip via the gated single-use ticket. + const ticketRes = await fetch(`${BASE}/api/v1/export/ticket`, { method: 'POST', headers: bearer }); + const { ticket } = await ticketRes.json(); + const dl = await fetch(`${BASE}/api/v1/export/html?ticket=${encodeURIComponent(ticket)}`); + expect(dl.status).toBe(200); + + const bytes = new Uint8Array(await dl.arrayBuffer()); + // Valid ZIP (PK local-file-header magic). + expect(Array.from(bytes.subarray(0, 2))).toEqual([0x50, 0x4b]); + + // ZIP entry names are stored verbatim (uncompressed) in the archive, so the video's + // media entry name appears as plaintext bytes. Its presence means the streamed + // video entry was written; if the stream had failed, entry.close() would have + // errored and the job would never have reached 'done'. + const needle = `media/${id}.mp4`; + const haystack = new TextDecoder('latin1').decode(bytes); + expect(haystack.includes(needle), `Memories.zip must contain ${needle}`).toBe(true); + }); +}); diff --git a/frontend/src/lib/sse.ts b/frontend/src/lib/sse.ts index 1fc2fa7..00a2d42 100644 --- a/frontend/src/lib/sse.ts +++ b/frontend/src/lib/sse.ts @@ -109,6 +109,17 @@ export function connectSse(): void { ); } + // `resync` is emitted by the server when our broadcast subscription fell + // behind and events were dropped. Rather than let those losses leave the feed + // stale, fetch the gap since the last event we actually saw and fan it out as + // a feed-delta (which reconciles new uploads AND deletions). Handled with its + // own listener — not via `dispatch` — so reading `lastEventTime` as the gap + // start isn't clobbered by dispatch bumping it to "now". + eventSource.addEventListener('resync', () => { + const since = lastEventTime; + if (since) void deltaFetchAndFan(since); + }); + eventSource.onerror = () => { // EventSource auto-reconnects but the connection state can stay broken; close // and try again ourselves with exponential backoff capped at 60s. Prevents diff --git a/frontend/src/routes/feed/+page.svelte b/frontend/src/routes/feed/+page.svelte index 4b731c5..c8610e5 100644 --- a/frontend/src/routes/feed/+page.svelte +++ b/frontend/src/routes/feed/+page.svelte @@ -249,6 +249,11 @@ const dead = new Set(delta.deleted_ids); uploads = uploads.filter((u) => !dead.has(u.id)); } + // A delta reconciles new uploads and deletions, but not like/comment + // counts that changed on already-visible cards while we were + // disconnected or lagged (a `resync`). Debounced page-1 refresh merges + // those fresh counts in place without disturbing scroll. + scheduleInPlaceRefresh(); } catch { /* ignore */ } }) );