sochdb-kernel 2.0.8

// 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/>.

//! WASM-Sandboxed Multi-Tenant Plugin Runtime
//!
//! This module replaces the unsafe `libloading`-based dynamic plugin loader
//! with a secure, sandboxed WASM runtime.
//!
//! ## Security Model
//!
//! - **Memory Isolation**: Each plugin runs in its own linear memory
//! - **Fuel Limits**: Instruction counting prevents infinite loops
//! - **Capability-Based Access**: Plugins can only access allowed resources
//! - **No Syscalls**: WASM code cannot directly access filesystem/network
//!
//! ## Performance
//!
//! - Target overhead: ~100ns per host function call
//! - Memory-mapped WASM modules for fast instantiation
//! - Pooled instances for frequently-used plugins
//!
//! ## Architecture
//!
//! ```text
//! ┌─────────────────────────────────────────────────────────────┐
//! │                    WASM Plugin Runtime                       │
//! │  ┌─────────────────────────────────────────────────────┐    │
//! │  │              WasmPluginRegistry                      │    │
//! │  │   ┌──────────┐ ┌──────────┐ ┌──────────┐            │    │
//! │  │   │Plugin A  │ │Plugin B  │ │Plugin C  │            │    │
//! │  │   │(WASM)    │ │(WASM)    │ │(WASM)    │            │    │
//! │  │   └────┬─────┘ └────┬─────┘ └────┬─────┘            │    │
//! │  └────────┼────────────┼────────────┼───────────────────┘    │
//! │           │            │            │                        │
//! │  ┌────────┴────────────┴────────────┴───────────────────┐    │
//! │  │              Host Function ABI                        │    │
//! │  │   soch_read, soch_write, vector_search, emit_metric   │    │
//! │  └──────────────────────────────────────────────────────┘    │
//! └─────────────────────────────────────────────────────────────┘
//! ```

use crate::error::{KernelError, KernelResult};
use crate::plugin::{Extension, ExtensionCapability, ExtensionInfo, ObservabilityExtension};
use parking_lot::RwLock;
use std::any::Any;
use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Instant;

// ============================================================================
// WASM Plugin Capabilities
// ============================================================================

/// Capabilities granted to a WASM plugin
///
/// This struct defines what a plugin can and cannot do.
/// Capability violations result in immediate trap.
#[derive(Debug, Clone)]
pub struct WasmPluginCapabilities {
    /// Tables the plugin can read (glob patterns supported)
    pub can_read_table: Vec<String>,
    /// Tables the plugin can write (glob patterns supported)
    pub can_write_table: Vec<String>,
    /// Can perform index searches
    pub can_index_search: bool,
    /// Can perform vector similarity search
    pub can_vector_search: bool,
    /// Can call other plugins
    pub can_call_plugin: Vec<String>,
    /// Memory limit in bytes
    pub memory_limit_bytes: u64,
    /// Fuel limit (instruction count)
    pub fuel_limit: u64,
    /// Timeout in milliseconds
    pub timeout_ms: u64,
}

impl Default for WasmPluginCapabilities {
    fn default() -> Self {
        Self {
            can_read_table: vec![],
            can_write_table: vec![],
            can_index_search: false,
            can_vector_search: false,
            can_call_plugin: vec![],
            memory_limit_bytes: 16 * 1024 * 1024, // 16 MB default
            fuel_limit: 1_000_000,                // 1M instructions
            timeout_ms: 100,                      // 100ms timeout
        }
    }
}

impl WasmPluginCapabilities {
    /// Create capabilities for an observability-only plugin
    pub fn observability_only() -> Self {
        Self {
            can_read_table: vec![],
            can_write_table: vec![],
            can_index_search: false,
            can_vector_search: false,
            can_call_plugin: vec![],
            memory_limit_bytes: 4 * 1024 * 1024, // 4 MB for metrics
            fuel_limit: 100_000,                 // 100K instructions
            timeout_ms: 10,                      // 10ms timeout
        }
    }

    /// Create capabilities for a read-only analytics plugin
    pub fn read_only(tables: Vec<String>) -> Self {
        Self {
            can_read_table: tables,
            can_write_table: vec![],
            can_index_search: true,
            can_vector_search: true,
            can_call_plugin: vec![],
            memory_limit_bytes: 64 * 1024 * 1024, // 64 MB for analytics
            fuel_limit: 10_000_000,               // 10M instructions
            timeout_ms: 1000,                     // 1s timeout
        }
    }

