calimero-node-primitives 0.11.0-rc.2

//! Application installation, uninstallation, and bundle extraction.

use std::io::{self, ErrorKind};
use std::sync::Arc;

use super::bundle;
use crate::bundle::{BundleManifest, ManifestVerification};
use calimero_primitives::application::{
    Application, ApplicationBlob, ApplicationId, ApplicationSource,
};
use calimero_primitives::blobs::BlobId;
use calimero_primitives::hash::Hash;
use calimero_store::{key, types};
use camino::{Utf8Path, Utf8PathBuf};
use eyre::bail;
use flate2::read::GzDecoder;
use futures_util::{io::Cursor, TryStreamExt};
use reqwest::Url;
use sha2::{Digest, Sha256};
use std::fs;
use tar::Archive;
use tokio::fs::File;
use tokio_util::compat::TokioAsyncReadCompatExt;
use tracing::{debug, trace, warn};

use crate::client::NodeClient;

const MAX_ERROR_BODY_LEN: usize = 256;

impl NodeClient {
    pub fn install_raw_wasm(
        &self,
        blob_id: &BlobId,
        size: u64,
        source: &ApplicationSource,
        metadata: Vec<u8>,
        package: &str,
        version: &str,
    ) -> eyre::Result<ApplicationId> {
        let application = types::ApplicationMeta::new(
            key::BlobMeta::new(*blob_id),
            size,
            source.to_string().into_boxed_str(),
            metadata.into_boxed_slice(),
            key::BlobMeta::new(BlobId::from([0; 32])),
            package.to_owned().into_boxed_str(),
            version.to_owned().into_boxed_str(),
            "".to_owned().into_boxed_str(),
        );

        let application_id = {
            let components = (
                application.bytecode,
                application.size,
                &application.source,
                &application.metadata,
            );
            ApplicationId::from(*Hash::hash_borsh(&components)?)
        };

        let mut handle = self.datastore.handle();
        let key = key::ApplicationMeta::new(application_id);
        handle.put(&key, &application)?;
        Ok(application_id)
    }

    fn install_bundle_application(
        &self,
        blob_id: &BlobId,
        size: u64,
        source: &ApplicationSource,
        metadata: Vec<u8>,
        package: &str,
        version: &str,
        signer_id: &str,
        services: Vec<types::ServiceMeta>,
    ) -> eyre::Result<ApplicationId> {
        let mut application = types::ApplicationMeta::new(
            key::BlobMeta::new(*blob_id),
            size,
            source.to_string().into_boxed_str(),
            metadata.into_boxed_slice(),
            key::BlobMeta::new(BlobId::from([0; 32])),
            package.to_owned().into_boxed_str(),
            version.to_owned().into_boxed_str(),
            signer_id.to_owned().into_boxed_str(),
        );
        application.services = services;

        let application_id = {
            let components = (&application.package, &application.signer_id);
            ApplicationId::from(*Hash::hash_borsh(&components)?)
        };

        let mut handle = self.datastore.handle();
        let key = key::ApplicationMeta::new(application_id);
        handle.put(&key, &application)?;
        Ok(application_id)
    }

