obelisk 0.36.1

use super::{config_holder::PathPrefixes, env_var::EnvVarConfig};
use crate::command::server::FrameFilesToSourceContent;
use crate::config::env_var::{
    EnvVarMissing, EnvVarsMissing, interpolate_env_vars_plaintext, interpolate_env_vars_secret,
};
use crate::github::{self, GH_SCHEMA_PREFIX, GitHubReleaseReference};
use crate::oci;
use crate::{
    config::config_holder::{CACHE_DIR_PREFIX, DATA_DIR_PREFIX},
    github::content_digest_to_wasm_file,
};
use anyhow::{Context, ensure};
use anyhow::{anyhow, bail};
use concepts::ContentDigest;
use concepts::ReturnType;
use concepts::component_id::Digest;
use concepts::{
    ComponentId, ComponentRetryConfig, ComponentType, FunctionFqn, InvalidNameError, StrVariant,
    check_name, component_id::ComponentDigest, prefixed_ulid::ExecutorId, storage::LogLevel,
};
use db_postgres::postgres_dao::{self, PostgresConfig};
use db_sqlite::sqlite_dao::SqliteConfig;
use hashbrown::HashMap;
use log::{LoggingConfig, LoggingStyle};
use schemars::JsonSchema;
use secrecy::SecretString;
use serde::{Deserialize, Deserializer, Serialize};
use serde_with::{DeserializeFromStr, SerializeDisplay};
use sha2::{Digest as _, Sha256};
use std::fmt::Display;
use std::str::FromStr;
use std::{
    net::SocketAddr,
    path::{Path, PathBuf},
    sync::Arc,
    time::Duration,
};
use tracing::{debug, info, instrument, trace, warn};
use utils::wasm_tools::WasmComponent;
use wasm_workers::http_request_policy::HostPatternError;
use wasm_workers::{
    activity::activity_worker::{ActivityConfig, ActivityDirectoriesConfig, ProcessProvider},
    envvar::EnvVar,
    http_request_policy::{AllowedHostConfig, HostPattern, MethodsPattern, ReplacementLocation},
    std_output_stream::StdOutputConfig,
    workflow::workflow_worker::{
        DEFAULT_NON_BLOCKING_EVENT_BATCHING, JoinNextBlockingStrategy, WorkflowConfig,
    },
};
use webhook::{HttpServer, WebhookJsComponentConfigToml, WebhookWasmComponentConfigToml};

const DEFAULT_SQLITE_DIR_IF_PROJECT_DIRS: &str =
    const_format::formatcp!("{}obelisk-sqlite", DATA_DIR_PREFIX);
const DEFAULT_SQLITE_DIR: &str = "obelisk-sqlite";
pub(crate) const SQLITE_FILE_NAME: &str = "obelisk.sqlite";
const DEFAULT_WASM_DIRECTORY_IF_PROJECT_DIRS: &str =
    const_format::formatcp!("{}wasm", CACHE_DIR_PREFIX);
const DEFAULT_WASM_DIRECTORY: &str = "cache/wasm";
const DEFAULT_CODEGEN_CACHE_DIRECTORY_IF_PROJECT_DIRS: &str =
    const_format::formatcp!("{}codegen", CACHE_DIR_PREFIX);
const DEFAULT_CODEGEN_CACHE_DIRECTORY: &str = "cache/codegen";

#[derive(Debug, Deserialize, Serialize, JsonSchema, Default, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct DeploymentToml {
    #[serde(default, rename = "activity_wasm")]
    pub(crate) activities_wasm: Vec<ActivityWasmComponentConfigToml>,
    #[serde(default, rename = "activity_stub")]
    pub(crate) activities_stub: Vec<ActivityStubComponentConfigToml>,
    #[serde(default, rename = "activity_external")]
    pub(crate) activities_external: Vec<ActivityExternalComponentConfigToml>,
    #[serde(default, rename = "activity_js")]
    pub(crate) activities_js: Vec<ActivityJsComponentConfigToml>,
    #[serde(default, rename = "workflow_wasm")]
    pub(crate) workflows: Vec<WorkflowWasmComponentConfigToml>,
    #[serde(default, rename = "workflow_js")]
    pub(crate) workflows_js: Vec<WorkflowJsComponentConfigToml>,
    #[serde(default, rename = "webhook_endpoint_wasm")]
    pub(crate) webhooks: Vec<WebhookWasmComponentConfigToml>,
    #[serde(default, rename = "webhook_endpoint_js")]
    pub(crate) webhooks_js: Vec<WebhookJsComponentConfigToml>,
}

#[derive(Debug, Default, Deserialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct ServerConfigToml {
    #[serde(default, rename = "obelisk-version")]
    pub(crate) obelisk_version: Option<String>,
    #[serde(default)]
    pub(crate) api: ApiConfig,
    #[serde(default)]
    pub(crate) database: DatabaseConfigToml,
    #[serde(default)]
    pub(crate) webui: WebUIConfig,
    #[serde(default)]
    pub(crate) external: ExternalServerConfig,
    #[serde(default, rename = "wasm")]
    pub(crate) wasm_global_config: WasmGlobalConfigToml,
    #[serde(default, rename = "activities")]
    pub(crate) activities_global_config: ActivitiesGlobalConfigToml,
    #[serde(default, rename = "workflows")]
    pub(crate) workflows_global_config: WorkflowsGlobalConfigToml,
    #[serde(default)]
    pub(crate) timers_watcher: TimersWatcherTomlConfig,
    #[serde(default)]
    pub(crate) cancel_watcher: CancelWatcherTomlConfig,
    #[cfg(feature = "otlp")]
    #[serde(default)]
    pub(crate) otlp: Option<otlp::OtlpConfig>,
    #[serde(default)]
    pub(crate) log: LoggingConfig,
    #[serde(default, rename = "http_server")]
    pub(crate) http_servers: Vec<HttpServer>,
}

/// Return a canonical JSON string of the deployment config for storage.
pub(crate) fn compute_config_json(deployment: &DeploymentCanonical) -> String {
    serde_json::to_string(deployment).expect("DeploymentCanonical is serializable")
}

#[derive(Debug, Deserialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct ApiConfig {
    #[serde(default = "default_true")]
    pub(crate) enabled: bool,
    #[serde(default = "default_api_listening_addr")]
    pub(crate) listening_addr: SocketAddr,
}
impl Default for ApiConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            listening_addr: default_api_listening_addr(),
        }
    }
}
fn default_api_listening_addr() -> SocketAddr {
    "127.0.0.1:5005".parse().expect("valid default address")
}

#[derive(Debug, Deserialize, JsonSchema, Clone)]
#[serde(rename_all = "snake_case")]
pub(crate) enum DatabaseConfigToml {
    Sqlite(SqliteConfigToml),
    Postgres(PostgresConfigToml),
}
impl DatabaseConfigToml {
    pub fn get_subscription_interruption(&self) -> Option<Duration> {
        match self {
            DatabaseConfigToml::Sqlite(_) => None,
            DatabaseConfigToml::Postgres(postgres_config_toml) => {
                postgres_config_toml.subscription_interruption.into()
            }
        }
    }
}
impl Default for DatabaseConfigToml {
    fn default() -> DatabaseConfigToml {
        DatabaseConfigToml::Sqlite(SqliteConfigToml::default())
    }
}

#[derive(Debug, Deserialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct PostgresConfigToml {
    host: String,
    user: String,
    password: String,
    db_name: String,
    /// Interrupts listening for notifications periodically, needed for Postgres with a local-only subscription mechanism.
    #[serde(default = "default_subscription_interruption")]
    pub subscription_interruption: DurationConfigOptional,
    #[serde(default)]
    provision_policy: PostgresProvisionPolicy,
}

impl PostgresConfigToml {
    pub fn as_config(&self) -> Result<PostgresConfig, anyhow::Error> {
        Ok(PostgresConfig {
            host: interpolate_env_vars_plaintext(&self.host)?,
            user: interpolate_env_vars_plaintext(&self.user)?,
            password: interpolate_env_vars_secret(&self.password)?,
            db_name: interpolate_env_vars_plaintext(&self.db_name)?,
        })
    }
    pub fn as_provision_policy(&self) -> postgres_dao::ProvisionPolicy {
        match self.provision_policy {
            PostgresProvisionPolicy::Never => postgres_dao::ProvisionPolicy::NeverCreate,
            PostgresProvisionPolicy::Auto => postgres_dao::ProvisionPolicy::Auto,
        }
    }
}

#[derive(Debug, Deserialize, JsonSchema, Clone, Default)]
#[serde(rename_all = "snake_case")]
pub enum PostgresProvisionPolicy {
    #[default]
    Never,
    /// Create database if it does not exist.
    Auto,
}

#[derive(Debug, Deserialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct SqliteConfigToml {
    #[serde(default)]
    directory: Option<String>,
    #[serde(default = "default_sqlite_queue_capacity")]
    queue_capacity: usize,
    #[serde(default)]
    pragma: std::collections::HashMap<String, String>, // hashbrown is not supported by schemars
    #[serde(default)]
    metrics_threshold: Option<DurationConfig>,
}
impl Default for SqliteConfigToml {
    fn default() -> Self {
        Self {
            directory: None,
            queue_capacity: default_sqlite_queue_capacity(),
            pragma: std::collections::HashMap::default(),
            metrics_threshold: Option::default(),
        }
    }
}
impl SqliteConfigToml {
    pub(crate) async fn get_sqlite_dir(
        &self,
        path_prefixes: &PathPrefixes,
    ) -> Result<PathBuf, anyhow::Error> {
        let sqlite_file = self.directory.as_deref().unwrap_or_else(|| {
            if path_prefixes.project_dirs.is_some() {
                DEFAULT_SQLITE_DIR_IF_PROJECT_DIRS
            } else {
                DEFAULT_SQLITE_DIR
            }
        });
        path_prefixes
            .server_config_replace_path_prefix_mkdir(sqlite_file)
            .await
    }

    pub(crate) fn as_sqlite_config(&self) -> SqliteConfig {
        SqliteConfig {
            queue_capacity: self.queue_capacity,
            pragma_override: Some(self.pragma.clone().into_iter().collect()),
            metrics_threshold: self.metrics_threshold.map(Duration::from),
        }
    }
}

#[derive(Debug, Deserialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct WebUIConfig {
    #[serde(default = "default_true")]
    pub(crate) enabled: bool,
    #[serde(default = "default_webui_listening_addr")]
    pub(crate) listening_addr: String,
}
impl Default for WebUIConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            listening_addr: default_webui_listening_addr(),
        }
    }
}
fn default_webui_listening_addr() -> String {
    "127.0.0.1:8080".to_string()
}

#[derive(Debug, Deserialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct ExternalServerConfig {
    #[serde(default = "default_true")]
    pub(crate) enabled: bool,
    #[serde(default = "default_external_listening_addr")]
    pub(crate) listening_addr: SocketAddr,
}
impl Default for ExternalServerConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            listening_addr: default_external_listening_addr(),
        }
    }
}
fn default_external_listening_addr() -> SocketAddr {
    "127.0.0.1:9090".parse().expect("valid default address")
}

#[derive(Debug, Deserialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct WasmGlobalConfigToml {
    #[serde(default)]
    pub(crate) codegen_cache: CodegenCache,
    #[serde(default)]
    pub(crate) backtrace: WasmGlobalBacktrace,
    #[serde(default)]
    cache_directory: Option<String>,
    #[serde(default)]
    pub(crate) allocator_config: WasmtimeAllocatorConfig,
    #[serde(default)]
    pub(crate) global_executor_instance_limiter: InflightSemaphore,
    #[serde(default)]
    pub(crate) global_webhook_instance_limiter: InflightSemaphore,
    #[serde(default)]
    pub(crate) fuel: ValueOrUnlimited<u64>,
    #[serde(default)]
    pub(crate) build_semaphore: ValueOrUnlimited<u64>,
    #[serde(default = "default_parallel_compilation")]
    pub(crate) parallel_compilation: bool,
    #[serde(default)]
    pub(crate) wasmtime_pooling_config: WasmtimePoolingAllocatorConfig,
    #[serde(default = "default_debug")]
    pub(crate) debug: bool,
}
impl Default for WasmGlobalConfigToml {
    fn default() -> Self {
        WasmGlobalConfigToml {
            codegen_cache: CodegenCache::default(),
            backtrace: WasmGlobalBacktrace::default(),
            cache_directory: Option::default(),
            allocator_config: WasmtimeAllocatorConfig::default(),
            global_executor_instance_limiter: InflightSemaphore::default(),
            global_webhook_instance_limiter: InflightSemaphore::default(),
            fuel: ValueOrUnlimited::default(),
            build_semaphore: ValueOrUnlimited::default(),
            parallel_compilation: default_parallel_compilation(),
            wasmtime_pooling_config: WasmtimePoolingAllocatorConfig::default(),
            debug: default_debug(),
        }
    }
}

impl WasmGlobalConfigToml {
    pub(crate) async fn get_wasm_cache_directory(
        &self,
        path_prefixes: &PathPrefixes,
    ) -> Result<PathBuf, anyhow::Error> {
        let wasm_directory = self.cache_directory.as_deref().unwrap_or_else(|| {
            if path_prefixes.project_dirs.is_some() {
                DEFAULT_WASM_DIRECTORY_IF_PROJECT_DIRS
            } else {
                DEFAULT_WASM_DIRECTORY
            }
        });
        path_prefixes
            .server_config_replace_path_prefix_mkdir(wasm_directory)
            .await
    }
}

#[derive(Debug, Deserialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct WasmGlobalBacktrace {
    #[serde(default = "default_global_backtrace_persist")]
    pub(crate) persist: bool,
}

impl Default for WasmGlobalBacktrace {
    fn default() -> Self {
        Self {
            persist: default_global_backtrace_persist(),
        }
    }
}

#[derive(Debug, Default, Deserialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct ActivitiesGlobalConfigToml {
    directories: ActivitiesDirectoriesGlobalConfigToml,
}
impl ActivitiesGlobalConfigToml {
    pub(crate) fn get_directories(&self) -> Option<&ActivitiesDirectoriesGlobalConfigToml> {
        if self.directories.enabled {
            Some(&self.directories)
        } else {
            None
        }
    }
}

#[derive(Debug, Deserialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct WorkflowsGlobalConfigToml {
    #[serde(default = "default_workflows_lock_extension_leeway")]
    pub(crate) lock_extension_leeway: DurationConfig,
}
impl Default for WorkflowsGlobalConfigToml {
    fn default() -> WorkflowsGlobalConfigToml {
        WorkflowsGlobalConfigToml {
            lock_extension_leeway: default_workflows_lock_extension_leeway(),
        }
    }
}

#[derive(Debug, Deserialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct ActivitiesDirectoriesGlobalConfigToml {
    #[serde(default = "default_activities_directories_enabled")]
    enabled: bool,
    #[serde(default = "default_activities_directories_parent_directory")]
    parent_directory: String,
    #[serde(default)]
    cleanup: ActivitiesDirectoriesCleanupConfigToml,
}
impl Default for ActivitiesDirectoriesGlobalConfigToml {
    fn default() -> Self {
        Self {
            enabled: default_activities_directories_enabled(),
            parent_directory: default_activities_directories_parent_directory(),
            cleanup: ActivitiesDirectoriesCleanupConfigToml::default(),
        }
    }
}
impl ActivitiesDirectoriesGlobalConfigToml {
    pub(crate) async fn get_parent_directory(
        &self,
        path_prefixes: &PathPrefixes,
    ) -> Result<Arc<Path>, anyhow::Error> {
        assert!(self.enabled); // see `ActivitiesGlobalConfigToml::get_directories`
        path_prefixes
            .server_config_replace_path_prefix_mkdir(&self.parent_directory)
            .await
            .map(Arc::from)
    }

