ch-rds 0.2.1

#![doc = include_str!("../README.md")]
#![forbid(unsafe_code)]
#![warn(missing_docs)]

/// Background-job scheduling and execution primitives.
pub mod cronjob;
/// Development commands for building and running an application workspace.
#[cfg(feature = "dev")]
pub mod dev;
mod docker;
mod project;

use std::fs;
use std::future::Future;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use crate::docker::{
    ensure_local_spacetime_module, ensure_local_stack, is_local_database,
    should_bootstrap_local_stack,
};
use anyhow::{Context as _, Result, anyhow};
use ch_config::config::AppConfig;
use ch_database::mariadb::url::redact_mysql_url;

pub use project::Project;

/// Selects whether the runtime may manage local development services.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RuntimeEnvironment {
    /// Resolve configured resources and manage loopback services when needed.
    Local,
    /// Use configured resources without managing local services or modules.
    Production,
}

impl RuntimeEnvironment {
    fn allows_local_resources(self) -> bool {
        matches!(self, Self::Local)
    }
}

/// Failures while discovering a project or loading its resources.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum RuntimeError {
    /// A filesystem or process-environment operation failed during discovery.
    #[error("failed to discover the runtime project: {0}")]
    Io(#[from] std::io::Error),
    /// Application environment settings could not be loaded.
    #[error(transparent)]
    Configuration(#[from] ch_config::config::ConfigError),
    /// No application root could be found from the provided path.
    #[error("failed to discover an application root from {0}")]
    ProjectNotFound(PathBuf),
    /// A resource could not be resolved or prepared.
    #[error("failed to load {resource}: {source}")]
    Resource {
        /// Resource or discovery operation that failed.
        resource: &'static str,
        /// Underlying failure, retained for inspection through `Error::source`.
        #[source]
        source: Box<dyn std::error::Error + Send + Sync>,
    },
}

/// Loads one resource using the selected runtime policy and project.
///
/// Implementations return a concrete output and a `Send` future without boxing.
pub trait Resource: Send {
    /// Value returned after the resource has been prepared.
    type Output: Send;

    /// Resolves and prepares this resource for the runtime.
    fn load(
        self,
        runtime: &Runtime,
    ) -> impl Future<Output = std::result::Result<Self::Output, RuntimeError>> + Send;
}

/// Discovers application settings and prepares runtime resources.
#[derive(Debug)]
pub struct Runtime {
    environment: RuntimeEnvironment,
    project: Project,
    environment_load: ch_config::environment::EnvLoad,
}

impl Runtime {
    /// Discovers the project and loads its conventional environment files.
    pub fn bootstrap() -> std::result::Result<Self, RuntimeError> {
        Self::bootstrap_with_environment(RuntimeEnvironment::Local)
    }

    /// Bootstraps resources with an explicit local-resource policy.
    pub fn bootstrap_with_environment(
        environment: RuntimeEnvironment,
    ) -> std::result::Result<Self, RuntimeError> {
        let project = Project::discover()?;
        let environment_load = project.load_environment()?;
        Ok(Self {
            environment,
            project,
            environment_load,
        })
    }

    /// Returns the resource policy selected for this runtime.
    pub fn environment(&self) -> RuntimeEnvironment {
        self.environment
    }

    /// Returns the discovered application layout.
    pub fn project(&self) -> &Project {
        &self.project
    }

    /// Reports which conventional dotenv files were loaded during bootstrap.
    pub fn environment_load(&self) -> ch_config::environment::EnvLoad {
        self.environment_load
    }

    /// Resolves and prepares one resource through its owning implementation.
    pub async fn load<R: Resource>(
        &self,
        resource: R,
    ) -> std::result::Result<R::Output, RuntimeError> {
        resource.load(self).await
    }
}

/// Settings for the bundled local Docker Compose stack.
#[derive(Debug, Clone)]
pub struct LocalStack {
    project_name: String,
    env_path: PathBuf,
    spacetime_module: Option<LocalSpacetimeModule>,
}

#[derive(Debug, Clone)]
struct LocalSpacetimeModule {
    name: String,
    path: PathBuf,
}

impl LocalStack {
    /// Creates local-stack settings from an explicit project name and env file.
    pub fn new(project_name: impl Into<String>, env_path: impl Into<PathBuf>) -> Self {
        Self {
            project_name: project_name.into(),
            env_path: env_path.into(),
            spacetime_module: None,
        }
    }

    /// Discovers the conventional local-stack settings from the current directory.
    pub fn discover() -> std::result::Result<Self, RuntimeError> {
        Self::discover_from(std::env::current_dir()?)
    }

    /// Discovers a project from an ancestor containing `services.env`.
    pub fn discover_from(start: impl AsRef<Path>) -> std::result::Result<Self, RuntimeError> {
        discover_local_stack(start.as_ref()).map_err(|source| RuntimeError::Resource {
            resource: "local stack",
            source: source.into_boxed_dyn_error(),
        })
    }

    /// Returns the caller-owned Docker Compose project name.
    pub fn project_name(&self) -> &str {
        &self.project_name
    }

    /// Returns the caller-owned Docker Compose env-file path.
    pub fn env_path(&self) -> &Path {
        &self.env_path
    }
}

fn discover_local_stack(start: &Path) -> Result<LocalStack> {
    let start = project::discovery_root(start);

    for project_root in start.ancestors() {
        let env_path = project_root.join("services.env");
        if !env_path.is_file() {
            continue;
        }

        let app_manifest = project_root.join("app").join("Cargo.toml");
        let root_manifest = project_root.join("Cargo.toml");
        let package_name = cargo_package_name(&app_manifest)?
            .or(cargo_package_name(&root_manifest)?)
            .ok_or_else(|| {
                anyhow!(
                    "local project at {} has no package name in app/Cargo.toml or Cargo.toml",
                    project_root.display()
                )
            })?;
        let spacetime_module = discover_spacetime_module(project_root)?;

        return Ok(LocalStack {
            project_name: format!("{package_name}-local"),
            env_path,
            spacetime_module,
        });
    }

    Err(anyhow!(
        "no local project with services.env was found from {}",
        start.display()
    ))
}

fn cargo_package_name(manifest_path: &Path) -> Result<Option<String>> {
    if !manifest_path.is_file() {
        return Ok(None);
    }

    let source = fs::read_to_string(manifest_path)
        .with_context(|| format!("failed to read Cargo manifest {}", manifest_path.display()))?;
    let manifest = toml::from_str::<toml::Table>(&source)
        .with_context(|| format!("failed to parse Cargo manifest {}", manifest_path.display()))?;

    Ok(manifest
        .get("package")
        .and_then(toml::Value::as_table)
        .and_then(|package| package.get("name"))
        .and_then(toml::Value::as_str)
        .map(str::trim)
        .filter(|name| !name.is_empty())
        .map(str::to_owned))
}

fn discover_spacetime_module(project_root: &Path) -> Result<Option<LocalSpacetimeModule>> {
    let path = project_root
        .join("crates")
        .join("database")
        .join("spacetimedb");
    let manifest_path = path.join("Cargo.toml");
    let Some(package_name) = cargo_package_name(&manifest_path)? else {
        return Ok(None);
    };
    let name = package_name
        .strip_suffix("-spacetimedb")
        .unwrap_or(package_name.as_str())
        .to_owned();

    Ok(Some(LocalSpacetimeModule { name, path }))
}

fn apply_local_project_name(
    local_stack: Option<&mut LocalStack>,
    project_name: Option<String>,
) -> Result<()> {
    let Some(project_name) = project_name else {
        return Ok(());
    };
    let project_name = project_name.trim();
    if project_name.is_empty() {
        return Err(anyhow!("local-stack project name must not be empty"));
    }
    if let Some(local_stack) = local_stack {
        local_stack.project_name = project_name.to_owned();
    }
    Ok(())
}

/// Configures how a MariaDB resource should be resolved for the runtime.
#[derive(Default, Clone)]
pub struct MariaDB {
    database_name: Option<String>,
    local_uri: Option<String>,
    port: Option<u16>,
    config: Option<Arc<AppConfig>>,
    local_stack: Option<LocalStack>,
    local_project_name: Option<String>,
}

impl MariaDB {
    /// Starts a builder without an implicit connection source.
    pub fn builder() -> Self {
        Self::default()
    }

    /// Overrides the logical database name used within the connection string.
    pub fn database_name(mut self, database_name: impl Into<String>) -> Self {
        self.database_name = Some(database_name.into());
        self
    }

    /// Supplies an explicit connection URI that bypasses the application configuration.
    pub fn local_uri(mut self, local_uri: impl Into<String>) -> Self {
        self.local_uri = Some(local_uri.into());
        self
    }

    /// Sets the local MariaDB host port in the resolved connection URI.
    pub fn port(mut self, port: u16) -> Self {
        self.port = Some(port);
        self
    }

    /// Supplies the application-owned configuration loaded before runtime setup.
    pub fn config(mut self, config: Arc<AppConfig>) -> Self {
        self.config = Some(config);
        self
    }

    /// Overrides the automatically discovered local Docker settings.
    pub fn local_stack(mut self, local_stack: LocalStack) -> Self {
        self.local_stack = Some(local_stack);
        self
    }

    /// Overrides only the Docker Compose project name of the resolved local stack.
    pub fn local_project_name(mut self, project_name: impl Into<String>) -> Self {
        self.local_project_name = Some(project_name.into());
        self
    }
}

/// Configures how a SpacetimeDB endpoint should be resolved for the runtime.
#[derive(Default, Clone)]
pub struct SpacetimeDB {
    local_uri: Option<String>,
    port: Option<u16>,
    config: Option<Arc<AppConfig>>,
    local_stack: Option<LocalStack>,
    local_project_name: Option<String>,
    module_name: Option<String>,
    module_path: Option<PathBuf>,
}

impl SpacetimeDB {
    /// Starts a builder without an implicit endpoint source.
    pub fn builder() -> Self {
        Self::default()
    }

    /// Supplies an explicit endpoint URI that bypasses application configuration.
    pub fn local_uri(mut self, local_uri: impl Into<String>) -> Self {
        self.local_uri = Some(local_uri.into());
        self
    }

    /// Sets the local host port used when the config URL is rewritten locally.
    pub fn port(mut self, port: u16) -> Self {
        self.port = Some(port);
        self
    }

    /// Supplies the application-owned configuration loaded before runtime setup.
    pub fn config(mut self, config: Arc<AppConfig>) -> Self {
        self.config = Some(config);
        self
    }

    /// Overrides the automatically discovered local Docker settings.
    pub fn local_stack(mut self, local_stack: LocalStack) -> Self {
        self.local_stack = Some(local_stack);
        self
    }

    /// Overrides only the Docker Compose project name of the resolved local stack.
    pub fn local_project_name(mut self, project_name: impl Into<String>) -> Self {
        self.local_project_name = Some(project_name.into());
        self
    }

    /// Overrides the discovered SpacetimeDB module identity.
    pub fn module_name(mut self, module_name: impl Into<String>) -> Self {
        self.module_name = Some(module_name.into());
        self
    }

    /// Overrides the discovered SpacetimeDB module source directory.
    pub fn module_path(mut self, path: impl Into<PathBuf>) -> Self {
        self.module_path = Some(path.into());
        self
    }
}

/// Configures the app web listener address for the runtime.
#[derive(Debug, Clone)]
pub struct WebBind {
    addr: String,
}

impl WebBind {
    /// Starts a builder with the fallback web listener address.
    pub fn builder() -> Self {
        Self {
            addr: "0.0.0.0:4000".to_string(),
        }
    }

    /// Sets the concrete listener address used by the app runtime.
    pub fn addr(mut self, addr: impl Into<String>) -> Self {
        self.addr = addr.into();
        self
    }
}

/// Intermediate configuration derived from [`MariaDB`] used during loading.
struct MariaDbResource {
    connection_string: String,
    local_stack: Option<LocalStack>,
    bootstrap_local_stack: bool,
}

/// Intermediate configuration derived from [`SpacetimeDB`] used during loading.
struct SpacetimeDbResource {
    url: String,
    local_stack: Option<LocalStack>,
    bootstrap_local_stack: bool,
    module_name: Option<String>,
    module_path: Option<PathBuf>,
}

/// Resolved SpacetimeDB endpoint plus the private module-owner token when the
/// shared runtime manages a local development module.
pub struct SpacetimeDbConnection {
    endpoint_url: String,
    auth_token: Option<String>,
}

impl SpacetimeDbConnection {
    /// Returns the endpoint URL used by generated SpacetimeDB clients.
    pub fn endpoint_url(&self) -> &str {
        self.endpoint_url.as_str()
    }

    /// Returns the module-owner token required for private host-owned tables
    /// and reducers. Remote deployments supply this separately.
    pub fn auth_token(&self) -> Option<&str> {
        self.auth_token.as_deref()
    }

    /// Supplies a deployment-owned module-owner token when the endpoint is not
    /// managed by the local development stack.
    pub fn with_auth_token(mut self, auth_token: impl Into<String>) -> Self {
        self.auth_token = Some(auth_token.into());
        self
    }
}

impl std::fmt::Debug for SpacetimeDbConnection {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("SpacetimeDbConnection")
            .field("endpoint_url", &redact_mysql_url(&self.endpoint_url))
            .field(
                "auth_token",
                &self.auth_token.as_ref().map(|_| "[redacted]"),
            )
            .finish()
    }
}

