forklaunch 1.35.1

//! Local Docker runtime for Rust and mixed applications: shared infrastructure,
//! per-project PostgreSQL roles, one-shot migrations, and git-ignored
//! development values in the root `.env` and each crate's `.env.local`.
//!
//! Rust services require a restricted runtime role, a separate migration owner and
//! explicit secrets. Compose keeps those required (`${VAR:?...}`); this module
//! provisions matching local infrastructure and roles and fills only the missing
//! `.env` values, so `docker compose up` works on a fresh scaffold without
//! weakening deployments. Generated infrastructure binds to loopback only.
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"),
];

/// A group of compose services sharing one `<PREFIX>_DATABASE_URL`.
struct DatabaseGroup {
    prefix: String,
    services: Vec<String>,
    migrator: String,
}

/// One `<service>-migrate` job: the owner database migration and/or Kafka topics.
#[derive(Default)]
struct Migration {
    database: Option<String>,
    kafka: bool,
    dependents: BTreeSet<String>,
}

/// A TypeScript service on the shared PostgreSQL (mixed applications).
struct TypescriptDatabase {
    service: String,
    database: String,
    role: String,
}

/// `${<PREFIX><suffix>:?...}` → `<PREFIX>`.
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();
            // Workers own their queue tables, so their binary runs the migration.
            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()
}

/// TypeScript services on the shared PostgreSQL. Processes of one project
/// (`ecommerce`, `ecommerce-worker`) share its database, so they share the
/// roles of the process its `<name>-migrate` job migrates for; separate roles
/// would each take over the schema and lock the other out.
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())
}

/// Idempotent provisioning for (owner, runtime, database) triples: roles, a
/// database owned by the migration role, and default grants so tables the owner
/// creates are usable by the runtime role.
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()
        });
}

/// Healthy when the dependency has a healthcheck, started otherwise.
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
}

/// Settings `.env` must hold beyond the per-name defaults: tenant, IAM JWKS and
/// the Redis databases already taken.
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
    }
}

/// A local value for a required compose variable, or None when only a person can supply it.
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();
    }
    // Each Redis-backed worker gets its own logical database.
    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 {
        // The internal service key is 64 hex characters (the producer policy requires at least 32 bytes).
        "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,
    }
}

/// `.env` as written: `KEY=value` lines (commented `# KEY=` placeholders separately).
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);
    // One local tenant for every service, so records line up across projects.
    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;
        }
        // Share the app's existing key ring and service signing secret.
        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),
        }
    }
    // Local Docker PostgreSQL has no TLS; deployments keep verified TLS.
    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());
        }
    }
    // Optional Stripe settings (billing) get the same placeholder, so the
    // whole stack starts without a Stripe account.
    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)
}

/// PostgreSQL, Kafka and MinIO buckets for the projects that need them.
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));
    }
    // Redis workers and caches point at local Redis even when no project asked
    // for the `redis` infrastructure, so the service must exist for them too.
    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();
    // The telemetry stack (Tempo, Loki, Mimir) stores in its own bucket.
    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];
        // A literal local URL, or a required per-project URL `.env` points at local Redis.
        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,
            );
        }
    }
}

/// Each service (and its worker/producer pair) gets its own Redis database, so
/// caches and queues never share keys by accident.
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}"));
        }
    }
}

/// Mixed applications: TypeScript services use restricted per-service roles
/// like Rust services, never the PostgreSQL superuser.
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());
    }
    // Kafka workers create their topics with `--migrate`.
    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
}

/// One built image for every generated Rust process. Only one service carries
/// the `build`; the rest run the same tag with `pull_policy: never`. When every
/// service declared the build, Compose exported the multi-gigabyte image once
/// per service at the same time, which filled the disk and stopped the Docker
/// daemon. `docker compose up` still builds before it starts anything; start a
/// single service with `docker compose up --build <builder> <service>`.
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 };
    // The first member by name builds; it is deterministic across regenerations.
    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());
        }
    }
}

/// `${VAR}` references resolved against `.env`; None when a required value is unset.
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: &regex::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())
}

/// Container addresses rewritten to their published loopback ports.
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
}

/// Each Rust crate's `.env.local` mirrors its Compose processes with host
/// addresses, so `cargo run` behaves like the container. Existing keys are kept.
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()));
            }
        }
        // A host build sees DATABASE_URL; keep sqlx macros offline.
        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(())
}

/// Projects whose Cargo manifest depends (directly or through an app library)
/// on another project's generated `<name>-sdk` crate.
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(&current) 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 {
                    // Libraries only: a service crate never re-exports clients.
                    pending.push(dependency.clone());
                }
            }
        }
        if !targets.is_empty() {
            result.insert(name.clone(), targets);
        }
    }
    Ok(result)
}

/// `<SVC>_URL` and the HMAC key every SDK dependency needs, plus the shared
/// internal key that worker producers and their callers sign with.
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(())
}

/// Generated encrypted cache/object-store helpers read the field key ring.
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(())
}

