fakecloud_cloudformation/

service.rs

use async_trait::async_trait;
use chrono::Utc;
use http::StatusCode;
use std::collections::HashMap;
use std::sync::Arc;

use fakecloud_core::delivery::DeliveryBus;
use fakecloud_core::service::{AwsRequest, AwsResponse, AwsService, AwsServiceError};
use fakecloud_dynamodb::state::SharedDynamoDbState;
use fakecloud_eventbridge::state::SharedEventBridgeState;
use fakecloud_iam::state::SharedIamState;
use fakecloud_logs::state::SharedLogsState;
use fakecloud_persistence::SnapshotStore;
use fakecloud_s3::state::SharedS3State;
use fakecloud_sns::state::SharedSnsState;
use fakecloud_sqs::state::SharedSqsState;
use fakecloud_ssm::state::SharedSsmState;
use tokio::sync::Mutex as AsyncMutex;

use crate::resource_provisioner::ResourceProvisioner;
use crate::state::{
    CloudFormationSnapshot, CloudFormationState, SharedCloudFormationState, Stack, StackResource,
    CLOUDFORMATION_SNAPSHOT_SCHEMA_VERSION,
};
use crate::template;
use crate::xml_responses;

/// Multi-pass provisioning for all resources in a parsed template.
///
/// Resources may `Ref` each other in either direction, and JSON object
/// iteration order isn't stable, so a single forward pass isn't enough
/// to resolve them. We loop: each pass tries every pending resource, and
/// any resource whose `Ref` targets are still unknown just stays pending
/// for the next pass. When no pass makes progress we report the first
/// pending failure and rollback.
fn provision_stack_resources(
    provisioner: &ResourceProvisioner,
    resource_defs: &[template::ResourceDefinition],
    template_body: &str,
    parameters: &HashMap<String, String>,
) -> Result<Vec<StackResource>, AwsServiceError> {
    let mut resources = Vec::new();
    let mut physical_ids: HashMap<String, String> = HashMap::new();
    let mut pending: Vec<&template::ResourceDefinition> = resource_defs.iter().collect();
    let max_passes = pending.len() + 1;

    for _ in 0..max_passes {
        if pending.is_empty() {
            break;
        }
        let mut still_pending = Vec::new();
        let mut made_progress = false;

        for resource_def in pending {
            let resolved_def = template::resolve_resource_properties(
                resource_def,
                template_body,
                parameters,
                &physical_ids,
            )
            .map_err(|e| {
                AwsServiceError::aws_error(StatusCode::BAD_REQUEST, "ValidationError", e)
            })?;

            match provisioner.create_resource(&resolved_def) {
                Ok(stack_resource) => {
                    physical_ids.insert(
                        stack_resource.logical_id.clone(),
                        stack_resource.physical_id.clone(),
                    );
                    resources.push(stack_resource);
                    made_progress = true;
                }
                Err(_) => still_pending.push(resource_def),
            }
        }

        pending = still_pending;
        if !made_progress && !pending.is_empty() {
            // No progress — report the first failure and rollback anything
            // we already created.
            let resource_def = pending[0];
            let resolved_def = template::resolve_resource_properties(
                resource_def,
                template_body,
                parameters,
                &physical_ids,
            )
            .unwrap_or_else(|_| resource_def.clone());
            let err = provisioner.create_resource(&resolved_def).unwrap_err();
            for r in &resources {
                let _ = provisioner.delete_resource(r);
            }
            return Err(AwsServiceError::aws_error(
                StatusCode::BAD_REQUEST,
                "ValidationError",
                format!(
                    "Failed to create resource {}: {err}",
                    resource_def.logical_id
                ),
            ));
        }
    }

    Ok(resources)
}

/// State references for every service CloudFormation can provision resources in.
pub struct CloudFormationDeps {
    pub sqs: SharedSqsState,
    pub sns: SharedSnsState,
    pub ssm: SharedSsmState,
    pub iam: SharedIamState,
    pub s3: SharedS3State,
    pub eventbridge: SharedEventBridgeState,
    pub dynamodb: SharedDynamoDbState,
    pub logs: SharedLogsState,
    pub delivery: Arc<DeliveryBus>,
}

pub struct CloudFormationService {
    state: SharedCloudFormationState,
    deps: CloudFormationDeps,
    snapshot_store: Option<Arc<dyn SnapshotStore>>,
    snapshot_lock: Arc<AsyncMutex<()>>,
}

impl CloudFormationService {
    pub fn new(state: SharedCloudFormationState, deps: CloudFormationDeps) -> Self {
        Self {
            state,
            deps,
            snapshot_store: None,
            snapshot_lock: Arc::new(AsyncMutex::new(())),
        }
    }