    pub(crate) fn get_cleanup(&self) -> Option<ActivitiesDirectoriesCleanupConfigToml> {
        assert!(self.enabled); // see `ActivitiesGlobalConfigToml::get_directories`
        if self.cleanup.enabled {
            Some(self.cleanup)
        } else {
            None
        }
    }
}

#[derive(Debug, Deserialize, JsonSchema, Clone, Copy)]
#[serde(deny_unknown_fields)]
pub(crate) struct ActivitiesDirectoriesCleanupConfigToml {
    #[serde(default = "default_dir_cleanup_enabled")]
    pub(crate) enabled: bool,
    #[serde(default = "default_dir_cleanup_run_every")]
    pub(crate) run_every: DurationConfig,
    #[serde(default = "default_dir_cleanup_older_than")]
    pub(crate) older_than: DurationConfig,
}
impl Default for ActivitiesDirectoriesCleanupConfigToml {
    fn default() -> Self {
        Self {
            enabled: default_dir_cleanup_enabled(),
            run_every: default_dir_cleanup_run_every(),
            older_than: default_dir_cleanup_older_than(),
        }
    }
}

#[derive(Debug, Deserialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct CodegenCache {
    #[serde(default = "default_codegen_enabled")]
    enabled: bool,
    #[serde(default)]
    directory: Option<String>,
}

impl Default for CodegenCache {
    fn default() -> Self {
        Self {
            enabled: default_codegen_enabled(),
            directory: None,
        }
    }
}

impl CodegenCache {
    pub(crate) async fn get_directory(
        &self,
        path_prefixes: &PathPrefixes,
    ) -> Result<Option<PathBuf>, anyhow::Error> {
        if self.enabled {
            let directory = self.directory.as_deref().unwrap_or_else(|| {
                if path_prefixes.project_dirs.is_some() {
                    DEFAULT_CODEGEN_CACHE_DIRECTORY_IF_PROJECT_DIRS
                } else {
                    DEFAULT_CODEGEN_CACHE_DIRECTORY
                }
            });
            path_prefixes
                .server_config_replace_path_prefix_mkdir(directory)
                .await
                .map(Some)
        } else {
            Ok(None)
        }
    }
}

#[derive(Debug, Deserialize, JsonSchema, Clone, Copy)]
#[serde(deny_unknown_fields)]
pub(crate) struct TimersWatcherTomlConfig {
    #[serde(default = "default_timers_watcher_enabled")]
    pub(crate) enabled: bool,

    #[serde(default = "default_timers_watcher_leeway")]
    // TODO: Add `derivative`,
    // #[derivative(Default(value = "default_timers_watcher_leeway()"))]
    pub(crate) leeway: DurationConfig,
    #[serde(default = "default_timers_watcher_tick_sleep")]
    pub(crate) tick_sleep: DurationConfig,
}
impl Default for TimersWatcherTomlConfig {
    fn default() -> Self {
        Self {
            enabled: default_timers_watcher_enabled(),
            leeway: default_timers_watcher_leeway(),
            tick_sleep: default_timers_watcher_tick_sleep(),
        }
    }
}

#[derive(Debug, Deserialize, JsonSchema, Clone, Copy)]
#[serde(deny_unknown_fields)]
pub(crate) struct CancelWatcherTomlConfig {
    #[serde(default = "default_cancel_watcher_enabled")]
    pub(crate) enabled: bool,
    #[serde(default = "default_cancel_watcher_tick_sleep")]
    pub(crate) tick_sleep: DurationConfig,
}
impl Default for CancelWatcherTomlConfig {
    fn default() -> Self {
        Self {
            enabled: default_cancel_watcher_enabled(),
            tick_sleep: default_cancel_watcher_tick_sleep(),
        }
    }
}

// Components

#[derive(Debug, Clone, Hash)]
pub(crate) struct ComponentCommonVerified {
    pub(crate) name: ConfigName,
    pub(crate) location: ComponentLocationToml,
    pub(crate) content_digest: ContentDigest,
}

#[derive(
    Debug, Clone, Hash, JsonSchema, serde_with::DeserializeFromStr, serde_with::SerializeDisplay,
)]
#[serde(rename_all = "snake_case")]
#[schemars(with = "String")]
pub(crate) enum ComponentLocationToml {
    Path(String), // String because it can contain path prefix
    Oci(
        // #[serde_as(as = "serde_with::DisplayFromStr")]
        oci_client::Reference,
    ),
    GitHub(GitHubReleaseReference),
}
impl ComponentLocationToml {
    /// Fetch wasm file, calculate its content digest.
    ///
    /// Read wasm file either from local fs, pull from an OCI registry, or pull from GitHub release and cache it.
    /// Calculate the `content_digest`. File is not converted from Core to Component format.
    /// If `expected_content_digest` is specified:
    /// - try to find it in cache instead of download. (We trust that the cache was not tampered with).
    /// - if downloaded, digests must match.
    pub(crate) async fn fetch(
        &self,
        wasm_cache_dir: &Path,
        metadata_dir: &Path,
        path_prefixes: &PathPrefixes,
        expected_digest: Option<&ContentDigest>,
    ) -> Result<(ContentDigest, PathBuf), anyhow::Error> {
        use utils::sha256sum::calculate_sha256_file;

        debug!("Fetching {self:?}");
        let stopwatch = std::time::Instant::now();
        // Happy path: if content_digest is known and file exists in cache, return immediately
        if let Some(expected_digest) = expected_digest
            && let wasm_path = content_digest_to_wasm_file(wasm_cache_dir, expected_digest)
            && wasm_path.exists()
        {
            debug!("Using cached file for known content digest");
            return Ok((expected_digest.clone(), wasm_path));
        }

        let (actual_digest, path) = match &self {
            ComponentLocationToml::Path(wasm_path) => {
                let wasm_path =
                    path_prefixes.deployment_config_replace_file_prefix_verify_exists(wasm_path)?;
                let actual_digest = calculate_sha256_file(&wasm_path)
                    .await
                    .with_context(|| format!("cannot compute hash of file `{wasm_path:?}`"))?;
                (actual_digest, wasm_path)
            }
            ComponentLocationToml::Oci(image) => {
                oci::pull_to_cache_dir(image, wasm_cache_dir, metadata_dir)
                    .await
                    .context("try cleaning the cache directory with `--clean-cache`")?
            }
            ComponentLocationToml::GitHub(github_ref) => {
                let (actual_digest, wasm_path) =
                    github::pull_to_cache_dir(github_ref, wasm_cache_dir)
                        .await
                        .context("try cleaning the cache directory with `--clean-cache`")?;
                // Suggest adding content_digest to config
                if expected_digest.is_none() {
                    info!(
                        r#"No content_digest specified for GitHub release component. Consider adding content_digest = "{}" to avoid refetching"#,
                        actual_digest.with_infix(":")
                    );
                }

                (actual_digest, wasm_path)
            }
        };
        if let Some(expected_digest) = expected_digest {
            ensure!(
                *expected_digest == actual_digest,
                "content digest mismatch: expected {expected_digest}, got {actual_digest}"
            );
        }
        let stopwatch = stopwatch.elapsed();
        debug!("Fetching done in {stopwatch:?}");
        Ok((actual_digest, path))
    }
}
pub(crate) const OCI_SCHEMA_PREFIX: &str = "oci://";
impl FromStr for ComponentLocationToml {
    type Err = anyhow::Error;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        if let Some(location) = s.strip_prefix(OCI_SCHEMA_PREFIX) {
            Ok(ComponentLocationToml::Oci(
                oci_client::Reference::from_str(location)
                    .map_err(|e| anyhow::anyhow!("invalid OCI reference: {e}"))?,
            ))
        } else if let Some(location) = s.strip_prefix(GH_SCHEMA_PREFIX) {
            Ok(ComponentLocationToml::GitHub(
                GitHubReleaseReference::from_str(location)?,
            ))
        } else {
            Ok(ComponentLocationToml::Path(s.to_string()))
        }
    }
}

impl std::fmt::Display for ComponentLocationToml {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ComponentLocationToml::Path(p) => write!(f, "{p}"),
            ComponentLocationToml::Oci(r) => write!(f, "{OCI_SCHEMA_PREFIX}{r}"),
            ComponentLocationToml::GitHub(gh) => write!(f, "{GH_SCHEMA_PREFIX}{gh}"),
        }
    }
}

/// Location of a JavaScript source file for JS activities.
/// Supports local file paths and `gh://` GitHub release references.
/// OCI references are not supported.
/// On-disk format only; replaced by [`JsLocationCanonical`] before transmission and hash computation.
#[derive(Debug, Clone, Hash, JsonSchema, SerializeDisplay, DeserializeFromStr)]
#[schemars(with = "String")]
pub(crate) enum JsLocationToml {
    Path(String),
    GitHub(GitHubReleaseReference),
}
impl Display for JsLocationToml {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            JsLocationToml::Path(p) => write!(f, "{p}"),
            JsLocationToml::GitHub(gh) => write!(f, "{GH_SCHEMA_PREFIX}{gh}"),
        }
    }
}
impl FromStr for JsLocationToml {
    type Err = anyhow::Error;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        if s.starts_with(OCI_SCHEMA_PREFIX) {
            bail!(
                "OCI references are not supported for JS activities. Use a local file path or gh:// reference."
            );
        } else if let Some(location) = s.strip_prefix(GH_SCHEMA_PREFIX) {
            Ok(JsLocationToml::GitHub(GitHubReleaseReference::from_str(
                location,
            )?))
        } else {
            Ok(JsLocationToml::Path(s.to_string()))
        }
    }
}

/// Activity, Webhook, Workflow or a Http server
#[derive(
    Debug,
    Clone,
    Hash,
    PartialEq,
    Eq,
    derive_more::Display,
    derive_more::Into,
    JsonSchema,
    derive_more::Deref,
)]
#[display("{_0}")]
pub struct ConfigName(#[schemars(with = "String")] StrVariant);
impl ConfigName {
    pub fn new(name: StrVariant) -> Result<Self, InvalidNameError<ConfigName>> {
        Ok(Self(check_name(name, "_")?))
    }
}
impl<'de> Deserialize<'de> for ConfigName {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let name = String::deserialize(deserializer)?;
        ConfigName::new(StrVariant::from(name)).map_err(serde::de::Error::custom)
    }
}

impl serde::Serialize for ConfigName {
    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        self.0.serialize(s)
    }
}

#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct ComponentCommon {
    pub(crate) name: ConfigName,
    pub(crate) location: ComponentLocationToml,
    /// Content digest of the WASM file.
    /// Recommended for GitHub releases to enable caching and reproducible builds.
    /// If the file is found in cache, the download is bypassed.
    #[serde(default)]
    #[schemars(with = "Option<String>")]
    pub(crate) content_digest: Option<ContentDigest>,
}

impl ComponentCommon {
    async fn fetch(
        self,
        wasm_cache_dir: &Path,
        metadata_dir: &Path,
        path_prefixes: &PathPrefixes,
    ) -> Result<(ComponentCommonVerified, PathBuf), anyhow::Error> {
        let (fetched_digest, wasm_path) = self
            .location
            .fetch(
                wasm_cache_dir,
                metadata_dir,
                path_prefixes,
                self.content_digest.as_ref(),
            )
            .await?;

        let verified = ComponentCommonVerified {
            name: self.name,
            location: self.location,
            content_digest: fetched_digest,
        };
        Ok((verified, wasm_path))
    }
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub(crate) enum LockingStrategy {
    ByFfqns,
    ByComponentDigest,
}
impl From<LockingStrategy> for executor::executor::LockingStrategy {
    fn from(value: LockingStrategy) -> Self {
        match value {
            LockingStrategy::ByFfqns => executor::executor::LockingStrategy::ByFfqns,
            LockingStrategy::ByComponentDigest => {
                executor::executor::LockingStrategy::ByComponentDigest
            }
        }
    }
}
impl From<executor::executor::LockingStrategy> for LockingStrategy {
    fn from(value: executor::executor::LockingStrategy) -> Self {
        match value {
            executor::executor::LockingStrategy::ByFfqns => LockingStrategy::ByFfqns,
            executor::executor::LockingStrategy::ByComponentDigest => {
                LockingStrategy::ByComponentDigest
            }
        }
    }
}

#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct ExecConfigToml {
    #[serde(default = "default_batch_size")]
    batch_size: u32,
    #[serde(default = "default_lock_expiry")]
    lock_expiry: DurationConfig,
    #[serde(default = "default_tick_sleep")]
    tick_sleep: DurationConfig,
    #[serde(default)]
    locking_strategy: Option<LockingStrategy>,
}

impl Default for ExecConfigToml {
    fn default() -> Self {
        Self {
            batch_size: default_batch_size(),
            lock_expiry: default_lock_expiry(),
            tick_sleep: default_tick_sleep(),
            locking_strategy: None,
        }
    }
}

impl ExecConfigToml {
    pub(crate) fn into_exec_exec_config(
        self,
        component_id: ComponentId,
        global_executor_instance_limiter: Option<Arc<tokio::sync::Semaphore>>,
        retry_config: ComponentRetryConfig,
    ) -> executor::executor::ExecConfig {
        executor::executor::ExecConfig {
            lock_expiry: self.lock_expiry.into(),
            tick_sleep: self.tick_sleep.into(),
            batch_size: self.batch_size,
            locking_strategy: locking_strategy(self.locking_strategy, component_id.component_type),
            component_id,
            task_limiter: global_executor_instance_limiter,
            executor_id: ExecutorId::generate(),
            retry_config,
        }
    }
}
fn locking_strategy(
    locking_strategy_override: Option<LockingStrategy>,
    component_type: ComponentType,
) -> executor::executor::LockingStrategy {
    locking_strategy_override.map(executor::executor::LockingStrategy::from).unwrap_or_else(||
    match component_type {
        ComponentType::Activity => executor::executor::LockingStrategy::ByFfqns,
        ComponentType::Workflow => executor::executor::LockingStrategy::ByComponentDigest,
        other => unreachable!(
            "unexpected type {other}, only worklows and activities (wasm,js) expose locking strategy"
        ),
    })
}

#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct ActivityWasmComponentConfigToml {
    #[serde(flatten)]
    pub(crate) common: ComponentCommon,
    /// Override the auto-computed component digest used for locking.
    /// If set, this value is used instead of the content digest of the WASM file.
    #[serde(default)]
    #[schemars(with = "Option<String>")]
    pub(crate) component_digest: Option<ComponentDigest>,
    #[serde(default)]
    pub(crate) exec: ExecConfigToml,
    #[serde(default = "default_max_retries")]
    pub(crate) max_retries: u32,
    #[serde(default = "default_retry_exp_backoff")]
    pub(crate) retry_exp_backoff: DurationConfig,
    #[serde(default)]
    pub(crate) forward_stdout: ComponentStdOutputToml,
    #[serde(default)]
    pub(crate) forward_stderr: ComponentStdOutputToml,
    #[serde(default)]
    pub(crate) env_vars: Vec<EnvVarConfig>,
    #[serde(default)]
    pub(crate) directories: ActivityDirectoriesConfigToml,
    #[serde(default)]
    pub(crate) logs_store_min_level: LogLevelToml,
    /// Allowed outgoing HTTP hosts with optional method restrictions and secrets.
    #[serde(default, rename = "allowed_host")]
    pub(crate) allowed_hosts: Vec<AllowedHostToml>,
}

