#![cfg(all(feature = "rapira-codec", feature = "var-collections"))]

use armdb::{Config, DbError, MigrateAction, RapiraCodec, VarTypedMap};
use rapira::Rapira;
use tempfile::tempdir;

#[derive(Debug, Clone, PartialEq, Rapira)]
struct Profile {
    name: String,
    email: String,
    scores: Vec<i64>,
}

type ProfileMap = VarTypedMap<[u8; 16], Profile, RapiraCodec>;

fn open_map(dir: &std::path::Path) -> ProfileMap {
    VarTypedMap::open(dir, Config::test(), RapiraCodec).unwrap()
}

fn key(id: u64) -> [u8; 16] {
    let mut k = [0u8; 16];
    k[8..].copy_from_slice(&id.to_be_bytes());
    k
}

fn profile(id: u64) -> Profile {
    Profile {
        name: format!("user_{id}"),
        email: format!("user_{id}@ex.com"),
        scores: vec![id as i64, (id * 2) as i64, (id * 3) as i64],
    }
}

#[test]
fn put_get_round_trip() {
    let dir = tempdir().unwrap();
    let map = open_map(dir.path());
    let p = profile(1);
    map.put(&key(1), &p).unwrap();
    assert_eq!(map.get(&key(1)).unwrap(), p);
    assert!(map.contains(&key(1)));
}

#[test]
fn update_and_fetch_update() {
    let dir = tempdir().unwrap();
    let map = open_map(dir.path());
    map.put(&key(1), &profile(1)).unwrap();

    let new = map
        .update(&key(1), |cur| Profile {
            name: cur.name.clone(),
            email: "new@ex.com".into(),
            scores: cur.scores.clone(),
        })
        .unwrap()
        .unwrap();
    assert_eq!(new.email, "new@ex.com");

    let old = map
        .fetch_update(&key(1), |cur| Profile {
            name: cur.name.clone(),
            email: "newer@ex.com".into(),
            scores: cur.scores.clone(),
        })
        .unwrap()
        .unwrap();
    assert_eq!(old.email, "new@ex.com");
    assert_eq!(map.get(&key(1)).unwrap().email, "newer@ex.com");
}

#[test]
fn cas_flow() {
    let dir = tempdir().unwrap();
    let map = open_map(dir.path());
    let p1 = profile(1);
    let p2 = Profile {
        name: "x".into(),
        email: "x@x".into(),
        scores: vec![],
    };
    map.put(&key(1), &p1).unwrap();
    map.cas(&key(1), &p1, &p2).unwrap();
    assert_eq!(map.get(&key(1)).unwrap(), p2);
    assert!(map.cas(&key(1), &p1, &p2).is_err());
}

#[test]
fn delete_flow() {
    let dir = tempdir().unwrap();
    let map = open_map(dir.path());
    map.put(&key(1), &profile(1)).unwrap();
    assert!(map.delete(&key(1)).unwrap());
    assert!(!map.contains(&key(1)));
    assert!(!map.delete(&key(1)).unwrap());
}

#[test]
fn migrate_flow() {
    let dir = tempdir().unwrap();
    let map = open_map(dir.path());
    for i in 0..6u64 {
        map.put(&key(i), &profile(i)).unwrap();
    }
    let mutated = map
        .migrate(|k, p| {
            let id = u64::from_be_bytes(k[8..].try_into().unwrap());
            if id % 2 == 0 {
                MigrateAction::Delete
            } else {
                MigrateAction::Update(Profile {
                    name: format!("m_{id}"),
                    ..p.clone()
                })
            }
        })
        .unwrap();
    assert_eq!(mutated, 6);
    assert_eq!(map.len(), 3);
    for i in [1u64, 3, 5] {
        assert_eq!(map.get(&key(i)).unwrap().name, format!("m_{i}"));
    }
}

#[test]
fn atomic_multi_ops() {
    let dir = tempdir().unwrap();
    let map = open_map(dir.path());
    let anchor = key(0);
    let shard = map.shard_for(&anchor);
    let same_shard: Vec<[u8; 16]> = (0u64..10_000)
        .map(key)
        .filter(|k| map.shard_for(k) == shard)
        .take(3)
        .collect();
    assert_eq!(same_shard.len(), 3);

    map.atomic(&same_shard[0], |s| {
        for k in &same_shard {
            s.put(k, &profile(u64::from_be_bytes(k[8..].try_into().unwrap())))?;
        }
        Ok(())
    })
    .unwrap();

    for k in &same_shard {
        assert!(map.contains(k));
    }
}

