// Copyright 2025 Sushanth (https://github.com/sushanthpy)
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use crate::database::{Database, TxnHandle};
use std::ffi::CStr;
use std::os::raw::{c_char, c_int};
use std::ptr;
use std::slice;
use std::sync::Arc;

/// Opaque pointer to Database
pub struct DatabasePtr(Arc<Database>);

/// C-compatible Transaction Handle
#[repr(C)]
pub struct C_TxnHandle {
    pub txn_id: u64,
    pub snapshot_ts: u64,
}

/// Open the database.
/// Returns a pointer to the database instance, or null on error.
/// # Safety
/// The path must be a valid C string.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_open(path: *const c_char) -> *mut DatabasePtr {
    if path.is_null() {
        return ptr::null_mut();
    }

    let c_str = unsafe { CStr::from_ptr(path) };
    let path_str = match c_str.to_str() {
        Ok(s) => s,
        Err(_) => return ptr::null_mut(),
    };

    // Use default config for now
    let config = crate::database::DatabaseConfig::default();

    // Database::open returns Result<Arc<Database>>
    match Database::open_with_config(path_str, config) {
        Ok(db) => {
            let ptr = Box::new(DatabasePtr(db));
            Box::into_raw(ptr)
        }
        Err(_) => ptr::null_mut(),
    }
}

/// Close the database and free the pointer.
/// # Safety
/// The ptr must be a valid pointer returned by toondb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_close(ptr: *mut DatabasePtr) {
    if !ptr.is_null() {
        unsafe {
            let _ = Box::from_raw(ptr);
        }
    }
}

/// Begin a transaction.
/// Returns C_TxnHandle. On error, txn_id will be 0.
/// # Safety
/// The ptr must be a valid pointer returned by toondb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_begin_txn(ptr: *mut DatabasePtr) -> C_TxnHandle {
    if ptr.is_null() {
        return C_TxnHandle {
            txn_id: 0,
            snapshot_ts: 0,
        };
    }
    let db = unsafe { &(*ptr).0 };
    match db.begin_transaction() {
        Ok(txn) => C_TxnHandle {
            txn_id: txn.txn_id,
            snapshot_ts: txn.snapshot_ts,
        },
        Err(_) => C_TxnHandle {
            txn_id: 0,
            snapshot_ts: 0,
        },
    }
}