    /// Check if the plugin can read a given table
    pub fn can_read(&self, table_name: &str) -> bool {
        self.can_read_table.iter().any(|pattern| {
            if pattern == "*" {
                true
            } else if pattern.ends_with('*') {
                table_name.starts_with(&pattern[..pattern.len() - 1])
            } else {
                pattern == table_name
            }
        })
    }

    /// Check if the plugin can write to a given table
    pub fn can_write(&self, table_name: &str) -> bool {
        self.can_write_table.iter().any(|pattern| {
            if pattern == "*" {
                true
            } else if pattern.ends_with('*') {
                table_name.starts_with(&pattern[..pattern.len() - 1])
            } else {
                pattern == table_name
            }
        })
    }
}

// ============================================================================
// WASM Plugin Instance
// ============================================================================

/// State of a WASM plugin instance
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WasmPluginState {
    /// Plugin is loading
    Loading,
    /// Plugin is ready to handle requests
    Ready,
    /// Plugin is currently executing
    Executing,
    /// Plugin execution trapped (error)
    Trapped,
    /// Plugin is being unloaded
    Unloading,
    /// Plugin is unloaded
    Unloaded,
}

/// Statistics for a WASM plugin instance
#[derive(Debug, Clone, Default)]
pub struct WasmPluginStats {
    /// Total number of calls
    pub total_calls: u64,
    /// Total fuel consumed
    pub total_fuel_consumed: u64,
    /// Total execution time in microseconds
    pub total_execution_us: u64,
    /// Number of traps
    pub trap_count: u64,
    /// Peak memory usage in bytes
    pub peak_memory_bytes: u64,
}

/// Configuration for a WASM plugin instance
#[derive(Debug, Clone)]
pub struct WasmInstanceConfig {
    /// Plugin capabilities
    pub capabilities: WasmPluginCapabilities,
    /// Enable debug mode (verbose logging)
    pub debug_mode: bool,
    /// Enable fuel metering
    pub enable_fuel: bool,
    /// Enable epoch-based interrupts
    pub enable_epochs: bool,
    /// Epoch interrupt interval in milliseconds
    pub epoch_interval_ms: u64,
}

impl Default for WasmInstanceConfig {
    fn default() -> Self {
        Self {
            capabilities: WasmPluginCapabilities::default(),
            debug_mode: false,
            enable_fuel: true,
            enable_epochs: true,
            epoch_interval_ms: 10, // 10ms epoch interrupt
        }
    }
}

// ============================================================================
// WASM Plugin Instance (Simulated)
// ============================================================================

/// A WASM plugin instance
///
/// In production, this wraps wasmtime's Instance. For now, we provide
/// a functional simulation that demonstrates the API and security model.
pub struct WasmPluginInstance {
    /// Plugin name
    name: String,
    /// Plugin state
    state: RwLock<WasmPluginState>,
    /// Plugin configuration
    config: WasmInstanceConfig,
    /// Plugin statistics
    stats: RwLock<WasmPluginStats>,
    /// Fuel remaining (for metering)
    fuel_remaining: AtomicU64,
    /// Extension info
    info: ExtensionInfo,
    /// Module bytes (for hot-reload verification)
    module_hash: [u8; 32],
}

impl WasmPluginInstance {
    /// Create a new WASM plugin instance from bytes
    pub fn new(name: &str, _wasm_bytes: &[u8], config: WasmInstanceConfig) -> KernelResult<Self> {
        // In production, this would compile the WASM module using wasmtime
        // For now, we create a simulated instance

        // Compute module hash for integrity verification
        let module_hash = Self::compute_hash(_wasm_bytes);

        Ok(Self {
            name: name.to_string(),
            state: RwLock::new(WasmPluginState::Loading),
            config: config.clone(),
            stats: RwLock::new(WasmPluginStats::default()),
            fuel_remaining: AtomicU64::new(config.capabilities.fuel_limit),
            info: ExtensionInfo {
                name: name.to_string(),
                version: "1.0.0".to_string(),
                description: format!("WASM plugin: {}", name),
                author: "SochDB".to_string(),
                capabilities: vec![ExtensionCapability::Custom("wasm".to_string())],
            },
            module_hash,
        })
    }

    /// Initialize the plugin
    pub fn init(&self) -> KernelResult<()> {
        *self.state.write() = WasmPluginState::Ready;
        Ok(())
    }

