use std::collections::{HashMap, HashSet, VecDeque};

use anyhow::Result;
use axum::body::{Body, to_bytes};
use axum::extract::{Path as AxumPath, State};
use axum::http::{HeaderValue, Request, StatusCode, header};
use axum::response::{IntoResponse, Response};

use crate::history;

use super::{AppState, WebResult, internal};

const MAX_OBJECT_QUERY_OBJECTS: usize = 128;
const MAX_OBJECT_QUERY_BYTES: usize = 16 << 20;
const MAX_COMMIT_PREFETCH: usize = 25;
const MAX_HISTORY_FRONTIER: usize = 4096;
/// Commits examined per path-history request; the client loops on the frontier.
const HISTORY_SCAN_LIMIT: usize = 5000;

#[derive(Clone, Copy, serde::Deserialize)]
#[serde(rename_all = "kebab-case")]
enum SmartObjectQuery {
    Tree,
    CommitDiff,
    CommitPagination,
    PathHistory,
}

#[derive(serde::Deserialize)]
struct ObjectQuery {
    oids: Vec<String>,
    #[serde(default)]
    depth: usize,
    smart: Option<SmartObjectQuery>,
    limit: Option<usize>,
    path: Option<Vec<String>>,
}

enum ObjectQueryMode {
    Generic(usize),
    Tree,
    CommitDiff,
    CommitPagination(usize),
    PathHistory { limit: usize, path: Vec<String> },
}

struct ObjectBundle {
    data: Vec<u8>,
    seen: HashSet<gix::ObjectId>,
    full: bool,
}

impl ObjectBundle {
    fn new() -> Self {
        Self {
            // SOBJ, version, flags (bit 0 = truncated), then
            // oid-length/oid/kind/u32be-size/data frames. Kinds 1-4 are git
            // objects; kind 5 is a payload-less "resume from here" marker.
            data: b"SOBJ\x01\0".to_vec(),
            seen: HashSet::new(),
            full: false,
        }
    }

    fn object<'repo>(
        &mut self,
        repo: &'repo gix::Repository,
        oid: gix::ObjectId,
    ) -> Option<gix::Object<'repo>> {
        if self.full {
            return None;
        }
        let object = repo.find_object(oid).ok()?;
        if self.seen.contains(&oid) {
            return Some(object);
        }
        let oid_bytes = oid.as_bytes();
        let frame_bytes = 1 + oid_bytes.len() + 1 + 4 + object.data.len();
        if self.seen.len() == MAX_OBJECT_QUERY_OBJECTS
            || self
                .data
                .len()
                .checked_add(frame_bytes)
                .is_none_or(|size| size > MAX_OBJECT_QUERY_BYTES)
        {
            self.data[5] = 1;
            self.full = true;
            return None;
        }
        let kind = match object.kind {
            gix::object::Kind::Commit => 1,
            gix::object::Kind::Tree => 2,
            gix::object::Kind::Blob => 3,
            gix::object::Kind::Tag => 4,
        };
        self.seen.insert(oid);
        self.data.push(oid_bytes.len() as u8);
        self.data.extend_from_slice(oid_bytes);
        self.data.push(kind);
        self.data
            .extend_from_slice(&(object.data.len() as u32).to_be_bytes());
        self.data.extend_from_slice(&object.data);
        Some(object)
    }

    fn frontier(&mut self, oid: gix::ObjectId) {
        let oid_bytes = oid.as_bytes();
        self.data.push(oid_bytes.len() as u8);
        self.data.extend_from_slice(oid_bytes);
        self.data.push(5);
        self.data.extend_from_slice(&0u32.to_be_bytes());
    }
}

