ch-rds 0.2.0

use std::fs;
use std::io::Write;
use std::path::{Component, Path, PathBuf};
use std::process::{Command, Output, Stdio};
use std::time::{Duration, Instant};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;

use crate::LocalStack;
use crate::RuntimeEnvironment;
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);
// Keep the tested CLI, server, app modules, SDKs, and generated bindings aligned.
// The next target is 2.12.0. Advancing only this server would leave FGuard and
// FoxFYP on 2.11.0 bindings and SDKs, so their coordinated compatibility check
// must precede a version change here.
const SPACETIME_VERSION: &str = "2.11.0";
const SPACETIME_DOCKER_IMAGE: &str = "clockworklabs/spacetime:v2.11.0";
#[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: RuntimeEnvironment,
    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;
    }

    true
}

/// Prepares the selected local service under the shared lifecycle lock.
pub(crate) async fn ensure_local_stack(
    local_stack: &LocalStack,
    profiles: &[&str],
    endpoint_url: &str,
) -> Result<()> {
    let _lifecycle_lock = local_stack_lifecycle_lock(local_stack).await?;
    ensure_docker_engine_available().await?;

    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)
        .await
        .with_context(|| {
            format!(
                "failed to run docker compose up using {}",
                compose_path.display()
            )
        })?;
    let service = if profiles.contains(&"spacetime") {
        LocalService::SpacetimeDb(endpoint_url)
    } else {
        LocalService::MariaDb(endpoint_url)
    };

    ensure_local_service_ready(&compose_path, &project_name, env_path, profiles, service).await?;
    Ok(())
}

async 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()));
    tokio::fs::create_dir_all(&directory)
        .await
        .with_context(|| {
            format!(
                "failed to create local-stack lifecycle directory at {}",
                directory.display()
            )
        })?;
    let path = directory.join("lifecycle.lock");
    let file = tokio::fs::OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(&path)
        .await
        .with_context(|| {
            format!(
                "failed to open local-stack lifecycle lock at {}",
                path.display()
            )
        })?;
    let file = file.into_std().await;
    lock_file_until(
        &file,
        LOCAL_SERVICE_RECOVERY_LOCK_TIMEOUT,
        "local-stack lifecycle",
    )
    .await?;
    Ok(file)
}

/// Ensures the app-owned local SpacetimeDB module exists on the local server.
pub(crate) async 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)?;
    tokio::fs::create_dir_all(&config_dir)
        .await
        .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 = tokio::fs::OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(&bootstrap_lock_path)
        .await
        .with_context(|| {
            format!(
                "failed to open SpacetimeDB module bootstrap lock at {}",
                bootstrap_lock_path.display()
            )
        })?;
    let bootstrap_lock = bootstrap_lock.into_std().await;
    lock_file_until(
        &bootstrap_lock,
        SPACETIME_MODULE_BOOTSTRAP_LOCK_TIMEOUT,
        "local SpacetimeDB module bootstrap",
    )
    .await?;

    let fingerprint_path = module_path.to_path_buf();
    let module_fingerprint = tokio::task::spawn_blocking(move || {
        spacetime_module_fingerprint(&fingerprint_path, SPACETIME_VERSION)
    })
    .await
    .context("SpacetimeDB source fingerprint task failed")??;
    let module_status = wait_for_spacetime_module_status(endpoint_url, module_name).await?;
    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).await?;
            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),
    )
    .await?;
    write_spacetime_cli_token(endpoint_url, credentials.token.as_str())?;
    publish_local_spacetime_module(endpoint_url, module_name, module_path).await?;

    let deadline = Instant::now() + SPACETIME_MODULE_READY_TIMEOUT;
    loop {
        match spacetime_module_status(endpoint_url, module_name).await? {
            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 => {
                tokio::time::sleep(COMMAND_POLL_INTERVAL).await;
            }
            SpacetimeModuleStatus::Missing => {
                return Err(anyhow!(
                    "SpacetimeDB module `{module_name}` was published but is still missing at {endpoint_url}"
                ));
            }
        }
    }
}

async 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 => {
                tokio::time::sleep(COMMAND_POLL_INTERVAL).await;
            }
            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"));
            }
        }
    }
}

