ch-rds 0.1.3

use std::fs::{self, OpenOptions};
use std::io::{Read, Write};
use std::net::{TcpStream, ToSocketAddrs};
use std::path::{Component, Path, PathBuf};
use std::process::{Command, Output, Stdio};
use std::thread;
use std::time::{Duration, Instant};

use crate::__internals::Environment;
use crate::LocalStack;
use anyhow::{Context, Result, anyhow};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use sqlx::{Connection, MySqlConnection};
use tracing::{debug, info, warn};
use url::Url;

const DOCKER_INFO_TIMEOUT: Duration = Duration::from_secs(15);
const COMMAND_POLL_INTERVAL: Duration = Duration::from_millis(200);
const LOCAL_DATABASE_PROBE_TIMEOUT: Duration = Duration::from_millis(300);
const LOCAL_SERVICE_PROTOCOL_TIMEOUT: Duration = Duration::from_secs(2);
const LOCAL_SERVICE_READY_TIMEOUT: Duration = Duration::from_secs(30);
const LOCAL_SERVICE_HEALTH_TIMEOUT: Duration = Duration::from_secs(120);
const LOCAL_SERVICE_RECOVERY_LOCK_TIMEOUT: Duration = Duration::from_secs(330);
const LOCAL_SERVICE_BROKEN_PROBE_LIMIT: usize = 10;
const SPACETIME_MODULE_READY_TIMEOUT: Duration = Duration::from_secs(30);
const SPACETIME_PUBLISH_TIMEOUT: Duration = Duration::from_secs(300);
const SPACETIME_MODULE_BOOTSTRAP_LOCK_TIMEOUT: Duration = Duration::from_secs(330);
#[cfg(any(target_os = "windows", target_os = "macos"))]
const DOCKER_DESKTOP_READY_TIMEOUT: Duration = Duration::from_secs(180);

/// Returns true when the runtime should attempt to start the local service stack.
pub(crate) fn should_bootstrap_local_stack(environment: Environment, connection_url: &str) -> bool {
    if !environment.allows_local_resources() {
        return false;
    }

    if std::env::var_os("CI").is_some() {
        debug!("skipping compose stack bootstrap in CI environment");
        return false;
    }

    if !is_local_database(connection_url) {
        debug!("service url points to remote host; skipping compose stack bootstrap");
        return false;
    }

    if cfg!(target_os = "linux") {
        debug!("skipping compose stack bootstrap on Linux; expecting local services");
        return false;
    }

    true
}

/// Ensures the Docker compose stack used for local development is running.
pub(crate) async fn ensure_local_stack(
    local_stack: &LocalStack,
    connection_url: &str,
) -> Result<()> {
    ensure_local_stack_with_profiles(local_stack, &[], Some(connection_url)).await
}

/// Ensures the Docker compose stack is running with additional optional profiles enabled.
pub(crate) async fn ensure_local_stack_with_profiles(
    local_stack: &LocalStack,
    profiles: &[&str],
    spacetime_endpoint_url: Option<&str>,
) -> Result<()> {
    let lifecycle_lock = local_stack_lifecycle_lock(local_stack)?;
    ensure_docker_desktop_running()?;

    let compose_path = bundled_compose_path();
    if !compose_path.exists() {
        return Err(anyhow!(
            "bundled compose file not found at {}",
            compose_path.display()
        ));
    }

    let project_name = compose_project_name(local_stack.project_name());
    let env_path = local_stack.env_path();

    if !env_path.exists() {
        return Err(anyhow!(
            "local stack env file not found at {}",
            env_path.display()
        ));
    }

    info!("ensuring development compose stack is running");
    run_docker_compose_up(&compose_path, &project_name, env_path, profiles).with_context(|| {
        format!(
            "failed to run docker compose up using {}",
            compose_path.display()
        )
    })?;
    let fallback_spacetime_endpoint;
    let service = if profiles.contains(&"spacetime") {
        let endpoint_url = match spacetime_endpoint_url {
            Some(endpoint_url) => endpoint_url,
            None => {
                let port = resolve_spacetime_compose_port(env_path);
                fallback_spacetime_endpoint = format!("http://127.0.0.1:{port}");
                fallback_spacetime_endpoint.as_str()
            }
        };
        LocalService::SpacetimeDb(endpoint_url)
    } else {
        let connection_url = spacetime_endpoint_url.ok_or_else(|| {
            anyhow!("local MariaDB protocol readiness requires its connection URL")
        })?;
        LocalService::MariaDb(connection_url)
    };

    ensure_local_service_ready(&compose_path, &project_name, env_path, profiles, service).await?;
    drop(lifecycle_lock);

    Ok(())
}

fn local_stack_lifecycle_lock(local_stack: &LocalStack) -> Result<fs::File> {
    let home = std::env::var_os("USERPROFILE")
        .or_else(|| std::env::var_os("HOME"))
        .ok_or_else(|| anyhow!("failed to resolve home directory for local-stack lifecycle"))?;
    let directory = PathBuf::from(home)
        .join(".config")
        .join("ch-rds")
        .join("local-stacks")
        .join(compose_project_name(local_stack.project_name()));
    fs::create_dir_all(&directory).with_context(|| {
        format!(
            "failed to create local-stack lifecycle directory at {}",
            directory.display()
        )
    })?;
    let path = directory.join("lifecycle.lock");
    let file = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(&path)
        .with_context(|| {
            format!(
                "failed to open local-stack lifecycle lock at {}",
                path.display()
            )
        })?;
    lock_file_until(
        &file,
        LOCAL_SERVICE_RECOVERY_LOCK_TIMEOUT,
        "local-stack lifecycle",
    )?;
    Ok(file)
}

/// Ensures the app-owned local SpacetimeDB module exists on the local server.
pub(crate) fn ensure_local_spacetime_module(
    endpoint_url: &str,
    module_name: &str,
    module_path: &Path,
) -> Result<Option<String>> {
    if !is_local_database(endpoint_url) {
        debug!(
            spacetime_url = endpoint_url,
            "spacetime endpoint is not local; skipping module bootstrap"
        );
        return Ok(None);
    }

    if !module_path.is_dir() {
        return Err(anyhow!(
            "SpacetimeDB module directory does not exist at {}",
            module_path.display()
        ));
    }

    let config_dir = spacetime_docker_config_dir(endpoint_url)?;
    fs::create_dir_all(&config_dir).with_context(|| {
        format!(
            "failed to create SpacetimeDB module bootstrap lock directory at {}",
            config_dir.display()
        )
    })?;
    let bootstrap_lock_path = config_dir.join("module-bootstrap.lock");
    let bootstrap_lock = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(&bootstrap_lock_path)
        .with_context(|| {
            format!(
                "failed to open SpacetimeDB module bootstrap lock at {}",
                bootstrap_lock_path.display()
            )
        })?;
    lock_file_until(
        &bootstrap_lock,
        SPACETIME_MODULE_BOOTSTRAP_LOCK_TIMEOUT,
        "local SpacetimeDB module bootstrap",
    )?;

    let module_fingerprint = spacetime_module_fingerprint(module_path)?;
    let module_status = wait_for_spacetime_module_status(endpoint_url, module_name)?;
    let publish_reason = match module_status {
        SpacetimeModuleStatus::Missing => "missing",
        SpacetimeModuleStatus::Available
            if spacetime_module_fingerprint_current(
                endpoint_url,
                module_name,
                module_fingerprint.as_str(),
            )? =>
        {
            let credentials = load_local_spacetime_owner_credentials(endpoint_url, false)?;
            write_spacetime_cli_token(endpoint_url, credentials.token.as_str())?;
            return Ok(Some(credentials.token));
        }
        SpacetimeModuleStatus::Available => "source fingerprint changed",
    };

    info!(
        spacetime_url = endpoint_url,
        module_name,
        module_path = %module_path.display(),
        publish_reason,
        "preparing local SpacetimeDB module"
    );
    // Module updates stay in the existing standalone database. The publish
    // command owns compatible migration and its on-conflict data policy;
    // replacing the server volume here would also rotate local identity state.
    let credentials = load_local_spacetime_owner_credentials(
        endpoint_url,
        matches!(module_status, SpacetimeModuleStatus::Missing),
    )?;
    write_spacetime_cli_token(endpoint_url, credentials.token.as_str())?;
    publish_local_spacetime_module(endpoint_url, module_name, module_path)?;

    let deadline = Instant::now() + SPACETIME_MODULE_READY_TIMEOUT;
    loop {
        match spacetime_module_status(endpoint_url, module_name)? {
            SpacetimeModuleStatus::Available => {
                write_spacetime_module_fingerprint(
                    endpoint_url,
                    module_name,
                    module_fingerprint.as_str(),
                )?;
                return Ok(Some(credentials.token));
            }
            SpacetimeModuleStatus::Missing if Instant::now() < deadline => {
                thread::sleep(COMMAND_POLL_INTERVAL);
            }
            SpacetimeModuleStatus::Missing => {
                return Err(anyhow!(
                    "SpacetimeDB module `{module_name}` was published but is still missing at {endpoint_url}"
                ));
            }
        }
    }
}

fn lock_file_until(file: &fs::File, timeout: Duration, operation: &str) -> Result<()> {
    let deadline = Instant::now() + timeout;
    loop {
        match file.try_lock() {
            Ok(()) => return Ok(()),
            Err(fs::TryLockError::WouldBlock) if Instant::now() < deadline => {
                thread::sleep(COMMAND_POLL_INTERVAL);
            }
            Err(fs::TryLockError::WouldBlock) => {
                return Err(anyhow!("timed out waiting for {operation} lock"));
            }
            Err(fs::TryLockError::Error(error)) => {
                return Err(error).with_context(|| format!("failed to acquire {operation} lock"));
            }
        }
    }
}

fn run_docker_compose_up(
    compose_path: &Path,
    project_name: &str,
    env_path: &Path,
    profiles: &[&str],
) -> Result<()> {
    let mut command = docker_compose_command(compose_path, project_name, env_path, profiles);
    let output = command
        .arg("up")
        .arg("--detach")
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .output()
        .with_context(|| "failed to spawn docker compose")?;

    if output.status.success() {
        return Ok(());
    }

    let stdout = String::from_utf8_lossy(&output.stdout).trim().to_string();
    let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
    let mut detail = format!("docker compose exited with {}", output.status);
    if !stdout.is_empty() {
        detail.push_str("\nstdout:\n");
        detail.push_str(&stdout);
    }
    if !stderr.is_empty() {
        detail.push_str("\nstderr:\n");
        detail.push_str(&stderr);
    }

    Err(anyhow!(detail))
}

