oxirs_stream/

wasm_edge_processor.rs

//! # WebAssembly Edge Computing Processor
//!
//! Ultra-low latency edge processing using WebAssembly for distributed streaming.
//! Enables hot-swappable processing plugins and edge-cloud hybrid architectures.

use crate::error::{StreamError, StreamResult};
use crate::StreamEvent;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{debug, error, info, warn};
// Note: wasmparser's `WasmFeatures` is aliased to avoid colliding with this
// module's own `WasmFeatures` (the runtime-config type used throughout this
// file), which has an unrelated, simpler shape.
use ed25519_dalek::{Signature, Verifier, VerifyingKey};
use wasmparser::{Validator, WasmFeatures as WasmParserFeatures};
// SECURITY WARNING: RSA crate (v0.9.10) has CVE-2023-49092 - Timing attack vulnerability
// No patch available as of 2026-02-09. Consider migrating to constant-time alternatives.
// Current usage: WASM module signature verification (non-critical timing path)
// TODO: Evaluate migration to RustCrypto's newer RSA implementation when available
// Gated behind `wasm-rsa` (opt-in, RUSTSEC-2023-0071) per the `rsa` dependency
// declaration in Cargo.toml.
#[cfg(feature = "wasm-rsa")]
// `.verify()` on the RSA `VerifyingKey` resolves through the `signature::Verifier`
// trait, but the two crates now disagree on its version: ed25519-dalek 3.0
// re-exports signature 3.0, while `rsa` 0.9.10 still implements signature 2.2.
// The module-level `ed25519_dalek::Verifier` import therefore no longer covers
// this call, so the RSA verify path brings in its own `rsa::signature::Verifier`
// locally (see `verify_rsa_signature`).
use rsa::{pkcs1v15::VerifyingKey as RsaVerifyingKey, RsaPublicKey};

/// WebAssembly edge processor for distributed streaming
pub struct WasmEdgeProcessor {
    pub id: String,
    pub runtime: WasmRuntime,
    pub modules: Arc<RwLock<HashMap<String, WasmModule>>>,
    pub execution_context: WasmExecutionContext,
    pub resource_manager: WasmResourceManager,
    pub security_manager: WasmSecurityManager,
}

/// WebAssembly runtime configuration
#[derive(Debug, Clone)]
pub struct WasmRuntime {
    pub engine: WasmEngine,
    pub memory_limit: usize,
    pub fuel_limit: u64,
    pub timeout: std::time::Duration,
    pub optimization_level: OptimizationLevel,
    pub features: WasmFeatures,
}

/// WebAssembly engine types
#[derive(Debug, Clone)]
pub enum WasmEngine {
    Wasmtime { config: WasmtimeConfig },
    Wasmer { compiler: WasmerCompiler },
    Wasm3 { stack_size: usize },
    Browser { worker_pool_size: usize },
}

/// Wasmtime-specific configuration
#[derive(Debug, Clone)]
pub struct WasmtimeConfig {
    pub cranelift_opt_level: CraneliftOptLevel,
    pub enable_parallel_compilation: bool,
    pub memory_init_cow: bool,
    pub generate_address_map: bool,
}

/// Cranelift optimization levels
#[derive(Debug, Clone)]
pub enum CraneliftOptLevel {
    None,
    Speed,
    SpeedAndSize,
}

/// Wasmer compiler backends
#[derive(Debug, Clone)]
pub enum WasmerCompiler {
    Cranelift,
    LLVM,
    Singlepass,
}

/// WebAssembly optimization levels
#[derive(Debug, Clone)]
pub enum OptimizationLevel {
    O0, // No optimization
    O1, // Basic optimization
    O2, // Full optimization
    O3, // Aggressive optimization
    Os, // Size optimization
    Oz, // Aggressive size optimization
}

/// WebAssembly features
#[derive(Debug, Clone)]
pub struct WasmFeatures {
    pub simd: bool,
    pub threads: bool,
    pub tail_call: bool,
    pub multi_value: bool,
    pub reference_types: bool,
    pub bulk_memory: bool,
    pub sign_extension: bool,
    pub saturating_float_to_int: bool,
}

/// WebAssembly module representation
#[derive(Debug, Clone)]
pub struct WasmModule {
    pub id: String,
    pub name: String,
    pub version: String,
    pub bytecode: Vec<u8>,
    pub metadata: WasmModuleMetadata,
    pub capabilities: WasmCapabilities,
    pub resource_requirements: ResourceRequirements,
    pub security_policy: SecurityPolicy,
}

/// Module metadata
#[derive(Debug, Clone)]
pub struct WasmModuleMetadata {
    pub author: String,
    pub description: String,
    pub created_at: chrono::DateTime<chrono::Utc>,
    pub checksum: String,
    pub signature: Option<DigitalSignature>,
    pub license: String,
    pub tags: Vec<String>,
}

/// Module capabilities
#[derive(Debug, Clone)]
pub struct WasmCapabilities {
    pub input_formats: Vec<DataFormat>,
    pub output_formats: Vec<DataFormat>,
    pub processing_types: Vec<ProcessingType>,
    pub supported_events: Vec<StreamEventType>,
    pub exports: Vec<WasmExport>,
    pub imports: Vec<WasmImport>,
}

