use rocksdb::WriteBatch;
use rocksdb::DB;

use bloock_storage::config::RocksDBConfig;
use bloock_storage::kv::kv_rocks::RocksDB;
use bloock_storage::kv::KeyValue;
use rand::Rng;
use std::any::type_name;
use std::fs::read_dir;
use std::path::{Path, PathBuf};
use std::{fs, io};

fn init_rocksdb(name: &str, path: Option<String>) -> RocksDB {
    let config: RocksDBConfig = RocksDBConfig {
        base_path: String::from(name),
        create_if_missing: true,
    };
    <RocksDB as KeyValue>::open(&config, path).expect("Something went wrong at init_rocksdb.")
}

fn check_empty_path(path: &str) -> bool {
    match read_dir(path) {
        Ok(contents) => {
            let len = contents.collect::<Vec<_>>().len();
            return len == 0;
        }
        Err(_) => false,
    }
}

fn dir_size(path: impl Into<PathBuf>) -> io::Result<u64> {
    fn dir_size(mut dir: fs::ReadDir) -> io::Result<u64> {
        dir.try_fold(0, |acc, file| {
            let file = file?;
            let size = match file.metadata()? {
                data if data.is_dir() => dir_size(fs::read_dir(file.path())?)?,
                data => data.len(),
            };
            Ok(acc + size)
        })
    }

    dir_size(fs::read_dir(path.into())?)
}

fn type_of<T>(_: T) -> &'static str {
    type_name::<T>()
}

#[test]
fn test_open_with_subdirectory() {
    {
        let mut rocksdb = init_rocksdb("open_with_subdirectory", Some(String::from("hola")));
        rocksdb.put(&[1; 32], &[1; 32]).unwrap();
        assert_eq!(
            rocksdb.get(&[1; 32]).unwrap(),
            Some(vec![1; 32]),
            "The value should be none"
        );
    }
    let _ = DB::destroy(&rocksdb::Options::default(), "get_exist_kv_rocks");
}

#[test]
fn test_get_exist() {
    {
        let mut rocksdb = init_rocksdb("get_exist_kv_rocks", None);
        rocksdb.put(&[1; 32], &[1; 32]).unwrap();
        assert_eq!(
            rocksdb.get(&[1; 32]).unwrap(),
            Some(vec![1; 32]),
            "The value should be none"
        );
    }
    let _ = DB::destroy(&rocksdb::Options::default(), "get_exist_kv_rocks");
}

#[test]
fn test_get_not_exist() {
    {
        let rocksdb = init_rocksdb("get_not_exist_kv_rocks", None);
        assert_eq!(
            rocksdb.get(&[1; 32]).unwrap().is_none(),
            true,
            "The value should be none"
        );
    }
    let _ = DB::destroy(&rocksdb::Options::default(), "get_not_exist_kv_rocks");
}

#[test]
fn test_get_error() {
    /* TO DO: fill, don't know how to make rocksdb return error" */
}

#[test]
fn test_put() {
    {
        let mut rocksdb = init_rocksdb("put_kv_rocks", None);
        rocksdb.put(&[1; 32], &[0; 32]).unwrap();
        assert_eq!(
            rocksdb.get(&[1; 32]).unwrap().unwrap(),
            [0; 32],
            "The value read was not written or unable to read"
        );
    }
    let _ = DB::destroy(&rocksdb::Options::default(), "put_kv_rocks");
}

#[test]
fn test_put_error() {
    {
        let opts = rocksdb::Options::default();
        let _ = init_rocksdb("put_error_kv_rocks", None);
        let rocksdb = DB::open_for_read_only(&opts, "put_error_kv_rocks", false).unwrap();
        assert_eq!(
            rocksdb.put(&[0u8; 32], vec![1u8; 32]).is_err(),
            true,
            "The write should have failed"
        );
    }
    let _ = DB::destroy(&rocksdb::Options::default(), "put_error_kv_rocks");
}

#[test]
fn test_ini_batch() {
    assert_eq!(
        "rocksdb::write_batch::WriteBatchWithTransaction<false>",
        type_of(RocksDB::ini_batch().unwrap()),
        "ini_batch does not work correctly"
    )
}

#[test]
fn test_put_batch() {
    let mut b = WriteBatch::default();
    assert_eq!(
        true,
        RocksDB::put_batch(&mut b, &[2u8], &[3u8]).is_ok(),
        "put batch not working"
    )
}

#[test]
fn test_end_batch() {
    {
        let mut db = init_rocksdb("end_batch_kv_rocks", None);
        let b = WriteBatch::default();
        assert_eq!(true, db.end_batch(b).is_ok(), "put batch not working");
    }
    let _ = DB::destroy(&rocksdb::Options::default(), "end_batch_kv_rocks");
}

