obelisk 0.41.3

use crate::args::{self, DeploymentSource};
use crate::client::ClientStartup;
use crate::config::deployment::sanitize_deployment_relative_path;
use crate::config::deployment::{
    PreparedDeploymentManifest, prepare_deployment_manifest_from_disk,
};
use anyhow::{Context as _, bail};
use chrono::DateTime;
use concepts::prefixed_ulid::DeploymentId;
use grpc::grpc_gen;
use grpc::grpc_gen::switch_deployment_response::Outcome;
use grpc::to_channel;
use std::path::PathBuf;
use tonic::transport::Channel;

/// Map the CLI `--allow-unavailable-runtime-config` flag to the wire enum.
fn runtime_config_check_from_bool(allow_unavailable: bool) -> grpc_gen::RuntimeConfigCheck {
    if allow_unavailable {
        grpc_gen::RuntimeConfigCheck::AllowUnavailable
    } else {
        grpc_gen::RuntimeConfigCheck::Strict
    }
}

impl args::Deployment {
    pub(crate) async fn run(self, client_startup: ClientStartup) -> Result<(), anyhow::Error> {
        match self {
            args::Deployment::Submit {
                file,
                empty,
                allow_unavailable_runtime_config,
                description,
                deployment_id,
                api_url,
            } => {
                let prepared = prepare_manifest_from_file_or_empty(file, empty).await?;
                let channel = to_channel(&api_url).await?;
                let mut client = client_startup.deployment_repository_client(channel)?;
                let id = upload_and_submit_manifest(
                    &mut client,
                    prepared,
                    runtime_config_check_from_bool(allow_unavailable_runtime_config),
                    description,
                    deployment_id,
                )
                .await?;
                println!("{id}");
                Ok(())
            }

            args::Deployment::Enqueue {
                source,
                empty,
                allow_unavailable_runtime_config,
                description,
                deployment_id,
                api_url,
            } => {
                let runtime_config_check =
                    runtime_config_check_from_bool(allow_unavailable_runtime_config);
                let channel = to_channel(&api_url).await?;
                let mut client = client_startup.deployment_repository_client(channel)?;
                let id = submit_deployment(
                    &mut client,
                    source,
                    empty,
                    runtime_config_check,
                    description,
                    deployment_id,
                )
                .await?;
                switch_deployment(
                    &mut client,
                    id,
                    runtime_config_check,
                    SwitchCommand::Enqueue,
                )
                .await
            }

            args::Deployment::Apply {
                source,
                empty,
                description,
                deployment_id,
                api_url,
            } => {
                // A hot redeploy always requires runtime config to be available.
                let runtime_config_check = grpc_gen::RuntimeConfigCheck::Strict;
                let channel = to_channel(&api_url).await?;
                let mut client = client_startup.deployment_repository_client(channel)?;
                let id = submit_deployment(
                    &mut client,
                    source,
                    empty,
                    runtime_config_check,
                    description,
                    deployment_id,
                )
                .await?;
                switch_deployment(&mut client, id, runtime_config_check, SwitchCommand::Apply).await
            }

            args::Deployment::List { api_url } => {
                let channel = to_channel(&api_url).await?;
                let mut client = client_startup.deployment_repository_client(channel)?;
                let resp = client
                    .list_deployments(grpc_gen::ListDeploymentsRequest {
                        pagination: None,
                        include_deployment_toml: false,
                        include_execution_counts: false,
                        include_component_summary: false,
                        include_derived: false,
                    })
                    .await?
                    .into_inner();

                if resp.deployments.is_empty() {
                    println!("No deployments found.");
                    return Ok(());
                }

                println!(
                    "{:<32}  {:<12}  {:<19}  {:<19}  DESCRIPTION",
                    "ID", "STATUS", "CREATED_AT", "LAST_ACTIVE_AT"
                );
                for summary in resp.deployments {
                    let dep = summary.deployment.context("missing deployment")?;
                    let id = dep
                        .deployment_id
                        .as_ref()
                        .map(|d| d.id.as_str())
                        .unwrap_or_default()
                        .to_string();
                    let status = format_status(dep.status());
                    let created: DateTime<_> = dep.created_at.expect("created_at is sent").into();
                    let created = created.format("%Y-%m-%d %H:%M:%S");
                    let last_active = dep
                        .last_active_at
                        .map(|t| {
                            let dt: DateTime<_> = t.into();
                            dt.format("%Y-%m-%d %H:%M:%S").to_string()
                        })
                        .unwrap_or_default();
                    println!(
                        "{id:<32}  {status:<12}  {created:<19}  {last_active:<19}  {}",
                        dep.description.unwrap_or_default()
                    );
                }
                Ok(())
            }

            args::Deployment::Gc { api_url } => {
                let channel = to_channel(&api_url).await?;
                let mut client = client_startup.deployment_repository_client(channel)?;
                let resp = client
                    .gc_orphan_files(grpc_gen::GcOrphanFilesRequest {})
                    .await?
                    .into_inner();
                println!("Deleted {} orphan file blob(s).", resp.deleted_count);
                Ok(())
            }

            args::Deployment::Active { api_url, json } => {
                let channel = to_channel(&api_url).await?;
                let mut client = client_startup.deployment_repository_client(channel)?;
                let resp = client
                    .get_current_deployment_id(grpc_gen::GetCurrentDeploymentIdRequest {})
                    .await?
                    .into_inner();
                let id = resp.deployment_id.context("missing deployment_id")?.id;
                if json {
                    println!("\"{id}\"");
                } else {
                    println!("{id}");
                }
                Ok(())
            }

            args::Deployment::Show {
                id,
                file,
                json,
                api_url,
            } => {
                let channel = to_channel(&api_url).await?;
                let mut client = client_startup.deployment_repository_client(channel)?;
                let resp = client
                    .get_deployment(grpc_gen::GetDeploymentRequest {
                        deployment_id: Some(grpc_gen::DeploymentId { id: id.to_string() }),
                        include_generated_metadata: Some(true),
                    })
                    .await?
                    .into_inner();
                let dep = resp.deployment.context("deployment not found")?;
                let deployment_toml = dep
                    .deployment_toml
                    .context("deployment_toml not available")?;

                if let Some(file) = file {
                    // Normalize the requested path the same way `get` writes it, so a
                    // `./scripts/x` or `scripts//x` still matches the stored ref path.
                    let rel = sanitize_deployment_relative_path(&file)
                        .with_context(|| format!("invalid source path `{file}`"))?;
                    let file_ref = dep.files.iter().find(|f| f.path == rel).with_context(|| {
                        format!("deployment {id} has no deployment-owned source file `{rel}`")
                    })?;
                    let bytes = fetch_file(&mut client, &file_ref.digest).await?;
                    print!("{}", String::from_utf8_lossy(&bytes));
                    return Ok(());
                }

                if json {
                    // The manifest is the source of truth; render it as JSON for tooling.
                    let value: toml::Value = toml::from_str(&deployment_toml)
                        .context("cannot parse stored deployment manifest")?;
                    println!("{}", serde_json::to_string_pretty(&value)?);
                    return Ok(());
                }

                print!("{deployment_toml}");
                Ok(())
            }

            args::Deployment::Get {
                id,
                output,
                force,
                include_generated_metadata,
                api_url,
            } => {
                let channel = to_channel(&api_url).await?;
                let mut client = client_startup.deployment_repository_client(channel)?;
                let resp = client
                    .get_deployment(grpc_gen::GetDeploymentRequest {
                        deployment_id: Some(grpc_gen::DeploymentId { id: id.to_string() }),
                        include_generated_metadata: Some(include_generated_metadata),
                    })
                    .await?
                    .into_inner();
                let dep = resp.deployment.context("deployment not found")?;
                let deployment_toml = dep
                    .deployment_toml
                    .context("deployment_toml not available")?;

                let output_dir = output.unwrap_or_else(|| PathBuf::from("."));
                tokio::fs::create_dir_all(&output_dir)
                    .await
                    .with_context(|| format!("cannot create output directory {output_dir:?}"))?;

                let toml_path = output_dir.join("deployment.toml");
                write_new_file(&toml_path, deployment_toml.as_bytes(), force).await?;
                let file_count = dep.files.len();
                for file_ref in &dep.files {
                    // Defensively re-validate the relative path so a malformed stored
                    // manifest can never write outside the output directory.
                    let rel =
                        sanitize_deployment_relative_path(&file_ref.path).with_context(|| {
                            format!("refusing to write unsafe source path `{}`", file_ref.path)
                        })?;
                    let path = output_dir.join(&rel);
                    if let Some(parent) = path.parent() {
                        tokio::fs::create_dir_all(parent).await.with_context(|| {
                            format!("cannot create source directory {parent:?}")
                        })?;
                    }
                    let bytes = fetch_file(&mut client, &file_ref.digest).await?;
                    write_new_file(&path, &bytes, force).await?;
                }
                println!(
                    "Wrote {} ({file_count} source file{}) for deployment {id}",
                    toml_path.display(),
                    if file_count == 1 { "" } else { "s" }
                );
                Ok(())
            }

            // `deployment verify` is dispatched through the local verification path in `main`,
            // so it never reaches the gRPC client here.
            args::Deployment::Verify(_) => unreachable!("handled in main before ClientStartup"),
        }
    }
}