/// Data formats supported by modules
#[derive(Debug, Clone, PartialEq)]
pub enum DataFormat {
    RdfTurtle,
    RdfXml,
    JsonLd,
    NTriples,
    NQuads,
    Json,
    MessagePack,
    Avro,
    Protobuf,
    Custom(String),
}

/// Processing types
#[derive(Debug, Clone, PartialEq)]
pub enum ProcessingType {
    Filter,
    Transform,
    Aggregate,
    Join,
    Validate,
    Enrich,
    Compress,
    Encrypt,
    Custom(String),
}

/// Stream event types for capability matching
#[derive(Debug, Clone, PartialEq)]
pub enum StreamEventType {
    TripleAdded,
    TripleRemoved,
    QuadAdded,
    QuadRemoved,
    GraphCreated,
    GraphCleared,
    SparqlUpdate,
    TransactionBegin,
    TransactionCommit,
    SchemaChanged,
    Heartbeat,
    Custom(String),
}

/// WebAssembly export definitions
#[derive(Debug, Clone)]
pub struct WasmExport {
    pub name: String,
    pub export_type: WasmExportType,
    pub signature: FunctionSignature,
}

/// WebAssembly import definitions
#[derive(Debug, Clone)]
pub struct WasmImport {
    pub module: String,
    pub name: String,
    pub import_type: WasmImportType,
}

/// Export types
#[derive(Debug, Clone)]
pub enum WasmExportType {
    Function,
    Memory,
    Global,
    Table,
}

/// Import types
#[derive(Debug, Clone)]
pub enum WasmImportType {
    Function(FunctionSignature),
    Memory(MemoryType),
    Global(GlobalType),
    Table(TableType),
}

/// Function signature
#[derive(Debug, Clone)]
pub struct FunctionSignature {
    pub parameters: Vec<WasmValueType>,
    pub results: Vec<WasmValueType>,
}

/// WebAssembly value types
#[derive(Debug, Clone, PartialEq)]
pub enum WasmValueType {
    I32,
    I64,
    F32,
    F64,
    V128, // SIMD
    FuncRef,
    ExternRef,
}

/// Memory type
#[derive(Debug, Clone)]
pub struct MemoryType {
    pub minimum: u32,
    pub maximum: Option<u32>,
    pub shared: bool,
}

/// Global type
#[derive(Debug, Clone)]
pub struct GlobalType {
    pub value_type: WasmValueType,
    pub mutable: bool,
}

/// Table type
#[derive(Debug, Clone)]
pub struct TableType {
    pub element_type: WasmValueType,
    pub minimum: u32,
    pub maximum: Option<u32>,
}

/// Resource requirements
#[derive(Debug, Clone)]
pub struct ResourceRequirements {
    pub memory_mb: u32,
    pub cpu_cores: f32,
    pub disk_mb: u32,
    pub network_mbps: u32,
    pub execution_time_ms: u32,
    pub fuel_consumption: u64,
}

/// Security policy
#[derive(Debug, Clone)]
pub struct SecurityPolicy {
    pub trusted: bool,
    pub sandbox_level: SandboxLevel,
    pub allowed_hosts: Vec<String>,
    pub allowed_syscalls: Vec<String>,
    pub resource_limits: ResourceLimits,
    pub network_access: NetworkAccess,
}

/// Sandbox security levels
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SandboxLevel {
    None,
    Basic,
    Strict,
    Paranoid,
}

/// Resource limits for security
#[derive(Debug, Clone)]
pub struct ResourceLimits {
    pub max_memory: usize,
    pub max_fuel: u64,
    pub max_stack_depth: u32,
    pub max_execution_time: std::time::Duration,
}

/// Network access permissions
#[derive(Debug, Clone)]
pub enum NetworkAccess {
    None,
    LocalOnly,
    Whitelist(Vec<String>),
    Full,
}

/// Digital signature for module verification
#[derive(Debug, Clone)]
pub struct DigitalSignature {
    pub algorithm: SignatureAlgorithm,
    pub signature: Vec<u8>,
    pub public_key: Vec<u8>,
    pub certificate_chain: Option<Vec<Vec<u8>>>,
}

/// Signature algorithms
#[derive(Debug, Clone)]
pub enum SignatureAlgorithm {
    Ed25519,
    ECDSA,
    RSA,
    Falcon,
    Dilithium,
}

/// Execution context for WebAssembly modules
#[derive(Debug, Clone)]
pub struct WasmExecutionContext {
    pub node_id: String,
    pub location: EdgeLocation,
    pub compute_tier: ComputeTier,
    pub network_conditions: NetworkConditions,
    pub available_resources: AvailableResources,
}

/// Edge computing location
#[derive(Debug, Clone)]
pub struct EdgeLocation {
    pub latitude: f64,
    pub longitude: f64,
    pub region: String,
    pub zone: String,
    pub provider: String,
}

/// Compute tier for edge deployment
#[derive(Debug, Clone)]
pub enum ComputeTier {
    Device,   // IoT devices, smartphones
    Edge,     // Edge servers, 5G edge
    Regional, // Regional data centers
    Cloud,    // Central cloud
    Hybrid,   // Distributed across tiers
}

/// Current network conditions
#[derive(Debug, Clone)]
pub struct NetworkConditions {
    pub bandwidth_mbps: f64,
    pub latency_ms: f64,
    pub packet_loss: f64,
    pub jitter_ms: f64,
    pub connection_type: ConnectionType,
}