fn docker_compose_command(
    compose_path: &Path,
    project_name: &str,
    env_path: &Path,
    profiles: &[&str],
) -> Command {
    let mut command = Command::new("docker");
    command
        .arg("compose")
        .arg("--file")
        .arg(compose_path)
        .arg("--project-name")
        .arg(project_name);
    if env_path.exists() {
        command.arg("--env-file").arg(env_path);
    }
    for profile in profiles {
        command.arg("--profile").arg(profile);
    }
    command
}

#[derive(Debug, Clone, Copy)]
enum LocalService<'a> {
    MariaDb(&'a str),
    SpacetimeDb(&'a str),
}

impl<'a> LocalService<'a> {
    fn compose_name(self) -> &'static str {
        match self {
            Self::MariaDb(_) => "mariadb",
            Self::SpacetimeDb(_) => "spacetimedb",
        }
    }

    fn display_name(self) -> &'static str {
        match self {
            Self::MariaDb(_) => "MariaDB",
            Self::SpacetimeDb(_) => "SpacetimeDB",
        }
    }

    fn url(self) -> &'a str {
        match self {
            Self::MariaDb(url) | Self::SpacetimeDb(url) => url,
        }
    }

    fn display_endpoint(self) -> String {
        match self {
            Self::MariaDb(url) => host_and_port_from_connection_url(url)
                .map(|(host, port)| format!("{host}:{port}"))
                .unwrap_or_else(|| "<invalid local MariaDB URL>".to_string()),
            Self::SpacetimeDb(url) => url.to_string(),
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ProtocolProbeFailureKind {
    Starting,
    Broken,
    Fatal,
}

#[derive(Debug)]
struct ProtocolProbeFailure {
    kind: ProtocolProbeFailureKind,
    error: anyhow::Error,
}

enum ServiceWaitFailure {
    Restart(anyhow::Error),
    Fatal(anyhow::Error),
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "PascalCase")]
struct ComposeServiceStatus {
    service: String,
    state: String,
    #[serde(default)]
    health: String,
    #[serde(default)]
    exit_code: i32,
}

async fn ensure_local_service_ready(
    compose_path: &Path,
    project_name: &str,
    env_path: &Path,
    profiles: &[&str],
    service: LocalService<'_>,
) -> Result<()> {
    wait_for_compose_service_health(compose_path, project_name, env_path, profiles, service)
        .await?;

    match wait_for_local_service_protocol(service, true).await {
        Ok(()) => return Ok(()),
        Err(ServiceWaitFailure::Fatal(error)) => {
            return Err(local_service_ready_error(service, error, false));
        }
        Err(ServiceWaitFailure::Restart(error)) => {
            warn!(
                service = service.display_name(),
                %error,
                "local service accepts connections but is not protocol-ready; restarting it once"
            );
        }
    }

    restart_local_service(
        compose_path,
        project_name,
        env_path,
        profiles,
        service.compose_name(),
    )
    .with_context(|| format!("failed to restart local {}", service.display_name()))?;

    wait_for_compose_service_health(compose_path, project_name, env_path, profiles, service)
        .await?;

    wait_for_local_service_protocol(service, false)
        .await
        .map_err(|failure| match failure {
            ServiceWaitFailure::Restart(error) | ServiceWaitFailure::Fatal(error) => {
                local_service_ready_error(service, error, true)
            }
        })
}

async fn wait_for_compose_service_health(
    compose_path: &Path,
    project_name: &str,
    env_path: &Path,
    profiles: &[&str],
    service: LocalService<'_>,
) -> Result<()> {
    let deadline = Instant::now() + LOCAL_SERVICE_HEALTH_TIMEOUT;

    loop {
        let observation = match compose_service_status(
            compose_path,
            project_name,
            env_path,
            profiles,
            service.compose_name(),
        )? {
            Some(status)
                if status.state.eq_ignore_ascii_case("running")
                    && status.health.eq_ignore_ascii_case("healthy") =>
            {
                return Ok(());
            }
            Some(status)
                if ["exited", "dead", "paused", "removing"]
                    .iter()
                    .any(|state| status.state.eq_ignore_ascii_case(state)) =>
            {
                return Err(local_service_health_error(
                    service,
                    format!(
                        "container state is `{}` with exit code {}",
                        status.state, status.exit_code
                    ),
                ));
            }
            Some(status) => match status.health.trim() {
                "" => format!(
                    "container state is `{}` and no health result is available yet",
                    status.state
                ),
                health => format!(
                    "container state is `{}` and health is `{health}`",
                    status.state
                ),
            },
            None => "Compose service has no container yet".to_string(),
        };

        if Instant::now() >= deadline {
            return Err(local_service_health_error(service, observation));
        }

        debug!(
            service = service.display_name(),
            observation, "local service container is not healthy yet; retrying"
        );
        tokio::time::sleep(COMMAND_POLL_INTERVAL).await;
    }
}

fn compose_service_status(
    compose_path: &Path,
    project_name: &str,
    env_path: &Path,
    profiles: &[&str],
    service_name: &str,
) -> Result<Option<ComposeServiceStatus>> {
    let mut command = docker_compose_command(compose_path, project_name, env_path, profiles);
    command
        .arg("ps")
        .arg("--all")
        .arg("--format")
        .arg("json")
        .arg(service_name)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    let output = run_command_with_timeout(
        &mut command,
        DOCKER_INFO_TIMEOUT,
        "`docker compose ps --format json`",
    )?;
    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
        return Err(anyhow!(
            "docker compose ps for `{service_name}` exited with {}: {stderr}",
            output.status
        ));
    }

    let statuses = parse_compose_service_statuses(&output.stdout)?;
    Ok(statuses
        .into_iter()
        .find(|status| status.service == service_name))
}

fn parse_compose_service_statuses(output: &[u8]) -> Result<Vec<ComposeServiceStatus>> {
    let output = String::from_utf8_lossy(output);
    let output = output.trim();
    if output.is_empty() {
        return Ok(Vec::new());
    }
    if output.starts_with('[') {
        return serde_json::from_str(output)
            .context("failed to parse Docker Compose service status");
    }

    serde_json::Deserializer::from_str(output)
        .into_iter::<ComposeServiceStatus>()
        .collect::<std::result::Result<Vec<_>, _>>()
        .context("failed to parse Docker Compose service status")
}

async fn wait_for_local_service_protocol(
    service: LocalService<'_>,
    restart_on_repeated_broken_response: bool,
) -> std::result::Result<(), ServiceWaitFailure> {
    let deadline = Instant::now() + LOCAL_SERVICE_READY_TIMEOUT;
    let mut broken_responses = 0;

    loop {
        let error = match local_service_protocol_probe(service).await {
            Ok(()) => return Ok(()),
            Err(ProtocolProbeFailure {
                kind: ProtocolProbeFailureKind::Fatal,
                error,
            }) => return Err(ServiceWaitFailure::Fatal(error)),
            Err(ProtocolProbeFailure {
                kind: ProtocolProbeFailureKind::Broken,
                error,
            }) => {
                broken_responses += 1;
                if restart_on_repeated_broken_response
                    && broken_responses >= LOCAL_SERVICE_BROKEN_PROBE_LIMIT
                {
                    return Err(ServiceWaitFailure::Restart(error));
                }
                error
            }
            Err(ProtocolProbeFailure {
                kind: ProtocolProbeFailureKind::Starting,
                error,
            }) => {
                broken_responses = 0;
                error
            }
        };

        if Instant::now() >= deadline {
            return Err(ServiceWaitFailure::Restart(error));
        }

        debug!(
            service = service.display_name(),
            "local service protocol is not ready; retrying"
        );
        tokio::time::sleep(COMMAND_POLL_INTERVAL).await;
    }
}

async fn local_service_protocol_probe(
    service: LocalService<'_>,
) -> std::result::Result<(), ProtocolProbeFailure> {
    match service {
        LocalService::MariaDb(connection_url) => mariadb_protocol_probe(connection_url).await,
        LocalService::SpacetimeDb(endpoint_url) => spacetime_protocol_probe(endpoint_url),
    }
}

async fn mariadb_protocol_probe(
    connection_url: &str,
) -> std::result::Result<(), ProtocolProbeFailure> {
    let connect = tokio::time::timeout(
        LOCAL_SERVICE_PROTOCOL_TIMEOUT,
        MySqlConnection::connect(connection_url),
    )
    .await;
    let mut connection = match connect {
        Ok(Ok(connection)) => connection,
        Ok(Err(error)) => return Err(mariadb_protocol_failure(error)),
        Err(error) => {
            return Err(ProtocolProbeFailure {
                kind: ProtocolProbeFailureKind::Starting,
                error: anyhow!(error).context("MariaDB protocol probe timed out"),
            });
        }
    };

    match tokio::time::timeout(
        LOCAL_SERVICE_PROTOCOL_TIMEOUT,
        sqlx::query_scalar::<_, i32>("SELECT 1").fetch_one(&mut connection),
    )
    .await
    {
        Ok(Ok(1)) => {
            let _ = connection.close().await;
            Ok(())
        }
        Ok(Ok(value)) => Err(ProtocolProbeFailure {
            kind: ProtocolProbeFailureKind::Broken,
            error: anyhow!("MariaDB protocol probe returned unexpected value {value}"),
        }),
        Ok(Err(error)) => Err(mariadb_protocol_failure(error)),
        Err(error) => Err(ProtocolProbeFailure {
            kind: ProtocolProbeFailureKind::Broken,
            error: anyhow!(error).context("MariaDB query probe timed out"),
        }),
    }
}

fn mariadb_protocol_failure(error: sqlx::Error) -> ProtocolProbeFailure {
    let kind = mariadb_protocol_failure_kind(&error);
    ProtocolProbeFailure {
        kind,
        error: anyhow!(error).context("MariaDB protocol probe failed"),
    }
}

fn mariadb_protocol_failure_kind(error: &sqlx::Error) -> ProtocolProbeFailureKind {
    match error {
        sqlx::Error::Configuration(_) | sqlx::Error::Database(_) => ProtocolProbeFailureKind::Fatal,
        sqlx::Error::Io(error)
            if matches!(
                error.kind(),
                std::io::ErrorKind::ConnectionRefused
                    | std::io::ErrorKind::TimedOut
                    | std::io::ErrorKind::NotConnected
                    | std::io::ErrorKind::AddrNotAvailable
                    | std::io::ErrorKind::WouldBlock
            ) =>
        {
            ProtocolProbeFailureKind::Starting
        }
        _ => ProtocolProbeFailureKind::Broken,
    }
}