async 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);
    command
        .arg("up")
        .arg("--detach")
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    let output =
        run_command_with_timeout(command, SPACETIME_PUBLISH_TIMEOUT, "`docker compose up`").await?;

    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(
    fallback_compose_path: &Path,
    project_name: &str,
    env_path: &Path,
    profiles: &[&str],
) -> Command {
    let app_compose_path = env_path
        .parent()
        .map(|directory| directory.join("compose.yaml"));
    let compose_path = app_compose_path
        .as_deref()
        .filter(|path| path.is_file())
        .unwrap_or(fallback_compose_path);
    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);
        // Use the same parsed values for Compose interpolation and the local
        // connection URL. This prevents the container and the application
        // from interpreting special characters in credentials differently.
        if let Ok(values) = dotenvy::from_path_iter(env_path) {
            for (key, value) in values.flatten() {
                command.env(key, value);
            }
        }
    }
    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).await);
        }
        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(),
    )
    .await
    .with_context(|| format!("failed to restart local {}", service.display_name()))?;

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

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

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(),
        )
        .await?
        {
            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;
    }
}

async 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(
        command,
        DOCKER_INFO_TIMEOUT,
        "`docker compose ps --format json`",
    )
    .await?;
    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).await,
    }
}

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

async fn spacetime_protocol_probe(
    endpoint_url: &str,
) -> std::result::Result<(), ProtocolProbeFailure> {
    match local_spacetime_http_request(endpoint_url, "GET", "/v1/ping", None).await {
        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,
    })
}

async 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(
        command,
        LOCAL_SERVICE_READY_TIMEOUT,
        "`docker compose restart`",
    )
    .await?;
    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)
}

async 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()).await
    {
        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"))
    })
}

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

async 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).await {
            Ok(status) => return Ok(status),
            Err(error) if Instant::now() < deadline => {
                debug!(
                    %error,
                    "local SpacetimeDB endpoint is not ready for module check; retrying"
                );
                tokio::time::sleep(COMMAND_POLL_INTERVAL).await;
            }
            Err(error) => {
                return Err(spacetime_module_ready_error(endpoint_url, error).await);
            }
        }
    }
}

async 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).await {
        message.push_str("; ");
        message.push_str(diagnostic.as_str());
    }
    anyhow!(message)
}

async 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).await
}

