//! WASM plugin runtime — loads and manages WASM lifecycle plugins.
//!
//! When compiled with the `wasmtime-runtime` feature, this module uses wasmtime
//! to compile WASM modules, instantiate them with registered host functions, and
//! manage their lifecycle. Without the feature, a stub runtime is provided.

use crate::adapter::WasmLifecycleAdapter;
use crate::sandbox::PluginSandbox;
use bext_plugin_api::types::{PluginManifest, WasmPluginPermissions};

/// Configuration for loading a single WASM plugin.
pub struct WasmPluginConfig {
    pub name: String,
    pub path: std::path::PathBuf,
    pub priority: u32,
    pub permissions: WasmPluginPermissions,
    pub config: serde_json::Value,
}

// ---------------------------------------------------------------------------
// Real wasmtime runtime (feature-gated)
// ---------------------------------------------------------------------------

#[cfg(feature = "wasmtime-runtime")]
pub(crate) mod real {
    use super::*;
    use bext_plugin_api::types::fuel;
    use hmac::{Hmac, Mac};
    use sha2::{Digest, Sha256};
    use std::path::{Path, PathBuf};
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::Arc;
    use wasmtime::*;

    type HmacSha256 = Hmac<Sha256>;
    const DEFAULT_MAX_CACHE_BYTES: u64 = 256 * 1024 * 1024;
    /// Length of the HMAC-SHA256 tag appended to cached .cwasm files.
    const HMAC_TAG_LEN: usize = 32;

    // ── Cached export handles ──────────────────────────────────────
    #[derive(Clone)]
    pub(crate) struct CachedExports {
        pub memory: Memory,
        pub alloc: TypedFunc<i32, i32>,
        pub init: TypedFunc<(), (i32, i32)>,
        pub on_server_start: Option<TypedFunc<(i32, i32), ()>>,
        pub on_server_stop: Option<TypedFunc<(), ()>>,
        pub on_request_complete: Option<TypedFunc<(i32, i32), ()>>,
        pub on_cache_write: Option<TypedFunc<(i32, i32, i32, i32), ()>>,
        pub on_cache_invalidate: Option<TypedFunc<(i32, i32, i32), ()>>,
        pub on_reload: Option<TypedFunc<(), ()>>,
        pub cleanup: Option<TypedFunc<(), ()>>,
    }

    impl CachedExports {
        pub(crate) fn resolve(
            instance: &Instance,
            store: &mut Store<HostState>,
        ) -> Result<Self, String> {
            let memory = instance
                .get_memory(&mut *store, "memory")
                .ok_or_else(|| "no memory export".to_string())?;
            let alloc = instance
                .get_typed_func::<i32, i32>(&mut *store, "alloc")
                .map_err(|e| format!("resolve alloc: {e}"))?;
            let init = instance
                .get_typed_func::<(), (i32, i32)>(&mut *store, "init")
                .map_err(|e| format!("resolve init: {e}"))?;

            // Optional exports — lifecycle hooks may not all be implemented by the guest
            Ok(Self {
                memory,
                alloc,
                init,
                on_server_start: instance
                    .get_typed_func::<(i32, i32), ()>(&mut *store, "on_server_start")
                    .ok(),
                on_server_stop: instance
                    .get_typed_func::<(), ()>(&mut *store, "on_server_stop")
                    .ok(),
                on_request_complete: instance
                    .get_typed_func::<(i32, i32), ()>(&mut *store, "on_request_complete")
                    .ok(),
                on_cache_write: instance
                    .get_typed_func::<(i32, i32, i32, i32), ()>(&mut *store, "on_cache_write")
                    .ok(),
                on_cache_invalidate: instance
                    .get_typed_func::<(i32, i32, i32), ()>(&mut *store, "on_cache_invalidate")
                    .ok(),
                on_reload: instance
                    .get_typed_func::<(), ()>(&mut *store, "on_reload")
                    .ok(),
                cleanup: instance
                    .get_typed_func::<(), ()>(&mut *store, "cleanup")
                    .ok(),
            })
        }
    }