fn spacetime_protocol_probe(endpoint_url: &str) -> std::result::Result<(), ProtocolProbeFailure> {
    match local_spacetime_http_request(endpoint_url, "GET", "/v1/ping", None) {
        Ok((200..=299, _)) => Ok(()),
        Ok((status, _)) => Err(ProtocolProbeFailure {
            kind: ProtocolProbeFailureKind::Broken,
            error: anyhow!("SpacetimeDB readiness probe returned HTTP {status}"),
        }),
        Err(error) => {
            let kind = protocol_io_failure_kind(&error).unwrap_or(ProtocolProbeFailureKind::Broken);
            Err(ProtocolProbeFailure { kind, error })
        }
    }
}

fn protocol_io_failure_kind(error: &anyhow::Error) -> Option<ProtocolProbeFailureKind> {
    let io_error = error
        .chain()
        .find_map(|cause| cause.downcast_ref::<std::io::Error>())?;
    Some(match io_error.kind() {
        std::io::ErrorKind::ConnectionRefused
        | std::io::ErrorKind::TimedOut
        | std::io::ErrorKind::NotConnected
        | std::io::ErrorKind::AddrNotAvailable
        | std::io::ErrorKind::WouldBlock => ProtocolProbeFailureKind::Starting,
        _ => ProtocolProbeFailureKind::Broken,
    })
}

fn restart_local_service(
    compose_path: &Path,
    project_name: &str,
    env_path: &Path,
    profiles: &[&str],
    service_name: &str,
) -> Result<()> {
    let mut command = docker_compose_command(compose_path, project_name, env_path, profiles);
    command
        .arg("restart")
        .arg(service_name)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());

    let output = run_command_with_timeout(
        &mut command,
        LOCAL_SERVICE_READY_TIMEOUT,
        "`docker compose restart`",
    )?;
    if output.status.success() {
        return Ok(());
    }

    let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
    let stdout = String::from_utf8_lossy(&output.stdout).trim().to_string();
    Err(anyhow!(
        "docker compose restart for `{service_name}` exited with {}\nstdout: {stdout}\nstderr: {stderr}",
        output.status
    ))
}

fn local_service_health_error(service: LocalService<'_>, observation: String) -> anyhow::Error {
    let mut message = format!(
        "local {} container did not become healthy within {} seconds: {observation}",
        service.display_name(),
        LOCAL_SERVICE_HEALTH_TIMEOUT.as_secs()
    );
    if matches!(service, LocalService::MariaDb(_)) {
        message.push_str(
            "; for an existing MariaDB volume, verify that its stored users still match \
             services.env; ch-rds did not remove or recreate the volume",
        );
    }
    anyhow!(message)
}

fn local_service_ready_error(
    service: LocalService<'_>,
    error: anyhow::Error,
    recovery_attempted: bool,
) -> anyhow::Error {
    let recovery = if recovery_attempted {
        " after one controlled recovery"
    } else {
        ""
    };
    let mut message = format!(
        "local {} endpoint {} did not become protocol-ready{recovery}: {error}",
        service.display_name(),
        service.display_endpoint(),
    );
    if matches!(service, LocalService::MariaDb(_)) && mariadb_credentials_rejected(&error) {
        message.push_str(
            "; MariaDB rejected the configured credentials; an existing local volume may have \
             been initialized with different services.env values, and ch-rds did not modify it",
        );
    }
    if matches!(service, LocalService::SpacetimeDb(_))
        && let Some(diagnostic) = local_spacetime_endpoint_diagnostic(service.url())
    {
        message.push_str("; ");
        message.push_str(diagnostic.as_str());
    }
    anyhow!(message)
}

fn mariadb_credentials_rejected(error: &anyhow::Error) -> bool {
    error.chain().any(|cause| {
        let Some(sqlx::Error::Database(database_error)) = cause.downcast_ref::<sqlx::Error>()
        else {
            return false;
        };
        database_error
            .code()
            .is_some_and(|code| code.eq_ignore_ascii_case("28000"))
    })
}

fn resolve_spacetime_compose_port(env_path: &Path) -> u16 {
    if let Ok(raw) = std::env::var("SPACETIME_PORT")
        && let Some(port) = parse_spacetime_port(raw.as_str())
    {
        return port;
    }

    fs::read_to_string(env_path)
        .ok()
        .and_then(|contents| parse_spacetime_compose_port(contents.as_str()))
        .unwrap_or(4530)
}

fn parse_spacetime_compose_port(contents: &str) -> Option<u16> {
    contents.lines().find_map(|line| {
        let trimmed = line.trim();
        if trimmed.is_empty() || trimmed.starts_with('#') {
            return None;
        }

        let (key, value) = trimmed.split_once('=')?;
        (key.trim() == "SPACETIME_PORT")
            .then(|| parse_spacetime_port(value))
            .flatten()
    })
}

fn parse_spacetime_port(value: &str) -> Option<u16> {
    let value = value
        .trim()
        .trim_matches(|character| character == '"' || character == '\'');
    value.parse::<u16>().ok().filter(|port| *port > 0)
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SpacetimeModuleStatus {
    Available,
    Missing,
}

fn wait_for_spacetime_module_status(
    endpoint_url: &str,
    module_name: &str,
) -> Result<SpacetimeModuleStatus> {
    let deadline = Instant::now() + SPACETIME_MODULE_READY_TIMEOUT;

    loop {
        match spacetime_module_status(endpoint_url, module_name) {
            Ok(status) => return Ok(status),
            Err(error) if Instant::now() < deadline => {
                debug!(
                    %error,
                    "local SpacetimeDB endpoint is not ready for module check; retrying"
                );
                thread::sleep(COMMAND_POLL_INTERVAL);
            }
            Err(error) => {
                return Err(spacetime_module_ready_error(endpoint_url, error));
            }
        }
    }
}

fn spacetime_module_ready_error(endpoint_url: &str, error: anyhow::Error) -> anyhow::Error {
    let mut message = format!(
        "local SpacetimeDB endpoint {endpoint_url} did not become ready for module check: {error}"
    );
    if let Some(diagnostic) = local_spacetime_endpoint_diagnostic(endpoint_url) {
        message.push_str("; ");
        message.push_str(diagnostic.as_str());
    }
    anyhow!(message)
}

fn local_spacetime_endpoint_diagnostic(endpoint_url: &str) -> Option<String> {
    let (host, port) = host_and_port_from_connection_url(endpoint_url)?;
    if !is_loopback(host.as_str()) {
        return None;
    }

    windows_tcp_excluded_port_diagnostic(port)
}

#[cfg(target_os = "windows")]
fn windows_tcp_excluded_port_diagnostic(port: u16) -> Option<String> {
    let output = Command::new("netsh")
        .arg("interface")
        .arg("ipv4")
        .arg("show")
        .arg("excludedportrange")
        .arg("protocol=tcp")
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .output()
        .ok()?;
    if !output.status.success() {
        return None;
    }

    let stdout = String::from_utf8_lossy(&output.stdout);
    let (start, end) = windows_excluded_tcp_port_range(stdout.as_ref(), port)?;
    Some(format!(
        "Windows TCP port {port} is inside excluded range {start}-{end}; Docker cannot publish the local SpacetimeDB host port until that reservation is repaired with elevated Windows network permissions"
    ))
}

#[cfg(not(target_os = "windows"))]
fn windows_tcp_excluded_port_diagnostic(_port: u16) -> Option<String> {
    None
}

#[cfg(any(target_os = "windows", test))]
fn windows_excluded_tcp_port_range(output: &str, port: u16) -> Option<(u16, u16)> {
    output.lines().find_map(|line| {
        let mut columns = line.split_whitespace();
        let start = columns.next()?.parse::<u16>().ok()?;
        let end = columns.next()?.parse::<u16>().ok()?;
        (start <= port && port <= end).then_some((start, end))
    })
}

fn spacetime_module_fingerprint(module_path: &Path) -> Result<String> {
    let mut files = Vec::new();
    collect_spacetime_module_fingerprint_files(module_path, module_path, &mut files)?;
    files.sort();

    let mut hasher = Sha256::new();
    hasher.update([1]);
    for file in files {
        let relative = file.strip_prefix(module_path).with_context(|| {
            format!(
                "failed to resolve SpacetimeDB module file {} relative to {}",
                file.display(),
                module_path.display()
            )
        })?;
        let relative = relative.to_string_lossy().replace('\\', "/");
        hasher.update(relative.as_bytes());
        hasher.update([0]);
        let contents = fs::read(&file).with_context(|| {
            format!(
                "failed to read SpacetimeDB module fingerprint file {}",
                file.display()
            )
        })?;
        hasher.update(contents);
        hasher.update([0]);
    }

    Ok(hex_digest(&hasher.finalize()))
}

fn collect_spacetime_module_fingerprint_files(
    root: &Path,
    current: &Path,
    files: &mut Vec<PathBuf>,
) -> Result<()> {
    for entry in fs::read_dir(current).with_context(|| {
        format!(
            "failed to read SpacetimeDB module fingerprint directory {}",
            current.display()
        )
    })? {
        let entry = entry?;
        let path = entry.path();
        if path.is_dir() {
            if should_skip_spacetime_module_fingerprint_dir(root, &path) {
                continue;
            }
            collect_spacetime_module_fingerprint_files(root, &path, files)?;
            continue;
        }

        if should_include_spacetime_module_fingerprint_file(root, &path) {
            files.push(path);
        }
    }

    Ok(())
}

fn should_skip_spacetime_module_fingerprint_dir(root: &Path, path: &Path) -> bool {
    let Ok(relative) = path.strip_prefix(root) else {
        return true;
    };
    if relative == Path::new("target") || relative == Path::new(".git") {
        return true;
    }

    relative == Path::new("src").join("generated")
}

fn should_include_spacetime_module_fingerprint_file(root: &Path, path: &Path) -> bool {
    let Ok(relative) = path.strip_prefix(root) else {
        return false;
    };
    if relative == Path::new("Cargo.toml") || relative == Path::new("Cargo.lock") {
        return true;
    }

    relative
        .components()
        .next()
        .is_some_and(|component| component.as_os_str() == "src")
        && path
            .extension()
            .and_then(|extension| extension.to_str())
            .is_some_and(|extension| extension == "rs")
}

fn spacetime_module_fingerprint_current(
    endpoint_url: &str,
    module_name: &str,
    module_fingerprint: &str,
) -> Result<bool> {
    let path = spacetime_module_fingerprint_path(endpoint_url, module_name)?;
    let Ok(stored) = fs::read_to_string(path) else {
        return Ok(false);
    };

    Ok(stored.trim() == module_fingerprint)
}

fn write_spacetime_module_fingerprint(
    endpoint_url: &str,
    module_name: &str,
    module_fingerprint: &str,
) -> Result<()> {
    let path = spacetime_module_fingerprint_path(endpoint_url, module_name)?;
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent).with_context(|| {
            format!(
                "failed to create SpacetimeDB module fingerprint directory {}",
                parent.display()
            )
        })?;
    }

    fs::write(&path, module_fingerprint).with_context(|| {
        format!(
            "failed to write SpacetimeDB module fingerprint at {}",
            path.display()
        )
    })
}

