use super::{
    AedbConfig, AedbError, AedbInstance, ColumnDef, ColumnType, ConsistencyMode, DdlOperation,
    DurabilityMode, Expr, Mutation, Query, RecoveryMode, Row, StorageMode, TransactionEnvelope,
    Value, WriteClass, WriteIntent, reclaim_eligible_wal_segments,
};
use crate::query::plan::QueryOptions;
use std::fs;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::{Duration, Instant};
use tempfile::tempdir;

#[test]
fn checkpoint_compression_level_is_validated() {
    let cfg = AedbConfig {
        checkpoint_compression_level: 23,
        ..AedbConfig::default()
    };
    let err = crate::lib_helpers::validate_config(&cfg)
        .expect_err("out-of-range compression level must be rejected");
    assert!(matches!(err, AedbError::InvalidConfig { .. }));

    let cfg = AedbConfig {
        checkpoint_compression_level: 1,
        ..AedbConfig::default()
    };
    crate::lib_helpers::validate_config(&cfg).expect("valid compression level");
}

#[tokio::test]
async fn disk_backed_checkpoint_is_self_contained_when_value_file_is_missing() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        storage_mode: StorageMode::DiskBacked,
        persistent_value_inline_threshold_bytes: 16,
        persistent_value_hot_cache_bytes: 0,
        max_kv_value_bytes: 256 * 1024,
        max_transaction_bytes: 512 * 1024,
        ..AedbConfig::default()
    };
    let value: Vec<u8> = (0..128 * 1024).map(|i| (i % 199) as u8).collect();

    let db = AedbInstance::open_anonymous(config.clone(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.kv_set("p", "app", b"blob".to_vec(), value.clone())
        .await
        .expect("set large value");
    db.checkpoint_now().await.expect("checkpoint");
    db.shutdown().await.expect("shutdown");
    drop(db);

    fs::remove_file(dir.path().join("values.aedbdat")).expect("remove value store");

    let reopened =
        AedbInstance::open_anonymous(config, dir.path()).expect("reopen from checkpoint");
    let recovered = reopened
        .kv_get_no_auth("p", "app", b"blob", ConsistencyMode::AtLatest)
        .await
        .expect("recovered get")
        .expect("recovered value");
    assert_eq!(recovered.value, value);
    reopened.shutdown().await.expect("shutdown reopened");
}

#[test]
fn wal_segment_gc_reclaims_checkpoint_covered_segments() {
    let dir = tempdir().expect("temp");
    for seq in 1_u64..=4 {
        std::fs::write(
            dir.path().join(format!("segment_{seq:016}.aedbwal")),
            format!("segment-{seq}"),
        )
        .expect("write segment");
    }
    let manifest = crate::manifest::schema::Manifest {
        durable_seq: 50,
        visible_seq: 50,
        active_segment_seq: 4,
        checkpoints: vec![crate::checkpoint::writer::CheckpointMeta {
            filename: "checkpoint_0000000000000050.aedb.zst".into(),
            seq: 50,
            sha256_hex: "00".repeat(32),
            created_at_micros: 1,
            key_id: None,
        }],
        segments: Vec::new(),
        ..Default::default()
    };
    crate::manifest::atomic::write_manifest_atomic_signed(&manifest, dir.path(), None)
        .expect("write manifest");
    let snapshot_manager = Arc::new(parking_lot::Mutex::new(
        crate::snapshot::gc::SnapshotManager::default(),
    ));

    let reclaimed = reclaim_eligible_wal_segments(dir.path(), &snapshot_manager, None).expect("gc");

    assert_eq!(reclaimed, 3);
    assert!(!dir.path().join("segment_0000000000000001.aedbwal").exists());
    assert!(!dir.path().join("segment_0000000000000002.aedbwal").exists());
    assert!(!dir.path().join("segment_0000000000000003.aedbwal").exists());
    assert!(dir.path().join("segment_0000000000000004.aedbwal").exists());
}

#[tokio::test]
async fn checkpoint_now_enables_clean_restart() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        owner_id: None,
        if_not_exists: false,
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        columns: vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("table");
    db.commit(Mutation::Upsert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        primary_key: vec![Value::Integer(5)],
        row: Row {
            values: vec![Value::Integer(5), Value::Text("bob".into())],
        },
    })
    .await
    .expect("insert");

    let seq = db.checkpoint_now().await.expect("checkpoint");
    assert!(seq >= 3);

    // Release the exclusive data-directory lock before reopening the same dir.
    drop(db);

    let reopened = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("reopen");
    let rows = reopened
        .query_no_auth(
            "p",
            "app",
            Query::select(&["*"])
                .from("users")
                .where_(Expr::Eq("id".into(), Value::Integer(5))),
            QueryOptions::default(),
        )
        .await
        .expect("query");
    assert_eq!(rows.rows.len(), 1);
}

