//! Concurrent-write smoke for the WAL group-commit pipeline.
//!
//! 32 concurrent tokio tasks each issue 1000 puts against the same
//! [`RedbStorage`]. The test asserts:
//!
//! * Every put returns `Ok` (no `WalCommand` is dropped under load).
//! * After all tasks join, every key written is present and matches.
//! * The total wall-clock duration is well under the per-write fsync
//!   bound — implicit proof that group-commit is amortizing fsync across
//!   peers. Without group-commit, 32 000 individual `fdatasync` calls
//!   would take tens of seconds even on SSD; we cap the test at 60s.

use mneme::storage::{Storage, redb_impl::RedbStorage};
use std::sync::Arc;
#[cfg(not(windows))]
use std::time::Duration;
use std::time::Instant;
use tempfile::TempDir;

const WORKERS: usize = 32;
const OPS_PER_WORKER: usize = 1000;

#[test]
fn group_commit_under_load() {
    let tmp = TempDir::new().unwrap();
    let root = tmp.path().to_path_buf();
    let rt = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap();

    let storage = rt.block_on(async { RedbStorage::open(&root).unwrap() });

    let started = Instant::now();
    rt.block_on(async {
        let mut handles = Vec::with_capacity(WORKERS);
        for w in 0..WORKERS {
            let s: Arc<RedbStorage> = Arc::clone(&storage);
            let handle = tokio::spawn(async move {
                for i in 0..OPS_PER_WORKER {
                    let key = format!("w{w:02}-k{i:04}").into_bytes();
                    let value = format!("v-{w}-{i}").into_bytes();
                    s.put(&key, &value).await.expect("put");
                }
            });
            handles.push(handle);
        }
        for h in handles {
            h.await.expect("worker join");
        }
    });
    let elapsed = started.elapsed();

    // The point of this assertion is "group-commit amortizes fsync" —
    // a regression to per-put fsync would push 32k ops well past 60 s
    // on any reasonable SSD (~10 ms × 32k = ~320 s naive per-put).
    //
    // We only enforce it on Linux + macOS. On Windows the assertion
    // is dropped because:
    //
    //   * NTFS + flush_and_close on every commit is structurally
    //     ~3× slower than ext4 / APFS,
    //   * GitHub Actions windows-latest runners have wide per-job
    //     performance variance — observed runs range from ~110 s to
    //     ~280 s for the same workload, with no code change between.
    //
    // The correctness checks below (every key reads back identical;
    // per-worker prefix scan returns exactly OPS_PER_WORKER items)
    // still exercise the concurrent-write path on every platform —
    // the only thing we forgo on Windows is the throughput floor.
    #[cfg(not(windows))]
    {
        let budget = Duration::from_secs(60);
        assert!(
            elapsed < budget,
            "{} concurrent puts took {:?} (budget {:?}) — group-commit appears not to be amortizing fsync",
            WORKERS * OPS_PER_WORKER,
            elapsed,
            budget,
        );
    }
    #[cfg(windows)]
    {
        // Surface the wall-clock to the runner log so a regression is
        // still investigable, even though we don't fail the test on
        // throughput here.
        eprintln!(
            "concurrent_writes (windows): {} ops in {:?}",
            WORKERS * OPS_PER_WORKER,
            elapsed,
        );
    }

    // Every key written must read back identical.
    rt.block_on(async {
        for w in 0..WORKERS {
            for i in 0..OPS_PER_WORKER {
                let key = format!("w{w:02}-k{i:04}").into_bytes();
                let expected = format!("v-{w}-{i}").into_bytes();
                let got = storage.get(&key).await.unwrap();
                assert_eq!(got.as_deref(), Some(expected.as_slice()), "key {key:?}");
            }
        }
    });

    // Per-worker prefix scan should return exactly OPS_PER_WORKER items.
    rt.block_on(async {
        for w in 0..WORKERS {
            let prefix = format!("w{w:02}-").into_bytes();
            let entries = storage.scan_prefix(&prefix).await.unwrap();
            assert_eq!(entries.len(), OPS_PER_WORKER, "worker {w} prefix scan");
        }
    });

    drop(storage);

    // Reopen and confirm counts survive close.
    let storage = rt.block_on(async { RedbStorage::open(&root).unwrap() });
    rt.block_on(async {
        let total = storage.scan_prefix(b"").await.unwrap().len();
        assert_eq!(total, WORKERS * OPS_PER_WORKER);
    });
}