fn spacetime_module_fingerprint_path(endpoint_url: &str, module_name: &str) -> Result<PathBuf> {
    let endpoint_slug = spacetime_endpoint_slug(endpoint_url)?;
    let safe_module_name = safe_path_segment(module_name);
    let home = std::env::var_os("USERPROFILE")
        .or_else(|| std::env::var_os("HOME"))
        .ok_or_else(|| {
            anyhow!("failed to resolve home directory for SpacetimeDB module fingerprint")
        })?;
    Ok(PathBuf::from(home)
        .join(".config")
        .join("spacetimedb-module-fingerprints")
        .join(endpoint_slug)
        .join(format!("{safe_module_name}.sha256")))
}

fn spacetime_module_status(endpoint_url: &str, module_name: &str) -> Result<SpacetimeModuleStatus> {
    let (mut stream, host, port) = connect_local_spacetime_http(endpoint_url, "schema probe")?;

    let path = format!("/v1/database/{module_name}/schema?version=9");
    let host_header = host_header(&host, port);
    let request =
        format!("GET {path} HTTP/1.1\r\nHost: {host_header}\r\nConnection: close\r\n\r\n");
    stream
        .write_all(request.as_bytes())
        .with_context(|| "failed to write SpacetimeDB schema probe request")?;

    let mut response = String::new();
    stream
        .read_to_string(&mut response)
        .with_context(|| "failed to read SpacetimeDB schema probe response")?;
    let status_line = response
        .lines()
        .next()
        .ok_or_else(|| anyhow!("SpacetimeDB schema probe returned an empty response"))?;
    let status_code = status_line
        .split_whitespace()
        .nth(1)
        .ok_or_else(|| anyhow!("SpacetimeDB schema probe returned invalid status: {status_line}"))?
        .parse::<u16>()
        .with_context(|| {
            format!("SpacetimeDB schema probe returned invalid status: {status_line}")
        })?;

    match status_code {
        200..=299 | 401 | 403 => Ok(SpacetimeModuleStatus::Available),
        404 => Ok(SpacetimeModuleStatus::Missing),
        code => Err(anyhow!(
            "SpacetimeDB schema probe for `{module_name}` returned HTTP {code}"
        )),
    }
}

fn connect_local_http(host: &str, port: u16) -> Result<TcpStream> {
    let mut candidates = Vec::new();
    if let Ok(ip) = host.parse::<std::net::IpAddr>() {
        candidates.push(std::net::SocketAddr::new(ip, port));
    } else if let Ok(resolved) = (host, port).to_socket_addrs() {
        candidates.extend(resolved);
    }

    let mut last_error = None;
    for addr in candidates {
        match TcpStream::connect_timeout(&addr, LOCAL_DATABASE_PROBE_TIMEOUT) {
            Ok(stream) => return Ok(stream),
            Err(error) => last_error = Some(error),
        }
    }

    Err(anyhow!(
        "failed to connect to local SpacetimeDB endpoint {host}:{port}: {}",
        last_error
            .map(|error| error.to_string())
            .unwrap_or_else(|| "host did not resolve".to_string())
    ))
}

fn host_header(host: &str, port: u16) -> String {
    if host.contains(':') && !host.starts_with('[') {
        format!("[{host}]:{port}")
    } else {
        format!("{host}:{port}")
    }
}

#[derive(Debug, Deserialize, Serialize)]
struct LocalSpacetimeOwnerCredentials {
    identity: String,
    token: String,
}

fn load_local_spacetime_owner_credentials(
    endpoint_url: &str,
    allow_create: bool,
) -> Result<LocalSpacetimeOwnerCredentials> {
    let credential_path = spacetime_docker_config_dir(endpoint_url)?.join("module-owner.json");
    if credential_path.is_file() {
        let bytes = fs::read(&credential_path).with_context(|| {
            format!(
                "failed to read local SpacetimeDB module-owner credentials at {}",
                credential_path.display()
            )
        })?;
        let credentials = serde_json::from_slice::<LocalSpacetimeOwnerCredentials>(&bytes)
            .with_context(|| {
                format!(
                    "failed to parse local SpacetimeDB module-owner credentials at {}",
                    credential_path.display()
                )
            })?;
        if verify_local_spacetime_identity(endpoint_url, &credentials)? {
            return Ok(credentials);
        }
        if !allow_create {
            return Err(anyhow!(
                "the saved local SpacetimeDB module-owner credential is no longer valid; the local module cannot be opened with a replacement identity"
            ));
        }
    } else if !allow_create {
        return Err(anyhow!(
            "the local SpacetimeDB module exists without its module-owner credential; reset the local SpacetimeDB development volume before continuing"
        ));
    }

    let credentials = create_local_spacetime_identity(endpoint_url)?;
    let parent = credential_path.parent().ok_or_else(|| {
        anyhow!(
            "local SpacetimeDB credential path has no parent: {}",
            credential_path.display()
        )
    })?;
    fs::create_dir_all(parent).with_context(|| {
        format!(
            "failed to create local SpacetimeDB credential directory at {}",
            parent.display()
        )
    })?;
    let encoded = serde_json::to_vec(&credentials)
        .with_context(|| "failed to encode local SpacetimeDB module-owner credentials")?;
    let temporary_path = credential_path.with_extension("json.tmp");
    write_private_file(&temporary_path, &encoded)?;
    if credential_path.exists() {
        fs::remove_file(&credential_path).with_context(|| {
            format!(
                "failed to replace invalid local SpacetimeDB module-owner credentials at {}",
                credential_path.display()
            )
        })?;
    }
    fs::rename(&temporary_path, &credential_path).with_context(|| {
        format!(
            "failed to persist local SpacetimeDB module-owner credentials at {}",
            credential_path.display()
        )
    })?;
    Ok(credentials)
}

fn create_local_spacetime_identity(endpoint_url: &str) -> Result<LocalSpacetimeOwnerCredentials> {
    let (status, body) = local_spacetime_http_request(endpoint_url, "POST", "/v1/identity", None)?;
    if !(200..=299).contains(&status) {
        return Err(anyhow!(
            "local SpacetimeDB identity creation returned HTTP {status}"
        ));
    }
    let credentials = serde_json::from_slice::<LocalSpacetimeOwnerCredentials>(&body)
        .with_context(|| "local SpacetimeDB identity creation returned invalid JSON")?;
    if credentials.identity.trim().is_empty() || credentials.token.trim().is_empty() {
        return Err(anyhow!(
            "local SpacetimeDB identity creation returned empty credentials"
        ));
    }
    Ok(credentials)
}

fn verify_local_spacetime_identity(
    endpoint_url: &str,
    credentials: &LocalSpacetimeOwnerCredentials,
) -> Result<bool> {
    if credentials.identity.trim().is_empty() || credentials.token.trim().is_empty() {
        return Ok(false);
    }
    let path = format!("/v1/identity/{}/verify", credentials.identity.trim());
    let (status, _) = local_spacetime_http_request(
        endpoint_url,
        "GET",
        path.as_str(),
        Some(credentials.token.trim()),
    )?;
    match status {
        200..=299 => Ok(true),
        400 | 401 | 403 | 404 => Ok(false),
        _ => Err(anyhow!(
            "local SpacetimeDB identity verification returned HTTP {status}"
        )),
    }
}

fn local_spacetime_http_request(
    endpoint_url: &str,
    method: &str,
    path: &str,
    bearer_token: Option<&str>,
) -> Result<(u16, Vec<u8>)> {
    let (mut stream, host, port) = connect_local_spacetime_http(endpoint_url, "identity request")?;

    let mut request = format!(
        "{method} {path} HTTP/1.1\r\nHost: {}\r\nAccept: application/json\r\nContent-Length: 0\r\nConnection: close\r\n",
        host_header(&host, port)
    );
    if let Some(token) = bearer_token {
        request.push_str("Authorization: Bearer ");
        request.push_str(token);
        request.push_str("\r\n");
    }
    request.push_str("\r\n");
    stream
        .write_all(request.as_bytes())
        .with_context(|| "failed to write local SpacetimeDB identity request")?;

    let mut response = Vec::new();
    stream
        .read_to_end(&mut response)
        .with_context(|| "failed to read local SpacetimeDB identity response")?;
    parse_http_response(&response)
}

fn connect_local_spacetime_http(
    endpoint_url: &str,
    operation: &'static str,
) -> Result<(TcpStream, String, u16)> {
    let parsed = Url::parse(endpoint_url)
        .with_context(|| format!("invalid SpacetimeDB endpoint URL: {endpoint_url}"))?;
    if parsed.scheme() != "http" {
        return Err(anyhow!(
            "local SpacetimeDB {operation} requires an http endpoint, got {endpoint_url}"
        ));
    }
    let host = parsed
        .host_str()
        .ok_or_else(|| anyhow!("SpacetimeDB endpoint URL has no host: {endpoint_url}"))?;
    let port = parsed
        .port_or_known_default()
        .ok_or_else(|| anyhow!("SpacetimeDB endpoint URL has no port: {endpoint_url}"))?;
    let stream = connect_local_http(host, port)?;
    stream
        .set_read_timeout(Some(Duration::from_secs(2)))
        .with_context(|| format!("failed to set local SpacetimeDB {operation} read timeout"))?;
    stream
        .set_write_timeout(Some(Duration::from_secs(2)))
        .with_context(|| format!("failed to set local SpacetimeDB {operation} write timeout"))?;

    Ok((stream, host.to_string(), port))
}

