//! Comprehensive integration tests for noxu-dbi.
//!
//! Covers:
//!   - DatabaseImpl: open, put, get, delete, cursor operations
//!   - EnvironmentImpl: create, open databases, close
//!   - CursorImpl: basic iteration, state machine
//!   - DbType: all variants
//!   - DatabaseId: creation, comparison

#![allow(clippy::field_reassign_with_default)]

use std::sync::Arc;

use noxu_dbi::{
    CursorImpl, DatabaseConfig, DatabaseId, DatabaseImpl, DbType,
    EnvironmentImpl, GetMode, OperationStatus, PutMode, SearchMode,
};
use tempfile::TempDir;

fn tmp_env() -> (TempDir, EnvironmentImpl) {
    let dir = TempDir::new().unwrap();
    let env = EnvironmentImpl::new(dir.path(), false, true).unwrap();
    (dir, env)
}
use noxu_sync::RwLock;

// ============================================================================
// 1. DatabaseId: creation, comparison, ordering
// ============================================================================

#[test]
fn database_id_creation_and_equality() {
    let id1 = DatabaseId::new(1);
    let id2 = DatabaseId::new(1);
    let id3 = DatabaseId::new(2);

    assert_eq!(id1, id2);
    assert_ne!(id1, id3);
    assert_eq!(id1.id(), 1);
    assert_eq!(id1.as_i64(), 1);
}

#[test]
fn database_id_ordering() {
    let id1 = DatabaseId::new(1);
    let id2 = DatabaseId::new(2);
    let id3 = DatabaseId::new(3);

    assert!(id1 < id2);
    assert!(id2 < id3);
    assert!(id3 > id1);
    assert_eq!(id1.cmp(&id1), std::cmp::Ordering::Equal);
}

#[test]
fn database_id_serialization_roundtrip() {
    let original = DatabaseId::new(12345);
    let mut buf = Vec::new();
    original.write_to_log(&mut buf);
    assert_eq!(buf.len(), 8);

    let restored = DatabaseId::read_from_log(&buf).unwrap();
    assert_eq!(original, restored);
}

#[test]
fn database_id_negative_value() {
    let id = DatabaseId::new(-1);
    assert_eq!(id.id(), -1);
    let mut buf = Vec::new();
    id.write_to_log(&mut buf);
    let restored = DatabaseId::read_from_log(&buf).unwrap();
    assert_eq!(id, restored);
}

// ============================================================================
// 2. DbType: all variants, is_internal, internal_name, Display
// ============================================================================

#[test]
fn db_type_is_internal() {
    assert!(DbType::Id.is_internal());
    assert!(DbType::Name.is_internal());
    assert!(DbType::Utilization.is_internal());
    assert!(!DbType::User.is_internal());
}

#[test]
fn db_type_internal_name() {
    assert_eq!(DbType::Id.internal_name(), Some("_jeIdMap"));
    assert_eq!(DbType::Name.internal_name(), Some("_jeNameMap"));
    assert_eq!(DbType::Utilization.internal_name(), Some("_jeUtilization"));
    assert_eq!(DbType::User.internal_name(), None);
}

#[test]
fn db_type_display() {
    assert_eq!(DbType::Id.to_string(), "ID");
    assert_eq!(DbType::Name.to_string(), "NAME");
    assert_eq!(DbType::Utilization.to_string(), "UTILIZATION");
    assert_eq!(DbType::User.to_string(), "USER");
}

#[test]
fn db_type_equality() {
    assert_eq!(DbType::User, DbType::User);
    assert_ne!(DbType::User, DbType::Id);
    assert_ne!(DbType::Id, DbType::Name);
}

// ============================================================================
// 3. DatabaseImpl: creation, flags, delete state, dirty, reference count
// ============================================================================

fn default_config() -> DatabaseConfig {
    DatabaseConfig::default()
}

fn make_db(id: i64, name: &str) -> DatabaseImpl {
    DatabaseImpl::new(
        DatabaseId::new(id),
        name.to_string(),
        DbType::User,
        &default_config(),
    )
}

#[test]
fn database_impl_new() {
    let db = make_db(1, "mydb");
    assert_eq!(db.get_id(), DatabaseId::new(1));
    assert_eq!(db.get_name(), "mydb");
    assert_eq!(db.get_db_type(), DbType::User);
    assert!(!db.is_deleted());
    assert!(!db.is_deleting());
    assert_eq!(db.reference_count(), 0);
    assert!(!db.is_dirty());
}

#[test]
fn database_impl_sorted_duplicates_flag() {
    let mut config = DatabaseConfig::default();
    config.sorted_duplicates = false;
    let db1 = DatabaseImpl::new(
        DatabaseId::new(1),
        "a".into(),
        DbType::User,
        &config,
    );
    assert!(!db1.get_sorted_duplicates());

    config.sorted_duplicates = true;
    let db2 = DatabaseImpl::new(
        DatabaseId::new(2),
        "b".into(),
        DbType::User,
        &config,
    );
    assert!(db2.get_sorted_duplicates());
}

#[test]
fn database_impl_temporary_flag() {
    let mut config = DatabaseConfig::default();
    config.temporary = true;
    let db = DatabaseImpl::new(
        DatabaseId::new(1),
        "t".into(),
        DbType::User,
        &config,
    );
    assert!(db.is_temporary());
}

#[test]
fn database_impl_key_prefixing_flag() {
    let mut config = DatabaseConfig::default();
    config.key_prefixing = true;
    let db = DatabaseImpl::new(
        DatabaseId::new(1),
        "p".into(),
        DbType::User,
        &config,
    );
    assert!(db.get_key_prefixing());
}

#[test]
fn database_impl_delete_state_transitions() {
    let mut db = make_db(1, "db");
    assert!(!db.is_deleted());
    assert!(!db.is_deleting());

    db.start_delete();
    assert!(!db.is_deleted());
    assert!(db.is_deleting());

    db.finish_delete();
    assert!(db.is_deleted());
    assert!(db.is_deleting()); // still deleting (Deleted state)
}

#[test]
fn database_impl_dirty_tracking() {
    let db = make_db(1, "db");
    assert!(!db.is_dirty());
    db.set_dirty();
    assert!(db.is_dirty());
    db.clear_dirty();
    assert!(!db.is_dirty());
}

#[test]
fn database_impl_reference_counting() {
    let db = make_db(1, "db");
    assert_eq!(db.reference_count(), 0);
    db.increment_reference_count();
    db.increment_reference_count();
    assert_eq!(db.reference_count(), 2);
    db.decrement_reference_count();
    assert_eq!(db.reference_count(), 1);
    db.decrement_reference_count();
    assert_eq!(db.reference_count(), 0);
}

#[test]
fn database_impl_tree_access() {
    let mut db = make_db(1, "db");
    {
        let tree = db.get_tree().unwrap();
        assert_eq!(tree.get_root_lsn(), noxu_util::NULL_LSN.as_u64());
    }
    {
        let tree = db.get_tree_mut().unwrap();
        tree.set_root_lsn(9999);
    }
    assert_eq!(db.get_tree().unwrap().get_root_lsn(), 9999);
}

#[test]
fn database_impl_serialization_roundtrip() {
    let mut config = DatabaseConfig::default();
    config.sorted_duplicates = true;
    config.key_prefixing = true;
    config.node_max_entries = 512;

    let db = DatabaseImpl::new(
        DatabaseId::new(77),
        "roundtrip_db".to_string(),
        DbType::User,
        &config,
    );

    let mut buf = Vec::new();
    db.write_to_log(&mut buf).unwrap();

    let db2 = DatabaseImpl::read_from_log(&buf).unwrap();
    assert_eq!(db2.get_id(), DatabaseId::new(77));
    assert_eq!(db2.get_name(), "roundtrip_db");
    assert!(db2.get_sorted_duplicates());
    assert!(db2.get_key_prefixing());
    assert_eq!(db2.max_tree_entries_per_node(), 512);
}

// ============================================================================
// 4. EnvironmentImpl: create, open databases, close
// ============================================================================

#[test]
fn environment_impl_create_and_open() {
    let (_dir, env) = tmp_env();
    assert!(!env.is_read_only());
    assert!(env.is_transactional());
    assert!(env.is_open());
    assert!(env.is_valid());
}