fn generic_object_bundle(
    repo: &gix::Repository,
    roots: Vec<gix::ObjectId>,
    depth: usize,
) -> Result<Vec<u8>> {
    let mut bundle = ObjectBundle::new();
    let mut queue = roots
        .into_iter()
        .map(|oid| (oid, 0usize))
        .collect::<VecDeque<_>>();
    while let Some((oid, object_depth)) = queue.pop_front() {
        if bundle.full {
            break;
        }
        if bundle.seen.contains(&oid) {
            continue;
        }
        let Some(object) = bundle.object(repo, oid) else {
            continue;
        };
        if object_depth == depth {
            continue;
        }
        let mut remaining =
            MAX_OBJECT_QUERY_OBJECTS.saturating_sub(bundle.seen.len() + queue.len());
        let mut enqueue = |oid| {
            if remaining == 0 {
                bundle.data[5] = 1;
                return false;
            }
            remaining -= 1;
            queue.push_back((oid, object_depth + 1));
            true
        };
        match object.kind {
            gix::object::Kind::Commit => {
                let commit = object.into_commit();
                if enqueue(commit.tree_id()?.detach()) {
                    for parent in commit.parent_ids() {
                        if !enqueue(parent.detach()) {
                            break;
                        }
                    }
                }
            }
            gix::object::Kind::Tree => {
                for entry in object.into_tree().iter() {
                    if !enqueue(entry?.oid().to_owned()) {
                        break;
                    }
                }
            }
            gix::object::Kind::Tag => {
                enqueue(object.into_tag().target_id()?.detach());
            }
            gix::object::Kind::Blob => {}
        }
    }
    Ok(bundle.data)
}

fn commit_pagination_object_bundle(
    repo: &gix::Repository,
    roots: Vec<gix::ObjectId>,
    limit: usize,
) -> Result<Vec<u8>> {
    let mut bundle = ObjectBundle::new();
    let mut queue = VecDeque::from(roots);
    let mut commits = 0;

    while commits < limit {
        let Some(oid) = queue.pop_front() else {
            break;
        };
        if bundle.seen.contains(&oid) {
            continue;
        }
        let Some(object) = bundle.object(repo, oid) else {
            continue;
        };
        match object.kind {
            gix::object::Kind::Commit => {
                commits += 1;
                queue.extend(object.into_commit().parent_ids().map(|id| id.detach()));
            }
            gix::object::Kind::Tag => {
                queue.push_front(object.into_tag().target_id()?.detach());
            }
            _ => {}
        }
    }
    Ok(bundle.data)
}

/// The commits of `git log -- path` from `frontier`, in order, then the
/// frontier to resume from, then (best effort) each commit's trees along the
/// path so the client can locate the object without further requests.
fn path_history_object_bundle(
    repo: &gix::Repository,
    frontier: &[gix::ObjectId],
    path: &[String],
    limit: usize,
) -> Result<Vec<u8>> {
    let page = history::page(repo, frontier, path, limit, HISTORY_SCAN_LIMIT)?;
    let mut bundle = ObjectBundle::new();
    for &oid in &page.changes {
        bundle.object(repo, oid);
    }
    for oid in page.frontier {
        bundle.frontier(oid);
    }
    'commits: for &oid in &page.changes {
        let Some(commit) = bundle.object(repo, oid) else {
            break;
        };
        let mut tree_oid = commit.into_commit().tree_id()?.detach();
        for component in path {
            let Some(tree) = bundle.object(repo, tree_oid) else {
                break 'commits;
            };
            let next = tree.into_tree().iter().find_map(|entry| {
                let entry = entry.ok()?;
                (entry.filename() == component.as_bytes() && entry.mode().is_tree())
                    .then(|| entry.oid().to_owned())
            });
            let Some(next) = next else {
                break;
            };
            tree_oid = next;
        }
    }
    Ok(bundle.data)
}

fn tree_object_bundle(repo: &gix::Repository, roots: Vec<gix::ObjectId>) -> Result<Vec<u8>> {
    let mut bundle = ObjectBundle::new();
    let mut queue = VecDeque::from(roots);

    while let Some(oid) = queue.pop_front() {
        if bundle.full {
            break;
        }
        if bundle.seen.contains(&oid) {
            continue;
        }
        let Some(object) = bundle.object(repo, oid) else {
            continue;
        };
        match object.kind {
            gix::object::Kind::Commit => queue.push_back(object.into_commit().tree_id()?.detach()),
            gix::object::Kind::Tree => {
                for entry in object.into_tree().iter() {
                    let entry = entry?;
                    if entry.mode().is_tree() {
                        if bundle.seen.len() + queue.len() == MAX_OBJECT_QUERY_OBJECTS {
                            bundle.data[5] = 1;
                            break;
                        }
                        queue.push_back(entry.oid().to_owned());
                    }
                }
            }
            gix::object::Kind::Tag => queue.push_back(object.into_tag().target_id()?.detach()),
            gix::object::Kind::Blob => {}
        }
    }
    Ok(bundle.data)
}

#[derive(Clone, Copy)]
struct DiffEntry {
    oid: gix::ObjectId,
    mode: u16,
    tree: bool,
    gitlink: bool,
}

