spin_loader/

local.rs

use std::path::{Path, PathBuf};

use anyhow::{Context, Result, anyhow, bail, ensure};
use futures::{StreamExt, future::try_join_all};
use reqwest::Url;
use spin_common::{paths::parent_dir, sloth, ui::quoted_path};
use spin_expressions::Resolver;
use spin_locked_app::{
    locked::{
        self, ContentPath, ContentRef, LockedApp, LockedComponent, LockedComponentDependency,
        LockedComponentSource, LockedTrigger,
    },
    values::{ValuesMap, ValuesMapBuilder},
};
use spin_manifest::schema::v2::{self, AppManifest, KebabId, WasiFilesMount};
use spin_outbound_networking_config::allowed_hosts::{AllowedHostConfig, AllowedHostsConfig};
use spin_serde::DependencyName;
use std::collections::BTreeMap;
use tokio::{io::AsyncWriteExt, sync::Semaphore};

use crate::{FilesMountStrategy, cache::Cache};

mod trigger_components;

#[derive(Debug)]
pub struct LocalLoader {
    app_root: PathBuf,
    files_mount_strategy: FilesMountStrategy,
    file_loading_permits: std::sync::Arc<Semaphore>,
    wasm_loader: WasmLoader,
    profile: Option<String>,
}

impl LocalLoader {
    pub async fn new(
        app_root: &Path,
        files_mount_strategy: FilesMountStrategy,
        profile: Option<&str>,
        cache_root: Option<PathBuf>,
    ) -> Result<Self> {
        let app_root = safe_canonicalize(app_root)
            .with_context(|| format!("Invalid manifest dir `{}`", app_root.display()))?;
        let file_loading_permits =
            std::sync::Arc::new(Semaphore::new(crate::MAX_FILE_LOADING_CONCURRENCY));
        Ok(Self {
            app_root: app_root.clone(),
            files_mount_strategy,
            // Limit concurrency to avoid hitting system resource limits
            file_loading_permits: file_loading_permits.clone(),
            wasm_loader: WasmLoader::new(app_root, cache_root, Some(file_loading_permits)).await?,
            profile: profile.map(|s| s.to_owned()),
        })
    }

    // Load the manifest file (spin.toml) at the given path into a LockedApp,
    // preparing all its content for execution.
    pub async fn load_file(&self, path: impl AsRef<Path>) -> Result<LockedApp> {
        // Parse manifest
        let path = path.as_ref();
        let manifest = spin_manifest::manifest_from_file(path).with_context(|| {
            format!(
                "Failed to read Spin app manifest from {}",
                quoted_path(path)
            )
        })?;
        let mut locked = self
            .load_manifest(manifest, self.profile())
            .await
            .with_context(|| format!("Failed to load Spin app from {}", quoted_path(path)))?;

        // Set origin metadata
        locked
            .metadata
            .insert("origin".into(), file_url(path)?.into());

        // Set build profile metadata
        if let Some(profile) = self.profile.as_ref() {
            locked
                .metadata
                .insert("profile".into(), profile.as_str().into());
        }

        Ok(locked)
    }

    // Load the given manifest into a LockedApp, ready for execution.
    pub(crate) async fn load_manifest(
        &self,
        mut manifest: AppManifest,
        profile: Option<&str>,
    ) -> Result<LockedApp> {
        spin_manifest::normalize::normalize_manifest(&mut manifest, profile)?;

        manifest.validate_dependencies()?;

        let AppManifest {
            spin_manifest_version: _,
            application,
            variables,
            triggers,
            components,
        } = manifest;

        let metadata = locked_metadata(application, triggers.keys().cloned())?;

        let variables = variables
            .into_iter()
            .map(|(name, v)| Ok((name.to_string(), locked_variable(v)?)))
            .collect::<Result<BTreeMap<_, _>>>()?;
        let resolver = Resolver::new(variables.clone())?;

        let triggers = try_join_all(triggers.into_iter().flat_map(|(trigger_type, configs)| {
            configs
                .into_iter()
                .map(|trigger| {
                    let trigger_type_2 = trigger_type.clone();
                    let id = trigger.id.clone();
                    async move {
                        self.load_trigger(&id, &trigger_type_2, trigger)
                            .await
                            .with_context(|| {
                                format!("Failed to load `{}` trigger `{}`", trigger_type_2, id)
                            })
                    }
                })
                .collect::<Vec<_>>()
        }))
        .await?;

        let sloth_guard = warn_if_component_load_slothful();

        // Load all components concurrently
        let components = try_join_all(components.into_iter().map(|(id, c)| {
            let resolver = &resolver;
            async move {
                self.load_component(&id, c, resolver)
                    .await
                    .with_context(|| format!("Failed to load component `{id}`"))
            }
        }))
        .await?;

        drop(sloth_guard);

        let locked = LockedApp {
            spin_lock_version: Default::default(),
            metadata,
            must_understand: vec![],
            host_requirements: spin_locked_app::values::ValuesMap::new(),
            variables,
            triggers,
            components,
        };

        let mut locked = trigger_components::reassign_trigger_deps(locked);

        if !locked.host_requirements.is_empty() {
            locked
                .must_understand
                .push(spin_locked_app::locked::MustUnderstand::HostRequirements);
        }
        if locked
            .components
            .iter()
            .any(|c| !c.host_requirements.is_empty())
        {
            locked
                .must_understand
                .push(spin_locked_app::locked::MustUnderstand::ComponentHostRequirements);
        }

        Ok(locked)
    }