#[test]
fn test_move_kv_directory() {
    let path = String::from("./tmp_kv_rocks/");
    let config: RocksDBConfig = RocksDBConfig {
        base_path: path.clone(),
        create_if_missing: true,
    };
    let mut kv =
        <RocksDB as KeyValue>::open(&config, None).expect("Something went wrong at init_rocksdb.");
    kv.put(&[0x45u8; 32], &vec![0; 32]).unwrap();
    assert_eq!(
        kv.get(&[0x45u8; 32]).unwrap().unwrap(),
        [0; 32],
        "The value read was not written or unable to read"
    );
    let path_2 = String::from("./tmp_kv_rocks_moved/");
    let res = kv.move_to(path_2.clone());
    //println!("{:?}", res.err().unwrap());
    assert!(res.is_ok(), "error moving rocks");
    assert!(
        Path::new("tmp_kv_rocks_moved").exists(),
        "blob folder was not moved"
    );
    assert!(check_empty_path(&path.clone()));
    assert!(std::fs::remove_dir_all(path).is_ok());
    assert!(std::fs::remove_dir_all(path_2).is_ok());
    //assert!(false);
}

#[test]
fn test_move_kv_directory_already_existing() {
    let path = String::from("./tmp_kv_rocks_1/");
    let path_2 = String::from("./tmp_kv_rocks_2/");
    let config: RocksDBConfig = RocksDBConfig {
        base_path: path.clone(),
        create_if_missing: true,
    };
    {
        let mut kv = <RocksDB as KeyValue>::open(&config, None)
            .expect("Something went wrong at init_rocksdb.");
        kv.put(&[0x45u8; 32], &vec![0; 32]).unwrap();
        let config_2: RocksDBConfig = RocksDBConfig {
            base_path: path_2.clone(),
            create_if_missing: true,
        };
        let mut kv_2 = <RocksDB as KeyValue>::open(&config_2, None)
            .expect("Something went wrong at init_rocksdb.");
        kv.put(&[0x45u8; 32], &vec![0; 32]).unwrap();
        kv_2.put(&[0x45u8; 32], &vec![1; 32]).unwrap();
        assert!(kv_2.move_to(path.clone()).is_ok(), "Error moving");
    }
    let kv =
        <RocksDB as KeyValue>::open(&config, None).expect("Something went wrong at init_rocksdb.");
    assert_eq!(
        kv.get(&[0x45u8; 32]).unwrap().unwrap(),
        [1; 32],
        "The value read was not written or unable to read"
    );
    assert!(check_empty_path(&path_2.clone()));
    assert!(std::fs::remove_dir_all(path).is_ok());
    assert!(std::fs::remove_dir_all(path_2).is_ok())
}

#[test]
fn test_delete_no_subpath() {
    {
        let mut rocksdb = init_rocksdb("delete_kv_rocks", None);
        rocksdb.put(&[1; 32], &[1; 32]).unwrap();
        assert_eq!(
            rocksdb.get(&[1; 32]).unwrap(),
            Some(vec![1; 32]),
            "Value not matching"
        );
    }
    assert!(
        <RocksDB as KeyValue>::destroy(
            &RocksDBConfig {
                base_path: String::from("delete_kv_rocks"),
                create_if_missing: false,
            },
            None
        )
        .is_ok(),
        "Should be okay"
    );
}

#[test]
fn test_delete_subpath() {
    {
        let mut rocksdb = init_rocksdb("delete_kv_rocks", Some("hello".to_string()));
        rocksdb.put(&[1; 32], &[1; 32]).unwrap();
        assert_eq!(
            rocksdb.get(&[1; 32]).unwrap(),
            Some(vec![1; 32]),
            "Value not matching"
        );
    }
    assert!(
        <RocksDB as KeyValue>::destroy(
            &RocksDBConfig {
                base_path: String::from("delete_kv_rocks"),
                create_if_missing: false,
            },
            Some("hello".to_string())
        )
        .is_ok(),
        "Should be okay"
    );
}

#[test]
fn test_compact() {
    {
        let mut rocksdb = init_rocksdb("compact", None);
        for _ in 0..5000 {
            rocksdb
                .put(
                    &rand::thread_rng().gen::<[u8; 32]>(),
                    &rand::thread_rng().gen::<[u8; 32]>(),
                )
                .unwrap();
        }
        let folder_size_precompact = dir_size("compact").unwrap();
        rocksdb.compact().unwrap();
        let folder_size_postcompact = dir_size("compact").unwrap();
        assert_eq!(
            (folder_size_precompact > folder_size_postcompact),
            true,
            "Compaction did not reduce the space used in disk"
        );
    }
    assert!(
        <RocksDB as KeyValue>::destroy(
            &RocksDBConfig {
                base_path: String::from("compact"),
                create_if_missing: false,
            },
            None
        )
        .is_ok(),
        "Should be okay"
    );
}