pub use super::Options;
use super::serde;
use crate::bson::DateTime;
use crate::storage::associations::{BelongsTo, HasMany};
use crate::storage::session_store::session_store::FindOptions;
use crate::storage::{Error, session_store::SessionStore};
use async_trait::async_trait;
use futures::stream::StreamExt;
use inflection_rs::inflection;
use pbbson::Model;
use redis::{AsyncIter, AsyncTypedCommands, ScanOptions};
use std::collections::HashMap;
use std::fmt::Debug;
use std::iter::zip;
use std::time::{Duration, SystemTime};
use url::Url;

#[allow(unused)]
pub struct RedisSessionStore<B: Clone + Send + Sync + ToString> {
    redis_client: redis::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 mut redis_client = redis::Client::open(url.as_str())?;

        // Validate the connection
        let mut conn = redis_client.get_multiplexed_async_connection().await?;
        conn.ping().await?;

        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 mut conn = self.redis_client.get_multiplexed_async_connection().await?;
        let fields: HashMap<String, String> = conn.hgetall(&key).await?;
        let id = match key.rfind(":") {
            None => key.to_string(),
            Some(index) => key[index + 1..].to_string(),
        };
        let model = serde::transform_from_redis(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 values = serde::transform_to_redis(&model)?;
        let key = to_key(bucket, &id);
        let mut conn = self.redis_client.get_multiplexed_async_connection().await?;
        let mut pipe = redis::pipe();
        let mut pipe = pipe.hset_multiple(&key, &values);
        if let Some(expires) = maybe_expires {
            pipe = pipe.expire(key, expires.as_secs() as i64);
        }
        let _: () = pipe.query_async(&mut conn).await?;
        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 mut conn = self.redis_client.get_multiplexed_async_connection().await?;
        let exists: bool = conn.exists(&key).await?;
        if !exists {
            return Err(Error::not_found("No such record"));
        }
        conn.del(key).await?;
        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, options: FindOptions) -> Result<Vec<Model>, Error> {
        // Collect keys
        let key_prefix = to_key(bucket, "");
        let pattern = format!("{key_prefix}*");
        let mut conn = self.redis_client.get_multiplexed_async_connection().await?;
        let scan_opts = ScanOptions::default().with_pattern(pattern).with_count(100);
        let scanner: AsyncIter<String> = conn.scan_options(scan_opts).await?;
        let mut keys: Vec<_> = scanner.map(Result::unwrap).collect().await;

        // Filter keys
        if let Ok(id_filter) = options.filter.get_str("id") {
            keys.retain(|key| {
                let id = match key.rfind(":") {
                    None => key,
                    Some(index) => &key[index + 1..].to_string(),
                };
                id == id_filter
            });
        }

        // Fetch models
        let mut models: Vec<Model> = vec![];
        if !keys.is_empty() {
            let mut pipeline = redis::pipe();
            let mut pipeline = &mut pipeline;
            for key in keys.iter() {
                pipeline = pipeline.hgetall(key);
            }
            let results: Vec<HashMap<String, String>> = pipeline.query_async(&mut conn).await?;
            for (key, fields) in zip(keys, results) {
                let id = match key.rfind(":") {
                    None => key,
                    Some(index) => key[index + 1..].to_string(),
                };
                let model = serde::transform_from_redis(fields, &id)?;
                models.push(model);
            }
        }

        // Apply filter
        models.retain(|model| {
            for (filter_key, filter_value) in options.filter.iter() {
                if filter_key == "id" {
                    continue;
                }
                if model.get(filter_key) != Some(filter_value) {
                    return false;
                }
            }
            true
        });

        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 mut conn = self.redis_client.get_multiplexed_async_connection().await?;
        let exists: bool = conn.exists(key).await?;
        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}")
}