#[cfg(target_os = "windows")]
async fn windows_tcp_excluded_port_diagnostic(port: u16) -> Option<String> {
    let mut command = Command::new("netsh");
    command
        .args([
            "interface",
            "ipv4",
            "show",
            "excludedportrange",
            "protocol=tcp",
        ])
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    let output =
        run_command_with_timeout(command, DOCKER_INFO_TIMEOUT, "`netsh excludedportrange`")
            .await
            .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"))]
async 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, runtime_version: &str) -> Result<String> {
    #[derive(Deserialize)]
    struct Manifest {
        package: Option<Package>,
        workspace: Option<toml::Table>,
    }
    #[derive(Deserialize)]
    struct Package {
        workspace: Option<PathBuf>,
    }

    let module_manifest = fs::canonicalize(module_path.join("Cargo.toml"))
        .context("failed to resolve SpacetimeDB module manifest")?;
    let read_manifest = |path: &Path| -> Result<Manifest> {
        let contents = fs::read_to_string(path)
            .with_context(|| format!("failed to read Cargo manifest {}", path.display()))?;
        toml::from_str(&contents)
            .with_context(|| format!("failed to parse Cargo manifest {}", path.display()))
    };
    let manifest = read_manifest(&module_manifest)?;
    // Follow Cargo's package.workspace/nearest-[workspace] contract without
    // requiring host Cargo: the existing Docker build also works without it.
    let workspace_manifest = match manifest.package.and_then(|package| package.workspace) {
        Some(workspace) => {
            let path = module_manifest
                .parent()
                .unwrap()
                .join(workspace)
                .join("Cargo.toml");
            let path = fs::canonicalize(&path)
                .with_context(|| format!("failed to resolve Cargo workspace {}", path.display()))?;
            if read_manifest(&path)?.workspace.is_none() {
                return Err(anyhow!(
                    "Cargo workspace manifest has no [workspace]: {}",
                    path.display()
                ));
            }
            path
        }
        None if manifest.workspace.is_some() => module_manifest.clone(),
        None => {
            let mut workspace_manifest = module_manifest.clone();
            for parent in module_manifest.parent().unwrap().ancestors().skip(1) {
                let path = parent.join("Cargo.toml");
                if path.is_file() && read_manifest(&path)?.workspace.is_some() {
                    workspace_manifest = path;
                    break;
                }
            }
            workspace_manifest
        }
    };

    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([2]);
    hasher.update(runtime_version.as_bytes());
    hasher.update([0]);
    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]);
    }

    if workspace_manifest != module_manifest {
        for (name, path) in [
            ("workspace/Cargo.toml", workspace_manifest.clone()),
            (
                "workspace/Cargo.lock",
                workspace_manifest.with_file_name("Cargo.lock"),
            ),
        ] {
            hasher.update(name.as_bytes());
            hasher.update([0]);
            match fs::read(&path) {
                Ok(contents) => {
                    hasher.update([1]);
                    hasher.update(contents);
                }
                Err(error)
                    if name == "workspace/Cargo.lock"
                        && error.kind() == std::io::ErrorKind::NotFound =>
                {
                    hasher.update([0]);
                }
                Err(error) => {
                    return Err(error).with_context(|| {
                        format!(
                            "failed to read SpacetimeDB workspace fingerprint file {}",
                            path.display()
                        )
                    });
                }
            }
            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")))
}

async fn spacetime_module_status(
    endpoint_url: &str,
    module_name: &str,
) -> Result<SpacetimeModuleStatus> {
    let path = format!("/v1/database/{module_name}/schema?version=9");
    let (status, _) = local_spacetime_http_request(endpoint_url, "GET", &path, None).await?;
    match status {
        200..=299 | 401 | 403 => Ok(SpacetimeModuleStatus::Available),
        404 => Ok(SpacetimeModuleStatus::Missing),
        code => Err(anyhow!(
            "SpacetimeDB schema probe for `{module_name}` returned HTTP {code}"
        )),
    }
}

async fn connect_local_http(host: &str, port: u16) -> Result<TcpStream> {
    let candidates = tokio::net::lookup_host((host, port))
        .await
        .context("failed to resolve local SpacetimeDB endpoint")?;
    let mut last_error = None;
    for addr in candidates {
        match tokio::time::timeout(LOCAL_DATABASE_PROBE_TIMEOUT, TcpStream::connect(addr)).await {
            Ok(Ok(stream)) => return Ok(stream),
            Ok(Err(error)) => last_error = Some(error),
            Err(_) => {
                last_error = Some(std::io::Error::new(
                    std::io::ErrorKind::TimedOut,
                    "connection timed out",
                ))
            }
        }
    }
    Err(last_error.unwrap_or_else(|| {
        std::io::Error::new(std::io::ErrorKind::NotFound, "host did not resolve")
    }))
    .with_context(|| format!("failed to connect to local SpacetimeDB endpoint {host}:{port}"))
}

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

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

async 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).await? {
            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).await?;
    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)
}

async fn create_local_spacetime_identity(
    endpoint_url: &str,
) -> Result<LocalSpacetimeOwnerCredentials> {
    let (status, body) =
        local_spacetime_http_request(endpoint_url, "POST", "/v1/identity", None).await?;
    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)
}

async 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()),
    )
    .await?;
    match status {
        200..=299 => Ok(true),
        400 | 401 | 403 | 404 => Ok(false),
        _ => Err(anyhow!(
            "local SpacetimeDB identity verification returned HTTP {status}"
        )),
    }
}

async fn local_spacetime_http_request(
    endpoint_url: &str,
    method: &str,
    path: &str,
    bearer_token: Option<&str>,
) -> Result<(u16, Vec<u8>)> {
    let request = async {
        let (mut stream, host, port) =
            connect_local_spacetime_http(endpoint_url, "HTTP request").await?;
        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())
            .await
            .context("failed to write local SpacetimeDB HTTP request")?;

        let mut response = Vec::new();
        stream
            .read_to_end(&mut response)
            .await
            .context("failed to read local SpacetimeDB HTTP response")?;
        parse_http_response(&response)
    };
    tokio::time::timeout(LOCAL_SERVICE_PROTOCOL_TIMEOUT, request)
        .await
        .map_err(|_| {
            std::io::Error::new(
                std::io::ErrorKind::TimedOut,
                "local SpacetimeDB HTTP request timed out",
            )
        })?
}