    async fn load_trigger(
        &self,
        id: &str,
        trigger_type: &str,
        trigger: v2::Trigger,
    ) -> Result<LockedTrigger> {
        fn reference_id(spec: v2::ComponentSpec) -> toml::Value {
            let v2::ComponentSpec::Reference(id) = spec else {
                unreachable!("should have already been normalized");
            };
            id.as_ref().into()
        }

        let mut config = trigger.config;

        if let Some(id) = trigger.component.map(reference_id) {
            config.insert("component".into(), id);
        }

        let locked_deps = if !trigger.dependencies.is_empty() {
            self.load_trigger_dependencies(id, trigger.dependencies.iter())
                .await?
        } else {
            BTreeMap::new()
        };

        Ok(LockedTrigger {
            id: trigger.id,
            trigger_type: trigger_type.to_string(),
            trigger_config: config.try_into()?,
            trigger_dependencies: locked_deps,
        })
    }

    // Load the given component into a LockedComponent, ready for execution.
    async fn load_component(
        &self,
        id: &KebabId,
        component: v2::Component,
        resolver: &Resolver,
    ) -> Result<LockedComponent> {
        let allowed_outbound_hosts = component
            .normalized_allowed_outbound_hosts()
            .context("`allowed_http_hosts` is malformed")?;
        AllowedHostsConfig::validate(&allowed_outbound_hosts, resolver)
            .context("`allowed_outbound_hosts` is malformed")?;

        let component_requires_service_chaining = requires_service_chaining(&component);

        let metadata = ValuesMapBuilder::new()
            .string("description", component.description)
            .string_array("allowed_outbound_hosts", allowed_outbound_hosts)
            .string_array("key_value_stores", component.key_value_stores)
            .string_array("databases", component.sqlite_databases)
            .string_array("ai_models", component.ai_models)
            .serializable("build", component.build)?
            .take();

        let source = self
            .load_component_source(id, component.source.clone())
            .await
            .with_context(|| format!("Failed to load Wasm source {}", component.source))?;

        let dependencies = self
            .load_component_dependencies(id, &component.dependencies)
            .await?;

        let env = component.environment.into_iter().collect();

        let files = if component.files.is_empty() {
            vec![]
        } else {
            match &self.files_mount_strategy {
                FilesMountStrategy::Copy(files_mount_root) => {
                    let component_mount_root = files_mount_root.join(id.as_ref());
                    // Copy mounted files into component mount root, concurrently
                    try_join_all(component.files.iter().map(|f| {
                        self.copy_file_mounts(f, &component_mount_root, &component.exclude_files)
                    }))
                    .await?;

                    // All component files (copies) are in `component_mount_root` now
                    vec![ContentPath {
                        content: file_content_ref(component_mount_root)?,
                        path: "/".into(),
                    }]
                }
                FilesMountStrategy::Direct => {
                    ensure!(
                        component.exclude_files.is_empty(),
                        "Cannot load a component with `exclude_files` using --direct-mounts"
                    );
                    let mut files = vec![];
                    for mount in &component.files {
                        // Validate (and canonicalize) direct mount directory
                        files.push(self.resolve_direct_mount(mount).await?);
                    }
                    files
                }
            }
        };

        let config = component
            .variables
            .into_iter()
            .map(|(k, v)| (k.into(), v))
            .collect();

        let mut host_requirements = ValuesMapBuilder::new();
        if component_requires_service_chaining {
            host_requirements.string(
                spin_locked_app::locked::SERVICE_CHAINING_KEY,
                spin_locked_app::locked::HOST_REQ_REQUIRED,
            );
        }
        let host_requirements = host_requirements.build();

        Ok(LockedComponent {
            id: id.as_ref().into(),
            metadata,
            source,
            env,
            files,
            config,
            dependencies,
            trigger_dependencies: Default::default(), // We will fix this up later
            host_requirements,
        })
    }

    async fn load_component_dependencies(
        &self,
        id: &KebabId,
        dependencies: &v2::ComponentDependencies,
    ) -> Result<BTreeMap<DependencyName, LockedComponentDependency>> {
        Ok(try_join_all(dependencies.inner.iter().map(
            |(dependency_name, dependency)| async move {
                let locked_dependency = self
                    .load_component_dependency(dependency_name.clone(), dependency.clone())
                    .await
                    .with_context(|| {
                        format!(
                            "Failed to load component dependency `{dependency_name}` for `{id}`"
                        )
                    })?;

                anyhow::Ok((dependency_name.clone(), locked_dependency))
            },
        ))
        .await?
        .into_iter()
        .collect())
    }

    async fn load_component_dependency(
        &self,
        dependency_name: DependencyName,
        dependency: v2::ComponentDependency,
    ) -> Result<LockedComponentDependency> {
        self.wasm_loader
            .load_component_dependency(&dependency_name, &dependency)
            .await
    }

