//! Image upload and serving handlers.

use axum::{
    body::Body,
    extract::{Multipart, Path, Query},
    http::{StatusCode, header},
    response::{IntoResponse, Response},
};
use serde::Deserialize;

use mt_core::types::{ModAction, ModActor};

use crate::AppState;
use crate::auth::MaybeUser;
use crate::storage;

use super::{
    check_community_access, check_write_access, db_error, get_community, get_role, is_mod_or_owner,
};

/// Max uploads per user per hour.
const UPLOAD_RATE_LIMIT: i64 = 20;
const UPLOAD_RATE_WINDOW_SECS: i64 = 3600;

/// POST /p/{slug}/upload, multipart image upload, returns JSON with markdown link.
#[tracing::instrument(skip_all)]
pub(super) async fn upload_image_handler(
    axum::extract::State(state): axum::extract::State<AppState>,
    Path(slug): Path<String>,
    MaybeUser(session_user): MaybeUser,
    mut multipart: Multipart,
) -> Result<impl IntoResponse, Response> {
    let user = session_user.ok_or_else(|| StatusCode::UNAUTHORIZED.into_response())?;

    let s3 = state.s3.as_ref().ok_or_else(|| {
        (
            StatusCode::SERVICE_UNAVAILABLE,
            "Image uploads are not configured.",
        )
            .into_response()
    })?;

    let community = get_community(&state.db, &slug).await?;
    // Uploading is a write: it stores an object, spends the caller's rate-limit
    // budget, and yields a hosted URL to embed in a post. Same gate as posting,
    // so a platform-suspended or muted user cannot stage images they are not
    // allowed to publish.
    check_write_access(
        &state.db,
        community.id,
        user.user_id,
        community.suspended_at.is_some(),
    )
    .await?;

    // Check membership
    let role = get_role(&state.db, user.user_id, community.id).await?;
    if role.is_none() {
        return Err((
            StatusCode::FORBIDDEN,
            "You must be a community member to upload.",
        )
            .into_response());
    }

    // Rate limit: uploads per hour
    let recent = mt_db::queries::count_recent_uploads_by_user(
        &state.db,
        user.user_id,
        UPLOAD_RATE_WINDOW_SECS,
    )
    .await
    .map_err(db_error)?;
    if recent >= UPLOAD_RATE_LIMIT {
        return Err((
            StatusCode::TOO_MANY_REQUESTS,
            "Upload limit reached. Try again later.",
        )
            .into_response());
    }

    // Read the multipart field
    let mut field = multipart
        .next_field()
        .await
        .map_err(|e| {
            tracing::error!(error = ?e, "multipart read error");
            (StatusCode::BAD_REQUEST, "Invalid upload.").into_response()
        })?
        .ok_or_else(|| (StatusCode::BAD_REQUEST, "No file provided.").into_response())?;

    let filename = field.file_name().unwrap_or("image").to_string();
    let content_type = field
        .content_type()
        .unwrap_or("application/octet-stream")
        .to_string();

    // Read the field chunk-by-chunk, bailing the instant we cross MAX_IMAGE_SIZE
    // rather than buffering the whole field first (`field.bytes()`) and checking
    // afterwards. The `/p/{slug}/upload` route already carries a `DefaultBodyLimit`
    // (routes/mod.rs), but that is a separately-configured layer that could drift;
    // enforcing the same cap here keeps the handler self-defending against a
    // memory-DoS regardless of the routing setup.
    let mut data: Vec<u8> = Vec::new();
    loop {
        match field.chunk().await {
            Ok(Some(chunk)) => {
                if data.len() + chunk.len() > storage::MAX_IMAGE_SIZE {
                    return Err((
                        StatusCode::PAYLOAD_TOO_LARGE,
                        "Image exceeds the 5 MB limit.",
                    )
                        .into_response());
                }
                data.extend_from_slice(&chunk);
            }
            Ok(None) => break,
            Err(e) => {
                tracing::error!(error = ?e, "failed to read upload bytes");
                return Err((StatusCode::BAD_REQUEST, "Failed to read file.").into_response());
            }
        }
    }

    let (ext, validated_ct) = storage::validate_image(&filename, &content_type, &data)
        .map_err(|msg| (StatusCode::UNPROCESSABLE_ENTITY, msg).into_response())?;

    // Strip EXIF from JPEG
    let data = if ext == "jpg" {
        storage::strip_exif_jpeg(&data)
    } else {
        data
    };

    let s3_key = storage::generate_image_key(&slug, ext);
    let data_len = data.len() as i64;

    // Record the DB row BEFORE the S3 upload. If we uploaded first and the
    // insert then failed, the S3 object would have no row and the reconcile
    // sweep (keyed on `images`) could never find it, an unbounded orphan.
    // Insert-first inverts the failure mode: a failed/absent upload leaves at
    // most a row with no object, which is bounded, queryable, and cleaned up
    // below (or by the sweep), never a silent S3 cost.
    let image_id = mt_db::mutations::insert_image(
        &state.db,
        user.user_id,
        community.id,
        &s3_key,
        &filename,
        validated_ct,
        data_len,
    )
    .await
    .map_err(db_error)?;

    // Upload to S3. On failure, drop the row we just inserted so neither store
    // is left holding a dangling reference; if the cleanup delete itself fails,
    // the row remains pointing at a missing object, recoverable, not a
    // leaked S3 object.
    if let Err(e) = s3.upload(&s3_key, validated_ct, data).await {
        tracing::error!(error = %e, "S3 upload failed");
        if let Err(del) = mt_db::mutations::delete_image_row(&state.db, image_id).await {
            tracing::warn!(error = ?del, image_id = %image_id, "failed to roll back image row after S3 upload failure");
        }
        return Err(StatusCode::INTERNAL_SERVER_ERROR.into_response());
    }

    // Return JSON with the image URL for markdown insertion
    let url = format!("/uploads/{image_id}");
    let markdown = format!("![{filename}]({url})");

    Ok(axum::Json(serde_json::json!({
        "url": url,
        "markdown": markdown,
        "id": image_id.to_string(),
    })))
}