/// Connection types
#[derive(Debug, Clone)]
pub enum ConnectionType {
    WiFi,
    Ethernet,
    LTE,
    FiveG,
    Satellite,
    Bluetooth,
    LoRaWAN,
}

/// Available computing resources
#[derive(Debug, Clone)]
pub struct AvailableResources {
    pub cpu_cores: u32,
    pub memory_mb: u32,
    pub storage_gb: u32,
    pub gpu_available: bool,
    pub specialized_hardware: Vec<SpecializedHardware>,
}

/// Specialized hardware types
#[derive(Debug, Clone)]
pub enum SpecializedHardware {
    TPU,
    FPGA,
    NPU,
    VPU,
    QuantumProcessor,
    Custom(String),
}

/// Resource manager for WebAssembly execution
pub struct WasmResourceManager {
    pub memory_pools: HashMap<String, MemoryPool>,
    pub cpu_scheduler: CpuScheduler,
    pub fuel_monitor: FuelMonitor,
    pub bandwidth_controller: BandwidthController,
}

/// Memory pool for efficient allocation
#[derive(Debug, Clone)]
pub struct MemoryPool {
    pub pool_id: String,
    pub total_size: usize,
    pub used_size: usize,
    pub allocation_strategy: AllocationStrategy,
    pub fragmentation_level: f64,
}

/// Memory allocation strategies
#[derive(Debug, Clone)]
pub enum AllocationStrategy {
    FirstFit,
    BestFit,
    WorstFit,
    NextFit,
    BuddySystem,
    SlabAllocator,
}

/// CPU scheduler for module execution
#[derive(Debug, Clone)]
pub struct CpuScheduler {
    pub algorithm: SchedulingAlgorithm,
    pub time_slice_ms: u32,
    pub priority_levels: u32,
    pub load_balancing: bool,
}

/// Scheduling algorithms
#[derive(Debug, Clone)]
pub enum SchedulingAlgorithm {
    RoundRobin,
    PriorityBased,
    WeightedFairQueuing,
    EarliestDeadlineFirst,
    ProportionalShare,
}

/// Fuel monitoring for execution limits
#[derive(Debug, Clone)]
pub struct FuelMonitor {
    pub total_fuel: u64,
    pub consumed_fuel: u64,
    pub fuel_rate: f64,
    pub low_fuel_threshold: u64,
}

/// Bandwidth controller for network operations
#[derive(Debug, Clone)]
pub struct BandwidthController {
    pub total_bandwidth: f64,
    pub allocated_bandwidth: f64,
    pub rate_limiting: bool,
    pub qos_policies: Vec<QosPolicy>,
}

/// Quality of Service policies
#[derive(Debug, Clone)]
pub struct QosPolicy {
    pub priority: QosPriority,
    pub bandwidth_guarantee: f64,
    pub latency_target: std::time::Duration,
    pub packet_loss_target: f64,
}

/// QoS priority levels
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum QosPriority {
    Critical,
    High,
    Normal,
    Low,
    Background,
}

/// Security manager for WebAssembly execution
pub struct WasmSecurityManager {
    pub sandbox_engine: SandboxEngine,
    pub code_verifier: CodeVerifier,
    pub access_controller: AccessController,
    pub threat_detector: ThreatDetector,
}

/// Sandbox engine for isolation
#[derive(Debug, Clone)]
pub struct SandboxEngine {
    pub isolation_level: IsolationLevel,
    pub syscall_filter: SyscallFilter,
    pub network_isolation: NetworkIsolation,
    pub filesystem_isolation: FilesystemIsolation,
}

/// Isolation levels
#[derive(Debug, Clone)]
pub enum IsolationLevel {
    Process,
    Container,
    Hypervisor,
    Hardware,
}

/// System call filtering
#[derive(Debug, Clone)]
pub struct SyscallFilter {
    pub allowed_syscalls: Vec<String>,
    pub blocked_syscalls: Vec<String>,
    pub audit_mode: bool,
}

/// Network isolation mechanisms
#[derive(Debug, Clone)]
pub struct NetworkIsolation {
    pub virtual_network: bool,
    pub firewall_rules: Vec<FirewallRule>,
    pub proxy_mode: bool,
}

/// Firewall rule
#[derive(Debug, Clone)]
pub struct FirewallRule {
    pub direction: TrafficDirection,
    pub protocol: NetworkProtocol,
    pub source: NetworkEndpoint,
    pub destination: NetworkEndpoint,
    pub action: FirewallAction,
}

/// Traffic direction
#[derive(Debug, Clone)]
pub enum TrafficDirection {
    Inbound,
    Outbound,
    Bidirectional,
}

/// Network protocols
#[derive(Debug, Clone)]
pub enum NetworkProtocol {
    TCP,
    UDP,
    ICMP,
    HTTP,
    HTTPS,
    WebSocket,
    Custom(String),
}

/// Network endpoint
#[derive(Debug, Clone)]
pub struct NetworkEndpoint {
    pub address: String,
    pub port: Option<u16>,
    pub port_range: Option<(u16, u16)>,
}

/// Firewall actions
#[derive(Debug, Clone)]
pub enum FirewallAction {
    Allow,
    Deny,
    Log,
    RateLimit(u32),
}