#[test]
fn environment_impl_open_database_with_create() {
    let (_dir, env) = tmp_env();
    let mut cfg = DatabaseConfig::new();
    cfg.set_allow_create(true);

    let db = env.open_database("testdb", &cfg).unwrap();
    assert_eq!(db.read().get_name(), "testdb");
    assert_eq!(db.read().reference_count(), 1);
}

#[test]
fn environment_impl_open_database_no_create_fails() {
    let (_dir, env) = tmp_env();
    let cfg = DatabaseConfig::new();
    let result = env.open_database("nonexistent", &cfg);
    assert!(result.is_err());
}

#[test]
fn environment_impl_open_same_database_twice_increments_ref() {
    let (_dir, env) = tmp_env();
    let mut cfg = DatabaseConfig::new();
    cfg.set_allow_create(true);

    let db1 = env.open_database("shared", &cfg).unwrap();
    let db2 = env.open_database("shared", &cfg).unwrap();

    assert_eq!(db1.read().get_id(), db2.read().get_id());
    assert_eq!(db1.read().reference_count(), 2);
}

#[test]
fn environment_impl_multiple_databases_unique_ids() {
    let (_dir, env) = tmp_env();
    let mut cfg = DatabaseConfig::new();
    cfg.set_allow_create(true);

    let db1 = env.open_database("db1", &cfg).unwrap();
    let db2 = env.open_database("db2", &cfg).unwrap();
    let db3 = env.open_database("db3", &cfg).unwrap();

    assert_ne!(db1.read().get_id(), db2.read().get_id());
    assert_ne!(db2.read().get_id(), db3.read().get_id());
    assert_eq!(env.n_databases(), 3);
}

#[test]
fn environment_impl_remove_database() {
    let (_dir, env) = tmp_env();
    let mut cfg = DatabaseConfig::new();
    cfg.set_allow_create(true);

    let db = env.open_database("to_remove", &cfg).unwrap();
    env.close_database(db.read().get_id()).unwrap();
    env.remove_database("to_remove").unwrap();

    let result = env.open_database("to_remove", &DatabaseConfig::new());
    assert!(result.is_err());
}

#[test]
fn environment_impl_rename_database() {
    let (_dir, env) = tmp_env();
    let mut cfg = DatabaseConfig::new();
    cfg.set_allow_create(true);

    let db = env.open_database("old", &cfg).unwrap();
    let original_id = db.read().get_id();
    env.close_database(original_id).unwrap();

    env.rename_database("old", "new").unwrap();

    let found = env.open_database("new", &cfg).unwrap();
    assert_eq!(found.read().get_id(), original_id);
    assert!(env.open_database("old", &DatabaseConfig::new()).is_err());
}

#[test]
fn environment_impl_get_database_names() {
    let (_dir, env) = tmp_env();
    let mut cfg = DatabaseConfig::new();
    cfg.set_allow_create(true);

    env.open_database("alpha", &cfg).unwrap();
    env.open_database("beta", &cfg).unwrap();

    let names = env.get_database_names();
    assert!(names.contains(&"alpha".to_string()));
    assert!(names.contains(&"beta".to_string()));
}

#[test]
fn environment_impl_begin_txn_tracks_active() {
    let (_dir, env) = tmp_env();
    assert_eq!(env.n_active_txns(), 0);
    let _t1 = env.begin_txn().unwrap();
    assert_eq!(env.n_active_txns(), 1);
    let _t2 = env.begin_txn().unwrap();
    assert_eq!(env.n_active_txns(), 2);
}

#[test]
fn environment_impl_close() {
    let (_dir, env) = tmp_env();
    assert!(env.is_open());
    env.close().unwrap();
    assert!(!env.is_open());
    // Closing twice is safe.
    env.close().unwrap();
}

#[test]
fn environment_impl_ops_on_closed_env_fail() {
    let (_dir, env) = tmp_env();
    env.close().unwrap();

    let mut cfg = DatabaseConfig::new();
    cfg.set_allow_create(true);

    assert!(env.open_database("x", &cfg).is_err());
    assert!(env.begin_txn().is_err());
}

// ============================================================================
// 5. CursorImpl: basic operations, state machine
// ============================================================================

fn make_cursor_db() -> Arc<RwLock<DatabaseImpl>> {
    let db = DatabaseImpl::new(
        DatabaseId::new(1),
        "cursor_db".to_string(),
        DbType::User,
        &DatabaseConfig::default(),
    );
    Arc::new(RwLock::new(db))
}

#[test]
fn cursor_impl_new_not_initialized() {
    let db = make_cursor_db();
    let cursor = CursorImpl::new(db, 100);
    assert!(!cursor.is_initialized());
    assert!(!cursor.is_closed());
    assert!(cursor.get_current_key().is_none());
    assert!(cursor.get_current_data().is_none());
    assert_eq!(cursor.get_locker_id(), 100);
}

#[test]
fn cursor_impl_search_positions_cursor() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);

    // Insert into the tree first, then search.
    cursor.put(b"key1", b"val1", PutMode::Overwrite).unwrap();
    let status =
        cursor.search(b"key1", Some(b"val1"), SearchMode::Set).unwrap();
    assert_eq!(status, OperationStatus::Success);
    assert!(cursor.is_initialized());
    assert_eq!(cursor.get_current_key(), Some(b"key1".as_slice()));
    assert_eq!(cursor.get_current_data(), Some(b"val1".as_slice()));
}

#[test]
fn cursor_impl_get_current_after_search() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);

    cursor.put(b"k", b"v", PutMode::Overwrite).unwrap();
    cursor.search(b"k", Some(b"v"), SearchMode::Set).unwrap();
    let (key, data) = cursor.get_current().unwrap();
    assert_eq!(key, b"k");
    assert_eq!(data, b"v");
}

#[test]
fn cursor_impl_get_current_uninitialized_fails() {
    let db = make_cursor_db();
    let cursor = CursorImpl::new(db, 1);
    assert!(cursor.get_current().is_err());
}

#[test]
fn cursor_impl_retrieve_next_from_uninitialized() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);
    let status = cursor.retrieve_next(GetMode::Next).unwrap();
    assert_eq!(status, OperationStatus::NotFound);
}

#[test]
fn cursor_impl_put_overwrite() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);
    let status = cursor.put(b"key", b"data", PutMode::Overwrite).unwrap();
    assert_eq!(status, OperationStatus::Success);
    assert!(cursor.is_initialized());
    assert_eq!(cursor.get_current_key(), Some(b"key".as_slice()));
}

#[test]
fn cursor_impl_put_no_overwrite_key_exists_returns_key_exist() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);

    cursor.put(b"key", b"v1", PutMode::Overwrite).unwrap();
    let status = cursor.put(b"key", b"v2", PutMode::NoOverwrite).unwrap();
    assert_eq!(status, OperationStatus::KeyExist);
}

#[test]
fn cursor_impl_put_current_requires_initialized() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);
    let result = cursor.put(b"k", b"v", PutMode::Current);
    assert!(result.is_err());
}

#[test]
fn cursor_impl_put_current_updates_data() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);
    cursor.put(b"k", b"v1", PutMode::Overwrite).unwrap();
    cursor.search(b"k", Some(b"v1"), SearchMode::Set).unwrap();
    cursor.put(b"k", b"v2", PutMode::Current).unwrap();
    assert_eq!(cursor.get_current_data(), Some(b"v2".as_slice()));
}

#[test]
fn cursor_impl_delete_resets_state() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);

    cursor.put(b"k", b"v", PutMode::Overwrite).unwrap();
    cursor.search(b"k", Some(b"v"), SearchMode::Set).unwrap();
    assert!(cursor.is_initialized());

    let status = cursor.delete().unwrap();
    assert_eq!(status, OperationStatus::Success);
    assert!(!cursor.is_initialized());
    assert!(cursor.get_current_key().is_none());
}

#[test]
fn cursor_impl_delete_uninitialized_fails() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);
    assert!(cursor.delete().is_err());
}

#[test]
fn cursor_impl_count_returns_one_after_search() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);
    cursor.put(b"k", b"v", PutMode::Overwrite).unwrap();
    cursor.search(b"k", Some(b"v"), SearchMode::Set).unwrap();
    assert_eq!(cursor.count().unwrap(), 1);
}

