use std::collections::{BTreeMap, BTreeSet};
use base64::Engine;
use indexmap::IndexMap;
use super::*;
use crate::core::docker::{
DependencyCondition, DependsOn, KAFKA_HOST_BROKERS, KAFKA_INTERNAL_BROKERS, MINIO_INIT,
};
const PROVISION: &str = "postgresql-provision";
const ENV_HEADER: &str = "# Local development values generated by ForkLaunch for docker compose.\n# Git-ignored and never used for deployments. Existing values are kept.\n";
const WORKER_OWNER: &str = "x-forklaunch-worker-for";
pub(super) const INTERNAL_HMAC: &str = "INTERNAL_HMAC_SECRET_KEY";
const STRIPE_PLACEHOLDERS: [(&str, &str); 2] =
[("STRIPE_API_KEY", "sk_test_placeholder"), ("STRIPE_WEBHOOK_SECRET", "whsec_placeholder")];
struct DatabaseGroup {
prefix: String,
services: Vec<String>,
migrator: String,
}
#[derive(Default)]
struct Migration {
database: Option<String>,
kafka: bool,
dependents: BTreeSet<String>,
}
struct TypescriptDatabase {
service: String,
database: String,
role: String,
}
fn required_prefix(value: &str, suffix: &str) -> Option<String> {
let rest = value.strip_prefix("${")?;
let (name, _) = rest.split_once(":?")?;
let prefix = name.strip_suffix(suffix)?;
(!prefix.is_empty() && prefix.bytes().all(|b| b.is_ascii_uppercase() || b.is_ascii_digit() || b == b'_'))
.then(|| prefix.to_owned())
}
fn database_prefix(value: &str) -> Option<String> {
required_prefix(value, "_DATABASE_URL")
}
fn binary(service: &DockerService) -> Option<&str> {
match &service.command {
Some(DockerCommand::Multiple(command)) => command.first().map(String::as_str),
_ => None,
}
}
fn binary_name(service: &DockerService) -> Option<&str> {
binary(service).map(|path| path.rsplit('/').next().unwrap_or(path))
}
fn env<'a>(service: &'a DockerService, key: &str) -> Option<&'a String> {
service.environment.as_ref().and_then(|env| env.get(key))
}
fn worker_owner(service: &DockerService) -> Option<&str> {
service.additional_properties.get(WORKER_OWNER).and_then(|value| value.as_str())
}
fn groups(compose: &DockerCompose) -> Vec<DatabaseGroup> {
let mut grouped: BTreeMap<String, Vec<String>> = BTreeMap::new();
for (name, service) in &compose.services {
if name.ends_with("-migrate") || binary(service).is_none() {
continue;
}
if let Some(prefix) = env(service, "DATABASE_URL").and_then(|value| database_prefix(value)) {
grouped.entry(prefix).or_default().push(name.clone());
}
}
grouped
.into_iter()
.map(|(prefix, mut services)| {
services.sort();
let migrator = services
.iter()
.find(|name| binary_name(&compose.services[*name]).is_some_and(|b| b.ends_with("-worker")))
.unwrap_or(&services[0])
.clone();
DatabaseGroup { prefix, services, migrator }
})
.collect()
}
fn typescript_databases(compose: &DockerCompose) -> Vec<TypescriptDatabase> {
let candidates: Vec<(&String, &String)> = compose
.services
.iter()
.filter(|(name, service)| {
!name.ends_with("-migrate")
&& env(service, "DB_HOST").is_some_and(|host| host == "postgresql")
&& env(service, "DB_NAME").is_some()
})
.map(|(name, service)| (name, env(service, "DB_NAME").unwrap()))
.collect();
let owner = |database: &String| {
let sharing = candidates.iter().filter(|(_, db)| *db == database).map(|(name, _)| *name);
sharing
.clone()
.find(|name| compose.services.contains_key(&format!("{name}-migrate")))
.or_else(|| sharing.min_by_key(|name| name.len()))
.expect("a database has at least one service")
.to_case(Case::Snake)
};
candidates
.iter()
.map(|(name, database)| TypescriptDatabase {
service: (*name).clone(),
database: (*database).clone(),
role: owner(database),
})
.collect()
}
fn role(prefix: &str, suffix: &str) -> String {
format!("{}_{suffix}", prefix.to_lowercase())
}
fn database(app: &str, prefix: &str) -> String {
format!("{}_{}", app.replace('-', "_"), prefix.to_lowercase())
}
fn provision_script(user: &str, databases: &[(String, String, String)]) -> String {
let mut script = String::from("set -eu\n");
for (owner, runtime, db) in databases {
script.push_str(&format!(
"psql -v ON_ERROR_STOP=1 -h postgresql -U {user} -d postgres <<'SQL'\n\
SELECT format('CREATE ROLE %I LOGIN', name) FROM unnest(ARRAY['{owner}', '{runtime}']) AS name WHERE NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = name)\\gexec\n\
SELECT format('CREATE DATABASE %I OWNER %I', '{db}', '{owner}') WHERE NOT EXISTS (SELECT 1 FROM pg_database WHERE datname = '{db}')\\gexec\n\
SQL\n\
psql -v ON_ERROR_STOP=1 -h postgresql -U {user} -d '{db}' <<'SQL'\n\
ALTER SCHEMA public OWNER TO {owner};\n\
ALTER DEFAULT PRIVILEGES FOR ROLE {owner} GRANT USAGE ON SCHEMAS TO {runtime};\n\
ALTER DEFAULT PRIVILEGES FOR ROLE {owner} GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO {runtime};\n\
ALTER DEFAULT PRIVILEGES FOR ROLE {owner} GRANT USAGE, SELECT ON SEQUENCES TO {runtime};\n\
SELECT format('GRANT USAGE ON SCHEMA %I TO %I', nspname, '{runtime}') FROM pg_namespace WHERE nspowner = '{owner}'::regrole\\gexec\n\
SELECT format('GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA %I TO %I', nspname, '{runtime}') FROM pg_namespace WHERE nspowner = '{owner}'::regrole\\gexec\n\
SELECT format('GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA %I TO %I', nspname, '{runtime}') FROM pg_namespace WHERE nspowner = '{owner}'::regrole\\gexec\n\
SQL\n"
));
}
script
}
fn depends(service: &mut DockerService, name: &str, condition: DependencyCondition) {
service
.depends_on
.get_or_insert_with(IndexMap::new)
.entry(name.to_owned())
.or_insert(DependsOn { condition, ..Default::default() });
}
fn depends_ready(compose: &mut DockerCompose, service: &str, dependency: &str) {
let condition = if compose.services.get(dependency).is_some_and(|d| d.healthcheck.is_some()) {
DependencyCondition::ServiceHealthy
} else {
DependencyCondition::ServiceStarted
};
if let Some(service) = compose.services.get_mut(service) {
depends(service, dependency, condition);
}
}
fn random(bytes: usize) -> Vec<u8> {
let mut value = vec![0; bytes];
getrandom::fill(&mut value).expect("Failed to generate random bytes");
value
}
struct LocalContext<'a> {
app: &'a str,
tenant: String,
jwks: Option<String>,
redis_databases: BTreeSet<u32>,
}
impl LocalContext<'_> {
fn next_redis_database(&mut self) -> u32 {
let database = (0..).find(|index| !self.redis_databases.contains(index)).unwrap();
self.redis_databases.insert(database);
database
}
}
fn local_value(name: &str, context: &mut LocalContext<'_>) -> Option<String> {
let app = context.app;
if let Some(value) = super::auth_env::local_value(name, app) {
return Some(value);
}
let url = |prefix: &str, suffix: &str| {
format!("postgresql://{}@postgresql:5432/{}", role(prefix, suffix), database(app, prefix))
};
if let Some(prefix) = name.strip_suffix("_MIGRATION_DATABASE_URL") {
return Some(url(prefix, "owner"));
}
if let Some(prefix) = name.strip_suffix("_DATABASE_URL") {
return Some(url(prefix, "app"));
}
if name.ends_with("TENANT_ID") {
return Some(context.tenant.clone());
}
if let Some(prefix) = name.strip_suffix("_STORAGE_ENCRYPTION_NAMESPACE") {
return Some(format!("{}-records", prefix.to_lowercase()));
}
if name.ends_with("_JWKS_PUBLIC_KEY_URL") {
return context.jwks.clone();
}
if name.ends_with("_REDIS_URL") {
return Some(format!("redis://redis:6379/{}", context.next_redis_database()));
}
if name.ends_with("_KAFKA_BROKERS") {
return Some(KAFKA_INTERNAL_BROKERS.to_owned());
}
let kebab = |prefix: &str| format!("{app}-{}", prefix.to_case(Case::Kebab));
if let Some(prefix) = name.strip_suffix("_KAFKA_TRANSACTIONAL_ID") {
return Some(format!("{}-1", kebab(prefix)));
}
if let Some(prefix) = name.strip_suffix("_KAFKA_PUBLISHER_ID") {
return Some(format!("{}-publisher-1", kebab(prefix)));
}
if let Some(prefix) = name.strip_suffix("_NAMESPACE") {
return Some(kebab(prefix));
}
match name {
"ENCRYPTION_KEY_HEX" | INTERNAL_HMAC => Some(random(32).iter().map(|b| format!("{b:02x}")).collect()),
_ if name.ends_with("ENCRYPTION_KEY") || name.ends_with("HMAC_SECRET_KEY") => {
Some(base64::engine::general_purpose::STANDARD.encode(random(32)))
}
_ => None,
}
}
fn variable_pattern() -> regex::Regex {
regex::Regex::new(r"\$\{([A-Z0-9_]+)(?:(:?[-?])([^}]*))?\}").expect("fixed variable pattern")
}
fn required_variables(compose: &DockerCompose) -> Vec<String> {
let pattern = regex::Regex::new(r"\$\{([A-Z0-9_]+):\?").expect("fixed variable pattern");
let mut names: Vec<String> = compose
.services
.values()
.filter_map(|service| service.environment.as_ref())
.flat_map(|env| env.values())
.flat_map(|value| pattern.captures_iter(value).map(|c| c[1].to_owned()).collect::<Vec<_>>())
.collect();
names.sort();
names.dedup();
names
}
fn redis_database(url: &str) -> Option<u32> {
let rest = url.strip_prefix("redis://redis:6379")?;
match rest.strip_prefix('/') {
Some(index) => index.parse().ok(),
None if rest.is_empty() => Some(0),
None => None,
}
}
fn parse_env(content: &str) -> (IndexMap<String, String>, BTreeSet<String>) {
let mut values = IndexMap::new();
let mut commented = BTreeSet::new();
for line in content.lines() {
let line = line.trim();
if let Some(rest) = line.strip_prefix('#') {
if let Some((key, _)) = rest.trim().split_once('=') {
commented.insert(key.trim().to_owned());
}
continue;
}
if let Some((key, value)) = line.split_once('=') {
values.insert(key.trim().to_owned(), value.trim().to_owned());
}
}
(values, commented)
}
fn write_env(root: &Path, modules_path: &str, app: &str, compose: &DockerCompose) -> Result<IndexMap<String, String>> {
let path = root.join(".env");
let existing = read_to_string(&path).unwrap_or_default();
let (mut values, commented) = parse_env(&existing);
let tenant = required_variables(compose)
.iter()
.filter(|name| name.ends_with("TENANT_ID"))
.find_map(|name| values.get(name).filter(|value| !value.is_empty()).cloned())
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
let jwks = compose.services.get("iam").map(|iam| {
let port = env(iam, "PORT").map_or("8000", String::as_str);
format!("http://iam:{port}/api/auth/jwks")
});
let mut redis_databases: BTreeSet<u32> = compose
.services
.values()
.filter_map(|service| env(service, "REDIS_URL"))
.filter_map(|url| redis_database(url))
.collect();
redis_databases.extend(
values.iter().filter(|(key, _)| key.ends_with("_REDIS_URL")).filter_map(|(_, url)| redis_database(url)),
);
let mut context = LocalContext { app, tenant, jwks, redis_databases };
let modules = root.join(modules_path);
let mut additions = String::new();
let mut unresolved = Vec::new();
let add = |additions: &mut String, values: &mut IndexMap<String, String>, name: &str, value: String| {
additions.push_str(&format!("{name}={value}\n"));
values.insert(name.to_owned(), value);
};
for name in required_variables(compose) {
if values.contains_key(&name) {
continue;
}
if let Some((_, placeholder)) = STRIPE_PLACEHOLDERS.iter().find(|(key, _)| *key == name) {
additions.push_str("# Placeholder so the stack starts; Stripe calls fail until a Stripe test key replaces it.\n");
add(&mut additions, &mut values, &name, (*placeholder).to_owned());
continue;
}
let shared = if name == "ENCRYPTION_KEY" {
crate::core::env_defaults::find_existing_encryption_key(&modules)
} else if name.ends_with("HMAC_SECRET_KEY") && name != INTERNAL_HMAC {
crate::core::env_defaults::find_existing_hmac_secret(&modules)
} else {
None
};
match shared.or_else(|| local_value(&name, &mut context)) {
Some(value) => add(&mut additions, &mut values, &name, value),
None => unresolved.push(name),
}
}
let referenced = |flag: &str| {
compose.services.values().filter_map(|s| s.environment.as_ref()).flat_map(|e| e.values()).any(|v| v.contains(&format!("${{{flag}")))
};
for flag in ["DATABASE_ALLOW_PLAINTEXT_DEVELOPMENT", "WORKER_ALLOW_PLAINTEXT_DEVELOPMENT"] {
if referenced(flag) && !values.contains_key(flag) {
add(&mut additions, &mut values, flag, "true".into());
}
}
for (name, placeholder) in STRIPE_PLACEHOLDERS {
if referenced(name) && !values.contains_key(name) {
additions.push_str("# Placeholder so the stack starts; Stripe calls fail until a Stripe test key replaces it.\n");
add(&mut additions, &mut values, name, placeholder.to_owned());
}
}
for name in unresolved {
if !commented.contains(&name) {
additions.push_str(&format!("# {name} has no local default; set it before docker compose up.\n# {name}=\n"));
}
}
if !additions.is_empty() {
let mut content = if existing.is_empty() { ENV_HEADER.to_owned() } else { existing };
if !content.ends_with('\n') {
content.push('\n');
}
content.push_str(&additions);
write(path, content)?;
}
Ok(values)
}
fn ensure_infrastructure(compose: &mut DockerCompose, app: &str, groups: &[DatabaseGroup], typescript: &[TypescriptDatabase]) {
if !groups.is_empty() || !typescript.is_empty() {
crate::core::docker::ensure_postgresql_service(compose, app);
}
let kafka = compose
.services
.values()
.any(|service| env(service, "KAFKA_BROKERS").is_some_and(|value| required_prefix(value, "_KAFKA_BROKERS").is_some()));
if kafka && !compose.services.contains_key("kafka") {
compose.services.insert("kafka".into(), crate::core::docker::kafka_service(app));
}
let redis = compose.services.iter().any(|(name, service)| {
name != "redis"
&& env(service, "REDIS_URL")
.is_some_and(|url| url.contains("//redis:") || required_prefix(url, "_REDIS_URL").is_some())
});
if redis && !compose.services.contains_key("redis") {
let _ = crate::core::docker::add_redis_to_docker_compose(app, compose, &mut IndexMap::new(), 0);
}
let mut buckets: BTreeSet<String> = compose
.services
.iter()
.filter(|(name, _)| name.as_str() != MINIO_INIT)
.filter(|(_, service)| env(service, "S3_URL").is_some_and(|url| url.contains("//minio:")))
.filter_map(|(_, service)| env(service, "S3_BUCKET").cloned())
.filter(|bucket| !bucket.contains('$'))
.collect();
if ["tempo", "loki", "mimir"].iter().any(|name| compose.services.contains_key(*name)) {
buckets.insert("forklaunch-telemetry".into());
}
if !buckets.is_empty() && compose.services.contains_key("minio") {
let buckets: Vec<&str> = buckets.iter().map(String::as_str).collect();
crate::core::docker::ensure_minio_buckets(compose, app, &buckets);
}
let names: Vec<String> = compose.services.keys().cloned().collect();
for name in names {
let service = &compose.services[&name];
let redis = env(service, "REDIS_URL")
.is_some_and(|url| url.contains("//redis:") || required_prefix(url, "_REDIS_URL").is_some());
let bucket = env(service, "S3_BUCKET").is_some_and(|_| env(service, "S3_URL").is_some_and(|url| url.contains("//minio:")));
if redis && compose.services.contains_key("redis") && name != "redis" {
depends_ready(compose, &name, "redis");
}
if bucket && compose.services.contains_key(MINIO_INIT) {
depends(compose.services.get_mut(&name).unwrap(), MINIO_INIT, DependencyCondition::ServiceCompletedSuccessfully);
}
}
}
fn isolate_redis(compose: &mut DockerCompose) {
let owner = |name: &str, service: &DockerService| {
worker_owner(service)
.map(str::to_owned)
.unwrap_or_else(|| {
name.trim_end_matches("-migrate").trim_end_matches("-server").trim_end_matches("-worker").to_owned()
})
};
let mut taken: BTreeMap<u32, String> = BTreeMap::new();
let mut assigned: BTreeMap<String, u32> = BTreeMap::new();
let names: Vec<String> = compose.services.keys().cloned().collect();
for name in names {
let service = &compose.services[&name];
let Some(index) = env(service, "REDIS_URL").and_then(|url| redis_database(url)) else { continue };
let group = owner(&name, service);
let wanted = match assigned.get(&group) {
Some(index) => *index,
None if taken.get(&index).is_none_or(|holder| *holder == group) => index,
None => (0..).find(|candidate| !taken.contains_key(candidate)).unwrap(),
};
taken.insert(wanted, group.clone());
assigned.insert(group, wanted);
if wanted != index {
compose.services.get_mut(&name).unwrap().environment.as_mut().unwrap()
.insert("REDIS_URL".into(), format!("redis://redis:6379/{wanted}"));
}
}
}
fn restrict_typescript_roles(compose: &mut DockerCompose, typescript: &[TypescriptDatabase]) {
for database in typescript {
let set = |service: Option<&mut DockerService>, user: String| {
if let Some(env) = service.and_then(|service| service.environment.as_mut()) {
env.insert("DB_USER".into(), user);
}
};
set(compose.services.get_mut(&database.service), format!("{}_app", database.role));
set(compose.services.get_mut(&format!("{}-migrate", database.service)), format!("{}_owner", database.role));
let service = database.service.clone();
if compose.services.contains_key(PROVISION) {
for name in [service.clone(), format!("{service}-migrate")] {
if let Some(service) = compose.services.get_mut(&name) {
depends(service, PROVISION, DependencyCondition::ServiceCompletedSuccessfully);
}
}
}
}
}
fn migrations(compose: &DockerCompose, groups: &[DatabaseGroup]) -> BTreeMap<String, Migration> {
let mut migrations: BTreeMap<String, Migration> = BTreeMap::new();
for group in groups {
let migration = migrations.entry(group.migrator.clone()).or_default();
migration.database = Some(group.prefix.clone());
migration.dependents.extend(group.services.iter().cloned());
}
for (name, service) in &compose.services {
let kafka = env(service, "KAFKA_BROKERS").is_some_and(|value| required_prefix(value, "_KAFKA_BROKERS").is_some());
if !kafka || !binary_name(service).is_some_and(|binary| binary.ends_with("-worker")) || name.ends_with("-migrate") {
continue;
}
let migration = migrations.entry(name.clone()).or_default();
migration.kafka = true;
migration.dependents.insert(name.clone());
for (producer, candidate) in &compose.services {
if worker_owner(candidate) == Some(name) {
migration.dependents.insert(producer.clone());
}
}
}
migrations
}
fn share_rust_image(compose: &mut DockerCompose, app: &str) {
let image = format!("{app}-rust:latest");
let shared = |service: &DockerService| {
service.build.as_ref().is_some_and(|build| build.dockerfile.trim_start_matches("./") == "Dockerfile")
&& binary(service).is_some_and(|binary| binary.starts_with("./target/release/"))
};
let previous: BTreeSet<String> = compose
.services
.values()
.filter(|service| shared(service) || service.image.as_ref() == Some(&image))
.filter_map(|service| service.image.clone())
.collect();
let members: BTreeSet<String> = compose
.services
.iter()
.filter(|(_, service)| shared(service) || service.image.as_ref().is_some_and(|i| previous.contains(i)))
.map(|(name, _)| name.clone())
.collect();
let build = compose
.services
.iter()
.filter(|(name, _)| members.contains(*name))
.find_map(|(_, service)| service.build.clone().filter(|_| shared(service)));
let Some(build) = build else { return };
let builder = members.iter().next().cloned();
for (name, service) in compose.services.iter_mut().filter(|(name, _)| members.contains(*name)) {
service.image = Some(image.clone());
if Some(name) == builder.as_ref() {
service.build = Some(build.clone());
service.additional_properties.shift_remove("pull_policy");
} else {
service.build = None;
service.additional_properties.insert("pull_policy".into(), "never".into());
}
}
}
fn resolve(value: &str, values: &IndexMap<String, String>) -> Option<String> {
let pattern = variable_pattern();
let mut missing = false;
let resolved = pattern.replace_all(value, |captures: ®ex::Captures<'_>| {
let name = &captures[1];
let operator = captures.get(2).map(|m| m.as_str());
match (values.get(name).filter(|v| !v.is_empty() || operator.is_none_or(|o| !o.starts_with(':'))), operator) {
(Some(value), _) => value.clone(),
(None, Some(op)) if op.ends_with('-') => captures.get(3).map_or("", |m| m.as_str()).to_owned(),
_ => {
missing = true;
String::new()
}
}
});
(!missing).then(|| resolved.into_owned())
}
fn host_addresses(compose: &DockerCompose) -> Vec<(String, String)> {
let mut rewrites = Vec::new();
for (name, service) in &compose.services {
for port in service.ports.iter().flatten() {
let parts: Vec<&str> = port.split('/').next().unwrap_or(port).split(':').collect();
let (published, target) = match parts.as_slice() {
[_, published, target] | [published, target] => (*published, *target),
_ => continue,
};
for host in std::iter::once(name.as_str()).chain(service.hostname.as_deref()) {
rewrites.push((format!("//{host}:{target}"), format!("//127.0.0.1:{published}")));
rewrites.push((format!("@{host}:{target}"), format!("@127.0.0.1:{published}")));
}
}
}
if compose.services.contains_key("kafka") {
rewrites.push((KAFKA_INTERNAL_BROKERS.to_owned(), KAFKA_HOST_BROKERS.to_owned()));
}
rewrites
}
fn write_local_envs(root: &Path, compose: &DockerCompose, values: &IndexMap<String, String>, migrations: &BTreeMap<String, Migration>) -> Result<()> {
let manifest = read_manifest_value(&root.join(".forklaunch/manifest.toml"))?;
let rewrites = host_addresses(compose);
let host = |value: String| rewrites.iter().fold(value, |value, (from, to)| value.replace(from, to));
for project in manifest.get("projects").and_then(Value::as_array).into_iter().flatten() {
let Some(name) = project.get("name").and_then(Value::as_str) else { continue };
let rust = project.get("metadata").and_then(|m| m.get("language")).and_then(Value::as_str) == Some("rust");
if !rust || !matches!(project.get("type").and_then(Value::as_str), Some("Service" | "Worker")) {
continue;
}
let Some(relative) = crate_path_for_project(&manifest, name) else { continue };
let directory = root.join(relative);
if !directory.is_dir() {
continue;
}
let mut processes: Vec<&DockerService> = compose.services.get(name).into_iter().collect();
processes.extend(compose.services.values().filter(|service| worker_owner(service) == Some(name)));
if processes.is_empty() {
continue;
}
let mut settings: IndexMap<String, String> = IndexMap::new();
for service in &processes {
for (key, value) in service.environment.iter().flatten() {
if settings.contains_key(key) || matches!(key.as_str(), "HOST" | "PORT") {
continue;
}
if let Some(value) = resolve(value, values) {
settings.insert(key.clone(), host(value));
}
}
}
if let Some(prefix) = migrations.iter().find(|(_, m)| m.dependents.contains(name)).and_then(|(_, m)| m.database.clone()) {
if let Some(owner) = values.get(&format!("{prefix}_MIGRATION_DATABASE_URL")) {
settings.entry("MIGRATION_DATABASE_URL".into()).or_insert_with(|| host(owner.clone()));
}
}
if settings.contains_key("DATABASE_URL") {
settings.insert("SQLX_OFFLINE".into(), "true".into());
}
let path = directory.join(".env.local");
let existing = read_to_string(&path).unwrap_or_default();
let (present, _) = parse_env(&existing);
let mut content = existing.clone();
if !content.is_empty() && !content.ends_with('\n') {
content.push('\n');
}
for (key, value) in settings {
if !present.contains_key(&key) {
content.push_str(&format!("{key}={value}\n"));
}
}
if content != existing {
write(&path, content)?;
}
}
Ok(())
}
fn sdk_dependencies(root: &Path) -> Result<BTreeMap<String, BTreeSet<String>>> {
let manifest = read_manifest_value(&root.join(".forklaunch/manifest.toml"))?;
let mut packages: BTreeMap<String, (String, BTreeSet<String>)> = BTreeMap::new();
for project in manifest.get("projects").and_then(Value::as_array).into_iter().flatten() {
let Some(name) = project.get("name").and_then(Value::as_str) else { continue };
let Some(relative) = crate_path_for_project(&manifest, name) else { continue };
let Ok(source) = read_to_string(root.join(relative).join("Cargo.toml")) else { continue };
let Ok(cargo) = toml::from_str::<Value>(&source) else { continue };
let package = cargo.get("package").and_then(|p| p.get("name")).and_then(Value::as_str).unwrap_or(name).to_owned();
let mut dependencies = BTreeSet::new();
for (key, dependency) in cargo.get("dependencies").and_then(Value::as_table).into_iter().flatten() {
dependencies.insert(dependency.get("package").and_then(Value::as_str).unwrap_or(key).to_owned());
}
packages.insert(package, (name.to_owned(), dependencies));
}
let by_package: BTreeMap<String, String> = packages.iter().map(|(package, (name, _))| (package.clone(), name.clone())).collect();
let mut result = BTreeMap::new();
for (package, (name, _)) in &packages {
let mut seen = BTreeSet::new();
let mut pending = vec![package.clone()];
let mut targets = BTreeSet::new();
while let Some(current) = pending.pop() {
if !seen.insert(current.clone()) {
continue;
}
let Some((_, dependencies)) = packages.get(¤t) else { continue };
for dependency in dependencies {
if let Some(target) = dependency.strip_suffix("-sdk").filter(|t| by_package.contains_key(*t) || packages.values().any(|(n, _)| n == t)) {
if target != name {
targets.insert(target.to_owned());
}
} else if packages.contains_key(dependency) && dependency != package {
pending.push(dependency.clone());
}
}
}
if !targets.is_empty() {
result.insert(name.clone(), targets);
}
}
Ok(result)
}
fn wire_service_dependencies(root: &Path, compose: &mut DockerCompose) -> Result<()> {
let internal = format!("${{{INTERNAL_HMAC}:?Configure the internal service HMAC key}}");
let producers: Vec<String> = compose.services.iter().filter(|(_, s)| worker_owner(s).is_some()).map(|(n, _)| n.clone()).collect();
for producer in &producers {
let env = compose.services.get_mut(producer).unwrap().environment.get_or_insert_with(IndexMap::new);
env.entry(INTERNAL_HMAC.into()).or_insert_with(|| internal.clone());
}
for (consumer, targets) in sdk_dependencies(root)? {
let mut settings = IndexMap::new();
for target in targets {
let worker = compose.services.iter().find(|(_, s)| worker_owner(s) == Some(target.as_str())).map(|(n, s)| (n.clone(), s.clone()));
let endpoint = worker.or_else(|| compose.services.get(&target).map(|s| (target.clone(), s.clone())));
let Some((service_name, service)) = endpoint else { continue };
let Some(port) = env(&service, "PORT") else { continue };
let hostname = service.hostname.clone().unwrap_or(service_name);
let prefix = target.to_case(Case::UpperSnake);
settings.insert(format!("{prefix}_URL"), format!("http://{hostname}:{port}"));
if let Some(key) = env(&service, "HMAC_SECRET_KEY").filter(|v| v.starts_with("${")) {
settings.insert(format!("{prefix}_HMAC_SECRET_KEY"), key.clone());
}
if worker_owner(&service).is_some() {
settings.insert(INTERNAL_HMAC.into(), internal.clone());
}
}
let names: Vec<String> = compose
.services
.iter()
.filter(|(name, service)| *name == &consumer || worker_owner(service) == Some(consumer.as_str()))
.map(|(name, _)| name.clone())
.collect();
for name in names {
let env = compose.services.get_mut(&name).unwrap().environment.get_or_insert_with(IndexMap::new);
for (key, value) in &settings {
env.entry(key.clone()).or_insert_with(|| value.clone());
}
}
}
Ok(())
}
fn wire_encrypted_helpers(root: &Path, compose: &mut DockerCompose) -> Result<()> {
let manifest = read_manifest_value(&root.join(".forklaunch/manifest.toml"))?;
let names: Vec<String> = compose.services.keys().cloned().collect();
for name in names {
let Some(relative) = crate_path_for_project(&manifest, &name) else { continue };
let helpers = ["infrastructure/redis.rs", "infrastructure/s3.rs"].iter().any(|helper| {
read_to_string(root.join(&relative).join(helper)).is_ok_and(|source| source.contains("ENCRYPTION_KEY"))
});
if helpers {
let env = compose.services.get_mut(&name).unwrap().environment.get_or_insert_with(IndexMap::new);
env.entry("ENCRYPTION_KEY".into()).or_insert_with(|| crate::infra::native_local::ENCRYPTION_KEY_REFERENCE.into());
}
}
Ok(())
}
pub(super) fn ensure_cargo_config(root: &Path) -> Result<()> {
if !root.join("Cargo.toml").is_file() {
return Ok(());
}
let path = root.join(".cargo/config.toml");
let existing = read_to_string(&path).unwrap_or_default();
let mut config: toml_edit::DocumentMut = existing.parse()?;
let env = config.entry("env").or_insert_with(toml_edit::table).as_table_mut().context("Cargo [env] must be a table")?;
if env.contains_key("SQLX_OFFLINE") {
return Ok(());
}
env.insert("SQLX_OFFLINE", toml_edit::value("true"));
create_dir_all(root.join(".cargo"))?;
write(path, config.to_string())?;
Ok(())
}
pub(super) fn framework_contexts(root: &Path) -> Option<(PathBuf, Vec<(&'static str, PathBuf)>)> {
let framework = crate::core::forklaunch_state::framework(root)?;
let path = PathBuf::from(framework.get("path")?.as_str()?);
let checkout = path.parent()?.to_path_buf();
let mut contexts = vec![("framework", path)];
let blueprint = checkout.join("blueprint-rs");
if blueprint.join("Cargo.toml").is_file() {
contexts.push(("blueprint", blueprint));
}
Some((checkout, contexts))
}
pub(super) fn wire_framework_contexts(root: &Path, compose: &mut DockerCompose, dockerfile: &str) {
let contexts = framework_contexts(root);
for service in compose.services.values_mut() {
let Some(build) = service.build.as_mut() else { continue };
if build.dockerfile.trim_start_matches("./") != dockerfile {
continue;
}
match &contexts {
Some((checkout, contexts)) => {
let mut map = serde_yml::Mapping::new();
for (name, path) in contexts {
map.insert(serde_yml::Value::String((*name).into()), serde_yml::Value::String(path.to_string_lossy().into_owned()));
}
build.additional_properties.insert("additional_contexts".into(), serde_yml::Value::Mapping(map));
let mut args = serde_yml::Mapping::new();
args.insert(serde_yml::Value::String("FRAMEWORK_ROOT".into()), serde_yml::Value::String(checkout.to_string_lossy().into_owned()));
build.additional_properties.insert("args".into(), serde_yml::Value::Mapping(args));
}
None => {
build.additional_properties.shift_remove("additional_contexts");
build.additional_properties.shift_remove("args");
}
}
}
}
pub(super) fn prune_stale_migrations(root: &Path) -> Result<()> {
let manifest = read_manifest_value(&root.join(".forklaunch/manifest.toml"))?;
let compose_path = root.join(
manifest.get("docker_compose_path").and_then(Value::as_str).unwrap_or("docker-compose.yaml"),
);
if !compose_path.is_file() {
return Ok(());
}
let mut compose = read_or_create_rust_docker_compose(&compose_path)?;
let stale: Vec<String> = compose
.services
.iter()
.filter(|(name, service)| {
let generated = matches!(&service.command, Some(DockerCommand::Multiple(command)) if command.last().map(String::as_str) == Some("--migrate"));
generated && name.strip_suffix("-migrate").is_some_and(|base| !compose.services.contains_key(base))
})
.map(|(name, _)| name.clone())
.collect();
if stale.is_empty() {
return Ok(());
}
for name in &stale {
compose.services.shift_remove(name);
}
for service in compose.services.values_mut() {
if let Some(depends) = service.depends_on.as_mut() {
depends.retain(|name, _| !stale.contains(name));
}
}
write(&compose_path, yaml_to_string(&compose)?)?;
Ok(())
}
pub(super) fn env_gaps(root: &Path) -> Result<Vec<(String, String)>> {
let manifest = read_manifest_value(&root.join(".forklaunch/manifest.toml"))?;
let compose_path = root.join(manifest.get("docker_compose_path").and_then(Value::as_str).unwrap_or("docker-compose.yaml"));
if !compose_path.is_file() {
return Ok(vec![]);
}
let compose = read_or_create_rust_docker_compose(&compose_path)?;
let (values, _) = parse_env(&read_to_string(root.join(".env")).unwrap_or_default());
let pattern = regex::Regex::new(r"\$\{([A-Z0-9_]+):\?").expect("fixed variable pattern");
let mut gaps = Vec::new();
for (name, service) in &compose.services {
if !binary(service).is_some_and(|binary| binary.starts_with("/app/bin/")) {
continue;
}
for value in service.environment.iter().flatten().map(|(_, value)| value) {
for captures in pattern.captures_iter(value) {
let variable = captures[1].to_owned();
if values.get(&variable).is_none_or(|value| value.is_empty()) && !gaps.contains(&(name.clone(), variable.clone())) {
gaps.push((name.clone(), variable));
}
}
}
}
Ok(gaps)
}
pub(super) fn release_unused_kafka(root: &Path) -> Result<()> {
let path = root.join("docker-compose.yaml");
if !path.is_file() {
return Ok(());
}
let mut compose = read_or_create_rust_docker_compose(&path)?;
let Some(kafka) = compose.services.get("kafka") else { return Ok(()) };
let generated = crate::core::docker::kafka_service(&rust_app_name(root)?);
if kafka.image != generated.image || kafka.hostname != generated.hostname {
return Ok(());
}
let used = compose.services.iter().any(|(name, service)| {
name != "kafka"
&& (service.environment.iter().flatten().any(|(key, _)| key.ends_with("KAFKA_BROKERS"))
|| service.depends_on.as_ref().is_some_and(|depends| depends.contains_key("kafka")))
});
if used {
return Ok(());
}
compose.services.shift_remove("kafka");
write(&path, yaml_to_string(&compose)?)?;
Ok(())
}
pub(super) fn prune_project_env(root: &Path, name: &str, local_keys: &[&str]) -> Result<()> {
let manifest = read_manifest_value(&root.join(".forklaunch/manifest.toml"))?;
let compose_path = root.join(manifest.get("docker_compose_path").and_then(Value::as_str).unwrap_or("docker-compose.yaml"));
let compose_text = read_to_string(&compose_path).unwrap_or_default();
let prefix = format!("{}_", name.to_case(Case::UpperSnake));
let referenced = |key: &str| ["}", ":", "-"].iter().any(|end| compose_text.contains(&format!("${{{key}{end}")));
let path = root.join(".env");
if let Ok(existing) = read_to_string(&path) {
let updated: String = existing
.split_inclusive('\n')
.filter(|line| {
let line = line.trim_start();
let key = line.trim_start_matches('#').trim_start().split('=').next().unwrap_or_default().trim();
!(key.starts_with(&prefix) && line.contains('=') && !referenced(key))
})
.collect();
if updated != existing {
write(&path, updated)?;
}
}
if local_keys.is_empty() {
return Ok(());
}
let Some(relative) = crate_path_for_project(&manifest, name) else { return Ok(()) };
let local = root.join(relative).join(".env.local");
let Ok(existing) = read_to_string(&local) else { return Ok(()) };
let compose = read_or_create_rust_docker_compose(&compose_path)?;
let mut set: BTreeSet<String> = compose
.services
.iter()
.filter(|(key, service)| *key == name || worker_owner(service) == Some(name))
.flat_map(|(_, service)| service.environment.iter().flatten().map(|(key, _)| key.clone()))
.collect();
if set.contains("DATABASE_URL") {
set.extend(["MIGRATION_DATABASE_URL".into(), "SQLX_OFFLINE".into()]);
}
let updated: String = existing
.split_inclusive('\n')
.filter(|line| {
let key = line.split('=').next().unwrap_or_default().trim();
!(local_keys.contains(&key) && !set.contains(key))
})
.collect();
if updated != existing {
write(&local, updated)?;
}
Ok(())
}
fn owned_by(key: &str, service: &DockerService, name: &str) -> bool {
key == name
|| [format!("{name}-migrate"), format!("{name}-seed"), format!("{name}-server"), format!("{name}-worker")].iter().any(|owned| owned == key)
|| ["x-forklaunch-worker-for", "x-forklaunch-ecommerce-worker-for"]
.iter()
.any(|tag| service.additional_properties.get(*tag).and_then(serde_yml::Value::as_str) == Some(name))
}
pub(super) fn renamed_segment(text: &str, old: &str, new: &str) -> String {
if old.is_empty() {
return text.to_owned();
}
let word = |c: char| c.is_ascii_alphanumeric() || c == '_';
let mut out = String::with_capacity(text.len());
let mut rest = 0;
let mut search = 0;
while let Some(found) = text[search..].find(old) {
let start = search + found;
let end = start + old.len();
let before = text[..start].chars().next_back();
let after = text[end..].chars().next();
if before.is_none_or(|c| !word(c)) && after.is_none_or(|c| !word(c)) {
out.push_str(&text[rest..start]);
out.push_str(new);
rest = end;
search = end;
} else {
search = start + text[start..].chars().next().map_or(1, char::len_utf8);
}
}
out.push_str(&text[rest..]);
out
}
fn renamed_project_value(value: &str, app: &str, old: &str, new: &str) -> String {
let (old_upper, new_upper) = (old.to_case(Case::UpperSnake), new.to_case(Case::UpperSnake));
let (old_snake, new_snake) = (old.to_case(Case::Snake), new.to_case(Case::Snake));
let app_snake = app.replace('-', "_");
let value = value.replace(&format!("${{{old_upper}_"), &format!("${{{new_upper}_"));
let value = value
.replace(&format!("/{app_snake}_{old_snake}"), &format!("/{app_snake}_{new_snake}"))
.replace(&format!("//{old_snake}_app@"), &format!("//{new_snake}_app@"))
.replace(&format!("//{old_snake}_owner@"), &format!("//{new_snake}_owner@"));
renamed_segment(&value, old, new)
}
fn renamed_host(value: &str, old: &str, new: &str) -> String {
value.replace(&format!("://{old}:"), &format!("://{new}:")).replace(&format!("://{old}-server:"), &format!("://{new}-server:"))
}
pub(super) fn rename_compose_project(root: &Path, old: &str, new: &str) -> Result<()> {
if old == new {
return Ok(());
}
let manifest = read_manifest_value(&root.join(".forklaunch/manifest.toml"))?;
let app = rust_app_name(root)?;
let compose_path = root.join(manifest.get("docker_compose_path").and_then(Value::as_str).unwrap_or("docker-compose.yaml"));
let (old_upper, new_upper) = (format!("{}_", old.to_case(Case::UpperSnake)), format!("{}_", new.to_case(Case::UpperSnake)));
let renamed_key = |key: &str| key.strip_prefix(&old_upper).map_or_else(|| key.to_owned(), |rest| format!("{new_upper}{rest}"));
if compose_path.is_file() {
let mut compose = read_or_create_rust_docker_compose(&compose_path)?;
let renames: BTreeMap<String, String> = compose
.services
.iter()
.filter(|(key, service)| owned_by(key, service, old))
.map(|(key, _)| (key.clone(), key.strip_prefix(old).map_or_else(|| key.clone(), |rest| format!("{new}{rest}"))))
.collect();
compose.services = std::mem::take(&mut compose.services)
.into_iter()
.map(|(key, mut service)| {
if let Some(target) = renames.get(&key) {
service.hostname = service.hostname.map(|hostname| if hostname == key { target.clone() } else { renamed_segment(&hostname, old, new) });
service.container_name = service.container_name.map(|name| renamed_segment(&name, old, new));
if let Some(command) = service.command.as_mut() {
match command {
DockerCommand::Simple(command) => *command = renamed_segment(command, old, new),
DockerCommand::Multiple(parts) => parts.iter_mut().for_each(|part| *part = renamed_segment(part, old, new)),
}
}
for tag in ["x-forklaunch-worker-for", "x-forklaunch-ecommerce-worker-for"] {
if service.additional_properties.get(tag).and_then(serde_yml::Value::as_str) == Some(old) {
service.additional_properties.insert(tag.into(), serde_yml::Value::String(new.into()));
}
}
if let Some(env) = service.environment.as_mut() {
env.values_mut().for_each(|value| *value = renamed_project_value(value, &app, old, new));
}
} else if let Some(env) = service.environment.take() {
service.environment = Some(
env.into_iter()
.map(|(key, value)| {
if key.starts_with(&old_upper) {
(renamed_key(&key), renamed_project_value(&value, &app, old, new))
} else {
(key, renamed_host(&value.replace(&format!("${{{old_upper}"), &format!("${{{new_upper}")), old, new))
}
})
.collect(),
);
}
if let Some(depends) = service.depends_on.take() {
service.depends_on = Some(depends.into_iter().map(|(key, condition)| (renames.get(&key).cloned().unwrap_or(key), condition)).collect());
}
(renames.get(&key).cloned().unwrap_or(key), service)
})
.collect();
write(&compose_path, yaml_to_string(&compose)?)?;
}
let mut env_files = vec![(root.join(".env"), false)];
for project in manifest.get("projects").and_then(Value::as_array).into_iter().flatten() {
let name = project.get("name").and_then(Value::as_str);
if let Some(relative) = name.and_then(|name| crate_path_for_project(&manifest, name)) {
env_files.push((root.join(relative).join(".env.local"), name == Some(new)));
}
}
for (path, own) in env_files {
let Ok(existing) = read_to_string(&path) else { continue };
let updated: String = existing
.split_inclusive('\n')
.map(|line| match line.split_once('=') {
Some((key, value)) if !key.trim_start().starts_with('#') && (own || key.trim().starts_with(&old_upper)) => {
format!("{}={}", renamed_key(key.trim()), renamed_project_value(value, &app, old, new))
}
Some((key, value)) if !key.trim_start().starts_with('#') => format!("{key}={}", renamed_host(value, old, new)),
_ => line.to_owned(),
})
.collect();
if updated != existing {
write(&path, updated)?;
}
}
Ok(())
}
fn renamed_application_text(text: &str, old: &str, new: &str) -> String {
let (old_kebab, new_kebab) = (old.to_case(Case::Kebab), new.to_case(Case::Kebab));
let (old_snake, new_snake) = (old.to_case(Case::Snake), new.to_case(Case::Snake));
let prefix = regex::Regex::new(&format!(r"(^|[^A-Za-z0-9_-]){}-", regex::escape(&old_kebab))).expect("escaped name");
let renamed = prefix.replace_all(text, |captures: ®ex::Captures| format!("{}{new_kebab}-", &captures[1]));
renamed.replace(&format!("/{old_snake}_"), &format!("/{new_snake}_"))
}
pub(super) fn rename_application(root: &Path, old: &str, new: &str) -> Result<()> {
if old == new {
return Ok(());
}
let manifest = read_manifest_value(&root.join(".forklaunch/manifest.toml"))?;
let compose_path = root.join(
manifest.get("docker_compose_path").and_then(Value::as_str).unwrap_or("docker-compose.yaml"),
);
let mut files = vec![compose_path, root.join(".env")];
let (old_scope, new_scope) = (format!("\"@{}/", old.to_case(Case::Kebab)), format!("\"@{}/", new.to_case(Case::Kebab)));
for project in manifest.get("projects").and_then(Value::as_array).into_iter().flatten() {
let rust = project.get("metadata").and_then(|m| m.get("language")).and_then(Value::as_str) == Some("rust");
if let Some(relative) = project.get("name").and_then(Value::as_str).and_then(|name| crate_path_for_project(&manifest, name)) {
files.push(root.join(&relative).join(".env.local"));
if !rust {
continue;
}
for entry in walkdir::WalkDir::new(root.join(&relative)).into_iter().filter_entry(|entry| !matches!(entry.file_name().to_str(), Some("node_modules" | "target"))).flatten() {
if entry.file_name() == "package.json"
&& let Ok(existing) = read_to_string(entry.path())
&& existing.contains(&old_scope)
{
write(entry.path(), existing.replace(&old_scope, &new_scope))?;
}
}
}
}
for path in files {
let Ok(existing) = read_to_string(&path) else { continue };
let updated = if path.extension().is_some_and(|extension| extension == "yaml" || extension == "yml") {
renamed_application_text(&existing, old, new)
} else {
existing
.split_inclusive('\n')
.map(|line| match line.split_once('=') {
Some((key, value)) if !key.trim_start().starts_with('#') => format!("{key}={}", renamed_application_text(value, old, new)),
_ => line.to_owned(),
})
.collect()
};
if updated != existing {
write(&path, updated)?;
}
}
Ok(())
}
pub(super) fn reconcile(root: &Path) -> Result<()> {
let manifest = read_manifest_value(&root.join(".forklaunch/manifest.toml"))?;
let compose_path = root.join(
manifest.get("docker_compose_path").and_then(Value::as_str).unwrap_or("docker-compose.yaml"),
);
if !compose_path.is_file() {
return Ok(());
}
let mut compose = read_or_create_rust_docker_compose(&compose_path)?;
let app = rust_app_name(root)?;
let mixed = manifest.get("language").and_then(Value::as_str) == Some("mixed");
if !mixed {
share_rust_image(&mut compose, &app);
}
wire_encrypted_helpers(root, &mut compose)?;
wire_service_dependencies(root, &mut compose)?;
let groups = groups(&compose);
let typescript = if mixed { typescript_databases(&compose) } else { vec![] };
ensure_infrastructure(&mut compose, &app, &groups, &typescript);
isolate_redis(&mut compose);
let migrations = migrations(&compose, &groups);
if let Some(postgres) = compose.services.get("postgresql").filter(|_| !groups.is_empty() || !typescript.is_empty()).cloned() {
let user = postgres
.environment
.as_ref()
.and_then(|env| env.get("POSTGRES_USER"))
.cloned()
.unwrap_or_else(|| "postgres".into());
let mut databases: Vec<(String, String, String)> = groups
.iter()
.map(|group| (role(&group.prefix, "owner"), role(&group.prefix, "app"), database(&app, &group.prefix)))
.collect();
for ts in &typescript {
let entry = (format!("{}_owner", ts.role), format!("{}_app", ts.role), ts.database.clone());
if !databases.contains(&entry) {
databases.push(entry);
}
}
let mut provision = DockerService {
image: postgres.image.clone(),
restart: Some(Restart::No),
networks: postgres.networks.clone(),
entrypoint: Some(DockerCommand::Multiple(vec!["sh".into(), "-c".into()])),
command: Some(DockerCommand::Multiple(vec![provision_script(&user, &databases)])),
..Default::default()
};
if postgres.healthcheck.is_some() {
depends(&mut provision, "postgresql", DependencyCondition::ServiceHealthy);
} else {
depends(&mut provision, "postgresql", DependencyCondition::ServiceStarted);
}
compose.services.insert(PROVISION.into(), provision);
} else if groups.is_empty() && typescript.is_empty() {
compose.services.shift_remove(PROVISION);
for service in compose.services.values_mut() {
if let Some(depends) = service.depends_on.as_mut() {
depends.shift_remove(PROVISION);
if depends.is_empty() {
service.depends_on = None;
}
}
}
}
restrict_typescript_roles(&mut compose, &typescript);
for (migrator, migration) in &migrations {
let source = compose.services[migrator].clone();
let mut environment = source.environment.clone().unwrap_or_default();
if let Some(prefix) = &migration.database {
let owner = format!("${{{prefix}_MIGRATION_DATABASE_URL:?Configure the migration owner role}}");
environment.insert("DATABASE_URL".into(), owner.clone());
environment.insert("MIGRATION_DATABASE_URL".into(), owner);
}
let mut command = match &source.command {
Some(DockerCommand::Multiple(command)) => vec![command[0].clone()],
_ => unreachable!("migrations require exec-form commands"),
};
command.push("--migrate".into());
let mut migrate = DockerService {
image: source.image.clone(),
build: source.build.clone(),
restart: Some(Restart::No),
environment: Some(environment),
networks: source.networks.clone(),
working_dir: source.working_dir.clone(),
command: Some(DockerCommand::Multiple(command)),
..Default::default()
};
if migration.database.is_some() && compose.services.contains_key(PROVISION) {
depends(&mut migrate, PROVISION, DependencyCondition::ServiceCompletedSuccessfully);
}
let migrate_name = format!("{migrator}-migrate");
compose.services.insert(migrate_name.clone(), migrate);
if migration.kafka && compose.services.contains_key("kafka") {
depends_ready(&mut compose, &migrate_name, "kafka");
}
for name in &migration.dependents {
if let Some(service) = compose.services.get_mut(name) {
depends(service, &migrate_name, DependencyCondition::ServiceCompletedSuccessfully);
}
}
}
let seeds: Vec<(String, String)> = compose
.services
.keys()
.filter_map(|name| Some((name.clone(), name.strip_suffix("-seed")?.to_owned())))
.filter(|(_, service)| compose.services.contains_key(service))
.collect();
for (seed, service) in seeds {
let migrate = format!("{service}-migrate");
if compose.services.contains_key(&migrate) {
depends(compose.services.get_mut(&seed).unwrap(), &migrate, DependencyCondition::ServiceCompletedSuccessfully);
}
depends(compose.services.get_mut(&service).unwrap(), &seed, DependencyCondition::ServiceCompletedSuccessfully);
}
if mixed {
wire_framework_contexts(root, &mut compose, "Dockerfile.rust");
wire_framework_contexts(root, &mut compose, "Dockerfile.typescript");
} else {
share_rust_image(&mut compose, &app);
wire_framework_contexts(root, &mut compose, "Dockerfile");
}
write(&compose_path, yaml_to_string(&compose)?)?;
let modules_path = manifest.get("modules_path").and_then(Value::as_str).unwrap_or("src");
let values = write_env(root, modules_path, &app, &compose)?;
let superuser = compose
.services
.get("postgresql")
.and_then(|postgres| env(postgres, "POSTGRES_USER"))
.cloned()
.unwrap_or_else(|| "postgres".into());
typescript_local_owners(&root.join(modules_path), &typescript, &superuser)?;
if !typescript.is_empty() && compose.services.contains_key(PROVISION) {
provision_host_scripts(&root.join(modules_path).join("package.json"))?;
}
write_local_envs(root, &compose, &values, &migrations)
}
fn provision_host_scripts(path: &Path) -> Result<()> {
const START: &str = "\"docker compose up -d --wait postgresql && ";
let Ok(existing) = read_to_string(path) else { return Ok(()) };
let provisioned = format!("{START}docker compose run --rm {PROVISION} && ");
let content = existing.replace(&provisioned, START).replace(START, &provisioned);
if content != existing {
write(path, content)?;
}
Ok(())
}
fn typescript_local_owners(modules: &Path, typescript: &[TypescriptDatabase], superuser: &str) -> Result<()> {
for database in typescript.iter().filter(|database| database.service.to_case(Case::Snake) == database.role) {
let path = modules.join(&database.service).join(".env.local");
let Ok(existing) = read_to_string(&path) else { continue };
let owner = format!("{}_owner", database.role);
let content: Vec<String> = existing
.lines()
.map(|line| match line.split_once('=') {
Some((key, value)) if key.trim() == "DB_USER" && value.trim() == superuser => format!("DB_USER={owner}"),
_ => line.to_owned(),
})
.collect();
let mut content = content.join("\n");
if existing.ends_with('\n') {
content.push('\n');
}
if content != existing {
write(&path, content)?;
}
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn renamed_segment_replaces_whole_segments_once() {
assert_eq!(renamed_segment("./target/release/jobs", "jobs", "email-jobs"), "./target/release/email-jobs");
assert_eq!(renamed_segment("app-jobs-rust", "jobs", "email-jobs"), "app-email-jobs-rust");
assert_eq!(renamed_segment("jobs-jobs", "jobs", "x"), "x-x");
assert_eq!(renamed_segment("myjobs jobs_db jobs", "jobs", "x"), "myjobs jobs_db x");
}
#[test]
fn project_rename_keeps_compose_settings_and_rewires_references() {
let root = tempfile::tempdir().unwrap();
let root = root.path();
create_dir_all(root.join(".forklaunch")).unwrap();
create_dir_all(root.join("src/ledgers")).unwrap();
write(root.join(".forklaunch/manifest.toml"), "app_name = \"kestrel\"\n[[projects]]\nname = \"ledgers\"\ntype = \"Service\"\n[projects.metadata]\nlanguage = \"rust\"\ncratePath = \"src/ledgers\"\n").unwrap();
write(root.join("docker-compose.yaml"), r#"services:
accounts:
hostname: accounts
container_name: kestrel-accounts-rust
environment:
PORT: '8000'
DATABASE_URL: ${ACCOUNTS_DATABASE_URL:?Configure}
depends_on:
accounts-migrate:
condition: service_completed_successfully
ports:
- 8000:8000
command:
- /app/bin/accounts
accounts-migrate:
command:
- /app/bin/accounts
- --migrate
accountsx:
environment:
ACCOUNTS_URL: http://accounts:8000
OTHER: keep-accounts-word
depends_on:
accounts:
condition: service_started
"#).unwrap();
write(root.join(".env"), "ACCOUNTS_DATABASE_URL=postgresql://accounts_app@postgresql:5432/kestrel_accounts\nACCOUNTS_STORAGE_ENCRYPTION_NAMESPACE=accounts-records\nACCOUNTSX_KEEP=1\nIAM_NOTE=accounts\nOTHER_URL=http://accounts:8000/x\n").unwrap();
write(root.join("src/ledgers/.env.local"), "DATABASE_URL=postgresql://accounts_app@127.0.0.1:5432/kestrel_accounts\nPORT=8000\n").unwrap();
rename_compose_project(root, "accounts", "ledgers").unwrap();
let compose = read_or_create_rust_docker_compose(&root.join("docker-compose.yaml")).unwrap();
let ledgers = &compose.services["ledgers"];
assert_eq!(ledgers.hostname.as_deref(), Some("ledgers"));
assert_eq!(ledgers.container_name.as_deref(), Some("kestrel-ledgers-rust"));
assert_eq!(ledgers.environment.as_ref().unwrap()["DATABASE_URL"], "${LEDGERS_DATABASE_URL:?Configure}");
assert_eq!(ledgers.ports.as_ref().unwrap(), &vec!["8000:8000".to_string()]);
assert!(ledgers.depends_on.as_ref().unwrap().contains_key("ledgers-migrate"));
assert!(matches!(&compose.services["ledgers-migrate"].command, Some(DockerCommand::Multiple(c)) if c[0] == "/app/bin/ledgers"));
let consumer = &compose.services["accountsx"];
assert_eq!(consumer.environment.as_ref().unwrap()["LEDGERS_URL"], "http://ledgers:8000");
assert_eq!(consumer.environment.as_ref().unwrap()["OTHER"], "keep-accounts-word");
assert!(consumer.depends_on.as_ref().unwrap().contains_key("ledgers"));
assert!(!compose.services.contains_key("accounts"));
assert_eq!(read_to_string(root.join(".env")).unwrap(), "LEDGERS_DATABASE_URL=postgresql://ledgers_app@postgresql:5432/kestrel_ledgers\nLEDGERS_STORAGE_ENCRYPTION_NAMESPACE=ledgers-records\nACCOUNTSX_KEEP=1\nIAM_NOTE=accounts\nOTHER_URL=http://ledgers:8000/x\n");
assert_eq!(read_to_string(root.join("src/ledgers/.env.local")).unwrap(), "DATABASE_URL=postgresql://ledgers_app@127.0.0.1:5432/kestrel_ledgers\nPORT=8000\n");
}
#[test]
fn unreferenced_project_settings_leave_env_files() {
let root = tempfile::tempdir().unwrap();
let root = root.path();
create_dir_all(root.join(".forklaunch")).unwrap();
create_dir_all(root.join("src/jobs")).unwrap();
write(root.join(".forklaunch/manifest.toml"), "app_name = \"kestrel\"\n[[projects]]\nname = \"jobs\"\ntype = \"Worker\"\n[projects.metadata]\nlanguage = \"rust\"\ncratePath = \"src/jobs\"\n").unwrap();
write(root.join("docker-compose.yaml"), "services:\n jobs:\n environment:\n REDIS_URL: ${JOBS_REDIS_URL:?Configure Redis}\n TENANT_ID: ${JOBS_TENANT_ID:?Configure}\n").unwrap();
write(root.join(".env"), "JOBS_DATABASE_URL=postgresql://jobs_app@postgresql:5432/kestrel_jobs\nJOBS_REDIS_URL=redis://redis:6379/1\nJOBS_TENANT_ID=t\nOTHER_DATABASE_URL=x\n").unwrap();
write(root.join("src/jobs/.env.local"), "DATABASE_URL=postgresql://x\nMIGRATION_DATABASE_URL=postgresql://y\nSQLX_OFFLINE=true\nREDIS_URL=redis://127.0.0.1:6379/1\nMINE=1\n").unwrap();
prune_project_env(root, "jobs", &["DATABASE_URL", "MIGRATION_DATABASE_URL", "SQLX_OFFLINE", "REDIS_URL"]).unwrap();
assert_eq!(read_to_string(root.join(".env")).unwrap(), "JOBS_REDIS_URL=redis://redis:6379/1\nJOBS_TENANT_ID=t\nOTHER_DATABASE_URL=x\n");
assert_eq!(read_to_string(root.join("src/jobs/.env.local")).unwrap(), "REDIS_URL=redis://127.0.0.1:6379/1\nMINE=1\n");
}
#[test]
fn generated_kafka_leaves_with_its_last_reader() {
let root = tempfile::tempdir().unwrap();
let root = root.path();
create_dir_all(root.join(".forklaunch")).unwrap();
write(root.join(".forklaunch/manifest.toml"), "app_name = \"kestrel\"\n").unwrap();
let mut compose = DockerCompose::default();
compose.services.insert("kafka".into(), crate::core::docker::kafka_service("kestrel"));
let mut reader = DockerService::default();
reader.environment = Some(IndexMap::from([("KAFKA_BROKERS".to_string(), "kafka:29092".to_string())]));
compose.services.insert("events".into(), reader);
write(root.join("docker-compose.yaml"), yaml_to_string(&compose).unwrap()).unwrap();
release_unused_kafka(root).unwrap();
assert!(read_or_create_rust_docker_compose(&root.join("docker-compose.yaml")).unwrap().services.contains_key("kafka"));
compose.services.shift_remove("events");
write(root.join("docker-compose.yaml"), yaml_to_string(&compose).unwrap()).unwrap();
release_unused_kafka(root).unwrap();
assert!(!read_or_create_rust_docker_compose(&root.join("docker-compose.yaml")).unwrap().services.contains_key("kafka"));
}
#[test]
fn application_rename_rewrites_generated_names_only() {
let text = "QUEUE_NAME=zebra-jobs-dev\nDATABASE_URL=postgresql://a@postgresql:5432/zebra_accounts\nPATH_HINT=/srv/old-zebra-x\nOTHER=zebras-keep\n";
assert_eq!(
renamed_application_text(text, "zebra", "kestrel"),
"QUEUE_NAME=kestrel-jobs-dev\nDATABASE_URL=postgresql://a@postgresql:5432/kestrel_accounts\nPATH_HINT=/srv/old-zebra-x\nOTHER=zebras-keep\n"
);
let root = tempfile::tempdir().unwrap();
let root = root.path();
create_dir_all(root.join(".forklaunch")).unwrap();
create_dir_all(root.join("src/jobs")).unwrap();
write(root.join(".forklaunch/manifest.toml"), "app_name = \"kestrel\"\n[[projects]]\nname = \"jobs\"\ntype = \"Worker\"\n[projects.metadata]\nlanguage = \"rust\"\ncratePath = \"src/jobs\"\n").unwrap();
write(root.join("docker-compose.yaml"), "networks:\n zebra-network:\n name: zebra-network\n").unwrap();
write(root.join(".env"), "# zebra-keep comment\nJOBS_NAMESPACE=zebra-jobs\n").unwrap();
write(root.join("src/jobs/.env.local"), "WORKER_NAMESPACE=zebra-jobs\nCUSTOM=mine\n").unwrap();
create_dir_all(root.join("src/jobs/sdk/typescript")).unwrap();
write(root.join("src/jobs/sdk/typescript/package.json"), "{\"name\": \"@zebra/jobs\"}").unwrap();
rename_application(root, "zebra", "kestrel").unwrap();
assert_eq!(read_to_string(root.join("src/jobs/sdk/typescript/package.json")).unwrap(), "{\"name\": \"@kestrel/jobs\"}");
assert_eq!(read_to_string(root.join("docker-compose.yaml")).unwrap(), "networks:\n kestrel-network:\n name: kestrel-network\n");
assert_eq!(read_to_string(root.join(".env")).unwrap(), "# zebra-keep comment\nJOBS_NAMESPACE=kestrel-jobs\n");
assert_eq!(read_to_string(root.join("src/jobs/.env.local")).unwrap(), "WORKER_NAMESPACE=kestrel-jobs\nCUSTOM=mine\n");
}
#[test]
fn database_prefix_reads_required_service_urls_only() {
assert_eq!(database_prefix("${NOTES_DATABASE_URL:?Configure}").as_deref(), Some("NOTES"));
assert_eq!(database_prefix("${NOTES_DATABASE_URL:-x}"), None);
assert_eq!(database_prefix("postgresql://localhost/app"), None);
}
fn context() -> LocalContext<'static> {
LocalContext {
app: "notes-app",
tenant: "t".into(),
jwks: Some("http://iam:8002/api/auth/jwks".into()),
redis_databases: BTreeSet::from([0]),
}
}
#[test]
fn local_values_cover_databases_tenants_and_keys() {
let mut context = context();
assert_eq!(
local_value("NOTES_DATABASE_URL", &mut context).unwrap(),
"postgresql://notes_app@postgresql:5432/notes_app_notes"
);
assert_eq!(
local_value("NOTES_MIGRATION_DATABASE_URL", &mut context).unwrap(),
"postgresql://notes_owner@postgresql:5432/notes_app_notes"
);
assert_eq!(local_value("WORKER_TENANT_ID", &mut context).unwrap(), "t");
assert_eq!(local_value("ENCRYPTION_KEY_HEX", &mut context).unwrap().len(), 64);
assert_eq!(local_value("BILLING_JWKS_PUBLIC_KEY_URL", &mut context).as_deref(), Some("http://iam:8002/api/auth/jwks"));
context.jwks = None;
assert_eq!(local_value("BILLING_JWKS_PUBLIC_KEY_URL", &mut context), None);
assert_eq!(local_value("STRIPE_API_KEY", &mut context), None);
}
#[test]
fn local_values_cover_worker_brokers_and_identities() {
let mut context = context();
assert_eq!(local_value("NOTIFY_REDIS_URL", &mut context).unwrap(), "redis://redis:6379/1");
assert_eq!(local_value("JOBS_REDIS_URL", &mut context).unwrap(), "redis://redis:6379/2");
assert_eq!(local_value("AUDIT_KAFKA_BROKERS", &mut context).unwrap(), "kafka:29092");
assert_eq!(local_value("AUDIT_NAMESPACE", &mut context).unwrap(), "notes-app-audit");
assert_eq!(local_value("AUDIT_KAFKA_TRANSACTIONAL_ID", &mut context).unwrap(), "notes-app-audit-1");
assert_eq!(local_value("AUDIT_KAFKA_PUBLISHER_ID", &mut context).unwrap(), "notes-app-audit-publisher-1");
assert_eq!(local_value("INTERNAL_HMAC_SECRET_KEY", &mut context).unwrap().len(), 64);
}
#[test]
fn resolve_uses_env_values_defaults_and_reports_missing() {
let values = IndexMap::from([("A".to_string(), "1".to_string()), ("E".to_string(), String::new())]);
assert_eq!(resolve("${A:?x}", &values).as_deref(), Some("1"));
assert_eq!(resolve("${B:-primary}", &values).as_deref(), Some("primary"));
assert_eq!(resolve("${E:-}", &values).as_deref(), Some(""));
assert_eq!(resolve("${B:?Configure}", &values), None);
assert_eq!(resolve("plain", &values).as_deref(), Some("plain"));
}
#[test]
fn redis_databases_are_isolated_per_service_pair() {
let mut compose: DockerCompose = serde_yml::from_str(
"services:\n api:\n environment: {REDIS_URL: 'redis://redis:6379/0'}\n web-migrate:\n environment: {REDIS_URL: 'redis://redis:6379/0'}\n web:\n environment: {REDIS_URL: 'redis://redis:6379/0'}\n jobs-worker:\n environment: {REDIS_URL: 'redis://redis:6379/0'}\n jobs-server:\n environment: {REDIS_URL: 'redis://redis:6379/0'}\n",
)
.unwrap();
isolate_redis(&mut compose);
let url = |name: &str| compose.services[name].environment.as_ref().unwrap()["REDIS_URL"].clone();
assert_eq!(url("api"), "redis://redis:6379/0");
assert_eq!(url("web"), "redis://redis:6379/1");
assert_eq!(url("web-migrate"), "redis://redis:6379/1");
assert_eq!(url("jobs-worker"), "redis://redis:6379/2");
assert_eq!(url("jobs-server"), "redis://redis:6379/2");
}
#[test]
fn host_addresses_use_published_loopback_ports() {
let compose: DockerCompose = serde_yml::from_str(
"services:\n postgresql:\n ports: ['127.0.0.1:5432:5432']\n iam:\n ports: ['8004:8004']\n coding-server:\n hostname: coding-server\n ports: ['127.0.0.1:8001:3000']\n kafka:\n image: kafka\n",
)
.unwrap();
let rewrites = host_addresses(&compose);
let host = |value: &str| rewrites.iter().fold(value.to_owned(), |value, (from, to)| value.replace(from, to));
assert_eq!(host("postgresql://a_app@postgresql:5432/db"), "postgresql://a_app@127.0.0.1:5432/db");
assert_eq!(host("http://iam:8004/api/auth/jwks"), "http://127.0.0.1:8004/api/auth/jwks");
assert_eq!(host("http://coding-server:3000"), "http://127.0.0.1:8001");
assert_eq!(host("kafka:29092"), "127.0.0.1:9092");
}
#[test]
fn typescript_database_scripts_provision_roles_first() {
let root = tempfile::tempdir().unwrap();
let path = root.path().join("package.json");
std::fs::write(&path, r#"{"scripts":{"migrate:init":"docker compose up -d --wait postgresql && pnpm -r run migrate:init","dev":"pnpm dev"}}"#).unwrap();
provision_host_scripts(&path).unwrap();
let expected = r#"{"scripts":{"migrate:init":"docker compose up -d --wait postgresql && docker compose run --rm postgresql-provision && pnpm -r run migrate:init","dev":"pnpm dev"}}"#;
assert_eq!(std::fs::read_to_string(&path).unwrap(), expected);
provision_host_scripts(&path).unwrap();
assert_eq!(std::fs::read_to_string(&path).unwrap(), expected);
}
#[test]
fn typescript_host_tooling_uses_the_owner_role() {
let root = tempfile::tempdir().unwrap();
for (project, user) in [("ecommerce", "postgresql"), ("intake", "custom")] {
std::fs::create_dir_all(root.path().join(project)).unwrap();
std::fs::write(root.path().join(project).join(".env.local"), format!("DB_HOST=localhost\nDB_USER={user}\n")).unwrap();
}
let database = |service: &str, role: &str| TypescriptDatabase { service: service.into(), database: "db".into(), role: role.into() };
let typescript = [database("ecommerce", "ecommerce"), database("ecommerce-worker", "ecommerce"), database("intake", "intake")];
typescript_local_owners(root.path(), &typescript, "postgresql").unwrap();
let read = |project: &str| std::fs::read_to_string(root.path().join(project).join(".env.local")).unwrap();
assert_eq!(read("ecommerce"), "DB_HOST=localhost\nDB_USER=ecommerce_owner\n");
assert_eq!(read("intake"), "DB_HOST=localhost\nDB_USER=custom\n");
}
#[test]
fn typescript_processes_of_one_database_share_its_migrated_roles() {
let compose: DockerCompose = yaml_from_str(r#"
services:
ecommerce:
environment: { DB_HOST: postgresql, DB_NAME: app-ecommerce-dev }
ecommerce-worker:
environment: { DB_HOST: postgresql, DB_NAME: app-ecommerce-dev }
ecommerce-migrate:
environment: { DB_HOST: postgresql, DB_NAME: app-ecommerce-dev }
intake:
environment: { DB_HOST: postgresql, DB_NAME: app-intake-dev }
"#).unwrap();
let roles: Vec<(String, String)> =
typescript_databases(&compose).into_iter().map(|db| (db.service, db.role)).collect();
assert_eq!(roles, [
("ecommerce".into(), "ecommerce".into()),
("ecommerce-worker".into(), "ecommerce".into()),
("intake".into(), "intake".into()),
]);
}
#[test]
fn rust_only_application_gets_infrastructure_migrations_and_complete_env() {
let root = tempfile::tempdir().unwrap();
let root = root.path();
create_dir_all(root.join(".forklaunch")).unwrap();
write(root.join(".forklaunch/manifest.toml"), "app_name='demo'\nlanguage='rust'\n[[projects]]\nname='notes'\ntype='Service'\n[projects.metadata]\nlanguage='rust'\ncratePath='src/notes'\n[[projects]]\nname='audit'\ntype='Worker'\n[projects.metadata]\nlanguage='rust'\ncratePath='src/audit'\n").unwrap();
for name in ["notes", "audit"] {
create_dir_all(root.join("src").join(name)).unwrap();
write(root.join("src").join(name).join("Cargo.toml"), format!("[package]\nname='{name}'\n")).unwrap();
}
write(root.join("Cargo.toml"), "[workspace]\nmembers=['src/*']\n").unwrap();
let service = |command: &str, env: &str| format!(" build: {{context: ., dockerfile: ./Dockerfile}}\n image: demo-x-rust:latest\n command: ['./target/release/{command}']\n environment: {{{env}}}\n");
let compose = format!(
"services:\n notes:\n{} ports: ['8000:8000']\n minio:\n image: minio\n audit:\n{} audit-server:\n{} x-forklaunch-worker-for: audit\n ports: ['127.0.0.1:8001:3000']\n",
service("notes", "PORT: '8000', DATABASE_URL: '${NOTES_DATABASE_URL:?x}', S3_URL: 'http://minio:9000', S3_BUCKET: demo-notes-dev"),
service("audit-worker", "KAFKA_BROKERS: '${AUDIT_KAFKA_BROKERS:?x}', WORKER_NAMESPACE: '${AUDIT_NAMESPACE:?x}', KAFKA_PUBLISHER_ID: '${AUDIT_KAFKA_PUBLISHER_ID:?x}'"),
service("audit", "PORT: '3000', KAFKA_BROKERS: '${AUDIT_KAFKA_BROKERS:?x}', WORKER_NAMESPACE: '${AUDIT_NAMESPACE:?x}', STRIPE_API_KEY: '${STRIPE_API_KEY:-}'"),
);
write(root.join("docker-compose.yaml"), compose).unwrap();
reconcile(root).unwrap();
let compose = read_or_create_rust_docker_compose(&root.join("docker-compose.yaml")).unwrap();
for name in ["postgresql", PROVISION, "kafka", MINIO_INIT, "notes-migrate", "audit-migrate"] {
assert!(compose.services.contains_key(name), "{name}");
}
assert!(compose.services["postgresql"].ports.as_ref().unwrap()[0].starts_with("127.0.0.1:"));
assert!(minio_bucket_created(&compose, "demo-notes-dev"));
assert_eq!(binary(&compose.services["audit-migrate"]), Some("./target/release/audit-worker"));
for (name, migration) in [("notes", "notes-migrate"), ("audit", "audit-migrate"), ("audit-server", "audit-migrate")] {
assert!(compose.services[name].depends_on.as_ref().unwrap().contains_key(migration), "{name}");
}
assert!(compose.services.values().filter(|s| s.build.is_some()).all(|s| s.image.as_deref() == Some("demo-rust:latest")));
assert!(env(&compose.services["audit-server"], INTERNAL_HMAC).is_some());
let (values, commented) = parse_env(&read_to_string(root.join(".env")).unwrap());
for name in required_variables(&compose) {
assert!(values.contains_key(&name), "{name} unset");
}
assert!(commented.is_empty() || !commented.iter().any(|name| !values.contains_key(name)));
assert_eq!(values["STRIPE_API_KEY"], "sk_test_placeholder");
let (local, _) = parse_env(&read_to_string(root.join("src/notes/.env.local")).unwrap());
assert_eq!(local["DATABASE_URL"], "postgresql://notes_app@127.0.0.1:5432/demo_notes");
assert_eq!(local["MIGRATION_DATABASE_URL"], "postgresql://notes_owner@127.0.0.1:5432/demo_notes");
let (worker, _) = parse_env(&read_to_string(root.join("src/audit/.env.local")).unwrap());
assert_eq!(worker["KAFKA_BROKERS"], "127.0.0.1:9092");
let before = read_to_string(root.join(".env")).unwrap();
reconcile(root).unwrap();
assert_eq!(read_to_string(root.join(".env")).unwrap(), before);
}
fn minio_bucket_created(compose: &DockerCompose, bucket: &str) -> bool {
crate::core::docker::minio_buckets(compose).iter().any(|name| name == bucket)
}
#[test]
fn provisioning_is_idempotent_sql() {
let databases = [("notes_owner".to_string(), "notes_app".to_string(), "notes_app_notes".to_string())];
let script = provision_script("postgresql", &databases);
assert!(script.contains("WHERE NOT EXISTS (SELECT 1 FROM pg_roles"));
assert!(script.contains("CREATE DATABASE %I OWNER %I', 'notes_app_notes', 'notes_owner'"));
assert!(script.contains("GRANT USAGE ON SCHEMAS TO notes_app;"));
assert!(script.contains("nspowner = 'notes_owner'::regrole"));
assert!(!script.contains('$'), "compose would interpolate dollar signs");
}
}