#[derive(Debug, Default, Deserialize, Serialize, JsonSchema, Clone, Copy)]
#[serde(rename_all = "snake_case")]
pub(crate) enum LogLevelToml {
    Off,
    Trace,
    #[default]
    Debug,
    Info,
    Warn,
    Error,
}
impl From<LogLevelToml> for Option<LogLevel> {
    fn from(value: LogLevelToml) -> Self {
        match value {
            LogLevelToml::Off => None,
            LogLevelToml::Trace => Some(LogLevel::Trace),
            LogLevelToml::Debug => Some(LogLevel::Debug),
            LogLevelToml::Info => Some(LogLevel::Info),
            LogLevelToml::Warn => Some(LogLevel::Warn),
            LogLevelToml::Error => Some(LogLevel::Error),
        }
    }
}

/// Where in the outgoing request placeholders are replaced.
#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone, Copy, PartialEq, Eq, Hash)]
#[serde(rename_all = "snake_case")]
pub(crate) enum ReplaceIn {
    Headers,
    Body,
    Params,
}

/// Input for method restrictions in TOML configuration.
/// Supports both `methods = "*"` and `methods = ["GET", "POST"]` syntax.
#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(untagged)]
pub(crate) enum MethodsInput {
    /// All methods allowed (from `methods = "*"`).
    Star(MethodsInputStar),
    /// Specific methods list (from `methods = ["GET", "POST"]`).
    List(Vec<String>),
}

#[derive(Debug, Default, Deserialize, Serialize, JsonSchema, Clone)]
pub(crate) struct MethodsInputStar(
    #[serde(
        deserialize_with = "deserialize_star",
        serialize_with = "serialize_star"
    )]
    (),
);

fn deserialize_star<'de, D>(deserializer: D) -> Result<(), D::Error>
where
    D: Deserializer<'de>,
{
    let s = String::deserialize(deserializer)?;
    if s == "*" {
        Ok(())
    } else {
        Err(serde::de::Error::custom(format!(
            "expected \"*\", got \"{s}\""
        )))
    }
}

fn serialize_star<S: serde::Serializer>(_: &(), s: S) -> Result<S::Ok, S::Error> {
    s.serialize_str("*")
}

/// An allowed outgoing HTTP host with optional method restrictions and secrets.
#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct AllowedHostToml {
    /// Host pattern (e.g. `"api.example.com"`, `"*.example.com"`, `"http://localhost:8080"`).
    pub pattern: String,
    /// Allowed HTTP methods.
    /// - Omit to allow nothing (warning emitted).
    /// - `methods = "*"` or `methods = ["*"]` to allow all methods.
    /// - `methods = ["GET", "POST"]` to allow specific methods.
    /// - `methods = []` to allow nothing (warning emitted).
    pub methods: Option<MethodsInput>,
    /// Optional secrets for this host.
    #[serde(default)]
    pub secrets: Option<AllowedHostSecretsToml>,
}

/// Secrets configuration for an allowed host.
#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct AllowedHostSecretsToml {
    /// Env vars using the same syntax as top-level `env_vars`.
    pub env_vars: Vec<EnvVarConfig>,
    /// Where in the request to perform replacement.
    /// Default: empty (no replacement — deny by default).
    #[serde(default)]
    pub replace_in: Vec<ReplaceIn>,
}

#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct ActivityStubFileConfigToml {
    pub(crate) name: ConfigName,
    pub(crate) location: ComponentLocationToml,
    #[serde(default)]
    #[schemars(with = "Option<String>")]
    pub(crate) content_digest: Option<ContentDigest>,
}

#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct ActivityStubInlineConfigToml {
    pub(crate) name: ConfigName,
    #[schemars(with = "String")]
    pub(crate) ffqn: FunctionFqn,
    #[serde(default)]
    pub(crate) params: Option<Vec<JsParamToml>>,
    #[serde(default)]
    pub(crate) return_type: Option<String>,
}

#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(untagged)]
pub(crate) enum ActivityStubComponentConfigToml {
    File(ActivityStubFileConfigToml),
    Inline(ActivityStubInlineConfigToml),
}
#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct ActivityExternalFileConfigToml {
    #[serde(flatten)]
    pub(crate) common: ComponentCommon,
    /// Override the auto-computed component digest used for locking.
    /// If set, this value is used instead of the content digest of the WASM file.
    #[serde(default)]
    #[schemars(with = "Option<String>")]
    pub(crate) component_digest: Option<ComponentDigest>,
}

#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(untagged)]
pub(crate) enum ActivityExternalComponentConfigToml {
    File(ActivityExternalFileConfigToml),
    Inline(ActivityStubInlineConfigToml),
}
#[derive(Debug)]
pub(crate) struct ActivityStubExtConfigVerified {
    pub(crate) wasm_path: PathBuf,
    pub(crate) component_id: ComponentId,
}

#[derive(Debug)]
pub(crate) struct ActivityStubExtInlineConfigVerified {
    pub(crate) component_id: ComponentId,
    pub(crate) ffqn: FunctionFqn,
    pub(crate) params: Vec<concepts::ParameterType>,
    pub(crate) return_type: concepts::ReturnTypeExtendable,
}

#[derive(Debug)]
pub(crate) enum ActivityStubConfigVerified {
    File(ActivityStubExtConfigVerified),
    Inline(ActivityStubExtInlineConfigVerified),
}

impl ActivityStubComponentConfigToml {
    fn name_str(&self) -> &str {
        match self {
            Self::File(f) => f.name.0.as_ref(),
            Self::Inline(i) => i.name.0.as_ref(),
        }
    }

    #[instrument(skip_all, fields(component_name = self.name_str(), component_id))]
    pub(crate) async fn fetch_and_verify(
        self,
        wasm_cache_dir: Arc<Path>,
        metadata_dir: Arc<Path>,
        path_prefixes: Arc<PathPrefixes>,
    ) -> Result<ActivityStubConfigVerified, anyhow::Error> {
        match self {
            Self::File(file) => {
                let common = ComponentCommon {
                    name: file.name,
                    location: file.location,
                    content_digest: file.content_digest,
                };
                let (common, wasm_path) = common
                    .fetch(&wasm_cache_dir, &metadata_dir, &path_prefixes)
                    .await?;
                let component_id = ComponentId::new(
                    ComponentType::ActivityStub,
                    StrVariant::from(common.name),
                    ComponentDigest(common.content_digest.0),
                )?;
                Ok(ActivityStubConfigVerified::File(
                    ActivityStubExtConfigVerified {
                        wasm_path,
                        component_id,
                    },
                ))
            }
            Self::Inline(inline) => {
                let ffqn = inline.ffqn;
                let parsed_params = match inline.params {
                    None => {
                        vec![concepts::ParameterType {
                            type_wrapper: val_json::type_wrapper::TypeWrapper::List(Box::new(
                                val_json::type_wrapper::TypeWrapper::String,
                            )),
                            name: StrVariant::Static("params"),
                            wit_type: StrVariant::Static("list<string>"),
                        }]
                    }
                    Some(params) => params
                        .iter()
                        .map(|p| {
                            let tw = val_json::type_wrapper::parse_wit_type(&p.wit_type)
                                .map_err(|e| anyhow!("invalid param type `{}`: {e}", p.wit_type))?;
                            Ok(concepts::ParameterType {
                                type_wrapper: tw,
                                name: StrVariant::from(p.name.clone()),
                                wit_type: StrVariant::from(p.wit_type.clone()),
                            })
                        })
                        .collect::<Result<Vec<_>, anyhow::Error>>()?,
                };

                const DEFAULT_RETURN_TYPE: &str = "result<string, string>";
                let return_type_str = inline.return_type.as_deref().unwrap_or(DEFAULT_RETURN_TYPE);
                let return_type_tw = val_json::type_wrapper::parse_wit_type(return_type_str)
                    .map_err(|e| anyhow!("invalid return_type `{return_type_str}`: {e}"))?;
                let return_type = concepts::ReturnType::detect(
                    return_type_tw,
                    StrVariant::from(return_type_str.to_string()),
                );
                let return_type = match return_type {
                    ReturnType::Extendable(rt) => rt,
                    ReturnType::NonExtendable(_) => bail!(
                        "return_type must be `result`, `result<T>`, `result<T, string>`, or \
                         `result<T, variant {{ execution-failed, ... }}>`, got `{return_type_str}`"
                    ),
                };

                // Compute component digest: SHA256 of prefix + ffqn + params + return_type
                let mut hasher = Sha256::new();
                hasher.update(b"activity_stub_inline:");
                hasher.update(ffqn.to_string().as_bytes());
                for p in &parsed_params {
                    hasher.update(p.wit_type.as_ref().as_bytes());
                }
                hasher.update(return_type.wit_type.as_bytes());
                let hash: [u8; 32] = hasher.finalize().into();
                let component_digest = ComponentDigest(Digest(hash));

                let component_id = ComponentId::new(
                    ComponentType::ActivityStub,
                    StrVariant::from(inline.name),
                    component_digest,
                )?;

                Ok(ActivityStubConfigVerified::Inline(
                    ActivityStubExtInlineConfigVerified {
                        component_id,
                        ffqn,
                        params: parsed_params,
                        return_type,
                    },
                ))
            }
        }
    }
}
#[derive(Debug)]
pub(crate) enum ActivityExternalConfigVerified {
    File(ActivityStubExtConfigVerified),
    Inline(ActivityStubExtInlineConfigVerified),
}

impl ActivityExternalComponentConfigToml {
    fn name_str(&self) -> &str {
        match self {
            Self::File(f) => f.common.name.0.as_ref(),
            Self::Inline(i) => i.name.0.as_ref(),
        }
    }

    #[instrument(skip_all, fields(component_name = self.name_str(), component_id))]
    pub(crate) async fn fetch_and_verify(
        self,
        wasm_cache_dir: Arc<Path>,
        metadata_dir: Arc<Path>,
        path_prefixes: Arc<PathPrefixes>,
    ) -> Result<ActivityExternalConfigVerified, anyhow::Error> {
        match self {
            Self::File(file) => {
                let component_digest_override = file.component_digest;
                let (common, wasm_path) = file
                    .common
                    .fetch(&wasm_cache_dir, &metadata_dir, &path_prefixes)
                    .await?;
                let component_digest =
                    component_digest_override.unwrap_or(ComponentDigest(common.content_digest.0));
                let component_id = ComponentId::new(
                    ComponentType::Activity,
                    StrVariant::from(common.name),
                    component_digest,
                )?;
                Ok(ActivityExternalConfigVerified::File(
                    ActivityStubExtConfigVerified {
                        wasm_path,
                        component_id,
                    },
                ))
            }
            Self::Inline(inline) => {
                let ffqn = inline.ffqn;
                let parsed_params = match inline.params {
                    None => {
                        vec![concepts::ParameterType {
                            type_wrapper: val_json::type_wrapper::TypeWrapper::List(Box::new(
                                val_json::type_wrapper::TypeWrapper::String,
                            )),
                            name: StrVariant::Static("params"),
                            wit_type: StrVariant::Static("list<string>"),
                        }]
                    }
                    Some(params) => params
                        .iter()
                        .map(|p| {
                            let tw = val_json::type_wrapper::parse_wit_type(&p.wit_type)
                                .map_err(|e| anyhow!("invalid param type `{}`: {e}", p.wit_type))?;
                            Ok(concepts::ParameterType {
                                type_wrapper: tw,
                                name: StrVariant::from(p.name.clone()),
                                wit_type: StrVariant::from(p.wit_type.clone()),
                            })
                        })
                        .collect::<Result<Vec<_>, anyhow::Error>>()?,
                };

                const DEFAULT_RETURN_TYPE: &str = "result<string, string>";
                let return_type_str = inline.return_type.as_deref().unwrap_or(DEFAULT_RETURN_TYPE);
                let return_type_tw = val_json::type_wrapper::parse_wit_type(return_type_str)
                    .map_err(|e| anyhow!("invalid return_type `{return_type_str}`: {e}"))?;
                let return_type = concepts::ReturnType::detect(
                    return_type_tw,
                    StrVariant::from(return_type_str.to_string()),
                );
                let return_type = match return_type {
                    ReturnType::Extendable(rt) => rt,
                    ReturnType::NonExtendable(_) => bail!(
                        "return_type must be `result`, `result<T>`, `result<T, string>`, or \
                         `result<T, variant {{ execution-failed, ... }}>`, got `{return_type_str}`"
                    ),
                };

                // Compute component digest: SHA256 of prefix + ffqn + params + return_type
                let mut hasher = Sha256::new();
                hasher.update(b"activity_external_inline:");
                hasher.update(ffqn.to_string().as_bytes());
                for p in &parsed_params {
                    hasher.update(p.wit_type.as_ref().as_bytes());
                }
                hasher.update(return_type.wit_type.as_bytes());
                let hash: [u8; 32] = hasher.finalize().into();
                let component_digest = ComponentDigest(Digest(hash));

                let component_id = ComponentId::new(
                    ComponentType::Activity,
                    StrVariant::from(inline.name),
                    component_digest,
                )?;

                Ok(ActivityExternalConfigVerified::Inline(
                    ActivityStubExtInlineConfigVerified {
                        component_id,
                        ffqn,
                        params: parsed_params,
                        return_type,
                    },
                ))
            }
        }
    }
}

#[derive(Debug, Default, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct ActivityDirectoriesConfigToml {
    #[serde(default)]
    enabled: bool,
    #[serde(default)]
    reuse_on_retry: bool,
    #[serde(default)]
    process_provider: ActivityDirectoriesProcessProvider,
}

#[derive(Debug, Default, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(rename_all = "snake_case")]
pub(crate) enum ActivityDirectoriesProcessProvider {
    #[default]
    None,
    Native,
}
impl From<ActivityDirectoriesProcessProvider> for Option<ProcessProvider> {
    fn from(value: ActivityDirectoriesProcessProvider) -> Self {
        match value {
            ActivityDirectoriesProcessProvider::None => None,
            ActivityDirectoriesProcessProvider::Native => Some(ProcessProvider::Native),
        }
    }
}

#[derive(Debug)]
pub(crate) struct ActivityWasmConfigVerified {
    pub(crate) wasm_path: PathBuf,
    pub(crate) activity_config: ActivityConfig,
    pub(crate) exec_config: executor::executor::ExecConfig,
    pub(crate) logs_store_min_level: Option<LogLevel>,
}

impl ActivityWasmConfigVerified {
    pub fn component_id(&self) -> &ComponentId {
        &self.activity_config.component_id
    }
}

