use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex, OnceLock};

use hayro::hayro_interpret::InterpreterSettings;
use hayro::hayro_syntax::Pdf;
use hayro::{RenderCache, RenderSettings, render};
use serde_json::json;
use tokio::sync::mpsc;

use crate::rpcgit::commit::types::TreeEntryMode;
use crate::rpcgit::error::BabyError;

use crate::lfs;
use crate::service::GitService;
use crate::service::blobs;
use crate::share::errors::{GitError, map_baby};

/// Page rasterisation for the in-repo PDF viewer.
///
/// The browser's own PDF surface cannot be reached from an embedded same-origin
/// frame, so the server rasterises the document and streams it back one page at
/// a time. Rendering is CPU-bound and runs on a blocking task; the channel is
/// shallow, so a reader that stops consuming also stops the renderer instead of
/// rasterising all 200+ pages for nothing.
pub const PAGE_SCALE_MIN: f32 = 0.5;
pub const PAGE_SCALE_MAX: f32 = 3.0;
pub const PAGE_SCALE_DEFAULT: f32 = 2.0;
/// Document ceiling: past this the stream ends early and says so.
pub const PAGE_LIMIT: usize = 400;
/// Frames the renderer may run ahead of the reader.
const CHANNEL_DEPTH: usize = 2;
/// Rasterised pages kept in the process for repeat views, shared by every
/// viewer of the same blob and scale.
const CACHE_BUDGET_BYTES: usize = 96 * 1024 * 1024;

pub fn clamp_scale(scale: Option<f32>) -> f32 {
    match scale {
        Some(v) if v.is_finite() => v.clamp(PAGE_SCALE_MIN, PAGE_SCALE_MAX),
        _ => PAGE_SCALE_DEFAULT,
    }
}

/// Frame the payload as `u32 header_len | header | u32 payload_len | payload`,
/// all big-endian. The header is JSON tagged by `kind`:
/// `doc` (once, upfront), `page` (PNG payload), `page_error`, `done`. Frames are
/// self-describing, so a single failed page can be reported inline instead of
/// killing the stream, and the reader needs no lookahead beyond the two lengths.
fn encode_frame(header: &str, payload: &[u8]) -> Vec<u8> {
    let mut out = Vec::with_capacity(header.len() + payload.len() + 8);
    out.extend_from_slice(&(header.len() as u32).to_be_bytes());
    out.extend_from_slice(header.as_bytes());
    out.extend_from_slice(&(payload.len() as u32).to_be_bytes());
    out.extend_from_slice(payload);
    out
}

fn frame(value: serde_json::Value, payload: &[u8]) -> Vec<u8> {
    encode_frame(&value.to_string(), payload)
}

struct CachedPage {
    png: Arc<Vec<u8>>,
    width: u16,
    height: u16,
}

#[derive(Default)]
struct PageCache {
    pages: HashMap<String, CachedPage>,
    order: VecDeque<String>,
    bytes: usize,
}

fn page_cache() -> &'static Mutex<PageCache> {
    static CACHE: OnceLock<Mutex<PageCache>> = OnceLock::new();
    CACHE.get_or_init(|| Mutex::new(PageCache::default()))
}

fn cache_get(key: &str) -> Option<(Arc<Vec<u8>>, u16, u16)> {
    let cache = page_cache().lock().ok()?;
    let hit = cache.pages.get(key)?;
    Some((hit.png.clone(), hit.width, hit.height))
}

fn cache_put(key: String, png: Arc<Vec<u8>>, width: u16, height: u16) {
    if png.len() > CACHE_BUDGET_BYTES {
        return;
    }
    let Ok(mut cache) = page_cache().lock() else {
        return;
    };
    if let Some(prev) = cache.pages.remove(&key) {
        cache.bytes = cache.bytes.saturating_sub(prev.png.len());
        cache.order.retain(|k| k != &key);
    }
    while cache.bytes + png.len() > CACHE_BUDGET_BYTES {
        let Some(oldest) = cache.order.pop_front() else {
            break;
        };
        if let Some(evicted) = cache.pages.remove(&oldest) {
            cache.bytes = cache.bytes.saturating_sub(evicted.png.len());
        }
    }
    cache.bytes += png.len();
    cache.order.push_back(key.clone());
    cache.pages.insert(key, CachedPage { png, width, height });
}