#[tokio::test]
async fn checkpoint_retention_keeps_recent_checkpoints_and_prunes_older_files() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig::default().with_checkpoint_retention_count(2);
    let db = AedbInstance::open_anonymous(config.clone(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");

    // Drive several checkpoints, each at a strictly higher seq.
    let mut checkpoint_seqs = Vec::new();
    for i in 0..4u64 {
        db.commit(Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: format!("k{i}").into_bytes(),
            value: format!("v{i}").into_bytes(),
        })
        .await
        .expect("write");
        checkpoint_seqs.push(db.checkpoint_now().await.expect("checkpoint"));
    }

    // Manifest retains exactly the two newest checkpoints, newest last.
    let manifest = crate::manifest::atomic::load_manifest(dir.path()).expect("manifest");
    assert_eq!(manifest.checkpoints.len(), 2, "retention count honored");
    let retained_seqs: Vec<u64> = manifest.checkpoints.iter().map(|cp| cp.seq).collect();
    let mut expected = checkpoint_seqs.clone();
    expected.sort_unstable();
    assert_eq!(retained_seqs, expected[expected.len() - 2..].to_vec());

    // Superseded checkpoint files are pruned from disk; only retained remain.
    let on_disk: Vec<String> = fs::read_dir(dir.path())
        .expect("read dir")
        .filter_map(|entry| entry.ok())
        .filter_map(|entry| entry.file_name().into_string().ok())
        .filter(|name| name.starts_with("checkpoint_") && name.ends_with(".aedb.zst"))
        .collect();
    assert_eq!(
        on_disk.len(),
        2,
        "older checkpoint files pruned: {on_disk:?}"
    );

    // A clean reopen still recovers the latest state.
    drop(db);
    let reopened = AedbInstance::open_anonymous(config, dir.path()).expect("reopen");
    let entry = reopened
        .kv_get_no_auth("p", "app", b"k3", ConsistencyMode::AtLatest)
        .await
        .expect("kv_get")
        .expect("present");
    assert_eq!(entry.value, b"v3".to_vec());
}

#[tokio::test]
async fn checkpoint_does_not_prune_older_checkpoints_when_prior_manifest_unreadable() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig::production([7u8; 32]).with_checkpoint_retention_count(3);
    let db = AedbInstance::open_anonymous(config.clone(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    for i in 0..2u64 {
        db.commit(Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: format!("k{i}").into_bytes(),
            value: b"v".to_vec(),
        })
        .await
        .expect("write");
        db.checkpoint_now().await.expect("checkpoint");
    }

    let count_checkpoints = || {
        fs::read_dir(dir.path())
            .expect("read dir")
            .filter_map(|entry| entry.ok())
            .filter_map(|entry| entry.file_name().into_string().ok())
            .filter(|name| name.starts_with("checkpoint_") && name.ends_with(".aedb.zst"))
            .count()
    };
    let before = count_checkpoints();
    assert!(before >= 2, "expected retained checkpoints, got {before}");

    // Make the signed manifest unreadable: under an HMAC key reconstruction is
    // disabled, so the next checkpoint sees no prior manifest.
    let _ = fs::remove_file(dir.path().join("manifest.hmac"));
    let _ = fs::remove_file(dir.path().join("manifest.hmac.prev"));

    db.commit(Mutation::KvSet {
        project_id: "p".into(),
        scope_id: "app".into(),
        key: b"k2".to_vec(),
        value: b"v".to_vec(),
    })
    .await
    .expect("write");
    db.checkpoint_now().await.expect("cp3");

    // The new checkpoint is written, but older files are NOT pruned away.
    assert!(
        count_checkpoints() > before,
        "older checkpoints must survive an unreadable prior manifest"
    );
}

#[tokio::test]
async fn checkpoint_now_in_batch_mode_flushes_wal_and_recovers_tail() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        durability_mode: DurabilityMode::Batch,
        batch_interval_ms: 60_000,
        batch_max_bytes: usize::MAX,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config.clone(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.commit(Mutation::KvSet {
        project_id: "p".into(),
        scope_id: "app".into(),
        key: b"tail".to_vec(),
        value: b"v1".to_vec(),
    })
    .await
    .expect("tail write");

    let before = db.head_state().await;
    assert!(before.visible_head_seq > before.durable_head_seq);

    let cp_seq = db.checkpoint_now().await.expect("checkpoint");
    let after = db.head_state().await;
    assert_eq!(after.visible_head_seq, after.durable_head_seq);
    assert!(after.durable_head_seq >= cp_seq);
    drop(db);

    let recovered = crate::recovery::recover_with_config(dir.path(), &config).expect("recover");
    let tail = recovered
        .keyspace
        .kv_get("p", "app", b"tail")
        .expect("tail recovered");
    assert_eq!(tail.value, b"v1".to_vec());
}

#[tokio::test]
async fn checkpoint_now_allows_commits_while_running() {
    let dir = tempdir().expect("temp");
    let db =
        Arc::new(AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open"));
    db.create_project("p").await.expect("project");

    // Build enough state to make checkpoint work measurable.
    for i in 0..2_000u32 {
        db.commit(Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: format!("seed:{i:06}").into_bytes(),
            value: vec![b'x'; 1024],
        })
        .await
        .expect("seed write");
    }

    let checkpoint_db = Arc::clone(&db);
    let checkpoint_task = tokio::spawn(async move { checkpoint_db.checkpoint_now().await });

    // Wait briefly for checkpoint to begin.
    for _ in 0..10 {
        if !checkpoint_task.is_finished() {
            break;
        }
        tokio::time::sleep(Duration::from_millis(1)).await;
    }
    assert!(
        !checkpoint_task.is_finished(),
        "checkpoint should still be running"
    );

    let commit = db
        .commit(Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: b"concurrent:write".to_vec(),
            value: b"ok".to_vec(),
        })
        .await
        .expect("commit should proceed during checkpoint");
    let checkpoint_seq = checkpoint_task
        .await
        .expect("checkpoint join")
        .expect("checkpoint");
    assert!(commit.commit_seq >= checkpoint_seq);
}