/// Commit a transaction.
/// Returns 0 on success, -1 on error.
/// # Safety
/// The ptr must be a valid pointer returned by toondb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_commit(ptr: *mut DatabasePtr, handle: C_TxnHandle) -> c_int {
    if ptr.is_null() {
        return -1;
    }
    let db = unsafe { &(*ptr).0 };
    let txn = TxnHandle {
        txn_id: handle.txn_id,
        snapshot_ts: handle.snapshot_ts,
    };
    match db.commit(txn) {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// Abort a transaction.
/// Returns 0 on success, -1 on error.
/// # Safety
/// The ptr must be a valid pointer returned by toondb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_abort(ptr: *mut DatabasePtr, handle: C_TxnHandle) -> c_int {
    if ptr.is_null() {
        return -1;
    }
    let db = unsafe { &(*ptr).0 };
    let txn = TxnHandle {
        txn_id: handle.txn_id,
        snapshot_ts: handle.snapshot_ts,
    };
    match db.abort(txn) {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// Put a key-value pair.
/// Returns 0 on success, -1 on error.
/// # Safety
/// The ptr must be a valid pointer returned by toondb_open.
/// key_ptr and val_ptr must be valid pointers with the specified lengths.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_put(
    ptr: *mut DatabasePtr,
    handle: C_TxnHandle,
    key_ptr: *const u8,
    key_len: usize,
    val_ptr: *const u8,
    val_len: usize,
) -> c_int {
    if ptr.is_null() || key_ptr.is_null() || val_ptr.is_null() {
        return -1;
    }
    let db = unsafe { &(*ptr).0 };
    let key = unsafe { slice::from_raw_parts(key_ptr, key_len) };
    let val = unsafe { slice::from_raw_parts(val_ptr, val_len) };
    let txn = TxnHandle {
        txn_id: handle.txn_id,
        snapshot_ts: handle.snapshot_ts,
    };

    match db.put(txn, key, val) {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// Get a value.
/// Writes pointer to val_out and length to len_out.
/// The caller must free the returned bytes using toondb_free_bytes.
/// Returns 0 on success (found), 1 on not found, -1 on error.
/// # Safety
/// All pointer arguments must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_get(
    ptr: *mut DatabasePtr,
    handle: C_TxnHandle,
    key_ptr: *const u8,
    key_len: usize,
    val_out: *mut *mut u8,
    len_out: *mut usize,
) -> c_int {
    if ptr.is_null() || key_ptr.is_null() || val_out.is_null() || len_out.is_null() {
        return -1;
    }
    let db = unsafe { &(*ptr).0 };
    let key = unsafe { slice::from_raw_parts(key_ptr, key_len) };
    let txn = TxnHandle {
        txn_id: handle.txn_id,
        snapshot_ts: handle.snapshot_ts,
    };

    match db.get(txn, key) {
        Ok(Some(val)) => {
            // Copy value to heap to pass to C
            let mut buf = val.into_boxed_slice();
            unsafe {
                *val_out = buf.as_mut_ptr();
                *len_out = buf.len();
            }
            let _ = Box::into_raw(buf); // Leak memory, caller must free
            0
        }
        Ok(None) => 1, // Not found
        Err(_) => -1,
    }
}

/// Free bytes allocated by toondb_get.
/// # Safety
/// ptr must be a valid pointer returned by toondb_get.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_free_bytes(ptr: *mut u8, len: usize) {
    if !ptr.is_null() {
        unsafe {
            let _ = Box::from_raw(std::ptr::slice_from_raw_parts_mut(ptr, len));
        }
    }
}

/// Delete a key.
/// Returns 0 on success, -1 on error.
/// # Safety
/// All pointer arguments must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_delete(
    ptr: *mut DatabasePtr,
    handle: C_TxnHandle,
    key_ptr: *const u8,
    key_len: usize,
) -> c_int {
    if ptr.is_null() || key_ptr.is_null() {
        return -1;
    }
    let db = unsafe { &(*ptr).0 };
    let key = unsafe { slice::from_raw_parts(key_ptr, key_len) };
    let txn = TxnHandle {
        txn_id: handle.txn_id,
        snapshot_ts: handle.snapshot_ts,
    };

    match db.delete(txn, key) {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// Put path.
/// # Safety
/// All pointer arguments must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_put_path(
    ptr: *mut DatabasePtr,
    handle: C_TxnHandle,
    path_ptr: *const c_char,
    val_ptr: *const u8,
    val_len: usize,
) -> c_int {
    if ptr.is_null() || path_ptr.is_null() || val_ptr.is_null() {
        return -1;
    }
    let db = unsafe { &(*ptr).0 };
    let c_str = unsafe { CStr::from_ptr(path_ptr) };
    let path_str = match c_str.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let val = unsafe { slice::from_raw_parts(val_ptr, val_len) };
    let txn = TxnHandle {
        txn_id: handle.txn_id,
        snapshot_ts: handle.snapshot_ts,
    };

    match db.put_path(txn, path_str, val) {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// Get path.
/// # Safety
/// All pointer arguments must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_get_path(
    ptr: *mut DatabasePtr,
    handle: C_TxnHandle,
    path_ptr: *const c_char,
    val_out: *mut *mut u8,
    len_out: *mut usize,
) -> c_int {
    if ptr.is_null() || path_ptr.is_null() || val_out.is_null() || len_out.is_null() {
        return -1;
    }
    let db = unsafe { &(*ptr).0 };
    let c_str = unsafe { CStr::from_ptr(path_ptr) };
    let path_str = match c_str.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let txn = TxnHandle {
        txn_id: handle.txn_id,
        snapshot_ts: handle.snapshot_ts,
    };

    match db.get_path(txn, path_str) {
        Ok(Some(val)) => {
            let mut buf = val.into_boxed_slice();
            unsafe {
                *val_out = buf.as_mut_ptr();
                *len_out = buf.len();
            }
            let _ = Box::into_raw(buf);
            0
        }
        Ok(None) => 1,
        Err(_) => -1,
    }
}

/// Opaque pointer to Scan Iterator
#[allow(clippy::type_complexity)]
pub struct ScanIteratorPtr(
    Box<dyn Iterator<Item = Result<(Vec<u8>, Vec<u8>), toondb_core::ToonDBError>>>,
);

/// Start a scan.
/// # Safety
/// All pointer arguments must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_scan(
    ptr: *mut DatabasePtr,
    handle: C_TxnHandle,
    start_ptr: *const u8,
    start_len: usize,
    end_ptr: *const u8,
    end_len: usize,
) -> *mut ScanIteratorPtr {
    if ptr.is_null() {
        return ptr::null_mut();
    }
    let db = unsafe { &(*ptr).0 };
    let txn = TxnHandle {
        txn_id: handle.txn_id,
        snapshot_ts: handle.snapshot_ts,
    };

    let start = if !start_ptr.is_null() && start_len > 0 {
        unsafe { slice::from_raw_parts(start_ptr, start_len).to_vec() }
    } else {
        vec![]
    };

    let end = if !end_ptr.is_null() && end_len > 0 {
        unsafe { slice::from_raw_parts(end_ptr, end_len).to_vec() }
    } else {
        vec![] // Empty end means unbounded in `scan` usually, or we need to handle it
    };

    // Note: The underlying `scan` method expects `Range<Vec<u8>>`.
    // We need to handle empty start/end correctly.
    // For now, let's assume the caller provides valid bounds or we use defaults.
    // Ideally, we'd pass optionals.

    // Using a simplified approach: if start is empty, use empty vec (start of db).
    // If end is empty, use a "max" key or handle in `scan` impl.
    // The `StorageEngine::scan` takes `Range<Vec<u8>>`.

    // Using a simplified approach: if start is empty, use empty vec (start of db).
    // If end is empty, use empty vec (unbounded).

    match db.scan_range(txn, &start, &end) {
        Ok(rows) => {
            // Convert rows to an iterator of (key, value)
            // scan_range returns Vec<(Vec<u8>, Vec<u8>)>
            let iter = Box::new(rows.into_iter().map(Ok));

            let ptr = Box::new(ScanIteratorPtr(iter));
            Box::into_raw(ptr)
        }
        Err(_) => ptr::null_mut(),
    }
}

/// Start a prefix scan - returns only keys that start with the given prefix.
/// This is the safe method for multi-tenant isolation.
/// # Safety
/// All pointer arguments must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_scan_prefix(
    ptr: *mut DatabasePtr,
    handle: C_TxnHandle,
    prefix_ptr: *const u8,
    prefix_len: usize,
) -> *mut ScanIteratorPtr {
    if ptr.is_null() {
        return ptr::null_mut();
    }
    let db = unsafe { &(*ptr).0 };
    let txn = TxnHandle {
        txn_id: handle.txn_id,
        snapshot_ts: handle.snapshot_ts,
    };

    let prefix = if !prefix_ptr.is_null() && prefix_len > 0 {
        unsafe { slice::from_raw_parts(prefix_ptr, prefix_len).to_vec() }
    } else {
        vec![]
    };

    // Use the proper scan method that filters by prefix
    match db.scan(txn, &prefix) {
        Ok(rows) => {
            // The underlying scan already filters by prefix, but double-check
            // to ensure no data leakage
            let prefix_owned = prefix.clone();
            let filtered: Vec<(Vec<u8>, Vec<u8>)> = rows
                .into_iter()
                .filter(|(k, _)| k.starts_with(&prefix_owned))
                .collect();
            
            let iter = Box::new(filtered.into_iter().map(Ok));
            let ptr = Box::new(ScanIteratorPtr(iter));
            Box::into_raw(ptr)
        }
        Err(_) => ptr::null_mut(),
    }
}

