//! Streams blobs from the `Storage` trait. Same endpoint serves manga
//! covers and chapter pages; the key embedded in the URL is whatever the
//! writer stored.
//!
//! The handler uses `Storage::get_stream` so a multi-MB page is piped to
//! the client a chunk at a time instead of buffered server-side.
//!
//! **Thumbnails.** `?w=<px>` serves a width-bounded variant (grids ask for a
//! small width so they download ~KB instead of the 1–5 MB original). The width
//! snaps to a small allow-list so the number of cached derivatives stays
//! bounded; the resized image is cached in storage under a `thumbs/w{W}/` prefix
//! and regenerated on demand. Only JPEG/PNG sources are thumbnailed (encoders we
//! ship); other formats fall back to the original.
//!
//! **Auth model — capability URLs by design.** This endpoint is
//! deliberately unauthenticated: reads stay public per the project
//! brief, and per-page authorisation would require either a per-request
//! ownership lookup (covers + pages are scoped by manga, not user) or a
//! signed-URL scheme. Mangalord instead relies on the keys being
//! unguessable — `mangas/{uuid}/...` and
//! `mangas/{uuid}/chapters/{uuid}/...` — so a leaked URL leaks at most
//! the one referenced file. A future feat/private-libraries branch
//! would gate this endpoint behind a `Storage::owner_of(key)` check;
//! the seam is intentional.

use std::io::Cursor;

use axum::body::Body;
use axum::extract::{Path, Query, State};
use axum::http::{header, HeaderName};
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use axum::Router;
use image::imageops::FilterType;
use image::{ImageFormat, ImageReader};
use serde::Deserialize;

use crate::app::AppState;
use crate::error::{AppError, AppResult};
use crate::storage::{Storage, StorageError};

/// Widths a thumbnail may be rendered at. A requested width snaps up to the
/// smallest of these so the set of cached derivatives stays tiny.
const ALLOWED_THUMB_WIDTHS: &[u32] = &[160, 320, 480, 640, 960];

/// Storage key prefix for cached thumbnails.
const THUMB_PREFIX: &str = "thumbs";

/// Decode allocation cap (mirrors `analysis::ocr`): a tiny file declaring huge
/// dimensions is rejected before the decoder allocates, not after (OOM guard).
const MAX_THUMB_DECODE_PIXELS: u64 = 40_000_000;

pub fn routes() -> Router<AppState> {
    Router::new().route("/files/*key", get(serve))
}

#[derive(Debug, Deserialize)]
struct ServeQuery {
    /// Requested thumbnail width in pixels; absent = serve the original.
    w: Option<String>,
}

async fn serve(
    State(state): State<AppState>,
    Path(key): Path<String>,
    Query(q): Query<ServeQuery>,
) -> AppResult<Response> {
    // Thumbnail request: only for source formats we can re-encode; anything
    // else falls through to serving the original.
    if let Some(width) = resolve_thumb_width(q.w.as_deref()) {
        if let Some(fmt) = thumb_format_for(&key) {
            return serve_thumbnail(&state, &key, width, fmt).await;
        }
    }
    serve_original(&state, &key).await
}

async fn serve_original(state: &AppState, key: &str) -> AppResult<Response> {
    let file = match state.storage.get_stream(key).await {
        Ok(f) => f,
        Err(StorageError::NotFound) => return Err(AppError::NotFound),
        Err(e) => return Err(e.into()),
    };
    Ok(image_response(
        state,
        content_type_for(key),
        file.size_bytes.to_string(),
        Body::from_stream(file.stream),
    ))
}

async fn serve_thumbnail(
    state: &AppState,
    key: &str,
    width: u32,
    fmt: ImageFormat,
) -> AppResult<Response> {
    let derived = thumb_key(key, width);

    // Serve the cached variant if it exists.
    match state.storage.get_stream(&derived).await {
        Ok(f) => {
            return Ok(image_response(
                state,
                content_type_for(key),
                f.size_bytes.to_string(),
                Body::from_stream(f.stream),
            ));
        }
        Err(StorageError::NotFound) => {}
        Err(e) => return Err(e.into()),
    }

    // Generate from the original.
    let original = match state.storage.get(key).await {
        Ok(b) => b,
        Err(StorageError::NotFound) => return Err(AppError::NotFound),
        Err(e) => return Err(e.into()),
    };
    // Resizing is CPU-bound; keep it off the async worker threads.
    let thumb = tokio::task::spawn_blocking(move || make_thumbnail(&original, width, fmt))
        .await
        .map_err(|e| AppError::Other(anyhow::anyhow!("thumbnail task join: {e}")))??;

    // Best-effort cache; a write failure just means we regenerate next time.
    let _ = state.storage.put(&derived, &thumb).await;

    let len = thumb.len().to_string();
    Ok(image_response(
        state,
        content_type_for(key),
        len,
        Body::from(thumb),
    ))
}