    pub fn with_snapshot_store(mut self, store: Arc<dyn SnapshotStore>) -> Self {
        self.snapshot_store = Some(store);
        self
    }

    async fn save_snapshot(&self) {
        let Some(store) = self.snapshot_store.clone() else {
            return;
        };
        let _guard = self.snapshot_lock.lock().await;
        let snapshot = CloudFormationSnapshot {
            schema_version: CLOUDFORMATION_SNAPSHOT_SCHEMA_VERSION,
            state: None,
            accounts: Some(self.state.read().clone()),
        };
        let join = tokio::task::spawn_blocking(move || -> std::io::Result<()> {
            let bytes = serde_json::to_vec(&snapshot)
                .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?;
            store.save(&bytes)
        })
        .await;
        match join {
            Ok(Ok(())) => {}
            Ok(Err(err)) => tracing::error!(%err, "failed to write cloudformation snapshot"),
            Err(err) => tracing::error!(%err, "cloudformation snapshot task panicked"),
        }
    }

    fn provisioner(&self, stack_id: &str, account_id: &str, region: &str) -> ResourceProvisioner {
        ResourceProvisioner {
            sqs_state: self.deps.sqs.clone(),
            sns_state: self.deps.sns.clone(),
            ssm_state: self.deps.ssm.clone(),
            iam_state: self.deps.iam.clone(),
            s3_state: self.deps.s3.clone(),
            eventbridge_state: self.deps.eventbridge.clone(),
            dynamodb_state: self.deps.dynamodb.clone(),
            logs_state: self.deps.logs.clone(),
            delivery: self.deps.delivery.clone(),
            account_id: account_id.to_string(),
            region: region.to_string(),
            stack_id: stack_id.to_string(),
        }
    }

    fn get_param(req: &AwsRequest, key: &str) -> Option<String> {
        // Check query params first (for Query protocol)
        if let Some(v) = req.query_params.get(key) {
            return Some(v.clone());
        }
        // Then check form-encoded body
        let body_params = fakecloud_core::protocol::parse_query_body(&req.body);
        body_params.get(key).cloned()
    }

    fn get_all_params(req: &AwsRequest) -> HashMap<String, String> {
        let mut params = req.query_params.clone();
        let body_params = fakecloud_core::protocol::parse_query_body(&req.body);
        for (k, v) in body_params {
            params.entry(k).or_insert(v);
        }
        params
    }

    fn extract_tags(params: &HashMap<String, String>) -> HashMap<String, String> {
        let mut tags = HashMap::new();
        for i in 1.. {
            let key_param = format!("Tags.member.{i}.Key");
            let value_param = format!("Tags.member.{i}.Value");
            match (params.get(&key_param), params.get(&value_param)) {
                (Some(k), Some(v)) => {
                    tags.insert(k.clone(), v.clone());
                }
                _ => break,
            }
        }
        tags
    }

    fn extract_parameters(params: &HashMap<String, String>) -> HashMap<String, String> {
        let mut result = HashMap::new();
        for i in 1.. {
            let key_param = format!("Parameters.member.{i}.ParameterKey");
            let value_param = format!("Parameters.member.{i}.ParameterValue");
            match (params.get(&key_param), params.get(&value_param)) {
                (Some(k), Some(v)) => {
                    result.insert(k.clone(), v.clone());
                }
                _ => break,
            }
        }
        result
    }

    fn extract_notification_arns(params: &HashMap<String, String>) -> Vec<String> {
        let mut arns = Vec::new();
        for i in 1.. {
            let key = format!("NotificationARNs.member.{i}");
            match params.get(&key) {
                Some(arn) => arns.push(arn.clone()),
                None => break,
            }
        }
        arns
    }

    fn send_stack_notification(
        delivery: &DeliveryBus,
        notification_arns: &[String],
        stack_name: &str,
        stack_id: &str,
        status: &str,
    ) {
        if notification_arns.is_empty() {
            return;
        }
        let message = format!(
            "StackId='{}'\nTimestamp='{}'\nEventId='{}'\nLogicalResourceId='{}'\nResourceStatus='{}'\nResourceType='AWS::CloudFormation::Stack'\nStackName='{}'",
            stack_id,
            chrono::Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ"),
            uuid::Uuid::new_v4(),
            stack_name,
            status,
            stack_name,
        );
        for arn in notification_arns {
            delivery.publish_to_sns(arn, &message, Some("AWS CloudFormation Notification"));
        }
    }

