use std::net::SocketAddr;

use clap::Parser;
use tonic::transport::Server;
use tonic::{Request, Response, Status};
use tracing::info;

mod diff;
mod protocol;

pub mod gitmirror {
    pub mod v1 {
        tonic::include_proto!("gitmirror.v1");
    }
}

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

use gix::bstr::ByteSlice as _;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tokio_stream::Stream;

use gix::revision::plumbing::Spec as RevSpec;
use gix::revision::walk::Sorting;

use gitmirror::v1::git_mirror_service_server::{GitMirrorService, GitMirrorServiceServer};
use gitmirror::v1::{
    BlobChunk, ByteChanges, CommitLogRequest, CommitLogResponse, CommitsRequest, CommitsResponse,
    DiffRequest, FileContent, FileDiff, GetBlobRequest, GitCommit, GitSignature, Hunk,
    InterdiffRequest, LinePair,
};

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;

#[derive(Parser)]
#[command(name = "gitmirror", about = "Git mirror gRPC service")]
struct Args {
    /// Address to bind the gRPC server to.
    #[arg(long, env = "GITMIRROR_ADDR", default_value = "127.0.0.1:9000")]
    addr: SocketAddr,

    /// Base directory holding bare mirror repos, one per DID (<base>/<did>).
    #[arg(long, env = "GITMIRROR_REPO_BASE", default_value = "repos")]
    repo_base: PathBuf,
}

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

}

/// 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()?)
}

#[tonic::async_trait]
impl GitMirrorService for GitMirror {
    async fn commits(
        &self,
        _request: Request<CommitsRequest>,
    ) -> Result<Response<CommitsResponse>, Status> {
        Err(Status::unimplemented("not implemented"))
    }

    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 old_oid = resolve_tree_oid(&repo, &req.base)?;
        let new_oid = resolve_tree_oid(&repo, &req.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 old = repo.find_tree(old_oid)?;
                let new = repo.find_tree(new_oid)?;
                for item in diff::diff(&repo, &old, &new)? {
                    let proto = to_proto(item?);
                    if tx.blocking_send(Ok(proto)).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> {
        Err(Status::unimplemented("not implemented"))
    }

    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();
        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 tree oid.
fn resolve_tree_oid(repo: &gix::Repository, spec: &str) -> Result<gix::ObjectId, Status> {
    let tree = repo
        .rev_parse_single(spec.as_bytes().as_bstr())
        .map_err(|e| Status::invalid_argument(format!("bad rev '{spec}': {e}")))?
        .object()
        .map_err(|e| Status::internal(e.to_string()))?
        .peel_to_tree()
        .map_err(|e| Status::internal(e.to_string()))?;
    Ok(tree.id)
}

fn to_proto(d: diff::Diff) -> FileDiff {
    FileDiff {
        lhs_src: Some(file_content(d.lhs_src)),
        rhs_src: Some(file_content(d.rhs_src)),
        hunks: d.hunks.into_iter().map(hunk).collect(),
        has_byte_changes: d.has_byte_changes.map(|(lhs, rhs)| ByteChanges {
            lhs: lhs as u64,
            rhs: rhs as u64,
        }),
        has_syntactic_changes: d.has_syntactic_changes,
    }
}

fn file_content(f: diff::FileContent) -> FileContent {
    FileContent {
        path: f.path,
        oid: f.oid,
        size: f.size as u64,
        is_binary: f.is_binary,
        is_submodule: f.is_submodule,
    }
}

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

    fn try_from(commit: &gix::Commit<'_>) -> anyhow::Result<Self> {
        Ok(GitCommit {
            oid: commit.id.to_string(),
            author: Some(commit.author()?.try_into()?),
            committer: Some(commit.committer()?.try_into()?),
            message: commit.message_raw()?.to_vec(),
            parents: commit.parent_ids().map(|id| id.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,
            }),
        })
    }
}

fn hunk(h: diff::Hunk) -> Hunk {
    // Sort the novel line sets for deterministic output.
    let sorted = |set: rustc_hash::FxHashSet<line_numbers::LineNumber>| -> Vec<u32> {
        let mut v: Vec<u32> = set.into_iter().map(|n| n.0).collect();
        v.sort_unstable();
        v
    };
    Hunk {
        novel_lhs: sorted(h.novel_lhs),
        novel_rhs: sorted(h.novel_rhs),
        lines: h
            .lines
            .into_iter()
            .map(|(lhs, rhs)| LinePair {
                lhs: lhs.map(|n| n.0),
                rhs: rhs.map(|n| n.0),
            })
            .collect(),
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()),
        )
        .init();

    let args = Args::parse();
    let service = GitMirror {
        repo_base: args.repo_base,
    };
    info!(addr = %args.addr, "gitmirror gRPC server listening");

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

    Ok(())
}