use crate::{
    get_account_by_id_lite, get_region_by_id, get_regions, insert_account, insert_account_idents,
    insert_region, insert_region_idents,
};
pub use bb8_redis::bb8::{Pool, PooledConnection};
pub use bb8_redis::RedisConnectionManager;
use cal_core::device::device::DeviceType;
use cal_core::{AccountLite, FlowState, RedisEvent, Region};
use moka::sync::Cache;
use redis::aio::MultiplexedConnection;
use redis::{AsyncCommands, PushKind, Value};
use std::collections::HashMap;
use std::env;
use std::time::Duration;
use tokio::sync::mpsc::unbounded_channel;

#[derive(Debug)]
pub struct CacheError {
    pub msg: String,
}

impl CacheError {
    pub fn new_string(s: String) -> CacheError {
        CacheError { msg: s }
    }
}

#[derive(Clone)]
pub struct CallableCache {
    pub local_cache: LocalCache,
    pub remote_cache: RedisCache,
}

#[derive(Clone)]
pub struct LocalCache {
    pub flow_state: Cache<String, FlowState>,
    pub regions: Cache<String, Region>,
    pub region_idents: Cache<String, String>,
    pub accounts: Cache<String, AccountLite>,
    pub account_idents: Cache<String, String>,
}

#[derive(Clone)]
pub struct RedisCache {
    pub connection: MultiplexedConnection,
}

impl RedisCache {
    pub async fn insert_region(self, value: Region) -> RedisCache {
        insert_region(self.connection.clone(), value).await;
        self
    }

    pub async fn insert_region_idents(self, items: Vec<(String, String)>) -> RedisCache {
        insert_region_idents(self.connection.clone(), items).await;
        self
    }

    pub async fn insert_account(self, value: AccountLite) -> RedisCache {
        insert_account(self.connection.clone(), value).await;
        self
    }

    pub async fn insert_account_idents(self, items: Vec<(String, String)>) -> RedisCache {
        insert_account_idents(self.connection.clone(), items).await;
        self
    }
}

impl CallableCache {
    pub async fn new() -> CallableCache {
        let (remote_cache, local_cache) = create_pool().await;
        CallableCache {
            local_cache,
            remote_cache,
        }
    }

    pub async fn get_region_by_id(self, id: &str) -> Option<Region> {
        match self.local_cache.region_idents.get(id) {
            Some(r) => {
                let region: Option<Region> = self.local_cache.regions.get(r.as_str());
                match region {
                    Some(region) => Some(region),
                    None => {
                        let region = get_region_by_id(
                            self.clone().remote_cache.connection.clone(),
                            id.to_string(),
                        )
                        .await?;
                        self.build_local_region_cache(region.clone());
                        Some(region)
                    }
                }
            }
            None => {
                let region =
                    get_region_by_id(self.remote_cache.connection.clone(), id.to_string()).await?;
                self.build_local_region_cache(region.clone());
                Some(region)
            }
        }
    }

    pub async fn get_account_by_id(self, id: &str) -> Option<AccountLite> {
        match self.local_cache.account_idents.get(id) {
            Some(r) => {
                let region: Option<AccountLite> = self.local_cache.accounts.get(r.as_str());
                match region {
                    Some(account) => Some(account),
                    None => {
                        let account = get_account_by_id_lite(
                            self.remote_cache.connection.clone(),
                            id.to_string(),
                        )
                        .await?;
                        self.build_local_account_cache(account.clone());
                        Some(account)
                    }
                }
            }
            None => {
                let account =
                    get_account_by_id_lite(self.remote_cache.connection.clone(), id.to_string())
                        .await?;
                self.build_local_account_cache(account.clone());
                Some(account)
            }
        }
    }

    pub fn build_local_region_cache(self, reg: Region) {
        self.local_cache.regions.insert(reg.id.clone(), reg.clone());
        self.local_cache
            .region_idents
            .insert(reg.id.clone(), reg.id.clone());
        for vs in &reg.voice_servers {
            self.local_cache
                .region_idents
                .insert(vs.id.clone(), reg.id.clone());
            self.local_cache
                .region_idents
                .insert(vs.public_ip.clone(), reg.id.clone());
            self.local_cache
                .region_idents
                .insert(vs.private_ip.clone(), reg.id.clone());
        }
    }

    pub async fn build_remote_region_cache(self, reg: Region) {
        let mut items: Vec<(String, String)> = Vec::new();
        items.push((reg.id.clone(), reg.id.clone()));
        for vs in &reg.voice_servers {
            items.push((vs.id.clone(), reg.id.clone()));
            items.push((vs.public_ip.clone(), reg.id.clone()));
            items.push((vs.private_ip.clone(), reg.id.clone()));
        }
        self.remote_cache
            .clone()
            .insert_region_idents(items)
            .await
            .insert_region(reg.clone())
            .await;
    }