/// Get next item from scan iterator.
/// Returns 0 on success, 1 on done, -1 on error.
/// # Safety
/// All pointer arguments must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_scan_next(
    iter_ptr: *mut ScanIteratorPtr,
    key_out: *mut *mut u8,
    key_len_out: *mut usize,
    val_out: *mut *mut u8,
    val_len_out: *mut usize,
) -> c_int {
    if iter_ptr.is_null() || key_out.is_null() || val_out.is_null() {
        return -1;
    }
    let iter = unsafe { &mut (*iter_ptr).0 };

    match iter.next() {
        Some(Ok((key, val))) => {
            let mut key_buf = key.into_boxed_slice();
            let mut val_buf = val.into_boxed_slice();
            unsafe {
                *key_out = key_buf.as_mut_ptr();
                *key_len_out = key_buf.len();
                *val_out = val_buf.as_mut_ptr();
                *val_len_out = val_buf.len();
            }
            let _ = Box::into_raw(key_buf);
            let _ = Box::into_raw(val_buf);
            0
        }
        Some(Err(_)) => -1,
        None => 1, // Done
    }
}

/// Free scan iterator.
/// # Safety
/// ptr must be a valid pointer returned by toondb_scan.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_scan_free(ptr: *mut ScanIteratorPtr) {
    if !ptr.is_null() {
        unsafe {
            let _ = Box::from_raw(ptr);
        }
    }
}

