sochdb_storage/

ffi.rs

// SPDX-License-Identifier: AGPL-3.0-or-later
// SochDB - LLM-Optimized Embedded Database
// Copyright (C) 2026 Sushanth Reddy Vanagala (https://github.com/sushanthpy)
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.

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;
use serde_json::Value;
use std::collections::HashMap;
use std::sync::Mutex;
use std::sync::OnceLock;
use sochdb_index::hnsw::{DistanceMetric, HnswConfig, HnswIndex};

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

// =========================================================================
// Collection Index Registry (in-memory)
// =========================================================================

struct CollectionIndex {
    index: Arc<HnswIndex>,
    dimension: usize,
    metric: DistanceMetric,
}

static COLLECTION_INDEXES: OnceLock<Mutex<HashMap<String, Arc<CollectionIndex>>>> = OnceLock::new();

fn collection_key(namespace: &str, collection: &str) -> String {
    format!("{}/{}", namespace, collection)
}

fn vector_bin_key(namespace: &str, collection: &str, id_hash: u128) -> String {
    format!("{}/collections/{}/vectors_bin/{:032x}", namespace, collection, id_hash)
}

fn metadata_key(namespace: &str, collection: &str, id_hash: u128) -> String {
    format!("{}/collections/{}/meta/{:032x}", namespace, collection, id_hash)
}

fn hash_id_to_u128(id: &str) -> u128 {
    let hash = blake3::hash(id.as_bytes());
    let bytes = hash.as_bytes();
    u128::from_le_bytes([
        bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
        bytes[8], bytes[9], bytes[10], bytes[11], bytes[12], bytes[13], bytes[14], bytes[15],
    ])
}

fn ensure_collection_index(
    db: &Database,
    namespace: &str,
    collection: &str,
    dimension: usize,
    metric: DistanceMetric,
) -> Arc<CollectionIndex> {
    let registry = COLLECTION_INDEXES.get_or_init(|| Mutex::new(HashMap::new()));
    let key = collection_key(namespace, collection);

    let mut registry_guard = registry.lock().unwrap();
    if let Some(existing) = registry_guard.get(&key) {
        return existing.clone();
    }

    let mut config = HnswConfig::default();
    config.metric = metric;
    let index = Arc::new(HnswIndex::new(dimension, config));

    let entry = Arc::new(CollectionIndex {
        index,
        dimension,
        metric,
    });
    registry_guard.insert(key, entry.clone());

    entry
}

fn resolve_collection_config(
    db: &Database,
    namespace: &str,
    collection: &str,
) -> Option<(usize, DistanceMetric)> {
    let key = format!("{}/_collections/{}", namespace, collection);
    let txn = db.begin_transaction().ok()?;
    let value = db.get(txn, key.as_bytes()).ok().flatten();
    let _ = db.commit(txn);
    let value = value?;

    let parsed: serde_json::Value = serde_json::from_slice(&value).ok()?;
    let dimension = parsed.get("dimension")?.as_u64()? as usize;
    let metric = match parsed.get("metric").and_then(|v| v.as_u64()).unwrap_or(0) {
        1 => DistanceMetric::Euclidean,
        2 => DistanceMetric::DotProduct,
        _ => DistanceMetric::Cosine,
    };
    Some((dimension, metric))
}

fn serialize_vector_binary(vector: &[f32]) -> Vec<u8> {
    let mut out = Vec::with_capacity(4 + vector.len() * 4);
    let len = vector.len() as u32;
    out.extend_from_slice(&len.to_le_bytes());
    for value in vector {
        out.extend_from_slice(&value.to_le_bytes());
    }
    out
}

fn decode_score(metric: DistanceMetric, distance: f32) -> f32 {
    match metric {
        DistanceMetric::Cosine => 1.0 - distance,
        DistanceMetric::DotProduct => -distance,
        DistanceMetric::Euclidean => -distance,
    }
}

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

/// C-compatible Commit Result
/// Returns commit_ts on success, or 0 with error_code on failure
#[repr(C)]
pub struct C_CommitResult {
    /// Commit timestamp (HLC-backed, monotonically increasing)
    /// This is 0 if the commit failed.
    pub commit_ts: u64,
    /// Error code: 0 = success, -1 = error, -2 = SSI conflict
    pub error_code: i32,
}

/// C-compatible Database Configuration
/// 
/// All fields have sensible defaults when set to 0/false.
/// This allows clients to only set the fields they care about.
#[repr(C)]
pub struct C_DatabaseConfig {
    /// Enable WAL for durability (default: true if wal_enabled_set is false)
    pub wal_enabled: bool,
    /// Whether wal_enabled was explicitly set
    pub wal_enabled_set: bool,
    /// Sync mode: 0=OFF, 1=NORMAL (default), 2=FULL
    pub sync_mode: u8,
    /// Whether sync_mode was explicitly set
    pub sync_mode_set: bool,
    /// Memtable size in bytes (0 = default 64MB)
    pub memtable_size_bytes: u64,
    /// Enable group commit for throughput (default: true if group_commit_set is false)
    pub group_commit: bool,
    /// Whether group_commit was explicitly set  
    pub group_commit_set: bool,
    /// Default index policy: 0=WriteOptimized, 1=Balanced (default), 2=ScanOptimized, 3=AppendOnly
    pub default_index_policy: u8,
    /// Whether default_index_policy was explicitly set
    pub default_index_policy_set: bool,
}

/// Open the database with configuration.
/// 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 sochdb_open_with_config(
    path: *const c_char, 
    config: C_DatabaseConfig
) -> *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(),
    };

    // Build config from C struct, using defaults for unset fields
    let mut db_config = crate::database::DatabaseConfig::default();
    
    if config.wal_enabled_set {
        db_config.wal_enabled = config.wal_enabled;
    }
    
    if config.sync_mode_set {
        db_config.sync_mode = match config.sync_mode {
            0 => crate::database::SyncMode::Off,
            1 => crate::database::SyncMode::Normal,
            _ => crate::database::SyncMode::Full,
        };
    }
    
    if config.memtable_size_bytes > 0 {
        db_config.memtable_size_limit = config.memtable_size_bytes as usize;
    }
    
    if config.group_commit_set {
        db_config.group_commit = config.group_commit;
    }
    
    if config.default_index_policy_set {
        db_config.default_index_policy = match config.default_index_policy {
            0 => crate::index_policy::IndexPolicy::WriteOptimized,
            1 => crate::index_policy::IndexPolicy::Balanced,
            2 => crate::index_policy::IndexPolicy::ScanOptimized,
            _ => crate::index_policy::IndexPolicy::AppendOnly,
        };
    }

    match Database::open_with_config(path_str, db_config) {
        Ok(db) => {
            let ptr = Box::new(DatabasePtr(db));
            Box::into_raw(ptr)
        }
        Err(_) => ptr::null_mut(),
    }
}

/// 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 sochdb_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(),
    }
}

/// Open the database in concurrent mode (multi-reader, single-writer).
/// 
/// This mode allows multiple processes to access the database simultaneously:
/// - Readers: Lock-free, concurrent access via MVCC snapshots
/// - Writers: Single-writer coordination through atomic locks
///
/// Use this for web applications, hot reloading, or any multi-process scenario.
///
/// # Safety
/// The path must be a valid C string.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_open_concurrent(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(),
    };

    match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        Database::open_concurrent(path_str)
    })) {
        Ok(Ok(db)) => {
            let ptr = Box::new(DatabasePtr(db));
            Box::into_raw(ptr)
        }
        Ok(Err(_)) | Err(_) => ptr::null_mut(),
    }
}

/// Check if database is in concurrent mode.
/// Returns 1 if concurrent mode, 0 otherwise.
/// # Safety
/// The ptr must be a valid pointer returned by sochdb_open or sochdb_open_concurrent.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_is_concurrent(ptr: *mut DatabasePtr) -> c_int {
    if ptr.is_null() {
        return 0;
    }
    let db = unsafe { &*ptr };
    if db.0.is_concurrent() { 1 } else { 0 }
}

/// Close the database and free the pointer.
/// # Safety
/// The ptr must be a valid pointer returned by sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_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 sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_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 C_CommitResult with commit_ts on success.
/// The commit_ts is HLC-backed and monotonically increasing, suitable for 
/// MVCC observability, replication, and audit trails.
/// # Safety
/// The ptr must be a valid pointer returned by sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_commit(ptr: *mut DatabasePtr, handle: C_TxnHandle) -> C_CommitResult {
    if ptr.is_null() {
        return C_CommitResult {
            commit_ts: 0,
            error_code: -1,
        };
    }
    let db = unsafe { &(*ptr).0 };
    let txn = TxnHandle {
        txn_id: handle.txn_id,
        snapshot_ts: handle.snapshot_ts,
    };
    match db.commit(txn) {
        Ok(commit_ts) => C_CommitResult {
            commit_ts,
            error_code: 0,
        },
        Err(_) => C_CommitResult {
            commit_ts: 0,
            error_code: -1,
        },
    }
}

/// Abort a transaction.
/// Returns 0 on success, -1 on error.
/// # Safety
/// The ptr must be a valid pointer returned by sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_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 sochdb_open.
/// key_ptr and val_ptr must be valid pointers with the specified lengths.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_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 sochdb_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 sochdb_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 sochdb_get.
/// # Safety
/// ptr must be a valid pointer returned by sochdb_get.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_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 sochdb_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 sochdb_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 sochdb_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>), sochdb_core::SochDBError>>>,
);

/// Start a scan.
/// # Safety
/// All pointer arguments must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_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 sochdb_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 sochdb_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 sochdb_scan.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_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 sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_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 sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_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 sochdb_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 sochdb_open
/// - `batch` must point to a valid CBatchPut with correct format
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_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 sochdb_put_many.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_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 sochdb_free_bytes.
///
/// # Safety
///
/// Same as sochdb_put_many.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_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 `sochdb_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.sochdb_scan(...)
/// while True:
///     result = lib.sochdb_scan_batch(iter_ptr, 1000, ...)
///     if result == 1:  # Done
///         break
///     # Parse batch buffer for up to 1000 results
/// lib.sochdb_scan_free(iter_ptr)
/// ```
///
/// # Safety
///
/// - `iter_ptr` must be a valid ScanIteratorPtr from sochdb_scan
/// - Output pointers must be valid
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_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
}

// ============================================================================
// Per-Table Index Policy API
// ============================================================================

/// Set index policy for a table.
///
/// # Policy Values
/// - 0: WriteOptimized - O(1) writes, O(N) scans. For write-heavy tables.
/// - 1: Balanced (default) - O(1) amortized writes, O(output + log K) scans.
/// - 2: ScanOptimized - O(log N) writes, O(log N + K) scans. For analytics.
/// - 3: AppendOnly - O(1) writes, O(N) forward-only scans. For time-series.
///
/// # Returns
/// - 0: Success
/// - -1: Invalid pointer or table name
/// - -2: Invalid policy value
///
/// # Safety
/// ptr must be a valid DatabasePtr, table_name must be a valid C string.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_set_table_index_policy(
    ptr: *mut DatabasePtr,
    table_name: *const c_char,
    policy: u8,
) -> c_int {
    if ptr.is_null() || table_name.is_null() {
        return -1;
    }
    
    let c_str = unsafe { CStr::from_ptr(table_name) };
    let table = match c_str.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    
    let index_policy = match policy {
        0 => crate::index_policy::IndexPolicy::WriteOptimized,
        1 => crate::index_policy::IndexPolicy::Balanced,
        2 => crate::index_policy::IndexPolicy::ScanOptimized,
        3 => crate::index_policy::IndexPolicy::AppendOnly,
        _ => return -2,
    };
    
    let db = unsafe { &(*ptr).0 };
    
    // Configure the table's index policy through the database registry
    let config = crate::index_policy::TableIndexConfig::new(table, index_policy);
    db.index_registry().configure_table(config);
    
    0
}

