Diashow completeness rewrite so every eligible upload is shown regardless of
bursts, disconnects, or library size:
- queue.ts: SlideQueue with live/shuffle queues, allKnown map, recentlyShown
ring; merge(dedup, live-first), remove/removeByUser (prunes recentlyShown),
knownIds for reconcile-eviction. Adds queue.test.ts (burst/completeness/race).
- diashow/+page.svelte: reconcile (full paginate + evict, pre-scan snapshot to
spare concurrent uploads) on mount/reconnect/periodic; catchUpNew paginate-
until-known for bursts with debounced maxWait; hard-cut removals; decode
timeout + candidate fallback + bounded skip so a broken image never stalls.
New ~2048px "display" derivative for big-screen sharpness, decoupled from the
data-saver preview (800px) used on phones:
- migration 016: upload.display_path + v_feed rebuilt (DROP+CREATE, not REPLACE,
to slot the column beside preview/thumbnail).
- compression: generate_image_derivatives emits preview+display (downscale-only
guard, no upscaling); backfill_missing_display regenerates on startup (safe:
logs on error, never soft-deletes).
- upload.rs/main.rs: GET /upload/{id}/display (mirrors preview auth/cache),
/media/displays direct-serve blocked.
- feed.rs + types.ts: display_url in feed/delta DTOs.
- diashow candidate chain: display -> original -> preview.
Verified on the running stack: migration applied, 10/10 existing images
backfilled (2048px cap honoured, small images not upscaled), /display serves
200, /feed returns display_url, diashow cycles.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
829 lines
33 KiB
Rust
829 lines
33 KiB
Rust
use std::time::Duration;
|
|
|
|
use axum::Json;
|
|
use axum::extract::{Multipart, Path, State};
|
|
use axum::http::StatusCode;
|
|
use chrono::{DateTime, Utc};
|
|
use serde::Deserialize;
|
|
use uuid::Uuid;
|
|
|
|
use crate::auth::middleware::AuthUser;
|
|
use crate::error::AppError;
|
|
use crate::models::hashtag::{self, Hashtag};
|
|
use crate::models::upload::{Upload, UploadDto};
|
|
use crate::models::user::User;
|
|
use crate::services::config;
|
|
use crate::state::AppState;
|
|
|
|
const MAX_CAPTION_LENGTH: usize = 2000;
|
|
|
|
/// Allowlist of accepted media types, keyed by the MIME that `infer` derives from
|
|
/// the file's magic bytes. The detected MIME (not the client-declared one) is what
|
|
/// we trust, store, and hand to the compression pipeline — so a text-based payload
|
|
/// (SVG/HTML/JS) can never be stored or served on-origin. Each entry maps to the
|
|
/// server-controlled file extension we persist the original under.
|
|
///
|
|
/// HEIC/HEIF are deliberately excluded: the preview pipeline (`image` crate, and
|
|
/// the bundled ffmpeg 6.1) cannot decode them, so accepting them would store files
|
|
/// that never get a thumbnail. iOS Safari already transcodes HEIC→JPEG when a photo
|
|
/// is selected via a file input, so this rejects only the rare HEIC-preserving
|
|
/// upload path — with a clear error rather than a silently broken post.
|
|
const ALLOWED_MEDIA: &[(&str, &str)] = &[
|
|
("image/jpeg", "jpg"),
|
|
("image/png", "png"),
|
|
("image/webp", "webp"),
|
|
("image/gif", "gif"),
|
|
("video/mp4", "mp4"),
|
|
("video/quicktime", "mov"),
|
|
("video/webm", "webm"),
|
|
];
|
|
|
|
pub async fn upload(
|
|
State(state): State<AppState>,
|
|
auth: AuthUser,
|
|
mut multipart: Multipart,
|
|
) -> Result<(StatusCode, Json<UploadDto>), AppError> {
|
|
// Rate limit: N uploads per hour per user. Gated by master + per-endpoint toggles.
|
|
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", 100).await as usize;
|
|
if let Err(retry_after_secs) = state.rate_limiter.check_with_retry(
|
|
format!("upload:{}", auth.user_id),
|
|
upload_rate,
|
|
Duration::from_secs(3600),
|
|
) {
|
|
drain_multipart(multipart).await;
|
|
return Err(AppError::TooManyRequests(
|
|
"Du hast dein Upload-Limit für diese Stunde erreicht.".into(),
|
|
Some(retry_after_secs),
|
|
));
|
|
}
|
|
}
|
|
|
|
// Check if user is banned
|
|
let user = User::find_by_id(&state.pool, auth.user_id)
|
|
.await?
|
|
.ok_or_else(|| AppError::NotFound("Benutzer nicht gefunden.".into()))?;
|
|
if user.is_banned {
|
|
drain_multipart(multipart).await;
|
|
return Err(AppError::Forbidden("Du bist gesperrt.".into()));
|
|
}
|
|
|
|
// Check if uploads are locked
|
|
let event = crate::models::event::Event::find_by_slug(&state.pool, &state.config.event_slug)
|
|
.await?
|
|
.ok_or_else(|| AppError::NotFound("Event nicht gefunden.".into()))?;
|
|
if event.uploads_locked_at.is_some() {
|
|
drain_multipart(multipart).await;
|
|
// Reversible: a host can reopen the event, so the client keeps the queued blob and
|
|
// retries on `event-opened` rather than purging it (UploadsLocked, not Forbidden).
|
|
return Err(AppError::UploadsLocked("Uploads sind gesperrt.".into()));
|
|
}
|
|
// Belt-and-suspenders on top of the lock (release ⇒ lock): once the gallery is
|
|
// released the export has been snapshotted, so a late upload could never make it into
|
|
// the keepsake. Reject it explicitly rather than silently diverging the live feed.
|
|
// Also reversible (reopen clears `export_released_at`), so likewise UploadsLocked.
|
|
if event.export_released_at.is_some() {
|
|
drain_multipart(multipart).await;
|
|
return Err(AppError::UploadsLocked(
|
|
"Galerie wurde bereits freigegeben.".into(),
|
|
));
|
|
}
|
|
|
|
// Read config limits from DB
|
|
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;
|
|
|
|
// 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<u8>)> = None; // (size, head bytes for sniffing)
|
|
let mut caption: Option<String> = None;
|
|
let mut hashtags_csv: Option<String> = None;
|
|
|
|
// 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()))?,
|
|
);
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
.await;
|
|
|
|
if let Err(e) = parse_result {
|
|
let _ = tokio::fs::remove_file(&temp_abs).await;
|
|
return Err(e);
|
|
}
|
|
|
|
// 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
|
|
&& 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
|
|
)));
|
|
}
|
|
|
|
// Determine the file type from its magic bytes and require it to be on the
|
|
// allowlist. `infer` returns None for text-based payloads (SVG/HTML/JS), so
|
|
// 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 = 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))
|
|
{
|
|
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 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)
|
|
)));
|
|
}
|
|
|
|
// 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.config_cache, "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
|
|
// pre-check and both increment, blowing past the quota. The pre-check stays as a
|
|
// fast path that avoids the disk write when the user is already clearly over.
|
|
let mut quota_limit: Option<i64> = None;
|
|
if quota_on && storage_quota_on {
|
|
let estimate = compute_storage_quota(&state).await;
|
|
if let Some(limit) = estimate.limit_bytes {
|
|
quota_limit = Some(limit);
|
|
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::QuotaExceeded(
|
|
"Du hast dein Upload-Limit für dieses Event erreicht.".into(),
|
|
));
|
|
}
|
|
}
|
|
}
|
|
|
|
// 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);
|
|
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<String> = Vec::new();
|
|
if let Some(ref cap) = caption {
|
|
tags.extend(hashtag::extract_hashtags(cap));
|
|
}
|
|
if let Some(ref csv) = hashtags_csv {
|
|
for tag in csv.split(',') {
|
|
let t = tag.trim().trim_start_matches('#').to_lowercase();
|
|
if !t.is_empty() {
|
|
tags.push(t);
|
|
}
|
|
}
|
|
}
|
|
tags.sort();
|
|
tags.dedup();
|
|
|
|
// 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 tx_result: Result<Upload, AppError> = async {
|
|
let mut tx = state.pool.begin().await?;
|
|
|
|
// RE-CHECK THE LOCK, UNDER A ROW LOCK, INSIDE THE COMMIT TX.
|
|
//
|
|
// The pre-flight check at the top of this handler ran BEFORE we streamed the body — which
|
|
// for a 500 MB video is minutes. Trusting it here is a TOCTOU that silently loses photos
|
|
// from the keepsake, and it is the real cause of the "stale keepsake" bug that survived
|
|
// three rounds of fixes inside the export state machine:
|
|
//
|
|
// 1. guest starts a big upload; the lock check passes (event open)
|
|
// 2. host releases the gallery → uploads lock, export workers snapshot the uploads table
|
|
// 3. this upload commits AFTER that snapshot → it shows up in the live feed but is
|
|
// MISSING from the downloaded keepsake, permanently (nothing ever regenerates it)
|
|
//
|
|
// `FOR SHARE` conflicts with the `UPDATE event` in `release_gallery`, which serializes us
|
|
// against it. Either we take the lock first — and release (hence the export snapshot) is
|
|
// strictly ordered after our commit, so the snapshot CONTAINS this upload — or release
|
|
// commits first and we observe the lock here and reject. Either way the keepsake is
|
|
// complete. `UploadsLocked` (not Forbidden) is reversible: the client keeps the blob and
|
|
// resumes it when the host reopens.
|
|
let (locked_at, released_at): (Option<DateTime<Utc>>, Option<DateTime<Utc>>) =
|
|
sqlx::query_as(
|
|
"SELECT uploads_locked_at, export_released_at FROM event WHERE id = $1 FOR SHARE",
|
|
)
|
|
.bind(auth.event_id)
|
|
.fetch_one(&mut *tx)
|
|
.await?;
|
|
if locked_at.is_some() || released_at.is_some() {
|
|
return Err(AppError::UploadsLocked("Uploads sind gesperrt.".into()));
|
|
}
|
|
|
|
// Increment the user's byte total. When a quota is in force, guard it atomically
|
|
// (`total + size <= limit`) so two concurrent uploads can't both slip past the
|
|
// stale pre-check — the loser's UPDATE matches 0 rows and we abort with the same
|
|
// terminal quota error (the tx rolls back on drop; the on-disk file is cleaned by
|
|
// the error path below).
|
|
let inc = if let Some(limit) = quota_limit {
|
|
sqlx::query(
|
|
"UPDATE \"user\" SET total_upload_bytes = total_upload_bytes + $2
|
|
WHERE id = $1 AND total_upload_bytes + $2 <= $3",
|
|
)
|
|
.bind(auth.user_id)
|
|
.bind(size)
|
|
.bind(limit)
|
|
.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?
|
|
};
|
|
if inc.rows_affected() == 0 {
|
|
return Err(AppError::QuotaExceeded(
|
|
"Du hast dein Upload-Limit für dieses Event erreicht.".into(),
|
|
));
|
|
}
|
|
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?;
|
|
}
|
|
tx.commit().await?;
|
|
Ok(upload)
|
|
}
|
|
.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
|
|
.compression
|
|
.process(upload.id, relative_path, mime.clone());
|
|
|
|
// Broadcast SSE event
|
|
let dto = UploadDto {
|
|
id: upload.id,
|
|
user_id: auth.user_id,
|
|
uploader_name: user.display_name,
|
|
preview_url: None,
|
|
thumbnail_url: None,
|
|
mime_type: mime,
|
|
caption,
|
|
hashtags: tags,
|
|
like_count: 0,
|
|
comment_count: 0,
|
|
liked_by_me: false,
|
|
created_at: upload.created_at,
|
|
};
|
|
|
|
let _ = state.sse_tx.send(crate::state::SseEvent::new(
|
|
"new-upload",
|
|
serde_json::to_string(&dto).unwrap_or_default(),
|
|
));
|
|
|
|
Ok((StatusCode::CREATED, Json(dto)))
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
pub struct EditUploadRequest {
|
|
pub caption: Option<String>,
|
|
pub hashtags: Option<Vec<String>>,
|
|
}
|
|
|
|
pub async fn edit_upload(
|
|
State(state): State<AppState>,
|
|
auth: AuthUser,
|
|
Path(upload_id): Path<Uuid>,
|
|
Json(body): Json<EditUploadRequest>,
|
|
) -> Result<StatusCode, AppError> {
|
|
// Banned users keep read access but cannot mutate (USER_JOURNEYS §10).
|
|
if auth.is_banned {
|
|
return Err(AppError::Forbidden("Du bist gesperrt.".into()));
|
|
}
|
|
let upload = Upload::find_by_id_and_event(&state.pool, upload_id, auth.event_id)
|
|
.await?
|
|
.ok_or_else(|| AppError::NotFound("Upload nicht gefunden.".into()))?;
|
|
|
|
if upload.user_id != auth.user_id {
|
|
return Err(AppError::Forbidden("Nur eigene Uploads bearbeiten.".into()));
|
|
}
|
|
|
|
// Caption update + hashtag wipe-then-relink in one transaction, so a crash
|
|
// mid-relink can't leave the upload with its hashtags stripped.
|
|
//
|
|
// Editing is intentionally allowed while uploads are locked or the gallery is released — like
|
|
// comments and likes, the lock freezes *new uploads* only (USER_JOURNEYS §9.3). But a caption
|
|
// is embedded in the HTML viewer keepsake (the ZIP holds media only — see export.rs), so an
|
|
// edit AFTER release must regenerate the viewer, or the downloadable keepsake keeps showing the
|
|
// old caption forever while the live feed shows the new one. Same atomicity as delete_upload:
|
|
// the edit and its invalidation share one tx so a dropped handler can't leave them disagreeing.
|
|
// `Affects::ViewerOnly` carries the finished ZIP forward (the media didn't change); when the
|
|
// gallery isn't released, `invalidate_and_arm` returns None and this is a no-op.
|
|
let mut tx = state.pool.begin().await?;
|
|
if let Some(ref caption) = body.caption {
|
|
Upload::update_caption(&mut *tx, upload_id, Some(caption)).await?;
|
|
}
|
|
if let Some(ref hashtags) = body.hashtags {
|
|
Hashtag::unlink_all_from_upload(&mut *tx, upload_id).await?;
|
|
for tag in hashtags {
|
|
let h = Hashtag::upsert(&mut *tx, auth.event_id, tag).await?;
|
|
Hashtag::link_to_upload(&mut *tx, upload_id, h.id).await?;
|
|
}
|
|
}
|
|
let regen = crate::services::export::invalidate_and_arm(
|
|
&mut tx,
|
|
&state.config.event_slug,
|
|
crate::services::export::Affects::ViewerOnly,
|
|
)
|
|
.await?;
|
|
tx.commit().await?;
|
|
if let Some(r) = regen {
|
|
crate::handlers::host::start_regen(&state, r);
|
|
}
|
|
|
|
Ok(StatusCode::OK)
|
|
}
|
|
|
|
pub async fn delete_upload(
|
|
State(state): State<AppState>,
|
|
auth: AuthUser,
|
|
Path(upload_id): Path<Uuid>,
|
|
) -> Result<StatusCode, AppError> {
|
|
// Banned users keep read access but cannot mutate (USER_JOURNEYS §10).
|
|
if auth.is_banned {
|
|
return Err(AppError::Forbidden("Du bist gesperrt.".into()));
|
|
}
|
|
let upload = Upload::find_by_id_and_event(&state.pool, upload_id, auth.event_id)
|
|
.await?
|
|
.ok_or_else(|| AppError::NotFound("Upload nicht gefunden.".into()))?;
|
|
|
|
if upload.user_id != auth.user_id {
|
|
return Err(AppError::Forbidden("Nur eigene Uploads löschen.".into()));
|
|
}
|
|
|
|
// Atomic with the keepsake invalidation: a guest removing their own photo must have it removed
|
|
// from the downloadable archive too, and a half-applied delete would leave it there forever.
|
|
let mut tx = state.pool.begin().await?;
|
|
Upload::soft_delete_in_event(&mut tx, upload_id, auth.event_id).await?;
|
|
let regen = crate::services::export::invalidate_and_arm(
|
|
&mut tx,
|
|
&state.config.event_slug,
|
|
crate::services::export::Affects::Both,
|
|
)
|
|
.await?;
|
|
tx.commit().await?;
|
|
if let Some(r) = regen {
|
|
crate::handlers::host::start_regen(&state, r);
|
|
}
|
|
|
|
// Evict the card live on every other feed + the projector diashow — otherwise
|
|
// a self-deleted post lingers until each viewer manually reloads. Same event
|
|
// the host-delete path already emits and the frontend already handles.
|
|
let _ = state.sse_tx.send(crate::state::SseEvent::new(
|
|
"upload-deleted",
|
|
serde_json::json!({ "upload_id": upload_id }).to_string(),
|
|
));
|
|
|
|
Ok(StatusCode::NO_CONTENT)
|
|
}
|
|
|
|
/// 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<(i64, Vec<u8>), AppError> {
|
|
use tokio::io::AsyncWriteExt;
|
|
|
|
let mut file = tokio::fs::File::create(dest)
|
|
.await
|
|
.map_err(|e| AppError::Internal(e.into()))?;
|
|
let mut total: usize = 0;
|
|
let mut head: Vec<u8> = 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)
|
|
)));
|
|
}
|
|
|
|
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()));
|
|
}
|
|
}
|
|
|
|
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.
|
|
/// Without draining, the client may still be sending the body after we've sent our response,
|
|
/// which can corrupt the keep-alive connection for subsequent requests.
|
|
async fn drain_multipart(mut mp: Multipart) {
|
|
while let Ok(Some(mut field)) = mp.next_field().await {
|
|
while field.chunk().await.ok().flatten().is_some() {}
|
|
}
|
|
}
|
|
|
|
/// Snapshot of the dynamic per-user quota used both by the upload pre-check and the
|
|
/// `GET /me/quota` endpoint. `limit_bytes = None` means quota enforcement is currently
|
|
/// off (the frontend hides the widget in that case).
|
|
pub struct QuotaEstimate {
|
|
pub limit_bytes: Option<i64>,
|
|
pub active_uploaders: i64,
|
|
pub free_disk_bytes: i64,
|
|
/// The tolerance factor the limit above was computed with. Carried on the snapshot so the
|
|
/// number is self-describing; no caller reads it back today.
|
|
#[allow(dead_code)]
|
|
pub tolerance: f64,
|
|
}
|
|
|
|
/// Pure per-user quota formula: `floor((free_disk * tolerance) / max(active, 1))`.
|
|
/// Extracted from `compute_storage_quota` so it's unit-testable without a DB or disk.
|
|
fn quota_limit_bytes(free_disk: i64, tolerance: f64, active_uploaders: i64) -> i64 {
|
|
let active = active_uploaders.max(1);
|
|
((free_disk as f64 * tolerance) / active as f64).floor() as i64
|
|
}
|
|
|
|
/// Computes the per-user storage quota using
|
|
/// `floor((free_disk * tolerance) / max(active_uploaders, 1))`. Returns `limit_bytes =
|
|
/// 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.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")
|
|
.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.
|
|
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 {
|
|
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
|
|
};
|
|
|
|
QuotaEstimate {
|
|
limit_bytes,
|
|
active_uploaders: active,
|
|
free_disk_bytes: free_disk,
|
|
tolerance,
|
|
}
|
|
}
|
|
|
|
/// Stream a media file from disk into an HTTP response with a fixed set of security
|
|
/// headers. Every media response (original, preview, thumbnail) goes through here so
|
|
/// they consistently carry `X-Content-Type-Options: nosniff` (defense-in-depth against
|
|
/// content-type confusion, even if the edge proxy is bypassed) plus an explicit
|
|
/// `Content-Disposition` and `Cache-Control`.
|
|
async fn stream_media_file(
|
|
absolute: &std::path::Path,
|
|
content_type: String,
|
|
disposition: &str,
|
|
cache_control: &str,
|
|
) -> Result<axum::response::Response, AppError> {
|
|
use axum::body::Body;
|
|
use axum::http::{Response, StatusCode, header};
|
|
use tokio_util::io::ReaderStream;
|
|
|
|
if !absolute.exists() {
|
|
return Err(AppError::NotFound("Datei nicht gefunden.".into()));
|
|
}
|
|
|
|
let file = tokio::fs::File::open(absolute)
|
|
.await
|
|
.map_err(|e| AppError::Internal(e.into()))?;
|
|
let metadata = file
|
|
.metadata()
|
|
.await
|
|
.map_err(|e| AppError::Internal(e.into()))?;
|
|
let stream = ReaderStream::new(file);
|
|
|
|
Response::builder()
|
|
.status(StatusCode::OK)
|
|
.header(header::CONTENT_TYPE, content_type)
|
|
.header(header::CONTENT_DISPOSITION, disposition)
|
|
.header(header::CONTENT_LENGTH, metadata.len())
|
|
.header(header::CACHE_CONTROL, cache_control)
|
|
.header(header::X_CONTENT_TYPE_OPTIONS, "nosniff")
|
|
.body(Body::from_stream(stream))
|
|
.map_err(|e| AppError::Internal(e.into()))
|
|
}
|
|
|
|
/// Streaming download of the original file behind an upload. Used by:
|
|
/// - the per-post "Original anzeigen" context action (`window.open`)
|
|
/// - `<img src>` / `<video src>` in the feed, lightbox, and diashow when the user is in
|
|
/// Data Mode = Original
|
|
///
|
|
/// **Auth model:** the route is intentionally unauthenticated so it works from
|
|
/// `<img src>` / `window.open`. The URL contains the upload's unguessable UUID. Unlike
|
|
/// raw `/media` files, this alias is the *only* way to fetch an original: direct
|
|
/// `/media/originals/**` access is blocked in the router, and this handler filters out
|
|
/// soft-deleted and ban-hidden uploads (via `find_visible_media`) so moderation actually
|
|
/// removes access to content. Preview and thumbnail variants are gated the same way (see
|
|
/// [`get_preview`] / [`get_thumbnail`]).
|
|
pub async fn get_original(
|
|
State(state): State<AppState>,
|
|
Path(upload_id): Path<Uuid>,
|
|
) -> Result<axum::response::Response, AppError> {
|
|
let media = Upload::find_visible_media(&state.pool, upload_id)
|
|
.await?
|
|
.ok_or_else(|| AppError::NotFound("Upload nicht gefunden.".into()))?;
|
|
|
|
let absolute = state.config.media_path.join(&media.original_path);
|
|
let filename = absolute
|
|
.file_name()
|
|
.and_then(|n| n.to_str())
|
|
.unwrap_or("original");
|
|
let disposition = format!("attachment; filename=\"{filename}\"");
|
|
|
|
// Full-res original: force download, never cache at the edge.
|
|
stream_media_file(&absolute, media.mime_type, &disposition, "no-store").await
|
|
}
|
|
|
|
/// Streaming access to an upload's compressed **preview** image. Gated exactly like
|
|
/// [`get_original`]: `find_visible_media` drops soft-deleted / ban-hidden uploads, so a
|
|
/// deleted or moderated post's preview 404s here even though the file may still be on
|
|
/// disk. Direct `/media/previews/**` is blocked in the router, making this the only path
|
|
/// to a preview.
|
|
///
|
|
/// Served inline (it's an `<img src>`) and privately cacheable for a short window. The
|
|
/// window is intentionally short (5 min) so a moderated image stops being served to a
|
|
/// direct-URL holder promptly; the live feed already evicts the card instantly via the
|
|
/// `upload-deleted` / `user-hidden` SSE events, so this only bounds the raw-URL edge case.
|
|
pub async fn get_preview(
|
|
State(state): State<AppState>,
|
|
Path(upload_id): Path<Uuid>,
|
|
) -> Result<axum::response::Response, AppError> {
|
|
let media = Upload::find_visible_media(&state.pool, upload_id)
|
|
.await?
|
|
.ok_or_else(|| AppError::NotFound("Upload nicht gefunden.".into()))?;
|
|
let rel = media
|
|
.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
|
|
}
|
|
|
|
/// Streaming access to an upload's big-screen **display** derivative (~2048px), used by the
|
|
/// diashow. Gated identically to [`get_preview`]. 404s when the derivative doesn't exist yet
|
|
/// (still compressing, or an old upload the backfill hasn't reached) — the diashow then falls
|
|
/// back to the original.
|
|
pub async fn get_display(
|
|
State(state): State<AppState>,
|
|
Path(upload_id): Path<Uuid>,
|
|
) -> Result<axum::response::Response, AppError> {
|
|
let media = Upload::find_visible_media(&state.pool, upload_id)
|
|
.await?
|
|
.ok_or_else(|| AppError::NotFound("Upload nicht gefunden.".into()))?;
|
|
let rel = media
|
|
.display_path
|
|
.ok_or_else(|| AppError::NotFound("Anzeige 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
|
|
}
|
|
|
|
/// Streaming access to an upload's **thumbnail** (video poster). Gated identically to
|
|
/// [`get_preview`].
|
|
pub async fn get_thumbnail(
|
|
State(state): State<AppState>,
|
|
Path(upload_id): Path<Uuid>,
|
|
) -> Result<axum::response::Response, AppError> {
|
|
let media = Upload::find_visible_media(&state.pool, upload_id)
|
|
.await?
|
|
.ok_or_else(|| AppError::NotFound("Upload nicht gefunden.".into()))?;
|
|
let rel = media
|
|
.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
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::quota_limit_bytes;
|
|
|
|
#[test]
|
|
fn divides_free_space_by_uploaders_with_tolerance() {
|
|
// 1000 * 0.75 / 3 = 250
|
|
assert_eq!(quota_limit_bytes(1000, 0.75, 3), 250);
|
|
}
|
|
|
|
#[test]
|
|
fn floors_fractional_results() {
|
|
// 1000 * 0.75 / 7 = 107.14… → 107
|
|
assert_eq!(quota_limit_bytes(1000, 0.75, 7), 107);
|
|
}
|
|
|
|
#[test]
|
|
fn active_uploaders_below_one_is_clamped_to_one() {
|
|
// Guards against divide-by-zero when no one has uploaded yet.
|
|
assert_eq!(quota_limit_bytes(1000, 1.0, 0), 1000);
|
|
assert_eq!(quota_limit_bytes(1000, 1.0, -5), 1000);
|
|
}
|
|
|
|
#[test]
|
|
fn zero_free_disk_yields_zero() {
|
|
assert_eq!(quota_limit_bytes(0, 0.75, 3), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn full_tolerance_is_identity_for_a_single_uploader() {
|
|
assert_eq!(quota_limit_bytes(500, 1.0, 1), 500);
|
|
}
|
|
}
|