#[tokio::test]
async fn checkpoint_now_serializes_checkpoint_writers() {
    let dir = tempdir().expect("temp");
    let db =
        Arc::new(AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open"));
    db.create_project("p").await.expect("project");

    db.commit(Mutation::KvSet {
        project_id: "p".into(),
        scope_id: "app".into(),
        key: b"k".to_vec(),
        value: b"v".to_vec(),
    })
    .await
    .expect("seed");

    let db1 = Arc::clone(&db);
    let db2 = Arc::clone(&db);
    let t1 = tokio::spawn(async move { db1.checkpoint_now().await });
    let t2 = tokio::spawn(async move { db2.checkpoint_now().await });

    let s1 = t1.await.expect("checkpoint task 1").expect("checkpoint 1");
    let s2 = t2.await.expect("checkpoint task 2").expect("checkpoint 2");
    assert_eq!(s1, s2);
}

#[tokio::test]
async fn checkpoint_captures_transaction_all_or_none() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig::default();
    let db = AedbInstance::open_anonymous(config.clone(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");

    let base_seq = db
        .snapshot_probe(ConsistencyMode::AtLatest)
        .await
        .expect("base");
    let commit = db
        .commit_envelope(TransactionEnvelope {
            caller: None,
            idempotency_key: None,
            write_class: WriteClass::Standard,
            assertions: Vec::new(),
            read_set: Default::default(),
            write_intent: WriteIntent {
                mutations: vec![
                    Mutation::KvSet {
                        project_id: "p".into(),
                        scope_id: "app".into(),
                        key: b"tx:part:a".to_vec(),
                        value: b"1".to_vec(),
                    },
                    Mutation::KvSet {
                        project_id: "p".into(),
                        scope_id: "app".into(),
                        key: b"tx:part:b".to_vec(),
                        value: b"2".to_vec(),
                    },
                ],
            },
            base_seq,
        })
        .await
        .expect("atomic multi-mutation commit");
    let checkpoint_seq = db.checkpoint_now().await.expect("checkpoint");
    assert!(checkpoint_seq >= commit.commit_seq);

    let recovered_at_cp =
        crate::recovery::recover_at_seq_with_config(dir.path(), checkpoint_seq, &config)
            .expect("recover at checkpoint");
    let snapshot_at_cp = recovered_at_cp.keyspace.snapshot();
    assert_eq!(
        snapshot_at_cp
            .kv_get("p", "app", b"tx:part:a")
            .map(|entry| entry.value.clone()),
        Some(b"1".to_vec())
    );
    assert_eq!(
        snapshot_at_cp
            .kv_get("p", "app", b"tx:part:b")
            .map(|entry| entry.value.clone()),
        Some(b"2".to_vec())
    );

    if commit.commit_seq > 0 {
        let recovered_before =
            crate::recovery::recover_at_seq_with_config(dir.path(), commit.commit_seq - 1, &config)
                .expect("recover before commit seq");
        let before_snapshot = recovered_before.keyspace.snapshot();
        assert!(before_snapshot.kv_get("p", "app", b"tx:part:a").is_none());
        assert!(before_snapshot.kv_get("p", "app", b"tx:part:b").is_none());
    }
}

#[tokio::test]
async fn checkpoint_manifest_trims_fully_covered_segments() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        max_segment_bytes: 4096,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config.clone(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");

    for i in 0..400u32 {
        db.commit(Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: format!("seg:{i:04}").into_bytes(),
            value: vec![b'x'; 256],
        })
        .await
        .expect("seed");
    }

    let all_segments = crate::lib_helpers::read_segments(dir.path()).expect("all segments");
    assert!(
        all_segments.len() > 1,
        "test must create multiple wal segments"
    );

    let checkpoint_seq = db.checkpoint_now().await.expect("checkpoint");
    let manifest = crate::manifest::atomic::load_manifest_signed(dir.path(), config.hmac_key())
        .expect("manifest");

    assert!(
        manifest.segments.len() < all_segments.len(),
        "checkpoint manifest should drop fully covered historical segments"
    );
    for segment in &manifest.segments {
        let path = dir.path().join(&segment.filename);
        let range = crate::lib_helpers::scan_segment_seq_range(&path).expect("scan segment");
        if let Some((_, max_seq)) = range {
            assert!(
                max_seq > checkpoint_seq || segment.segment_seq == manifest.active_segment_seq,
                "manifest retained a segment fully covered by checkpoint"
            );
        }
    }
}

#[tokio::test]
#[ignore = "manual perf probe: commit latency with and without concurrent checkpoint"]
async fn benchmark_commit_latency_during_checkpoint() {
    fn percentile(sorted: &[u128], p: f64) -> u128 {
        if sorted.is_empty() {
            return 0;
        }
        let percentile_index = ((sorted.len() as f64 - 1.0) * p).round() as usize;
        sorted[percentile_index.min(sorted.len() - 1)]
    }

    async fn run_phase(
        db: &Arc<AedbInstance>,
        start: usize,
        count: usize,
        with_checkpoint: bool,
    ) -> (f64, u128, u128) {
        for i in 0..10_000usize {
            db.commit(Mutation::KvSet {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: format!("warmup:{i:05}").into_bytes(),
                value: vec![b'w'; 256],
            })
            .await
            .expect("warmup seed");
        }

        let checkpoint_task = if with_checkpoint {
            Some(tokio::spawn({
                let db = Arc::clone(db);
                async move { db.checkpoint_now().await }
            }))
        } else {
            None
        };

        let phase_started = Instant::now();
        let mut latencies_us = Vec::with_capacity(count);
        for i in 0..count {
            let started = Instant::now();
            db.commit(Mutation::KvSet {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: format!(
                    "bench:{}:{:06}",
                    if with_checkpoint { "cp" } else { "base" },
                    start + i
                )
                .into_bytes(),
                value: vec![b'x'; 512],
            })
            .await
            .expect("bench commit");
            latencies_us.push(started.elapsed().as_micros());
        }

        if let Some(task) = checkpoint_task {
            let _ = task
                .await
                .expect("checkpoint task join")
                .expect("checkpoint");
        }

        latencies_us.sort_unstable();
        let elapsed_secs = phase_started.elapsed().as_secs_f64().max(0.000_001);
        let tps = count as f64 / elapsed_secs;
        let p50_us = percentile(&latencies_us, 0.50);
        let p99_us = percentile(&latencies_us, 0.99);
        (tps, p50_us, p99_us)
    }

    let dir = tempdir().expect("temp");
    let config = AedbConfig::default();
    let db = Arc::new(AedbInstance::open_anonymous(config, dir.path()).expect("open"));
    db.create_project("p").await.expect("project");

    let (base_tps, base_p50, base_p99) = run_phase(&db, 0, 800, false).await;
    let (cp_tps, cp_p50, cp_p99) = run_phase(&db, 1_000_000, 800, true).await;

    eprintln!(
        "checkpoint_perf: base_tps={:.2} base_p50_us={} base_p99_us={} | cp_tps={:.2} cp_p50_us={} cp_p99_us={} | tps_ratio={:.3} p50_ratio={:.3} p99_ratio={:.3}",
        base_tps,
        base_p50,
        base_p99,
        cp_tps,
        cp_p50,
        cp_p99,
        cp_tps / base_tps.max(0.000_001),
        (cp_p50 as f64) / (base_p50.max(1) as f64),
        (cp_p99 as f64) / (base_p99.max(1) as f64),
    );
}

#[tokio::test]
#[ignore = "manual perf probe: parallel commit throughput with and without concurrent checkpoint"]
async fn benchmark_parallel_commit_throughput_during_checkpoint() {
    fn percentile(sorted: &[u128], p: f64) -> u128 {
        if sorted.is_empty() {
            return 0;
        }
        let percentile_index = ((sorted.len() as f64 - 1.0) * p).round() as usize;
        sorted[percentile_index.min(sorted.len() - 1)]
    }

    async fn run_parallel_phase(
        db: &Arc<AedbInstance>,
        workers: usize,
        commits_per_worker: usize,
        with_checkpoint: bool,
        offset: usize,
    ) -> (f64, u128, u128) {
        let checkpoint_task = if with_checkpoint {
            Some(tokio::spawn({
                let db = Arc::clone(db);
                async move { db.checkpoint_now().await }
            }))
        } else {
            None
        };

        let phase_started = Instant::now();
        let mut tasks = Vec::with_capacity(workers);
        for worker in 0..workers {
            let db = Arc::clone(db);
            tasks.push(tokio::spawn(async move {
                let mut latencies = Vec::with_capacity(commits_per_worker);
                for i in 0..commits_per_worker {
                    let started = Instant::now();
                    db.commit(Mutation::KvSet {
                        project_id: "p".into(),
                        scope_id: "app".into(),
                        key: format!(
                            "par:{}:{:02}:{:06}",
                            if with_checkpoint { "cp" } else { "base" },
                            worker,
                            offset + i
                        )
                        .into_bytes(),
                        value: vec![b'p'; 256],
                    })
                    .await
                    .expect("parallel bench commit");
                    latencies.push(started.elapsed().as_micros());
                }
                latencies
            }));
        }

        let mut all_latencies = Vec::with_capacity(workers * commits_per_worker);
        for task in tasks {
            let mut worker_latencies = task.await.expect("worker join");
            all_latencies.append(&mut worker_latencies);
        }

        if let Some(task) = checkpoint_task {
            let _ = task
                .await
                .expect("checkpoint task join")
                .expect("checkpoint");
        }

        all_latencies.sort_unstable();
        let elapsed_secs = phase_started.elapsed().as_secs_f64().max(0.000_001);
        let total = workers * commits_per_worker;
        let tps = total as f64 / elapsed_secs;
        let p50_us = percentile(&all_latencies, 0.50);
        let p99_us = percentile(&all_latencies, 0.99);
        (tps, p50_us, p99_us)
    }

    let dir = tempdir().expect("temp");
    let mut config = AedbConfig {
        durability_mode: DurabilityMode::Batch,
        batch_interval_ms: 10,
        batch_max_bytes: usize::MAX,
        recovery_mode: RecoveryMode::Permissive,
        hash_chain_required: false,
        ..AedbConfig::default()
    };
    config.manifest_hmac_key = None;
    let db = Arc::new(AedbInstance::open_anonymous(config, dir.path()).expect("open"));
    db.create_project("p").await.expect("project");

    // Seed state so the checkpoint has meaningful work.
    for i in 0..12_000usize {
        db.commit(Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: format!("parallel-seed:{i:05}").into_bytes(),
            value: vec![b's'; 256],
        })
        .await
        .expect("seed");
    }

    let workers = 8usize;
    let commits_per_worker = 300usize;
    let (base_tps, base_p50, base_p99) =
        run_parallel_phase(&db, workers, commits_per_worker, false, 0).await;
    let (cp_tps, cp_p50, cp_p99) =
        run_parallel_phase(&db, workers, commits_per_worker, true, 1_000_000).await;

    eprintln!(
        "checkpoint_parallel_perf: workers={} commits_per_worker={} | base_tps={:.2} base_p50_us={} base_p99_us={} | cp_tps={:.2} cp_p50_us={} cp_p99_us={} | tps_ratio={:.3} p50_ratio={:.3} p99_ratio={:.3}",
        workers,
        commits_per_worker,
        base_tps,
        base_p50,
        base_p99,
        cp_tps,
        cp_p50,
        cp_p99,
        cp_tps / base_tps.max(0.000_001),
        (cp_p50 as f64) / (base_p50.max(1) as f64),
        (cp_p99 as f64) / (base_p99.max(1) as f64),
    );
}

