//! Shared tests for keyvaluedb functionality, to be executed against actual implementations.

#![deny(clippy::all)]

use keyvaluedb::{DBKey, DBKeyRef, DBKeyValue, DBKeyValueRef, IoStatsKind, KeyValueDB};
use std::io;

/// A test for `KeyValueDB::get`.
pub async fn test_put_and_get<DB: KeyValueDB>(db: DB) -> io::Result<()> {
    let key1 = b"key1";

    let mut transaction = db.transaction();
    transaction.put(0, key1.to_vec(), b"horse");
    db.write(transaction).await.map_err(|e| e.error)?;
    assert_eq!(db.get(0, key1).await?.unwrap(), b"horse");
    Ok(())
}

/// A test for `KeyValueDB::get` and delete with transaction.
pub async fn test_delete_and_get<DB: KeyValueDB>(db: DB) -> io::Result<()> {
    let key1 = b"key1";

    let mut transaction = db.transaction();
    transaction.put(0, key1, b"horse");
    db.write(transaction).await.map_err(|e| e.error)?;
    assert_eq!(db.get(0, key1).await?.unwrap(), b"horse");

    let mut transaction = db.transaction();
    transaction.delete(0, key1);
    db.write(transaction).await.map_err(|e| e.error)?;
    assert!(db.get(0, key1).await?.is_none());
    Ok(())
}

/// A test for `KeyValueDB::get` and delete without transaction.
pub async fn test_delete_and_get_single<DB: KeyValueDB>(db: DB) -> io::Result<()> {
    let key1 = b"key1";

    let mut transaction = db.transaction();
    transaction.put(0, key1, b"horse");
    db.write(transaction).await.map_err(|e| e.error)?;
    assert_eq!(db.get(0, key1).await?.unwrap(), b"horse");

    assert_eq!(db.delete(0, key1).await?, Some(b"horse".to_vec()));
    assert!(db.get(0, key1).await?.is_none());
    assert_eq!(db.delete(0, key1).await?, None);

    Ok(())
}

/// A test for `KeyValueDB::get`.
/// Assumes the `db` has only 1 column.
pub async fn test_get_fails_with_non_existing_column<DB: KeyValueDB>(db: DB) -> io::Result<()> {
    assert!(db.get(1, b"").await.is_err());
    Ok(())
}

/// A test for `KeyValueDB::write`.
pub async fn test_write_clears_buffered_ops<DB: KeyValueDB>(db: DB) -> io::Result<()> {
    let mut batch = db.transaction();
    batch.put(0, b"foo", b"bar");
    db.write(batch).await.map_err(|e| e.error)?;

    assert_eq!(db.get(0, b"foo").await?.unwrap(), b"bar");

    let mut batch = db.transaction();
    batch.put(0, b"foo", b"baz");
    db.write(batch).await.map_err(|e| e.error)?;

    assert_eq!(db.get(0, b"foo").await?.unwrap(), b"baz");
    Ok(())
}

/// A test for `KeyValueDB::iter`.
pub async fn test_iter<DB: KeyValueDB>(db: DB) -> io::Result<()> {
    let key1 = b"key1";
    let key2 = b"key2";

    let mut transaction = db.transaction();
    transaction.put(0, key1, key1);
    transaction.put(0, key2, key2);
    db.write(transaction).await.map_err(|e| e.error)?;

    let mut contents: Vec<DBKeyValue> = Vec::new();
    let out = db
        .iter(0, None, |kv: DBKeyValueRef| {
            contents.push((kv.0.clone(), kv.1.clone()));
            Ok(Option::<()>::None)
        })
        .await?;
    assert!(out.is_none());
    assert_eq!(contents.len(), 2);
    assert_eq!(contents[0].0, key1);
    assert_eq!(contents[0].1, key1);
    assert_eq!(contents[1].0, key2);
    assert_eq!(contents[1].1, key2);

    // test iter with early return
    let mut contents: Vec<DBKeyValue> = Vec::new();
    let out = db
        .iter(0, None, |kv: DBKeyValueRef| {
            contents.push((kv.0.clone(), kv.1.clone()));
            Ok(Option::<()>::Some(()))
        })
        .await?;
    assert!(out.is_some());
    assert_eq!(contents.len(), 1);
    assert_eq!(contents[0].0, key1);
    assert_eq!(contents[0].1, key1);
    Ok(())
}

