lean_lmdb

use super::*;

#[test]
fn batch_bounds_are_checked_before_execution() {
    assert_eq!(validate_batch_bounds(0, 0), Ok(()));
    assert_eq!(
        validate_batch_bounds(MAX_BATCH_OPERATIONS, MAX_BATCH_BYTES),
        Ok(())
    );
    assert_eq!(
        validate_batch_bounds(MAX_BATCH_OPERATIONS + 1, 0),
        Err(CrudError::BatchTooLarge)
    );
    assert_eq!(
        validate_batch_bounds(1, MAX_BATCH_BYTES + 1),
        Err(CrudError::BatchTooLarge)
    );
}

#[test]
fn ordered_batch_spans_databases_and_missing_deletes_are_idempotent() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("ordered_batch");
    let canonical = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), true)
        .expect("canonical path");
    let state =
        open_shared_environment(canonical.clone(), options(), true).expect("environment open");
    let first = database_resource(&state, "first");
    let second = database_resource(&state, "second");
    put_binary(&first, b"old", b"present").expect("initial put");

    let operations = vec![
        BatchOperation::Put {
            database: first.database,
            key: b"same".to_vec(),
            value: b"one".to_vec(),
        },
        BatchOperation::Put {
            database: second.database,
            key: vec![0xff],
            value: vec![0, 0xff],
        },
        BatchOperation::Put {
            database: first.database,
            key: b"same".to_vec(),
            value: b"two".to_vec(),
        },
        BatchOperation::Delete {
            database: first.database,
            key: b"same".to_vec(),
        },
        BatchOperation::Delete {
            database: first.database,
            key: b"missing".to_vec(),
        },
        BatchOperation::Delete {
            database: first.database,
            key: b"old".to_vec(),
        },
    ];
    write_batch_operations(&state, &operations).expect("atomic batch");
    assert_eq!(get_binary(&first, b"same"), Ok(None));
    assert_eq!(get_binary(&first, b"old"), Ok(None));
    assert_eq!(get_binary(&second, &[0xff]), Ok(Some(vec![0, 0xff])));

    drop(first);
    drop(second);
    drop(state);
    cleanup_environment(&canonical);
}

#[test]
fn map_full_batch_rolls_back_operations_across_databases() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("map_full_batch");
    let canonical = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), true)
        .expect("canonical path");
    let small = EnvironmentOptions {
        map_size: MIN_MAP_SIZE as usize,
        ..options()
    };
    let state = open_shared_environment(canonical.clone(), small, true).expect("environment open");
    let first = database_resource(&state, "first");
    let second = database_resource(&state, "second");
    put_binary(&first, b"one", b"before-one").expect("initial first value");
    put_binary(&second, b"two", b"before-two").expect("initial second value");

    let operations = vec![
        BatchOperation::Put {
            database: first.database,
            key: b"one".to_vec(),
            value: b"changed".to_vec(),
        },
        BatchOperation::Delete {
            database: second.database,
            key: b"two".to_vec(),
        },
        BatchOperation::Put {
            database: first.database,
            key: b"large".to_vec(),
            value: vec![7; 2 * MIN_MAP_SIZE as usize],
        },
    ];
    assert_eq!(
        write_batch_operations(&state, &operations),
        Err(CrudError::MapFull)
    );
    assert_eq!(get_binary(&first, b"one"), Ok(Some(b"before-one".to_vec())));
    assert_eq!(
        get_binary(&second, b"two"),
        Ok(Some(b"before-two".to_vec()))
    );
    assert_eq!(get_binary(&first, b"large"), Ok(None));

    drop(first);
    drop(second);
    drop(state);
    cleanup_environment(&canonical);
}

#[test]
fn compare_exchange_handles_empty_missing_delete_and_concurrent_contenders() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("compare_exchange");
    let canonical = canonical_environment_path(dir.to_str().expect("UTF-8 temp path"), true)
        .expect("canonical path");
    let state =
        open_shared_environment(canonical.clone(), options(), true).expect("environment open");
    let database = Arc::new(database_resource(&state, "data"));

    assert_eq!(
        compare_exchange_binary(
            &database,
            b"empty",
            &ExpectedValue::Missing,
            &ReplacementValue::Put(&[]),
            copy_bytes,
        ),
        Ok(CompareExchangeResult::Exchanged)
    );
    assert_eq!(
        compare_exchange_binary(
            &database,
            b"empty",
            &ExpectedValue::Missing,
            &ReplacementValue::Delete,
            copy_bytes,
        ),
        Ok(CompareExchangeResult::Conflict(Some(Vec::new())))
    );
    assert_eq!(
        compare_exchange_binary(
            &database,
            b"empty",
            &ExpectedValue::Value(&[]),
            &ReplacementValue::Delete,
            copy_bytes,
        ),
        Ok(CompareExchangeResult::Exchanged)
    );
    assert_eq!(get_binary(&database, b"empty"), Ok(None));

    let barrier = Arc::new(Barrier::new(12));
    let mut workers = Vec::new();
    for index in 0..12_u8 {
        let database = Arc::clone(&database);
        let barrier = Arc::clone(&barrier);
        workers.push(thread::spawn(move || {
            barrier.wait();
            let value = vec![0xff, index];
            let result = compare_exchange_binary(
                &database,
                b"contended",
                &ExpectedValue::Missing,
                &ReplacementValue::Put(value.as_slice()),
                copy_bytes,
            );
            (value, result)
        }));
    }
    let results: Vec<_> = workers
        .into_iter()
        .map(|worker| worker.join().expect("CAS worker did not panic"))
        .collect();
    let winners: Vec<_> = results
        .iter()
        .filter_map(|(value, result)| {
            (result == &Ok(CompareExchangeResult::Exchanged)).then_some(value.clone())
        })
        .collect();
    assert_eq!(winners.len(), 1);
    let winner = &winners[0];
    assert!(results.iter().all(|(value, result)| {
        (value == winner && result == &Ok(CompareExchangeResult::Exchanged))
            || result == &Ok(CompareExchangeResult::Conflict(Some(winner.clone())))
    }));
    assert_eq!(
        get_binary(&database, b"contended"),
        Ok(Some(winner.clone()))
    );

    drop(database);
    drop(state);
    cleanup_environment(&canonical);
}