impl MariaDB {
    fn resolve(self, context: &Runtime) -> Result<MariaDbResource> {
        let mut local_stack = self.local_stack;
        let local_project_name = self.local_project_name;
        let has_configured_database = self
            .config
            .as_ref()
            .is_some_and(|config| config.mariadb.is_configured());
        if self.local_uri.is_none()
            && !has_configured_database
            && context.environment.allows_local_resources()
            && local_stack.is_none()
        {
            local_stack = Some(discover_local_stack(context.project.root())?);
        }
        let mut connection_string = match self.local_uri {
            Some(override_uri) => override_uri,
            None => resolve_connection_string(
                context.environment,
                self.config.as_deref(),
                local_stack.as_ref(),
            )?,
        };

        if context.environment.allows_local_resources()
            && let Some(port) = self.port
        {
            connection_string = rewrite_connection_port(&connection_string, port)?;
        }

        if let Some(database_name) = self.database_name {
            connection_string = rewrite_database_name(&connection_string, &database_name)?;
        }

        let (mut local_stack, bootstrap_local_stack) = resolve_local_stack(
            context.environment,
            &connection_string,
            local_stack,
            context.project.root(),
        )?;
        apply_local_project_name(local_stack.as_mut(), local_project_name)?;

        Ok(MariaDbResource {
            connection_string,
            local_stack,
            bootstrap_local_stack,
        })
    }
}