impl ActivityWasmComponentConfigToml {
    #[instrument(skip_all, fields(component_name = self.common.name.0.as_ref()))]
    #[expect(clippy::too_many_arguments)]
    pub(crate) async fn fetch_and_verify(
        self,
        wasm_cache_dir: Arc<Path>,
        metadata_dir: Arc<Path>,
        path_prefixes: Arc<PathPrefixes>,
        ignore_missing_env_vars: bool,
        parent_preopen_dir: Option<Arc<Path>>,
        global_executor_instance_limiter: Option<Arc<tokio::sync::Semaphore>>,
        fuel: Option<u64>,
    ) -> Result<ActivityWasmConfigVerified, anyhow::Error> {
        let (common, wasm_path) = self
            .common
            .fetch(&wasm_cache_dir, &metadata_dir, &path_prefixes)
            .await?;

        let env_vars = resolve_env_vars_plaintext(self.env_vars, ignore_missing_env_vars)?;
        let allowed_hosts = resolve_allowed_hosts(self.allowed_hosts, ignore_missing_env_vars)?;

        // Validate no collision between env_vars and secret env names
        validate_no_env_collision(&env_vars, &allowed_hosts)?;

        let directories_config = match (parent_preopen_dir, self.directories.enabled) {
            (Some(parent_preopen_dir), true) => Some(ActivityDirectoriesConfig {
                parent_preopen_dir,
                reuse_on_retry: self.directories.reuse_on_retry,
                process_provider: self.directories.process_provider.into(),
            }),
            (None, true) => {
                bail!(
                    "`directories.enabled` set to true for activity `{}` while the global setting `activities.directories.enabled` is false",
                    common.name.0
                );
            }
            (_, false) => None,
        };

        let component_digest = self
            .component_digest
            .unwrap_or(ComponentDigest(common.content_digest.0));
        let component_id = ComponentId::new(
            ComponentType::Activity,
            StrVariant::from(common.name),
            component_digest,
        )?;
        let activity_config = ActivityConfig {
            component_id: component_id.clone(),
            forward_stdout: self.forward_stdout.into(),
            forward_stderr: self.forward_stderr.into(),
            env_vars,
            directories_config,
            fuel,
            allowed_hosts,
        };
        let retry_config = ComponentRetryConfig {
            max_retries: Some(self.max_retries),
            retry_exp_backoff: self.retry_exp_backoff.into(),
        };
        Ok(ActivityWasmConfigVerified {
            wasm_path,
            activity_config,
            exec_config: self.exec.into_exec_exec_config(
                component_id,
                global_executor_instance_limiter,
                retry_config,
            ),
            logs_store_min_level: self.logs_store_min_level.into(),
        })
    }
}

#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct ActivityJsComponentConfigToml {
    pub(crate) name: ConfigName,
    /// Location of the JavaScript source file.
    /// Supports local file paths and `gh://` GitHub release references.
    /// OCI references are not supported for JS activities.
    pub(crate) location: JsLocationToml,
    /// Content digest of the JS source file.
    /// Recommended for GitHub releases to enable caching and reproducible builds.
    #[serde(default)]
    #[schemars(with = "Option<String>")]
    pub(crate) content_digest: Option<ContentDigest>,
    /// Override the auto-computed component digest used for locking.
    /// If set, this value is used instead of the digest derived from the JS source, ffqn, and params.
    #[serde(default)]
    #[schemars(with = "Option<String>")]
    pub(crate) component_digest: Option<ComponentDigest>,
    #[schemars(with = "String")]
    pub(crate) ffqn: FunctionFqn,
    /// Custom parameters for the JS function.
    /// Each entry has a `name` and a WIT `type` (e.g. `string`, `u32`, `list<string>`).
    /// If omitted, defaults to no parameters.
    #[serde(default)]
    pub(crate) params: Option<Vec<JsParamToml>>,
    #[serde(default)]
    pub(crate) exec: ExecConfigToml,
    #[serde(default = "default_max_retries")]
    pub(crate) max_retries: u32,
    #[serde(default = "default_retry_exp_backoff")]
    pub(crate) retry_exp_backoff: DurationConfig,
    #[serde(default)]
    pub(crate) forward_stdout: ComponentStdOutputToml,
    #[serde(default)]
    pub(crate) forward_stderr: ComponentStdOutputToml,
    #[serde(default)]
    pub(crate) logs_store_min_level: LogLevelToml,
    #[serde(default)]
    pub(crate) env_vars: Vec<EnvVarConfig>,
    /// Allowed outgoing HTTP hosts with optional method restrictions and secrets.
    #[serde(default, rename = "allowed_host")]
    pub(crate) allowed_hosts: Vec<AllowedHostToml>,
    /// WIT return type. Defaults to `result`.
    /// Must be `result<T, string>` — the error type must be `string` since JS throws strings.
    #[serde(default)]
    pub(crate) return_type: Option<String>,
}
/// A parameter declaration for a JS activity function.
#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct JsParamToml {
    /// Parameter name (used in WIT metadata).
    pub(crate) name: String,
    /// WIT type string, e.g. `string`, `u32`, `list<string>`, `option<u64>`.
    #[serde(rename = "type")]
    pub(crate) wit_type: String,
}

#[derive(Debug)]
pub(crate) struct ActivityJsConfigVerified {
    pub(crate) wasm_path: Arc<Path>, // same for all JS activities
    pub(crate) js_source: String,
    pub(crate) ffqn: FunctionFqn,
    pub(crate) params: Vec<concepts::ParameterType>,
    pub(crate) return_type: concepts::ReturnTypeExtendable,
    pub(crate) activity_config: ActivityConfig,
    pub(crate) exec_config: executor::executor::ExecConfig,
    pub(crate) logs_store_min_level: Option<LogLevel>,
}

impl ActivityJsConfigVerified {
    pub fn component_id(&self) -> &ComponentId {
        &self.activity_config.component_id
    }
}

#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct WorkflowJsComponentConfigToml {
    pub(crate) name: ConfigName,
    /// Location of the JavaScript source file.
    /// Supports local file paths and `gh://` GitHub release references.
    /// OCI references are not supported for JS workflows.
    pub(crate) location: JsLocationToml,
    /// Content digest of the JS source file.
    /// Recommended for GitHub releases to enable caching and reproducible builds.
    #[serde(default)]
    #[schemars(with = "Option<String>")]
    pub(crate) content_digest: Option<ContentDigest>,
    /// Override the auto-computed component digest used for locking.
    /// If set, this value is used instead of the digest derived from the JS source, ffqn, and params.
    #[serde(default)]
    #[schemars(with = "Option<String>")]
    pub(crate) component_digest: Option<ComponentDigest>,
    #[schemars(with = "String")]
    pub(crate) ffqn: FunctionFqn,
    /// Custom parameters for the JS workflow function.
    /// Each entry has a `name` and a WIT `type` (e.g. `string`, `u32`, `list<string>`).
    /// If omitted, defaults to no parameters.
    #[serde(default)]
    pub(crate) params: Option<Vec<JsParamToml>>,
    #[serde(default)]
    pub(crate) exec: ExecConfigToml,
    #[serde(default = "default_retry_exp_backoff")]
    pub(crate) retry_exp_backoff: DurationConfig,
    #[serde(default)]
    pub(crate) blocking_strategy: BlockingStrategyConfigToml,
    #[serde(default = "default_lock_extension")]
    pub(crate) lock_extension: bool,
    #[serde(default)]
    pub(crate) logs_store_min_level: LogLevelToml,
    /// WIT return type. Defaults to `result`.
    /// Must be `result`, `result<T>`, `result<T, string>`, or
    /// `result<T, variant { execution-failed, ... }>`.
    #[serde(default)]
    pub(crate) return_type: Option<String>,
}

#[derive(Debug)]
pub(crate) struct WorkflowJsConfigVerified {
    pub(crate) wasm_path: Arc<Path>, // same for all JS workflows
    pub(crate) js_source: String,
    pub(crate) js_file_name: String,
    pub(crate) ffqn: FunctionFqn,
    pub(crate) params: Vec<concepts::ParameterType>,
    pub(crate) return_type: concepts::ReturnTypeExtendable,
    pub(crate) workflow_config: WorkflowConfig,
    pub(crate) exec_config: executor::executor::ExecConfig,
    pub(crate) logs_store_min_level: Option<LogLevel>,
}

impl WorkflowJsConfigVerified {
    pub fn component_id(&self) -> &ComponentId {
        &self.workflow_config.component_id
    }

    pub(crate) fn as_frame_sources(&self) -> FrameFilesToSourceContent {
        FrameFilesToSourceContent::from([(self.js_file_name.clone(), self.js_source.clone())])
    }
}

#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct WorkflowWasmComponentConfigToml {
    #[serde(flatten)]
    pub(crate) common: ComponentCommon,
    /// Override the auto-computed component digest used for locking.
    /// If set, this value is used instead of the content digest of the WASM file.
    #[serde(default)]
    #[schemars(with = "Option<String>")]
    pub(crate) component_digest: Option<ComponentDigest>,
    #[serde(default)]
    pub(crate) exec: ExecConfigToml,
    #[serde(default = "default_retry_exp_backoff")]
    pub(crate) retry_exp_backoff: DurationConfig,
    #[serde(default)]
    pub(crate) blocking_strategy: BlockingStrategyConfigToml,
    #[serde(default)]
    pub(crate) backtrace: ComponentBacktraceConfig,
    #[serde(default)]
    pub(crate) stub_wasi: bool,
    #[serde(default = "default_lock_extension")]
    pub(crate) lock_extension: bool,
    #[serde(default)]
    pub(crate) logs_store_min_level: LogLevelToml,
}

#[derive(Debug, Deserialize, Serialize, Clone, Copy, JsonSchema, PartialEq)]
#[serde(untagged)] // Try variants without needing a specific outer tag
pub(crate) enum BlockingStrategyConfigToml {
    // Try the more specific map format first
    Tagged(BlockingStrategyConfigCustomized),
    // If it's not the map format, try the simple string format
    Simple(BlockingStrategyConfigSimple),
}
impl Default for BlockingStrategyConfigToml {
    fn default() -> Self {
        Self::Simple(BlockingStrategyConfigSimple::default())
    }
}
// Enum to handle the tagged map case ({ kind = "await", ... })
#[derive(Debug, Deserialize, Serialize, Clone, Copy, JsonSchema, PartialEq)]
#[serde(tag = "kind", rename_all = "snake_case")] // Expects a map with "kind" field
pub(crate) enum BlockingStrategyConfigCustomized {
    Await(BlockingStrategyAwaitConfig),
}

#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, JsonSchema)]
#[serde(deny_unknown_fields)]
pub(crate) struct BlockingStrategyAwaitConfig {
    #[serde(default = "default_non_blocking_event_batching")]
    non_blocking_event_batching: u32,
}
#[derive(Debug, Deserialize, Serialize, Clone, Copy, JsonSchema, Default, PartialEq)]
#[serde(rename_all = "snake_case")]
pub(crate) enum BlockingStrategyConfigSimple {
    Interrupt,
    #[default]
    Await,
}
impl From<BlockingStrategyConfigToml> for JoinNextBlockingStrategy {
    fn from(input: BlockingStrategyConfigToml) -> Self {
        match input {
            BlockingStrategyConfigToml::Tagged(BlockingStrategyConfigCustomized::Await(
                BlockingStrategyAwaitConfig {
                    non_blocking_event_batching,
                },
            )) => JoinNextBlockingStrategy::Await {
                non_blocking_event_batching,
            },
            BlockingStrategyConfigToml::Simple(BlockingStrategyConfigSimple::Interrupt) => {
                JoinNextBlockingStrategy::Interrupt
            }
            BlockingStrategyConfigToml::Simple(BlockingStrategyConfigSimple::Await) => {
                JoinNextBlockingStrategy::Await {
                    non_blocking_event_batching: DEFAULT_NON_BLOCKING_EVENT_BATCHING,
                }
            }
        }
    }
}
/// Location of a WASM backtrace source file.
/// On-disk format only; resolved to `ComponentBacktraceConfigCanonical` before transmission
/// and hash computation.
///
/// Serialised as: `"path/to/file.ts"` → `Path`
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(untagged)]
#[schemars(with = "String")]
pub(crate) enum BacktraceSourceLocation {
    /// Local file path — resolved at submit/startup time via `resolve_to_canonical`.
    Path(String),
}

#[derive(Debug, Deserialize, Serialize, JsonSchema, Default, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct ComponentBacktraceConfig {
    #[serde(rename = "sources")]
    #[schemars(with = "std::collections::HashMap<String, String>")]
    pub(crate) frame_files_to_sources: HashMap<String, BacktraceSourceLocation>,
}
/// Canonical JS location — no local file paths.
/// Used for hash computation, wire format in deployment submission, and DB storage.
#[derive(Debug, Clone, Hash, Serialize, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub(crate) enum JsLocationCanonical {
    #[schemars(with = "String")]
    GitHub(GitHubReleaseReference),
    Content {
        content: String,
        file_name: String,
    },
}
impl JsLocationCanonical {
    pub(crate) fn file_name(&self) -> String {
        match self {
            JsLocationCanonical::GitHub(gh) => gh.asset_name.clone(),
            JsLocationCanonical::Content { file_name, .. } => file_name.clone(),
        }
    }

    /// Return the JS source content. For `Content`, returns it directly (validating digest if
    /// provided). For `GitHub`, downloads to `wasm_cache_dir`.
    pub(crate) async fn get_content(
        &self,
        wasm_cache_dir: &Path,
        expected_digest: Option<&ContentDigest>,
    ) -> anyhow::Result<String> {
        match self {
            JsLocationCanonical::Content { content, file_name } => {
                if let Some(expected) = expected_digest {
                    let hash: [u8; 32] = Sha256::digest(content.as_bytes()).into();
                    let actual = ContentDigest(Digest(hash));
                    ensure!(
                        *expected == actual,
                        "content digest mismatch for inline JS `{file_name}`: expected {expected}, got {actual}"
                    );
                }
                Ok(content.clone())
            }
            JsLocationCanonical::GitHub(github_ref) => {
                if let Some(expected) = expected_digest {
                    let cached = content_digest_to_wasm_file(wasm_cache_dir, expected);
                    if cached.exists() {
                        trace!("Using cached JS source for known content digest");
                        return tokio::fs::read_to_string(&cached).await.with_context(|| {
                            format!("cannot read cached JS source file `{cached:?}`")
                        });
                    }
                }
                let (actual_digest, cached_path) =
                    github::pull_to_cache_dir(github_ref, wasm_cache_dir)
                        .await
                        .context("cannot fetch JS source from GitHub release")?;
                if let Some(expected) = expected_digest {
                    ensure!(
                        *expected == actual_digest,
                        "content digest mismatch for JS source: expected {expected}, got {actual_digest}"
                    );
                } else {
                    info!(
                        r#"No content_digest specified for GitHub release JS source. Consider adding content_digest = "{}" to avoid refetching"#,
                        actual_digest.with_infix(":")
                    );
                }
                tokio::fs::read_to_string(&cached_path)
                    .await
                    .with_context(|| format!("cannot read cached JS source file `{cached_path:?}`"))
            }
        }
    }
}

#[derive(Debug)]
pub(crate) struct WorkflowConfigVerified {
    pub(crate) wasm_path: PathBuf,
    pub(crate) workflow_config: WorkflowConfig,
    pub(crate) exec_config: executor::executor::ExecConfig,
    pub(crate) frame_files_to_sources: FrameFilesToSourceContent,
    pub(crate) logs_store_min_level: Option<LogLevel>,
}

impl WorkflowConfigVerified {
    pub fn component_id(&self) -> &ComponentId {
        &self.workflow_config.component_id
    }
}

// Canonical component config types
// Used for wire format, hash computation, and DB storage.

#[derive(schemars::JsonSchema, Debug, Default, Clone, Serialize, Deserialize)]
pub(crate) struct ComponentBacktraceConfigCanonical {
    #[schemars(with = "std::collections::HashMap<String, String>")]
    pub(crate) frame_files_to_sources: HashMap<String, String>,
}

impl ComponentBacktraceConfigCanonical {
    pub(crate) fn into_frame_files(self) -> FrameFilesToSourceContent {
        self.frame_files_to_sources
    }
}