    /// Call a function in the plugin
    pub fn call(&self, func_name: &str, args: &[WasmValue]) -> KernelResult<Vec<WasmValue>> {
        // Check state
        {
            let state = self.state.read();
            if *state != WasmPluginState::Ready {
                return Err(KernelError::Plugin {
                    message: format!("plugin {} not ready, state: {:?}", self.name, *state),
                });
            }
        }

        // Set executing state
        *self.state.write() = WasmPluginState::Executing;
        let start = Instant::now();

        // Check fuel
        if self.config.enable_fuel {
            let remaining = self.fuel_remaining.load(Ordering::Acquire);
            if remaining == 0 {
                *self.state.write() = WasmPluginState::Trapped;
                return Err(KernelError::Plugin {
                    message: format!("plugin {} exhausted fuel limit", self.name),
                });
            }
        }

        // Simulate function call with fuel consumption
        // In production, this would invoke the WASM function via wasmtime
        let result = self.simulate_call(func_name, args);

        // Update stats
        {
            let mut stats = self.stats.write();
            stats.total_calls += 1;
            stats.total_execution_us += start.elapsed().as_micros() as u64;

            // Simulate fuel consumption (10 fuel per arg + 100 base)
            let fuel_used = 100 + args.len() as u64 * 10;
            stats.total_fuel_consumed += fuel_used;
            self.fuel_remaining.fetch_sub(
                fuel_used.min(self.fuel_remaining.load(Ordering::Acquire)),
                Ordering::AcqRel,
            );
        }

        // Restore ready state
        *self.state.write() = WasmPluginState::Ready;

        result
    }

    /// Refuel the plugin (reset fuel to limit)
    pub fn refuel(&self) {
        self.fuel_remaining
            .store(self.config.capabilities.fuel_limit, Ordering::Release);
    }

    /// Get current statistics
    pub fn stats(&self) -> WasmPluginStats {
        self.stats.read().clone()
    }

    /// Get current state
    pub fn state(&self) -> WasmPluginState {
        *self.state.read()
    }

    /// Get plugin name
    pub fn name(&self) -> &str {
        &self.name
    }

    /// Get capabilities
    pub fn capabilities(&self) -> &WasmPluginCapabilities {
        &self.config.capabilities
    }

    /// Get module hash
    pub fn module_hash(&self) -> &[u8; 32] {
        &self.module_hash
    }

    /// Simulate a function call (placeholder for real WASM execution)
    fn simulate_call(&self, func_name: &str, args: &[WasmValue]) -> KernelResult<Vec<WasmValue>> {
        // Simulated responses for common plugin functions
        match func_name {
            "on_insert" | "on_update" | "on_delete" => {
                // Table hook - return success (0)
                Ok(vec![WasmValue::I32(0)])
            }
            "get_metrics" => {
                // Return a simulated metric value
                Ok(vec![WasmValue::F64(42.0)])
            }
            "transform" => {
                // Echo back the first argument
                if args.is_empty() {
                    Ok(vec![WasmValue::I32(0)])
                } else {
                    Ok(vec![args[0].clone()])
                }
            }
            _ => Err(KernelError::Plugin {
                message: format!("unknown function: {}", func_name),
            }),
        }
    }

    /// Compute SHA-256 hash of module bytes
    fn compute_hash(bytes: &[u8]) -> [u8; 32] {
        // Simple hash using CRC32 repeated (not cryptographic, just for demo)
        // In production, use SHA-256
        let mut hash = [0u8; 32];
        let crc = crc32fast::hash(bytes);
        for i in 0..8 {
            hash[i * 4..(i + 1) * 4].copy_from_slice(&crc.to_le_bytes());
        }
        hash
    }
}

// ============================================================================
// WASM Value Types
// ============================================================================

/// WASM value types for function arguments and returns
#[derive(Debug, Clone, PartialEq)]
pub enum WasmValue {
    /// 32-bit integer
    I32(i32),
    /// 64-bit integer
    I64(i64),
    /// 32-bit float
    F32(f32),
    /// 64-bit float
    F64(f64),
    /// External reference (for host objects)
    ExternRef(u64),
}

impl WasmValue {
    /// Convert to i32
    pub fn as_i32(&self) -> Option<i32> {
        match self {
            WasmValue::I32(v) => Some(*v),
            _ => None,
        }
    }

