//! opfs-btree-backed Storage implementation.
//!
//! Uses a single opfs-btree instance with key-encoded namespaces for all data:
//! objects, commits, ack tiers, and indices.
//!
//! Key encoding scheme (all keys are UTF-8 strings with hex-encoded binary parts):
//!
//! ```text
//! "obj:{uuid}:meta"                                       → JSON metadata
//! "obj:{uuid}:br:{branch}:tips"                           → JSON HashSet<CommitId>
//! "obj:{uuid}:br:{branch}:c:{commit_uuid}"                → JSON Commit
//! "ack:{commit_hex}"                                      → JSON HashSet<PersistenceTier>
//! "idx:{table}:{col}:{branch}:{hex_encoded_value}:{uuid}" → empty (existence is the signal)
//! ```

use std::cell::RefCell;
use std::collections::{HashMap, HashSet};
use std::ops::Bound;
#[cfg(not(target_arch = "wasm32"))]
use std::path::Path;

#[cfg(target_arch = "wasm32")]
use opfs_btree::OpfsFile;
#[cfg(not(target_arch = "wasm32"))]
use opfs_btree::StdFile;
use opfs_btree::{BTreeError, BTreeOptions, MemoryFile, OpfsBTree, SyncFile};

use crate::commit::{Commit, CommitId};
use crate::object::{BranchName, ObjectId};
use crate::query_manager::types::Value;
use crate::sync_manager::PersistenceTier;

use super::{LoadedBranch, Storage, StorageError, encode_value};

const MIN_CACHE_SIZE_BYTES: usize = 4 * 1024 * 1024;

#[derive(Clone, Debug)]
enum AnyFile {
    Memory(MemoryFile),
    #[cfg(not(target_arch = "wasm32"))]
    Std(StdFile),
    #[cfg(target_arch = "wasm32")]
    Opfs(OpfsFile),
}

impl SyncFile for AnyFile {
    fn len(&self) -> Result<u64, BTreeError> {
        match self {
            Self::Memory(file) => file.len(),
            #[cfg(not(target_arch = "wasm32"))]
            Self::Std(file) => file.len(),
            #[cfg(target_arch = "wasm32")]
            Self::Opfs(file) => file.len(),
        }
    }

    fn read_exact_at(&self, offset: u64, buf: &mut [u8]) -> Result<(), BTreeError> {
        match self {
            Self::Memory(file) => file.read_exact_at(offset, buf),
            #[cfg(not(target_arch = "wasm32"))]
            Self::Std(file) => file.read_exact_at(offset, buf),
            #[cfg(target_arch = "wasm32")]
            Self::Opfs(file) => file.read_exact_at(offset, buf),
        }
    }

    fn write_all_at(&self, offset: u64, buf: &[u8]) -> Result<(), BTreeError> {
        match self {
            Self::Memory(file) => file.write_all_at(offset, buf),
            #[cfg(not(target_arch = "wasm32"))]
            Self::Std(file) => file.write_all_at(offset, buf),
            #[cfg(target_arch = "wasm32")]
            Self::Opfs(file) => file.write_all_at(offset, buf),
        }
    }

    fn truncate(&self, len: u64) -> Result<(), BTreeError> {
        match self {
            Self::Memory(file) => file.truncate(len),
            #[cfg(not(target_arch = "wasm32"))]
            Self::Std(file) => file.truncate(len),
            #[cfg(target_arch = "wasm32")]
            Self::Opfs(file) => file.truncate(len),
        }
    }

    fn flush(&self) -> Result<(), BTreeError> {
        match self {
            Self::Memory(file) => file.flush(),
            #[cfg(not(target_arch = "wasm32"))]
            Self::Std(file) => file.flush(),
            #[cfg(target_arch = "wasm32")]
            Self::Opfs(file) => file.flush(),
        }
    }
}

pub struct OpfsBTreeStorage {
    tree: RefCell<OpfsBTree<AnyFile>>,
}