    fn create_stack(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
        let params = Self::get_all_params(req);

        let stack_name = params.get("StackName").ok_or_else(|| {
            AwsServiceError::aws_error(
                StatusCode::BAD_REQUEST,
                "ValidationError",
                "StackName is required",
            )
        })?;

        let template_body = params.get("TemplateBody").ok_or_else(|| {
            AwsServiceError::aws_error(
                StatusCode::BAD_REQUEST,
                "ValidationError",
                "TemplateBody is required",
            )
        })?;

        // Check if stack already exists and is not deleted
        {
            let accounts = self.state.read();
            let empty = CloudFormationState::new(&req.account_id, &req.region);
            let state = accounts.get(&req.account_id).unwrap_or(&empty);
            if let Some(existing) = state.stacks.get(stack_name.as_str()) {
                if existing.status != "DELETE_COMPLETE" {
                    return Err(AwsServiceError::aws_error(
                        StatusCode::BAD_REQUEST,
                        "AlreadyExistsException",
                        format!("Stack [{stack_name}] already exists"),
                    ));
                }
            }
        }

        let tags = Self::extract_tags(&params);
        let parameters = Self::extract_parameters(&params);
        let notification_arns = Self::extract_notification_arns(&params);

        // First pass: parse to get resource definitions (without physical ID resolution)
        let parsed = template::parse_template(template_body, &parameters).map_err(|e| {
            AwsServiceError::aws_error(StatusCode::BAD_REQUEST, "ValidationError", e)
        })?;

        let stack_id = format!(
            "arn:aws:cloudformation:{}:{}:stack/{}/{}",
            req.region,
            req.account_id,
            stack_name,
            uuid::Uuid::new_v4()
        );

        let provisioner = self.provisioner(&stack_id, &req.account_id, &req.region);
        let resources =
            provision_stack_resources(&provisioner, &parsed.resources, template_body, &parameters)?;

        let stack = Stack {
            name: stack_name.clone(),
            stack_id: stack_id.clone(),
            template: template_body.clone(),
            status: "CREATE_COMPLETE".to_string(),
            resources,
            parameters,
            tags,
            created_at: Utc::now(),
            updated_at: None,
            description: parsed.description,
            notification_arns: notification_arns.clone(),
        };

        {
            let mut accounts = self.state.write();
            let state = accounts.get_or_create(&req.account_id);
            state.stacks.insert(stack_name.clone(), stack);
        }

        Self::send_stack_notification(
            &self.deps.delivery,
            &notification_arns,
            stack_name,
            &stack_id,
            "CREATE_COMPLETE",
        );

        Ok(AwsResponse::xml(
            StatusCode::OK,
            xml_responses::create_stack_response(&stack_id, &req.request_id),
        ))
    }

    fn delete_stack(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
        let stack_name = Self::get_param(req, "StackName").ok_or_else(|| {
            AwsServiceError::aws_error(
                StatusCode::BAD_REQUEST,
                "ValidationError",
                "StackName is required",
            )
        })?;

        let mut accounts = self.state.write();
        let state = accounts.get_or_create(&req.account_id);

        // Find stack by name or stack ID
        let stack = state.stacks.values_mut().find(|s| {
            (s.name == stack_name || s.stack_id == stack_name) && s.status != "DELETE_COMPLETE"
        });

        if let Some(stack) = stack {
            let stack_id = stack.stack_id.clone();
            let stack_name_for_notif = stack.name.clone();
            let notification_arns = stack.notification_arns.clone();
            let resources: Vec<_> = stack.resources.clone();

            // Build the provisioner while we still have the stack_id
            // Drop the write lock temporarily so the provisioner can read state
            drop(accounts);
            let provisioner = self.provisioner(&stack_id, &req.account_id, &req.region);

            // Delete resources in reverse order
            for resource in resources.iter().rev() {
                let _ = provisioner.delete_resource(resource);
            }

            // Re-acquire the write lock to update stack status
            let mut accounts = self.state.write();
            let state = accounts.get_or_create(&req.account_id);
            if let Some(stack) = state.stacks.values_mut().find(|s| s.stack_id == stack_id) {
                stack.status = "DELETE_COMPLETE".to_string();
                stack.resources.clear();
            }
            drop(accounts);

            Self::send_stack_notification(
                &self.deps.delivery,
                &notification_arns,
                &stack_name_for_notif,
                &stack_id,
                "DELETE_COMPLETE",
            );
        }

        Ok(AwsResponse::xml(
            StatusCode::OK,
            xml_responses::delete_stack_response(&req.request_id),
        ))
    }