#[test]
fn cursor_impl_dup_same_position() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);
    cursor.put(b"k", b"v", PutMode::Overwrite).unwrap();
    cursor.search(b"k", Some(b"v"), SearchMode::Set).unwrap();

    let dup = cursor.dup(true).unwrap();
    assert!(dup.is_initialized());
    assert_eq!(dup.get_current_key(), Some(b"k".as_slice()));
    assert_eq!(dup.get_locker_id(), 1);
    assert_ne!(cursor.get_id(), dup.get_id());
}

#[test]
fn cursor_impl_dup_no_position() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);
    cursor.put(b"k", b"v", PutMode::Overwrite).unwrap();
    cursor.search(b"k", Some(b"v"), SearchMode::Set).unwrap();

    let dup = cursor.dup(false).unwrap();
    assert!(!dup.is_initialized());
    assert!(dup.get_current_key().is_none());
}

#[test]
fn cursor_impl_close_sets_state() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);
    cursor.close().unwrap();
    assert!(cursor.is_closed());
}

#[test]
fn cursor_impl_ops_after_close_fail() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);
    cursor.close().unwrap();

    assert!(cursor.search(b"k", None, SearchMode::Set).is_err());
    assert!(cursor.get_current().is_err());
    assert!(cursor.retrieve_next(GetMode::Next).is_err());
    assert!(cursor.put(b"k", b"v", PutMode::Overwrite).is_err());
    assert!(cursor.delete().is_err());
    assert!(cursor.count().is_err());
    assert!(cursor.dup(true).is_err());
}

#[test]
fn cursor_impl_close_idempotent() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);
    cursor.close().unwrap();
    cursor.close().unwrap(); // Must not panic.
    assert!(cursor.is_closed());
}

#[test]
fn cursor_impl_unique_ids() {
    let db = make_cursor_db();
    let c1 = CursorImpl::new(db.clone(), 1);
    let c2 = CursorImpl::new(db.clone(), 1);
    let c3 = CursorImpl::new(db, 1);
    assert_ne!(c1.get_id(), c2.get_id());
    assert_ne!(c2.get_id(), c3.get_id());
    assert_ne!(c1.get_id(), c3.get_id());
}

#[test]
fn cursor_impl_search_modes_all_succeed() {
    let db = make_cursor_db();
    // Pre-insert the key so search can find it in all modes.
    {
        let mut cursor = CursorImpl::new(db.clone(), 1);
        cursor.put(b"key", b"val", PutMode::Overwrite).unwrap();
    }
    for mode in [
        SearchMode::Set,
        SearchMode::Both,
        SearchMode::SetRange,
        SearchMode::BothRange,
    ] {
        let mut cursor = CursorImpl::new(db.clone(), 1);
        let status = cursor.search(b"key", Some(b"val"), mode).unwrap();
        assert_eq!(
            status,
            OperationStatus::Success,
            "mode {:?} should succeed",
            mode
        );
    }
}

// ============================================================================
// 5b. CursorImpl traversal — get_first, get_last, retrieve_next
//
// These tests mirror CursorImplTest for the basic traversal path:
//   positionFirstOrLast / getNext / getPrev.
// ============================================================================

/// `get_first` on an empty database returns NotFound.
#[test]
fn cursor_get_first_empty_returns_not_found() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);
    let status = cursor.get_first().unwrap();
    assert_eq!(status, OperationStatus::NotFound);
}

/// `get_last` on an empty database returns NotFound.
#[test]
fn cursor_get_last_empty_returns_not_found() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);
    let status = cursor.get_last().unwrap();
    assert_eq!(status, OperationStatus::NotFound);
}

/// `get_first` positions the cursor at the smallest key.
///
/// CursorImplTest: after inserting multiple keys, getFirst()
/// must land on the smallest one.
#[test]
fn cursor_get_first_positions_at_smallest_key() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);

    // Insert three keys in non-sorted order.
    cursor.put(b"cherry", b"c", PutMode::Overwrite).unwrap();
    cursor.put(b"apple", b"a", PutMode::Overwrite).unwrap();
    cursor.put(b"banana", b"b", PutMode::Overwrite).unwrap();

    let status = cursor.get_first().unwrap();
    assert_eq!(status, OperationStatus::Success);
    assert!(cursor.is_initialized());

    // The cursor should be on "apple" (byte-lexicographic minimum).
    assert_eq!(cursor.get_current_key(), Some(b"apple".as_slice()));
    assert_eq!(cursor.get_current_data(), Some(b"a".as_slice()));
}

/// `get_last` positions the cursor at the largest key.
///
/// CursorImplTest: after inserting multiple keys, getLast()
/// must land on the largest one.
#[test]
fn cursor_get_last_positions_at_largest_key() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);

    cursor.put(b"apple", b"a", PutMode::Overwrite).unwrap();
    cursor.put(b"cherry", b"c", PutMode::Overwrite).unwrap();
    cursor.put(b"banana", b"b", PutMode::Overwrite).unwrap();

    let status = cursor.get_last().unwrap();
    assert_eq!(status, OperationStatus::Success);
    assert!(cursor.is_initialized());

    // "cherry" is the largest key byte-lexicographically.
    assert_eq!(cursor.get_current_key(), Some(b"cherry".as_slice()));
    assert_eq!(cursor.get_current_data(), Some(b"c".as_slice()));
}

/// `get_first` + sequential `retrieve_next` (GetMode::Next) iterates in
/// sorted order.
///
/// CursorImplTest scan: get first, then repeatedly getNext until
/// NotFound.
#[test]
fn cursor_iterate_forward_with_get_first_and_retrieve_next() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);

    cursor.put(b"aardvark", b"1", PutMode::Overwrite).unwrap();
    cursor.put(b"zebra", b"3", PutMode::Overwrite).unwrap();
    cursor.put(b"meerkat", b"2", PutMode::Overwrite).unwrap();

    // Position at first.
    let s = cursor.get_first().unwrap();
    assert_eq!(s, OperationStatus::Success);
    let (k, _) = cursor.get_current().unwrap();
    assert_eq!(k, b"aardvark");

    // Advance to second.
    let s = cursor.retrieve_next(GetMode::Next).unwrap();
    assert_eq!(s, OperationStatus::Success);
    let (k, _) = cursor.get_current().unwrap();
    assert_eq!(k, b"meerkat");

    // Advance to third.
    let s = cursor.retrieve_next(GetMode::Next).unwrap();
    assert_eq!(s, OperationStatus::Success);
    let (k, _) = cursor.get_current().unwrap();
    assert_eq!(k, b"zebra");

    // No more entries.
    let s = cursor.retrieve_next(GetMode::Next).unwrap();
    assert_eq!(s, OperationStatus::NotFound);
}

/// `get_last` + sequential `retrieve_next` (GetMode::Prev) iterates in
/// reverse sorted order.
///
/// CursorImplTest reverse scan.
#[test]
fn cursor_iterate_backward_with_get_last_and_retrieve_next_prev() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);

    cursor.put(b"aardvark", b"1", PutMode::Overwrite).unwrap();
    cursor.put(b"meerkat", b"2", PutMode::Overwrite).unwrap();
    cursor.put(b"zebra", b"3", PutMode::Overwrite).unwrap();

    // Position at last.
    let s = cursor.get_last().unwrap();
    assert_eq!(s, OperationStatus::Success);
    let (k, _) = cursor.get_current().unwrap();
    assert_eq!(k, b"zebra");

    // Move backward to second.
    let s = cursor.retrieve_next(GetMode::Prev).unwrap();
    assert_eq!(s, OperationStatus::Success);
    let (k, _) = cursor.get_current().unwrap();
    assert_eq!(k, b"meerkat");

    // Move backward to first.
    let s = cursor.retrieve_next(GetMode::Prev).unwrap();
    assert_eq!(s, OperationStatus::Success);
    let (k, _) = cursor.get_current().unwrap();
    assert_eq!(k, b"aardvark");

    // No more entries backward.
    let s = cursor.retrieve_next(GetMode::Prev).unwrap();
    assert_eq!(s, OperationStatus::NotFound);
}

/// `retrieve_next` from an uninitialized cursor returns NotFound.
///
/// : CursorImpl.getNext() asserts mustBeInitialized.
#[test]
fn cursor_retrieve_next_uninitialized_returns_not_found() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);
    let s = cursor.retrieve_next(GetMode::Next).unwrap();
    assert_eq!(s, OperationStatus::NotFound);
}

