/// Service basics
use elbus::client::AsyncClient;
use elbus::rpc::{Rpc, RpcClient, RpcError, RpcEvent, RpcResult};
use elbus::QoS;
use eva_common::acl::OIDMaskList;
use eva_common::events::{
    ANY_STATE_TOPIC, LOCAL_STATE_TOPIC, LOG_INPUT_TOPIC, REMOTE_STATE_TOPIC, SERVICE_STATUS_TOPIC,
};
use eva_common::prelude::*;
use eva_common::services;
use eva_common::SLEEP_STEP;
use lazy_static::lazy_static;
use log::{Level, Log};
use serde::Deserialize;
use std::collections::HashMap;
use std::io::Read;
use std::path::Path;
use std::sync::atomic;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::{Duration, Instant};
use tokio::io::AsyncReadExt;
use tokio::signal::unix::{signal, SignalKind};
use tokio::time::sleep;
pub use eva_sdk_derive::svc_main;

#[derive(Debug, Copy, Clone, Eq, PartialEq)]
pub enum EventKind {
    Local,
    Remote,
    Any,
}

impl EventKind {
    #[inline]
    pub fn topic(&self) -> &str {
        match self {
            EventKind::Local => LOCAL_STATE_TOPIC,
            EventKind::Remote => REMOTE_STATE_TOPIC,
            EventKind::Any => ANY_STATE_TOPIC,
        }
    }
}

pub async fn subscribe_oids<R: Rpc>(rpc: &R, masks: &OIDMaskList, kind: EventKind) -> EResult<()> {
    let topics: Vec<String> = masks
        .oid_masks()
        .iter()
        .map(|mask| format!("{}{}", kind.topic(), mask.to_path()))
        .collect();
    rpc.client()
        .lock()
        .await
        .subscribe_bulk(topics.iter().map(String::as_str).collect(), QoS::No)
        .await?;
    Ok(())
}

lazy_static! {
    static ref NEED_PANIC: Mutex<Option<Duration>> = <_>::default();
}

/// # Panics
///
/// Will panic if the mutex is poisoned
pub fn set_poc(panic_in: Option<Duration>) {
    *NEED_PANIC.lock().unwrap() = panic_in;
}

/// # Panics
///
/// Will panic if the mutex is poisoned
pub fn poc() {
    svc_terminate();
    if let Some(delay) = *NEED_PANIC.lock().unwrap() {
        bmart::process::suicide(delay + Duration::from_secs(1), false);
        std::thread::spawn(move || {
            std::thread::sleep(delay);
            std::process::exit(1);
        });
    }
}

pub fn svc_handle_default_rpc(method: &str, info: &services::ServiceInfo) -> RpcResult {
    match method {
        "test" => Ok(None),
        "info" => Ok(Some(rmp_serde::to_vec_named(info)?)),
        "stop" => {
            svc_terminate();
            Ok(None)
        }
        _ => Err(RpcError::method(None)),
    }
}

#[inline]
pub async fn safe_rpc_call(
    rpc: &RpcClient,
    target: &str,
    method: &str,
    params: elbus::borrow::Cow<'_>,
    qos: QoS,
    timeout: Duration,
) -> EResult<RpcEvent> {
    tokio::time::timeout(timeout, rpc.call(target, method, params, qos))
        .await?
        .map_err(Into::into)
}

#[inline]
pub async fn svc_wait_core(rpc: &RpcClient, timeout: Duration, wait_forever: bool) -> EResult<()> {
    #[derive(Deserialize)]
    struct TR {
        active: bool,
    }
    let wait_until = Instant::now() + timeout;
    loop {
        if svc_is_terminating() {
            return Err(Error::failed(
                "core load wait aborted, the service is not active",
            ));
        }
        if let Ok(ev) = safe_rpc_call(
            rpc,
            "eva.core",
            "test",
            elbus::empty_payload!(),
            QoS::Processed,
            timeout,
        )
        .await
        {
            if let Ok(result) = rmp_serde::from_slice::<TR>(ev.payload()) {
                if result.active {
                    return Ok(());
                }
            }
        }
        tokio::time::sleep(SLEEP_STEP).await;
        if !wait_forever && wait_until >= Instant::now() {
            return Err(Error::timeout());
        }
    }
}

