lean_lmdb

use super::*;

#[test]
fn scan_tokens_are_versioned_bound_and_authenticated() {
    let mode = ScanMode::Range {
        from: Some(vec![0, 1]),
        to: None,
        from_inclusive: false,
        to_inclusive: true,
    };
    let token = make_scan_token(42, "data", &mode, &[0xff, 0]).expect("make token");
    assert_eq!(
        parse_scan_token(&token, 42, "data", &mode, 511),
        Ok(vec![0xff, 0])
    );
    let stale_token =
        make_scan_token(u64::MAX, "data", &mode, &[0xff, 0]).expect("make stale token");
    assert_eq!(
        parse_scan_token(&stale_token, 42, "data", &mode, 511),
        Err(ScanError::StaleContinuation)
    );
    assert_eq!(
        parse_scan_token(&token, 42, "other", &mode, 511),
        Err(ScanError::InvalidContinuation)
    );
    assert_eq!(
        parse_scan_token(&token, 42, "data", &ScanMode::Prefix(vec![0, 1]), 511,),
        Err(ScanError::InvalidContinuation)
    );

    let empty_key_token = make_scan_token(42, "data", &mode, b"").expect("make malformed token");
    assert_eq!(
        parse_scan_token(&empty_key_token, 42, "data", &mode, 511),
        Err(ScanError::InvalidContinuation)
    );

    let prefix_mode = ScanMode::Prefix(b"ab".to_vec());
    let outside_prefix =
        make_scan_token(42, "data", &prefix_mode, b"ac").expect("make out-of-prefix token");
    assert_eq!(
        parse_scan_token(&outside_prefix, 42, "data", &prefix_mode, 511),
        Err(ScanError::InvalidContinuation)
    );
    let bounded_mode = ScanMode::Range {
        from: Some(b"b".to_vec()),
        to: Some(b"d".to_vec()),
        from_inclusive: false,
        to_inclusive: false,
    };
    for outside_key in [&b"b"[..], &b"d"[..], &b"z"[..]] {
        let outside_range = make_scan_token(42, "data", &bounded_mode, outside_key)
            .expect("make out-of-range token");
        assert_eq!(
            parse_scan_token(&outside_range, 42, "data", &bounded_mode, 511),
            Err(ScanError::InvalidContinuation)
        );
    }

    let mut corrupt = token.clone();
    corrupt[15] ^= 0x80;
    assert_eq!(
        parse_scan_token(&corrupt, 42, "data", &mode, 511),
        Err(ScanError::InvalidContinuation)
    );
    assert_eq!(
        parse_scan_token(&token[..token.len() - 1], 42, "data", &mode, 511),
        Err(ScanError::InvalidContinuation)
    );
    let mut trailing = token.clone();
    trailing.insert(trailing.len() - 8, 0);
    replace_scan_token_tag(&mut trailing);
    assert_eq!(
        parse_scan_token(&trailing, 42, "data", &mode, 511),
        Err(ScanError::InvalidContinuation)
    );

    let mut oversized_binding_length = token.clone();
    oversized_binding_length[12..16].copy_from_slice(&u32::MAX.to_be_bytes());
    replace_scan_token_tag(&mut oversized_binding_length);
    assert_eq!(
        parse_scan_token(&oversized_binding_length, 42, "data", &mode, 511),
        Err(ScanError::InvalidContinuation)
    );

    let binding_length = u32::from_be_bytes(token[12..16].try_into().expect("binding length"));
    let last_key_length_offset = 16 + binding_length as usize;
    let mut oversized_key_length = token.clone();
    oversized_key_length[last_key_length_offset..last_key_length_offset + 4]
        .copy_from_slice(&u32::MAX.to_be_bytes());
    replace_scan_token_tag(&mut oversized_key_length);
    assert_eq!(
        parse_scan_token(&oversized_key_length, 42, "data", &mode, 511),
        Err(ScanError::InvalidContinuation)
    );
}