impl OpfsBTreeStorage {
    #[cfg(not(target_arch = "wasm32"))]
    pub fn open(path: impl AsRef<Path>, cache_size_bytes: usize) -> Result<Self, StorageError> {
        let file = StdFile::open(path).map_err(map_storage_err)?;
        Self::open_with_file(AnyFile::Std(file), cache_size_bytes)
    }

    pub fn memory(cache_size_bytes: usize) -> Result<Self, StorageError> {
        Self::open_with_file(AnyFile::Memory(MemoryFile::new()), cache_size_bytes)
    }

    #[cfg(target_arch = "wasm32")]
    pub fn with_opfs(file: OpfsFile, cache_size_bytes: usize) -> Result<Self, StorageError> {
        Self::open_with_file(AnyFile::Opfs(file), cache_size_bytes)
    }

    #[cfg(target_arch = "wasm32")]
    pub async fn open_opfs(namespace: &str, cache_size_bytes: usize) -> Result<Self, StorageError> {
        let file = OpfsFile::open(namespace).await.map_err(map_storage_err)?;
        Self::with_opfs(file, cache_size_bytes)
    }

    #[cfg(target_arch = "wasm32")]
    pub async fn destroy_opfs(namespace: &str) -> Result<(), StorageError> {
        OpfsFile::destroy(namespace).await.map_err(map_storage_err)
    }

    fn open_with_file(file: AnyFile, cache_size_bytes: usize) -> Result<Self, StorageError> {
        let options = Self::options(cache_size_bytes);
        let tree = OpfsBTree::open(file, options).map_err(map_storage_err)?;
        let storage = Self {
            tree: RefCell::new(tree),
        };
        storage.log_key_stats();
        Ok(storage)
    }

    fn options(cache_size_bytes: usize) -> BTreeOptions {
        BTreeOptions {
            cache_bytes: cache_size_bytes.max(MIN_CACHE_SIZE_BYTES),
            pin_internal_pages: true,
            read_coalesce_pages: 4,
            ..Default::default()
        }
    }

    fn with_tree_mut<R>(
        &self,
        f: impl FnOnce(&mut OpfsBTree<AnyFile>) -> Result<R, StorageError>,
    ) -> Result<R, StorageError> {
        let mut tree = self
            .tree
            .try_borrow_mut()
            .map_err(|_| StorageError::IoError("opfs-btree already borrowed".to_string()))?;
        f(&mut tree)
    }

    fn log_key_stats(&self) {
        let count_prefix =
            |pfx: &str| -> usize { self.tree_scan_keys(pfx).map(|v| v.len()).unwrap_or(0) };
        let obj_count = count_prefix("obj:");
        let idx_count = count_prefix("idx:");
        let ack_count = count_prefix("ack:");
        tracing::info!(obj_count, idx_count, ack_count, "OpfsBTreeStorage opened");
    }

    fn obj_meta_key(id: ObjectId) -> String {
        format!("obj:{}:meta", format_uuid(id))
    }

    fn branch_tips_key(object_id: ObjectId, branch: &BranchName) -> String {
        format!("obj:{}:br:{}:tips", format_uuid(object_id), branch)
    }

    fn commit_key(object_id: ObjectId, branch: &BranchName, commit_id: CommitId) -> String {
        format!(
            "obj:{}:br:{}:c:{}",
            format_uuid(object_id),
            branch,
            hex::encode(commit_id.0)
        )
    }

    fn commit_prefix(object_id: ObjectId, branch: &BranchName) -> String {
        format!("obj:{}:br:{}:c:", format_uuid(object_id), branch)
    }

    fn ack_key(commit_id: CommitId) -> String {
        format!("ack:{}", hex::encode(commit_id.0))
    }

    fn index_entry_key(
        table: &str,
        column: &str,
        branch: &str,
        value: &Value,
        row_id: ObjectId,
    ) -> String {
        format!(
            "idx:{}:{}:{}:{}:{}",
            table,
            column,
            branch,
            hex::encode(encode_value(value)),
            format_uuid(row_id)
        )
    }

    fn index_value_prefix(table: &str, column: &str, branch: &str, value: &Value) -> String {
        format!(
            "idx:{}:{}:{}:{}:",
            table,
            column,
            branch,
            hex::encode(encode_value(value))
        )
    }

