reflow_assets 0.2.1

//! File-based asset database for Reflow workflows.
//!
//! Inspired by CastleDB — a structured, queryable asset store that persists
//! across actor invocations and workflow runs. Binary data is content-addressed
//! for automatic deduplication.
//!
//! ## Storage backends
//!
//! - **FileBackend** (native) — directory with manifest.json + blobs/
//! - **MemoryBackend** (wasm/testing) — in-memory HashMap
//! - **S3Backend** (remote workflows) — to be implemented
//!
//! ## Directory structure (FileBackend)
//!
//! ```text
//! assets.db/
//!   manifest.json       ← asset index
//!   blobs/
//!     ab/abcdef1234...  ← binary data keyed by content hash
//! ```

pub mod graph_projection;
pub mod layout;

use anyhow::Result;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::collections::HashMap;
use std::sync::{Arc, Mutex, RwLock};

// ═══════════════════════════════════════════════════════════════════════════
// Types
// ═══════════════════════════════════════════════════════════════════════════

pub type AssetId = String;

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
#[serde(rename_all = "lowercase")]
pub enum AssetType {
    Mesh,
    Texture,
    Skeleton,
    Animation,
    SkinWeights,
    Material,
    Scene,
    Audio,
    Video,
    Generic,
}

impl AssetType {
    pub fn parse(s: &str) -> Self {
        match s.to_lowercase().as_str() {
            "mesh" => Self::Mesh,
            "texture" | "image" => Self::Texture,
            "skeleton" => Self::Skeleton,
            "animation" | "clip" => Self::Animation,
            "skinweights" | "skin_weights" | "skin" => Self::SkinWeights,
            "material" => Self::Material,
            "scene" => Self::Scene,
            "audio" => Self::Audio,
            "video" => Self::Video,
            _ => Self::Generic,
        }
    }

    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Mesh => "mesh",
            Self::Texture => "texture",
            Self::Skeleton => "skeleton",
            Self::Animation => "animation",
            Self::SkinWeights => "skinweights",
            Self::Material => "material",
            Self::Scene => "scene",
            Self::Audio => "audio",
            Self::Video => "video",
            Self::Generic => "generic",
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AssetEntry {
    pub id: AssetId,
    pub name: String,
    #[serde(rename = "assetType")]
    pub asset_type: AssetType,
    #[serde(rename = "blobHash")]
    pub blob_hash: String,
    #[serde(rename = "blobSize")]
    pub blob_size: u64,
    #[serde(default)]
    pub metadata: Value,
    #[serde(default)]
    pub tags: Vec<String>,
    #[serde(rename = "createdAt")]
    pub created_at: String,
    #[serde(rename = "updatedAt")]
    pub updated_at: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub inline_data: Option<Value>,
    /// True if the blob is LZ4-compressed. Decompression is transparent on read.
    #[serde(default)]
    pub compressed: bool,
    /// Original uncompressed size (for pre-allocating decompress buffer).
    #[serde(default, rename = "rawSize", skip_serializing_if = "is_zero")]
    pub raw_size: u64,
}

fn is_zero(v: &u64) -> bool {
    *v == 0
}

pub struct Asset {
    pub entry: AssetEntry,
    pub data: Vec<u8>,
}

// ═══════════════════════════════════════════════════════════════════════════
// Storage backend trait
// ═══════════════════════════════════════════════════════════════════════════

/// Pluggable storage backend for AssetDB.
/// Implementations: FileBackend (native), MemoryBackend (wasm), S3Backend (remote).
pub trait StorageBackend: Send + Sync {
    fn read_manifest(&self) -> Result<Vec<AssetEntry>>;
    fn write_manifest(&self, entries: &[AssetEntry]) -> Result<()>;
    fn read_blob(&self, hash: &str) -> Result<Vec<u8>>;
    fn write_blob(&self, hash: &str, data: &[u8]) -> Result<()>;
    fn blob_exists(&self, hash: &str) -> bool;
    fn delete_blob(&self, hash: &str) -> Result<()>;
    fn backend_name(&self) -> &'static str;
}

// ═══════════════════════════════════════════════════════════════════════════
// FileBackend (native)
// ═══════════════════════════════════════════════════════════════════════════

#[cfg(not(target_arch = "wasm32"))]
pub struct FileBackend {
    root: std::path::PathBuf,
}

#[cfg(not(target_arch = "wasm32"))]
impl FileBackend {
    pub fn new(root: impl AsRef<std::path::Path>) -> Result<Self> {
        let root = root.as_ref().to_path_buf();
        std::fs::create_dir_all(root.join("blobs"))?;

        // Ensure manifest exists
        let manifest_path = root.join("manifest.json");
        if !manifest_path.exists() {
            std::fs::write(&manifest_path, "[]")?;
        }
        Ok(Self { root })
    }

    fn blob_path(&self, hash: &str) -> std::path::PathBuf {
        let prefix = &hash[..2.min(hash.len())];
        self.root.join("blobs").join(prefix).join(hash)
    }
}

#[cfg(not(target_arch = "wasm32"))]
impl StorageBackend for FileBackend {
    fn read_manifest(&self) -> Result<Vec<AssetEntry>> {
        let path = self.root.join("manifest.json");
        let data = std::fs::read_to_string(&path)?;
        Ok(serde_json::from_str(&data).unwrap_or_default())
    }

    fn write_manifest(&self, entries: &[AssetEntry]) -> Result<()> {
        let data = serde_json::to_string_pretty(entries)?;
        let tmp = self.root.join(".manifest.tmp");
        std::fs::write(&tmp, &data)?;
        std::fs::rename(&tmp, self.root.join("manifest.json"))?;
        Ok(())
    }

    fn read_blob(&self, hash: &str) -> Result<Vec<u8>> {
        Ok(std::fs::read(self.blob_path(hash))?)
    }

    fn write_blob(&self, hash: &str, data: &[u8]) -> Result<()> {
        let path = self.blob_path(hash);
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        std::fs::write(&path, data)?;
        Ok(())
    }

    fn blob_exists(&self, hash: &str) -> bool {
        self.blob_path(hash).exists()
    }

    fn delete_blob(&self, hash: &str) -> Result<()> {
        let _ = std::fs::remove_file(self.blob_path(hash));
        Ok(())
    }

    fn backend_name(&self) -> &'static str {
        "file"
    }
}

// ═══════════════════════════════════════════════════════════════════════════
// MemoryBackend (wasm + testing)
// ═══════════════════════════════════════════════════════════════════════════

pub struct MemoryBackend {
    manifest: RwLock<Vec<AssetEntry>>,
    blobs: RwLock<HashMap<String, Vec<u8>>>,
}

impl Default for MemoryBackend {
    fn default() -> Self {
        Self::new()
    }
}

impl MemoryBackend {
    pub fn new() -> Self {
        Self {
            manifest: RwLock::new(Vec::new()),
            blobs: RwLock::new(HashMap::new()),
        }
    }
}

impl StorageBackend for MemoryBackend {
    fn read_manifest(&self) -> Result<Vec<AssetEntry>> {
        Ok(self
            .manifest
            .read()
            .map_err(|e| anyhow::anyhow!("{}", e))?
            .clone())
    }

    fn write_manifest(&self, entries: &[AssetEntry]) -> Result<()> {
        let mut m = self
            .manifest
            .write()
            .map_err(|e| anyhow::anyhow!("{}", e))?;
        *m = entries.to_vec();
        Ok(())
    }