/// A workspace Cargo setting so no build (host, image or dev container) switches
/// sqlx macros to online checking because DATABASE_URL happens to be set.
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(())
}

/// Build contexts for a local framework checkout (`[framework] path`): the
/// checkout's root, and the `framework` / `blueprint` workspaces the generated
/// Dockerfiles copy to that same absolute path. None for published crates.
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))
}

/// Point every build of the given Dockerfile at the local framework contexts,
/// or remove them once the application uses published crates again.
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");
            }
        }
    }
}

/// Drop generated `<service>-migrate` jobs whose service was deleted or renamed;
/// reconcile regenerates them under the current names.
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(())
}

/// Required `${VAR:?…}` settings of native processes that the root `.env`
/// leaves unset or empty, as (Compose service, variable).
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)
}

/// The generated local Kafka broker leaves with the last process that read it.
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)?);
    // The generated broker (an application rename keeps its first cluster id).
    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(())
}

/// Drop root `.env` settings for project `name` (`<NAME>_*`) that no Compose
/// service references any more, after its backend changed or it was deleted.
/// With `local_keys`, the project's `.env.local` loses those generated keys
/// once its processes no longer set them.
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(())
}

/// Compose services that run project `name`: its process, migrate/seed jobs and
/// worker or module processes tagged with it.
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)
        })
}

/// `old` as a whole name segment in `text` (delimited by anything but a
/// letter, digit or underscore) becomes `new`.
pub(super) fn renamed_segment(text: &str, old: &str, new: &str) -> String {
    // One scan over the original text: adjacent occurrences may share a
    // delimiter, and the replacement is never rescanned (a new name that
    // contains the old one, such as jobs -> email-jobs, must not grow again).
    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
}

/// Values generated for a project that a rename changes: `${OLD_…}` references,
/// roles, database names, namespaces, hostnames and binaries.
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)
}

/// `http://<old>:port` and `http://<old>-server:port` addresses of a renamed project.
fn renamed_host(value: &str, old: &str, new: &str) -> String {
    value
        .replace(&format!("://{old}:"), &format!("://{new}:"))
        .replace(&format!("://{old}-server:"), &format!("://{new}-server:"))
}

/// Rename a project's Compose processes in place, keeping their settings,
/// ports and infrastructure, and point every reference (depends_on, consumer
/// `<OLD>_*` settings) at the new name. The root `.env` keys follow, and the
/// generated values in it and in `.env.local` files are rewritten.
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() {
                    // Consumers address the renamed project by `<OLD>_*` settings.
                    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)?)?;
    }
    // The renamed crate's own `.env.local` holds only its settings; elsewhere
    // only `<OLD>_*` settings belong to it.
    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)
                    )
                }
                // Other settings that address the project's processes by hostname.
                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(())
}

/// Replace the application-derived prefixes an application rename changes:
/// `<old>-` names (containers, networks, volumes, queues, namespaces, Kafka
/// and JWT identities) and `/<old>_` database names in URLs.
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: &regex::Captures| {
        format!("{}{new_kebab}-", &captures[1])
    });
    renamed.replace(&format!("/{old_snake}_"), &format!("/{new_snake}_"))
}

/// An application rename keeps local runtime files consistent with a fresh
/// scaffold of the new name: Compose names and the generated values in the
/// root `.env` and each crate's `.env.local` (database names, namespaces,
/// identities). Values a person changed away from the old defaults stay.
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;
            }
            // Generated TypeScript SDK and event packages carry the
            // application scope; build scripts keep the name they find.
            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 {
            // Only values change; keys and comments stay as written.
            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() {
        // The last database project is gone; so is its role provisioning.
        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,
                );
            }
        }
    }
    // A `<svc>-seed` job runs after its service's migration, and the service
    // waits for it. Compose `up --wait` also treats a one-shot job that nothing
    // waits on as a failure once it exits.
    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 {
        // The TypeScript image compiles native SDK crates too.
        wire_framework_contexts(root, &mut compose, "Dockerfile.rust");
        wire_framework_contexts(root, &mut compose, "Dockerfile.typescript");
    } else {
        // Migrate and seed jobs added above copy their service's build; fold
        // them back into the single shared builder.
        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)
}

/// The TypeScript workspace's database scripts (`migrate:init`, `seed`, ...)
/// start PostgreSQL and connect as the owner roles, which only exist once
/// `postgresql-provision` has run.
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(())
}

/// Host tooling of a TypeScript project (`pnpm migrate:*`, seeds) connects
/// with its `.env.local` `DB_USER`. The generator writes the PostgreSQL
/// superuser there, whose tables the restricted roles cannot see; use the
/// owner role the `<name>-migrate` job migrates as. Other users are kept.
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();
        // Database 0 is taken by a service cache; each Redis worker gets its own.
        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);
        // Idempotent.
        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}"
            );
        }
        // One shared image; worker producers get the internal service key.
        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");
        // Reconciling again changes nothing.
        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"
        );
    }
}