#![allow(dead_code, unused_imports)]

mod common;

use std::hint::black_box;
use std::time::Duration;

use armdb::{
    ConstMap, ConstTree, FixedMap, FixedTree, RapiraCodec, TypedMap, TypedTree, VarMap, VarTree,
};
use criterion::{Criterion, Throughput, criterion_group, criterion_main};
use rand::RngExt;
use tempfile::TempDir;

use common::config::{bitcask_config, fixed_config};
use common::data::{make_const_value, make_key, make_record, make_var_value};
use common::latency::Probe;
use common::naming::id_with_mode;
use common::populate::*;
use common::runner::{mix_batch, op_batch, read_seq, write_rand, write_seq};
use common::{
    FULL_NUM_ENTRIES, FULL_WRITE_BATCH, HOT_RANGE, NUM_GROUPS, READ_BATCH, Record, SCAN_TAKE,
};

const MODES: [bool; 2] = [false, true]; // page_cache, direct_io

// ---------------------------------------------------------------------------
// seq-write — Bitcask collections, both modes
// ---------------------------------------------------------------------------

fn bench_seq_write(c: &mut Criterion) {
    let mut group = c.benchmark_group("full/seq-write");
    group.throughput(Throughput::Elements(FULL_WRITE_BATCH));
    group.sample_size(10);

    for &direct_io in &MODES {
        group.bench_function(id_with_mode("const_tree", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir = TempDir::new().unwrap();
            let const_tree = ConstTree::<[u8; 8], 16>::open(dir.path(), cfg).unwrap();
            write_seq(
                b,
                FULL_WRITE_BATCH,
                |i| {
                    const_tree.put(&make_key(i), &make_const_value(i)).unwrap();
                },
                || const_tree.flush_buffers().unwrap(),
            )
        });

        group.bench_function(id_with_mode("typed_tree", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_t = TempDir::new().unwrap();
            let typed_tree =
                TypedTree::<[u8; 8], Record, RapiraCodec>::open(dir_t.path(), cfg, RapiraCodec)
                    .unwrap();
            write_seq(
                b,
                FULL_WRITE_BATCH,
                |i| {
                    typed_tree.put(&make_key(i), make_record(i)).unwrap();
                },
                || typed_tree.flush_buffers().unwrap(),
            )
        });

        group.bench_function(id_with_mode("var_tree", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_v = TempDir::new().unwrap();
            let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), cfg).unwrap();
            write_seq(
                b,
                FULL_WRITE_BATCH,
                |i| {
                    var_tree.put(&make_key(i), &make_var_value(i)).unwrap();
                },
                || var_tree.flush_buffers().unwrap(),
            )
        });

        group.bench_function(id_with_mode("const_map", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_cm = TempDir::new().unwrap();
            let const_map = ConstMap::<[u8; 8], 16>::open(dir_cm.path(), cfg).unwrap();
            write_seq(
                b,
                FULL_WRITE_BATCH,
                |i| {
                    const_map.put(&make_key(i), &make_const_value(i)).unwrap();
                },
                || const_map.flush_buffers().unwrap(),
            )
        });

        group.bench_function(id_with_mode("typed_map", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_tm = TempDir::new().unwrap();
            let typed_map =
                TypedMap::<[u8; 8], Record, RapiraCodec>::open(dir_tm.path(), cfg, RapiraCodec)
                    .unwrap();
            write_seq(
                b,
                FULL_WRITE_BATCH,
                |i| {
                    typed_map.put(&make_key(i), make_record(i)).unwrap();
                },
                || typed_map.flush_buffers().unwrap(),
            )
        });

        group.bench_function(id_with_mode("var_map", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_vm = TempDir::new().unwrap();
            let var_map = VarMap::<[u8; 8]>::open(dir_vm.path(), cfg).unwrap();
            write_seq(
                b,
                FULL_WRITE_BATCH,
                |i| {
                    var_map.put(&make_key(i), &make_var_value(i)).unwrap();
                },
                || var_map.flush_buffers().unwrap(),
            )
        });
    }

    group.finish();
}

// ---------------------------------------------------------------------------
// rand-read-hot — Bitcask collections, both modes
// ---------------------------------------------------------------------------