    fn read_blob(&self, hash: &str) -> Result<Vec<u8>> {
        self.blobs
            .read()
            .map_err(|e| anyhow::anyhow!("{}", e))?
            .get(hash)
            .cloned()
            .ok_or_else(|| anyhow::anyhow!("Blob not found: {}", hash))
    }

    fn write_blob(&self, hash: &str, data: &[u8]) -> Result<()> {
        self.blobs
            .write()
            .map_err(|e| anyhow::anyhow!("{}", e))?
            .insert(hash.to_string(), data.to_vec());
        Ok(())
    }

    fn blob_exists(&self, hash: &str) -> bool {
        self.blobs
            .read()
            .map(|b| b.contains_key(hash))
            .unwrap_or(false)
    }

    fn delete_blob(&self, hash: &str) -> Result<()> {
        self.blobs
            .write()
            .map_err(|e| anyhow::anyhow!("{}", e))?
            .remove(hash);
        Ok(())
    }

    fn backend_name(&self) -> &'static str {
        "memory"
    }
}

// ═══════════════════════════════════════════════════════════════════════════
// IndexedDbBackend (wasm — persistent browser storage)
// ═══════════════════════════════════════════════════════════════════════════

/// Wrap an IDB request as a Promise-backed `JsFuture`.
///
/// `wasm-bindgen-futures` only converts `Promise` → `JsFuture` directly.
/// IDB requests use `onsuccess`/`onerror` instead of being thenable, so
/// we have to bridge them through a manually constructed Promise.
#[cfg(target_arch = "wasm32")]
async fn idb_request_to_future(
    req: web_sys::IdbRequest,
) -> std::result::Result<wasm_bindgen::JsValue, wasm_bindgen::JsValue> {
    use wasm_bindgen::prelude::*;
    use wasm_bindgen_futures::JsFuture;

    let promise = js_sys::Promise::new(&mut |resolve, reject| {
        let success_target = req.clone();
        let success = Closure::once(Box::new(move |_event: web_sys::Event| {
            let value = success_target.result().unwrap_or(JsValue::NULL);
            let _ = resolve.call1(&JsValue::NULL, &value);
        }) as Box<dyn FnOnce(web_sys::Event)>);

        // We don't surface the IDB error object — wasm-bindgen-futures
        // signals failure via the rejected JsValue, callers only treat it
        // as Result<_, JsValue>::Err. A NULL placeholder is enough.
        let error = Closure::once(Box::new(move |_event: web_sys::Event| {
            let _ = reject.call1(&JsValue::NULL, &JsValue::NULL);
        }) as Box<dyn FnOnce(web_sys::Event)>);

        req.set_onsuccess(Some(success.as_ref().unchecked_ref()));
        req.set_onerror(Some(error.as_ref().unchecked_ref()));
        success.forget();
        error.forget();
    });

    JsFuture::from(promise).await
}

/// IndexedDB-backed storage for wasm. Uses in-memory cache for sync trait
/// methods and flushes writes to IndexedDB asynchronously.
///
/// Two object stores: "manifest" (single entry, key "entries") and "blobs" (keyed by hash).
///
/// Call `IndexedDbBackend::open(name).await` to create + hydrate from IDB.
#[cfg(target_arch = "wasm32")]
pub struct IndexedDbBackend {
    memory: MemoryBackend,
    db_name: String,
}

#[cfg(target_arch = "wasm32")]
impl IndexedDbBackend {
    /// Open (or create) an IndexedDB database and hydrate the in-memory cache.
    pub async fn open(name: &str) -> Result<Self> {
        use js_sys::{Array, Uint8Array};
        use wasm_bindgen::prelude::*;
        use wasm_bindgen_futures::JsFuture;
        use web_sys::{IdbObjectStoreParameters, IdbTransactionMode};

        let window = web_sys::window().ok_or_else(|| anyhow::anyhow!("No window"))?;
        let idb_factory = window
            .indexed_db()
            .map_err(|_| anyhow::anyhow!("IndexedDB not available"))?
            .ok_or_else(|| anyhow::anyhow!("IndexedDB not available"))?;

        // Open DB (version 1)
        let open_req = idb_factory
            .open_with_u32(name, 1)
            .map_err(|_| anyhow::anyhow!("Failed to open IndexedDB"))?;

        // Handle upgrade (create object stores)
        let on_upgrade = Closure::once(move |event: web_sys::IdbVersionChangeEvent| {
            let db: web_sys::IdbDatabase = event
                .target()
                .unwrap()
                .dyn_into::<web_sys::IdbOpenDbRequest>()
                .unwrap()
                .result()
                .unwrap()
                .dyn_into()
                .unwrap();

            if !db.object_store_names().contains("manifest") {
                db.create_object_store("manifest").unwrap();
            }
            if !db.object_store_names().contains("blobs") {
                db.create_object_store("blobs").unwrap();
            }
        });
        open_req.set_onupgradeneeded(Some(on_upgrade.as_ref().unchecked_ref()));
        on_upgrade.forget();

        let db: web_sys::IdbDatabase = idb_request_to_future((*open_req).clone())
            .await
            .map_err(|_| anyhow::anyhow!("IndexedDB open failed"))?
            .dyn_into()
            .map_err(|_| anyhow::anyhow!("IndexedDB cast failed"))?;

        // Hydrate manifest
        let tx = db
            .transaction_with_str_and_mode("manifest", IdbTransactionMode::Readonly)
            .map_err(|_| anyhow::anyhow!("Failed to create transaction"))?;
        let store = tx
            .object_store("manifest")
            .map_err(|_| anyhow::anyhow!("No manifest store"))?;
        let get_req = store
            .get(&"entries".into())
            .map_err(|_| anyhow::anyhow!("get failed"))?;
        let result = idb_request_to_future(get_req).await;

        let memory = MemoryBackend::new();

        if let Ok(val) = result {
            if !val.is_undefined() && !val.is_null() {
                if let Some(s) = val.as_string() {
                    if let Ok(entries) = serde_json::from_str::<Vec<AssetEntry>>(&s) {
                        let mut m = memory
                            .manifest
                            .write()
                            .map_err(|e| anyhow::anyhow!("{}", e))?;
                        *m = entries;
                    }
                }
            }
        }

        // Hydrate blobs — read all keys and values
        let tx = db
            .transaction_with_str_and_mode("blobs", IdbTransactionMode::Readonly)
            .map_err(|_| anyhow::anyhow!("Failed to create blob transaction"))?;
        let store = tx
            .object_store("blobs")
            .map_err(|_| anyhow::anyhow!("No blobs store"))?;
        let keys_req = store
            .get_all_keys()
            .map_err(|_| anyhow::anyhow!("getAllKeys failed"))?;
        let keys_result = idb_request_to_future(keys_req).await;

        if let Ok(keys_val) = keys_result {
            let keys: Array = keys_val.dyn_into().unwrap_or_else(|_| Array::new());
            for i in 0..keys.length() {
                let key = keys.get(i);
                if let Some(hash) = key.as_string() {
                    let get_req = store.get(&key).ok();
                    if let Some(req) = get_req {
                        if let Ok(blob_val) = idb_request_to_future(req).await {
                            if let Ok(arr) = blob_val.dyn_into::<Uint8Array>() {
                                let data = arr.to_vec();
                                let _ = memory.write_blob(&hash, &data);
                            }
                        }
                    }
                }
            }
        }

        db.close();

        Ok(Self {
            memory,
            db_name: name.to_string(),
        })
    }