/// Canonical form of `ActivityJsComponentConfigToml`.
#[derive(schemars::JsonSchema, Debug, Deserialize, Serialize, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct ActivityJsComponentConfigCanonical {
    pub(crate) name: ConfigName,
    pub(crate) location: JsLocationCanonical,
    pub(crate) content_digest: Option<ContentDigest>,
    pub(crate) component_digest: Option<ComponentDigest>,
    pub(crate) ffqn: FunctionFqn,
    pub(crate) params: Option<Vec<JsParamToml>>,
    pub(crate) exec: ExecConfigToml,
    pub(crate) max_retries: u32,
    pub(crate) retry_exp_backoff: DurationConfig,
    pub(crate) forward_stdout: ComponentStdOutputToml,
    pub(crate) forward_stderr: ComponentStdOutputToml,
    pub(crate) logs_store_min_level: LogLevelToml,
    pub(crate) env_vars: Vec<EnvVarConfig>,
    pub(crate) allowed_hosts: Vec<AllowedHostToml>,
    pub(crate) return_type: Option<String>,
}

impl ActivityJsComponentConfigCanonical {
    #[instrument(skip_all, fields(component_name = self.name.0.as_ref()))]
    pub(crate) async fn fetch_and_verify(
        self,
        wasm_path: Arc<Path>,
        wasm_cache_dir: Arc<Path>,
        ignore_missing_env_vars: bool,
        global_executor_instance_limiter: Option<Arc<tokio::sync::Semaphore>>,
        fuel: Option<u64>,
    ) -> Result<ActivityJsConfigVerified, anyhow::Error> {
        let parsed_params = match self.params {
            None => vec![],
            Some(params) => params
                .iter()
                .map(|p| {
                    let tw = val_json::type_wrapper::parse_wit_type(&p.wit_type)
                        .map_err(|e| anyhow!("invalid param type `{}`: {e}", p.wit_type))?;
                    Ok(concepts::ParameterType {
                        type_wrapper: tw,
                        name: StrVariant::from(p.name.clone()),
                        wit_type: StrVariant::from(p.wit_type.clone()),
                    })
                })
                .collect::<Result<Vec<_>, anyhow::Error>>()?,
        };
        let js_source = self
            .location
            .get_content(&wasm_cache_dir, self.content_digest.as_ref())
            .await?;
        const DEFAULT_RETURN_TYPE: &str = "result";
        let return_type_str = self.return_type.as_deref().unwrap_or(DEFAULT_RETURN_TYPE);
        let return_type_tw = val_json::type_wrapper::parse_wit_type(return_type_str)
            .map_err(|e| anyhow!("invalid return_type `{return_type_str}`: {e}"))?;
        let return_type = concepts::ReturnType::detect(
            return_type_tw,
            StrVariant::from(return_type_str.to_string()),
        );
        let return_type = match return_type {
            ReturnType::Extendable(rt) => rt,
            ReturnType::NonExtendable(_) => bail!(
                "return_type must be `result`, `result<T>`, `result<T, string>`, or \
                 `result<T, variant {{ execution-failed, ... }}>`, got `{return_type_str}`"
            ),
        };
        let component_digest = self.component_digest.unwrap_or_else(|| {
            let mut hasher = Sha256::new();
            hasher.update(b"activity_js:");
            hasher.update(js_source.as_bytes());
            hasher.update(self.ffqn.to_string().as_bytes());
            for p in &parsed_params {
                hasher.update(p.wit_type.as_ref().as_bytes());
            }
            hasher.update(return_type.wit_type.as_bytes());
            let hash: [u8; 32] = hasher.finalize().into();
            ComponentDigest(Digest(hash))
        });
        let component_id = ComponentId::new(
            ComponentType::Activity,
            StrVariant::from(self.name),
            component_digest,
        )?;
        let env_vars = resolve_env_vars_plaintext(self.env_vars, ignore_missing_env_vars)?;
        let allowed_hosts = resolve_allowed_hosts(self.allowed_hosts, ignore_missing_env_vars)?;
        validate_no_env_collision(&env_vars, &allowed_hosts)?;
        let activity_config = ActivityConfig {
            component_id: component_id.clone(),
            forward_stdout: self.forward_stdout.into(),
            forward_stderr: self.forward_stderr.into(),
            env_vars,
            directories_config: None,
            fuel,
            allowed_hosts,
        };
        let retry_config = ComponentRetryConfig {
            max_retries: Some(self.max_retries),
            retry_exp_backoff: self.retry_exp_backoff.into(),
        };
        Ok(ActivityJsConfigVerified {
            wasm_path,
            js_source,
            ffqn: self.ffqn,
            params: parsed_params,
            return_type,
            activity_config,
            exec_config: self.exec.into_exec_exec_config(
                component_id,
                global_executor_instance_limiter,
                retry_config,
            ),
            logs_store_min_level: self.logs_store_min_level.into(),
        })
    }
}

/// Canonical form of `WorkflowWasmComponentConfigToml`.
#[derive(schemars::JsonSchema, Debug, Deserialize, Serialize, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct WorkflowWasmComponentConfigCanonical {
    pub(crate) common: ComponentCommon,
    pub(crate) component_digest: Option<ComponentDigest>,
    pub(crate) exec: ExecConfigToml,
    pub(crate) retry_exp_backoff: DurationConfig,
    pub(crate) blocking_strategy: BlockingStrategyConfigToml,
    pub(crate) backtrace: ComponentBacktraceConfigCanonical,
    pub(crate) stub_wasi: bool,
    pub(crate) lock_extension: bool,
    pub(crate) logs_store_min_level: LogLevelToml,
}

impl WorkflowWasmComponentConfigCanonical {
    #[instrument(skip_all, fields(component_name = self.common.name.0.as_ref()))]
    #[expect(clippy::too_many_arguments)]
    pub(crate) async fn fetch_and_verify(
        self,
        wasm_cache_dir: Arc<Path>,
        metadata_dir: Arc<Path>,
        path_prefixes: Arc<PathPrefixes>,
        global_backtrace_persist: bool,
        global_executor_instance_limiter: Option<Arc<tokio::sync::Semaphore>>,
        fuel: Option<u64>,
        subscription_interruption: Option<Duration>,
    ) -> Result<WorkflowConfigVerified, anyhow::Error> {
        let retry_exp_backoff = Duration::from(self.retry_exp_backoff);
        if retry_exp_backoff == Duration::ZERO {
            bail!(
                "invalid `retry_exp_backoff` setting for workflow `{}` - duration must not be zero",
                self.common.name.0
            );
        }
        let (common, wasm_path) = self
            .common
            .fetch(&wasm_cache_dir, &metadata_dir, &path_prefixes)
            .await?;
        let wasm_path = WasmComponent::convert_core_module_to_component(
            &wasm_path,
            &common.content_digest,
            &wasm_cache_dir,
        )
        .await?
        .unwrap_or(wasm_path);
        let component_digest = self
            .component_digest
            .unwrap_or(ComponentDigest(common.content_digest.0));
        let component_id = ComponentId::new(
            ComponentType::Workflow,
            StrVariant::from(common.name),
            component_digest,
        )?;
        let workflow_config = WorkflowConfig {
            component_id: component_id.clone(),
            join_next_blocking_strategy: self.blocking_strategy.into(),
            backtrace_persist: global_backtrace_persist,
            stub_wasi: self.stub_wasi,
            fuel,
            lock_extension: self.lock_extension.then_some(self.exec.lock_expiry.into()),
            subscription_interruption,
        };
        let frame_files_to_sources = self.backtrace.into_frame_files();
        let retry_config = ComponentRetryConfig {
            max_retries: None,
            retry_exp_backoff,
        };
        Ok(WorkflowConfigVerified {
            wasm_path,
            workflow_config,
            exec_config: self.exec.into_exec_exec_config(
                component_id,
                global_executor_instance_limiter,
                retry_config,
            ),
            frame_files_to_sources,
            logs_store_min_level: self.logs_store_min_level.into(),
        })
    }
}

/// Canonical form of `WorkflowJsComponentConfigToml`.
#[derive(schemars::JsonSchema, Debug, Deserialize, Serialize, Clone)]
#[serde(deny_unknown_fields)]
pub(crate) struct WorkflowJsComponentConfigCanonical {
    pub(crate) name: ConfigName,
    pub(crate) location: JsLocationCanonical,
    pub(crate) content_digest: Option<ContentDigest>,
    pub(crate) component_digest: Option<ComponentDigest>,
    pub(crate) ffqn: FunctionFqn,
    pub(crate) params: Option<Vec<JsParamToml>>,
    pub(crate) exec: ExecConfigToml,
    pub(crate) retry_exp_backoff: DurationConfig,
    pub(crate) blocking_strategy: BlockingStrategyConfigToml,
    pub(crate) logs_store_min_level: LogLevelToml,
    pub(crate) return_type: Option<String>,
    pub(crate) lock_extension: bool,
}

impl WorkflowJsComponentConfigCanonical {
    #[instrument(skip_all, fields(component_name = self.name.0.as_ref()))]
    pub(crate) async fn fetch_and_verify(
        self,
        wasm_path: Arc<Path>,
        wasm_cache_dir: Arc<Path>,
        global_executor_instance_limiter: Option<Arc<tokio::sync::Semaphore>>,
    ) -> Result<WorkflowJsConfigVerified, anyhow::Error> {
        let parsed_params = match self.params {
            None => vec![],
            Some(params) => params
                .iter()
                .map(|p| {
                    let tw = val_json::type_wrapper::parse_wit_type(&p.wit_type)
                        .map_err(|e| anyhow!("invalid param type `{}`: {e}", p.wit_type))?;
                    Ok(concepts::ParameterType {
                        type_wrapper: tw,
                        name: StrVariant::from(p.name.clone()),
                        wit_type: StrVariant::from(p.wit_type.clone()),
                    })
                })
                .collect::<Result<Vec<_>, anyhow::Error>>()?,
        };
        let js_source = self
            .location
            .get_content(&wasm_cache_dir, self.content_digest.as_ref())
            .await?;
        const DEFAULT_RETURN_TYPE: &str = "result";
        let return_type_str = self.return_type.as_deref().unwrap_or(DEFAULT_RETURN_TYPE);
        let return_type_tw = val_json::type_wrapper::parse_wit_type(return_type_str)
            .map_err(|e| anyhow!("invalid return_type `{return_type_str}`: {e}"))?;
        let return_type = concepts::ReturnType::detect(
            return_type_tw,
            StrVariant::from(return_type_str.to_string()),
        );
        let return_type = match return_type {
            ReturnType::Extendable(rt) => rt,
            ReturnType::NonExtendable(_) => bail!(
                "return_type must be `result`, `result<T>`, `result<T, string>`, or \
                 `result<T, variant {{ execution-failed, ... }}>`, got `{return_type_str}`"
            ),
        };
        let component_digest = self.component_digest.unwrap_or_else(|| {
            let mut hasher = Sha256::new();
            hasher.update(b"workflow_js:");
            hasher.update(js_source.as_bytes());
            hasher.update(self.ffqn.to_string().as_bytes());
            for p in &parsed_params {
                hasher.update(p.wit_type.as_ref().as_bytes());
            }
            hasher.update(return_type.wit_type.as_bytes());
            let hash: [u8; 32] = hasher.finalize().into();
            ComponentDigest(Digest(hash))
        });
        let component_id = ComponentId::new(
            ComponentType::Workflow,
            StrVariant::from(self.name),
            component_digest,
        )?;
        let workflow_config = WorkflowConfig {
            component_id: component_id.clone(),
            join_next_blocking_strategy: self.blocking_strategy.into(),
            backtrace_persist: false,
            stub_wasi: false,
            fuel: None,
            lock_extension: self.lock_extension.then_some(self.exec.lock_expiry.into()),
            subscription_interruption: None,
        };
        let retry_config = ComponentRetryConfig {
            max_retries: None,
            retry_exp_backoff: self.retry_exp_backoff.into(),
        };
        let js_file_name = self.location.file_name();
        Ok(WorkflowJsConfigVerified {
            wasm_path,
            js_source,
            js_file_name,
            ffqn: self.ffqn,
            params: parsed_params,
            return_type,
            workflow_config,
            exec_config: self.exec.into_exec_exec_config(
                component_id,
                global_executor_instance_limiter,
                retry_config,
            ),
            logs_store_min_level: self.logs_store_min_level.into(),
        })
    }
}

/// Canonical deployment configuration — no local file paths.
/// Used for hash computation, wire format, and DB storage.
#[derive(Debug, Deserialize, Serialize, Default, Clone, schemars::JsonSchema)]
pub(crate) struct DeploymentCanonical {
    pub(crate) activities_wasm: Vec<ActivityWasmComponentConfigToml>,
    pub(crate) activities_stub: Vec<ActivityStubComponentConfigToml>,
    pub(crate) activities_external: Vec<ActivityExternalComponentConfigToml>,
    pub(crate) activities_js: Vec<ActivityJsComponentConfigCanonical>,
    pub(crate) workflows: Vec<WorkflowWasmComponentConfigCanonical>,
    pub(crate) workflows_js: Vec<WorkflowJsComponentConfigCanonical>,
    pub(crate) webhooks: Vec<webhook::WebhookWasmComponentConfigCanonical>,
    pub(crate) webhooks_js: Vec<webhook::WebhookJsComponentConfigCanonical>,
}