#[tokio::test]
#[ignore = "manual perf probe: compare checkpoint compression levels under parallel load"]
async fn benchmark_parallel_checkpoint_compression_levels() {
    async fn seed(db: &Arc<AedbInstance>) {
        for i in 0..12_000usize {
            db.commit(Mutation::KvSet {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: format!("parallel-seed:{i:05}").into_bytes(),
                value: vec![b's'; 256],
            })
            .await
            .expect("seed");
        }
    }

    async fn run_parallel_phase(
        db: &Arc<AedbInstance>,
        workers: usize,
        commits_per_worker: usize,
        with_checkpoint: bool,
        offset: usize,
    ) -> f64 {
        let checkpoint_task = if with_checkpoint {
            Some(tokio::spawn({
                let db = Arc::clone(db);
                async move { db.checkpoint_now().await }
            }))
        } else {
            None
        };

        let phase_started = Instant::now();
        let mut tasks = Vec::with_capacity(workers);
        for worker in 0..workers {
            let db = Arc::clone(db);
            tasks.push(tokio::spawn(async move {
                for i in 0..commits_per_worker {
                    db.commit(Mutation::KvSet {
                        project_id: "p".into(),
                        scope_id: "app".into(),
                        key: format!(
                            "cmp:{}:{:02}:{:06}",
                            if with_checkpoint { "cp" } else { "base" },
                            worker,
                            offset + i
                        )
                        .into_bytes(),
                        value: vec![b'c'; 256],
                    })
                    .await
                    .expect("parallel bench commit");
                }
            }));
        }
        for task in tasks {
            task.await.expect("worker join");
        }
        if let Some(task) = checkpoint_task {
            let _ = task
                .await
                .expect("checkpoint task join")
                .expect("checkpoint");
        }

        let elapsed_secs = phase_started.elapsed().as_secs_f64().max(0.000_001);
        let total = workers * commits_per_worker;
        total as f64 / elapsed_secs
    }

    let workers = 8usize;
    let commits_per_worker = 300usize;
    let levels = [3, 1, 0];

    let mut baseline_tps = 0.0f64;
    for (idx, level) in levels.iter().copied().enumerate() {
        let dir = tempdir().expect("temp");
        let mut config = AedbConfig {
            durability_mode: DurabilityMode::Batch,
            batch_interval_ms: 10,
            batch_max_bytes: usize::MAX,
            recovery_mode: RecoveryMode::Permissive,
            hash_chain_required: false,
            ..AedbConfig::default()
        };
        config.manifest_hmac_key = None;
        config.checkpoint_compression_level = level;
        let db = Arc::new(AedbInstance::open_anonymous(config, dir.path()).expect("open"));
        db.create_project("p").await.expect("project");
        seed(&db).await;

        let base =
            run_parallel_phase(&db, workers, commits_per_worker, false, idx * 1_000_000).await;
        let cp = run_parallel_phase(
            &db,
            workers,
            commits_per_worker,
            true,
            idx * 1_000_000 + 500_000,
        )
        .await;
        if idx == 0 {
            baseline_tps = base;
        }

        eprintln!(
            "checkpoint_compression_perf: level={} workers={} commits_per_worker={} base_tps={:.2} cp_tps={:.2} cp_to_base={:.3} cp_to_level3_base={:.3}",
            level,
            workers,
            commits_per_worker,
            base,
            cp,
            cp / base.max(0.000_001),
            cp / baseline_tps.max(0.000_001),
        );
    }
}