#[test]
fn continuation_environment_mismatch_distinguishes_live_from_stale() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let first_dir = fresh_dir("token_first_env");
    let first_path = canonical_environment_path(first_dir.to_str().expect("UTF-8 temp path"), true)
        .expect("first canonical path");
    let first =
        open_shared_environment(first_path.clone(), options(), true).expect("first environment");
    let second_dir = fresh_dir("token_second_env");
    let second_path =
        canonical_environment_path(second_dir.to_str().expect("UTF-8 temp path"), true)
            .expect("second canonical path");
    let second =
        open_shared_environment(second_path.clone(), options(), true).expect("second environment");
    let mode = ScanMode::Prefix(Vec::new());
    let token = make_scan_token(first.id, "data", &mode, b"key").expect("make token");

    assert_eq!(
        parse_scan_token(&token, second.id, "data", &mode, second.env.max_key_size()),
        Err(ScanError::InvalidContinuation)
    );

    drop(first);
    cleanup_environment(&first_path);
    assert_eq!(
        parse_scan_token(&token, second.id, "data", &mode, second.env.max_key_size()),
        Err(ScanError::StaleContinuation)
    );

    drop(second);
    cleanup_environment(&second_path);
}

#[test]
fn prefix_successors_cover_empty_and_all_ff_prefixes() {
    assert_eq!(prefix_upper_bound(b""), Ok(None));
    assert_eq!(prefix_upper_bound(&[0xff]), Ok(None));
    assert_eq!(prefix_upper_bound(&[0xff, 0xff]), Ok(None));
    assert_eq!(prefix_upper_bound(b"abc"), Ok(Some(b"abd".to_vec())));
    assert_eq!(
        prefix_upper_bound(&[0x12, 0xff, 0xff]),
        Ok(Some(vec![0x13]))
    );
    assert_eq!(
        prefix_upper_bound(&[0x12, 0xfe, 0xff]),
        Ok(Some(vec![0x12, 0xff]))
    );
}

#[test]
fn scans_prefixes_in_pages_and_resumes_exclusively() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("prefix_scan");
    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 = database_resource(&state, "data");
    for (key, value) in [
        (&b"a"[..], &b"zero"[..]),
        (&b"aa"[..], &b"one"[..]),
        (&b"ab"[..], &b"two"[..]),
        (&b"b"[..], &b"three"[..]),
        (&[0xff][..], &b"ff"[..]),
        (&[0xff, 0][..], &b"ff-zero"[..]),
    ] {
        put_binary(&database, key, value).expect("seed scan row");
    }

    let mode = ScanMode::Prefix(b"a".to_vec());
    let first = scan_binary(&database, &mode, 2, MAX_SCAN_BYTES, None).expect("first page");
    assert_eq!(
        first.rows,
        vec![
            (b"a".to_vec(), b"zero".to_vec()),
            (b"aa".to_vec(), b"one".to_vec())
        ]
    );
    let token = first.continuation.expect("more prefix rows");
    // Limits are intentionally excluded from token binding.
    let second = scan_binary(&database, &mode, 10, 5, Some(&token)).expect("second page");
    assert_eq!(second.rows, vec![(b"ab".to_vec(), b"two".to_vec())]);
    assert_eq!(second.continuation, None);

    let all_ff = scan_binary(
        &database,
        &ScanMode::Prefix(vec![0xff]),
        10,
        MAX_SCAN_BYTES,
        None,
    )
    .expect("all-FF prefix");
    assert_eq!(
        all_ff.rows,
        vec![
            (vec![0xff], b"ff".to_vec()),
            (vec![0xff, 0], b"ff-zero".to_vec())
        ]
    );
    let all = scan_binary(
        &database,
        &ScanMode::Prefix(Vec::new()),
        10,
        MAX_SCAN_BYTES,
        None,
    )
    .expect("empty prefix");
    assert_eq!(all.rows.len(), 6);

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