/// Single-element database: get_first, retrieve_next returns NotFound.
#[test]
fn cursor_single_element_get_first_then_next_is_not_found() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);
    cursor.put(b"only", b"one", PutMode::Overwrite).unwrap();

    let s = cursor.get_first().unwrap();
    assert_eq!(s, OperationStatus::Success);
    assert_eq!(cursor.get_current_key(), Some(b"only".as_slice()));

    let s = cursor.retrieve_next(GetMode::Next).unwrap();
    assert_eq!(s, OperationStatus::NotFound);
}

/// Single-element database: get_last, retrieve_next(Prev) returns NotFound.
#[test]
fn cursor_single_element_get_last_then_prev_is_not_found() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);
    cursor.put(b"only", b"one", PutMode::Overwrite).unwrap();

    let s = cursor.get_last().unwrap();
    assert_eq!(s, OperationStatus::Success);
    assert_eq!(cursor.get_current_key(), Some(b"only".as_slice()));

    let s = cursor.retrieve_next(GetMode::Prev).unwrap();
    assert_eq!(s, OperationStatus::NotFound);
}

/// Range search (SetRange) positions at first key >= search key.
///
/// CursorImpl.searchRange(): with keys "apple", "banana", "cherry",
/// searching for "b" should position at "banana".
#[test]
fn cursor_search_range_positions_at_first_ge_key() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);

    cursor.put(b"apple", b"a", PutMode::Overwrite).unwrap();
    cursor.put(b"banana", b"b", PutMode::Overwrite).unwrap();
    cursor.put(b"cherry", b"c", PutMode::Overwrite).unwrap();

    // Search for "b" — should land on "banana" (first key >= "b").
    let s = cursor.search(b"b", None, SearchMode::SetRange).unwrap();
    assert_eq!(s, OperationStatus::Success);
    assert_eq!(cursor.get_current_key(), Some(b"banana".as_slice()));
}

/// Range search beyond all keys returns NotFound.
#[test]
fn cursor_search_range_beyond_last_key_returns_not_found() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);

    cursor.put(b"apple", b"a", PutMode::Overwrite).unwrap();
    cursor.put(b"banana", b"b", PutMode::Overwrite).unwrap();

    // "z" is beyond all keys.
    let s = cursor.search(b"z", None, SearchMode::SetRange).unwrap();
    assert_eq!(s, OperationStatus::NotFound);
}

/// After insert, search finds the record and the data matches.
#[test]
fn cursor_put_then_search_retrieves_correct_data() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);

    cursor.put(b"hello", b"world", PutMode::Overwrite).unwrap();

    let s = cursor.search(b"hello", None, SearchMode::Set).unwrap();
    assert_eq!(s, OperationStatus::Success);
    let (k, v) = cursor.get_current().unwrap();
    assert_eq!(k, b"hello");
    assert_eq!(v, b"world");
}

/// After delete, search returns NotFound.
#[test]
fn cursor_delete_then_search_returns_not_found() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);

    cursor.put(b"key", b"val", PutMode::Overwrite).unwrap();
    cursor.search(b"key", None, SearchMode::Set).unwrap();
    cursor.delete().unwrap();

    let s = cursor.search(b"key", None, SearchMode::Set).unwrap();
    assert_eq!(s, OperationStatus::NotFound);
}

/// get_first on a database that had a key deleted returns NotFound when empty.
#[test]
fn cursor_get_first_after_all_keys_deleted_returns_not_found() {
    let db = make_cursor_db();
    let mut cursor = CursorImpl::new(db, 1);

    cursor.put(b"only", b"one", PutMode::Overwrite).unwrap();
    cursor.search(b"only", None, SearchMode::Set).unwrap();
    cursor.delete().unwrap();

    // Tree is now empty.
    let s = cursor.get_first().unwrap();
    assert_eq!(s, OperationStatus::NotFound);
}

// ============================================================================
// 6. EnvironmentImpl + CursorImpl end-to-end
// ============================================================================

#[test]
fn environment_cursor_put_get_delete() {
    let (_dir, env) = tmp_env();
    let mut cfg = DatabaseConfig::new();
    cfg.set_allow_create(true);

    let db_arc = env.open_database("e2e", &cfg).unwrap();

    let mut cursor = CursorImpl::new(db_arc, 1);

    // Put.
    cursor.put(b"hello", b"world", PutMode::Overwrite).unwrap();
    assert_eq!(cursor.get_current_key(), Some(b"hello".as_slice()));
    assert_eq!(cursor.get_current_data(), Some(b"world".as_slice()));

    // Search re-positions.
    cursor.search(b"hello", None, SearchMode::Set).unwrap();
    let (k, _) = cursor.get_current().unwrap();
    assert_eq!(k, b"hello");

    // Delete.
    cursor.delete().unwrap();
    assert!(!cursor.is_initialized());

    cursor.close().unwrap();
    env.close().unwrap();
}

// ============================================================================
// UtilizationTracker wiring
// ============================================================================

/// Writes through `EnvironmentImpl` update the `UtilizationTracker`.
///
/// Uses `CursorImpl::with_log_manager` so LN writes flow through the real
/// `LogManager`, which notifies the `UtilizationTrackerObserver` on each
/// `log()` call.  Verifies that after 10 puts the tracker holds at least one
/// file summary with a non-zero `total_count`.
#[test]
fn utilization_tracker_is_populated_after_writes() {
    let dir = TempDir::new().unwrap();
    let env = EnvironmentImpl::new(dir.path(), false, true).unwrap();

    let lm = env
        .get_log_manager()
        .expect("transactional env must have a LogManager");

    let mut cfg = DatabaseConfig::new();
    cfg.set_allow_create(true);
    let db_arc = env.open_database("util_test", &cfg).unwrap();

    // Use with_log_manager so LN writes go through LogManager (and trigger
    // the UtilizationTrackerObserver on each write).
    let mut cursor = CursorImpl::with_log_manager(db_arc, 1, lm);

    for i in 0u8..10 {
        cursor.put(&[i], b"value", PutMode::Overwrite).unwrap();
    }
    cursor.close().unwrap();

    // The UtilizationTracker must be populated with at least one file having
    // a non-zero total_count.
    let tracker_lock = env
        .get_utilization_tracker()
        .expect("UtilizationTracker must be wired into EnvironmentImpl");

    let tracker = tracker_lock.lock();
    let has_entries = tracker
        .get_tracked_files()
        .values()
        .any(|t| t.get_summary().total_count > 0);

    assert!(
        has_entries,
        "UtilizationTracker has no tracked entries after writes — \
         count_new_log_entry is not being called from the write path"
    );
}

/// Reproduce-first (OBSOLETE-ACCOUNTING / CLN-9/10 + L-10): overwriting
/// every record makes the prior versions obsolete in BOTH the per-file
/// FileSummary AND the per-DB DbFileSummary, and the per-DB axis records the
/// db that owns the bytes.
///
/// FAIL-PRE (on main): the `LogWriteObserver` carries no db_id, so the
/// per-DB `DbFileSummary` map has NO live producer — `get_db_file_summary`
/// would return `None` even after many overwrites.  Per-file obsolete
/// counting worked, but the per-DB axis (CLN-9) did not exist, and there was
/// only one obsolete-counting method (CLN-10).
///
/// PASS-POST: each auto-commit overwrite counts the prior version obsolete
/// (exact, with db_id), so both the per-file and per-DB obsolete counters
/// climb and `get_db_file_summary(db_id, file)` is populated.
#[test]
fn overwrites_count_prior_versions_obsolete_per_db() {
    let dir = TempDir::new().unwrap();
    let env = EnvironmentImpl::new(dir.path(), false, true).unwrap();
    let lm = env.get_log_manager().expect("writable env has LogManager");

    let mut cfg = DatabaseConfig::new();
    cfg.set_allow_create(true);
    let db_arc = env.open_database("obsolete_test", &cfg).unwrap();
    let db_id = db_arc.read().get_id().id() as u32;

    let mut cursor = CursorImpl::with_log_manager(db_arc, 1, lm);

    const N: u8 = 20;
    // Initial write of N records (auto-commit).
    for i in 0u8..N {
        cursor.put(&[i], b"v0", PutMode::Overwrite).unwrap();
    }
    // Overwrite every record 3 times.  Each overwrite supersedes the prior
    // version, which must be counted obsolete.
    for round in 1u8..=3 {
        for i in 0u8..N {
            let v = [b'v', round];
            cursor.put(&[i], &v, PutMode::Overwrite).unwrap();
        }
    }
    cursor.close().unwrap();

    let tracker = env.get_utilization_tracker().unwrap().lock();

    // Per-file: the obsolete LN count across all tracked files must equal the
    // number of superseded versions: N records * 3 overwrite rounds = 60.
    let total_obsolete_ln: i32 = tracker
        .get_tracked_files()
        .values()
        .map(|t| t.get_summary().obsolete_ln_count)
        .sum();
    assert_eq!(
        total_obsolete_ln,
        (N as i32) * 3,
        "each of the {N} records overwritten 3 times must produce 60 \
         obsolete prior versions in the per-file summaries"
    );

    // Per-DB axis (CLN-9): the DbFileSummary for this db must exist and have
    // recorded the obsolete LNs.  On main this map has no producer.
    let per_db_obsolete: i32 = tracker
        .get_tracked_files()
        .keys()
        .filter_map(|&file| tracker.get_db_file_summary(db_id, file))
        .map(|s| s.obsolete_ln_count)
        .sum();
    assert_eq!(
        per_db_obsolete,
        (N as i32) * 3,
        "CLN-9: the per-DB DbFileSummary must record the same obsolete LN \
         count (the per-DB axis has a live producer)"
    );
}

