ch_rds/

lib.rs

#![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;
mod runtime;

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

use crate::docker::{
    ensure_local_spacetime_module, ensure_local_stack, ensure_local_stack_with_profiles,
    is_local_database, should_bootstrap_local_stack,
};
use crate::runtime::{
    Context as RuntimeContext, Environment, IntoResource, LoaderAndRunner, ResourceInputBuilder,
};
use anyhow::{Context as AnyhowContext, Result, anyhow};
use async_trait::async_trait;
use ch_config::config::AppConfig;
use serde::Serialize;
use sqlx::MySqlPool;
use sqlx::mysql::MySqlPoolOptions;

pub use crate::runtime::Environment as RuntimeEnvironment;
pub use project::Project;

#[doc(hidden)]
pub mod __internals {
    pub use crate::runtime::{
        Context, Environment, IntoResource, LoaderAndRunner, MetadataMap, ResourceInputBuilder,
    };
}

/// Runtime facade used by app entrypoints to load shared resources before
/// handing control to the app server future.
#[derive(Debug)]
pub struct Runtime {
    runner: LoaderAndRunner,
    project: Project,
    environment_load: ch_config::environment::EnvLoad,
}

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

    /// Bootstraps resources with an explicit local-resource policy.
    pub fn bootstrap_with_environment(environment: RuntimeEnvironment) -> Result<Self> {
        let project = Project::discover()?;
        let environment_load = project.load_environment()?;
        let mut context = RuntimeContext::with_environment(environment);
        context.set_project_root(project.root().to_path_buf());
        Ok(Self {
            runner: LoaderAndRunner::with_context(context),
            project,
            environment_load,
        })
    }

    /// 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
    }

    /// Loads one shared resource through its builder.
    pub async fn load<RB, Input, Output>(&self, builder: RB) -> Result<Output>
    where
        RB: ResourceInputBuilder<Input = Input>,
        Input: IntoResource<Output> + Send,
        Output: Send,
    {
        self.runner.load(builder).await
    }
}

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

#[derive(Debug, Clone, Serialize)]
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() -> Result<Self> {
        let current_dir = std::env::current_dir()
            .context("failed to determine the current directory for local-stack discovery")?;
        Self::discover_from(current_dir)
    }

    /// Discovers a project from an ancestor containing `services.env`.
    ///
    /// The Docker Compose project name is derived from `app/Cargo.toml`. A
    /// module at `crates/database/spacetimedb` is detected automatically.
    pub fn discover_from(start: impl AsRef<Path>) -> Result<Self> {
        let start = project::discovery_root(start.as_ref());

        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(Self {
                project_name: format!("{package_name}-local"),
                env_path,
                spacetime_module,
            });
        }

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

    /// 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 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 }))
}

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

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
    }

    /// Adjusts the maximum connection count when constructing SQLx pools.
    pub fn max_connections(mut self, max: u32) -> Self {
        self.max_connections = Some(max);
        self
    }

    /// Returns the configured database override if present.
    pub fn database_name_override(&self) -> Option<&str> {
        self.database_name.as_deref()
    }

    /// Returns the configured URI override if present.
    pub fn local_uri_override(&self) -> Option<&str> {
        self.local_uri.as_deref()
    }

    /// Returns the configured local port override if present.
    pub fn port_override(&self) -> Option<u16> {
        self.port
    }
}

/// Configures how a SpacetimeDB endpoint should be resolved for the runtime.
#[derive(Debug, Default, Clone)]
pub struct SpacetimeDB {
    local_uri: Option<String>,
    port: Option<u16>,
    config: Option<Arc<AppConfig>>,
    local_stack: Option<LocalStack>,
    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 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.
#[derive(Debug, Clone, Serialize)]
pub struct MariaDbResource {
    connection_string: String,
    max_connections: Option<u32>,
    local_stack: Option<LocalStack>,
    bootstrap_local_stack: bool,
}

impl MariaDbResource {
    /// Exposes the resolved connection string, mainly for diagnostics.
    pub fn connection_string(&self) -> &str {
        &self.connection_string
    }
}

/// Intermediate configuration derived from [`SpacetimeDB`] used during loading.
#[derive(Debug, Clone, Serialize)]
pub struct SpacetimeDbResource {
    url: String,
    local_stack: Option<LocalStack>,
    bootstrap_local_stack: bool,
    module_name: Option<String>,
    module_path: Option<PathBuf>,
}

impl SpacetimeDbResource {
    /// Exposes the resolved endpoint URL, mainly for diagnostics.
    pub fn url(&self) -> &str {
        &self.url
    }
}

/// 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", &self.endpoint_url)
            .field(
                "auth_token",
                &self.auth_token.as_ref().map(|_| "[redacted]"),
            )
            .finish()
    }
}