fn bundle_tree_entries(
    repo: &gix::Repository,
    bundle: &mut ObjectBundle,
    oid: Option<gix::ObjectId>,
) -> Result<HashMap<Vec<u8>, DiffEntry>> {
    let Some(oid) = oid else {
        return Ok(HashMap::new());
    };
    let Some(object) = bundle.object(repo, oid) else {
        return Ok(HashMap::new());
    };
    let mut entries = HashMap::new();
    for entry in object.into_tree().iter() {
        let entry = entry?;
        let mode = entry.mode();
        entries.insert(
            entry.filename().to_vec(),
            DiffEntry {
                oid: entry.oid().to_owned(),
                mode: mode.value(),
                tree: mode.is_tree(),
                gitlink: mode.is_commit(),
            },
        );
    }
    Ok(entries)
}

fn commit_diff_object_bundle(repo: &gix::Repository, roots: Vec<gix::ObjectId>) -> Result<Vec<u8>> {
    let mut bundle = ObjectBundle::new();
    let mut trees = VecDeque::new();

    for mut oid in roots {
        if bundle.full {
            break;
        }
        loop {
            let Some(object) = bundle.object(repo, oid) else {
                break;
            };
            match object.kind {
                gix::object::Kind::Tag => oid = object.into_tag().target_id()?.detach(),
                gix::object::Kind::Commit => {
                    let commit = object.into_commit();
                    let new_tree = commit.tree_id()?.detach();
                    let old_tree = if let Some(parent) = commit.parent_ids().next() {
                        let parent = parent.detach();
                        bundle
                            .object(repo, parent)
                            .map(|object| object.into_commit().tree_id().map(|id| id.detach()))
                            .transpose()?
                    } else {
                        None
                    };
                    trees.push_back((old_tree, Some(new_tree)));
                    break;
                }
                _ => break,
            }
        }
    }

    let mut visited = HashSet::new();
    while let Some((old_tree, new_tree)) = trees.pop_front() {
        if bundle.full {
            break;
        }
        if old_tree == new_tree || !visited.insert((old_tree, new_tree)) {
            continue;
        }
        let mut old_entries = bundle_tree_entries(repo, &mut bundle, old_tree)?;
        let new_entries = bundle_tree_entries(repo, &mut bundle, new_tree)?;

        let mut changed = |old: Option<DiffEntry>, new: Option<DiffEntry>| -> Result<()> {
            if old
                .zip(new)
                .is_some_and(|(old, new)| old.oid == new.oid && old.mode == new.mode)
            {
                return Ok(());
            }
            if old.is_some_and(|entry| entry.tree) || new.is_some_and(|entry| entry.tree) {
                trees.push_back((
                    old.filter(|entry| entry.tree).map(|entry| entry.oid),
                    new.filter(|entry| entry.tree).map(|entry| entry.oid),
                ));
            }
            if old.is_some_and(|entry| entry.gitlink) || new.is_some_and(|entry| entry.gitlink) {
                return Ok(());
            }
            for entry in [old, new].into_iter().flatten() {
                if !entry.tree {
                    let _ = bundle.object(repo, entry.oid);
                }
            }
            Ok(())
        };

        for (name, new) in new_entries {
            changed(old_entries.remove(&name), Some(new))?;
        }
        for old in old_entries.into_values() {
            changed(Some(old), None)?;
        }
    }
    Ok(bundle.data)
}

