cufflink-cli 0.19.0

use crate::commands::version;
use crate::config::CliConfig;
use crate::project_config::ProjectConfig;
use std::collections::{BTreeSet, HashMap};
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::{Duration, Instant};

const CLI_DEPLOY_API_VERSION: u32 = 1;

/// Package the bundled node_modules must contain for the web-runner to accept
/// the artifact. Validated at deploy time so a broken bundle fails the CLI
/// (non-zero, with the real cause) instead of being rejected after upload.
const REQUIRED_PACKAGE: &str = "next";

/// What a deploy asks of the platform beyond the manifest.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct DeployFlags {
    pub allow_destructive: bool,
    /// Under an enforced schema guard, leave what the database contradicts
    /// as it is.
    pub accept_drift: bool,
    /// Under an enforced schema guard, drop what would be archived.
    pub purge: bool,
    /// Print what the deploy would do and deploy nothing.
    pub plan: bool,
}

/// A plan the platform refused (422): the deploy it describes would not
/// go through. The CLI exits with [`Refused::EXIT_CODE`] for it, and 1 for
/// any other failure.
#[derive(Debug)]
pub struct Refused {
    pub body: String,
}

impl Refused {
    pub const EXIT_CODE: u8 = 3;
}

impl std::fmt::Display for Refused {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "Plan refused: {}", self.body)
    }
}

impl std::error::Error for Refused {}

/// How a deploy the platform runs beyond the request is followed: a poll
/// every `interval`, up to `attempts` in a row that fail for a reason
/// that passes (the network, a platform restarting) before it gives up,
/// and for `lost_grace` after the platform takes the deploy's replica for
/// lost, since a replica that was only slow to be heard from records how
/// the deploy ended over that verdict.
#[derive(Debug, Clone, Copy)]
struct Following {
    interval: Duration,
    attempts: u32,
    lost_grace: Duration,
}

const FOLLOWING: Following = Following {
    interval: Duration::from_secs(2),
    attempts: 5,
    lost_grace: Duration::from_secs(120),
};

/// What the platform records of a deploy whose replica went unheard from.
const REPLICA_LOST: &str = "replica lost";

/// A poll of a deploy that failed: for a reason that may pass, or for
/// good.
enum Unfollowed {
    Passing(String),
    Lasting(String),
}

/// What the platform answered a deploy or a plan it did not make. Only a
/// 422 refuses a plan; a 409 (another deploy runs), a 401 or a 503 says
/// nothing of whether the deploy would go through.
fn unanswered(plan: bool, status: reqwest::StatusCode, body: String) -> eyre::Report {
    match (plan, status) {
        (true, reqwest::StatusCode::UNPROCESSABLE_ENTITY) => Refused { body }.into(),
        (true, _) => eyre::eyre!("Plan failed ({status}): {body}"),
        (false, _) => eyre::eyre!("Deploy failed ({status}): {body}"),
    }
}

async fn poll(
    config: &CliConfig,
    client: &reqwest::Client,
    url: &str,
) -> Result<serde_json::Value, Unfollowed> {
    let passing = |e: reqwest::Error| Unfollowed::Passing(e.to_string());
    let resp = config
        .auth_request(client, reqwest::Method::GET, url)
        .send()
        .await
        .map_err(passing)?;
    let status = resp.status();
    if status.is_success() {
        return resp.json().await.map_err(passing);
    }
    let failure = format!("{status}: {}", resp.text().await.unwrap_or_default());
    Err(
        if status.is_server_error() || status == reqwest::StatusCode::TOO_MANY_REQUESTS {
            Unfollowed::Passing(failure)
        } else {
            Unfollowed::Lasting(failure)
        },
    )
}

/// Where a deploy is sent: as a plan, or to run beyond its request when
/// the platform can follow one (`async_deploy`), else as it always was.
fn deploy_url(api_url: &str, flags: &DeployFlags, supports: &[String]) -> String {
    let query = if flags.plan {
        "?plan=true"
    } else if supports
        .iter()
        .any(|s| s == cufflink_types::features::ASYNC_DEPLOY)
    {
        "?async=true"
    } else {
        ""
    };
    format!("{api_url}/api/services/deploy{query}")
}

/// A deploy's progress as one line.
fn progress_line(progress: &serde_json::Value) -> Option<String> {
    let phase = progress["phase"].as_str()?;
    let details: Vec<String> = progress
        .as_object()
        .into_iter()
        .flatten()
        .filter(|(key, _)| *key != "phase")
        .map(|(key, value)| format!("{key} {value}"))
        .collect();
    Some(if details.is_empty() {
        phase.to_string()
    } else {
        format!("{phase} ({})", details.join(", "))
    })
}

/// Follow a deploy the platform runs beyond the request, printing each
/// phase it reaches, until it ends; its result is the deploy response.
async fn await_deploy(
    config: &CliConfig,
    client: &reqwest::Client,
    deploy_id: &str,
    following: Following,
) -> eyre::Result<serde_json::Value> {
    let url = format!("{}/api/deploys/{deploy_id}", config.api_url);
    let mut last_progress = String::new();
    let mut failed = 0;
    let mut lost: Option<Instant> = None;
    loop {
        tokio::time::sleep(following.interval).await;
        let job = match poll(config, client, &url).await {
            Ok(job) => job,
            Err(Unfollowed::Passing(reason)) if failed + 1 < following.attempts => {
                failed += 1;
                eprintln!("  Deploy {deploy_id} not heard from ({reason}); asking again");
                continue;
            }
            Err(Unfollowed::Passing(reason) | Unfollowed::Lasting(reason)) => eyre::bail!(
                "Deploy {deploy_id} could not be followed ({reason}); it may still be running: \
                 GET /api/deploys/{deploy_id} says how it ends"
            ),
        };
        failed = 0;
        if let Some(line) = progress_line(&job["progress"]).filter(|line| *line != last_progress) {
            println!("  {line}");
            last_progress = line;
        }
        match (job["status"].as_str(), job["error"].as_str()) {
            (Some("succeeded"), _) => return Ok(job["result"].clone()),
            (Some("failed"), Some(REPLICA_LOST)) => {
                let since = *lost.get_or_insert_with(|| {
                    eprintln!(
                        "  The platform lost the heartbeat of the replica running deploy \
                         {deploy_id}; following it a while longer in case it goes on"
                    );
                    Instant::now()
                });
                if since.elapsed() >= following.lost_grace {
                    eyre::bail!(
                        "Deploy failed: {REPLICA_LOST}. It may still end: \
                         GET /api/deploys/{deploy_id} says how"
                    );
                }
            }
            (Some("failed"), error) => {
                eyre::bail!("Deploy failed: {}", error.unwrap_or("no reason recorded"))
            }
            _ => lost = None,
        }
    }
}

impl DeployFlags {
    /// The flags as the deploy request carries them, each with the
    /// platform feature it needs. A flag that is not set is not sent: a
    /// platform that predates it refuses a request that names it.
    fn requested(&self) -> Vec<(&'static str, &'static str)> {
        [
            (
                self.accept_drift,
                "accept_drift",
                cufflink_types::features::SCHEMA_GUARD,
            ),
            (self.purge, "purge", cufflink_types::features::SCHEMA_GUARD),
        ]
        .into_iter()
        .filter_map(|(set, flag, feature)| set.then_some((flag, feature)))
        .collect()
    }

    /// A platform that predates plans reads none in the request's query
    /// and would deploy what it was asked to plan.
    fn verify(&self, supports: &[String]) -> eyre::Result<()> {
        let plan = self
            .plan
            .then_some(("plan", cufflink_types::features::ASYNC_DEPLOY));
        match self
            .requested()
            .into_iter()
            .chain(plan)
            .find(|(_, feature)| !supports.iter().any(|s| s == feature))
        {
            Some((flag, feature)) => eyre::bail!(
                "--{} needs a platform that supports '{}'; this one does not. \
                 Upgrade the platform or deploy without the flag.",
                flag.replace('_', "-"),
                feature
            ),
            None => Ok(()),
        }
    }

    fn write(&self, request: &mut serde_json::Value) {
        self.requested().into_iter().for_each(|(flag, _)| {
            request[flag] = serde_json::Value::Bool(true);
        });
    }
}

pub async fn run(
    allow_destructive: bool,
    commit_hash_override: Option<String>,
    env: Option<&str>,
    tenant_override: Option<&str>,
) -> eyre::Result<()> {
    let flags = DeployFlags {
        allow_destructive,
        ..DeployFlags::default()
    };
    run_with(flags, commit_hash_override, env, tenant_override).await
}

pub async fn run_with(
    flags: DeployFlags,
    commit_hash_override: Option<String>,
    env: Option<&str>,
    tenant_override: Option<&str>,
) -> eyre::Result<()> {
    let mut config = CliConfig::load_with_env(env)?;
    if let Some(tenant) = tenant_override {
        config.tenant_override = Some(tenant.to_string());
    }

    if let Some(ref name) = config.env_name {
        println!("Environment: {}", name);
    }

    // Check if this is a web project
    let project = ProjectConfig::find_and_load()?;
    let mode = project
        .as_ref()
        .and_then(|p| p.service.mode.as_deref())
        .unwrap_or("rust");

    let commit_hash = commit_hash_override.or_else(detect_git_commit_hash);

    if mode == "web" {
        let service_name = project
            .as_ref()
            .and_then(|p| p.service.name.as_deref())
            .ok_or_else(|| eyre::eyre!("Web mode requires [service].name in Cufflink.toml"))?;
        deploy_web(
            &config,
            service_name,
            project.as_ref(),
            commit_hash.as_deref(),
            env,
        )
        .await
    } else {
        deploy_rust(
            &config,
            flags,
            project.as_ref(),
            commit_hash.as_deref(),
            env,
        )
        .await
    }
}