async 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).await?;

    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..];
    }
}

async fn publish_local_spacetime_module(
    endpoint_url: &str,
    module_name: &str,
    module_path: &Path,
) -> Result<()> {
    if host_spacetime_cli_matches_runtime().await {
        let host_command = host_spacetime_publish_command(endpoint_url, module_name, module_path)?;
        let host_output = run_command_with_timeout(
            host_command,
            SPACETIME_PUBLISH_TIMEOUT,
            "`spacetime publish`",
        )
        .await?;
        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 docker_command =
                docker_spacetime_publish_command(endpoint_url, module_name, module_path).await?;
            let docker_output = run_command_with_timeout(
                docker_command,
                SPACETIME_PUBLISH_TIMEOUT,
                "`spacetime publish` (docker fallback)",
            )
            .await?;
            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 docker_command =
        docker_spacetime_publish_command(endpoint_url, module_name, module_path).await?;
    let docker_output = run_command_with_timeout(
        docker_command,
        SPACETIME_PUBLISH_TIMEOUT,
        "`spacetime publish`",
    )
    .await?;
    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);
    configure_non_destructive_spacetime_publish(&mut command);
    command
        .arg(module_name)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    Ok(command)
}

async fn docker_spacetime_publish_command(
    endpoint_url: &str,
    module_name: &str,
    module_path: &Path,
) -> Result<Command> {
    ensure_docker_engine_available().await?;
    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).await?;
    if let Some(network) = publish_target.network_arg.as_deref() {
        command.arg("--network").arg(network);
    }
    command
        .arg(SPACETIME_DOCKER_IMAGE)
        .arg("publish")
        .arg("--server")
        .arg(&publish_target.server_url)
        .arg("--module-path")
        .arg(module_path_in_container);
    configure_non_destructive_spacetime_publish(&mut command);
    command
        .arg(module_name)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    Ok(command)
}

fn configure_non_destructive_spacetime_publish(command: &mut Command) {
    command
        .arg("--break-clients")
        .arg("--delete-data=never")
        .arg("--yes=migrate,break-clients,skip-login");
}

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
}

async fn host_spacetime_cli_matches_runtime() -> bool {
    let mut command = Command::new("spacetime");
    command
        .arg("--version")
        .stdout(Stdio::piped())
        .stderr(Stdio::null());
    let Ok(output) =
        run_command_with_timeout(command, DOCKER_INFO_TIMEOUT, "`spacetime --version`").await
    else {
        return false;
    };
    output.status.success()
        && spacetime_cli_version_matches_runtime(String::from_utf8_lossy(&output.stdout).as_ref())
}

fn spacetime_cli_version_matches_runtime(output: &str) -> bool {
    output.lines().any(|line| {
        line.trim() == format!(
            "spacetimedb tool version {SPACETIME_VERSION}; spacetimedb-lib version {SPACETIME_VERSION};"
        )
    })
}

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

async 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).await? {
        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,
    })
}

async fn docker_container_name_for_host_port(host_port: u16) -> Result<Option<String>> {
    let mut command = Command::new("docker");
    command
        .arg("ps")
        .arg("--format")
        .arg("{{.Names}}\t{{.Ports}}")
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    let output = run_command_with_timeout(command, DOCKER_INFO_TIMEOUT, "`docker ps`").await?;
    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()
    }
}

async fn ensure_docker_engine_available() -> Result<()> {
    match docker_info_probe().await {
        DockerInfoProbe::Ready => return Ok(()),
        DockerInfoProbe::NotReady(detail) => {
            if !detail.is_empty() {
                debug!("docker info failed: {detail}");
            }
        }
        DockerInfoProbe::Unavailable(error) => return Err(error),
    }

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

    let guidance = if cfg!(target_os = "linux") {
        "Start the Docker service or Docker Desktop, then try again."
    } else if cfg!(any(target_os = "windows", target_os = "macos")) {
        "Start Docker Desktop, then try again."
    } else {
        "Start a Docker-compatible engine, then try again."
    };
    Err(anyhow!("Docker Engine is unavailable. {guidance}"))
}

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

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

