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

use armdb::codec::Codec;
use armdb::{Config, DbError, MigrateAction, RapiraCodec, VarTree, VarTypedTree};
use parking_lot::Mutex;
use rapira::Rapira;
use std::sync::Arc;
use tempfile::tempdir;

#[derive(Debug, Clone, PartialEq, Rapira)]
struct Msg {
    id: u64,
    body: String,
    tags: Vec<String>,
}

type MsgTree = VarTypedTree<[u8; 16], Msg, RapiraCodec>;

fn open_tree(dir: &std::path::Path) -> MsgTree {
    VarTypedTree::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 msg(id: u64) -> Msg {
    Msg {
        id,
        body: format!("hello from message #{id}"),
        tags: vec![format!("t{id}"), "bench".into()],
    }
}

#[test]
fn put_get_round_trip() {
    let dir = tempdir().unwrap();
    let tree = open_tree(dir.path());

    let m = msg(42);
    tree.put(&key(42), &m).unwrap();

    let got = tree.get(&key(42)).unwrap();
    assert_eq!(got, m);
    assert!(tree.contains(&key(42)));
    assert_eq!(tree.len(), 1);
}

#[test]
fn missing_key_returns_none() {
    let dir = tempdir().unwrap();
    let tree = open_tree(dir.path());
    assert!(tree.get(&key(99)).is_none());
    assert!(!tree.contains(&key(99)));
}

#[test]
fn insert_twice_fails() {
    let dir = tempdir().unwrap();
    let tree = open_tree(dir.path());
    tree.insert(&key(1), &msg(1)).unwrap();
    assert!(tree.insert(&key(1), &msg(1)).is_err());
}

#[test]
fn delete_works() {
    let dir = tempdir().unwrap();
    let tree = open_tree(dir.path());
    tree.put(&key(1), &msg(1)).unwrap();
    assert!(tree.delete(&key(1)).unwrap());
    assert!(!tree.contains(&key(1)));
    assert!(!tree.delete(&key(1)).unwrap()); // second delete → false
}

#[test]
fn cas_success_and_mismatch() {
    let dir = tempdir().unwrap();
    let tree = open_tree(dir.path());
    let m1 = msg(7);
    let m2 = Msg {
        id: 7,
        body: "updated".into(),
        tags: vec!["u".into()],
    };
    tree.put(&key(7), &m1).unwrap();

    tree.cas(&key(7), &m1, &m2).unwrap();
    assert_eq!(tree.get(&key(7)).unwrap(), m2);

    // now current != m1 — mismatch
    assert!(tree.cas(&key(7), &m1, &m2).is_err());
}

#[test]
fn update_returns_new_value() {
    let dir = tempdir().unwrap();
    let tree = open_tree(dir.path());
    tree.put(&key(1), &msg(1)).unwrap();

    let new = tree
        .update(&key(1), |cur| Msg {
            id: cur.id,
            body: format!("{}+", cur.body),
            tags: cur.tags.clone(),
        })
        .unwrap()
        .unwrap();
    assert!(new.body.ends_with('+'));

    let got = tree.get(&key(1)).unwrap();
    assert_eq!(got.body, new.body);
}

#[test]
fn fetch_update_returns_old_value() {
    let dir = tempdir().unwrap();
    let tree = open_tree(dir.path());
    tree.put(&key(1), &msg(1)).unwrap();

    let old = tree
        .fetch_update(&key(1), |_| Msg {
            id: 1,
            body: "new".into(),
            tags: vec![],
        })
        .unwrap()
        .unwrap();
    assert_eq!(old, msg(1));
    assert_eq!(tree.get(&key(1)).unwrap().body, "new");
}

#[test]
fn update_missing_key_returns_none() {
    let dir = tempdir().unwrap();
    let tree = open_tree(dir.path());
    let out = tree.update(&key(999), |m| m.clone()).unwrap();
    assert!(out.is_none());
}

#[test]
fn iteration_desc_by_default() {
    let dir = tempdir().unwrap();
    let tree = open_tree(dir.path());
    for i in 0..10 {
        tree.put(&key(i), &msg(i)).unwrap();
    }

    let keys: Vec<u64> = tree
        .iter()
        .map(|(k, _)| u64::from_be_bytes(k[8..].try_into().unwrap()))
        .collect();
    // reversed=true by default → DESC
    assert_eq!(keys, (0..10).rev().collect::<Vec<_>>());
}

#[test]
fn range_iteration() {
    let dir = tempdir().unwrap();
    let tree = open_tree(dir.path());
    for i in 0..20 {
        tree.put(&key(i), &msg(i)).unwrap();
    }

    // range with default reversed=true: from larger towards smaller
    let collected: Vec<u64> = tree
        .range(&key(10), &key(5))
        .map(|(k, _)| u64::from_be_bytes(k[8..].try_into().unwrap()))
        .collect();
    // With reversed=true DESC, range(start=10, end=5) → 10..5 exclusive of 5.
    // See var_tree behavior: first arg is upper (included), second is lower (excluded).
    assert!(collected.iter().all(|&x| (6..=10).contains(&x)));
}

#[test]
fn first_and_last() {
    let dir = tempdir().unwrap();
    let tree = open_tree(dir.path());
    for i in 0..5 {
        tree.put(&key(i), &msg(i)).unwrap();
    }
    // reversed=true → first is the largest key, last is the smallest
    let (first_k, _) = tree.first().unwrap();
    let (last_k, _) = tree.last().unwrap();
    assert_eq!(u64::from_be_bytes(first_k[8..].try_into().unwrap()), 4);
    assert_eq!(u64::from_be_bytes(last_k[8..].try_into().unwrap()), 0);
}

#[test]
fn atomic_multi_ops() {
    let dir = tempdir().unwrap();
    let tree = open_tree(dir.path());
    let anchor = key(0);
    let shard = tree.shard_for(&anchor);

    // pick keys that land on the same shard
    let mut same_shard_keys: Vec<[u8; 16]> = (0u64..10_000)
        .map(key)
        .filter(|k| tree.shard_for(k) == shard)
        .take(3)
        .collect();
    assert_eq!(same_shard_keys.len(), 3);

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

    for k in &same_shard_keys {
        assert!(tree.contains(k));
    }

    // ShardMismatch when passing a foreign key
    let foreign = same_shard_keys.remove(0);
    let other: [u8; 16] = (0u64..10_000)
        .map(key)
        .find(|k| tree.shard_for(k) != shard)
        .unwrap();
    let err = tree.atomic(&foreign, |s| s.put(&other, &msg(0)));
    assert!(err.is_err());
}

/// Codec that fails decode for values whose first byte is 0xFF.
#[derive(Clone)]
struct FailingCodec;

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

    fn decode_from(&self, bytes: &[u8]) -> armdb::DbResult<Msg> {
        if bytes.first() == Some(&0xFF) {
            return Err(DbError::CorruptedEntry { offset: 0 });
        }
        RapiraCodec.decode_from(bytes)
    }
}