static ACTIVE: atomic::AtomicBool = atomic::AtomicBool::new(false);
static TERMINATING: atomic::AtomicBool = atomic::AtomicBool::new(false);

#[inline]
pub fn svc_is_active() -> bool {
    ACTIVE.load(atomic::Ordering::SeqCst)
}

#[inline]
pub fn svc_is_terminating() -> bool {
    TERMINATING.load(atomic::Ordering::SeqCst)
}

#[macro_export]
macro_rules! svc_need_ready {
    () => {
        if !svc_is_active() {
            return;
        }
    };
}

#[macro_export]
macro_rules! svc_rpc_need_ready {
    () => {
        if !svc_is_active() {
            return Err(Error::not_ready("service is not ready").into());
        }
    };
}

lazy_static::lazy_static! {
    static ref LOG_TOPICS: HashMap<Level, String> = {
        let mut topics = HashMap::new();
        topics.insert(Level::Trace, format!("{}{}", LOG_INPUT_TOPIC, "trace"));
        topics.insert(Level::Debug, format!("{}{}", LOG_INPUT_TOPIC, "debug"));
        topics.insert(Level::Info, format!("{}{}", LOG_INPUT_TOPIC, "info"));
        topics.insert(Level::Warn, format!("{}{}", LOG_INPUT_TOPIC, "warn"));
        topics.insert(Level::Error, format!("{}{}", LOG_INPUT_TOPIC, "error"));
        topics
    };
    static ref LOG_TX:
            Mutex<Option<async_channel::Sender<(log::Level, String)>>> = Mutex::new(None);
}

static BUS_LOGGER: BusLogger = BusLogger {};

struct BusLogger {}

/// Terminate the service (canceling block)
#[inline]
pub fn svc_terminate() {
    ACTIVE.store(false, atomic::Ordering::SeqCst);
    TERMINATING.store(true, atomic::Ordering::SeqCst);
}

/// Block the service until terminate is called
#[inline]
pub async fn svc_block(rpc: &RpcClient) {
    while svc_is_active() && rpc.is_connected() {
        sleep(SLEEP_STEP).await;
    }
}

impl Log for BusLogger {
    #[inline]
    fn enabled(&self, metadata: &log::Metadata) -> bool {
        !metadata.target().starts_with("elbus::") && !metadata.target().starts_with("mio::")
    }
    #[inline]
    fn log(&self, record: &log::Record) {
        if self.enabled(record.metadata()) {
            if let Some(tx) = LOG_TX.lock().unwrap().as_ref() {
                let _r = tx.try_send((record.level(), format!("{}", record.args())));
            }
        }
    }
    #[inline]
    fn flush(&self) {}
}

async fn handle_logs<C: ?Sized>(
    client: Arc<tokio::sync::Mutex<C>>,
    rx: async_channel::Receiver<(Level, String)>,
) where
    C: AsyncClient,
{
    while let Ok((level, message)) = rx.recv().await {
        client
            .lock()
            .await
            .publish(
                LOG_TOPICS.get(&level).unwrap(),
                message.as_bytes().into(),
                QoS::No,
            )
            .await
            .unwrap();
    }
}

/// Initializing service logs
///
/// After calling, log macros can be used, all records are transferred to bus LOG/IN/ topics
///
/// # Panics
///
/// Will panic if the mutex is poisoned
pub fn svc_init_logs<C: ?Sized>(
    initial: &services::Initial,
    client: Arc<tokio::sync::Mutex<C>>,
) -> EResult<()>
where
    C: AsyncClient + 'static,
{
    let (tx, rx) = async_channel::bounded(initial.elbus_queue_size());
    LOG_TX.lock().unwrap().replace(tx);
    tokio::spawn(async move {
        handle_logs(client, rx).await;
    });
    log::set_logger(&BUS_LOGGER)
        .map(|()| log::set_max_level(initial.eva_log_level_filter()))
        .map_err(Error::failed)?;
    Ok(())
}

