use anyhow::Result;
use log::info;
use tokio::{
    sync::{mpsc::UnboundedReceiver, oneshot},
    task::JoinHandle,
};

pub fn unwrap_grpc_error<R>(
    r: Result<tonic::Response<R>, tonic::Status>,
) -> Result<tonic::Response<R>> {
    match r {
        Ok(r) => Ok(r),
        Err(s) => Err(anyhow::format_err!(
            "receive error response of {}, code={} message={} details={:?}",
            std::any::type_name::<R>(),
            s.code(),
            s.message(),
            s.details()
        )),
    }
}

pub struct BgRunner {
    tf: Option<oneshot::Sender<()>>,
    name: String,
    handle: Option<JoinHandle<()>>,
}

impl BgRunner {
    pub fn spawn<F, Fut>(f: F) -> Self
    where
        F: FnOnce(oneshot::Receiver<()>) -> Fut + Send + 'static,
        Fut: std::future::Future<Output = ()> + Send + 'static,
    {
        let (tf, rf) = oneshot::channel();
        let handle = tokio::spawn(f(rf));
        Self {
            tf: Some(tf),
            name: String::new(),
            handle: Some(handle),
        }
    }

    pub fn spawn_name<F, Fut>(name: String, f: F) -> Self
    where
        F: FnOnce(oneshot::Receiver<()>) -> Fut + Send + 'static,
        Fut: std::future::Future<Output = ()> + Send + 'static,
    {
        let (tf, rf) = oneshot::channel();
        let handle = tokio::spawn(f(rf));
        Self {
            tf: Some(tf),
            name,
            handle: Some(handle),
        }
    }

    pub fn wrapper<Initial, R, Handler, Quit, Fut>(
        tag: String,
        mut new_task_fetcher: UnboundedReceiver<R>,
        mut handler: Handler,
        quit: Quit,
    ) -> (BgRunner, tokio::sync::oneshot::Sender<Initial>)
    where
        Initial: Clone + Send + 'static,
        R: Send + 'static,
        Handler: FnMut(Initial, R) -> Fut + Send + 'static,
        Quit: FnOnce(Initial) + Send + 'static,
        Fut: std::future::Future<Output = bool> + Send + 'static,
    {
        let (initial_tx, initial_rx) = tokio::sync::oneshot::channel();
        let runner = BgRunner::spawn_name(tag.clone(), move |mut rf| async move {
            // We allow this unwrap because we can't propagate the error anymore. The thread
            // must panic, and according to panic hook, the process will abort.
            #[allow(clippy::unwrap_used)]
            let initial_ctx = initial_rx.await.unwrap();
            info!("{}: BgRunner start", tag);
            loop {
                tokio::select! {
                    _ = &mut rf => {
                        // Myself is dropped
                        info!("{}: quit", tag);
                        quit(initial_ctx);
                        return;
                    },
                    new_task = new_task_fetcher.recv() => {
                        if let Some(x) = new_task {
                            if handler(initial_ctx.clone(), x).await {
                                info!("{}: handler quit", tag);
                                quit(initial_ctx);
                                return;
                            }
                        } else {
                            // Sender is dropped
                            info!("{}: sender quit", tag);
                            quit(initial_ctx);
                            return;
                        }
                    },
                }
            }
        });
        (runner, initial_tx)
    }

    // Some users may shutdown a future task by calling abort on join handle,
    // this may cause leak of user defined structures, and leave the state before
    // shutdown uncertain, so we use explicit shutdown in BgRunner.
    pub fn explicit_shutdown(&mut self) -> Option<JoinHandle<()>> {
        if let Some(tf) = self.tf.take() {
            let e = tf.send(());
            if !self.name.is_empty() {
                if e.is_ok() {
                    info!("explicitly shutdown BgRunner {}", self.name);
                } else {
                    info!("explicitly shutdown BgRunner {}, error={:?}", self.name, e);
                }
            }
        }
        self.handle.take()
    }
}

impl Drop for BgRunner {
    fn drop(&mut self) {
        info!("{}: BgRunner dropped", self.name);
        let _ = self.explicit_shutdown();
    }
}