/// KenShell resolves refs through its git backend, so an unknown revision surfaces as a
/// non-zero exit rather than a typed error. On this read path the only thing
/// that matters is the resulting status: an unresolvable ref is a 404, exactly
/// like a missing path.
fn is_missing_ref(error: &BabyError) -> bool {
    match error {
        BabyError::RevisionNotFound(_)
        | BabyError::RevisionInvalid(_)
        | BabyError::RefNotFound(_)
        | BabyError::PathNotFound { .. } => true,
        BabyError::Command(cmd) => has_missing_ref_text(&cmd.to_string()),
        // KenShell only classifies the git message shapes it knows; anything
        // else (such as `fatal: Not a valid object name`) lands here, so the
        // stderr has to be inspected once more.
        BabyError::Spawn { stderr, .. } => has_missing_ref_text(stderr),
        _ => false,
    }
}

fn has_missing_ref_text(text: &str) -> bool {
    let text = text.to_ascii_lowercase();
    text.contains("not a valid object name")
        || text.contains("unknown revision")
        || text.contains("bad revision")
        || text.contains("does not exist")
        || text.contains("needed a single revision")
}

/// Where the bytes of a previewable file actually live, and under which cache
/// key they belong.
struct Source {
    bytes: Vec<u8>,
    /// Content-addressed identity: the git blob id, or the LFS object id. A
    /// branch that moves or a tag that is rewritten can therefore never replay
    /// an older file's pages.
    key: String,
}

/// Whole bytes for a PDF, resolving through Git LFS when the blob is a pointer.
///
/// A pointer blob is ~130 bytes of text, so an LFS-tracked `paper.pdf` must be
/// followed to object storage or the rasteriser only ever sees the pointer. A
/// partial file cannot be parsed either way, so the size cap rejects the
/// document instead of truncating it.
async fn pdf_bytes(
    service: &GitService,
    space: &str,
    repo: &str,
    revision: &str,
    path: &str,
) -> Result<Source, GitError> {
    let ctx = service.context(space.to_string(), repo.to_string()).await?;
    let baby = service.baby(&ctx);
    // A missing ref, a missing path and a directory all read as not-found here;
    // only a file that really is not a readable PDF is a bad request.
    let entry = match baby
        .tree_entry_at(revision, std::path::Path::new(path))
        .await
    {
        Ok(Some(entry)) => entry,
        Ok(None) => return Err(GitError::NotFound),
        Err(e) if is_missing_ref(&e) => return Err(GitError::NotFound),
        Err(e) => return Err(map_baby(e)),
    };
    if !matches!(
        entry.mode,
        TreeEntryMode::Blob | TreeEntryMode::Exec | TreeEntryMode::Symlink
    ) {
        return Err(GitError::NotFound);
    }

    let max = service.state.config.git.max_blob_size;
    let size = baby
        .get_blob_size(entry.oid.clone())
        .await
        .map_err(map_baby)?;

    // Sniff before reading the whole file: `parse_lfs_pointer` only needs its
    // first lines, and a plain multi-hundred-megabyte PDF should be rejected by
    // the size cap rather than paged into memory first.
    let sniff = baby
        .get_blob(
            entry.oid.clone(),
            crate::rpcgit::blob::types::GetBlobOptions {
                limit: Some(blobs::SNIFF_LIMIT),
            },
        )
        .await
        .map_err(map_baby)?;
    if let Some(pointer) = blobs::parse_lfs_pointer(&sniff.data) {
        return lfs_source(service, &ctx, space, repo, pointer).await;
    }

    if size.max(0) as u64 > max {
        return Err(GitError::InvalidArgument(format!(
            "pdf blob is {size} bytes, above the {max} byte preview limit"
        )));
    }
    let blob = baby
        .get_blob(
            entry.oid.clone(),
            crate::rpcgit::blob::types::GetBlobOptions { limit: None },
        )
        .await
        .map_err(map_baby)?;
    Ok(Source {
        bytes: blob.data,
        key: entry.oid.to_string(),
    })
}