fn parse_http_response(response: &[u8]) -> Result<(u16, Vec<u8>)> {
    let header_end = response
        .windows(4)
        .position(|window| window == b"\r\n\r\n")
        .ok_or_else(|| anyhow!("local SpacetimeDB returned an invalid HTTP response"))?;
    let headers = std::str::from_utf8(&response[..header_end])
        .with_context(|| "local SpacetimeDB returned non-UTF-8 HTTP headers")?;
    let status_line = headers
        .lines()
        .next()
        .ok_or_else(|| anyhow!("local SpacetimeDB returned an empty HTTP response"))?;
    let status = status_line
        .split_whitespace()
        .nth(1)
        .ok_or_else(|| anyhow!("local SpacetimeDB returned invalid HTTP status: {status_line}"))?
        .parse::<u16>()
        .with_context(|| {
            format!("local SpacetimeDB returned invalid HTTP status: {status_line}")
        })?;
    let body = &response[header_end + 4..];
    if headers.lines().any(|line| {
        line.split_once(':').is_some_and(|(name, value)| {
            name.eq_ignore_ascii_case("transfer-encoding")
                && value
                    .split(',')
                    .any(|encoding| encoding.trim().eq_ignore_ascii_case("chunked"))
        })
    }) {
        return Ok((status, decode_chunked_http_body(body)?));
    }
    Ok((status, body.to_vec()))
}

fn decode_chunked_http_body(mut body: &[u8]) -> Result<Vec<u8>> {
    let mut decoded = Vec::new();
    loop {
        let line_end = body
            .windows(2)
            .position(|window| window == b"\r\n")
            .ok_or_else(|| anyhow!("local SpacetimeDB returned an invalid chunked response"))?;
        let size = std::str::from_utf8(&body[..line_end])
            .with_context(|| "local SpacetimeDB returned a non-UTF-8 chunk size")?
            .split(';')
            .next()
            .unwrap_or_default()
            .trim();
        let size = usize::from_str_radix(size, 16)
            .with_context(|| "local SpacetimeDB returned an invalid chunk size")?;
        body = &body[line_end + 2..];
        if size == 0 {
            return Ok(decoded);
        }
        if body.len() < size + 2 || &body[size..size + 2] != b"\r\n" {
            return Err(anyhow!(
                "local SpacetimeDB returned a truncated chunked response"
            ));
        }
        decoded.extend_from_slice(&body[..size]);
        body = &body[size + 2..];
    }
}

fn publish_local_spacetime_module(
    endpoint_url: &str,
    module_name: &str,
    module_path: &Path,
) -> Result<()> {
    if command_available("spacetime") {
        let mut host_command =
            host_spacetime_publish_command(endpoint_url, module_name, module_path)?;
        let host_output = run_command_with_timeout(
            &mut host_command,
            SPACETIME_PUBLISH_TIMEOUT,
            "`spacetime publish`",
        )?;
        if host_output.status.success() {
            return Ok(());
        }

        let host_stdout = String::from_utf8_lossy(&host_output.stdout)
            .trim()
            .to_string();
        let host_stderr = String::from_utf8_lossy(&host_output.stderr)
            .trim()
            .to_string();
        if should_retry_spacetime_publish_with_docker(&host_stdout, &host_stderr) {
            let mut docker_command =
                docker_spacetime_publish_command(endpoint_url, module_name, module_path)?;
            let docker_output = run_command_with_timeout(
                &mut docker_command,
                SPACETIME_PUBLISH_TIMEOUT,
                "`spacetime publish` (docker fallback)",
            )?;
            if docker_output.status.success() {
                return Ok(());
            }

            return Err(anyhow!(
                "{}\n\ndocker fallback:\n{}",
                format_spacetime_publish_failure(&host_output),
                format_spacetime_publish_failure(&docker_output),
            ));
        }

        return Err(anyhow!(format_spacetime_publish_failure(&host_output)));
    }

    let mut docker_command =
        docker_spacetime_publish_command(endpoint_url, module_name, module_path)?;
    let docker_output = run_command_with_timeout(
        &mut docker_command,
        SPACETIME_PUBLISH_TIMEOUT,
        "`spacetime publish`",
    )?;
    if docker_output.status.success() {
        return Ok(());
    }

    Err(anyhow!(format_spacetime_publish_failure(&docker_output)))
}

fn should_retry_spacetime_publish_with_docker(stdout: &str, stderr: &str) -> bool {
    let combined = format!("{stdout}\n{stderr}");
    combined.contains("failed to load manifest for dependency")
        && combined.contains("/workspace/")
        && combined.contains("No such file or directory")
}

fn format_spacetime_publish_failure(output: &std::process::Output) -> String {
    let stdout = String::from_utf8_lossy(&output.stdout).trim().to_string();
    let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
    let mut detail = format!("spacetime publish exited with {}", output.status);
    if !stdout.is_empty() {
        detail.push_str("\nstdout:\n");
        detail.push_str(&stdout);
    }
    if !stderr.is_empty() {
        detail.push_str("\nstderr:\n");
        detail.push_str(&stderr);
    }
    detail
}

fn host_spacetime_publish_command(
    endpoint_url: &str,
    module_name: &str,
    module_path: &Path,
) -> Result<Command> {
    let config_dir = spacetime_docker_config_dir(endpoint_url)?;
    fs::create_dir_all(&config_dir).with_context(|| {
        format!(
            "failed to create SpacetimeDB CLI config directory at {}",
            config_dir.display()
        )
    })?;
    let config_path = config_dir.join("cli.toml");

    let mut command = Command::new("spacetime");
    command
        .arg("--config-path")
        .arg(config_path)
        .arg("publish")
        .arg("--server")
        .arg(endpoint_url)
        .arg("--module-path")
        .arg(module_path)
        .arg("--break-clients")
        .arg("--delete-data=on-conflict")
        .arg("--yes")
        .arg(module_name)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    Ok(command)
}

fn docker_spacetime_publish_command(
    endpoint_url: &str,
    module_name: &str,
    module_path: &Path,
) -> Result<Command> {
    ensure_docker_desktop_running()?;
    let mount_root = spacetime_docker_mount_root(module_path)?;
    let module_path_in_container = container_workspace_path(&mount_root, module_path)?;
    let config_dir = spacetime_docker_config_dir(endpoint_url)?;
    fs::create_dir_all(&config_dir).with_context(|| {
        format!(
            "failed to create SpacetimeDB docker config directory at {}",
            config_dir.display()
        )
    })?;

    let mut command = Command::new("docker");
    command
        .arg("run")
        .arg("--rm")
        .arg("-e")
        .arg("CARGO_TARGET_DIR=/tmp/spacetimedb-module-target")
        .arg("-v")
        .arg(format!("{}:/workspace:ro", mount_root.display()))
        .arg("-v")
        .arg(format!(
            "{}:/home/spacetime/.config/spacetime",
            config_dir.display()
        ))
        .arg("-w")
        .arg("/workspace");
    let publish_target = docker_spacetime_publish_target(endpoint_url)?;
    if let Some(network) = publish_target.network_arg.as_deref() {
        command.arg("--network").arg(network);
    }
    command
        .arg("clockworklabs/spacetime:v2.6.1")
        .arg("publish")
        .arg("--server")
        .arg(&publish_target.server_url)
        .arg("--module-path")
        .arg(module_path_in_container)
        .arg("--break-clients")
        .arg("--delete-data=on-conflict")
        .arg("--yes")
        .arg(module_name)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    Ok(command)
}

fn safe_path_segment(value: &str) -> String {
    value
        .chars()
        .map(|character| {
            if character.is_ascii_alphanumeric() {
                character.to_ascii_lowercase()
            } else {
                '-'
            }
        })
        .collect::<String>()
        .trim_matches('-')
        .to_string()
}

fn hex_digest(bytes: &[u8]) -> String {
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let mut output = String::with_capacity(bytes.len() * 2);
    for byte in bytes {
        output.push(HEX[(byte >> 4) as usize] as char);
        output.push(HEX[(byte & 0x0f) as usize] as char);
    }
    output
}

fn command_available(name: &str) -> bool {
    Command::new(name)
        .arg("--version")
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status()
        .map(|status| status.success())
        .unwrap_or(false)
}

fn spacetime_docker_mount_root(module_path: &Path) -> Result<PathBuf> {
    let app_root = app_repo_root_from_module_path(module_path).ok_or_else(|| {
        anyhow!(
            "failed to resolve app repository root from SpacetimeDB module path {}",
            module_path.display()
        )
    })?;
    Ok(app_root.parent().map(Path::to_path_buf).unwrap_or(app_root))
}

fn app_repo_root_from_module_path(module_path: &Path) -> Option<PathBuf> {
    for ancestor in module_path.ancestors() {
        if ancestor.join("Cargo.toml").is_file() && ancestor.join("services.env").is_file() {
            return Some(ancestor.to_path_buf());
        }
    }

    None
}

fn container_workspace_path(mount_root: &Path, host_path: &Path) -> Result<String> {
    let relative = host_path.strip_prefix(mount_root).with_context(|| {
        format!(
            "SpacetimeDB module path {} is not under docker mount root {}",
            host_path.display(),
            mount_root.display()
        )
    })?;
    let mut path = String::from("/workspace");
    for component in relative.components() {
        if let Component::Normal(value) = component {
            path.push('/');
            path.push_str(&value.to_string_lossy().replace('\\', "/"));
        }
    }
    Ok(path)
}

fn spacetime_docker_config_dir(endpoint_url: &str) -> Result<PathBuf> {
    let endpoint_slug = spacetime_endpoint_slug(endpoint_url)?;
    let home = std::env::var_os("USERPROFILE")
        .or_else(|| std::env::var_os("HOME"))
        .ok_or_else(|| anyhow!("failed to resolve home directory for SpacetimeDB docker config"))?;
    Ok(PathBuf::from(home)
        .join(".config")
        .join("spacetime-docker")
        .join(endpoint_slug))
}

fn spacetime_endpoint_slug(endpoint_url: &str) -> Result<String> {
    let parsed = Url::parse(endpoint_url)
        .with_context(|| format!("invalid SpacetimeDB endpoint URL: {endpoint_url}"))?;
    let host = parsed.host_str().unwrap_or("local");
    let port = parsed
        .port_or_known_default()
        .unwrap_or(if parsed.scheme() == "https" { 443 } else { 80 });
    let safe_host = safe_path_segment(host);
    Ok(format!("{safe_host}-{port}"))
}

