use std::net::SocketAddr;
use std::path::PathBuf;
use std::pin::Pin;

use gix::bstr::ByteSlice as _;
use gix::revision::plumbing::Spec as RevSpec;
use gix::revision::walk::Sorting;
use tempfile::TempDir;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tokio_stream::Stream;
use tonic::transport::Server;
use tonic::{Request, Response, Status};
use tracing::{debug, info};

use crate::diff::{self, prepare_interdiff};
use crate::protocol::v1::diff_request::ComparisonType;
use crate::protocol::v1::git_mirror_service_server::{GitMirrorService, GitMirrorServiceServer};
use crate::protocol::v1::{
    BlobChunk, CommitLogRequest, CommitLogResponse, DiffRequest, FileDiff, GetBlobRequest,
    GitCommit, GitSignature, InterdiffRequest, MergeCheckRequest, MergeCheckResponse,
    ResolveRevRequest, ResolveRevResponse,
};

type FileDiffStream = Pin<Box<dyn Stream<Item = Result<FileDiff, Status>> + Send>>;
type BlobChunkStream = Pin<Box<dyn Stream<Item = Result<BlobChunk, Status>> + Send>>;
type CommitLogResponseStream =
    Pin<Box<dyn Stream<Item = Result<CommitLogResponse, Status>> + Send>>;

const BLOB_CHUNK_SIZE: usize = 64 * 1024;
const COMMIT_LOG_BATCH_SIZE: usize = 128;

struct GitMirror {
    repo_base: PathBuf,
}

/// A [`gix::Repository`] backed by a temporary directory that is removed on drop. Derefs to
/// `gix::Repository`, so it is used just like one; keeping it alive keeps the scratch dir alive.
struct TempRepository {
    repo: gix::Repository,
    // Declared AFTER `repo` so `repo` drops first: any file handles into the scratch dir close
    // before the dir itself is removed (Rust drops struct fields in declaration order).
    _dir: TempDir,
}

impl TempRepository {
    fn new(repo: gix::Repository, dir: TempDir) -> Self {
        Self { repo, _dir: dir }
    }
}

impl std::ops::Deref for TempRepository {
    type Target = gix::Repository;
    fn deref(&self) -> &gix::Repository {
        &self.repo
    }
}

impl GitMirror {
    fn open_repo(&self, did: &str) -> Result<gix::Repository, Status> {
        if !did.starts_with("did:") {
            return Err(Status::invalid_argument("repo must be a DID"));
        }
        gix::open(self.repo_base.join(did))
            .map_err(|e| Status::not_found(format!("repo not found: {e}")))
    }

    /// Build a throwaway bare repo whose `objects/info/alternates` points read-only at each of the
    /// given git repositories, so a single `gix::Repository` can see objects from all of them
    /// without ever mutating them.
    fn open_scratch(&self, dids: &[&str]) -> Result<TempRepository, Status> {
        let mut seen: Vec<&str> = Vec::new();
        let mut object_dirs = Vec::new();
        for &did in dids {
            if seen.contains(&did) {
                continue
            }
            if !did.starts_with("did:") {
                return Err(Status::invalid_argument("repo must be a DID"));
            }
            seen.push(did);
            let object_dir = std::fs::canonicalize(self.repo_base.join(did).join("objects"))
                .map_err(|_| Status::not_found(format!("repo not found: {did}")))?;
            object_dirs.push(object_dir);
        }
        if object_dirs.is_empty() {
            return Err(Status::internal("open_scratch requires at least one repo"));
        }

        let scratch = tempfile::tempdir().map_err(|e| Status::internal(e.to_string()))?;
        gix::init_bare(scratch.path()).map_err(|e| Status::internal(e.to_string()))?;

        let info_dir = scratch.path().join("objects").join("info");
        std::fs::create_dir_all(&info_dir).map_err(|e| Status::internal(e.to_string()))?;
        let alternates = object_dirs
            .iter()
            .map(|p| p.display().to_string())
            .collect::<Vec<_>>()
            .join("\n");
        std::fs::write(info_dir.join("alternates"), format!("{alternates}\n"))
            .map_err(|e| Status::internal(e.to_string()))?;

        let repo = gix::open(scratch.path())
            .map_err(|e| Status::internal(e.to_string()))?
            .with_object_memory();
        Ok(TempRepository::new(repo, scratch))
    }
}

#[tonic::async_trait]
impl GitMirrorService for GitMirror {
    type DiffStream = FileDiffStream;

