use crate::models;
use crate::storage::Error;
use pbbson::{Model, bson};
use serde_json::{Map, Value};
use std::collections::{HashMap, VecDeque};

const EMPTY_FIELDS_PSEUDOFIELD: &str = "_exists";

pub fn transform_to_redis(model: &Model) -> Result<Vec<(String, String)>, Error> {
    let hashmap = models::to_json(model, false)?;
    let mut map = serde_json::Map::new();
    if hashmap.is_empty() {
        map.insert(EMPTY_FIELDS_PSEUDOFIELD.to_string(), serde_json::Value::Bool(true));
    } else {
        for (k, v) in hashmap.into_iter() {
            map.insert(k, v);
        }
    }
    let mut values = Vec::new();
    add_values(&mut values, &map, "");
    Ok(values)
}

fn add_values(values: &mut Vec<(String, String)>, map: &Map<String, Value>, prefix: &str) {
    for (k, v) in map.iter() {
        let key = {
            if prefix.is_empty() {
                k.clone()
            } else {
                format!("{prefix}.{k}")
            }
        };
        match v {
            serde_json::Value::Array(a) => {
                if !a.is_empty() {
                    values.push((key.clone(), v.to_string()));
                    values.push((format!("{key}$type"), "array".to_string()));
                }
            }
            serde_json::Value::Bool(b) => {
                values.push((key.clone(), b.to_string()));
                values.push((format!("{key}$type"), "bool".to_string()));
            }
            serde_json::Value::Number(n) => {
                values.push((key.clone(), n.to_string()));
                values.push((format!("{key}$type"), "number".to_string()));
            }
            serde_json::Value::Object(object) => add_values(values, object, &key),
            serde_json::Value::String(s) => values.push((key, s.clone())),
            _ => values.push((key, v.to_string())),
        }
    }
}

pub fn transform_from_redis(fields: HashMap<String, String>, id: &str) -> Result<Model, Error> {
    if fields.is_empty() {
        return Err(Error::not_found("No such record"));
    }

    // Collect the type hints
    let mut type_hints = HashMap::new();
    for (k, v) in fields.iter() {
        if let Some(key) = k.strip_suffix("$type") {
            type_hints.insert(key.to_string(), v.to_string());
        }
    }

    // Deserialize fields
    let mut fields = fields;
    fields.remove(EMPTY_FIELDS_PSEUDOFIELD);
    let mut map: Map<String, serde_json::Value> = Map::new();
    for (k, v) in fields.into_iter() {
        if k.ends_with("$type") {
            continue;
        }
        match type_hints.get(&k) {
            Some(type_hint) => match type_hint.as_str() {
                "bool" => set(&mut map, k, serde_json::Value::Bool(v == "true")),
                "number" => {
                    let f = v.as_str().parse::<f64>().map_err(|e| Error::internal(e.to_string()))?;
                    set(&mut map, k, f.into());
                }
                _ => set(&mut map, k, serde_json::Value::String(v)),
            },
            None => set(&mut map, k, serde_json::Value::String(v)),
        }
    }
    let doc = bson::serialize_to_document(&map).map_err(|e| Error::internal(e.to_string()))?;
    let mut model = Model::from(doc);
    model.insert("id", id);
    Ok(model)
}

fn set(map: &mut Map<String, serde_json::Value>, k: String, v: serde_json::Value) {
    let mut k_segments = k.split('.').collect::<VecDeque<_>>();
    if let Some(prefix) = k_segments.pop_front()
        && !k_segments.is_empty()
    {
        if let Some(sub_v) = map.get_mut(prefix) {
            let k = Into::<Vec<_>>::into(k_segments).join(".");
            let maybe_object = sub_v.as_object_mut();
            if let Some(object) = maybe_object {
                set(object, k, v);
            }
        } else {
            let k = Into::<Vec<_>>::into(k_segments).join(".");
            let mut sub_map = Map::new();
            set(&mut sub_map, k, v);
            map.insert(prefix.to_string(), serde_json::Value::Object(sub_map));
        }
    } else {
        map.insert(k, v);
    }
}

#[cfg(test)]
mod tests {
    use super::{transform_from_redis, transform_to_redis};
    use pbbson::{
        Model,
        bson::{Bson, doc},
    };
    use std::collections::HashMap;

    #[test]
    fn can_serialize_flat_object() {
        let model = Model::from(doc! {"id": "123", "name": "abc"});
        let values = transform_to_redis(&model).unwrap();
        assert_eq!(values.len(), 1);
        assert_eq!(values.first(), Some(("name".to_string(), "abc".to_string())).as_ref());
    }

    #[test]
    fn can_deserialize_flat_object() {
        let fields = HashMap::from([("name".to_string(), "abc".to_string())]);
        let model = transform_from_redis(fields, "123").unwrap();
        assert_eq!(model.len(), 2);
        assert_eq!(model.id().unwrap(), "123".to_string());
        assert_eq!(model.get("name"), Some(Bson::String("abc".to_string())).as_ref());
    }

    #[test]
    fn can_serialize_nested_object() {
        let model = Model::from(doc! {"id": "123", "name": "abc", "metadata": doc! {"foo": "bar"}});
        let values = transform_to_redis(&model).unwrap();
        assert_eq!(values.len(), 2);
        for (k, v) in values.iter() {
            if k == "name" {
                assert_eq!(v, "abc");
            } else {
                assert_eq!(k, "metadata.foo");
                assert_eq!(v, "bar");
            }
        }
    }

    #[test]
    fn can_serialize_nested_object_2() {
        let model = Model::from(doc! {
            "name": "abc",
            "metadata": doc! {
                "vendor": doc! {
                    "foo": "bar",
                }
            }
        });
        let values = transform_to_redis(&model).unwrap();
        assert_eq!(values.len(), 2);
        for (k, v) in values.iter() {
            if k == "name" {
                assert_eq!(v, "abc");
            } else {
                assert_eq!(k, "metadata.vendor.foo");
                assert_eq!(v, "bar");
            }
        }
    }

    #[test]
    fn can_deserialize_nested_object() {
        let fields = HashMap::from([
            ("name".to_string(), "abc".to_string()),
            ("metadata.foo".to_string(), "bar".to_string()),
        ]);
        let model = transform_from_redis(fields, "123").unwrap();
        assert_eq!(model.len(), 3);
        assert_eq!(model.id().unwrap(), "123".to_string());
        assert_eq!(model.get("name"), Some(Bson::String("abc".to_string())).as_ref());
        assert_eq!(
            model.get("metadata"),
            Some(Bson::Document(doc! {"foo": "bar"})).as_ref()
        );
    }

    #[test]
    fn can_deserialize_nested_object_2() {
        let fields = HashMap::from([
            ("name".to_string(), "abc".to_string()),
            ("metadata.vendor.first".to_string(), "John".to_string()),
            ("metadata.vendor.last".to_string(), "Doe".to_string()),
            ("metadata.vendor.age".to_string(), "123".to_string()),
            ("metadata.vendor.age$type".to_string(), "number".to_string()),
        ]);
        let model = transform_from_redis(fields, "123").unwrap();
        assert_eq!(model.len(), 3);
        assert_eq!(model.id().unwrap(), "123".to_string());
        assert_eq!(model.get("name"), Some(Bson::String("abc".to_string())).as_ref());
        assert_eq!(
            model.get("metadata"),
            Some(Bson::Document(
                doc! {"vendor": doc! {"first": "John", "last": "Doe", "age": 123_f32}}
            ))
            .as_ref()
        );
    }
}
