wasmeld/crates/wasmeld-package/src/dev.rs

//! Local build, watch, and deployment loop for Component development.

use std::{
    collections::BTreeSet,
    env, fs, io,
    path::{Path, PathBuf},
    thread,
    time::Duration,
};

use reqwest::{
    StatusCode,
    blocking::{Client, multipart},
};
use serde::Deserialize;
use sha2::{Digest, Sha256};
use wasmeld_package::module::{MODULE_MANIFEST_FILE, ModuleManifest};

use crate::{BuildProfile, PackRequest, PackageIdentity, PackedComponent, pack_component};

const DEFAULT_CONSOLE: &str = "http://127.0.0.1:8080";
const DEFAULT_POLL_MS: u64 = 350;

struct DevOptions {
    manifest_path: PathBuf,
    console: String,
    service_id: Option<String>,
    locked: bool,
    release: bool,
    once: bool,
    poll_interval: Duration,
}

#[derive(Debug, Deserialize)]
struct ServiceList {
    services: Vec<ServiceIdentity>,
}

#[derive(Debug, Deserialize)]
struct ServiceIdentity {
    id: String,
    revision: String,
}

#[derive(Debug, Deserialize)]
struct Deployment {
    active_revision: String,
}

pub(crate) fn run(arguments: Vec<String>) -> Result<(), Box<dyn std::error::Error>> {
    let options = parse_options(arguments)?;
    let manifest_path = fs::canonicalize(&options.manifest_path)?;
    let watch_roots = watch_roots(&manifest_path)?;
    let client = Client::builder()
        .user_agent(format!("wasmeld-dev/{}", env!("CARGO_PKG_VERSION")))
        .timeout(Duration::from_secs(120))
        .build()?;
    ensure_console(&client, &options.console)?;

    let temporary_dir = env::temp_dir().join(format!("wasmeld-dev-{}", std::process::id()));
    fs::create_dir_all(&temporary_dir)?;
    let output = temporary_dir.join("component.wasmpkg");
    let mut observed = None;

    println!("watching:");
    for root in &watch_roots {
        println!("  {}", root.display());
    }
    println!("console: {}", options.console);

    loop {
        let current = source_fingerprint(&watch_roots)?;
        if observed.as_ref() == Some(&current) {
            thread::sleep(options.poll_interval);
            continue;
        }

        println!("change detected; building Component");
        let result = build_and_deploy(&client, &options, &manifest_path, &output);
        observed = Some(source_fingerprint(&watch_roots)?);
        match result {
            Ok(packed) => {
                println!(
                    "ready: {}@{} ({})",
                    packed.manifest.id, packed.manifest.revision, packed.manifest.sha256
                );
                if options.once {
                    return Ok(());
                }
            }
            Err(error) if options.once => return Err(error),
            Err(error) => {
                eprintln!("wasmeld dev: {error}");
                eprintln!("waiting for the next source change; the previous deployment is intact");
            }
        }
    }
}

fn parse_options(arguments: Vec<String>) -> Result<DevOptions, Box<dyn std::error::Error>> {
    let mut arguments = arguments.into_iter();
    let manifest_path = arguments.next().ok_or("dev requires a Cargo.toml path")?;
    let mut console = env::var("WASMELD_CONSOLE").unwrap_or_else(|_| DEFAULT_CONSOLE.to_owned());
    let mut service_id = None;
    let mut locked = false;
    let mut release = false;
    let mut once = false;
    let mut poll_ms = DEFAULT_POLL_MS;

    while let Some(argument) = arguments.next() {
        match argument.as_str() {
            "--console" => {
                console = arguments.next().ok_or("--console requires an HTTP URL")?;
            }
            "--id" => {
                service_id = Some(arguments.next().ok_or("--id requires a service ID")?);
            }
            "--locked" => locked = true,
            "--release" => release = true,
            "--once" => once = true,
            "--poll-ms" => {
                poll_ms = arguments
                    .next()
                    .ok_or("--poll-ms requires a number")?
                    .parse::<u64>()?;
                if poll_ms < 100 {
                    return Err("--poll-ms must be at least 100".into());
                }
            }
            _ => return Err(format!("unknown dev argument {argument:?}").into()),
        }
    }

    Ok(DevOptions {
        manifest_path: PathBuf::from(manifest_path),
        console: console.trim_end_matches('/').to_owned(),
        service_id,
        locked,
        release,
        once,
        poll_interval: Duration::from_millis(poll_ms),
    })
}