/// Checkpoint the database.
/// # Safety
/// ptr must be a valid pointer returned by toondb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_checkpoint(ptr: *mut DatabasePtr) -> c_int {
    if ptr.is_null() {
        return -1;
    }
    let db = unsafe { &(*ptr).0 };
    match db.flush() {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// Storage statistics
#[repr(C)]
pub struct CStorageStats {
    pub memtable_size_bytes: u64,
    pub wal_size_bytes: u64,
    pub active_transactions: usize,
    pub min_active_snapshot: u64,
    pub last_checkpoint_lsn: u64,
}

/// Get storage statistics.
/// # Safety
/// ptr must be a valid pointer returned by toondb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_stats(ptr: *mut DatabasePtr) -> CStorageStats {
    if ptr.is_null() {
        return CStorageStats {
            memtable_size_bytes: 0,
            wal_size_bytes: 0,
            active_transactions: 0,
            min_active_snapshot: 0,
            last_checkpoint_lsn: 0,
        };
    }
    let db = unsafe { &(*ptr).0 };
    let stats = db.storage_stats();

    CStorageStats {
        memtable_size_bytes: stats.memtable_size_bytes,
        wal_size_bytes: stats.wal_size_bytes,
        active_transactions: stats.active_transactions,
        min_active_snapshot: stats.min_active_snapshot,
        last_checkpoint_lsn: stats.last_checkpoint_lsn,
    }
}

// ============================================================================
// Batched Operations - Minimize FFI Call Overhead
// ============================================================================

/// Batch descriptor for put_many operation
///
/// Memory layout for batch:
/// ```text
/// [num_entries: u32]
/// For each entry:
///   [key_len: u32][value_len: u32][key_bytes: ...][value_bytes: ...]
/// ```
///
/// This packed format minimizes FFI crossing overhead:
/// - One call instead of N calls
/// - No per-entry pointer chasing
/// - Contiguous memory for CPU cache efficiency
#[repr(C)]
pub struct CBatchPut {
    /// Pointer to packed batch data
    pub data: *const u8,
    /// Total length of packed data
    pub len: usize,
}