#[tokio::test]
async fn snapshot_probe_returns_snapshot_seq() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    let seq = db
        .snapshot_probe(ConsistencyMode::AtLatest)
        .await
        .expect("probe");
    assert!(seq >= 1);
}

#[tokio::test]
async fn snapshot_probe_does_not_consume_snapshot_slot() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        max_concurrent_snapshots: 1,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
    db.create_project("p").await.expect("project");

    let _lease = db
        .acquire_snapshot(ConsistencyMode::AtLatest)
        .await
        .expect("lease");

    let seq = db
        .snapshot_probe(ConsistencyMode::AtLatest)
        .await
        .expect("probe");
    assert!(seq >= 1);
}

#[tokio::test]
async fn snapshot_probe_remains_live_under_snapshot_capacity_pressure() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        max_concurrent_snapshots: 1,
        ..AedbConfig::default()
    };
    let db = Arc::new(AedbInstance::open_anonymous(config, dir.path()).expect("open"));
    db.create_project("p").await.expect("project");

    let _lease = db
        .acquire_snapshot(ConsistencyMode::AtLatest)
        .await
        .expect("lease");

    let mut tasks = Vec::new();
    for _ in 0..4 {
        let db = Arc::clone(&db);
        tasks.push(tokio::spawn(async move {
            for _ in 0..128 {
                let seq = db
                    .snapshot_probe(ConsistencyMode::AtLatest)
                    .await
                    .expect("probe");
                assert!(seq >= 1);
                tokio::task::yield_now().await;
            }
        }));
    }

    tokio::time::timeout(Duration::from_secs(2), async {
        for task in tasks {
            task.await.expect("probe task");
        }
    })
    .await
    .expect("snapshot probes timed out under snapshot pressure");
}