/// End-to-end (OBSOLETE-ACCOUNTING): overwriting every record makes the
/// original file's utilization drop toward zero, so the cleaner can reclaim
/// it.
///
/// Fill file 0 with records, then overwrite each record repeatedly so the new
/// versions roll into later files.  Every prior version in file 0 is counted
/// obsolete, so file 0's obsolete bytes approach its total bytes (utilization
/// -> ~0).
///
/// FAIL-PRE (on main): the obsolete LSN passed at the write site was the
/// cursor's stale `current_lsn` (belonging to a different key), so file 0's
/// obsolete count did not track the real superseded versions — utilization
/// stayed high and the file was never selected for cleaning.
#[test]
fn overwriting_all_records_makes_original_file_cleanable() {
    use noxu_dbi::DbiEnvConfig;

    let dir = TempDir::new().unwrap();
    // Small log files so the records spread across several files and the
    // overwrites roll past file 0.
    let cfg = DbiEnvConfig {
        transactional: true,
        log_file_max_bytes: 4096,
        run_cleaner: false,
        run_checkpointer: false,
        run_in_compressor: false,
        ..DbiEnvConfig::default()
    };
    let env = EnvironmentImpl::from_dbi_config(dir.path(), &cfg).unwrap();
    let lm = env.get_log_manager().expect("writable env has LogManager");

    let mut db_cfg = DatabaseConfig::new();
    db_cfg.set_allow_create(true);
    let db_arc = env.open_database("cleanable", &db_cfg).unwrap();
    let mut cursor = CursorImpl::with_log_manager(db_arc, 1, lm);

    // Phase 1: write enough records to fill file 0 and spill into file 1+.
    const N: u16 = 60;
    for i in 0u16..N {
        let key = format!("key{i:04}");
        cursor
            .put(key.as_bytes(), b"initial-value-payload", PutMode::Overwrite)
            .unwrap();
    }

    // Snapshot file 0's totals after the initial fill.
    let (file0_total, file0_obsolete_before) = {
        let t = env.get_utilization_tracker().unwrap().lock();
        let s = t.get_tracked_summary(0).expect("file 0 must be tracked");
        (s.get_summary().total_ln_count, s.get_summary().obsolete_ln_count)
    };
    assert!(file0_total > 0, "file 0 must hold LN records after the fill");

    // Phase 2: overwrite every record several times so all of file 0's
    // versions are superseded by versions in later files.
    for round in 0u16..5 {
        for i in 0u16..N {
            let key = format!("key{i:04}");
            let val = format!("round{round}-value-payload");
            cursor
                .put(key.as_bytes(), val.as_bytes(), PutMode::Overwrite)
                .unwrap();
        }
    }
    cursor.close().unwrap();

    // File 0's obsolete LN count must now cover (nearly) all of its records:
    // every initial version was superseded.  Utilization (active / total)
    // drops toward zero, making the file cleanable.
    let t = env.get_utilization_tracker().unwrap().lock();
    let s = t.get_tracked_summary(0).unwrap();
    let obsolete_after = s.get_summary().obsolete_ln_count;
    assert!(
        obsolete_after > file0_obsolete_before,
        "file 0 obsolete count must rise after overwrites"
    );
    // Nearly every record originally in file 0 is now obsolete (allow a
    // one-record boundary slack: a record straddling the file 0/1 flip may
    // have its only surviving version counted against the later file).
    assert!(
        obsolete_after >= file0_total - 1,
        "nearly every record originally in file 0 ({file0_total}) must be \
         obsolete after overwriting all keys; got {obsolete_after}"
    );
}

/// L-6: a deleted LN is counted obsolete at WRITE time (the logged deletion
/// is immediately obsolete — it can be ignored by the cleaner).
///
/// JE `LNLogEntry.isImmediatelyObsolete`: `ln.isDeleted() || ...` -> the
/// write path counts the new entry obsolete via `countObsoleteNodeInexact`.
///
/// FAIL-PRE (on main): the DeleteLN write path never counted the deletion
/// entry obsolete, so the deletion records accumulated as "active" bytes the
/// cleaner could not reclaim.
#[test]
fn deleted_ln_is_counted_obsolete_at_write() {
    let dir = TempDir::new().unwrap();
    let env = EnvironmentImpl::new(dir.path(), false, true).unwrap();
    let lm = env.get_log_manager().expect("writable env has LogManager");

    let mut cfg = DatabaseConfig::new();
    cfg.set_allow_create(true);
    let db_arc = env.open_database("del_test", &cfg).unwrap();
    let mut cursor = CursorImpl::with_log_manager(db_arc, 1, lm);

    // Insert then delete a record.
    cursor.put(b"k", b"v", PutMode::Overwrite).unwrap();
    cursor.search(b"k", Some(b"v"), SearchMode::Set).unwrap();
    let obsolete_before: i32 = {
        let t = env.get_utilization_tracker().unwrap().lock();
        t.get_tracked_files()
            .values()
            .map(|s| s.get_summary().obsolete_ln_count)
            .sum()
    };
    cursor.delete().unwrap();
    cursor.close().unwrap();

    let obsolete_after: i32 = {
        let t = env.get_utilization_tracker().unwrap().lock();
        t.get_tracked_files()
            .values()
            .map(|s| s.get_summary().obsolete_ln_count)
            .sum()
    };
    // The delete must produce at least one new obsolete LN: the deletion
    // entry itself (immediately obsolete, inexact) plus the superseded live
    // version.
    assert!(
        obsolete_after > obsolete_before,
        "deleting a record must count at least one obsolete LN at write time \
         (L-6); before={obsolete_before}, after={obsolete_after}"
    );
}

// ============================================================================
// CLN-4 (HEADLINE): cleaner relies on persisted utilization after restart
// ============================================================================