fn bench_rand_read_hot(c: &mut Criterion) {
    let mut group = c.benchmark_group("full/rand-read-hot");
    group.throughput(Throughput::Elements(READ_BATCH));

    for &direct_io in &MODES {
        group.bench_function(id_with_mode("const_tree", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir = TempDir::new().unwrap();
            let const_tree = ConstTree::<[u8; 8], 16>::open(dir.path(), cfg).unwrap();
            populate_const(&const_tree, FULL_NUM_ENTRIES);
            let mut rng = rand::rng();
            op_batch(b, READ_BATCH, || {
                black_box(const_tree.get(&make_key(rng.random_range(0..HOT_RANGE))));
            });
            const_tree.close().unwrap();
        });

        group.bench_function(id_with_mode("typed_tree", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_t = TempDir::new().unwrap();
            let typed_tree =
                TypedTree::<[u8; 8], Record, RapiraCodec>::open(dir_t.path(), cfg, RapiraCodec)
                    .unwrap();
            populate_typed(&typed_tree, FULL_NUM_ENTRIES);
            let mut rng = rand::rng();
            op_batch(b, READ_BATCH, || {
                black_box(typed_tree.get(&make_key(rng.random_range(0..HOT_RANGE))));
            });
            typed_tree.close().unwrap();
        });

        group.bench_function(id_with_mode("var_tree", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_v = TempDir::new().unwrap();
            let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), cfg).unwrap();
            populate_var(&var_tree, FULL_NUM_ENTRIES);
            let mut rng = rand::rng();
            op_batch(b, READ_BATCH, || {
                black_box(var_tree.get(&make_key(rng.random_range(0..HOT_RANGE))));
            });
            var_tree.close().unwrap();
        });

        group.bench_function(id_with_mode("const_map", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_cm = TempDir::new().unwrap();
            let const_map = ConstMap::<[u8; 8], 16>::open(dir_cm.path(), cfg).unwrap();
            populate_const_map(&const_map, FULL_NUM_ENTRIES);
            let mut rng = rand::rng();
            op_batch(b, READ_BATCH, || {
                black_box(const_map.get(&make_key(rng.random_range(0..HOT_RANGE))));
            });
            const_map.close().unwrap();
        });

        group.bench_function(id_with_mode("typed_map", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_tm = TempDir::new().unwrap();
            let typed_map =
                TypedMap::<[u8; 8], Record, RapiraCodec>::open(dir_tm.path(), cfg, RapiraCodec)
                    .unwrap();
            populate_typed_map(&typed_map, FULL_NUM_ENTRIES);
            let mut rng = rand::rng();
            op_batch(b, READ_BATCH, || {
                black_box(typed_map.get(&make_key(rng.random_range(0..HOT_RANGE))));
            });
            typed_map.close().unwrap();
        });

        group.bench_function(id_with_mode("var_map", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_vm = TempDir::new().unwrap();
            let var_map = VarMap::<[u8; 8]>::open(dir_vm.path(), cfg).unwrap();
            populate_var_map(&var_map, FULL_NUM_ENTRIES);
            let mut rng = rand::rng();
            op_batch(b, READ_BATCH, || {
                black_box(var_map.get(&make_key(rng.random_range(0..HOT_RANGE))));
            });
            var_map.close().unwrap();
        });
    }

    group.finish();
}

// ---------------------------------------------------------------------------
// FixedStore line — seq-write + rand-read-hot (no direct-io label)
// ---------------------------------------------------------------------------

fn bench_fixed(c: &mut Criterion) {
    // seq-write
    {
        let mut group = c.benchmark_group("full/seq-write");
        group.throughput(Throughput::Elements(FULL_WRITE_BATCH));
        group.sample_size(10);

        group.bench_function("fixed_tree", |b| {
            let dir_f = TempDir::new().unwrap();
            let fixed_tree = FixedTree::<[u8; 8], 16>::open(dir_f.path(), fixed_config()).unwrap();
            write_seq(
                b,
                FULL_WRITE_BATCH,
                |i| {
                    fixed_tree.put(&make_key(i), &make_const_value(i)).unwrap();
                },
                || {},
            )
        });

        group.bench_function("fixed_map", |b| {
            let dir_fm = TempDir::new().unwrap();
            let fixed_map = FixedMap::<[u8; 8], 16>::open(dir_fm.path(), fixed_config()).unwrap();
            write_seq(
                b,
                FULL_WRITE_BATCH,
                |i| {
                    fixed_map.put(&make_key(i), &make_const_value(i)).unwrap();
                },
                || {},
            )
        });

        group.finish();
    }

    // rand-read-hot
    {
        let mut group = c.benchmark_group("full/rand-read-hot");
        group.throughput(Throughput::Elements(READ_BATCH));

        group.bench_function("fixed_tree", |b| {
            let dir_f = TempDir::new().unwrap();
            let fixed_tree = FixedTree::<[u8; 8], 16>::open(dir_f.path(), fixed_config()).unwrap();
            populate_fixed_tree(&fixed_tree, FULL_NUM_ENTRIES);
            let mut rng = rand::rng();
            op_batch(b, READ_BATCH, || {
                black_box(fixed_tree.get(&make_key(rng.random_range(0..HOT_RANGE))));
            });
            fixed_tree.close().unwrap();
        });

        group.bench_function("fixed_map", |b| {
            let dir_fm = TempDir::new().unwrap();
            let fixed_map = FixedMap::<[u8; 8], 16>::open(dir_fm.path(), fixed_config()).unwrap();
            populate_fixed_map(&fixed_map, FULL_NUM_ENTRIES);
            let mut rng = rand::rng();
            op_batch(b, READ_BATCH, || {
                black_box(fixed_map.get(&make_key(rng.random_range(0..HOT_RANGE))));
            });
            fixed_map.close().unwrap();
        });

        group.finish();
    }
}

// ---------------------------------------------------------------------------
// rand-read-cold — Var collections only, working set >> block cache
// ---------------------------------------------------------------------------

fn bench_rand_read_cold(c: &mut Criterion) {
    let mut group = c.benchmark_group("full/rand-read-cold");
    group.throughput(Throughput::Elements(READ_BATCH));

    for &direct_io in &MODES {
        group.bench_function(id_with_mode("var_tree", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_v = TempDir::new().unwrap();
            let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), cfg).unwrap();
            populate_var(&var_tree, FULL_NUM_ENTRIES);
            let mut rng = rand::rng();
            op_batch(b, READ_BATCH, || {
                black_box(var_tree.get(&make_key(rng.random_range(0..FULL_NUM_ENTRIES))));
            });
            var_tree.close().unwrap();
        });

        group.bench_function(id_with_mode("var_map", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_vm = TempDir::new().unwrap();
            let var_map = VarMap::<[u8; 8]>::open(dir_vm.path(), cfg).unwrap();
            populate_var_map(&var_map, FULL_NUM_ENTRIES);
            let mut rng = rand::rng();
            op_batch(b, READ_BATCH, || {
                black_box(var_map.get(&make_key(rng.random_range(0..FULL_NUM_ENTRIES))));
            });
            var_map.close().unwrap();
        });
    }

    group.finish();
}