#[tokio::test]
async fn checkpoint_now_completes_under_hot_load() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        durability_mode: DurabilityMode::Batch,
        batch_interval_ms: 60_000,
        batch_max_bytes: usize::MAX,
        ..AedbConfig::default()
    };
    let db = Arc::new(AedbInstance::open_anonymous(config, dir.path()).expect("open"));
    db.create_project("p").await.expect("project");

    let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let writes = Arc::new(std::sync::atomic::AtomicU64::new(0));
    let reads = Arc::new(std::sync::atomic::AtomicU64::new(0));

    let writer_db = Arc::clone(&db);
    let writer_stop = Arc::clone(&stop);
    let writes_ref = Arc::clone(&writes);
    let writer = tokio::spawn(async move {
        let mut i = 0u64;
        while !writer_stop.load(Ordering::Relaxed) {
            let _ = writer_db
                .commit(Mutation::KvSet {
                    project_id: "p".into(),
                    scope_id: "app".into(),
                    key: format!("checkpoint-hot:{i}").into_bytes(),
                    value: i.to_string().into_bytes(),
                })
                .await;
            writes_ref.fetch_add(1, Ordering::Relaxed);
            i = i.saturating_add(1);
            tokio::task::yield_now().await;
        }
    });

    let reader_db = Arc::clone(&db);
    let reader_stop = Arc::clone(&stop);
    let reads_ref = Arc::clone(&reads);
    let reader = tokio::spawn(async move {
        while !reader_stop.load(Ordering::Relaxed) {
            let _ = reader_db.snapshot_probe(ConsistencyMode::AtLatest).await;
            reads_ref.fetch_add(1, Ordering::Relaxed);
            tokio::task::yield_now().await;
        }
    });

    tokio::time::sleep(Duration::from_millis(50)).await;
    let writes_before = writes.load(Ordering::Relaxed);
    let reads_before = reads.load(Ordering::Relaxed);

    let checkpoint_seq = tokio::time::timeout(Duration::from_secs(10), db.checkpoint_now())
        .await
        .expect("checkpoint_now timed out")
        .expect("checkpoint_now");
    assert!(checkpoint_seq >= 1);

    let writes_after = writes.load(Ordering::Relaxed);
    let reads_after = reads.load(Ordering::Relaxed);
    assert!(
        writes_after > writes_before,
        "writes stalled during checkpoint"
    );
    assert!(
        reads_after > reads_before,
        "reads stalled during checkpoint"
    );

    stop.store(true, Ordering::Relaxed);
    tokio::time::timeout(Duration::from_secs(5), writer)
        .await
        .expect("writer join timed out")
        .expect("writer join");
    tokio::time::timeout(Duration::from_secs(5), reader)
        .await
        .expect("reader join timed out")
        .expect("reader join");
}

#[tokio::test]
async fn at_checkpoint_falls_back_when_version_evicted() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        max_versions: 2,
        min_version_age_ms: 0,
        version_gc_interval_ms: 1,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.commit(Mutation::KvSet {
        project_id: "p".into(),
        scope_id: "app".into(),
        key: b"checkpoint:key".to_vec(),
        value: b"v1".to_vec(),
    })
    .await
    .expect("seed checkpoint value");
    let checkpoint_seq = db.checkpoint_now().await.expect("checkpoint");
    for i in 0..8 {
        db.commit(Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: format!("tail:{i}").into_bytes(),
            value: vec![i as u8],
        })
        .await
        .expect("tail write");
    }
    tokio::time::sleep(Duration::from_millis(20)).await;
    let cp_view_seq = db
        .snapshot_probe(ConsistencyMode::AtCheckpoint)
        .await
        .expect("checkpoint snapshot");
    assert_eq!(cp_view_seq, checkpoint_seq);
}