    /// Convert to i64
    pub fn as_i64(&self) -> Option<i64> {
        match self {
            WasmValue::I64(v) => Some(*v),
            _ => None,
        }
    }

    /// Convert to f32
    pub fn as_f32(&self) -> Option<f32> {
        match self {
            WasmValue::F32(v) => Some(*v),
            _ => None,
        }
    }

    /// Convert to f64
    pub fn as_f64(&self) -> Option<f64> {
        match self {
            WasmValue::F64(v) => Some(*v),
            _ => None,
        }
    }
}

// ============================================================================
// WASM Plugin Registry
// ============================================================================

/// Registry for WASM plugins
///
/// Thread-safe registry that manages plugin lifecycle and provides
/// fast lookup for plugin invocation.
pub struct WasmPluginRegistry {
    /// Registered plugins by name
    plugins: RwLock<HashMap<String, Arc<WasmPluginInstance>>>,
    /// Plugin load order (for deterministic shutdown)
    load_order: RwLock<Vec<String>>,
    /// Global statistics
    total_calls: AtomicU64,
    total_traps: AtomicU64,
}

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

impl WasmPluginRegistry {
    /// Create a new registry
    pub fn new() -> Self {
        Self {
            plugins: RwLock::new(HashMap::new()),
            load_order: RwLock::new(Vec::new()),
            total_calls: AtomicU64::new(0),
            total_traps: AtomicU64::new(0),
        }
    }

    /// Load a plugin from WASM bytes
    pub fn load(
        &self,
        name: &str,
        wasm_bytes: &[u8],
        config: WasmInstanceConfig,
    ) -> KernelResult<()> {
        // Check if already registered
        if self.plugins.read().contains_key(name) {
            return Err(KernelError::Plugin {
                message: format!("plugin '{}' already registered", name),
            });
        }

        // Create and initialize instance
        let instance = WasmPluginInstance::new(name, wasm_bytes, config)?;
        instance.init()?;

        // Register
        let arc = Arc::new(instance);
        self.plugins.write().insert(name.to_string(), arc);
        self.load_order.write().push(name.to_string());

        Ok(())
    }

    /// Load a plugin from a file
    pub fn load_from_file(
        &self,
        name: &str,
        path: &Path,
        config: WasmInstanceConfig,
    ) -> KernelResult<()> {
        let wasm_bytes = std::fs::read(path).map_err(|e| KernelError::Plugin {
            message: format!("failed to read WASM file: {}", e),
        })?;
        self.load(name, &wasm_bytes, config)
    }

    /// Unload a plugin
    pub fn unload(&self, name: &str) -> KernelResult<()> {
        let mut plugins = self.plugins.write();

        if !plugins.contains_key(name) {
            return Err(KernelError::Plugin {
                message: format!("plugin '{}' not found", name),
            });
        }

        // Mark as unloading
        if let Some(plugin) = plugins.get(name) {
            *plugin.state.write() = WasmPluginState::Unloading;
        }

        // Remove from registry
        plugins.remove(name);
        self.load_order.write().retain(|n| n != name);

        Ok(())
    }

    /// Get a plugin by name
    pub fn get(&self, name: &str) -> Option<Arc<WasmPluginInstance>> {
        self.plugins.read().get(name).cloned()
    }

    /// Call a function on a plugin
    pub fn call(
        &self,
        plugin_name: &str,
        func_name: &str,
        args: &[WasmValue],
    ) -> KernelResult<Vec<WasmValue>> {
        let plugin = self.get(plugin_name).ok_or_else(|| KernelError::Plugin {
            message: format!("plugin '{}' not found", plugin_name),
        })?;

        self.total_calls.fetch_add(1, Ordering::Relaxed);

        match plugin.call(func_name, args) {
            Ok(result) => Ok(result),
            Err(e) => {
                self.total_traps.fetch_add(1, Ordering::Relaxed);
                Err(e)
            }
        }
    }

    /// List all registered plugins
    pub fn list(&self) -> Vec<String> {
        self.load_order.read().clone()
    }

    /// Get plugin count
    pub fn count(&self) -> usize {
        self.plugins.read().len()
    }

    /// Get global statistics
    pub fn global_stats(&self) -> (u64, u64) {
        (
            self.total_calls.load(Ordering::Relaxed),
            self.total_traps.load(Ordering::Relaxed),
        )
    }