fn write_spacetime_cli_token(endpoint_url: &str, token: &str) -> Result<()> {
    if token.trim().is_empty()
        || token
            .chars()
            .any(|character| character.is_control() || matches!(character, '"' | '\\'))
    {
        return Err(anyhow!(
            "local SpacetimeDB module-owner token cannot be encoded in the CLI config"
        ));
    }
    let config_dir = spacetime_docker_config_dir(endpoint_url)?;
    fs::create_dir_all(&config_dir).with_context(|| {
        format!(
            "failed to create SpacetimeDB CLI config directory at {}",
            config_dir.display()
        )
    })?;
    let config_path = config_dir.join("cli.toml");
    let existing = fs::read_to_string(&config_path).unwrap_or_else(|_| {
        concat!(
            "default_server = \"maincloud\"\n\n",
            "[[server_configs]]\n",
            "nickname = \"maincloud\"\n",
            "host = \"maincloud.spacetimedb.com\"\n",
            "protocol = \"https\"\n\n",
            "[[server_configs]]\n",
            "nickname = \"local\"\n",
            "host = \"127.0.0.1:3000\"\n",
            "protocol = \"http\"\n",
        )
        .to_string()
    });
    let token_line = format!("spacetimedb_token = \"{}\"", token.trim());
    let mut replaced = false;
    let mut lines = existing
        .lines()
        .map(|line| {
            if line.trim_start().starts_with("spacetimedb_token") {
                replaced = true;
                token_line.clone()
            } else {
                line.to_string()
            }
        })
        .collect::<Vec<_>>();
    if !replaced {
        let insert_at = lines
            .iter()
            .position(|line| line.trim_start().starts_with("default_server"))
            .map(|index| index + 1)
            .unwrap_or(0);
        lines.insert(insert_at, token_line);
    }
    write_private_file(&config_path, format!("{}\n", lines.join("\n")).as_bytes())
}

fn write_private_file(path: &Path, contents: &[u8]) -> Result<()> {
    let mut options = fs::OpenOptions::new();
    options.create(true).truncate(true).write(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        options.mode(0o600);
    }

    let mut file = options
        .open(path)
        .with_context(|| format!("failed to open private file {}", path.display()))?;
    file.write_all(contents)
        .with_context(|| format!("failed to write private file {}", path.display()))?;
    file.sync_all()
        .with_context(|| format!("failed to flush private file {}", path.display()))?;

    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(path, fs::Permissions::from_mode(0o600))
            .with_context(|| format!("failed to restrict private file {}", path.display()))?;
    }

    Ok(())
}

fn docker_spacetime_server_url(endpoint_url: &str) -> Result<String> {
    if cfg!(target_os = "linux") {
        return Ok(endpoint_url.to_string());
    }

    let parsed = Url::parse(endpoint_url)
        .with_context(|| format!("invalid SpacetimeDB endpoint URL: {endpoint_url}"))?;
    let Some(host) = parsed.host_str() else {
        return Ok(endpoint_url.to_string());
    };
    if !is_loopback(host) {
        return Ok(endpoint_url.to_string());
    }

    let port = parsed
        .port()
        .map(|port| format!(":{port}"))
        .unwrap_or_default();
    let replacement = format!("{}://host.docker.internal{port}", parsed.scheme());
    Ok(replacement)
}

#[derive(Debug, Clone)]
struct DockerSpacetimePublishTarget {
    server_url: String,
    network_arg: Option<String>,
}

fn docker_spacetime_publish_target(endpoint_url: &str) -> Result<DockerSpacetimePublishTarget> {
    if cfg!(target_os = "linux") {
        return Ok(DockerSpacetimePublishTarget {
            server_url: endpoint_url.to_string(),
            network_arg: Some("host".to_string()),
        });
    }

    let parsed = Url::parse(endpoint_url)
        .with_context(|| format!("invalid SpacetimeDB endpoint URL: {endpoint_url}"))?;
    let Some(host) = parsed.host_str() else {
        return Ok(DockerSpacetimePublishTarget {
            server_url: endpoint_url.to_string(),
            network_arg: None,
        });
    };
    if !is_loopback(host) {
        return Ok(DockerSpacetimePublishTarget {
            server_url: endpoint_url.to_string(),
            network_arg: None,
        });
    }

    let host_port = parsed
        .port_or_known_default()
        .ok_or_else(|| anyhow!("SpacetimeDB endpoint URL has no port: {endpoint_url}"))?;
    if let Some(container_name) = docker_container_name_for_host_port(host_port)? {
        return Ok(DockerSpacetimePublishTarget {
            server_url: "http://127.0.0.1:3000".to_string(),
            network_arg: Some(format!("container:{container_name}")),
        });
    }

    Ok(DockerSpacetimePublishTarget {
        server_url: docker_spacetime_server_url(endpoint_url)?,
        network_arg: None,
    })
}

fn docker_container_name_for_host_port(host_port: u16) -> Result<Option<String>> {
    let output = Command::new("docker")
        .arg("ps")
        .arg("--format")
        .arg("{{.Names}}\t{{.Ports}}")
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .output()
        .with_context(|| "failed to inspect running docker containers for local SpacetimeDB")?;
    if !output.status.success() {
        return Err(anyhow!(
            "docker ps exited with {} while inspecting local SpacetimeDB containers",
            output.status
        ));
    }

    let needle = format!(":{host_port}->");
    for line in String::from_utf8_lossy(&output.stdout).lines() {
        let mut parts = line.splitn(2, '\t');
        let Some(name) = parts
            .next()
            .map(str::trim)
            .filter(|value| !value.is_empty())
        else {
            continue;
        };
        let Some(ports) = parts.next().map(str::trim) else {
            continue;
        };
        if ports.contains(&needle) {
            return Ok(Some(name.to_string()));
        }
    }

    Ok(None)
}

/// Returns the Docker Compose file bundled with this crate.
fn bundled_compose_path() -> PathBuf {
    Path::new(env!("CARGO_MANIFEST_DIR"))
        .join("docker")
        .join("docker-compose.yml")
}

/// Computes the docker compose project name for an app name.
fn compose_project_name(app_name: &str) -> String {
    let lower = app_name.trim().to_ascii_lowercase();
    let mut out = String::with_capacity(lower.len());
    let mut last_was_dash = false;

    for ch in lower.chars() {
        if ch.is_ascii_alphanumeric() {
            out.push(ch);
            last_was_dash = false;
            continue;
        }

        if !last_was_dash {
            out.push('-');
            last_was_dash = true;
        }
    }

    let trimmed = out.trim_matches('-');
    if trimmed.is_empty() {
        "workspace".to_string()
    } else {
        trimmed.to_string()
    }
}

fn ensure_docker_desktop_running() -> Result<()> {
    match docker_info_probe() {
        DockerInfoProbe::Ready => return Ok(()),
        DockerInfoProbe::NotReady(detail) => {
            if !detail.is_empty() {
                debug!("docker info failed: {detail}");
            }
        }
        DockerInfoProbe::Unavailable(error) => {
            #[cfg(any(target_os = "windows", target_os = "macos"))]
            {
                if auto_start_docker_desktop()? {
                    return Ok(());
                }
            }

            return Err(error);
        }
    }

    #[cfg(any(target_os = "windows", target_os = "macos"))]
    {
        if auto_start_docker_desktop()? {
            return Ok(());
        }
    }

    Err(anyhow!(
        "Docker Desktop does not appear to be running. Please start Docker Desktop and try again."
    ))
}

/// Executes `docker info` and returns the captured process output.
fn docker_info() -> Result<Output> {
    let mut command = Command::new("docker");
    command
        .arg("info")
        .stdout(Stdio::null())
        .stderr(Stdio::piped());
    run_command_with_timeout(&mut command, DOCKER_INFO_TIMEOUT, "`docker info`")
        .context("failed to execute `docker info`; is Docker CLI installed?")
}

enum DockerInfoProbe {
    Ready,
    NotReady(String),
    Unavailable(anyhow::Error),
}

fn docker_info_probe() -> DockerInfoProbe {
    match docker_info() {
        Ok(output) if output.status.success() => DockerInfoProbe::Ready,
        Ok(output) => {
            let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
            let detail = if stderr.is_empty() {
                format!("docker info exited with status {}", output.status)
            } else {
                stderr
            };
            DockerInfoProbe::NotReady(detail)
        }
        Err(error) => {
            if docker_info_timeout_like(&error) {
                DockerInfoProbe::NotReady(error.to_string())
            } else {
                DockerInfoProbe::Unavailable(error)
            }
        }
    }
}

fn docker_info_timeout_like(error: &anyhow::Error) -> bool {
    error
        .chain()
        .any(|cause| cause.to_string().contains("`docker info` timed out after"))
}

fn run_command_with_timeout(
    command: &mut Command,
    timeout: Duration,
    description: &str,
) -> Result<Output> {
    let mut child = command
        .spawn()
        .with_context(|| format!("failed to spawn {description}"))?;
    let started = Instant::now();

    loop {
        if let Some(status) = child
            .try_wait()
            .with_context(|| format!("failed to poll {description}"))?
        {
            let mut stdout = Vec::new();
            let mut stderr = Vec::new();
            if let Some(mut stdout_reader) = child.stdout.take() {
                let _ = stdout_reader.read_to_end(&mut stdout);
            }
            if let Some(mut stderr_reader) = child.stderr.take() {
                let _ = stderr_reader.read_to_end(&mut stderr);
            }
            return Ok(Output {
                status,
                stdout,
                stderr,
            });
        }

        if started.elapsed() >= timeout {
            let _ = child.kill();
            let _ = child.wait();
            return Err(anyhow!(
                "{description} timed out after {} seconds",
                timeout.as_secs()
            ));
        }

        thread::sleep(COMMAND_POLL_INTERVAL);
    }
}

#[cfg(any(target_os = "windows", target_os = "macos"))]
/// Attempts to start Docker Desktop and waits until it becomes ready.
fn auto_start_docker_desktop() -> Result<bool> {
    if matches!(docker_info_probe(), DockerInfoProbe::Ready) {
        return Ok(true);
    }

    if std::env::var_os("RDS_SKIP_DOCKER_AUTOSTART").is_some() {
        info!("docker desktop auto-start disabled via RDS_SKIP_DOCKER_AUTOSTART");
        return Ok(false);
    }

    if should_suppress_autostart() {
        debug!("skipping Docker Desktop auto-start; recent attempt detected");
        return Ok(false);
    }

    if launch_docker_desktop()? {
        info!("waiting for Docker Desktop to become ready after automatic launch");
        let timeout = DOCKER_DESKTOP_READY_TIMEOUT;
        let start = Instant::now();

        while start.elapsed() < timeout {
            thread::sleep(Duration::from_secs(2));

            match docker_info_probe() {
                DockerInfoProbe::Ready => {
                    info!("Docker Desktop became ready after automatic launch");
                    return Ok(true);
                }
                DockerInfoProbe::NotReady(detail) => {
                    if !detail.is_empty() {
                        debug!("docker info failed: {detail}");
                    }
                    debug!("Docker Desktop is still starting up; retrying docker info");
                }
                DockerInfoProbe::Unavailable(err) => {
                    debug!(
                        "failed to execute docker info while waiting for Docker Desktop: {err:?}"
                    );
                    return Ok(false);
                }
            }
        }

        warn!("timed out while waiting for Docker Desktop to start");
        return Ok(false);
    }

    debug!("could not locate Docker Desktop installation; skipping auto-start attempt");
    Ok(false)
}