/// Write `contents` to `path`. Refuses to overwrite an existing file unless `force`.
async fn write_new_file(
    path: &std::path::Path,
    contents: &[u8],
    force: bool,
) -> anyhow::Result<()> {
    use tokio::io::AsyncWriteExt as _;
    let mut file = tokio::fs::OpenOptions::new()
        .write(true)
        .create(true) // allow creating new files
        .truncate(true) // truncate when overwriting is permitted
        .create_new(!force) // when set, only new-file creation is allowed (no overwrite)
        .open(path)
        .await
        .with_context(|| {
            format!(
                "cannot open {path:?} for writing{}",
                if force { "" } else { ", try using `--force`" }
            )
        })?;
    file.write_all(contents)
        .await
        .with_context(|| format!("cannot write {path:?}"))?;
    Ok(())
}

type DeploymentClient = grpc::grpc_gen::deployment_repository_client::DeploymentRepositoryClient<
    tonic::service::interceptor::InterceptedService<Channel, crate::client::ClientInterceptor>,
>;

/// If the source is a file, submit it and return the new ID. If it's an ID, return it directly.
/// If `empty`, submit an empty deployment and return the new ID.
async fn submit_deployment(
    client: &mut DeploymentClient,
    source: Option<DeploymentSource>,
    empty: bool,
    runtime_config_check: grpc_gen::RuntimeConfigCheck,
    description: Option<String>,
    deployment_id: Option<DeploymentId>,
) -> anyhow::Result<DeploymentId> {
    assert_ne!(source.is_some(), empty);
    let prepared = match source {
        Some(DeploymentSource::Id(id)) => {
            if description.is_some() {
                bail!("--description cannot be used with an existing deployment ID");
            }
            if deployment_id.is_some() {
                bail!("--deployment-id cannot be used with an existing deployment ID source");
            }
            return Ok(id);
        }
        Some(DeploymentSource::File(path)) => prepare_deployment_manifest_from_disk(&path).await?,
        None => prepare_manifest_from_file_or_empty(None, empty).await?,
    };
    let id = upload_and_submit_manifest(
        client,
        prepared,
        runtime_config_check,
        description,
        deployment_id,
    )
    .await?;
    println!("Submitted as {id}");
    Ok(id)
}