/// Sends a broadcast event to mark the service ready at launcher and announce neighbors
///
/// # Panics
///
/// Will panic only if rmp_serde is broken
pub async fn svc_mark_ready<C: ?Sized>(client: &tokio::sync::Mutex<C>) -> EResult<()>
where
    C: AsyncClient + 'static,
{
    client
        .lock()
        .await
        .publish(
            SERVICE_STATUS_TOPIC,
            rmp_serde::to_vec_named(&services::ServiceStatusBroadcastEvent::ready())
                .unwrap()
                .into(),
            QoS::Processed,
        )
        .await?
        .unwrap()
        .await??;
    ACTIVE.store(true, atomic::Ordering::SeqCst);
    Ok(())
}

/// Sends a broadcast event to mark the service terminating at launcher and announce neighbors
///
/// # Panics
///
/// Will panic only if rmp_serde is broken
pub async fn svc_mark_terminating<C: ?Sized>(client: &tokio::sync::Mutex<C>) -> EResult<()>
where
    C: AsyncClient + 'static,
{
    ACTIVE.store(false, atomic::Ordering::SeqCst);
    TERMINATING.store(true, atomic::Ordering::SeqCst);
    client
        .lock()
        .await
        .publish(
            SERVICE_STATUS_TOPIC,
            rmp_serde::to_vec_named(&services::ServiceStatusBroadcastEvent::terminating())
                .unwrap()
                .into(),
            QoS::Processed,
        )
        .await?
        .unwrap()
        .await??;
    Ok(())
}

/// Start service signal handlers (SIGTERM and SIGINT)
///
/// Calls the terminate method when received
pub fn svc_start_signal_handlers() {
    macro_rules! handle_signal {
        ($signal: expr) => {{
            tokio::spawn(async move {
                signal($signal).unwrap().recv().await;
                svc_terminate();
            });
        }};
    }
    macro_rules! ignore_signal {
        ($signal: expr) => {{
            tokio::spawn(async move {
                loop {
                    signal($signal).unwrap().recv().await;
                }
            });
        }};
    }
    handle_signal!(SignalKind::terminate());
    handle_signal!(SignalKind::hangup());
    ignore_signal!(SignalKind::interrupt());
}

fn process_initial(buf: &[u8]) -> EResult<services::Initial> {
    let initial: services::Initial = rmp_serde::from_read_ref(buf)?;
    if initial.config_version() != services::SERVICE_CONFIG_VERSION {
        return Err(Error::not_implemented(format!(
            "config version not supported: {}",
            initial.config_version()
        )));
    }
    if initial.eapi_version() != crate::EAPI_VERSION {
        return Err(Error::not_implemented(format!(
            "EAPI version not supported: {}",
            initial.config_version(),
        )));
    }
    Ok(initial)
}

pub async fn read_initial() -> EResult<services::Initial> {
    let mut stdin = tokio::io::stdin();
    let mut buf: Vec<u8> = Vec::new();
    tokio::time::timeout(eva_common::DEFAULT_TIMEOUT, stdin.read_to_end(&mut buf)).await??;
    process_initial(&buf)
}

pub fn read_initial_sync() -> EResult<services::Initial> {
    let mut buf: Vec<u8> = Vec::new();
    std::io::stdin().read_to_end(&mut buf)?;
    process_initial(&buf)
}

pub fn svc_launch<L, LFut>(launcher: L) -> EResult<()>
where
    L: FnMut(services::Initial) -> LFut,
    LFut: std::future::Future<Output = EResult<()>>,
{
    let initial = read_initial_sync()?;
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(initial.workers() as usize)
        .enable_all()
        .build()?;
    rt.block_on(launch(launcher, initial))?;
    Ok(())
}

async fn launch<L, LFut>(mut launcher: L, mut initial: services::Initial) -> EResult<()>
where
    L: FnMut(services::Initial) -> LFut,
    LFut: std::future::Future<Output = EResult<()>>,
{
    let op = eva_common::op::Op::new(initial.startup_timeout());
    let eva_dir = initial.eva_dir().to_owned();
    if let Some(prepare_command) = initial.prepare_command() {
        let cmd = format!("cd \"{}\" && {}", eva_dir, prepare_command);
        let res = bmart::process::command("sh", ["-c", &cmd], op.timeout()?, None, None).await?;
        if !res.ok() {
            return Err(Error::failed(format!(
                "prepare command failed: {}",
                res.err.join("\n")
            )));
        }
    }
    initial
        .extend_config(op.timeout()?, Path::new(&eva_dir))
        .await?;
    launcher(initial).await
}