/// Put multiple key-value pairs in a single FFI call.
///
/// This is the high-performance path for Python and other FFI users.
/// Instead of N individual toondb_put calls (each with FFI overhead),
/// the caller packs all writes into a single buffer and makes one call.
///
/// ## Performance
///
/// For N writes with per-call overhead c:
/// - Individual puts: N × c (e.g., 100 × 500ns = 50µs overhead)
/// - put_many: 1 × c (e.g., 1 × 500ns = 0.5µs overhead)
/// - Speedup: 100× for FFI overhead alone
///
/// ## Buffer Format
///
/// ```text
/// ┌────────────────────────────────────────────────────────────────┐
/// │  num_entries (4 bytes, little-endian u32)                      │
/// ├────────────────────────────────────────────────────────────────┤
/// │  Entry 1:                                                      │
/// │    key_len (4 bytes, u32) | val_len (4 bytes, u32)             │
/// │    key_bytes (key_len bytes)                                   │
/// │    value_bytes (val_len bytes)                                 │
/// ├────────────────────────────────────────────────────────────────┤
/// │  Entry 2: ...                                                  │
/// ├────────────────────────────────────────────────────────────────┤
/// │  Entry N: ...                                                  │
/// └────────────────────────────────────────────────────────────────┘
/// ```
///
/// ## Returns
///
/// - 0: All entries written successfully
/// - -1: Error (null pointer, invalid format, write failure)
/// - >0: Number of entries successfully written before error
///
/// # Safety
///
/// - `ptr` must be a valid DatabasePtr from toondb_open
/// - `batch` must point to a valid CBatchPut with correct format
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_put_many(
    ptr: *mut DatabasePtr,
    handle: C_TxnHandle,
    batch: CBatchPut,
) -> c_int {
    if ptr.is_null() || batch.data.is_null() || batch.len < 4 {
        return -1;
    }

    let db = unsafe { &(*ptr).0 };
    let txn = TxnHandle {
        txn_id: handle.txn_id,
        snapshot_ts: handle.snapshot_ts,
    };

    // Parse batch
    let data = unsafe { slice::from_raw_parts(batch.data, batch.len) };
    
    // Read number of entries
    if data.len() < 4 {
        return -1;
    }
    let num_entries = u32::from_le_bytes([data[0], data[1], data[2], data[3]]) as usize;
    
    let mut offset = 4;
    let mut success_count = 0;

    for _ in 0..num_entries {
        // Read key_len and value_len
        if offset + 8 > data.len() {
            return success_count;
        }
        let key_len = u32::from_le_bytes([
            data[offset], data[offset + 1], data[offset + 2], data[offset + 3]
        ]) as usize;
        let val_len = u32::from_le_bytes([
            data[offset + 4], data[offset + 5], data[offset + 6], data[offset + 7]
        ]) as usize;
        offset += 8;

        // Read key and value
        if offset + key_len + val_len > data.len() {
            return success_count;
        }
        let key = &data[offset..offset + key_len];
        offset += key_len;
        let value = &data[offset..offset + val_len];
        offset += val_len;

        // Write to database
        match db.put(txn, key, value) {
            Ok(_) => success_count += 1,
            Err(_) => return success_count,
        }
    }

    success_count
}

/// Delete multiple keys in a single FFI call.
///
/// ## Buffer Format
///
/// ```text
/// ┌────────────────────────────────────────────────────────────────┐
/// │  num_entries (4 bytes, little-endian u32)                      │
/// ├────────────────────────────────────────────────────────────────┤
/// │  Entry 1:                                                      │
/// │    key_len (4 bytes, u32)                                      │
/// │    key_bytes (key_len bytes)                                   │
/// ├────────────────────────────────────────────────────────────────┤
/// │  Entry 2: ...                                                  │
/// └────────────────────────────────────────────────────────────────┘
/// ```
///
/// # Safety
///
/// Same as toondb_put_many.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_delete_many(
    ptr: *mut DatabasePtr,
    handle: C_TxnHandle,
    keys_data: *const u8,
    keys_len: usize,
) -> c_int {
    if ptr.is_null() || keys_data.is_null() || keys_len < 4 {
        return -1;
    }

    let db = unsafe { &(*ptr).0 };
    let txn = TxnHandle {
        txn_id: handle.txn_id,
        snapshot_ts: handle.snapshot_ts,
    };

    let data = unsafe { slice::from_raw_parts(keys_data, keys_len) };
    
    let num_entries = u32::from_le_bytes([data[0], data[1], data[2], data[3]]) as usize;
    
    let mut offset = 4;
    let mut success_count = 0;

    for _ in 0..num_entries {
        if offset + 4 > data.len() {
            return success_count;
        }
        let key_len = u32::from_le_bytes([
            data[offset], data[offset + 1], data[offset + 2], data[offset + 3]
        ]) as usize;
        offset += 4;

        if offset + key_len > data.len() {
            return success_count;
        }
        let key = &data[offset..offset + key_len];
        offset += key_len;

        match db.delete(txn, key) {
            Ok(_) => success_count += 1,
            Err(_) => return success_count,
        }
    }

    success_count
}