    async fn diff(
        &self,
        request: Request<DiffRequest>,
    ) -> Result<Response<Self::DiffStream>, Status> {
        let req = request.into_inner();
        let head = req
            .head
            .ok_or_else(|| Status::invalid_argument("head is required"))?;
        let base = req.base;
        debug!(
            base_repo = ?base.as_ref().map(|b| b.repo.as_str()),
            base_commit = ?base.as_ref().map(|b| b.commit.as_bstr()),
            head_repo = %head.repo,
            head_commit = %head.commit.as_bstr(),
            "diff request",
        );
        let scratch = match &base {
            Some(base) => self.open_scratch(&[head.repo.as_str(), base.repo.as_str()])?,
            None => self.open_scratch(&[head.repo.as_str()])?,
        };
        let head_commit_id = find_commit_by_sha(&scratch, &head.commit)?;
        let base_commit_id = base
            .as_ref()
            .map(|b| find_commit_by_sha(&scratch, &b.commit))
            .transpose()?;
        let comparison_type =
            ComparisonType::try_from(req.comparison_type).unwrap_or(ComparisonType::Unspecified);

        let (tx, rx) = mpsc::channel::<Result<FileDiff, Status>>(16);
        tokio::task::spawn_blocking(move || {
            let repo = &*scratch;

            let run = || -> anyhow::Result<()> {
                let head_commit = repo.find_commit(head_commit_id)?;
                let old = match base_commit_id {
                    Some(base_commit_id) => match comparison_type {
                        ComparisonType::OnlyInHead => {
                            repo.find_tree(repo.find_commit(base_commit_id)?.tree_id()?)?
                        }
                        ComparisonType::Unspecified | ComparisonType::Intersection => {
                            let merge_base =
                                repo.merge_base(base_commit_id, head_commit_id)?.detach();
                            repo.find_tree(repo.find_commit(merge_base)?.tree_id()?)?
                        }
                    },
                    None => match head_commit.parent_ids().next() {
                        Some(parent_id) => {
                            repo.find_tree(repo.find_commit(parent_id.detach())?.tree_id()?)?
                        }
                        None => repo.empty_tree(),
                    },
                };
                let new = repo.find_tree(head_commit.tree_id()?)?;
                for item in diff::diff(repo, &old, &new, false)? {
                    if tx.blocking_send(Ok(item?.into())).is_err() {
                        break;
                    }
                }
                Ok(())
            };
            if let Err(e) = run() {
                let _ = tx.blocking_send(Err(Status::internal(e.to_string())));
            }
        });

        Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
    }

    type InterdiffStream = FileDiffStream;

    async fn interdiff(
        &self,
        request: Request<InterdiffRequest>,
    ) -> Result<Response<Self::InterdiffStream>, Status> {
        let req = request.into_inner();
        let repo = self.open_repo(&req.repo)?;
        let from_base_id = resolve_commit_oid(&repo, &req.from_base)?;
        let from_head_id = resolve_commit_oid(&repo, &req.from_head)?;
        let to_base_id = resolve_commit_oid(&repo, &req.to_base)?;
        let to_head_id = resolve_commit_oid(&repo, &req.to_head)?;
        let safe = repo.into_sync();

        let (tx, rx) = mpsc::channel::<Result<FileDiff, Status>>(16);
        tokio::task::spawn_blocking(move || {
            let repo = safe.to_thread_local();
            let run = || -> anyhow::Result<()> {
                let to_head = repo.find_commit(to_head_id)?;
                let to_head_tree = to_head.tree()?;
                let rebased_tree =
                    prepare_interdiff(&repo, (from_base_id, from_head_id), to_base_id)?;
                for item in diff::diff(&repo, &rebased_tree, &to_head_tree, true)? {
                    if tx.blocking_send(Ok(item?.into())).is_err() {
                        break;
                    }
                }
                Ok(())
            };
            if let Err(e) = run() {
                let _ = tx.blocking_send(Err(Status::internal(e.to_string())));
            }
        });

        Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
    }

    type GetBlobStream = BlobChunkStream;

    async fn get_blob(
        &self,
        request: Request<GetBlobRequest>,
    ) -> Result<Response<Self::GetBlobStream>, Status> {
        let req = request.into_inner();
        let repo = self.open_repo(&req.repo)?;
        let oid = gix::ObjectId::from_hex(req.oid.as_bytes())
            .map_err(|e| Status::invalid_argument(format!("bad oid '{}': {e}", req.oid)))?;
        let safe = repo.into_sync();

        let (tx, rx) = mpsc::channel::<Result<BlobChunk, Status>>(16);
        tokio::task::spawn_blocking(move || {
            let repo = safe.to_thread_local();
            let run = || -> anyhow::Result<()> {
                let blob = repo.find_object(oid)?.try_into_blob()?;
                for chunk in blob.data.chunks(BLOB_CHUNK_SIZE) {
                    let msg = BlobChunk {
                        data: chunk.to_vec(),
                    };
                    if tx.blocking_send(Ok(msg)).is_err() {
                        break;
                    }
                }
                Ok(())
            };
            if let Err(e) = run() {
                let _ = tx.blocking_send(Err(Status::internal(e.to_string())));
            }
        });

        Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
    }