    /// Flush a blob write to IndexedDB (fire-and-forget).
    fn flush_blob(&self, hash: &str, data: &[u8]) {
        let db_name = self.db_name.clone();
        let hash = hash.to_string();
        let data = data.to_vec();
        wasm_bindgen_futures::spawn_local(async move {
            let _ = Self::idb_put_blob(&db_name, &hash, &data).await;
        });
    }

    /// Flush manifest to IndexedDB (fire-and-forget).
    fn flush_manifest(&self, entries: &[AssetEntry]) {
        let db_name = self.db_name.clone();
        let json = serde_json::to_string(entries).unwrap_or_default();
        wasm_bindgen_futures::spawn_local(async move {
            let _ = Self::idb_put_manifest(&db_name, &json).await;
        });
    }

    async fn idb_put_blob(db_name: &str, hash: &str, data: &[u8]) -> Result<()> {
        use js_sys::Uint8Array;
        use wasm_bindgen::prelude::*;
        use wasm_bindgen_futures::JsFuture;
        use web_sys::IdbTransactionMode;

        let window = web_sys::window().ok_or_else(|| anyhow::anyhow!("No window"))?;
        let factory = window
            .indexed_db()
            .ok()
            .flatten()
            .ok_or_else(|| anyhow::anyhow!("No IDB"))?;
        let open_req = factory
            .open(db_name)
            .map_err(|_| anyhow::anyhow!("open failed"))?;
        let db: web_sys::IdbDatabase = idb_request_to_future((*open_req).clone())
            .await
            .map_err(|_| anyhow::anyhow!("open await failed"))?
            .dyn_into()
            .map_err(|_| anyhow::anyhow!("cast failed"))?;

        let tx = db
            .transaction_with_str_and_mode("blobs", IdbTransactionMode::Readwrite)
            .map_err(|_| anyhow::anyhow!("tx failed"))?;
        let store = tx
            .object_store("blobs")
            .map_err(|_| anyhow::anyhow!("store failed"))?;
        let arr = Uint8Array::from(data);
        store
            .put_with_key(&arr, &hash.into())
            .map_err(|_| anyhow::anyhow!("put failed"))?;

        db.close();
        Ok(())
    }

    async fn idb_put_manifest(db_name: &str, json: &str) -> Result<()> {
        use wasm_bindgen::prelude::*;
        use wasm_bindgen_futures::JsFuture;
        use web_sys::IdbTransactionMode;

        let window = web_sys::window().ok_or_else(|| anyhow::anyhow!("No window"))?;
        let factory = window
            .indexed_db()
            .ok()
            .flatten()
            .ok_or_else(|| anyhow::anyhow!("No IDB"))?;
        let open_req = factory
            .open(db_name)
            .map_err(|_| anyhow::anyhow!("open failed"))?;
        let db: web_sys::IdbDatabase = idb_request_to_future((*open_req).clone())
            .await
            .map_err(|_| anyhow::anyhow!("open await failed"))?
            .dyn_into()
            .map_err(|_| anyhow::anyhow!("cast failed"))?;

        let tx = db
            .transaction_with_str_and_mode("manifest", IdbTransactionMode::Readwrite)
            .map_err(|_| anyhow::anyhow!("tx failed"))?;
        let store = tx
            .object_store("manifest")
            .map_err(|_| anyhow::anyhow!("store failed"))?;
        store
            .put_with_key(&json.into(), &"entries".into())
            .map_err(|_| anyhow::anyhow!("put failed"))?;

        db.close();
        Ok(())
    }
}

#[cfg(target_arch = "wasm32")]
impl StorageBackend for IndexedDbBackend {
    fn read_manifest(&self) -> Result<Vec<AssetEntry>> {
        self.memory.read_manifest()
    }

    fn write_manifest(&self, entries: &[AssetEntry]) -> Result<()> {
        self.memory.write_manifest(entries)?;
        self.flush_manifest(entries);
        Ok(())
    }

    fn read_blob(&self, hash: &str) -> Result<Vec<u8>> {
        self.memory.read_blob(hash)
    }

    fn write_blob(&self, hash: &str, data: &[u8]) -> Result<()> {
        self.memory.write_blob(hash, data)?;
        self.flush_blob(hash, data);
        Ok(())
    }

    fn blob_exists(&self, hash: &str) -> bool {
        self.memory.blob_exists(hash)
    }

    fn delete_blob(&self, hash: &str) -> Result<()> {
        self.memory.delete_blob(hash)?;
        // IDB delete is fire-and-forget
        let db_name = self.db_name.clone();
        let hash = hash.to_string();
        wasm_bindgen_futures::spawn_local(async move {
            let _ = Self::idb_put_blob(&db_name, &hash, &[]).await; // overwrite with empty
        });
        Ok(())
    }

    fn backend_name(&self) -> &'static str {
        "indexeddb"
    }
}

// ═══════════════════════════════════════════════════════════════════════════
// AssetDB
// ═══════════════════════════════════════════════════════════════════════════