// ---------------------------------------------------------------------------
// Latency probes — diagnostic stdout, gated behind ARMDB_BENCH_LATENCY
// ---------------------------------------------------------------------------

const PROBE_OPS: usize = 50_000;

fn latency_probes(_c: &mut Criterion) {
    if std::env::var("ARMDB_BENCH_LATENCY").is_err() {
        return;
    }
    println!("[latency] running probes (per-op Instant overhead applies; relative tail only)");

    // ConstTree seq-write (page_cache)
    {
        let dir = TempDir::new().unwrap();
        let t = ConstTree::<[u8; 8], 16>::open(dir.path(), bitcask_config(false)).unwrap();
        let mut p = Probe::with_capacity(PROBE_OPS);
        for i in 0..PROBE_OPS as u64 {
            p.record(|| t.put(&make_key(i), &make_const_value(i)).unwrap());
        }
        t.flush_buffers().unwrap();
        p.report("const_tree/page_cache/seq-write");
        t.close().unwrap();
    }

    // VarTree rand-read-hot (page_cache)
    {
        let dir = TempDir::new().unwrap();
        let t = VarTree::<[u8; 8]>::open(dir.path(), bitcask_config(false)).unwrap();
        populate_var(&t, FULL_NUM_ENTRIES);
        let mut rng = rand::rng();
        let mut p = Probe::with_capacity(PROBE_OPS);
        for _ in 0..PROBE_OPS {
            let k = make_key(rng.random_range(0..HOT_RANGE));
            p.record(|| black_box(t.get(&k)));
        }
        p.report("var_tree/page_cache/rand-read-hot");
        t.close().unwrap();
    }

    // VarTree rand-read-cold (direct_io — the only truly device-cold path)
    {
        let dir = TempDir::new().unwrap();
        let t = VarTree::<[u8; 8]>::open(dir.path(), bitcask_config(true)).unwrap();
        populate_var(&t, FULL_NUM_ENTRIES);
        let mut rng = rand::rng();
        let mut p = Probe::with_capacity(PROBE_OPS);
        for _ in 0..PROBE_OPS {
            let k = make_key(rng.random_range(0..FULL_NUM_ENTRIES));
            p.record(|| black_box(t.get(&k)));
        }
        p.report("var_tree/direct_io/rand-read-cold");
        t.close().unwrap();
    }

    // ConstTree overwrite (page_cache)
    {
        let dir = TempDir::new().unwrap();
        let t = ConstTree::<[u8; 8], 16>::open(dir.path(), bitcask_config(false)).unwrap();
        populate_const(&t, FULL_NUM_ENTRIES);
        let mut p = Probe::with_capacity(PROBE_OPS);
        for i in 0..PROBE_OPS as u64 {
            let k = make_key(i % FULL_NUM_ENTRIES);
            p.record(|| t.put(&k, &make_const_value(i)).unwrap());
        }
        p.report("const_tree/page_cache/overwrite");
        t.close().unwrap();
    }

    // ConstTree read-write-mix (page_cache) — per-op, labeled 80r/20w
    {
        let dir = TempDir::new().unwrap();
        let t = ConstTree::<[u8; 8], 16>::open(dir.path(), bitcask_config(false)).unwrap();
        populate_const(&t, FULL_NUM_ENTRIES);
        let mut rng = rand::rng();
        let mut p = Probe::with_capacity(PROBE_OPS);
        for i in 0..PROBE_OPS as u64 {
            if i % 5 == 4 {
                let k = make_key(i % FULL_NUM_ENTRIES);
                p.record(|| t.put(&k, &make_const_value(FULL_NUM_ENTRIES + i)).unwrap());
            } else {
                let k = make_key(rng.random_range(0..FULL_NUM_ENTRIES));
                p.record(|| black_box(t.get(&k)));
            }
        }
        p.report("const_tree/page_cache/read-write-mix(80r/20w)");
        t.close().unwrap();
    }
}

// ---------------------------------------------------------------------------
// rand-write — Bitcask 6 × 2 modes + FixedStore 2 (long-lived, precomputed keys)
// ---------------------------------------------------------------------------

/// Precomputed key list replayed identically each Criterion iteration
/// (access pattern is fixed, not re-seeded per iteration).
fn rand_keys(n: u64) -> Vec<u64> {
    let mut rng = rand::rng();
    (0..n).map(|_| rng.random()).collect()
}