/// Get index policy for a table.
///
/// # Returns
/// - 0: WriteOptimized
/// - 1: Balanced  
/// - 2: ScanOptimized
/// - 3: AppendOnly
/// - 255: Error (invalid pointer)
///
/// # Safety
/// ptr must be a valid DatabasePtr, table_name must be a valid C string.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_get_table_index_policy(
    ptr: *mut DatabasePtr,
    table_name: *const c_char,
) -> u8 {
    if ptr.is_null() || table_name.is_null() {
        return 255;
    }
    
    let c_str = unsafe { CStr::from_ptr(table_name) };
    let table = match c_str.to_str() {
        Ok(s) => s,
        Err(_) => return 255,
    };
    
    let db = unsafe { &(*ptr).0 };
    let config = db.index_registry().get_config(table);
    
    match config.policy {
        crate::index_policy::IndexPolicy::WriteOptimized => 0,
        crate::index_policy::IndexPolicy::Balanced => 1,
        crate::index_policy::IndexPolicy::ScanOptimized => 2,
        crate::index_policy::IndexPolicy::AppendOnly => 3,
    }
}

/// C-compatible Temporal Edge
#[repr(C)]
pub struct C_TemporalEdge {
    pub from_id: *const c_char,
    pub edge_type: *const c_char,
    pub to_id: *const c_char,
    pub valid_from: u64,
    pub valid_until: u64,
    pub properties_json: *const c_char,  // JSON string of properties
}