/// Filesystem isolation
#[derive(Debug, Clone)]
pub struct FilesystemIsolation {
    pub chroot_enabled: bool,
    pub readonly_filesystem: bool,
    pub allowed_paths: Vec<String>,
    pub temp_directory: Option<String>,
}

/// Code verifier for module validation
#[derive(Debug, Clone)]
pub struct CodeVerifier {
    pub signature_verification: bool,
    pub static_analysis: bool,
    pub dynamic_analysis: bool,
    pub reputation_checking: bool,
}

/// Access controller for permissions
#[derive(Debug, Clone)]
pub struct AccessController {
    pub permission_model: PermissionModel,
    pub capability_based: bool,
    pub role_based: bool,
    pub attribute_based: bool,
}

/// Permission models
#[derive(Debug, Clone)]
pub enum PermissionModel {
    Discretionary,
    Mandatory,
    RoleBased,
    AttributeBased,
    CapabilityBased,
}

/// Threat detector for security monitoring
#[derive(Debug, Clone)]
pub struct ThreatDetector {
    pub anomaly_detection: bool,
    pub behavioral_analysis: bool,
    pub signature_detection: bool,
    pub ml_detection: bool,
}

impl WasmEdgeProcessor {
    /// Create a new WebAssembly edge processor
    pub fn new(id: String, runtime: WasmRuntime) -> Self {
        Self {
            id,
            runtime,
            modules: Arc::new(RwLock::new(HashMap::new())),
            execution_context: WasmExecutionContext::default(),
            resource_manager: WasmResourceManager::default(),
            security_manager: WasmSecurityManager::default(),
        }
    }

    /// Load a WebAssembly module
    pub async fn load_module(&self, module: WasmModule) -> StreamResult<()> {
        // Verify module security
        self.verify_module_security(&module).await?;

        // Check resource requirements
        self.check_resource_requirements(&module.resource_requirements)
            .await?;

        // Validate module bytecode
        self.validate_module_bytecode(&module.bytecode).await?;

        let mut modules = self.modules.write().await;
        modules.insert(module.id.clone(), module.clone());

        info!("Loaded WebAssembly module: {} ({})", module.name, module.id);
        Ok(())
    }

    /// Process stream event using WebAssembly module
    pub async fn process_event(
        &self,
        event: StreamEvent,
        module_id: &str,
        function_name: &str,
    ) -> StreamResult<Vec<StreamEvent>> {
        let modules = self.modules.read().await;
        let module = modules
            .get(module_id)
            .ok_or_else(|| StreamError::InvalidOperation("Module not found".to_string()))?;

        // Check if module can handle this event type
        let event_type = self.event_to_type(&event);
        if !module.capabilities.supported_events.contains(&event_type) {
            return Err(StreamError::InvalidOperation(
                "Module does not support this event type".to_string(),
            ));
        }

        // Serialize event for WASM processing
        let event_bytes = self.serialize_event(&event)?;

        // Execute WASM function (simulated)
        let result_bytes = self
            .execute_wasm_function(&module.bytecode, function_name, &event_bytes)
            .await?;

        // Deserialize results
        let result_events = self.deserialize_events(&result_bytes)?;

        debug!(
            "Processed event using module {} function {}: {} -> {} events",
            module_id,
            function_name,
            1,
            result_events.len()
        );

        Ok(result_events)
    }

    /// Verify module security before loading
    async fn verify_module_security(&self, module: &WasmModule) -> StreamResult<()> {
        // Verify digital signature if present
        if let Some(signature) = &module.metadata.signature {
            self.verify_digital_signature(&module.bytecode, signature)
                .await?;
        }

        // Check security policy compliance
        if module.security_policy.sandbox_level == SandboxLevel::None {
            warn!("Module {} has no sandboxing - security risk", module.id);
        }

        // Validate against security policies
        if !module.security_policy.trusted {
            return Err(StreamError::SecurityViolation(
                "Untrusted module not allowed".to_string(),
            ));
        }

        Ok(())
    }

    /// Check if sufficient resources are available
    async fn check_resource_requirements(
        &self,
        requirements: &ResourceRequirements,
    ) -> StreamResult<()> {
        let available = &self.execution_context.available_resources;

        if requirements.memory_mb > available.memory_mb {
            return Err(StreamError::InsufficientResources(format!(
                "Insufficient memory: need {} MB, have {} MB",
                requirements.memory_mb, available.memory_mb
            )));
        }

        if requirements.cpu_cores > available.cpu_cores as f32 {
            return Err(StreamError::InsufficientResources(format!(
                "Insufficient CPU: need {} cores, have {} cores",
                requirements.cpu_cores, available.cpu_cores
            )));
        }

        Ok(())
    }