async fn deploy_rust(
    config: &CliConfig,
    flags: DeployFlags,
    project: Option<&ProjectConfig>,
    commit_hash: Option<&str>,
    env: Option<&str>,
) -> eyre::Result<()> {
    let client = config.http_client();

    let platform = verify_platform_version(config, &client).await?;
    flags.verify(&platform.supports)?;
    let sdk = version::sdk_version();

    println!("Building service...");
    let output = Command::new("cargo")
        .args(["run", "--", "--emit-manifest"])
        .output()?;

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        eyre::bail!("Build failed:\n{}", stderr);
    }

    let stdout = String::from_utf8(output.stdout)?;
    let mut manifest_json: serde_json::Value = serde_json::from_str(stdout.trim())
        .map_err(|e| eyre::eyre!("Failed to parse manifest JSON: {}", e))?;
    let manifest: cufflink_types::ServiceManifest =
        serde_json::from_value(manifest_json.clone())
            .map_err(|e| eyre::eyre!("Manifest is not a valid ServiceManifest: {}", e))?;

    let (negotiation, verdict) = negotiate(
        &platform,
        sdk.as_deref(),
        &manifest_json,
        &manifest.required_platform_features,
    );
    if verdict.is_err() {
        println!("  {negotiation}");
    }
    verdict?;
    verify_migrations(&std::env::current_dir()?, &manifest.migrations)?;

    let mode_label = match manifest.mode {
        cufflink_types::ServiceMode::Wasm => "wasm",
        cufflink_types::ServiceMode::Container => "container",
        cufflink_types::ServiceMode::Web => "web",
        cufflink_types::ServiceMode::Crud => "crud",
    };
    println!("Deploying {} (mode: {})...", manifest.name, mode_label);

    let wasm_artifact = if manifest.requires_wasm() {
        println!(
            "  WASM required ({}); building...",
            wasm_required_reason(&manifest)
        );
        let artifact = build_wasm_artifact()?;
        println!(
            "  Bundled WASM: {} bytes, hash: {}",
            artifact.size,
            &artifact.hash[..16]
        );
        Some(artifact)
    } else {
        None
    };
    let required = wasm_artifact.as_ref().map_or_else(
        || manifest.required_platform_features.clone(),
        |artifact| {
            require_module_features(
                &mut manifest_json,
                &manifest.required_platform_features,
                &artifact.features,
            )
        },
    );
    let (negotiation, verdict) = negotiate(&platform, sdk.as_deref(), &manifest_json, &required);
    println!("  {negotiation}");
    verdict?;

    let expected_wasm_hash = wasm_artifact.as_ref().map(|a| a.hash.clone());

    let mut deploy_req = serde_json::json!({
        "manifest": manifest_json,
        "allow_destructive": flags.allow_destructive,
        "commit_hash": commit_hash,
        "wasm_base64": wasm_artifact.as_ref().map(|a| a.base64.clone()),
        "cli_version": env!("CARGO_PKG_VERSION"),
        "api_version": CLI_DEPLOY_API_VERSION,
    });
    flags.write(&mut deploy_req);

    let resp = config
        .auth_request(
            &client,
            reqwest::Method::POST,
            &deploy_url(&config.api_url, &flags, &platform.supports),
        )
        .json(&deploy_req)
        .send()
        .await?;

    if !resp.status().is_success() {
        let status = resp.status();
        let body = resp.text().await.unwrap_or_default();
        return Err(unanswered(flags.plan, status, body));
    }

    let body: serde_json::Value = resp.json().await?;
    if flags.plan {
        println!(
            "Plan for {} v{}{}",
            body["name"].as_str().unwrap_or("?"),
            body["version"],
            if body["new"].as_bool() == Some(true) {
                " (a new service)"
            } else {
                ""
            }
        );
        if body["skipped"].as_bool() == Some(true) {
            println!("  Nothing to do: manifest and code unchanged");
        }
        for statement in texts(&body["plan"]["schema_changes"]) {
            println!("  {statement}");
        }
        plan_lines(&body["plan"])
            .iter()
            .for_each(|line| println!("  {line}"));
        return Ok(());
    }
    let body = match body["deploy_id"].as_str() {
        Some(deploy_id) => {
            println!("  Deploy {deploy_id} runs on the platform; following it");
            await_deploy(config, &client, deploy_id, FOLLOWING).await?
        }
        None => body,
    };
    let service_id = body["service_id"]
        .as_str()
        .ok_or_else(|| eyre::eyre!("No service_id in deploy response"))?
        .to_string();

    // Skipped deploys echo the previous deployment's wasm_hash, not ours.
    // Upload the new binary via the dedicated endpoint instead.
    if body["skipped"].as_bool() == Some(true) {
        println!(
            "No schema changes for {} v{} — manifest unchanged",
            body["name"].as_str().unwrap_or("?"),
            body["version"]
        );
        maybe_sync_configs(config, &service_id, project, env).await?;
        if manifest.requires_wasm() {
            upload_wasm(
                config,
                &client,
                &service_id,
                &platform.supports,
                commit_hash,
            )
            .await?;
        }
        verify_deploy_ready(config, &client, &service_id).await?;
        return Ok(());
    }

    if let Some(ref expected) = expected_wasm_hash {
        match body["wasm_hash"].as_str() {
            Some(h) if h == expected => {}
            Some(h) => {
                eyre::bail!(
                    "WASM hash mismatch: CLI built {} but platform stored {}. \
                     This indicates corruption somewhere in the upload path.",
                    &expected[..16],
                    &h[..16.min(h.len())]
                );
            }
            None => {
                eyre::bail!(
                    "Platform did not echo wasm_hash in the deploy response. \
                     This usually means the platform is older than expected; \
                     check `curl {}/api/version`.",
                    config.api_url
                );
            }
        }
    }

    maybe_sync_configs(config, &service_id, project, env).await?;

    println!(
        "Deployed {} v{}",
        body["name"].as_str().unwrap_or("?"),
        body["version"]
    );
    println!("  Schema changes: {}", body["schema_changes"]);
    println!("  Deployment ID:  {}", body["deployment_id"]);
    println!("  Tenant:         {}", body["tenant_slug"]);
    if let Some(hash) = body["commit_hash"].as_str() {
        println!("  Commit:         {}", hash);
    }
    if let Some(hash) = body["wasm_hash"].as_str() {
        println!("  WASM hash:      {}", &hash[..16.min(hash.len())]);
    }
    if let Some(ref handler) = manifest.on_migrate {
        println!("  on_migrate:     {} (ran between phases)", handler);
    }
    plan_lines(&body["plan"])
        .iter()
        .for_each(|line| println!("  {line}"));

    verify_deploy_ready(config, &client, &service_id).await?;

    Ok(())
}

/// Directories a service keeps no migrations in.
const NOT_SOURCES: [&str; 3] = ["target", "node_modules", ".git"];
/// How far below the service's directory its migrations are looked for.
const MIGRATIONS_DEPTH: usize = 3;

/// `root` and the directories under it, at most `depth` levels down and
/// outside [`NOT_SOURCES`].
fn directories(root: &Path, depth: usize) -> Vec<PathBuf> {
    let mut below: Vec<PathBuf> = std::fs::read_dir(root)
        .into_iter()
        .filter(|_| depth > 0)
        .flatten()
        .filter_map(Result::ok)
        .map(|entry| entry.path())
        .filter(|path| {
            path.is_dir()
                && !path
                    .file_name()
                    .and_then(|name| name.to_str())
                    .is_some_and(|name| NOT_SOURCES.contains(&name))
        })
        .collect();
    below.sort();
    std::iter::once(root.to_path_buf())
        .chain(below.iter().flat_map(|dir| directories(dir, depth - 1)))
        .collect()
}

/// Whether `dir` has a file of `migration`, whatever zeros pad its
/// version (`V007__name.sql` is `V7__name.sql`).
fn holds(dir: &Path, migration: &cufflink_types::SqlMigration) -> bool {
    let named = format!("__{}.sql", migration.name);
    std::fs::read_dir(dir)
        .into_iter()
        .flatten()
        .filter_map(Result::ok)
        .filter_map(|entry| entry.file_name().into_string().ok())
        .any(|file| {
            file.strip_prefix('V')
                .and_then(|rest| rest.strip_suffix(named.as_str()))
                .is_some_and(|digits| {
                    digits.bytes().all(|b| b.is_ascii_digit())
                        && digits.parse::<u32>() == Ok(migration.version)
                })
        })
}