#[test]
fn range_scan_honors_bounds_lookahead_byte_budget_and_limits() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("range_scan");
    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 = database_resource(&state, "data");
    for key in [b"a", b"b", b"c", b"d"] {
        let value = if key == b"b" {
            vec![7; 20]
        } else {
            key.to_vec()
        };
        put_binary(&database, key, &value).expect("seed range row");
    }
    let mode = ScanMode::Range {
        from: Some(b"a".to_vec()),
        to: Some(b"d".to_vec()),
        from_inclusive: false,
        to_inclusive: false,
    };
    assert_eq!(
        scan_binary(&database, &mode, 10, 2, None),
        Err(ScanError::RowTooLarge)
    );
    let first = scan_binary(&database, &mode, 10, 21, None).expect("bounded first row");
    assert_eq!(first.rows, vec![(b"b".to_vec(), vec![7; 20])]);
    let token = first.continuation.expect("lookahead found c");
    let second = scan_binary(&database, &mode, 10, 2, Some(&token)).expect("resume range");
    assert_eq!(second.rows, vec![(b"c".to_vec(), b"c".to_vec())]);
    assert_eq!(second.continuation, None);

    assert_eq!(
        scan_binary(&database, &mode, 0, 1, None),
        Err(ScanError::InvalidLimit)
    );
    assert_eq!(
        scan_binary(&database, &mode, MAX_SCAN_LIMIT + 1, 1, None),
        Err(ScanError::InvalidLimit)
    );
    assert_eq!(
        scan_binary(&database, &mode, 1, 0, None),
        Err(ScanError::InvalidMaxBytes)
    );
    assert_eq!(
        scan_binary(&database, &mode, 1, MAX_SCAN_BYTES + 1, None),
        Err(ScanError::InvalidMaxBytes)
    );

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

#[test]
fn scans_normalize_oversized_prefixes_and_range_bounds() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("oversized_scan_bounds");
    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 = database_resource(&state, "data");
    let max_key_size = state.env.max_key_size();
    let max_prefix = vec![0x10; max_key_size];
    let mut successor = max_prefix.clone();
    *successor.last_mut().expect("nonzero max key size") += 1;
    let high = vec![0x20; max_key_size];
    for key in [
        vec![0x01],
        max_prefix.clone(),
        successor.clone(),
        high.clone(),
    ] {
        put_binary(&database, &key, &key).expect("seed boundary key");
    }

    let mut oversized = max_prefix.clone();
    oversized.push(0);
    for from_inclusive in [false, true] {
        let mode = ScanMode::Range {
            from: Some(oversized.clone()),
            to: None,
            from_inclusive,
            to_inclusive: false,
        };
        assert_eq!(
            collect_scan_pages(&database, &mode, 10, MAX_SCAN_BYTES),
            vec![
                (successor.clone(), successor.clone()),
                (high.clone(), high.clone())
            ]
        );
    }
    for to_inclusive in [false, true] {
        let mode = ScanMode::Range {
            from: None,
            to: Some(oversized.clone()),
            from_inclusive: false,
            to_inclusive,
        };
        assert_eq!(
            collect_scan_pages(&database, &mode, 10, MAX_SCAN_BYTES),
            vec![
                (vec![0x01], vec![0x01]),
                (max_prefix.clone(), max_prefix.clone())
            ]
        );
    }

    let oversized_prefix = ScanMode::Prefix(vec![0x10; max_key_size + 1]);
    assert_eq!(
        scan_binary(&database, &oversized_prefix, 10, MAX_SCAN_BYTES, None),
        Ok(ScanPage {
            rows: Vec::new(),
            continuation: None
        })
    );
    let mut ff_tail = vec![0x10; max_key_size];
    *ff_tail.last_mut().expect("nonzero max key size") = 0xff;
    let ff_successor = prefix_upper_bound(&ff_tail)
        .expect("FF-tail successor")
        .expect("FF-tail is not all FF");
    put_binary(&database, &ff_tail, b"ff-tail").expect("seed FF-tail key");
    put_binary(&database, &ff_successor, b"ff-successor").expect("seed FF successor");
    let mut ff_tail_oversized = ff_tail.clone();
    ff_tail_oversized.push(0);
    let ff_lower = ScanMode::Range {
        from: Some(ff_tail_oversized.clone()),
        to: None,
        from_inclusive: true,
        to_inclusive: false,
    };
    let ff_lower_rows = collect_scan_pages(&database, &ff_lower, 10, MAX_SCAN_BYTES);
    assert_eq!(ff_lower_rows.first().map(|row| &row.0), Some(&ff_successor));

    let ff_upper = ScanMode::Range {
        from: None,
        to: Some(ff_tail_oversized),
        from_inclusive: true,
        to_inclusive: false,
    };
    let ff_upper_rows = collect_scan_pages(&database, &ff_upper, 10, MAX_SCAN_BYTES);
    assert!(ff_upper_rows.iter().any(|row| row.0 == ff_tail));
    assert!(!ff_upper_rows.iter().any(|row| row.0 == ff_successor));

    let mut beyond_all_keys = vec![0xff; max_key_size];
    beyond_all_keys.push(0);
    let empty_range = ScanMode::Range {
        from: Some(beyond_all_keys.clone()),
        to: None,
        from_inclusive: true,
        to_inclusive: false,
    };
    assert_eq!(
        scan_binary(&database, &empty_range, 10, MAX_SCAN_BYTES, None),
        Ok(ScanPage {
            rows: Vec::new(),
            continuation: None
        })
    );
    let all_keys_upper = ScanMode::Range {
        from: None,
        to: Some(beyond_all_keys),
        from_inclusive: true,
        to_inclusive: false,
    };
    assert_eq!(
        collect_scan_pages(&database, &all_keys_upper, 10, MAX_SCAN_BYTES),
        collect_scan_pages(
            &database,
            &ScanMode::Range {
                from: None,
                to: None,
                from_inclusive: true,
                to_inclusive: false,
            },
            10,
            MAX_SCAN_BYTES
        )
    );

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