/// HEADLINE (CLN-4 + C7 + cleaner-relies): write+overwrite records to
/// accumulate obsolete bytes, checkpoint (persist_file_summaries writes the
/// FileSummaryLN records), CLOSE, REOPEN (recovery rebuilds the profile), then
/// the cleaner IMMEDIATELY sees the obsolete bytes for file 0 WITHOUT having
/// re-warmed from any new writes.
///
/// FAIL-PRE (on main): recovery never read FileSummaryLN back, so the cleaner's
/// UtilizationProfile was empty after restart — `get_profile_summary(0)` was
/// `None` and the cleaner was blind until live writes re-warmed it (CLN-6).
///
/// PASS-POST (CLN-4): recovery rebuilds the profile from the persisted
/// FileSummaryLN (latest per file), env-open seeds the cleaner via
/// `seed_profile`, so `get_profile_summary(0)` shows the obsolete LN bytes
/// immediately and the file is selectable for cleaning.
///
/// JE: UtilizationProfile.populateCache / RecoveryUtilizationTracker.
#[test]
fn cleaner_sees_persisted_obsolete_immediately_after_restart() {
    use noxu_dbi::DbiEnvConfig;

    let dir = TempDir::new().unwrap();
    let cfg = DbiEnvConfig {
        transactional: true,
        // Small files so the overwrites roll past file 0.
        log_file_max_bytes: 4096,
        run_cleaner: false,
        run_checkpointer: false,
        run_in_compressor: false,
        ..DbiEnvConfig::default()
    };

    // --- Session 1: write, overwrite, checkpoint, close. ---
    {
        let env = EnvironmentImpl::from_dbi_config(dir.path(), &cfg).unwrap();
        let lm = env.get_log_manager().expect("writable env has LogManager");
        let mut db_cfg = DatabaseConfig::new();
        db_cfg.set_allow_create(true);
        let db_arc = env.open_database("cleanable", &db_cfg).unwrap();
        let mut cursor = CursorImpl::with_log_manager(db_arc, 1, lm);

        const N: u16 = 60;
        for i in 0u16..N {
            let key = format!("key{i:04}");
            cursor
                .put(
                    key.as_bytes(),
                    b"initial-value-payload",
                    PutMode::Overwrite,
                )
                .unwrap();
        }
        // Overwrite every record several times so file 0's versions are
        // all superseded by later files.
        for round in 0u16..5 {
            for i in 0u16..N {
                let key = format!("key{i:04}");
                let val = format!("round{round}-value-payload");
                cursor
                    .put(key.as_bytes(), val.as_bytes(), PutMode::Overwrite)
                    .unwrap();
            }
        }
        cursor.close().unwrap();

        // File 0 must have accumulated obsolete LNs in the live tracker.
        let obsolete_live = {
            let t = env.get_utilization_tracker().unwrap().lock();
            t.get_tracked_summary(0)
                .map(|s| s.get_summary().obsolete_ln_count)
                .unwrap_or(0)
        };
        assert!(
            obsolete_live > 0,
            "file 0 must have obsolete LNs before checkpoint; got \
             {obsolete_live}"
        );

        // Checkpoint persists the FileSummaryLN records (C7 form).
        env.run_checkpoint().unwrap();
        env.close().unwrap();
    }

    // --- Session 2: reopen.  Recovery rebuilds the profile; the cleaner
    //     sees the persisted obsolete bytes WITHOUT any new writes. ---
    let env2 = EnvironmentImpl::from_dbi_config(dir.path(), &cfg).unwrap();
    let cleaner =
        env2.get_cleaner().expect("writable env has a cleaner after reopen");

    // CLN-4: the cleaner's profile is seeded from the persisted FileSummaryLN
    // BEFORE any live write.  On main this is None (profile empty / blind).
    let seeded = cleaner.get_profile_summary(0);
    assert!(
        seeded.is_some(),
        "CLN-4: after restart the cleaner's profile must be seeded from the \
         persisted FileSummaryLN for file 0 (was None/blind on main)"
    );
    let summary = seeded.unwrap();
    assert!(
        summary.obsolete_ln_count > 0,
        "CLN-4: the seeded profile must show file 0's obsolete LNs \
         immediately after restart; got {}",
        summary.obsolete_ln_count
    );
    // The file's utilization must be measurably below full (obsolete bytes
    // present), so it is selectable for cleaning without re-warming.
    assert!(
        summary.get_obsolete_size() > 0,
        "CLN-4: file 0 must report obsolete bytes so the cleaner can select \
         it for cleaning right after restart"
    );

    env2.close().unwrap();
}

// ============================================================================
// X-11: log_flush_no_sync_interval_ms daemon
// ============================================================================
/// page cache within the configured interval.
///
/// Test strategy:
/// 1. Open a writable env with `log_flush_no_sync_interval_ms = 50` ms.
/// 2. Write a record using CommitNoSync (no flush/fsync from the committer).
/// 3. Wait 200 ms (4× the flush interval) for the daemon to run.
/// 4. Assert that the LogManager's `last_flush_lsn` has advanced past 0,
///    meaning at least one flush_no_sync() has fired.  We verify via
///    `get_log_manager().unwrap().get_last_flush_lsn()`.
#[test]
fn test_x11_log_flush_no_sync_daemon_fires() {
    use noxu_dbi::{DatabaseConfig, DbiEnvConfig, EnvironmentImpl};

    let dir = TempDir::new().unwrap();

    let cfg = DbiEnvConfig {
        transactional: true,
        log_flush_no_sync_interval_ms: 50, // 50 ms flush interval
        run_cleaner: false,
        run_checkpointer: false,
        run_in_compressor: false,
        ..DbiEnvConfig::default()
    };
    let env = EnvironmentImpl::from_dbi_config(dir.path(), &cfg).unwrap();

    // Open a database and write a record.
    let db_cfg = DatabaseConfig::new().set_allow_create(true).clone();
    let db_arc = env.open_database("test", &db_cfg).unwrap();

    {
        let db = db_arc.read();
        let tree = db.get_real_tree().expect("tree must be present");
        let lsn = noxu_util::Lsn::from_u64(1);
        tree.insert(b"k1".to_vec(), b"v1".to_vec(), lsn).unwrap();
    }

    // Write something to the log manager directly to ensure there's data to flush.
    // Use a raw log write to simulate CommitNoSync (flush=false, fsync=false).
    let lm = env.get_log_manager().expect("log manager must be present");
    let mut buf = bytes::BytesMut::with_capacity(32);
    let entry = noxu_log::entry::LnLogEntry::new(
        1,
        None,
        noxu_util::lsn::NULL_LSN,
        false,
        None,
        None,
        noxu_util::vlsn::NULL_VLSN,
        0,
        false,
        b"k1".to_vec(),
        Some(b"v1".to_vec()),
        0,
        noxu_util::vlsn::NULL_VLSN,
    );
    entry.write_to_log(&mut buf);
    lm.log(
        noxu_log::LogEntryType::UpdateLN,
        &buf,
        noxu_log::Provisional::No,
        false, // flush=false (CommitNoSync)
        false, // fsync=false
    )
    .expect("log write should succeed");

    // Record last_flush_lsn before waiting.
    let before = lm.get_last_flush_lsn();

    // Wait long enough for the 50-ms daemon to have run multiple times.
    std::thread::sleep(std::time::Duration::from_millis(300));

    let after = lm.get_last_flush_lsn();

    env.close().unwrap();

    assert!(
        after > before,
        "LogFlushTask daemon must have advanced last_flush_lsn: \
         before={before:?} after={after:?}"
    );
}

/// X-11: When log_flush_no_sync_interval_ms = 0 the daemon must be a no-op
/// (the thread exits immediately without firing any flushes beyond env open).
#[test]
fn test_x11_disabled_interval_no_spurious_flush() {
    use noxu_dbi::{DbiEnvConfig, EnvironmentImpl};

    let dir = TempDir::new().unwrap();
    let cfg = DbiEnvConfig {
        transactional: true,
        log_flush_no_sync_interval_ms: 0, // disabled
        run_cleaner: false,
        run_checkpointer: false,
        run_in_compressor: false,
        ..DbiEnvConfig::default()
    };
    // Should open and close without error.
    let env = EnvironmentImpl::from_dbi_config(dir.path(), &cfg).unwrap();
    env.close().unwrap();
}

// ============================================================================
// X-12: cache_size total budget model
// ============================================================================

/// X-12: Verify that the Arbiter budget = cache_size - log_buf_total.
///
/// When cache_size = 64 MiB, log_buffer_size = 1 MiB (×3 buffers), and
/// max_off_heap_memory = 0, the Arbiter must use 64 - 3 = 61 MiB, not 64 MiB.
#[test]
fn test_x12_arbiter_budget_subtracts_log_buffers() {
    use noxu_dbi::{DbiEnvConfig, EnvironmentImpl};

    let dir = TempDir::new().unwrap();

    let cache_size: u64 = 64 * 1024 * 1024;
    let buf_size: usize = 1024 * 1024; // 1 MiB per buffer
    let num_bufs: usize = 3;

    let cfg = DbiEnvConfig {
        cache_size,
        log_buffer_size: buf_size,
        log_num_buffers: num_bufs,
        max_off_heap_memory: 0,
        run_cleaner: false,
        run_checkpointer: false,
        run_in_compressor: false,
        log_flush_no_sync_interval_ms: 0,
        ..DbiEnvConfig::default()
    };
    let env = EnvironmentImpl::from_dbi_config(dir.path(), &cfg).unwrap();

    let expected_budget = (cache_size as i64) - (num_bufs * buf_size) as i64;
    let actual_budget = env.get_arbiter_max_memory();
    env.close().unwrap();

    assert_eq!(
        actual_budget, expected_budget,
        "X-12: Arbiter budget must equal cache_size - log_buf_total. \
         expected={expected_budget} actual={actual_budget}"
    );
}