/// Resolve a `DeploymentToml` to `DeploymentCanonical` by reading all local JS and backtrace
/// source files. GitHub JS locations are left as references (downloaded lazily at verify time).
pub(crate) async fn resolve_local_refs_to_canonical(
    deployment: &DeploymentToml,
    path_prefixes: &PathPrefixes,
) -> anyhow::Result<DeploymentCanonical> {
    let mut activities_js = Vec::with_capacity(deployment.activities_js.len());
    for a in &deployment.activities_js {
        activities_js.push(ActivityJsComponentConfigCanonical {
            name: a.name.clone(),
            location: resolve_js_to_canonical(&a.location, path_prefixes).await?,
            content_digest: a.content_digest.clone(),
            component_digest: a.component_digest.clone(),
            ffqn: a.ffqn.clone(),
            params: a.params.clone(),
            exec: a.exec.clone(),
            max_retries: a.max_retries,
            retry_exp_backoff: a.retry_exp_backoff,
            forward_stdout: a.forward_stdout,
            forward_stderr: a.forward_stderr,
            logs_store_min_level: a.logs_store_min_level,
            env_vars: a.env_vars.clone(),
            allowed_hosts: a.allowed_hosts.clone(),
            return_type: a.return_type.clone(),
        });
    }

    let mut workflows = Vec::with_capacity(deployment.workflows.len());
    for w in &deployment.workflows {
        workflows.push(WorkflowWasmComponentConfigCanonical {
            common: w.common.clone(),
            component_digest: w.component_digest.clone(),
            exec: w.exec.clone(),
            retry_exp_backoff: w.retry_exp_backoff,
            blocking_strategy: w.blocking_strategy,
            backtrace: resolve_backtrace_to_canonical(&w.backtrace, path_prefixes).await,
            stub_wasi: w.stub_wasi,
            lock_extension: w.lock_extension,
            logs_store_min_level: w.logs_store_min_level,
        });
    }

    let mut workflows_js = Vec::with_capacity(deployment.workflows_js.len());
    for w in &deployment.workflows_js {
        workflows_js.push(WorkflowJsComponentConfigCanonical {
            name: w.name.clone(),
            location: resolve_js_to_canonical(&w.location, path_prefixes).await?,
            content_digest: w.content_digest.clone(),
            component_digest: w.component_digest.clone(),
            ffqn: w.ffqn.clone(),
            params: w.params.clone(),
            exec: w.exec.clone(),
            retry_exp_backoff: w.retry_exp_backoff,
            blocking_strategy: w.blocking_strategy,
            lock_extension: w.lock_extension,
            logs_store_min_level: w.logs_store_min_level,
            return_type: w.return_type.clone(),
        });
    }

    let mut webhooks = Vec::with_capacity(deployment.webhooks.len());
    for w in &deployment.webhooks {
        webhooks.push(webhook::WebhookWasmComponentConfigCanonical {
            common: w.common.clone(),
            http_server: w.http_server.clone(),
            routes: w.routes.clone(),
            forward_stdout: w.forward_stdout,
            forward_stderr: w.forward_stderr,
            env_vars: w.env_vars.clone(),
            backtrace: resolve_backtrace_to_canonical(&w.backtrace, path_prefixes).await,
            logs_store_min_level: w.logs_store_min_level,
            allowed_hosts: w.allowed_hosts.clone(),
        });
    }

    let mut webhooks_js = Vec::with_capacity(deployment.webhooks_js.len());
    for w in &deployment.webhooks_js {
        webhooks_js.push(webhook::WebhookJsComponentConfigCanonical {
            name: w.name.clone(),
            location: resolve_js_to_canonical(&w.location, path_prefixes).await?,
            content_digest: w.content_digest.clone(),
            http_server: w.http_server.clone(),
            routes: w.routes.clone(),
            forward_stdout: w.forward_stdout,
            forward_stderr: w.forward_stderr,
            logs_store_min_level: w.logs_store_min_level,
            env_vars: w.env_vars.clone(),
            allowed_hosts: w.allowed_hosts.clone(),
        });
    }

    Ok(DeploymentCanonical {
        activities_wasm: deployment.activities_wasm.clone(),
        activities_stub: deployment.activities_stub.clone(),
        activities_external: deployment.activities_external.clone(),
        activities_js,
        workflows,
        workflows_js,
        webhooks,
        webhooks_js,
    })
}

async fn resolve_js_to_canonical(
    location: &JsLocationToml,
    path_prefixes: &PathPrefixes,
) -> anyhow::Result<JsLocationCanonical> {
    match location {
        JsLocationToml::Path(path) => {
            let file_name = std::path::Path::new(path)
                .file_name()
                .and_then(|n| n.to_str())
                .unwrap_or(path)
                .to_string();
            let full_path =
                path_prefixes.deployment_config_replace_file_prefix_verify_exists(path)?;
            let content = tokio::fs::read_to_string(&full_path)
                .await
                .with_context(|| format!("cannot read JS file {full_path:?}"))?;
            Ok(JsLocationCanonical::Content { content, file_name })
        }
        JsLocationToml::GitHub(gh) => Ok(JsLocationCanonical::GitHub(gh.clone())),
    }
}

async fn resolve_backtrace_to_canonical(
    backtrace: &ComponentBacktraceConfig,
    path_prefixes: &PathPrefixes,
) -> ComponentBacktraceConfigCanonical {
    let mut frame_files_to_sources = HashMap::new();
    for (key, location) in &backtrace.frame_files_to_sources {
        let BacktraceSourceLocation::Path(path) = location;
        let content = async {
            let full_path = path_prefixes
                .deployment_config_replace_file_prefix_verify_exists(path)
                .inspect_err(|err| warn!("Ignoring missing backtrace source - {err:?}"))
                .ok()?;
            tokio::fs::read_to_string(&full_path)
                .await
                .inspect_err(|err| warn!("Cannot read backtrace source {full_path:?} - {err:?}"))
                .ok()
        }
        .await;
        if let Some(content) = content {
            frame_files_to_sources.insert(key.clone(), content);
        }
    }
    ComponentBacktraceConfigCanonical {
        frame_files_to_sources,
    }
}

#[derive(Debug, Deserialize, JsonSchema, Clone, Copy, Default)]
#[serde(rename_all = "snake_case")]
pub(crate) enum WasmtimeAllocatorConfig {
    #[default]
    Auto,
    OnDemand,
    Pooling,
}

#[derive(Debug, Deserialize, JsonSchema, Clone, Copy, Default)]
#[serde(deny_unknown_fields)]
pub(crate) struct WasmtimePoolingAllocatorConfig {
    /// How many bytes to keep resident between instantiations for the
    /// pooling allocator in linear memories.
    #[serde(default)]
    pooling_memory_keep_resident: Option<usize>,

    /// How many bytes to keep resident between instantiations for the
    /// pooling allocator in tables.
    #[serde(default)]
    pooling_table_keep_resident: Option<usize>,

    /// Enable memory protection keys for the pooling allocator; this can
    /// optimize the size of memory slots.
    #[serde(default)]
    memory_protection_keys: Option<bool>,

    /// The maximum number of WebAssembly instances which can be created
    /// with the pooling allocator.
    #[serde(default)]
    pooling_total_core_instances: Option<u32>,

    /// The maximum number of WebAssembly components which can be created
    /// with the pooling allocator.
    #[serde(default)]
    pooling_total_component_instances: Option<u32>,

    /// The maximum number of WebAssembly memories which can be created with
    /// the pooling allocator.
    #[serde(default)]
    pooling_total_memories: Option<u32>,

    /// The maximum number of WebAssembly tables which can be created with
    /// the pooling allocator.
    #[serde(default)]
    pooling_total_tables: Option<u32>,

    /// The maximum number of WebAssembly stacks which can be created with
    /// the pooling allocator.
    #[serde(default)]
    pooling_total_stacks: Option<u32>,

    /// The maximum runtime size of each linear memory in the pooling
    /// allocator, in bytes.
    #[serde(default)]
    pooling_max_memory_size: Option<usize>,
}

impl From<WasmtimePoolingAllocatorConfig> for wasm_workers::engines::PoolingOptions {
    fn from(value: WasmtimePoolingAllocatorConfig) -> wasm_workers::engines::PoolingOptions {
        wasm_workers::engines::PoolingOptions {
            pooling_memory_keep_resident: value.pooling_memory_keep_resident,
            pooling_table_keep_resident: value.pooling_table_keep_resident,
            memory_protection_keys: value.memory_protection_keys,
            pooling_total_core_instances: value.pooling_total_core_instances,
            pooling_total_component_instances: value.pooling_total_component_instances,
            pooling_total_memories: value.pooling_total_memories,
            pooling_total_tables: value.pooling_total_tables,
            pooling_total_stacks: value.pooling_total_stacks,
            pooling_max_memory_size: value.pooling_max_memory_size,
        }
    }
}

#[cfg(feature = "otlp")]
pub(crate) mod otlp {
    use super::{Deserialize, log};
    use log::EnvFilter;
    use schemars::JsonSchema;

    #[derive(Debug, Deserialize, JsonSchema, Clone)]
    #[serde(deny_unknown_fields)]
    pub(crate) struct OtlpConfig {
        pub(crate) enabled: bool,
        #[serde(default)]
        pub(crate) level: EnvFilter,
        #[serde(default = "default_service_name")]
        pub(crate) service_name: String,
        #[serde(default = "default_otlp_endpoint")]
        pub(crate) otlp_endpoint: String,
    }

    fn default_service_name() -> String {
        "obelisk-server".to_string()
    }

    fn default_otlp_endpoint() -> String {
        // Default port as per https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/protocol/exporter.md
        "http://localhost:4317".to_string()
    }
}

#[derive(Debug, Clone, Copy, Deserialize, Serialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub(crate) enum DurationConfig {
    Milliseconds(u64),
    Seconds(u64),
    Minutes(u64),
    Hours(u64),
}
impl From<DurationConfig> for Duration {
    fn from(value: DurationConfig) -> Self {
        match value {
            DurationConfig::Milliseconds(millis) => Duration::from_millis(millis),
            DurationConfig::Seconds(secs) => Duration::from_secs(secs),
            DurationConfig::Minutes(mins) => Duration::from_secs(mins * 60),
            DurationConfig::Hours(hrs) => Duration::from_secs(hrs * 60 * 60),
        }
    }
}

#[derive(Debug, Clone, Copy, Deserialize, Serialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub(crate) enum DurationConfigOptional {
    None,
    Milliseconds(u64),
    Seconds(u64),
    Minutes(u64),
    Hours(u64),
}
impl From<DurationConfigOptional> for Option<Duration> {
    fn from(value: DurationConfigOptional) -> Self {
        match value {
            DurationConfigOptional::None => None,
            DurationConfigOptional::Milliseconds(millis) => Some(Duration::from_millis(millis)),
            DurationConfigOptional::Seconds(secs) => Some(Duration::from_secs(secs)),
            DurationConfigOptional::Minutes(mins) => Some(Duration::from_secs(mins * 60)),
            DurationConfigOptional::Hours(hrs) => Some(Duration::from_secs(hrs * 60 * 60)),
        }
    }
}
pub(crate) mod log {
    use crate::config::toml::default_console_enabled;

    use super::{Deserialize, JsonSchema, default_console_style};
    use serde_with::serde_as;
    use std::str::FromStr;

    #[derive(Debug, Deserialize, JsonSchema, Default, Clone)]
    #[serde(deny_unknown_fields)]
    pub(crate) struct LoggingConfig {
        #[serde(default)]
        pub(crate) file: Option<AppenderRollingFile>,
        #[serde(default)]
        pub(crate) console: AppenderConsole,
    }

    #[derive(Debug, Deserialize, JsonSchema, Default, Copy, Clone)]
    #[serde(rename_all = "snake_case")]
    pub(crate) enum SpanConfig {
        /// spans are ignored (this is the default)
        #[default]
        None,
        /// one event when span is created
        New,
        /// one event per enter of a span
        Enter,
        /// one event per exit of a span
        Exit,
        /// one event when the span is dropped
        Close,
        /// one event per enter/exit of a span
        Active,
        /// events at all points (new, enter, exit, drop)
        Full,
    }

    impl From<SpanConfig> for tracing_subscriber::fmt::format::FmtSpan {
        fn from(value: SpanConfig) -> Self {
            match value {
                SpanConfig::None => Self::NONE,
                SpanConfig::New => Self::NEW,
                SpanConfig::Enter => Self::ENTER,
                SpanConfig::Exit => Self::EXIT,
                SpanConfig::Close => Self::CLOSE,
                SpanConfig::Active => Self::ACTIVE,
                SpanConfig::Full => Self::FULL,
            }
        }
    }

    #[derive(Debug, Deserialize, JsonSchema, Default, Clone)]
    #[serde(rename_all = "snake_case")]
    pub(crate) enum LoggingStyle {
        #[default]
        Plain,
        PlainCompact,
        Json,
    }

    #[serde_as]
    #[derive(Debug, Deserialize, JsonSchema, Default, Clone)]
    #[serde(deny_unknown_fields)]
    pub(crate) struct AppenderCommon {
        #[serde(default)]
        pub(crate) level: EnvFilter,
        #[serde(default)]
        pub(crate) span: SpanConfig,
        #[serde(default)]
        pub(crate) target: bool,
    }

    #[derive(Debug, serde_with::DeserializeFromStr, JsonSchema, Clone)]
    pub(crate) struct EnvFilter(
        #[schemars(with = "String")] pub(crate) tracing_subscriber::EnvFilter,
    );
    impl FromStr for EnvFilter {
        type Err = tracing_subscriber::filter::ParseError;

        fn from_str(directives: &str) -> Result<Self, Self::Err> {
            tracing_subscriber::EnvFilter::builder()
                .parse(directives)
                .map(Self)
        }
    }
    impl Default for EnvFilter {
        fn default() -> Self {
            Self::from_str("info,app=debug").expect("empty directive must not fail to parse")
        }
    }

    #[derive(Copy, Clone, Debug, Default, Deserialize, JsonSchema, PartialEq, Eq)]
    pub(crate) enum AppenderConsoleWriter {
        #[default]
        Stderr,
        Stdout,
    }

    #[derive(Debug, Deserialize, JsonSchema, Clone)]
    #[serde(deny_unknown_fields)]
    pub(crate) struct AppenderConsole {
        #[serde(default = "default_console_enabled")]
        pub(crate) enabled: bool,
        #[serde(flatten, default)]
        pub(crate) common: AppenderCommon,
        #[serde(default = "default_console_style")]
        pub(crate) style: LoggingStyle,
        #[serde(default)]
        pub(crate) writer: AppenderConsoleWriter,
    }
    impl Default for AppenderConsole {
        fn default() -> Self {
            Self {
                enabled: default_console_enabled(),
                common: AppenderCommon::default(),
                style: default_console_style(),
                writer: AppenderConsoleWriter::default(),
            }
        }
    }

    #[derive(Debug, Deserialize, JsonSchema, Clone)]
    #[serde(deny_unknown_fields)]
    pub(crate) struct AppenderRollingFile {
        pub(crate) enabled: bool,
        #[serde(flatten, default)]
        pub(crate) common: AppenderCommon,
        pub(crate) directory: String,
        pub(crate) prefix: String,
        #[serde(default)]
        pub(crate) rotation: Rotation,
        #[serde(default)]
        pub(crate) style: LoggingStyle,
    }

    #[derive(Debug, Deserialize, JsonSchema, Clone, Copy, Default)]
    #[serde(rename_all = "snake_case")]
    pub(crate) enum Rotation {
        Minutely,
        Hourly,
        Daily,
        #[default]
        Never,
    }
    impl From<Rotation> for tracing_appender::rolling::Rotation {
        fn from(value: Rotation) -> Self {
            match value {
                Rotation::Minutely => Self::MINUTELY,
                Rotation::Hourly => Self::HOURLY,
                Rotation::Daily => Self::DAILY,
                Rotation::Never => Self::NEVER,
            }
        }
    }
}

#[derive(Debug, Deserialize, Serialize, JsonSchema, Clone, Copy, Default)]
#[serde(rename_all = "snake_case")]
pub(crate) enum ComponentStdOutputToml {
    None,
    Stdout,
    Stderr,
    #[default]
    Db,
}
impl From<ComponentStdOutputToml> for Option<StdOutputConfig> {
    fn from(value: ComponentStdOutputToml) -> Self {
        match value {
            ComponentStdOutputToml::None => None,
            ComponentStdOutputToml::Stdout => Some(StdOutputConfig::Stdout),
            ComponentStdOutputToml::Stderr => Some(StdOutputConfig::Stderr),
            ComponentStdOutputToml::Db => Some(StdOutputConfig::Db),
        }
    }
}

pub(crate) mod webhook {
    use super::{
        AllowedHostToml, ComponentBacktraceConfig, ComponentBacktraceConfigCanonical,
        ComponentCommon, ComponentStdOutputToml, ConfigName, JsLocationCanonical, JsLocationToml,
        resolve_allowed_hosts, resolve_env_vars_plaintext, validate_no_env_collision,
    };
    use crate::{
        command::server::FrameFilesToSourceContent,
        config::{config_holder::PathPrefixes, env_var::EnvVarConfig, toml::LogLevelToml},
    };
    use anyhow::Context;
    use concepts::{
        ComponentId, ComponentType, ContentDigest, StrVariant,
        component_id::{ComponentDigest, Digest},
        storage::LogLevel,
    };
    use schemars::JsonSchema;
    use serde::{Deserialize, Serialize};
    use sha2::{Digest as _, Sha256};
    use std::{
        net::SocketAddr,
        path::{Path, PathBuf},
        sync::Arc,
        time::Duration,
    };
    use tracing::instrument;
    use wasm_workers::{
        envvar::EnvVar, http_request_policy::AllowedHostConfig, std_output_stream::StdOutputConfig,
    };