/// Get multiple values in a single FFI call.
///
/// ## Input Format
///
/// Same as delete_many: packed keys.
///
/// ## Output Format
///
/// ```text
/// ┌────────────────────────────────────────────────────────────────┐
/// │  num_results (4 bytes, u32)                                    │
/// ├────────────────────────────────────────────────────────────────┤
/// │  Entry 1:                                                      │
/// │    status (1 byte): 0=found, 1=not_found, 2=error              │
/// │    if found: val_len (4 bytes, u32), value_bytes               │
/// ├────────────────────────────────────────────────────────────────┤
/// │  Entry 2: ...                                                  │
/// └────────────────────────────────────────────────────────────────┘
/// ```
///
/// ## Returns
///
/// Pointer to allocated result buffer. Caller must free with toondb_free_bytes.
///
/// # Safety
///
/// Same as toondb_put_many.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_get_many(
    ptr: *mut DatabasePtr,
    handle: C_TxnHandle,
    keys_data: *const u8,
    keys_len: usize,
    result_out: *mut *mut u8,
    result_len_out: *mut usize,
) -> c_int {
    if ptr.is_null() || keys_data.is_null() || keys_len < 4 
        || result_out.is_null() || result_len_out.is_null() {
        return -1;
    }

    let db = unsafe { &(*ptr).0 };
    let txn = TxnHandle {
        txn_id: handle.txn_id,
        snapshot_ts: handle.snapshot_ts,
    };

    let data = unsafe { slice::from_raw_parts(keys_data, keys_len) };
    
    let num_entries = u32::from_le_bytes([data[0], data[1], data[2], data[3]]) as usize;
    
    // Build result buffer
    let mut result = Vec::with_capacity(4 + num_entries * 10); // Estimate
    result.extend_from_slice(&(num_entries as u32).to_le_bytes());
    
    let mut offset = 4;

    for _ in 0..num_entries {
        if offset + 4 > data.len() {
            result.push(2); // Error
            continue;
        }
        let key_len = u32::from_le_bytes([
            data[offset], data[offset + 1], data[offset + 2], data[offset + 3]
        ]) as usize;
        offset += 4;

        if offset + key_len > data.len() {
            result.push(2); // Error
            continue;
        }
        let key = &data[offset..offset + key_len];
        offset += key_len;

        match db.get(txn, key) {
            Ok(Some(value)) => {
                result.push(0); // Found
                result.extend_from_slice(&(value.len() as u32).to_le_bytes());
                result.extend_from_slice(&value);
            }
            Ok(None) => {
                result.push(1); // Not found
            }
            Err(_) => {
                result.push(2); // Error
            }
        }
    }

    // Return result
    let mut boxed = result.into_boxed_slice();
    unsafe {
        *result_out = boxed.as_mut_ptr();
        *result_len_out = boxed.len();
    }
    let _ = Box::into_raw(boxed); // Leak for caller to free
    
    0
}

// ============================================================================
// Batched Scan - Minimize FFI Call Overhead for Iterations
// ============================================================================

