EventSnap/backend/src/handlers/upload.rs

use std::time::Duration;

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> {
    // Rate limit: N uploads per hour per user
    let upload_rate = get_config_i64(&state.pool, "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, 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;
        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)
}

/// 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() {}
    }
}

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)
}