    /// Shutdown all plugins in reverse load order
    pub fn shutdown_all(&self) -> KernelResult<()> {
        let order = self.load_order.read().clone();
        for name in order.iter().rev() {
            if let Err(e) = self.unload(name) {
                // Log error but continue shutdown
                eprintln!("warning: failed to unload plugin {}: {}", name, e);
            }
        }
        Ok(())
    }
}

// ============================================================================
// WASM Observability Plugin Wrapper
// ============================================================================

/// Wrapper to use a WASM plugin as an ObservabilityExtension
pub struct WasmObservabilityPlugin {
    /// Underlying WASM instance
    instance: Arc<WasmPluginInstance>,
}

impl WasmObservabilityPlugin {
    /// Create a new wrapper from a WASM instance
    pub fn new(instance: Arc<WasmPluginInstance>) -> Self {
        Self { instance }
    }
}

impl Extension for WasmObservabilityPlugin {
    fn info(&self) -> ExtensionInfo {
        self.instance.info.clone()
    }

    fn as_any(&self) -> &dyn Any {
        self
    }

    fn as_any_mut(&mut self) -> &mut dyn Any {
        self
    }
}

impl ObservabilityExtension for WasmObservabilityPlugin {
    fn counter_inc(&self, name: &str, value: u64, labels: &[(&str, &str)]) {
        // Serialize labels for WASM
        let _ = self.instance.call(
            "counter_inc",
            &[
                WasmValue::I64(name.as_ptr() as i64),
                WasmValue::I32(name.len() as i32),
                WasmValue::I64(value as i64),
                WasmValue::I32(labels.len() as i32),
            ],
        );
    }

    fn gauge_set(&self, name: &str, value: f64, labels: &[(&str, &str)]) {
        let _ = self.instance.call(
            "gauge_set",
            &[
                WasmValue::I64(name.as_ptr() as i64),
                WasmValue::I32(name.len() as i32),
                WasmValue::F64(value),
                WasmValue::I32(labels.len() as i32),
            ],
        );
    }

    fn histogram_observe(&self, name: &str, value: f64, labels: &[(&str, &str)]) {
        let _ = self.instance.call(
            "histogram_observe",
            &[
                WasmValue::I64(name.as_ptr() as i64),
                WasmValue::I32(name.len() as i32),
                WasmValue::F64(value),
                WasmValue::I32(labels.len() as i32),
            ],
        );
    }

    fn span_start(&self, name: &str, parent: Option<u64>) -> u64 {
        match self.instance.call(
            "span_start",
            &[
                WasmValue::I64(name.as_ptr() as i64),
                WasmValue::I32(name.len() as i32),
                WasmValue::I64(parent.unwrap_or(0) as i64),
            ],
        ) {
            Ok(results) => results.first().and_then(|v| v.as_i64()).unwrap_or(0) as u64,
            Err(_) => 0,
        }
    }

    fn span_end(&self, span_id: u64) {
        let _ = self
            .instance
            .call("span_end", &[WasmValue::I64(span_id as i64)]);
    }

    fn span_event(&self, span_id: u64, name: &str, _attributes: &[(&str, &str)]) {
        let _ = self.instance.call(
            "span_event",
            &[
                WasmValue::I64(span_id as i64),
                WasmValue::I64(name.as_ptr() as i64),
                WasmValue::I32(name.len() as i32),
            ],
        );
    }
}