    // ── AOT module cache ──────────────────────────────────────────
    struct ModuleCache {
        cache_dir: PathBuf,
        version: String,
        max_bytes: u64,
        /// HMAC key derived from (engine version + storage root path) at startup.
        /// Used to sign/verify cached .cwasm files so that a tampered file
        /// cannot trick `Module::deserialize` into loading attacker-controlled
        /// native code.
        hmac_key: Vec<u8>,
    }

    impl ModuleCache {
        fn new(storage_root: &Path, version: &str) -> Self {
            // Derive a per-installation HMAC key from material that is stable
            // across restarts but unique per storage root (different deploys get
            // different keys). This doesn't need to be a secret: the threat
            // model is a local attacker who can write to the cache dir but
            // doesn't have read access to server memory. The key is never
            // persisted on disk.
            let mut key_hasher = Sha256::new();
            key_hasher.update(b"bext-wasm-cache-hmac-key-v1:");
            key_hasher.update(version.as_bytes());
            key_hasher.update(b":");
            key_hasher.update(storage_root.to_string_lossy().as_bytes());
            let hmac_key = key_hasher.finalize().to_vec();

            Self {
                cache_dir: storage_root.join(".module-cache"),
                version: version.to_string(),
                max_bytes: DEFAULT_MAX_CACHE_BYTES,
                hmac_key,
            }
        }

        fn cache_key(&self, wasm_bytes: &[u8]) -> String {
            let mut hasher = Sha256::new();
            hasher.update(self.version.as_bytes());
            hasher.update(b":");
            hasher.update(wasm_bytes);
            hex::encode(hasher.finalize())
        }

        /// Compute HMAC-SHA256 over the serialized module bytes.
        fn compute_hmac(&self, data: &[u8]) -> Vec<u8> {
            let mut mac = HmacSha256::new_from_slice(&self.hmac_key)
                .expect("HMAC can take key of any size");
            mac.update(data);
            mac.finalize().into_bytes().to_vec()
        }