// ═══════════════════════════════════════════════════════════════════════════
// Delta events — KyuGraph-compatible change stream
// ═══════════════════════════════════════════════════════════════════════════
//
// The delta system is designed for compatibility with KyuGraph's delta
// upsert protocol. Every AssetDB mutation emits a Delta that can be
// projected into Cypher MERGE/DELETE statements and applied to KyuGraph
// incrementally — no full graph rebuilds.
//
// KyuGraph supports both in-memory and S3-backed storage, so the
// knowledge graph works in development (instant) and production
// (persistent, cross-session).
//
// See `graph_projection.rs` for the Cypher projection layer.

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Delta {
    pub op: DeltaOp,
    pub id: String,
    pub entity: String,
    pub component: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub data: Option<Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub metadata: Option<Value>,
    pub timestamp: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum DeltaOp {
    Put,
    Merge,
    Delete,
    Tag,
    Spawn,
    Destroy,
}

/// Listener that receives deltas from AssetDB mutations.
pub trait DeltaListener: Send + Sync {
    fn on_delta(&self, delta: &Delta);
}

pub struct AssetDB {
    backend: Box<dyn StorageBackend>,
    cache: RwLock<Vec<AssetEntry>>,
    listeners: RwLock<Vec<Arc<dyn DeltaListener>>>,
}

impl AssetDB {
    /// Create a new AssetDB with the given storage backend.
    pub fn with_backend(backend: Box<dyn StorageBackend>) -> Result<Arc<Self>> {
        let entries = backend.read_manifest()?;
        Ok(Arc::new(Self {
            backend,
            cache: RwLock::new(entries),
            listeners: RwLock::new(Vec::new()),
        }))
    }

    /// Register a delta listener. Returns index for removal.
    pub fn add_listener(&self, listener: Arc<dyn DeltaListener>) {
        if let Ok(mut listeners) = self.listeners.write() {
            listeners.push(listener);
        }
    }

    fn emit_delta(&self, op: DeltaOp, id: &str, data: Option<Value>, metadata: Option<Value>) {
        let (entity, component) = if let Some(colon) = id.rfind(':') {
            (id[..colon].to_string(), id[colon + 1..].to_string())
        } else {
            (id.to_string(), String::new())
        };

        let delta = Delta {
            op,
            id: id.to_string(),
            entity,
            component,
            data,
            metadata,
            timestamp: now_iso(),
        };

        if let Ok(listeners) = self.listeners.read() {
            for listener in listeners.iter() {
                listener.on_delta(&delta);
            }
        }
    }

    /// Open a file-backed AssetDB at the given path (native only).
    #[cfg(not(target_arch = "wasm32"))]
    pub fn open(path: impl AsRef<std::path::Path>) -> Result<Arc<Self>> {
        let backend = FileBackend::new(path)?;
        Self::with_backend(Box::new(backend))
    }

    /// Create an in-memory AssetDB (wasm or testing).
    pub fn in_memory() -> Result<Arc<Self>> {
        Self::with_backend(Box::new(MemoryBackend::new()))
    }

    // ─── Entity-style access (ECS pattern) ───
    //
    // Assets have stable, user-defined IDs like "snake:mesh" or
    // "character/idle:animation". Put is an upsert. Get is direct lookup.
    //
    //   db.put("snake:mesh", &mesh_bytes, json!({ "stride": 24 }))?;
    //   let asset = db.get("snake:mesh")?;
    //
    // The ID encodes intent. Convention: "entity:component" or "path/name:type".

    /// Put (upsert) an asset by ID. If the ID already exists, the data
    /// and metadata are replaced. The blob is content-addressed, so
    /// re-storing identical data is free.
    pub fn put(&self, id: &str, data: &[u8], metadata: Value) -> Result<()> {
        let hash = content_hash(data);
        let (asset_type, name) = parse_entity_id(id);

        // Compress if beneficial
        let (blob, compressed) = compress_blob(data, &asset_type);
        let raw_size = data.len() as u64;

        if !self.backend.blob_exists(&hash) {
            self.backend.write_blob(&hash, &blob)?;
        }

        let now = now_iso();
        let mut cache = self.cache.write().map_err(|e| anyhow::anyhow!("{}", e))?;

        if let Some(existing) = cache.iter_mut().find(|a| a.id == id) {
            let old_hash = existing.blob_hash.clone();
            existing.blob_hash = hash;
            existing.blob_size = blob.len() as u64;
            existing.raw_size = raw_size;
            existing.compressed = compressed;
            existing.metadata = metadata;
            existing.updated_at = now;
            existing.inline_data = None;

            if !cache.iter().any(|a| a.blob_hash == old_hash) {
                let _ = self.backend.delete_blob(&old_hash);
            }
        } else {
            cache.push(AssetEntry {
                id: id.to_string(),
                name,
                asset_type,
                blob_hash: hash,
                blob_size: blob.len() as u64,
                raw_size,
                compressed,
                metadata,
                tags: Vec::new(),
                created_at: now.clone(),
                updated_at: now,
                inline_data: None,
            });
        }

        let delta_meta = json!({});
        self.backend.write_manifest(&cache)?;
        self.emit_delta(DeltaOp::Put, id, None, Some(delta_meta));
        Ok(())
    }

    /// Put JSON inline (materials, skeletons, configs). Upsert by ID.
    pub fn put_json(&self, id: &str, json_data: Value, metadata: Value) -> Result<()> {
        let serialized = serde_json::to_vec(&json_data)?;
        let hash = content_hash(&serialized);
        let (asset_type, name) = parse_entity_id(id);
        let now = now_iso();
        let delta_data = json_data.clone();

        let mut cache = self.cache.write().map_err(|e| anyhow::anyhow!("{}", e))?;

        if let Some(existing) = cache.iter_mut().find(|a| a.id == id) {
            existing.blob_hash = hash;
            existing.blob_size = serialized.len() as u64;
            existing.raw_size = 0;
            existing.compressed = false;
            existing.metadata = metadata;
            existing.updated_at = now;
            existing.inline_data = Some(json_data);
        } else {
            cache.push(AssetEntry {
                id: id.to_string(),
                name,
                asset_type,
                blob_hash: hash,
                blob_size: serialized.len() as u64,
                raw_size: 0,
                compressed: false,
                metadata,
                tags: Vec::new(),
                created_at: now.clone(),
                updated_at: now,
                inline_data: Some(json_data),
            });
        }

        self.backend.write_manifest(&cache)?;
        self.emit_delta(DeltaOp::Put, id, Some(delta_data), None);
        Ok(())
    }

    /// Tag an entity. Additive — doesn't remove existing tags.
    pub fn tag(&self, id: &str, tags: &[&str]) -> Result<()> {
        let mut cache = self.cache.write().map_err(|e| anyhow::anyhow!("{}", e))?;
        let entry = cache
            .iter_mut()
            .find(|a| a.id == id)
            .ok_or_else(|| anyhow::anyhow!("Asset not found: {}", id))?;
        for t in tags {
            let s = t.to_string();
            if !entry.tags.contains(&s) {
                entry.tags.push(s);
            }
        }
        entry.updated_at = now_iso();
        self.backend.write_manifest(&cache)?;
        self.emit_delta(DeltaOp::Tag, id, Some(json!(tags)), None);
        Ok(())
    }

    /// Get a single asset by exact ID. This is the primary access path.
    ///
    /// ```ignore
    /// let mesh = db.get("snake:mesh")?;
    /// ```
    pub fn get(&self, id: &str) -> Result<Asset> {
        let cache = self.cache.read().map_err(|e| anyhow::anyhow!("{}", e))?;
        let entry = cache
            .iter()
            .find(|a| a.id == id)
            .ok_or_else(|| anyhow::anyhow!("Entity not found: {}", id))?
            .clone();
        drop(cache);
        self.load_entry(&entry)
    }

    /// Check if an entity exists.
    pub fn has(&self, id: &str) -> bool {
        self.cache
            .read()
            .map(|c| c.iter().any(|a| a.id == id))
            .unwrap_or(false)
    }

    /// Get just the metadata for an entity (no blob load).
    pub fn get_entry(&self, id: &str) -> Result<AssetEntry> {
        let cache = self.cache.read().map_err(|e| anyhow::anyhow!("{}", e))?;
        cache
            .iter()
            .find(|a| a.id == id)
            .cloned()
            .ok_or_else(|| anyhow::anyhow!("Entity not found: {}", id))
    }

    // ─── Legacy name-based access (kept for convenience) ───

    /// Store with auto-generated ID. Returns the ID.
    pub fn store(
        &self,
        asset_type: &str,
        name: &str,
        data: &[u8],
        metadata: Value,
        tags: &[&str],
    ) -> Result<AssetId> {
        let id = format!("{}:{}", name, asset_type);
        self.put(&id, data, metadata)?;
        if !tags.is_empty() {
            self.tag(&id, tags)?;
        }
        Ok(id)
    }

    /// Store JSON with auto-generated ID. Returns the ID.
    pub fn store_json(
        &self,
        asset_type: &str,
        name: &str,
        json_data: Value,
        metadata: Value,
        tags: &[&str],
    ) -> Result<AssetId> {
        let id = format!("{}:{}", name, asset_type);
        self.put_json(&id, json_data, metadata)?;
        if !tags.is_empty() {
            self.tag(&id, tags)?;
        }
        Ok(id)
    }

    /// Load by name (convenience — looks up "name:*" pattern).
    pub fn load_by_name(&self, name: &str) -> Result<Asset> {
        let cache = self.cache.read().map_err(|e| anyhow::anyhow!("{}", e))?;
        let entry = cache
            .iter()
            .find(|a| a.name == name || a.id.starts_with(&format!("{}:", name)))
            .ok_or_else(|| anyhow::anyhow!("Asset not found: {}", name))?
            .clone();
        drop(cache);
        self.load_entry(&entry)
    }

    // ─── ECS-style component access ───
    //
    // Entities are name prefixes. Components are type suffixes.
    //
    //   db.set_component("player", "transform", &transform_bytes, json!({}))?;
    //   db.set_component("player", "rigidbody", &rb_bytes, json!({"mass": 1.0}))?;
    //   db.set_component("sun", "light", &light_bytes, json!({"type": "directional"}))?;
    //
    //   let tf = db.get_component("player", "transform")?;
    //   let entities = db.entities_with(&["rigidbody", "transform"])?;
    //   let components = db.components_of("player")?;

    /// Set a component on an entity. Upserts "entity:component" in the DB.
    pub fn set_component(
        &self,
        entity: &str,
        component: &str,
        data: &[u8],
        metadata: Value,
    ) -> Result<()> {
        let id = format!("{}:{}", entity, component);
        self.put(&id, data, metadata)
    }

    /// Set a JSON component on an entity.
    pub fn set_component_json(
        &self,
        entity: &str,
        component: &str,
        data: Value,
        metadata: Value,
    ) -> Result<()> {
        let id = format!("{}:{}", entity, component);
        self.put_json(&id, data, metadata)
    }

    /// Merge fields into an existing JSON component. Creates if missing.
    /// Only the keys present in `patch` are overwritten; all other fields preserved.
    pub fn merge_component_json(
        &self,
        entity: &str,
        component: &str,
        patch: Value,
        metadata: Value,
    ) -> Result<()> {
        let id = format!("{}:{}", entity, component);

        // Load existing
        let mut base = match self.get(&id) {
            Ok(asset) => {
                if let Some(ref inline) = asset.entry.inline_data {
                    inline.clone()
                } else {
                    serde_json::from_slice(&asset.data).unwrap_or(Value::Object(Default::default()))
                }
            }
            Err(_) => Value::Object(Default::default()),
        };

        // Merge patch on top
        if let (Value::Object(ref mut base_map), Value::Object(patch_map)) = (&mut base, &patch) {
            for (k, v) in patch_map {
                base_map.insert(k.clone(), v.clone());
            }
        } else {
            base = patch;
        }

        self.put_json(&id, base, metadata)
    }

    /// Get a component from an entity.
    pub fn get_component(&self, entity: &str, component: &str) -> Result<Asset> {
        self.get(&format!("{}:{}", entity, component))
    }

    /// Check if an entity has a specific component.
    pub fn has_component(&self, entity: &str, component: &str) -> bool {
        self.has(&format!("{}:{}", entity, component))
    }

    /// Remove a component from an entity.
    pub fn remove_component(&self, entity: &str, component: &str) -> Result<()> {
        self.delete(&format!("{}:{}", entity, component))
    }

    /// List all component types attached to an entity.
    ///
    /// ```ignore
    /// db.components_of("player") → ["transform", "rigidbody", "collider", "mesh"]
    /// ```
    pub fn components_of(&self, entity: &str) -> Result<Vec<String>> {
        let prefix = format!("{}:", entity);
        let cache = self.cache.read().map_err(|e| anyhow::anyhow!("{}", e))?;
        Ok(cache
            .iter()
            .filter(|a| a.id.starts_with(&prefix))
            .map(|a| a.id[prefix.len()..].to_string())
            .collect())
    }

    /// Find all entity names that have ALL of the specified components.
    ///
    /// ```ignore
    /// db.entities_with(&["rigidbody", "transform"])
    /// → ["player", "enemy_1", "crate_3"]
    /// ```
    pub fn entities_with(&self, components: &[&str]) -> Result<Vec<String>> {
        let cache = self.cache.read().map_err(|e| anyhow::anyhow!("{}", e))?;

        // Collect all entity names (everything before the last ':')
        let mut entity_names: HashMap<&str, Vec<&str>> = HashMap::new();
        for entry in cache.iter() {
            if let Some(colon) = entry.id.rfind(':') {
                let entity = &entry.id[..colon];
                let comp = &entry.id[colon + 1..];
                entity_names.entry(entity).or_default().push(comp);
            }
        }

        // Filter to entities that have ALL requested components
        Ok(entity_names
            .into_iter()
            .filter(|(_, comps)| components.iter().all(|c| comps.contains(c)))
            .map(|(name, _)| name.to_string())
            .collect())
    }

    /// Find all entity names that have ANY of the specified components.
    pub fn entities_with_any(&self, components: &[&str]) -> Result<Vec<String>> {
        let cache = self.cache.read().map_err(|e| anyhow::anyhow!("{}", e))?;

        let mut seen = std::collections::HashSet::new();
        let mut result = Vec::new();

        for entry in cache.iter() {
            if let Some(colon) = entry.id.rfind(':') {
                let entity = &entry.id[..colon];
                let comp = &entry.id[colon + 1..];
                if components.contains(&comp) && seen.insert(entity) {
                    result.push(entity.to_string());
                }
            }
        }
        Ok(result)
    }

    /// Get all components for an entity as a JSON object.
    /// Loads inline_data for JSON components, metadata for binary ones.
    ///
    /// ```ignore
    /// db.entity_snapshot("player")
    /// → { "transform": { "position": [0,1,0], ... },
    ///      "rigidbody": { "mass": 1.0, ... } }
    /// ```
    pub fn entity_snapshot(&self, entity: &str) -> Result<Value> {
        let prefix = format!("{}:", entity);
        let cache = self.cache.read().map_err(|e| anyhow::anyhow!("{}", e))?;

        let mut snapshot = serde_json::Map::new();
        for entry in cache.iter() {
            if entry.id.starts_with(&prefix) {
                let comp = &entry.id[prefix.len()..];
                let val = if let Some(ref inline) = entry.inline_data {
                    inline.clone()
                } else {
                    // For binary components, expose metadata
                    json!({
                        "$binary": true,
                        "size": entry.blob_size,
                        "metadata": entry.metadata,
                    })
                };
                snapshot.insert(comp.to_string(), val);
            }
        }
        Ok(Value::Object(snapshot))
    }

    /// Spawn an entity from a template (prefab). Copies all components
    /// from source entity to a new entity name.
    pub fn spawn_from(&self, template_entity: &str, new_entity: &str) -> Result<()> {
        let prefix = format!("{}:", template_entity);
        let cache = self.cache.read().map_err(|e| anyhow::anyhow!("{}", e))?;

        let entries: Vec<AssetEntry> = cache
            .iter()
            .filter(|a| a.id.starts_with(&prefix))
            .cloned()
            .collect();
        drop(cache);

        for entry in entries {
            let comp = &entry.id[prefix.len()..];
            let new_id = format!("{}:{}", new_entity, comp);

            if let Some(ref inline) = entry.inline_data {
                self.put_json(&new_id, inline.clone(), entry.metadata.clone())?;
            } else {
                let data = self.backend.read_blob(&entry.blob_hash)?;
                // If compressed, we need the raw data
                let raw = if entry.compressed {
                    decompress_blob(&data, entry.raw_size)?
                } else {
                    data
                };
                self.put(&new_id, &raw, entry.metadata.clone())?;
            }

            // Copy tags
            if !entry.tags.is_empty() {
                let tag_refs: Vec<&str> = entry.tags.iter().map(|s| s.as_str()).collect();
                self.tag(&new_id, &tag_refs)?;
            }
        }
        self.emit_delta(
            DeltaOp::Spawn,
            new_entity,
            Some(json!({"template": template_entity})),
            None,
        );
        Ok(())
    }

    /// Delete all components for an entity.
    pub fn destroy_entity(&self, entity: &str) -> Result<()> {
        let prefix = format!("{}:", entity);
        let mut cache = self.cache.write().map_err(|e| anyhow::anyhow!("{}", e))?;

        let to_remove: Vec<usize> = cache
            .iter()
            .enumerate()
            .filter(|(_, a)| a.id.starts_with(&prefix))
            .map(|(i, _)| i)
            .collect();

        // Remove in reverse order to preserve indices
        for &idx in to_remove.iter().rev() {
            let entry = cache.remove(idx);
            let hash_used = cache.iter().any(|a| a.blob_hash == entry.blob_hash);
            if !hash_used && entry.inline_data.is_none() {
                let _ = self.backend.delete_blob(&entry.blob_hash);
            }
        }

        self.backend.write_manifest(&cache)?;
        self.emit_delta(DeltaOp::Destroy, entity, None, None);
        Ok(())
    }

    /// Query from JSON DSL — primary interface for DAG actors.
    ///
    /// ```json
    /// { "type": "mesh", "tags": ["snake"], "$sort": "newest", "$limit": 5 }
    /// ```
    pub fn query_dsl(&self, dsl: &Value) -> Result<Vec<AssetEntry>> {
        self.query(&AssetQuery::from_json(dsl))
    }

    /// Query metadata (no blob data).
    pub fn query(&self, q: &AssetQuery) -> Result<Vec<AssetEntry>> {
        let cache = self.cache.read().map_err(|e| anyhow::anyhow!("{}", e))?;

        let mut results: Vec<AssetEntry> = cache
            .iter()
            .filter(|a| q.filters.iter().all(|f| matches_filter(a, f)))
            .cloned()
            .collect();

        // Sort
        match q.sort {
            SortOrder::Newest => results.sort_by(|a, b| b.created_at.cmp(&a.created_at)),
            SortOrder::Oldest => results.sort_by(|a, b| a.created_at.cmp(&b.created_at)),
            SortOrder::Largest => results.sort_by_key(|entry| std::cmp::Reverse(entry.blob_size)),
            SortOrder::Smallest => results.sort_by_key(|entry| entry.blob_size),
            SortOrder::Name => results.sort_by(|a, b| a.name.cmp(&b.name)),
            SortOrder::None => {}
        }

        if let Some(limit) = q.limit {
            results.truncate(limit);
        }
        Ok(results)
    }

    /// Delete by ID.
    pub fn delete(&self, id: &str) -> Result<()> {
        let mut cache = self.cache.write().map_err(|e| anyhow::anyhow!("{}", e))?;
        let idx = cache
            .iter()
            .position(|a| a.id == id)
            .ok_or_else(|| anyhow::anyhow!("Asset not found: {}", id))?;

        let entry = cache.remove(idx);
        let hash_still_used = cache.iter().any(|a| a.blob_hash == entry.blob_hash);
        if !hash_still_used && entry.inline_data.is_none() {
            let _ = self.backend.delete_blob(&entry.blob_hash);
        }
        self.backend.write_manifest(&cache)?;
        self.emit_delta(DeltaOp::Delete, id, None, None);
        Ok(())
    }

    /// Update metadata/tags.
    pub fn update_metadata(
        &self,
        id: &str,
        metadata: Option<Value>,
        tags: Option<Vec<String>>,
    ) -> Result<()> {
        let mut cache = self.cache.write().map_err(|e| anyhow::anyhow!("{}", e))?;
        let entry = cache
            .iter_mut()
            .find(|a| a.id == id)
            .ok_or_else(|| anyhow::anyhow!("Asset not found: {}", id))?;

        if let Some(meta) = metadata {
            entry.metadata = meta;
        }
        if let Some(t) = tags {
            entry.tags = t;
        }
        entry.updated_at = now_iso();
        self.backend.write_manifest(&cache)?;
        Ok(())
    }

    /// Stats summary.
    pub fn stats(&self) -> Result<Value> {
        let cache = self.cache.read().map_err(|e| anyhow::anyhow!("{}", e))?;
        let mut type_counts: HashMap<String, usize> = HashMap::new();
        let mut total_bytes: u64 = 0;

        for a in cache.iter() {
            *type_counts
                .entry(a.asset_type.as_str().to_string())
                .or_default() += 1;
            total_bytes += a.blob_size;
        }

        Ok(json!({
            "assetCount": cache.len(),
            "totalBytes": total_bytes,
            "types": type_counts,
            "backend": self.backend.backend_name(),
        }))
    }

    fn load_entry(&self, entry: &AssetEntry) -> Result<Asset> {
        let data = if let Some(ref inline) = entry.inline_data {
            serde_json::to_vec(inline)?
        } else {
            let raw = self.backend.read_blob(&entry.blob_hash)?;
            if entry.compressed {
                decompress_blob(&raw, entry.raw_size)?
            } else {
                raw
            }
        };
        Ok(Asset {
            entry: entry.clone(),
            data,
        })
    }
}

// ═══════════════════════════════════════════════════════════════════════════
// Query DSL — document-style, DAG-friendly
// ═══════════════════════════════════════════════════════════════════════════
//
// The query describes the shape of what you're looking for.
// Plain values are exact matches. Operator objects refine the match.
//
//   // Find all meshes tagged "snake"
//   { "type": "mesh", "tags": ["snake"] }
//
//   // Find assets whose name contains "body"
//   { "name": { "$contains": "body" } }
//
//   // Find large meshes with specific metadata
//   { "type": "mesh", "size": { "$gt": 10000 }, "metadata.stride": 24 }
//
//   // Combine conditions
//   {
//     "type": { "$in": ["mesh", "texture"] },
//     "tags": { "$all": ["snake", "procedural"] },
//     "name": { "$startsWith": "snake_" },
//     "$sort": "newest",
//     "$limit": 10
//   }
//
// ## Operators
//
//   $gt, $gte, $lt, $lte  — numeric comparison
//   $in                    — value in set
//   $contains              — substring match (strings) or has-any (tags)
//   $startsWith            — prefix match
//   $all                   — tags: has all listed values
//   $exists                — metadata field exists (true/false)
//   $between               — [lo, hi] range
//   $not                   — negate: { "$not": "mesh" }
//
// ## Control keys
//
//   $sort   — "newest" | "oldest" | "largest" | "smallest" | "name"
//   $limit  — max results
//

/// Internal query representation parsed from JSON DSL.
#[derive(Default)]
pub struct AssetQuery {
    pub filters: Vec<Filter>,
    pub sort: SortOrder,
    pub limit: Option<usize>,
}

#[derive(Clone)]
pub struct Filter {
    pub field: String,
    pub op: FilterOp,
    pub value: Value,
}

#[derive(Clone)]
pub enum FilterOp {
    Eq,
    Neq,
    Contains,
    StartsWith,
    In,
    Gt,
    Gte,
    Lt,
    Lte,
    Between,
    HasAny,
    HasAll,
    Exists,
}

#[derive(Default, Clone)]
pub enum SortOrder {
    #[default]
    None,
    Newest,
    Oldest,
    Largest,
    Smallest,
    Name,
}

#[derive(Default, Clone)]
pub enum TagMatch {
    #[default]
    Any,
    All,
}

impl AssetQuery {
    pub fn new() -> Self {
        Self::default()
    }

    /// Parse a query from the document-style JSON DSL.
    ///
    /// ```json
    /// { "type": "mesh", "tags": ["snake"], "metadata.stride": 24, "$limit": 10 }
    /// ```
    pub fn from_json(v: &Value) -> Self {
        let mut q = Self::default();

        let map = match v.as_object() {
            Some(m) => m,
            None => return q,
        };

        // Control keys
        q.limit = map
            .get("$limit")
            .and_then(|v| v.as_u64())
            .map(|n| n as usize);
        if let Some(s) = map.get("$sort").and_then(|v| v.as_str()) {
            q.sort = match s {
                "newest" => SortOrder::Newest,
                "oldest" => SortOrder::Oldest,
                "largest" => SortOrder::Largest,
                "smallest" => SortOrder::Smallest,
                "name" => SortOrder::Name,
                _ => SortOrder::None,
            };
        }

        // Each non-$ key is a field filter
        for (key, val) in map {
            if key.starts_with('$') {
                continue;
            }
            parse_field_filter(&mut q.filters, key, val);
        }

        q
    }

    /// Rust builder — still available for programmatic use.
    pub fn asset_type(mut self, t: &str) -> Self {
        self.filters.push(Filter {
            field: "type".into(),
            op: FilterOp::Eq,
            value: json!(t),
        });
        self
    }
    pub fn name_contains(mut self, n: &str) -> Self {
        self.filters.push(Filter {
            field: "name".into(),
            op: FilterOp::Contains,
            value: json!(n),
        });
        self
    }
    pub fn tag(mut self, t: &str) -> Self {
        self.filters.push(Filter {
            field: "tags".into(),
            op: FilterOp::HasAny,
            value: json!([t]),
        });
        self
    }
    pub fn tags(mut self, tags: &[&str]) -> Self {
        self.filters.push(Filter {
            field: "tags".into(),
            op: FilterOp::HasAny,
            value: json!(tags),
        });
        self
    }
    pub fn match_all_tags(mut self) -> Self {
        // Convert last HasAny to HasAll
        if let Some(f) = self.filters.last_mut() {
            if matches!(f.op, FilterOp::HasAny) && f.field == "tags" {
                f.op = FilterOp::HasAll;
            }
        }
        self
    }
    pub fn limit(mut self, n: usize) -> Self {
        self.limit = Some(n);
        self
    }
}

/// Parse a field:value pair from the document-style DSL.
///
/// Plain value → exact match: `"type": "mesh"`
/// Array value → has-any for tags, in for others: `"tags": ["snake", "body"]`
/// Object value → operator: `"name": { "$contains": "body" }`
fn parse_field_filter(filters: &mut Vec<Filter>, field: &str, val: &Value) {
    match val {
        // Operator object: { "$contains": "body", "$gt": 100 }
        Value::Object(ops) => {
            for (op_key, op_val) in ops {
                let op = match op_key.as_str() {
                    "$gt" => FilterOp::Gt,
                    "$gte" => FilterOp::Gte,
                    "$lt" => FilterOp::Lt,
                    "$lte" => FilterOp::Lte,
                    "$in" => FilterOp::In,
                    "$contains" => FilterOp::Contains,
                    "$startsWith" => FilterOp::StartsWith,
                    "$all" => FilterOp::HasAll,
                    "$between" => FilterOp::Between,
                    "$exists" => FilterOp::Exists,
                    "$not" => FilterOp::Neq,
                    _ => continue,
                };
                filters.push(Filter {
                    field: field.to_string(),
                    op,
                    value: op_val.clone(),
                });
            }
        }
        // Array: tags → has-any, others → in
        Value::Array(_) => {
            let op = if field == "tags" {
                FilterOp::HasAny
            } else {
                FilterOp::In
            };
            filters.push(Filter {
                field: field.to_string(),
                op,
                value: val.clone(),
            });
        }
        // Scalar: exact match
        _ => {
            filters.push(Filter {
                field: field.to_string(),
                op: FilterOp::Eq,
                value: val.clone(),
            });
        }
    }
}

/// Evaluate a filter against an asset entry.
fn matches_filter(entry: &AssetEntry, filter: &Filter) -> bool {
    match filter.field.as_str() {
        "type" => match filter.op {
            FilterOp::Eq => filter
                .value
                .as_str()
                .map(|s| AssetType::parse(s) == entry.asset_type)
                .unwrap_or(false),
            FilterOp::Neq => filter
                .value
                .as_str()
                .map(|s| AssetType::parse(s) != entry.asset_type)
                .unwrap_or(false),
            FilterOp::In => filter
                .value
                .as_array()
                .map(|arr| {
                    arr.iter().any(|v| {
                        v.as_str()
                            .map(|s| AssetType::parse(s) == entry.asset_type)
                            .unwrap_or(false)
                    })
                })
                .unwrap_or(false),
            _ => true,
        },
        "name" => {
            let name = &entry.name;
            match filter.op {
                FilterOp::Eq => filter.value.as_str().map(|s| name == s).unwrap_or(false),
                FilterOp::Neq => filter.value.as_str().map(|s| name != s).unwrap_or(false),
                FilterOp::Contains => filter
                    .value
                    .as_str()
                    .map(|s| name.to_lowercase().contains(&s.to_lowercase()))
                    .unwrap_or(false),
                FilterOp::StartsWith => filter
                    .value
                    .as_str()
                    .map(|s| name.to_lowercase().starts_with(&s.to_lowercase()))
                    .unwrap_or(false),
                FilterOp::In => filter
                    .value
                    .as_array()
                    .map(|arr| arr.iter().any(|v| v.as_str() == Some(name.as_str())))
                    .unwrap_or(false),
                _ => true,
            }
        }
        "tags" => {
            let tags = &entry.tags;
            match filter.op {
                FilterOp::HasAny | FilterOp::Contains => {
                    if let Some(arr) = filter.value.as_array() {
                        arr.iter().any(|v| {
                            v.as_str()
                                .map(|s| tags.contains(&s.to_string()))
                                .unwrap_or(false)
                        })
                    } else if let Some(s) = filter.value.as_str() {
                        tags.contains(&s.to_string())
                    } else {
                        false
                    }
                }
                FilterOp::HasAll => {
                    if let Some(arr) = filter.value.as_array() {
                        arr.iter().all(|v| {
                            v.as_str()
                                .map(|s| tags.contains(&s.to_string()))
                                .unwrap_or(false)
                        })
                    } else {
                        false
                    }
                }
                FilterOp::Eq => filter
                    .value
                    .as_str()
                    .map(|s| tags.contains(&s.to_string()))
                    .unwrap_or(false),
                _ => true,
            }
        }
        "size" => {
            let size = entry.blob_size as f64;
            match_numeric(&filter.op, size, &filter.value)
        }
        "id" => match filter.op {
            FilterOp::Eq => filter
                .value
                .as_str()
                .map(|s| entry.id == s)
                .unwrap_or(false),
            FilterOp::In => filter
                .value
                .as_array()
                .map(|arr| arr.iter().any(|v| v.as_str() == Some(entry.id.as_str())))
                .unwrap_or(false),
            _ => true,
        },
        field if field.starts_with("metadata.") => {
            let path = &field["metadata.".len()..];
            let val = get_json_path(&entry.metadata, path);
            match filter.op {
                FilterOp::Eq => val.map(|v| v == &filter.value).unwrap_or(false),
                FilterOp::Neq => val.map(|v| v != &filter.value).unwrap_or(true),
                FilterOp::Exists => val.is_some(),
                FilterOp::Gt | FilterOp::Gte | FilterOp::Lt | FilterOp::Lte | FilterOp::Between => {
                    val.and_then(|v| v.as_f64())
                        .map(|n| match_numeric(&filter.op, n, &filter.value))
                        .unwrap_or(false)
                }
                FilterOp::Contains => val
                    .and_then(|v| v.as_str())
                    .map(|s| {
                        filter
                            .value
                            .as_str()
                            .map(|q| s.to_lowercase().contains(&q.to_lowercase()))
                            .unwrap_or(false)
                    })
                    .unwrap_or(false),
                _ => true,
            }
        }
        _ => true, // unknown field = no filter
    }
}

fn match_numeric(op: &FilterOp, actual: f64, value: &Value) -> bool {
    match op {
        FilterOp::Gt => value.as_f64().map(|v| actual > v).unwrap_or(false),
        FilterOp::Gte => value.as_f64().map(|v| actual >= v).unwrap_or(false),
        FilterOp::Lt => value.as_f64().map(|v| actual < v).unwrap_or(false),
        FilterOp::Lte => value.as_f64().map(|v| actual <= v).unwrap_or(false),
        FilterOp::Eq => value
            .as_f64()
            .map(|v| (actual - v).abs() < f64::EPSILON)
            .unwrap_or(false),
        FilterOp::Between => {
            if let Some(arr) = value.as_array() {
                let lo = arr.first().and_then(|v| v.as_f64()).unwrap_or(f64::MIN);
                let hi = arr.get(1).and_then(|v| v.as_f64()).unwrap_or(f64::MAX);
                actual >= lo && actual <= hi
            } else {
                false
            }
        }
        _ => true,
    }
}

fn get_json_path<'a>(v: &'a Value, path: &str) -> Option<&'a Value> {
    let mut current = v;
    for key in path.split('.') {
        current = current.get(key)?;
    }
    Some(current)
}