/// A test for `KeyValueDB::iter_keys`.
pub async fn test_iter_keys<DB: KeyValueDB>(db: DB) -> io::Result<()> {
    let key1 = b"key1";
    let key2 = b"key2";

    let mut transaction = db.transaction();
    transaction.put(0, key1, key1);
    transaction.put(0, key2, key2);
    db.write(transaction).await.map_err(|e| e.error)?;

    let mut contents: Vec<DBKey> = Vec::new();
    let out = db
        .iter_keys(0, None, |k: DBKeyRef| {
            contents.push(k.clone());
            Ok(Option::<()>::None)
        })
        .await?;
    assert!(out.is_none());
    assert_eq!(contents.len(), 2);
    assert_eq!(contents[0], key1);
    assert_eq!(contents[1], key2);

    // test iter keys with early return
    let mut contents: Vec<DBKey> = Vec::new();
    let out = db
        .iter_keys(0, None, |k: DBKeyRef| {
            contents.push(k.clone());
            Ok(Option::<()>::Some(()))
        })
        .await?;
    assert!(out.is_some());
    assert_eq!(contents.len(), 1);
    assert_eq!(contents[0], key1);

    Ok(())
}

/// A test for `KeyValueDB::iter` with a prefix.
pub async fn test_iter_with_prefix<DB: KeyValueDB>(db: DB) -> io::Result<()> {
    let key1 = b"0";
    let key2 = b"ab";
    let key3 = b"abc";
    let key4 = b"abcd";

    let mut batch = db.transaction();
    batch.put(0, key1, key1);
    batch.put(0, key2, key2);
    batch.put(0, key3, key3);
    batch.put(0, key4, key4);
    db.write(batch).await.map_err(|e| e.error)?;

    // empty prefix
    let mut contents: Vec<DBKeyValue> = Vec::new();
    let out = db
        .iter(0, Some(b""), |kv: DBKeyValueRef| {
            contents.push((kv.0.clone(), kv.1.clone()));
            Ok(Option::<()>::None)
        })
        .await?;
    assert!(out.is_none());
    assert_eq!(contents.len(), 4);
    assert_eq!(contents[0].0, key1);
    assert_eq!(contents[1].0, key2);
    assert_eq!(contents[2].0, key3);
    assert_eq!(contents[3].0, key4);

    // empty prefix with early return
    let mut contents: Vec<DBKeyValue> = Vec::new();
    let out = db
        .iter(0, Some(b""), |kv: DBKeyValueRef| {
            contents.push((kv.0.clone(), kv.1.clone()));
            Ok(Option::<()>::Some(()))
        })
        .await?;
    assert!(out.is_some());
    assert_eq!(contents.len(), 1);
    assert_eq!(contents[0].0, key1);

    // prefix a
    let mut contents: Vec<DBKeyValue> = Vec::new();
    db.iter(0, Some(b"a"), |kv: DBKeyValueRef| {
        contents.push((kv.0.clone(), kv.1.clone()));
        Ok(Option::<()>::None)
    })
    .await?;
    assert_eq!(contents.len(), 3);
    assert_eq!(contents[0].0, key2);
    assert_eq!(contents[1].0, key3);
    assert_eq!(contents[2].0, key4);

    // prefix abc
    let mut contents: Vec<DBKeyValue> = Vec::new();
    db.iter(0, Some(b"abc"), |kv: DBKeyValueRef| {
        contents.push((kv.0.clone(), kv.1.clone()));
        Ok(Option::<()>::None)
    })
    .await?;
    assert_eq!(contents.len(), 2);
    assert_eq!(contents[0].0, key3);
    assert_eq!(contents[1].0, key4);

    // prefix abcde
    let mut contents: Vec<DBKeyValue> = Vec::new();
    db.iter(0, Some(b"abcde"), |kv: DBKeyValueRef| {
        contents.push((kv.0.clone(), kv.1.clone()));
        Ok(Option::<()>::None)
    })
    .await?;
    assert_eq!(contents.len(), 0);

    // prefix 0
    let mut contents: Vec<DBKeyValue> = Vec::new();
    db.iter(0, Some(b"0"), |kv: DBKeyValueRef| {
        contents.push((kv.0.clone(), kv.1.clone()));
        Ok(Option::<()>::None)
    })
    .await?;
    assert_eq!(contents.len(), 1);
    assert_eq!(contents[0].0, key1);
    Ok(())
}