    async fn load_trigger_dependencies(
        &self,
        id: &str,
        dependencies: impl Iterator<Item = (&String, &v2::TriggerDependencies)>,
    ) -> Result<BTreeMap<String, Vec<LockedComponentDependency>>> {
        let mut loaded = BTreeMap::new();

        for (role, deps) in dependencies {
            let locked_deps = self.load_trigger_dependencies_vec(id, &deps.0).await?;
            loaded.insert(role.clone(), locked_deps);
        }

        Ok(loaded)
    }

    async fn load_trigger_dependencies_vec(
        &self,
        id: &str,
        dependencies: &[v2::TriggerDependency],
    ) -> Result<Vec<LockedComponentDependency>> {
        Ok(
            try_join_all(dependencies.iter().map(|dependency| async move {
                let locked_dependency = self
                    .load_trigger_dependency(dependency.clone())
                    .await
                    .with_context(|| {
                        format!("Failed to load trigger dependency `{dependency:?}` for `{id}`")
                    })?;

                anyhow::Ok(locked_dependency)
            }))
            .await?
            .into_iter()
            .collect(),
        )
    }

    async fn load_trigger_dependency(
        &self,
        dependency: v2::TriggerDependency,
    ) -> Result<LockedComponentDependency> {
        self.wasm_loader.load_trigger_dependency(&dependency).await
    }

    // Load a Wasm source from the given ContentRef and update the source
    // URL with an absolute path to the content.
    async fn load_component_source(
        &self,
        component_id: &KebabId,
        source: v2::ComponentSource,
    ) -> Result<LockedComponentSource> {
        let path = self
            .wasm_loader
            .load_component_source(component_id.as_ref(), &source)
            .await?;
        Ok(LockedComponentSource {
            content_type: "application/wasm".into(),
            content: file_content_ref(path)?,
        })
    }

    // Copy content(s) from the given `mount`
    async fn copy_file_mounts(
        &self,
        mount: &WasiFilesMount,
        dest_root: &Path,
        exclude_files: &[String],
    ) -> Result<()> {
        match mount {
            WasiFilesMount::Pattern(pattern) => {
                self.copy_glob_or_path(pattern, dest_root, exclude_files)
                    .await
            }
            WasiFilesMount::Placement {
                source,
                destination,
            } => {
                let src = Path::new(source);
                let dest = dest_root.join(destination.trim_start_matches('/'));
                self.copy_file_or_directory(src, &dest, destination, exclude_files)
                    .await
            }
        }
    }

    // Copy files matching glob pattern or single file/directory path.
    async fn copy_glob_or_path(
        &self,
        glob_or_path: &str,
        dest_root: &Path,
        exclude_files: &[String],
    ) -> Result<()> {
        if glob_or_path == ".." || glob_or_path.ends_with("/..") {
            bail!(
                "A file pattern can't end in a parent directory path (..)\nIf you want to include a directory, use source-destination form, or a glob pattern ending in **/*.\nLearn more: https://spinframework.dev/writing-apps#including-files-with-components"
            );
        }
        if glob_or_path == "." || glob_or_path.ends_with("/.") {
            bail!(
                "A file pattern can't end in a current directory path (.)\nIf you want to include a directory, use source-destination form, or a glob pattern ending in **/*.\nLearn more: https://spinframework.dev/writing-apps#including-files-with-components"
            );
        }

        if glob_or_path == "*" {
            tracing::warn!(
                alert_in_dev = true,
                "A component is including the entire application directory as asset files. This is unlikely to be what you want.\nIf this is what you want, use the pattern \"./*\" to avoid this warning.\nLearn more: https://spinframework.dev/writing-apps#including-files-with-components\n"
            );
        }
        if glob_or_path == "**/*" {
            tracing::warn!(
                alert_in_dev = true,
                "A component is including the entire application directory tree as asset files. This is unlikely to be what you want.\nIf this is what you want, use the pattern \"./**/*\" to avoid this warning.\nLearn more: https://spinframework.dev/writing-apps#including-files-with-components\n"
            );
        }

        let path = self.app_root.join(glob_or_path);
        if path.exists() {
            let dest = dest_root.join(glob_or_path);
            if path.is_dir() {
                // "single/dir"
                let pattern = path.join("**/*");
                self.copy_glob(&pattern, &self.app_root, &dest, exclude_files)
                    .await?;
            } else {
                // "single/file.txt"
                self.copy_single_file(&path, &dest, glob_or_path).await?;
            }
        } else if looks_like_glob_pattern(glob_or_path) {
            // "glob/pattern/*"
            self.copy_glob(&path, &self.app_root, dest_root, exclude_files)
                .await?;
        } else {
            bail!("{glob_or_path:?} does not exist and doesn't appear to be a glob pattern");
        }
        Ok(())
    }

    // Copy a single file or entire directory from `src` to `dest`
    async fn copy_file_or_directory(
        &self,
        src: &Path,
        dest: &Path,
        guest_dest: &str,
        exclude_files: &[String],
    ) -> Result<()> {
        let src_path = self.app_root.join(src);
        let meta = crate::fs::metadata(&src_path)
            .await
            .map_err(|e| explain_file_mount_source_error(e, src))?;
        if meta.is_dir() {
            // { source = "host/dir", destination = "guest/dir" }
            let pattern = src_path.join("**/*");
            self.copy_glob(&pattern, &src_path, dest, exclude_files)
                .await?;
        } else {
            // { source = "host/file.txt", destination = "guest/file.txt" }
            self.copy_single_file(&src_path, dest, guest_dest).await?;
        }
        Ok(())
    }