    /// Install a bundle given an already-resolved manifest and verification.
    async fn install_bundle_with_manifest(
        &self,
        bundle_data: Arc<Vec<u8>>,
        blob_id: &BlobId,
        stored_size: u64,
        source: &ApplicationSource,
        verification: ManifestVerification,
        manifest: BundleManifest,
    ) -> eyre::Result<ApplicationId> {
        let signer_id = verification.signer_id;
        let package = &manifest.package;
        let version = &manifest.app_version;

        let blobstore_root = self.blob_manager.root_path();
        let node_root = blobstore_root
            .parent()
            .ok_or_else(|| eyre::eyre!("blobstore root has no parent"))?
            .to_path_buf();
        let extract_dir = node_root
            .join("applications")
            .join(package)
            .join(version)
            .join("extracted");

        let bundle_data_clone = Arc::clone(&bundle_data);
        let manifest_clone = manifest.clone();
        let extract_dir_clone = extract_dir.clone();
        let node_root_clone = node_root.clone();
        let package_clone = package.to_string();
        let version_clone = version.to_string();
        tokio::task::spawn_blocking(move || {
            Self::extract_bundle_artifacts(
                &bundle_data_clone,
                &manifest_clone,
                &extract_dir_clone,
                &node_root_clone,
                &package_clone,
                &version_clone,
            )
        })
        .await??;

        let mut services = Vec::new();
        for artifact in manifest.wasm_artifacts() {
            if let Some(name) = artifact.name {
                let wasm_path = extract_dir.join(&artifact.wasm.path);
                let wasm_bytes = tokio::fs::read(&wasm_path).await?;
                let cursor = Cursor::new(wasm_bytes.as_slice());
                let (svc_blob_id, _svc_size) = self
                    .add_blob(cursor, Some(wasm_bytes.len() as u64), None)
                    .await?;
                services.push(types::ServiceMeta {
                    name: name.to_owned().into_boxed_str(),
                    bytecode: key::BlobMeta::new(svc_blob_id),
                    compiled: key::BlobMeta::new(BlobId::from([0; 32])),
                });
            }
        }

        let bundle_metadata = manifest.to_metadata_json()?;

        self.install_bundle_application(
            blob_id,
            stored_size,
            source,
            bundle_metadata,
            package,
            version,
            &signer_id,
            services,
        )
    }

    /// Install a bundle from a registry. Signature is mandatory.
    async fn install_verified_bundle(
        &self,
        bundle_data: Arc<Vec<u8>>,
        blob_id: &BlobId,
        stored_size: u64,
        source: &ApplicationSource,
    ) -> eyre::Result<ApplicationId> {
        let bundle_data_clone = Arc::clone(&bundle_data);
        let (verification, manifest) = tokio::task::spawn_blocking(move || {
            bundle::verify_and_extract_manifest(&bundle_data_clone)
        })
        .await??;

        self.install_bundle_with_manifest(
            bundle_data,
            blob_id,
            stored_size,
            source,
            verification,
            manifest,
        )
        .await
    }

    /// Install a bundle from a local dev path.
    ///
    /// Signature is optional: verified if present, unsigned allowed otherwise.
    /// Production installs from registries go through `install_verified_bundle`
    /// which always requires a valid signature.
    async fn install_dev_bundle(
        &self,
        bundle_data: Arc<Vec<u8>>,
        blob_id: &BlobId,
        stored_size: u64,
        source: &ApplicationSource,
    ) -> eyre::Result<ApplicationId> {
        let bundle_data_clone = Arc::clone(&bundle_data);
        let (verification, manifest) = tokio::task::spawn_blocking(move || {
            bundle::extract_manifest_allow_unsigned(&bundle_data_clone)
        })
        .await??;

        self.install_bundle_with_manifest(
            bundle_data,
            blob_id,
            stored_size,
            source,
            verification,
            manifest,
        )
        .await
    }

    /// Check if a path points to a bundle archive (.mpk - Mero Package Kit)
    fn is_bundle_archive(path: &Utf8Path) -> bool {
        path.extension().map(|ext| ext == "mpk").unwrap_or(false)
    }

    pub async fn install_application_from_path(
        &self,
        path: Utf8PathBuf,
        metadata: Vec<u8>,
        package: Option<String>,
        version: Option<String>,
    ) -> eyre::Result<ApplicationId> {
        let metadata_len = metadata.len();
        debug!(
            path = %path,
            metadata_len,
            "install_application_from_path started"
        );

        let path = match path.canonicalize_utf8() {
            Ok(canonicalized) => canonicalized,
            Err(err) if err.kind() == ErrorKind::NotFound => {
                bail!("application file not found at {}", path);
            }
            Err(err) => return Err(err.into()),
        };
        trace!(path = %path, "application path canonicalized");

        // Detect bundle vs single WASM
        if bundle::is_bundle_archive(&path) {
            return self.install_bundle_from_path(path, metadata).await;
        }

        // For non-bundle installations, use provided package/version or defaults
        let package = package.as_deref().unwrap_or("unknown");
        let version = version.as_deref().unwrap_or("0.0.0");

        // Existing single WASM installation path
        let file = match File::open(&path).await {
            Ok(file) => file,
            Err(err) if err.kind() == ErrorKind::NotFound => {
                bail!("application file not found at {}", path);
            }
            Err(err) => return Err(err.into()),
        };
        trace!(path = %path, "application file opened");

        let expected_size = file.metadata().await?.len();
        debug!(
            path = %path,
            expected_size,
            "install_application_from_path discovered file size"
        );

        let (blob_id, size) = self
            .add_blob(file.compat(), Some(expected_size), None)
            .await?;
        debug!(
            %blob_id,
            expected_size,
            stored_size = size,
            "application blob added via add_blob"
        );

        let Ok(uri) = Url::from_file_path(path) else {
            bail!("non-absolute path")
        };

        self.install_raw_wasm(
            &blob_id,
            size,
            &uri.as_str().parse()?,
            metadata,
            package,
            version,
        )
    }