    /// Validate WebAssembly module bytecode
    async fn validate_module_bytecode(&self, bytecode: &[u8]) -> StreamResult<()> {
        // Basic WASM magic number check
        if bytecode.len() < 8 {
            return Err(StreamError::InvalidModule("Bytecode too short".to_string()));
        }

        let magic = &bytecode[0..4];
        let version = &bytecode[4..8];

        if magic != b"\x00asm" {
            return Err(StreamError::InvalidModule(
                "Invalid WASM magic number".to_string(),
            ));
        }

        if version != [0x01, 0x00, 0x00, 0x00] {
            return Err(StreamError::InvalidModule(
                "Unsupported WASM version".to_string(),
            ));
        }

        // Enhanced validation using wasmparser. `WasmFeatures` is a bitflags
        // type (wasmparser 0.252+), so build it flag-by-flag rather than via
        // struct-literal field initialization.
        let mut features = WasmParserFeatures::empty();
        features.set(WasmParserFeatures::MUTABLE_GLOBAL, true);
        features.set(WasmParserFeatures::SATURATING_FLOAT_TO_INT, true);
        features.set(WasmParserFeatures::SIGN_EXTENSION, true);
        features.set(WasmParserFeatures::REFERENCE_TYPES, true);
        features.set(WasmParserFeatures::MULTI_VALUE, true);
        features.set(WasmParserFeatures::BULK_MEMORY, true);
        features.set(WasmParserFeatures::SIMD, true);
        features.set(WasmParserFeatures::RELAXED_SIMD, false);
        features.set(WasmParserFeatures::THREADS, false);
        features.set(WasmParserFeatures::SHARED_EVERYTHING_THREADS, false);
        features.set(WasmParserFeatures::TAIL_CALL, false);
        features.set(WasmParserFeatures::FLOATS, true);
        features.set(WasmParserFeatures::MULTI_MEMORY, false);
        features.set(WasmParserFeatures::EXCEPTIONS, false);
        features.set(WasmParserFeatures::MEMORY64, false);
        features.set(WasmParserFeatures::EXTENDED_CONST, false);
        features.set(WasmParserFeatures::COMPONENT_MODEL, false);
        features.set(WasmParserFeatures::FUNCTION_REFERENCES, false);
        features.set(WasmParserFeatures::MEMORY_CONTROL, false);
        features.set(WasmParserFeatures::GC, false);
        features.set(WasmParserFeatures::CUSTOM_PAGE_SIZES, false);
        features.set(WasmParserFeatures::WIDE_ARITHMETIC, false);
        let mut validator = Validator::new_with_features(features);

        match validator.validate_all(bytecode) {
            Ok(_) => {
                debug!("WASM module validation successful");
                Ok(())
            }
            Err(e) => {
                error!("WASM module validation failed: {}", e);
                Err(StreamError::InvalidModule(format!(
                    "Validation failed: {}",
                    e
                )))
            }
        }
    }

    /// Convert stream event to event type
    fn event_to_type(&self, event: &StreamEvent) -> StreamEventType {
        match event {
            StreamEvent::TripleAdded { .. } => StreamEventType::TripleAdded,
            StreamEvent::TripleRemoved { .. } => StreamEventType::TripleRemoved,
            StreamEvent::QuadAdded { .. } => StreamEventType::QuadAdded,
            StreamEvent::QuadRemoved { .. } => StreamEventType::QuadRemoved,
            StreamEvent::GraphCreated { .. } => StreamEventType::GraphCreated,
            StreamEvent::GraphCleared { .. } => StreamEventType::GraphCleared,
            StreamEvent::SparqlUpdate { .. } => StreamEventType::SparqlUpdate,
            StreamEvent::TransactionBegin { .. } => StreamEventType::TransactionBegin,
            StreamEvent::TransactionCommit { .. } => StreamEventType::TransactionCommit,
            StreamEvent::SchemaChanged { .. } => StreamEventType::SchemaChanged,
            StreamEvent::Heartbeat { .. } => StreamEventType::Heartbeat,
            _ => StreamEventType::Custom("unknown".to_string()),
        }
    }

    /// Serialize stream event for WASM processing
    fn serialize_event(&self, event: &StreamEvent) -> StreamResult<Vec<u8>> {
        serde_json::to_vec(event).map_err(|e| StreamError::Serialization(e.to_string()))
    }

    /// Deserialize events from WASM output
    fn deserialize_events(&self, bytes: &[u8]) -> StreamResult<Vec<StreamEvent>> {
        serde_json::from_slice(bytes).map_err(|e| StreamError::Deserialization(e.to_string()))
    }

    /// Execute a WebAssembly function.
    ///
    /// `WasmRuntime::engine` (see [`WasmEngine`]) only describes *which*
    /// runtime and configuration a module would run under; this type does not
    /// embed an actual execution engine instance (no `wasmtime::Engine`
    /// handle, no Wasmer store, no wasm3 VM). There is therefore no way to
    /// genuinely execute WASM bytecode here. Rather than fabricating a result
    /// (e.g. echoing the input back as if it were the function's real
    /// output — which would silently corrupt callers relying on the
    /// transformation actually happening), this returns an explicit
    /// `UnsupportedOperation` error so failures are loud, not silent.
    ///
    /// Real WASM execution (via `wasmtime`, gated behind the crate's `wasm`
    /// feature) is implemented by the sibling `wasm_edge_computing` module,
    /// which does hold a live `wasmtime::Engine`; callers that need actual
    /// execution should use `wasm_edge_computing::WasmEdgeProcessor` instead.
    async fn execute_wasm_function(
        &self,
        _bytecode: &[u8],
        function_name: &str,
        _input: &[u8],
    ) -> StreamResult<Vec<u8>> {
        Err(StreamError::UnsupportedOperation(format!(
            "WASM function execution ('{function_name}') is not available: this processor's \
             WasmRuntime ({:?}) does not embed a real execution engine. Use \
             wasm_edge_computing::WasmEdgeProcessor (built on wasmtime, feature = \"wasm\") for \
             genuine WASM execution.",
            self.runtime.engine
        )))
    }