    type CommitLogStream = CommitLogResponseStream;

    async fn commit_log(
        &self,
        request: Request<CommitLogRequest>,
    ) -> Result<Response<Self::CommitLogStream>, Status> {
        let req = request.into_inner();
        debug!(
            repo = %req.repo,
            all_refs = req.all_refs,
            ranges = ?req.ranges,
            after = ?req.after,
            before = ?req.before,
            max_commits = req.max_commits,
            skip = req.skip,
            "commit_log request",
        );
        if req.all_refs != req.ranges.is_empty() {
            return Err(Status::invalid_argument(
                "specify exactly one of `ranges` or `all_refs`",
            ));
        }
        let repo = self.open_repo(&req.repo)?;
        let safe = repo.into_sync();

        let (tx, rx) = mpsc::channel::<Result<CommitLogResponse, Status>>(16);
        tokio::task::spawn_blocking(move || {
            let repo = safe.to_thread_local();
            let run = || -> anyhow::Result<()> {
                let after = req.after.map(|t| t.seconds);
                let before = req.before.map(|t| t.seconds);
                let max_commits = req.max_commits as usize;
                let mut skip = req.skip as usize;
                let mut sent = 0usize;
                let mut batch = Vec::with_capacity(COMMIT_LOG_BATCH_SIZE);

                for info in commit_log_walk(&repo, &req)? {
                    let info = info?;
                    let commit_time = info.commit_time();
                    if before.is_some_and(|b| commit_time > b) {
                        continue;
                    }
                    if after.is_some_and(|a| commit_time < a) {
                        break;
                    }
                    if skip > 0 {
                        skip -= 1;
                        continue;
                    }
                    if max_commits != 0 && sent >= max_commits {
                        break;
                    }
                    batch.push(GitCommit::try_from(&info.object()?)?);
                    sent += 1;
                    if batch.len() == COMMIT_LOG_BATCH_SIZE {
                        let msg = CommitLogResponse {
                            commits: std::mem::take(&mut batch),
                        };
                        if tx.blocking_send(Ok(msg)).is_err() {
                            return Ok(()); // client hung up
                        }
                    }
                }
                if !batch.is_empty() {
                    let _ = tx.blocking_send(Ok(CommitLogResponse { commits: batch }));
                }
                Ok(())
            };
            if let Err(e) = run() {
                let _ = tx.blocking_send(Err(Status::internal(e.to_string())));
            }
        });

        Ok(Response::new(Box::pin(ReceiverStream::new(rx))))
    }

    async fn merge_check(
        &self,
        request: Request<MergeCheckRequest>,
    ) -> Result<Response<MergeCheckResponse>, Status> {
        let req = request.into_inner();
        let target = req
            .target
            .ok_or_else(|| Status::invalid_argument("target is required"))?;
        let source = req
            .source
            .ok_or_else(|| Status::invalid_argument("source is required"))?;
        debug!(
            target_repo = %target.repo,
            target_commit = %target.commit.as_bstr(),
            source_repo = %source.repo,
            source_commit = %source.commit.as_bstr(),
            "merge_check request",
        );

        let scratch = self.open_scratch(&[target.repo.as_str(), source.repo.as_str()])?;

        let response =
            tokio::task::spawn_blocking(move || -> Result<MergeCheckResponse, Status> {
                let target_oid = find_commit_by_sha(&scratch, &target.commit)?;
                let source_oid = find_commit_by_sha(&scratch, &source.commit)?;
                crate::merge::merge_check(&scratch, target_oid, source_oid)
                    .map_err(|e| Status::internal(e.to_string()))
            })
            .await
            .map_err(|e| Status::internal(format!("merge task panicked: {e}")))??;

        Ok(Response::new(response))
    }

    async fn resolve_rev(
        &self,
        request: Request<ResolveRevRequest>,
    ) -> Result<Response<ResolveRevResponse>, Status> {
        let req = request.into_inner();
        debug!(repo = %req.repo, rev = %req.rev.as_bstr(), "resolve_rev request");
        let repo = self.open_repo(&req.repo)?;
        let commit = tokio::task::spawn_blocking(move || -> Result<String, Status> {
            Ok(resolve_simple_rev(&repo, &req.rev)?.to_string())
        })
        .await
        .map_err(|e| Status::internal(format!("resolve task panicked: {e}")))??;
        Ok(Response::new(ResolveRevResponse { commit }))
    }
}