    pub async fn install_application_from_url(
        &self,
        url: Url,
        metadata: Vec<u8>,
        expected_hash: Option<&Hash>,
    ) -> eyre::Result<ApplicationId> {
        let uri = url.as_str().parse()?;

        let response = reqwest::Client::new().get(url.clone()).send().await?;

        if !response.status().is_success() {
            let status = response.status();
            let body = response
                .text()
                .await
                .unwrap_or_else(|_| "<unreadable body>".to_owned());
            // Truncate to avoid leaking sensitive data that may appear in error responses.
            let truncated: String = body.chars().take(MAX_ERROR_BODY_LEN).collect();
            eyre::bail!(
                "Registry returned HTTP {} for {}: {}",
                status,
                url,
                truncated
            );
        }

        let expected_size = response.content_length();

        // Check if URL indicates a bundle archive (.mpk - Mero Package Kit)
        let is_bundle = url.path().ends_with(".mpk");

        if is_bundle {
            let bundle_data = Arc::new(response.bytes().await?.to_vec());

            let cursor = Cursor::new(bundle_data.as_slice());
            let (bundle_blob_id, stored_size) = self
                .add_blob(cursor, Some(bundle_data.len() as u64), expected_hash)
                .await?;

            debug!(
                %bundle_blob_id,
                bundle_size = bundle_data.len(),
                stored_size,
                "bundle downloaded and stored as blob"
            );

            return self
                .install_verified_bundle(bundle_data, &bundle_blob_id, stored_size, &uri)
                .await;
        }

        // Single WASM installation (existing behavior)
        // For non-bundle installations, use defaults (package/version are not part of ApplicationId)
        let package = "unknown";
        let version = "0.0.0";

        let (blob_id, size) = self
            .add_blob(
                response
                    .bytes_stream()
                    .map_err(io::Error::other)
                    .into_async_read(),
                expected_size,
                expected_hash,
            )
            .await?;

        self.install_raw_wasm(&blob_id, size, &uri, metadata, package, version)
    }

    /// Install a bundle archive (.mpk - Mero Package Kit) containing WASM, ABI, and migrations
    /// Note: metadata parameter is ignored for bundles - metadata is always extracted from manifest
    async fn install_bundle_from_path(
        &self,
        path: Utf8PathBuf,
        _metadata: Vec<u8>,
    ) -> eyre::Result<ApplicationId> {
        debug!(path = %path, "install_bundle_from_path started");

        let bundle_data = Arc::new(tokio::fs::read(&path).await?);

        let cursor = Cursor::new(bundle_data.as_slice());
        let (bundle_blob_id, stored_size) = self
            .add_blob(cursor, Some(bundle_data.len() as u64), None)
            .await?;

        debug!(
            %bundle_blob_id,
            bundle_size = bundle_data.len(),
            stored_size,
            "bundle stored as blob"
        );

        let Ok(uri) = Url::from_file_path(&path) else {
            bail!("non-absolute path")
        };

        // Dev installs from local path skip signature verification — the
        // bundle may be unsigned during development/testing.
        self.install_dev_bundle(
            bundle_data,
            &bundle_blob_id,
            stored_size,
            &uri.as_str().parse()?,
        )
        .await
    }

