use log::{info, warn};
use tici_common::index::IndexKey;
use tici_grpc::{IndexInfo, TableInfo};

use crate::{
    ShardInfo, ShardInfoAppendUpdate,
    meta::{metrics::SHARD_EVENT_COUNTER_VEC, util::BgRunner, writer::FTSIndex},
};

pub trait ShardEventSubscriberTrait {
    fn on_event(
        &mut self,
        e: ShardEvent,
    ) -> impl std::future::Future<Output = ()> + std::marker::Send;
}

#[derive(Clone, Debug)]
pub struct ShardEventIndexInfo {
    pub table_info: TableInfo,
    pub index_info: IndexInfo,
}

impl ShardEventIndexInfo {
    pub fn new(table_info: TableInfo, index_info: IndexInfo) -> Self {
        Self {
            table_info,
            index_info,
        }
    }

    pub fn index_key(&self) -> IndexKey {
        IndexKey::from(&self.index_info)
    }
}

impl From<&FTSIndex> for ShardEventIndexInfo {
    fn from(value: &FTSIndex) -> Self {
        Self {
            table_info: value.table_info.clone(),
            index_info: value.index_info.clone(),
        }
    }
}

#[derive(Clone, Debug)]
pub enum ShardEvent {
    FullSync(Vec<ShardEventIndexInfo>, Vec<ShardInfo>),
    Add(ShardEventIndexInfo, ShardInfo),
    Remove(ShardEventIndexInfo, Vec<u64>),
    Update(ShardEventIndexInfo, ShardInfo),
    Append(ShardEventIndexInfo, ShardInfoAppendUpdate),
    Split(ShardEventIndexInfo, ShardInfo, Vec<ShardInfo>),
    Merge(ShardEventIndexInfo, Vec<ShardInfo>, ShardInfo),
    Stop,
}

pub struct ShardEventPublishMgr {
    tx: tokio::sync::broadcast::Sender<ShardEvent>,
    rx: tokio::sync::broadcast::Receiver<ShardEvent>,
}

impl Default for ShardEventPublishMgr {
    fn default() -> Self {
        let (tx, rx) = tokio::sync::broadcast::channel(1024);
        Self { tx, rx }
    }
}

impl ShardEventPublishMgr {
    // Note: The events before will not be resend!
    #[must_use]
    pub fn resubscribe(
        &self,
        s: impl ShardEventSubscriberTrait + Send + 'static,
        name: String,
    ) -> ShardEventSubscriber {
        let rx = self.rx.resubscribe();
        ShardEventSubscriber::new(s, rx, name)
    }

    pub fn on_shard_append(&self, index: ShardEventIndexInfo, append: ShardInfoAppendUpdate) {
        let e = ShardEvent::Append(index, append);
        if let Err(e) = self.tx.send(e) {
            SHARD_EVENT_COUNTER_VEC
                .with_label_values(&["pubsub_failed"])
                .inc();
            warn!("pubsub err, on_shard_append, e={:?}", e);
        }
    }

    pub fn on_shard_update(&self, index: ShardEventIndexInfo, shard: ShardInfo) {
        let e = ShardEvent::Update(index, shard);
        if let Err(e) = self.tx.send(e) {
            SHARD_EVENT_COUNTER_VEC
                .with_label_values(&["pubsub_failed"])
                .inc();
            warn!("pubsub err, on_shard_update, e={:?}", e);
        }
    }

    pub fn on_shard_split(&self, index: ShardEventIndexInfo, x: ShardInfo, y: Vec<ShardInfo>) {
        let e = ShardEvent::Split(index, x, y);
        if let Err(e) = self.tx.send(e) {
            SHARD_EVENT_COUNTER_VEC
                .with_label_values(&["pubsub_failed"])
                .inc();
            warn!("pubsub err, on_shard_split, e={:?}", e);
        }
    }

    pub fn on_shard_merge(&self, index: ShardEventIndexInfo, x: Vec<ShardInfo>, y: ShardInfo) {
        let e = ShardEvent::Merge(index, x, y);
        if let Err(e) = self.tx.send(e) {
            SHARD_EVENT_COUNTER_VEC
                .with_label_values(&["pubsub_failed"])
                .inc();
            warn!("pubsub err, on_shard_merge, e={:?}", e);
        }
    }

    pub fn on_shard_add(&self, index: ShardEventIndexInfo, x: ShardInfo) {
        let e = ShardEvent::Add(index, x);
        if let Err(e) = self.tx.send(e) {
            SHARD_EVENT_COUNTER_VEC
                .with_label_values(&["pubsub_failed"])
                .inc();
            warn!("pubsub err, on_shard_add, e={:?}", e);
        }
    }

    pub fn on_shard_remove(&self, index: ShardEventIndexInfo, x: Vec<u64>) {
        let e = ShardEvent::Remove(index, x);
        if let Err(e) = self.tx.send(e) {
            SHARD_EVENT_COUNTER_VEC
                .with_label_values(&["pubsub_failed"])
                .inc();
            warn!("pubsub err, on_shard_remove, e={:?}", e);
        }
    }

    pub fn on_full_sync(&self, x: Vec<ShardEventIndexInfo>, y: Vec<ShardInfo>) {
        let e = ShardEvent::FullSync(x, y);
        if let Err(e) = self.tx.send(e) {
            SHARD_EVENT_COUNTER_VEC
                .with_label_values(&["pubsub_failed"])
                .inc();
            warn!("pubsub err, on_full_sync, e={:?}", e);
        }
    }
}

impl Drop for ShardEventPublishMgr {
    fn drop(&mut self) {
        info!("ShardEventSubscriber: dropped");
        let _ = self.tx.send(ShardEvent::Stop);
    }
}

pub struct ShardEventSubscriber {
    _event_listen_runner: BgRunner,
}

impl ShardEventSubscriber {
    fn new(
        mut s: impl ShardEventSubscriberTrait + Send + 'static,
        mut rx: tokio::sync::broadcast::Receiver<ShardEvent>,
        name: String,
    ) -> Self {
        let event_listen_runner = BgRunner::spawn(move |mut rf| async move {
            loop {
                tokio::select! {
                    _ = &mut rf => {
                        info!("ShardEventSubscriber[{}]: recv signal, quit", name);
                        return;
                    },
                    result = rx.recv() => {
                        match result {
                            Ok(x) => {
                                match x {
                                    ShardEvent::Stop => {
                                        info!("ShardEventSubscriber[{}]: recv stop, quit", name);
                                        return;
                                    }
                                    _ => {
                                        s.on_event(x).await;
                                    }
                                }
                            }
                            Err(e) => {
                                info!("ShardEventSubscriber[{}]: recv error={:?}, quit", name, e);
                                return;
                            }
                        }
                    }
                };
            }
        });
        Self {
            _event_listen_runner: event_listen_runner,
        }
    }
}