/// Submit the enriched manifest as a CAS-efficient package: first without file
/// blobs, and if the server reports missing files, retry the same submit with only
/// those blobs attached. The requested TOML is identical on both attempts.
async fn upload_and_submit_manifest(
    client: &mut DeploymentClient,
    prepared: PreparedDeploymentManifest,
    runtime_config_check: grpc_gen::RuntimeConfigCheck,
    description: Option<String>,
    deployment_id: Option<DeploymentId>,
) -> anyhow::Result<DeploymentId> {
    // Preflight: no blobs, so digests already in the CAS are not re-uploaded.
    let missing = match submit_attempt(
        client,
        &prepared,
        runtime_config_check,
        description.as_deref(),
        deployment_id,
        Vec::new(),
    )
    .await?
    {
        SubmitAttempt::Stored(id) => return Ok(id),
        SubmitAttempt::Missing(digests) => digests,
    };

    // Retry with only the blobs the server is missing.
    let files = prepared
        .files
        .iter()
        .filter(|file| missing.contains(&file.digest.to_string()))
        .map(|file| grpc_gen::DeploymentFileContent {
            path: file.path.clone(),
            digest: Some(file.digest.to_string()),
            content: file.bytes.clone(),
        })
        .collect();
    match submit_attempt(
        client,
        &prepared,
        runtime_config_check,
        description.as_deref(),
        deployment_id,
        files,
    )
    .await?
    {
        SubmitAttempt::Stored(id) => Ok(id),
        SubmitAttempt::Missing(digests) => bail!(
            "server is still missing {} file blob(s) after upload: {}",
            digests.len(),
            digests.join(", ")
        ),
    }
}