    // Bundle verification, manifest extraction, and path validation functions
    // are in the `bundle` submodule. The methods below delegate for backward
    // compatibility with external callers that use `NodeClient::method()`.

    /// Check if a blob contains a bundle archive.
    /// Delegates to [`bundle::is_bundle_blob`].
    pub fn is_bundle_blob(blob_bytes: &[u8]) -> bool {
        bundle::is_bundle_blob(blob_bytes)
    }

    /// Install an application from a bundle blob that's already in the blobstore.
    /// This is used when a bundle blob is received via blob sharing or discovery.
    pub async fn install_application_from_bundle_blob(
        &self,
        blob_id: &BlobId,
        source: &ApplicationSource,
    ) -> eyre::Result<ApplicationId> {
        debug!(%blob_id, "install_application_from_bundle_blob started");

        let Some(bundle_bytes) = self.get_blob_bytes(blob_id, None).await? else {
            bail!("bundle blob not found");
        };

        let stored_size = bundle_bytes.len() as u64;
        let bundle_data = Arc::new(bundle_bytes.to_vec());

        self.install_verified_bundle(bundle_data, blob_id, stored_size, source)
            .await
    }

    /// Find duplicate artifact in other versions by hash and relative path
    /// Only matches files with the same relative path within the bundle to avoid
    /// collisions between files with the same name in different directories
    fn find_duplicate_artifact(
        node_root: &Utf8Path,
        package: &str,
        current_version: &str,
        hash: &[u8; 32],
        relative_path: &str,
    ) -> Option<Utf8PathBuf> {
        // Check other versions for the same hash at the same relative path
        let package_dir = node_root.join("applications").join(package);

        if let Ok(entries) = fs::read_dir(package_dir.as_std_path()) {
            for entry in entries.flatten() {
                if let Ok(version_name) = entry.file_name().into_string() {
                    if version_name == current_version {
                        continue; // Skip current version
                    }

                    // Check extracted directory in this version at the same relative path
                    let extracted_dir = package_dir.join(&version_name).join("extracted");
                    let candidate_path = extracted_dir.join(relative_path);

                    if candidate_path.exists() {
                        // Compute hash of candidate file
                        if let Ok(candidate_content) = fs::read(candidate_path.as_std_path()) {
                            let candidate_hash = Sha256::digest(&candidate_content);
                            let candidate_array: [u8; 32] = candidate_hash.into();
                            if candidate_array == *hash {
                                return Some(candidate_path);
                            }
                        }
                    }
                }
            }
        }

        None
    }

