use std::fs;
use std::io::{Read, Write};
use std::net::{TcpStream, ToSocketAddrs};
use std::path::{Component, Path, PathBuf};
use std::process::{Command, Output, Stdio};
use std::thread;
use std::time::{Duration, Instant};
use crate::__internals::Environment;
use crate::LocalStack;
use anyhow::{Context, Result, anyhow};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use 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 SPACETIME_MODULE_READY_TIMEOUT: Duration = Duration::from_secs(30);
const SPACETIME_PUBLISH_TIMEOUT: Duration = Duration::from_secs(300);
#[cfg(any(target_os = "windows", target_os = "macos"))]
const DOCKER_DESKTOP_READY_TIMEOUT: Duration = Duration::from_secs(180);
pub(crate) fn should_bootstrap_local_stack(environment: Environment, connection_url: &str) -> bool {
if !environment.allows_local_resources() {
return false;
}
if std::env::var_os("CI").is_some() {
debug!("skipping compose stack bootstrap in CI environment");
return false;
}
if !is_local_database(connection_url) {
debug!("service url points to remote host; skipping compose stack bootstrap");
return false;
}
if local_database_available(connection_url) {
debug!(
"service url already resolves to a reachable local endpoint; skipping compose stack bootstrap"
);
return false;
}
if cfg!(target_os = "linux") {
debug!("skipping compose stack bootstrap on Linux; expecting local services");
return false;
}
true
}
pub(crate) async fn ensure_local_stack(local_stack: &LocalStack) -> Result<()> {
ensure_local_stack_with_profiles(local_stack, &[], None).await
}
pub(crate) async fn ensure_local_stack_with_profiles(
local_stack: &LocalStack,
profiles: &[&str],
spacetime_endpoint_url: Option<&str>,
) -> Result<()> {
ensure_docker_desktop_running()?;
let compose_path = bundled_compose_path();
if !compose_path.exists() {
return Err(anyhow!(
"bundled compose file not found at {}",
compose_path.display()
));
}
let project_name = compose_project_name(local_stack.project_name());
let env_path = local_stack.env_path();
if !env_path.exists() {
return Err(anyhow!(
"local stack env file not found at {}",
env_path.display()
));
}
info!("ensuring development compose stack is running");
run_docker_compose_up(&compose_path, &project_name, env_path, profiles).with_context(|| {
format!(
"failed to run docker compose up using {}",
compose_path.display()
)
})?;
if profiles.contains(&"spacetime") {
let fallback_endpoint;
let endpoint_url = match spacetime_endpoint_url {
Some(endpoint_url) => endpoint_url,
None => {
let port = resolve_spacetime_compose_port(env_path);
fallback_endpoint = format!("http://127.0.0.1:{port}");
fallback_endpoint.as_str()
}
};
ensure_local_spacetime_host_endpoint(endpoint_url)?;
}
Ok(())
}
pub(crate) fn ensure_local_spacetime_module(
local_stack: &LocalStack,
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 module_fingerprint = spacetime_module_fingerprint(module_path)?;
let module_status = wait_for_spacetime_module_status(endpoint_url, module_name)?;
let reset_before_publish = matches!(module_status, SpacetimeModuleStatus::Available);
let publish_reason = match module_status {
SpacetimeModuleStatus::Missing => "missing",
SpacetimeModuleStatus::Available
if spacetime_module_fingerprint_current(
endpoint_url,
module_name,
module_fingerprint.as_str(),
)? =>
{
let credentials = load_local_spacetime_owner_credentials(endpoint_url, false)?;
write_spacetime_cli_token(endpoint_url, credentials.token.as_str())?;
return Ok(Some(credentials.token));
}
SpacetimeModuleStatus::Available => "source fingerprint changed",
};
info!(
spacetime_url = endpoint_url,
module_name,
module_path = %module_path.display(),
publish_reason,
"preparing local SpacetimeDB module"
);
if reset_before_publish {
reset_local_spacetime_module_state(local_stack, endpoint_url)?;
let _ = wait_for_spacetime_module_status(endpoint_url, module_name)?;
}
let credentials = load_local_spacetime_owner_credentials(endpoint_url, true)?;
write_spacetime_cli_token(endpoint_url, credentials.token.as_str())?;
publish_local_spacetime_module(endpoint_url, module_name, module_path)?;
let deadline = Instant::now() + SPACETIME_MODULE_READY_TIMEOUT;
loop {
match spacetime_module_status(endpoint_url, module_name)? {
SpacetimeModuleStatus::Available => {
write_spacetime_module_fingerprint(
endpoint_url,
module_name,
module_fingerprint.as_str(),
)?;
return Ok(Some(credentials.token));
}
SpacetimeModuleStatus::Missing if Instant::now() < deadline => {
thread::sleep(COMMAND_POLL_INTERVAL);
}
SpacetimeModuleStatus::Missing => {
return Err(anyhow!(
"SpacetimeDB module `{module_name}` was published but is still missing at {endpoint_url}"
));
}
}
}
}
fn reset_local_spacetime_module_state(local_stack: &LocalStack, endpoint_url: &str) -> Result<()> {
if !is_local_database(endpoint_url) {
return Ok(());
}
ensure_docker_desktop_running()?;
let compose_path = bundled_compose_path();
if !compose_path.exists() {
return Err(anyhow!(
"compose file not found at {} for local SpacetimeDB reset",
compose_path.display()
));
}
let project_name = compose_project_name(local_stack.project_name());
let env_path = local_stack.env_path();
info!(
spacetime_url = endpoint_url,
project_name, "resetting local SpacetimeDB module state after source fingerprint drift"
);
run_docker_compose_best_effort(
&compose_path,
&project_name,
env_path,
&["spacetime"],
&["rm", "--force", "--stop", "spacetimedb", "spacetimedb-init"],
"`docker compose rm spacetimedb`",
);
run_docker_best_effort(
&[
"volume",
"rm",
"--force",
&format!("{project_name}_spacetimedb-data"),
],
"`docker volume rm spacetimedb-data`",
);
run_docker_compose_up(&compose_path, &project_name, env_path, &["spacetime"])
.with_context(|| "failed to restart local SpacetimeDB after reset")?;
Ok(())
}
fn run_docker_compose_up(
compose_path: &Path,
project_name: &str,
env_path: &Path,
profiles: &[&str],
) -> Result<()> {
let mut command = Command::new("docker");
command
.arg("compose")
.arg("--file")
.arg(compose_path)
.arg("--project-name")
.arg(project_name);
if env_path.exists() {
command.arg("--env-file").arg(env_path);
}
for profile in profiles {
command.arg("--profile").arg(profile);
}
let output = command
.arg("up")
.arg("--detach")
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.output()
.with_context(|| "failed to spawn docker compose")?;
if output.status.success() {
return Ok(());
}
let stdout = String::from_utf8_lossy(&output.stdout).trim().to_string();
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
let mut detail = format!("docker compose exited with {}", output.status);
if !stdout.is_empty() {
detail.push_str("\nstdout:\n");
detail.push_str(&stdout);
}
if !stderr.is_empty() {
detail.push_str("\nstderr:\n");
detail.push_str(&stderr);
}
Err(anyhow!(detail))
}
fn ensure_local_spacetime_host_endpoint(endpoint_url: &str) -> Result<()> {
if !is_local_database(endpoint_url) {
return Ok(());
}
if local_database_available(endpoint_url) {
return Ok(());
}
let (host, port) = host_and_port_from_connection_url(endpoint_url)
.ok_or_else(|| anyhow!("invalid local SpacetimeDB endpoint URL: {endpoint_url}"))?;
let mut message = format!(
"local SpacetimeDB endpoint {endpoint_url} did not become reachable after docker compose up"
);
if is_loopback(host.as_str()) && docker_container_name_for_host_port(port)?.is_none() {
message.push_str(&format!(
"; docker ps does not show a published host binding for port {port}"
));
}
if let Some(diagnostic) = local_spacetime_endpoint_diagnostic(endpoint_url) {
message.push_str("; ");
message.push_str(diagnostic.as_str());
}
Err(anyhow!(message))
}
fn resolve_spacetime_compose_port(env_path: &Path) -> u16 {
if let Ok(raw) = std::env::var("SPACETIME_PORT")
&& let Some(port) = parse_spacetime_port(raw.as_str())
{
return port;
}
fs::read_to_string(env_path)
.ok()
.and_then(|contents| parse_spacetime_compose_port(contents.as_str()))
.unwrap_or(4530)
}
fn parse_spacetime_compose_port(contents: &str) -> Option<u16> {
contents.lines().find_map(|line| {
let trimmed = line.trim();
if trimmed.is_empty() || trimmed.starts_with('#') {
return None;
}
let (key, value) = trimmed.split_once('=')?;
(key.trim() == "SPACETIME_PORT")
.then(|| parse_spacetime_port(value))
.flatten()
})
}
fn parse_spacetime_port(value: &str) -> Option<u16> {
let value = value
.trim()
.trim_matches(|character| character == '"' || character == '\'');
value.parse::<u16>().ok().filter(|port| *port > 0)
}
fn run_docker_compose_best_effort(
compose_path: &Path,
project_name: &str,
env_path: &Path,
profiles: &[&str],
args: &[&str],
label: &str,
) {
let mut command = Command::new("docker");
command
.arg("compose")
.arg("--file")
.arg(compose_path)
.arg("--project-name")
.arg(project_name);
if env_path.exists() {
command.arg("--env-file").arg(env_path);
}
for profile in profiles {
command.arg("--profile").arg(profile);
}
for arg in args {
command.arg(arg);
}
run_docker_command_best_effort(command, label);
}
fn run_docker_best_effort(args: &[&str], label: &str) {
let mut command = Command::new("docker");
for arg in args {
command.arg(arg);
}
run_docker_command_best_effort(command, label);
}
fn run_docker_command_best_effort(mut command: Command, label: &str) {
match run_command_with_timeout(&mut command, SPACETIME_MODULE_READY_TIMEOUT, label) {
Ok(output) if output.status.success() => {}
Ok(output) => warn!(
status = %output.status,
stderr = %String::from_utf8_lossy(&output.stderr).trim(),
stdout = %String::from_utf8_lossy(&output.stdout).trim(),
"{label} did not complete cleanly during best-effort local SpacetimeDB reset"
),
Err(error) => warn!(%error, "{label} failed during best-effort local SpacetimeDB reset"),
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SpacetimeModuleStatus {
Available,
Missing,
}
fn wait_for_spacetime_module_status(
endpoint_url: &str,
module_name: &str,
) -> Result<SpacetimeModuleStatus> {
let deadline = Instant::now() + SPACETIME_MODULE_READY_TIMEOUT;
loop {
match spacetime_module_status(endpoint_url, module_name) {
Ok(status) => return Ok(status),
Err(error) if Instant::now() < deadline => {
debug!(
%error,
"local SpacetimeDB endpoint is not ready for module check; retrying"
);
thread::sleep(COMMAND_POLL_INTERVAL);
}
Err(error) => {
return Err(spacetime_module_ready_error(endpoint_url, error));
}
}
}
}
fn spacetime_module_ready_error(endpoint_url: &str, error: anyhow::Error) -> anyhow::Error {
let mut message = format!(
"local SpacetimeDB endpoint {endpoint_url} did not become ready for module check: {error}"
);
if let Some(diagnostic) = local_spacetime_endpoint_diagnostic(endpoint_url) {
message.push_str("; ");
message.push_str(diagnostic.as_str());
}
anyhow!(message)
}
fn local_spacetime_endpoint_diagnostic(endpoint_url: &str) -> Option<String> {
let (host, port) = host_and_port_from_connection_url(endpoint_url)?;
if !is_loopback(host.as_str()) {
return None;
}
windows_tcp_excluded_port_diagnostic(port)
}
#[cfg(target_os = "windows")]
fn windows_tcp_excluded_port_diagnostic(port: u16) -> Option<String> {
let output = Command::new("netsh")
.arg("interface")
.arg("ipv4")
.arg("show")
.arg("excludedportrange")
.arg("protocol=tcp")
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.output()
.ok()?;
if !output.status.success() {
return None;
}
let stdout = String::from_utf8_lossy(&output.stdout);
let (start, end) = windows_excluded_tcp_port_range(stdout.as_ref(), port)?;
Some(format!(
"Windows TCP port {port} is inside excluded range {start}-{end}; Docker cannot publish the local SpacetimeDB host port until that reservation is repaired with elevated Windows network permissions"
))
}
#[cfg(not(target_os = "windows"))]
fn windows_tcp_excluded_port_diagnostic(_port: u16) -> Option<String> {
None
}
#[cfg(any(target_os = "windows", test))]
fn windows_excluded_tcp_port_range(output: &str, port: u16) -> Option<(u16, u16)> {
output.lines().find_map(|line| {
let mut columns = line.split_whitespace();
let start = columns.next()?.parse::<u16>().ok()?;
let end = columns.next()?.parse::<u16>().ok()?;
(start <= port && port <= end).then_some((start, end))
})
}
fn spacetime_module_fingerprint(module_path: &Path) -> Result<String> {
let mut files = Vec::new();
collect_spacetime_module_fingerprint_files(module_path, module_path, &mut files)?;
files.sort();
let mut hasher = Sha256::new();
hasher.update([1]);
for file in files {
let relative = file.strip_prefix(module_path).with_context(|| {
format!(
"failed to resolve SpacetimeDB module file {} relative to {}",
file.display(),
module_path.display()
)
})?;
let relative = relative.to_string_lossy().replace('\\', "/");
hasher.update(relative.as_bytes());
hasher.update([0]);
let contents = fs::read(&file).with_context(|| {
format!(
"failed to read SpacetimeDB module fingerprint file {}",
file.display()
)
})?;
hasher.update(contents);
hasher.update([0]);
}
Ok(hex_digest(&hasher.finalize()))
}
fn collect_spacetime_module_fingerprint_files(
root: &Path,
current: &Path,
files: &mut Vec<PathBuf>,
) -> Result<()> {
for entry in fs::read_dir(current).with_context(|| {
format!(
"failed to read SpacetimeDB module fingerprint directory {}",
current.display()
)
})? {
let entry = entry?;
let path = entry.path();
if path.is_dir() {
if should_skip_spacetime_module_fingerprint_dir(root, &path) {
continue;
}
collect_spacetime_module_fingerprint_files(root, &path, files)?;
continue;
}
if should_include_spacetime_module_fingerprint_file(root, &path) {
files.push(path);
}
}
Ok(())
}
fn should_skip_spacetime_module_fingerprint_dir(root: &Path, path: &Path) -> bool {
let Ok(relative) = path.strip_prefix(root) else {
return true;
};
if relative == Path::new("target") || relative == Path::new(".git") {
return true;
}
relative == Path::new("src").join("generated")
}
fn should_include_spacetime_module_fingerprint_file(root: &Path, path: &Path) -> bool {
let Ok(relative) = path.strip_prefix(root) else {
return false;
};
if relative == Path::new("Cargo.toml") || relative == Path::new("Cargo.lock") {
return true;
}
relative
.components()
.next()
.is_some_and(|component| component.as_os_str() == "src")
&& path
.extension()
.and_then(|extension| extension.to_str())
.is_some_and(|extension| extension == "rs")
}
fn spacetime_module_fingerprint_current(
endpoint_url: &str,
module_name: &str,
module_fingerprint: &str,
) -> Result<bool> {
let path = spacetime_module_fingerprint_path(endpoint_url, module_name)?;
let Ok(stored) = fs::read_to_string(path) else {
return Ok(false);
};
Ok(stored.trim() == module_fingerprint)
}
fn write_spacetime_module_fingerprint(
endpoint_url: &str,
module_name: &str,
module_fingerprint: &str,
) -> Result<()> {
let path = spacetime_module_fingerprint_path(endpoint_url, module_name)?;
if let Some(parent) = path.parent() {
fs::create_dir_all(parent).with_context(|| {
format!(
"failed to create SpacetimeDB module fingerprint directory {}",
parent.display()
)
})?;
}
fs::write(&path, module_fingerprint).with_context(|| {
format!(
"failed to write SpacetimeDB module fingerprint at {}",
path.display()
)
})
}
fn spacetime_module_fingerprint_path(endpoint_url: &str, module_name: &str) -> Result<PathBuf> {
let endpoint_slug = spacetime_endpoint_slug(endpoint_url)?;
let safe_module_name = safe_path_segment(module_name);
let home = std::env::var_os("USERPROFILE")
.or_else(|| std::env::var_os("HOME"))
.ok_or_else(|| {
anyhow!("failed to resolve home directory for SpacetimeDB module fingerprint")
})?;
Ok(PathBuf::from(home)
.join(".config")
.join("spacetimedb-module-fingerprints")
.join(endpoint_slug)
.join(format!("{safe_module_name}.sha256")))
}
fn spacetime_module_status(endpoint_url: &str, module_name: &str) -> Result<SpacetimeModuleStatus> {
let (mut stream, host, port) = connect_local_spacetime_http(endpoint_url, "schema probe")?;
let path = format!("/v1/database/{module_name}/schema?version=9");
let host_header = host_header(&host, port);
let request =
format!("GET {path} HTTP/1.1\r\nHost: {host_header}\r\nConnection: close\r\n\r\n");
stream
.write_all(request.as_bytes())
.with_context(|| "failed to write SpacetimeDB schema probe request")?;
let mut response = String::new();
stream
.read_to_string(&mut response)
.with_context(|| "failed to read SpacetimeDB schema probe response")?;
let status_line = response
.lines()
.next()
.ok_or_else(|| anyhow!("SpacetimeDB schema probe returned an empty response"))?;
let status_code = status_line
.split_whitespace()
.nth(1)
.ok_or_else(|| anyhow!("SpacetimeDB schema probe returned invalid status: {status_line}"))?
.parse::<u16>()
.with_context(|| {
format!("SpacetimeDB schema probe returned invalid status: {status_line}")
})?;
match status_code {
200..=299 | 401 | 403 => Ok(SpacetimeModuleStatus::Available),
404 => Ok(SpacetimeModuleStatus::Missing),
code => Err(anyhow!(
"SpacetimeDB schema probe for `{module_name}` returned HTTP {code}"
)),
}
}
fn connect_local_http(host: &str, port: u16) -> Result<TcpStream> {
let mut candidates = Vec::new();
if let Ok(ip) = host.parse::<std::net::IpAddr>() {
candidates.push(std::net::SocketAddr::new(ip, port));
} else if let Ok(resolved) = (host, port).to_socket_addrs() {
candidates.extend(resolved);
}
let mut last_error = None;
for addr in candidates {
match TcpStream::connect_timeout(&addr, LOCAL_DATABASE_PROBE_TIMEOUT) {
Ok(stream) => return Ok(stream),
Err(error) => last_error = Some(error),
}
}
Err(anyhow!(
"failed to connect to local SpacetimeDB endpoint {host}:{port}: {}",
last_error
.map(|error| error.to_string())
.unwrap_or_else(|| "host did not resolve".to_string())
))
}
fn host_header(host: &str, port: u16) -> String {
if host.contains(':') && !host.starts_with('[') {
format!("[{host}]:{port}")
} else {
format!("{host}:{port}")
}
}
#[derive(Debug, Deserialize, Serialize)]
struct LocalSpacetimeOwnerCredentials {
identity: String,
token: String,
}
fn load_local_spacetime_owner_credentials(
endpoint_url: &str,
allow_create: bool,
) -> Result<LocalSpacetimeOwnerCredentials> {
let credential_path = spacetime_docker_config_dir(endpoint_url)?.join("module-owner.json");
if credential_path.is_file() {
let bytes = fs::read(&credential_path).with_context(|| {
format!(
"failed to read local SpacetimeDB module-owner credentials at {}",
credential_path.display()
)
})?;
let credentials = serde_json::from_slice::<LocalSpacetimeOwnerCredentials>(&bytes)
.with_context(|| {
format!(
"failed to parse local SpacetimeDB module-owner credentials at {}",
credential_path.display()
)
})?;
if verify_local_spacetime_identity(endpoint_url, &credentials)? {
return Ok(credentials);
}
if !allow_create {
return Err(anyhow!(
"the saved local SpacetimeDB module-owner credential is no longer valid; the local module cannot be opened with a replacement identity"
));
}
} else if !allow_create {
return Err(anyhow!(
"the local SpacetimeDB module exists without its module-owner credential; reset the local SpacetimeDB development volume before continuing"
));
}
let credentials = create_local_spacetime_identity(endpoint_url)?;
let parent = credential_path.parent().ok_or_else(|| {
anyhow!(
"local SpacetimeDB credential path has no parent: {}",
credential_path.display()
)
})?;
fs::create_dir_all(parent).with_context(|| {
format!(
"failed to create local SpacetimeDB credential directory at {}",
parent.display()
)
})?;
let encoded = serde_json::to_vec(&credentials)
.with_context(|| "failed to encode local SpacetimeDB module-owner credentials")?;
let temporary_path = credential_path.with_extension("json.tmp");
write_private_file(&temporary_path, &encoded)?;
if credential_path.exists() {
fs::remove_file(&credential_path).with_context(|| {
format!(
"failed to replace invalid local SpacetimeDB module-owner credentials at {}",
credential_path.display()
)
})?;
}
fs::rename(&temporary_path, &credential_path).with_context(|| {
format!(
"failed to persist local SpacetimeDB module-owner credentials at {}",
credential_path.display()
)
})?;
Ok(credentials)
}
fn create_local_spacetime_identity(endpoint_url: &str) -> Result<LocalSpacetimeOwnerCredentials> {
let (status, body) = local_spacetime_http_request(endpoint_url, "POST", "/v1/identity", None)?;
if !(200..=299).contains(&status) {
return Err(anyhow!(
"local SpacetimeDB identity creation returned HTTP {status}"
));
}
let credentials = serde_json::from_slice::<LocalSpacetimeOwnerCredentials>(&body)
.with_context(|| "local SpacetimeDB identity creation returned invalid JSON")?;
if credentials.identity.trim().is_empty() || credentials.token.trim().is_empty() {
return Err(anyhow!(
"local SpacetimeDB identity creation returned empty credentials"
));
}
Ok(credentials)
}
fn verify_local_spacetime_identity(
endpoint_url: &str,
credentials: &LocalSpacetimeOwnerCredentials,
) -> Result<bool> {
if credentials.identity.trim().is_empty() || credentials.token.trim().is_empty() {
return Ok(false);
}
let path = format!("/v1/identity/{}/verify", credentials.identity.trim());
let (status, _) = local_spacetime_http_request(
endpoint_url,
"GET",
path.as_str(),
Some(credentials.token.trim()),
)?;
match status {
200..=299 => Ok(true),
400 | 401 | 403 | 404 => Ok(false),
_ => Err(anyhow!(
"local SpacetimeDB identity verification returned HTTP {status}"
)),
}
}
fn local_spacetime_http_request(
endpoint_url: &str,
method: &str,
path: &str,
bearer_token: Option<&str>,
) -> Result<(u16, Vec<u8>)> {
let (mut stream, host, port) = connect_local_spacetime_http(endpoint_url, "identity request")?;
let mut request = format!(
"{method} {path} HTTP/1.1\r\nHost: {}\r\nAccept: application/json\r\nContent-Length: 0\r\nConnection: close\r\n",
host_header(&host, port)
);
if let Some(token) = bearer_token {
request.push_str("Authorization: Bearer ");
request.push_str(token);
request.push_str("\r\n");
}
request.push_str("\r\n");
stream
.write_all(request.as_bytes())
.with_context(|| "failed to write local SpacetimeDB identity request")?;
let mut response = Vec::new();
stream
.read_to_end(&mut response)
.with_context(|| "failed to read local SpacetimeDB identity response")?;
parse_http_response(&response)
}
fn connect_local_spacetime_http(
endpoint_url: &str,
operation: &'static str,
) -> Result<(TcpStream, String, u16)> {
let parsed = Url::parse(endpoint_url)
.with_context(|| format!("invalid SpacetimeDB endpoint URL: {endpoint_url}"))?;
if parsed.scheme() != "http" {
return Err(anyhow!(
"local SpacetimeDB {operation} requires an http endpoint, got {endpoint_url}"
));
}
let host = parsed
.host_str()
.ok_or_else(|| anyhow!("SpacetimeDB endpoint URL has no host: {endpoint_url}"))?;
let port = parsed
.port_or_known_default()
.ok_or_else(|| anyhow!("SpacetimeDB endpoint URL has no port: {endpoint_url}"))?;
let stream = connect_local_http(host, port)?;
stream
.set_read_timeout(Some(Duration::from_secs(2)))
.with_context(|| format!("failed to set local SpacetimeDB {operation} read timeout"))?;
stream
.set_write_timeout(Some(Duration::from_secs(2)))
.with_context(|| format!("failed to set local SpacetimeDB {operation} write timeout"))?;
Ok((stream, host.to_string(), port))
}
fn parse_http_response(response: &[u8]) -> Result<(u16, Vec<u8>)> {
let header_end = response
.windows(4)
.position(|window| window == b"\r\n\r\n")
.ok_or_else(|| anyhow!("local SpacetimeDB returned an invalid HTTP response"))?;
let headers = std::str::from_utf8(&response[..header_end])
.with_context(|| "local SpacetimeDB returned non-UTF-8 HTTP headers")?;
let status_line = headers
.lines()
.next()
.ok_or_else(|| anyhow!("local SpacetimeDB returned an empty HTTP response"))?;
let status = status_line
.split_whitespace()
.nth(1)
.ok_or_else(|| anyhow!("local SpacetimeDB returned invalid HTTP status: {status_line}"))?
.parse::<u16>()
.with_context(|| {
format!("local SpacetimeDB returned invalid HTTP status: {status_line}")
})?;
let body = &response[header_end + 4..];
if headers.lines().any(|line| {
line.split_once(':').is_some_and(|(name, value)| {
name.eq_ignore_ascii_case("transfer-encoding")
&& value
.split(',')
.any(|encoding| encoding.trim().eq_ignore_ascii_case("chunked"))
})
}) {
return Ok((status, decode_chunked_http_body(body)?));
}
Ok((status, body.to_vec()))
}
fn decode_chunked_http_body(mut body: &[u8]) -> Result<Vec<u8>> {
let mut decoded = Vec::new();
loop {
let line_end = body
.windows(2)
.position(|window| window == b"\r\n")
.ok_or_else(|| anyhow!("local SpacetimeDB returned an invalid chunked response"))?;
let size = std::str::from_utf8(&body[..line_end])
.with_context(|| "local SpacetimeDB returned a non-UTF-8 chunk size")?
.split(';')
.next()
.unwrap_or_default()
.trim();
let size = usize::from_str_radix(size, 16)
.with_context(|| "local SpacetimeDB returned an invalid chunk size")?;
body = &body[line_end + 2..];
if size == 0 {
return Ok(decoded);
}
if body.len() < size + 2 || &body[size..size + 2] != b"\r\n" {
return Err(anyhow!(
"local SpacetimeDB returned a truncated chunked response"
));
}
decoded.extend_from_slice(&body[..size]);
body = &body[size + 2..];
}
}
fn publish_local_spacetime_module(
endpoint_url: &str,
module_name: &str,
module_path: &Path,
) -> Result<()> {
if command_available("spacetime") {
let mut host_command =
host_spacetime_publish_command(endpoint_url, module_name, module_path)?;
let host_output = run_command_with_timeout(
&mut host_command,
SPACETIME_PUBLISH_TIMEOUT,
"`spacetime publish`",
)?;
if host_output.status.success() {
return Ok(());
}
let host_stdout = String::from_utf8_lossy(&host_output.stdout)
.trim()
.to_string();
let host_stderr = String::from_utf8_lossy(&host_output.stderr)
.trim()
.to_string();
if should_retry_spacetime_publish_with_docker(&host_stdout, &host_stderr) {
let mut docker_command =
docker_spacetime_publish_command(endpoint_url, module_name, module_path)?;
let docker_output = run_command_with_timeout(
&mut docker_command,
SPACETIME_PUBLISH_TIMEOUT,
"`spacetime publish` (docker fallback)",
)?;
if docker_output.status.success() {
return Ok(());
}
return Err(anyhow!(
"{}\n\ndocker fallback:\n{}",
format_spacetime_publish_failure(&host_output),
format_spacetime_publish_failure(&docker_output),
));
}
return Err(anyhow!(format_spacetime_publish_failure(&host_output)));
}
let mut docker_command =
docker_spacetime_publish_command(endpoint_url, module_name, module_path)?;
let docker_output = run_command_with_timeout(
&mut docker_command,
SPACETIME_PUBLISH_TIMEOUT,
"`spacetime publish`",
)?;
if docker_output.status.success() {
return Ok(());
}
Err(anyhow!(format_spacetime_publish_failure(&docker_output)))
}
fn should_retry_spacetime_publish_with_docker(stdout: &str, stderr: &str) -> bool {
let combined = format!("{stdout}\n{stderr}");
combined.contains("failed to load manifest for dependency")
&& combined.contains("/workspace/")
&& combined.contains("No such file or directory")
}
fn format_spacetime_publish_failure(output: &std::process::Output) -> String {
let stdout = String::from_utf8_lossy(&output.stdout).trim().to_string();
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
let mut detail = format!("spacetime publish exited with {}", output.status);
if !stdout.is_empty() {
detail.push_str("\nstdout:\n");
detail.push_str(&stdout);
}
if !stderr.is_empty() {
detail.push_str("\nstderr:\n");
detail.push_str(&stderr);
}
detail
}
fn host_spacetime_publish_command(
endpoint_url: &str,
module_name: &str,
module_path: &Path,
) -> Result<Command> {
let config_dir = spacetime_docker_config_dir(endpoint_url)?;
fs::create_dir_all(&config_dir).with_context(|| {
format!(
"failed to create SpacetimeDB CLI config directory at {}",
config_dir.display()
)
})?;
let config_path = config_dir.join("cli.toml");
let mut command = Command::new("spacetime");
command
.arg("--config-path")
.arg(config_path)
.arg("publish")
.arg("--server")
.arg(endpoint_url)
.arg("--module-path")
.arg(module_path)
.arg("--break-clients")
.arg("--delete-data=on-conflict")
.arg("--yes")
.arg(module_name)
.stdout(Stdio::piped())
.stderr(Stdio::piped());
Ok(command)
}
fn docker_spacetime_publish_command(
endpoint_url: &str,
module_name: &str,
module_path: &Path,
) -> Result<Command> {
ensure_docker_desktop_running()?;
let mount_root = spacetime_docker_mount_root(module_path)?;
let module_path_in_container = container_workspace_path(&mount_root, module_path)?;
let config_dir = spacetime_docker_config_dir(endpoint_url)?;
fs::create_dir_all(&config_dir).with_context(|| {
format!(
"failed to create SpacetimeDB docker config directory at {}",
config_dir.display()
)
})?;
let mut command = Command::new("docker");
command
.arg("run")
.arg("--rm")
.arg("-e")
.arg("CARGO_TARGET_DIR=/tmp/spacetimedb-module-target")
.arg("-v")
.arg(format!("{}:/workspace:ro", mount_root.display()))
.arg("-v")
.arg(format!(
"{}:/home/spacetime/.config/spacetime",
config_dir.display()
))
.arg("-w")
.arg("/workspace");
let publish_target = docker_spacetime_publish_target(endpoint_url)?;
if let Some(network) = publish_target.network_arg.as_deref() {
command.arg("--network").arg(network);
}
command
.arg("clockworklabs/spacetime:v2.6.1")
.arg("publish")
.arg("--server")
.arg(&publish_target.server_url)
.arg("--module-path")
.arg(module_path_in_container)
.arg("--break-clients")
.arg("--delete-data=on-conflict")
.arg("--yes")
.arg(module_name)
.stdout(Stdio::piped())
.stderr(Stdio::piped());
Ok(command)
}
fn safe_path_segment(value: &str) -> String {
value
.chars()
.map(|character| {
if character.is_ascii_alphanumeric() {
character.to_ascii_lowercase()
} else {
'-'
}
})
.collect::<String>()
.trim_matches('-')
.to_string()
}
fn hex_digest(bytes: &[u8]) -> String {
const HEX: &[u8; 16] = b"0123456789abcdef";
let mut output = String::with_capacity(bytes.len() * 2);
for byte in bytes {
output.push(HEX[(byte >> 4) as usize] as char);
output.push(HEX[(byte & 0x0f) as usize] as char);
}
output
}
fn command_available(name: &str) -> bool {
Command::new(name)
.arg("--version")
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.map(|status| status.success())
.unwrap_or(false)
}
fn spacetime_docker_mount_root(module_path: &Path) -> Result<PathBuf> {
let app_root = app_repo_root_from_module_path(module_path).ok_or_else(|| {
anyhow!(
"failed to resolve app repository root from SpacetimeDB module path {}",
module_path.display()
)
})?;
Ok(app_root.parent().map(Path::to_path_buf).unwrap_or(app_root))
}
fn app_repo_root_from_module_path(module_path: &Path) -> Option<PathBuf> {
for ancestor in module_path.ancestors() {
if ancestor.join("Cargo.toml").is_file() && ancestor.join("services.env").is_file() {
return Some(ancestor.to_path_buf());
}
}
None
}
fn container_workspace_path(mount_root: &Path, host_path: &Path) -> Result<String> {
let relative = host_path.strip_prefix(mount_root).with_context(|| {
format!(
"SpacetimeDB module path {} is not under docker mount root {}",
host_path.display(),
mount_root.display()
)
})?;
let mut path = String::from("/workspace");
for component in relative.components() {
if let Component::Normal(value) = component {
path.push('/');
path.push_str(&value.to_string_lossy().replace('\\', "/"));
}
}
Ok(path)
}
fn spacetime_docker_config_dir(endpoint_url: &str) -> Result<PathBuf> {
let endpoint_slug = spacetime_endpoint_slug(endpoint_url)?;
let home = std::env::var_os("USERPROFILE")
.or_else(|| std::env::var_os("HOME"))
.ok_or_else(|| anyhow!("failed to resolve home directory for SpacetimeDB docker config"))?;
Ok(PathBuf::from(home)
.join(".config")
.join("spacetime-docker")
.join(endpoint_slug))
}
fn spacetime_endpoint_slug(endpoint_url: &str) -> Result<String> {
let parsed = Url::parse(endpoint_url)
.with_context(|| format!("invalid SpacetimeDB endpoint URL: {endpoint_url}"))?;
let host = parsed.host_str().unwrap_or("local");
let port = parsed
.port_or_known_default()
.unwrap_or(if parsed.scheme() == "https" { 443 } else { 80 });
let safe_host = safe_path_segment(host);
Ok(format!("{safe_host}-{port}"))
}
fn write_spacetime_cli_token(endpoint_url: &str, token: &str) -> Result<()> {
if token.trim().is_empty()
|| token
.chars()
.any(|character| character.is_control() || matches!(character, '"' | '\\'))
{
return Err(anyhow!(
"local SpacetimeDB module-owner token cannot be encoded in the CLI config"
));
}
let config_dir = spacetime_docker_config_dir(endpoint_url)?;
fs::create_dir_all(&config_dir).with_context(|| {
format!(
"failed to create SpacetimeDB CLI config directory at {}",
config_dir.display()
)
})?;
let config_path = config_dir.join("cli.toml");
let existing = fs::read_to_string(&config_path).unwrap_or_else(|_| {
concat!(
"default_server = \"maincloud\"\n\n",
"[[server_configs]]\n",
"nickname = \"maincloud\"\n",
"host = \"maincloud.spacetimedb.com\"\n",
"protocol = \"https\"\n\n",
"[[server_configs]]\n",
"nickname = \"local\"\n",
"host = \"127.0.0.1:3000\"\n",
"protocol = \"http\"\n",
)
.to_string()
});
let token_line = format!("spacetimedb_token = \"{}\"", token.trim());
let mut replaced = false;
let mut lines = existing
.lines()
.map(|line| {
if line.trim_start().starts_with("spacetimedb_token") {
replaced = true;
token_line.clone()
} else {
line.to_string()
}
})
.collect::<Vec<_>>();
if !replaced {
let insert_at = lines
.iter()
.position(|line| line.trim_start().starts_with("default_server"))
.map(|index| index + 1)
.unwrap_or(0);
lines.insert(insert_at, token_line);
}
write_private_file(&config_path, format!("{}\n", lines.join("\n")).as_bytes())
}
fn write_private_file(path: &Path, contents: &[u8]) -> Result<()> {
let mut options = fs::OpenOptions::new();
options.create(true).truncate(true).write(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
options.mode(0o600);
}
let mut file = options
.open(path)
.with_context(|| format!("failed to open private file {}", path.display()))?;
file.write_all(contents)
.with_context(|| format!("failed to write private file {}", path.display()))?;
file.sync_all()
.with_context(|| format!("failed to flush private file {}", path.display()))?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
fs::set_permissions(path, fs::Permissions::from_mode(0o600))
.with_context(|| format!("failed to restrict private file {}", path.display()))?;
}
Ok(())
}
fn docker_spacetime_server_url(endpoint_url: &str) -> Result<String> {
if cfg!(target_os = "linux") {
return Ok(endpoint_url.to_string());
}
let parsed = Url::parse(endpoint_url)
.with_context(|| format!("invalid SpacetimeDB endpoint URL: {endpoint_url}"))?;
let Some(host) = parsed.host_str() else {
return Ok(endpoint_url.to_string());
};
if !is_loopback(host) {
return Ok(endpoint_url.to_string());
}
let port = parsed
.port()
.map(|port| format!(":{port}"))
.unwrap_or_default();
let replacement = format!("{}://host.docker.internal{port}", parsed.scheme());
Ok(replacement)
}
#[derive(Debug, Clone)]
struct DockerSpacetimePublishTarget {
server_url: String,
network_arg: Option<String>,
}
fn docker_spacetime_publish_target(endpoint_url: &str) -> Result<DockerSpacetimePublishTarget> {
if cfg!(target_os = "linux") {
return Ok(DockerSpacetimePublishTarget {
server_url: endpoint_url.to_string(),
network_arg: Some("host".to_string()),
});
}
let parsed = Url::parse(endpoint_url)
.with_context(|| format!("invalid SpacetimeDB endpoint URL: {endpoint_url}"))?;
let Some(host) = parsed.host_str() else {
return Ok(DockerSpacetimePublishTarget {
server_url: endpoint_url.to_string(),
network_arg: None,
});
};
if !is_loopback(host) {
return Ok(DockerSpacetimePublishTarget {
server_url: endpoint_url.to_string(),
network_arg: None,
});
}
let host_port = parsed
.port_or_known_default()
.ok_or_else(|| anyhow!("SpacetimeDB endpoint URL has no port: {endpoint_url}"))?;
if let Some(container_name) = docker_container_name_for_host_port(host_port)? {
return Ok(DockerSpacetimePublishTarget {
server_url: "http://127.0.0.1:3000".to_string(),
network_arg: Some(format!("container:{container_name}")),
});
}
Ok(DockerSpacetimePublishTarget {
server_url: docker_spacetime_server_url(endpoint_url)?,
network_arg: None,
})
}
fn docker_container_name_for_host_port(host_port: u16) -> Result<Option<String>> {
let output = Command::new("docker")
.arg("ps")
.arg("--format")
.arg("{{.Names}}\t{{.Ports}}")
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.output()
.with_context(|| "failed to inspect running docker containers for local SpacetimeDB")?;
if !output.status.success() {
return Err(anyhow!(
"docker ps exited with {} while inspecting local SpacetimeDB containers",
output.status
));
}
let needle = format!(":{host_port}->");
for line in String::from_utf8_lossy(&output.stdout).lines() {
let mut parts = line.splitn(2, '\t');
let Some(name) = parts
.next()
.map(str::trim)
.filter(|value| !value.is_empty())
else {
continue;
};
let Some(ports) = parts.next().map(str::trim) else {
continue;
};
if ports.contains(&needle) {
return Ok(Some(name.to_string()));
}
}
Ok(None)
}
fn bundled_compose_path() -> PathBuf {
Path::new(env!("CARGO_MANIFEST_DIR"))
.join("docker")
.join("docker-compose.yml")
}
fn compose_project_name(app_name: &str) -> String {
let lower = app_name.trim().to_ascii_lowercase();
let mut out = String::with_capacity(lower.len());
let mut last_was_dash = false;
for ch in lower.chars() {
if ch.is_ascii_alphanumeric() {
out.push(ch);
last_was_dash = false;
continue;
}
if !last_was_dash {
out.push('-');
last_was_dash = true;
}
}
let trimmed = out.trim_matches('-');
if trimmed.is_empty() {
"workspace".to_string()
} else {
trimmed.to_string()
}
}
fn ensure_docker_desktop_running() -> Result<()> {
match docker_info_probe() {
DockerInfoProbe::Ready => return Ok(()),
DockerInfoProbe::NotReady(detail) => {
if !detail.is_empty() {
debug!("docker info failed: {detail}");
}
}
DockerInfoProbe::Unavailable(error) => {
#[cfg(any(target_os = "windows", target_os = "macos"))]
{
if auto_start_docker_desktop()? {
return Ok(());
}
}
return Err(error);
}
}
#[cfg(any(target_os = "windows", target_os = "macos"))]
{
if auto_start_docker_desktop()? {
return Ok(());
}
}
Err(anyhow!(
"Docker Desktop does not appear to be running. Please start Docker Desktop and try again."
))
}
fn local_database_available(url: &str) -> bool {
let Some((host, port)) = host_and_port_from_connection_url(url) else {
return false;
};
if !is_loopback(host.as_str()) {
return false;
}
let mut candidates = Vec::new();
if let Ok(ip) = host.parse::<std::net::IpAddr>() {
candidates.push(std::net::SocketAddr::new(ip, port));
} else if let Ok(resolved) = (host.as_str(), port).to_socket_addrs() {
candidates.extend(resolved);
}
candidates
.into_iter()
.any(|addr| TcpStream::connect_timeout(&addr, LOCAL_DATABASE_PROBE_TIMEOUT).is_ok())
}
fn docker_info() -> Result<Output> {
let mut command = Command::new("docker");
command
.arg("info")
.stdout(Stdio::null())
.stderr(Stdio::piped());
run_command_with_timeout(&mut command, DOCKER_INFO_TIMEOUT, "`docker info`")
.context("failed to execute `docker info`; is Docker CLI installed?")
}
enum DockerInfoProbe {
Ready,
NotReady(String),
Unavailable(anyhow::Error),
}
fn docker_info_probe() -> DockerInfoProbe {
match docker_info() {
Ok(output) if output.status.success() => DockerInfoProbe::Ready,
Ok(output) => {
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
let detail = if stderr.is_empty() {
format!("docker info exited with status {}", output.status)
} else {
stderr
};
DockerInfoProbe::NotReady(detail)
}
Err(error) => {
if docker_info_timeout_like(&error) {
DockerInfoProbe::NotReady(error.to_string())
} else {
DockerInfoProbe::Unavailable(error)
}
}
}
}
fn docker_info_timeout_like(error: &anyhow::Error) -> bool {
error
.chain()
.any(|cause| cause.to_string().contains("`docker info` timed out after"))
}
fn run_command_with_timeout(
command: &mut Command,
timeout: Duration,
description: &str,
) -> Result<Output> {
let mut child = command
.spawn()
.with_context(|| format!("failed to spawn {description}"))?;
let started = Instant::now();
loop {
if let Some(status) = child
.try_wait()
.with_context(|| format!("failed to poll {description}"))?
{
let mut stdout = Vec::new();
let mut stderr = Vec::new();
if let Some(mut stdout_reader) = child.stdout.take() {
let _ = stdout_reader.read_to_end(&mut stdout);
}
if let Some(mut stderr_reader) = child.stderr.take() {
let _ = stderr_reader.read_to_end(&mut stderr);
}
return Ok(Output {
status,
stdout,
stderr,
});
}
if started.elapsed() >= timeout {
let _ = child.kill();
let _ = child.wait();
return Err(anyhow!(
"{description} timed out after {} seconds",
timeout.as_secs()
));
}
thread::sleep(COMMAND_POLL_INTERVAL);
}
}
#[cfg(any(target_os = "windows", target_os = "macos"))]
fn auto_start_docker_desktop() -> Result<bool> {
if matches!(docker_info_probe(), DockerInfoProbe::Ready) {
return Ok(true);
}
if std::env::var_os("RDS_SKIP_DOCKER_AUTOSTART").is_some() {
info!("docker desktop auto-start disabled via RDS_SKIP_DOCKER_AUTOSTART");
return Ok(false);
}
if should_suppress_autostart() {
debug!("skipping Docker Desktop auto-start; recent attempt detected");
return Ok(false);
}
if launch_docker_desktop()? {
info!("waiting for Docker Desktop to become ready after automatic launch");
let timeout = DOCKER_DESKTOP_READY_TIMEOUT;
let start = Instant::now();
while start.elapsed() < timeout {
thread::sleep(Duration::from_secs(2));
match docker_info_probe() {
DockerInfoProbe::Ready => {
info!("Docker Desktop became ready after automatic launch");
return Ok(true);
}
DockerInfoProbe::NotReady(detail) => {
if !detail.is_empty() {
debug!("docker info failed: {detail}");
}
debug!("Docker Desktop is still starting up; retrying docker info");
}
DockerInfoProbe::Unavailable(err) => {
debug!(
"failed to execute docker info while waiting for Docker Desktop: {err:?}"
);
return Ok(false);
}
}
}
warn!("timed out while waiting for Docker Desktop to start");
return Ok(false);
}
debug!("could not locate Docker Desktop installation; skipping auto-start attempt");
Ok(false)
}
#[cfg(any(target_os = "windows", target_os = "macos"))]
fn autostart_marker_path() -> Option<PathBuf> {
Some(std::env::temp_dir().join("local-data-services-docker-desktop-autostart"))
}
#[cfg(any(target_os = "windows", target_os = "macos"))]
fn should_suppress_autostart() -> bool {
let Some(path) = autostart_marker_path() else {
return false;
};
let Ok(metadata) = fs::metadata(&path) else {
return false;
};
let Ok(modified) = metadata.modified() else {
return false;
};
let Ok(elapsed) = modified.elapsed() else {
return false;
};
elapsed < Duration::from_secs(90)
}
#[cfg(any(target_os = "windows", target_os = "macos"))]
fn mark_autostart_attempt() {
let Some(path) = autostart_marker_path() else {
return;
};
if let Some(parent) = path.parent() {
let _ = fs::create_dir_all(parent);
}
let _ = fs::write(path, b"1");
}
#[cfg(target_os = "windows")]
fn docker_desktop_executable() -> Option<PathBuf> {
const EXECUTABLE_NAME: &str = "Docker Desktop.exe";
let mut candidates = Vec::new();
if let Ok(program_files) = std::env::var("ProgramFiles") {
candidates.push(
PathBuf::from(program_files)
.join("Docker")
.join("Docker")
.join(EXECUTABLE_NAME),
);
}
if let Ok(program_files_x86) = std::env::var("ProgramFiles(x86)") {
candidates.push(
PathBuf::from(program_files_x86)
.join("Docker")
.join("Docker")
.join(EXECUTABLE_NAME),
);
}
if let Ok(local_app_data) = std::env::var("LOCALAPPDATA") {
candidates.push(
PathBuf::from(local_app_data)
.join("Docker")
.join("Docker Desktop.exe"),
);
}
candidates.into_iter().find(|path| path.exists())
}
#[cfg(target_os = "windows")]
fn launch_docker_desktop() -> Result<bool> {
let Some(executable) = docker_desktop_executable() else {
return Ok(false);
};
mark_autostart_attempt();
info!(
"attempting to launch Docker Desktop automatically via {}",
executable.display()
);
Command::new(&executable)
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.with_context(|| {
format!(
"failed to launch Docker Desktop executable at {}",
executable.display()
)
})?;
Ok(true)
}
#[cfg(target_os = "macos")]
fn launch_docker_desktop() -> Result<bool> {
let app_bundle = PathBuf::from("/Applications/Docker.app");
if !app_bundle.exists() {
return Ok(false);
}
mark_autostart_attempt();
info!(
"attempting to launch Docker Desktop automatically via {}",
app_bundle.display()
);
Command::new("open")
.arg("-g")
.arg(app_bundle.as_os_str())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.with_context(|| {
format!(
"failed to launch Docker Desktop app bundle at {}",
app_bundle.display()
)
})?;
Ok(true)
}
pub(crate) fn is_local_database(url: &str) -> bool {
host_from_connection_url(url)
.map(|host| is_loopback(&host))
.unwrap_or(false)
}
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)
}
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()
}
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::*;
#[test]
fn detects_local_hosts() {
assert!(is_local_database("mysql://user:pass@localhost:4520/db"));
assert!(is_local_database("mysql://user:pass@127.0.0.1:4520/db"));
assert!(!is_local_database("mysql://user:pass@192.168.1.10:4520/db"));
}
#[test]
fn accepts_local_hosts_with_unescaped_credentials() {
let url = "mysql://app:R7#vM}tQw@1Ls4eZ@127.0.0.1:4520/app";
assert!(is_local_database(url));
}
#[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() {
assert!(bundled_compose_path().is_file());
}
#[test]
fn docker_info_timeout_is_treated_as_not_ready() {
let error = anyhow!(
"failed to execute `docker info`; is Docker CLI installed?: `docker info` timed out after 15 seconds"
);
assert!(docker_info_timeout_like(&error));
}
#[test]
fn windows_excluded_tcp_port_range_finds_spacetime_port() {
let output = r#"
Portausschlussbereiche fuer das Protokoll "tcp"
Startport Endport
---------- --------
4348 4447
4448 4547
50000 50059 *
"#;
assert_eq!(
windows_excluded_tcp_port_range(output, 4530),
Some((4448, 4547))
);
}
#[test]
fn windows_excluded_tcp_port_range_ignores_unmatched_ports() {
let output = r#"
Startport Endport
---------- --------
4448 4547
"#;
assert_eq!(windows_excluded_tcp_port_range(output, 4550), None);
}
#[test]
fn parses_spacetime_compose_port_from_env_file_contents() {
let contents = r#"
MARIADB_PORT=4520
SPACETIME_PORT="4530"
MAILPIT_UI_PORT=8025
"#;
assert_eq!(parse_spacetime_compose_port(contents), Some(4530));
}
#[test]
fn ignores_invalid_spacetime_compose_port_values() {
assert_eq!(parse_spacetime_compose_port("SPACETIME_PORT=0"), None);
assert_eq!(parse_spacetime_compose_port("SPACETIME_PORT=invalid"), None);
assert_eq!(parse_spacetime_compose_port("# SPACETIME_PORT=4530"), None);
}
#[test]
fn parses_spacetime_identity_http_response() {
let response = concat!(
"HTTP/1.1 200 OK\r\n",
"Content-Type: application/json\r\n",
"Content-Length: 35\r\n",
"\r\n",
"{\"identity\":\"owner\",\"token\":\"jwt\"}"
);
let (status, body) = parse_http_response(response.as_bytes()).unwrap();
assert_eq!(status, 200);
let credentials = serde_json::from_slice::<LocalSpacetimeOwnerCredentials>(&body).unwrap();
assert_eq!(credentials.identity, "owner");
assert_eq!(credentials.token, "jwt");
}
#[test]
fn parses_chunked_spacetime_identity_http_response() {
let response = concat!(
"HTTP/1.1 200 OK\r\n",
"Transfer-Encoding: chunked\r\n",
"\r\n",
"10\r\n{\"identity\":\"own\r\n",
"12\r\ner\",\"token\":\"jwt\"}\r\n",
"0\r\n\r\n"
);
let (status, body) = parse_http_response(response.as_bytes()).unwrap();
assert_eq!(status, 200);
assert_eq!(body, br#"{"identity":"owner","token":"jwt"}"#);
}
#[test]
fn spacetime_module_fingerprint_tracks_module_sources_only() {
let root = std::env::temp_dir().join(format!(
"spacetime-module-fingerprint-{}",
std::process::id()
));
let _ = fs::remove_dir_all(&root);
fs::create_dir_all(root.join("src").join("generated")).unwrap();
fs::create_dir_all(root.join("target")).unwrap();
fs::write(root.join("Cargo.toml"), "[package]\nname = \"demo\"\n").unwrap();
fs::write(root.join("src").join("lib.rs"), "mod reducers;\n").unwrap();
fs::write(
root.join("src").join("reducers.rs"),
"pub fn reducer() {}\n",
)
.unwrap();
fs::write(
root.join("src").join("generated").join("client.rs"),
"generated-v1\n",
)
.unwrap();
fs::write(root.join("target").join("scratch.txt"), "scratch-v1\n").unwrap();
let original = spacetime_module_fingerprint(&root).unwrap();
fs::write(
root.join("src").join("generated").join("client.rs"),
"generated-v2\n",
)
.unwrap();
fs::write(root.join("target").join("scratch.txt"), "scratch-v2\n").unwrap();
assert_eq!(spacetime_module_fingerprint(&root).unwrap(), original);
fs::write(
root.join("src").join("reducers.rs"),
"pub fn reducer2() {}\n",
)
.unwrap();
assert_ne!(spacetime_module_fingerprint(&root).unwrap(), original);
fs::remove_dir_all(&root).unwrap();
}
}