    // Copy files matching glob `pattern` into `dest_root`.
    async fn copy_glob(
        &self,
        pattern: &Path,
        src_prefix: &Path,
        dest_root: &Path,
        exclude_files: &[String],
    ) -> Result<()> {
        let pattern = pattern
            .to_str()
            .with_context(|| format!("invalid (non-utf8) file pattern {pattern:?}"))?;

        let paths = glob::glob(pattern)
            .with_context(|| format!("Failed to resolve glob pattern {pattern:?}"))?;

        let exclude_patterns = exclude_files
            .iter()
            .map(|pattern| {
                glob::Pattern::new(pattern)
                    .with_context(|| format!("Invalid exclude_files glob pattern {pattern:?}"))
            })
            .collect::<Result<Vec<_>>>()?;

        crate::fs::create_dir_all(dest_root)
            .await
            .with_context(|| {
                format!(
                    "Failed to create parent directory {}",
                    quoted_path(&dest_root)
                )
            })?;

        for path_res in paths {
            let src = path_res?;
            if !src.is_file() {
                continue;
            }

            let Ok(app_root_path) = src.strip_prefix(&self.app_root) else {
                bail!(
                    "{pattern} cannot be mapped because it is outside the application directory. Files must be within the application directory."
                );
            };

            if exclude_patterns
                .iter()
                .any(|pattern| pattern.matches_path(app_root_path))
            {
                tracing::debug!(
                    "File {app_root_path:?} excluded by exclude_files {exclude_files:?}"
                );
                continue;
            }

            let relative_path = src.strip_prefix(src_prefix)?;
            let dest = dest_root.join(relative_path);
            self.copy_single_file(&src, &dest, &relative_path.to_string_lossy())
                .await?;
        }
        Ok(())
    }

    // Copy a single file from `src` to `dest`, creating parent directories.
    async fn copy_single_file(&self, src: &Path, dest: &Path, guest_dest: &str) -> Result<()> {
        // Sanity checks: src is in app_root...
        src.strip_prefix(&self.app_root)?;
        // ...and dest is in the Copy root.
        if let FilesMountStrategy::Copy(files_mount_root) = &self.files_mount_strategy {
            dest.strip_prefix(files_mount_root)?;
        } else {
            unreachable!();
        }

        let _loading_permit = self.file_loading_permits.acquire().await?;
        let dest_parent = parent_dir(dest)?;
        crate::fs::create_dir_all(&dest_parent)
            .await
            .with_context(|| {
                format!(
                    "Failed to create parent directory {}",
                    quoted_path(&dest_parent)
                )
            })?;
        crate::fs::copy(src, dest)
            .await
            .or_else(|e| Self::failed_to_copy_single_file_error(src, dest, guest_dest, e))?;
        tracing::debug!("Copied {src:?} to {dest:?}");
        Ok(())
    }

    fn failed_to_copy_single_file_error<T>(
        src: &Path,
        dest: &Path,
        guest_dest: &str,
        e: anyhow::Error,
    ) -> anyhow::Result<T> {
        let src_text = quoted_path(src);
        let dest_text = quoted_path(dest);
        let base_msg = format!("Failed to copy {src_text} to working path {dest_text}");

        if let Some(io_error) = e.downcast_ref::<std::io::Error>()
            && (Self::is_directory_like(guest_dest)
                || io_error.kind() == std::io::ErrorKind::NotFound)
        {
            return Err(anyhow::anyhow!(
                r#""{guest_dest}" is not a valid destination file name"#
            ))
            .context(base_msg);
        }

        Err(e).with_context(|| format!("{base_msg} (for destination path \"{guest_dest}\")"))
    }

    /// Does a guest path appear to be a directory name, e.g. "/" or ".."? This is for guest
    /// paths *only* and does not consider Windows separators.
    fn is_directory_like(guest_path: &str) -> bool {
        guest_path.ends_with('/') || guest_path.ends_with('.') || guest_path.ends_with("..")
    }

    // Resolve the given direct mount directory, checking that it is valid for
    // direct mounting and returning its canonicalized source path.
    async fn resolve_direct_mount(&self, mount: &WasiFilesMount) -> Result<ContentPath> {
        let (src, dest) = match mount {
            WasiFilesMount::Pattern(pattern) => (pattern, pattern),
            WasiFilesMount::Placement {
                source,
                destination,
            } => (source, destination),
        };
        let path = self.app_root.join(src);
        if !path.is_dir() {
            bail!(
                "Only directory mounts are supported with `--direct-mounts`; {src:?} is not a directory."
            );
        }
        Ok(ContentPath {
            content: file_content_ref(src)?,
            path: dest.into(),
        })
    }

    fn profile(&self) -> Option<&str> {
        self.profile.as_deref()
    }
}