    /// Extract bundle artifacts with deduplication
    ///
    /// This function is synchronized per package-version to prevent race conditions
    /// when multiple concurrent calls try to extract the same bundle.
    pub(crate) fn extract_bundle_artifacts(
        bundle_data: &[u8],
        _manifest: &BundleManifest,
        extract_dir: &Utf8Path,
        node_root: &Utf8Path,
        package: &str,
        current_version: &str,
    ) -> eyre::Result<()> {
        // Create extraction directory
        fs::create_dir_all(extract_dir)?;

        // Use a lock file to prevent concurrent extraction of the same bundle version
        // Lock file path: extract_dir/.extracting.lock
        let lock_file_path = extract_dir.join(".extracting.lock");
        let marker_file_path = extract_dir.join(".extracted");

        // Check if extraction is already complete
        // Only skip if marker exists AND the expected WASM file exists
        // This handles the case where files were deleted but marker remains
        if marker_file_path.exists() {
            // Check if WASM file exists (using manifest to determine path)
            // If marker exists but WASM doesn't, marker is stale - remove it and re-extract
            let wasm_relative_path = _manifest
                .wasm
                .as_ref()
                .map(|w| w.path.as_str())
                .unwrap_or("app.wasm");

            // Validate WASM path to prevent path traversal attacks before checking existence
            if wasm_relative_path.contains("..") {
                bail!(
                    "WASM path traversal detected in manifest: {} contains '..' component",
                    wasm_relative_path
                );
            }

            let wasm_path = extract_dir.join(wasm_relative_path);

            // Additional validation: ensure the resolved path stays within extract_dir
            if wasm_path.exists() {
                // Validate path traversal even if file exists
                let canonical_wasm = wasm_path.canonicalize_utf8()?;

                // extract_dir might not exist if wasm_relative_path contains subdirectories
                // Reconstruct canonical extract_dir from wasm_path by removing relative path components
                let canonical_extract = if extract_dir.exists() {
                    extract_dir.canonicalize_utf8()?
                } else {
                    // Reconstruct extract_dir from wasm_path by removing wasm_relative_path components
                    // Since we validated wasm_relative_path doesn't contain "..", this is safe
                    let wasm_parent = wasm_path
                        .parent()
                        .ok_or_else(|| eyre::eyre!("WASM path has no parent directory"))?;
                    let wasm_parent_canonical = wasm_parent.canonicalize_utf8()?;

                    // Count depth of wasm_relative_path (number of path components)
                    let relative_depth = wasm_relative_path
                        .split('/')
                        .filter(|s| !s.is_empty())
                        .count()
                        .saturating_sub(1); // Subtract 1 for the filename itself

                    // Go up relative_depth levels from wasm_parent to get extract_dir
                    let mut canonical_extract_candidate = wasm_parent_canonical.clone();
                    for _ in 0..relative_depth {
                        if let Some(parent) = canonical_extract_candidate.parent() {
                            canonical_extract_candidate = parent.to_path_buf();
                        } else {
                            bail!("Cannot reconstruct extract_dir from WASM path");
                        }
                    }

                    canonical_extract_candidate.try_into().map_err(|_| {
                        eyre::eyre!("Failed to convert extract_dir path to Utf8PathBuf")
                    })?
                };

                if !canonical_wasm.starts_with(&canonical_extract) {
                    bail!(
                        "WASM path traversal detected: {} escapes extraction directory {}",
                        wasm_relative_path,
                        extract_dir
                    );
                }

                debug!(
                    package,
                    version = current_version,
                    "Bundle already extracted (marker file and WASM exist), skipping"
                );
                return Ok(());
            } else {
                // Marker exists but WASM doesn't - remove stale marker and re-extract
                debug!(
                    package,
                    version = current_version,
                    "Marker file exists but WASM not found, removing stale marker"
                );
                let _ = fs::remove_file(&marker_file_path);
            }
        }

        // Try to acquire exclusive lock by creating lock file atomically
        // create_new() is atomic - fails if file exists (works on Unix and Windows)
        let lock_acquired = match std::fs::OpenOptions::new()
            .create_new(true)
            .write(true)
            .open(lock_file_path.as_std_path())
        {
            Ok(_) => {
                // Lock file created - we're the first to extract
                true
            }
            Err(_) => {
                // Lock file already exists - another extraction is in progress
                // Wait and check if extraction completes
                for _ in 0..20 {
                    std::thread::sleep(std::time::Duration::from_millis(100));
                    if marker_file_path.exists() {
                        debug!(
                            package,
                            version = current_version,
                            "Bundle extraction completed by another process"
                        );
                        return Ok(());
                    }
                }
                // If marker still doesn't exist after waiting, proceed anyway
                // (lock file might be stale from crashed process)
                warn!(
                    package,
                    version = current_version,
                    "Lock file exists but extraction not complete, proceeding anyway"
                );
                false
            }
        };

        // Track if we created a lock file that needs cleanup
        let mut lock_created_by_us = lock_acquired;

        // Only proceed with extraction if we acquired the lock
        // (or if lock is stale and we're proceeding anyway)
        if !lock_acquired {
            // Try to remove stale lock and retry
            let _ = fs::remove_file(&lock_file_path);
            // Check marker one more time
            if marker_file_path.exists() {
                return Ok(());
            }
            // Create lock file again - handle race condition where another thread
            // might have created it between removal and this creation
            match std::fs::OpenOptions::new()
                .create_new(true)
                .write(true)
                .open(lock_file_path.as_std_path())
            {
                Ok(_) => {
                    // Successfully acquired lock, proceed with extraction
                    lock_created_by_us = true;
                }
                Err(_) => {
                    // Another thread created the lock between removal and creation
                    // Wait briefly and check if extraction completed
                    std::thread::sleep(std::time::Duration::from_millis(100));
                    if marker_file_path.exists() {
                        debug!(
                            package,
                            version = current_version,
                            "Bundle extraction completed by another process after lock retry"
                        );
                        return Ok(());
                    }
                    // If marker still doesn't exist, the other thread is still extracting
                    // Return error to avoid concurrent extraction
                    bail!(
                        "Failed to acquire extraction lock after retry - another process is extracting"
                    );
                }
            }
        }

        // Ensure lock file is cleaned up even if extraction fails
        // Use a guard to clean up on early return or error
        struct LockGuard {
            path: Utf8PathBuf,
            should_remove: std::cell::Cell<bool>,
        }

        impl Drop for LockGuard {
            fn drop(&mut self) {
                if self.should_remove.get() {
                    let _ = fs::remove_file(&self.path);
                }
            }
        }

        let lock_guard = LockGuard {
            path: lock_file_path.clone(),
            should_remove: std::cell::Cell::new(lock_created_by_us),
        };

        let tar = GzDecoder::new(bundle_data);
        let mut archive = Archive::new(tar);

        // Extract all files from bundle
        for entry_result in archive.entries()? {
            let mut entry = entry_result?;

            // Extract path_bytes first, converting to owned to drop borrow
            let path_bytes_owned = {
                let header = entry.header();
                header.path_bytes().into_owned()
            };

            let relative_path = {
                let path_str = std::str::from_utf8(&path_bytes_owned)
                    .map_err(|_| eyre::eyre!("invalid UTF-8 in file path"))?;
                path_str.to_string()
            };

            // Skip directory entries and macOS resource fork files
            if entry.header().entry_type().is_dir() {
                continue;
            }
            if let Some(file_name) = std::path::Path::new(&relative_path)
                .file_name()
                .and_then(|n| n.to_str())
            {
                if file_name.starts_with("._") {
                    continue;
                }
            }

            // Read content (header borrow is dropped)
            let mut content = Vec::new();
            std::io::copy(&mut entry, &mut content)?;

            // Preserve directory structure from bundle
            let dest_path = extract_dir.join(&relative_path);

            // Validate path to prevent path traversal attacks
            // Check that the relative path doesn't contain ".." components that would escape
            if relative_path.contains("..") {
                bail!(
                    "Path traversal detected: {} contains '..' component",
                    relative_path
                );
            }

            // Additional validation: ensure the resolved path stays within extract_dir
            // Always validate by constructing expected path, regardless of whether it exists
            // This prevents path traversal even when parent directories don't exist yet
            let canonical_extract = extract_dir.canonicalize_utf8()?;

            // Construct what the canonical dest_path should be
            // Since we already checked relative_path doesn't contain "..",
            // joining extract_dir with relative_path is safe
            let expected_dest = canonical_extract.join(&relative_path);

            // Verify the expected path stays within extract_dir
            // This works even if the path doesn't exist yet because we're constructing
            // it from the canonical extract_dir and a validated relative_path
            if !expected_dest.starts_with(&canonical_extract) {
                bail!(
                    "Path traversal detected: {} would escape extraction directory {}",
                    relative_path,
                    extract_dir
                );
            }

            // If dest_path exists, also verify the actual canonicalized path matches expected
            if dest_path.exists() {
                let canonical_dest = dest_path.canonicalize_utf8()?;
                if !canonical_dest.starts_with(&canonical_extract) {
                    bail!(
                        "Path traversal detected: {} escapes extraction directory {}",
                        relative_path,
                        extract_dir
                    );
                }
            }

            // Create parent directories if needed
            if let Some(parent) = dest_path.parent() {
                fs::create_dir_all(parent)?;
            }

            // Compute hash
            let hash = Sha256::digest(&content);
            let hash_array: [u8; 32] = hash.into();

            // Check for duplicates in other versions at the same relative path
            if let Some(duplicate_path) = Self::find_duplicate_artifact(
                node_root,
                package,
                current_version,
                &hash_array,
                &relative_path,
            ) {
                // Create hardlink to duplicate file
                if let Err(e) = fs::hard_link(duplicate_path.as_std_path(), dest_path.as_std_path())
                {
                    // If hardlink fails (e.g., cross-filesystem), fall back to copying
                    warn!(
                        file = %relative_path,
                        duplicate = %duplicate_path,
                        error = %e,
                        "hardlink failed, copying instead"
                    );
                    fs::write(&dest_path, &content)?;
                } else {
                    debug!(
                        file = %relative_path,
                        hash = hex::encode(hash),
                        duplicate = %duplicate_path,
                        "deduplicated artifact via hardlink"
                    );
                }
            } else {
                // No duplicate found, write new file
                fs::write(&dest_path, &content)?;
                debug!(
                    file = %relative_path,
                    hash = hex::encode(hash),
                    "extracted artifact"
                );
            }
        }

        // Write marker file to indicate extraction is complete
        fs::write(&marker_file_path, b"extracted")?;

        // Remove lock file explicitly on success (guard will skip removal if we already did it)
        if lock_guard.should_remove.get() {
            let _ = fs::remove_file(&lock_file_path);
            lock_guard.should_remove.set(false); // Prevent guard from removing it again
        }

        Ok(())
    }

