use std::io::Write;
use anyhow::{Context, Result, bail};
use dialoguer::{Confirm, theme::ColorfulTheme};
use termcolor::{Color, ColorChoice, StandardStream, WriteColor};
use crate::{
constants::{get_platform_ui_url, get_resource_management_api_url},
core::{
hmac::AuthMode,
http_client::{patch, post},
},
deploy::utils::stream_deployment_status,
};
use super::{
resource_resolver::encode_resource_id_for_url,
types::{
DeployResourceRequest, DeployResourceResponse, PatchResourceRequest, ResourceConfig,
ResourceDetailResponse,
},
};
pub(crate) struct MutationRequest {
pub(crate) resource_id: String,
pub(crate) current: ResourceDetailResponse,
pub(crate) requested_config: ResourceConfig,
pub(crate) distribution_strategy: Option<String>,
pub(crate) primary_region: Option<String>,
pub(crate) snapshot_before_change: Option<bool>,
pub(crate) skip_confirm: bool,
pub(crate) dry_run: bool,
}
pub(crate) fn nothing_to_change(
config: &ResourceConfig,
distribution_strategy: &Option<String>,
primary_region: &Option<String>,
) -> bool {
is_metadata_only(config) && distribution_strategy.is_none() && primary_region.is_none()
}
fn is_metadata_only(config: &ResourceConfig) -> bool {
let ResourceConfig {
instance_class,
engine,
allocated_storage,
num_cache_nodes,
number_of_broker_nodes,
ebs_storage_size,
visibility_timeout,
message_retention_seconds,
port,
multi_az,
node_type,
broker_node_type,
kafka_version,
queue_type,
encryption,
} = config;
instance_class.is_none()
&& engine.is_none()
&& allocated_storage.is_none()
&& num_cache_nodes.is_none()
&& number_of_broker_nodes.is_none()
&& ebs_storage_size.is_none()
&& visibility_timeout.is_none()
&& message_retention_seconds.is_none()
&& port.is_none()
&& multi_az.is_none()
&& node_type.is_none()
&& broker_node_type.is_none()
&& kafka_version.is_none()
&& queue_type.is_none()
&& encryption.is_none()
}
fn print_config_diff(stdout: &mut StandardStream, current: &ResourceConfig, requested: &ResourceConfig) -> Result<()> {
writeln!(stdout, " The following will change:")?;
macro_rules! diff_field {
($label:expr, $cur:expr, $req:expr) => {
if let Some(new_value) = &$req {
let cur_display = $cur
.as_ref()
.map(|v| format!("{:?}", v))
.unwrap_or_else(|| "unset".to_string());
writeln!(
stdout,
" {:<24} {} -> {:?}",
$label,
cur_display,
new_value
)?;
}
};
}
diff_field!("instance_class:", current.instance_class, requested.instance_class);
diff_field!("engine:", current.engine, requested.engine);
diff_field!("allocated_storage:", current.allocated_storage, requested.allocated_storage);
diff_field!("num_cache_nodes:", current.num_cache_nodes, requested.num_cache_nodes);
diff_field!("number_of_broker_nodes:", current.number_of_broker_nodes, requested.number_of_broker_nodes);
diff_field!("ebs_storage_size:", current.ebs_storage_size, requested.ebs_storage_size);
diff_field!("visibility_timeout:", current.visibility_timeout, requested.visibility_timeout);
diff_field!("message_retention_seconds:", current.message_retention_seconds, requested.message_retention_seconds);
diff_field!("port:", current.port, requested.port);
diff_field!("multi_az:", current.multi_az, requested.multi_az);
diff_field!("node_type:", current.node_type, requested.node_type);
diff_field!("broker_node_type:", current.broker_node_type, requested.broker_node_type);
diff_field!("kafka_version:", current.kafka_version, requested.kafka_version);
diff_field!("queue_type:", current.queue_type, requested.queue_type);
diff_field!("encryption:", current.encryption, requested.encryption);
Ok(())
}
pub(crate) fn run_mutation(req: MutationRequest) -> Result<()> {
let mut stdout = StandardStream::stdout(ColorChoice::Always);
if is_metadata_only(&req.requested_config) {
if req.distribution_strategy.is_none() && req.primary_region.is_none() {
bail!("nothing to change — no fields were set");
}
if req.dry_run {
writeln!(stdout, "Dry run: would PATCH distributionStrategy/primaryRegion. No changes applied.")?;
return Ok(());
}
let updated = patch_resource(
&req.resource_id,
req.distribution_strategy,
req.primary_region,
)?;
writeln!(stdout, "Updated {} (status: {})", updated.name, updated.status)?;
return Ok(());
}
writeln!(stdout)?;
print_config_diff(&mut stdout, &req.current.manifest_config, &req.requested_config)?;
writeln!(stdout)?;
if req.dry_run {
writeln!(stdout, "Dry run: no changes applied.")?;
return Ok(());
}
if !req.skip_confirm {
let confirmed = Confirm::with_theme(&ColorfulTheme::default())
.with_prompt("Apply this change?")
.default(false)
.interact()
.with_context(|| "Failed to read confirmation")?;
if !confirmed {
writeln!(stdout, "Aborted — no changes applied.")?;
return Ok(());
}
}
let response = deploy_resource(
&req.resource_id,
Some(req.requested_config),
req.distribution_strategy,
req.primary_region,
req.snapshot_before_change,
)?;
let dashboard_url = format!(
"{}/dashboard/deployments/{}",
get_platform_ui_url(),
response.deployment_id
);
writeln!(stdout)?;
stdout.set_color(termcolor::ColorSpec::new().set_fg(Some(Color::Green)))?;
writeln!(stdout, "Change submitted: {}", dashboard_url)?;
stdout.reset()?;
writeln!(stdout, "(Check this URL if the live status stream below is interrupted.)")?;
writeln!(stdout)?;
stream_deployment_status(&AuthMode::Jwt, &response.deployment_id, None, None, &mut stdout)?;
Ok(())
}
fn deploy_resource(
resource_id: &str,
manifest_config: Option<ResourceConfig>,
distribution_strategy: Option<String>,
primary_region: Option<String>,
snapshot_before_change: Option<bool>,
) -> Result<DeployResourceResponse> {
let url = format!(
"{}/platform-resources/{}/deploy",
get_resource_management_api_url(),
encode_resource_id_for_url(resource_id)
);
let body = DeployResourceRequest {
manifest_config,
distribution_strategy,
primary_region,
snapshot_before_change,
};
let response = post(&url, serde_json::to_value(&body)?)
.with_context(|| "Failed to reach resource-management API")?;
let status = response.status();
if status.as_u16() == 409 {
bail!(
"another operation is already in progress for this application/environment/region — try again shortly"
);
}
if !status.is_success() {
let body_text = response
.text()
.unwrap_or_else(|_| "unknown error".to_string());
bail!("resource-management API returned {} — {}", status, body_text);
}
response
.json()
.with_context(|| "Failed to parse deploy response")
}
fn patch_resource(
resource_id: &str,
distribution_strategy: Option<String>,
primary_region: Option<String>,
) -> Result<ResourceDetailResponse> {
let url = format!(
"{}/platform-resources/{}",
get_resource_management_api_url(),
encode_resource_id_for_url(resource_id)
);
let body = PatchResourceRequest {
distribution_strategy,
primary_region,
};
let response = patch(&url, serde_json::to_value(&body)?)
.with_context(|| "Failed to reach resource-management API")?;
let status = response.status();
if !status.is_success() {
let body_text = response
.text()
.unwrap_or_else(|_| "unknown error".to_string());
bail!("resource-management API returned {} — {}", status, body_text);
}
response
.json()
.with_context(|| "Failed to parse patch response")
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn metadata_only_true_when_all_config_fields_unset() {
assert!(is_metadata_only(&ResourceConfig::default()));
}
#[test]
fn nothing_to_change_true_when_everything_unset() {
assert!(nothing_to_change(&ResourceConfig::default(), &None, &None));
}
#[test]
fn nothing_to_change_false_when_distribution_strategy_set() {
assert!(!nothing_to_change(
&ResourceConfig::default(),
&Some("centralized".to_string()),
&None
));
}
#[test]
fn metadata_only_false_when_any_config_field_set() {
let config = ResourceConfig {
instance_class: Some("db.t3.small".to_string()),
..Default::default()
};
assert!(!is_metadata_only(&config));
}
}