// ═══════════════════════════════════════════════════════════════════════════
// Global DB registry
// ═══════════════════════════════════════════════════════════════════════════

static DB_REGISTRY: std::sync::OnceLock<Mutex<HashMap<String, Arc<AssetDB>>>> =
    std::sync::OnceLock::new();

fn registry() -> &'static Mutex<HashMap<String, Arc<AssetDB>>> {
    DB_REGISTRY.get_or_init(|| Mutex::new(HashMap::new()))
}

/// Get or create a file-backed AssetDB (native).
#[cfg(not(target_arch = "wasm32"))]
pub fn get_or_create_db(path: &str) -> Result<Arc<AssetDB>> {
    let mut reg = registry().lock().map_err(|e| anyhow::anyhow!("{}", e))?;
    if let Some(db) = reg.get(path) {
        return Ok(Arc::clone(db));
    }
    let db = AssetDB::open(path)?;
    reg.insert(path.to_string(), Arc::clone(&db));
    Ok(db)
}

/// Get or create an AssetDB (wasm — returns in-memory if not yet hydrated).
/// For full IndexedDB persistence, use `get_or_create_db_async` instead.
#[cfg(target_arch = "wasm32")]
pub fn get_or_create_db(path: &str) -> Result<Arc<AssetDB>> {
    let mut reg = registry().lock().map_err(|e| anyhow::anyhow!("{}", e))?;
    if let Some(db) = reg.get(path) {
        return Ok(Arc::clone(db));
    }
    // Sync fallback: in-memory only. Use get_or_create_db_async for IndexedDB.
    let db = AssetDB::in_memory()?;
    reg.insert(path.to_string(), Arc::clone(&db));
    Ok(db)
}