/// The number of columns required to run `test_io_stats`.
pub const IO_STATS_NUM_COLUMNS: u32 = 3;

/// A test for `KeyValueDB::io_stats`.
/// Assumes that the `db` has at least 3 columns.
pub async fn test_io_stats<DB: KeyValueDB>(db: DB) -> io::Result<()> {
    let key1 = b"kkk";
    let mut batch = db.transaction();
    batch.put(0, key1, key1);
    batch.put(1, key1, key1);
    batch.put(2, key1, key1);

    for _ in 0..10 {
        db.get(0, key1).await?;
    }

    db.write(batch).await.map_err(|e| e.error)?;

    let io_stats = db.io_stats(IoStatsKind::SincePrevious);
    assert_eq!(io_stats.transactions, 1);
    assert_eq!(io_stats.writes, 3);
    assert_eq!(io_stats.bytes_written, 18);
    assert_eq!(io_stats.reads, 10);
    assert_eq!(io_stats.bytes_read, 30);

    let new_io_stats = db.io_stats(IoStatsKind::SincePrevious);
    // Since we taken previous statistic period,
    // this is expected to be totally empty.
    assert_eq!(new_io_stats.transactions, 0);

    // but the overall should be there
    let new_io_stats = db.io_stats(IoStatsKind::Overall);
    assert_eq!(new_io_stats.bytes_written, 18);

    let mut batch = db.transaction();
    batch.delete(0, key1);
    batch.delete(1, key1);
    batch.delete(2, key1);

    // transaction is not commited yet
    assert_eq!(db.io_stats(IoStatsKind::SincePrevious).writes, 0);

    db.write(batch).await.map_err(|e| e.error)?;
    // now it is, and delete is counted as write
    assert_eq!(db.io_stats(IoStatsKind::SincePrevious).writes, 3);
    Ok(())
}

/// The number of columns required to run `test_delete_prefix`.
pub const DELETE_PREFIX_NUM_COLUMNS: u32 = 7;

/// A test for `KeyValueDB::delete_prefix`.
pub async fn test_delete_prefix<DB: KeyValueDB + 'static>(db: DB) -> io::Result<()> {
    let keys = [
        &[][..],
        &[0u8][..],
        &[0, 1][..],
        &[1][..],
        &[1, 0][..],
        &[1, 255][..],
        &[1, 255, 255][..],
        &[2][..],
        &[2, 0][..],
        &[2, 255][..],
        &[255; 16][..],
    ];
    let tests: [_; DELETE_PREFIX_NUM_COLUMNS as usize] = [
        // standard
        (
            &[1u8][..],
            [
                true, true, true, false, false, false, false, true, true, true, true,
            ],
        ),
        // edge
        (
            &[1u8, 255, 255][..],
            [
                true, true, true, true, true, true, false, true, true, true, true,
            ],
        ),
        // none 1
        (
            &[1, 2][..],
            [
                true, true, true, true, true, true, true, true, true, true, true,
            ],
        ),
        // none 2
        (
            &[8][..],
            [
                true, true, true, true, true, true, true, true, true, true, true,
            ],
        ),
        // last value
        (
            &[255, 255][..],
            [
                true, true, true, true, true, true, true, true, true, true, false,
            ],
        ),
        // last value, limit prefix
        (
            &[255][..],
            [
                true, true, true, true, true, true, true, true, true, true, false,
            ],
        ),
        // all
        (
            &[][..],
            [
                false, false, false, false, false, false, false, false, false, false, false,
            ],
        ),
    ];
    for (ix, test) in tests.iter().enumerate() {
        let ix = ix as u32;

        // Init Transaction
        let mut batch = db.transaction();
        for (i, key) in keys.iter().enumerate() {
            batch.put(ix, key, &[i as u8]);
        }
        db.write(batch).await.map_err(|e| e.error)?;

        // Delete Transaction
        let mut batch = db.transaction();
        batch.delete_prefix(ix, test.0);
        db.write(batch).await.map_err(|e| e.error)?;

        // Check Test
        let mut state = [true; 11];
        for (c, key) in keys.iter().enumerate() {
            state[c] = db.get(ix, key).await?.is_some();
        }
        assert_eq!(state, test.1, "at {}", ix);
    }

    Ok(())
}

