use std::{cmp::Reverse, sync::Arc, time::SystemTime};

use log::{error, info, warn};
use priority_queue::PriorityQueue;
use tokio::sync::mpsc::UnboundedSender;

use crate::meta::{util::BgRunner, writer::WorkerNodeMgr};

pub enum HealthCheckerCommands {
    Hup,
    Add(String),
    Ping(String),
}

#[derive(Default)]
pub struct HealthCheckMgr {
    heap: PriorityQueue<String, Reverse<SystemTime>>,
}

impl HealthCheckMgr {
    pub fn add(&mut self, addr: String) {
        let now = SystemTime::now();
        self.heap.push(addr, Reverse(now));
    }

    pub fn update(&mut self, addr: &String, time: SystemTime) {
        self.heap.change_priority(addr, Reverse(time));
    }

    pub fn ping(&mut self, addr: &String) {
        self.update(addr, SystemTime::now());
    }

    pub fn get_all_timeout(&mut self, milliseconds: u64) -> Vec<String> {
        let mut result = vec![];
        let now = SystemTime::now();
        while let Some((addr, Reverse(time))) = self.heap.peek() {
            let dur = now.duration_since(*time);
            match dur {
                Err(e) => {
                    warn!("error: e={:?}", e);
                }
                Ok(dur) => {
                    if dur.as_millis() > milliseconds as u128 {
                        result.push(addr.clone());
                        self.heap.pop();
                    } else {
                        break;
                    }
                }
            }
        }
        result
    }
}

impl HealthCheckMgr {
    pub fn create_runner() -> (
        BgRunner,
        UnboundedSender<HealthCheckerCommands>,
        tokio::sync::oneshot::Sender<Arc<WorkerNodeMgr>>,
    ) {
        let (stx2, srx2) = tokio::sync::oneshot::channel();
        let (health_tx, mut health_rx) = tokio::sync::mpsc::unbounded_channel();
        let health_runner = BgRunner::spawn_name(
            "HealthChecker".to_string(),
            move |mut rf| async move {
                let mgr: Arc<WorkerNodeMgr> = match srx2.await {
                    Ok(a) => a,
                    Err(e) => {
                        error!("HealthCheckMgr: init failed, can not run, quit, e={:?}", e);
                        return;
                    }
                };
                let mut checker = HealthCheckMgr::default();
                let check_millis = mgr
                    .config
                    .load()
                    .worker_node
                    .health_check_interval
                    .as_millis();
                let mut interval =
                    tokio::time::interval(tokio::time::Duration::from_millis(check_millis));
                interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
                info!("HealthChecker: start");
                loop {
                    tokio::select! {
                        _ = &mut rf => {
                            // Since it hold an reference of Arc<WorkerNodeMgr>,
                            // it can't be dropped.
                            info!("HealthChecker: quit");
                            return;
                        },
                        msg = health_rx.recv() => {
                            if let Some(cmd) = msg {
                                match cmd {
                                    HealthCheckerCommands::Hup => {
                                        info!("HealthChecker: closed");
                                        return;
                                    }
                                    HealthCheckerCommands::Add(addr) => {
                                        checker.add(addr.clone());
                                        checker.ping(&addr);
                                    }
                                    HealthCheckerCommands::Ping(addr) => {
                                        checker.ping(&addr);
                                    }
                                }
                            } else {
                                info!("HealthChecker: sender closed");
                                return;
                            }
                        }
                        _ = interval.tick() => {
                            let timeouts = checker
                                .get_all_timeout(mgr.config.load().worker_node.worker_node_down_timeout.as_millis());
                            for addr in timeouts {
                                mgr.remove_worker_node(&addr, "timeout").await;
                            }
                        },
                    }
                }
            },
        );
        (health_runner, health_tx, stx2)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_timeout() {
        let mut m = HealthCheckMgr::default();
        m.add("a".to_string());
        m.add("b".to_string());
        m.update(
            &"a".to_string(),
            SystemTime::now() - std::time::Duration::from_secs(2),
        );
        assert_eq!(m.get_all_timeout(1000), vec!["a".to_string()]);
    }
}