    fn index_prefix(table: &str, column: &str, branch: &str) -> String {
        format!("idx:{}:{}:{}:", table, column, branch)
    }

    fn tree_insert(&self, key: &str, value: &[u8]) -> Result<(), StorageError> {
        self.with_tree_mut(|tree| tree.put(key.as_bytes(), value).map_err(map_storage_err))
    }

    fn tree_read(&self, key: &str) -> Result<Option<Vec<u8>>, StorageError> {
        self.with_tree_mut(|tree| tree.get(key.as_bytes()).map_err(map_storage_err))
    }

    fn tree_delete(&self, key: &str) -> Result<(), StorageError> {
        self.with_tree_mut(|tree| tree.delete(key.as_bytes()).map_err(map_storage_err))
    }

    fn tree_scan_prefix(&self, prefix: &str) -> Result<Vec<(String, Vec<u8>)>, StorageError> {
        let start = prefix.as_bytes();
        let mut end = start.to_vec();
        increment_bytes(&mut end);
        self.tree_scan_range_bytes(start, &end)
    }

    fn tree_scan_keys(&self, prefix: &str) -> Result<Vec<String>, StorageError> {
        Ok(self
            .tree_scan_prefix(prefix)?
            .into_iter()
            .map(|(key, _)| key)
            .collect())
    }

    fn tree_scan_key_range(&self, start: &str, end: &str) -> Result<Vec<String>, StorageError> {
        Ok(self
            .tree_scan_range_bytes(start.as_bytes(), end.as_bytes())?
            .into_iter()
            .map(|(key, _)| key)
            .collect())
    }

    fn tree_scan_range_bytes(
        &self,
        start: &[u8],
        end: &[u8],
    ) -> Result<Vec<(String, Vec<u8>)>, StorageError> {
        if start >= end {
            return Ok(Vec::new());
        }

        self.with_tree_mut(|tree| {
            let entries = tree
                .range(start, end, usize::MAX)
                .map_err(map_storage_err)?;

            entries
                .into_iter()
                .map(|(key, value)| {
                    let key = String::from_utf8(key)
                        .map_err(|e| StorageError::IoError(format!("invalid key utf8: {}", e)))?;
                    Ok((key, value))
                })
                .collect()
        })
    }
}

impl Storage for OpfsBTreeStorage {
    fn create_object(
        &mut self,
        id: ObjectId,
        metadata: HashMap<String, String>,
    ) -> Result<(), StorageError> {
        let key = Self::obj_meta_key(id);
        let json = serde_json::to_vec(&metadata)
            .map_err(|e| StorageError::IoError(format!("serialize metadata: {}", e)))?;
        self.tree_insert(&key, &json)
    }

    fn load_object_metadata(
        &self,
        id: ObjectId,
    ) -> Result<Option<HashMap<String, String>>, StorageError> {
        let key = Self::obj_meta_key(id);
        match self.tree_read(&key)? {
            Some(data) => {
                let meta: HashMap<String, String> = serde_json::from_slice(&data)
                    .map_err(|e| StorageError::IoError(format!("deserialize metadata: {}", e)))?;
                Ok(Some(meta))
            }
            None => Ok(None),
        }
    }

    fn load_branch(
        &self,
        object_id: ObjectId,
        branch: &BranchName,
    ) -> Result<Option<LoadedBranch>, StorageError> {
        let meta_key = Self::obj_meta_key(object_id);
        if self.tree_read(&meta_key)?.is_none() {
            return Ok(None);
        }

        let commit_prefix = Self::commit_prefix(object_id, branch);
        let commit_entries = self.tree_scan_prefix(&commit_prefix)?;

        if commit_entries.is_empty() {
            let tips_key = Self::branch_tips_key(object_id, branch);
            if self.tree_read(&tips_key)?.is_none() {
                return Ok(None);
            }
        }

        let mut commits = Vec::new();
        for (_key, data) in &commit_entries {
            let mut commit: Commit = serde_json::from_slice(data)
                .map_err(|e| StorageError::IoError(format!("deserialize commit: {}", e)))?;

            let ack_key = Self::ack_key(commit.id());
            if let Some(ack_data) = self.tree_read(&ack_key)? {
                let tiers: HashSet<PersistenceTier> = serde_json::from_slice(&ack_data)
                    .map_err(|e| StorageError::IoError(format!("deserialize ack: {}", e)))?;
                commit.ack_state.confirmed_tiers = tiers;
            }

            commits.push(commit);
        }

        let tips_key = Self::branch_tips_key(object_id, branch);
        let tails = match self.tree_read(&tips_key)? {
            Some(data) => serde_json::from_slice(&data)
                .map_err(|e| StorageError::IoError(format!("deserialize tips: {}", e)))?,
            None => HashSet::new(),
        };

        Ok(Some(LoadedBranch { commits, tails }))
    }

