use crate::bson::DateTime;
use crate::storage::associations::{BelongsTo, HasMany};
use crate::storage::{Error, models, session_store::SessionStore};
use async_trait::async_trait;
use fred::prelude::{ClientLike, HashesInterface, KeysInterface, TransactionInterface, Value};
use fred::types::{InfoKind, scan::Scanner};
use futures_util::TryStreamExt;
use inflection_rs::inflection;
use log::debug;
use pbbson::{Model, bson};
use std::collections::HashMap;
use std::fmt::Debug;
use std::iter::zip;
use std::time::{Duration, SystemTime};
use url::Url;

pub use super::Options;

const EMPTY_FIELDS_PSEUDOFIELD: &str = "_exists";

#[allow(unused)]
pub struct RedisSessionStore<B: Clone + Send + Sync + ToString> {
    redis_client: fred::clients::Client,

    #[allow(unused)]
    belongs_tos_by_bucket: HashMap<String, Vec<BelongsTo<B>>>,

    #[allow(unused)]
    has_manys_by_bucket: HashMap<String, Vec<HasMany<B>>>,
}

impl<B: Clone + Debug + Send + Sync + ToString> RedisSessionStore<B> {
    #[allow(unused)]
    pub async fn new(url: &Url, options: Options<B>) -> Result<Self, Error> {
        let redis_client = {
            let config = fred::prelude::Config::from_url(url.as_str()).map_err(|e| Error::internal(e.to_string()))?;
            debug!(config:?; "Using Redis config");
            let client = fred::prelude::Builder::from_config(config)
                .build()
                .map_err(|e| Error::internal(e.to_string()))?;
            client.init().await.map_err(|e| Error::internal(e.to_string()))?;
            client
        };

        // Validate the connection
        redis_client
            .info::<Value>(Some(InfoKind::Server))
            .await
            .map_err(|e| Error::internal(e.to_string()))?;

        Ok(Self {
            redis_client,
            belongs_tos_by_bucket: options.belongs_tos_by_bucket,
            has_manys_by_bucket: options.has_manys_by_bucket,
        })
    }

    async fn find_by_key(&self, key: String) -> Result<Model, Error> {
        let fields: HashMap<String, String> = self
            .redis_client
            .hgetall(&key)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        let id = match key.rfind(":") {
            None => key.to_string(),
            Some(index) => key[index + 1..].to_string(),
        };
        let model = to_model(fields, id)?;
        Ok(model)
    }

    async fn store(
        &self,
        bucket: B,
        model: Model,
        maybe_expires: Option<Duration>,
        update: bool,
    ) -> Result<Model, Error> {
        let mut model = model.clone();
        let id = match model.id() {
            Ok(id) => id,
            _ => {
                let id = xid::new().to_string();
                model.insert("id", id.clone());
                id
            }
        };
        let now = DateTime::from_system_time(SystemTime::now());
        if update {
            model.set_datetime("updatedAt", now);
        } else {
            model.set_datetime("createdAt", now);
        }
        let mut map = models::to_json(&model, false)?;
        if map.is_empty() {
            map.insert(EMPTY_FIELDS_PSEUDOFIELD.to_string(), serde_json::Value::Bool(true));
        }
        let mut values = Vec::new();
        for (k, v) in map.iter() {
            let v = match v {
                serde_json::Value::String(s) => s.to_string(),
                _ => v.to_string(),
            };
            values.push((k.clone(), v.to_string()));
        }
        let key = to_key(bucket, &id);
        let tx = self.redis_client.multi();
        let _: Value = tx
            .hset(key.clone(), values)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        if let Some(expires) = maybe_expires {
            let _: Value = tx
                .expire(key, expires.as_secs() as i64, None)
                .await
                .map_err(|e| Error::internal(e.to_string()))?;
        }
        let _: Value = tx.exec(true).await.map_err(|e| Error::internal(e.to_string()))?;
        Ok(model)
    }
}

#[async_trait]
impl<B: Clone + Debug + Send + Sync + ToString> SessionStore<B> for RedisSessionStore<B> {
    async fn create(&self, bucket: B, model: Model, maybe_expires: Option<Duration>) -> Result<Model, Error> {
        self.store(bucket, model, maybe_expires, false).await
    }

    async fn delete(&self, bucket: B, id: &str) -> Result<(), Error> {
        let key = to_key(bucket, id);
        let _: () = self
            .redis_client
            .del(&key)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        Ok(())
    }

    async fn find(&self, bucket: B, id: &str) -> Result<Model, Error> {
        let key = to_key(&bucket, id);
        self.find_by_key(key).await
    }

    async fn find_many(&self, bucket: B, _filter: Model, _ordering: Option<Model>) -> Result<Vec<Model>, Error> {
        // Collect keys
        let key_prefix = to_key(bucket, "");
        let pattern = format!("{key_prefix}*");
        let mut scanner = self.redis_client.scan(pattern, Some(100), None);
        let mut keys = vec![];
        while let Some(mut page) = scanner.try_next().await.map_err(|e| Error::internal(e.to_string()))? {
            if let Some(page) = page.take_results() {
                for key in page.iter() {
                    keys.push(key.clone());
                }
            }
        }

        // Fetch models
        let mut models: Vec<Model> = vec![];
        let pipeline = self.redis_client.pipeline();
        for key in keys.iter() {
            let _: () = pipeline
                .hgetall(key)
                .await
                .map_err(|e| Error::internal(e.to_string()))?;
        }
        let results: Vec<HashMap<String, String>> = pipeline.all().await.map_err(|e| Error::internal(e.to_string()))?;
        for (key, fields) in zip(keys, results) {
            let key = match key.as_str() {
                None => continue,
                Some(key) => key.to_string(),
            };
            let id = match key.rfind(":") {
                None => key,
                Some(index) => key[index + 1..].to_string(),
            };
            let model = to_model(fields, id)?;
            models.push(model);
        }
        Ok(models)
    }

    async fn update(&self, bucket: B, model: Model, maybe_expires: Option<Duration>) -> Result<Model, Error> {
        let key = to_key(&bucket, &model.id()?);
        let exists: bool = self
            .redis_client
            .exists(key)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        if !exists {
            return Err(Error::not_found("No such record"));
        }
        self.store(bucket, model, maybe_expires, true).await
    }
}

fn to_key<B: Debug>(bucket: B, id: &str) -> String {
    let prefix = inflection::singularize(change_case::lower_case_first(&format!("{bucket:?}")));
    format!("{prefix}:{id}")
}

fn to_model(mut fields: HashMap<String, String>, id: String) -> Result<Model, Error> {
    if fields.is_empty() {
        return Err(Error::not_found("No such record"));
    }
    fields.remove(EMPTY_FIELDS_PSEUDOFIELD);
    let doc = bson::serialize_to_document(&fields).map_err(|e| Error::internal(e.to_string()))?;
    let mut model = Model::from(doc);
    model.insert("id", id);
    Ok(model)
}