    pub async fn build_local_account_cache(self, acc: AccountLite) {
        let mut items: Vec<(String, String)> = Vec::new();
        items.push((acc.get_id_string().clone(), acc.get_id_string().clone()));

        items.push((
            format!("{}.connect.callable.io", acc.domain),
            acc.get_id_string().clone(),
        ));

        if !acc.teams.domain.is_empty() {
            items.push((
                format!("{}.connect.callable.io", acc.teams.domain),
                acc.get_id_string().clone(),
            ));
        }

        for ddi in &acc.ddis {
            items.push((ddi.name.clone(), acc.get_id_string().clone()));
        }

        for device in &acc.devices {
            if device.is_device_type(DeviceType::Client) {
                if let Some(client) = &device.client {
                    items.push((
                        format!("{}@{}.connect.callable.io", client.username, acc.domain),
                        acc.get_id_string().clone(),
                    ));
                }
            }
        }

        self.remote_cache
            .clone()
            .insert_account_idents(items)
            .await
            .insert_account(acc.clone())
            .await;
    }

    pub async fn get_regions(self) -> HashMap<String, Region> {
        get_regions(self.remote_cache.connection).await
    }

    pub async fn get_str(self, key: &str) -> Option<String> {
        get_str(self.remote_cache.connection, key).await
    }

    pub async fn get_hash(self, key: &str, field: &str) -> Option<String> {
        get_hash(self.remote_cache.connection, key, field).await
    }
}

fn get_redis_host() -> String {
    let name = "CAL_VALKEY_HOST";
    match env::var(name) {
        Ok(v) => v,
        Err(e) => panic!("${} is not set ({})", name, e),
    }
}

pub async fn create_pool() -> (RedisCache, LocalCache) {
    let client = redis::Client::open(get_redis_host()).unwrap();

    let (tx, mut rx) = unbounded_channel();
    let config = redis::AsyncConnectionConfig::new().set_push_sender(tx);
    let mut con = client
        .get_multiplexed_async_connection_with_config(&config)
        .await
        .unwrap();

    con.subscribe(&["cb:events"]).await.unwrap();

    let flow_state: Cache<String, FlowState> = Cache::builder()
        .max_capacity(50 * 1024 * 1024)
        .time_to_live(Duration::from_secs(60 * 60 * 4))
        .build();

    let region_idents: Cache<String, String> = Cache::builder()
        .max_capacity(5 * 1024 * 1024)
        .time_to_live(Duration::from_secs(60))
        .build();

    let regions: Cache<String, Region> = Cache::builder()
        .max_capacity(5 * 1024 * 1024)
        .time_to_live(Duration::from_secs(60))
        .build();

    let account_idents: Cache<String, String> = Cache::builder()
        .max_capacity(5 * 1024 * 1024)
        .time_to_live(Duration::from_secs(60))
        .build();

    let accounts: Cache<String, AccountLite> = Cache::builder()
        .max_capacity(50 * 1024 * 1024)
        .time_to_live(Duration::from_secs(60))
        .build();

    let local_cache = LocalCache {
        flow_state,
        regions: regions.clone(),
        region_idents: region_idents.clone(),
        accounts: accounts.clone(),
        account_idents: account_idents.clone(),
    };

    tokio::spawn(async move {
        loop {
            let push_info = rx.recv().await.unwrap();
            if push_info.kind == PushKind::Message {
                let message = push_info.data.get(1).unwrap().clone();
                if let Value::BulkString(bytes) = message {
                    let str = String::from_utf8(bytes).unwrap();
                    let event: RedisEvent = serde_json::from_str(str.as_str()).unwrap();
                    match event {
                        RedisEvent::AccountUpdate(update) => {
                            accounts.insert(update.payload.get_id_string(), update.payload.clone())
                        }
                        RedisEvent::AccountDelete(delete) => accounts.invalidate(&delete.id),
                        RedisEvent::AccountCreate(create) => {
                            accounts.insert(create.payload.get_id_string(), create.payload.clone())
                        }
                        RedisEvent::AccountSync => {
                            accounts.invalidate_all();
                            account_idents.invalidate_all();
                        }
                        RedisEvent::RegionUpdate(update) => {
                            regions.insert(update.payload.id.clone(), update.payload.clone())
                        }
                        RedisEvent::RegionDelete(delete) => regions.invalidate(&delete.id),
                        RedisEvent::RegionCreate(create) => {
                            regions.insert(create.payload.id.clone(), create.payload.clone())
                        }
                        RedisEvent::RegionSync => {
                            regions.invalidate_all();
                            region_idents.invalidate_all();
                        }
                    }
                }
            }
        }
    });

    (RedisCache { connection: con }, local_cache)
}

pub async fn get_str(mut con: MultiplexedConnection, key: &str) -> Option<String> {
    con.get(key).await.unwrap_or_else(|e| {
        println!("Cache Error {:?}", e);
        None
    })
}

pub async fn get_hash(mut con: MultiplexedConnection, key: &str, field: &str) -> Option<String> {
    let _field = format!("\"{}\"", field);
    con.hget(key, _field).await.unwrap_or_else(|e| {
        println!("Cache Error {:?}", e);
        None
    })
}

pub async fn get_hash_all_values(
    mut con: MultiplexedConnection,
    key: &str,
) -> HashMap<String, String> {
    con.hvals(key).await.unwrap_or_else(|e| {
        println!("Cache Error {:?}", e);
        HashMap::new()
    })
}