pub(super) async fn query(
    State(state): State<AppState>,
    AxumPath((user, repo)): AxumPath<(String, String)>,
    request: Request<Body>,
) -> WebResult<Response> {
    if request.method().as_str() != "QUERY" {
        return Err((StatusCode::METHOD_NOT_ALLOWED, "QUERY required".into()));
    }
    if request
        .headers()
        .get(header::CONTENT_TYPE)
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.split(';').next())
        != Some("application/json")
    {
        return Err((
            StatusCode::UNSUPPORTED_MEDIA_TYPE,
            "expected application/json".into(),
        ));
    }
    let body = to_bytes(request.into_body(), 256 << 10)
        .await
        .map_err(|_| {
            (
                StatusCode::PAYLOAD_TOO_LARGE,
                "object query is too large".into(),
            )
        })?;
    let ObjectQuery {
        oids,
        depth,
        smart,
        limit,
        path,
    } = serde_json::from_slice(&body)
        .map_err(|_| (StatusCode::BAD_REQUEST, "invalid object query".into()))?;
    let mode = match (smart, limit, path) {
        (None, None, None) if depth <= 2 => ObjectQueryMode::Generic(depth),
        (Some(SmartObjectQuery::Tree), None, None) if depth == 0 => ObjectQueryMode::Tree,
        (Some(SmartObjectQuery::CommitDiff), None, None) if depth == 0 => {
            ObjectQueryMode::CommitDiff
        }
        (Some(SmartObjectQuery::CommitPagination), Some(limit), None)
            if depth == 0 && (1..=MAX_COMMIT_PREFETCH).contains(&limit) =>
        {
            ObjectQueryMode::CommitPagination(limit)
        }
        (Some(SmartObjectQuery::PathHistory), Some(limit), Some(path))
            if depth == 0
                && (1..=MAX_COMMIT_PREFETCH).contains(&limit)
                && path.iter().all(|component| {
                    !component.is_empty()
                        && component != "."
                        && component != ".."
                        && !component.contains('/')
                        && !component.contains('\0')
                }) =>
        {
            ObjectQueryMode::PathHistory { limit, path }
        }
        _ => {
            return Err((
                StatusCode::UNPROCESSABLE_ENTITY,
                "invalid object query bounds".into(),
            ));
        }
    };
    // a path-history frontier can fan out past the object cap; it's oids only
    let max_oids = match mode {
        ObjectQueryMode::PathHistory { .. } => MAX_HISTORY_FRONTIER,
        _ => MAX_OBJECT_QUERY_OBJECTS,
    };
    if oids.is_empty() || oids.len() > max_oids {
        return Err((
            StatusCode::UNPROCESSABLE_ENTITY,
            "invalid object query bounds".into(),
        ));
    }
    let roots = oids
        .iter()
        .map(|oid| {
            gix::ObjectId::from_hex(oid.as_bytes())
                .map_err(|_| (StatusCode::UNPROCESSABLE_ENTITY, "invalid object id".into()))
        })
        .collect::<WebResult<Vec<_>>>()?;

    let path = state.resolve(&user, &repo)?.repository.path;
    let permit = state.git_permit()?;
    let data = tokio::task::spawn_blocking(move || -> Result<Vec<u8>> {
        let _permit = permit;
        let mut repo = gix::open(path)?;
        // deep tree delta chains in large repos make repeated lookups expensive
        repo.object_cache_size_if_unset(64 << 20);
        match mode {
            ObjectQueryMode::Generic(depth) => generic_object_bundle(&repo, roots, depth),
            ObjectQueryMode::Tree => tree_object_bundle(&repo, roots),
            ObjectQueryMode::CommitDiff => commit_diff_object_bundle(&repo, roots),
            ObjectQueryMode::CommitPagination(limit) => {
                commit_pagination_object_bundle(&repo, roots, limit)
            }
            ObjectQueryMode::PathHistory { limit, path } => {
                path_history_object_bundle(&repo, &roots, &path, limit)
            }
        }
    })
    .await
    .map_err(internal)?
    .map_err(internal)?;

    let mut response = Body::from(data).into_response();
    response.headers_mut().insert(
        header::CONTENT_TYPE,
        HeaderValue::from_static("application/x-git-object-bundle"),
    );
    response
        .headers_mut()
        .insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store"));
    response
        .headers_mut()
        .insert("accept-query", HeaderValue::from_static("application/json"));
    Ok(response)
}

#[cfg(test)]
mod tests {
    use std::fs;

    use anyhow::Result;

    use crate::testutil::{TempDir, commit, init_sha256_repo};

    use super::generic_object_bundle;

    #[test]
    fn bundles_sha256_object_ids() -> Result<()> {
        let root = TempDir::new("sha256-bundle");
        let repo_path = root.join("repo");
        init_sha256_repo(&repo_path)?;
        fs::write(repo_path.join("file"), "contents")?;
        let head = commit(&repo_path, "initial")?;

        let repo = gix::open(repo_path)?;
        let bundle =
            generic_object_bundle(&repo, vec![gix::ObjectId::from_hex(head.as_bytes())?], 2)?;
        let mut position = 6;
        let mut frames = 0;
        while position < bundle.len() {
            let oid_bytes = bundle[position] as usize;
            assert_eq!(oid_bytes, 32);
            let size_at = position + 1 + oid_bytes + 1;
            let size = u32::from_be_bytes(bundle[size_at..size_at + 4].try_into()?) as usize;
            position = size_at + 4 + size;
            frames += 1;
        }
        assert_eq!(position, bundle.len());
        assert_eq!(frames, 3);
        Ok(())
    }
}