    fn append_commit(
        &mut self,
        object_id: ObjectId,
        branch: &BranchName,
        commit: Commit,
    ) -> Result<(), StorageError> {
        let commit_id = commit.id();

        let commit_key = Self::commit_key(object_id, branch, commit_id);
        let commit_json = serde_json::to_vec(&commit)
            .map_err(|e| StorageError::IoError(format!("serialize commit: {}", e)))?;
        self.tree_insert(&commit_key, &commit_json)?;

        let tips_key = Self::branch_tips_key(object_id, branch);
        let mut tips: HashSet<CommitId> = match self.tree_read(&tips_key)? {
            Some(data) => serde_json::from_slice(&data)
                .map_err(|e| StorageError::IoError(format!("deserialize tips: {}", e)))?,
            None => HashSet::new(),
        };

        for parent in &commit.parents {
            tips.remove(parent);
        }
        tips.insert(commit_id);

        let tips_json = serde_json::to_vec(&tips)
            .map_err(|e| StorageError::IoError(format!("serialize tips: {}", e)))?;
        self.tree_insert(&tips_key, &tips_json)?;

        Ok(())
    }

    fn delete_commit(
        &mut self,
        object_id: ObjectId,
        branch: &BranchName,
        commit_id: CommitId,
    ) -> Result<(), StorageError> {
        let commit_key = Self::commit_key(object_id, branch, commit_id);
        self.tree_delete(&commit_key)?;

        let tips_key = Self::branch_tips_key(object_id, branch);
        if let Some(data) = self.tree_read(&tips_key)? {
            let mut tips: HashSet<CommitId> = serde_json::from_slice(&data)
                .map_err(|e| StorageError::IoError(format!("deserialize tips: {}", e)))?;
            tips.remove(&commit_id);
            let tips_json = serde_json::to_vec(&tips)
                .map_err(|e| StorageError::IoError(format!("serialize tips: {}", e)))?;
            self.tree_insert(&tips_key, &tips_json)?;
        }

        Ok(())
    }

    fn set_branch_tails(
        &mut self,
        object_id: ObjectId,
        branch: &BranchName,
        tails: Option<HashSet<CommitId>>,
    ) -> Result<(), StorageError> {
        let tips_key = Self::branch_tips_key(object_id, branch);
        match tails {
            Some(t) => {
                let json = serde_json::to_vec(&t)
                    .map_err(|e| StorageError::IoError(format!("serialize tails: {}", e)))?;
                self.tree_insert(&tips_key, &json)?;
            }
            None => {
                self.tree_delete(&tips_key)?;
            }
        }
        Ok(())
    }

    fn store_ack_tier(
        &mut self,
        commit_id: CommitId,
        tier: PersistenceTier,
    ) -> Result<(), StorageError> {
        let key = Self::ack_key(commit_id);
        let mut tiers: HashSet<PersistenceTier> = match self.tree_read(&key)? {
            Some(data) => serde_json::from_slice(&data)
                .map_err(|e| StorageError::IoError(format!("deserialize ack: {}", e)))?,
            None => HashSet::new(),
        };
        tiers.insert(tier);
        let json = serde_json::to_vec(&tiers)
            .map_err(|e| StorageError::IoError(format!("serialize ack: {}", e)))?;
        self.tree_insert(&key, &json)
    }

