use super::*;
fn config() -> Config {
    Config {
        bank_size: 4096,
        max_key: 32,
        max_value: 1024,
    }
}
#[test]
fn crud_reopen_limits() {
    let mut d = Database::open(MemoryStorage::default(), config()).unwrap();
    assert_eq!(d.get(b"a").unwrap(), None);
    d.put(b"b", &[0, 255, 0]).unwrap();
    d.put(b"a", b"").unwrap();
    assert_eq!(d.get(b"a").unwrap(), Some(vec![]));
    assert!(d.put(b"", b"x").is_err());
    assert!(d.put(&[0; 33], b"x").is_err());
    assert!(d.put(b"x", &[0; 1025]).is_err());
    d.put(&[1; 32], &[2; 1024]).unwrap();
    for n in 0..20 {
        d.put(b"b", &[n]).unwrap();
        d = Database::open(d.into_storage(), config()).unwrap();
        assert_eq!(d.get(b"b").unwrap(), Some(vec![n]));
    }
    d.delete(b"a").unwrap();
    assert_eq!(d.get(b"a").unwrap(), None);
    let mut n = 0;
    d.scan(|_, _| {
        n += 1;
        false
    })
    .unwrap();
    assert_eq!(n, 1);
}
// Volatile writes and durable image are separate. Failures can persist a torn prefix.
struct Fault {
    volatile: MemoryStorage,
    durable: MemoryStorage,
    step: usize,
    fail: usize,
    tear: bool,
}
impl Fault {
    fn tick(&mut self) -> Result<()> {
        self.step += 1;
        if self.step == self.fail {
            Err(Error::Io)
        } else {
            Ok(())
        }
    }
}
impl Storage for Fault {
    fn len(&mut self) -> Result<u64> {
        self.tick()?;
        self.volatile.len()
    }
    fn resize(&mut self, n: u64) -> Result<()> {
        if self.tick().is_err() {
            if self.tear {
                self.durable.resize(n / 2)?;
            }
            return Err(Error::Io);
        }
        self.volatile.resize(n)
    }
    fn read(&mut self, p: u64, b: &mut [u8]) -> Result<()> {
        self.tick()?;
        self.volatile.read(p, b)
    }
    fn write(&mut self, p: u64, b: &[u8]) -> Result<()> {
        if self.tick().is_err() {
            if self.tear {
                let n = b.len() / 2;
                self.volatile.write(p, &b[..n])?;
                self.durable = self.volatile.clone();
            }
            return Err(Error::Io);
        }
        self.volatile.write(p, b)
    }
    fn sync(&mut self) -> Result<()> {
        let result = self.tick();
        if result.is_err() && !self.tear {
            return result;
        }
        self.durable = self.volatile.clone();
        result?;
        Ok(())
    }
}
#[test]
fn every_commit_failure_recovers_old_or_new() {
    let mut d = Database::open(MemoryStorage::default(), config()).unwrap();
    d.put(b"a", b"old").unwrap();
    d.put(b"z", b"kept").unwrap();
    let base = d.into_storage();
    for tear in [false, true] {
        for fail in 1..40 {
            let s = Fault {
                volatile: base.clone(),
                durable: base.clone(),
                step: 0,
                fail: usize::MAX,
                tear,
            };
            let mut d = Database::open(s, config()).unwrap();
            d.storage.step = 0;
            d.storage.fail = fail;
            let result = d.put(b"a", b"new");
            if result.is_err() {
                assert_eq!(d.get(b"a"), Err(Error::Poisoned));
            }
            let s = d.into_storage();
            let mut recovered = Database::inspect(s.durable, config()).unwrap();
            let v = recovered.get(b"a").unwrap().unwrap();
            assert!(v == b"old" || v == b"new");
            if result.is_ok() {
                assert_eq!(v, b"new");
            }
            assert_eq!(recovered.get(b"z").unwrap().unwrap(), b"kept");
        }
    }
}
#[test]
fn corruption_version_truncation() {
    let mut d = Database::open(MemoryStorage::default(), config()).unwrap();
    d.put(b"a", b"b").unwrap();
    let bank = d.head.bank as usize;
    let s = d.into_storage();
    let mut damaged = s.clone();
    damaged.0[bank * 4096 + 76] ^= 1;
    assert!(matches!(
        Database::open(damaged, config()),
        Err(Error::Corrupt)
    ));
    let mut short = s.clone();
    short.0.pop();
    assert!(Database::open(short, config()).is_err());
    let mut future = s;
    let h = &mut future.0[bank * 4096..bank * 4096 + 64];
    h[8..12].copy_from_slice(&2u32.to_le_bytes());
    let c = crc(&h[..60]);
    h[60..].copy_from_slice(&c.to_le_bytes());
    assert!(matches!(
        Database::open(future, config()),
        Err(Error::Version)
    ));
}
#[test]
fn full_preserves_committed_state() {
    let c = Config {
        bank_size: 128,
        max_key: 32,
        max_value: 64,
    };
    let mut d = Database::open(MemoryStorage::default(), c).unwrap();
    d.put(b"a", b"saved").unwrap();
    assert_eq!(d.put(b"b", &[1; 64]), Err(Error::Full));
    let mut d = Database::open(d.into_storage(), c).unwrap();
    assert_eq!(d.get(b"a").unwrap().unwrap(), b"saved");
}
#[test]
fn crc_standard_vector() {
    assert_eq!(crc(b"123456789"), 0xcbf43926);
}
#[cfg(feature = "std")]
#[test]
fn file_exclusive_lock_and_reopen() {
    let p = std::env::temp_dir().join(format!("ratdb-test-{}.ratdb", std::process::id()));
    {
        let s = file::FileStorage::open(&p).unwrap();
        assert!(file::FileStorage::open(&p).is_err());
        let mut d = Database::open(s, config()).unwrap();
        d.put(b"key", b"value").unwrap();
    }
    {
        let mut d = Database::open(file::FileStorage::open(&p).unwrap(), config()).unwrap();
        assert_eq!(d.get(b"key").unwrap().unwrap(), b"value");
    }
    std::fs::remove_file(p).unwrap();
}

