use std::sync::Arc;

use cache::AppCache;
use tokio_stream::wrappers::ReceiverStream;
use tonic::{Request, Response, Status};

use crate::rpc::{
    error::{spawn_blocking_error, to_status},
    proto as p,
    registry::RepoRegistry,
};

pub struct BlameServiceImpl {
    pub registry: Arc<RepoRegistry>,
    pub cache: AppCache,
}

type BlameStream = ReceiverStream<Result<p::CommitBlameHunk, Status>>;

#[tonic::async_trait]
impl p::blame_service_server::BlameService for BlameServiceImpl {
    type BlameStreamStream = BlameStream;

    async fn blame_file(
        &self,
        req: Request<p::BlameFileRequest>,
    ) -> Result<Response<p::BlameFileResponse>, Status> {
        let inner = req.into_inner();
        let repo_id = inner.repo_id.clone();
        let rev_str = inner.rev.unwrap_or_default();
        let path = inner.path.clone();
        let cache_key = format!(
            "git:rpc:cache:blame:file:{}:{}:{}",
            repo_id, path, rev_str
        );

        if let Ok(Some(cached)) =
            self.cache.get::<p::BlameFileResponse>(&cache_key).await
        {
            return Ok(Response::new(cached));
        }

        let bare = self.registry.get(&repo_id).await?;
        let oid = crate::cmd::oid::ObjectId::new(&rev_str);
        let opts = inner.options.map(Into::into);
        let result = tokio::task::spawn_blocking(move || {
            bare.blame_file(oid, &path, opts)
        })
        .await
        .map_err(spawn_blocking_error)?
        .map_err(to_status)?;
        let resp = p::BlameFileResponse {
            hunks: result.into_iter().map(Into::into).collect(),
        };
        let _ = self.cache.set(&cache_key, &resp).await;
        Ok(Response::new(resp))
    }

    async fn blame_stream(
        &self,
        req: Request<p::BlameFileRequest>,
    ) -> Result<Response<BlameStream>, Status> {
        let inner = req.into_inner();
        let bare = self.registry.get(&inner.repo_id).await?;
        let rev_str = inner.rev.unwrap_or_default();
        let oid = crate::cmd::oid::ObjectId::new(&rev_str);
        let path = inner.path.clone();
        let opts = inner.options.map(Into::into);
        let (tx, rx) = tokio::sync::mpsc::channel(128);
        tokio::task::spawn_blocking(move || {
            let result = bare.blame_file(oid, &path, opts);
            match result {
                Ok(hunks) => {
                    for hunk in hunks {
                        if tx.blocking_send(Ok(hunk.into())).is_err() {
                            break;
                        }
                    }
                }
                Err(e) => {
                    let _ = tx.blocking_send(Err(to_status(e)));
                }
            }
        });
        Ok(Response::new(ReceiverStream::new(rx)))
    }

    async fn blame_hunk(
        &self,
        req: Request<p::BlameHunkRequest>,
    ) -> Result<Response<p::BlameHunkResponse>, Status> {
        let inner = req.into_inner();
        let bare = self.registry.get(&inner.repo_id).await?;
        let rev_str = inner.rev.unwrap_or_default();
        let oid = crate::cmd::oid::ObjectId::new(&rev_str);
        let path = inner.path.clone();
        let line_on = inner.start_line as usize;
        let result = tokio::task::spawn_blocking(move || {
            bare.blame_hunk(&oid, &path, line_on)
        })
        .await
        .map_err(spawn_blocking_error)?
        .map_err(to_status)?;
        Ok(Response::new(p::BlameHunkResponse {
            hunks: vec![result.into()],
        }))
    }

    async fn blame_lines(
        &self,
        req: Request<p::BlameLinesRequest>,
    ) -> Result<Response<p::BlameLinesResponse>, Status> {
        let inner = req.into_inner();
        let bare = self.registry.get(&inner.repo_id).await?;
        let rev_str = inner.rev.unwrap_or_default();
        let oid = crate::cmd::oid::ObjectId::new(&rev_str);
        let path = inner.path.clone();
        let opts = Some(crate::cmd::blame::BlameOptions {
            min_line: Some(inner.start_line as usize),
            max_line: Some(inner.end_line as usize),
            track_copies_same_file: false,
            track_copies_same_commit_moves: false,
            ignore_whitespace: false,
        });
        let result = tokio::task::spawn_blocking(move || {
            bare.blame_lines(oid, &path, opts)
        })
        .await
        .map_err(spawn_blocking_error)?
        .map_err(to_status)?;
        Ok(Response::new(p::BlameLinesResponse {
            lines: result.into_iter().map(Into::into).collect(),
        }))
    }
}