fn bench_rand_write(c: &mut Criterion) {
    let mut group = c.benchmark_group("full/rand-write");
    group.throughput(Throughput::Elements(FULL_WRITE_BATCH));
    group.sample_size(10);

    for &direct_io in &MODES {
        group.bench_function(id_with_mode("const_tree", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            // Precomputed key list replayed identically each Criterion iteration
            // (access pattern is fixed, not re-seeded per iteration).
            let keys = rand_keys(FULL_WRITE_BATCH);
            let dir = TempDir::new().unwrap();
            let const_tree = ConstTree::<[u8; 8], 16>::open(dir.path(), cfg).unwrap();
            write_rand(
                b,
                &keys,
                |k| {
                    const_tree.put(&make_key(k), &make_const_value(k)).unwrap();
                },
                || const_tree.flush_buffers().unwrap(),
            )
        });

        group.bench_function(id_with_mode("typed_tree", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let keys = rand_keys(FULL_WRITE_BATCH);
            let dir_t = TempDir::new().unwrap();
            let typed_tree =
                TypedTree::<[u8; 8], Record, RapiraCodec>::open(dir_t.path(), cfg, RapiraCodec)
                    .unwrap();
            write_rand(
                b,
                &keys,
                |k| {
                    typed_tree.put(&make_key(k), make_record(k)).unwrap();
                },
                || typed_tree.flush_buffers().unwrap(),
            )
        });

        group.bench_function(id_with_mode("var_tree", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let keys = rand_keys(FULL_WRITE_BATCH);
            let dir_v = TempDir::new().unwrap();
            let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), cfg).unwrap();
            write_rand(
                b,
                &keys,
                |k| {
                    var_tree.put(&make_key(k), &make_var_value(k)).unwrap();
                },
                || var_tree.flush_buffers().unwrap(),
            )
        });

        group.bench_function(id_with_mode("const_map", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let keys = rand_keys(FULL_WRITE_BATCH);
            let dir_cm = TempDir::new().unwrap();
            let const_map = ConstMap::<[u8; 8], 16>::open(dir_cm.path(), cfg).unwrap();
            write_rand(
                b,
                &keys,
                |k| {
                    const_map.put(&make_key(k), &make_const_value(k)).unwrap();
                },
                || const_map.flush_buffers().unwrap(),
            )
        });

        group.bench_function(id_with_mode("typed_map", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let keys = rand_keys(FULL_WRITE_BATCH);
            let dir_tm = TempDir::new().unwrap();
            let typed_map =
                TypedMap::<[u8; 8], Record, RapiraCodec>::open(dir_tm.path(), cfg, RapiraCodec)
                    .unwrap();
            write_rand(
                b,
                &keys,
                |k| {
                    typed_map.put(&make_key(k), make_record(k)).unwrap();
                },
                || typed_map.flush_buffers().unwrap(),
            )
        });

        group.bench_function(id_with_mode("var_map", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let keys = rand_keys(FULL_WRITE_BATCH);
            let dir_vm = TempDir::new().unwrap();
            let var_map = VarMap::<[u8; 8]>::open(dir_vm.path(), cfg).unwrap();
            write_rand(
                b,
                &keys,
                |k| {
                    var_map.put(&make_key(k), &make_var_value(k)).unwrap();
                },
                || var_map.flush_buffers().unwrap(),
            )
        });
    }

    // FixedStore — no mode label, no-op sync
    group.bench_function("fixed_tree", |b| {
        let keys = rand_keys(FULL_WRITE_BATCH);
        let dir_f = TempDir::new().unwrap();
        let fixed_tree = FixedTree::<[u8; 8], 16>::open(dir_f.path(), fixed_config()).unwrap();
        write_rand(
            b,
            &keys,
            |k| {
                fixed_tree.put(&make_key(k), &make_const_value(k)).unwrap();
            },
            || {},
        )
    });

    group.bench_function("fixed_map", |b| {
        let keys = rand_keys(FULL_WRITE_BATCH);
        let dir_fm = TempDir::new().unwrap();
        let fixed_map = FixedMap::<[u8; 8], 16>::open(dir_fm.path(), fixed_config()).unwrap();
        write_rand(
            b,
            &keys,
            |k| {
                fixed_map.put(&make_key(k), &make_const_value(k)).unwrap();
            },
            || {},
        )
    });

    group.finish();
}

// ---------------------------------------------------------------------------
// seq-read — Bitcask 6 × 2 modes + FixedStore 2
// ---------------------------------------------------------------------------

fn bench_seq_read(c: &mut Criterion) {
    let mut group = c.benchmark_group("full/seq-read");
    group.throughput(Throughput::Elements(READ_BATCH));

    for &direct_io in &MODES {
        group.bench_function(id_with_mode("const_tree", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir = TempDir::new().unwrap();
            let const_tree = ConstTree::<[u8; 8], 16>::open(dir.path(), cfg).unwrap();
            populate_const(&const_tree, FULL_NUM_ENTRIES);
            read_seq(b, READ_BATCH, |i| {
                black_box(const_tree.get(&make_key(i % FULL_NUM_ENTRIES)));
            });
            const_tree.close().unwrap();
        });

        group.bench_function(id_with_mode("typed_tree", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_t = TempDir::new().unwrap();
            let typed_tree =
                TypedTree::<[u8; 8], Record, RapiraCodec>::open(dir_t.path(), cfg, RapiraCodec)
                    .unwrap();
            populate_typed(&typed_tree, FULL_NUM_ENTRIES);
            read_seq(b, READ_BATCH, |i| {
                black_box(typed_tree.get(&make_key(i % FULL_NUM_ENTRIES)));
            });
            typed_tree.close().unwrap();
        });

        group.bench_function(id_with_mode("var_tree", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_v = TempDir::new().unwrap();
            let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), cfg).unwrap();
            populate_var(&var_tree, FULL_NUM_ENTRIES);
            read_seq(b, READ_BATCH, |i| {
                black_box(var_tree.get(&make_key(i % FULL_NUM_ENTRIES)));
            });
            var_tree.close().unwrap();
        });

        group.bench_function(id_with_mode("const_map", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_cm = TempDir::new().unwrap();
            let const_map = ConstMap::<[u8; 8], 16>::open(dir_cm.path(), cfg).unwrap();
            populate_const_map(&const_map, FULL_NUM_ENTRIES);
            read_seq(b, READ_BATCH, |i| {
                black_box(const_map.get(&make_key(i % FULL_NUM_ENTRIES)));
            });
            const_map.close().unwrap();
        });

        group.bench_function(id_with_mode("typed_map", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_tm = TempDir::new().unwrap();
            let typed_map =
                TypedMap::<[u8; 8], Record, RapiraCodec>::open(dir_tm.path(), cfg, RapiraCodec)
                    .unwrap();
            populate_typed_map(&typed_map, FULL_NUM_ENTRIES);
            read_seq(b, READ_BATCH, |i| {
                black_box(typed_map.get(&make_key(i % FULL_NUM_ENTRIES)));
            });
            typed_map.close().unwrap();
        });

        group.bench_function(id_with_mode("var_map", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_vm = TempDir::new().unwrap();
            let var_map = VarMap::<[u8; 8]>::open(dir_vm.path(), cfg).unwrap();
            populate_var_map(&var_map, FULL_NUM_ENTRIES);
            read_seq(b, READ_BATCH, |i| {
                black_box(var_map.get(&make_key(i % FULL_NUM_ENTRIES)));
            });
            var_map.close().unwrap();
        });
    }

    // FixedStore — no mode label
    group.bench_function("fixed_tree", |b| {
        let dir_f = TempDir::new().unwrap();
        let fixed_tree = FixedTree::<[u8; 8], 16>::open(dir_f.path(), fixed_config()).unwrap();
        populate_fixed_tree(&fixed_tree, FULL_NUM_ENTRIES);
        read_seq(b, READ_BATCH, |i| {
            black_box(fixed_tree.get(&make_key(i % FULL_NUM_ENTRIES)));
        });
        fixed_tree.close().unwrap();
    });

    group.bench_function("fixed_map", |b| {
        let dir_fm = TempDir::new().unwrap();
        let fixed_map = FixedMap::<[u8; 8], 16>::open(dir_fm.path(), fixed_config()).unwrap();
        populate_fixed_map(&fixed_map, FULL_NUM_ENTRIES);
        read_seq(b, READ_BATCH, |i| {
            black_box(fixed_map.get(&make_key(i % FULL_NUM_ENTRIES)));
        });
        fixed_map.close().unwrap();
    });

    group.finish();
}