    fn index_insert(
        &mut self,
        table: &str,
        column: &str,
        branch: &str,
        value: &Value,
        row_id: ObjectId,
    ) -> Result<(), StorageError> {
        let key = Self::index_entry_key(table, column, branch, value, row_id);
        self.tree_insert(&key, &[0x01])
    }

    fn index_remove(
        &mut self,
        table: &str,
        column: &str,
        branch: &str,
        value: &Value,
        row_id: ObjectId,
    ) -> Result<(), StorageError> {
        let key = Self::index_entry_key(table, column, branch, value, row_id);
        self.tree_delete(&key)
    }

    fn index_lookup(
        &self,
        table: &str,
        column: &str,
        branch: &str,
        value: &Value,
    ) -> Vec<ObjectId> {
        let prefix = Self::index_value_prefix(table, column, branch, value);
        match self.tree_scan_keys(&prefix) {
            Ok(keys) => keys
                .iter()
                .filter_map(|k| parse_uuid_from_index_key(k))
                .collect(),
            Err(_) => Vec::new(),
        }
    }

    fn index_range(
        &self,
        table: &str,
        column: &str,
        branch: &str,
        start: Bound<&Value>,
        end: Bound<&Value>,
    ) -> Vec<ObjectId> {
        let base_prefix = Self::index_prefix(table, column, branch);

        let start_key = match start {
            Bound::Included(v) => {
                format!("{}{}", base_prefix, hex::encode(encode_value(v)))
            }
            Bound::Excluded(v) => {
                let encoded = hex::encode(encode_value(v));
                let mut key = format!("{}{}:", base_prefix, encoded);
                increment_string(&mut key);
                key
            }
            Bound::Unbounded => base_prefix.clone(),
        };

        let end_key = match end {
            Bound::Included(v) => {
                let encoded = hex::encode(encode_value(v));
                let mut key = format!("{}{}:", base_prefix, encoded);
                increment_string(&mut key);
                key
            }
            Bound::Excluded(v) => {
                format!("{}{}", base_prefix, hex::encode(encode_value(v)))
            }
            Bound::Unbounded => {
                let mut end = base_prefix.clone();
                increment_string(&mut end);
                end
            }
        };

        if start_key >= end_key {
            return Vec::new();
        }

        match self.tree_scan_key_range(&start_key, &end_key) {
            Ok(keys) => keys
                .iter()
                .filter_map(|k| parse_uuid_from_index_key(k))
                .collect(),
            Err(_) => Vec::new(),
        }
    }

    fn index_scan_all(&self, table: &str, column: &str, branch: &str) -> Vec<ObjectId> {
        let prefix = Self::index_prefix(table, column, branch);
        match self.tree_scan_keys(&prefix) {
            Ok(keys) => keys
                .iter()
                .filter_map(|k| parse_uuid_from_index_key(k))
                .collect(),
            Err(_) => Vec::new(),
        }
    }

    fn flush(&self) {
        if let Err(error) = self.with_tree_mut(|tree| tree.checkpoint().map_err(map_storage_err)) {
            tracing::warn!(?error, "OpfsBTreeStorage flush failed");
        }
    }

    fn flush_wal(&self) {
        // opfs-btree has no separate WAL; flush_wal maps to an incremental checkpoint.
        self.flush();
    }
}

fn map_storage_err(error: BTreeError) -> StorageError {
    StorageError::IoError(format!("opfs-btree: {}", error))
}

fn format_uuid(id: ObjectId) -> String {
    hex::encode(id.uuid().as_bytes())
}

fn parse_uuid_from_index_key(key: &str) -> Option<ObjectId> {
    let uuid_hex = key.rsplit(':').next()?;
    let bytes = hex::decode(uuid_hex).ok()?;
    if bytes.len() != 16 {
        return None;
    }
    let uuid = uuid::Uuid::from_bytes(bytes.try_into().ok()?);
    Some(ObjectId(internment::Intern::new(uuid)))
}

fn increment_bytes(bytes: &mut Vec<u8>) {
    for i in (0..bytes.len()).rev() {
        if bytes[i] < 0xFF {
            bytes[i] += 1;
            bytes.truncate(i + 1);
            return;
        }
    }
    bytes.push(0x00);
}