    #[derive(Debug, Deserialize, JsonSchema, Clone)]
    #[serde(deny_unknown_fields)]
    pub(crate) struct HttpServer {
        pub(crate) name: ConfigName,
        pub(crate) listening_addr: SocketAddr,
    }

    fn default_external_server_name() -> ConfigName {
        ConfigName::new(StrVariant::Static("external")).expect("valid name")
    }

    #[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
    #[serde(deny_unknown_fields)]
    pub(crate) struct WebhookWasmComponentConfigToml {
        #[serde(flatten)]
        pub(crate) common: ComponentCommon,
        #[serde(default = "default_external_server_name")]
        pub(crate) http_server: ConfigName,
        pub(crate) routes: Vec<WebhookRoute>,
        #[serde(default)]
        pub(crate) forward_stdout: ComponentStdOutputToml,
        #[serde(default)]
        pub(crate) forward_stderr: ComponentStdOutputToml,
        #[serde(default)]
        pub(crate) env_vars: Vec<EnvVarConfig>,
        #[serde(default)]
        pub(crate) backtrace: ComponentBacktraceConfig,
        #[serde(default)]
        pub(crate) logs_store_min_level: LogLevelToml,
        /// Allowed outgoing HTTP hosts with optional method restrictions and secrets.
        #[serde(default, rename = "allowed_host")]
        pub(crate) allowed_hosts: Vec<AllowedHostToml>,
    }

    #[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
    #[serde(untagged)]
    pub(crate) enum WebhookRoute {
        String(String),
        WebhookRouteDetail(WebhookRouteDetail),
    }

    impl Default for WebhookRoute {
        fn default() -> Self {
            WebhookRoute::String(String::new())
        }
    }

    #[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
    #[serde(deny_unknown_fields)]
    pub(crate) struct WebhookRouteDetail {
        // Empty means all methods.
        #[serde(default)]
        pub(crate) methods: Vec<String>,
        pub(crate) route: String,
    }

    #[derive(Debug)]
    pub(crate) struct WebhookWasmComponentConfigVerified {
        pub(crate) component_id: ComponentId,
        pub(crate) wasm_path: PathBuf,
        pub(crate) routes: Vec<WebhookRouteVerified>,
        pub(crate) forward_stdout: Option<StdOutputConfig>,
        pub(crate) forward_stderr: Option<StdOutputConfig>,
        pub(crate) env_vars: Arc<[EnvVar]>,
        pub(crate) frame_files_to_sources: FrameFilesToSourceContent,
        pub(crate) subscription_interruption: Option<Duration>,
        pub(crate) logs_store_min_level: Option<LogLevel>,
        pub(crate) allowed_hosts: Arc<[AllowedHostConfig]>,
    }

    #[derive(Debug)]
    pub(crate) struct WebhookRouteVerified {
        pub(crate) methods: Vec<http::Method>,
        pub(crate) route: String,
    }

    impl TryFrom<WebhookRoute> for WebhookRouteVerified {
        type Error = anyhow::Error;

        fn try_from(value: WebhookRoute) -> Result<Self, Self::Error> {
            Ok(match value {
                WebhookRoute::String(route) => Self {
                    methods: Vec::new(),
                    route,
                },
                WebhookRoute::WebhookRouteDetail(WebhookRouteDetail { methods, route }) => {
                    let methods = methods
                        .into_iter()
                        .map(|method| {
                            http::Method::from_bytes(method.as_bytes())
                                .with_context(|| format!("cannot parse route method `{method}`",))
                        })
                        .collect::<Result<Vec<_>, _>>()?;
                    Self { methods, route }
                }
            })
        }
    }

    #[derive(Debug, Deserialize, Serialize, JsonSchema, Clone)]
    #[serde(deny_unknown_fields)]
    pub(crate) struct WebhookJsComponentConfigToml {
        pub(crate) name: ConfigName,
        /// Location of the JavaScript source file.
        /// Supports local file paths and `gh://` GitHub release references.
        pub(crate) location: JsLocationToml,
        /// Content digest of the JS source file.
        #[serde(default)]
        #[schemars(with = "Option<String>")]
        pub(crate) content_digest: Option<ContentDigest>,
        /// The HTTP server to bind this webhook to.
        #[serde(default = "default_external_server_name")]
        pub(crate) http_server: ConfigName,
        /// Routes that this webhook responds to.
        pub(crate) routes: Vec<WebhookRoute>,
        #[serde(default)]
        pub(crate) forward_stdout: ComponentStdOutputToml,
        #[serde(default)]
        pub(crate) forward_stderr: ComponentStdOutputToml,
        #[serde(default)]
        pub(crate) logs_store_min_level: LogLevelToml,
        #[serde(default)]
        pub(crate) env_vars: Vec<EnvVarConfig>,
        /// Allowed outgoing HTTP hosts with optional method restrictions and secrets.
        #[serde(default, rename = "allowed_host")]
        pub(crate) allowed_hosts: Vec<AllowedHostToml>,
    }

    #[derive(Debug)]
    pub(crate) struct WebhookJsConfigVerified {
        pub(crate) wasm_path: Arc<Path>,
        pub(crate) component_id: ComponentId,
        pub(crate) js_source: String,
        pub(crate) js_file_name: String,
        pub(crate) routes: Vec<WebhookRouteVerified>,
        pub(crate) forward_stdout: Option<StdOutputConfig>,
        pub(crate) forward_stderr: Option<StdOutputConfig>,
        pub(crate) env_vars: Arc<[EnvVar]>,
        pub(crate) logs_store_min_level: Option<LogLevel>,
        pub(crate) allowed_hosts: Arc<[AllowedHostConfig]>,
    }

    impl WebhookJsConfigVerified {
        pub(crate) fn as_frame_sources(&self) -> FrameFilesToSourceContent {
            FrameFilesToSourceContent::from([(self.js_file_name.clone(), self.js_source.clone())])
        }
    }

    /// Canonical form of `WebhookWasmComponentConfigToml`.
    #[derive(Debug, Deserialize, Serialize, Clone, schemars::JsonSchema)]
    #[serde(deny_unknown_fields)]
    pub(crate) struct WebhookWasmComponentConfigCanonical {
        #[serde(flatten)]
        pub(crate) common: ComponentCommon,
        #[serde(default = "default_external_server_name")]
        pub(crate) http_server: ConfigName,
        pub(crate) routes: Vec<WebhookRoute>,
        #[serde(default)]
        pub(crate) forward_stdout: ComponentStdOutputToml,
        #[serde(default)]
        pub(crate) forward_stderr: ComponentStdOutputToml,
        #[serde(default)]
        pub(crate) env_vars: Vec<EnvVarConfig>,
        #[serde(default)]
        pub(crate) backtrace: ComponentBacktraceConfigCanonical,
        #[serde(default)]
        pub(crate) logs_store_min_level: LogLevelToml,
        #[serde(default, rename = "allowed_host")]
        pub(crate) allowed_hosts: Vec<AllowedHostToml>,
    }

    impl WebhookWasmComponentConfigCanonical {
        #[instrument(skip_all, fields(component_name = self.common.name.0.as_ref()), err)]
        pub(crate) async fn fetch_and_verify(
            self,
            wasm_cache_dir: Arc<Path>,
            metadata_dir: Arc<Path>,
            ignore_missing_env_vars: bool,
            path_prefixes: Arc<PathPrefixes>,
            subscription_interruption: Option<Duration>,
        ) -> Result<(ConfigName, WebhookWasmComponentConfigVerified), anyhow::Error> {
            let (common, wasm_path) = self
                .common
                .fetch(&wasm_cache_dir, &metadata_dir, &path_prefixes)
                .await?;
            let frame_files_to_sources = self.backtrace.into_frame_files();
            let component_id = ComponentId::new(
                ComponentType::WebhookEndpoint,
                StrVariant::from(common.name.clone()),
                ComponentDigest(common.content_digest.0),
            )?;
            let env_vars = resolve_env_vars_plaintext(self.env_vars, ignore_missing_env_vars)?;
            let allowed_hosts = resolve_allowed_hosts(self.allowed_hosts, ignore_missing_env_vars)?;
            validate_no_env_collision(&env_vars, &allowed_hosts)?;
            Ok((
                common.name,
                WebhookWasmComponentConfigVerified {
                    component_id,
                    wasm_path,
                    routes: self
                        .routes
                        .into_iter()
                        .map(WebhookRouteVerified::try_from)
                        .collect::<Result<Vec<_>, _>>()?,
                    forward_stdout: self.forward_stdout.into(),
                    forward_stderr: self.forward_stderr.into(),
                    env_vars,
                    frame_files_to_sources,
                    subscription_interruption,
                    logs_store_min_level: self.logs_store_min_level.into(),
                    allowed_hosts,
                },
            ))
        }
    }

    /// Canonical form of `WebhookJsComponentConfigToml`.
    #[derive(Debug, Deserialize, Serialize, Clone, schemars::JsonSchema)]
    #[serde(deny_unknown_fields)]
    pub(crate) struct WebhookJsComponentConfigCanonical {
        pub(crate) name: ConfigName,
        pub(crate) location: JsLocationCanonical,
        #[serde(default)]
        pub(crate) content_digest: Option<ContentDigest>,
        #[serde(default = "default_external_server_name")]
        pub(crate) http_server: ConfigName,
        pub(crate) routes: Vec<WebhookRoute>,
        #[serde(default)]
        pub(crate) forward_stdout: ComponentStdOutputToml,
        #[serde(default)]
        pub(crate) forward_stderr: ComponentStdOutputToml,
        #[serde(default)]
        pub(crate) logs_store_min_level: LogLevelToml,
        #[serde(default)]
        pub(crate) env_vars: Vec<EnvVarConfig>,
        #[serde(default, rename = "allowed_host")]
        pub(crate) allowed_hosts: Vec<AllowedHostToml>,
    }

    impl WebhookJsComponentConfigCanonical {
        #[instrument(skip_all, fields(component_name = self.name.0.as_ref()))]
        pub(crate) async fn fetch_and_verify(
            self,
            wasm_path: Arc<Path>,
            wasm_cache_dir: Arc<Path>,
            ignore_missing_env_vars: bool,
        ) -> Result<(ConfigName, WebhookJsConfigVerified), anyhow::Error> {
            let js_source = self
                .location
                .get_content(&wasm_cache_dir, self.content_digest.as_ref())
                .await?;
            let mut hasher = Sha256::new();
            hasher.update(b"webhook_js:");
            hasher.update(js_source.as_bytes());
            let hash: [u8; 32] = hasher.finalize().into();
            let component_id = ComponentId::new(
                ComponentType::WebhookEndpoint,
                StrVariant::from(self.name.clone()),
                ComponentDigest(Digest(hash)),
            )?;
            let env_vars = resolve_env_vars_plaintext(self.env_vars, ignore_missing_env_vars)?;
            let allowed_hosts = resolve_allowed_hosts(self.allowed_hosts, ignore_missing_env_vars)?;
            validate_no_env_collision(&env_vars, &allowed_hosts)?;
            Ok((
                self.name,
                WebhookJsConfigVerified {
                    wasm_path,
                    component_id,
                    js_file_name: self.location.file_name(),
                    js_source,
                    routes: self
                        .routes
                        .into_iter()
                        .map(WebhookRouteVerified::try_from)
                        .collect::<Result<Vec<_>, _>>()?,
                    forward_stdout: self.forward_stdout.into(),
                    forward_stderr: self.forward_stderr.into(),
                    env_vars,
                    logs_store_min_level: self.logs_store_min_level.into(),
                    allowed_hosts,
                },
            ))
        }
    }
}

#[derive(Debug, Deserialize, JsonSchema, Clone, Copy)]
#[serde(untagged)]
pub(crate) enum ValueOrUnlimited<T> {
    Unlimited(Unlimited),
    Some(T),
}
impl<T> Default for ValueOrUnlimited<T> {
    fn default() -> Self {
        Self::Unlimited(Unlimited::Unlimited)
    }
}
impl<T> From<ValueOrUnlimited<T>> for Option<T> {
    fn from(value: ValueOrUnlimited<T>) -> Self {
        match value {
            ValueOrUnlimited::Some(val) => Some(val),
            ValueOrUnlimited::Unlimited(Unlimited::Unlimited) => None,
        }
    }
}

// TODO: Unify with ValueOrUnlimited
#[derive(Debug, Deserialize, JsonSchema, Clone, Copy)]
#[serde(untagged)]
pub(crate) enum InflightSemaphore {
    Unlimited(Unlimited),
    Some(u32),
}
impl Default for InflightSemaphore {
    fn default() -> Self {
        Self::Unlimited(Unlimited::Unlimited)
    }
}

#[derive(Debug, Default, Deserialize, JsonSchema, Clone, Copy)]
#[serde(rename_all = "snake_case")]
pub(crate) enum Unlimited {
    #[default]
    Unlimited,
}

impl InflightSemaphore {
    pub(crate) fn as_semaphore(&self) -> Option<Arc<tokio::sync::Semaphore>> {
        match self {
            InflightSemaphore::Unlimited(_) => None,
            InflightSemaphore::Some(permits) => Some(Arc::new(tokio::sync::Semaphore::new(
                usize::try_from(*permits).expect("usize >= u32"),
            ))),
        }
    }
}

// TODO: Move to env_var module
fn resolve_env_vars_plaintext(
    env_vars: Vec<EnvVarConfig>,
    ignore_missing: bool,
) -> Result<Arc<[EnvVar]>, EnvVarMissing> {
    env_vars
        .into_iter()
        .map(|env_var| match env_var {
            EnvVarConfig::KeyValue { key, value } => Ok(EnvVar {
                key,
                val: interpolate_env_vars_plaintext(&value)?,
            }),
            EnvVarConfig::Key(key) => match std::env::var(&key) {
                Ok(val) => Ok(EnvVar { key, val }),
                Err(_err) => {
                    if ignore_missing {
                        Ok(EnvVar {
                            key,
                            val: String::new(),
                        })
                    } else {
                        Err(EnvVarMissing(key))
                    }
                }
            },
        })
        .collect::<Result<_, _>>()
}