/// Shared response builder for both the original and thumbnail paths.
fn image_response(
    state: &AppState,
    content_type: &str,
    content_length: String,
    body: Body,
) -> Response {
    let headers = [
        (header::CONTENT_TYPE, content_type.to_string()),
        (header::CONTENT_LENGTH, content_length),
        // `nosniff` makes the contract explicit: the browser must trust the
        // Content-Type we declared (and that the magic-byte sniff at upload
        // time produced) instead of trying to detect HTML/JS in the body.
        (
            HeaderName::from_static("x-content-type-options"),
            "nosniff".to_string(),
        ),
        // Blobs are content-addressed by unguessable, immutable keys (a
        // re-upload mints new UUIDs), so a fetched page/cover never changes.
        // Cache it for a year and mark it `immutable` so browsers skip
        // revalidation entirely.
        //
        // BUT under PRIVATE_MODE these blobs are auth-gated, so they must NOT be
        // marked `public`: a shared cache / CDN would store the object and serve
        // it to unauthenticated clients. Use `private` so only the requesting
        // user's browser caches it.
        (
            header::CACHE_CONTROL,
            if state.auth.private_mode {
                "private, max-age=31536000, immutable".to_string()
            } else {
                "public, max-age=31536000, immutable".to_string()
            },
        ),
    ];
    (headers, body).into_response()
}

/// Parse and clamp a requested thumbnail width. Returns `None` for absent /
/// unparseable / zero widths (serve the original); otherwise snaps the request
/// up to the smallest allowed width (capped at the largest) so cached variants
/// stay bounded.
fn resolve_thumb_width(raw: Option<&str>) -> Option<u32> {
    let requested: u32 = raw?.trim().parse().ok()?;
    if requested == 0 {
        return None;
    }
    Some(
        ALLOWED_THUMB_WIDTHS
            .iter()
            .copied()
            .find(|&w| w >= requested)
            .unwrap_or_else(|| *ALLOWED_THUMB_WIDTHS.last().expect("non-empty")),
    )
}

/// The re-encode format for a source key, or `None` when it isn't one we ship an
/// encoder for (gif/avif → serve the original instead of a broken thumbnail).
fn thumb_format_for(key: &str) -> Option<ImageFormat> {
    match content_type_for(key) {
        "image/jpeg" => Some(ImageFormat::Jpeg),
        "image/png" => Some(ImageFormat::Png),
        _ => None,
    }
}

/// The storage key a width-`w` thumbnail of `key` is cached under.
fn thumb_key(key: &str, width: u32) -> String {
    format!("{THUMB_PREFIX}/w{width}/{key}")
}

/// Every cached thumbnail key for an original `key`, across all allowed widths.
/// Used by the cover handlers to purge stale variants when a cover (whose key is
/// reused, unlike content-addressed pages) is replaced or deleted.
pub(crate) fn thumbnail_keys(key: &str) -> Vec<String> {
    ALLOWED_THUMB_WIDTHS
        .iter()
        .map(|&w| thumb_key(key, w))
        .collect()
}

/// Best-effort deletion of every cached thumbnail for `key`. Call after the
/// underlying blob at `key` changes or is removed.
pub(crate) async fn purge_thumbnails(storage: &dyn Storage, key: &str) {
    for derived in thumbnail_keys(key) {
        let _ = storage.delete(&derived).await;
    }
}