// ============================================================================
// Tests
// ============================================================================

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

    #[test]
    fn test_capabilities_default() {
        let caps = WasmPluginCapabilities::default();
        assert_eq!(caps.memory_limit_bytes, 16 * 1024 * 1024);
        assert_eq!(caps.fuel_limit, 1_000_000);
        assert!(!caps.can_read("any_table"));
        assert!(!caps.can_write("any_table"));
    }

    #[test]
    fn test_capabilities_read_patterns() {
        let caps = WasmPluginCapabilities {
            can_read_table: vec!["users".to_string(), "logs_*".to_string()],
            ..Default::default()
        };

        assert!(caps.can_read("users"));
        assert!(caps.can_read("logs_2024"));
        assert!(caps.can_read("logs_"));
        assert!(!caps.can_read("orders"));
    }

    #[test]
    fn test_capabilities_wildcard() {
        let caps = WasmPluginCapabilities {
            can_read_table: vec!["*".to_string()],
            ..Default::default()
        };

        assert!(caps.can_read("any_table"));
        assert!(caps.can_read("another_table"));
    }

    #[test]
    fn test_wasm_instance_creation() {
        let config = WasmInstanceConfig::default();
        let instance = WasmPluginInstance::new("test_plugin", b"fake wasm bytes", config).unwrap();

        assert_eq!(instance.name(), "test_plugin");
        assert_eq!(instance.state(), WasmPluginState::Loading);

        instance.init().unwrap();
        assert_eq!(instance.state(), WasmPluginState::Ready);
    }

    #[test]
    fn test_wasm_instance_call() {
        let config = WasmInstanceConfig::default();
        let instance = WasmPluginInstance::new("test_plugin", b"fake wasm bytes", config).unwrap();
        instance.init().unwrap();

        let result = instance.call("on_insert", &[WasmValue::I32(42)]).unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0], WasmValue::I32(0));

        let stats = instance.stats();
        assert_eq!(stats.total_calls, 1);
        assert!(stats.total_fuel_consumed > 0);
    }

    #[test]
    fn test_wasm_registry() {
        let registry = WasmPluginRegistry::new();

        // Load a plugin
        registry
            .load("plugin1", b"fake wasm", WasmInstanceConfig::default())
            .unwrap();
        assert_eq!(registry.count(), 1);

        // Call the plugin
        let result = registry.call("plugin1", "on_insert", &[]).unwrap();
        assert_eq!(result[0], WasmValue::I32(0));

        // Check stats
        let (calls, traps) = registry.global_stats();
        assert_eq!(calls, 1);
        assert_eq!(traps, 0);

        // Unload
        registry.unload("plugin1").unwrap();
        assert_eq!(registry.count(), 0);
    }

    #[test]
    fn test_wasm_registry_duplicate() {
        let registry = WasmPluginRegistry::new();

        registry
            .load("plugin1", b"fake wasm", WasmInstanceConfig::default())
            .unwrap();

        let result = registry.load("plugin1", b"fake wasm", WasmInstanceConfig::default());
        assert!(result.is_err());
    }

    #[test]
    fn test_wasm_value_conversions() {
        let v = WasmValue::I32(42);
        assert_eq!(v.as_i32(), Some(42));
        assert_eq!(v.as_i64(), None);

        let v = WasmValue::F64(2.5);
        assert_eq!(v.as_f64(), Some(2.5));
        assert_eq!(v.as_f32(), None);
    }

    #[test]
    fn test_fuel_exhaustion() {
        let config = WasmInstanceConfig {
            capabilities: WasmPluginCapabilities {
                fuel_limit: 100, // Very low fuel
                ..Default::default()
            },
            enable_fuel: true,
            ..Default::default()
        };

        let instance = WasmPluginInstance::new("test", b"fake wasm", config).unwrap();
        instance.init().unwrap();

        // First call succeeds
        let _ = instance.call("on_insert", &[]);

        // After enough calls, fuel exhausts
        // (Each call consumes 100+ fuel, so second call should fail)
        let result = instance.call("on_insert", &[]);
        assert!(result.is_err());
    }

    #[test]
    fn test_refuel() {
        let config = WasmInstanceConfig {
            capabilities: WasmPluginCapabilities {
                fuel_limit: 150,
                ..Default::default()
            },
            enable_fuel: true,
            ..Default::default()
        };

        let instance = WasmPluginInstance::new("test", b"fake wasm", config).unwrap();
        instance.init().unwrap();

        // Use up fuel
        let _ = instance.call("on_insert", &[]);

        // Refuel
        instance.refuel();

        // Now should work again
        let result = instance.call("on_insert", &[]);
        assert!(result.is_ok());
    }

    #[test]
    fn test_observability_wrapper() {
        let config = WasmInstanceConfig::default();
        let instance =
            Arc::new(WasmPluginInstance::new("obs_plugin", b"fake wasm", config).unwrap());
        instance.init().unwrap();

        let wrapper = WasmObservabilityPlugin::new(instance.clone());

        // These should not panic (they're fire-and-forget)
        wrapper.counter_inc("test_counter", 1, &[]);
        wrapper.gauge_set("test_gauge", 42.0, &[]);
        wrapper.histogram_observe("test_histogram", 0.5, &[]);

        let span = wrapper.span_start("test_span", None);
        wrapper.span_event(span, "event1", &[]);
        wrapper.span_end(span);
    }
}