/// Intermediate configuration derived from [`WebBind`] used during loading.
#[derive(Debug, Clone, Serialize)]
pub struct WebBindResource {
    addr: String,
}

impl WebBindResource {
    /// Exposes the configured listener address, mainly for diagnostics.
    pub fn addr(&self) -> &str {
        &self.addr
    }
}

impl ResourceInputBuilder for MariaDB {
    type Input = MariaDbResource;

    fn build(self, context: &RuntimeContext) -> Result<Self::Input> {
        let mut local_stack = self.local_stack;
        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(match context.project_root() {
                Some(project_root) => LocalStack::discover_from(project_root)?,
                None => LocalStack::discover()?,
            });
        }
        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 (local_stack, bootstrap_local_stack) = resolve_local_stack(
            context.environment(),
            &connection_string,
            local_stack,
            context.project_root(),
        )?;

        Ok(MariaDbResource {
            connection_string,
            max_connections: self.max_connections,
            local_stack,
            bootstrap_local_stack,
        })
    }
}

impl ResourceInputBuilder for SpacetimeDB {
    type Input = SpacetimeDbResource;

    fn build(self, context: &RuntimeContext) -> Result<Self::Input> {
        let url = resolve_spacetime_url(
            context.environment(),
            self.local_uri,
            self.port,
            self.config.as_deref(),
        )?;
        let (local_stack, bootstrap_local_stack) = resolve_local_stack(
            context.environment(),
            &url,
            self.local_stack,
            context.project_root(),
        )?;
        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 ResourceInputBuilder for WebBind {
    type Input = WebBindResource;

    fn build(self, _context: &RuntimeContext) -> Result<Self::Input> {
        let addr = match ch_config::environment::value("PORT")? {
            Some(port) => {
                let port = port
                    .parse::<u16>()
                    .context("PORT must be an integer TCP port")?;
                let mut addr = self
                    .addr
                    .parse::<SocketAddr>()
                    .context("the fallback web listener address must be a socket address")?;
                addr.set_port(port);
                addr.to_string()
            }
            None => self.addr,
        };

        Ok(WebBindResource { addr })
    }
}

#[async_trait]
impl IntoResource<String> for MariaDbResource {
    async fn into_resource(self, _context: &RuntimeContext) -> Result<String> {
        let connection_string = apply_local_stack_overrides(
            self.connection_string,
            self.local_stack.as_ref(),
            self.bootstrap_local_stack,
        )
        .await?;

        Ok(connection_string)
    }
}

#[async_trait]
impl IntoResource<MySqlPool> for MariaDbResource {
    async fn into_resource(self, _context: &RuntimeContext) -> Result<MySqlPool> {
        let mut builder = MySqlPoolOptions::new();
        if let Some(max) = self.max_connections {
            builder = builder.max_connections(max);
        }

        let connection_string = apply_local_stack_overrides(
            self.connection_string,
            self.local_stack.as_ref(),
            self.bootstrap_local_stack,
        )
        .await?;

        builder
            .connect(&connection_string)
            .await
            .with_context(|| anyhow!("failed to connect to MariaDB"))
    }
}

#[async_trait]
impl IntoResource<String> for SpacetimeDbResource {
    async fn into_resource(self, _context: &RuntimeContext) -> Result<String> {
        Ok(resolve_spacetime_resource(self).await?.endpoint_url)
    }
}

#[async_trait]
impl IntoResource<SpacetimeDbConnection> for SpacetimeDbResource {
    async fn into_resource(self, _context: &RuntimeContext) -> Result<SpacetimeDbConnection> {
        resolve_spacetime_resource(self).await
    }
}

async fn resolve_spacetime_resource(
    resource: SpacetimeDbResource,
) -> Result<SpacetimeDbConnection> {
    let manage_local_module = resource.local_stack.is_some();
    let url = apply_local_stack_with_profiles(
        resource.url,
        resource.local_stack.as_ref(),
        resource.bootstrap_local_stack,
        &["spacetime"],
    )
    .await?;

    let mut auth_token = None;
    if manage_local_module && let Some(module_path) = resource.module_path.as_deref() {
        let module_name = resource.module_name.as_deref().ok_or_else(|| {
            anyhow!("SpacetimeDB module name must accompany an explicit module path")
        })?;
        auth_token = ensure_local_spacetime_module(url.as_str(), module_name, module_path)?;
    }

    Ok(SpacetimeDbConnection {
        endpoint_url: url,
        auth_token,
    })
}

#[async_trait]
impl IntoResource<SocketAddr> for WebBindResource {
    async fn into_resource(self, _context: &RuntimeContext) -> Result<SocketAddr> {
        self.addr
            .parse()
            .with_context(|| format!("invalid web bind address `{}`", self.addr))
    }
}

/// Ensures the local development stack is running when requested.
async fn apply_local_stack_overrides(
    connection_string: String,
    local_stack: Option<&LocalStack>,
    bootstrap: bool,
) -> Result<String> {
    if !bootstrap {
        return Ok(connection_string);
    }
    let local_stack = local_stack
        .ok_or_else(|| anyhow!("local-stack settings are required for local bootstrap"))?;

    ensure_local_stack(local_stack, connection_string.as_str()).await?;
    Ok(connection_string)
}

/// Ensures the local development stack is running with optional compose profiles.
async fn apply_local_stack_with_profiles(
    value: String,
    local_stack: Option<&LocalStack>,
    bootstrap: bool,
    profiles: &[&str],
) -> Result<String> {
    if !bootstrap {
        return Ok(value);
    }
    let local_stack = local_stack
        .ok_or_else(|| anyhow!("local-stack settings are required for local bootstrap"))?;

    ensure_local_stack_with_profiles(local_stack, profiles, Some(value.as_str())).await?;
    Ok(value)
}

fn resolve_connection_string(
    environment: Environment,
    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: Environment,
    local_uri: Option<String>,
    port: Option<u16>,
    config: Option<&AppConfig>,
) -> 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),
        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>) -> String {
    let port = port.unwrap_or(4530);

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

fn resolve_local_stack(
    environment: Environment,
    connection_url: &str,
    local_stack: Option<LocalStack>,
    project_root: Option<&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 = match project_root {
                Some(project_root) => LocalStack::discover_from(project_root),
                None => LocalStack::discover(),
            };
            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: {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: {current}"))?;
    let authority_start = scheme_end + "://".len();
    let relative_path_start = current[authority_start..]
        .find('/')
        .ok_or_else(|| anyhow!("invalid database 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: {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: {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 crate::__internals::{Context, Environment};
    use std::sync::atomic::{AtomicU64, Ordering};
    use tempfile::tempdir;

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

    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)
    }

    #[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 = Context::with_environment(Environment::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())
            .build(&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 = Context::with_environment(Environment::Local);
        let resource = MariaDB::builder()
            .local_stack(test_local_stack())
            .build(&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 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 = Context::with_environment(Environment::Production);
        let error = MariaDB::builder()
            .local_stack(test_local_stack())
            .build(&context)
            .expect_err("production must not read credentials from a local stack file");

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

    fn docker_available() -> bool {
        std::process::Command::new("docker")
            .arg("--version")
            .output()
            .is_ok()
    }

    #[test]
    fn builds_spacetime_resource_from_explicit_url() {
        let context = Context::with_environment(Environment::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")
            .build(&context)
            .expect("resource build should succeed");

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

    #[test]
    fn builds_spacetime_resource_from_config() {
        let context = Context::with_environment(Environment::Local);
        let resource = SpacetimeDB::builder()
            .config(test_config())
            .local_stack(test_local_stack())
            .module_name("example-module")
            .module_path("crates/database/spacetimedb")
            .build(&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 = Context::with_environment(Environment::Local);
        let resource = SpacetimeDB::builder()
            .config(test_config())
            .local_stack(test_local_stack())
            .module_name("example-module")
            .module_path("crates/database/spacetimedb")
            .port(4531)
            .build(&context)
            .expect("resource build should succeed");

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

    #[test]
    #[ignore = "requires a local Docker daemon and compose stack"]
    fn loads_default_connection_from_config() {
        if !docker_available() {
            eprintln!("skipping docker-backed test because Docker is not available");
            return;
        }

        let builder = MariaDB::builder()
            .config(test_config())
            .local_stack(test_local_stack());
        let context = Context::with_environment(Environment::Local);
        let resource = builder
            .build(&context)
            .expect("resource build should succeed");

        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("failed to create runtime");
        let connection: String = runtime
            .block_on(resource.into_resource(&context))
            .expect("into_resource should return a connection string");

        assert!(connection.starts_with("mysql://"));
    }

    #[test]
    #[ignore = "requires a local Docker daemon and compose stack"]
    fn applies_database_name_override() {
        if !docker_available() {
            eprintln!("skipping docker-backed test because Docker is not available");
            return;
        }

        let builder = MariaDB::builder()
            .config(test_config())
            .local_stack(test_local_stack())
            .database_name("custom_db");
        let context = Context::with_environment(Environment::Local);
        let resource = builder
            .build(&context)
            .expect("resource build should succeed");

        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("failed to create runtime");
        let connection: String = runtime
            .block_on(resource.into_resource(&context))
            .expect("into_resource should return a connection string");

        assert!(connection.ends_with("/custom_db"));
    }
}