/// F-5: migrate keeps entries that fail to decode; user callback is not invoked.
#[test]
fn test_var_typed_migrate_keeps_undecoded_entries() {
    let dir = tempdir().unwrap();
    let good_key = key(1);
    let bad_key = key(2);

    {
        let tree = VarTypedTree::open(dir.path(), Config::test(), RapiraCodec).unwrap();
        tree.put(&good_key, &msg(1)).unwrap();
        tree.close().unwrap();
    }

    {
        let raw = VarTree::<[u8; 16]>::open(dir.path(), Config::test()).unwrap();
        raw.put(&bad_key, &[0xFF, 0x00, 0x01, 0x02]).unwrap();
        raw.close().unwrap();
    }

    let called = Arc::new(Mutex::new(Vec::<[u8; 16]>::new()));
    {
        let tree = VarTypedTree::<[u8; 16], Msg, FailingCodec>::open(
            dir.path(),
            Config::test(),
            FailingCodec,
        )
        .unwrap();
        assert_eq!(tree.len(), 2);

        let called_clone = Arc::clone(&called);
        let mutated = tree
            .migrate(|k, m| {
                called_clone.lock().push(*k);
                MigrateAction::Update(Msg {
                    id: m.id,
                    body: format!("migrated-{}", m.id),
                    tags: m.tags.clone(),
                })
            })
            .unwrap();
        assert_eq!(mutated, 1);
        tree.close().unwrap();
    }

    assert_eq!(*called.lock(), vec![good_key]);

    let tree =
        VarTypedTree::<[u8; 16], Msg, FailingCodec>::open(dir.path(), Config::test(), FailingCodec)
            .unwrap();
    assert_eq!(tree.len(), 2);
    assert_eq!(tree.get(&good_key).unwrap().body, "migrated-1".to_string());
    assert!(tree.get(&bad_key).is_none());

    let raw = VarTree::<[u8; 16]>::open(dir.path(), Config::test()).unwrap();
    assert_eq!(raw.get(&bad_key).unwrap().as_ref()[0], 0xFF);
}