/// Get or create an IndexedDB-backed AssetDB (wasm, async).
/// Hydrates from IndexedDB on first call, returns cached instance after.
#[cfg(target_arch = "wasm32")]
pub async fn get_or_create_db_async(name: &str) -> Result<Arc<AssetDB>> {
    // Check cache first (sync)
    {
        let reg = registry().lock().map_err(|e| anyhow::anyhow!("{}", e))?;
        if let Some(db) = reg.get(name) {
            return Ok(Arc::clone(db));
        }
    }
    // Hydrate from IndexedDB (async)
    let backend = IndexedDbBackend::open(name).await?;
    let db = AssetDB::with_backend(Box::new(backend))?;
    let mut reg = registry().lock().map_err(|e| anyhow::anyhow!("{}", e))?;
    reg.insert(name.to_string(), Arc::clone(&db));
    Ok(db)
}

// ═══════════════════════════════════════════════════════════════════════════
// Helpers
// ═══════════════════════════════════════════════════════════════════════════

// ═══════════════════════════════════════════════════════════════════════════
// LZ4 compression — transparent, selective
// ═══════════════════════════════════════════════════════════════════════════

/// Minimum blob size to bother compressing (small blobs have overhead > savings).
const COMPRESS_THRESHOLD: usize = 256;