/// The manifest's migrations are the files of the directory they were
/// read from. A file added since the last build is one the build did not
/// see: cargo rebuilds for a file that changed, never for a new one. Every
/// directory that holds the first migration is a candidate, so a copy of
/// the migrations elsewhere under the service is no refusal.
fn verify_migrations(root: &Path, built: &[cufflink_types::SqlMigration]) -> eyre::Result<()> {
    let Some(first) = built.first() else {
        return Ok(());
    };
    let mut candidates: Vec<PathBuf> = directories(root, MIGRATIONS_DEPTH)
        .into_iter()
        .filter(|dir| holds(dir, first))
        .collect();
    candidates.sort_by_key(|dir| dir.components().count());
    let scanned: Vec<(&PathBuf, Result<Vec<_>, _>)> = candidates
        .iter()
        .map(|dir| (dir, cufflink_migrate::scan_dir(dir)))
        .collect();
    if scanned
        .iter()
        .any(|(_, on_disk)| on_disk.as_deref() == Ok(built))
    {
        return Ok(());
    }
    match scanned.into_iter().next() {
        None => eyre::bail!(
            "the manifest has SQL migrations, and {}.sql is in no directory under {}; \
             deploy from the service's directory",
            first.file_stem(),
            root.display()
        ),
        Some((dir, Err(e))) => eyre::bail!("migrations in {}: {e}", dir.display()),
        Some((dir, Ok(on_disk))) => eyre::bail!(
            "migrations directory changed since the last build ({} holds {}, the manifest {}); \
             run `cargo clean -p <crate>` or touch src/main.rs",
            dir.display(),
            stems(&on_disk),
            stems(built)
        ),
    }
}

fn stems(migrations: &[cufflink_types::SqlMigration]) -> String {
    migrations
        .iter()
        .map(cufflink_types::SqlMigration::file_stem)
        .collect::<Vec<_>>()
        .join(", ")
}

fn migration_names(list: &serde_json::Value) -> Vec<String> {
    list.as_array()
        .into_iter()
        .flatten()
        .map(|m| {
            format!(
                "V{}__{} ({})",
                m["version"],
                m["name"].as_str().unwrap_or("?"),
                m["checksum"]
            )
        })
        .collect()
}

fn texts(list: &serde_json::Value) -> Vec<String> {
    list.as_array()
        .into_iter()
        .flatten()
        .filter_map(|item| item.as_str().map(str::to_string))
        .collect()
}

/// What the platform found in the database, as the deploy response's
/// `plan` reports it.
fn plan_lines(plan: &serde_json::Value) -> Vec<String> {
    let guard = plan["schema_guard"].as_str();
    let drift = plan["drift"].as_array().into_iter().flatten().map(|d| {
        let at = [&d["table"], &d["column"], &d["index"]]
            .into_iter()
            .filter_map(|part| part.as_str())
            .collect::<Vec<_>>()
            .join(".");
        format!("drift: {} {at}", d["kind"].as_str().unwrap_or("differs"))
    });
    let dependents = plan["dependents"]
        .as_array()
        .into_iter()
        .flatten()
        .map(|d| {
            format!(
                "unmanaged: {} {} depends on {}",
                d["kind"].as_str().unwrap_or("object").replace('_', " "),
                d["name"].as_str().unwrap_or("?"),
                d["on"].as_str().unwrap_or("?")
            )
        });
    let migrations = &plan["sql_migrations"];
    let history = migrations["history"].as_str().map(|history| {
        format!(
            "SQL migrations: {history} ({}, {} verified)",
            migrations["mode"].as_str().unwrap_or("apply"),
            migrations["verified"]
        )
    });
    history
        .into_iter()
        .chain(
            migration_names(&migrations["applied"])
                .into_iter()
                .map(|m| format!("applied: {m}")),
        )
        .chain(
            migration_names(&migrations["pending"])
                .into_iter()
                .map(|m| format!("pending: {m}")),
        )
        .chain(guard.map(|mode| format!("Schema guard:   {mode}")))
        .chain(
            texts(&plan["satisfied"])
                .into_iter()
                .map(|change| format!("already there: {change}")),
        )
        .chain(drift)
        .chain(dependents)
        .chain(
            texts(&plan["extra_columns"])
                .into_iter()
                .map(|column| format!("not declared: {column}")),
        )
        .chain(
            texts(&plan["warnings"])
                .into_iter()
                .map(|warning| format!("warning: {warning}")),
        )
        .collect()
}

/// Fetch and validate the platform's deploy API version, returning the
/// platform's version and capabilities (`supports`) so the caller can check the
/// manifest's `required_platform_features` against them.
async fn verify_platform_version(
    config: &CliConfig,
    client: &reqwest::Client,
) -> eyre::Result<version::PlatformVersion> {
    let url = format!("{}/api/version", config.api_url);
    let resp = match client.get(&url).send().await {
        Ok(r) => r,
        Err(e) => {
            eyre::bail!(
                "Could not reach platform at {}: {}\n\
                 Check your CUFFLINK_API_URL / Cufflink.toml api_url and network connectivity.",
                url,
                e
            );
        }
    };

    if resp.status().as_u16() == 404 {
        // Old platform that predates `/api/version`. Refuse — we have no
        // way to know whether it understands bundled-WASM or not, and
        // silently sending fields it drops is exactly the failure mode
        // we're fixing.
        eyre::bail!(
            "Platform at {} does not expose /api/version. \
             This CLI (v{}) requires a platform that supports deploy api_version >= {}. \
             Upgrade the platform image before deploying.",
            config.api_url,
            env!("CARGO_PKG_VERSION"),
            CLI_DEPLOY_API_VERSION
        );
    }

    if !resp.status().is_success() {
        let status = resp.status();
        let body = resp.text().await.unwrap_or_default();
        eyre::bail!("Platform /api/version returned {}: {}", status, body);
    }

    let info: version::PlatformVersion = resp.json().await?;
    let platform_api = info.api_version;
    if platform_api < CLI_DEPLOY_API_VERSION {
        eyre::bail!(
            "Platform deploy api_version is {} but this CLI (v{}) requires >= {}. \
             Upgrade the platform image before deploying.",
            platform_api,
            env!("CARGO_PKG_VERSION"),
            CLI_DEPLOY_API_VERSION
        );
    }
    Ok(info)
}

/// Refuse to deploy when the platform does not advertise a capability the
/// service requires — catching the case an old platform cannot: it would
/// silently ignore both `required_platform_features` and the feature itself
/// (e.g. running posture-annotated routes unenforced).
/// What a deploy that needs `required` negotiates, as the line it prints,
/// and whether the platform refuses it. The line is printed whatever the
/// verdict: it says why a deploy is refused.
fn negotiate(
    platform: &version::PlatformVersion,
    sdk: Option<&str>,
    manifest_json: &serde_json::Value,
    required: &[String],
) -> (String, eyre::Result<()>) {
    (
        version::negotiated(platform, sdk, manifest_json),
        verify_required_features(required, &platform.supports),
    )
}

fn verify_required_features(required: &[String], supports: &[String]) -> eyre::Result<()> {
    let missing: Vec<&String> = required
        .iter()
        .filter(|f| !supports.iter().any(|s| s == *f))
        .collect();
    if !missing.is_empty() {
        eyre::bail!(
            "This service requires platform capabilities the target does not advertise: {}. \
             Upgrade the platform image before deploying (it advertises: {}).",
            missing
                .iter()
                .map(|s| s.as_str())
                .collect::<Vec<_>>()
                .join(", "),
            if supports.is_empty() {
                "none".to_string()
            } else {
                supports.join(", ")
            }
        );
    }
    Ok(())
}

/// Add the features the module declares to the manifest it is deployed with,
/// so a platform that cannot read the module's section still refuses a
/// capability it lacks; the requirements to verify. The manifest JSON is left
/// untouched when the module declares none.
fn require_module_features(
    manifest_json: &mut serde_json::Value,
    required: &[String],
    declared: &BTreeSet<String>,
) -> Vec<String> {
    if declared.is_empty() {
        return required.to_vec();
    }
    let union: Vec<String> = required
        .iter()
        .chain(declared)
        .cloned()
        .collect::<BTreeSet<_>>()
        .into_iter()
        .collect();
    if let Some(manifest) = manifest_json.as_object_mut() {
        manifest.insert(
            "required_platform_features".into(),
            serde_json::json!(union),
        );
    }
    union
}

/// Poll `wasm/status` until `db_state: true` (binary durable in S3).
/// No-op for non-WASM services.
async fn verify_deploy_ready(
    config: &CliConfig,
    client: &reqwest::Client,
    service_id: &str,
) -> eyre::Result<()> {
    let url = format!("{}/api/services/{}/wasm/status", config.api_url, service_id);

    let delays = [
        Duration::from_millis(100),
        Duration::from_millis(200),
        Duration::from_millis(500),
        Duration::from_secs(1),
        Duration::from_secs(2),
        Duration::from_secs(4),
        Duration::from_secs(8),
        Duration::from_secs(14),
    ];

    let mut last_body = String::new();
    for delay in delays {
        let resp = config
            .auth_request(client, reqwest::Method::GET, &url)
            .send()
            .await?;
        if !resp.status().is_success() {
            let status = resp.status();
            let body = resp.text().await.unwrap_or_default();
            eyre::bail!(
                "Deploy verification failed ({} on wasm/status): {}",
                status,
                body
            );
        }
        let body: serde_json::Value = resp.json().await?;
        last_body = serde_json::to_string(&body).unwrap_or_default();

        let is_wasm_service = body["wasm_hash"]
            .as_str()
            .map(|s| !s.is_empty())
            .unwrap_or(false);
        if !is_wasm_service {
            return Ok(());
        }

        if body["db_state"].as_bool() == Some(true) {
            println!(
                "  Deploy verified: db_state=true, local_cache={}",
                body["local_cache"].as_bool().unwrap_or(false)
            );
            return Ok(());
        }

        tokio::time::sleep(delay).await;
    }

    eyre::bail!(
        "Deploy verification timed out: wasm/status never reported db_state=true \
         within 30s. Last status body: {}. The platform accepted the deploy but \
         the WASM binary is not durably persisted — this is a bug. Check platform \
         logs (`cufflink_deployment_wasm_missing_total` metric) and retry.",
        last_body
    );
}