#[derive(Debug, thiserror::Error)]
enum ResolveAllowedHostsError {
    #[error(transparent)]
    HostPattern(#[from] HostPatternError),
    #[error(transparent)]
    EnvVarsMissing(#[from] EnvVarsMissing),
    #[error("cannot parse HTTP method `{0}`")]
    InvalidMethod(String),
    #[error("use `methods = \"*\"` to allow all methods, not `methods = [\"*\"]`")]
    InvalidMethodStar,
}

fn resolve_allowed_hosts(
    entries: Vec<AllowedHostToml>,
    ignore_missing_env_vars: bool,
) -> Result<Arc<[AllowedHostConfig]>, ResolveAllowedHostsError> {
    entries
        .into_iter()
        .filter_map(|entry| {
            // Convert MethodsInput to MethodsPattern
            let methods = match entry.methods {
                None => {
                    // Omitted methods: nothing allowed, warn and skip
                    warn!(
                        "allowed_host `{}` has no `methods` field - no requests will be allowed; \
                         use `methods = \"*\"` to allow all methods",
                        entry.pattern
                    );
                    return None;
                }
                Some(MethodsInput::Star(_)) => {
                    // `methods = "*"` - all methods allowed
                    MethodsPattern::AllMethods
                }
                Some(MethodsInput::List(list)) => {
                    if list.is_empty() {
                        // Empty list: nothing allowed, warn and skip
                        warn!(
                            "allowed_host `{}` has empty `methods = []` - no requests will be allowed",
                            entry.pattern
                        );
                        return None;
                    }
                    // Parse specific methods
                    match list
                        .into_iter()
                        .map(|m| {
                            http::Method::from_bytes(m.as_bytes()).map_err(|_| {
                                if m == "*" {
                                    ResolveAllowedHostsError::InvalidMethodStar
                                } else {
                                    ResolveAllowedHostsError::InvalidMethod(m)
                                }
                            })
                        })
                        .collect::<Result<Vec<_>, _>>()
                    {
                        Ok(methods) => MethodsPattern::Specific(methods),
                        Err(e) => return Some(Err(e)),
                    }
                }
            };

            let pattern_str = match interpolate_env_vars_plaintext(&entry.pattern) {
                Ok(s) => s,
                Err(EnvVarMissing(var)) => {
                    if ignore_missing_env_vars {
                        warn!(
                            "allowed_host pattern `{}` references missing env var `{var}`, skipping",
                            entry.pattern
                        );
                        return None;
                    }
                    return Some(Err(ResolveAllowedHostsError::EnvVarsMissing(
                        EnvVarsMissing(vec![var]),
                    )));
                }
            };
            let pattern = match HostPattern::parse_with_methods(&pattern_str, methods) {
                Ok(p) => p,
                Err(e) => return Some(Err(e.into())),
            };

            let (secret_env_mappings, replace_in) = if let Some(secrets) = entry.secrets {
                if secrets.env_vars.is_empty() {
                    warn!(
                        "allowed_host `{}` has empty `secrets.env_vars`",
                        entry.pattern
                    );
                }
                if secrets.replace_in.is_empty() {
                    warn!(
                        "allowed_host `{}` has empty `secrets.replace_in` - secrets will never be injected",
                        entry.pattern
                    );
                }
                if pattern.scheme.allows_unencrypted() {
                    warn!("secrets allowed for potentially unencrypted host `{pattern}`");
                }

                let env_mappings =
                    match resolve_secret_env_vars(secrets.env_vars, ignore_missing_env_vars) {
                        Ok(m) => m,
                        Err(e) => return Some(Err(e)),
                    };
                let replace_in = secrets
                    .replace_in
                    .into_iter()
                    .map(|r| match r {
                        ReplaceIn::Headers => ReplacementLocation::Headers,
                        ReplaceIn::Body => ReplacementLocation::Body,
                        ReplaceIn::Params => ReplacementLocation::Params,
                    })
                    .collect();
                (env_mappings, replace_in)
            } else {
                (Vec::new(), hashbrown::HashSet::new())
            };

            Some(Ok(AllowedHostConfig {
                pattern,
                secret_env_mappings,
                replace_in,
            }))
        })
        .collect::<Result<_, _>>()
}

fn resolve_secret_env_vars(
    env_vars: Vec<EnvVarConfig>,
    ignore_missing: bool,
) -> Result<Vec<(String, SecretString)>, ResolveAllowedHostsError> {
    let mut missing = vec![];
    let mut env_mappings = Vec::new();
    for env_var in env_vars {
        match env_var {
            EnvVarConfig::KeyValue { key, value } => match interpolate_env_vars_secret(&value) {
                Ok(real_value) => env_mappings.push((key, real_value)),
                Err(err) => missing.push(err.0),
            },
            EnvVarConfig::Key(key) => match std::env::var(&key) {
                Ok(val) => env_mappings.push((key, SecretString::from(val))),
                Err(_err) => missing.push(key),
            },
        }
    }
    if !missing.is_empty() && !ignore_missing {
        return Err(EnvVarsMissing(missing).into());
    }
    Ok(env_mappings)
}

fn validate_no_env_collision(
    env_vars: &[EnvVar],
    allowed_hosts: &[AllowedHostConfig],
) -> Result<(), anyhow::Error> {
    let env_var_keys: hashbrown::HashSet<_> = env_vars.iter().map(|e| e.key.as_str()).collect();
    for host in allowed_hosts {
        for (key, _) in &host.secret_env_mappings {
            ensure!(
                !env_var_keys.contains(key.as_str()),
                "secret env var `{key}` collides with an `env_vars` entry"
            );
        }
    }
    Ok(())
}

const fn default_true() -> bool {
    true
}
const fn default_parallel_compilation() -> bool {
    true
}
const fn default_debug() -> bool {
    false
}
const fn default_global_backtrace_persist() -> bool {
    true
}

const fn default_codegen_enabled() -> bool {
    true
}

const fn default_max_retries() -> u32 {
    5
}

const fn default_retry_exp_backoff() -> DurationConfig {
    DurationConfig::Milliseconds(100)
}

const fn default_non_blocking_event_batching() -> u32 {
    DEFAULT_NON_BLOCKING_EVENT_BATCHING
}

const fn default_batch_size() -> u32 {
    5
}

const fn default_lock_expiry() -> DurationConfig {
    DurationConfig::Seconds(1)
}

const fn default_tick_sleep() -> DurationConfig {
    DurationConfig::Milliseconds(200)
}

const fn default_lock_extension() -> bool {
    true
}

const fn default_subscription_interruption() -> DurationConfigOptional {
    DurationConfigOptional::Seconds(1)
}
fn default_console_enabled() -> bool {
    true
}
fn default_console_style() -> LoggingStyle {
    LoggingStyle::PlainCompact
}

fn default_sqlite_queue_capacity() -> usize {
    SqliteConfig::default().queue_capacity
}
fn default_workflows_lock_extension_leeway() -> DurationConfig {
    DurationConfig::Milliseconds(100)
}
fn default_activities_directories_enabled() -> bool {
    false
}

fn default_activities_directories_parent_directory() -> String {
    "${TEMP_DIR}/obelisk".to_string()
}

fn default_dir_cleanup_run_every() -> DurationConfig {
    DurationConfig::Minutes(1)
}
fn default_dir_cleanup_older_than() -> DurationConfig {
    DurationConfig::Minutes(5)
}
fn default_dir_cleanup_enabled() -> bool {
    true
}

fn default_timers_watcher_enabled() -> bool {
    true
}
fn default_timers_watcher_leeway() -> DurationConfig {
    DurationConfig::Milliseconds(500)
}
fn default_timers_watcher_tick_sleep() -> DurationConfig {
    DurationConfig::Milliseconds(100)
}

fn default_cancel_watcher_enabled() -> bool {
    true
}
fn default_cancel_watcher_tick_sleep() -> DurationConfig {
    DurationConfig::Seconds(1)
}

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

        // Helper struct to deserialize into
        #[derive(Deserialize, Debug, PartialEq)]
        struct TestConfig {
            strategy: BlockingStrategyConfigToml,
        }

        #[test]
        fn deserialize_simple_interrupt() {
            let toml_str = r#"
strategy = "interrupt"
"#;
            let expected = TestConfig {
                strategy: BlockingStrategyConfigToml::Simple(
                    BlockingStrategyConfigSimple::Interrupt,
                ),
            };
            let actual: TestConfig =
                toml::from_str(toml_str).expect("Should parse interrupt string");
            assert_eq!(actual, expected);

            // Verify From impl result
            assert_eq!(
                JoinNextBlockingStrategy::from(actual.strategy),
                JoinNextBlockingStrategy::Interrupt
            );
        }

        #[test]
        fn deserialize_simple_await() {
            let toml_str = r#"
strategy = "await"
"#;
            let expected = TestConfig {
                strategy: BlockingStrategyConfigToml::Simple(
                    BlockingStrategyConfigSimple::Await, // The default variant of Simple
                ),
            };
            let actual: TestConfig = toml::from_str(toml_str).expect("Should parse await string");
            assert_eq!(actual, expected);

            // Verify From impl result (uses default batching)
            assert_eq!(
                JoinNextBlockingStrategy::from(actual.strategy),
                JoinNextBlockingStrategy::Await {
                    non_blocking_event_batching: DEFAULT_NON_BLOCKING_EVENT_BATCHING
                }
            );
        }

        #[test]
        fn deserialize_tagged_await_default_batching() {
            let toml_str = r#"
strategy = { kind = "await" }
"#;
            let expected = TestConfig {
                strategy: BlockingStrategyConfigToml::Tagged(
                    BlockingStrategyConfigCustomized::Await(BlockingStrategyAwaitConfig {
                        non_blocking_event_batching: default_non_blocking_event_batching(),
                    }),
                ),
            };
            let actual: TestConfig =
                toml::from_str(toml_str).expect("Should parse tagged await with default batching");
            assert_eq!(actual, expected);

            // Verify From impl result (uses default batching)
            assert_eq!(
                JoinNextBlockingStrategy::from(actual.strategy),
                JoinNextBlockingStrategy::Await {
                    non_blocking_event_batching: DEFAULT_NON_BLOCKING_EVENT_BATCHING
                }
            );
        }

        #[test]
        fn deserialize_tagged_await_custom_batching() {
            let toml_str = r#"
strategy = { kind = "await", non_blocking_event_batching = 99 }
"#;
            let expected = TestConfig {
                strategy: BlockingStrategyConfigToml::Tagged(
                    BlockingStrategyConfigCustomized::Await(BlockingStrategyAwaitConfig {
                        non_blocking_event_batching: 99,
                    }),
                ),
            };
            let actual: TestConfig =
                toml::from_str(toml_str).expect("Should parse tagged await with custom batching");
            assert_eq!(actual, expected);

            // Verify From impl result (uses custom batching)
            assert_eq!(
                JoinNextBlockingStrategy::from(actual.strategy),
                JoinNextBlockingStrategy::Await {
                    non_blocking_event_batching: 99
                }
            );
        }

        #[test]
        fn deserialize_invalid_string_should_fail() {
            let toml_str = r#"
strategy = "unknown"
"#;
            let result = toml::from_str::<TestConfig>(toml_str);
            assert!(result.is_err(), "Should fail on unknown string");
            // Check for a more specific error if needed, e.g., contains "unknown variant"
        }

        #[test]
        fn deserialize_invalid_kind_in_tagged_should_fail() {
            let toml_str = r#"
strategy = { kind = "interrupt", non_blocking_event_batching = 10 }
"#;
            let result = toml::from_str::<TestConfig>(toml_str);
            assert!(result.is_err(), "Should fail on invalid kind in map");
        }

        #[test]
        fn deserialize_invalid_structure_missing_kind_should_fail() {
            let toml_str = r#"
strategy = { name = "await", non_blocking_event_batching = 10 } # Missing 'kind'
"#;
            let result = toml::from_str::<TestConfig>(toml_str);
            // Fails `Tagged` because 'kind' is missing. Fails `Simple` because it's not a string.
            assert!(result.is_err(), "Should fail on map missing 'kind'");
        }

        #[test]
        fn deserialize_invalid_type_should_fail() {
            let toml_str = r"
strategy = 123
";
            let result = toml::from_str::<TestConfig>(toml_str);
            // Fails `Tagged` because not a map. Fails `Simple` because not a string.
            assert!(result.is_err(), "Should fail on incorrect type (integer)");
        }

        #[test]
        fn deserialize_tagged_await_with_extra_field_should_fail() {
            // TOML allows extra fields by default, Serde ignores them if not in the struct
            let toml_str = r#"
strategy = { kind = "await", non_blocking_event_batching = 25, extra_stuff = "hello" }
"#;
            let result = toml::from_str::<TestConfig>(toml_str);
            assert!(result.is_err(), "Should fail on `extra_stuff`");
        }
    }

    mod component_location {
        use super::super::*;
        use crate::github::GitHubReleaseTag;

        #[test]
        fn parse_local_path() {
            let location: ComponentLocationToml = "./my-component.wasm".parse().unwrap();
            assert!(
                matches!(location, ComponentLocationToml::Path(p) if p == "./my-component.wasm")
            );
        }

        #[test]
        fn parse_oci_reference() {
            let location: ComponentLocationToml =
                "oci://ghcr.io/obeli-sk/obelisk:v0.34.1".parse().unwrap();
            assert!(matches!(location, ComponentLocationToml::Oci(_)));
        }

        #[test]
        fn parse_github_release_specific_tag() {
            let location: ComponentLocationToml = "gh://obeli-sk/obelisk@v0.34.1/my-component.wasm"
                .parse()
                .unwrap();
            match location {
                ComponentLocationToml::GitHub(ref gh_ref) => {
                    assert_eq!(gh_ref.owner, "obeli-sk");
                    assert_eq!(gh_ref.repo, "obelisk");
                    assert_eq!(
                        gh_ref.tag,
                        GitHubReleaseTag::Specific("v0.34.1".to_string())
                    );
                    assert_eq!(gh_ref.asset_name, "my-component.wasm");
                }
                _ => panic!("expected GitHub variant"),
            }
        }

        #[test]
        fn parse_github_release_latest() {
            let location: ComponentLocationToml = "gh://obeli-sk/obelisk@latest/my-component.wasm"
                .parse()
                .unwrap();
            match location {
                ComponentLocationToml::GitHub(ref gh_ref) => {
                    assert_eq!(gh_ref.tag, GitHubReleaseTag::Latest);
                }
                _ => panic!("expected GitHub variant"),
            }
        }

        #[test]
        fn parse_github_release_invalid() {
            let result: Result<ComponentLocationToml, _> = "gh://invalid-format".parse();
            assert!(result.is_err());
        }

        #[test]
        fn deserialize_component_common_with_content_digest() {
            let toml_str = r#"
name = "my_component"
location = "gh://owner/repo@v1.0.0/component.wasm"
content_digest = "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
"#;
            let common: ComponentCommon = toml::from_str(toml_str).unwrap();
            assert!(common.content_digest.is_some());
        }

        #[test]
        fn deserialize_component_common_without_content_digest() {
            let toml_str = r#"
name = "my_component"
location = "gh://owner/repo@v1.0.0/component.wasm"
"#;
            let common: ComponentCommon = toml::from_str(toml_str).unwrap();
            assert!(common.content_digest.is_none());
        }
    }

    mod activity_stub {
        use super::super::*;

        #[test]
        fn deserialize_file_mode() {
            let toml_str = r#"
name = "my_stub"
location = "./stub.wasm"
"#;
            let stub: ActivityStubComponentConfigToml = toml::from_str(toml_str).unwrap();
            assert!(matches!(stub, ActivityStubComponentConfigToml::File(_)));
        }

        #[test]
        fn deserialize_inline_mode() {
            let toml_str = r#"
name = "my_stub"
ffqn = "ns:pkg/ifc.fn"
params = [{ name = "id", type = "u64" }]
return_type = "result<string, string>"
"#;
            let stub: ActivityStubComponentConfigToml = toml::from_str(toml_str).unwrap();
            assert!(matches!(stub, ActivityStubComponentConfigToml::Inline(_)));
        }

        #[test]
        fn reject_both_location_and_ffqn() {
            let toml_str = r#"
name = "my_stub"
location = "./stub.wasm"
ffqn = "ns:pkg/ifc.fn"
"#;
            toml::from_str::<ActivityStubComponentConfigToml>(toml_str).unwrap_err();
        }

        #[test]
        fn reject_neither_location_nor_ffqn() {
            let toml_str = r#"
name = "my_stub"
"#;
            toml::from_str::<ActivityStubComponentConfigToml>(toml_str).unwrap_err();
        }
    }
}