// ---------------------------------------------------------------------------
// read-write-mix — 80% read / 20% write, Bitcask 6 × 2 modes + FixedStore 2
// ---------------------------------------------------------------------------

fn bench_read_write_mix(c: &mut Criterion) {
    let mut group = c.benchmark_group("full/read-write-mix");
    group.throughput(Throughput::Elements(READ_BATCH));
    group.sample_size(10);

    for &direct_io in &MODES {
        group.bench_function(
            format!("{}/80r_20w", id_with_mode("const_tree", direct_io)),
            |b| {
                let cfg = bitcask_config(direct_io);
                let dir = TempDir::new().unwrap();
                let const_tree = ConstTree::<[u8; 8], 16>::open(dir.path(), cfg).unwrap();
                populate_const(&const_tree, FULL_NUM_ENTRIES);
                let mut rng = rand::rng();
                mix_batch(
                    b,
                    800,
                    200,
                    || {
                        black_box(const_tree.get(&make_key(rng.random_range(0..FULL_NUM_ENTRIES))));
                    },
                    |s| {
                        const_tree
                            .put(
                                &make_key(s % FULL_NUM_ENTRIES),
                                &make_const_value(FULL_NUM_ENTRIES + s),
                            )
                            .unwrap();
                    },
                );
                const_tree.close().unwrap();
            },
        );

        group.bench_function(
            format!("{}/80r_20w", id_with_mode("typed_tree", direct_io)),
            |b| {
                let cfg = bitcask_config(direct_io);
                let dir_t = TempDir::new().unwrap();
                let typed_tree =
                    TypedTree::<[u8; 8], Record, RapiraCodec>::open(dir_t.path(), cfg, RapiraCodec)
                        .unwrap();
                populate_typed(&typed_tree, FULL_NUM_ENTRIES);
                let mut rng = rand::rng();
                mix_batch(
                    b,
                    800,
                    200,
                    || {
                        black_box(typed_tree.get(&make_key(rng.random_range(0..FULL_NUM_ENTRIES))));
                    },
                    |s| {
                        typed_tree
                            .put(
                                &make_key(s % FULL_NUM_ENTRIES),
                                make_record(FULL_NUM_ENTRIES + s),
                            )
                            .unwrap();
                    },
                );
                typed_tree.close().unwrap();
            },
        );

        group.bench_function(
            format!("{}/80r_20w", id_with_mode("var_tree", direct_io)),
            |b| {
                let cfg = bitcask_config(direct_io);
                let dir_v = TempDir::new().unwrap();
                let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), cfg).unwrap();
                populate_var(&var_tree, FULL_NUM_ENTRIES);
                let mut rng = rand::rng();
                mix_batch(
                    b,
                    800,
                    200,
                    || {
                        black_box(var_tree.get(&make_key(rng.random_range(0..FULL_NUM_ENTRIES))));
                    },
                    |s| {
                        var_tree
                            .put(
                                &make_key(s % FULL_NUM_ENTRIES),
                                &make_var_value(FULL_NUM_ENTRIES + s),
                            )
                            .unwrap();
                    },
                );
                var_tree.close().unwrap();
            },
        );

        group.bench_function(
            format!("{}/80r_20w", id_with_mode("const_map", direct_io)),
            |b| {
                let cfg = bitcask_config(direct_io);
                let dir_cm = TempDir::new().unwrap();
                let const_map = ConstMap::<[u8; 8], 16>::open(dir_cm.path(), cfg).unwrap();
                populate_const_map(&const_map, FULL_NUM_ENTRIES);
                let mut rng = rand::rng();
                mix_batch(
                    b,
                    800,
                    200,
                    || {
                        black_box(const_map.get(&make_key(rng.random_range(0..FULL_NUM_ENTRIES))));
                    },
                    |s| {
                        const_map
                            .put(
                                &make_key(s % FULL_NUM_ENTRIES),
                                &make_const_value(FULL_NUM_ENTRIES + s),
                            )
                            .unwrap();
                    },
                );
                const_map.close().unwrap();
            },
        );

        group.bench_function(
            format!("{}/80r_20w", id_with_mode("typed_map", direct_io)),
            |b| {
                let cfg = bitcask_config(direct_io);
                let dir_tm = TempDir::new().unwrap();
                let typed_map =
                    TypedMap::<[u8; 8], Record, RapiraCodec>::open(dir_tm.path(), cfg, RapiraCodec)
                        .unwrap();
                populate_typed_map(&typed_map, FULL_NUM_ENTRIES);
                let mut rng = rand::rng();
                mix_batch(
                    b,
                    800,
                    200,
                    || {
                        black_box(typed_map.get(&make_key(rng.random_range(0..FULL_NUM_ENTRIES))));
                    },
                    |s| {
                        typed_map
                            .put(
                                &make_key(s % FULL_NUM_ENTRIES),
                                make_record(FULL_NUM_ENTRIES + s),
                            )
                            .unwrap();
                    },
                );
                typed_map.close().unwrap();
            },
        );

        group.bench_function(
            format!("{}/80r_20w", id_with_mode("var_map", direct_io)),
            |b| {
                let cfg = bitcask_config(direct_io);
                let dir_vm = TempDir::new().unwrap();
                let var_map = VarMap::<[u8; 8]>::open(dir_vm.path(), cfg).unwrap();
                populate_var_map(&var_map, FULL_NUM_ENTRIES);
                let mut rng = rand::rng();
                mix_batch(
                    b,
                    800,
                    200,
                    || {
                        black_box(var_map.get(&make_key(rng.random_range(0..FULL_NUM_ENTRIES))));
                    },
                    |s| {
                        var_map
                            .put(
                                &make_key(s % FULL_NUM_ENTRIES),
                                &make_var_value(FULL_NUM_ENTRIES + s),
                            )
                            .unwrap();
                    },
                );
                var_map.close().unwrap();
            },
        );
    }

    // FixedStore — no mode label
    group.bench_function("fixed_tree/80r_20w", |b| {
        let dir_f = TempDir::new().unwrap();
        let fixed_tree = FixedTree::<[u8; 8], 16>::open(dir_f.path(), fixed_config()).unwrap();
        populate_fixed_tree(&fixed_tree, FULL_NUM_ENTRIES);
        let mut rng = rand::rng();
        mix_batch(
            b,
            800,
            200,
            || {
                black_box(fixed_tree.get(&make_key(rng.random_range(0..FULL_NUM_ENTRIES))));
            },
            |s| {
                fixed_tree
                    .put(
                        &make_key(s % FULL_NUM_ENTRIES),
                        &make_const_value(FULL_NUM_ENTRIES + s),
                    )
                    .unwrap();
            },
        );
        fixed_tree.close().unwrap();
    });

    group.bench_function("fixed_map/80r_20w", |b| {
        let dir_fm = TempDir::new().unwrap();
        let fixed_map = FixedMap::<[u8; 8], 16>::open(dir_fm.path(), fixed_config()).unwrap();
        populate_fixed_map(&fixed_map, FULL_NUM_ENTRIES);
        let mut rng = rand::rng();
        mix_batch(
            b,
            800,
            200,
            || {
                black_box(fixed_map.get(&make_key(rng.random_range(0..FULL_NUM_ENTRIES))));
            },
            |s| {
                fixed_map
                    .put(
                        &make_key(s % FULL_NUM_ENTRIES),
                        &make_const_value(FULL_NUM_ENTRIES + s),
                    )
                    .unwrap();
            },
        );
        fixed_map.close().unwrap();
    });

    group.finish();
}