#[cfg(any(target_os = "windows", target_os = "macos"))]
/// Returns the marker path used to suppress repeated Docker Desktop auto-start attempts.
fn autostart_marker_path() -> Option<PathBuf> {
    Some(std::env::temp_dir().join("local-data-services-docker-desktop-autostart"))
}

#[cfg(any(target_os = "windows", target_os = "macos"))]
/// Returns true when a recent auto-start attempt should suppress another launch.
fn should_suppress_autostart() -> bool {
    let Some(path) = autostart_marker_path() else {
        return false;
    };
    let Ok(metadata) = fs::metadata(&path) else {
        return false;
    };
    let Ok(modified) = metadata.modified() else {
        return false;
    };
    let Ok(elapsed) = modified.elapsed() else {
        return false;
    };

    elapsed < Duration::from_secs(90)
}

#[cfg(any(target_os = "windows", target_os = "macos"))]
/// Records a Docker Desktop auto-start attempt for throttling.
fn mark_autostart_attempt() {
    let Some(path) = autostart_marker_path() else {
        return;
    };
    if let Some(parent) = path.parent() {
        let _ = fs::create_dir_all(parent);
    }
    let _ = fs::write(path, b"1");
}

#[cfg(target_os = "windows")]
/// Resolves the Docker Desktop executable path using common installation locations.
fn docker_desktop_executable() -> Option<PathBuf> {
    const EXECUTABLE_NAME: &str = "Docker Desktop.exe";

    let mut candidates = Vec::new();

    if let Ok(program_files) = std::env::var("ProgramFiles") {
        candidates.push(
            PathBuf::from(program_files)
                .join("Docker")
                .join("Docker")
                .join(EXECUTABLE_NAME),
        );
    }

    if let Ok(program_files_x86) = std::env::var("ProgramFiles(x86)") {
        candidates.push(
            PathBuf::from(program_files_x86)
                .join("Docker")
                .join("Docker")
                .join(EXECUTABLE_NAME),
        );
    }

    if let Ok(local_app_data) = std::env::var("LOCALAPPDATA") {
        candidates.push(
            PathBuf::from(local_app_data)
                .join("Docker")
                .join("Docker Desktop.exe"),
        );
    }

    candidates.into_iter().find(|path| path.exists())
}

#[cfg(target_os = "windows")]
fn launch_docker_desktop() -> Result<bool> {
    let Some(executable) = docker_desktop_executable() else {
        return Ok(false);
    };

    mark_autostart_attempt();
    info!(
        "attempting to launch Docker Desktop automatically via {}",
        executable.display()
    );

    Command::new(&executable)
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .spawn()
        .with_context(|| {
            format!(
                "failed to launch Docker Desktop executable at {}",
                executable.display()
            )
        })?;

    Ok(true)
}

#[cfg(target_os = "macos")]
fn launch_docker_desktop() -> Result<bool> {
    let app_bundle = PathBuf::from("/Applications/Docker.app");
    if !app_bundle.exists() {
        return Ok(false);
    }

    mark_autostart_attempt();
    info!(
        "attempting to launch Docker Desktop automatically via {}",
        app_bundle.display()
    );

    Command::new("open")
        .arg("-g")
        .arg(app_bundle.as_os_str())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .spawn()
        .with_context(|| {
            format!(
                "failed to launch Docker Desktop app bundle at {}",
                app_bundle.display()
            )
        })?;

    Ok(true)
}

/// Determines whether the provided connection URL points to a local host.
pub(crate) fn is_local_database(url: &str) -> bool {
    host_from_connection_url(url)
        .map(|host| is_loopback(&host))
        .unwrap_or(false)
}

/// Extracts the host component from a database connection URL.
fn host_from_connection_url(url: &str) -> Option<String> {
    if let Ok(parsed) = Url::parse(url)
        && let Some(host) = parsed.host_str()
    {
        return Some(host.to_string());
    }

    extract_host_from_authority(url)
}

fn host_and_port_from_connection_url(url: &str) -> Option<(String, u16)> {
    if let Ok(parsed) = Url::parse(url)
        && let Some(host) = parsed.host_str()
    {
        return Some((host.to_string(), parsed.port().unwrap_or(3306)));
    }

    extract_host_port_from_authority(url)
}

/// Extracts the host component from the authority section of a URL string.
fn extract_host_from_authority(url: &str) -> Option<String> {
    let host_port = connection_host_port(url)?;

    if let Some(end) = host_port.find(']')
        && let Some(start) = host_port.find('[')
        && start < end
    {
        return Some(host_port[start + 1..end].to_string());
    }

    let host_only = host_port.split(':').next().unwrap_or(host_port);
    if host_only.is_empty() {
        None
    } else {
        Some(host_only.to_string())
    }
}

fn extract_host_port_from_authority(url: &str) -> Option<(String, u16)> {
    let host_port = connection_host_port(url)?;

    if let Some(end) = host_port.find(']')
        && let Some(start) = host_port.find('[')
        && start < end
    {
        let host = host_port[start + 1..end].to_string();
        let port = host_port[end + 1..]
            .strip_prefix(':')
            .and_then(|value| value.parse::<u16>().ok())
            .unwrap_or(3306);
        return Some((host, port));
    }

    if let Some((host, port)) = host_port.rsplit_once(':')
        && !host.is_empty()
        && let Ok(port) = port.parse::<u16>()
    {
        return Some((host.to_string(), port));
    }

    Some((host_port.to_string(), 3306))
}

fn connection_host_port(url: &str) -> Option<&str> {
    let (_, remainder) = url.split_once("://")?;
    let authority = remainder.split(['/', '?']).next()?;
    (!authority.is_empty())
        .then(|| authority.rsplit('@').next())
        .flatten()
}