        /// Verify the HMAC tag appended to cached data. Returns the module
        /// bytes (without the tag) on success, or None on mismatch.
        fn verify_and_strip_hmac<'a>(&self, data: &'a [u8]) -> Option<&'a [u8]> {
            if data.len() < HMAC_TAG_LEN {
                return None;
            }
            let (module_bytes, tag) = data.split_at(data.len() - HMAC_TAG_LEN);
            let mut mac = HmacSha256::new_from_slice(&self.hmac_key)
                .expect("HMAC can take key of any size");
            mac.update(module_bytes);
            mac.verify_slice(tag).ok()?;
            Some(module_bytes)
        }

        fn get(&self, engine: &Engine, wasm_bytes: &[u8]) -> Option<Module> {
            let key = self.cache_key(wasm_bytes);
            let path = self.cache_dir.join(format!("{key}.cwasm"));
            if !path.exists() {
                return None;
            }

            // Security: verify the cache file is not world-writable before
            // deserializing. A writable cache file could be tampered with to
            // inject arbitrary native code, since Module::deserialize trusts
            // the input completely (it loads pre-compiled machine code).
            //
            // The cache directory should be owned by the server user with
            // mode 0o700 (rwx------). Operators must ensure this via deployment
            // configuration.
            #[cfg(unix)]
            {
                use std::os::unix::fs::PermissionsExt;
                if let Ok(meta) = std::fs::metadata(&path) {
                    let mode = meta.permissions().mode();
                    // Reject if world-writable (o+w) or group-writable (g+w)
                    if mode & 0o022 != 0 {
                        tracing::warn!(
                            cache_key = %key,
                            mode = format!("{:o}", mode),
                            "rejecting AOT cache file: insecure permissions (group/world-writable)"
                        );
                        return None;
                    }
                }
            }

            let data = std::fs::read(&path).ok()?;

            // Security: verify HMAC-SHA256 integrity tag before deserializing.
            // Module::deserialize loads pre-compiled native code and trusts the
            // input completely — a tampered file means arbitrary code execution.
            let module_bytes = match self.verify_and_strip_hmac(&data) {
                Some(bytes) => bytes,
                None => {
                    tracing::warn!(
                        cache_key = %key,
                        "rejecting AOT cache file: HMAC integrity check failed (possible tampering)"
                    );
                    let _ = std::fs::remove_file(&path);
                    return None;
                }
            };

            // SAFETY: We compiled, HMAC-signed, and cached this module
            // ourselves. The HMAC verification above ensures the file has not
            // been modified since we wrote it. The cache key is derived from
            // SHA-256(engine_version + wasm_bytes), so a valid cache hit
            // implies the file was produced by this engine version for this
            // exact WASM input.
            match unsafe { Module::deserialize(engine, module_bytes) } {
                Ok(module) => {
                    tracing::debug!(cache_key = %key, "loaded module from AOT cache");
                    Some(module)
                }
                Err(e) => {
                    tracing::warn!(cache_key = %key, error = %e, "AOT cache stale, recompiling");
                    let _ = std::fs::remove_file(&path);
                    None
                }
            }
        }

        fn put(&self, wasm_bytes: &[u8], module: &Module) {
            let key = self.cache_key(wasm_bytes);
            if std::fs::create_dir_all(&self.cache_dir).is_err() {
                return;
            }
            let final_path = self.cache_dir.join(format!("{key}.cwasm"));
            let tmp_path = self.cache_dir.join(format!("{key}.cwasm.tmp"));
            if let Ok(serialized) = module.serialize() {
                // Append HMAC-SHA256 tag so we can verify integrity on load.
                let tag = self.compute_hmac(&serialized);
                let mut signed = serialized;
                signed.extend_from_slice(&tag);

                // Atomic write: tmp + rename prevents corruption on crash
                if std::fs::write(&tmp_path, &signed).is_ok() {
                    let _ = std::fs::rename(&tmp_path, &final_path);
                    tracing::debug!(cache_key = %key, "stored module in AOT cache");
                }
            }
            self.prune();
        }

        fn prune(&self) {
            let entries = match std::fs::read_dir(&self.cache_dir) {
                Ok(rd) => rd,
                Err(_) => return,
            };
            let mut files: Vec<(PathBuf, u64, std::time::SystemTime)> = Vec::new();
            let mut total: u64 = 0;
            for entry in entries.flatten() {
                let path = entry.path();
                if path.extension().is_some_and(|ext| ext == "cwasm") {
                    if let Ok(meta) = entry.metadata() {
                        let size = meta.len();
                        let mtime = meta.modified().unwrap_or(std::time::SystemTime::UNIX_EPOCH);
                        total += size;
                        files.push((path, size, mtime));
                    }
                }
            }
            if total <= self.max_bytes {
                return;
            }
            files.sort_by_key(|(_, _, mtime)| *mtime);
            for (path, size, _) in &files {
                if total <= self.max_bytes {
                    break;
                }
                if std::fs::remove_file(path).is_ok() {
                    total = total.saturating_sub(*size);
                }
            }
        }
    }

    // ── Host state ────────────────────────────────────────────────
    pub(crate) struct HostState {
        pub(crate) sandbox: PluginSandbox,
        pub(crate) plugin_config: serde_json::Value,
        pub(crate) memory: Option<Memory>,
        pub(crate) limiter: StoreLimits,
    }

    fn default_store_limits() -> StoreLimits {
        StoreLimitsBuilder::new()
            .memory_size(64 * 1024 * 1024) // 64 MB
            .table_elements(10_000)
            .instances(1)
            .build()
    }

    // ── Loaded plugin ─────────────────────────────────────────────
    struct WasmPlugin {
        store: Store<HostState>,
        exports: CachedExports,
        manifest: PluginManifest,
        priority: u32,
    }

    // ── Runtime ───────────────────────────────────────────────────
    pub struct WasmPluginRuntime {
        engine: Engine,
        linker: Linker<HostState>,
        module_cache: ModuleCache,
        storage_root: PathBuf,
        plugins: Vec<WasmPlugin>,
        epoch_shutdown: Arc<AtomicBool>,
        _epoch_thread: Option<std::thread::JoinHandle<()>>,
    }

    impl Drop for WasmPluginRuntime {
        fn drop(&mut self) {
            self.epoch_shutdown.store(true, Ordering::Relaxed);
        }
    }

    impl WasmPluginRuntime {
        pub fn new(storage_root: PathBuf) -> Result<Self, String> {
            let mut config = Config::new();
            config.consume_fuel(true);
            config.epoch_interruption(true);
            config.cranelift_opt_level(OptLevel::Speed);

            let engine = Engine::new(&config).map_err(|e| format!("create WASM engine: {e}"))?;

            let mut linker = Linker::new(&engine);
            register_host_functions(&mut linker)?;

            // Epoch thread (~50ms ticks for wall-clock timeouts)
            let epoch_shutdown = Arc::new(AtomicBool::new(false));
            let epoch_thread = {
                let eng = engine.clone();
                let shutdown = epoch_shutdown.clone();
                Some(
                    std::thread::Builder::new()
                        .name("bext-wasm-epoch".into())
                        .spawn(move || {
                            while !shutdown.load(Ordering::Relaxed) {
                                std::thread::sleep(std::time::Duration::from_millis(50));
                                eng.increment_epoch();
                            }
                        })
                        .map_err(|e| format!("spawn epoch thread: {e}"))?,
                )
            };

            let module_cache = ModuleCache::new(&storage_root, env!("CARGO_PKG_VERSION"));

            Ok(Self {
                engine,
                linker,
                module_cache,
                storage_root,
                plugins: Vec::new(),
                epoch_shutdown,
                _epoch_thread: epoch_thread,
            })
        }

        pub fn load_plugin(&mut self, config: WasmPluginConfig) -> Result<(), String> {
            let wasm_path = &config.path;
            if !wasm_path.exists() {
                return Err(format!("WASM plugin not found: {}", wasm_path.display()));
            }

            tracing::info!(name = %config.name, path = %wasm_path.display(), "loading WASM plugin");

            let wasm_bytes = std::fs::read(wasm_path).map_err(|e| format!("read WASM: {e}"))?;

            // AOT cache: deserialize (~1ms) vs compile (~100-200ms)
            let module = match self.module_cache.get(&self.engine, &wasm_bytes) {
                Some(m) => m,
                None => {
                    let m = Module::new(&self.engine, &wasm_bytes)
                        .map_err(|e| format!("compile WASM: {e}"))?;
                    self.module_cache.put(&wasm_bytes, &m);
                    m
                }
            };

            // Create sandbox + store
            let sandbox =
                PluginSandbox::new(config.name.clone(), config.permissions, &self.storage_root);
            let host_state = HostState {
                sandbox,
                plugin_config: config.config.clone(),
                memory: None,
                limiter: default_store_limits(),
            };

            let mut store = Store::new(&self.engine, host_state);
            store.limiter(|state| &mut state.limiter);
            store
                .set_fuel(fuel::ON_SERVER_START)
                .map_err(|e| format!("set fuel: {e}"))?;
            store.set_epoch_deadline(200); // 10s

            // Instantiate
            let instance = self
                .linker
                .instantiate(&mut store, &module)
                .map_err(|e| format!("instantiate: {e}"))?;
            let exports = CachedExports::resolve(&instance, &mut store)?;
            store.data_mut().memory = Some(exports.memory);

            // Call init() to get manifest
            let (ptr, len) = exports
                .init
                .call(&mut store, ())
                .map_err(|e| format!("init call failed: {e}"))?;
            let json = read_string_from_memory(&exports.memory, &store, ptr, len);
            let manifest: PluginManifest =
                serde_json::from_str(&json).map_err(|e| format!("parse manifest: {e}"))?;

            tracing::info!(
                name = %manifest.name,
                version = %manifest.version,
                "WASM plugin loaded"
            );

            self.plugins.push(WasmPlugin {
                store,
                exports,
                manifest,
                priority: config.priority,
            });
            Ok(())
        }

        pub fn into_lifecycle_plugins(
            &mut self,
        ) -> Vec<Box<dyn bext_plugin_api::lifecycle::LifecyclePlugin>> {
            std::mem::take(&mut self.plugins)
                .into_iter()
                .map(|p| {
                    let wasm_state = crate::adapter::real::WasmState {
                        store: p.store,
                        exports: p.exports,
                    };
                    Box::new(WasmLifecycleAdapter::new(
                        p.manifest, p.priority, wasm_state,
                    )) as Box<dyn bext_plugin_api::lifecycle::LifecyclePlugin>
                })
                .collect()
        }

        pub fn len(&self) -> usize {
            self.plugins.len()
        }
        pub fn is_empty(&self) -> bool {
            self.plugins.is_empty()
        }
    }

    // ── Host function registration ─────────────────────────────────
    fn register_host_functions(linker: &mut Linker<HostState>) -> Result<(), String> {
        linker
            .func_wrap(
                "bext",
                "log",
                |mut caller: Caller<'_, HostState>,
                 level_ptr: i32,
                 level_len: i32,
                 msg_ptr: i32,
                 msg_len: i32| {
                    let level = read_guest_string(&mut caller, level_ptr, level_len);
                    let message = read_guest_string(&mut caller, msg_ptr, msg_len);
                    let plugin = &caller.data().sandbox.plugin_id;
                    match level.as_str() {
                        "error" => tracing::error!(plugin = %plugin, "{}", message),
                        "warn" => tracing::warn!(plugin = %plugin, "{}", message),
                        "info" => tracing::info!(plugin = %plugin, "{}", message),
                        "debug" => tracing::debug!(plugin = %plugin, "{}", message),
                        _ => tracing::trace!(plugin = %plugin, "{}", message),
                    }
                },
            )
            .map_err(|e| format!("link bext.log: {e}"))?;

        linker
            .func_wrap(
                "bext",
                "storage-get",
                |mut caller: Caller<'_, HostState>, key_ptr: i32, key_len: i32| -> (i32, i32) {
                    let key = read_guest_string(&mut caller, key_ptr, key_len);
                    let sanitized = match PluginSandbox::sanitize_key(&key) {
                        Some(k) => k,
                        None => {
                            return write_result_to_guest(&mut caller, Err("invalid key".into()))
                        }
                    };
                    let dir = caller.data().sandbox.storage_dir.clone();
                    let path = dir.join(&sanitized);
                    let result = match std::fs::read_to_string(&path) {
                        Ok(val) => Ok(serde_json::json!(val).to_string()),
                        Err(_) => Ok("null".to_string()),
                    };
                    write_result_to_guest(&mut caller, result)
                },
            )
            .map_err(|e| format!("link bext.storage-get: {e}"))?;

        linker
            .func_wrap(
                "bext",
                "storage-set",
                |mut caller: Caller<'_, HostState>,
                 key_ptr: i32,
                 key_len: i32,
                 val_ptr: i32,
                 val_len: i32|
                 -> i32 {
                    let key = read_guest_string(&mut caller, key_ptr, key_len);
                    let val = read_guest_string(&mut caller, val_ptr, val_len);
                    let sanitized = match PluginSandbox::sanitize_key(&key) {
                        Some(k) => k,
                        None => return 1,
                    };
                    if !caller.data().sandbox.check_storage_quota(val.len() as u64) {
                        return 2; // quota exceeded
                    }
                    let dir = caller.data().sandbox.storage_dir.clone();
                    let _ = std::fs::create_dir_all(&dir);
                    match std::fs::write(dir.join(&sanitized), val.as_bytes()) {
                        Ok(()) => {
                            caller.data().sandbox.record_storage(val.len() as u64);
                            0
                        }
                        Err(_) => 1,
                    }
                },
            )
            .map_err(|e| format!("link bext.storage-set: {e}"))?;

        linker
            .func_wrap(
                "bext",
                "storage-delete",
                |mut caller: Caller<'_, HostState>, key_ptr: i32, key_len: i32| -> i32 {
                    let key = read_guest_string(&mut caller, key_ptr, key_len);
                    let sanitized = match PluginSandbox::sanitize_key(&key) {
                        Some(k) => k,
                        None => return 1,
                    };
                    let dir = caller.data().sandbox.storage_dir.clone();
                    match std::fs::remove_file(dir.join(&sanitized)) {
                        Ok(()) => 0,
                        Err(_) => 1,
                    }
                },
            )
            .map_err(|e| format!("link bext.storage-delete: {e}"))?;

        linker
            .func_wrap(
                "bext",
                "get-config",
                |mut caller: Caller<'_, HostState>| -> (i32, i32) {
                    let config = caller.data().plugin_config.to_string();
                    write_result_to_guest(&mut caller, Ok(config))
                },
            )
            .map_err(|e| format!("link bext.get-config: {e}"))?;

        linker
            .func_wrap(
                "bext",
                "emit-metric",
                |mut caller: Caller<'_, HostState>,
                 name_ptr: i32,
                 name_len: i32,
                 value: f64,
                 tags_ptr: i32,
                 tags_len: i32| {
                    let name = read_guest_string(&mut caller, name_ptr, name_len);
                    let tags = read_guest_string(&mut caller, tags_ptr, tags_len);
                    let plugin = &caller.data().sandbox.plugin_id;
                    tracing::info!(
                        target: "bext::plugin_metric",
                        plugin = %plugin,
                        metric = %name,
                        value = value,
                        tags = %tags,
                        "plugin_metric"
                    );
                },
            )
            .map_err(|e| format!("link bext.emit-metric: {e}"))?;

        // bext.http-fetch(req_ptr, req_len) -> (ptr, len)
        // Perform a rate-limited, URL-allowlisted HTTP request.
        linker.func_wrap("bext", "http-fetch",
            |mut caller: Caller<'_, HostState>, req_ptr: i32, req_len: i32| -> (i32, i32) {
                let req_json = read_guest_string(&mut caller, req_ptr, req_len);
                let plugin_id = caller.data().sandbox.plugin_id.clone();

                // Parse request
                let req: crate::host_functions::FetchRequest = match serde_json::from_str(&req_json) {
                    Ok(r) => r,
                    Err(e) => {
                        return write_result_to_guest(
                            &mut caller,
                            Err(format!("invalid fetch request: {e}")),
                        );
                    }
                };

                // URL allowlist check
                if !caller.data().sandbox.is_url_allowed(&req.url) {
                    return write_result_to_guest(
                        &mut caller,
                        Err(format!("URL not in allowlist: {}", req.url)),
                    );
                }

                // Rate limit check
                if !caller.data().sandbox.try_fetch() {
                    return write_result_to_guest(
                        &mut caller,
                        Err("rate limit exceeded".into()),
                    );
                }

                // SSRF prevention: resolve DNS once, validate all IPs are public,
                // then pin the URL to the resolved IP to prevent DNS rebinding.
                let (pinned_url, original_host) = match crate::sandbox::resolve_and_pin_url(&req.url) {
                    Ok(pair) => pair,
                    Err(reason) => {
                        return write_result_to_guest(
                            &mut caller,
                            Err(format!("SSRF blocked: {reason}")),
                        );
                    }
                };

                tracing::debug!(plugin = %plugin_id, url = %req.url, pinned = %pinned_url, method = %req.method, "plugin http-fetch");

                // Perform blocking HTTP request via ureq using the pinned URL.
                // Pass the original Host header so the upstream server sees the
                // expected hostname, not the raw IP.
                let result = do_http_fetch(&req, &pinned_url, &original_host);
                let response_json = match result {
                    Ok(resp) => match serde_json::to_string(&resp) {
                        Ok(j) => Ok(j),
                        Err(e) => Err(format!("serialize response: {e}")),
                    },
                    Err(e) => Err(e),
                };

                write_result_to_guest(&mut caller, response_json)
            },
        ).map_err(|e| format!("link bext.http-fetch: {e}"))?;

        Ok(())
    }

    /// Maximum response body size (1 MB).
    const MAX_RESPONSE_BYTES: u64 = 1024 * 1024;
    /// HTTP request timeout (5 seconds).
    const FETCH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);

    /// Perform a blocking HTTP fetch via ureq. Called from the http-fetch host function.
    ///
    /// `pinned_url` has the hostname replaced with the resolved IP (SSRF-safe).
    /// `original_host` is the original Host header value to send to the server.
    ///
    /// Enforces a 5-second timeout and 1 MB response body limit.
    fn do_http_fetch(
        req: &crate::host_functions::FetchRequest,
        pinned_url: &str,
        original_host: &str,
    ) -> Result<crate::host_functions::FetchResponse, String> {
        use std::io::Read;

        // Build base request with timeout — use the pinned (IP-based) URL to
        // prevent DNS rebinding between the SSRF check and the actual request.
        let mut request = match req.method.to_uppercase().as_str() {
            "GET" => ureq::get(pinned_url),
            "POST" => ureq::post(pinned_url),
            "PUT" => ureq::put(pinned_url),
            "DELETE" => ureq::delete(pinned_url),
            "PATCH" => ureq::patch(pinned_url),
            "HEAD" => ureq::head(pinned_url),
            other => return Err(format!("unsupported HTTP method: {other}")),
        };

        // Set timeout
        request = request.timeout(FETCH_TIMEOUT);

        // Set the original Host header so the upstream server sees the expected
        // hostname, not the raw IP from the pinned URL.
        request = request.set("Host", original_host);

        // Set headers (user-provided headers can override the Host we just set)
        for (key, value) in &req.headers {
            request = request.set(key.as_str(), value.as_str());
        }

        // Execute request
        let response = if let Some(ref body) = req.body {
            request.send_string(body)
        } else {
            request.call()
        };

        let resp = response.map_err(|e| format!("fetch failed: {e}"))?;

        // Extract status and headers before consuming the body
        let status = resp.status();
        let headers: Vec<(String, String)> = resp
            .headers_names()
            .into_iter()
            .filter_map(|name| resp.header(&name).map(|v| (name, v.to_string())))
            .collect();

        // Read body with size limit (1 MB)
        let mut reader = resp.into_reader().take(MAX_RESPONSE_BYTES + 1);
        let mut body_bytes = Vec::new();
        reader
            .read_to_end(&mut body_bytes)
            .map_err(|e| format!("read response body: {e}"))?;

        if body_bytes.len() as u64 > MAX_RESPONSE_BYTES {
            return Err(format!(
                "response body exceeds {} byte limit",
                MAX_RESPONSE_BYTES
            ));
        }

        let body = String::from_utf8(body_bytes)
            .unwrap_or_else(|e| String::from_utf8_lossy(e.as_bytes()).to_string());

        Ok(crate::host_functions::FetchResponse {
            status,
            headers,
            body,
        })
    }

    // ── Guest memory helpers ───────────────────────────────────────

    fn read_guest_string(caller: &mut Caller<'_, HostState>, ptr: i32, len: i32) -> String {
        // Validate that ptr and len are non-negative before casting to usize.
        // A negative value from the guest would wrap to a very large usize,
        // potentially causing out-of-bounds reads or panics.
        if ptr < 0 || len < 0 {
            return String::new();
        }

        let memory = match caller.data().memory {
            Some(mem) => mem,
            None => match caller.get_export("memory").and_then(|e| e.into_memory()) {
                Some(mem) => mem,
                None => return String::new(),
            },
        };
        let data = memory.data(&*caller);
        let start = ptr as usize;
        let end = start + len as usize;
        if end <= data.len() {
            match std::str::from_utf8(&data[start..end]) {
                Ok(s) => s.to_string(),
                Err(_) => String::from_utf8_lossy(&data[start..end]).to_string(),
            }
        } else {
            String::new()
        }
    }

    fn write_result_to_guest(
        caller: &mut Caller<'_, HostState>,
        result: Result<String, String>,
    ) -> (i32, i32) {
        let (data, is_err) = match result {
            Ok(s) => (s, false),
            Err(s) => (s, true),
        };
        let memory = match caller.data().memory {
            Some(mem) => mem,
            None => return (0, 0),
        };
        let alloc_func = match caller.get_export("alloc").and_then(|e| e.into_func()) {
            Some(af) => af,
            None => return (0, 0),
        };
        let alloc_fn = match alloc_func.typed::<i32, i32>(&*caller) {
            Ok(f) => f,
            Err(_) => return (0, 0),
        };
        let bytes = data.as_bytes();
        let len = bytes.len() as i32;
        match alloc_fn.call(&mut *caller, len) {
            Ok(ptr) => {
                let mem_data = memory.data_mut(&mut *caller);
                let start = ptr as usize;
                let end = start + bytes.len();
                if end <= mem_data.len() {
                    mem_data[start..end].copy_from_slice(bytes);
                }
                (ptr, if is_err { -len } else { len })
            }
            Err(_) => (0, 0),
        }
    }

    fn read_string_from_memory(
        memory: &Memory,
        store: &Store<HostState>,
        ptr: i32,
        len: i32,
    ) -> String {
        // Validate that ptr and len are non-negative before casting to usize.
        // A negative value from the guest would wrap to a very large usize,
        // potentially causing out-of-bounds reads or panics.
        if ptr < 0 || len < 0 {
            return String::new();
        }
        let data = memory.data(store);
        let start = ptr as usize;
        let end = start + len as usize;
        if end > data.len() {
            return String::new();
        }
        match std::str::from_utf8(&data[start..end]) {
            Ok(s) => s.to_string(),
            Err(_) => String::from_utf8_lossy(&data[start..end]).to_string(),
        }
    }
}