/// Decode, downscale to `width` (aspect-preserving, never upscaling), and
/// re-encode in `fmt`. Pure + synchronous so it runs under `spawn_blocking` and
/// is unit-testable without a server.
fn make_thumbnail(bytes: &[u8], width: u32, fmt: ImageFormat) -> anyhow::Result<Vec<u8>> {
    use anyhow::Context;
    let mut reader = ImageReader::new(Cursor::new(bytes))
        .with_guessed_format()
        .context("guess image format for thumbnail")?;
    let mut limits = image::Limits::default();
    limits.max_alloc = Some(MAX_THUMB_DECODE_PIXELS.saturating_mul(4));
    reader.limits(limits);
    let img = reader.decode().context("decode image for thumbnail")?;

    // Only downscale; a source narrower than the target is served as-is.
    let out = if img.width() > width {
        img.resize(width, u32::MAX, FilterType::Lanczos3)
    } else {
        img
    };

    let mut buf = Cursor::new(Vec::new());
    out.write_to(&mut buf, fmt).context("encode thumbnail")?;
    Ok(buf.into_inner())
}

fn content_type_for(key: &str) -> &'static str {
    let ext = key.rsplit('.').next().unwrap_or("").to_ascii_lowercase();
    match ext.as_str() {
        "jpg" | "jpeg" => "image/jpeg",
        "png" => "image/png",
        "webp" => "image/webp",
        "gif" => "image/gif",
        "avif" => "image/avif",
        _ => "application/octet-stream",
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn resolve_thumb_width_snaps_and_validates() {
        // Absent / unparseable / zero → serve the original.
        assert_eq!(resolve_thumb_width(None), None);
        assert_eq!(resolve_thumb_width(Some("")), None);
        assert_eq!(resolve_thumb_width(Some("abc")), None);
        assert_eq!(resolve_thumb_width(Some("0")), None);
        // Snap up to the smallest allowed width.
        assert_eq!(resolve_thumb_width(Some("1")), Some(160));
        assert_eq!(resolve_thumb_width(Some("160")), Some(160));
        assert_eq!(resolve_thumb_width(Some("161")), Some(320));
        assert_eq!(resolve_thumb_width(Some("640")), Some(640));
        // Above the max → capped at the largest allowed width.
        assert_eq!(resolve_thumb_width(Some("5000")), Some(960));
    }

    #[test]
    fn thumb_format_only_for_encodable_sources() {
        assert_eq!(thumb_format_for("a/b/cover.jpg"), Some(ImageFormat::Jpeg));
        assert_eq!(thumb_format_for("a/b/cover.png"), Some(ImageFormat::Png));
        // Formats we can't re-encode fall back to the original.
        assert_eq!(thumb_format_for("a/b/cover.webp"), None);
        assert_eq!(thumb_format_for("a/b/cover.gif"), None);
        assert_eq!(thumb_format_for("a/b/cover.avif"), None);
    }

    #[test]
    fn thumb_key_is_prefixed_by_width() {
        assert_eq!(
            thumb_key("mangas/x/cover.png", 320),
            "thumbs/w320/mangas/x/cover.png"
        );
        assert_eq!(thumbnail_keys("k.png").len(), ALLOWED_THUMB_WIDTHS.len());
    }

    #[test]
    fn make_thumbnail_downscales_and_preserves_aspect() {
        // 100x50 red PNG → thumbnail width 40 → 40x20, still decodable PNG.
        let mut src = Cursor::new(Vec::new());
        image::RgbImage::from_pixel(100, 50, image::Rgb([255, 0, 0]))
            .write_to(&mut src, ImageFormat::Png)
            .unwrap();
        let out = make_thumbnail(src.get_ref(), 40, ImageFormat::Png).unwrap();
        let decoded = image::load_from_memory(&out).unwrap();
        assert_eq!(decoded.width(), 40);
        assert_eq!(decoded.height(), 20, "aspect ratio preserved");
    }

    #[test]
    fn make_thumbnail_does_not_upscale() {
        // A 30px-wide source requested at 40 stays 30 wide (no upscaling).
        let mut src = Cursor::new(Vec::new());
        image::RgbImage::from_pixel(30, 30, image::Rgb([0, 255, 0]))
            .write_to(&mut src, ImageFormat::Png)
            .unwrap();
        let out = make_thumbnail(src.get_ref(), 40, ImageFormat::Png).unwrap();
        let decoded = image::load_from_memory(&out).unwrap();
        assert_eq!(decoded.width(), 30);
    }
}