fn build_and_deploy(
    client: &Client,
    options: &DevOptions,
    manifest_path: &Path,
    output: &Path,
) -> Result<PackedComponent, Box<dyn std::error::Error>> {
    let packed = pack_component(PackRequest {
        manifest_path: manifest_path.to_owned(),
        output: Some(output.to_owned()),
        no_build: false,
        locked: options.locked,
        profile: if options.release {
            BuildProfile::Release
        } else {
            BuildProfile::Debug
        },
        identity: PackageIdentity::Development {
            id: options.service_id.clone(),
        },
    })?;
    let previous = active_revision(client, &options.console, &packed.manifest.id)?;
    register(client, &options.console, &packed)?;
    activate(
        client,
        &options.console,
        &packed.manifest.id,
        &packed.manifest.revision,
    )?;

    if let Some(previous) = previous
        && previous != packed.manifest.revision
        && is_matching_dev_revision(&previous, &packed.manifest.revision)
        && let Err(error) = unregister(client, &options.console, &packed.manifest.id, &previous)
    {
        eprintln!(
            "wasmeld dev: deployed successfully but could not unregister {}@{}: {error}",
            packed.manifest.id, previous
        );
    }
    Ok(packed)
}

fn ensure_console(client: &Client, console: &str) -> Result<(), Box<dyn std::error::Error>> {
    let response = client.get(format!("{console}/healthz")).send()?;
    if response.status().is_success() {
        Ok(())
    } else {
        Err(http_error("Console health check", response))
    }
}

fn active_revision(
    client: &Client,
    console: &str,
    service_id: &str,
) -> Result<Option<String>, Box<dyn std::error::Error>> {
    let response = client
        .get(format!("{console}/api/v1/deployments/{service_id}"))
        .send()?;
    match response.status() {
        status if status.is_success() => Ok(Some(response.json::<Deployment>()?.active_revision)),
        StatusCode::NOT_FOUND => Ok(None),
        _ => Err(http_error("deployment lookup", response)),
    }
}

fn register(
    client: &Client,
    console: &str,
    packed: &PackedComponent,
) -> Result<(), Box<dyn std::error::Error>> {
    let package = fs::read(&packed.output)?;
    let part = multipart::Part::bytes(package)
        .file_name("component.wasmpkg")
        .mime_str("application/vnd.wasmeld.package")?;
    let response = client
        .post(format!("{console}/api/v1/services"))
        .multipart(multipart::Form::new().part("package", part))
        .send()?;
    if response.status().is_success() {
        return Ok(());
    }
    if response.status() == StatusCode::CONFLICT
        && service_exists(
            client,
            console,
            &packed.manifest.id,
            &packed.manifest.revision,
        )?
    {
        return Ok(());
    }
    Err(http_error("Component registration", response))
}

fn service_exists(
    client: &Client,
    console: &str,
    service_id: &str,
    revision: &str,
) -> Result<bool, Box<dyn std::error::Error>> {
    let response = client.get(format!("{console}/api/v1/services")).send()?;
    if !response.status().is_success() {
        return Err(http_error("service lookup", response));
    }
    Ok(response
        .json::<ServiceList>()?
        .services
        .iter()
        .any(|service| service.id == service_id && service.revision == revision))
}

fn activate(
    client: &Client,
    console: &str,
    service_id: &str,
    revision: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let response = client
        .post(format!(
            "{console}/api/v1/deployments/{service_id}/activate"
        ))
        .json(&serde_json::json!({ "revision": revision }))
        .send()?;
    if response.status().is_success() {
        Ok(())
    } else {
        Err(http_error("development deployment", response))
    }
}

fn unregister(
    client: &Client,
    console: &str,
    service_id: &str,
    revision: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let response = client
        .delete(format!("{console}/api/v1/services/{service_id}/{revision}"))
        .send()?;
    if response.status().is_success() || response.status() == StatusCode::NOT_FOUND {
        Ok(())
    } else {
        Err(http_error("old development revision cleanup", response))
    }
}

fn http_error(action: &str, response: reqwest::blocking::Response) -> Box<dyn std::error::Error> {
    let status = response.status();
    let body = response
        .text()
        .unwrap_or_else(|error| format!("failed to read response: {error}"));
    format!("{action} returned HTTP {status}: {body}").into()
}