impl SpacetimeDB {
    fn resolve(self, context: &Runtime) -> Result<SpacetimeDbResource> {
        let mut local_stack = self.local_stack;
        let local_project_name = self.local_project_name;
        let has_configured_spacetime = self
            .config
            .as_ref()
            .is_some_and(|config| config.spacetimedb.is_configured());
        if self.local_uri.is_none()
            && !has_configured_spacetime
            && context.environment.allows_local_resources()
            && local_stack.is_none()
        {
            local_stack = Some(discover_local_stack(context.project.root())?);
        }
        let url = resolve_spacetime_url(
            context.environment,
            self.local_uri,
            self.port,
            self.config.as_deref(),
            local_stack.as_ref(),
        )?;
        let (mut local_stack, bootstrap_local_stack) = resolve_local_stack(
            context.environment,
            &url,
            local_stack,
            context.project.root(),
        )?;
        apply_local_project_name(local_stack.as_mut(), local_project_name)?;
        if self.module_path.is_some() != self.module_name.is_some() {
            return Err(anyhow!(
                "SpacetimeDB module name and path must be supplied together"
            ));
        }
        let (module_name, module_path) = match (self.module_name, self.module_path) {
            (Some(name), Some(path)) => (Some(name), Some(path)),
            (None, None) => local_stack
                .as_ref()
                .and_then(|stack| stack.spacetime_module.as_ref())
                .map_or((None, None), |module| {
                    (Some(module.name.clone()), Some(module.path.clone()))
                }),
            _ => unreachable!("module name/path pairing was validated above"),
        };

        Ok(SpacetimeDbResource {
            url,
            local_stack,
            bootstrap_local_stack,
            module_name,
            module_path,
        })
    }
}

impl Resource for MariaDB {
    type Output = String;