fn explain_file_mount_source_error(e: anyhow::Error, src: &Path) -> anyhow::Error {
    if let Some(io_error) = e.downcast_ref::<std::io::Error>()
        && io_error.kind() == std::io::ErrorKind::NotFound
    {
        return anyhow::anyhow!("File or directory {} does not exist", quoted_path(src));
    }
    e.context(format!("invalid file mount source {}", quoted_path(src)))
}

#[cfg(feature = "async-io")]
async fn verified_download(
    url: &str,
    digest: &str,
    dest: &Path,
    convention: crate::http::DestinationConvention,
) -> Result<()> {
    crate::http::verified_download(url, digest, dest, convention)
        .await
        .with_context(|| format!("Error fetching source URL {url:?}"))
}

#[cfg(not(feature = "async-io"))]
async fn verified_download(
    _url: &str,
    _digest: &str,
    _dest: &Path,
    _convention: crate::http::DestinationConvention,
) -> Result<()> {
    panic!("async-io feature is required for downloading Wasm sources")
}

fn safe_canonicalize(path: &Path) -> std::io::Result<PathBuf> {
    use path_absolutize::Absolutize;
    Ok(path.absolutize()?.into_owned())
}

fn locked_metadata(
    details: v2::AppDetails,
    trigger_types: impl Iterator<Item = String>,
) -> Result<ValuesMap> {
    let mut builder = ValuesMapBuilder::new();
    builder
        .string("name", details.name)
        .string("version", details.version)
        .string("description", details.description)
        .string_array("authors", details.authors)
        .serializable("triggers", &details.trigger_global_configs)?;

    // Duplicate single-trigger global options into "trigger" with "type"
    // key to maintain backward compatibility for a while.
    let types = trigger_types.collect::<Vec<_>>();
    if types.len() == 1 {
        let trigger_type = types.into_iter().next().unwrap();
        let mut single_trigger = details
            .trigger_global_configs
            .get(&trigger_type)
            .cloned()
            .unwrap_or_default();
        single_trigger.insert("type".into(), trigger_type.into());
        builder.serializable("trigger", single_trigger).unwrap();
    }

    Ok(builder.build())
}

fn locked_variable(variable: v2::Variable) -> Result<locked::Variable> {
    ensure!(
        variable.required ^ variable.default.is_some(),
        "must be `required` OR have a `default`"
    );
    Ok(locked::Variable {
        description: variable.description,
        default: variable.default.clone(),
        secret: variable.secret,
    })
}

#[derive(Debug)]
/// Handles loading of component Wasm from different sources.
pub struct WasmLoader {
    app_root: PathBuf,
    cache: Cache,
    file_loading_permits: std::sync::Arc<Semaphore>,
}

impl WasmLoader {
    /// Create a new instance of WasmLoader.
    pub async fn new(
        app_root: PathBuf,
        cache_root: Option<PathBuf>,
        file_loading_permits: Option<std::sync::Arc<Semaphore>>,
    ) -> Result<Self> {
        let file_loading_permits = file_loading_permits.unwrap_or_else(|| {
            std::sync::Arc::new(Semaphore::new(crate::MAX_FILE_LOADING_CONCURRENCY))
        });
        Ok(Self {
            app_root,
            cache: Cache::new(cache_root).await?,
            file_loading_permits,
        })
    }

    /// Load a Wasm source from the given ComponentSource and return a path
    /// to a file location from where it can be read.
    pub async fn load_component_source(
        &self,
        component_id: &str,
        source: &v2::ComponentSource,
    ) -> Result<PathBuf> {
        let content = match source {
            v2::ComponentSource::Local(path) => self.app_root.join(path),
            v2::ComponentSource::Remote { url, digest } => {
                self.load_http_source(url, digest).await?
            }
            v2::ComponentSource::Registry {
                registry,
                package,
                version,
            } => {
                let version = semver::Version::parse(version).with_context(|| format!("Component {component_id} specifies an invalid semantic version ({version:?}) for its package version"))?;
                let version_req = format!("={version}").parse().expect("version");

                self.load_registry_source(registry.as_ref(), package, &version_req)
                    .await?
            }
        };
        Ok(content)
    }

    // Load a Wasm source from the given HTTP ContentRef source URL and
    // return a ContentRef an absolute path to the local copy.
    async fn load_http_source(&self, url: &str, digest: &str) -> Result<PathBuf> {
        ensure!(
            digest.starts_with("sha256:"),
            "invalid `digest` {digest:?}; must start with 'sha256:'"
        );
        let path = if let Ok(cached_path) = self.cache.wasm_file(digest) {
            cached_path
        } else {
            let _loading_permit = self.file_loading_permits.acquire().await?;

            self.cache.ensure_dirs().await?;
            let dest = self.cache.wasm_path(digest);
            verified_download(
                url,
                digest,
                &dest,
                crate::http::DestinationConvention::ContentIndexed,
            )
            .await
            .with_context(|| format!("Error fetching source URL {url:?}"))?;
            dest
        };
        Ok(path)
    }