fn is_matching_dev_revision(previous: &str, current: &str) -> bool {
    let Some((previous_base, previous_hash)) = previous.rsplit_once("-dev.h") else {
        return false;
    };
    let Some((current_base, current_hash)) = current.rsplit_once("-dev.h") else {
        return false;
    };
    previous_base == current_base && valid_dev_hash(previous_hash) && valid_dev_hash(current_hash)
}

fn valid_dev_hash(value: &str) -> bool {
    value.len() == 12 && value.bytes().all(|byte| byte.is_ascii_hexdigit())
}

fn watch_roots(manifest_path: &Path) -> Result<Vec<PathBuf>, Box<dyn std::error::Error>> {
    let component_root = manifest_path
        .parent()
        .ok_or("Cargo manifest must have a parent directory")?
        .to_owned();
    let mut roots = BTreeSet::from([component_root.clone()]);
    let module_path = component_root.join(MODULE_MANIFEST_FILE);
    if module_path.is_file() {
        let module = ModuleManifest::read(&module_path)?;
        for replacement in module.replacements.values() {
            let path = component_root.join(&replacement.path);
            if path.exists() {
                roots.insert(fs::canonicalize(path)?);
            }
        }
    }
    Ok(roots.into_iter().collect())
}

fn source_fingerprint(roots: &[PathBuf]) -> Result<Vec<u8>, Box<dyn std::error::Error>> {
    let mut files = BTreeSet::new();
    for root in roots {
        collect_source_files(root, &mut files)?;
    }
    let mut digest = Sha256::new();
    for path in files {
        digest.update(path.to_string_lossy().as_bytes());
        match fs::read(&path) {
            Ok(bytes) => digest.update(bytes),
            Err(error) if error.kind() == io::ErrorKind::NotFound => {}
            Err(error) => return Err(error.into()),
        }
    }
    Ok(digest.finalize().to_vec())
}

fn collect_source_files(
    path: &Path,
    files: &mut BTreeSet<PathBuf>,
) -> Result<(), Box<dyn std::error::Error>> {
    if path.is_dir() {
        if ignored_directory(path) {
            return Ok(());
        }
        for entry in fs::read_dir(path)? {
            collect_source_files(&entry?.path(), files)?;
        }
    } else if is_source_file(path) {
        files.insert(path.to_owned());
    }
    Ok(())
}

fn ignored_directory(path: &Path) -> bool {
    let name = path.file_name().and_then(|name| name.to_str());
    if matches!(
        name,
        Some(".git" | ".wasmeld" | "deps" | "dist" | "node_modules" | "target")
    ) {
        return name != Some("deps")
            || path
                .parent()
                .and_then(Path::file_name)
                .and_then(|name| name.to_str())
                == Some("wit");
    }
    false
}

fn is_source_file(path: &Path) -> bool {
    matches!(
        path.extension().and_then(|extension| extension.to_str()),
        Some("rs" | "wit")
    ) || matches!(
        path.file_name().and_then(|name| name.to_str()),
        Some("Cargo.lock" | "Cargo.toml" | "config.toml" | "wasmeld.toml" | "wit.lock")
    )
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn recognizes_only_matching_hash_development_revisions() {
        assert!(is_matching_dev_revision(
            "0.1.0-dev.h0123456789ab",
            "0.1.0-dev.habcdef012345"
        ));
        assert!(!is_matching_dev_revision(
            "0.2.0-dev.h0123456789ab",
            "0.1.0-dev.habcdef012345"
        ));
        assert!(!is_matching_dev_revision(
            "0.1.0",
            "0.1.0-dev.habcdef012345"
        ));
    }

    #[test]
    fn fingerprints_sources_but_ignores_materialized_wit_dependencies() {
        let root = tempfile::tempdir().unwrap();
        let source = root.path().join("src/lib.rs");
        let generated = root.path().join("wit/deps/example/package.wit");
        fs::create_dir_all(source.parent().unwrap()).unwrap();
        fs::create_dir_all(generated.parent().unwrap()).unwrap();
        fs::write(&source, "pub fn first() {}").unwrap();
        fs::write(&generated, "package generated:dependency@1.0.0;").unwrap();
        let roots = vec![root.path().to_owned()];
        let initial = source_fingerprint(&roots).unwrap();

        fs::write(&generated, "package generated:dependency@2.0.0;").unwrap();
        assert_eq!(source_fingerprint(&roots).unwrap(), initial);

        fs::write(&source, "pub fn second() {}").unwrap();
        assert_ne!(source_fingerprint(&roots).unwrap(), initial);
    }
}