    async fn load(self, runtime: &Runtime) -> std::result::Result<String, RuntimeError> {
        let result = async {
            let resource = self.resolve(runtime)?;
            if resource.bootstrap_local_stack {
                let stack = resource.local_stack.as_ref().ok_or_else(|| {
                    anyhow!("local-stack settings are required for local bootstrap")
                })?;
                ensure_local_stack(stack, &[], &resource.connection_string).await?;
            }
            Ok::<_, anyhow::Error>(resource.connection_string)
        }
        .await;
        result.map_err(|source| RuntimeError::Resource {
            resource: "MariaDB",
            source: source.into_boxed_dyn_error(),
        })
    }
}

impl Resource for SpacetimeDB {
    type Output = SpacetimeDbConnection;

    async fn load(
        self,
        runtime: &Runtime,
    ) -> std::result::Result<SpacetimeDbConnection, RuntimeError> {
        let result = async {
            let resource = self.resolve(runtime)?;
            if resource.bootstrap_local_stack {
                let stack = resource.local_stack.as_ref().ok_or_else(|| {
                    anyhow!("local-stack settings are required for local bootstrap")
                })?;
                ensure_local_stack(stack, &["spacetime"], &resource.url).await?;
            }
            let auth_token =
                if runtime.environment.allows_local_resources() && resource.local_stack.is_some() {
                    match (
                        resource.module_name.as_deref(),
                        resource.module_path.as_deref(),
                    ) {
                        (Some(name), Some(path)) => {
                            ensure_local_spacetime_module(&resource.url, name, path).await?
                        }
                        _ => None,
                    }
                } else {
                    None
                };
            Ok::<_, anyhow::Error>(SpacetimeDbConnection {
                endpoint_url: resource.url,
                auth_token,
            })
        }
        .await;
        result.map_err(|source| RuntimeError::Resource {
            resource: "SpacetimeDB",
            source: source.into_boxed_dyn_error(),
        })
    }
}

impl Resource for WebBind {
    type Output = SocketAddr;

    async fn load(self, _runtime: &Runtime) -> std::result::Result<SocketAddr, RuntimeError> {
        let result = (|| -> anyhow::Result<SocketAddr> {
            let mut addr = self
                .addr
                .parse::<SocketAddr>()
                .context("the web listener address must be a socket address")?;
            if let Some(port) = ch_config::environment::value("PORT")? {
                addr.set_port(
                    port.parse::<u16>()
                        .context("PORT must be an integer TCP port")?,
                );
            }
            Ok(addr)
        })();
        result.map_err(|source| RuntimeError::Resource {
            resource: "web listener",
            source: source.into_boxed_dyn_error(),
        })
    }
}

impl std::fmt::Debug for MariaDB {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MariaDB")
            .field("database_name", &self.database_name)
            .field(
                "local_uri",
                &self.local_uri.as_deref().map(redact_mysql_url),
            )
            .field("port", &self.port)
            .field("config", &self.config.as_ref().map(|_| "[configured]"))
            .field("local_stack", &self.local_stack)
            .field("local_project_name", &self.local_project_name)
            .finish()
    }
}

impl std::fmt::Debug for SpacetimeDB {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SpacetimeDB")
            .field(
                "local_uri",
                &self.local_uri.as_deref().map(redact_mysql_url),
            )
            .field("port", &self.port)
            .field("config", &self.config.as_ref().map(|_| "[configured]"))
            .field("local_stack", &self.local_stack)
            .field("local_project_name", &self.local_project_name)
            .field("module_name", &self.module_name)
            .field("module_path", &self.module_path)
            .finish()
    }
}

impl std::fmt::Debug for MariaDbResource {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MariaDbResource")
            .field(
                "connection_string",
                &redact_mysql_url(&self.connection_string),
            )
            .field("local_stack", &self.local_stack)
            .field("bootstrap_local_stack", &self.bootstrap_local_stack)
            .finish()
    }
}

impl std::fmt::Debug for SpacetimeDbResource {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SpacetimeDbResource")
            .field("url", &redact_mysql_url(&self.url))
            .field("local_stack", &self.local_stack)
            .field("bootstrap_local_stack", &self.bootstrap_local_stack)
            .field("module_name", &self.module_name)
            .field("module_path", &self.module_path)
            .finish()
    }
}

fn resolve_connection_string(
    environment: RuntimeEnvironment,
    config: Option<&AppConfig>,
    local_stack: Option<&LocalStack>,
) -> Result<String> {
    if let Some(config) = config.filter(|config| config.mariadb.is_configured()) {
        return Ok(config.mariadb.connection_url()?);
    }

    if environment.allows_local_resources()
        && let Some(local_stack) = local_stack
    {
        return local_mariadb_url(local_stack.env_path());
    }

    Err(anyhow!("MariaDB configuration must be supplied explicitly"))
}

fn local_mariadb_url(env_path: &Path) -> Result<String> {
    let values = dotenvy::from_path_iter(env_path)
        .with_context(|| {
            format!(
                "failed to read local MariaDB environment from {}",
                env_path.display()
            )
        })?
        .collect::<std::result::Result<std::collections::HashMap<_, _>, _>>()
        .with_context(|| {
            format!(
                "failed to parse local MariaDB environment from {}",
                env_path.display()
            )
        })?;
    let required = |name: &'static str| {
        values
            .get(name)
            .map(String::as_str)
            .filter(|value| !value.trim().is_empty())
            .ok_or_else(|| {
                anyhow!(
                    "{name} must be set in the local MariaDB environment file {}",
                    env_path.display()
                )
            })
    };
    let user = required("MARIADB_USER")?;
    let password = required("MARIADB_PASSWORD")?;
    let database = required("MARIADB_DATABASE")?;
    let port = values
        .get("MARIADB_PORT")
        .map(String::as_str)
        .filter(|value| !value.trim().is_empty())
        .unwrap_or("3306")
        .parse::<u16>()
        .with_context(|| {
            format!(
                "MARIADB_PORT must be a valid TCP port in {}",
                env_path.display()
            )
        })?;

    let mut url = url::Url::parse("mysql://127.0.0.1")
        .expect("the static local MariaDB base URL must be valid");
    url.set_username(user)
        .map_err(|()| anyhow!("MARIADB_USER contains unsupported URL data"))?;
    url.set_password(Some(password))
        .map_err(|()| anyhow!("MARIADB_PASSWORD contains unsupported URL data"))?;
    url.set_port(Some(port))
        .map_err(|()| anyhow!("MARIADB_PORT cannot be represented in the connection URL"))?;
    url.set_path(database);

    Ok(url.into())
}

