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));
}
}