#[test]
fn deterministic_model_stress() {
    use alloc::collections::BTreeMap;
    let c = Config {
        bank_size: 65536,
        max_key: 8,
        max_value: 32,
    };
    let mut d = Database::open(MemoryStorage::default(), c).unwrap();
    let mut model = BTreeMap::new();
    let mut rng = 7u32;
    for step in 0..1200u32 {
        rng = rng.wrapping_mul(1664525).wrapping_add(1013904223);
        let k = (rng % 180).to_be_bytes();
        if rng.is_multiple_of(7) {
            d.delete(&k).unwrap();
            model.remove(k.as_slice());
        } else {
            let v = step.to_le_bytes().to_vec();
            d.put(&k, &v).unwrap();
            model.insert(k.to_vec(), v);
        }
        if step % 100 == 0 {
            d = Database::open(d.into_storage(), c).unwrap();
            let mut actual = BTreeMap::new();
            d.scan(|k, v| {
                actual.insert(k.to_vec(), v.to_vec());
                true
            })
            .unwrap();
            assert_eq!(actual, model);
        }
    }
}

#[test]
fn torn_commit_header_all_prefixes() {
    let mut d = Database::open(MemoryStorage::default(), config()).unwrap();
    d.put(b"a", b"old").unwrap();
    let old = d.into_storage();
    let mut d = Database::open(old.clone(), config()).unwrap();
    d.put(b"a", b"new").unwrap();
    let bank = d.head.bank as usize;
    let committed = d.into_storage();
    // Data was synced, invalidation was synced; header writes persist any prefix.
    for cut in 0..=64 {
        let mut torn = committed.clone();
        let pos = bank * 4096;
        torn.0[pos + cut..pos + 64].fill(0);
        let mut d = Database::inspect(torn, config()).unwrap();
        let v = d.get(b"a").unwrap().unwrap();
        assert!(v == b"old" || v == b"new");
        if cut == 64 {
            assert_eq!(v, b"new");
        }
    }
}