/// Returns true when the provided host string references a loopback address.
fn is_loopback(host: &str) -> bool {
    if host.eq_ignore_ascii_case("localhost") {
        return true;
    }

    let trimmed = host.trim_matches(|c| c == '[' || c == ']');
    trimmed
        .parse::<std::net::IpAddr>()
        .map(|addr| addr.is_loopback())
        .unwrap_or(false)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::borrow::Cow;
    use std::error::Error as StdError;
    use std::fmt::{Display, Formatter};

    #[test]
    fn detects_local_hosts() {
        assert!(is_local_database("mysql://user:pass@localhost:4520/db"));
        assert!(is_local_database("mysql://user:pass@127.0.0.1:4520/db"));
        assert!(!is_local_database("mysql://user:pass@192.168.1.10:4520/db"));
    }

    #[test]
    fn accepts_local_hosts_with_unescaped_credentials() {
        let url = "mysql://app:R7#vM}tQw@1Ls4eZ@127.0.0.1:4520/app";
        assert!(is_local_database(url));
        assert_eq!(
            LocalService::MariaDb(url).display_endpoint(),
            "127.0.0.1:4520"
        );
    }

    #[test]
    fn rejects_remote_hosts_with_unescaped_credentials() {
        let url = "mysql://app:R7#vM}tQw@db.example.com:4520/app";
        assert!(!is_local_database(url));
    }

    #[test]
    fn recognises_ipv6_loopback_hosts() {
        assert!(is_local_database("mysql://user:pass@[::1]:4520/db"));
    }

    #[test]
    fn compose_file_is_bundled_with_the_crate() {
        let compose_path = bundled_compose_path();
        assert!(compose_path.is_file());
        let compose = fs::read_to_string(compose_path).unwrap();
        assert!(compose.contains("http://127.0.0.1:3000/v1/ping"));
        assert!(compose.contains("healthcheck.sh"));
        assert!(compose.contains("--innodb_initialized"));
        assert!(compose.contains("start_period: 30s"));
        assert!(!compose.contains("mariadb-admin"));
    }

    #[test]
    fn docker_info_timeout_is_treated_as_not_ready() {
        let error = anyhow!(
            "failed to execute `docker info`; is Docker CLI installed?: `docker info` timed out after 15 seconds"
        );
        assert!(docker_info_timeout_like(&error));
    }

    #[test]
    fn windows_excluded_tcp_port_range_finds_spacetime_port() {
        let output = r#"
Portausschlussbereiche fuer das Protokoll "tcp"

Startport      Endport
----------    --------
      4348        4447
      4448        4547
     50000       50059     *
"#;

        assert_eq!(
            windows_excluded_tcp_port_range(output, 4530),
            Some((4448, 4547))
        );
    }

    #[test]
    fn windows_excluded_tcp_port_range_ignores_unmatched_ports() {
        let output = r#"
Startport      Endport
----------    --------
      4448        4547
"#;

        assert_eq!(windows_excluded_tcp_port_range(output, 4550), None);
    }

    #[test]
    fn parses_spacetime_compose_port_from_env_file_contents() {
        let contents = r#"
MARIADB_PORT=4520
SPACETIME_PORT="4530"
MAILPIT_UI_PORT=8025
"#;

        assert_eq!(parse_spacetime_compose_port(contents), Some(4530));
    }

    #[test]
    fn ignores_invalid_spacetime_compose_port_values() {
        assert_eq!(parse_spacetime_compose_port("SPACETIME_PORT=0"), None);
        assert_eq!(parse_spacetime_compose_port("SPACETIME_PORT=invalid"), None);
        assert_eq!(parse_spacetime_compose_port("# SPACETIME_PORT=4530"), None);
    }

    #[test]
    fn parses_spacetime_identity_http_response() {
        let response = concat!(
            "HTTP/1.1 200 OK\r\n",
            "Content-Type: application/json\r\n",
            "Content-Length: 35\r\n",
            "\r\n",
            "{\"identity\":\"owner\",\"token\":\"jwt\"}"
        );
        let (status, body) = parse_http_response(response.as_bytes()).unwrap();
        assert_eq!(status, 200);
        let credentials = serde_json::from_slice::<LocalSpacetimeOwnerCredentials>(&body).unwrap();
        assert_eq!(credentials.identity, "owner");
        assert_eq!(credentials.token, "jwt");
    }

    #[test]
    fn parses_chunked_spacetime_identity_http_response() {
        let response = concat!(
            "HTTP/1.1 200 OK\r\n",
            "Transfer-Encoding: chunked\r\n",
            "\r\n",
            "10\r\n{\"identity\":\"own\r\n",
            "12\r\ner\",\"token\":\"jwt\"}\r\n",
            "0\r\n\r\n"
        );
        let (status, body) = parse_http_response(response.as_bytes()).unwrap();
        assert_eq!(status, 200);
        assert_eq!(body, br#"{"identity":"owner","token":"jwt"}"#);
    }

    #[test]
    fn mariadb_protocol_failures_distinguish_startup_from_broken_proxy() {
        let starting = sqlx::Error::Io(std::io::Error::from(std::io::ErrorKind::ConnectionRefused));
        assert_eq!(
            mariadb_protocol_failure_kind(&starting),
            ProtocolProbeFailureKind::Starting
        );

        let broken = sqlx::Error::Io(std::io::Error::from(std::io::ErrorKind::UnexpectedEof));
        assert_eq!(
            mariadb_protocol_failure_kind(&broken),
            ProtocolProbeFailureKind::Broken
        );
    }

    #[test]
    fn parses_compose_service_health_from_object_stream_and_array() {
        let object_stream = concat!(
            r#"{"Service":"mariadb","State":"running","Health":"starting","ExitCode":0}"#,
            "\n",
            r#"{"Service":"mailpit","State":"running","Health":"","ExitCode":0}"#,
        );
        let statuses = parse_compose_service_statuses(object_stream.as_bytes()).unwrap();
        assert_eq!(statuses.len(), 2);
        assert_eq!(statuses[0].service, "mariadb");
        assert_eq!(statuses[0].health, "starting");

        let array = br#"[{"Service":"mariadb","State":"running","Health":"healthy","ExitCode":0}]"#;
        let statuses = parse_compose_service_statuses(array).unwrap();
        assert_eq!(statuses.len(), 1);
        assert_eq!(statuses[0].health, "healthy");
    }

    #[test]
    fn mariadb_authentication_failure_points_to_existing_volume_credentials() {
        let error = anyhow!(sqlx::Error::Database(Box::new(AuthenticationDatabaseError)))
            .context("MariaDB protocol probe failed");
        assert!(mariadb_credentials_rejected(&error));

        let message = local_service_ready_error(
            LocalService::MariaDb("mysql://app:secret@127.0.0.1:4542/app"),
            error,
            false,
        )
        .to_string();
        assert!(message.contains("MariaDB rejected the configured credentials"));
        assert!(message.contains("existing local volume"));
        assert!(message.contains("ch-rds did not modify it"));
        assert!(!message.contains("after one controlled recovery"));
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    #[ignore = "requires a running Docker engine and creates an isolated MariaDB volume"]
    async fn fresh_mariadb_volume_initializes_without_restart_and_reopens() {
        assert!(
            matches!(docker_info_probe(), DockerInfoProbe::Ready),
            "Docker must be running for this ignored integration test"
        );

        let directory = tempfile::tempdir().unwrap();
        let env_path = directory.path().join("services.env");
        let listeners = (0..3)
            .map(|_| std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap())
            .collect::<Vec<_>>();
        let ports = listeners
            .iter()
            .map(|listener| listener.local_addr().unwrap().port())
            .collect::<Vec<_>>();
        drop(listeners);

        fs::write(
            &env_path,
            format!(
                concat!(
                    "MARIADB_ROOT_PASSWORD=integration-root\n",
                    "MARIADB_DATABASE=integration\n",
                    "MARIADB_USER=integration\n",
                    "MARIADB_PASSWORD=integration-pass\n",
                    "MARIADB_PORT={}\n",
                    "MAILPIT_SMTP_PORT={}\n",
                    "MAILPIT_UI_PORT={}\n"
                ),
                ports[0], ports[1], ports[2]
            ),
        )
        .unwrap();

        let compose_path = bundled_compose_path();
        let project_name = compose_project_name(
            format!("mariadb-readiness-{}-{}", std::process::id(), ports[0]).as_str(),
        );
        let cleanup = ComposeStackCleanup {
            compose_path: compose_path.clone(),
            project_name: project_name.clone(),
            env_path: env_path.clone(),
        };
        cleanup.remove();

        run_docker_compose_up(&compose_path, &project_name, &env_path, &[]).unwrap();
        let started_at =
            compose_service_started_at(&compose_path, &project_name, &env_path, "mariadb");
        let connection_url = format!(
            "mysql://integration:integration-pass@127.0.0.1:{}/integration",
            ports[0]
        );
        ensure_local_service_ready(
            &compose_path,
            &project_name,
            &env_path,
            &[],
            LocalService::MariaDb(connection_url.as_str()),
        )
        .await
        .unwrap();
        assert_eq!(
            compose_service_started_at(&compose_path, &project_name, &env_path, "mariadb"),
            started_at,
            "the fresh MariaDB container was restarted during initialization"
        );
        mariadb_protocol_probe(connection_url.as_str())
            .await
            .unwrap();

        run_docker_compose_up(&compose_path, &project_name, &env_path, &[]).unwrap();
        ensure_local_service_ready(
            &compose_path,
            &project_name,
            &env_path,
            &[],
            LocalService::MariaDb(connection_url.as_str()),
        )
        .await
        .unwrap();
        assert_eq!(
            compose_service_started_at(&compose_path, &project_name, &env_path, "mariadb"),
            started_at,
            "the existing MariaDB volume caused an unnecessary restart"
        );
        mariadb_protocol_probe(connection_url.as_str())
            .await
            .unwrap();
    }

    #[test]
    fn spacetime_protocol_probe_requires_a_real_http_response() {
        let endpoint = local_http_probe_endpoint(Vec::new());
        let failure = spacetime_protocol_probe(endpoint.as_str()).unwrap_err();
        assert_eq!(failure.kind, ProtocolProbeFailureKind::Broken);
    }

    #[test]
    fn spacetime_protocol_probe_accepts_the_ping_success_status() {
        let endpoint = local_http_probe_endpoint(
            b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_vec(),
        );
        spacetime_protocol_probe(endpoint.as_str()).unwrap();
    }

    fn local_http_probe_endpoint(response: Vec<u8>) -> String {
        let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap();
        let address = listener.local_addr().unwrap();
        thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            let mut request = [0_u8; 1024];
            let _ = stream.read(&mut request);
            if !response.is_empty() {
                stream.write_all(&response).unwrap();
            }
        });
        format!("http://{address}")
    }

    #[derive(Debug)]
    struct AuthenticationDatabaseError;

    impl Display for AuthenticationDatabaseError {
        fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
            formatter.write_str("access denied")
        }
    }

    impl StdError for AuthenticationDatabaseError {}

    impl sqlx::error::DatabaseError for AuthenticationDatabaseError {
        fn message(&self) -> &str {
            "access denied"
        }

        fn code(&self) -> Option<Cow<'_, str>> {
            Some(Cow::Borrowed("28000"))
        }

        fn as_error(&self) -> &(dyn StdError + Send + Sync + 'static) {
            self
        }

        fn as_error_mut(&mut self) -> &mut (dyn StdError + Send + Sync + 'static) {
            self
        }

        fn into_error(self: Box<Self>) -> Box<dyn StdError + Send + Sync + 'static> {
            self
        }

        fn kind(&self) -> sqlx::error::ErrorKind {
            sqlx::error::ErrorKind::Other
        }
    }

    struct ComposeStackCleanup {
        compose_path: PathBuf,
        project_name: String,
        env_path: PathBuf,
    }

    impl ComposeStackCleanup {
        fn remove(&self) {
            let mut command =
                docker_compose_command(&self.compose_path, &self.project_name, &self.env_path, &[]);
            let _ = command
                .arg("down")
                .arg("--volumes")
                .arg("--remove-orphans")
                .stdout(Stdio::null())
                .stderr(Stdio::null())
                .status();
        }
    }

    impl Drop for ComposeStackCleanup {
        fn drop(&mut self) {
            self.remove();
        }
    }

    fn compose_service_started_at(
        compose_path: &Path,
        project_name: &str,
        env_path: &Path,
        service_name: &str,
    ) -> String {
        let mut compose = docker_compose_command(compose_path, project_name, env_path, &[]);
        let output = compose
            .arg("ps")
            .arg("--quiet")
            .arg(service_name)
            .output()
            .unwrap();
        assert!(output.status.success());
        let container_id = String::from_utf8(output.stdout).unwrap();
        let container_id = container_id.trim();
        assert!(!container_id.is_empty());

        let output = Command::new("docker")
            .arg("inspect")
            .arg("--format")
            .arg("{{.State.StartedAt}}")
            .arg(container_id)
            .output()
            .unwrap();
        assert!(output.status.success());
        String::from_utf8(output.stdout).unwrap().trim().to_string()
    }

    #[test]
    fn local_spacetime_bootstrap_lock_serializes_processes() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("module-bootstrap.lock");
        let first = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&path)
            .unwrap();
        let second = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&path)
            .unwrap();

        first.lock().unwrap();
        let error = lock_file_until(&second, Duration::ZERO, "test bootstrap").unwrap_err();
        assert_eq!(
            error.to_string(),
            "timed out waiting for test bootstrap lock"
        );

        drop(first);
        lock_file_until(&second, Duration::from_secs(1), "test bootstrap").unwrap();
    }

    #[test]
    fn spacetime_module_fingerprint_tracks_module_sources_only() {
        let root = std::env::temp_dir().join(format!(
            "spacetime-module-fingerprint-{}",
            std::process::id()
        ));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(root.join("src").join("generated")).unwrap();
        fs::create_dir_all(root.join("target")).unwrap();
        fs::write(root.join("Cargo.toml"), "[package]\nname = \"demo\"\n").unwrap();
        fs::write(root.join("src").join("lib.rs"), "mod reducers;\n").unwrap();
        fs::write(
            root.join("src").join("reducers.rs"),
            "pub fn reducer() {}\n",
        )
        .unwrap();
        fs::write(
            root.join("src").join("generated").join("client.rs"),
            "generated-v1\n",
        )
        .unwrap();
        fs::write(root.join("target").join("scratch.txt"), "scratch-v1\n").unwrap();

        let original = spacetime_module_fingerprint(&root).unwrap();
        fs::write(
            root.join("src").join("generated").join("client.rs"),
            "generated-v2\n",
        )
        .unwrap();
        fs::write(root.join("target").join("scratch.txt"), "scratch-v2\n").unwrap();
        assert_eq!(spacetime_module_fingerprint(&root).unwrap(), original);

        fs::write(
            root.join("src").join("reducers.rs"),
            "pub fn reducer2() {}\n",
        )
        .unwrap();
        assert_ne!(spacetime_module_fingerprint(&root).unwrap(), original);

        fs::remove_dir_all(&root).unwrap();
    }
}