fn increment_string(s: &mut String) {
    let mut bytes = std::mem::take(s).into_bytes();
    increment_bytes(&mut bytes);
    *s = String::from_utf8(bytes).unwrap_or_default();
}

#[cfg(test)]
mod tests {
    use super::*;
    use smallvec::smallvec;

    fn make_commit(content: &[u8]) -> Commit {
        Commit {
            parents: smallvec![],
            content: content.to_vec(),
            timestamp: 12345,
            author: ObjectId::new(),
            metadata: None,
            stored_state: Default::default(),
            ack_state: Default::default(),
        }
    }

    fn test_storage() -> OpfsBTreeStorage {
        OpfsBTreeStorage::memory(4 * 1024 * 1024).unwrap()
    }

    #[test]
    fn opfs_btree_object_roundtrip() {
        let mut storage = test_storage();

        let id = ObjectId::new();
        let mut metadata = HashMap::new();
        metadata.insert(
            crate::metadata::MetadataKey::Table.to_string(),
            "users".to_string(),
        );
        metadata.insert("app".to_string(), "test".to_string());

        storage.create_object(id, metadata.clone()).unwrap();

        let loaded = storage.load_object_metadata(id).unwrap();
        assert_eq!(loaded, Some(metadata));

        let other = ObjectId::new();
        assert_eq!(storage.load_object_metadata(other).unwrap(), None);
    }

    #[test]
    fn opfs_btree_commit_roundtrip() {
        let mut storage = test_storage();

        let id = ObjectId::new();
        let branch = BranchName::new("main");
        storage.create_object(id, HashMap::new()).unwrap();

        assert_eq!(storage.load_branch(id, &branch).unwrap(), None);

        let commit = make_commit(b"first");
        let commit_id = commit.id();
        storage.append_commit(id, &branch, commit).unwrap();

        let loaded = storage.load_branch(id, &branch).unwrap().unwrap();
        assert_eq!(loaded.commits.len(), 1);
        assert!(loaded.tails.contains(&commit_id));
        assert_eq!(loaded.commits[0].content, b"first");

        let mut commit2 = make_commit(b"second");
        commit2.parents = smallvec![commit_id];
        let commit2_id = commit2.id();
        storage.append_commit(id, &branch, commit2).unwrap();

        let loaded = storage.load_branch(id, &branch).unwrap().unwrap();
        assert_eq!(loaded.commits.len(), 2);
        assert!(!loaded.tails.contains(&commit_id));
        assert!(loaded.tails.contains(&commit2_id));

        storage.delete_commit(id, &branch, commit_id).unwrap();
        let loaded = storage.load_branch(id, &branch).unwrap().unwrap();
        assert_eq!(loaded.commits.len(), 1);
        assert_eq!(loaded.commits[0].content, b"second");
    }

    #[test]
    fn opfs_btree_index_ops() {
        let mut storage = test_storage();

        let row1 = ObjectId::new();
        let row2 = ObjectId::new();
        let row3 = ObjectId::new();
        let row4 = ObjectId::new();

        storage
            .index_insert("users", "age", "main", &Value::Integer(20), row1)
            .unwrap();
        storage
            .index_insert("users", "age", "main", &Value::Integer(25), row2)
            .unwrap();
        storage
            .index_insert("users", "age", "main", &Value::Integer(25), row3)
            .unwrap();
        storage
            .index_insert("users", "age", "main", &Value::Integer(30), row4)
            .unwrap();

        let results = storage.index_lookup("users", "age", "main", &Value::Integer(25));
        assert_eq!(results.len(), 2);
        assert!(results.contains(&row2));
        assert!(results.contains(&row3));

        let results = storage.index_lookup("users", "age", "main", &Value::Integer(99));
        assert!(results.is_empty());

        let results = storage.index_range(
            "users",
            "age",
            "main",
            Bound::Included(&Value::Integer(25)),
            Bound::Excluded(&Value::Integer(30)),
        );
        assert_eq!(results.len(), 2);
        assert!(results.contains(&row2));
        assert!(results.contains(&row3));

        let results = storage.index_range(
            "users",
            "age",
            "main",
            Bound::Unbounded,
            Bound::Excluded(&Value::Integer(26)),
        );
        assert_eq!(results.len(), 3);
        assert!(results.contains(&row1));
        assert!(results.contains(&row2));
        assert!(results.contains(&row3));

        let results = storage.index_range(
            "users",
            "age",
            "main",
            Bound::Included(&Value::Integer(30)),
            Bound::Unbounded,
        );
        assert_eq!(results.len(), 1);
        assert!(results.contains(&row4));

        let results = storage.index_scan_all("users", "age", "main");
        assert_eq!(results.len(), 4);

        storage
            .index_remove("users", "age", "main", &Value::Integer(25), row2)
            .unwrap();
        let results = storage.index_lookup("users", "age", "main", &Value::Integer(25));
        assert_eq!(results.len(), 1);
        assert!(results.contains(&row3));
    }