#[tokio::test]
async fn strict_restore_rejects_older_backup_version() {
    let dir = tempdir().expect("data dir");
    let backup_dir = tempdir().expect("backup dir");
    let config = AedbConfig::production([7u8; 32]);
    let db = AedbInstance::open_anonymous(config.clone(), dir.path()).expect("open");
    db.create_project("arcana").await.expect("create project");
    db.create_scope("arcana", "app")
        .await
        .expect("create scope");
    db.backup_full(backup_dir.path())
        .await
        .expect("backup full");

    let mut manifest = crate::backup::load_backup_manifest(backup_dir.path(), config.hmac_key())
        .expect("manifest");
    // An older major.minor line (0.2.x) predates the signed-I256 key encoding
    // introduced in 0.3.0, so its keys are unreadable by this build — restore
    // must refuse it outright as incompatible.
    manifest.aedb_version = "0.2.0".into();
    crate::backup::write_backup_manifest(backup_dir.path(), &manifest, config.hmac_key())
        .expect("rewrite manifest");

    let restore_dir = tempdir().expect("restore dir");
    let err = AedbInstance::restore_from_backup_chain(
        &[backup_dir.path().to_path_buf()],
        restore_dir.path(),
        &config,
        None,
    )
    .expect_err("strict restore must reject older backup version");
    assert!(format!("{err}").contains("incompatible AEDB version"));
}

#[tokio::test]
async fn restore_at_time_rejects_backup_wal_with_invalid_hash_chain() {
    let dir = tempdir().expect("data dir");
    let backup_dir = tempdir().expect("backup dir");
    let restore_dir = tempdir().expect("restore dir");
    let config = AedbConfig::production([8u8; 32]);
    let db = AedbInstance::open_anonymous(config.clone(), dir.path()).expect("open");
    db.create_project("arcana").await.expect("create project");
    db.create_scope("arcana", "app")
        .await
        .expect("create scope");
    db.commit(Mutation::KvSet {
        project_id: "arcana".into(),
        scope_id: "app".into(),
        key: b"k".to_vec(),
        value: b"v".to_vec(),
    })
    .await
    .expect("write kv");
    db.backup_full(backup_dir.path())
        .await
        .expect("backup full");

    let manifest = crate::backup::load_backup_manifest(backup_dir.path(), config.hmac_key())
        .expect("manifest");
    let wal_name = manifest.wal_segments.first().expect("wal segment");
    let wal_path = backup_dir.path().join("wal_tail").join(wal_name);
    let mut wal_bytes = fs::read(&wal_path).expect("read wal");
    wal_bytes[0] ^= 0xFF;
    fs::write(&wal_path, wal_bytes).expect("corrupt wal");
    let mut manifest = crate::backup::load_backup_manifest(backup_dir.path(), config.hmac_key())
        .expect("manifest");
    manifest.file_sha256.insert(
        format!("wal_tail/{wal_name}"),
        crate::backup::sha256_file_hex(&wal_path).expect("rehash corrupted wal"),
    );
    crate::backup::write_backup_manifest(backup_dir.path(), &manifest, config.hmac_key())
        .expect("rewrite manifest");

    let err = AedbInstance::restore_from_backup_chain_at_time(
        &[backup_dir.path().to_path_buf()],
        restore_dir.path(),
        &config,
        u64::MAX,
    )
    .expect_err("time restore must reject invalid hash chain");
    assert!(
        format!("{err}").contains("bad segment header")
            || format!("{err}").contains("segment hash chain mismatch")
    );
}

#[tokio::test]
async fn checkpoint_and_restore_round_trips_spilled_table_rows() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        storage_mode: StorageMode::DiskBacked,
        // Tiny ceiling so table-row payloads spill to disk before the checkpoint.
        max_memory_estimate_bytes: 8 * 1024,
        persistent_value_hot_cache_bytes: 4 * 1024,
        table_row_spill_enabled: true,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config.clone(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        owner_id: None,
        if_not_exists: false,
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "items".into(),
        columns: vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "blob".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("table");

    let payload = "y".repeat(512);
    for i in 0..64i64 {
        db.commit(Mutation::Upsert {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "items".into(),
            primary_key: vec![Value::Integer(i)],
            row: Row {
                values: vec![
                    Value::Integer(i),
                    Value::Text(format!("{payload}-{i}").into()),
                ],
            },
        })
        .await
        .expect("insert spills under pressure");
    }

    // Take a checkpoint while rows are spilled. This forces
    // `materialized_for_checkpoint` to page every spilled row back inline so the
    // checkpoint payload is self-contained (the value store is detached in it).
    db.checkpoint_now().await.expect("checkpoint");
    drop(db);

    // Restore from the checkpoint and confirm every spilled row round-tripped.
    let reopened = AedbInstance::open_anonymous(config, dir.path()).expect("reopen");
    let rows = reopened
        .query_no_auth(
            "p",
            "app",
            Query::select(&["id", "blob"]).from("items").limit(1000),
            QueryOptions::default(),
        )
        .await
        .expect("query after restore");
    assert_eq!(
        rows.rows.len(),
        64,
        "all spilled rows must survive checkpoint+restore"
    );

    let one = reopened
        .query_no_auth(
            "p",
            "app",
            Query::select(&["id", "blob"])
                .from("items")
                .where_(Expr::Eq("id".into(), Value::Integer(42))),
            QueryOptions::default(),
        )
        .await
        .expect("point query after restore");
    assert_eq!(one.rows.len(), 1);
    assert_eq!(
        one.rows[0].values[1],
        Value::Text(format!("{payload}-42").into())
    );
}

// ---------------------------------------------------------------------------
// WriteClass::Volatile — WAL-skipping, checkpoint-only durability.
// ---------------------------------------------------------------------------