enum SubmitAttempt {
    Stored(DeploymentId),
    /// Digests the server is missing; the deployment was not stored.
    Missing(Vec<String>),
}

/// One submit request. A `FAILED_PRECONDITION` carrying only `missing_files`
/// becomes `Missing`; any other structured issue or status is a hard error.
async fn submit_attempt(
    client: &mut DeploymentClient,
    prepared: &PreparedDeploymentManifest,
    runtime_config_check: grpc_gen::RuntimeConfigCheck,
    description: Option<&str>,
    deployment_id: Option<DeploymentId>,
    files: Vec<grpc_gen::DeploymentFileContent>,
) -> anyhow::Result<SubmitAttempt> {
    let resp = client
        .submit_deployment(grpc_gen::SubmitDeploymentRequest {
            deployment_toml: prepared.deployment_toml.clone(),
            created_by: Some("cli".to_string()),
            runtime_config_check: runtime_config_check.into(),
            description: description.map(str::to_string),
            deployment_id: deployment_id.map(grpc_gen::DeploymentId::from),
            files,
        })
        .await;
    match resp {
        Ok(resp) => {
            let resp = resp.into_inner();
            Ok(SubmitAttempt::Stored(DeploymentId::try_from(
                resp.deployment_id.context("missing deployment_id")?,
            )?))
        }
        Err(status) => {
            if let Some(detail) = decode_submit_error_detail(&status) {
                let only_missing = detail.unexpected_files.is_empty()
                    && detail.digest_mismatches.is_empty()
                    && detail.oversized_files.is_empty()
                    && detail.missing_digest_fields.is_empty()
                    && !detail.missing_files.is_empty();
                if only_missing {
                    let digests = detail
                        .missing_files
                        .iter()
                        .filter_map(|issue| issue.digest.clone())
                        .collect();
                    return Ok(SubmitAttempt::Missing(digests));
                }
                bail!(
                    "deployment submit rejected: {}",
                    format_submit_detail(&detail)
                );
            }
            Err(status.into())
        }
    }
}