    #[test]
    fn opfs_btree_index_branch_isolation() {
        let mut storage = test_storage();

        let row1 = ObjectId::new();
        let row2 = ObjectId::new();

        storage
            .index_insert("users", "age", "main", &Value::Integer(25), row1)
            .unwrap();
        storage
            .index_insert("users", "age", "feature", &Value::Integer(25), row2)
            .unwrap();

        let main_results = storage.index_lookup("users", "age", "main", &Value::Integer(25));
        assert_eq!(main_results.len(), 1);
        assert!(main_results.contains(&row1));

        let feature_results = storage.index_lookup("users", "age", "feature", &Value::Integer(25));
        assert_eq!(feature_results.len(), 1);
        assert!(feature_results.contains(&row2));
    }

    #[test]
    fn opfs_btree_ack_tier_roundtrip() {
        let mut storage = test_storage();

        let commit_id = CommitId([99u8; 32]);

        storage
            .store_ack_tier(commit_id, PersistenceTier::Worker)
            .unwrap();
        storage
            .store_ack_tier(commit_id, PersistenceTier::EdgeServer)
            .unwrap();

        let ack_key = OpfsBTreeStorage::ack_key(commit_id);
        let data = storage.tree_read(&ack_key).unwrap().unwrap();
        let tiers: HashSet<PersistenceTier> = serde_json::from_slice(&data).unwrap();
        assert!(tiers.contains(&PersistenceTier::Worker));
        assert!(tiers.contains(&PersistenceTier::EdgeServer));
    }

    #[test]
    fn opfs_btree_persistence() {
        let temp_dir = tempfile::TempDir::new().unwrap();
        let db_path = temp_dir.path().join("test.opfsbtree");

        let id = ObjectId::new();
        let mut metadata = HashMap::new();
        metadata.insert(
            crate::metadata::MetadataKey::Table.to_string(),
            "users".to_string(),
        );

        let commit_content = b"persistent data";
        let branch = BranchName::new("main");

        {
            let mut storage = OpfsBTreeStorage::open(&db_path, 4 * 1024 * 1024).unwrap();
            storage.create_object(id, metadata.clone()).unwrap();

            let commit = make_commit(commit_content);
            storage.append_commit(id, &branch, commit).unwrap();

            storage
                .index_insert(
                    "users",
                    "name",
                    "main",
                    &Value::Text("Alice".to_string()),
                    id,
                )
                .unwrap();

            storage.flush();
        }

        {
            let storage = OpfsBTreeStorage::open(&db_path, 4 * 1024 * 1024).unwrap();

            let loaded_meta = storage.load_object_metadata(id).unwrap();
            assert_eq!(loaded_meta, Some(metadata));

            let loaded_branch = storage.load_branch(id, &branch).unwrap().unwrap();
            assert_eq!(loaded_branch.commits.len(), 1);
            assert_eq!(loaded_branch.commits[0].content, commit_content);

            let results =
                storage.index_lookup("users", "name", "main", &Value::Text("Alice".to_string()));
            assert_eq!(results.len(), 1);
            assert!(results.contains(&id));
        }
    }
}