fn resolve_commit_oid(repo: &gix::Repository, spec: &[u8]) -> Result<gix::ObjectId, Status> {
    let commit = repo
        .rev_parse_single(spec.as_bstr())
        .map_err(|e| {
            Status::invalid_argument(format!("bad rev '{}': {e}", String::from_utf8_lossy(spec)))
        })?
        .object()
        .map_err(|e| Status::internal(e.to_string()))?
        .peel_to_commit()
        .map_err(|e| Status::internal(e.to_string()))?;
    Ok(commit.id)
}

fn resolve_simple_rev(repo: &gix::Repository, rev: &[u8]) -> Result<gix::ObjectId, Status> {
    let unsupported = rev == b"HEAD"
        || rev == b"@"
        || rev.windows(2).any(|w| w == b"@{")
        || rev.iter().any(|&b| matches!(b, b'~' | b'^' | b':'));
    if unsupported {
        return Err(Status::invalid_argument(format!(
            "unsupported rev '{}': only a branch, tag, or commit oid is allowed",
            rev.as_bstr()
        )));
    }
    resolve_commit_oid(repo, rev)
}

fn find_commit_by_sha(repo: &gix::Repository, sha: &[u8]) -> Result<gix::ObjectId, Status> {
    let oid = gix::ObjectId::from_hex(sha).map_err(|e| {
        Status::invalid_argument(format!("bad commit sha '{}': {e}", sha.as_bstr()))
    })?;
    repo.find_commit(oid)
        .map_err(|_| Status::not_found(format!("commit not found: {oid}")))?;
    Ok(oid)
}

fn commit_log_walk<'repo>(
    repo: &'repo gix::Repository,
    req: &CommitLogRequest,
) -> anyhow::Result<gix::revision::Walk<'repo>> {
    let mut tips = Vec::new();
    let mut hidden = Vec::new();

    if req.all_refs {
        for r in repo.references()?.all()? {
            let mut r = r.map_err(|e| anyhow::anyhow!(e))?;
            if let Ok(commit) = r.peel_to_commit() {
                tips.push(commit.id);
            }
        }
    } else {
        for range in &req.ranges {
            let revspec = repo.rev_parse(range.as_bstr())?;
            let spec = revspec.detach();
            match spec {
                RevSpec::Include(id) => tips.push(id),
                RevSpec::Range { from, to } => {
                    tips.push(to);
                    hidden.push(from);
                }
                _ => {
                    anyhow::bail!("The spec isn't currently supported: {spec:?}")
                }
            }
        }
    }

    Ok(repo
        .rev_walk(tips)
        .sorting(Sorting::ByCommitTime(Default::default()))
        .with_hidden(hidden)
        .all()?)
}

impl TryFrom<&gix::Commit<'_>> for GitCommit {
    type Error = anyhow::Error;

    fn try_from(commit: &gix::Commit<'_>) -> anyhow::Result<Self> {
        let decoded = commit.decode()?;
        Ok(GitCommit {
            oid: commit.id.to_string(),
            author: Some(decoded.author()?.try_into()?),
            committer: Some(decoded.committer()?.try_into()?),
            message: decoded.message.to_vec(),
            parents: decoded
                .parents
                .into_iter()
                .map(|id| id.to_string())
                .collect(),
            extra_headers: decoded
                .extra_headers
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect(),
        })
    }
}

impl TryFrom<gix::actor::SignatureRef<'_>> for GitSignature {
    type Error = anyhow::Error;

    fn try_from(sig: gix::actor::SignatureRef<'_>) -> anyhow::Result<Self> {
        let time = sig.time()?;
        Ok(GitSignature {
            name: sig.name.to_vec(),
            email: sig.email.to_vec(),
            date: Some(prost_types::Timestamp {
                seconds: time.seconds,
                nanos: 0,
            }),
        })
    }
}

pub async fn serve(addr: SocketAddr, repo_base: PathBuf) -> anyhow::Result<()> {
    let service = GitMirror { repo_base };
    info!(addr = %addr, "gitmirror gRPC server listening");

    Server::builder()
        .add_service(GitMirrorServiceServer::new(service))
        .serve(addr)
        .await?;

    Ok(())
}

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

    #[test]
    fn temp_repository_removes_scratch_dir_on_drop() {
        let dir = tempfile::tempdir().unwrap();
        gix::init_bare(dir.path()).unwrap();
        let repo = gix::open(dir.path()).unwrap().with_object_memory();
        let temp = TempRepository::new(repo, dir);
        let path = temp.git_dir().to_path_buf();
        assert!(path.exists());
        // Derefs to gix::Repository.
        assert!(temp.object_hash() == gix::hash::Kind::Sha1);
        drop(temp);
        assert!(!path.exists(), "scratch dir should be gone after drop");
    }
}