feat: implement upload pipeline with compression and SSE

Backend:
- POST /api/v1/upload: multipart file upload with caption + hashtags
  - Validates file size against DB config limits (image/video separate)
  - Checks user ban status and event upload lock
  - Saves original to disk under {media_path}/originals/{slug}/
  - Tracks user total_upload_bytes for quota enforcement
  - Extracts hashtags from caption text and explicit CSV field
  - Upserts hashtags and links them to uploads
- PATCH /api/v1/upload/{id}: edit caption and hashtags (owner only)
- DELETE /api/v1/upload/{id}: soft-delete (owner only)
- GET /api/v1/stream: SSE endpoint with 30s keepalive
  - Broadcasts new-upload events to all connected clients
  - Uses tokio broadcast channel for fan-out

Services:
- CompressionWorker: Tokio semaphore-bounded (concurrency=2) background processor
  - Images: resize to 800px wide JPEG preview via image crate
  - PNG originals: lossless compression via oxipng
  - Videos: ffmpeg thumbnail extraction (1 frame at 1s, scaled to 800px)
  - Updates upload record with preview_path/thumbnail_path on completion

Models:
- Upload with full CRUD (create, find, update caption, soft delete, set paths)
- Hashtag with upsert, link/unlink, extract_hashtags() text parser
- UploadDto for API serialization with like/comment counts

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

backend/src/handlers/mod.rs

@@ -0,0 +1,2 @@
+pub mod sse;
+pub mod upload;

backend/src/handlers/sse.rs

@@ -0,0 +1,33 @@
+use std::convert::Infallible;
+use std::time::Duration;
+
+use axum::extract::State;
+use axum::response::sse::{Event, KeepAlive, Sse};
+use futures::stream::Stream;
+use tokio_stream::wrappers::BroadcastStream;
+use tokio_stream::StreamExt;
+
+use crate::auth::middleware::AuthUser;
+use crate::error::AppError;
+use crate::state::AppState;
+
+pub async fn stream(
+    State(state): State<AppState>,
+    _auth: AuthUser,
+) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, AppError> {
+    let rx = state.sse_tx.subscribe();
+    let stream = BroadcastStream::new(rx).filter_map(|msg| {
+        match msg {
+            Ok(sse_event) => Some(Ok(Event::default()
+                .event(sse_event.event_type)
+                .data(sse_event.data))),
+            Err(_) => None, // Lagged — skip missed events
+        }
+    });
+
+    Ok(Sse::new(stream).keep_alive(
+        KeepAlive::new()
+            .interval(Duration::from_secs(30))
+            .text("ping"),
+    ))
+}

backend/src/handlers/upload.rs

@@ -0,0 +1,237 @@
+use axum::extract::{Multipart, Path, State};
+use axum::http::StatusCode;
+use axum::Json;
+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::state::AppState;
+
+pub async fn upload(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    mut multipart: Multipart,
+) -> Result<(StatusCode, Json<UploadDto>), AppError> {
+    // 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 {
+        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() {
+        return Err(AppError::Forbidden("Uploads sind gesperrt.".into()));
+    }
+
+    // Read config limits from DB
+    let max_image_mb: i64 = get_config_i64(&state.pool, "max_image_size_mb", 20).await;
+    let max_video_mb: i64 = get_config_i64(&state.pool, "max_video_size_mb", 500).await;
+
+    let mut file_data: Option<Vec<u8>> = None;
+    let mut file_name: Option<String> = None;
+    let mut content_type: Option<String> = None;
+    let mut caption: Option<String> = None;
+    let mut hashtags_csv: Option<String> = 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" => {
+                file_name = field.file_name().map(|s| s.to_string());
+                content_type = field.content_type().map(|s| s.to_string());
+                file_data = Some(
+                    field.bytes().await
+                        .map_err(|e| AppError::BadRequest(format!("Datei konnte nicht gelesen werden: {e}")))?
+                        .to_vec(),
+                );
+            }
+            "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()))?,
+                );
+            }
+            _ => {}
+        }
+    }
+
+    let data = file_data.ok_or_else(|| AppError::BadRequest("Keine Datei hochgeladen.".into()))?;
+    let mime = content_type.unwrap_or_else(|| "application/octet-stream".to_string());
+    let size = data.len() as i64;
+
+    // Validate file size
+    let max_bytes = if mime.starts_with("video/") {
+        max_video_mb * 1024 * 1024
+    } else {
+        max_image_mb * 1024 * 1024
+    };
+    if size > max_bytes {
+        return Err(AppError::BadRequest(format!(
+            "Datei ist zu groß. Maximum: {} MB.",
+            max_bytes / (1024 * 1024)
+        )));
+    }
+
+    // Determine file extension
+    let ext = file_name
+        .as_deref()
+        .and_then(|n| n.rsplit('.').next())
+        .unwrap_or(if mime.starts_with("video/") { "mp4" } else { "jpg" });
+
+    let upload_id = Uuid::new_v4();
+    let event_slug = &state.config.event_slug;
+    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()))?;
+
+    // Update user's total upload bytes
+    sqlx::query("UPDATE \"user\" SET total_upload_bytes = total_upload_bytes + $2 WHERE id = $1")
+        .bind(auth.user_id)
+        .bind(size)
+        .execute(&state.pool)
+        .await?;
+
+    // Insert upload record
+    let upload = Upload::create(
+        &state.pool,
+        auth.event_id,
+        auth.user_id,
+        &relative_path,
+        &mime,
+        size,
+        caption.as_deref(),
+    )
+    .await?;
+
+    // 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();
+
+    for tag in &tags {
+        let h = Hashtag::upsert(&state.pool, auth.event_id, tag).await?;
+        Hashtag::link_to_upload(&state.pool, upload.id, h.id).await?;
+    }
+
+    // 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 {
+        event_type: "new-upload".to_string(),
+        data: 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> {
+    let upload = Upload::find_by_id(&state.pool, upload_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()));
+    }
+
+    if let Some(ref caption) = body.caption {
+        Upload::update_caption(&state.pool, upload_id, Some(caption)).await?;
+    }
+
+    if let Some(ref hashtags) = body.hashtags {
+        Hashtag::unlink_all_from_upload(&state.pool, upload_id).await?;
+        for tag in hashtags {
+            let h = Hashtag::upsert(&state.pool, auth.event_id, tag).await?;
+            Hashtag::link_to_upload(&state.pool, upload_id, h.id).await?;
+        }
+    }
+
+    Ok(StatusCode::OK)
+}
+
+pub async fn delete_upload(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    Path(upload_id): Path<Uuid>,
+) -> Result<StatusCode, AppError> {
+    let upload = Upload::find_by_id(&state.pool, upload_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()));
+    }
+
+    Upload::soft_delete(&state.pool, upload_id).await?;
+
+    Ok(StatusCode::NO_CONTENT)
+}
+
+async fn get_config_i64(pool: &sqlx::PgPool, key: &str, default: i64) -> i64 {
+    let row: Option<(String,)> =
+        sqlx::query_as("SELECT value FROM config WHERE key = $1")
+            .bind(key)
+            .fetch_optional(pool)
+            .await
+            .unwrap_or(None);
+    row.and_then(|r| r.0.parse().ok()).unwrap_or(default)
+}