    async fn load_registry_source(
        &self,
        registry: Option<&wasm_pkg_client::Registry>,
        package: &wasm_pkg_client::PackageRef,
        version: &semver::VersionReq,
    ) -> Result<PathBuf> {
        let mut client_config = wasm_pkg_client::Config::global_defaults().await?;

        if let Some(registry) = registry.cloned() {
            let mapping = wasm_pkg_client::RegistryMapping::Registry(registry);
            client_config.set_package_registry_override(package.clone(), mapping);
        }
        let pkg_loader = wasm_pkg_client::Client::new(client_config);

        let mut releases = pkg_loader.list_all_versions(package).await.map_err(|e| {
            if matches!(e, wasm_pkg_client::Error::NoRegistryForNamespace(_)) && registry.is_none() {
                anyhow!("No default registry specified for wasm-pkg-loader. Create a default config, or set `registry` for package {package:?}")
            } else {
                e.into()
            }
        })?;

        releases.sort();

        let release_version = releases
            .iter()
            .rev()
            .find(|release| version.matches(&release.version) && !release.yanked)
            .with_context(|| format!("No matching version found for {package} {version}",))?;

        let release = pkg_loader
            .get_release(package, &release_version.version)
            .await?;

        let digest = match &release.content_digest {
            wasm_pkg_client::ContentDigest::Sha256 { hex } => format!("sha256:{hex}"),
        };

        let path = if let Ok(cached_path) = self.cache.wasm_file(&digest) {
            cached_path
        } else {
            let mut stm = pkg_loader.stream_content(package, &release).await?;

            self.cache.ensure_dirs().await?;
            let dest = self.cache.wasm_path(&digest);

            let mut file = tokio::fs::File::create(&dest).await?;
            while let Some(block) = stm.next().await {
                let bytes = block.context("Failed to get content from registry")?;
                file.write_all(&bytes)
                    .await
                    .context("Failed to save registry content to cache")?;
            }

            dest
        };

        Ok(path)
    }

    /// Loads a dependency and returns a fully resolved locked component dependency.
    pub async fn load_component_dependency(
        &self,
        dependency_name: &DependencyName,
        dependency: &v2::ComponentDependency,
    ) -> Result<locked::LockedComponentDependency> {
        let inherit = locked_inherit(dependency);

        let (content, export) = self
            .load_dependency_content(dependency_name, dependency)
            .await?;

        Ok(locked::LockedComponentDependency {
            source: locked::LockedComponentSource {
                content_type: "application/wasm".into(),
                content: file_content_ref(content)?,
            },
            export,
            inherit,
        })
    }

    /// Loads a dependency and returns a fully resolved locked component dependency.
    pub async fn load_trigger_dependency(
        &self,
        dependency: &v2::TriggerDependency,
    ) -> Result<locked::LockedComponentDependency> {
        let as_component_dep = match dependency.clone() {
            v2::TriggerDependency::Package {
                version,
                registry,
                package,
                inherit_configuration,
            } => v2::ComponentDependency::Package {
                version,
                registry,
                package: Some(package),
                export: None,
                inherit_configuration,
            },
            v2::TriggerDependency::Local {
                path,
                inherit_configuration,
            } => v2::ComponentDependency::Local {
                path,
                export: None,
                inherit_configuration,
            },
            v2::TriggerDependency::HTTP {
                url,
                digest,
                inherit_configuration,
            } => v2::ComponentDependency::HTTP {
                url,
                digest,
                export: None,
                inherit_configuration,
            },
            v2::TriggerDependency::AppComponent {
                component,
                inherit_configuration,
            } => v2::ComponentDependency::AppComponent {
                component,
                export: None,
                inherit_configuration,
            },
        };

        let inherit = locked_inherit(&as_component_dep);

        let fake_dep_name =
            DependencyName::Plain(KebabId::try_from("to-do-fix-fix-fix".to_string()).unwrap());

        let (content, export) = self
            .load_dependency_content(&fake_dep_name, &as_component_dep)
            .await?;

        Ok(locked::LockedComponentDependency {
            source: locked::LockedComponentSource {
                content_type: "application/wasm".into(),
                content: file_content_ref(content)?,
            },
            export,
            inherit,
        })
    }

    /// Loads the content path and export for a dependency.
    pub async fn load_dependency_content(
        &self,
        dependency_name: &DependencyName,
        dependency: &v2::ComponentDependency,
    ) -> Result<(PathBuf, Option<String>)> {
        match dependency.clone() {
            v2::ComponentDependency::Version(version) => {
                let version = semver::VersionReq::parse(&version).with_context(|| format!("Component dependency {dependency_name:?} specifies an invalid semantic version requirement ({version:?}) for its package version"))?;

                // This `unwrap()` should be OK because we've already validated
                // this form of dependency requires a package name, i.e. the
                // dependency name is not a kebab id.
                let package = dependency_name.package().unwrap();

                let content = self.load_registry_source(None, package, &version).await?;
                Ok((content, None))
            }
            v2::ComponentDependency::Package {
                version,
                registry,
                package,
                export,
                ..
            } => {
                let version = semver::VersionReq::parse(&version).with_context(|| format!("Component dependency {dependency_name:?} specifies an invalid semantic version requirement ({version:?}) for its package version"))?;

                let package = match package {
                    Some(package) => {
                        package.parse().with_context(|| format!("Component dependency {dependency_name:?} specifies an invalid package name ({package:?})"))?
                    }
                    None => {
                        // This `unwrap()` should be OK because we've already validated
                        // this form of dependency requires a package name, i.e. the
                        // dependency name is not a kebab id.
                        dependency_name
                            .package()
                            .cloned()
                            .unwrap()
                    }
                };

                let registry = match registry {
                    Some(registry) => {
                        registry
                            .parse()
                            .map(Some)
                            .with_context(|| format!("Component dependency {dependency_name:?} specifies an invalid registry name ({registry:?})"))?
                    }
                    None => None,
                };

                let content = self
                    .load_registry_source(registry.as_ref(), &package, &version)
                    .await?;
                Ok((content, export))
            }
            v2::ComponentDependency::Local { path, export, .. } => {
                let content = self.app_root.join(path);
                Ok((content, export))
            }
            v2::ComponentDependency::HTTP {
                url,
                digest,
                export,
                ..
            } => {
                let content = self.load_http_source(&url, &digest).await?;
                Ok((content, export))
            }
            v2::ComponentDependency::AppComponent { .. } => {
                panic!("Internal error: component ID dependency was not resolved to a source");
            }
        }
    }
}