fn resolve_spacetime_url(
    environment: RuntimeEnvironment,
    local_uri: Option<String>,
    port: Option<u16>,
    config: Option<&AppConfig>,
    local_stack: Option<&LocalStack>,
) -> Result<String> {
    if let Some(local_uri) = local_uri {
        return Ok(local_uri);
    }

    let mut url = match config.filter(|config| config.spacetimedb.is_configured()) {
        Some(config) => config.spacetimedb.endpoint_url()?.to_owned(),
        None if environment.allows_local_resources() => local_spacetime_url(port, local_stack)?,
        None => {
            return Err(anyhow!(
                "SpacetimeDB configuration must be supplied explicitly outside local development"
            ));
        }
    };

    if environment.allows_local_resources()
        && let Some(port) = port
    {
        url = rewrite_url_port(&url, port)?;
    }

    Ok(url)
}

fn local_spacetime_url(port: Option<u16>, local_stack: Option<&LocalStack>) -> Result<String> {
    let port = match (port, local_stack) {
        (Some(port), _) => port,
        (None, Some(local_stack)) => local_spacetime_port(local_stack.env_path())?,
        (None, None) => 4530,
    };

    Ok(format!("http://127.0.0.1:{port}"))
}

fn local_spacetime_port(env_path: &Path) -> Result<u16> {
    let entries = dotenvy::from_path_iter(env_path).with_context(|| {
        format!(
            "failed to read local SpacetimeDB environment from {}",
            env_path.display()
        )
    })?;

    for entry in entries {
        let (name, value) = entry.with_context(|| {
            format!(
                "failed to parse local SpacetimeDB environment from {}",
                env_path.display()
            )
        })?;
        if name == "SPACETIME_PORT" {
            return value.parse::<u16>().with_context(|| {
                format!(
                    "SPACETIME_PORT must be a valid TCP port in {}",
                    env_path.display()
                )
            });
        }
    }

    Ok(4530)
}

fn resolve_local_stack(
    environment: RuntimeEnvironment,
    connection_url: &str,
    local_stack: Option<LocalStack>,
    project_root: &Path,
) -> Result<(Option<LocalStack>, bool)> {
    let bootstrap = should_bootstrap_local_stack(environment, connection_url);
    let local_stack = match local_stack {
        Some(local_stack) => Some(local_stack),
        None if environment.allows_local_resources() && is_local_database(connection_url) => {
            let discovered = discover_local_stack(project_root);
            match discovered {
                Ok(local_stack) => Some(local_stack),
                Err(error) if bootstrap => return Err(error),
                Err(_) => None,
            }
        }
        None => None,
    };
    let Some(local_stack) = local_stack else {
        return Ok((None, false));
    };
    if local_stack.project_name().trim().is_empty() {
        return Err(anyhow!("local-stack project name must not be empty"));
    }
    if local_stack.env_path().as_os_str().is_empty() {
        return Err(anyhow!("local-stack env-file path must not be empty"));
    }

    Ok((Some(local_stack), bootstrap))
}

fn rewrite_url_port(current: &str, port: u16) -> Result<String> {
    let scheme_end = current
        .find("://")
        .ok_or_else(|| anyhow!("invalid url: {}", redact_mysql_url(current)))?;
    let authority_start = scheme_end + "://".len();
    let suffix_start = current[authority_start..]
        .find(['/', '?', '#'])
        .map(|index| authority_start + index)
        .unwrap_or(current.len());
    rewrite_authority_port(current, authority_start, suffix_start, port)
}

/// Rewrites the authority port while preserving credentials, host, path and suffix.
fn rewrite_connection_port(current: &str, port: u16) -> Result<String> {
    let scheme_end = current
        .find("://")
        .ok_or_else(|| anyhow!("invalid database url: {}", redact_mysql_url(current)))?;
    let authority_start = scheme_end + "://".len();
    let relative_path_start = current[authority_start..]
        .find('/')
        .ok_or_else(|| anyhow!("invalid database url: {}", redact_mysql_url(current)))?;
    let path_start = authority_start + relative_path_start;
    rewrite_authority_port(current, authority_start, path_start, port)
}

fn rewrite_authority_port(
    current: &str,
    authority_start: usize,
    suffix_start: usize,
    port: u16,
) -> Result<String> {
    let authority = &current[authority_start..suffix_start];
    let host_start = authority.rfind('@').map(|index| index + 1).unwrap_or(0);
    let (userinfo, host_port) = authority.split_at(host_start);
    let host_port = rewrite_host_port(host_port, port)?;

    let mut rebuilt = String::with_capacity(current.len() + port.to_string().len() + ":".len());
    rebuilt.push_str(&current[..authority_start]);
    rebuilt.push_str(userinfo);
    rebuilt.push_str(&host_port);
    rebuilt.push_str(&current[suffix_start..]);

    Ok(rebuilt)
}