// ---------------------------------------------------------------------------
// Stub runtime (when wasmtime-runtime feature is not enabled)
// ---------------------------------------------------------------------------

#[cfg(not(feature = "wasmtime-runtime"))]
mod stub {
    use super::*;

    pub struct WasmPluginRuntime {
        storage_root: std::path::PathBuf,
        plugins: Vec<(WasmLifecycleAdapter, PluginSandbox)>,
    }

    impl WasmPluginRuntime {
        pub fn new(storage_root: std::path::PathBuf) -> Result<Self, String> {
            Ok(Self {
                storage_root,
                plugins: Vec::new(),
            })
        }

        pub fn load_plugin(&mut self, config: WasmPluginConfig) -> Result<(), String> {
            if !config.path.exists() {
                return Err(format!("WASM plugin not found: {}", config.path.display()));
            }
            tracing::warn!(
                name = %config.name,
                "WASM plugin loaded in STUB mode (compile with wasmtime-runtime feature for real execution)"
            );
            let manifest = PluginManifest {
                name: config.name.clone(),
                version: "0.0.0-stub".into(),
                description: format!("Stub: {}", config.path.display()),
                capabilities: vec![bext_plugin_api::types::PluginCapability::Lifecycle],
                provides_capabilities: Vec::new(),
                requires_capabilities: Vec::new(),
            };
            let sandbox = PluginSandbox::new(config.name, config.permissions, &self.storage_root);
            let adapter = WasmLifecycleAdapter::new(manifest, config.priority);
            self.plugins.push((adapter, sandbox));
            Ok(())
        }