    fn describe_stacks(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
        let stack_name = Self::get_param(req, "StackName");

        let accounts = self.state.read();
        let empty = CloudFormationState::new(&req.account_id, &req.region);
        let state = accounts.get(&req.account_id).unwrap_or(&empty);
        let stacks: Vec<Stack> = if let Some(ref name) = stack_name {
            state
                .stacks
                .values()
                .filter(|s| {
                    (s.name == *name || s.stack_id == *name) && s.status != "DELETE_COMPLETE"
                })
                .cloned()
                .collect()
        } else {
            state
                .stacks
                .values()
                .filter(|s| s.status != "DELETE_COMPLETE")
                .cloned()
                .collect()
        };

        if let Some(ref name) = stack_name {
            if stacks.is_empty() {
                return Err(AwsServiceError::aws_error(
                    StatusCode::BAD_REQUEST,
                    "ValidationError",
                    format!("Stack with id {name} does not exist"),
                ));
            }
        }

        Ok(AwsResponse::xml(
            StatusCode::OK,
            xml_responses::describe_stacks_response(&stacks, &req.request_id),
        ))
    }

    fn list_stacks(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
        let accounts = self.state.read();
        let empty = CloudFormationState::new(&req.account_id, &req.region);
        let state = accounts.get(&req.account_id).unwrap_or(&empty);
        let stacks: Vec<Stack> = state.stacks.values().cloned().collect();

        Ok(AwsResponse::xml(
            StatusCode::OK,
            xml_responses::list_stacks_response(&stacks, &req.request_id),
        ))
    }

    fn list_stack_resources(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
        let stack_name = Self::get_param(req, "StackName").ok_or_else(|| {
            AwsServiceError::aws_error(
                StatusCode::BAD_REQUEST,
                "ValidationError",
                "StackName is required",
            )
        })?;

        let accounts = self.state.read();
        let empty = CloudFormationState::new(&req.account_id, &req.region);
        let state = accounts.get(&req.account_id).unwrap_or(&empty);
        let stack = state
            .stacks
            .values()
            .find(|s| {
                (s.name == stack_name || s.stack_id == stack_name) && s.status != "DELETE_COMPLETE"
            })
            .ok_or_else(|| {
                AwsServiceError::aws_error(
                    StatusCode::BAD_REQUEST,
                    "ValidationError",
                    format!("Stack [{stack_name}] does not exist"),
                )
            })?;

        Ok(AwsResponse::xml(
            StatusCode::OK,
            xml_responses::list_stack_resources_response(&stack.resources, &req.request_id),
        ))
    }

    fn describe_stack_resources(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
        let stack_name = Self::get_param(req, "StackName").ok_or_else(|| {
            AwsServiceError::aws_error(
                StatusCode::BAD_REQUEST,
                "ValidationError",
                "StackName is required",
            )
        })?;

        let accounts = self.state.read();
        let empty = CloudFormationState::new(&req.account_id, &req.region);
        let state = accounts.get(&req.account_id).unwrap_or(&empty);
        let stack = state
            .stacks
            .values()
            .find(|s| {
                (s.name == stack_name || s.stack_id == stack_name) && s.status != "DELETE_COMPLETE"
            })
            .ok_or_else(|| {
                AwsServiceError::aws_error(
                    StatusCode::BAD_REQUEST,
                    "ValidationError",
                    format!("Stack [{stack_name}] does not exist"),
                )
            })?;

        Ok(AwsResponse::xml(
            StatusCode::OK,
            xml_responses::describe_stack_resources_response(
                &stack.resources,
                &stack.name,
                &req.request_id,
            ),
        ))
    }