/// Fetch a batch of results from scan iterator.
///
/// This dramatically reduces FFI overhead for scan operations.
/// Instead of N calls to `toondb_scan_next` (each with FFI overhead),
/// fetch up to `batch_size` results in a single call.
///
/// ## Performance
///
/// For N results with per-call overhead c:
/// - Individual next calls: N × c (e.g., 10000 × 500ns = 5ms overhead)
/// - Batched (size=1000): 10 × c (e.g., 10 × 500ns = 5µs overhead)
/// - Speedup: 1000× for FFI overhead
///
/// ## Output Format
///
/// ```text
/// ┌────────────────────────────────────────────────────────────────┐
/// │  num_results (4 bytes, little-endian u32)                      │
/// │  is_done (1 byte): 0=more results available, 1=scan complete   │
/// ├────────────────────────────────────────────────────────────────┤
/// │  Entry 1:                                                      │
/// │    key_len (4 bytes, u32)                                      │
/// │    val_len (4 bytes, u32)                                      │
/// │    key_bytes (key_len bytes)                                   │
/// │    value_bytes (val_len bytes)                                 │
/// ├────────────────────────────────────────────────────────────────┤
/// │  Entry 2: ...                                                  │
/// └────────────────────────────────────────────────────────────────┘
/// ```
///
/// ## Returns
///
/// - 0: Batch fetched successfully (check is_done flag for completion)
/// - 1: Scan complete (no more results)
/// - -1: Error
///
/// ## Usage from Python
///
/// ```python
/// iter_ptr = lib.toondb_scan(...)
/// while True:
///     result = lib.toondb_scan_batch(iter_ptr, 1000, ...)
///     if result == 1:  # Done
///         break
///     # Parse batch buffer for up to 1000 results
/// lib.toondb_scan_free(iter_ptr)
/// ```
///
/// # Safety
///
/// - `iter_ptr` must be a valid ScanIteratorPtr from toondb_scan
/// - Output pointers must be valid
#[unsafe(no_mangle)]
pub unsafe extern "C" fn toondb_scan_batch(
    iter_ptr: *mut ScanIteratorPtr,
    batch_size: usize,
    result_out: *mut *mut u8,
    result_len_out: *mut usize,
) -> c_int {
    if iter_ptr.is_null() || result_out.is_null() || result_len_out.is_null() || batch_size == 0 {
        return -1;
    }

    let iter = unsafe { &mut (*iter_ptr).0 };
    
    // Pre-allocate result buffer
    // Estimate: header (5 bytes) + batch_size * (8 bytes header + ~100 bytes avg data)
    let estimated_size = 5 + batch_size * 108;
    let mut result = Vec::with_capacity(estimated_size);
    
    // Reserve space for header (will fill in at end)
    result.extend_from_slice(&[0u8; 5]); // 4 bytes count + 1 byte is_done
    
    let mut count = 0u32;
    let mut is_done = false;
    
    for _ in 0..batch_size {
        match iter.next() {
            Some(Ok((key, val))) => {
                // Write key_len, val_len, key, value
                result.extend_from_slice(&(key.len() as u32).to_le_bytes());
                result.extend_from_slice(&(val.len() as u32).to_le_bytes());
                result.extend_from_slice(&key);
                result.extend_from_slice(&val);
                count += 1;
            }
            Some(Err(_)) => {
                // Write header with current count and return error
                result[0..4].copy_from_slice(&count.to_le_bytes());
                result[4] = 0; // Not done (error case)
                
                let mut boxed = result.into_boxed_slice();
                unsafe {
                    *result_out = boxed.as_mut_ptr();
                    *result_len_out = boxed.len();
                }
                let _ = Box::into_raw(boxed);
                return -1;
            }
            None => {
                is_done = true;
                break;
            }
        }
    }
    
    // Fill in header
    result[0..4].copy_from_slice(&count.to_le_bytes());
    result[4] = if is_done { 1 } else { 0 };
    
    // If no results and done, signal completion
    if count == 0 && is_done {
        // Still allocate minimal buffer so caller can free consistently
        let mut boxed = result.into_boxed_slice();
        unsafe {
            *result_out = boxed.as_mut_ptr();
            *result_len_out = boxed.len();
        }
        let _ = Box::into_raw(boxed);
        return 1; // Done
    }
    
    // Return buffer
    let mut boxed = result.into_boxed_slice();
    unsafe {
        *result_out = boxed.as_mut_ptr();
        *result_len_out = boxed.len();
    }
    let _ = Box::into_raw(boxed);
    
    0 // Success, check is_done flag for completion
}

