freenet 0.2.68

use std::{fs::File, io::Write, path::PathBuf, sync::Arc};

use dashmap::DashMap;
use freenet_stdlib::prelude::*;
use moka::sync::Cache as MokaCache;

use crate::contract::storages::Storage;

use super::RuntimeResult;

/// Handle contract blob storage on the file system.
pub struct ContractStore {
    contracts_dir: PathBuf,
    contract_cache: MokaCache<CodeHash, Arc<ContractCode<'static>>>,
    /// In-memory index: ContractInstanceId -> CodeHash
    /// This is populated from ReDb on startup and kept in sync
    key_to_code_part: Arc<DashMap<ContractInstanceId, CodeHash>>,
    /// ReDb storage for persistent index
    db: Storage,
}
// Eviction-driven reclamation of unused contracts now exists: the hosting
// cache evicts least-valuable contracts past its budget and the resulting
// `EvictContract` event drives `Executor::reclaim_contract_storage`, which
// calls `ContractStore::remove_contract` to delete the on-disk `.wasm` blob.

impl ContractStore {
    /// # Arguments
    /// - contracts_dir: directory where contract WASM files are stored
    /// - max_size: max size in bytes of the contracts being cached
    /// - db: ReDb storage for persistent index
    pub fn new(contracts_dir: PathBuf, max_size: u64, db: Storage) -> RuntimeResult<Self> {
        std::fs::create_dir_all(&contracts_dir).map_err(|err| {
            tracing::error!("error creating contract dir: {err}");
            err
        })?;

        // Load index from ReDb
        let key_to_code_part = Arc::new(DashMap::new());
        match db.load_all_contract_index() {
            Ok(entries) => {
                for (instance_id, code_hash) in entries {
                    key_to_code_part.insert(instance_id, code_hash);
                }
                tracing::debug!(
                    "Loaded {} contract index entries from ReDb",
                    key_to_code_part.len()
                );
            }
            Err(e) => {
                tracing::warn!("Failed to load contract index from ReDb: {e}");
            }
        }

        // Migrate any contract WASM files written under the legacy lowercased
        // Base58 name to the canonical mixed-case name (issue #4214) so the
        // fetch paths below, which use `code_hash.encode()`, still find code
        // persisted before the stdlib CodeHash::encode case-fix.
        for entry in key_to_code_part.iter() {
            super::migrate_legacy_lowercased_code_file(
                &contracts_dir,
                &entry.value().encode(),
                "wasm",
            );
        }

        Ok(Self {
            contract_cache: MokaCache::builder()
                .max_capacity(max_size)
                .weigher(
                    |key: &CodeHash, value: &Arc<ContractCode<'static>>| -> u32 {
                        // Saturate to u32::MAX on overflow as moka recommends.
                        // A contract WASM module larger than 4 GiB would indicate
                        // a bug in upstream size validation — log it loudly.
                        let len = value.data().len();
                        u32::try_from(len).unwrap_or_else(|_| {
                            tracing::warn!(
                                code_hash = %key,
                                size_bytes = len,
                                "Contract code exceeds u32::MAX in cache weigher; \
                                 saturating. This should be impossible."
                            );
                            u32::MAX
                        })
                    },
                )
                .build(),
            contracts_dir,
            key_to_code_part,
            db,
        })
    }

    /// Returns a copy of the contract bytes if available, none otherwise.
    // todo: instead return Result<Option<_>, _> to handle IO errors upstream
    //
    // Known limitation: `key_to_code_part` is a per-`ContractStore` index
    // (a fresh `Arc<DashMap>` per `ContractStore::new`), so when the
    // executor pool holds multiple `ContractStore` instances each carries
    // its own copy of the index. A contract stored via executor A may
    // therefore be reported as "missing" by `fetch_contract` on executor
    // B even though both share the same on-disk `.wasm` and the same
    // process-wide `contract_cache`. The cache-hit fast path below
    // masks this for shared-code reads when the code hash happens to be
    // warm in the cache, which is the only reason cross-executor
    // fetches currently work in production. This divergence is tracked
    // as #4218; a proper fix would either share the index across
    // pool members or rebuild it from the on-disk state at startup
    // per executor.
    pub fn fetch_contract(
        &self,
        key: &ContractKey,
        params: &Parameters<'_>,
    ) -> Option<ContractContainer> {
        let code_hash = key.code_hash();
        if let Some(data) = self.contract_cache.get(code_hash) {
            return Some(ContractContainer::Wasm(ContractWasmAPIVersion::V1(
                WrappedContract::new(data, params.clone().into_owned()),
            )));
        }

        self.key_to_code_part.get(key.id()).and_then(|entry| {
            let code_hash = *entry.value();
            let path = code_hash.encode();
            let key_path = self.contracts_dir.join(path).with_extension("wasm");
            // Load with version prefix stripping (fixes #2924)
            // Files are stored with to_bytes_versioned() which adds a version prefix.
            // Must use load_versioned_from_path() to strip it before compilation.
            let (code, _ver) = ContractCode::load_versioned_from_path(&key_path)
                .map_err(|err| {
                    tracing::debug!("contract not found: {err}");
                    err
                })
                .ok()?;
            let params = params.clone().into_owned();
            // add back the contract part to the mem store
            self.contract_cache
                .insert(code_hash, Arc::new(code.clone()));
            Some(ContractContainer::Wasm(ContractWasmAPIVersion::V1(
                WrappedContract::new(Arc::new(code), params),
            )))
        })
    }

    /// Store a copy of the contract in the local store, in case it hasn't been stored previously.
    pub fn store_contract(&mut self, contract: ContractContainer) -> RuntimeResult<()> {
        let (key, code) = match contract.clone() {
            ContractContainer::Wasm(ContractWasmAPIVersion::V1(contract_v1)) => {
                (*contract_v1.key(), contract_v1.code().clone())
            }
            ContractContainer::Wasm(_) | _ => unimplemented!(),
        };
        let code_hash = key.code_hash();
        let key_path = code_hash.encode();
        let key_path = self.contracts_dir.join(key_path).with_extension("wasm");
        if self.contract_cache.get(code_hash).is_some() && key_path.exists() {
            // WASM code is cached AND the blob is still on disk: fast path. We
            // still need to ensure this instance_id is indexed (different
            // ContractInstanceIds with the same code each need their own
            // mapping — see issue #2380).
            //
            // We MUST also verify the blob is on disk: this `contract_cache` is
            // per-`ContractStore` (per pool executor), and a `remove_contract`
            // on a sibling executor can delete the shared blob without
            // invalidating our cache. Falling through to the disk-write branch
            // below restores the blob in that case so the new instance is
            // durably stored. See issue #4218 for the underlying per-executor
            // map divergence and Codex's round-4 finding.
            self.ensure_key_indexed(&key)?;
            return Ok(());
        }
        if let Ok((code, _ver)) = ContractCode::load_versioned_from_path(&key_path) {
            // WASM file exists on disk. Add to cache AND ensure the index is updated.
            // See issue #2344 for why this is critical after crash recovery.
            self.ensure_key_indexed(&key)?;
            self.contract_cache.insert(*code_hash, Arc::new(code));
            return Ok(());
        }

        // CRITICAL ORDER: Write disk first, then index, then cache.
        // This ensures fetch_contract() can always fall back to disk lookup.

        // Step 1: Save to disk first (ensures data is persisted)
        let version = APIVersion::from(contract);
        let output: Vec<u8> = code
            .to_bytes_versioned(version)
            .map_err(|e| anyhow::anyhow!(e))?;
        let mut file = File::create(&key_path)?;
        file.write_all(output.as_slice())?;
        file.sync_all()?; // Ensure durability before updating index

        // Step 2: Update index in ReDb (persistent, crash-safe)
        self.db
            .store_contract_index(key.id(), code_hash)
            .map_err(|e| anyhow::anyhow!("Failed to store contract index: {e}"))?;

        // Step 3: Update in-memory index
        self.key_to_code_part.insert(*key.id(), *code_hash);

        // Step 4: Insert into memory cache
        let data = code.data().to_vec();
        self.contract_cache
            .insert(*code_hash, Arc::new(ContractCode::from(data)));

        Ok(())
    }

    pub fn get_contract_path(&mut self, key: &ContractKey) -> RuntimeResult<PathBuf> {
        let contract_hash = *key.code_hash();
        let key_path = contract_hash.encode();
        Ok(self.contracts_dir.join(key_path).with_extension("wasm"))
    }

    /// Remove a contract instance from the store.
    ///
    /// Removes this instance's index entries (ReDb + in-memory) and any
    /// delegate subscriptions unconditionally. The on-disk `.wasm` blob and
    /// the code cache entry are keyed by `code_hash`, which is shared across
    /// every `ContractInstanceId` using the same code, so they are removed
    /// only once no remaining instance references that code hash. File
    /// removal is idempotent — an already-missing `.wasm` is not an error.
    ///
    /// The "still referenced?" decision is made against the **shared,
    /// persistent ReDb `contract_index`**, not the per-`ContractStore`
    /// in-memory `key_to_code_part` map. Each runtime-pool executor owns a
    /// *separate* `ContractStore` with its own `key_to_code_part` built
    /// fresh in [`ContractStore::new`], so an instance stored via a
    /// different pool executor is invisible to an in-memory scan. Deciding
    /// from the in-memory map would wrongly delete a `.wasm` blob still
    /// referenced by an instance owned by another executor — corrupting
    /// every surviving contract that shares the code (e.g. every River
    /// room shares one room-contract WASM, see issue #2380). The ReDb
    /// index is the single shared source of truth across all executors.
    pub fn remove_contract(&mut self, key: &ContractKey) -> RuntimeResult<()> {
        let contract_hash = *key.code_hash();

        // Remove this instance's index entries first. The ReDb removal must
        // happen before the `load_all_contract_index()` scan below so this
        // instance's own entry is not counted as a remaining reference.
        self.db
            .remove_contract_index(key.id())
            .map_err(|e| anyhow::anyhow!("Failed to remove contract index: {e}"))?;
        self.key_to_code_part.remove(key.id());

        // Clean up any delegate subscriptions for this contract instance.
        super::DELEGATE_SUBSCRIPTIONS.remove(key.id());

        // The WASM blob on disk is keyed by code hash and shared by every
        // contract instance with the same code (e.g. all River rooms share
        // one room-contract WASM — see issue #2380). Only delete the blob
        // and its cache entry once no remaining instance references this
        // code hash, otherwise the surviving instances break on cache miss.
        //
        // Decide from the shared persistent index, NOT the per-executor
        // in-memory map (see the doc comment above for why).
        let code_still_referenced = match self.db.load_all_contract_index() {
            Ok(entries) => entries
                .iter()
                .any(|(_, code_hash)| code_hash == &contract_hash),
            Err(e) => {
                // If we cannot read the shared index we cannot prove the
                // blob is unreferenced. A leaked blob is recoverable disk
                // space; deleting a still-referenced shared blob corrupts
                // every surviving contract using that code. Fail safe:
                // keep the blob — and return an error so the caller knows
                // the code half was NOT fully reclaimed and will requeue
                // for retry rather than clearing the pending entry.
                tracing::warn!(
                    code_hash = %contract_hash,
                    error = %e,
                    "Could not load shared contract index while removing a \
                     contract; keeping the .wasm blob to avoid corrupting \
                     contracts that may still reference this code"
                );
                return Err(anyhow::anyhow!(
                    "kept WASM blob for {contract_hash}: shared contract \
                     index read failed: {e}"
                )
                .into());
            }
        };
        if !code_still_referenced {
            // Invalidate the code cache so a removed contract is never served
            // as a "ghost" (issue #3487).
            self.contract_cache.invalidate(&contract_hash);
            let key_path = self
                .contracts_dir
                .join(contract_hash.encode())
                .with_extension("wasm");
            match std::fs::remove_file(&key_path) {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                Err(e) => return Err(e.into()),
            }
        }
        Ok(())
    }

    pub fn code_hash_from_key(&self, key: &ContractKey) -> Option<CodeHash> {
        self.key_to_code_part.get(key.id()).map(|r| *r.value())
    }

    /// Look up the code hash for a contract given only its instance ID.
    /// Used when clients request contracts without knowing the code hash.
    pub fn code_hash_from_id(&self, id: &ContractInstanceId) -> Option<CodeHash> {
        self.key_to_code_part.get(id).map(|r| *r.value())
    }

    /// Ensures the key_to_code_part mapping exists for the given contract key.
    /// This is needed because when contract code is already cached, store_contract
    /// may not be called, but we still need to index new ContractInstanceIds that
    /// use the same code (different parameters = different rooms).
    /// See issue #2380.
    pub fn ensure_key_indexed(&mut self, key: &ContractKey) -> RuntimeResult<()> {
        let code_hash = key.code_hash();
        if !self.key_to_code_part.contains_key(key.id()) {
            // Store in ReDb
            self.db
                .store_contract_index(key.id(), code_hash)
                .map_err(|e| anyhow::anyhow!("Failed to store contract index: {e}"))?;

            // Update in-memory map
            self.key_to_code_part.insert(*key.id(), *code_hash);

            tracing::debug!(
                contract = %key,
                instance_id = %key.id(),
                code_hash = %code_hash,
                "Indexed contract instance (same code, different params)"
            );
        }
        Ok(())
    }
}