fn model_scan(entries: &BTreeMap<Vec<u8>, Vec<u8>>, mode: &ScanMode) -> Vec<(Vec<u8>, Vec<u8>)> {
    entries
        .iter()
        .filter(|(key, _)| match mode {
            ScanMode::Prefix(prefix) => key.starts_with(prefix),
            ScanMode::Range {
                from,
                to,
                from_inclusive,
                to_inclusive,
            } => {
                let above_from = from.as_ref().is_none_or(|from| {
                    if *from_inclusive {
                        key.as_slice() >= from.as_slice()
                    } else {
                        key.as_slice() > from.as_slice()
                    }
                });
                let below_to = to.as_ref().is_none_or(|to| {
                    if *to_inclusive {
                        key.as_slice() <= to.as_slice()
                    } else {
                        key.as_slice() < to.as_slice()
                    }
                });
                above_from && below_to
            }
        })
        .map(|(key, value)| (key.clone(), value.clone()))
        .collect()
}

fn collect_scan_pages(
    database: &DatabaseResource,
    mode: &ScanMode,
    limit: u64,
    max_bytes: u64,
) -> Vec<(Vec<u8>, Vec<u8>)> {
    let mut rows = Vec::new();
    let mut continuation = None;
    for _page_index in 0..20_000 {
        let resumed = continuation.is_some();
        let page = scan_binary(database, mode, limit, max_bytes, continuation.as_deref())
            .expect("model scan page");
        if resumed {
            assert!(!page.rows.is_empty(), "a continuation must make progress");
        }
        rows.extend(page.rows);
        match page.continuation {
            Some(token) => continuation = Some(token),
            None => return rows,
        }
    }
    panic!("model scan did not finish within the deterministic page bound")
}