/// GET /uploads/{id}, serve an uploaded image (proxied from S3).
///
/// Enforces the owning community's access policy: a suspended community or a
/// caller banned from it cannot read the bytes. The cache is `private` so the
/// access decision is never stored in a shared cache and replayed to an
/// unauthorized viewer.
#[tracing::instrument(skip_all)]
pub(super) async fn serve_image_handler(
    axum::extract::State(state): axum::extract::State<AppState>,
    MaybeUser(session_user): MaybeUser,
    Path(image_id_str): Path<String>,
) -> Result<Response, Response> {
    let image_id = super::parse_uuid(&image_id_str)?;

    // Check DB first, return 404 before checking S3 availability
    let image = mt_db::queries::get_image(&state.db, image_id)
        .await
        .map_err(db_error)?
        .ok_or_else(|| StatusCode::NOT_FOUND.into_response())?;

    // Don't serve removed images
    if image.removed_at.is_some() {
        return Err(StatusCode::GONE.into_response());
    }

    // Enforce the owning community's access policy before serving any bytes.
    let community = mt_db::queries::get_community_by_id(&state.db, image.community_id)
        .await
        .map_err(db_error)?
        .ok_or_else(|| StatusCode::NOT_FOUND.into_response())?;
    check_community_access(
        &state.db,
        &community,
        session_user.as_ref().map(|u| u.user_id),
    )
    .await?;

    let s3 = state
        .s3
        .as_ref()
        .ok_or_else(|| StatusCode::SERVICE_UNAVAILABLE.into_response())?;

    // Stream the object straight from S3 to the client instead of buffering the
    // whole image into a Vec<u8> per request (ultra-fuzz S1). The Content-Type
    // comes from the stored row (set from upload-time magic-byte validation), so
    // we don't need S3's metadata round-trip.
    let body = s3.download_stream(&image.s3_key).await.map_err(|e| {
        tracing::error!(error = %e, "S3 stream open failed");
        StatusCode::INTERNAL_SERVER_ERROR.into_response()
    })?;

    // Serve hardening. `nosniff` pins the browser to the stored Content-Type
    // (which upload-time magic-byte validation guarantees is a real image), so
    // a content-spoofed file can never be reinterpreted as HTML/JS. Disposition
    // stays `inline` because these are embedded in post bodies via <img>;
    // `attachment` would force a download and break the feature. The byte-level
    // validation, not a disposition flag, is what makes inline serving safe.
    Ok(Response::builder()
        .status(StatusCode::OK)
        .header(header::CONTENT_TYPE, image.content_type)
        .header(header::CACHE_CONTROL, "private, max-age=86400, immutable")
        .header(header::X_CONTENT_TYPE_OPTIONS, "nosniff")
        .header(header::CONTENT_DISPOSITION, "inline")
        .body(body)
        .unwrap())
}

