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 tokio::sync::mpsc;
use tokio_stream::Stream;
use tokio_stream::wrappers::ReceiverStream;
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,
};

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>>;

/// Blob bytes are streamed in chunks of this size.
const BLOB_CHUNK_SIZE: usize = 64 * 1024;
/// GitCommits are streamed in batches of this size.
const COMMIT_LOG_BATCH_SIZE: usize = 128;

struct GitMirror {
    repo_base: PathBuf,
}

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}")))
    }
}

#[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 repo = self.open_repo(&req.repo)?;
        // Resolve refs up-front so bad input surfaces as the RPC status, then move the
        // owned repo + oids into a blocking task for the (synchronous) diff loop.
        let base_oid = req
            .base_rev_spec
            .as_deref()
            .map(|b| resolve_commit_oid(&repo, b))
            .transpose()?;
        let head_oid = resolve_commit_oid(&repo, &req.head_rev_spec)?;
        let comparison_type =
            ComparisonType::try_from(req.comparison_type).unwrap_or(ComparisonType::Unspecified);
        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 head_commit = repo.find_commit(head_oid)?;
                let old = match base_oid {
                    Some(base_oid) => match comparison_type {
                        ComparisonType::OnlyInHead => {
                            repo.find_tree(repo.find_commit(base_oid)?.tree_id()?)?
                        }
                        ComparisonType::Unspecified | ComparisonType::Intersection => {
                            let merge_base = repo.merge_base(base_oid, head_oid)?.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; // client hung up
                    }
                }
                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; // client hung up
                    }
                }
                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; // client hung up
                    }
                }
                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; // newest-first order: nothing older will match either
                    }
                    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))))
    }
}

/// Resolve a commit-ish ref string (oid, short oid, branch/tag) to its commit oid.
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)
}

/// Resolve tips/hidden commits from a [`CommitLogRequest`] and return the resulting gix walk,
/// sorted newest-first to match `git log`'s default order.
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,
            }),
        })
    }
}

/// Build the gRPC service and serve it until the process is stopped.
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(())
}