// ---------------------------------------------------------------------------
// overwrite — re-put existing keys, Bitcask 6 × 2 modes + FixedStore 2
// ---------------------------------------------------------------------------

fn bench_overwrite(c: &mut Criterion) {
    let mut group = c.benchmark_group("full/overwrite");
    group.throughput(Throughput::Elements(READ_BATCH));
    group.sample_size(10);

    for &direct_io in &MODES {
        group.bench_function(id_with_mode("const_tree", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir = TempDir::new().unwrap();
            let const_tree = ConstTree::<[u8; 8], 16>::open(dir.path(), cfg).unwrap();
            populate_const(&const_tree, FULL_NUM_ENTRIES);
            let mut ctr = 0u64;
            op_batch(b, READ_BATCH, || {
                const_tree
                    .put(&make_key(ctr % FULL_NUM_ENTRIES), &make_const_value(ctr))
                    .unwrap();
                ctr += 1;
            });
            const_tree.close().unwrap();
        });

        group.bench_function(id_with_mode("typed_tree", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_t = TempDir::new().unwrap();
            let typed_tree =
                TypedTree::<[u8; 8], Record, RapiraCodec>::open(dir_t.path(), cfg, RapiraCodec)
                    .unwrap();
            populate_typed(&typed_tree, FULL_NUM_ENTRIES);
            let mut ctr = 0u64;
            op_batch(b, READ_BATCH, || {
                typed_tree
                    .put(&make_key(ctr % FULL_NUM_ENTRIES), make_record(ctr))
                    .unwrap();
                ctr += 1;
            });
            typed_tree.close().unwrap();
        });

        group.bench_function(id_with_mode("var_tree", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_v = TempDir::new().unwrap();
            let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), cfg).unwrap();
            populate_var(&var_tree, FULL_NUM_ENTRIES);
            let mut ctr = 0u64;
            op_batch(b, READ_BATCH, || {
                var_tree
                    .put(&make_key(ctr % FULL_NUM_ENTRIES), &make_var_value(ctr))
                    .unwrap();
                ctr += 1;
            });
            var_tree.close().unwrap();
        });

        group.bench_function(id_with_mode("const_map", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_cm = TempDir::new().unwrap();
            let const_map = ConstMap::<[u8; 8], 16>::open(dir_cm.path(), cfg).unwrap();
            populate_const_map(&const_map, FULL_NUM_ENTRIES);
            let mut ctr = 0u64;
            op_batch(b, READ_BATCH, || {
                const_map
                    .put(&make_key(ctr % FULL_NUM_ENTRIES), &make_const_value(ctr))
                    .unwrap();
                ctr += 1;
            });
            const_map.close().unwrap();
        });

        group.bench_function(id_with_mode("typed_map", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_tm = TempDir::new().unwrap();
            let typed_map =
                TypedMap::<[u8; 8], Record, RapiraCodec>::open(dir_tm.path(), cfg, RapiraCodec)
                    .unwrap();
            populate_typed_map(&typed_map, FULL_NUM_ENTRIES);
            let mut ctr = 0u64;
            op_batch(b, READ_BATCH, || {
                typed_map
                    .put(&make_key(ctr % FULL_NUM_ENTRIES), make_record(ctr))
                    .unwrap();
                ctr += 1;
            });
            typed_map.close().unwrap();
        });

        group.bench_function(id_with_mode("var_map", direct_io), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_vm = TempDir::new().unwrap();
            let var_map = VarMap::<[u8; 8]>::open(dir_vm.path(), cfg).unwrap();
            populate_var_map(&var_map, FULL_NUM_ENTRIES);
            let mut ctr = 0u64;
            op_batch(b, READ_BATCH, || {
                var_map
                    .put(&make_key(ctr % FULL_NUM_ENTRIES), &make_var_value(ctr))
                    .unwrap();
                ctr += 1;
            });
            var_map.close().unwrap();
        });
    }

    // FixedStore — no mode label
    group.bench_function("fixed_tree", |b| {
        let dir_f = TempDir::new().unwrap();
        let fixed_tree = FixedTree::<[u8; 8], 16>::open(dir_f.path(), fixed_config()).unwrap();
        populate_fixed_tree(&fixed_tree, FULL_NUM_ENTRIES);
        let mut ctr = 0u64;
        op_batch(b, READ_BATCH, || {
            fixed_tree
                .put(&make_key(ctr % FULL_NUM_ENTRIES), &make_const_value(ctr))
                .unwrap();
            ctr += 1;
        });
        fixed_tree.close().unwrap();
    });

    group.bench_function("fixed_map", |b| {
        let dir_fm = TempDir::new().unwrap();
        let fixed_map = FixedMap::<[u8; 8], 16>::open(dir_fm.path(), fixed_config()).unwrap();
        populate_fixed_map(&fixed_map, FULL_NUM_ENTRIES);
        let mut ctr = 0u64;
        op_batch(b, READ_BATCH, || {
            fixed_map
                .put(&make_key(ctr % FULL_NUM_ENTRIES), &make_const_value(ctr))
                .unwrap();
            ctr += 1;
        });
        fixed_map.close().unwrap();
    });

    group.finish();
}