#[cfg(test)]
mod test {
    //! Tests for ContractStore
    //!
    //! Key invariant: For every contract stored, `code_hash_from_id(instance_id)`
    //! must return the correct CodeHash. This is critical because `lookup_key()`
    //! uses this to reconstruct ContractKey from just an instance ID.
    use super::*;

    async fn create_test_db(path: &std::path::Path) -> Storage {
        Storage::new(path).await.expect("failed to create test db")
    }

    #[tokio::test]
    async fn store_and_load() -> Result<(), Box<dyn std::error::Error>> {
        let contract_dir = crate::util::tests::get_temp_dir();
        std::fs::create_dir_all(contract_dir.path())?;
        let db = create_test_db(contract_dir.path()).await;
        let mut store = ContractStore::new(contract_dir.path().into(), 10_000, db)?;
        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![0, 1, 2])),
            [0, 1].as_ref().into(),
        );
        let container = ContractContainer::Wasm(ContractWasmAPIVersion::V1(contract.clone()));
        store.store_contract(container)?;
        let f = store.fetch_contract(contract.key(), &[0, 1].as_ref().into());
        assert!(f.is_some());
        Ok(())
    }

    /// Test that simulates the actual contract store flow to see if
    /// contracts can be "lost" between store and fetch
    #[tokio::test]
    async fn test_contract_store_fetch_reliability() -> Result<(), Box<dyn std::error::Error>> {
        let contract_dir = crate::util::tests::get_temp_dir();
        std::fs::create_dir_all(contract_dir.path())?;
        let db = create_test_db(contract_dir.path()).await;

        // Use realistic-ish cache size
        let mut store = ContractStore::new(contract_dir.path().into(), 100_000, db)?;

        // Store multiple contracts with varying sizes, track their keys
        let mut keys = Vec::new();
        for i in 0..10u8 {
            // Create contracts of different sizes
            let size = ((i as usize) + 1) * 1000;
            let code = vec![i; size];
            let params = Parameters::from(vec![i, i + 1]);
            let contract = WrappedContract::new(Arc::new(ContractCode::from(code)), params.clone());
            let key = *contract.key();
            keys.push((key, params));
            let container = ContractContainer::Wasm(ContractWasmAPIVersion::V1(contract));
            store.store_contract(container)?;
        }

        // Immediately try to fetch all contracts - this is the critical path
        // where issue #2306 manifests
        let mut fetch_failures = 0;
        for (key, params) in &keys {
            let fetched = store.fetch_contract(key, params);
            if fetched.is_none() {
                eprintln!("FETCH FAILED for contract {key} immediately after store!");
                fetch_failures += 1;
            }
        }

        assert_eq!(
            fetch_failures, 0,
            "Contracts should be fetchable immediately after store"
        );

        Ok(())
    }

    /// Test for issue #2344: Contract store index must be persisted to disk.
    /// This test simulates a node restart by creating a new ContractStore from
    /// the same directory, then verifies contracts are still fetchable.
    #[tokio::test]
    async fn test_index_persistence_after_restart() -> Result<(), Box<dyn std::error::Error>> {
        let contract_dir = crate::util::tests::get_temp_dir();
        std::fs::create_dir_all(contract_dir.path())?;

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1, 2, 3, 4, 5])),
            [10, 20].as_ref().into(),
        );
        let key = *contract.key();
        let params: Parameters = [10, 20].as_ref().into();

        // Store the contract
        {
            let db = create_test_db(contract_dir.path()).await;
            let mut store = ContractStore::new(contract_dir.path().into(), 10_000, db)?;
            let container = ContractContainer::Wasm(ContractWasmAPIVersion::V1(contract));
            store.store_contract(container)?;

            // Verify it's fetchable in the same instance
            assert!(
                store.fetch_contract(&key, &params).is_some(),
                "Contract should be fetchable immediately after store"
            );
        }
        // ContractStore dropped here - simulates process exit

        // Create a NEW ContractStore from the same directory - simulates node restart
        {
            let db = create_test_db(contract_dir.path()).await;
            let store = ContractStore::new(contract_dir.path().into(), 10_000, db)?;

            // The contract should be fetchable because both:
            // 1. The WASM file was persisted to disk
            // 2. The index (ReDb) was persisted to disk
            // Issue #2344: Before the fix, the index wasn't synced, so the contract
            // would not be found after restart.
            let fetched = store.fetch_contract(&key, &params);
            assert!(
                fetched.is_some(),
                "Contract should be fetchable after simulated restart - index must be persisted"
            );
        }

        Ok(())
    }

    /// Test for issue #2344: When WASM file exists but index entry is missing
    /// (e.g., after a crash), store_contract should add the missing index entry.
    #[tokio::test]
    async fn test_wasm_exists_but_index_missing() -> Result<(), Box<dyn std::error::Error>> {
        use std::io::Write;

        let contract_dir = crate::util::tests::get_temp_dir();
        std::fs::create_dir_all(contract_dir.path())?;

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![7, 8, 9])),
            [30, 40].as_ref().into(),
        );
        let key = *contract.key();
        let code_hash = key.code_hash();
        let params: Parameters = [30, 40].as_ref().into();

        // Manually create the WASM file on disk (simulating a crash scenario
        // where WASM was synced but index wasn't)
        let wasm_path = contract_dir
            .path()
            .join(code_hash.encode())
            .with_extension("wasm");
        {
            let code_bytes = contract
                .code()
                .to_bytes_versioned(freenet_stdlib::prelude::APIVersion::Version0_0_1)
                .unwrap();
            let mut file = std::fs::File::create(&wasm_path)?;
            file.write_all(&code_bytes)?;
            file.sync_all()?;
        }

        // Create a ContractStore - the ReDb will be empty (no index entries)
        let db = create_test_db(contract_dir.path()).await;
        let mut store = ContractStore::new(contract_dir.path().into(), 10_000, db)?;

        // The contract is NOT fetchable yet because the index doesn't have the entry
        // and it's not in cache
        assert!(
            store.fetch_contract(&key, &params).is_none(),
            "Contract should NOT be fetchable when WASM exists but index entry is missing"
        );

        // Now call store_contract - this should detect the WASM file exists,
        // add the missing index entry, and add to cache
        let container = ContractContainer::Wasm(ContractWasmAPIVersion::V1(contract));
        store.store_contract(container)?;

        // Drop the store and create a new one to verify the index was persisted
        drop(store);
        let db = create_test_db(contract_dir.path()).await;
        let store = ContractStore::new(contract_dir.path().into(), 10_000, db)?;

        // Now the contract should be fetchable because the fix adds the index entry
        let fetched = store.fetch_contract(&key, &params);
        assert!(
            fetched.is_some(),
            "Contract should be fetchable after store_contract adds missing index entry"
        );

        Ok(())
    }

    /// Regression test for issue #2380: Multiple contracts with same WASM code
    /// but different parameters must all be indexed correctly.
    ///
    /// This bug manifested in River when creating multiple chat rooms:
    /// - All rooms use the same room-contract WASM (same code_hash)
    /// - Different parameters (owner key) create different ContractInstanceIds
    /// - Only the first room's instance_id was indexed
    /// - Subscribe to 2nd+ rooms failed because lookup_key() returned None
    #[tokio::test]
    async fn test_multiple_contracts_same_code_different_params()
    -> Result<(), Box<dyn std::error::Error>> {
        let contract_dir = crate::util::tests::get_temp_dir();
        std::fs::create_dir_all(contract_dir.path())?;
        let db = create_test_db(contract_dir.path()).await;

        let mut store = ContractStore::new(contract_dir.path().into(), 10_000, db)?;

        // Same WASM code for all contracts (like River's room-contract)
        let shared_code = vec![1, 2, 3, 4, 5];

        // Create multiple contracts with SAME code but DIFFERENT parameters
        // This simulates creating multiple River rooms
        let mut contracts = Vec::new();
        for i in 0..5u8 {
            let params = Parameters::from(vec![i, i + 10, i + 20]); // Different params each time
            let contract = WrappedContract::new(
                Arc::new(ContractCode::from(shared_code.clone())),
                params.clone(),
            );
            contracts.push((contract.clone(), params));

            let container = ContractContainer::Wasm(ContractWasmAPIVersion::V1(contract));
            store.store_contract(container)?;
        }

        // All contracts share the same code_hash
        let expected_code_hash = contracts[0].0.key().code_hash();
        for (contract, _) in &contracts {
            assert_eq!(
                contract.key().code_hash(),
                expected_code_hash,
                "All contracts should have the same code_hash"
            );
        }

        // Critical assertion: code_hash_from_id must work for ALL instance IDs
        // This is what lookup_key() uses, and what failed before the fix
        for (i, (contract, _)) in contracts.iter().enumerate() {
            let instance_id = contract.key().id();
            let lookup_result = store.code_hash_from_id(instance_id);
            assert!(
                lookup_result.is_some(),
                "code_hash_from_id() failed for contract {i} (instance_id: {instance_id}) - \
                 this would cause Subscribe to fail!"
            );
            assert_eq!(
                lookup_result.unwrap(),
                *expected_code_hash,
                "code_hash_from_id() returned wrong hash for contract {i}"
            );
        }

        // Also verify fetch_contract works for all
        for (i, (contract, params)) in contracts.iter().enumerate() {
            let fetched = store.fetch_contract(contract.key(), params);
            assert!(
                fetched.is_some(),
                "fetch_contract() failed for contract {i}"
            );
        }

        Ok(())
    }

    /// Regression test for issue #2924: Versioned contract files must be
    /// properly loaded without the version prefix before compilation.
    ///
    /// The bug:
    /// - Contracts are stored with to_bytes_versioned() which adds a version prefix
    /// - fetch_contract() was using ContractContainer::try_from which read raw bytes
    /// - The prefix caused wasmtime to fail auto-detection (no WASM magic number)
    /// - Module::new tried to parse as WAT, failed with "input bytes aren't valid utf-8"
    ///
    /// The fix:
    /// - Use ContractCode::load_versioned_from_path() which strips the prefix
    /// - The WASM magic number is now at offset 0, so Module::new works correctly
    #[tokio::test]
    async fn test_versioned_contract_loading_issue_2924() -> Result<(), Box<dyn std::error::Error>>
    {
        use crate::wasm_runtime::engine::{Engine, WasmEngine};
        use crate::wasm_runtime::runtime::RuntimeConfig;

        let contract_dir = crate::util::tests::get_temp_dir();
        std::fs::create_dir_all(contract_dir.path())?;
        let db = create_test_db(contract_dir.path()).await;

        // Valid WASM binary (exports "memory" and "answer" function that returns 42)
        // WAT equivalent:
        // (module
        //   (memory 1)
        //   (export "memory" (memory 0))
        //   (func (export "answer") (result i32)
        //     i32.const 42
        //   )
        // )
        const VALID_WASM: &[u8] = &[
            0x00, 0x61, 0x73, 0x6d, 0x01, 0x00, 0x00, 0x00, 0x01, 0x05, 0x01, 0x60, 0x00, 0x01,
            0x7f, 0x03, 0x02, 0x01, 0x00, 0x05, 0x03, 0x01, 0x00, 0x01, 0x07, 0x13, 0x02, 0x06,
            0x6d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x02, 0x00, 0x06, 0x61, 0x6e, 0x73, 0x77, 0x65,
            0x72, 0x00, 0x00, 0x0a, 0x06, 0x01, 0x04, 0x00, 0x41, 0x2a, 0x0b,
        ];

        // Verify the WASM starts with magic number
        assert_eq!(
            &VALID_WASM[0..4],
            &[0x00, 0x61, 0x73, 0x6d],
            "Test WASM should start with magic number"
        );

        let mut store = ContractStore::new(contract_dir.path().into(), 10_000, db)?;
        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(VALID_WASM.to_vec())),
            [0, 1].as_ref().into(),
        );
        let key = *contract.key();
        let params: Parameters = [0, 1].as_ref().into();
        let container = ContractContainer::Wasm(ContractWasmAPIVersion::V1(contract.clone()));

        // Store the contract (this adds version prefix to disk file)
        store.store_contract(container)?;

        // Drop the store to clear the cache, forcing fetch_contract to read from disk
        drop(store);

        // Create a new store and fetch the contract from disk
        let db = create_test_db(contract_dir.path()).await;
        let store = ContractStore::new(contract_dir.path().into(), 10_000, db)?;
        let fetched = store
            .fetch_contract(&key, &params)
            .expect("Contract should be fetchable after store");

        // Extract the code bytes from the fetched contract
        let ContractContainer::Wasm(ContractWasmAPIVersion::V1(fetched_contract)) = fetched else {
            panic!("Expected WASM V1 contract");
        };

        // Verify the fetched code matches the original WASM (without version prefix)
        let fetched_bytes = fetched_contract.code().data();
        assert_eq!(
            fetched_bytes, VALID_WASM,
            "Fetched contract bytes should match original WASM without version prefix"
        );

        // Verify the WASM magic number is at the start (issue #2924 would fail here)
        assert_eq!(
            &fetched_bytes[0..4],
            &[0x00, 0x61, 0x73, 0x6d],
            "Fetched WASM should start with magic number (no version prefix)"
        );

        // Critical test: Verify the fetched contract can be compiled
        // Before the fix, this would fail with "Error when converting wat: input bytes aren't valid utf-8"
        let mut engine = Engine::new(&RuntimeConfig::default(), false)?;
        let compile_result = engine.compile(fetched_bytes);
        assert!(
            compile_result.is_ok(),
            "Contract should compile successfully without 'converting wat' error. Error: {:?}",
            compile_result.err()
        );

        Ok(())
    }

    /// Regression test for issue #3487: removed contracts must not be served from cache.
    ///
    /// Before the fix, remove_contract() deleted from disk and index but did not
    /// invalidate the cache, so fetch_contract() could still return "ghost" contracts.
    #[tokio::test]
    async fn test_remove_contract_invalidates_cache() -> Result<(), Box<dyn std::error::Error>> {
        let contract_dir = crate::util::tests::get_temp_dir();
        std::fs::create_dir_all(contract_dir.path())?;
        let db = create_test_db(contract_dir.path()).await;
        let mut store = ContractStore::new(contract_dir.path().into(), 10_000, db)?;

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![5, 6, 7])),
            [0, 1].as_ref().into(),
        );
        let key = *contract.key();
        let params: Parameters = [0, 1].as_ref().into();
        let container = ContractContainer::Wasm(ContractWasmAPIVersion::V1(contract));

        // Store and verify fetchable
        store.store_contract(container)?;
        assert!(
            store.fetch_contract(&key, &params).is_some(),
            "Contract should be fetchable after store"
        );

        // Remove contract
        store.remove_contract(&key)?;

        // Must NOT be fetchable from cache
        assert!(
            store.fetch_contract(&key, &params).is_none(),
            "Removed contract must not be served from cache (ghost contract)"
        );

        Ok(())
    }

    /// Regression test for the latent shared-WASM deletion bug: removing one
    /// contract instance must NOT delete the `.wasm` blob while another
    /// instance still references the same code hash.
    ///
    /// Multiple `ContractInstanceId`s can share one code hash (e.g. every
    /// River chat room shares one room-contract WASM — see issue #2380).
    /// The on-disk blob is keyed by code hash, so an unconditional delete
    /// would break the surviving instances after a cache miss / restart.
    #[tokio::test]
    async fn test_remove_contract_keeps_shared_wasm() -> Result<(), Box<dyn std::error::Error>> {
        let contract_dir = crate::util::tests::get_temp_dir();
        std::fs::create_dir_all(contract_dir.path())?;
        let db = create_test_db(contract_dir.path()).await;
        let mut store = ContractStore::new(contract_dir.path().into(), 10_000, db)?;

        // Two instances: SAME code, DIFFERENT params -> same code_hash,
        // different ContractInstanceIds.
        let shared_code = vec![1, 2, 3, 4, 5];
        let params1 = Parameters::from(vec![1, 1, 1]);
        let params2 = Parameters::from(vec![2, 2, 2]);
        let contract1 = WrappedContract::new(
            Arc::new(ContractCode::from(shared_code.clone())),
            params1.clone(),
        );
        let contract2 = WrappedContract::new(
            Arc::new(ContractCode::from(shared_code.clone())),
            params2.clone(),
        );
        let key1 = *contract1.key();
        let key2 = *contract2.key();

        // Both instances share one code hash.
        assert_eq!(key1.code_hash(), key2.code_hash());

        store.store_contract(ContractContainer::Wasm(ContractWasmAPIVersion::V1(
            contract1,
        )))?;
        store.store_contract(ContractContainer::Wasm(ContractWasmAPIVersion::V1(
            contract2,
        )))?;

        let wasm_path = contract_dir
            .path()
            .join(key1.code_hash().encode())
            .with_extension("wasm");
        assert!(wasm_path.exists(), "WASM file should exist after store");

        // Remove the first instance — the shared WASM must survive because
        // the second instance still references the code hash.
        store.remove_contract(&key1)?;
        assert!(
            wasm_path.exists(),
            "Shared WASM file must NOT be deleted while another instance references it"
        );

        // The second instance must still be fetchable.
        assert!(
            store.fetch_contract(&key2, &params2).is_some(),
            "Surviving instance must still be fetchable after the other is removed"
        );

        Ok(())
    }

    /// Regression test for the cross-executor shared-WASM deletion bug.
    ///
    /// Each runtime-pool executor owns a SEPARATE `ContractStore` with its
    /// own in-memory `key_to_code_part` map, but they all share one ReDb
    /// `contract_index`. If `remove_contract` decided "is this code still
    /// referenced?" from its own in-memory map, an instance stored via a
    /// different executor would be invisible — and removing the locally
    /// known instance would wrongly delete the shared `.wasm` blob,
    /// corrupting the instance owned by the other executor.
    ///
    /// This test reproduces that: two `ContractStore`s sharing ONE `db`,
    /// each storing a different instance of the SAME code. Removing one
    /// instance via store A must NOT delete the blob, and the instance
    /// stored via store B must remain fetchable.
    #[tokio::test]
    async fn test_remove_contract_keeps_shared_wasm_across_executors()
    -> Result<(), Box<dyn std::error::Error>> {
        let contract_dir = crate::util::tests::get_temp_dir();
        std::fs::create_dir_all(contract_dir.path())?;
        let db = create_test_db(contract_dir.path()).await;

        // Two separate ContractStores (simulating two runtime-pool
        // executors) sharing ONE db / ReDb contract_index.
        let mut store_a = ContractStore::new(contract_dir.path().into(), 10_000, db.clone())?;
        let mut store_b = ContractStore::new(contract_dir.path().into(), 10_000, db)?;

        // Same code, different params -> same code_hash, different
        // ContractInstanceIds.
        let shared_code = vec![10, 20, 30, 40, 50];
        let params1 = Parameters::from(vec![1, 1, 1]);
        let params2 = Parameters::from(vec![2, 2, 2]);
        let contract1 = WrappedContract::new(
            Arc::new(ContractCode::from(shared_code.clone())),
            params1.clone(),
        );
        let contract2 = WrappedContract::new(
            Arc::new(ContractCode::from(shared_code.clone())),
            params2.clone(),
        );
        let key1 = *contract1.key();
        let key2 = *contract2.key();
        assert_eq!(key1.code_hash(), key2.code_hash());

        // Instance X1 stored via executor A, instance X2 via executor B.
        store_a.store_contract(ContractContainer::Wasm(ContractWasmAPIVersion::V1(
            contract1,
        )))?;
        store_b.store_contract(ContractContainer::Wasm(ContractWasmAPIVersion::V1(
            contract2,
        )))?;

        let wasm_path = contract_dir
            .path()
            .join(key1.code_hash().encode())
            .with_extension("wasm");
        assert!(wasm_path.exists(), "WASM file should exist after store");

        // Remove X1 via store A. Store A's in-memory map never saw X2, so
        // the OLD code would delete the shared blob here.
        store_a.remove_contract(&key1)?;

        assert!(
            wasm_path.exists(),
            "Shared WASM file must NOT be deleted while another executor's \
             ContractStore still references the code hash"
        );

        // X2, owned by executor B, must still be fetchable.
        assert!(
            store_b.fetch_contract(&key2, &params2).is_some(),
            "Instance stored via another executor must survive removal of a \
             different instance sharing the same code"
        );

        Ok(())
    }

    /// Removing the last instance referencing a code hash must delete the
    /// `.wasm` blob from disk to reclaim space.
    #[tokio::test]
    async fn test_remove_contract_deletes_last_instance_wasm()
    -> Result<(), Box<dyn std::error::Error>> {
        let contract_dir = crate::util::tests::get_temp_dir();
        std::fs::create_dir_all(contract_dir.path())?;
        let db = create_test_db(contract_dir.path()).await;
        let mut store = ContractStore::new(contract_dir.path().into(), 10_000, db)?;

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![9, 8, 7, 6])),
            [4, 2].as_ref().into(),
        );
        let key = *contract.key();
        store.store_contract(ContractContainer::Wasm(ContractWasmAPIVersion::V1(
            contract,
        )))?;

        let wasm_path = contract_dir
            .path()
            .join(key.code_hash().encode())
            .with_extension("wasm");
        assert!(wasm_path.exists(), "WASM file should exist after store");

        // Removing the only instance must delete the blob.
        store.remove_contract(&key)?;
        assert!(
            !wasm_path.exists(),
            "WASM file must be deleted when the last instance is removed"
        );

        Ok(())
    }

    /// `remove_contract` must tolerate an already-missing `.wasm` file
    /// (idempotent file removal).
    #[tokio::test]
    async fn test_remove_contract_idempotent_when_file_missing()
    -> Result<(), Box<dyn std::error::Error>> {
        let contract_dir = crate::util::tests::get_temp_dir();
        std::fs::create_dir_all(contract_dir.path())?;
        let db = create_test_db(contract_dir.path()).await;
        let mut store = ContractStore::new(contract_dir.path().into(), 10_000, db)?;

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![3, 3, 3])),
            [7, 7].as_ref().into(),
        );
        let key = *contract.key();
        store.store_contract(ContractContainer::Wasm(ContractWasmAPIVersion::V1(
            contract,
        )))?;

        // Manually delete the WASM file out from under the store.
        let wasm_path = contract_dir
            .path()
            .join(key.code_hash().encode())
            .with_extension("wasm");
        std::fs::remove_file(&wasm_path)?;

        // remove_contract must still succeed despite the missing file.
        store
            .remove_contract(&key)
            .expect("remove_contract must be Ok when the WASM file is already gone");

        Ok(())
    }

    /// Regression test for Codex's round-4 finding: the `contract_cache` is
    /// per-`ContractStore` (one per pool executor), but the `.wasm` blob on
    /// disk is shared. A sibling executor's `remove_contract` can delete the
    /// blob without invalidating this store's cache. A subsequent
    /// `store_contract` for a new instance with the same code hash must NOT
    /// take the cache-hit fast path silently — it must verify the blob still
    /// exists on disk and re-write it if not. Otherwise the new instance is
    /// indexed but blobless until the cache evicts and a fetch fails.
    #[tokio::test]
    async fn test_store_contract_rewrites_blob_when_cache_hit_but_disk_missing()
    -> Result<(), Box<dyn std::error::Error>> {
        let contract_dir = crate::util::tests::get_temp_dir();
        std::fs::create_dir_all(contract_dir.path())?;
        let db = create_test_db(contract_dir.path()).await;
        let mut store = ContractStore::new(contract_dir.path().into(), 10_000, db)?;

        let shared_code = vec![9, 9, 9, 9];

        // Store instance X1: caches the code AND writes the blob.
        let x1 = WrappedContract::new(
            Arc::new(ContractCode::from(shared_code.clone())),
            [1, 1].as_ref().into(),
        );
        let x1_code_hash = *x1.key().code_hash();
        store.store_contract(ContractContainer::Wasm(ContractWasmAPIVersion::V1(x1)))?;

        let wasm_path = contract_dir
            .path()
            .join(x1_code_hash.encode())
            .with_extension("wasm");
        assert!(wasm_path.exists(), "blob must exist after first store");

        // Simulate a sibling executor's `remove_contract` deleting the shared
        // blob WITHOUT invalidating this store's `contract_cache`.
        std::fs::remove_file(&wasm_path)?;
        assert!(!wasm_path.exists());
        assert!(
            store.contract_cache.get(&x1_code_hash).is_some(),
            "this store's cache still has the code (sibling did not invalidate it)"
        );

        // Store a NEW instance X2 with the SAME code hash. Pre-fix, the
        // cache-hit fast path returned Ok without re-writing the blob,
        // leaving X2 indexed but blobless. Post-fix, the disk-existence
        // check forces a re-write.
        let x2 = WrappedContract::new(
            Arc::new(ContractCode::from(shared_code)),
            [2, 2].as_ref().into(),
        );
        let x2_key = *x2.key();
        let params: Parameters = [2, 2].as_ref().into();
        store.store_contract(ContractContainer::Wasm(ContractWasmAPIVersion::V1(x2)))?;

        assert!(
            wasm_path.exists(),
            "blob must be re-written after store_contract for a new instance \
             whose shared code was in cache but missing from disk"
        );
        // And the new instance is fetchable end-to-end.
        assert!(
            store.fetch_contract(&x2_key, &params).is_some(),
            "new instance must be fetchable after rewrite"
        );

        Ok(())
    }
}