    /// Verify digital signature
    async fn verify_digital_signature(
        &self,
        data: &[u8],
        signature: &DigitalSignature,
    ) -> StreamResult<()> {
        debug!(
            "Verifying digital signature using {:?} algorithm",
            signature.algorithm
        );

        match signature.algorithm {
            SignatureAlgorithm::Ed25519 => self.verify_ed25519_signature(data, signature).await,
            SignatureAlgorithm::RSA => {
                #[cfg(feature = "wasm-rsa")]
                {
                    self.verify_rsa_signature(data, signature).await
                }
                #[cfg(not(feature = "wasm-rsa"))]
                {
                    Err(StreamError::UnsupportedOperation(
                        "RSA signature verification requires the `wasm-rsa` cargo feature \
                         (opt-in due to RUSTSEC-2023-0071)"
                            .to_string(),
                    ))
                }
            }
            SignatureAlgorithm::ECDSA => {
                warn!("ECDSA signature verification not yet implemented");
                Err(StreamError::UnsupportedOperation(
                    "ECDSA verification not implemented".to_string(),
                ))
            }
            SignatureAlgorithm::Falcon => {
                warn!("Falcon signature verification not yet implemented");
                Err(StreamError::UnsupportedOperation(
                    "Falcon verification not implemented".to_string(),
                ))
            }
            SignatureAlgorithm::Dilithium => {
                warn!("Dilithium signature verification not yet implemented");
                Err(StreamError::UnsupportedOperation(
                    "Dilithium verification not implemented".to_string(),
                ))
            }
        }
    }

    /// Verify Ed25519 signature
    async fn verify_ed25519_signature(
        &self,
        data: &[u8],
        signature: &DigitalSignature,
    ) -> StreamResult<()> {
        // Parse the public key
        if signature.public_key.len() != 32 {
            return Err(StreamError::InvalidSignature(
                "Invalid Ed25519 public key length".to_string(),
            ));
        }

        let public_key_bytes: [u8; 32] =
            signature.public_key.as_slice().try_into().map_err(|_| {
                StreamError::InvalidSignature("Failed to parse Ed25519 public key".to_string())
            })?;

        let verifying_key = VerifyingKey::from_bytes(&public_key_bytes).map_err(|e| {
            StreamError::InvalidSignature(format!("Invalid Ed25519 public key: {}", e))
        })?;

        // Parse the signature
        if signature.signature.len() != 64 {
            return Err(StreamError::InvalidSignature(
                "Invalid Ed25519 signature length".to_string(),
            ));
        }

        let signature_bytes: [u8; 64] =
            signature.signature.as_slice().try_into().map_err(|_| {
                StreamError::InvalidSignature("Failed to parse Ed25519 signature".to_string())
            })?;

        let sig = Signature::from_bytes(&signature_bytes);

        // Verify the signature
        match verifying_key.verify(data, &sig) {
            Ok(_) => {
                debug!("Ed25519 signature verification successful");
                Ok(())
            }
            Err(e) => {
                error!("Ed25519 signature verification failed: {}", e);
                Err(StreamError::InvalidSignature(
                    "Ed25519 signature verification failed".to_string(),
                ))
            }
        }
    }

    /// Verify RSA signature (opt-in via the `wasm-rsa` feature; RUSTSEC-2023-0071).
    #[cfg(feature = "wasm-rsa")]
    async fn verify_rsa_signature(
        &self,
        data: &[u8],
        signature: &DigitalSignature,
    ) -> StreamResult<()> {
        use rsa::pkcs1::DecodeRsaPublicKey;
        // `rsa` 0.9.10 implements signature 2.2's `Verifier`, whereas the
        // module-level `ed25519_dalek::Verifier` is signature 3.0's. Bring the
        // matching trait into scope (methods only, `as _` avoids clashing with
        // the ed25519 import) so `verifying_key.verify(..)` resolves below.
        use rsa::signature::Verifier as _;
        use sha2::{Digest, Sha256};

        // Parse the RSA public key from DER format
        let public_key = RsaPublicKey::from_pkcs1_der(&signature.public_key)
            .map_err(|e| StreamError::InvalidSignature(format!("Invalid RSA public key: {}", e)))?;

        // NOTE: use `rsa::sha2::Sha256` (re-exported by `rsa` via its `sha2`
        // feature), NOT the workspace `sha2` crate. The workspace pins
        // `sha2 = "0.11"` (digest 0.11) but `rsa` 0.9 builds on digest 0.10;
        // mixing them yields an unsatisfied `Digest`/`HashMarker` trait bound.
        let verifying_key = RsaVerifyingKey::<rsa::sha2::Sha256>::new(public_key);

        // Hash the data
        let mut hasher = Sha256::new();
        hasher.update(data);
        let hash = hasher.finalize();

        // Verify the signature
        match verifying_key.verify(
            &hash,
            &signature.signature.as_slice().try_into().map_err(|_| {
                StreamError::InvalidSignature("Invalid RSA signature format".to_string())
            })?,
        ) {
            Ok(_) => {
                debug!("RSA signature verification successful");
                Ok(())
            }
            Err(e) => {
                error!("RSA signature verification failed: {}", e);
                Err(StreamError::InvalidSignature(
                    "RSA signature verification failed".to_string(),
                ))
            }
        }
    }