    pub fn uninstall_application(&self, application_id: &ApplicationId) -> eyre::Result<()> {
        let mut handle = self.datastore.handle();

        let key = key::ApplicationMeta::new(*application_id);

        // Get application metadata before deleting to check if it's a bundle
        let application_meta = handle.get(&key)?;

        // Delete the ApplicationMeta entry
        handle.delete(&key)?;

        // Clean up extracted bundle files if this is a bundle
        if let Some(application) = application_meta {
            // Check if this is a bundle by checking package/version
            // Bundles have meaningful package/version (not "unknown"/"0.0.0")
            let is_bundle = application.package.as_ref() != "unknown"
                && application.version.as_ref() != "0.0.0";

            if is_bundle {
                // Construct path to extracted bundle directory
                let blobstore_root = self.blob_manager.root_path();
                let node_root = blobstore_root
                    .parent()
                    .ok_or_else(|| eyre::eyre!("blobstore root has no parent"))?;
                let bundle_dir = node_root
                    .join("applications")
                    .join(application.package.as_ref())
                    .join(application.version.as_ref());

                // Delete the entire version directory (includes extracted/ subdirectory)
                if bundle_dir.exists() {
                    debug!(
                        package = %application.package,
                        version = %application.version,
                        path = %bundle_dir,
                        "Removing extracted bundle directory"
                    );
                    if let Err(e) = fs::remove_dir_all(bundle_dir.as_std_path()) {
                        warn!(
                            package = %application.package,
                            version = %application.version,
                            path = %bundle_dir,
                            error = %e,
                            "Failed to remove extracted bundle directory"
                        );
                        // Don't fail uninstallation if cleanup fails - metadata is already deleted
                    } else {
                        debug!(
                            package = %application.package,
                            version = %application.version,
                            "Successfully removed extracted bundle directory"
                        );
                    }

                    // Also try to remove parent package directory if it's empty
                    let package_dir = node_root
                        .join("applications")
                        .join(application.package.as_ref());
                    if package_dir.exists() {
                        // Check if package directory is empty
                        if let Ok(mut entries) = fs::read_dir(package_dir.as_std_path()) {
                            if entries.next().is_none() {
                                // Directory is empty, remove it
                                if let Err(e) = fs::remove_dir(package_dir.as_std_path()) {
                                    debug!(
                                        package = %application.package,
                                        error = %e,
                                        "Failed to remove empty package directory (non-fatal)"
                                    );
                                }
                            }
                        }
                    }
                }
            }
        }

        Ok(())
    }

    // Query and management functions (list_applications, list_packages,
    // list_versions, get_latest_version, update_compiled_app) are in the
    // `query` submodule.

    /// Install application by package and version
    pub async fn install_by_package_version(
        &self,
        _package: &str,
        _version: &str,
        source: &ApplicationSource,
        metadata: Vec<u8>,
    ) -> eyre::Result<ApplicationId> {
        // For now, we'll use the source URL to download the application
        // In a real implementation, you might want to resolve the package/version to a URL
        let url = source.to_string().parse()?;
        self.install_application_from_url(url, metadata, None).await
    }
}