        pub fn into_lifecycle_plugins(
            &mut self,
        ) -> Vec<Box<dyn bext_plugin_api::lifecycle::LifecyclePlugin>> {
            std::mem::take(&mut self.plugins)
                .into_iter()
                .map(|(a, _)| Box::new(a) as Box<dyn bext_plugin_api::lifecycle::LifecyclePlugin>)
                .collect()
        }

        pub fn len(&self) -> usize {
            self.plugins.len()
        }
        pub fn is_empty(&self) -> bool {
            self.plugins.is_empty()
        }
    }
}

// Re-export the active runtime implementation.
#[cfg(feature = "wasmtime-runtime")]
pub use real::WasmPluginRuntime;
#[cfg(not(feature = "wasmtime-runtime"))]
pub use stub::WasmPluginRuntime;

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

    #[test]
    fn runtime_empty() {
        let rt = WasmPluginRuntime::new(std::path::PathBuf::from("/tmp/bext-plugins")).unwrap();
        assert!(rt.is_empty());
        assert_eq!(rt.len(), 0);
    }

    #[test]
    fn load_nonexistent_fails() {
        let mut rt = WasmPluginRuntime::new(std::path::PathBuf::from("/tmp/bext-plugins")).unwrap();
        let result = rt.load_plugin(WasmPluginConfig {
            name: "test".into(),
            path: "/nonexistent/plugin.wasm".into(),
            priority: 1000,
            permissions: WasmPluginPermissions::default(),
            config: serde_json::Value::Null,
        });
        assert!(result.is_err());
    }
}