/// Deploy a Next.js web app: run checks, build, create tarball, upload
async fn deploy_web(
    config: &CliConfig,
    service_name: &str,
    project: Option<&ProjectConfig>,
    commit_hash: Option<&str>,
    env: Option<&str>,
) -> eyre::Result<()> {
    println!("Deploying web app: {}", service_name);

    // Step 1: Pre-deploy checks
    let pkg_json = read_package_json()?;
    let scripts = pkg_json["scripts"].as_object();

    if let Some(s) = scripts {
        if s.contains_key("typecheck") {
            run_npm_script("typecheck", "Type checking")?;
        }
        if s.contains_key("lint") {
            run_npm_script("lint", "Linting")?;
        }
        if s.contains_key("test") {
            run_npm_script("test", "Running tests")?;
        }
    }

    // Step 2: Register service on platform (POST manifest before build so service exists)
    let version = pkg_json["version"].as_str().unwrap_or("0.1.0");
    let manifest = serde_json::json!({
        "name": service_name,
        "mode": "web",
        "version": version,
        "tables": [],
    });

    let deploy_req = serde_json::json!({
        "manifest": manifest,
        "allow_destructive": false,
        "commit_hash": commit_hash,
    });

    let client = config.http_client();
    let resp = config
        .auth_request(
            &client,
            reqwest::Method::POST,
            &format!("{}/api/services/deploy", config.api_url),
        )
        .json(&deploy_req)
        .send()
        .await?;

    if !resp.status().is_success() {
        let status = resp.status();
        let body = resp.text().await.unwrap_or_default();
        eyre::bail!("Deploy failed ({}): {}", status, body);
    }

    let body: serde_json::Value = resp.json().await?;
    let service_id = body["service_id"]
        .as_str()
        .ok_or_else(|| eyre::eyre!("No service_id in deploy response"))?;

    println!(
        "Registered {} v{}",
        body["name"].as_str().unwrap_or("?"),
        body["version"]
    );
    println!("  Deployment ID: {}", body["deployment_id"]);
    println!("  Tenant:        {}", body["tenant_slug"]);
    if let Some(hash) = body["commit_hash"].as_str() {
        println!("  Commit:        {}", hash);
    }

    // Step 3: Sync configs from Cufflink.toml to platform
    maybe_sync_configs(config, service_id, project, env).await?;

    // Step 4: Resolve build-time env vars. The platform `GET /config` masks
    // every is_secret=true value as "***", so for any `NEXT_PUBLIC_*` secret
    // the bundle would ship the literal "***". Local Cufflink.toml +
    // securestore holds the declared source of truth — merge that over the
    // platform response so local wins on collision; platform-only extras
    // (admin-set, not declared locally) come through as a fallback.
    println!("Resolving service configs...");
    let local_envs = resolve_local_build_envs(project, env)?;
    let mut build_envs: HashMap<String, String> = fetch_service_configs(config, service_name)
        .await
        .into_iter()
        .collect();
    build_envs.extend(local_envs);

    if !build_envs.is_empty() {
        println!(
            "  Injecting {} service configs into build",
            build_envs.len()
        );
    }

    // Step 5: Check pnpm workspace membership before building
    check_pnpm_workspace()?;

    // Step 6: Build
    println!("Building Next.js app...");
    let pm = detect_package_manager();
    let mut build_cmd = Command::new(&pm);
    build_cmd.args(["run", "build"]);
    for (key, value) in &build_envs {
        build_cmd.env(key, value);
    }
    let build_output = build_cmd.output()?;

    if !build_output.status.success() {
        eyre::bail!("Build failed:\n{}", format_command_output(&build_output));
    }
    println!("  Build complete");

    // Step 6: Verify build output exists
    if !Path::new(".next").exists() {
        eyre::bail!("Next.js build output not found at .next/");
    }

    // Step 7: Create tarball
    println!("Creating deployment artifact...");
    let tarball = create_web_tarball()?;
    let tarball_size = tarball.len();
    println!(
        "  Artifact size: {} bytes ({:.1} MB)",
        tarball_size,
        tarball_size as f64 / 1_048_576.0
    );

    // Step 8: Upload tarball
    println!("Uploading web artifact...");
    let resp = config
        .auth_request(
            &client,
            reqwest::Method::POST,
            &format!("{}/api/services/{}/web", config.api_url, service_id),
        )
        .header("Content-Type", "application/gzip")
        .body(tarball)
        .send()
        .await?;

    if resp.status().is_success() {
        let body: serde_json::Value = resp.json().await?;
        println!(
            "  Artifact hash: {}",
            body["artifact_hash"].as_str().unwrap_or("?")
        );
        println!("  Size:          {} bytes", body["size_bytes"]);
    } else {
        let status = resp.status();
        let body = resp.text().await.unwrap_or_default();
        eyre::bail!("Web artifact upload failed ({}): {}", status, body);
    }

    println!("\nWeb app deployed successfully!");
    Ok(())
}

fn check_pnpm_workspace() -> eyre::Result<()> {
    use crate::pnpm::{check_workspace_membership, WorkspaceCheck};

    let cwd = std::env::current_dir()?;
    if let WorkspaceCheck::NotIncluded {
        workspace_file,
        suggested_glob,
        ..
    } = check_workspace_membership(&cwd)
    {
        eyre::bail!(
            "This package is inside a pnpm workspace but is not listed in {}\n\n\
             pnpm install will not have installed dependencies for this package,\n\
             so the build will fail.\n\n\
             Add \"{}\" to the packages list in {} and run pnpm install",
            workspace_file.display(),
            suggested_glob,
            workspace_file.display(),
        );
    }
    Ok(())
}

fn read_package_json() -> eyre::Result<serde_json::Value> {
    let content = std::fs::read_to_string("package.json")
        .map_err(|_| eyre::eyre!("package.json not found in current directory"))?;
    serde_json::from_str(&content).map_err(|e| eyre::eyre!("Failed to parse package.json: {}", e))
}

fn run_npm_script(script: &str, label: &str) -> eyre::Result<()> {
    println!("{}...", label);
    let pm = detect_package_manager();
    let output = Command::new(&pm).args(["run", script]).output()?;

    if !output.status.success() {
        eyre::bail!("{} failed:\n{}", label, format_command_output(&output));
    }
    println!("  {} passed", label);
    Ok(())
}

/// Detect the package manager by searching for lock files upward from CWD.
fn detect_package_manager() -> String {
    let mut dir = std::env::current_dir().unwrap_or_default();
    loop {
        if dir.join("pnpm-lock.yaml").exists() {
            return "pnpm".to_string();
        }
        if dir.join("yarn.lock").exists() {
            return "yarn".to_string();
        }
        if dir.join("bun.lockb").exists() || dir.join("bun.lock").exists() {
            return "bun".to_string();
        }
        if dir.join("package-lock.json").exists() {
            return "npm".to_string();
        }
        if !dir.pop() {
            break;
        }
    }
    "npm".to_string()
}

/// Render a finished command's captured streams into one block. pnpm writes
/// its actionable errors (e.g. ERR_PNPM_CATALOG_ENTRY_NOT_FOUND_FOR_SPEC) to
/// stdout, so surfacing stderr alone hides the cause in CI logs.
fn format_command_output(output: &std::process::Output) -> String {
    [&output.stdout, &output.stderr]
        .into_iter()
        .map(|bytes| String::from_utf8_lossy(bytes).trim().to_owned())
        .filter(|stream| !stream.is_empty())
        .collect::<Vec<_>>()
        .join("\n")
}

/// Create a gzipped tarball of the Next.js app for deployment.
/// Bundles .next/, node_modules/, package.json, and public/ — same approach
/// as a standard Next.js Docker deployment (no standalone mode needed).
///
/// Uses the system `tar` command instead of the Rust tar crate because the
/// crate fails to archive files cloned by pnpm on macOS APFS filesystems.
fn create_web_tarball() -> eyre::Result<Vec<u8>> {
    let cwd = std::env::current_dir()?;
    let nm_dir = resolve_node_modules()?;

    // Create a staging dir with symlinks so tar can bundle files from
    // different locations (CWD and the pnpm deploy temp dir) into one archive.
    let staging = tempfile::tempdir()?;
    let stage = staging.path();

    // Symlink .next/ from CWD
    let next_dir = cwd.join(".next");
    if next_dir.exists() {
        std::os::unix::fs::symlink(&next_dir, stage.join(".next"))?;
    }

    // Symlink node_modules/ from pnpm deploy output (or nearest)
    std::os::unix::fs::symlink(&nm_dir, stage.join("node_modules"))?;

    // Symlink package.json from CWD
    std::os::unix::fs::symlink(cwd.join("package.json"), stage.join("package.json"))?;

    // Symlink public/ from CWD if it exists
    let public_dir = cwd.join("public");
    let has_public = public_dir.exists();
    if has_public {
        std::os::unix::fs::symlink(&public_dir, stage.join("public"))?;
    }

    let output = Command::new("tar")
        .args(web_tarball_args(has_public))
        .current_dir(stage)
        .output()?;

    if !output.status.success() {
        eyre::bail!("tar command failed:\n{}", format_command_output(&output));
    }

    // Clean up pnpm deploy temp dir
    if let Some(parent) = nm_dir.parent() {
        if parent.starts_with(std::env::temp_dir()) {
            let _ = std::fs::remove_dir_all(parent);
        }
    }

    Ok(output.stdout)
}