fn locked_inherit(dependency: &v2::ComponentDependency) -> locked::InheritConfiguration {
    match dependency.inherit_configuration() {
        Some(v2::InheritConfiguration::All(true)) => locked::InheritConfiguration::All,
        Some(v2::InheritConfiguration::Some(keys)) => {
            locked::InheritConfiguration::Some(keys.clone())
        }
        Some(v2::InheritConfiguration::All(false)) | None => {
            locked::InheritConfiguration::Some(vec![])
        }
    }
}

fn looks_like_glob_pattern(s: impl AsRef<str>) -> bool {
    let s = s.as_ref();
    glob::Pattern::escape(s) != s
}

fn file_content_ref(path: impl AsRef<Path>) -> Result<ContentRef> {
    Ok(ContentRef {
        source: Some(file_url(path)?),
        ..Default::default()
    })
}

fn file_url(path: impl AsRef<Path>) -> Result<String> {
    let path = path.as_ref();
    let abs_path = safe_canonicalize(path)
        .with_context(|| format!("Couldn't resolve `{}`", path.display()))?;
    Ok(Url::from_file_path(abs_path).unwrap().to_string())
}

/// Determines if a component requires the host to support local
/// service chaining.
pub fn requires_service_chaining(component: &spin_manifest::schema::v2::Component) -> bool {
    component
        .normalized_allowed_outbound_hosts()
        .unwrap_or_default()
        .iter()
        .any(|h| is_chaining_host(h))
}

fn is_chaining_host(pattern: &str) -> bool {
    AllowedHostConfig::parse(pattern).is_ok_and(|config| config.is_for_service_chaining())
}

const SLOTH_WARNING_DELAY_MILLIS: u64 = 1250;