// ---------------------------------------------------------------------------
// scan — tree collections only: Bitcask 3 × 2 modes + FixedTree (no mode)
// ---------------------------------------------------------------------------

fn bench_scan(c: &mut Criterion) {
    use common::data::make_group_key;

    let mut group = c.benchmark_group("full/scan");
    group.throughput(Throughput::Elements(SCAN_TAKE as u64));

    // Each bench_function is self-contained: populate and close happen inside
    // the closure. On a full run the grouped dataset is populated twice per
    // tree (once for prefix_take_100, once for iter_take_100). That is the
    // accepted cost of filter-correctness — skipped benches do zero setup.

    for &direct_io in &MODES {
        let id = id_with_mode("const_tree", direct_io);

        group.bench_function(format!("{id}/prefix_take_100"), |b| {
            let cfg = bitcask_config(direct_io);
            let dir = TempDir::new().unwrap();
            let const_tree = ConstTree::<[u8; 8], 16>::open(dir.path(), cfg).unwrap();
            // Use populate_const_grouped helper (DRY: avoids inline for-loop).
            populate_const_grouped(&const_tree, NUM_GROUPS, SCAN_TAKE as u32);
            let mut rng = rand::rng();
            op_batch(b, 1, || {
                let prefix = rng.random_range(0..NUM_GROUPS).to_be_bytes();
                black_box(
                    const_tree
                        .prefix_iter(&prefix)
                        .take(SCAN_TAKE)
                        .collect::<Vec<_>>(),
                );
            });
            const_tree.close().unwrap();
        });

        group.bench_function(format!("{id}/iter_take_100"), |b| {
            let cfg = bitcask_config(direct_io);
            let dir = TempDir::new().unwrap();
            let const_tree = ConstTree::<[u8; 8], 16>::open(dir.path(), cfg).unwrap();
            // Use populate_const_grouped helper (DRY: avoids inline for-loop).
            populate_const_grouped(&const_tree, NUM_GROUPS, SCAN_TAKE as u32);
            op_batch(b, 1, || {
                black_box(const_tree.iter().take(SCAN_TAKE).collect::<Vec<_>>());
            });
            const_tree.close().unwrap();
        });

        let id = id_with_mode("typed_tree", direct_io);
        group.bench_function(format!("{id}/prefix_take_100"), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_t = TempDir::new().unwrap();
            let typed_tree =
                TypedTree::<[u8; 8], Record, RapiraCodec>::open(dir_t.path(), cfg, RapiraCodec)
                    .unwrap();
            for g in 0..NUM_GROUPS {
                for id_in_group in 0..SCAN_TAKE as u32 {
                    typed_tree
                        .put(
                            &make_group_key(g, id_in_group),
                            make_record(g as u64 * 1000 + id_in_group as u64),
                        )
                        .unwrap();
                }
            }
            typed_tree.flush_buffers().unwrap();
            let mut rng = rand::rng();
            op_batch(b, 1, || {
                let prefix = rng.random_range(0..NUM_GROUPS).to_be_bytes();
                black_box(
                    typed_tree
                        .prefix_iter(&prefix)
                        .take(SCAN_TAKE)
                        .collect::<Vec<_>>(),
                );
            });
            typed_tree.close().unwrap();
        });

        group.bench_function(format!("{id}/iter_take_100"), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_t = TempDir::new().unwrap();
            let typed_tree =
                TypedTree::<[u8; 8], Record, RapiraCodec>::open(dir_t.path(), cfg, RapiraCodec)
                    .unwrap();
            for g in 0..NUM_GROUPS {
                for id_in_group in 0..SCAN_TAKE as u32 {
                    typed_tree
                        .put(
                            &make_group_key(g, id_in_group),
                            make_record(g as u64 * 1000 + id_in_group as u64),
                        )
                        .unwrap();
                }
            }
            typed_tree.flush_buffers().unwrap();
            op_batch(b, 1, || {
                black_box(typed_tree.iter().take(SCAN_TAKE).collect::<Vec<_>>());
            });
            typed_tree.close().unwrap();
        });

        let id = id_with_mode("var_tree", direct_io);
        group.bench_function(format!("{id}/prefix_take_100"), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_v = TempDir::new().unwrap();
            let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), cfg).unwrap();
            for g in 0..NUM_GROUPS {
                for id_in_group in 0..SCAN_TAKE as u32 {
                    var_tree
                        .put(
                            &make_group_key(g, id_in_group),
                            &make_var_value(g as u64 * 1000 + id_in_group as u64),
                        )
                        .unwrap();
                }
            }
            var_tree.flush_buffers().unwrap();
            let mut rng = rand::rng();
            op_batch(b, 1, || {
                let prefix = rng.random_range(0..NUM_GROUPS).to_be_bytes();
                black_box(
                    var_tree
                        .prefix_iter(&prefix)
                        .take(SCAN_TAKE)
                        .collect::<Vec<_>>(),
                );
            });
            var_tree.close().unwrap();
        });

        group.bench_function(format!("{id}/iter_take_100"), |b| {
            let cfg = bitcask_config(direct_io);
            let dir_v = TempDir::new().unwrap();
            let var_tree = VarTree::<[u8; 8]>::open(dir_v.path(), cfg).unwrap();
            for g in 0..NUM_GROUPS {
                for id_in_group in 0..SCAN_TAKE as u32 {
                    var_tree
                        .put(
                            &make_group_key(g, id_in_group),
                            &make_var_value(g as u64 * 1000 + id_in_group as u64),
                        )
                        .unwrap();
                }
            }
            var_tree.flush_buffers().unwrap();
            op_batch(b, 1, || {
                black_box(var_tree.iter().take(SCAN_TAKE).collect::<Vec<_>>());
            });
            var_tree.close().unwrap();
        });
    }

    // FixedTree — no mode label
    group.bench_function("fixed_tree/prefix_take_100", |b| {
        let dir_f = TempDir::new().unwrap();
        let fixed_tree = FixedTree::<[u8; 8], 16>::open(dir_f.path(), fixed_config()).unwrap();
        for g in 0..NUM_GROUPS {
            for id_in_group in 0..SCAN_TAKE as u32 {
                fixed_tree
                    .put(
                        &make_group_key(g, id_in_group),
                        &make_const_value(g as u64 * 1000 + id_in_group as u64),
                    )
                    .unwrap();
            }
        }
        let mut rng = rand::rng();
        op_batch(b, 1, || {
            let prefix = rng.random_range(0..NUM_GROUPS).to_be_bytes();
            black_box(
                fixed_tree
                    .prefix_iter(&prefix)
                    .take(SCAN_TAKE)
                    .collect::<Vec<_>>(),
            );
        });
        fixed_tree.close().unwrap();
    });

    group.bench_function("fixed_tree/iter_take_100", |b| {
        let dir_f = TempDir::new().unwrap();
        let fixed_tree = FixedTree::<[u8; 8], 16>::open(dir_f.path(), fixed_config()).unwrap();
        for g in 0..NUM_GROUPS {
            for id_in_group in 0..SCAN_TAKE as u32 {
                fixed_tree
                    .put(
                        &make_group_key(g, id_in_group),
                        &make_const_value(g as u64 * 1000 + id_in_group as u64),
                    )
                    .unwrap();
            }
        }
        op_batch(b, 1, || {
            black_box(fixed_tree.iter().take(SCAN_TAKE).collect::<Vec<_>>());
        });
        fixed_tree.close().unwrap();
    });

    group.finish();
}

criterion_group! {
    name = benches;
    config = Criterion::default()
        .measurement_time(Duration::from_secs(5))
        .warm_up_time(Duration::from_secs(2));
    targets = bench_seq_write, bench_rand_write, bench_seq_read, bench_rand_read_hot,
        bench_rand_read_cold, bench_read_write_mix, bench_overwrite, bench_scan, bench_fixed,
        latency_probes,
}

criterion_main!(benches);