#[test]
fn recovery_reopen() {
    let dir = tempdir().unwrap();
    {
        let map = open_map(dir.path());
        for i in 0..30 {
            map.put(&key(i), &profile(i)).unwrap();
        }
        map.close().unwrap();
    }
    let map = open_map(dir.path());
    assert_eq!(map.len(), 30);
    for i in 0..30 {
        assert_eq!(map.get(&key(i)).unwrap(), profile(i));
    }
}

#[test]
fn update_decode_failure_no_side_effects() {
    let dir = tempdir().unwrap();
    let map = open_map(dir.path());

    let k = key(1);
    map.as_inner().put(&k, b"not-valid-rapira").unwrap();
    let len_before = map.entry_len(&k).unwrap();

    let result = map.update(&k, |p| Profile {
        name: "changed".into(),
        ..p.clone()
    });
    assert!(matches!(result, Err(DbError::CorruptedEntry { .. })));
    assert_eq!(map.entry_len(&k).unwrap(), len_before);
}

#[test]
fn fetch_update_decode_failure_no_side_effects() {
    let dir = tempdir().unwrap();
    let map = open_map(dir.path());

    let k = key(2);
    map.as_inner().put(&k, b"not-valid-rapira").unwrap();
    let len_before = map.entry_len(&k).unwrap();

    let result = map.fetch_update(&k, |p| Profile {
        name: "changed".into(),
        ..p.clone()
    });
    assert!(matches!(result, Err(DbError::CorruptedEntry { .. })));
    assert_eq!(map.entry_len(&k).unwrap(), len_before);
}

#[test]
fn update_normal_path_after_refactor() {
    let dir = tempdir().unwrap();
    let map = open_map(dir.path());

    map.put(&key(3), &profile(3)).unwrap();

    let new = map
        .update(&key(3), |p| Profile {
            name: "updated".into(),
            ..p.clone()
        })
        .unwrap()
        .unwrap();
    assert_eq!(new.name, "updated");

    let fetched_old = map
        .fetch_update(&key(3), |p| Profile {
            name: "again".into(),
            ..p.clone()
        })
        .unwrap()
        .unwrap();
    assert_eq!(fetched_old.name, "updated");
    assert_eq!(map.get(&key(3)).unwrap().name, "again");
}

mod nohook_decode_counter {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    static DECODE_COUNT: AtomicUsize = AtomicUsize::new(0);

    #[derive(Clone)]
    struct CountingCodec;

    impl armdb::Codec<Profile> for CountingCodec {
        fn encode_to(&self, value: &Profile, buf: &mut Vec<u8>) -> armdb::DbResult<()> {
            RapiraCodec.encode_to(value, buf)
        }

        fn decode_from(&self, bytes: &[u8]) -> armdb::DbResult<Profile> {
            DECODE_COUNT.fetch_add(1, Ordering::Relaxed);
            RapiraCodec.decode_from(bytes)
        }
    }

    fn reset() {
        DECODE_COUNT.store(0, Ordering::Relaxed);
    }

    fn count() -> usize {
        DECODE_COUNT.load(Ordering::Relaxed)
    }

    #[test]
    fn put_no_redundant_decode() {
        let dir = tempdir().unwrap();
        let map: VarTypedMap<[u8; 16], Profile, CountingCodec> =
            VarTypedMap::open(dir.path(), Config::test(), CountingCodec).unwrap();

        reset();
        map.put(&key(1), &profile(1)).unwrap();
        assert_eq!(count(), 0, "NoHook path should not decode on put");
    }

    #[test]
    fn update_one_decode_only() {
        let dir = tempdir().unwrap();
        let map: VarTypedMap<[u8; 16], Profile, CountingCodec> =
            VarTypedMap::open(dir.path(), Config::test(), CountingCodec).unwrap();

        map.put(&key(1), &profile(1)).unwrap();
        reset();
        map.update(&key(1), |p| Profile {
            name: "updated".into(),
            ..p.clone()
        })
        .unwrap();
        assert_eq!(
            count(),
            1,
            "update on NoHook should decode current value once"
        );
    }
}