    /// Deploy module to edge location
    pub async fn deploy_to_edge(
        &self,
        module_id: &str,
        target_location: EdgeLocation,
    ) -> StreamResult<String> {
        let modules = self.modules.read().await;
        let module = modules
            .get(module_id)
            .ok_or_else(|| StreamError::InvalidOperation("Module not found".to_string()))?;

        // Simulate edge deployment
        let deployment_id = uuid::Uuid::new_v4().to_string();

        info!(
            "Deployed module {} to edge location: {} (deployment: {})",
            module.name, target_location.region, deployment_id
        );

        Ok(deployment_id)
    }
}

impl Default for WasmExecutionContext {
    fn default() -> Self {
        Self {
            node_id: uuid::Uuid::new_v4().to_string(),
            location: EdgeLocation {
                latitude: 0.0,
                longitude: 0.0,
                region: "unknown".to_string(),
                zone: "unknown".to_string(),
                provider: "local".to_string(),
            },
            compute_tier: ComputeTier::Edge,
            network_conditions: NetworkConditions {
                bandwidth_mbps: 100.0,
                latency_ms: 10.0,
                packet_loss: 0.001,
                jitter_ms: 1.0,
                connection_type: ConnectionType::Ethernet,
            },
            available_resources: AvailableResources {
                cpu_cores: 4,
                memory_mb: 8192,
                storage_gb: 256,
                gpu_available: false,
                specialized_hardware: Vec::new(),
            },
        }
    }
}

impl Default for WasmResourceManager {
    fn default() -> Self {
        Self {
            memory_pools: HashMap::new(),
            cpu_scheduler: CpuScheduler {
                algorithm: SchedulingAlgorithm::RoundRobin,
                time_slice_ms: 10,
                priority_levels: 8,
                load_balancing: true,
            },
            fuel_monitor: FuelMonitor {
                total_fuel: 1_000_000,
                consumed_fuel: 0,
                fuel_rate: 1.0,
                low_fuel_threshold: 100_000,
            },
            bandwidth_controller: BandwidthController {
                total_bandwidth: 1000.0,
                allocated_bandwidth: 0.0,
                rate_limiting: true,
                qos_policies: Vec::new(),
            },
        }
    }
}