/// Decode the `SubmitDeploymentErrorDetail` attached to a submit status, if present.
fn decode_submit_error_detail(
    status: &tonic::Status,
) -> Option<grpc_gen::SubmitDeploymentErrorDetail> {
    use prost::Message as _;
    let details = status.details();
    if details.is_empty() {
        return None;
    }
    grpc_gen::SubmitDeploymentErrorDetail::decode(details).ok()
}

fn format_submit_detail(detail: &grpc_gen::SubmitDeploymentErrorDetail) -> String {
    let mut lines = Vec::new();
    for issue in &detail.missing_digest_fields {
        lines.push(format!("missing content_digest at {}", issue.field_path));
    }
    for issue in &detail.missing_files {
        lines.push(format!(
            "missing file at {} ({})",
            issue.field_path, issue.message
        ));
    }
    for issue in &detail.unexpected_files {
        lines.push(format!("unexpected file {}", issue.field_path));
    }
    for mismatch in &detail.digest_mismatches {
        lines.push(format!(
            "digest mismatch for {}: supplied {}, actual {}",
            mismatch.file.as_ref().map_or("", |file| &file.field_path),
            mismatch.supplied_digest,
            mismatch.actual_digest
        ));
    }
    for issue in &detail.oversized_files {
        lines.push(format!("oversized file {}", issue.field_path));
    }
    lines.join("; ")
}

/// Fetch a deployment file blob from the server's content-addressed store.
async fn fetch_file(client: &mut DeploymentClient, digest: &str) -> anyhow::Result<Vec<u8>> {
    let resp = client
        .get_file(grpc_gen::GetFileRequest {
            digest: digest.to_string(),
        })
        .await
        .with_context(|| format!("cannot fetch deployment file `{digest}`"))?
        .into_inner();
    Ok(resp.content)
}

async fn switch_deployment(
    client: &mut DeploymentClient,
    id: DeploymentId,
    runtime_config_check: grpc_gen::RuntimeConfigCheck,
    command: SwitchCommand,
) -> anyhow::Result<()> {
    let resp = client
        .switch_deployment(grpc_gen::SwitchDeploymentRequest {
            deployment_id: Some(grpc_gen::DeploymentId::from(id)),
            runtime_config_check: runtime_config_check.into(),
            apply: command == SwitchCommand::Apply,
        })
        .await?
        .into_inner();

    match (command, resp.outcome()) {
        (SwitchCommand::Apply, Outcome::SwitchOutcomeSwitched) => {
            println!("Applied successfully.");
        }
        (SwitchCommand::Apply, Outcome::SwitchOutcomeRestartRequired) => {
            bail!("Could not apply immediately; deployment enqueued. Restart the server to apply.");
        }
        (SwitchCommand::Enqueue, Outcome::SwitchOutcomeSwitched) => {
            println!("Deployment already active; it will remain active after restart.");
        }
        (SwitchCommand::Enqueue, Outcome::SwitchOutcomeRestartRequired) => {
            println!("Deployment enqueued. Restart the server to apply.");
        }
        (_, Outcome::SwitchOutcomeUnspecified) => {
            bail!("Unexpected outcome from server.");
        }
    }
    Ok(())
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SwitchCommand {
    Apply,
    Enqueue,
}

async fn prepare_manifest_from_file_or_empty(
    file: Option<std::path::PathBuf>,
    empty: bool,
) -> anyhow::Result<PreparedDeploymentManifest> {
    assert_ne!(file.is_some(), empty);
    if let Some(path) = file {
        prepare_deployment_manifest_from_disk(&path).await
    } else {
        Ok(PreparedDeploymentManifest::empty())
    }
}

fn format_status(status: grpc_gen::DeploymentStatus) -> &'static str {
    match status {
        grpc_gen::DeploymentStatus::Inactive => "Inactive",
        grpc_gen::DeploymentStatus::Enqueued => "Enqueued",
        grpc_gen::DeploymentStatus::Active => "Active",
        grpc_gen::DeploymentStatus::Unspecified => "Unknown",
    }
}