fn warn_if_component_load_slothful() -> sloth::SlothGuard {
    let message = "Loading Wasm components is taking a few seconds...";
    sloth::warn_if_slothful(SLOTH_WARNING_DELAY_MILLIS, format!("{message}\n"))
}

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

    async fn load_test_case(
        testcase_dir: &str,
        manifest_file: &str,
    ) -> anyhow::Result<(tempfile::TempDir, LockedApp)> {
        let app_root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
            .join("tests")
            .join(testcase_dir);
        let wd = tempfile::tempdir()?;
        let loader = LocalLoader::new(
            &app_root,
            FilesMountStrategy::Copy(wd.path().to_owned()),
            None,
            None,
        )
        .await?;
        let locked_app = loader.load_file(app_root.join(manifest_file)).await;
        locked_app.map(|locked| (wd, locked))
    }

    #[tokio::test]
    async fn bad_destination_filename_is_explained() -> anyhow::Result<()> {
        let err = load_test_case("file-errors", "bad.toml")
            .await
            .expect_err("loader should not have succeeded");
        let err_ctx = format!("{err:#}");
        assert!(
            err_ctx.contains(r#""/" is not a valid destination file name"#),
            "expected error to show destination file name but got {err_ctx}",
        );
        Ok(())
    }

    fn trigger_by_route<'a>(locked: &'a LockedApp, route: &str) -> &'a LockedTrigger {
        fn route_of(trigger: &LockedTrigger) -> &str {
            trigger
                .trigger_config
                .get("route")
                .and_then(|v| v.as_str())
                .unwrap()
        }
        locked
            .triggers
            .iter()
            .find(|t| route_of(t) == route)
            .unwrap()
    }

    fn component_for_route<'a>(locked: &'a LockedApp, route: &str) -> &'a LockedComponent {
        let component_id = component_id(trigger_by_route(locked, route));
        locked
            .components
            .iter()
            .find(|c| c.id == component_id)
            .unwrap()
    }

    fn component_id(trigger: &LockedTrigger) -> &str {
        trigger
            .trigger_config
            .get("component")
            .and_then(|v| v.as_str())
            .unwrap()
    }

    fn component_trigger_deps_for_route<'a>(
        locked: &'a LockedApp,
        route: &str,
        key: &str,
    ) -> &'a Vec<LockedComponentDependency> {
        let component = component_for_route(locked, route);
        component
            .trigger_dependencies
            .get(key)
            .expect("should have had trigger deps for key")
    }

    #[tokio::test]
    async fn unenriched_lockfile_is_unchanged() {
        let (_wd, locked_app) = load_test_case("extra-components", "vanilla.toml")
            .await
            .unwrap();
        assert_eq!(3, locked_app.triggers.len());
        assert_eq!(2, locked_app.components.len());
    }

    #[tokio::test]
    async fn enriched_lockfile_only_one_trigger_per_component_no_changes() {
        let (_wd, locked_app) = load_test_case("extra-components", "inoffensive.toml")
            .await
            .unwrap();
        assert_eq!(2, locked_app.triggers.len());
        assert_eq!("a", component_id(trigger_by_route(&locked_app, "/a")));
        assert_eq!("b", component_id(trigger_by_route(&locked_app, "/b")));
        assert_eq!(5, locked_app.components.len());
    }

    #[tokio::test]
    async fn enriched_lockfile_multiple_enriched_triggers_per_component_get_split() {
        let (_wd, locked_app) = load_test_case("extra-components", "three-to-one.toml")
            .await
            .unwrap();
        assert_eq!(4, locked_app.triggers.len());
        // Splitting should result in triggers pointing to different IDs, but the same primary source
        assert_ne!("a", component_id(trigger_by_route(&locked_app, "/a1")));
        assert!(
            component_for_route(&locked_app, "/a1")
                .source
                .content
                .source
                .as_ref()
                .unwrap()
                .ends_with("/a.dummy.wasm.txt")
        );
        assert_ne!("a", component_id(trigger_by_route(&locked_app, "/a2")));
        assert!(
            component_for_route(&locked_app, "/a3")
                .source
                .content
                .source
                .as_ref()
                .unwrap()
                .ends_with("/a.dummy.wasm.txt")
        );
        assert_ne!("a", component_id(trigger_by_route(&locked_app, "/a3")));
        assert!(
            component_for_route(&locked_app, "/a3")
                .source
                .content
                .source
                .as_ref()
                .unwrap()
                .ends_with("/a.dummy.wasm.txt")
        );
        // Triggers that don't need splitting should be unaffected
        assert_eq!("b", component_id(trigger_by_route(&locked_app, "/b")));
        // There should be new components inserted for the split
        assert_eq!(8, locked_app.components.len());
    }

    // Content refs have `file:` URIs with absolute paths. This is great for
    // definiteness but not so convenient for testing!
    fn content_ref_to_file_name(content_ref: &ContentRef) -> String {
        let url = content_ref.source.as_ref().unwrap();
        let url = Url::parse(url).unwrap();
        let file_path = PathBuf::from(url.path());
        file_path.file_name().unwrap().to_string_lossy().to_string()
    }

    #[tokio::test]
    async fn enriched_lockfile_captures_composition_graph_in_split_component() {
        let (_wd, locked_app) = load_test_case("extra-components", "three-to-one.toml")
            .await
            .unwrap();
        assert_eq!(4, locked_app.triggers.len());

        let a1_mw = component_trigger_deps_for_route(&locked_app, "/a1", "middleware");
        assert_eq!(2, a1_mw.len());
        assert_eq!(
            "a.dummy.wasm.txt",
            content_ref_to_file_name(&a1_mw[0].source.content)
        );
        assert_eq!(
            "b.dummy.wasm.txt",
            content_ref_to_file_name(&a1_mw[1].source.content)
        );

        let a2_mw = component_trigger_deps_for_route(&locked_app, "/a2", "middleware");
        assert_eq!(2, a2_mw.len());
        assert_eq!(
            "b.dummy.wasm.txt",
            content_ref_to_file_name(&a2_mw[0].source.content)
        );
        assert_eq!(
            "c.dummy.wasm.txt",
            content_ref_to_file_name(&a2_mw[1].source.content)
        );

        let a3_mw = component_trigger_deps_for_route(&locked_app, "/a3", "middleware");
        assert_eq!(3, a3_mw.len());
        assert_eq!(
            "c.dummy.wasm.txt",
            content_ref_to_file_name(&a3_mw[0].source.content)
        );
        assert_eq!(
            "b.dummy.wasm.txt",
            content_ref_to_file_name(&a3_mw[1].source.content)
        );
        assert_eq!(
            "a.dummy.wasm.txt",
            content_ref_to_file_name(&a3_mw[2].source.content)
        );

        // Unsplit things should still get the shunt
        let b_mw = component_trigger_deps_for_route(&locked_app, "/b", "middleware");
        assert_eq!(2, b_mw.len());
        assert_eq!(
            "a.dummy.wasm.txt",
            content_ref_to_file_name(&b_mw[0].source.content)
        );
        assert_eq!(
            "c.dummy.wasm.txt",
            content_ref_to_file_name(&b_mw[1].source.content)
        );
    }

    #[tokio::test]
    async fn deps_moved_off_trigger() {
        let (_wd, locked_app) = load_test_case("extra-components", "three-to-one.toml")
            .await
            .unwrap();
        assert_eq!(4, locked_app.triggers.len());

        for t in &locked_app.triggers {
            assert!(t.trigger_dependencies.is_empty());
        }
    }
}