/// POST /p/{slug}/uploads/{id}/remove, mod removes an uploaded image.
#[tracing::instrument(skip_all)]
pub(super) async fn remove_image_handler(
    axum::extract::State(state): axum::extract::State<AppState>,
    Path((slug, image_id_str)): Path<(String, String)>,
    MaybeUser(session_user): MaybeUser,
) -> Result<impl IntoResponse, Response> {
    let user = session_user.ok_or_else(|| StatusCode::UNAUTHORIZED.into_response())?;

    let community = get_community(&state.db, &slug).await?;
    let role = get_role(&state.db, user.user_id, community.id).await?;

    if !is_mod_or_owner(role) {
        return Err(StatusCode::FORBIDDEN.into_response());
    }

    let image_id = super::parse_uuid(&image_id_str)?;

    // Fetch first so we have the S3 key and can confirm the image belongs to
    // the community whose mod is acting (authorize against the resource, not
    // just the URL slug).
    let image = mt_db::queries::get_image(&state.db, image_id)
        .await
        .map_err(db_error)?
        .ok_or_else(|| StatusCode::NOT_FOUND.into_response())?;
    if image.community_id != community.id {
        return Err(StatusCode::NOT_FOUND.into_response());
    }

    // Mark the image removed and write the audit row on one transaction, so the
    // removal can never land without its log entry.
    let mut tx = super::begin_tx(&state.db).await?;
    mt_db::mutations::remove_image(&mut *tx, image_id, user.user_id)
        .await
        .map_err(db_error)?;
    super::audit(
        &mut tx,
        Some(community.id),
        ModActor::User(user.user_id),
        ModAction::RemoveImage,
        None,
        Some(image_id),
        None,
    )
    .await?;
    super::commit_tx(tx).await?;

    // Delete the backing S3 object so removed images don't accumulate in the
    // bucket forever. Best-effort, after the removal+log have committed: the DB
    // row is already marked removed (serve returns 410). On success, record
    // s3_purged_at so the reconcile sweep skips it; on failure, leave it unmarked
    // so the sweep retries it later.
    if let Some(s3) = state.s3.as_ref() {
        match s3.delete(&image.s3_key).await {
            Ok(()) => {
                if let Err(e) =
                    mt_db::mutations::mark_images_s3_purged(&state.db, &[image_id]).await
                {
                    tracing::warn!(error = ?e, "failed to mark image S3-purged (sweep will retry)");
                }
            }
            Err(e) => {
                tracing::warn!(error = %e, s3_key = %image.s3_key, "failed to delete removed image from S3 (sweep will retry)");
            }
        }
    }

    Ok(StatusCode::OK)
}

/// Query for [`image_proxy_handler`].
#[derive(Deserialize)]
pub(super) struct ImageProxyQuery {
    /// The external image URL to fetch (percent-encoded by the renderer).
    u: String,
}

/// GET /img-proxy?u=<url>, same-origin proxy for external Fan+ images.
///
/// The Fan+ markdown renderer rewrites external `<img src="https://…">` to point
/// here so the page CSP can stay `img-src 'self'` (M-UX1). This handler fetches
/// the URL through the SSRF-safe link-preview client (private addresses refused
/// at connect time, even across redirects), caps the body at 1 MB / 5 s, and
/// re-serves only recognised image content-types.
///
/// Login-gated so it can't be driven as an open image-fetch relay by anonymous
/// clients, and it sits in the per-IP image rate-limit group.
#[tracing::instrument(skip_all)]
pub(super) async fn image_proxy_handler(
    axum::extract::State(state): axum::extract::State<AppState>,
    MaybeUser(session_user): MaybeUser,
    Query(query): Query<ImageProxyQuery>,
) -> Result<Response, Response> {
    // Authenticated members only, an anonymous open proxy is the abuse vector.
    let _user = session_user.ok_or_else(|| StatusCode::UNAUTHORIZED.into_response())?;

    let client = match &state.link_preview {
        crate::link_preview::LinkPreviewFetcher::Http(c) => c,
        crate::link_preview::LinkPreviewFetcher::Noop => {
            return Err(StatusCode::SERVICE_UNAVAILABLE.into_response());
        }
    };

    let (bytes, content_type) = crate::link_preview::fetch_image(client, &query.u)
        .await
        .ok_or_else(|| StatusCode::BAD_GATEWAY.into_response())?;

    // `nosniff` pins the browser to the validated image content-type, so even if
    // an upstream served image bytes under a benign type, it can't be reframed as
    // HTML/JS. 1 MB cap means buffering here is bounded.
    Ok(Response::builder()
        .status(StatusCode::OK)
        .header(header::CONTENT_TYPE, content_type)
        .header(header::CACHE_CONTROL, "private, max-age=86400")
        .header(header::X_CONTENT_TYPE_OPTIONS, "nosniff")
        .header(header::CONTENT_DISPOSITION, "inline")
        .body(Body::from(bytes))
        .unwrap())
}