/// X-12: off-heap reservation is also subtracted from the arbiter budget.
#[test]
fn test_x12_arbiter_budget_subtracts_off_heap() {
    use noxu_dbi::{DbiEnvConfig, EnvironmentImpl};

    let dir = TempDir::new().unwrap();

    let cache_size: u64 = 128 * 1024 * 1024;
    let buf_size: usize = 512 * 1024;
    let num_bufs: usize = 3;
    let off_heap: u64 = 32 * 1024 * 1024;

    let cfg = DbiEnvConfig {
        cache_size,
        log_buffer_size: buf_size,
        log_num_buffers: num_bufs,
        max_off_heap_memory: off_heap,
        run_cleaner: false,
        run_checkpointer: false,
        run_in_compressor: false,
        log_flush_no_sync_interval_ms: 0,
        ..DbiEnvConfig::default()
    };
    let env = EnvironmentImpl::from_dbi_config(dir.path(), &cfg).unwrap();

    let expected_budget =
        (cache_size as i64) - (num_bufs * buf_size) as i64 - off_heap as i64;
    let actual_budget = env.get_arbiter_max_memory();
    env.close().unwrap();

    assert_eq!(
        actual_budget, expected_budget,
        "X-12: Arbiter budget must subtract off-heap reservation too. \
         expected={expected_budget} actual={actual_budget}"
    );
}

// ── R-4 TOCTOU / O(N) scan fix tests ─────────────────────────────────────────

/// Regression test for re-audit-keith R-4.
///
/// Verifies that `commit_pending_database` + `abort_pending_database` are
/// correct and that the pending_names HashMap stores the db_id so no O(N)
/// scan is needed.
#[test]
fn test_commit_pending_database_no_toctou() {
    use noxu_dbi::{DatabaseConfig, DbiEnvConfig, EnvironmentImpl};
    let dir = tempfile::TempDir::new().unwrap();
    let cfg = DbiEnvConfig {
        transactional: true,
        run_cleaner: false,
        run_checkpointer: false,
        run_in_compressor: false,
        ..DbiEnvConfig::default()
    };
    let env = EnvironmentImpl::from_dbi_config(dir.path(), &cfg).unwrap();

    let db_cfg = {
        let mut c = DatabaseConfig::new();
        c.set_allow_create(true);
        c
    };

    // Simulate transactional database creation:
    // open with txn_id=42 → name goes to pending_names (not name_map)
    let txn_id: u64 = 42;
    let _db_arc =
        env.open_database_transactional("pending_db", &db_cfg, txn_id).unwrap();

    // Database is not visible in committed names yet.
    assert!(
        !env.get_database_names().contains(&"pending_db".to_string()),
        "pending db should not appear in committed names before commit"
    );

    // Commit: name moves atomically from pending_names → name_map.
    env.commit_pending_database("pending_db");

    // Database is now visible.
    assert!(
        env.get_database_names().contains(&"pending_db".to_string()),
        "committed db should appear in get_database_names()"
    );

    // Idempotent: calling commit again is a no-op.
    env.commit_pending_database("pending_db");

    env.close().unwrap();
}

#[test]
fn test_abort_pending_database() {
    use noxu_dbi::{DatabaseConfig, DbiEnvConfig, EnvironmentImpl};
    let dir = tempfile::TempDir::new().unwrap();
    let cfg = DbiEnvConfig {
        transactional: true,
        run_cleaner: false,
        run_checkpointer: false,
        run_in_compressor: false,
        ..DbiEnvConfig::default()
    };
    let env = EnvironmentImpl::from_dbi_config(dir.path(), &cfg).unwrap();

    let db_cfg = {
        let mut c = DatabaseConfig::new();
        c.set_allow_create(true);
        c
    };
    let txn_id: u64 = 99;
    let _db_arc =
        env.open_database_transactional("aborted_db", &db_cfg, txn_id).unwrap();

    // Abort: name is removed from pending_names and db_map.
    env.abort_pending_database("aborted_db");

    // Database is not visible.
    assert!(
        !env.get_database_names().contains(&"aborted_db".to_string()),
        "aborted db should not appear in committed names"
    );

    // Idempotent abort.
    env.abort_pending_database("aborted_db");

    env.close().unwrap();
}

/// Concurrent test: a thread committing a pending database races with a
/// thread trying to open the same name.  The concurrent open must either
/// see the committed name or get DatabaseAlreadyExists — never create a
/// duplicate DatabaseImpl with a different db_id.
#[test]
fn test_commit_pending_concurrent_open_no_duplicate_db_id() {
    use noxu_dbi::{DatabaseConfig, DbiEnvConfig, EnvironmentImpl};
    use std::sync::Barrier;
    let dir = tempfile::TempDir::new().unwrap();
    let cfg = DbiEnvConfig {
        transactional: true,
        run_cleaner: false,
        run_checkpointer: false,
        run_in_compressor: false,
        ..DbiEnvConfig::default()
    };
    let env =
        Arc::new(EnvironmentImpl::from_dbi_config(dir.path(), &cfg).unwrap());

    let db_cfg = {
        let mut c = DatabaseConfig::new();
        c.set_allow_create(true);
        c
    };
    // Create the pending database in txn 1.
    let txn_id: u64 = 1;
    let db_arc =
        env.open_database_transactional("race_db", &db_cfg, txn_id).unwrap();
    let committed_id = db_arc.read().get_id();

    // Barrier: both threads start at the same moment.
    let barrier = Arc::new(Barrier::new(2));

    let env2 = Arc::clone(&env);
    let bar2 = Arc::clone(&barrier);
    let handle = std::thread::spawn(move || {
        bar2.wait();
        // This open may race with commit below.
        let mut cfg = DatabaseConfig::new();
        cfg.set_allow_create(true);
        env2.open_database_transactional("race_db", &cfg, 2)
    });

    barrier.wait();
    env.commit_pending_database("race_db");

    let concurrent_result = handle.join().unwrap();

    // The concurrent open must either succeed (returning the same db_id)
    // or return DatabaseAlreadyExists — never silently create a second one.
    match concurrent_result {
        Ok(arc) => {
            // If it succeeded, it must have returned the same db_id.
            assert_eq!(
                arc.read().get_id(),
                committed_id,
                "concurrent open returned a different db_id — duplicate!"
            );
        }
        Err(e) => {
            // DatabaseAlreadyExists is the expected failure mode.
            assert!(
                e.to_string().contains("already exists")
                    || e.to_string().contains("AlreadyExists"),
                "unexpected error: {e}"
            );
        }
    }

    // Exactly one database named "race_db" exists.
    assert_eq!(
        env.get_database_names()
            .iter()
            .filter(|n| n.as_str() == "race_db")
            .count(),
        1,
        "exactly one race_db should exist after concurrent commit"
    );

    env.close().unwrap();
}

/// GAP-8 regression: the cleaner built by EnvironmentImpl must have the
/// TxnManager wired so do_clean clamps file selection to the first-active-txn
/// window (CLN-4). Before the fix, EnvironmentImpl wired the tree-registry and
/// utilization-tracker but NOT the txn_manager, leaving first_active_txn_file
/// permanently None and the clamp inert.
#[test]
fn gap8_production_cleaner_has_txn_manager_wired() {
    use noxu_dbi::EnvironmentImpl;
    let dir = tempfile::tempdir().unwrap();
    // EnvironmentImpl::new(path, read_only=false, transactional=true)
    let env = EnvironmentImpl::new(dir.path(), false, true).expect("open env");
    let cleaner = env.get_cleaner().expect("writable env has a cleaner");
    assert!(
        cleaner.has_txn_manager(),
        "GAP-8: production cleaner must have TxnManager wired for the \
         first-active-txn clamp (CLN-4)"
    );
}