#[test]
fn damaged_header_is_reported() {
    let mut d = Database::open(MemoryStorage::default(), config()).unwrap();
    d.put(b"key", b"old").unwrap();
    d.put(b"key", b"new").unwrap();
    let bank = d.head.bank as usize;
    let mut s = d.into_storage();
    s.0[bank * 4096 + 60] ^= 1;
    assert!(matches!(
        Database::open(s.clone(), config()),
        Err(Error::Ambiguous)
    ));
    let mut recovered = Database::inspect(s.clone(), config()).unwrap();
    assert!(recovered.suspect_headers());
    assert_eq!(recovered.delete(b"key"), Err(Error::Invalid));
    assert_eq!(recovered.get(b"key").unwrap().unwrap(), b"old");
    assert_eq!(s.0, recovered.into_storage().0);
}
#[test]
fn interrupted_initialization_never_repairs_nonempty_media() {
    // Replay every storage call in open, including both post-publication reads.
    use alloc::rc::Rc;
    use core::cell::RefCell;
    struct Shared(Rc<RefCell<Fault>>);
    impl Storage for Shared {
        fn len(&mut self) -> Result<u64> {
            self.0.borrow_mut().len()
        }
        fn resize(&mut self, n: u64) -> Result<()> {
            self.0.borrow_mut().resize(n)
        }
        fn read(&mut self, p: u64, b: &mut [u8]) -> Result<()> {
            self.0.borrow_mut().read(p, b)
        }
        fn write(&mut self, p: u64, b: &[u8]) -> Result<()> {
            self.0.borrow_mut().write(p, b)
        }
        fn sync(&mut self) -> Result<()> {
            self.0.borrow_mut().sync()
        }
    }
    for tear in [false, true] {
        for fail in 1..=8 {
            let shared = Rc::new(RefCell::new(Fault {
                volatile: MemoryStorage::default(),
                durable: MemoryStorage::default(),
                step: 0,
                fail,
                tear,
            }));
            let result = Database::open(Shared(shared.clone()), config());
            assert_eq!(result.is_ok(), fail == 8);
            let image = shared.borrow().durable.clone();
            if image.0.is_empty() {
                assert!(Database::inspect(image, config()).is_err());
            } else {
                let bytes = image.0.clone();
                if let Ok(db) = Database::open(image, config()) {
                    assert_eq!(db.stats(), (0, 0, 1));
                    assert_eq!(db.into_storage().0, bytes);
                } else {
                    assert!(fail <= 5);
                }
            }
        }
    }
}
#[test]
fn repeated_failed_publications_then_acknowledgment() {
    let mut db = Database::open(MemoryStorage::default(), config()).unwrap();
    db.put(b"a", b"ack").unwrap();
    let mut image = db.into_storage();
    for fail in [2, 4, 3, 5, 2, 4] {
        let mut db = Database::open(
            Fault {
                volatile: image.clone(),
                durable: image,
                step: 0,
                fail: usize::MAX,
                tear: false,
            },
            config(),
        )
        .unwrap();
        assert_eq!(db.get(b"a").unwrap().unwrap(), b"ack");
        db.storage.step = 0;
        db.storage.fail = fail;
        assert_eq!(db.put(b"a", b"pending"), Err(Error::Io));
        image = db.into_storage().durable;
    }
    let mut db = Database::open(image, config()).unwrap();
    db.put(b"a", b"final").unwrap();
    assert_eq!(
        Database::open(db.into_storage(), config())
            .unwrap()
            .get(b"a")
            .unwrap()
            .unwrap(),
        b"final"
    );
}
#[test]
fn mixed_batch_is_atomic_at_every_adapter_call() {
    let base_changes = [
        Change {
            key: b"b",
            value: Some(b"old"),
        },
        Change {
            key: b"z",
            value: Some(b"kept"),
        },
    ];
    let next = [
        Change {
            key: b"a",
            value: Some(b"1"),
        },
        Change {
            key: b"b",
            value: None,
        },
        Change {
            key: b"y",
            value: Some(b"2"),
        },
    ];
    let mut db = Database::open(MemoryStorage::default(), config()).unwrap();
    db.apply(&base_changes).unwrap();
    let base = db.into_storage();
    for tear in [false, true] {
        for fail in 1.. {
            let mut d = Database::open(
                Fault {
                    volatile: base.clone(),
                    durable: base.clone(),
                    step: 0,
                    fail: usize::MAX,
                    tear,
                },
                config(),
            )
            .unwrap();
            d.storage.step = 0;
            d.storage.fail = fail;
            let ack = d.apply(&next).is_ok();
            let mut view = Database::inspect(d.into_storage().durable, config()).unwrap();
            let old = view.get(b"b").unwrap() == Some(b"old".to_vec());
            assert!(!ack || !old);
            assert_eq!(view.get(b"a").unwrap(), (!old).then(|| b"1".to_vec()));
            assert_eq!(view.get(b"y").unwrap(), (!old).then(|| b"2".to_vec()));
            assert_eq!(view.get(b"z").unwrap(), Some(b"kept".to_vec()));
            if ack {
                break;
            }
        }
    }
}
#[test]
fn bounded_seek_and_batch_validation() {
    let cfg = Config {
        bank_size: 32768,
        ..config()
    };
    let mut d = Database::open(MemoryStorage::default(), cfg).unwrap();
    let keys: Vec<_> = (0..1024u32).map(|i| (2 * i).to_be_bytes()).collect();
    let changes: Vec<_> = keys
        .iter()
        .map(|k| Change {
            key: k,
            value: Some(k),
        })
        .collect();
    assert_eq!(d.apply(&[changes[0]; MAX_BATCH + 1]), Err(Error::Invalid));
    assert_eq!(d.apply(&[changes[1], changes[0]]), Err(Error::Invalid));
    assert_eq!(d.apply(&[changes[0]; 2]), Err(Error::Invalid));
    assert_eq!(d.stats().2, 1);
    d.apply(&changes).unwrap();
    for _ in 0..2 {
        assert!(d.offsets.len() <= 64);
        for i in 0..2050u32 {
            assert_eq!(
                d.get(&i.to_be_bytes()).unwrap(),
                (i < 2048 && i % 2 == 0).then(|| i.to_be_bytes().to_vec())
            );
        }
        d = Database::open(d.into_storage(), cfg).unwrap();
    }
}
#[test]
fn tiny_index_limits_preserve_seeks_and_updates() {
    let mut db = Database::open(MemoryStorage::default(), config()).unwrap();
    for i in 0..50u8 {
        db.put(&[i], &[i + 1]).unwrap();
    }
    let mut storage = db.into_storage();
    for limit in [0, 1, 3, 16] {
        let mut db = Database::open_with_index_limit(storage, config(), limit).unwrap();
        assert!(db.offsets.len() <= usize::from(limit));
        assert_eq!(db.get(&[48]).unwrap(), Some(vec![49]));
        db.put(&[48], &[49]).unwrap();
        assert!(db.offsets.len() <= usize::from(limit));
        storage = db.into_storage();
    }
    assert!(matches!(
        Database::open_with_index_limit(storage, config(), 65),
        Err(Error::Invalid)
    ));
}
#[cfg(feature = "std")]
#[test]
fn existing_file_open_never_initializes() {
    let dir = std::env::temp_dir().join(format!("ratdb-open-{}", std::process::id()));
    std::fs::create_dir(&dir).unwrap();
    let path = dir.join("db");
    assert!(file::FileStorage::open_existing(&path).is_err());
    assert!(!path.exists());
    let s = file::FileStorage::create_new(&path).unwrap();
    assert!(file::FileStorage::create_new(&path).is_err());
    drop(s);
    assert!(file::FileStorage::open_existing(&path).is_err());
    assert_eq!(std::fs::metadata(&path).unwrap().len(), 0);
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        assert_eq!(
            std::fs::metadata(&path).unwrap().permissions().mode() & 0o077,
            0
        );
    }
    std::fs::remove_file(path).unwrap();
    std::fs::remove_dir(dir).unwrap();
}
#[test]
fn all_invalidation_prefixes_preserve_active_data() {
    let mut db = Database::open(MemoryStorage::default(), config()).unwrap();
    db.put(b"k", b"previous").unwrap();
    db.put(b"k", b"acknowledged").unwrap();
    let target = (1 - db.head.bank) as usize * 4096;
    let original = db.into_storage();
    for cut in 0..=64 {
        let mut s = original.clone();
        s.0[target..target + cut].fill(0);
        let mut inspection = Database::inspect(s.clone(), config()).unwrap();
        assert_eq!(inspection.get(b"k").unwrap().unwrap(), b"acknowledged");
        assert_eq!(
            Database::open(s, config()).is_err(),
            inspection.suspect_headers()
        );
    }
}
#[test]
fn valid_crc_does_not_authorize_impossible_generations() {
    let mut db = Database::open(MemoryStorage::default(), config()).unwrap();
    db.put(b"k", b"value").unwrap();
    let original = db.into_storage();
    for gen in [0u64, 1, 4, u64::MAX] {
        let mut image = original.clone();
        let h = &mut image.0[4096..4160];
        h[16..24].copy_from_slice(&gen.to_le_bytes());
        let sum = crc(&h[..60]);
        h[60..].copy_from_slice(&sum.to_le_bytes());
        assert!(matches!(
            Database::open(image, config()),
            Err(Error::Corrupt)
        ));
    }
}
// Unsynced bytes may independently reach media before the next successful barrier.
// Sample non-prefix tears deterministically, including write/sync returning an error
// after storing some bytes. Previously synced bytes outside the write range survive.
#[test]
fn scattered_unsynced_writes_never_mix_generations() {
    struct Device {
        pending: MemoryStorage,
        media: MemoryStorage,
        rng: u32,
        calls: usize,
        cut: usize,
    }
    impl Device {
        fn boundary(&mut self) -> Result<()> {
            self.calls += 1;
            if self.calls == self.cut {
                Err(Error::Io)
            } else {
                Ok(())
            }
        }
        fn spill(&mut self, p: usize, n: usize) {
            for i in p..p + n {
                self.rng = self.rng.wrapping_mul(1664525).wrapping_add(1013904223);
                if self.rng >> 31 != 0 {
                    self.media.0[i] = self.pending.0[i];
                }
            }
        }
    }
    impl Storage for Device {
        fn len(&mut self) -> Result<u64> {
            self.pending.len()
        }
        fn resize(&mut self, _: u64) -> Result<()> {
            panic!("commit unexpectedly resized storage")
        }
        fn read(&mut self, p: u64, b: &mut [u8]) -> Result<()> {
            self.pending.read(p, b)
        }
        fn write(&mut self, p: u64, b: &[u8]) -> Result<()> {
            self.pending.write(p, b)?;
            self.spill(p as usize, b.len());
            self.boundary()
        }
        fn sync(&mut self) -> Result<()> {
            if self.boundary().is_err() {
                self.spill(0, self.pending.0.len());
                return Err(Error::Io);
            }
            self.media = self.pending.clone();
            Ok(())
        }
    }
    let cfg = config();
    let mut db = Database::open(MemoryStorage::default(), cfg).unwrap();
    db.put(b"one", b"old").unwrap();
    db.put(b"two", b"old").unwrap();
    let base = db.into_storage();
    for rng in 1..=32 {
        for cut in 1.. {
            let mut d = Database::open(
                Device {
                    pending: base.clone(),
                    media: base.clone(),
                    rng,
                    calls: 0,
                    cut,
                },
                cfg,
            )
            .unwrap();
            let ack = d
                .apply(&[
                    Change {
                        key: b"one",
                        value: Some(b"updated"),
                    },
                    Change {
                        key: b"two",
                        value: None,
                    },
                ])
                .is_ok();
            let image = d.into_storage().media;
            let mut db = match Database::open(image.clone(), cfg) {
                Ok(db) => db,
                Err(Error::Ambiguous) if !ack => Database::inspect(image, cfg).unwrap(),
                _ => panic!("unexpected open outcome at seed {rng}, boundary {cut}"),
            };
            let a = db.get(b"one").unwrap().unwrap();
            let b = db.get(b"two").unwrap();
            assert!(
                (a == b"old" && b.as_deref() == Some(b"old")) || (a == b"updated" && b.is_none())
            );
            if ack {
                assert_eq!(a, b"updated");
                break;
            }
        }
    }
}