use crate::utils::{db_put_simple, tmpdir};
use leveldb::database::filter_policy::FilterPolicy;
use leveldb::database::kv::KV;
use leveldb::database::slice::Slice;
use leveldb::options::{Options, ReadOptions};

/// A simple test filter policy that always returns true for key matching
/// and creates empty filters
struct AlwaysMatchFilterPolicy;

impl FilterPolicy for AlwaysMatchFilterPolicy {
    fn create_filter(&self, _keys: &[&[u8]]) -> Vec<u8> {
        vec![0]
    }

    fn key_may_match(&self, _key: &[u8], _filter: &[u8]) -> bool {
        true
    }

    fn name(&self) -> &'static str {
        "AlwaysMatchFilterPolicy"
    }
}

/// A test filter policy that never matches any keys
struct NeverMatchFilterPolicy;

impl FilterPolicy for NeverMatchFilterPolicy {
    fn create_filter(&self, _keys: &[&[u8]]) -> Vec<u8> {
        vec![1]
    }

    fn key_may_match(&self, _key: &[u8], _filter: &[u8]) -> bool {
        false
    }

    fn name(&self) -> &'static str {
        "NeverMatchFilterPolicy"
    }
}

/// Helper function to test filter policy persistence
fn test_filter_policy<F>(test_name: &str, setup_filter: F, expected_results: &[Option<Vec<u8>>])
where
    F: Fn(&mut Options),
{
    let tmp_dir = tmpdir(test_name);
    let path = tmp_dir.path();

    // First, create a database with the specified filter policy
    {
        let mut opts = Options::new();
        opts.create_if_missing = true;
        setup_filter(&mut opts);
        let db = leveldb::database::Database::open(path, opts).unwrap();

        // Write some data using utility function
        db_put_simple(&db, b"key1", b"value1");
        db_put_simple(&db, b"key2", b"value2");
        db_put_simple(&db, b"key3", b"value3");
    }

    // Reopen the database with the same filter policy
    {
        let mut opts = Options::new();
        opts.create_if_missing = false;
        setup_filter(&mut opts);
        let db = leveldb::database::Database::open(path, opts).unwrap();

        // Read the data back and verify expected results
        let keys = ["key1", "key2", "key3"];
        for (key, expected) in keys.iter().zip(expected_results) {
            let result = db
                .get(ReadOptions::new(), Slice::from(key.as_bytes()))
                .unwrap();
            let expected_slice = expected.as_ref().map(|v| Slice::from(v.as_ref()));
            assert_eq!(
                result, expected_slice,
                "Key {} should return {:?}",
                key, expected
            );
        }
    }
}

#[test]
fn test_bloom_filter_policy_should_find_persisted_data() {
    test_filter_policy(
        "bloom_filter_test",
        |opts| opts.set_bloom_filter(10),
        &[
            Some(b"value1".to_vec()),
            Some(b"value2".to_vec()),
            Some(b"value3".to_vec()),
        ],
    );
}

#[test]
fn test_always_match_filter_policy_should_find_persisted_data() {
    test_filter_policy(
        "always_match_test",
        |opts| opts.set_filter_policy(AlwaysMatchFilterPolicy),
        &[
            Some(b"value1".to_vec()),
            Some(b"value2".to_vec()),
            Some(b"value3".to_vec()),
        ],
    );
}

#[test]
fn test_never_match_filter_policy_should_not_find_persisted_data() {
    // NeverMatchFilterPolicy would cause iterator won't try
    // to look into sst files because of incorrect hint
    test_filter_policy(
        "never_match_test",
        |opts| opts.set_filter_policy(NeverMatchFilterPolicy),
        &[None, None, None],
    );
}