// ─── Evictor F1 + F2: keystone cache-evictor wiring ──────────────────────────
//
// F1: the evictor's LRU policy lists must be populated through the production
//     insert/access path (JE INList → Evictor.addBack/moveBack), otherwise
//     evict_batch's phase quotas are all 0 and it selects nothing.
// F2: eviction must decrement the shared cache_usage counter the Arbiter reads
//     (JE IN.updateMemorySize(-bytes) → MemoryBudget.updateTreeMemoryUsage),
//     otherwise the engine can never get back under budget by evicting.
//
// These tests FAIL against current main (evictor inert) and pass after the fix.

/// Build a tiny-cache, daemon-free environment for deterministic eviction.
fn small_cache_env(dir: &std::path::Path) -> EnvironmentImpl {
    use noxu_dbi::DbiEnvConfig;
    let cfg = DbiEnvConfig {
        // Small total budget. The arbiter floors at 1 MiB, so use a 1-buffer
        // 64 KiB log pool and a 1.5 MiB cache -> arbiter budget ~1.4 MiB.
        cache_size: 3 * 1024 * 1024 / 2, // 1.5 MiB
        log_num_buffers: 1,
        log_buffer_size: 64 * 1024,
        max_off_heap_memory: 0,
        transactional: true,
        // Disable the *background* daemons that would prune/move nodes and
        // confound the counter assertions; we drive eviction (and the
        // checkpoint that cleans dirty slots) manually.
        run_cleaner: false,
        run_checkpointer: false,
        run_in_compressor: false,
        log_flush_no_sync_interval_ms: 0,
        // LRU-only so dirty nodes are stripped/evicted without needing a
        // checkpointer daemon to log them first.
        evictor_lru_only: true,
        ..DbiEnvConfig::default()
    };
    EnvironmentImpl::from_dbi_config(dir, &cfg).unwrap()
}

/// F1: inserting records through the production cursor path must populate the
/// evictor's LRU lists (note_ins_* wired). Against main the lists stay empty.
#[test]
fn evictor_f1_lru_lists_populated_by_production_inserts() {
    let dir = TempDir::new().unwrap();
    let env = small_cache_env(dir.path());

    let db_cfg = {
        let mut c = DatabaseConfig::new();
        c.set_allow_create(true);
        c
    };
    let db = env.open_database("f1_db", &db_cfg).unwrap();

    let evictor = env.get_evictor();
    let (before_pri, before_scan, _before_pri2) = evictor.get_policy_sizes();
    let before = before_pri + before_scan;

    // Insert enough records to create several BINs (node_max_entries=128).
    let mut cursor = CursorImpl::new(Arc::clone(&db), 1);
    for i in 0..1000u32 {
        let key = format!("key-{i:08}");
        let val = vec![b'v'; 256];
        cursor.put(key.as_bytes(), &val, PutMode::Overwrite).unwrap();
    }

    let (after_pri, after_scan, _after_pri2) = evictor.get_policy_sizes();
    let after = after_pri + after_scan;

    assert!(
        after > before,
        "F1: production inserts must populate the evictor LRU lists \
         (note_ins_* wired). before={before} after={after}. \
         If after==0 the evictor is still inert."
    );
    // Multiple BINs should be tracked after 1000 inserts.
    assert!(after > 1, "F1: expected several BINs tracked, got {after}");

    env.close().unwrap();
}

/// F1 + F2: insert past the cache budget, run eviction, and assert that
/// (a) the evictor actually evicted > 0 bytes AND (b) the shared cache_usage
/// counter dropped after eviction.
#[test]
fn evictor_f1_f2_eviction_reduces_cache_usage() {
    let dir = TempDir::new().unwrap();
    let env = small_cache_env(dir.path());

    let db_cfg = {
        let mut c = DatabaseConfig::new();
        c.set_allow_create(true);
        c
    };
    let db = env.open_database("f2_db", &db_cfg).unwrap();

    // Insert enough data to exceed the (~1.4 MiB) arbiter budget. Each entry
    // accounts key+data+48 ~ 1 KiB; 4000 entries ~ 4 MiB >> budget.
    let mut cursor = CursorImpl::new(Arc::clone(&db), 1);
    for i in 0..4000u32 {
        let key = format!("key-{i:08}");
        let val = vec![b'v'; 1000];
        cursor.put(key.as_bytes(), &val, PutMode::Overwrite).unwrap();
    }

    let budget = env.get_arbiter_max_memory();
    let usage_before = env.get_cache_usage();
    assert!(
        usage_before > budget,
        "precondition: cache usage ({usage_before}) must exceed the budget \
         ({budget}) so eviction is warranted"
    );

    // Run a checkpoint so the freshly inserted (dirty) BIN slots are logged
    // and marked clean; the evictor's PartialEvict path only strips clean
    // LN data (stripping a dirty, unlogged slot would lose the write).
    // JE: the checkpointer logs dirty BINs, after which the evictor can free
    // their in-memory LN bytes.
    env.run_checkpoint().unwrap();

    // Drive eviction manually. The background daemon may also run; either
    // way we assert on the cumulative evictor stats (which count every
    // source) plus the shared counter, so the test is robust to the daemon
    // racing us.
    for _ in 0..200 {
        env.evict_memory();
        if env.get_cache_usage() <= budget {
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(2));
    }

    let usage_after = env.get_cache_usage();

    // Cumulative work done by the evictor across all sources (daemon +
    // manual). nodes_stripped counts PartialEvict (LN strip), nodes_evicted
    // counts full evicts.
    let evictor = env.get_evictor();
    let st = evictor.get_stats();
    let nodes_worked = st.get(&st.nodes_stripped) + st.get(&st.nodes_evicted);

    assert!(
        nodes_worked > 0,
        "F1: the evictor must actually evict/strip > 0 nodes once the LRU \
         lists are populated (against main: 0 — lists empty, quotas 0). \
         stripped={} evicted={}",
        st.get(&st.nodes_stripped),
        st.get(&st.nodes_evicted),
    );
    assert!(
        usage_after < usage_before,
        "F2: eviction must reduce the shared cache_usage counter the Arbiter \
         reads. before={usage_before} after={usage_after} \
         (against main: unchanged — eviction never fetch_sub's the counter)"
    );
    assert!(
        usage_after >= 0,
        "F2: the counter must be clamped at >= 0, got {usage_after}"
    );

    env.close().unwrap();
}

/// IC-3 end-to-end (fail-pre / pass-post): the predicate that `compress_all`
/// / the INCompressor daemon supply to `Tree::compress_bin_with_lock_check`
/// consults the environment's REAL `LockManager`.  A write-locked LSN reports
/// `is_locked == true` (so the compressor would SKIP that slot, mirroring JE
/// `BIN.compress` isLockUncontended), while an unlocked LSN reports `false`
/// (so a committed/unlocked tombstone IS compressed).
///
/// This exercises the dbi wiring seam (closure over the live LockManager)
/// with a real held lock; the slot-removal behavior itself is proven in the
/// noxu-tree unit headline `test_ic3_compress_skips_write_locked_slot`.
#[test]
fn ic3_compress_predicate_consults_real_lock_manager() {
    use noxu_txn::LockType;

    let (_dir, env) = tmp_env();
    let lm = env.get_lock_manager().clone();

    let locked_lsn: u64 = 0x1234_5678;
    let unlocked_lsn: u64 = 0x9abc_def0;
    let locker_id: i64 = 4242;

    // Hold a write lock on `locked_lsn`, simulating an in-flight txn that
    // still references the slot.
    lm.lock(locked_lsn, locker_id, LockType::Write, true, false)
        .expect("lock should be granted");

    // Build the EXACT predicate the compressor paths construct.
    let lm_ref = &lm;
    let is_locked = move |lsn: u64| lm_ref.get_lock_info(lsn) != (0, 0);

    assert!(
        is_locked(locked_lsn),
        "IC-3: a write-locked LSN must report locked -> compressor SKIPS it"
    );
    assert!(
        !is_locked(unlocked_lsn),
        "IC-3: an unlocked LSN must report unlocked -> compressor REMOVES it"
    );

    // After release, the slot becomes compressible (txn resolved).
    lm.release(locked_lsn, locker_id).unwrap();
    assert!(
        !is_locked(locked_lsn),
        "IC-3: once the txn releases the lock the slot is compressible"
    );

    env.close().unwrap();
}