/// `tar` arguments for the web artifact: `-h` dereferences the staging
/// symlinks, and `.next/cache` (Next's build cache) is never read at runtime.
fn web_tarball_args(has_public: bool) -> impl Iterator<Item = &'static str> {
    [
        "czfh",
        "-",
        "--exclude=.next/cache",
        ".next",
        "node_modules",
        "package.json",
    ]
    .into_iter()
    .chain(has_public.then_some("public"))
}

/// Resolve the node_modules directory to bundle for the web artifact.
///
/// In a pnpm monorepo, `pnpm deploy --prod` produces a pruned, flat
/// node_modules with only production dependencies. If that fails — or yields a
/// tree missing the runtime — its full output is surfaced and we fall back to
/// the nearest ancestor node_modules that actually contains the runtime.
fn resolve_node_modules() -> eyre::Result<std::path::PathBuf> {
    if let Some(root) = find_pnpm_workspace_root() {
        println!("  Detected pnpm monorepo, creating pruned node_modules...");
        match pnpm_deploy_node_modules(&root) {
            Ok(node_modules) => return Ok(node_modules),
            Err(err) => eprintln!("  {err}\n  Falling back to nearest node_modules..."),
        }
    }

    find_node_modules()
}

/// Build a pruned production node_modules via `pnpm deploy`. Fails (with the
/// command's combined output) when pnpm errors or omits the required runtime,
/// cleaning up the temp dir so a partial tree is never bundled.
fn pnpm_deploy_node_modules(root: &Path) -> eyre::Result<std::path::PathBuf> {
    let pkg_name = read_package_name()?;
    let deploy_dir = tempfile::tempdir()?.keep();

    let output = Command::new("pnpm")
        .args([
            "--filter",
            &pkg_name,
            "deploy",
            "--prod",
            &deploy_dir.display().to_string(),
        ])
        .current_dir(root)
        .output()?;

    let node_modules = deploy_dir.join("node_modules");
    if output.status.success() && node_modules.join(REQUIRED_PACKAGE).exists() {
        println!("  Pruned node_modules ready");
        return Ok(node_modules);
    }

    let _ = std::fs::remove_dir_all(&deploy_dir);
    if output.status.success() {
        eyre::bail!("pnpm deploy succeeded but produced no `{REQUIRED_PACKAGE}` package");
    }
    eyre::bail!("pnpm deploy failed:\n{}", format_command_output(&output));
}

fn read_package_name() -> eyre::Result<String> {
    read_package_json()?
        .get("name")
        .and_then(serde_json::Value::as_str)
        .map(str::to_owned)
        .ok_or_else(|| eyre::eyre!("package.json missing 'name' field"))
}

/// Find the pnpm workspace root by searching upward for pnpm-lock.yaml.
fn find_pnpm_workspace_root() -> Option<std::path::PathBuf> {
    let mut dir = std::env::current_dir().ok()?;
    loop {
        if dir.join("pnpm-lock.yaml").exists() {
            // Only consider it a monorepo if we're not already at the root
            // (i.e., there's a parent with a different package.json)
            return Some(dir);
        }
        if !dir.pop() {
            break;
        }
    }
    None
}

/// Find the nearest ancestor node_modules that contains the required runtime
/// package. Keying on `REQUIRED_PACKAGE` rather than "any entry" skips pnpm
/// metadata-only dirs and `node-linker=hoisted` workspace dirs that omit
/// hoisted dependencies, mirroring the web-runner's post-extract validation.
fn find_node_modules() -> eyre::Result<std::path::PathBuf> {
    find_node_modules_from(&std::env::current_dir()?)
}

fn find_node_modules_from(start: &Path) -> eyre::Result<std::path::PathBuf> {
    let mut dir = start.to_path_buf();
    loop {
        let candidate = dir.join("node_modules");
        if candidate.join(REQUIRED_PACKAGE).exists() {
            return Ok(candidate);
        }
        if !dir.pop() {
            break;
        }
    }
    eyre::bail!(
        "No node_modules containing `{REQUIRED_PACKAGE}` found in any ancestor of {}.\n\
         Run a dependency install (e.g. `pnpm install`) before deploying.",
        start.display()
    )
}

/// Resolve the declared `[config]` + `[secrets]` for the active environment
/// into a flat env-var map for build-time injection. Returns an empty map if
/// no Cufflink.toml or no environment is selected — callers fall back to
/// platform-fetched configs only.
fn resolve_local_build_envs(
    project: Option<&ProjectConfig>,
    env: Option<&str>,
) -> eyre::Result<HashMap<String, String>> {
    let project = match project {
        Some(p) => p,
        None => return Ok(HashMap::new()),
    };
    let env_name = env
        .map(|s| s.to_string())
        .or_else(|| project.service.default_env.clone());
    let env_name = match env_name {
        Some(n) => n,
        None => return Ok(HashMap::new()),
    };
    super::config_cmd::resolve_local_envs(project, &env_name)
}

/// Sync configs and secrets from Cufflink.toml to the platform, if any are defined.
async fn maybe_sync_configs(
    config: &CliConfig,
    service_id: &str,
    project: Option<&ProjectConfig>,
    env: Option<&str>,
) -> eyre::Result<()> {
    let project = match project {
        Some(p) => p,
        None => return Ok(()),
    };

    let env_name = env
        .map(|s| s.to_string())
        .or_else(|| project.service.default_env.clone());

    let env_name = match env_name {
        Some(n) => n,
        None => return Ok(()),
    };

    let env_config = match project.environments.get(&env_name) {
        Some(c) => c,
        None => return Ok(()),
    };

    if env_config.config.is_empty() && env_config.secrets.is_empty() {
        return Ok(());
    }

    println!("Syncing configs...");
    super::config_cmd::sync_to_platform(
        config,
        service_id,
        &env_config.config,
        &env_config.secrets,
        project,
        &env_name,
    )
    .await
}

/// Fetch service configs from the platform to inject as build-time env vars.
/// Returns empty vec on any failure (first deploy, platform unreachable, etc.)
/// so the build proceeds with local env vars.
async fn fetch_service_configs(config: &CliConfig, service_name: &str) -> Vec<(String, String)> {
    let client = config.http_client();

    // Step 1: Find service ID by name
    let resp = match config
        .auth_request(
            &client,
            reqwest::Method::GET,
            &format!("{}/api/services", config.api_url),
        )
        .send()
        .await
    {
        Ok(r) => r,
        Err(_) => return Vec::new(),
    };

    let services: serde_json::Value = match resp.json().await {
        Ok(v) => v,
        Err(_) => return Vec::new(),
    };

    let service_id = match services["services"].as_array().and_then(|arr| {
        arr.iter()
            .find(|s| s["name"].as_str() == Some(service_name))
            .and_then(|s| s["id"].as_str())
    }) {
        Some(id) => id.to_string(),
        None => return Vec::new(),
    };

    // Step 2: Fetch configs for this service
    let resp = match config
        .auth_request(
            &client,
            reqwest::Method::GET,
            &format!("{}/api/services/{}/config", config.api_url, service_id),
        )
        .send()
        .await
    {
        Ok(r) if r.status().is_success() => r,
        _ => return Vec::new(),
    };

    let body: serde_json::Value = match resp.json().await {
        Ok(v) => v,
        Err(_) => return Vec::new(),
    };

    // Step 3: Extract key/value pairs
    body["configs"]
        .as_array()
        .map(|configs| {
            configs
                .iter()
                .filter_map(|c| {
                    let key = c["key"].as_str()?;
                    let value = c["value"].as_str()?;
                    Some((key.to_string(), value.to_string()))
                })
                .collect()
        })
        .unwrap_or_default()
}

fn detect_git_commit_hash() -> Option<String> {
    Command::new("git")
        .args(["rev-parse", "HEAD"])
        .output()
        .ok()
        .filter(|o| o.status.success())
        .and_then(|o| String::from_utf8(o.stdout).ok())
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty())
}

struct WasmArtifact {
    base64: String,
    size: usize,
    hash: String,
    features: BTreeSet<String>,
}