#[test]
fn range_scan_matches_binary_order_model_for_every_boundary_combination() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("range_model");
    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 = database_resource(&state, "data");

    // Fixed edge cases plus deterministic generated bytes exercise LMDB's unsigned,
    // byte-exact lexicographic ordering (including NUL, high-bit, and FF bytes).
    let mut model = BTreeMap::new();
    let edge_keys = [
        vec![0x00],
        vec![0x00, 0x00],
        vec![0x00, 0xff],
        vec![0x01],
        vec![0x7f, 0xff],
        vec![0x80],
        vec![0x80, 0x00],
        vec![0xfe, 0xff],
        vec![0xff],
        vec![0xff, 0x00],
        vec![0xff, 0xff],
    ];
    for (index, key) in edge_keys.into_iter().enumerate() {
        model.insert(key, vec![0x00, index as u8, 0xff]);
    }
    let mut seed = 0x6d_2b_79_f5_u32;
    for index in 0..48_u8 {
        seed = seed.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
        let length = (seed as usize % 6) + 1;
        let mut key = Vec::with_capacity(length);
        for _ in 0..length {
            seed = seed.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
            key.push((seed >> 24) as u8);
        }
        model.insert(key, vec![index, 0x00, 0x80, 0xff]);
    }
    for (key, value) in &model {
        put_binary(&database, key, value).expect("seed model row");
    }

    // None and exact byte edges cover unbounded, equal, empty, and inverted
    // ranges. Both booleans are varied even where an unbounded side ignores it.
    let bounds = [
        None,
        Some(Vec::new()),
        Some(vec![0x00]),
        Some(vec![0x00, 0xff]),
        Some(vec![0x80]),
        Some(vec![0xff]),
        Some(vec![0xff, 0xff]),
        Some(vec![0xff, 0xff, 0xff]),
    ];
    for from in &bounds {
        for to in &bounds {
            for from_inclusive in [false, true] {
                for to_inclusive in [false, true] {
                    let mode = ScanMode::Range {
                        from: from.clone(),
                        to: to.clone(),
                        from_inclusive,
                        to_inclusive,
                    };
                    let expected = model_scan(&model, &mode);
                    let actual = collect_scan_pages(&database, &mode, 1, 16);
                    assert_eq!(
                        actual, expected,
                        "range mismatch: {mode:?}; each bounded page must return every row exactly once"
                    );
                }
            }
        }
    }

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

#[test]
fn continuation_has_documented_weak_snapshot_behavior_across_writes() {
    let _serial = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
    let dir = fresh_dir("scan_weak_snapshot");
    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 = database_resource(&state, "data");
    for key in [vec![0x10], vec![0x20], vec![0x30], vec![0x40]] {
        put_binary(&database, &key, &key).expect("seed weak snapshot row");
    }
    let mode = ScanMode::Range {
        from: Some(vec![0x10]),
        to: Some(vec![0x50]),
        from_inclusive: true,
        to_inclusive: false,
    };
    let first = scan_binary(&database, &mode, 1, MAX_SCAN_BYTES, None).expect("first page");
    assert_eq!(first.rows, vec![(vec![0x10], vec![0x10])]);
    let token = first.continuation.expect("remaining rows");

    // A continuation is an exclusive last-key cursor, not a pinned snapshot:
    // writes behind it are not revisited, while subsequent pages see later writes.
    put_binary(&database, &[0x05], b"behind-range").expect("insert below range");
    put_binary(&database, &[0x08], b"behind-cursor").expect("insert behind cursor");
    put_binary(&database, &[0x18], b"new").expect("insert ahead of cursor");
    put_binary(&database, &[0x30], b"updated").expect("update ahead of cursor");
    delete_binary(&database, &[0x20]).expect("delete ahead of cursor");
    put_binary(&database, &[0x50], b"outside").expect("insert at excluded upper edge");

    let mut actual = first.rows;
    let mut continuation = Some(token);
    while let Some(token) = continuation {
        let page = scan_binary(&database, &mode, 1, 16, Some(&token)).expect("changed page");
        actual.extend(page.rows);
        continuation = page.continuation;
    }
    assert_eq!(
        actual,
        vec![
            (vec![0x10], vec![0x10]),
            (vec![0x18], b"new".to_vec()),
            (vec![0x30], b"updated".to_vec()),
            (vec![0x40], vec![0x40]),
        ]
    );

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