#[test]
fn migrate_keep_update_delete() {
    let dir = tempdir().unwrap();
    let tree = open_tree(dir.path());
    for i in 0..6 {
        tree.put(&key(i), &msg(i)).unwrap();
    }

    let mutated = tree
        .migrate(|k, m| {
            let id = u64::from_be_bytes(k[8..].try_into().unwrap());
            if id % 3 == 0 {
                MigrateAction::Delete
            } else if id % 3 == 1 {
                MigrateAction::Update(Msg {
                    id: m.id,
                    body: "migrated".into(),
                    tags: m.tags.clone(),
                })
            } else {
                MigrateAction::Keep
            }
        })
        .unwrap();

    // 0,3 deleted; 1,4 updated → 4 mutations
    assert_eq!(mutated, 4);

    // verify
    for i in [0u64, 3] {
        assert!(!tree.contains(&key(i)));
    }
    for i in [1u64, 4] {
        assert_eq!(tree.get(&key(i)).unwrap().body, "migrated");
    }
    for i in [2u64, 5] {
        assert_eq!(tree.get(&key(i)).unwrap(), msg(i));
    }
}

#[test]
fn recovery_after_close_reopen() {
    let dir = tempdir().unwrap();
    {
        let tree = open_tree(dir.path());
        for i in 0..50 {
            tree.put(&key(i), &msg(i)).unwrap();
        }
        tree.close().unwrap();
    }
    // reopen
    let tree = open_tree(dir.path());
    assert_eq!(tree.len(), 50);
    for i in 0..50 {
        assert_eq!(tree.get(&key(i)).unwrap(), msg(i));
    }
}

#[test]
fn prefix_iter() {
    let dir = tempdir().unwrap();
    let tree = open_tree(dir.path());
    // build keys with shared prefix
    for i in 0..5u8 {
        let mut k = [0u8; 16];
        k[0] = 0xAA;
        k[8..].copy_from_slice(&(i as u64).to_be_bytes());
        tree.put(
            &k,
            &Msg {
                id: i as u64,
                body: "p".into(),
                tags: vec![],
            },
        )
        .unwrap();
    }
    // unrelated
    for i in 0..3u8 {
        let mut k = [0u8; 16];
        k[0] = 0xBB;
        k[8..].copy_from_slice(&(i as u64).to_be_bytes());
        tree.put(
            &k,
            &Msg {
                id: i as u64,
                body: "q".into(),
                tags: vec![],
            },
        )
        .unwrap();
    }
    let bodies: Vec<String> = tree.prefix_iter(&[0xAA]).map(|(_, v)| v.body).collect();
    assert_eq!(bodies.len(), 5);
    assert!(bodies.iter().all(|b| b == "p"));
}

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

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

    let result = tree.update(&k, |m| Msg {
        id: m.id + 1,
        ..m.clone()
    });
    assert!(matches!(result, Err(DbError::CorruptedEntry { .. })));
    assert_eq!(tree.entry_len(&k).unwrap(), len_before);
}

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

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

    let result = tree.fetch_update(&k, |m| Msg {
        id: m.id + 1,
        ..m.clone()
    });
    assert!(matches!(result, Err(DbError::CorruptedEntry { .. })));
    assert_eq!(tree.entry_len(&k).unwrap(), len_before);
}

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

    tree.put(&key(3), &msg(3)).unwrap();

    let new = tree
        .update(&key(3), |m| Msg {
            id: m.id + 100,
            body: "updated".into(),
            tags: m.tags.clone(),
        })
        .unwrap()
        .unwrap();
    assert_eq!(new.id, 103);
    assert_eq!(new.body, "updated");

    let fetched_old = tree
        .fetch_update(&key(3), |m| Msg {
            id: m.id + 1,
            ..m.clone()
        })
        .unwrap()
        .unwrap();
    assert_eq!(fetched_old.id, 103);
    assert_eq!(tree.get(&key(3)).unwrap().id, 104);
}

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<Msg> for CountingCodec {
        fn encode_to(&self, value: &Msg, buf: &mut Vec<u8>) -> armdb::DbResult<()> {
            RapiraCodec.encode_to(value, buf)
        }

        fn decode_from(&self, bytes: &[u8]) -> armdb::DbResult<Msg> {
            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 tree: VarTypedTree<[u8; 16], Msg, CountingCodec> =
            VarTypedTree::open(dir.path(), Config::test(), CountingCodec).unwrap();

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

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

        tree.put(&key(1), &msg(1)).unwrap();
        reset();
        tree.update(&key(1), |m| Msg {
            id: m.id + 1,
            ..m.clone()
        })
        .unwrap();
        assert_eq!(
            count(),
            1,
            "update on NoHook should decode current value once"
        );
    }
}