/// Human-readable explanation of why a manifest needs a WASM bundle.
/// Mirrors the predicate in [`cufflink_types::ServiceManifest::requires_wasm`]
/// so the CLI's "I'm building WASM" message matches the platform's reasoning.
fn wasm_required_reason(manifest: &cufflink_types::ServiceManifest) -> String {
    use cufflink_types::{HandlerType, ServiceMode};
    if manifest.mode == ServiceMode::Wasm {
        return "mode: wasm".into();
    }
    if manifest.on_migrate.is_some() {
        return "on_migrate hook declared".into();
    }
    if let Some(t) = manifest.tables.iter().find(|t| t.hooks.is_some()) {
        return format!("table '{}' declares CRUD hooks", t.name);
    }
    if !manifest.custom_routes.is_empty() {
        return "custom_routes declared".into();
    }
    if manifest
        .subscriptions
        .iter()
        .any(|s| matches!(s.handler_type, HandlerType::Wasm))
    {
        return "wasm subscription declared".into();
    }
    "user code declared".into()
}

fn build_wasm_artifact() -> eyre::Result<WasmArtifact> {
    let rustup_check = Command::new("rustup")
        .args(["target", "list", "--installed"])
        .output()?;
    let installed_targets = String::from_utf8_lossy(&rustup_check.stdout);
    if !installed_targets.contains("wasm32-unknown-unknown") {
        eyre::bail!("WASM target not installed. Run:\n  rustup target add wasm32-unknown-unknown");
    }

    println!("Building WASM module...");
    let build_output = Command::new("cargo")
        .args([
            "build",
            "--lib",
            "--target",
            "wasm32-unknown-unknown",
            "--release",
        ])
        .output()?;

    if !build_output.status.success() {
        let stderr = String::from_utf8_lossy(&build_output.stderr);
        eyre::bail!("WASM build failed:\n{}", stderr);
    }

    let wasm_path = find_wasm_artifact()?;
    let wasm_bytes = std::fs::read(&wasm_path)?;
    let size = wasm_bytes.len();
    println!("  WASM artifact: {} ({} bytes)", wasm_path.display(), size);

    use sha2::{Digest, Sha256};
    let mut hasher = Sha256::new();
    hasher.update(&wasm_bytes);
    let hash = hex::encode(hasher.finalize());

    let features = cufflink_types::wasm_features::declared(&wasm_bytes)
        .map_err(|e| eyre::eyre!("{}: {e}", wasm_path.display()))?;

    use base64::Engine as _;
    let base64 = base64::engine::general_purpose::STANDARD.encode(&wasm_bytes);

    Ok(WasmArtifact {
        base64,
        size,
        hash,
        features,
    })
}

/// Legacy WASM upload via the dedicated `POST /api/services/:id/wasm`
/// endpoint. Kept for backwards compat with platforms that don't yet
/// support bundled WASM in the deploy request, and for the manifest-
/// unchanged path where the deploy endpoint short-circuits before any
/// WASM handling.
async fn upload_wasm(
    config: &CliConfig,
    client: &reqwest::Client,
    service_id: &str,
    supports: &[String],
    commit_hash: Option<&str>,
) -> eyre::Result<()> {
    let artifact = build_wasm_artifact()?;
    verify_required_features(
        &artifact.features.iter().cloned().collect::<Vec<_>>(),
        supports,
    )?;
    let expected_hash = artifact.hash.clone();
    let upload_req = serde_json::json!({
        "wasm_base64": artifact.base64,
        "commit_hash": commit_hash,
    });

    println!("Uploading WASM module...");
    let resp = config
        .auth_request(
            client,
            reqwest::Method::POST,
            &format!("{}/api/services/{}/wasm", config.api_url, service_id),
        )
        .json(&upload_req)
        .send()
        .await?;

    if resp.status().is_success() {
        let body: serde_json::Value = resp.json().await?;
        let echoed = body["wasm_hash"].as_str().unwrap_or("");
        println!("  WASM hash:  {}", echoed);
        println!("  Size:       {} bytes", body["size_bytes"]);
        println!("{}", upload_outcome(&body));
        if echoed != expected_hash {
            eyre::bail!(
                "WASM hash mismatch on upload: CLI built {} but platform stored {}",
                &expected_hash[..16],
                &echoed[..16.min(echoed.len())]
            );
        }
    } else {
        let status = resp.status();
        let body = resp.text().await.unwrap_or_default();
        eyre::bail!("WASM upload failed ({}): {}", status, body);
    }

    Ok(())
}

/// What an upload did: the running version already had the module, or the
/// module became a new version. Platforms before R2 replaced the running
/// version's module and report neither.
fn upload_outcome(body: &serde_json::Value) -> String {
    match (
        body["unchanged"].as_bool(),
        body["previous_version"].as_i64(),
    ) {
        (Some(true), _) if body["restored"].as_bool() == Some(true) => format!(
            "  Version:    v{} (module unchanged, restored to storage)",
            body["version"]
        ),
        (Some(true), _) => format!("  Version:    v{} (module unchanged)", body["version"]),
        (_, Some(previous)) => format!("  Version:    v{} (was v{previous})", body["version"]),
        _ => format!("  Version:    v{}", body["version"]),
    }
}