    fn update_stack(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
        let input = UpdateStackInput::from_params(req)?;
        let parsed =
            template::parse_template(&input.template_body, &input.parameters).map_err(|e| {
                AwsServiceError::aws_error(StatusCode::BAD_REQUEST, "ValidationError", e)
            })?;

        // Get stack_id before write lock for the provisioner
        let found_stack_id = {
            let accounts = self.state.read();
            let empty = CloudFormationState::new(&req.account_id, &req.region);
            let state = accounts.get(&req.account_id).unwrap_or(&empty);
            state
                .stacks
                .values()
                .find(|s| {
                    (s.name == input.stack_name || s.stack_id == input.stack_name)
                        && s.status != "DELETE_COMPLETE"
                })
                .map(|s| s.stack_id.clone())
                .unwrap_or_default()
        };

        let provisioner = self.provisioner(&found_stack_id, &req.account_id, &req.region);

        let mut accounts = self.state.write();
        let state = accounts.get_or_create(&req.account_id);
        let stack = state
            .stacks
            .values_mut()
            .find(|s| {
                (s.name == input.stack_name || s.stack_id == input.stack_name)
                    && s.status != "DELETE_COMPLETE"
            })
            .ok_or_else(|| {
                AwsServiceError::aws_error(
                    StatusCode::BAD_REQUEST,
                    "ValidationError",
                    format!("Stack [{}] does not exist", input.stack_name),
                )
            })?;

        let update_result = apply_resource_updates(
            stack,
            &parsed.resources,
            &input.template_body,
            &input.parameters,
            &provisioner,
        );

        let stack_id = stack.stack_id.clone();
        stack.template = input.template_body.clone();
        stack.status = if update_result.is_err() {
            "UPDATE_FAILED".to_string()
        } else {
            "UPDATE_COMPLETE".to_string()
        };
        stack.parameters = input.parameters;
        if !input.tags.is_empty() {
            stack.tags = input.tags;
        }
        stack.updated_at = Some(Utc::now());
        stack.description = parsed.description;
        if !input.notification_arns.is_empty() {
            stack.notification_arns = input.notification_arns.clone();
        }
        let notification_arns = stack.notification_arns.clone();
        let stack_name_for_notif = stack.name.clone();

        if let Err(error_msg) = update_result {
            drop(accounts);
            Self::send_stack_notification(
                &self.deps.delivery,
                &notification_arns,
                &stack_name_for_notif,
                &stack_id,
                "UPDATE_FAILED",
            );
            return Err(AwsServiceError::aws_error(
                StatusCode::BAD_REQUEST,
                "ValidationError",
                error_msg,
            ));
        }

        drop(accounts);
        Self::send_stack_notification(
            &self.deps.delivery,
            &notification_arns,
            &stack_name_for_notif,
            &stack_id,
            "UPDATE_COMPLETE",
        );

        Ok(AwsResponse::xml(
            StatusCode::OK,
            xml_responses::update_stack_response(&stack_id, &req.request_id),
        ))
    }

    fn get_template(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
        let stack_name = Self::get_param(req, "StackName").ok_or_else(|| {
            AwsServiceError::aws_error(
                StatusCode::BAD_REQUEST,
                "ValidationError",
                "StackName is required",
            )
        })?;

        let accounts = self.state.read();
        let empty = CloudFormationState::new(&req.account_id, &req.region);
        let state = accounts.get(&req.account_id).unwrap_or(&empty);
        let stack = state
            .stacks
            .values()
            .find(|s| {
                (s.name == stack_name || s.stack_id == stack_name) && s.status != "DELETE_COMPLETE"
            })
            .ok_or_else(|| {
                AwsServiceError::aws_error(
                    StatusCode::BAD_REQUEST,
                    "ValidationError",
                    format!("Stack [{stack_name}] does not exist"),
                )
            })?;

        Ok(AwsResponse::xml(
            StatusCode::OK,
            xml_responses::get_template_response(&stack.template, &req.request_id),
        ))
    }
}

#[async_trait]
impl AwsService for CloudFormationService {
    fn service_name(&self) -> &str {
        "cloudformation"
    }

    async fn handle(&self, req: AwsRequest) -> Result<AwsResponse, AwsServiceError> {
        let mutates = matches!(
            req.action.as_str(),
            "CreateStack" | "DeleteStack" | "UpdateStack"
        );
        let result = match req.action.as_str() {
            "CreateStack" => self.create_stack(&req),
            "DeleteStack" => self.delete_stack(&req),
            "DescribeStacks" => self.describe_stacks(&req),
            "ListStacks" => self.list_stacks(&req),
            "ListStackResources" => self.list_stack_resources(&req),
            "DescribeStackResources" => self.describe_stack_resources(&req),
            "UpdateStack" => self.update_stack(&req),
            "GetTemplate" => self.get_template(&req),
            _ => Err(AwsServiceError::action_not_implemented(
                "cloudformation",
                &req.action,
            )),
        };
        if mutates && matches!(result.as_ref(), Ok(resp) if resp.status.is_success()) {
            self.save_snapshot().await;
        }
        result
    }

    fn supported_actions(&self) -> &[&str] {
        &[
            "CreateStack",
            "DeleteStack",
            "DescribeStacks",
            "ListStacks",
            "ListStackResources",
            "DescribeStackResources",
            "UpdateStack",
            "GetTemplate",
        ]
    }
}

/// Parsed + validated inputs for `UpdateStack`.
struct UpdateStackInput {
    stack_name: String,
    template_body: String,
    parameters: HashMap<String, String>,
    tags: HashMap<String, String>,
    notification_arns: Vec<String>,
}