/// Follow one LFS pointer to object storage.
async fn lfs_source(
    service: &GitService,
    ctx: &crate::context::facade::GitContext,
    space: &str,
    repo: &str,
    pointer: blobs::LfsPointerDto,
) -> Result<Source, GitError> {
    if !ctx.model.enable_lfs {
        return Err(GitError::InvalidArgument(
            "file is an lfs pointer but lfs is disabled on this repo".to_string(),
        ));
    }
    let raw = pointer
        .oid
        .ok_or_else(|| GitError::InvalidArgument("lfs pointer has no object id".to_string()))?;
    let oid = lfs::canonical_oid(&raw)?;

    let max = service.state.config.git.max_blob_size;
    if let Some(size) = pointer.size
        && size.max(0) as u64 > max
    {
        return Err(GitError::InvalidArgument(format!(
            "lfs object is {size} bytes, above the {max} byte preview limit"
        )));
    }

    let mut bytes = Vec::new();
    let written = service
        .lfs_download_to(space.to_string(), repo.to_string(), oid.clone(), &mut bytes)
        .await?;
    if written > max {
        return Err(GitError::InvalidArgument(format!(
            "lfs object is {written} bytes, above the {max} byte preview limit"
        )));
    }
    Ok(Source { bytes, key: oid })
}

/// Stream `path` as framed pages. `Err` covers everything that has to fail
/// before the response starts (missing blob, unreadable document).
pub async fn render_stream(
    service: &GitService,
    space: &str,
    repo: &str,
    revision: &str,
    path: &str,
    scale: f32,
) -> Result<mpsc::Receiver<Vec<u8>>, GitError> {
    let source = pdf_bytes(service, space, repo, revision, path).await?;
    let pdf = tokio::task::spawn_blocking(move || Pdf::new(source.bytes))
        .await
        .map_err(|e| GitError::Internal(format!("pdf task failed: {e}")))?
        .map_err(|e| GitError::InvalidArgument(format!("unreadable pdf: {e:?}")))?;

    let (tx, rx) = mpsc::channel::<Vec<u8>>(CHANNEL_DEPTH);
    // Content-addressed cache key (git oid, or LFS object id): two paths that
    // hold the same bytes share one render, and no revision name can replay an
    // older file's pages.
    let prefix = format!("pdf.{space}/{repo}/{}@{scale}", source.key);
    tokio::task::spawn_blocking(move || {
        let pages = pdf.pages();
        let available = pages.len();
        let total = available.min(PAGE_LIMIT);
        let doc = json!({
            "kind": "doc",
            "pages": total,
            "available": available,
            "truncated": available > total,
        });
        if tx.blocking_send(frame(doc, &[])).is_err() {
            return;
        }

        let settings = RenderSettings {
            x_scale: scale,
            y_scale: scale,
            bg_color: hayro::vello_cpu::color::palette::css::WHITE,
            ..Default::default()
        };
        let cache = RenderCache::new();
        let interpreter = InterpreterSettings::default();
        for index in 0..total {
            let key = format!("{prefix}:{index}");
            let (png, width, height) = match cache_get(&key) {
                Some(hit) => hit,
                None => {
                    let pixmap = render(&pages[index], &cache, &interpreter, &settings);
                    let (width, height) = (pixmap.width(), pixmap.height());
                    match pixmap.into_png() {
                        Ok(data) => {
                            let png = Arc::new(data);
                            cache_put(key, png.clone(), width, height);
                            (png, width, height)
                        }
                        Err(e) => {
                            let err = json!({
                                "kind": "page_error",
                                "page": index + 1,
                                "total": total,
                                "error": e.to_string(),
                            });
                            if tx.blocking_send(frame(err, &[])).is_err() {
                                return;
                            }
                            continue;
                        }
                    }
                }
            };
            let header = json!({
                "kind": "page",
                "page": index + 1,
                "total": total,
                "width": width,
                "height": height,
            });
            if tx.blocking_send(frame(header, &png)).is_err() {
                return;
            }
        }
        let done = json!({ "kind": "done", "pages": total });
        let _ = tx.blocking_send(frame(done, &[]));
    });
    Ok(rx)
}