/// A complex test.
pub async fn test_complex<DB: KeyValueDB>(db: DB) -> io::Result<()> {
    let key1 = b"02c69be41d0b7e40352fc85be1cd65eb03d40ef8427a0ca4596b1ead9a00e9fc";
    let key2 = b"03c69be41d0b7e40352fc85be1cd65eb03d40ef8427a0ca4596b1ead9a00e9fc";
    let key3 = b"04c00000000b7e40352fc85be1cd65eb03d40ef8427a0ca4596b1ead9a00e9fc";
    let key4 = b"04c01111110b7e40352fc85be1cd65eb03d40ef8427a0ca4596b1ead9a00e9fc";
    let key5 = b"04c02222220b7e40352fc85be1cd65eb03d40ef8427a0ca4596b1ead9a00e9fc";

    let mut batch = db.transaction();
    batch.put(0, key1, b"cat");
    batch.put(0, key2, b"dog");
    batch.put(0, key3, b"caterpillar");
    batch.put(0, key4, b"beef");
    batch.put(0, key5, b"fish");
    db.write(batch).await.map_err(|e| e.error)?;

    assert_eq!(db.get(0, key1).await?.unwrap(), b"cat");

    let mut contents: Vec<DBKeyValue> = Vec::new();
    db.iter(0, None, |kv: DBKeyValueRef| {
        contents.push((kv.0.clone(), kv.1.clone()));
        Ok(Option::<()>::None)
    })
    .await?;
    assert_eq!(contents.len(), 5);
    assert_eq!(contents[0].0, key1.to_vec());
    assert_eq!(contents[0].1, b"cat");
    assert_eq!(contents[1].0, key2.to_vec());
    assert_eq!(contents[1].1, b"dog");

    let mut contents: Vec<DBKeyValue> = Vec::new();
    db.iter(0, Some(b"04c0"), |kv: DBKeyValueRef| {
        contents.push((kv.0.clone(), kv.1.clone()));
        Ok(Option::<()>::None)
    })
    .await?;
    assert_eq!(contents[0].1, b"caterpillar");
    assert_eq!(contents[1].1, b"beef");
    assert_eq!(contents[2].1, b"fish");

    let mut batch = db.transaction();
    batch.delete(0, key1);
    db.write(batch).await.map_err(|e| e.error)?;

    assert!(db.get(0, key1).await?.is_none());

    let mut batch = db.transaction();
    batch.put(0, key1, b"cat");
    db.write(batch).await.map_err(|e| e.error)?;

    let mut transaction = db.transaction();
    transaction.put(0, key3, b"elephant");
    transaction.delete(0, key1);
    db.write(transaction).await.map_err(|e| e.error)?;
    assert!(db.get(0, key1).await?.is_none());
    assert_eq!(db.get(0, key3).await?.unwrap(), b"elephant");

    assert_eq!(
        db.first_with_prefix(0, key3).await?.unwrap(),
        (key3.to_vec(), b"elephant".to_vec())
    );
    assert_eq!(
        db.first_with_prefix(0, key2).await?.unwrap(),
        (key2.to_vec(), b"dog".to_vec())
    );

    let mut transaction = db.transaction();
    transaction.put(0, key1, b"horse");
    transaction.delete(0, key3);
    db.write(transaction).await.map_err(|e| e.error)?;
    assert!(db.get(0, key3).await?.is_none());
    assert_eq!(db.get(0, key1).await?.unwrap(), b"horse");

    assert!(db.get(0, key3).await?.is_none());
    assert_eq!(db.get(0, key1).await?.unwrap(), b"horse");
    Ok(())
}