/// Add a temporal edge with validity interval.
/// # Safety
/// All pointers must be valid C strings. properties_json can be null.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_add_temporal_edge(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    edge: C_TemporalEdge,
) -> c_int {
    if ptr.is_null() || namespace.is_null() || edge.from_id.is_null() 
        || edge.edge_type.is_null() || edge.to_id.is_null() {
        return -1;
    }

    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let from = match unsafe { CStr::from_ptr(edge.from_id) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let etype = match unsafe { CStr::from_ptr(edge.edge_type) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let to = match unsafe { CStr::from_ptr(edge.to_id) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };

    let db = unsafe { &(*ptr).0 };
    
    // Begin transaction for atomic write
    let txn = match db.begin_transaction() {
        Ok(t) => t,
        Err(_) => return -1,
    };
    
    // Store temporal edge: _graph/{ns}/temporal/{from}/{type}/{to}/{valid_from}
    let key = format!(
        "_graph/{}/temporal/{}/{}/{}/{:016x}",
        ns, from, etype, to, edge.valid_from
    );
    
    let props_str = if edge.properties_json.is_null() {
        "{}".to_string()
    } else {
        match unsafe { CStr::from_ptr(edge.properties_json) }.to_str() {
            Ok(s) => s.to_string(),
            Err(_) => return -1,
        }
    };
    
    let value = format!(
        r#"{{"from_id":"{}","edge_type":"{}","to_id":"{}","valid_from":{},"valid_until":{},"properties":{}}}"#,
        from, etype, to, edge.valid_from, edge.valid_until, props_str
    );
    
    if let Err(_) = db.put(txn, key.as_bytes(), value.as_bytes()) {
        let _ = db.abort(txn);
        return -1;
    }
    
    match db.commit(txn) {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// Query temporal graph edges. Returns a JSON array of matching edges.
/// Caller must free the returned string with sochdb_free_string.
/// 
/// query_mode: 0=POINT_IN_TIME, 1=RANGE, 2=CURRENT
/// # Safety
/// All pointers must be valid C strings. edge_type can be null for no filter.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_query_temporal_graph(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    node_id: *const c_char,
    query_mode: u8,
    timestamp: u64,      // For POINT_IN_TIME
    start_time: u64,     // For RANGE
    end_time: u64,       // For RANGE
    edge_type: *const c_char,  // Optional filter (null = all types)
    out_len: *mut usize,
) -> *mut c_char {
    if ptr.is_null() || namespace.is_null() || node_id.is_null() || out_len.is_null() {
        return ptr::null_mut();
    }

    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s,
        Err(_) => return ptr::null_mut(),
    };
    let node = match unsafe { CStr::from_ptr(node_id) }.to_str() {
        Ok(s) => s,
        Err(_) => return ptr::null_mut(),
    };
    
    let edge_filter = if edge_type.is_null() {
        None
    } else {
        match unsafe { CStr::from_ptr(edge_type) }.to_str() {
            Ok(s) => Some(s),
            Err(_) => return ptr::null_mut(),
        }
    };

    let db = unsafe { &(*ptr).0 };
    
    // Begin transaction for scan
    let txn = match db.begin_transaction() {
        Ok(t) => t,
        Err(_) => return ptr::null_mut(),
    };
    
    // Scan prefix: _graph/{ns}/temporal/{node}/
    let prefix = format!("_graph/{}/temporal/{}/", ns, node);
    let pairs = match db.scan(txn, prefix.as_bytes()) {
        Ok(p) => p,
        Err(_) => {
            let _ = db.abort(txn);
            return ptr::null_mut();
        }
    };
    
    // Commit read transaction
    if let Err(_) = db.commit(txn) {
        return ptr::null_mut();
    }
    
    let mut results = Vec::new();
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_millis() as u64;
    
    for (_key, value) in pairs {
        // Parse the JSON value
        let value_str = match std::str::from_utf8(&value) {
            Ok(s) => s,
            Err(_) => continue,
        };
        
        // Simple JSON parsing (in production, use serde_json)
        if let Some(valid_from_pos) = value_str.find(r#""valid_from":"#) {
            if let Some(valid_until_pos) = value_str.find(r#""valid_until":"#) {
                let vf_start = valid_from_pos + r#""valid_from":"#.len();
                let vf_end = value_str[vf_start..].find(',').unwrap_or(0) + vf_start;
                let vu_start = valid_until_pos + r#""valid_until":"#.len();
                let vu_end = value_str[vu_start..].find(',').unwrap_or(0) + vu_start;
                
                let valid_from: u64 = value_str[vf_start..vf_end].parse().unwrap_or(0);
                let valid_until: u64 = value_str[vu_start..vu_end].parse().unwrap_or(0);
                
                // Filter by edge_type if specified
                if let Some(filter) = edge_filter {
                    if !value_str.contains(&format!(r#""edge_type":"{}""#, filter)) {
                        continue;
                    }
                }
                
                // Filter by query mode
                let matches = match query_mode {
                    0 => timestamp >= valid_from && (valid_until == 0 || timestamp < valid_until),
                    1 => {
                        let edge_end = if valid_until == 0 { u64::MAX } else { valid_until };
                        valid_from < end_time && edge_end > start_time
                    }
                    2 => now >= valid_from && (valid_until == 0 || now < valid_until),
                    _ => false,
                };
                
                if matches {
                    results.push(value_str.to_string());
                }
            }
        }
    }
    
    // Build JSON array
    let json = format!("[{}]", results.join(","));
    let c_string = match std::ffi::CString::new(json) {
        Ok(s) => s,
        Err(_) => return ptr::null_mut(),
    };
    
    unsafe { *out_len = c_string.as_bytes().len() };
    c_string.into_raw()
}

/// Free a string returned by sochdb_query_temporal_graph.
/// # Safety
/// The ptr must be a valid pointer returned by sochdb_query_temporal_graph.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_free_string(ptr: *mut c_char) {
    if !ptr.is_null() {
        unsafe {
            let _ = std::ffi::CString::from_raw(ptr);
        }
    }
}

// ============================================================================
// Graph Overlay FFI - Nodes, Edges, Traversal
// ============================================================================

/// Add a node to the graph overlay.
/// 
/// Stores node as: _graph/{namespace}/nodes/{node_id}
/// 
/// # Returns
/// - 0: Success
/// - -1: Error
/// 
/// # Safety
/// All pointers must be valid C strings. properties_json can be null.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_graph_add_node(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    node_id: *const c_char,
    node_type: *const c_char,
    properties_json: *const c_char,
) -> c_int {
    if ptr.is_null() || namespace.is_null() || node_id.is_null() || node_type.is_null() {
        return -1;
    }

    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let id = match unsafe { CStr::from_ptr(node_id) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let ntype = match unsafe { CStr::from_ptr(node_type) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let props = if properties_json.is_null() {
        "{}".to_string()
    } else {
        match unsafe { CStr::from_ptr(properties_json) }.to_str() {
            Ok(s) => s.to_string(),
            Err(_) => return -1,
        }
    };

    let db = unsafe { &(*ptr).0 };
    
    let txn = match db.begin_transaction() {
        Ok(t) => t,
        Err(_) => return -1,
    };
    
    let key = format!("_graph/{}/nodes/{}", ns, id);
    let value = format!(
        r#"{{"id":"{}","node_type":"{}","properties":{}}}"#,
        id, ntype, props
    );
    
    if let Err(_) = db.put(txn, key.as_bytes(), value.as_bytes()) {
        let _ = db.abort(txn);
        return -1;
    }
    
    match db.commit(txn) {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// Add an edge between nodes in the graph overlay.
/// 
/// Stores edge as: _graph/{namespace}/edges/{from_id}/{edge_type}/{to_id}
/// 
/// # Returns
/// - 0: Success
/// - -1: Error
/// 
/// # Safety
/// All pointers must be valid C strings. properties_json can be null.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_graph_add_edge(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    from_id: *const c_char,
    edge_type: *const c_char,
    to_id: *const c_char,
    properties_json: *const c_char,
) -> c_int {
    if ptr.is_null() || namespace.is_null() || from_id.is_null() 
        || edge_type.is_null() || to_id.is_null() {
        return -1;
    }

    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let from = match unsafe { CStr::from_ptr(from_id) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let etype = match unsafe { CStr::from_ptr(edge_type) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let to = match unsafe { CStr::from_ptr(to_id) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let props = if properties_json.is_null() {
        "{}".to_string()
    } else {
        match unsafe { CStr::from_ptr(properties_json) }.to_str() {
            Ok(s) => s.to_string(),
            Err(_) => return -1,
        }
    };

    let db = unsafe { &(*ptr).0 };
    
    let txn = match db.begin_transaction() {
        Ok(t) => t,
        Err(_) => return -1,
    };
    
    let key = format!("_graph/{}/edges/{}/{}/{}", ns, from, etype, to);
    let value = format!(
        r#"{{"from_id":"{}","edge_type":"{}","to_id":"{}","properties":{}}}"#,
        from, etype, to, props
    );
    
    if let Err(_) = db.put(txn, key.as_bytes(), value.as_bytes()) {
        let _ = db.abort(txn);
        return -1;
    }
    
    match db.commit(txn) {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// Traverse the graph from a starting node.
/// 
/// Returns JSON: {"nodes": [...], "edges": [...]}
/// Caller must free the returned string with sochdb_free_string.
/// 
/// order: 0=BFS, 1=DFS
/// 
/// # Safety
/// All pointers must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_graph_traverse(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    start_node: *const c_char,
    max_depth: usize,
    order: u8,  // 0=BFS, 1=DFS
    out_len: *mut usize,
) -> *mut c_char {
    if ptr.is_null() || namespace.is_null() || start_node.is_null() || out_len.is_null() {
        return ptr::null_mut();
    }

    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s,
        Err(_) => return ptr::null_mut(),
    };
    let start = match unsafe { CStr::from_ptr(start_node) }.to_str() {
        Ok(s) => s,
        Err(_) => return ptr::null_mut(),
    };

    let db = unsafe { &(*ptr).0 };
    
    let txn = match db.begin_transaction() {
        Ok(t) => t,
        Err(_) => return ptr::null_mut(),
    };
    
    // Collect nodes and edges through traversal
    let mut visited_nodes = std::collections::HashSet::new();
    let mut nodes_json = Vec::new();
    let mut edges_json = Vec::new();
    
    // Use queue for BFS, stack for DFS
    let mut frontier: Vec<(String, usize)> = vec![(start.to_string(), 0)];
    
    while let Some((current_node, depth)) = if order == 0 {
        // BFS: remove from front
        if frontier.is_empty() { None } else { Some(frontier.remove(0)) }
    } else {
        // DFS: remove from back
        frontier.pop()
    } {
        if depth > max_depth || visited_nodes.contains(&current_node) {
            continue;
        }
        visited_nodes.insert(current_node.clone());
        
        // Get node data
        let node_key = format!("_graph/{}/nodes/{}", ns, current_node);
        if let Ok(Some(node_data)) = db.get(txn, node_key.as_bytes()) {
            if let Ok(s) = std::str::from_utf8(&node_data) {
                nodes_json.push(s.to_string());
            }
        }
        
        // Get outgoing edges
        let edge_prefix = format!("_graph/{}/edges/{}/", ns, current_node);
        if let Ok(edges) = db.scan(txn, edge_prefix.as_bytes()) {
            for (_key, value) in edges {
                if let Ok(edge_str) = std::str::from_utf8(&value) {
                    edges_json.push(edge_str.to_string());
                    
                    // Extract to_id for traversal
                    if let Some(to_pos) = edge_str.find(r#""to_id":""#) {
                        let start_idx = to_pos + r#""to_id":""#.len();
                        if let Some(end_idx) = edge_str[start_idx..].find('"') {
                            let to_id = &edge_str[start_idx..start_idx + end_idx];
                            if !visited_nodes.contains(to_id) {
                                frontier.push((to_id.to_string(), depth + 1));
                            }
                        }
                    }
                }
            }
        }
    }
    
    if let Err(_) = db.commit(txn) {
        return ptr::null_mut();
    }
    
    let result = format!(
        r#"{{"nodes":[{}],"edges":[{}]}}"#,
        nodes_json.join(","),
        edges_json.join(",")
    );
    
    let c_string = match std::ffi::CString::new(result) {
        Ok(s) => s,
        Err(_) => return ptr::null_mut(),
    };
    
    unsafe { *out_len = c_string.as_bytes().len() };
    c_string.into_raw()
}

// ============================================================================
// Semantic Cache FFI
// ============================================================================

/// Store a value in the semantic cache with its embedding.
/// 
/// # Returns
/// - 0: Success
/// - -1: Error
/// 
/// # Safety
/// All pointers must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_cache_put(
    ptr: *mut DatabasePtr,
    cache_name: *const c_char,
    key: *const c_char,
    value: *const c_char,
    embedding_ptr: *const f32,
    embedding_len: usize,
    ttl_seconds: u64,
) -> c_int {
    if ptr.is_null() || cache_name.is_null() || key.is_null() 
        || value.is_null() || embedding_ptr.is_null() {
        return -1;
    }

    let cache = match unsafe { CStr::from_ptr(cache_name) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let k = match unsafe { CStr::from_ptr(key) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let v = match unsafe { CStr::from_ptr(value) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let embedding = unsafe { slice::from_raw_parts(embedding_ptr, embedding_len) };

    let db = unsafe { &(*ptr).0 };
    
    let txn = match db.begin_transaction() {
        Ok(t) => t,
        Err(_) => return -1,
    };
    
    // Compute expiry timestamp
    let expires_at = if ttl_seconds > 0 {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_secs() + ttl_seconds
    } else {
        0 // No expiry
    };
    
    // Store cache entry: _cache/{cache_name}/{key_hash}
    let key_hash = format!("{:016x}", twox_hash::xxh3::hash64(k.as_bytes()));
    let cache_key = format!("_cache/{}/{}", cache, key_hash);
    
    // Serialize embedding as JSON array
    let embedding_json: Vec<String> = embedding.iter().map(|f| f.to_string()).collect();
    
    let cache_value = format!(
        r#"{{"key":"{}","value":"{}","embedding":[{}],"expires_at":{}}}"#,
        k, v, embedding_json.join(","), expires_at
    );
    
    if let Err(_) = db.put(txn, cache_key.as_bytes(), cache_value.as_bytes()) {
        let _ = db.abort(txn);
        return -1;
    }
    
    match db.commit(txn) {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// Look up a value in the semantic cache by embedding similarity.
/// 
/// Returns the cached value if similarity >= threshold, null otherwise.
/// Caller must free the returned string with sochdb_free_string.
/// 
/// # Safety
/// All pointers must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_cache_get(
    ptr: *mut DatabasePtr,
    cache_name: *const c_char,
    query_embedding_ptr: *const f32,
    embedding_len: usize,
    threshold: f32,
    out_len: *mut usize,
) -> *mut c_char {
    if ptr.is_null() || cache_name.is_null() || query_embedding_ptr.is_null() || out_len.is_null() {
        return ptr::null_mut();
    }

    let cache = match unsafe { CStr::from_ptr(cache_name) }.to_str() {
        Ok(s) => s,
        Err(_) => return ptr::null_mut(),
    };
    let query = unsafe { slice::from_raw_parts(query_embedding_ptr, embedding_len) };

    let db = unsafe { &(*ptr).0 };
    
    let txn = match db.begin_transaction() {
        Ok(t) => t,
        Err(_) => return ptr::null_mut(),
    };
    
    let prefix = format!("_cache/{}/", cache);
    let entries = match db.scan(txn, prefix.as_bytes()) {
        Ok(e) => e,
        Err(_) => {
            let _ = db.abort(txn);
            return ptr::null_mut();
        }
    };
    
    let _ = db.commit(txn);
    
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_secs();
    
    let mut best_match: Option<(f32, String)> = None;
    
    for (_key, value) in entries {
        let value_str = match std::str::from_utf8(&value) {
            Ok(s) => s,
            Err(_) => continue,
        };
        
        // Parse expires_at
        if let Some(exp_pos) = value_str.find(r#""expires_at":"#) {
            let exp_start = exp_pos + r#""expires_at":"#.len();
            if let Some(exp_end) = value_str[exp_start..].find('}') {
                let expires_at: u64 = value_str[exp_start..exp_start + exp_end]
                    .parse()
                    .unwrap_or(0);
                if expires_at > 0 && now > expires_at {
                    continue; // Expired
                }
            }
        }
        
        // Parse embedding and compute cosine similarity
        if let Some(emb_pos) = value_str.find(r#""embedding":["#) {
            let emb_start = emb_pos + r#""embedding":["#.len();
            if let Some(emb_end) = value_str[emb_start..].find(']') {
                let emb_str = &value_str[emb_start..emb_start + emb_end];
                let cached_embedding: Vec<f32> = emb_str
                    .split(',')
                    .filter_map(|s| s.trim().parse().ok())
                    .collect();
                
                if cached_embedding.len() == query.len() {
                    let similarity = cosine_similarity(query, &cached_embedding);
                    if similarity >= threshold {
                        if best_match.is_none() || similarity > best_match.as_ref().unwrap().0 {
                            // Extract value field
                            if let Some(val_pos) = value_str.find(r#""value":""#) {
                                let val_start = val_pos + r#""value":""#.len();
                                if let Some(val_end) = value_str[val_start..].find('"') {
                                    let cached_value = &value_str[val_start..val_start + val_end];
                                    best_match = Some((similarity, cached_value.to_string()));
                                }
                            }
                        }
                    }
                }
            }
        }
    }
    
    match best_match {
        Some((_, value)) => {
            let c_string = match std::ffi::CString::new(value) {
                Ok(s) => s,
                Err(_) => return ptr::null_mut(),
            };
            unsafe { *out_len = c_string.as_bytes().len() };
            c_string.into_raw()
        }
        None => ptr::null_mut(),
    }
}

/// Compute cosine similarity between two vectors
/// Returns normalized similarity in [0, 1] range for threshold comparisons
fn cosine_similarity(a: &[f32], b: &[f32]) -> f32 {
    let dot: f32 = a.iter().zip(b.iter()).map(|(x, y)| x * y).sum();
    let norm_a: f32 = a.iter().map(|x| x * x).sum::<f32>().sqrt();
    let norm_b: f32 = b.iter().map(|x| x * x).sum::<f32>().sqrt();
    if norm_a == 0.0 || norm_b == 0.0 {
        0.0
    } else {
        let similarity = dot / (norm_a * norm_b);
        // Normalize from [-1, 1] to [0, 1] for threshold comparisons
        // This ensures consistent scoring across all SDKs (Python/Node.js/Go)
        (similarity + 1.0) / 2.0
    }
}

// ============================================================================
// Trace Service FFI
// ============================================================================

/// Start a new trace. Returns trace_id and root_span_id.
/// 
/// Caller must free the returned strings with sochdb_free_string.
/// 
/// # Returns
/// - 0: Success
/// - -1: Error
/// 
/// # Safety
/// All pointers must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_trace_start(
    ptr: *mut DatabasePtr,
    name: *const c_char,
    trace_id_out: *mut *mut c_char,
    span_id_out: *mut *mut c_char,
) -> c_int {
    if ptr.is_null() || name.is_null() || trace_id_out.is_null() || span_id_out.is_null() {
        return -1;
    }

    let trace_name = match unsafe { CStr::from_ptr(name) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };

    let db = unsafe { &(*ptr).0 };
    
    // Generate unique IDs
    let trace_id = format!("trace_{:016x}", rand_u64());
    let span_id = format!("span_{:016x}", rand_u64());
    
    let txn = match db.begin_transaction() {
        Ok(t) => t,
        Err(_) => return -1,
    };
    
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_micros() as u64;
    
    // Store trace: _traces/{trace_id}
    let trace_key = format!("_traces/{}", trace_id);
    let trace_value = format!(
        r#"{{"trace_id":"{}","name":"{}","start_us":{},"root_span_id":"{}"}}"#,
        trace_id, trace_name, now, span_id
    );
    
    if let Err(_) = db.put(txn, trace_key.as_bytes(), trace_value.as_bytes()) {
        let _ = db.abort(txn);
        return -1;
    }
    
    // Store root span: _traces/{trace_id}/spans/{span_id}
    let span_key = format!("_traces/{}/spans/{}", trace_id, span_id);
    let span_value = format!(
        r#"{{"span_id":"{}","name":"{}","start_us":{},"parent_span_id":null,"status":"active"}}"#,
        span_id, trace_name, now
    );
    
    if let Err(_) = db.put(txn, span_key.as_bytes(), span_value.as_bytes()) {
        let _ = db.abort(txn);
        return -1;
    }
    
    if let Err(_) = db.commit(txn) {
        return -1;
    }
    
    // Return trace_id and span_id
    let trace_c = match std::ffi::CString::new(trace_id) {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let span_c = match std::ffi::CString::new(span_id) {
        Ok(s) => s,
        Err(_) => return -1,
    };
    
    unsafe {
        *trace_id_out = trace_c.into_raw();
        *span_id_out = span_c.into_raw();
    }
    
    0
}

/// Start a child span within a trace.
/// 
/// Caller must free the returned span_id with sochdb_free_string.
/// 
/// # Safety
/// All pointers must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_trace_span_start(
    ptr: *mut DatabasePtr,
    trace_id: *const c_char,
    parent_span_id: *const c_char,
    name: *const c_char,
    span_id_out: *mut *mut c_char,
) -> c_int {
    if ptr.is_null() || trace_id.is_null() || parent_span_id.is_null() 
        || name.is_null() || span_id_out.is_null() {
        return -1;
    }

    let tid = match unsafe { CStr::from_ptr(trace_id) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let pid = match unsafe { CStr::from_ptr(parent_span_id) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let span_name = match unsafe { CStr::from_ptr(name) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };

    let db = unsafe { &(*ptr).0 };
    let span_id = format!("span_{:016x}", rand_u64());
    
    let txn = match db.begin_transaction() {
        Ok(t) => t,
        Err(_) => return -1,
    };
    
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_micros() as u64;
    
    let span_key = format!("_traces/{}/spans/{}", tid, span_id);
    let span_value = format!(
        r#"{{"span_id":"{}","name":"{}","start_us":{},"parent_span_id":"{}","status":"active"}}"#,
        span_id, span_name, now, pid
    );
    
    if let Err(_) = db.put(txn, span_key.as_bytes(), span_value.as_bytes()) {
        let _ = db.abort(txn);
        return -1;
    }
    
    if let Err(_) = db.commit(txn) {
        return -1;
    }
    
    let span_c = match std::ffi::CString::new(span_id) {
        Ok(s) => s,
        Err(_) => return -1,
    };
    
    unsafe { *span_id_out = span_c.into_raw() };
    0
}

/// End a span and record its duration.
/// 
/// status: 0=unset, 1=ok, 2=error
/// 
/// # Returns
/// Duration in microseconds on success, -1 on error.
/// 
/// # Safety
/// All pointers must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_trace_span_end(
    ptr: *mut DatabasePtr,
    trace_id: *const c_char,
    span_id: *const c_char,
    status: u8,
) -> i64 {
    if ptr.is_null() || trace_id.is_null() || span_id.is_null() {
        return -1;
    }

    let tid = match unsafe { CStr::from_ptr(trace_id) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let sid = match unsafe { CStr::from_ptr(span_id) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };

    let db = unsafe { &(*ptr).0 };
    
    let txn = match db.begin_transaction() {
        Ok(t) => t,
        Err(_) => return -1,
    };
    
    let span_key = format!("_traces/{}/spans/{}", tid, sid);
    
    // Read current span
    let span_data = match db.get(txn, span_key.as_bytes()) {
        Ok(Some(data)) => data,
        _ => {
            let _ = db.abort(txn);
            return -1;
        }
    };
    
    let span_str = match std::str::from_utf8(&span_data) {
        Ok(s) => s,
        Err(_) => {
            let _ = db.abort(txn);
            return -1;
        }
    };
    
    // Parse start_us
    let start_us = if let Some(pos) = span_str.find(r#""start_us":"#) {
        let start = pos + r#""start_us":"#.len();
        if let Some(end) = span_str[start..].find(',') {
            span_str[start..start + end].parse().unwrap_or(0u64)
        } else {
            0u64
        }
    } else {
        0u64
    };
    
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_micros() as u64;
    
    let duration_us = now.saturating_sub(start_us);
    let status_str = match status {
        1 => "ok",
        2 => "error",
        _ => "unset",
    };
    
    // Update span with end time and duration
    let new_span = span_str
        .replace(r#""status":"active""#, &format!(r#""status":"{}","end_us":{},"duration_us":{}"#, status_str, now, duration_us));
    
    if let Err(_) = db.put(txn, span_key.as_bytes(), new_span.as_bytes()) {
        let _ = db.abort(txn);
        return -1;
    }
    
    if let Err(_) = db.commit(txn) {
        return -1;
    }
    
    duration_us as i64
}

/// Generate a pseudo-random u64 (simple XorShift for trace IDs)
fn rand_u64() -> u64 {
    use std::sync::atomic::{AtomicU64, Ordering};
    static STATE: AtomicU64 = AtomicU64::new(0x853c49e6748fea9b);
    
    let mut s = STATE.load(Ordering::Relaxed);
    if s == 0 {
        s = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos() as u64;
    }
    s ^= s >> 12;
    s ^= s << 25;
    s ^= s >> 27;
    STATE.store(s, Ordering::Relaxed);
    s.wrapping_mul(0x2545F4914F6CDD1D)
}

// =========================================================================
// Vector Index Operations (KV-based, native Rust performance)
// =========================================================================

/// Create a vector collection for storing embeddings
/// 
/// # Returns
/// - 0: Success (or already exists)
/// - -1: Error
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_collection_create(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    collection: *const c_char,
    dimension: usize,
    dist_type: u8, // 0=Cosine, 1=Euclidean, 2=Dot
) -> c_int {
    if ptr.is_null() || namespace.is_null() || collection.is_null() {
        return -1;
    }
    
    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let col = match unsafe { CStr::from_ptr(collection) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    
    let db = unsafe { &(*ptr).0 };
    let txn = match db.begin_transaction() {
        Ok(t) => t,
        Err(_) => return -1,
    };
    
    // Store collection config
    let config_key = format!("{}/_collections/{}", ns, col);
    let config_value = format!(
        r#"{{"dimension":{},"metric":{}}}"#,
        dimension, dist_type
    );
    
    if let Err(_) = db.put(txn, config_key.as_bytes(), config_value.as_bytes()) {
        let _ = db.abort(txn);
        return -1;
    }
    
    let result = match db.commit(txn) {
        Ok(_) => 0,
        Err(_) => -1,
    };

    if result == 0 {
        let metric = match dist_type {
            1 => DistanceMetric::Euclidean,
            2 => DistanceMetric::DotProduct,
            _ => DistanceMetric::Cosine,
        };
        let _ = ensure_collection_index(db, ns, col, dimension, metric);
    }

    result
}

/// Insert a vector into a collection
/// 
/// # Returns
/// - 0: Success
/// - -1: Error
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_collection_insert(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    collection: *const c_char,
    id: *const c_char,
    vector_ptr: *const f32,
    vector_len: usize,
    metadata_json: *const c_char, // Optional JSON metadata
) -> c_int {
    if ptr.is_null() || namespace.is_null() || collection.is_null() 
        || id.is_null() || vector_ptr.is_null() {
        return -1;
    }
    
    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let col = match unsafe { CStr::from_ptr(collection) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let doc_id = match unsafe { CStr::from_ptr(id) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let vector = unsafe { slice::from_raw_parts(vector_ptr, vector_len) };
    let db = unsafe { &(*ptr).0 };

    let (dimension, metric) = match resolve_collection_config(db, ns, col) {
        Some(config) => config,
        None => (vector_len, DistanceMetric::Cosine),
    };
    if vector_len != dimension {
        return -1;
    }
    
    let metadata = if !metadata_json.is_null() {
        match unsafe { CStr::from_ptr(metadata_json) }.to_str() {
            Ok(s) => s.to_string(),
            Err(_) => "{}".to_string(),
        }
    } else {
        "{}".to_string()
    };
    
    let txn = match db.begin_transaction() {
        Ok(t) => t,
        Err(_) => return -1,
    };
    
    let id_hash = hash_id_to_u128(doc_id);
    let vec_key = vector_bin_key(ns, col, id_hash);
    let vec_value = serialize_vector_binary(vector);

    if let Err(_) = db.put(txn, vec_key.as_bytes(), &vec_value) {
        let _ = db.abort(txn);
        return -1;
    }

    let metadata_value = match serde_json::from_str::<serde_json::Value>(&metadata) {
        Ok(value) => serde_json::json!({"id": doc_id, "metadata": value}),
        Err(_) => serde_json::json!({"id": doc_id, "metadata": serde_json::json!({})}),
    };
    let meta_key = metadata_key(ns, col, id_hash);
    if let Ok(meta_bytes) = serde_json::to_vec(&metadata_value) {
        if let Err(_) = db.put(txn, meta_key.as_bytes(), &meta_bytes) {
            let _ = db.abort(txn);
            return -1;
        }
    }
    
    if let Err(_) = db.commit(txn) {
        return -1;
    }

    let index = ensure_collection_index(db, ns, col, dimension, metric);
    let _ = index.index.insert(id_hash, vector.to_vec());

    0
}

/// Batch insert vectors into a collection.
///
/// Takes parallel arrays of ids, vectors, and optional metadata.
/// Uses a single transaction for all inserts → much faster than per-vector.
///
/// # Returns
/// - >= 0: Number of successfully inserted vectors
/// - -1: Error
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_collection_insert_batch(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    collection: *const c_char,
    ids: *const *const c_char,        // Array of C strings
    vectors: *const f32,               // Flat array: count * dimension floats
    dimension: usize,
    metadata_jsons: *const *const c_char, // Array of C strings (nullable entries)
    count: usize,
) -> c_int {
    if ptr.is_null() || namespace.is_null() || collection.is_null()
        || ids.is_null() || vectors.is_null() || count == 0
    {
        return -1;
    }

    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let col = match unsafe { CStr::from_ptr(collection) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let db = unsafe { &(*ptr).0 };

    let (expected_dim, metric) = match resolve_collection_config(db, ns, col) {
        Some(config) => config,
        None => (dimension, DistanceMetric::Cosine),
    };
    if dimension != expected_dim {
        return -1;
    }

    // Single transaction for the entire batch
    let txn = match db.begin_transaction() {
        Ok(t) => t,
        Err(_) => return -1,
    };

    let ids_slice = unsafe { slice::from_raw_parts(ids, count) };
    let vectors_flat = unsafe { slice::from_raw_parts(vectors, count * dimension) };

    let mut inserted = 0i32;
    let mut id_hashes = Vec::with_capacity(count);
    let mut vector_copies = Vec::with_capacity(count);

    for i in 0..count {
        let doc_id = match unsafe { CStr::from_ptr(ids_slice[i]) }.to_str() {
            Ok(s) => s,
            Err(_) => continue,
        };
        let vec_start = i * dimension;
        let vector = &vectors_flat[vec_start..vec_start + dimension];

        let id_hash = hash_id_to_u128(doc_id);
        let vec_key = vector_bin_key(ns, col, id_hash);
        let vec_value = serialize_vector_binary(vector);

        if db.put(txn, vec_key.as_bytes(), &vec_value).is_err() {
            continue;
        }

        // Metadata
        let metadata = if !metadata_jsons.is_null() {
            let meta_ptr = unsafe { *metadata_jsons.add(i) };
            if !meta_ptr.is_null() {
                match unsafe { CStr::from_ptr(meta_ptr) }.to_str() {
                    Ok(s) => s.to_string(),
                    Err(_) => "{}".to_string(),
                }
            } else {
                "{}".to_string()
            }
        } else {
            "{}".to_string()
        };

        let metadata_value = match serde_json::from_str::<serde_json::Value>(&metadata) {
            Ok(value) => serde_json::json!({"id": doc_id, "metadata": value}),
            Err(_) => serde_json::json!({"id": doc_id, "metadata": serde_json::json!({})}),
        };
        let meta_key = metadata_key(ns, col, id_hash);
        if let Ok(meta_bytes) = serde_json::to_vec(&metadata_value) {
            let _ = db.put(txn, meta_key.as_bytes(), &meta_bytes);
        }

        id_hashes.push(id_hash);
        vector_copies.push(vector.to_vec());
        inserted += 1;
    }

    // Commit single transaction for entire batch
    if db.commit(txn).is_err() {
        return -1;
    }

    // Insert into HNSW index (after commit succeeds)
    let index = ensure_collection_index(db, ns, col, dimension, metric);
    for (id_hash, vector) in id_hashes.into_iter().zip(vector_copies.into_iter()) {
        let _ = index.index.insert(id_hash, vector);
    }

    inserted
}

/// C-compatible search result
#[repr(C)]
pub struct CSearchResult {
    pub id_ptr: *mut c_char,
    pub score: f32,
    pub metadata_ptr: *mut c_char,
}

/// Search a collection for nearest vectors
/// 
/// # Returns
/// - >= 0: Number of results
/// - -1: Error
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_collection_search(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    collection: *const c_char,
    query_ptr: *const f32,
    query_len: usize,
    k: usize,
    results_out: *mut CSearchResult,
) -> c_int {
    if ptr.is_null() || namespace.is_null() || collection.is_null() 
        || query_ptr.is_null() || results_out.is_null() {
        return -1;
    }
    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let col = match unsafe { CStr::from_ptr(collection) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let query = unsafe { slice::from_raw_parts(query_ptr, query_len) };
    let db = unsafe { &(*ptr).0 };
    let (dimension, metric) = match resolve_collection_config(db, ns, col) {
        Some(config) => config,
        None => return 0,
    };

    if query_len != dimension {
        return -1;
    }

    let index = ensure_collection_index(db, ns, col, dimension, metric);
    let mut scored = match index.index.search(query, k) {
        Ok(results) => results,
        Err(_) => return -1,
    };

    let result_count = scored.len().min(k);
    for (i, (id_hash, distance)) in scored.drain(..result_count).enumerate() {
        let meta_key = metadata_key(ns, col, id_hash);
        let txn = match db.begin_transaction() {
            Ok(t) => t,
            Err(_) => return -1,
        };
        let meta_value = db.get(txn, meta_key.as_bytes()).ok().flatten();
        let _ = db.commit(txn);

        let mut id_value = String::new();
        let mut metadata_json = serde_json::json!({});
        if let Some(bytes) = meta_value.as_deref() {
            if let Ok(parsed) = serde_json::from_slice::<serde_json::Value>(bytes) {
                id_value = parsed.get("id").and_then(|v| v.as_str()).unwrap_or("").to_string();
                metadata_json = parsed.get("metadata").cloned().unwrap_or(serde_json::json!({}));
            }
        }
        let metadata = serde_json::to_string(&metadata_json).unwrap_or_else(|_| "{}".to_string());

        let c_id = match std::ffi::CString::new(id_value) {
            Ok(s) => s.into_raw(),
            Err(_) => ptr::null_mut(),
        };
        let c_meta = match std::ffi::CString::new(metadata) {
            Ok(s) => s.into_raw(),
            Err(_) => ptr::null_mut(),
        };

        unsafe {
            (*results_out.add(i)).id_ptr = c_id;
            (*results_out.add(i)).score = decode_score(metric, distance);
            (*results_out.add(i)).metadata_ptr = c_meta;
        }
    }

    result_count as c_int
}

/// Search a collection and return results as struct-of-arrays (ids + scores)
///
/// - ids_out: pointer to u64 array (allocated by Rust)
/// - scores_out: pointer to f32 array (allocated by Rust)
/// - len_out: number of results
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_collection_search_soa(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    collection: *const c_char,
    query_ptr: *const f32,
    query_len: usize,
    k: usize,
    min_score: f32,
    filter_json: *const c_char,
    ids_hi_out: *mut *mut u64,
    ids_lo_out: *mut *mut u64,
    scores_out: *mut *mut f32,
    len_out: *mut usize,
) -> c_int {
    if ptr.is_null() || namespace.is_null() || collection.is_null()
        || query_ptr.is_null() || ids_hi_out.is_null() || ids_lo_out.is_null()
        || scores_out.is_null() || len_out.is_null() {
        return -1;
    }

    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let col = match unsafe { CStr::from_ptr(collection) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let query = unsafe { slice::from_raw_parts(query_ptr, query_len) };
    let db = unsafe { &(*ptr).0 };

    let (dimension, metric) = match resolve_collection_config(db, ns, col) {
        Some(config) => config,
        None => return 0,
    };
    if query_len != dimension {
        return -1;
    }

    let filter = if !filter_json.is_null() {
        match unsafe { CStr::from_ptr(filter_json) }.to_str() {
            Ok(s) => serde_json::from_str::<serde_json::Value>(s).ok(),
            Err(_) => None,
        }
    } else {
        None
    };

    let index = ensure_collection_index(db, ns, col, dimension, metric);
    let results = match index.index.search(query, k) {
        Ok(results) => results,
        Err(_) => return -1,
    };

    let mut ids_hi: Vec<u64> = Vec::with_capacity(results.len());
    let mut ids_lo: Vec<u64> = Vec::with_capacity(results.len());
    let mut scores: Vec<f32> = Vec::with_capacity(results.len());

    for (id_hash, distance) in results {
        let score = decode_score(metric, distance);
        if min_score > 0.0 && score < min_score {
            continue;
        }

        if let Some(filter_value) = &filter {
            let meta_key = metadata_key(ns, col, id_hash);
            let txn = match db.begin_transaction() {
                Ok(t) => t,
                Err(_) => return -1,
            };
            let meta_value = db.get(txn, meta_key.as_bytes()).ok().flatten();
            let _ = db.commit(txn);
            let meta_value = match meta_value {
                Some(value) => value,
                None => continue,
            };
            let parsed = match serde_json::from_slice::<serde_json::Value>(&meta_value) {
                Ok(value) => value,
                Err(_) => continue,
            };
            let metadata = parsed.get("metadata").cloned().unwrap_or(Value::Null);

            if !metadata_matches_filter(&metadata, filter_value) {
                continue;
            }
        }

        ids_hi.push((id_hash >> 64) as u64);
        ids_lo.push((id_hash & u128::from(u64::MAX)) as u64);
        scores.push(score);
        if ids_hi.len() >= k {
            break;
        }
    }

    let len = ids_hi.len();
    let mut ids_hi_box = ids_hi.into_boxed_slice();
    let mut ids_lo_box = ids_lo.into_boxed_slice();
    let mut scores_box = scores.into_boxed_slice();

    unsafe {
        *len_out = len;
        *ids_hi_out = ids_hi_box.as_mut_ptr();
        *ids_lo_out = ids_lo_box.as_mut_ptr();
        *scores_out = scores_box.as_mut_ptr();
    }

    std::mem::forget(ids_hi_box);
    std::mem::forget(ids_lo_box);
    std::mem::forget(scores_box);

    len as c_int
}

/// Fetch metadata JSON for a list of ids (u64 hashes)
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_collection_fetch_metadata_json(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    collection: *const c_char,
    ids_hi_ptr: *const u64,
    ids_lo_ptr: *const u64,
    ids_len: usize,
) -> *mut c_char {
    if ptr.is_null() || namespace.is_null() || collection.is_null()
        || ids_hi_ptr.is_null() || ids_lo_ptr.is_null() {
        return ptr::null_mut();
    }

    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s,
        Err(_) => return ptr::null_mut(),
    };
    let col = match unsafe { CStr::from_ptr(collection) }.to_str() {
        Ok(s) => s,
        Err(_) => return ptr::null_mut(),
    };
    let ids_hi = unsafe { slice::from_raw_parts(ids_hi_ptr, ids_len) };
    let ids_lo = unsafe { slice::from_raw_parts(ids_lo_ptr, ids_len) };
    let db = unsafe { &(*ptr).0 };

    let mut results = Vec::with_capacity(ids_len);
    for i in 0..ids_len {
        let id_hash = ((ids_hi[i] as u128) << 64) | (ids_lo[i] as u128);
        let meta_key = metadata_key(ns, col, id_hash);
        let txn = match db.begin_transaction() {
            Ok(t) => t,
            Err(_) => return ptr::null_mut(),
        };
        let meta_value = db.get(txn, meta_key.as_bytes()).ok().flatten();
        let _ = db.commit(txn);
        if let Some(bytes) = meta_value {
            if let Ok(parsed) = serde_json::from_slice::<serde_json::Value>(&bytes) {
                results.push(parsed);
                continue;
            }
        }
        results.push(serde_json::json!({"id": "", "metadata": {}}));
    }

    match serde_json::to_string(&results) {
        Ok(json) => match std::ffi::CString::new(json) {
            Ok(cstr) => cstr.into_raw(),
            Err(_) => ptr::null_mut(),
        },
        Err(_) => ptr::null_mut(),
    }
}

/// Free arrays returned by sochdb_collection_search_soa
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_collection_free_u64(ptr: *mut u64, len: usize) {
    if ptr.is_null() || len == 0 {
        return;
    }
    unsafe {
        let _ = Vec::from_raw_parts(ptr, len, len);
    }
}

#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_collection_free_f32(ptr: *mut f32, len: usize) {
    if ptr.is_null() || len == 0 {
        return;
    }
    unsafe {
        let _ = Vec::from_raw_parts(ptr, len, len);
    }
}

fn metadata_matches_filter(metadata: &Value, filter: &Value) -> bool {
    let filter_obj = match filter.as_object() {
        Some(obj) => obj,
        None => return true,
    };
    let metadata_obj = match metadata.as_object() {
        Some(obj) => obj,
        None => return false,
    };

    for (key, expected) in filter_obj.iter() {
        match metadata_obj.get(key) {
            Some(actual) if actual == expected => {}
            _ => return false,
        }
    }

    true
}

/// Search a collection for keywords (simple term match)
/// 
/// # Returns
/// - >= 0: Number of results
/// - -1: Error
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_collection_keyword_search(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    collection: *const c_char,
    query_ptr: *const c_char,
    k: usize,
    results_out: *mut CSearchResult,
) -> c_int {
    if ptr.is_null() || namespace.is_null() || collection.is_null() 
        || query_ptr.is_null() || results_out.is_null() {
        return -1;
    }
    
    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let col = match unsafe { CStr::from_ptr(collection) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };
    let query_str = match unsafe { CStr::from_ptr(query_ptr) }.to_str() {
        Ok(s) => s.to_lowercase(),
        Err(_) => return -1,
    };
    let terms: Vec<&str> = query_str.split_whitespace().collect();
    if terms.is_empty() {
        return 0;
    }
    
    let db = unsafe { &(*ptr).0 };
    let txn = match db.begin_transaction() {
        Ok(t) => t,
        Err(_) => return -1,
    };
    
    // Scan all vectors in collection (we assume vectors & metadata are stored together)
    let prefix = format!("{}/collections/{}/vectors/", ns, col);
    let entries = match db.scan(txn, prefix.as_bytes()) {
        Ok(e) => e,
        Err(_) => {
            let _ = db.abort(txn);
            return -1;
        }
    };
    let _ = db.commit(txn);
    
    // Score documents based on term frequency
    let mut scored: Vec<(f32, String, String)> = Vec::new();
    
    for (_key, value) in entries {
        // Parse whole JSON (robust)
        let doc: Value = match serde_json::from_slice(&value) {
            Ok(v) => v,
            Err(_) => continue,
        };
        
        // Search in metadata string (includes values)
        let metadata_val = doc.get("metadata");
        let metadata_str = metadata_val.map(|v| v.to_string()).unwrap_or("{}".to_string());
        
        // Also check "content" field if present (fallback compat)
        let content_str = doc.get("content").and_then(|v| v.as_str()).unwrap_or("");
        
        // Combine text to search
        let search_text = format!("{} {}", metadata_str, content_str).to_lowercase();
         
        let mut score = 0.0;
        for term in &terms {
            score += search_text.matches(term).count() as f32;
        }
        
        if score > 0.0 {
            let id = doc.get("id").and_then(|v| v.as_str()).unwrap_or("").to_string();
            if id.is_empty() { continue; }
            
            scored.push((score, id, metadata_str));
        }
    }
    
    // Sort by score descending
    scored.sort_by(|a, b| b.0.partial_cmp(&a.0).unwrap_or(std::cmp::Ordering::Equal));
    
    // Return top k
    let result_count = scored.len().min(k);
    for (i, (score, id, metadata)) in scored.into_iter().take(k).enumerate() {
        let c_id = match std::ffi::CString::new(id) {
            Ok(s) => s.into_raw(),
            Err(_) => ptr::null_mut(),
        };
        let c_meta = match std::ffi::CString::new(metadata) {
            Ok(s) => s.into_raw(),
            Err(_) => ptr::null_mut(),
        };
        
        unsafe {
            (*results_out.add(i)).id_ptr = c_id;
            (*results_out.add(i)).score = score;
            (*results_out.add(i)).metadata_ptr = c_meta;
        }
    }
    
    result_count as c_int
}

/// Free a search result
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_search_result_free(result: *mut CSearchResult, count: usize) {
    if result.is_null() {
        return;
    }
    
    for i in 0..count {
        let r = unsafe { &mut *result.add(i) };
        if !r.id_ptr.is_null() {
            let _ = unsafe { std::ffi::CString::from_raw(r.id_ptr) };
        }
        if !r.metadata_ptr.is_null() {
            let _ = unsafe { std::ffi::CString::from_raw(r.metadata_ptr) };
        }
    }
}

// ============================================================================
// NEW FFI: Key Existence Check
// ============================================================================

/// Check if a key exists without retrieving its value.
///
/// # Returns
/// - 1: Key exists
/// - 0: Key does not exist
/// - -1: Error
///
/// # Safety
/// All pointer arguments must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_exists(
    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.get(txn, key) {
        Ok(Some(_)) => 1,
        Ok(None) => 0,
        Err(_) => -1,
    }
}

// ============================================================================
// NEW FFI: Path-based delete and scan
// ============================================================================

/// Delete a key by path.
/// Returns 0 on success, -1 on error.
/// # Safety
/// All pointer arguments must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_delete_path(
    ptr: *mut DatabasePtr,
    handle: C_TxnHandle,
    path_ptr: *const c_char,
) -> c_int {
    if ptr.is_null() || path_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 txn = TxnHandle {
        txn_id: handle.txn_id,
        snapshot_ts: handle.snapshot_ts,
    };

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

/// Scan keys by path prefix. Returns JSON array of [{"path":"...", "value":"base64..."}]
/// Caller must free the returned string with sochdb_free_string.
///
/// # Safety
/// All pointer arguments must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_scan_path(
    ptr: *mut DatabasePtr,
    handle: C_TxnHandle,
    prefix_ptr: *const c_char,
    out_len: *mut usize,
) -> *mut c_char {
    if ptr.is_null() || prefix_ptr.is_null() || out_len.is_null() {
        return ptr::null_mut();
    }
    let db = unsafe { &(*ptr).0 };
    let c_str = unsafe { CStr::from_ptr(prefix_ptr) };
    let prefix_str = match c_str.to_str() {
        Ok(s) => s,
        Err(_) => return ptr::null_mut(),
    };
    let txn = TxnHandle {
        txn_id: handle.txn_id,
        snapshot_ts: handle.snapshot_ts,
    };

    match db.scan_path(txn, prefix_str) {
        Ok(pairs) => {
            let entries: Vec<String> = pairs.iter().map(|(path, value)| {
                let val_str = String::from_utf8_lossy(value);
                format!(r#"{{"path":"{}","value":"{}"}}"#, path, val_str)
            }).collect();
            let json = format!("[{}]", entries.join(","));
            let c_string = match std::ffi::CString::new(json) {
                Ok(s) => s,
                Err(_) => return ptr::null_mut(),
            };
            unsafe { *out_len = c_string.as_bytes().len() };
            c_string.into_raw()
        }
        Err(_) => ptr::null_mut(),
    }
}

// ============================================================================
// NEW FFI: Transaction Modes
// ============================================================================

/// Begin a read-only transaction (no write-set tracking, lower overhead).
/// Returns C_TxnHandle. On error, txn_id will be 0.
/// # Safety
/// ptr must be a valid pointer returned by sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_begin_read_only(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_read_only() {
        Ok(txn) => C_TxnHandle {
            txn_id: txn.txn_id,
            snapshot_ts: txn.snapshot_ts,
        },
        Err(_) => C_TxnHandle { txn_id: 0, snapshot_ts: 0 },
    }
}

/// Begin a write-only transaction (no snapshot reads, optimized for bulk loads).
/// Returns C_TxnHandle. On error, txn_id will be 0.
/// # Safety
/// ptr must be a valid pointer returned by sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_begin_write_only(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_write_only() {
        Ok(txn) => C_TxnHandle {
            txn_id: txn.txn_id,
            snapshot_ts: txn.snapshot_ts,
        },
        Err(_) => C_TxnHandle { txn_id: 0, snapshot_ts: 0 },
    }
}

// ============================================================================
// NEW FFI: Database Maintenance
// ============================================================================

/// Gracefully shut down the database, flushing all pending writes.
/// After this call the DatabasePtr is still valid but no new operations should be started.
/// Returns 0 on success, -1 on error.
/// # Safety
/// ptr must be a valid pointer returned by sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_shutdown(ptr: *mut DatabasePtr) -> c_int {
    if ptr.is_null() { return -1; }
    let db = unsafe { &(*ptr).0 };
    match db.shutdown() {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// Force fsync all data to durable storage.
/// Returns 0 on success, -1 on error.
/// # Safety
/// ptr must be a valid pointer returned by sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_fsync(ptr: *mut DatabasePtr) -> c_int {
    if ptr.is_null() { return -1; }
    let db = unsafe { &(*ptr).0 };
    match db.fsync() {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// Truncate the write-ahead log (WAL) up to the last checkpoint.
/// Returns 0 on success, -1 on error.
/// # Safety
/// ptr must be a valid pointer returned by sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_truncate_wal(ptr: *mut DatabasePtr) -> c_int {
    if ptr.is_null() { return -1; }
    let db = unsafe { &(*ptr).0 };
    match db.truncate_wal() {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// Run garbage collection (remove dead MVCC versions).
/// Returns the number of versions reclaimed, or -1 on error.
/// # Safety
/// ptr must be a valid pointer returned by sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_gc(ptr: *mut DatabasePtr) -> i64 {
    if ptr.is_null() { return -1; }
    let db = unsafe { &(*ptr).0 };
    db.gc() as i64
}

/// Perform a full checkpoint (flush memtable + dirty pages to durable storage).
/// Returns the checkpoint LSN on success, or 0 on error.
/// # Safety
/// ptr must be a valid pointer returned by sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_checkpoint_full(ptr: *mut DatabasePtr) -> u64 {
    if ptr.is_null() { return 0; }
    let db = unsafe { &(*ptr).0 };
    match db.checkpoint() {
        Ok(lsn) => lsn,
        Err(_) => 0,
    }
}

/// Extended storage statistics (JSON).
/// Returns JSON string with all stats. Caller must free with sochdb_free_string.
/// # Safety
/// ptr must be a valid pointer returned by sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_stats_json(
    ptr: *mut DatabasePtr,
    out_len: *mut usize,
) -> *mut c_char {
    if ptr.is_null() || out_len.is_null() { return ptr::null_mut(); }
    let db = unsafe { &(*ptr).0 };
    let storage_stats = db.storage_stats();
    let db_stats = db.stats();

    let json = format!(
        r#"{{"memtable_size_bytes":{},"wal_size_bytes":{},"active_transactions":{},"min_active_snapshot":{},"last_checkpoint_lsn":{},"transactions_started":{},"transactions_committed":{},"transactions_aborted":{},"queries_executed":{},"bytes_written":{},"bytes_read":{}}}"#,
        storage_stats.memtable_size_bytes,
        storage_stats.wal_size_bytes,
        storage_stats.active_transactions,
        storage_stats.min_active_snapshot,
        storage_stats.last_checkpoint_lsn,
        db_stats.transactions_started,
        db_stats.transactions_committed,
        db_stats.transactions_aborted,
        db_stats.queries_executed,
        db_stats.bytes_written,
        db_stats.bytes_read,
    );

    let c_string = match std::ffi::CString::new(json) {
        Ok(s) => s,
        Err(_) => return ptr::null_mut(),
    };
    unsafe { *out_len = c_string.as_bytes().len() };
    c_string.into_raw()
}

/// Get the database file path. Caller must free with sochdb_free_string.
/// # Safety
/// ptr must be a valid pointer returned by sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_path(
    ptr: *mut DatabasePtr,
    out_len: *mut usize,
) -> *mut c_char {
    if ptr.is_null() || out_len.is_null() { return ptr::null_mut(); }
    let db = unsafe { &(*ptr).0 };
    let path_str = db.path().to_string_lossy().to_string();
    let c_string = match std::ffi::CString::new(path_str) {
        Ok(s) => s,
        Err(_) => return ptr::null_mut(),
    };
    unsafe { *out_len = c_string.as_bytes().len() };
    c_string.into_raw()
}

// ============================================================================
// NEW FFI: Backup & Snapshot Operations
// ============================================================================

/// Create a backup of the database.
/// Returns 0 on success, -1 on error.
/// # Safety
/// All pointer arguments must be valid C strings.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_backup_create(
    ptr: *mut DatabasePtr,
    destination: *const c_char,
) -> c_int {
    if ptr.is_null() || destination.is_null() { return -1; }
    let db = unsafe { &(*ptr).0 };
    let dest = match unsafe { CStr::from_ptr(destination) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };

    // Flush before backup to ensure all data is on disk
    let _ = db.flush();

    let manager = crate::backup::BackupManager::new(db.path());
    match manager.create_backup(dest) {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// Restore a database from a backup.
/// Returns 0 on success, -1 on error.
/// # Safety
/// All pointer arguments must be valid C strings.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_backup_restore(
    ptr: *mut DatabasePtr,
    backup_path: *const c_char,
) -> c_int {
    if ptr.is_null() || backup_path.is_null() { return -1; }
    let db = unsafe { &(*ptr).0 };
    let path = match unsafe { CStr::from_ptr(backup_path) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };

    let manager = crate::backup::BackupManager::new(db.path());
    match manager.restore_backup(path) {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// List backups in a directory. Returns JSON array of backup metadata.
/// Caller must free the returned string with sochdb_free_string.
/// # Safety
/// backup_dir must be a valid C string. out_len must be a valid pointer.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_backup_list(
    backup_dir: *const c_char,
    out_len: *mut usize,
) -> *mut c_char {
    if backup_dir.is_null() || out_len.is_null() { return ptr::null_mut(); }
    let dir = match unsafe { CStr::from_ptr(backup_dir) }.to_str() {
        Ok(s) => s,
        Err(_) => return ptr::null_mut(),
    };

    match crate::backup::BackupManager::list_backups(dir) {
        Ok(backups) => {
            let entries: Vec<String> = backups.iter().map(|b| {
                format!(
                    r#"{{"name":"{}","timestamp":"{}","size_bytes":{}}}"#,
                    b.generate_name(),
                    b.generate_name(),
                    0  // BackupMetadata may not expose size, use 0 placeholder
                )
            }).collect();
            let json = format!("[{}]", entries.join(","));
            let c_string = match std::ffi::CString::new(json) {
                Ok(s) => s,
                Err(_) => return ptr::null_mut(),
            };
            unsafe { *out_len = c_string.as_bytes().len() };
            c_string.into_raw()
        }
        Err(_) => ptr::null_mut(),
    }
}

/// Verify a backup is valid and not corrupted.
/// Returns 1 if valid, 0 if invalid, -1 on error.
/// # Safety
/// backup_path must be a valid C string.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_backup_verify(
    backup_path: *const c_char,
) -> c_int {
    if backup_path.is_null() { return -1; }
    let path = match unsafe { CStr::from_ptr(backup_path) }.to_str() {
        Ok(s) => s,
        Err(_) => return -1,
    };

    match crate::backup::BackupManager::verify_backup(path) {
        Ok(true) => 1,
        Ok(false) => 0,
        Err(_) => -1,
    }
}

// ============================================================================
// NEW FFI: Graph Operations (delete, neighbors, find_path)
// ============================================================================

/// Delete a node from the graph overlay.
/// Also removes all edges involving this node.
/// Returns 0 on success, -1 on error.
/// # Safety
/// All pointers must be valid C strings.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_graph_delete_node(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    node_id: *const c_char,
) -> c_int {
    if ptr.is_null() || namespace.is_null() || node_id.is_null() { return -1; }
    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let id = match unsafe { CStr::from_ptr(node_id) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let db = unsafe { &(*ptr).0 };

    let txn = match db.begin_transaction() {
        Ok(t) => t, Err(_) => return -1,
    };

    // Delete the node itself
    let node_key = format!("_graph/{}/nodes/{}", ns, id);
    let _ = db.delete(txn, node_key.as_bytes());

    // Delete outgoing edges
    let edge_prefix_out = format!("_graph/{}/edges/{}/", ns, id);
    if let Ok(edges) = db.scan(txn, edge_prefix_out.as_bytes()) {
        for (key, _) in edges {
            let _ = db.delete(txn, &key);
        }
    }

    // Delete temporal edges from this node
    let temporal_prefix = format!("_graph/{}/temporal/{}/", ns, id);
    if let Ok(edges) = db.scan(txn, temporal_prefix.as_bytes()) {
        for (key, _) in edges {
            let _ = db.delete(txn, &key);
        }
    }

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

/// Delete a specific edge from the graph overlay.
/// Returns 0 on success, -1 on error.
/// # Safety
/// All pointers must be valid C strings.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_graph_delete_edge(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    from_id: *const c_char,
    edge_type: *const c_char,
    to_id: *const c_char,
) -> c_int {
    if ptr.is_null() || namespace.is_null() || from_id.is_null()
        || edge_type.is_null() || to_id.is_null() { return -1; }

    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let from = match unsafe { CStr::from_ptr(from_id) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let etype = match unsafe { CStr::from_ptr(edge_type) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let to = match unsafe { CStr::from_ptr(to_id) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let db = unsafe { &(*ptr).0 };

    let txn = match db.begin_transaction() {
        Ok(t) => t, Err(_) => return -1,
    };

    let key = format!("_graph/{}/edges/{}/{}/{}", ns, from, etype, to);
    match db.delete(txn, key.as_bytes()) {
        Ok(_) => match db.commit(txn) {
            Ok(_) => 0,
            Err(_) => -1,
        },
        Err(_) => {
            let _ = db.abort(txn);
            -1
        }
    }
}

/// Get neighbors of a node. Returns JSON: {"neighbors": [...]}
/// direction: 0=outgoing, 1=incoming, 2=both
/// edge_type can be null for all edge types.
/// Caller must free the returned string with sochdb_free_string.
/// # Safety
/// All pointers must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_graph_get_neighbors(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    node_id: *const c_char,
    direction: u8,
    edge_type_filter: *const c_char,
    out_len: *mut usize,
) -> *mut c_char {
    if ptr.is_null() || namespace.is_null() || node_id.is_null() || out_len.is_null() {
        return ptr::null_mut();
    }
    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s, Err(_) => return ptr::null_mut(),
    };
    let node = match unsafe { CStr::from_ptr(node_id) }.to_str() {
        Ok(s) => s, Err(_) => return ptr::null_mut(),
    };
    let et_filter = if edge_type_filter.is_null() {
        None
    } else {
        match unsafe { CStr::from_ptr(edge_type_filter) }.to_str() {
            Ok(s) => Some(s), Err(_) => None,
        }
    };
    let db = unsafe { &(*ptr).0 };

    let txn = match db.begin_transaction() {
        Ok(t) => t, Err(_) => return ptr::null_mut(),
    };

    let mut neighbors = Vec::new();

    // Outgoing edges (direction 0 or 2)
    if direction == 0 || direction == 2 {
        let prefix = format!("_graph/{}/edges/{}/", ns, node);
        if let Ok(edges) = db.scan(txn, prefix.as_bytes()) {
            for (_key, value) in edges {
                if let Ok(edge_str) = std::str::from_utf8(&value) {
                    if let Some(filter) = et_filter {
                        if !edge_str.contains(&format!(r#""edge_type":"{}""#, filter)) {
                            continue;
                        }
                    }
                    // Extract to_id
                    if let Some(to_pos) = edge_str.find(r#""to_id":""#) {
                        let start = to_pos + r#""to_id":""#.len();
                        if let Some(end) = edge_str[start..].find('"') {
                            let to_id = &edge_str[start..start + end];
                            neighbors.push(format!(
                                r#"{{"node_id":"{}","direction":"outgoing","edge":{}}}"#,
                                to_id, edge_str
                            ));
                        }
                    }
                }
            }
        }
    }

    // Incoming edges (direction 1 or 2) — scan ALL edges and filter by to_id
    if direction == 1 || direction == 2 {
        let all_edges_prefix = format!("_graph/{}/edges/", ns);
        if let Ok(edges) = db.scan(txn, all_edges_prefix.as_bytes()) {
            for (_key, value) in edges {
                if let Ok(edge_str) = std::str::from_utf8(&value) {
                    let has_to = edge_str.contains(&format!(r#""to_id":"{}""#, node));
                    if !has_to { continue; }
                    if let Some(filter) = et_filter {
                        if !edge_str.contains(&format!(r#""edge_type":"{}""#, filter)) {
                            continue;
                        }
                    }
                    if let Some(from_pos) = edge_str.find(r#""from_id":""#) {
                        let start = from_pos + r#""from_id":""#.len();
                        if let Some(end) = edge_str[start..].find('"') {
                            let from_id = &edge_str[start..start + end];
                            neighbors.push(format!(
                                r#"{{"node_id":"{}","direction":"incoming","edge":{}}}"#,
                                from_id, edge_str
                            ));
                        }
                    }
                }
            }
        }
    }

    let _ = db.commit(txn);

    let json = format!(r#"{{"neighbors":[{}]}}"#, neighbors.join(","));
    let c_string = match std::ffi::CString::new(json) {
        Ok(s) => s, Err(_) => return ptr::null_mut(),
    };
    unsafe { *out_len = c_string.as_bytes().len() };
    c_string.into_raw()
}

/// Find shortest path between two nodes using BFS.
/// Returns JSON: {"path": ["node1","node2",...], "edges": [...]}
/// Returns null if no path found.
/// Caller must free the returned string with sochdb_free_string.
/// # Safety
/// All pointers must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_graph_find_path(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    from_node: *const c_char,
    to_node: *const c_char,
    max_depth: usize,
    out_len: *mut usize,
) -> *mut c_char {
    if ptr.is_null() || namespace.is_null() || from_node.is_null()
        || to_node.is_null() || out_len.is_null() {
        return ptr::null_mut();
    }
    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s, Err(_) => return ptr::null_mut(),
    };
    let from = match unsafe { CStr::from_ptr(from_node) }.to_str() {
        Ok(s) => s, Err(_) => return ptr::null_mut(),
    };
    let to = match unsafe { CStr::from_ptr(to_node) }.to_str() {
        Ok(s) => s, Err(_) => return ptr::null_mut(),
    };
    let db = unsafe { &(*ptr).0 };

    let txn = match db.begin_transaction() {
        Ok(t) => t, Err(_) => return ptr::null_mut(),
    };

    // BFS with parent tracking
    let mut visited: std::collections::HashMap<String, (String, String)> = std::collections::HashMap::new(); // node -> (parent, edge_json)
    let mut queue: std::collections::VecDeque<(String, usize)> = std::collections::VecDeque::new();
    queue.push_back((from.to_string(), 0));
    visited.insert(from.to_string(), ("".to_string(), "".to_string()));
    let mut found = false;

    while let Some((current, depth)) = queue.pop_front() {
        if current == to {
            found = true;
            break;
        }
        if depth >= max_depth { continue; }

        let prefix = format!("_graph/{}/edges/{}/", ns, current);
        if let Ok(edges) = db.scan(txn, prefix.as_bytes()) {
            for (_key, value) in edges {
                if let Ok(edge_str) = std::str::from_utf8(&value) {
                    if let Some(to_pos) = edge_str.find(r#""to_id":""#) {
                        let start = to_pos + r#""to_id":""#.len();
                        if let Some(end) = edge_str[start..].find('"') {
                            let next = &edge_str[start..start + end];
                            if !visited.contains_key(next) {
                                visited.insert(next.to_string(), (current.clone(), edge_str.to_string()));
                                queue.push_back((next.to_string(), depth + 1));
                            }
                        }
                    }
                }
            }
        }
    }

    let _ = db.commit(txn);

    if !found { return ptr::null_mut(); }

    // Reconstruct path
    let mut path = Vec::new();
    let mut edges = Vec::new();
    let mut current = to.to_string();
    while !current.is_empty() {
        path.push(format!(r#""{}""#, current));
        if let Some((parent, edge)) = visited.get(&current) {
            if !edge.is_empty() {
                edges.push(edge.clone());
            }
            current = parent.clone();
        } else {
            break;
        }
    }
    path.reverse();
    edges.reverse();

    let json = format!(
        r#"{{"path":[{}],"edges":[{}]}}"#,
        path.join(","),
        edges.join(",")
    );
    let c_string = match std::ffi::CString::new(json) {
        Ok(s) => s, Err(_) => return ptr::null_mut(),
    };
    unsafe { *out_len = c_string.as_bytes().len() };
    c_string.into_raw()
}

// ============================================================================
// NEW FFI: End Temporal Edge (set valid_until)
// ============================================================================

/// End a temporal edge by setting its valid_until to the current time.
/// Returns 0 on success, -1 on error, 1 if edge not found.
/// # Safety
/// All pointers must be valid C strings.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_end_temporal_edge(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    from_id: *const c_char,
    edge_type: *const c_char,
    to_id: *const c_char,
) -> c_int {
    if ptr.is_null() || namespace.is_null() || from_id.is_null()
        || edge_type.is_null() || to_id.is_null() { return -1; }
    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let from = match unsafe { CStr::from_ptr(from_id) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let etype = match unsafe { CStr::from_ptr(edge_type) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let to = match unsafe { CStr::from_ptr(to_id) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let db = unsafe { &(*ptr).0 };

    let txn = match db.begin_transaction() {
        Ok(t) => t, Err(_) => return -1,
    };

    // Scan for temporal edges matching from/type/to
    let prefix = format!("_graph/{}/temporal/{}/{}/{}/", ns, from, etype, to);
    let edges = match db.scan(txn, prefix.as_bytes()) {
        Ok(e) => e,
        Err(_) => { let _ = db.abort(txn); return -1; }
    };

    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_millis() as u64;

    let mut found = false;
    for (key, value) in edges {
        if let Ok(val_str) = std::str::from_utf8(&value) {
            // Only end edges that are currently active (valid_until == 0)
            if val_str.contains(r#""valid_until":0"#) {
                let new_val = val_str.replace(r#""valid_until":0"#, &format!(r#""valid_until":{}"#, now));
                if db.put(txn, &key, new_val.as_bytes()).is_ok() {
                    found = true;
                }
            }
        }
    }

    match db.commit(txn) {
        Ok(_) => if found { 0 } else { 1 },
        Err(_) => -1,
    }
}

// ============================================================================
// NEW FFI: Cache Management (delete, clear, stats)
// ============================================================================

/// Delete a specific entry from the semantic cache.
/// Returns 0 on success, 1 if not found, -1 on error.
/// # Safety
/// All pointers must be valid C strings.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_cache_delete(
    ptr: *mut DatabasePtr,
    cache_name: *const c_char,
    key: *const c_char,
) -> c_int {
    if ptr.is_null() || cache_name.is_null() || key.is_null() { return -1; }
    let cache = match unsafe { CStr::from_ptr(cache_name) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let k = match unsafe { CStr::from_ptr(key) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let db = unsafe { &(*ptr).0 };

    let txn = match db.begin_transaction() {
        Ok(t) => t, Err(_) => return -1,
    };

    let key_hash = format!("{:016x}", twox_hash::xxh3::hash64(k.as_bytes()));
    let cache_key = format!("_cache/{}/{}", cache, key_hash);

    match db.get(txn, cache_key.as_bytes()) {
        Ok(Some(_)) => {
            match db.delete(txn, cache_key.as_bytes()) {
                Ok(_) => match db.commit(txn) {
                    Ok(_) => 0,
                    Err(_) => -1,
                },
                Err(_) => { let _ = db.abort(txn); -1 }
            }
        }
        Ok(None) => { let _ = db.commit(txn); 1 }
        Err(_) => { let _ = db.abort(txn); -1 }
    }
}

/// Clear all entries from a semantic cache.
/// Returns number of entries deleted, or -1 on error.
/// # Safety
/// All pointers must be valid C strings.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_cache_clear(
    ptr: *mut DatabasePtr,
    cache_name: *const c_char,
) -> i64 {
    if ptr.is_null() || cache_name.is_null() { return -1; }
    let cache = match unsafe { CStr::from_ptr(cache_name) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let db = unsafe { &(*ptr).0 };

    let txn = match db.begin_transaction() {
        Ok(t) => t, Err(_) => return -1,
    };

    let prefix = format!("_cache/{}/", cache);
    let entries = match db.scan(txn, prefix.as_bytes()) {
        Ok(e) => e,
        Err(_) => { let _ = db.abort(txn); return -1; }
    };

    let mut count = 0i64;
    for (key, _) in &entries {
        if db.delete(txn, key).is_ok() {
            count += 1;
        }
    }

    match db.commit(txn) {
        Ok(_) => count,
        Err(_) => -1,
    }
}

/// Get cache statistics. Returns JSON string.
/// Caller must free with sochdb_free_string.
/// # Safety
/// All pointers must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_cache_stats(
    ptr: *mut DatabasePtr,
    cache_name: *const c_char,
    out_len: *mut usize,
) -> *mut c_char {
    if ptr.is_null() || cache_name.is_null() || out_len.is_null() { return ptr::null_mut(); }
    let cache = match unsafe { CStr::from_ptr(cache_name) }.to_str() {
        Ok(s) => s, Err(_) => return ptr::null_mut(),
    };
    let db = unsafe { &(*ptr).0 };

    let txn = match db.begin_transaction() {
        Ok(t) => t, Err(_) => return ptr::null_mut(),
    };

    let prefix = format!("_cache/{}/", cache);
    let entries = match db.scan(txn, prefix.as_bytes()) {
        Ok(e) => e,
        Err(_) => { let _ = db.abort(txn); return ptr::null_mut(); }
    };
    let _ = db.commit(txn);

    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_secs();

    let total = entries.len();
    let mut expired = 0usize;
    let mut total_bytes = 0usize;

    for (_key, value) in &entries {
        total_bytes += value.len();
        if let Ok(val_str) = std::str::from_utf8(value) {
            if let Some(exp_pos) = val_str.find(r#""expires_at":"#) {
                let exp_start = exp_pos + r#""expires_at":"#.len();
                if let Some(exp_end) = val_str[exp_start..].find('}') {
                    let expires_at: u64 = val_str[exp_start..exp_start + exp_end]
                        .parse().unwrap_or(0);
                    if expires_at > 0 && now > expires_at {
                        expired += 1;
                    }
                }
            }
        }
    }

    let json = format!(
        r#"{{"cache_name":"{}","total_entries":{},"expired_entries":{},"active_entries":{},"total_bytes":{}}}"#,
        cache, total, expired, total - expired, total_bytes
    );
    let c_string = match std::ffi::CString::new(json) {
        Ok(s) => s, Err(_) => return ptr::null_mut(),
    };
    unsafe { *out_len = c_string.as_bytes().len() };
    c_string.into_raw()
}

// ============================================================================
// NEW FFI: Collection Management (delete, count, list)
// ============================================================================

/// Delete a vector collection and all its data.
/// Returns 0 on success, -1 on error.
/// # Safety
/// All pointers must be valid C strings.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_collection_delete(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    collection: *const c_char,
) -> c_int {
    if ptr.is_null() || namespace.is_null() || collection.is_null() { return -1; }
    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let col = match unsafe { CStr::from_ptr(collection) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let db = unsafe { &(*ptr).0 };

    let txn = match db.begin_transaction() {
        Ok(t) => t, Err(_) => return -1,
    };

    // Delete config
    let config_key = format!("{}/_collections/{}", ns, col);
    let _ = db.delete(txn, config_key.as_bytes());

    // Delete all vectors
    let vec_prefix = format!("{}/collections/{}/vectors_bin/", ns, col);
    if let Ok(entries) = db.scan(txn, vec_prefix.as_bytes()) {
        for (key, _) in entries {
            let _ = db.delete(txn, &key);
        }
    }

    // Delete all metadata
    let meta_prefix = format!("{}/collections/{}/meta/", ns, col);
    if let Ok(entries) = db.scan(txn, meta_prefix.as_bytes()) {
        for (key, _) in entries {
            let _ = db.delete(txn, &key);
        }
    }

    // Remove from in-memory index registry
    let registry = COLLECTION_INDEXES.get_or_init(|| Mutex::new(HashMap::new()));
    let key = collection_key(ns, col);
    registry.lock().unwrap().remove(&key);

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

/// Count vectors in a collection.
/// Returns count on success, -1 on error.
/// # Safety
/// All pointers must be valid C strings.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_collection_count(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    collection: *const c_char,
) -> i64 {
    if ptr.is_null() || namespace.is_null() || collection.is_null() { return -1; }
    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let col = match unsafe { CStr::from_ptr(collection) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let db = unsafe { &(*ptr).0 };

    let txn = match db.begin_transaction() {
        Ok(t) => t, Err(_) => return -1,
    };

    let meta_prefix = format!("{}/collections/{}/meta/", ns, col);
    match db.scan(txn, meta_prefix.as_bytes()) {
        Ok(entries) => {
            let count = entries.len() as i64;
            let _ = db.commit(txn);
            count
        }
        Err(_) => {
            let _ = db.abort(txn);
            -1
        }
    }
}

/// List all collections in a namespace. Returns JSON array of collection names.
/// Caller must free the returned string with sochdb_free_string.
/// # Safety
/// All pointers must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_collection_list(
    ptr: *mut DatabasePtr,
    namespace: *const c_char,
    out_len: *mut usize,
) -> *mut c_char {
    if ptr.is_null() || namespace.is_null() || out_len.is_null() { return ptr::null_mut(); }
    let ns = match unsafe { CStr::from_ptr(namespace) }.to_str() {
        Ok(s) => s, Err(_) => return ptr::null_mut(),
    };
    let db = unsafe { &(*ptr).0 };

    let txn = match db.begin_transaction() {
        Ok(t) => t, Err(_) => return ptr::null_mut(),
    };

    let prefix = format!("{}/_collections/", ns);
    let entries = match db.scan(txn, prefix.as_bytes()) {
        Ok(e) => e,
        Err(_) => { let _ = db.abort(txn); return ptr::null_mut(); }
    };
    let _ = db.commit(txn);

    let names: Vec<String> = entries.iter().filter_map(|(key, _)| {
        let key_str = std::str::from_utf8(key).ok()?;
        let name = key_str.strip_prefix(&prefix)?;
        Some(format!(r#""{}""#, name))
    }).collect();

    let json = format!("[{}]", names.join(","));
    let c_string = match std::ffi::CString::new(json) {
        Ok(s) => s, Err(_) => return ptr::null_mut(),
    };
    unsafe { *out_len = c_string.as_bytes().len() };
    c_string.into_raw()
}

// ============================================================================
// NEW FFI: Schema / Table Operations
// ============================================================================

/// List all registered tables. Returns JSON array of table names.
/// Caller must free the returned string with sochdb_free_string.
/// # Safety
/// ptr must be a valid pointer returned by sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_list_tables(
    ptr: *mut DatabasePtr,
    out_len: *mut usize,
) -> *mut c_char {
    if ptr.is_null() || out_len.is_null() { return ptr::null_mut(); }
    let db = unsafe { &(*ptr).0 };
    let tables = db.list_tables();

    let names: Vec<String> = tables.iter().map(|t| format!(r#""{}""#, t)).collect();
    let json = format!("[{}]", names.join(","));
    let c_string = match std::ffi::CString::new(json) {
        Ok(s) => s, Err(_) => return ptr::null_mut(),
    };
    unsafe { *out_len = c_string.as_bytes().len() };
    c_string.into_raw()
}

/// Get a table schema as JSON. Returns null if table not found.
/// Caller must free the returned string with sochdb_free_string.
/// # Safety
/// ptr and table_name must be valid pointers.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_get_table_schema(
    ptr: *mut DatabasePtr,
    table_name: *const c_char,
    out_len: *mut usize,
) -> *mut c_char {
    if ptr.is_null() || table_name.is_null() || out_len.is_null() { return ptr::null_mut(); }
    let db = unsafe { &(*ptr).0 };
    let name = match unsafe { CStr::from_ptr(table_name) }.to_str() {
        Ok(s) => s, Err(_) => return ptr::null_mut(),
    };

    match db.get_table_schema(name) {
        Some(schema) => {
            // Serialize schema to JSON
            let columns: Vec<String> = schema.columns.iter().map(|col| {
                format!(r#"{{"name":"{}","type":"{}","nullable":{}}}"#,
                    col.name,
                    format!("{:?}", col.col_type),
                    col.nullable
                )
            }).collect();
            let json = format!(
                r#"{{"table":"{}","columns":[{}]}}"#,
                schema.name,
                columns.join(",")
            );
            let c_string = match std::ffi::CString::new(json) {
                Ok(s) => s, Err(_) => return ptr::null_mut(),
            };
            unsafe { *out_len = c_string.as_bytes().len() };
            c_string.into_raw()
        }
        None => ptr::null_mut(),
    }
}

// ============================================================================
// NEW FFI: Compression Configuration
// ============================================================================

/// Set compression type for the database.
/// compression: 0=None, 1=Lz4, 2=ZstdFast, 3=ZstdMax
/// Returns 0 on success, -1 on error.
/// # Safety
/// ptr must be a valid pointer returned by sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_set_compression(
    ptr: *mut DatabasePtr,
    compression: u8,
) -> c_int {
    if ptr.is_null() { return -1; }
    let db = unsafe { &(*ptr).0 };
    let comp_type = crate::compression::CompressionType::from_u8(compression);
    
    // Store compression preference via KV
    let txn = match db.begin_transaction() {
        Ok(t) => t, Err(_) => return -1,
    };
    let key = b"_config/compression";
    let val = format!("{}", compression);
    if db.put(txn, key, val.as_bytes()).is_err() {
        let _ = db.abort(txn);
        return -1;
    }
    match db.commit(txn) {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// Get current compression type.
/// Returns 0=None, 1=Lz4, 2=ZstdFast, 3=ZstdMax, 255 on error.
/// # Safety
/// ptr must be a valid pointer returned by sochdb_open.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_get_compression(
    ptr: *mut DatabasePtr,
) -> u8 {
    if ptr.is_null() { return 255; }
    let db = unsafe { &(*ptr).0 };
    
    let txn = match db.begin_transaction() {
        Ok(t) => t, Err(_) => return 255,
    };
    let key = b"_config/compression";
    match db.get(txn, key) {
        Ok(Some(val)) => {
            let _ = db.commit(txn);
            std::str::from_utf8(&val).ok()
                .and_then(|s| s.parse::<u8>().ok())
                .unwrap_or(0)
        }
        _ => { let _ = db.commit(txn); 0 }
    }
}

// ============================================================================
// NEW FFI: SQL Execute
// ============================================================================

/// Execute a SQL query and return results as JSON.
/// Returns JSON: {"columns": [...], "rows": [[...], ...]}
/// Caller must free the returned string with sochdb_free_string.
/// # Safety
/// All pointers must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_execute_sql(
    ptr: *mut DatabasePtr,
    handle: C_TxnHandle,
    sql_ptr: *const c_char,
    out_len: *mut usize,
) -> *mut c_char {
    if ptr.is_null() || sql_ptr.is_null() || out_len.is_null() {
        return ptr::null_mut();
    }
    let db = unsafe { &(*ptr).0 };
    let sql = match unsafe { CStr::from_ptr(sql_ptr) }.to_str() {
        Ok(s) => s,
        Err(_) => return ptr::null_mut(),
    };
    let txn = TxnHandle {
        txn_id: handle.txn_id,
        snapshot_ts: handle.snapshot_ts,
    };

    // Use the query builder's path-based interface for SQL
    // The Database has an execute method or we can use the query builder
    let result = db.query(txn, "").execute();
    
    // For now, return the scan results as JSON
    // This is a simplified SQL handler - the real implementation
    // should parse SQL and map to appropriate operations
    match result {
        Ok(qr) => {
            let toon = qr.to_toon();
            let c_string = match std::ffi::CString::new(toon) {
                Ok(s) => s,
                Err(_) => return ptr::null_mut(),
            };
            unsafe { *out_len = c_string.as_bytes().len() };
            c_string.into_raw()
        }
        Err(_) => ptr::null_mut(),
    }
}

// ============================================================================
// NEW FFI: Namespace Management
// ============================================================================

/// Create a namespace. Returns 0 on success, -1 on error.
/// # Safety
/// All pointer arguments must be valid C strings.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_namespace_create(
    ptr: *mut DatabasePtr,
    name: *const c_char,
) -> c_int {
    if ptr.is_null() || name.is_null() { return -1; }
    let ns = match unsafe { CStr::from_ptr(name) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let db = unsafe { &(*ptr).0 };

    let txn = match db.begin_transaction() {
        Ok(t) => t, Err(_) => return -1,
    };

    let key = format!("_namespaces/{}", ns);
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_secs();
    let value = format!(r#"{{"name":"{}","created_at":{}}}"#, ns, now);

    if db.put(txn, key.as_bytes(), value.as_bytes()).is_err() {
        let _ = db.abort(txn);
        return -1;
    }
    match db.commit(txn) {
        Ok(_) => 0,
        Err(_) => -1,
    }
}

/// Delete a namespace and all its data. Returns 0 on success, -1 on error.
/// # Safety
/// All pointer arguments must be valid C strings.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_namespace_delete(
    ptr: *mut DatabasePtr,
    name: *const c_char,
) -> c_int {
    if ptr.is_null() || name.is_null() { return -1; }
    let ns = match unsafe { CStr::from_ptr(name) }.to_str() {
        Ok(s) => s, Err(_) => return -1,
    };
    let db = unsafe { &(*ptr).0 };

    let txn = match db.begin_transaction() {
        Ok(t) => t, Err(_) => return -1,
    };

    // Delete namespace metadata
    let ns_key = format!("_namespaces/{}", ns);
    let _ = db.delete(txn, ns_key.as_bytes());

    // Delete all data under namespace prefix
    let ns_prefix = format!("{}/", ns);
    if let Ok(entries) = db.scan(txn, ns_prefix.as_bytes()) {
        for (key, _) in entries {
            let _ = db.delete(txn, &key);
        }
    }

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

/// List all namespaces. Returns JSON array of namespace names.
/// Caller must free the returned string with sochdb_free_string.
/// # Safety
/// All pointers must be valid.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn sochdb_namespace_list(
    ptr: *mut DatabasePtr,
    out_len: *mut usize,
) -> *mut c_char {
    if ptr.is_null() || out_len.is_null() { return ptr::null_mut(); }
    let db = unsafe { &(*ptr).0 };

    let txn = match db.begin_transaction() {
        Ok(t) => t, Err(_) => return ptr::null_mut(),
    };

    let prefix = b"_namespaces/";
    let entries = match db.scan(txn, prefix) {
        Ok(e) => e,
        Err(_) => { let _ = db.abort(txn); return ptr::null_mut(); }
    };
    let _ = db.commit(txn);

    let names: Vec<String> = entries.iter().filter_map(|(key, _)| {
        let key_str = std::str::from_utf8(key).ok()?;
        let name = key_str.strip_prefix("_namespaces/")?;
        Some(format!(r#""{}""#, name))
    }).collect();

    let json = format!("[{}]", names.join(","));
    let c_string = match std::ffi::CString::new(json) {
        Ok(s) => s, Err(_) => return ptr::null_mut(),
    };
    unsafe { *out_len = c_string.as_bytes().len() };
    c_string.into_raw()
}