/// A volatile KvSet envelope for project "p", scope "app".
fn volatile_kv_set(key: &[u8], value: &[u8]) -> TransactionEnvelope {
    TransactionEnvelope {
        caller: None,
        idempotency_key: None,
        write_class: WriteClass::Volatile,
        assertions: vec![],
        read_set: crate::commit::tx::ReadSet::default(),
        write_intent: WriteIntent {
            mutations: vec![Mutation::KvSet {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: key.to_vec(),
                value: value.to_vec(),
            }],
        },
        base_seq: 0,
    }
}

/// A volatile commit is immediately visible to readers but does not advance the
/// durable-seq watermark (it is never written to the WAL).
#[tokio::test]
async fn volatile_commit_is_visible_but_not_durable() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        durability_mode: DurabilityMode::Full,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
    db.create_project("p").await.expect("project");

    // In Full mode after a durable commit, durable == visible.
    let before = db.head_state().await;
    assert_eq!(before.visible_head_seq, before.durable_head_seq);

    db.commit_envelope(volatile_kv_set(b"pos", b"v1"))
        .await
        .expect("volatile commit");

    let after = db.head_state().await;
    // Visible advanced by exactly the one volatile commit; durable did not move.
    assert_eq!(after.visible_head_seq, before.visible_head_seq + 1);
    assert_eq!(after.durable_head_seq, before.durable_head_seq);
    assert!(after.visible_head_seq > after.durable_head_seq);

    // ...but the write is readable right now.
    let entry = db
        .kv_get_no_auth("p", "app", b"pos", ConsistencyMode::AtLatest)
        .await
        .expect("kv_get")
        .expect("present");
    assert_eq!(entry.value, b"v1".to_vec());
}

/// A volatile write followed by a durable commit (which advances the durable
/// horizon past it) is captured by the next checkpoint and survives a restart.
#[tokio::test]
async fn volatile_write_anchored_by_later_durable_survives_restart() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        durability_mode: DurabilityMode::Full,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config.clone(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");

    // durable A, volatile V, then durable B (advances durable past V).
    db.commit(Mutation::KvSet {
        project_id: "p".into(),
        scope_id: "app".into(),
        key: b"durable".to_vec(),
        value: b"a".to_vec(),
    })
    .await
    .expect("durable A");
    db.commit_envelope(volatile_kv_set(b"volatile", b"v"))
        .await
        .expect("volatile V");
    db.commit(Mutation::KvSet {
        project_id: "p".into(),
        scope_id: "app".into(),
        key: b"anchor".to_vec(),
        value: b"b".to_vec(),
    })
    .await
    .expect("durable B anchor");

    db.checkpoint_now().await.expect("checkpoint");
    drop(db);

    let recovered = crate::recovery::recover_with_config(dir.path(), &config).expect("recover");
    // The volatile write was below the durable horizon when the checkpoint ran,
    // so it is captured and recovered.
    assert_eq!(
        recovered
            .keyspace
            .kv_get("p", "app", b"volatile")
            .expect("volatile recovered")
            .value,
        b"v".to_vec()
    );
    assert_eq!(
        recovered
            .keyspace
            .kv_get("p", "app", b"durable")
            .expect("durable recovered")
            .value,
        b"a".to_vec()
    );
}

/// A volatile write made after the last durable commit/checkpoint is held only in
/// memory and is lost on restart, while durable state is preserved.
#[tokio::test]
async fn volatile_write_without_durable_anchor_is_lost_on_restart() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        durability_mode: DurabilityMode::Full,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config.clone(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");

    db.commit(Mutation::KvSet {
        project_id: "p".into(),
        scope_id: "app".into(),
        key: b"durable".to_vec(),
        value: b"a".to_vec(),
    })
    .await
    .expect("durable A");
    db.checkpoint_now()
        .await
        .expect("checkpoint anchors at durable A");

    // Volatile write happens AFTER the checkpoint's durable horizon.
    db.commit_envelope(volatile_kv_set(b"volatile", b"v"))
        .await
        .expect("volatile V");
    drop(db);

    let recovered = crate::recovery::recover_with_config(dir.path(), &config).expect("recover");
    assert!(
        recovered.keyspace.kv_get("p", "app", b"volatile").is_none(),
        "volatile write after the durable horizon must not survive restart"
    );
    assert_eq!(
        recovered
            .keyspace
            .kv_get("p", "app", b"durable")
            .expect("durable recovered")
            .value,
        b"a".to_vec()
    );
}

/// Requesting Durable finality on a volatile commit must not block (there is no
/// WAL durability to wait for) — it returns promptly and the write is visible.
#[tokio::test]
async fn durable_finality_on_volatile_commit_does_not_block() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        durability_mode: DurabilityMode::Full,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
    db.create_project("p").await.expect("project");

    let committed = tokio::time::timeout(
        Duration::from_secs(5),
        db.commit_envelope_with_finality(
            volatile_kv_set(b"pos", b"v1"),
            crate::CommitFinality::Durable,
        ),
    )
    .await
    .expect("durable finality on a volatile commit must not hang")
    .expect("commit");
    // Durable head is reported but was not required to reach the volatile seq.
    assert!(committed.commit_seq > 0);

    let entry = db
        .kv_get_no_auth("p", "app", b"pos", ConsistencyMode::AtLatest)
        .await
        .expect("kv_get")
        .expect("present");
    assert_eq!(entry.value, b"v1".to_vec());
}