fn find_wasm_artifact() -> eyre::Result<std::path::PathBuf> {
    // Use cargo metadata to find the target directory and crate name
    let metadata_output = Command::new("cargo")
        .args(["metadata", "--format-version", "1", "--no-deps"])
        .output()?;

    if !metadata_output.status.success() {
        eyre::bail!("Failed to run cargo metadata");
    }

    let metadata: serde_json::Value = serde_json::from_slice(&metadata_output.stdout)?;
    let target_dir = metadata["target_directory"]
        .as_str()
        .ok_or_else(|| eyre::eyre!("Cannot determine target directory"))?;

    // Find the lib target name from the current package
    let cwd = std::env::current_dir()?;
    let cwd_str = cwd.to_string_lossy();

    let crate_name = metadata["packages"]
        .as_array()
        .and_then(|pkgs| {
            pkgs.iter().find(|p| {
                p["manifest_path"]
                    .as_str()
                    .map(|mp| mp.starts_with(cwd_str.as_ref()))
                    .unwrap_or(false)
            })
        })
        .and_then(|pkg| {
            // Look for a cdylib target first, then fall back to package name
            pkg["targets"]
                .as_array()
                .and_then(|targets| {
                    targets.iter().find(|t| {
                        t["crate_types"]
                            .as_array()
                            .map(|types| types.iter().any(|ct| ct.as_str() == Some("cdylib")))
                            .unwrap_or(false)
                    })
                })
                .and_then(|t| t["name"].as_str())
                .or_else(|| pkg["name"].as_str())
        })
        .unwrap_or("unknown");

    let wasm_name = crate_name.replace('-', "_");
    let wasm_path = std::path::PathBuf::from(target_dir)
        .join("wasm32-unknown-unknown")
        .join("release")
        .join(format!("{}.wasm", wasm_name));

    if wasm_path.exists() {
        return Ok(wasm_path);
    }

    eyre::bail!(
        "WASM artifact not found at {:?}.\n\
         Ensure your Cargo.toml has:\n  \
         [lib]\n  \
         crate-type = [\"cdylib\"]",
        wasm_path
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::os::unix::process::ExitStatusExt;
    use std::process::{ExitStatus, Output};

    #[test]
    fn a_flag_is_sent_only_when_set_and_only_to_a_platform_that_knows_it() {
        let guarded = vec!["bundled_wasm".to_string(), "schema_guard".to_string()];
        let older = vec!["bundled_wasm".to_string()];
        let plain = DeployFlags {
            allow_destructive: true,
            ..DeployFlags::default()
        };
        let mut request = serde_json::json!({"allow_destructive": true});
        assert!(plain.verify(&older).is_ok());
        plain.write(&mut request);
        assert_eq!(request, serde_json::json!({"allow_destructive": true}));

        let flags = DeployFlags {
            accept_drift: true,
            purge: true,
            ..plain
        };
        assert!(flags.verify(&guarded).is_ok());
        flags.write(&mut request);
        assert_eq!(
            request,
            serde_json::json!({"allow_destructive": true, "accept_drift": true, "purge": true})
        );
        let refused = flags.verify(&older).unwrap_err().to_string();
        assert!(refused.contains("--accept-drift"), "{refused}");
        assert!(refused.contains("'schema_guard'"), "{refused}");
    }

    fn sql(version: u32, name: &str, sql: &str) -> cufflink_types::SqlMigration {
        cufflink_types::SqlMigration {
            version,
            name: name.into(),
            sql: sql.into(),
            no_transaction: false,
        }
    }

    #[test]
    fn the_manifests_migrations_are_the_files_on_disk() {
        let root = tempfile::tempdir().expect("tempdir");
        let dir = root.path().join("db").join("migrations");
        std::fs::create_dir_all(&dir).expect("dir");
        std::fs::create_dir_all(root.path().join("target").join("old")).expect("target");
        let built = vec![
            sql(1, "initial_tables", "CREATE TABLE notes ();\r\n"),
            sql(
                6,
                "add_note_kind",
                "ALTER TABLE notes ADD COLUMN kind text;",
            ),
        ];
        built.iter().for_each(|m| {
            std::fs::write(dir.join(format!("{}.sql", m.file_stem())), &m.sql).expect("write");
            std::fs::write(
                root.path()
                    .join("target")
                    .join("old")
                    .join(format!("{}.sql", m.file_stem())),
                "stale",
            )
            .expect("write");
        });
        assert!(verify_migrations(root.path(), &[]).is_ok());
        assert!(verify_migrations(root.path(), &built).is_ok());

        std::fs::write(
            dir.join("V7__callback_index.sql"),
            "CREATE INDEX i ON notes (kind);",
        )
        .expect("a file added after the build");
        let added = verify_migrations(root.path(), &built)
            .expect_err("added")
            .to_string();
        assert!(added.contains("changed since the last build"), "{added}");
        assert!(added.contains("V7__callback_index"), "{added}");
        assert!(added.contains("cargo clean"), "{added}");
        std::fs::remove_file(dir.join("V7__callback_index.sql")).expect("remove");

        std::fs::write(
            dir.join("V6__add_note_kind.sql"),
            "ALTER TABLE notes ADD COLUMN kind text;\n",
        )
        .expect("edit");
        assert!(verify_migrations(root.path(), &built).is_err());

        let elsewhere = tempfile::tempdir().expect("tempdir");
        let lost = verify_migrations(elsewhere.path(), &built)
            .expect_err("no directory")
            .to_string();
        assert!(lost.contains("V1__initial_tables.sql"), "{lost}");
    }

    #[test]
    fn a_copy_of_the_migrations_elsewhere_and_a_padded_version_are_no_refusal() {
        let root = tempfile::tempdir().expect("tempdir");
        let write = |dir: &str, file: &str, sql: &str| {
            let dir = root.path().join(dir);
            std::fs::create_dir_all(&dir).expect("dir");
            std::fs::write(dir.join(file), sql).expect("write");
        };
        let built = vec![
            sql(7, "initial_tables", "CREATE TABLE notes ();\n"),
            sql(8, "add_kind", "ALTER TABLE notes ADD COLUMN kind text;\n"),
        ];
        write(
            "a/fixtures",
            "V7__initial_tables.sql",
            "CREATE TABLE notes ();\n",
        );
        write(
            "migrations",
            "V007__initial_tables.sql",
            "CREATE TABLE notes ();\n",
        );
        write(
            "migrations",
            "V008__add_kind.sql",
            "ALTER TABLE notes ADD COLUMN kind text;\n",
        );
        assert!(
            verify_migrations(root.path(), &built).is_ok(),
            "a decoy found first, and versions padded with zeros"
        );

        write("migrations", "V9__later.sql", "SELECT 1;\n");
        let changed = verify_migrations(root.path(), &built)
            .expect_err("a file added after the build")
            .to_string();
        assert!(
            changed.contains("migrations holds V7__initial_tables, V8__add_kind, V9__later"),
            "the nearest directory is named: {changed}"
        );
        write("migrations", "notes.sql", "SELECT 1;\n");
        let unreadable = verify_migrations(root.path(), &built)
            .expect_err("a file that is no migration")
            .to_string();
        assert!(unreadable.contains("notes.sql"), "{unreadable}");
    }

    #[test]
    fn only_a_422_refuses_a_plan() {
        use reqwest::StatusCode;
        let refused = |plan: bool, status: StatusCode| {
            let report = unanswered(plan, status, "{}".into());
            (
                report.downcast_ref::<Refused>().is_some(),
                report.to_string(),
            )
        };
        assert_eq!(
            refused(true, StatusCode::UNPROCESSABLE_ENTITY),
            (true, "Plan refused: {}".to_string())
        );
        for status in [
            StatusCode::CONFLICT,
            StatusCode::UNAUTHORIZED,
            StatusCode::SERVICE_UNAVAILABLE,
            StatusCode::INTERNAL_SERVER_ERROR,
        ] {
            assert_eq!(
                refused(true, status),
                (false, format!("Plan failed ({status}): {{}}"))
            );
        }
        assert_eq!(
            refused(false, StatusCode::UNPROCESSABLE_ENTITY),
            (
                false,
                "Deploy failed (422 Unprocessable Entity): {}".to_string()
            )
        );
    }

    /// A platform that answers each request with the next of `answers`,
    /// closing the connection unanswered for `None`, and says how many it
    /// was asked.
    async fn answering(
        answers: Vec<Option<(u16, serde_json::Value)>>,
    ) -> (CliConfig, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let address = listener.local_addr().expect("address");
        let asked = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let counted = asked.clone();
        tokio::spawn(async move {
            for answer in answers {
                let Ok((mut socket, _)) = listener.accept().await else {
                    return;
                };
                counted.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                let mut request = Vec::new();
                let mut buffer = [0u8; 1024];
                while !request.windows(4).any(|w| w == b"\r\n\r\n") {
                    match socket.read(&mut buffer).await {
                        Ok(0) | Err(_) => break,
                        Ok(n) => request.extend_from_slice(&buffer[..n]),
                    }
                }
                if let Some((status, body)) = answer {
                    let body = body.to_string();
                    let response = format!(
                        "HTTP/1.1 {status} X\r\ncontent-type: application/json\r\n\
                         content-length: {}\r\nconnection: close\r\n\r\n{body}",
                        body.len()
                    );
                    let _ = socket.write_all(response.as_bytes()).await;
                }
            }
        });
        let config = CliConfig {
            api_url: format!("http://{address}"),
            tenant_slug: "acme".into(),
            api_key: Some("key".into()),
            env_name: None,
            keycloak_url: None,
            keycloak_realm: None,
            keycloak_client_id: None,
            tenant_override: None,
        };
        (config, asked)
    }

    const QUICKLY: Following = Following {
        interval: Duration::from_millis(10),
        attempts: 3,
        lost_grace: Duration::from_millis(200),
    };

    #[tokio::test]
    async fn a_deploy_taken_for_lost_is_followed_until_it_ends_or_the_grace_does() {
        let lost = || {
            Some((
                200,
                serde_json::json!({"status": "failed", "error": "replica lost"}),
            ))
        };
        let (config, _) = answering(vec![
            lost(),
            lost(),
            Some((
                200,
                serde_json::json!({"status": "succeeded", "result": {"version": 5}}),
            )),
        ])
        .await;
        let result = await_deploy(&config, &reqwest::Client::new(), "d1", QUICKLY)
            .await
            .expect("the replica went on and ended it");
        assert_eq!(result, serde_json::json!({"version": 5}));

        let (config, asked) = answering(std::iter::repeat_with(lost).take(200).collect()).await;
        let error = await_deploy(&config, &reqwest::Client::new(), "d1", QUICKLY)
            .await
            .expect_err("lost for longer than the grace")
            .to_string();
        assert!(error.starts_with("Deploy failed: replica lost"), "{error}");
        assert!(error.contains("GET /api/deploys/d1"), "{error}");
        assert!(asked.load(std::sync::atomic::Ordering::SeqCst) < 200);
    }

    #[tokio::test]
    async fn a_poll_that_fails_for_a_while_is_asked_again() {
        let running = serde_json::json!({"status": "running", "progress": {"phase": "additive"}});
        let (config, asked) = answering(vec![
            Some((503, serde_json::json!({"error": "restarting"}))),
            None,
            Some((200, running.clone())),
            Some((502, serde_json::json!({}))),
            Some((
                200,
                serde_json::json!({"status": "succeeded", "result": {"version": 4}}),
            )),
        ])
        .await;
        let result = await_deploy(&config, &reqwest::Client::new(), "d1", QUICKLY)
            .await
            .expect("followed to its end");
        assert_eq!(result, serde_json::json!({"version": 4}));
        assert_eq!(asked.load(std::sync::atomic::Ordering::SeqCst), 5);
    }

    #[tokio::test]
    async fn a_poll_that_keeps_failing_or_is_refused_gives_up() {
        let (config, asked) = answering(vec![
            Some((503, serde_json::json!({}))),
            None,
            Some((503, serde_json::json!({"error": "down"}))),
        ])
        .await;
        let error = await_deploy(&config, &reqwest::Client::new(), "d1", QUICKLY)
            .await
            .expect_err("three in a row")
            .to_string();
        assert!(error.contains("503"), "{error}");
        assert!(error.contains("may still be running"), "{error}");
        assert_eq!(asked.load(std::sync::atomic::Ordering::SeqCst), 3);

        let (config, asked) =
            answering(vec![Some((404, serde_json::json!({"error": "not found"})))]).await;
        let error = await_deploy(&config, &reqwest::Client::new(), "d1", QUICKLY)
            .await
            .expect_err("not there")
            .to_string();
        assert!(error.contains("404"), "{error}");
        assert_eq!(asked.load(std::sync::atomic::Ordering::SeqCst), 1);

        let (config, _) = answering(vec![Some((
            200,
            serde_json::json!({"status": "failed", "error": "V2__broken failed"}),
        ))])
        .await;
        let error = await_deploy(&config, &reqwest::Client::new(), "d1", QUICKLY)
            .await
            .expect_err("failed")
            .to_string();
        assert_eq!(error, "Deploy failed: V2__broken failed");
    }

    #[test]
    fn the_migrations_of_a_deploy_are_printed_with_their_checksums() {
        let plan = serde_json::json!({
            "schema_changes": [],
            "sql_migrations": {
                "history": "note_service.schema_history", "mode": "verify_only", "verified": 2,
                "baselined": [], "applied": [{"version": 6, "name": "add_note_kind", "checksum": 11}],
                "pending": [{"version": 7, "name": "callback_index", "checksum": 42}],
                "recovered_indexes": []
            },
            "warnings": ["migration_history.mode is verify_only: V7__callback_index left to the legacy authority"]
        });
        assert_eq!(
            plan_lines(&plan),
            [
                "SQL migrations: note_service.schema_history (verify_only, 2 verified)",
                "applied: V6__add_note_kind (11)",
                "pending: V7__callback_index (42)",
                "warning: migration_history.mode is verify_only: V7__callback_index left to the legacy authority",
            ]
        );
    }

    #[test]
    fn a_deploy_is_sent_where_the_platform_can_run_it() {
        let flags = DeployFlags::default();
        let older = vec!["bundled_wasm".to_string()];
        let newer = vec!["bundled_wasm".to_string(), "async_deploy".to_string()];
        assert_eq!(
            deploy_url("https://p", &flags, &older),
            "https://p/api/services/deploy"
        );
        assert_eq!(
            deploy_url("https://p", &flags, &newer),
            "https://p/api/services/deploy?async=true"
        );
        let plan = DeployFlags {
            plan: true,
            ..flags
        };
        assert_eq!(
            deploy_url("https://p", &plan, &newer),
            "https://p/api/services/deploy?plan=true"
        );
        assert!(plan.verify(&newer).is_ok());
        let refused = plan.verify(&older).unwrap_err().to_string();
        assert!(refused.contains("--plan"), "{refused}");
        assert!(refused.contains("'async_deploy'"), "{refused}");
        let mut request = serde_json::json!({});
        plan.write(&mut request);
        assert_eq!(
            request,
            serde_json::json!({}),
            "a plan is asked in the query"
        );
    }

    #[test]
    fn a_deploys_progress_is_one_line() {
        assert_eq!(
            progress_line(
                &serde_json::json!({"phase": "sql_migrations", "applied": 3, "pending": 2})
            ),
            Some("sql_migrations (applied 3, pending 2)".to_string())
        );
        assert_eq!(
            progress_line(&serde_json::json!({"phase": "on_migrate"})),
            Some("on_migrate".to_string())
        );
        assert_eq!(progress_line(&serde_json::json!({})), None);
    }

    #[test]
    fn the_plan_of_a_deploy_is_printed_line_by_line() {
        assert!(plan_lines(&serde_json::Value::Null).is_empty());
        assert!(plan_lines(&serde_json::json!({"schema_changes": ["CREATE TABLE x"]})).is_empty());
        let plan = serde_json::json!({
            "schema_changes": [],
            "schema_guard": "warn",
            "satisfied": ["DROP COLUMN pickups.midpoint"],
            "drift": [
                {"kind": "column_missing", "table": "pickups", "column": "notes"},
                {"kind": "table_missing", "table": "batches"}
            ],
            "dependents": [{
                "on": "leads_service.lead.status",
                "kind": "materialized_view",
                "name": "reporting.lead_totals",
                "description": "rule _RETURN on materialized view reporting.lead_totals"
            }],
            "extra_columns": ["pickups.legacy_ref"],
            "warnings": ["schema_guard could not read the schema: timeout"]
        });
        assert_eq!(
            plan_lines(&plan),
            [
                "Schema guard:   warn",
                "already there: DROP COLUMN pickups.midpoint",
                "drift: column_missing pickups.notes",
                "drift: table_missing batches",
                "unmanaged: materialized view reporting.lead_totals depends on leads_service.lead.status",
                "not declared: pickups.legacy_ref",
                "warning: schema_guard could not read the schema: timeout",
            ]
        );
    }

    #[test]
    fn upload_outcomes_name_the_version() {
        use serde_json::json;
        assert_eq!(
            upload_outcome(&json!({"version": 4, "unchanged": true})),
            "  Version:    v4 (module unchanged)"
        );
        assert_eq!(
            upload_outcome(&json!({"version": 4, "unchanged": true, "restored": true})),
            "  Version:    v4 (module unchanged, restored to storage)"
        );
        assert_eq!(
            upload_outcome(&json!({"version": 5, "previous_version": 4})),
            "  Version:    v5 (was v4)"
        );
        assert_eq!(upload_outcome(&json!({"version": 4})), "  Version:    v4");
    }

    #[test]
    fn required_features_accept_when_all_advertised() {
        let supports = vec![
            "route_posture".to_string(),
            "nats_publish_authz".to_string(),
        ];
        assert!(verify_required_features(&["route_posture".to_string()], &supports).is_ok());
        assert!(verify_required_features(&[], &supports).is_ok());
    }

    #[test]
    fn module_features_join_the_manifest_requirements_sorted() {
        let legacy =
            serde_json::json!({"name": "svc", "required_platform_features": ["route_posture"]});
        let mut untouched = legacy.clone();
        let required = require_module_features(
            &mut untouched,
            &["route_posture".to_string()],
            &BTreeSet::new(),
        );
        assert_eq!(untouched, legacy);
        assert_eq!(required, vec!["route_posture"]);

        let mut manifest = legacy.clone();
        let declared = BTreeSet::from(["response_v2".to_string(), "db_v2".to_string()]);
        let required =
            require_module_features(&mut manifest, &["route_posture".to_string()], &declared);
        assert_eq!(required, vec!["db_v2", "response_v2", "route_posture"]);
        assert_eq!(
            manifest["required_platform_features"],
            serde_json::json!(["db_v2", "response_v2", "route_posture"])
        );

        let mut crud_only = serde_json::json!({"name": "svc"});
        require_module_features(&mut crud_only, &[], &declared);
        assert_eq!(
            crud_only["required_platform_features"],
            serde_json::json!(["db_v2", "response_v2"])
        );
    }

    #[test]
    fn required_features_reject_when_missing() {
        let supports = vec!["bundled_wasm".to_string()];
        let err = verify_required_features(&["route_posture".to_string()], &supports)
            .unwrap_err()
            .to_string();
        assert!(err.contains("route_posture"), "got: {err}");
    }

    #[test]
    fn a_refused_deploy_is_told_what_was_negotiated() {
        let platform = version::PlatformVersion {
            api_version: 1,
            platform_version: Some("sha-4f2a".to_string()),
            supports: vec!["bundled_wasm".to_string()],
        };
        let manifest = serde_json::json!({"required_platform_features": ["streams"]});
        let (line, verdict) = negotiate(
            &platform,
            Some("v0.27.0"),
            &manifest,
            &["streams".to_string()],
        );
        assert_eq!(
            line,
            "platform sha-4f2a (features bundled_wasm) ← sdk v0.27.0 (needs streams)"
        );
        assert!(
            verdict
                .expect_err("refused")
                .to_string()
                .contains("streams"),
            "the line says what the refusal is about"
        );
        let (_, admitted) = negotiate(&platform, None, &serde_json::json!({}), &[]);
        assert!(admitted.is_ok());
    }

    fn install_package(node_modules: &Path, name: &str) {
        std::fs::create_dir_all(node_modules.join(name)).unwrap();
    }

    fn output(code: i32, stdout: &str, stderr: &str) -> Output {
        Output {
            status: ExitStatus::from_raw(code),
            stdout: stdout.as_bytes().to_vec(),
            stderr: stderr.as_bytes().to_vec(),
        }
    }

    #[test]
    fn find_node_modules_walks_past_hoisted_workspace_dir_to_root() {
        let root = tempfile::tempdir().unwrap();
        let root_nm = root.path().join("node_modules");
        install_package(&root_nm, REQUIRED_PACKAGE);

        let app = root.path().join("apps").join("portal");
        install_package(&app.join("node_modules"), "@scope-only");

        assert_eq!(find_node_modules_from(&app).unwrap(), root_nm);
    }

    #[test]
    fn find_node_modules_uses_local_dir_when_it_has_the_runtime() {
        let root = tempfile::tempdir().unwrap();
        let app = root.path().join("app");
        let app_nm = app.join("node_modules");
        install_package(&app_nm, REQUIRED_PACKAGE);

        assert_eq!(find_node_modules_from(&app).unwrap(), app_nm);
    }

    #[test]
    fn find_node_modules_errors_when_runtime_absent() {
        let root = tempfile::tempdir().unwrap();
        let app = root.path().join("app");
        install_package(&app.join("node_modules"), "other");

        let err = find_node_modules_from(&app).unwrap_err().to_string();
        assert!(err.contains(REQUIRED_PACKAGE));
    }

    #[test]
    fn format_command_output_combines_and_trims_streams() {
        assert_eq!(
            format_command_output(&output(0, "  out line  \n", "err line")),
            "out line\nerr line"
        );
    }

    #[test]
    fn format_command_output_skips_empty_streams() {
        assert_eq!(
            format_command_output(&output(0, "", "only stderr\n")),
            "only stderr"
        );
    }

    #[test]
    fn web_tarball_args_exclude_the_build_cache() {
        assert_eq!(
            web_tarball_args(false).collect::<Vec<_>>(),
            [
                "czfh",
                "-",
                "--exclude=.next/cache",
                ".next",
                "node_modules",
                "package.json"
            ]
        );
    }

    #[test]
    fn web_tarball_args_add_public_only_when_present() {
        assert_eq!(web_tarball_args(true).last(), Some("public"));
        assert_eq!(web_tarball_args(false).last(), Some("package.json"));
    }
}