fn rewrite_host_port(host_port: &str, port: u16) -> Result<String> {
    if host_port.trim().is_empty() {
        return Err(anyhow!("invalid database url host"));
    }

    if let Some(rest) = host_port.strip_prefix('[') {
        let bracket_end = rest
            .find(']')
            .ok_or_else(|| anyhow!("invalid bracketed database url host: {host_port}"))?;
        let host_end = bracket_end + 2;
        let host = &host_port[..host_end];
        return Ok(format!("{host}:{port}"));
    }

    if let Some((host, candidate_port)) = host_port.rsplit_once(':')
        && !host.is_empty()
        && candidate_port.chars().all(|ch| ch.is_ascii_digit())
    {
        return Ok(format!("{host}:{port}"));
    }

    Ok(format!("{host_port}:{port}"))
}

/// Rewrites the database name segment while preserving any URL suffix.
fn rewrite_database_name(current: &str, database_name: &str) -> Result<String> {
    let trimmed = database_name.trim();
    if trimmed.is_empty() {
        return Err(anyhow!("database_name must not be empty"));
    }

    let last_slash = current
        .rfind('/')
        .ok_or_else(|| anyhow!("invalid database url: {}", redact_mysql_url(current)))?;

    let suffix_start = current[last_slash..]
        .find(|ch| ['?', '#'].contains(&ch))
        .map(|relative| last_slash + relative);

    let (base, suffix) = match suffix_start {
        Some(index) => current.split_at(index),
        None => (current, ""),
    };

    let (prefix, _) = base
        .rsplit_once('/')
        .ok_or_else(|| anyhow!("invalid database url: {}", redact_mysql_url(current)))?;

    let mut rebuilt = String::with_capacity(prefix.len() + 1 + trimmed.len() + suffix.len());
    rebuilt.push_str(prefix);
    rebuilt.push('/');
    rebuilt.push_str(trimmed);
    rebuilt.push_str(suffix);

    Ok(rebuilt)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicU64, Ordering};
    use tempfile::tempdir;

    static LOCAL_STACK_SEQUENCE: AtomicU64 = AtomicU64::new(0);

    fn test_runtime(environment: RuntimeEnvironment) -> Runtime {
        Runtime {
            environment,
            project: Project::discover_from(env!("CARGO_MANIFEST_DIR")).unwrap(),
            environment_load: Default::default(),
        }
    }

    fn test_config() -> Arc<AppConfig> {
        use ch_config::config::{CaptchaConfig, MariadbConfig, SpacetimeDbConfig};

        Arc::new(AppConfig {
            mariadb: MariadbConfig::new("mysql://user:pass@127.0.0.1:4520/app"),
            spacetimedb: SpacetimeDbConfig::new("http://127.0.0.1:4530"),
            captcha: CaptchaConfig::standard(),
            ..AppConfig::default()
        })
    }

    fn test_local_stack() -> LocalStack {
        let sequence = LOCAL_STACK_SEQUENCE.fetch_add(1, Ordering::Relaxed);
        let env_path =
            std::env::temp_dir().join(format!("local-stack-{}-{sequence}.env", std::process::id()));
        std::fs::write(
            &env_path,
            "MARIADB_ROOT_PASSWORD=root\nMARIADB_DATABASE=app\nMARIADB_USER=user\nMARIADB_PASSWORD=pass\n",
        )
        .expect("test stack env should be writable");
        LocalStack::new("test-data-services", env_path)
    }

    fn test_local_stack_with_spacetime_port(port: u16) -> LocalStack {
        let stack = test_local_stack();
        let mut contents =
            fs::read_to_string(stack.env_path()).expect("test stack env should remain readable");
        contents.push_str(format!("SPACETIME_PORT={port}\n").as_str());
        fs::write(stack.env_path(), contents).expect("SpacetimeDB port should be persisted");
        stack
    }

    #[test]
    fn preserves_query_params_when_rewriting_database_name() {
        let url = "mysql://user:pa#ss@127.0.0.1:4520/demo?ssl-mode=REQUIRED&pool_min=3";
        let rewritten =
            rewrite_database_name(url, "custom_db").expect("database name rewrite should succeed");

        assert_eq!(
            rewritten,
            "mysql://user:pa#ss@127.0.0.1:4520/custom_db?ssl-mode=REQUIRED&pool_min=3"
        );
    }

    #[test]
    fn preserves_credentials_and_suffix_when_rewriting_port() {
        let url = "mysql://user:pa#ss@word@127.0.0.1:3306/demo?ssl-mode=REQUIRED&pool_min=3";
        let rewritten = rewrite_connection_port(url, 4521).expect("port rewrite should succeed");

        assert_eq!(
            rewritten,
            "mysql://user:pa#ss@word@127.0.0.1:4521/demo?ssl-mode=REQUIRED&pool_min=3"
        );
    }

    #[test]
    fn appends_missing_connection_port() {
        let url = "mysql://user:pass@localhost/demo";
        let rewritten = rewrite_connection_port(url, 4524).expect("port rewrite should succeed");

        assert_eq!(rewritten, "mysql://user:pass@localhost:4524/demo");
    }

    #[test]
    fn applies_local_port_override_to_connection_url() {
        let context = test_runtime(RuntimeEnvironment::Local);
        let resource = MariaDB::builder()
            .local_uri("mysql://user:pass@127.0.0.1:3306/app?ssl-mode=DISABLED")
            .port(4521)
            .local_stack(test_local_stack())
            .resolve(&context)
            .expect("resource build should succeed");

        assert_eq!(
            resource.connection_string,
            "mysql://user:pass@127.0.0.1:4521/app?ssl-mode=DISABLED"
        );
    }

    #[test]
    fn resolves_local_mariadb_from_the_explicit_stack_env_file() {
        let context = test_runtime(RuntimeEnvironment::Local);
        let resource = MariaDB::builder()
            .local_stack(test_local_stack())
            .resolve(&context)
            .expect("the explicit local stack should provide its MariaDB settings");

        assert_eq!(
            resource.connection_string,
            "mysql://user:pass@127.0.0.1:3306/app"
        );
    }

    #[test]
    fn local_resource_builders_override_only_the_compose_project_name() {
        let context = test_runtime(RuntimeEnvironment::Local);
        let mariadb = MariaDB::builder()
            .local_stack(test_local_stack())
            .local_project_name("foxfyp")
            .resolve(&context)
            .expect("MariaDB local project name override should build");
        let spacetime = SpacetimeDB::builder()
            .local_stack(test_local_stack_with_spacetime_port(4531))
            .local_project_name("foxfyp")
            .resolve(&context)
            .expect("SpacetimeDB local project name override should build");

        assert_eq!(
            mariadb
                .local_stack
                .as_ref()
                .expect("MariaDB local stack should remain available")
                .project_name(),
            "foxfyp"
        );
        assert_eq!(
            spacetime
                .local_stack
                .as_ref()
                .expect("SpacetimeDB local stack should remain available")
                .project_name(),
            "foxfyp"
        );
    }

    #[test]
    fn local_resource_builders_reject_an_empty_compose_project_name() {
        let context = test_runtime(RuntimeEnvironment::Local);
        let mariadb_error = MariaDB::builder()
            .local_stack(test_local_stack())
            .local_project_name("   ")
            .resolve(&context)
            .expect_err("MariaDB must reject an empty local project name");
        let spacetime_error = SpacetimeDB::builder()
            .local_stack(test_local_stack_with_spacetime_port(4531))
            .local_project_name("\t")
            .resolve(&context)
            .expect_err("SpacetimeDB must reject an empty local project name");

        assert!(
            mariadb_error
                .to_string()
                .contains("local-stack project name must not be empty")
        );
        assert!(
            spacetime_error
                .to_string()
                .contains("local-stack project name must not be empty")
        );
    }

    #[test]
    fn configured_local_resources_override_the_late_discovered_compose_project_name() {
        let temp = tempdir().expect("temporary project should be created");
        let project_root = temp.path().join("example");
        let app_dir = project_root.join("app");
        let module_dir = project_root
            .join("crates")
            .join("database")
            .join("spacetimedb");
        fs::create_dir_all(&app_dir).expect("app directory should be created");
        fs::create_dir_all(&module_dir).expect("module directory should be created");
        fs::write(
            project_root.join("services.env"),
            "MARIADB_ROOT_PASSWORD=root\nMARIADB_DATABASE=app\nMARIADB_USER=user\nMARIADB_PASSWORD=pass\nSPACETIME_PORT=4530\n",
        )
        .expect("local env should be written");
        fs::write(
            app_dir.join("Cargo.toml"),
            "[package]\nname = \"ffyp\"\nversion = \"0.1.0\"\n",
        )
        .expect("app manifest should be written");
        fs::write(
            module_dir.join("Cargo.toml"),
            "[package]\nname = \"ffyp-spacetimedb\"\nversion = \"0.1.0\"\n",
        )
        .expect("module manifest should be written");

        let mut context = test_runtime(RuntimeEnvironment::Local);
        context.project = Project::discover_from(&project_root).unwrap();
        let mariadb = MariaDB::builder()
            .config(test_config())
            .local_project_name("foxfyp")
            .resolve(&context)
            .expect("configured MariaDB resource should discover its local stack");
        let spacetime = SpacetimeDB::builder()
            .config(test_config())
            .local_project_name("foxfyp")
            .resolve(&context)
            .expect("configured SpacetimeDB resource should discover its local stack");

        assert_eq!(
            mariadb
                .local_stack
                .as_ref()
                .expect("MariaDB local stack should be discovered")
                .project_name(),
            "foxfyp"
        );
        assert_eq!(
            mariadb
                .local_stack
                .as_ref()
                .expect("MariaDB local stack should be discovered")
                .env_path(),
            project_root.join("services.env")
        );
        assert_eq!(
            spacetime
                .local_stack
                .as_ref()
                .expect("SpacetimeDB local stack should be discovered")
                .project_name(),
            "foxfyp"
        );
        assert_eq!(
            spacetime
                .local_stack
                .as_ref()
                .expect("SpacetimeDB local stack should be discovered")
                .env_path(),
            project_root.join("services.env")
        );
        assert_eq!(spacetime.module_name.as_deref(), Some("ffyp"));
        assert_eq!(spacetime.module_path.as_deref(), Some(module_dir.as_path()));
    }

    #[test]
    fn discovers_local_stack_and_spacetime_module_from_project_layout() {
        let temp = tempdir().expect("temporary project should be created");
        let project_root = temp.path().join("example");
        let nested_start = project_root.join("app").join("src");
        let app_dir = project_root.join("app");
        let module_dir = project_root
            .join("crates")
            .join("database")
            .join("spacetimedb");
        fs::create_dir_all(&nested_start).expect("nested project path should be created");
        fs::create_dir_all(&module_dir).expect("module directory should be created");
        fs::write(
            project_root.join("services.env"),
            "MARIADB_DATABASE=example\n",
        )
        .expect("local env should be written");
        fs::write(
            app_dir.join("Cargo.toml"),
            "[package]\nname = \"example-service\"\nversion = \"0.1.0\"\n",
        )
        .expect("app manifest should be written");
        fs::write(
            module_dir.join("Cargo.toml"),
            "[package]\nname = \"example-spacetimedb\"\nversion = \"0.1.0\"\n",
        )
        .expect("module manifest should be written");

        let stack = LocalStack::discover_from(&nested_start)
            .expect("the conventional project layout should be discovered");

        assert_eq!(stack.project_name(), "example-service-local");
        assert_eq!(stack.env_path(), project_root.join("services.env"));
        let module = stack
            .spacetime_module
            .expect("the conventional SpacetimeDB module should be discovered");
        assert_eq!(module.name, "example");
        assert_eq!(module.path, module_dir);
    }

    #[test]
    fn production_mariadb_still_requires_explicit_app_config() {
        let context = test_runtime(RuntimeEnvironment::Production);
        let error = MariaDB::builder()
            .local_stack(test_local_stack())
            .resolve(&context)
            .expect_err("production must not read credentials from a local stack file");

        assert!(
            error
                .to_string()
                .contains("MariaDB configuration must be supplied explicitly")
        );
    }

    #[test]
    fn builds_spacetime_resource_from_explicit_url() {
        let context = test_runtime(RuntimeEnvironment::Local);
        let resource = SpacetimeDB::builder()
            .local_uri("http://127.0.0.1:4531")
            .local_stack(test_local_stack())
            .module_name("example-module")
            .module_path("crates/database/spacetimedb")
            .resolve(&context)
            .expect("resource build should succeed");

        assert_eq!(resource.url, "http://127.0.0.1:4531");
    }

    #[test]
    fn resolves_local_spacetime_from_the_explicit_stack_env_file() {
        let context = test_runtime(RuntimeEnvironment::Local);
        let resource = SpacetimeDB::builder()
            .local_stack(test_local_stack_with_spacetime_port(4531))
            .resolve(&context)
            .expect("the explicit local stack should provide its SpacetimeDB port");

        assert_eq!(resource.url, "http://127.0.0.1:4531");
    }

    #[test]
    fn builds_spacetime_resource_from_config() {
        let context = test_runtime(RuntimeEnvironment::Local);
        let resource = SpacetimeDB::builder()
            .config(test_config())
            .local_stack(test_local_stack())
            .module_name("example-module")
            .module_path("crates/database/spacetimedb")
            .resolve(&context)
            .expect("resource build should succeed");

        assert_eq!(resource.url, "http://127.0.0.1:4530");
    }

    #[test]
    fn applies_local_spacetime_port_override_to_config_url() {
        let context = test_runtime(RuntimeEnvironment::Local);
        let resource = SpacetimeDB::builder()
            .config(test_config())
            .local_stack(test_local_stack())
            .module_name("example-module")
            .module_path("crates/database/spacetimedb")
            .port(4531)
            .resolve(&context)
            .expect("resource build should succeed");

        assert_eq!(resource.url, "http://127.0.0.1:4531");
    }

    #[tokio::test]
    async fn production_never_prepares_explicit_local_spacetime_modules() {
        let directory = tempfile::tempdir().unwrap();
        let runtime = test_runtime(RuntimeEnvironment::Production);
        let connection = runtime
            .load(
                SpacetimeDB::builder()
                    .local_uri("http://127.0.0.1:19381")
                    .local_stack(LocalStack::new(
                        "production-policy-test",
                        directory.path().join("absent.env"),
                    ))
                    .module_name("production-policy-test")
                    .module_path(directory.path().join("absent-module")),
            )
            .await
            .unwrap();
        assert_eq!(connection.endpoint_url(), "http://127.0.0.1:19381");
        assert!(connection.auth_token().is_none());
        assert_eq!(fs::read_dir(directory.path()).unwrap().count(), 0);
    }

    #[tokio::test]
    async fn production_resolves_explicit_mariadb_without_a_local_stack_file() {
        let directory = tempfile::tempdir().unwrap();
        let runtime = test_runtime(RuntimeEnvironment::Production);
        let url = runtime
            .load(
                MariaDB::builder()
                    .local_uri("mysql://app:secret@127.0.0.1/database")
                    .local_stack(LocalStack::new(
                        "production-policy-test",
                        directory.path().join("absent.env"),
                    )),
            )
            .await
            .unwrap();
        assert_eq!(url, "mysql://app:secret@127.0.0.1/database");
        assert_eq!(fs::read_dir(directory.path()).unwrap().count(), 0);
    }

    #[test]
    fn resource_debug_output_redacts_passwords_and_tokens() {
        let config = Arc::new(AppConfig {
            mariadb: ch_config::config::MariadbConfig::new(
                "mysql://app:config-secret@localhost/database",
            ),
            ..AppConfig::default()
        });
        let maria = MariaDB::builder()
            .local_uri("mysql://app:url-secret@localhost/database")
            .config(Arc::clone(&config));
        let spacetime = SpacetimeDB::builder()
            .local_uri("http://app:url-secret@localhost:3000")
            .config(config);
        let connection = SpacetimeDbConnection {
            endpoint_url: "http://app:url-secret@localhost:3000".into(),
            auth_token: Some("owner-token-secret".into()),
        };
        for debug in [
            format!("{maria:?}"),
            format!("{spacetime:?}"),
            format!("{connection:?}"),
        ] {
            for secret in ["config-secret", "url-secret", "owner-token-secret"] {
                assert!(!debug.contains(secret), "secret leaked in Debug: {debug}");
            }
        }
    }

    #[test]
    fn runtime_error_preserves_the_original_error_type() {
        fn assert_error<T: std::error::Error + Send + Sync + 'static>() {}
        assert_error::<RuntimeError>();
        let error = RuntimeError::Io(std::io::Error::new(
            std::io::ErrorKind::NotFound,
            "missing project",
        ));
        let source = std::error::Error::source(&error)
            .unwrap()
            .downcast_ref::<std::io::Error>()
            .unwrap();
        assert_eq!(source.kind(), std::io::ErrorKind::NotFound);
    }
}