impl Default for WasmSecurityManager {
    fn default() -> Self {
        Self {
            sandbox_engine: SandboxEngine {
                isolation_level: IsolationLevel::Container,
                syscall_filter: SyscallFilter {
                    allowed_syscalls: vec!["read".to_string(), "write".to_string()],
                    blocked_syscalls: vec!["execve".to_string(), "fork".to_string()],
                    audit_mode: true,
                },
                network_isolation: NetworkIsolation {
                    virtual_network: true,
                    firewall_rules: Vec::new(),
                    proxy_mode: true,
                },
                filesystem_isolation: FilesystemIsolation {
                    chroot_enabled: true,
                    readonly_filesystem: true,
                    allowed_paths: vec![std::env::temp_dir().to_string_lossy().into_owned()],
                    temp_directory: Some(
                        std::env::temp_dir()
                            .join("oxirs-wasm")
                            .to_string_lossy()
                            .into_owned(),
                    ),
                },
            },
            code_verifier: CodeVerifier {
                signature_verification: true,
                static_analysis: true,
                dynamic_analysis: false,
                reputation_checking: true,
            },
            access_controller: AccessController {
                permission_model: PermissionModel::CapabilityBased,
                capability_based: true,
                role_based: true,
                attribute_based: false,
            },
            threat_detector: ThreatDetector {
                anomaly_detection: true,
                behavioral_analysis: true,
                signature_detection: true,
                ml_detection: false,
            },
        }
    }
}

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

    #[tokio::test]
    async fn test_wasm_processor_creation() {
        let runtime = WasmRuntime {
            engine: WasmEngine::Wasmtime {
                config: WasmtimeConfig {
                    cranelift_opt_level: CraneliftOptLevel::Speed,
                    enable_parallel_compilation: true,
                    memory_init_cow: true,
                    generate_address_map: false,
                },
            },
            memory_limit: 64 * 1024 * 1024, // 64MB
            fuel_limit: 1_000_000,
            timeout: std::time::Duration::from_secs(30),
            optimization_level: OptimizationLevel::O2,
            features: WasmFeatures {
                simd: true,
                threads: false,
                tail_call: false,
                multi_value: true,
                reference_types: true,
                bulk_memory: true,
                sign_extension: true,
                saturating_float_to_int: true,
            },
        };

        let processor = WasmEdgeProcessor::new("test_processor".to_string(), runtime);
        assert_eq!(processor.id, "test_processor");
    }

    #[tokio::test]
    async fn test_module_loading() {
        let processor = WasmEdgeProcessor::new(
            "test".to_string(),
            WasmRuntime {
                engine: WasmEngine::Wasm3 { stack_size: 1024 },
                memory_limit: 1024 * 1024,
                fuel_limit: 100_000,
                timeout: std::time::Duration::from_secs(10),
                optimization_level: OptimizationLevel::O1,
                features: WasmFeatures {
                    simd: false,
                    threads: false,
                    tail_call: false,
                    multi_value: false,
                    reference_types: false,
                    bulk_memory: false,
                    sign_extension: false,
                    saturating_float_to_int: false,
                },
            },
        );

        let module = WasmModule {
            id: "test_module".to_string(),
            name: "Test Module".to_string(),
            version: "1.0.0".to_string(),
            bytecode: b"\x00asm\x01\x00\x00\x00".to_vec(), // Minimal WASM header
            metadata: WasmModuleMetadata {
                author: "Test Author".to_string(),
                description: "Test module".to_string(),
                created_at: chrono::Utc::now(),
                checksum: "abc123".to_string(),
                signature: None,
                license: "MIT".to_string(),
                tags: vec!["test".to_string()],
            },
            capabilities: WasmCapabilities {
                input_formats: vec![DataFormat::Json],
                output_formats: vec![DataFormat::Json],
                processing_types: vec![ProcessingType::Filter],
                supported_events: vec![StreamEventType::TripleAdded],
                exports: Vec::new(),
                imports: Vec::new(),
            },
            resource_requirements: ResourceRequirements {
                memory_mb: 16,
                cpu_cores: 0.5,
                disk_mb: 1,
                network_mbps: 1,
                execution_time_ms: 100,
                fuel_consumption: 1000,
            },
            security_policy: SecurityPolicy {
                trusted: true,
                sandbox_level: SandboxLevel::Basic,
                allowed_hosts: Vec::new(),
                allowed_syscalls: Vec::new(),
                resource_limits: ResourceLimits {
                    max_memory: 1024 * 1024,
                    max_fuel: 10_000,
                    max_stack_depth: 1024,
                    max_execution_time: std::time::Duration::from_secs(1),
                },
                network_access: NetworkAccess::None,
            },
        };

        let result = processor.load_module(module).await;
        assert!(result.is_ok());
    }

    /// Regression test for wasm_edge_processor.rs:832 — `process_event` used to
    /// echo the serialized input event back as if it were genuine WASM
    /// function output, so `deserialize_events` happily "recovered" the
    /// original input as a fake successful transformation. This asserts the
    /// processor instead fails loudly with `UnsupportedOperation` because it
    /// has no embedded execution engine, so no caller can mistake "did
    /// nothing" for "ran the WASM function".
    #[tokio::test]
    async fn test_process_event_fails_loudly_without_real_wasm_runtime() {
        let processor = WasmEdgeProcessor::new(
            "no_engine".to_string(),
            WasmRuntime {
                engine: WasmEngine::Wasm3 { stack_size: 1024 },
                memory_limit: 1024 * 1024,
                fuel_limit: 100_000,
                timeout: std::time::Duration::from_secs(10),
                optimization_level: OptimizationLevel::O1,
                features: WasmFeatures {
                    simd: false,
                    threads: false,
                    tail_call: false,
                    multi_value: false,
                    reference_types: false,
                    bulk_memory: false,
                    sign_extension: false,
                    saturating_float_to_int: false,
                },
            },
        );

        let module = WasmModule {
            id: "echo_module".to_string(),
            name: "Echo Module".to_string(),
            version: "1.0.0".to_string(),
            bytecode: b"\x00asm\x01\x00\x00\x00".to_vec(),
            metadata: WasmModuleMetadata {
                author: "Test Author".to_string(),
                description: "Test module".to_string(),
                created_at: chrono::Utc::now(),
                checksum: "abc123".to_string(),
                signature: None,
                license: "MIT".to_string(),
                tags: vec!["test".to_string()],
            },
            capabilities: WasmCapabilities {
                input_formats: vec![DataFormat::Json],
                output_formats: vec![DataFormat::Json],
                processing_types: vec![ProcessingType::Filter],
                supported_events: vec![StreamEventType::TripleAdded],
                exports: Vec::new(),
                imports: Vec::new(),
            },
            resource_requirements: ResourceRequirements {
                memory_mb: 16,
                cpu_cores: 0.5,
                disk_mb: 1,
                network_mbps: 1,
                execution_time_ms: 100,
                fuel_consumption: 1000,
            },
            security_policy: SecurityPolicy {
                trusted: true,
                sandbox_level: SandboxLevel::Basic,
                allowed_hosts: Vec::new(),
                allowed_syscalls: Vec::new(),
                resource_limits: ResourceLimits {
                    max_memory: 1024 * 1024,
                    max_fuel: 10_000,
                    max_stack_depth: 1024,
                    max_execution_time: std::time::Duration::from_secs(1),
                },
                network_access: NetworkAccess::None,
            },
        };
        processor.load_module(module).await.unwrap();

        let event = StreamEvent::TripleAdded {
            subject: "http://test.org/s".to_string(),
            predicate: "http://test.org/p".to_string(),
            object: "\"o\"".to_string(),
            graph: None,
            metadata: crate::EventMetadata {
                event_id: uuid::Uuid::new_v4().to_string(),
                timestamp: chrono::Utc::now(),
                source: "test".to_string(),
                user: None,
                context: None,
                caused_by: None,
                version: "1.0".to_string(),
                properties: HashMap::new(),
                checksum: None,
            },
        };

        let result = processor
            .process_event(event, "echo_module", "transform")
            .await;
        assert!(
            result.is_err(),
            "process_event must not silently echo the input back as fake WASM output"
        );
        assert!(matches!(
            result.unwrap_err(),
            StreamError::UnsupportedOperation(_)
        ));
    }
}