/// Asset types that are already compressed (PNG, JPEG, H.264, etc.) — skip LZ4.
fn is_precompressed(asset_type: &AssetType) -> bool {
    matches!(
        asset_type,
        AssetType::Texture | AssetType::Audio | AssetType::Video
    )
}

/// Compress data with LZ4 if it's worth it. Returns (compressed_data, is_compressed).
fn compress_blob(data: &[u8], asset_type: &AssetType) -> (Vec<u8>, bool) {
    if data.len() < COMPRESS_THRESHOLD || is_precompressed(asset_type) {
        return (data.to_vec(), false);
    }

    let compressed = lz4_flex::compress_prepend_size(data);

    // Only use compression if it actually saves space
    if compressed.len() < data.len() {
        (compressed, true)
    } else {
        (data.to_vec(), false)
    }
}

/// Decompress LZ4 data.
fn decompress_blob(data: &[u8], _raw_size: u64) -> Result<Vec<u8>> {
    lz4_flex::decompress_size_prepended(data)
        .map_err(|e| anyhow::anyhow!("LZ4 decompress failed: {}", e))
}

/// Parse entity ID convention: "name:type" → (AssetType, name).
/// If no colon, type defaults to Generic and the whole string is the name.
fn parse_entity_id(id: &str) -> (AssetType, String) {
    if let Some(colon) = id.rfind(':') {
        let name = &id[..colon];
        let type_str = &id[colon + 1..];
        (AssetType::parse(type_str), name.to_string())
    } else {
        (AssetType::Generic, id.to_string())
    }
}

fn content_hash(data: &[u8]) -> String {
    use std::collections::hash_map::DefaultHasher;
    use std::hash::{Hash, Hasher};

    let mut h1 = DefaultHasher::new();
    data.hash(&mut h1);
    let a = h1.finish();

    let mut h2 = DefaultHasher::new();
    data.len().hash(&mut h2);
    data.hash(&mut h2);
    let b = h2.finish();

    format!("{:016x}{:016x}", a, b)
}

#[allow(dead_code)]
fn gen_id() -> String {
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::time::SystemTime;
    static CTR: AtomicU64 = AtomicU64::new(0);

    let ts = SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos() as u64;
    let c = CTR.fetch_add(1, Ordering::Relaxed);
    let a = ts ^ c.wrapping_mul(6364136223846793005);
    let b = ts.wrapping_mul(2654435761) ^ c;

    format!(
        "{:08x}-{:04x}-4{:03x}-{:04x}-{:012x}",
        (a >> 32) as u32,
        (a >> 16) as u16,
        a as u16 & 0x0FFF,
        (b >> 48) as u16 & 0x3FFF | 0x8000,
        b & 0xFFFF_FFFF_FFFF,
    )
}

fn now_iso() -> String {
    chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string()
}