impl UpdateStackInput {
    fn from_params(req: &AwsRequest) -> Result<Self, AwsServiceError> {
        let params = CloudFormationService::get_all_params(req);

        let stack_name = params
            .get("StackName")
            .ok_or_else(|| {
                AwsServiceError::aws_error(
                    StatusCode::BAD_REQUEST,
                    "ValidationError",
                    "StackName is required",
                )
            })?
            .to_string();

        let template_body = params
            .get("TemplateBody")
            .ok_or_else(|| {
                AwsServiceError::aws_error(
                    StatusCode::BAD_REQUEST,
                    "ValidationError",
                    "TemplateBody is required",
                )
            })?
            .to_string();

        Ok(Self {
            stack_name,
            template_body,
            parameters: CloudFormationService::extract_parameters(&params),
            tags: CloudFormationService::extract_tags(&params),
            notification_arns: CloudFormationService::extract_notification_arns(&params),
        })
    }
}

/// Apply resource updates: delete removed resources, create new ones.
/// Returns Err(msg) if any resource operation fails.
fn apply_resource_updates(
    stack: &mut crate::state::Stack,
    new_resource_defs: &[template::ResourceDefinition],
    template_body: &str,
    parameters: &HashMap<String, String>,
    provisioner: &crate::resource_provisioner::ResourceProvisioner,
) -> Result<(), String> {
    let old_logical_ids: std::collections::HashSet<String> = stack
        .resources
        .iter()
        .map(|r| r.logical_id.clone())
        .collect();
    let new_logical_ids: std::collections::HashSet<String> = new_resource_defs
        .iter()
        .map(|r| r.logical_id.clone())
        .collect();

    // Delete resources no longer in template
    let to_remove: Vec<_> = stack
        .resources
        .iter()
        .filter(|r| !new_logical_ids.contains(&r.logical_id))
        .cloned()
        .collect();
    for resource in &to_remove {
        let _ = provisioner.delete_resource(resource);
    }
    stack
        .resources
        .retain(|r| new_logical_ids.contains(&r.logical_id));

    // Build physical ID map from existing resources
    let mut physical_ids: HashMap<String, String> = stack
        .resources
        .iter()
        .map(|r| (r.logical_id.clone(), r.physical_id.clone()))
        .collect();

    // Create new resources
    for resource_def in new_resource_defs {
        if !old_logical_ids.contains(&resource_def.logical_id) {
            let resolved_def = template::resolve_resource_properties(
                resource_def,
                template_body,
                parameters,
                &physical_ids,
            )
            .map_err(|e| {
                format!(
                    "Failed to resolve resource {}: {e}",
                    resource_def.logical_id
                )
            })?;

            match provisioner.create_resource(&resolved_def) {
                Ok(stack_resource) => {
                    physical_ids.insert(
                        stack_resource.logical_id.clone(),
                        stack_resource.physical_id.clone(),
                    );
                    stack.resources.push(stack_resource);
                }
                Err(e) => {
                    tracing::warn!(
                        "Failed to create resource {} during update: {e}",
                        resource_def.logical_id
                    );
                    return Err(format!(
                        "Failed to create resource {}: {e}",
                        resource_def.logical_id
                    ));
                }
            }
        }
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use http::HeaderMap;
    use parking_lot::RwLock;
    use std::sync::Arc;

    fn make_service() -> CloudFormationService {
        let cf_state = Arc::new(RwLock::new(
            fakecloud_core::multi_account::MultiAccountState::new(
                "123456789012",
                "us-east-1",
                "http://localhost:4566",
            ),
        ));
        let deps = CloudFormationDeps {
            sqs: Arc::new(RwLock::new(
                fakecloud_core::multi_account::MultiAccountState::new(
                    "123456789012",
                    "us-east-1",
                    "http://localhost:4566",
                ),
            )),
            sns: Arc::new(RwLock::new(
                fakecloud_core::multi_account::MultiAccountState::new(
                    "123456789012",
                    "us-east-1",
                    "http://localhost:4566",
                ),
            )),
            ssm: Arc::new(RwLock::new(
                fakecloud_core::multi_account::MultiAccountState::new(
                    "123456789012",
                    "us-east-1",
                    "http://localhost:4566",
                ),
            )),
            iam: Arc::new(RwLock::new(
                fakecloud_core::multi_account::MultiAccountState::new(
                    "123456789012",
                    "us-east-1",
                    "",
                ),
            )),
            s3: Arc::new(RwLock::new(
                fakecloud_core::multi_account::MultiAccountState::new(
                    "123456789012",
                    "us-east-1",
                    "",
                ),
            )),
            eventbridge: Arc::new(RwLock::new(
                fakecloud_core::multi_account::MultiAccountState::new(
                    "123456789012",
                    "us-east-1",
                    "",
                ),
            )),
            dynamodb: Arc::new(RwLock::new(
                fakecloud_core::multi_account::MultiAccountState::new(
                    "123456789012",
                    "us-east-1",
                    "",
                ),
            )),
            logs: Arc::new(RwLock::new(
                fakecloud_core::multi_account::MultiAccountState::new(
                    "123456789012",
                    "us-east-1",
                    "",
                ),
            )),
            delivery: Arc::new(DeliveryBus::new()),
        };
        CloudFormationService::new(cf_state, deps)
    }

    fn make_request(action: &str, params: HashMap<String, String>) -> AwsRequest {
        AwsRequest {
            service: "cloudformation".to_string(),
            action: action.to_string(),
            region: "us-east-1".to_string(),
            account_id: "123456789012".to_string(),
            request_id: "test-request-id".to_string(),
            headers: HeaderMap::new(),
            query_params: params,
            body: bytes::Bytes::new(),
            path_segments: vec![],
            raw_path: "/".to_string(),
            raw_query: String::new(),
            method: http::Method::POST,
            is_query_protocol: true,
            access_key_id: None,
            principal: None,
        }
    }

    #[test]
    fn update_stack_sets_failed_status_on_resource_error() {
        let svc = make_service();

        // Create a stack with just a queue
        let mut create_params = HashMap::new();
        create_params.insert("StackName".to_string(), "test-stack".to_string());
        create_params.insert(
            "TemplateBody".to_string(),
            r#"{"Resources":{"MyQueue":{"Type":"AWS::SQS::Queue","Properties":{"QueueName":"q1"}}}}"#.to_string(),
        );
        let req = make_request("CreateStack", create_params);
        let result = svc.create_stack(&req);
        assert!(result.is_ok());

        // Update stack adding an SNS subscription with a non-existent topic
        let mut update_params = HashMap::new();
        update_params.insert("StackName".to_string(), "test-stack".to_string());
        update_params.insert(
            "TemplateBody".to_string(),
            r#"{"Resources":{"MyQueue":{"Type":"AWS::SQS::Queue","Properties":{"QueueName":"q1"}},"BadSub":{"Type":"AWS::SNS::Subscription","Properties":{"TopicArn":"arn:aws:sns:us-east-1:123456789012:nope","Protocol":"sqs","Endpoint":"arn:aws:sqs:us-east-1:123456789012:q1"}}}}"#.to_string(),
        );
        let req = make_request("UpdateStack", update_params);
        let result = svc.update_stack(&req);

        // Should return an error
        assert!(result.is_err());

        // Stack status should be UPDATE_FAILED
        let accounts = svc.state.read();
        let state = accounts.get("123456789012").unwrap();
        let stack = state.stacks.get("test-stack").unwrap();
        assert_eq!(stack.status, "UPDATE_FAILED");
    }

    #[test]
    fn create_stack_resolves_ref_to_physical_id() {
        let svc = make_service();

        // Template where subscription Refs the topic
        let template = r#"{
            "Resources": {
                "MyTopic": {
                    "Type": "AWS::SNS::Topic",
                    "Properties": { "TopicName": "ref-test-topic" }
                },
                "MySub": {
                    "Type": "AWS::SNS::Subscription",
                    "Properties": {
                        "TopicArn": { "Ref": "MyTopic" },
                        "Protocol": "sqs",
                        "Endpoint": "arn:aws:sqs:us-east-1:123456789012:some-queue"
                    }
                }
            }
        }"#;

        let mut params = HashMap::new();
        params.insert("StackName".to_string(), "ref-stack".to_string());
        params.insert("TemplateBody".to_string(), template.to_string());
        let req = make_request("CreateStack", params);
        let result = svc.create_stack(&req);
        assert!(result.is_ok(), "CreateStack failed: {:?}", result.err());

        // Verify both resources were created
        let accounts = svc.state.read();
        let state = accounts.get("123456789012").unwrap();
        let stack = state.stacks.get("ref-stack").unwrap();
        assert_eq!(stack.resources.len(), 2);
        assert_eq!(stack.status, "CREATE_COMPLETE");

        // The subscription's physical ID should be an ARN (not just "MyTopic")
        let sub = stack
            .resources
            .iter()
            .find(|r| r.logical_id == "MySub")
            .unwrap();
        assert!(
            sub.physical_id.contains("ref-test-topic"),
            "Subscription physical ID should reference the topic ARN, got: {}",
            sub.physical_id
        );
    }

    // ── Service error paths ──

    #[test]
    fn create_stack_missing_name_errors() {
        let svc = make_service();
        let mut params = HashMap::new();
        params.insert("TemplateBody".to_string(), "{}".to_string());
        let req = make_request("CreateStack", params);
        assert!(svc.create_stack(&req).is_err());
    }

    #[test]
    fn create_stack_missing_template_errors() {
        let svc = make_service();
        let mut params = HashMap::new();
        params.insert("StackName".to_string(), "s".to_string());
        let req = make_request("CreateStack", params);
        assert!(svc.create_stack(&req).is_err());
    }

    #[test]
    fn create_stack_duplicate_errors() {
        let svc = make_service();
        let mut params = HashMap::new();
        params.insert("StackName".to_string(), "dup".to_string());
        params.insert(
            "TemplateBody".to_string(),
            r#"{"Resources":{"Q":{"Type":"AWS::SQS::Queue","Properties":{"QueueName":"dq"}}}}"#
                .to_string(),
        );
        let req = make_request("CreateStack", params.clone());
        svc.create_stack(&req).unwrap();
        let req = make_request("CreateStack", params);
        assert!(svc.create_stack(&req).is_err());
    }

    #[test]
    fn create_stack_invalid_template_errors() {
        let svc = make_service();
        let mut params = HashMap::new();
        params.insert("StackName".to_string(), "bad".to_string());
        params.insert("TemplateBody".to_string(), "not json".to_string());
        let req = make_request("CreateStack", params);
        assert!(svc.create_stack(&req).is_err());
    }

    #[test]
    fn delete_stack_unknown_is_noop() {
        let svc = make_service();
        let mut params = HashMap::new();
        params.insert("StackName".to_string(), "ghost".to_string());
        let req = make_request("DeleteStack", params);
        assert!(svc.delete_stack(&req).is_ok());
    }

    #[test]
    fn describe_stacks_nonexistent_errors() {
        let svc = make_service();
        let mut params = HashMap::new();
        params.insert("StackName".to_string(), "ghost".to_string());
        let req = make_request("DescribeStacks", params);
        assert!(svc.describe_stacks(&req).is_err());
    }

    #[test]
    fn describe_stacks_empty_returns_all() {
        let svc = make_service();
        let req = make_request("DescribeStacks", HashMap::new());
        let resp = svc.describe_stacks(&req).unwrap();
        let b = std::str::from_utf8(resp.body.expect_bytes()).unwrap();
        assert!(b.contains("DescribeStacksResult"));
    }

    #[test]
    fn list_stacks_empty_returns_ok() {
        let svc = make_service();
        let req = make_request("ListStacks", HashMap::new());
        let resp = svc.list_stacks(&req).unwrap();
        let b = std::str::from_utf8(resp.body.expect_bytes()).unwrap();
        assert!(b.contains("ListStacksResult"));
    }

    #[test]
    fn list_stack_resources_missing_name_errors() {
        let svc = make_service();
        let req = make_request("ListStackResources", HashMap::new());
        assert!(svc.list_stack_resources(&req).is_err());
    }

    #[test]
    fn list_stack_resources_unknown_stack_errors() {
        let svc = make_service();
        let mut params = HashMap::new();
        params.insert("StackName".to_string(), "ghost".to_string());
        let req = make_request("ListStackResources", params);
        assert!(svc.list_stack_resources(&req).is_err());
    }

    #[test]
    fn describe_stack_resources_missing_name_errors() {
        let svc = make_service();
        let req = make_request("DescribeStackResources", HashMap::new());
        assert!(svc.describe_stack_resources(&req).is_err());
    }

    #[test]
    fn get_template_missing_name_errors() {
        let svc = make_service();
        let req = make_request("GetTemplate", HashMap::new());
        assert!(svc.get_template(&req).is_err());
    }

    #[test]
    fn get_template_unknown_stack_errors() {
        let svc = make_service();
        let mut params = HashMap::new();
        params.insert("StackName".to_string(), "ghost".to_string());
        let req = make_request("GetTemplate", params);
        assert!(svc.get_template(&req).is_err());
    }

    #[test]
    fn update_stack_missing_name_errors() {
        let svc = make_service();
        let mut params = HashMap::new();
        params.insert("TemplateBody".to_string(), "{}".to_string());
        let req = make_request("UpdateStack", params);
        assert!(svc.update_stack(&req).is_err());
    }

    #[test]
    fn update_stack_unknown_stack_errors() {
        let svc = make_service();
        let mut params = HashMap::new();
        params.insert("StackName".to_string(), "ghost".to_string());
        params.insert(
            "TemplateBody".to_string(),
            r#"{"Resources":{}}"#.to_string(),
        );
        let req = make_request("UpdateStack", params);
        assert!(svc.update_stack(&req).is_err());
    }
}