async fn docker_info_probe() -> DockerInfoProbe {
    match docker_info().await {
        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"))
}

async fn run_command_with_timeout(
    command: Command,
    timeout: Duration,
    description: &str,
) -> Result<Output> {
    let mut command = tokio::process::Command::from(command);
    command.kill_on_drop(true);
    tokio::time::timeout(timeout, command.output())
        .await
        .with_context(|| {
            format!(
                "{description} timed out after {} seconds",
                timeout.as_secs_f64()
            )
        })?
        .with_context(|| format!("failed to run {description}"))
}

#[cfg(any(target_os = "windows", target_os = "macos"))]
async fn start_docker_desktop() -> Result<bool> {
    if matches!(docker_info_probe().await, DockerInfoProbe::Ready) {
        return Ok(true);
    }

    let mut command = Command::new("docker");
    command
        .arg("desktop")
        .arg("start")
        .stdout(Stdio::null())
        .stderr(Stdio::piped());
    let output = match run_command_with_timeout(
        command,
        DOCKER_DESKTOP_READY_TIMEOUT,
        "`docker desktop start`",
    )
    .await
    {
        Ok(output) => output,
        Err(error) => {
            debug!(%error, "Docker Desktop CLI is unavailable");
            return Ok(false);
        }
    };
    if !output.status.success() {
        let detail = String::from_utf8_lossy(&output.stderr);
        debug!(
            status = %output.status,
            detail = detail.trim(),
            "Docker Desktop CLI did not start the engine"
        );
        return Ok(false);
    }

    info!("waiting for Docker Desktop to become ready");
    let start = Instant::now();
    while start.elapsed() < DOCKER_DESKTOP_READY_TIMEOUT {
        tokio::time::sleep(Duration::from_secs(2)).await;

        match docker_info_probe().await {
            DockerInfoProbe::Ready => {
                info!("Docker Desktop is ready");
                return Ok(true);
            }
            DockerInfoProbe::NotReady(detail) => {
                if !detail.is_empty() {
                    debug!("docker info failed: {detail}");
                }
                debug!("Docker Desktop is still starting; retrying docker info");
            }
            DockerInfoProbe::Unavailable(error) => {
                debug!(%error, "docker info became unavailable while Docker Desktop was starting");
                return Ok(false);
            }
        }
    }

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

/// 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};
    use std::fs::OpenOptions;
    use std::io::Read;
    use std::thread;

    #[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 local_stack_bootstrap_policy_is_platform_neutral() {
        let local = "mysql://user:pass@127.0.0.1:4520/db";
        assert_eq!(
            should_bootstrap_local_stack(RuntimeEnvironment::Local, local),
            std::env::var_os("CI").is_none()
        );
        assert!(!should_bootstrap_local_stack(
            RuntimeEnvironment::Production,
            local
        ));
        assert!(!should_bootstrap_local_stack(
            RuntimeEnvironment::Local,
            "mysql://user:pass@db.example.com: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("mkdir -p /data /config"));
        assert!(compose.contains("chown -R 1000:1000 /data /config"));
        assert!(compose.contains(SPACETIME_DOCKER_IMAGE));
        assert_eq!(
            compose
                .matches("spacetimedb-config:/home/spacetime/.config/spacetime")
                .count(),
            1
        );
        assert_eq!(compose.matches("spacetimedb-config:/config").count(), 1);
        assert!(!compose.contains("mariadb-admin"));
    }

    #[test]
    fn spacetime_publish_never_authorizes_data_deletion() {
        let mut command = Command::new("spacetime");
        configure_non_destructive_spacetime_publish(&mut command);
        let arguments = command
            .get_args()
            .map(|argument| argument.to_string_lossy().into_owned())
            .collect::<Vec<_>>();

        assert_eq!(
            arguments,
            [
                "--break-clients",
                "--delete-data=never",
                "--yes=migrate,break-clients,skip-login",
            ]
        );
    }

    #[test]
    fn host_spacetime_cli_must_match_the_runtime_version_exactly() {
        assert!(spacetime_cli_version_matches_runtime(
            "spacetime Path: /opt/spacetime\nspacetimedb tool version 2.11.0; spacetimedb-lib version 2.11.0;\n"
        ));
        assert!(!spacetime_cli_version_matches_runtime(
            "spacetimedb tool version 2.6.1; spacetimedb-lib version 2.6.1;"
        ));
        assert!(!spacetime_cli_version_matches_runtime(
            "spacetimedb tool version 2.11.0; spacetimedb-lib version 2.10.0;"
        ));
    }

    #[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_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");
    }

    #[tokio::test]
    async 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,
        )
        .await
        .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"));
    }

    #[test]
    fn compose_command_uses_the_same_parsed_env_file_values_as_the_client() {
        let directory = tempfile::tempdir().unwrap();
        let env_path = directory.path().join("services.env");
        fs::write(&env_path, "MARIADB_PASSWORD=local$$password\n").unwrap();

        let command = docker_compose_command(
            Path::new("docker-compose.yml"),
            "local-database",
            &env_path,
            &[],
        );
        let password = command.get_envs().find_map(|(key, value)| {
            (key == "MARIADB_PASSWORD")
                .then(|| value.and_then(|value| value.to_str()))
                .flatten()
        });

        assert_eq!(password, Some("local"));
    }

    #[test]
    fn compose_command_prefers_app_manifest_and_preserves_explicit_options() {
        let directory = tempfile::tempdir().unwrap();
        let env_path = directory.path().join("services.env");
        let app_compose_path = directory.path().join("compose.yaml");
        let fallback_compose_path = directory.path().join("docker-compose.yml");
        fs::write(&env_path, "APP_COMPOSE_VALUE=selected\n").unwrap();
        fs::write(&app_compose_path, "services: {}\n").unwrap();

        let command = docker_compose_command(
            &fallback_compose_path,
            "explicit-project",
            &env_path,
            &["mail", "spacetime"],
        );
        let arguments = command
            .get_args()
            .map(|argument| argument.to_string_lossy().into_owned())
            .collect::<Vec<_>>();

        assert_eq!(
            arguments,
            [
                "compose",
                "--file",
                app_compose_path.to_str().unwrap(),
                "--project-name",
                "explicit-project",
                "--env-file",
                env_path.to_str().unwrap(),
                "--profile",
                "mail",
                "--profile",
                "spacetime",
            ]
        );
        assert_eq!(
            command
                .get_envs()
                .find_map(|(key, value)| (key == "APP_COMPOSE_VALUE").then_some(value))
                .flatten()
                .and_then(|value| value.to_str()),
            Some("selected")
        );
    }

    #[test]
    fn compose_command_uses_fallback_manifest_when_app_manifest_is_absent() {
        let directory = tempfile::tempdir().unwrap();
        let env_path = directory.path().join("services.env");
        let fallback_compose_path = directory.path().join("docker-compose.yml");
        fs::write(&env_path, "APP_COMPOSE_VALUE=fallback\n").unwrap();

        let command = docker_compose_command(
            &fallback_compose_path,
            "fallback-project",
            &env_path,
            &["mail"],
        );
        let arguments = command
            .get_args()
            .map(|argument| argument.to_string_lossy().into_owned())
            .collect::<Vec<_>>();

        assert_eq!(
            arguments,
            [
                "compose",
                "--file",
                fallback_compose_path.to_str().unwrap(),
                "--project-name",
                "fallback-project",
                "--env-file",
                env_path.to_str().unwrap(),
                "--profile",
                "mail",
            ]
        );
    }

    #[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().await, 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 ports = {
            let listeners = (0..3)
                .map(|_| std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap())
                .collect::<Vec<_>>();
            listeners
                .iter()
                .map(|listener| listener.local_addr().unwrap().port())
                .collect::<Vec<_>>()
        };

        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, &[])
            .await
            .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, &[])
            .await
            .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();
    }

    #[tokio::test]
    async 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())
            .await
            .unwrap_err();
        assert_eq!(failure.kind, ProtocolProbeFailureKind::Broken);
    }

    #[tokio::test]
    async 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()).await.unwrap();
    }

    #[tokio::test(start_paused = true)]
    async fn timed_out_spacetime_probe_remains_a_startup_failure() {
        let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
            .await
            .unwrap();
        let endpoint = format!("http://{}", listener.local_addr().unwrap());
        let server = tokio::spawn(async move {
            let (_connection, _) = listener.accept().await.unwrap();
            std::future::pending::<()>().await;
        });
        let failure = spacetime_protocol_probe(&endpoint).await.unwrap_err();
        server.abort();
        assert_eq!(failure.kind, ProtocolProbeFailureKind::Starting);
    }

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

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

        {
            let first = 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")
                .await
                .unwrap_err();
            assert_eq!(
                error.to_string(),
                "timed out waiting for test bootstrap lock"
            );
        }
        lock_file_until(&second, Duration::from_secs(1), "test bootstrap")
            .await
            .unwrap();
    }

    #[test]
    fn spacetime_module_fingerprint_tracks_effective_build_inputs() {
        let directory = tempfile::tempdir().unwrap();
        let workspace = directory.path();
        let root = workspace.join("module");
        fs::create_dir_all(root.join("src").join("generated")).unwrap();
        fs::create_dir_all(root.join("target")).unwrap();
        let workspace_manifest = "[workspace]\nmembers = [\"module\"]\nresolver = \"3\"\n[workspace.dependencies]\nspacetimedb = \"=2.8.3\"\n";
        fs::write(workspace.join("Cargo.toml"), workspace_manifest).unwrap();
        fs::write(workspace.join("Cargo.lock"), "version = 4\n").unwrap();
        fs::write(root.join("Cargo.toml"), "[package]\nname = \"demo\"\nversion = \"0.1.0\"\nedition = \"2024\"\n[dependencies]\nspacetimedb.workspace = true\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, SPACETIME_VERSION).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, SPACETIME_VERSION).unwrap(),
            original
        );
        assert_ne!(
            spacetime_module_fingerprint(&root, "different-runtime").unwrap(),
            original
        );

        fs::write(
            workspace.join("Cargo.toml"),
            workspace_manifest.replace("2.8.3", "2.11.0"),
        )
        .unwrap();
        assert_ne!(
            spacetime_module_fingerprint(&root, SPACETIME_VERSION).unwrap(),
            original
        );
        fs::write(workspace.join("Cargo.toml"), workspace_manifest).unwrap();
        fs::write(
            workspace.join("Cargo.lock"),
            "version = 4\n# dependency revision changed\n",
        )
        .unwrap();
        assert_ne!(
            spacetime_module_fingerprint(&root, SPACETIME_VERSION).unwrap(),
            original
        );
        fs::write(workspace.join("Cargo.lock"), "version = 4\n").unwrap();
        assert_eq!(
            spacetime_module_fingerprint(&root, SPACETIME_VERSION).unwrap(),
            original
        );

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

        let standalone = workspace.join("standalone");
        fs::create_dir_all(standalone.join("src")).unwrap();
        fs::write(standalone.join("Cargo.toml"), "[workspace]\n[package]\nname = \"standalone\"\nversion = \"0.1.0\"\nedition = \"2024\"\n").unwrap();
        fs::write(standalone.join("src/lib.rs"), "pub fn reducer() {}\n").unwrap();
        let isolated = spacetime_module_fingerprint(&standalone, SPACETIME_VERSION).unwrap();
        fs::write(
            workspace.join("Cargo.toml"),
            workspace_manifest.replace("2.8.3", "2.11.0"),
        )
        .unwrap();
        assert_eq!(
            spacetime_module_fingerprint(&standalone, SPACETIME_VERSION).unwrap(),
            isolated
        );

        let selected = workspace.join("selected");
        let external = workspace.join("external");
        fs::create_dir_all(&selected).unwrap();
        fs::create_dir_all(external.join("src")).unwrap();
        fs::write(
            selected.join("Cargo.toml"),
            "[workspace]\nmembers = [\"../external\"]\n",
        )
        .unwrap();
        fs::write(
            external.join("Cargo.toml"),
            "[package]\nname = \"external\"\nversion = \"0.1.0\"\nworkspace = \"../selected\"\n",
        )
        .unwrap();
        fs::write(external.join("src/lib.rs"), "pub fn reducer() {}\n").unwrap();
        let explicit = spacetime_module_fingerprint(&external, SPACETIME_VERSION).unwrap();
        fs::write(workspace.join("Cargo.toml"), workspace_manifest).unwrap();
        assert_eq!(
            spacetime_module_fingerprint(&external, SPACETIME_VERSION).unwrap(),
            explicit
        );
        fs::write(selected.join("Cargo.lock"), "version = 4\n").unwrap();
        assert_ne!(
            spacetime_module_fingerprint(&external, SPACETIME_VERSION).unwrap(),
            explicit
        );
    }
    #[test]
    #[ignore = "subprocess fixture for process lifecycle regression tests"]
    fn process_fixture() {
        let Ok(case) = std::env::var("RUNTIME_TEST_PROCESS_CASE") else {
            return;
        };
        match case.as_str() {
            "output" => {
                let bytes = vec![b'x'; 1024 * 1024];
                std::io::stdout().write_all(&bytes).unwrap();
                std::io::stderr().write_all(&bytes).unwrap();
            }
            "wait" => {
                let path = std::env::var_os("RUNTIME_TEST_PROCESS_PID").unwrap();
                fs::write(path, std::process::id().to_string()).unwrap();
                thread::sleep(Duration::from_secs(30));
            }
            other => panic!("unknown subprocess fixture case: {other}"),
        }
    }

    fn process_fixture_command(case: &str) -> Command {
        let mut command = Command::new(std::env::current_exe().unwrap());
        command
            .args([
                "--ignored",
                "--exact",
                "docker::tests::process_fixture",
                "--nocapture",
            ])
            .env("RUNTIME_TEST_PROCESS_CASE", case)
            .stdout(Stdio::piped())
            .stderr(Stdio::piped());
        command
    }

    #[tokio::test]
    async fn command_drains_large_stdout_and_stderr_without_a_false_timeout() {
        let command = process_fixture_command("output");
        let result = run_command_with_timeout(command, Duration::from_secs(10), "output fixture")
            .await
            .unwrap();
        assert!(result.status.success());
        assert!(result.stdout.len() >= 1024 * 1024);
        assert_eq!(result.stderr.len(), 1024 * 1024);
    }

    #[tokio::test]
    async fn command_timeout_yields_to_the_runtime() {
        let directory = tempfile::tempdir().unwrap();
        let mut command = process_fixture_command("wait");
        command.env(
            "RUNTIME_TEST_PROCESS_PID",
            directory.path().join("process.pid"),
        );
        let command = run_command_with_timeout(command, Duration::from_secs(1), "waiting fixture");
        let started = Instant::now();
        let timer = async {
            tokio::time::sleep(Duration::from_millis(20)).await;
            assert!(
                started.elapsed() < Duration::from_millis(500),
                "process waiting blocked the Tokio worker"
            );
        };
        let (result, ()) = tokio::join!(command, timer);
        assert!(result.unwrap_err().to_string().contains("timed out"));
    }

    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn cancelling_a_command_terminates_its_child() {
        let directory = tempfile::tempdir().unwrap();
        let pid_file = directory.path().join("process.pid");
        let mut command = process_fixture_command("wait");
        command.env("RUNTIME_TEST_PROCESS_PID", &pid_file);
        let task = tokio::spawn(run_command_with_timeout(
            command,
            Duration::from_secs(20),
            "cancel fixture",
        ));
        let pid = tokio::time::timeout(Duration::from_secs(10), async {
            loop {
                if let Ok(value) = tokio::fs::read_to_string(&pid_file).await {
                    break value.parse::<u32>().unwrap();
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .unwrap();
        task.abort();
        assert!(task.await.unwrap_err().is_cancelled());
        tokio::time::timeout(Duration::from_secs(5), async {
            while Path::new(&format!("/proc/{pid}")).exists() {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("cancelled process remained alive or unreaped");
    }

    #[tokio::test]
    async fn contended_file_lock_yields_until_its_owner_finishes() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("lifecycle.lock");
        let first = OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&path)
            .unwrap();
        let second = OpenOptions::new()
            .read(true)
            .write(true)
            .open(&path)
            .unwrap();
        first.lock().unwrap();
        let owner = tokio::spawn(async move {
            let _guard = first;
            tokio::time::sleep(Duration::from_millis(30)).await;
        });
        lock_file_until(&second, Duration::from_secs(2), "contended fixture")
            .await
            .unwrap();
        owner.await.unwrap();
    }
}