tryaudex_core/

cleanup.rs

//! Phase 3: delete resources that `--tag-session` stamped during a session.
//!
//! **AWS-only**: Discovery uses the AWS Resource Groups Tagging API, and
//! delete commands shell out to the `aws` CLI. GCP and Azure sessions do not
//! yet support `--ephemeral` / `tryaudex cleanup`. Attempting to clean up a
//! GCP or Azure session will return an error with guidance.
//!
//! Credentials: this module shells out to the user's `aws` CLI, which picks
//! up ambient credentials (AWS_PROFILE, env vars, instance role, etc.).
//! Cleanup runs OUTSIDE the expired session — by definition, the short-lived
//! STS creds are already gone by the time the user asks to clean up.

use crate::error::{AvError, Result};
use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
use std::process::Command;

/// A single tagged resource returned by Resource Groups Tagging API.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct TaggedResource {
    /// Full ARN as reported by AWS.
    pub arn: String,
    /// Parsed service name ("s3", "dynamodb", ...).
    pub service: String,
    /// Parsed resource type within the service ("bucket", "table", ...).
    pub resource_type: String,
    /// Human-friendly name/identifier extracted from the ARN.
    pub name: String,
}

impl TaggedResource {
    /// Parse an ARN into its (service, type, name) parts.
    ///
    /// ARN formats we handle:
    /// - arn:aws:s3:::bucket-name
    /// - arn:aws:dynamodb:us-east-1:123:table/TableName
    /// - arn:aws:sqs:us-east-1:123:queue-name
    /// - arn:aws:sns:us-east-1:123:topic-name
    /// - arn:aws:lambda:us-east-1:123:function:FuncName
    /// - arn:aws:iam::123:role/RoleName
    /// - arn:aws:secretsmanager:us-east-1:123:secret:name-AbCdEf
    /// - arn:aws:logs:us-east-1:123:log-group:group-name
    /// - arn:aws:ecr:us-east-1:123:repository/repo
    /// - arn:aws:kms:us-east-1:123:key/uuid
    /// - arn:aws:cloudformation:us-east-1:123:stack/name/uuid
    /// - arn:aws:rds:us-east-1:123:db:instance-name
    pub fn from_arn(arn: &str) -> Option<Self> {
        // Split into: ["arn", "aws", service, region, account, resource...]
        let parts: Vec<&str> = arn.splitn(6, ':').collect();
        if parts.len() < 6 || parts[0] != "arn" {
            return None;
        }
        let service = parts[2].to_string();
        let resource_part = parts[5];

        // R6-M1: Split on whichever separator ('/' or ':') appears FIRST.
        // Previously `/` was preferred unconditionally, so a log-group ARN
        // like `log-group:groupname/substream` produced type=
        // `log-group:groupname`, name=`substream` — losing the type. And
        // Lambda function/alias ARNs or Secrets Manager secret ARNs could
        // also drop type information. Preserve the full resource portion
        // after the first separator by using whichever delimiter appears
        // earliest in `resource_part`.
        let slash_pos = resource_part.find('/');
        let colon_pos = resource_part.find(':');
        let separator = match (slash_pos, colon_pos) {
            (Some(s), Some(c)) if s < c => Some(('/', s)),
            (Some(_), Some(c)) => Some((':', c)),
            (Some(s), None) => Some(('/', s)),
            (None, Some(c)) => Some((':', c)),
            (None, None) => None,
        };
        let (resource_type, name) = if let Some((_, idx)) = separator {
            let (t, rest) = resource_part.split_at(idx);
            // `rest` begins with the separator byte; skip it.
            (t.to_string(), rest[1..].to_string())
        } else {
            // S3 bucket ARNs have no type prefix — the whole thing is the name.
            let default_type = match service.as_str() {
                "s3" => "bucket",
                "sqs" => "queue",
                "sns" => "topic",
                _ => "resource",
            };
            (default_type.to_string(), resource_part.to_string())
        };

        Some(TaggedResource {
            arn: arn.to_string(),
            service,
            resource_type,
            name,
        })
    }
}

/// Query Resource Groups Tagging API for every resource bearing
/// `tryaudex-session=<session_id>`. Returns the raw list (may be empty).
///
/// R6-M8: the tagging API is *regional*. A single call only returns
/// resources in the AWS CLI's default region, so sessions that created
/// resources in multiple regions leak everything outside the default.
/// Callers can set `AUDEX_CLEANUP_REGIONS=us-east-1,eu-west-1,…` to
/// scan an explicit list; the results are merged and deduplicated by
/// ARN. Without the env var, we fall back to the ambient default region
/// for backwards compatibility.
pub fn discover(session_id: &str) -> Result<Vec<TaggedResource>> {
    // R6-M2: reject comma/quote/shell-metachar before interpolating into
    // the `Key=…,Values=…` filter expression that goes to the `aws` CLI.
    crate::validate::session_id(session_id)?;

    let regions: Vec<Option<String>> = match std::env::var("AUDEX_CLEANUP_REGIONS") {
        Ok(csv) if !csv.trim().is_empty() => csv
            .split(',')
            .map(|r| r.trim().to_string())
            .filter(|r| !r.is_empty())
            .map(Some)
            .collect(),
        _ => vec![None],
    };

    let mut seen = std::collections::HashSet::new();
    let mut all_resources = Vec::new();
    for region in &regions {
        let found = discover_in_region(session_id, region.as_deref())?;
        for r in found {
            if seen.insert(r.arn.clone()) {
                all_resources.push(r);
            }
        }
    }
    Ok(all_resources)
}

/// Query the tagging API in a single region.
fn discover_in_region(session_id: &str, region: Option<&str>) -> Result<Vec<TaggedResource>> {
    let filter = format!("Key=tryaudex-session,Values={session_id}");

    #[derive(Deserialize)]
    struct Response {
        #[serde(rename = "ResourceTagMappingList", default)]
        mappings: Vec<Mapping>,
        #[serde(rename = "PaginationToken", default)]
        pagination_token: Option<String>,
    }
    #[derive(Deserialize)]
    struct Mapping {
        #[serde(rename = "ResourceARN")]
        arn: String,
    }

    let mut all_resources = Vec::new();
    let mut pagination_token: Option<String> = None;

    loop {
        let mut cmd = Command::new("aws");
        if let Some(r) = region {
            cmd.args(["--region", r]);
        }
        cmd.args([
            "resourcegroupstaggingapi",
            "get-resources",
            "--tag-filters",
            &filter,
            "--output",
            "json",
        ]);
        if let Some(ref token) = pagination_token {
            cmd.args(["--pagination-token", token]);
        }

        let output = cmd.output().map_err(AvError::Io)?;

        if !output.status.success() {
            return Err(AvError::InvalidPolicy(format!(
                "aws resourcegroupstaggingapi failed: {}",
                String::from_utf8_lossy(&output.stderr).trim()
            )));
        }

        let parsed: Response = serde_json::from_slice(&output.stdout).map_err(|e| {
            AvError::InvalidPolicy(format!("Failed to parse tagging API response: {e}"))
        })?;

        all_resources.extend(
            parsed
                .mappings
                .into_iter()
                .filter_map(|m| TaggedResource::from_arn(&m.arn)),
        );

        match parsed.pagination_token {
            Some(ref t) if !t.is_empty() => pagination_token = Some(t.clone()),
            _ => break,
        }
    }

    Ok(all_resources)
}

/// The outcome of attempting to delete one resource.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DeleteOutcome {
    Deleted,
    DryRun,
    Unsupported,
    Failed(String),
}

/// Extract the region from an ARN (arn:partition:service:region:account:resource).
fn region_from_arn(arn: &str) -> Option<&str> {
    let parts: Vec<&str> = arn.split(':').collect();
    if parts.len() >= 4 && !parts[3].is_empty() {
        Some(parts[3])
    } else {
        None
    }
}

/// Build the aws-CLI argv to delete this resource. None if unsupported.
/// Includes `--region` extracted from the resource ARN so deletes target
/// the correct region regardless of the user's default.
pub fn delete_command(r: &TaggedResource) -> Option<Vec<String>> {
    let s = |s: &str| s.to_string();
    let mut cmd = match (r.service.as_str(), r.resource_type.as_str()) {
        ("s3", "bucket") => Some(vec![
            s("aws"),
            s("s3"),
            s("rb"),
            format!("s3://{}", r.name),
            // Intentionally no --force: non-empty buckets fail safely.
            // Use `audex cleanup --force` to override.
        ]),
        ("dynamodb", "table") => Some(vec![
            s("aws"),
            s("dynamodb"),
            s("delete-table"),
            s("--table-name"),
            r.name.clone(),
        ]),
        ("sqs", "queue") => Some(vec![
            s("aws"),
            s("sqs"),
            s("delete-queue"),
            s("--queue-url"),
            // SQS delete-queue needs the URL, not the name.
            // We rebuild it from the ARN — arn:aws:sqs:region:account:name.
            sqs_url_from_arn(&r.arn)?,
        ]),
        ("sns", "topic") => Some(vec![
            s("aws"),
            s("sns"),
            s("delete-topic"),
            s("--topic-arn"),
            r.arn.clone(),
        ]),
        ("lambda", "function") => Some(vec![
            s("aws"),
            s("lambda"),
            s("delete-function"),
            s("--function-name"),
            r.name.clone(),
        ]),
        ("rds", "db") => {
            // Create a final snapshot before deletion to prevent data loss.
            // Snapshot ID must be <= 255 chars, alphanumeric + hyphens.
            let snap_id = format!("audex-cleanup-{}", r.name);
            let snap_id: String = snap_id
                .chars()
                .map(|c| {
                    if c.is_ascii_alphanumeric() || c == '-' {
                        c
                    } else {
                        '-'
                    }
                })
                .take(255)
                .collect();
            Some(vec![
                s("aws"),
                s("rds"),
                s("delete-db-instance"),
                s("--db-instance-identifier"),
                r.name.clone(),
                s("--final-db-snapshot-identifier"),
                snap_id,
            ])
        }
        ("iam", "role") => Some(vec![
            s("aws"),
            s("iam"),
            s("delete-role"),
            s("--role-name"),
            r.name.clone(),
        ]),
        ("iam", "user") => Some(vec![
            s("aws"),
            s("iam"),
            s("delete-user"),
            s("--user-name"),
            r.name.clone(),
        ]),
        ("iam", "policy") => Some(vec![
            s("aws"),
            s("iam"),
            s("delete-policy"),
            s("--policy-arn"),
            r.arn.clone(),
        ]),
        ("secretsmanager", "secret") => Some(vec![
            s("aws"),
            s("secretsmanager"),
            s("delete-secret"),
            s("--secret-id"),
            r.arn.clone(),
            // No --force-delete-without-recovery: uses default 30-day
            // recovery window so accidentally tagged secrets can be restored.
        ]),
        ("ssm", "parameter") => Some(vec![
            s("aws"),
            s("ssm"),
            s("delete-parameter"),
            s("--name"),
            r.name.clone(),
        ]),
        ("logs", "log-group") => Some(vec![
            s("aws"),
            s("logs"),
            s("delete-log-group"),
            s("--log-group-name"),
            r.name.clone(),
        ]),
        ("cloudformation", "stack") => Some(vec![
            s("aws"),
            s("cloudformation"),
            s("delete-stack"),
            s("--stack-name"),
            r.name.clone(),
        ]),
        ("ecr", "repository") => Some(vec![
            s("aws"),
            s("ecr"),
            s("delete-repository"),
            s("--repository-name"),
            r.name.clone(),
            s("--force"),
        ]),
        ("kms", "key") => Some(vec![
            s("aws"),
            s("kms"),
            s("schedule-key-deletion"),
            s("--key-id"),
            r.name.clone(),
            s("--pending-window-in-days"),
            s("7"),
        ]),
        // EC2 resources — tagged by tag_inject.rs but were previously missing from cleanup
        ("ec2", "instance") => Some(vec![
            s("aws"),
            s("ec2"),
            s("terminate-instances"),
            s("--instance-ids"),
            r.name.clone(),
        ]),
        ("ec2", "volume") => Some(vec![
            s("aws"),
            s("ec2"),
            s("delete-volume"),
            s("--volume-id"),
            r.name.clone(),
        ]),
        ("ec2", "security-group") => Some(vec![
            s("aws"),
            s("ec2"),
            s("delete-security-group"),
            s("--group-id"),
            r.name.clone(),
        ]),
        ("ec2", "vpc") => Some(vec![
            s("aws"),
            s("ec2"),
            s("delete-vpc"),
            s("--vpc-id"),
            r.name.clone(),
        ]),
        ("ec2", "subnet") => Some(vec![
            s("aws"),
            s("ec2"),
            s("delete-subnet"),
            s("--subnet-id"),
            r.name.clone(),
        ]),
        ("ec2", "internet-gateway") => Some(vec![
            s("aws"),
            s("ec2"),
            s("delete-internet-gateway"),
            s("--internet-gateway-id"),
            r.name.clone(),
        ]),
        ("ec2", "natgateway") => Some(vec![
            s("aws"),
            s("ec2"),
            s("delete-nat-gateway"),
            s("--nat-gateway-id"),
            r.name.clone(),
        ]),
        ("ec2", "route-table") => Some(vec![
            s("aws"),
            s("ec2"),
            s("delete-route-table"),
            s("--route-table-id"),
            r.name.clone(),
        ]),
        ("ec2", "key-pair") => Some(vec![
            s("aws"),
            s("ec2"),
            s("delete-key-pair"),
            s("--key-pair-id"),
            r.name.clone(),
        ]),
        ("ec2", "snapshot") => Some(vec![
            s("aws"),
            s("ec2"),
            s("delete-snapshot"),
            s("--snapshot-id"),
            r.name.clone(),
        ]),
        ("ec2", "image") => Some(vec![
            s("aws"),
            s("ec2"),
            s("deregister-image"),
            s("--image-id"),
            r.name.clone(),
        ]),
        _ => None,
    }?;

    // Append --region from the ARN so deletes target the correct region.
    // IAM is global (no region in ARN), so skip for IAM resources.
    if r.service != "iam" {
        if let Some(region) = region_from_arn(&r.arn) {
            cmd.push("--region".to_string());
            cmd.push(region.to_string());
        }
    }

    Some(cmd)
}

/// Reconstruct the SQS queue URL from its ARN.
///
/// # Limitation (R3-L9)
/// This reconstruction assumes the standard regional SQS endpoint format
/// (`https://sqs.{region}.amazonaws.com/{account}/{name}`). It will produce
/// incorrect URLs for FIPS endpoints (`sqs-fips.{region}.amazonaws.com`) and
/// VPC interface endpoints, which use custom DNS names that cannot be derived
/// from the ARN alone. A complete fix would require an extra
/// `aws sqs get-queue-url --queue-name {name} --queue-owner-aws-account-id {account}`
/// call; this function is retained as a best-effort fallback for standard endpoints.
fn sqs_url_from_arn(arn: &str) -> Option<String> {
    // arn:aws:sqs:us-east-1:123456789012:queue-name
    let parts: Vec<&str> = arn.split(':').collect();
    if parts.len() < 6 || parts[2] != "sqs" {
        return None;
    }
    let region = parts[3];
    let account = parts[4];
    let name = parts[5];
    // Detect partition from the ARN for correct endpoint suffix
    let partition = if parts.len() > 1 { parts[1] } else { "aws" };
    let suffix = match partition {
        "aws-cn" => "amazonaws.com.cn",
        "aws-us-gov" => "amazonaws.com",
        "aws" => "amazonaws.com",
        _ => "amazonaws.com",
    };
    Some(format!("https://sqs.{region}.{suffix}/{account}/{name}"))
}

/// Attempt to delete a single resource. `dry_run=true` skips the actual call.
///
/// For IAM roles we first detach attached policies + delete inline policies,
/// since AWS rejects `delete-role` if anything is still attached. Without
/// this pre-step users hit cryptic "cannot delete entity because it is
/// currently attached to 1 entities" errors and have to drop into the
/// AWS console.
pub fn delete(r: &TaggedResource, dry_run: bool) -> DeleteOutcome {
    let argv = match delete_command(r) {
        Some(a) => a,
        None => return DeleteOutcome::Unsupported,
    };
    if dry_run {
        return DeleteOutcome::DryRun;
    }
    if r.service == "iam" && r.resource_type == "role" {
        if let Err(msg) = detach_iam_role_policies(&r.name) {
            // Don't hard-fail — try delete-role anyway; if the role had
            // nothing attached the pre-step failure is irrelevant.
            tracing::warn!(role = %r.name, error = %msg, "role policy pre-detach failed");
        }
    }
    // CloudFormation: check termination protection before attempting delete.
    // Stacks with protection enabled will silently fail otherwise, and may
    // also contain nested resources that would be destroyed.
    if r.service == "cloudformation" && r.resource_type == "stack" {
        if let Ok(output) = Command::new("aws")
            .args(["cloudformation", "describe-stacks", "--stack-name", &r.name])
            .output()
        {
            // Parse the JSON response rather than relying on brittle string matching,
            // which can produce false negatives due to whitespace or field ordering.
            #[derive(serde::Deserialize)]
            struct DescribeStacksResponse {
                #[serde(rename = "Stacks")]
                stacks: Vec<StackEntry>,
            }
            #[derive(serde::Deserialize)]
            struct StackEntry {
                #[serde(rename = "EnableTerminationProtection", default)]
                enable_termination_protection: bool,
            }
            let protected = serde_json::from_slice::<DescribeStacksResponse>(&output.stdout)
                .ok()
                .and_then(|r| r.stacks.into_iter().next())
                .map(|s| s.enable_termination_protection)
                .unwrap_or(false);
            if protected {
                return DeleteOutcome::Failed(format!(
                    "Stack '{}' has termination protection enabled. \
                     Disable it first with: aws cloudformation update-termination-protection \
                     --no-enable-termination-protection --stack-name {}",
                    r.name, r.name
                ));
            }
        }
    }
    // EC2: check disableApiTermination before attempting terminate-instances.
    // Unlike CloudFormation, the EC2 API does not return a helpful error when
    // termination protection is enabled — it silently succeeds (exit 0) but
    // leaves the instance running, making the failure invisible to callers.
    if r.service == "ec2" && r.resource_type == "instance" {
        if let Ok(output) = Command::new("aws")
            .args([
                "ec2",
                "describe-instance-attribute",
                "--instance-id",
                &r.name,
                "--attribute",
                "disableApiTermination",
            ])
            .output()
        {
            #[derive(serde::Deserialize)]
            struct DisableApiTermination {
                #[serde(rename = "Value", default)]
                value: bool,
            }
            #[derive(serde::Deserialize)]
            struct DescribeInstanceAttributeResponse {
                #[serde(rename = "DisableApiTermination")]
                disable_api_termination: DisableApiTermination,
            }
            let protected =
                serde_json::from_slice::<DescribeInstanceAttributeResponse>(&output.stdout)
                    .ok()
                    .map(|r| r.disable_api_termination.value)
                    .unwrap_or(false);
            if protected {
                tracing::warn!(
                    instance_id = %r.name,
                    "EC2 instance has termination protection enabled; skipping termination. \
                     Disable it first with: aws ec2 modify-instance-attribute \
                     --instance-id {} --no-disable-api-termination",
                    r.name
                );
                return DeleteOutcome::Failed(format!(
                    "Instance '{}' has termination protection enabled. \
                     Disable it first with: aws ec2 modify-instance-attribute \
                     --instance-id {} --no-disable-api-termination",
                    r.name, r.name
                ));
            }
        }
    }
    let output = match Command::new(&argv[0]).args(&argv[1..]).output() {
        Ok(o) => o,
        Err(e) => return DeleteOutcome::Failed(e.to_string()),
    };
    if output.status.success() {
        DeleteOutcome::Deleted
    } else {
        DeleteOutcome::Failed(String::from_utf8_lossy(&output.stderr).trim().to_string())
    }
}

/// For an IAM role, list+detach all managed policies and list+delete all
/// inline policies so `delete-role` can succeed. Best-effort: partial
/// failures return the first error but keep going.
fn detach_iam_role_policies(role_name: &str) -> std::result::Result<(), String> {
    // Managed policies: list-attached-role-policies → detach each (paginated).
    {
        #[derive(Deserialize)]
        struct Attached {
            #[serde(rename = "AttachedPolicies", default)]
            policies: Vec<AttachedPolicy>,
            #[serde(rename = "Marker")]
            marker: Option<String>,
        }
        #[derive(Deserialize)]
        struct AttachedPolicy {
            #[serde(rename = "PolicyArn")]
            arn: String,
        }
        let mut marker: Option<String> = None;
        loop {
            let mut args = vec![
                "iam",
                "list-attached-role-policies",
                "--role-name",
                role_name,
                "--output",
                "json",
            ];
            let marker_val;
            if let Some(ref m) = marker {
                marker_val = m.clone();
                args.push("--marker");
                args.push(&marker_val);
            }
            let output = Command::new("aws")
                .args(&args)
                .output()
                .map_err(|e| e.to_string())?;
            if !output.status.success() {
                break;
            }
            let parsed: Attached = match serde_json::from_slice(&output.stdout) {
                Ok(p) => p,
                Err(_) => break,
            };
            for p in &parsed.policies {
                let _ = Command::new("aws")
                    .args([
                        "iam",
                        "detach-role-policy",
                        "--role-name",
                        role_name,
                        "--policy-arn",
                        &p.arn,
                    ])
                    .output();
            }
            match parsed.marker {
                Some(m) if !m.is_empty() => marker = Some(m),
                _ => break,
            }
        }
    }

    // Inline policies: list-role-policies → delete each (paginated).
    {
        #[derive(Deserialize)]
        struct Inline {
            #[serde(rename = "PolicyNames", default)]
            names: Vec<String>,
            #[serde(rename = "Marker")]
            marker: Option<String>,
        }
        let mut marker: Option<String> = None;
        loop {
            let mut args = vec![
                "iam",
                "list-role-policies",
                "--role-name",
                role_name,
                "--output",
                "json",
            ];
            let marker_val;
            if let Some(ref m) = marker {
                marker_val = m.clone();
                args.push("--marker");
                args.push(&marker_val);
            }
            let output = Command::new("aws")
                .args(&args)
                .output()
                .map_err(|e| e.to_string())?;
            if !output.status.success() {
                break;
            }
            let parsed: Inline = match serde_json::from_slice(&output.stdout) {
                Ok(p) => p,
                Err(_) => break,
            };
            for name in &parsed.names {
                let _ = Command::new("aws")
                    .args([
                        "iam",
                        "delete-role-policy",
                        "--role-name",
                        role_name,
                        "--policy-name",
                        name,
                    ])
                    .output();
            }
            match parsed.marker {
                Some(m) if !m.is_empty() => marker = Some(m),
                _ => break,
            }
        }
    }

    Ok(())
}

/// Delete priority for dependency ordering. Lower tiers are leaf resources
/// (nothing else in our set refers to them) and get deleted first. Higher
/// tiers sit upstream and must be torn down only after their dependents.
///
/// IAM policies are the main upstream case: `aws iam delete-policy` refuses
/// if any role/user/group still holds a reference. So we delete roles
/// (tier 1, which auto-detaches policies in `delete()`'s pre-step) before
/// policies (tier 2). Every other resource is a leaf at tier 0.
pub fn delete_tier(r: &TaggedResource) -> u8 {
    match (r.service.as_str(), r.resource_type.as_str()) {
        ("iam", "policy") => 2,
        ("iam", "role") => 1,
        _ => 0,
    }
}

/// Stable-sort resources into delete order. Same-tier ordering is
/// preserved from the caller's input (discovery order).
pub fn sort_for_deletion(resources: &mut [TaggedResource]) {
    resources.sort_by_key(delete_tier);
}

/// Rough daily-cost hint for a single tagged resource, used in the pre-delete
/// preview so users can see "oh wait, that's a $40/day RDS instance".
///
/// These are intentionally conservative *floor* estimates based on the
/// cheapest common SKU (e.g. RDS t3.micro, EC2 t3.micro, KMS single-region).
/// Storage- and request-driven services (S3, DynamoDB, CloudWatch Logs)
/// report usage-dependent because we can't see the bytes from the tag API
/// alone — the number we show is a rock-bottom floor assuming idle.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct DailyCostHint {
    /// Conservative lower bound, USD/day.
    pub usd_per_day: f64,
    /// True when actual cost scales with size or traffic beyond this floor.
    pub usage_dependent: bool,
}

/// Look up a daily cost hint for the given resource, or None if it's free
/// / unknown. Prices are approximate us-east-1 list prices circa 2025 —
/// precise to within "right order of magnitude", nothing more.
pub fn estimate_daily_cost(r: &TaggedResource) -> Option<DailyCostHint> {
    let hint = match (r.service.as_str(), r.resource_type.as_str()) {
        // RDS on-demand, t3.micro single-AZ: ~$0.017/hr = $0.41/day floor.
        // Anything bigger or multi-AZ is dramatically more.
        ("rds", "db") => DailyCostHint {
            usd_per_day: 0.41,
            usage_dependent: true,
        },
        // EC2 t3.micro on-demand: $0.0104/hr = $0.25/day. Plus EBS.
        ("ec2", "instance") => DailyCostHint {
            usd_per_day: 0.25,
            usage_dependent: true,
        },
        // KMS customer-managed key: $1/month = $0.033/day flat.
        ("kms", "key") => DailyCostHint {
            usd_per_day: 0.033,
            usage_dependent: false,
        },
        // Secrets Manager: $0.40/secret/month = $0.013/day.
        ("secretsmanager", "secret") => DailyCostHint {
            usd_per_day: 0.013,
            usage_dependent: false,
        },
        // S3 bucket: cost scales with stored bytes. Storage is $0.023/GB/mo
        // = $0.00077/GB/day. Show the floor as ~free and flag usage-dependent.
        ("s3", "bucket") => DailyCostHint {
            usd_per_day: 0.0,
            usage_dependent: true,
        },
        // DynamoDB: on-demand is pay-per-request, provisioned has a base.
        // Idle on-demand is effectively free but storage is $0.25/GB/mo.
        ("dynamodb", "table") => DailyCostHint {
            usd_per_day: 0.0,
            usage_dependent: true,
        },
        // CloudWatch Logs: ingest $0.50/GB, storage $0.03/GB/mo.
        ("logs", "log-group") => DailyCostHint {
            usd_per_day: 0.0,
            usage_dependent: true,
        },
        // ECR: $0.10/GB/mo storage + data transfer.
        ("ecr", "repository") => DailyCostHint {
            usd_per_day: 0.0,
            usage_dependent: true,
        },
        // Free/idle-free services: IAM, Lambda (idle), SQS/SNS (idle),
        // CloudFormation, SSM parameters (standard tier).
        ("iam", _)
        | ("lambda", _)
        | ("sqs", _)
        | ("sns", _)
        | ("cloudformation", _)
        | ("ssm", _) => return None,
        _ => return None,
    };
    Some(hint)
}

/// Sum the known daily-cost floors across a set of resources.
/// Returns (total_usd_per_day, any_usage_dependent).
pub fn estimate_daily_cost_total(resources: &[TaggedResource]) -> (f64, bool) {
    let mut total = 0.0;
    let mut any_usage = false;
    for r in resources {
        if let Some(h) = estimate_daily_cost(r) {
            total += h.usd_per_day;
            any_usage |= h.usage_dependent;
        }
    }
    (total, any_usage)
}

/// One stale-session entry: the session already ended (expired / completed
/// / failed / revoked) but still has resources tagged `tryaudex-session`
/// that are alive in the account.
#[derive(Debug, Clone)]
pub struct OrphanedSession {
    pub session_id: String,
    pub status: String,
    pub ended_at: chrono::DateTime<chrono::Utc>,
    pub resources: Vec<TaggedResource>,
    pub daily_cost: f64,
    pub usage_dependent: bool,
}

/// Cross-reference the local session store with the tagging API to find
/// every non-active session that still has billing resources. Each entry
/// can be drained with `tryaudex cleanup <session_id>`.
///
/// Hits the AWS tagging API once per non-active session in the input;
/// callers should scope the input list (recent N, filter by status) when
/// the store grows. Discovery errors on individual sessions are swallowed
/// — one failing session shouldn't block orphan reporting for the rest.
pub fn find_orphans(sessions: &[crate::session::Session]) -> Vec<OrphanedSession> {
    use crate::session::SessionStatus;
    let mut out = Vec::new();
    for s in sessions {
        if s.status == SessionStatus::Active {
            continue;
        }
        let Ok(resources) = discover(&s.id) else {
            continue;
        };
        if resources.is_empty() {
            continue;
        }
        let (daily_cost, usage_dependent) = estimate_daily_cost_total(&resources);
        out.push(OrphanedSession {
            session_id: s.id.clone(),
            status: s.status.to_string(),
            ended_at: s.expires_at,
            resources,
            daily_cost,
            usage_dependent,
        });
    }
    // Sort by cost descending so the expensive orphans surface first.
    out.sort_by(|a, b| {
        b.daily_cost
            .partial_cmp(&a.daily_cost)
            .unwrap_or(std::cmp::Ordering::Equal)
    });
    out
}

/// Summary of a cleanup run.
#[derive(Debug, Clone, Default)]
pub struct CleanupReport {
    pub deleted: Vec<TaggedResource>,
    pub failed: Vec<(TaggedResource, String)>,
    pub unsupported: Vec<TaggedResource>,
    pub dry_run: bool,
}

/// Run `delete()` over every discovered resource and aggregate results.
///
/// Currently AWS-only. Returns an error for GCP/Azure sessions.
pub fn cleanup_session(session_id: &str, dry_run: bool) -> Result<CleanupReport> {
    // Check if session is non-AWS and bail with a clear message
    let session_store = crate::session::SessionStore::new()?;
    if let Ok(session) = session_store.load(session_id) {
        match session.provider {
            crate::session::CloudProvider::Gcp => {
                return Err(AvError::InvalidPolicy(
                    "Cleanup is not yet supported for GCP sessions. \
                     Use `gcloud` CLI to manage GCP resources directly."
                        .to_string(),
                ));
            }
            crate::session::CloudProvider::Azure => {
                return Err(AvError::InvalidPolicy(
                    "Cleanup is not yet supported for Azure sessions. \
                     Use `az` CLI to manage Azure resources directly."
                        .to_string(),
                ));
            }
            crate::session::CloudProvider::Aws => {} // proceed
        }
    }

    let mut resources = discover(session_id)?;
    sort_for_deletion(&mut resources);
    let mut report = CleanupReport {
        dry_run,
        ..Default::default()
    };
    for r in resources {
        match delete(&r, dry_run) {
            DeleteOutcome::Deleted => report.deleted.push(r),
            DeleteOutcome::DryRun => report.deleted.push(r),
            DeleteOutcome::Unsupported => report.unsupported.push(r),
            DeleteOutcome::Failed(err) => report.failed.push((r, err)),
        }
    }
    Ok(report)
}

// --- Partial-cleanup state persistence ---
//
// Cleanup can crash mid-run (network blip, AWS throttle, user Ctrl-C). When
// that happens we need to know which resources were already deleted on the
// next invocation, otherwise the user either wastes API calls re-issuing
// deletes or, worse, gets misleading "resource not found" errors for things
// we handled last time. State files on disk give us that idempotency —
// re-running `tryaudex cleanup <id>` picks up from where we left off.

/// On-disk record of an in-flight cleanup. Written before any deletion
/// happens, updated after each attempt, removed once nothing is left.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct CleanupState {
    pub session_id: String,
    pub started_at: chrono::DateTime<chrono::Utc>,
    /// ARNs that are known to be deleted (successful outcomes from prior runs).
    pub deleted_arns: Vec<String>,
    /// Resources whose delete call returned an error — retry candidates.
    pub failed: Vec<(TaggedResource, String)>,
}

impl CleanupState {
    fn new(session_id: &str) -> Self {
        Self {
            session_id: session_id.to_string(),
            started_at: chrono::Utc::now(),
            deleted_arns: Vec::new(),
            failed: Vec::new(),
        }
    }

    pub fn is_deleted(&self, arn: &str) -> bool {
        self.deleted_arns.iter().any(|a| a == arn)
    }

    pub fn record_deleted(&mut self, arn: &str) {
        if !self.is_deleted(arn) {
            self.deleted_arns.push(arn.to_string());
        }
        // If the resource was previously failed, clear it from that list.
        self.failed.retain(|(r, _)| r.arn != arn);
    }

    pub fn record_failed(&mut self, resource: &TaggedResource, reason: &str) {
        self.failed.retain(|(r, _)| r.arn != resource.arn);
        self.failed.push((resource.clone(), reason.to_string()));
    }
}

/// Manages cleanup state files keyed by session id.
pub struct CleanupStateStore {
    dir: PathBuf,
}

impl CleanupStateStore {
    pub fn new() -> Result<Self> {
        let base = dirs::data_local_dir().ok_or_else(|| {
            AvError::InvalidPolicy(
                "Could not determine local data directory. Set XDG_DATA_HOME or HOME.".to_string(),
            )
        })?;
        let dir = base.join("audex").join("cleanup_state");
        std::fs::create_dir_all(&dir)?;
        Ok(Self { dir })
    }

    pub fn with_dir(dir: impl AsRef<Path>) -> Result<Self> {
        let dir = dir.as_ref().to_path_buf();
        std::fs::create_dir_all(&dir)?;
        Ok(Self { dir })
    }

    fn path_for(&self, session_id: &str) -> Result<PathBuf> {
        crate::validate::session_id(session_id)?;
        Ok(self.dir.join(format!("{session_id}.json")))
    }

    /// Load state for a session, or return a freshly-initialized state if
    /// none exists yet.
    pub fn load_or_new(&self, session_id: &str) -> Result<CleanupState> {
        let path = self.path_for(session_id)?;
        if !path.exists() {
            return Ok(CleanupState::new(session_id));
        }
        let json = std::fs::read_to_string(&path)?;
        serde_json::from_str(&json).map_err(|e| {
            AvError::InvalidPolicy(format!("corrupt cleanup state file {path:?}: {e}"))
        })
    }

    pub fn save(&self, state: &CleanupState) -> Result<()> {
        let path = self.path_for(&state.session_id)?;
        let json = serde_json::to_string_pretty(state)?;
        std::fs::write(&path, json)?;
        Ok(())
    }

    pub fn remove(&self, session_id: &str) {
        if let Ok(path) = self.path_for(session_id) {
            let _ = std::fs::remove_file(path);
        }
    }

    pub fn exists(&self, session_id: &str) -> bool {
        self.path_for(session_id)
            .map(|p| p.exists())
            .unwrap_or(false)
    }

    /// Return every cleanup state currently on disk that still has at least
    /// one unresolved failure. Used as a "cleanup backlog" guard before
    /// starting a fresh --ephemeral session so users don't quietly pile up
    /// orphaned resources. States with only successful deletes (which is an
    /// interim state during cleanup) are skipped — the file would be removed
    /// on next full success anyway.
    pub fn list_pending(&self) -> Vec<CleanupState> {
        let mut out = Vec::new();
        let entries = match std::fs::read_dir(&self.dir) {
            Ok(e) => e,
            Err(_) => return out,
        };
        for entry in entries.flatten() {
            let path = entry.path();
            if path.extension().and_then(|s| s.to_str()) != Some("json") {
                continue;
            }
            let Ok(json) = std::fs::read_to_string(&path) else {
                continue;
            };
            let Ok(state) = serde_json::from_str::<CleanupState>(&json) else {
                continue;
            };
            if !state.failed.is_empty() {
                out.push(state);
            }
        }
        // Sort by started_at for predictable listing order.
        out.sort_by_key(|s| s.started_at);
        out
    }
}

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

    #[test]
    fn parses_s3_bucket_arn() {
        let r = TaggedResource::from_arn("arn:aws:s3:::my-bucket").unwrap();
        assert_eq!(r.service, "s3");
        assert_eq!(r.resource_type, "bucket");
        assert_eq!(r.name, "my-bucket");
    }

    #[test]
    fn parses_dynamodb_table_arn() {
        let r = TaggedResource::from_arn("arn:aws:dynamodb:us-east-1:123:table/Users").unwrap();
        assert_eq!(r.service, "dynamodb");
        assert_eq!(r.resource_type, "table");
        assert_eq!(r.name, "Users");
    }

    #[test]
    fn parses_lambda_function_arn() {
        let r = TaggedResource::from_arn("arn:aws:lambda:us-east-1:123:function:my-fn").unwrap();
        assert_eq!(r.service, "lambda");
        assert_eq!(r.resource_type, "function");
        assert_eq!(r.name, "my-fn");
    }

    #[test]
    fn parses_iam_role_arn() {
        let r = TaggedResource::from_arn("arn:aws:iam::123:role/MyRole").unwrap();
        assert_eq!(r.service, "iam");
        assert_eq!(r.resource_type, "role");
        assert_eq!(r.name, "MyRole");
    }

    #[test]
    fn parses_sqs_queue_arn() {
        let r = TaggedResource::from_arn("arn:aws:sqs:us-east-1:123:my-queue").unwrap();
        assert_eq!(r.service, "sqs");
        assert_eq!(r.resource_type, "queue");
        assert_eq!(r.name, "my-queue");
    }

    #[test]
    fn rejects_bad_arn() {
        assert!(TaggedResource::from_arn("not-an-arn").is_none());
        assert!(TaggedResource::from_arn("arn:aws:s3").is_none());
    }

    #[test]
    fn sqs_url_rebuilt_from_arn() {
        let url = sqs_url_from_arn("arn:aws:sqs:us-east-1:123456789012:foo").unwrap();
        assert_eq!(url, "https://sqs.us-east-1.amazonaws.com/123456789012/foo");
    }

    #[test]
    fn sqs_url_govcloud() {
        let url = sqs_url_from_arn("arn:aws-us-gov:sqs:us-gov-west-1:123456789012:bar").unwrap();
        assert_eq!(
            url,
            "https://sqs.us-gov-west-1.amazonaws.com/123456789012/bar"
        );
    }

    #[test]
    fn sqs_url_china() {
        let url = sqs_url_from_arn("arn:aws-cn:sqs:cn-north-1:123456789012:baz").unwrap();
        assert_eq!(
            url,
            "https://sqs.cn-north-1.amazonaws.com.cn/123456789012/baz"
        );
    }

    #[test]
    fn delete_command_for_s3_bucket() {
        let r = TaggedResource::from_arn("arn:aws:s3:::mybk").unwrap();
        let cmd = delete_command(&r).unwrap();
        assert_eq!(cmd[0..3], vec!["aws", "s3", "rb"]);
        assert!(cmd.contains(&"s3://mybk".to_string()));
        // No --force: non-empty buckets fail safely
        assert!(!cmd.contains(&"--force".to_string()));
    }

    #[test]
    fn delete_command_for_dynamodb() {
        let r = TaggedResource::from_arn("arn:aws:dynamodb:us-east-1:1:table/T").unwrap();
        let cmd = delete_command(&r).unwrap();
        assert_eq!(cmd[0..3], vec!["aws", "dynamodb", "delete-table"]);
        assert!(cmd.contains(&"T".to_string()));
    }

    #[test]
    fn delete_command_for_sqs_uses_url() {
        let r = TaggedResource::from_arn("arn:aws:sqs:us-east-1:1:q1").unwrap();
        let cmd = delete_command(&r).unwrap();
        assert!(cmd.iter().any(|s| s.starts_with("https://sqs.")));
    }

    #[test]
    fn delete_command_for_kms_schedules_deletion() {
        let r = TaggedResource::from_arn("arn:aws:kms:us-east-1:1:key/uuid").unwrap();
        let cmd = delete_command(&r).unwrap();
        assert!(cmd.contains(&"schedule-key-deletion".to_string()));
        assert!(cmd.contains(&"7".to_string())); // 7-day pending window
    }

    #[test]
    fn delete_command_for_unsupported_returns_none() {
        let r = TaggedResource {
            arn: "arn:aws:exotic:us-east-1:1:thing/x".to_string(),
            service: "exotic".to_string(),
            resource_type: "thing".to_string(),
            name: "x".to_string(),
        };
        assert!(delete_command(&r).is_none());
    }

    #[test]
    fn delete_dry_run_returns_dryrun_outcome() {
        let r = TaggedResource::from_arn("arn:aws:dynamodb:us-east-1:1:table/T").unwrap();
        assert_eq!(delete(&r, true), DeleteOutcome::DryRun);
    }

    #[test]
    fn cleanup_state_records_deleted_arns() {
        let mut state = CleanupState::new("sess-1");
        assert!(!state.is_deleted("arn:aws:s3:::bk"));
        state.record_deleted("arn:aws:s3:::bk");
        assert!(state.is_deleted("arn:aws:s3:::bk"));
    }

    #[test]
    fn cleanup_state_record_deleted_is_idempotent() {
        let mut state = CleanupState::new("s");
        state.record_deleted("arn:x");
        state.record_deleted("arn:x");
        assert_eq!(state.deleted_arns.len(), 1);
    }

    #[test]
    fn cleanup_state_record_deleted_clears_prior_failure() {
        let mut state = CleanupState::new("s");
        let r = TaggedResource::from_arn("arn:aws:s3:::bk").unwrap();
        state.record_failed(&r, "temporary throttle");
        assert_eq!(state.failed.len(), 1);
        state.record_deleted(&r.arn);
        assert!(state.failed.is_empty());
        assert!(state.is_deleted(&r.arn));
    }

    #[test]
    fn cleanup_state_record_failed_updates_reason() {
        let mut state = CleanupState::new("s");
        let r = TaggedResource::from_arn("arn:aws:s3:::bk").unwrap();
        state.record_failed(&r, "first error");
        state.record_failed(&r, "second error");
        assert_eq!(state.failed.len(), 1);
        assert_eq!(state.failed[0].1, "second error");
    }

    #[test]
    fn cleanup_state_roundtrips_through_store() {
        let tmp = tempfile::tempdir().unwrap();
        let store = CleanupStateStore::with_dir(tmp.path()).unwrap();
        let mut state = store.load_or_new("sess-xyz").unwrap();
        assert!(state.deleted_arns.is_empty());
        state.record_deleted("arn:aws:s3:::bk1");
        store.save(&state).unwrap();

        let reloaded = store.load_or_new("sess-xyz").unwrap();
        assert_eq!(reloaded.deleted_arns, vec!["arn:aws:s3:::bk1".to_string()]);
        assert!(store.exists("sess-xyz"));

        store.remove("sess-xyz");
        assert!(!store.exists("sess-xyz"));
    }

    #[test]
    fn cleanup_state_store_returns_fresh_state_for_unknown_session() {
        let tmp = tempfile::tempdir().unwrap();
        let store = CleanupStateStore::with_dir(tmp.path()).unwrap();
        let state = store.load_or_new("brand-new").unwrap();
        assert_eq!(state.session_id, "brand-new");
        assert!(state.deleted_arns.is_empty());
        assert!(state.failed.is_empty());
    }

    #[test]
    fn delete_tier_puts_iam_policies_last() {
        let role = TaggedResource::from_arn("arn:aws:iam::1:role/R").unwrap();
        let policy = TaggedResource::from_arn("arn:aws:iam::1:policy/P").unwrap();
        let leaf = TaggedResource::from_arn("arn:aws:s3:::bk").unwrap();
        assert!(delete_tier(&leaf) < delete_tier(&role));
        assert!(delete_tier(&role) < delete_tier(&policy));
    }

    #[test]
    fn sort_for_deletion_orders_leaves_roles_policies() {
        let mut items = vec![
            TaggedResource::from_arn("arn:aws:iam::1:policy/P").unwrap(),
            TaggedResource::from_arn("arn:aws:s3:::bk").unwrap(),
            TaggedResource::from_arn("arn:aws:iam::1:role/R").unwrap(),
            TaggedResource::from_arn("arn:aws:dynamodb:us-east-1:1:table/T").unwrap(),
        ];
        sort_for_deletion(&mut items);
        // Leaves (s3, dynamodb) first, role next, policy last.
        assert_eq!(items[0].service, "s3");
        assert_eq!(items[1].service, "dynamodb");
        assert_eq!(
            (items[2].service.as_str(), items[2].resource_type.as_str()),
            ("iam", "role")
        );
        assert_eq!(
            (items[3].service.as_str(), items[3].resource_type.as_str()),
            ("iam", "policy")
        );
    }

    #[test]
    fn sort_for_deletion_is_stable_within_tier() {
        // Two leaves; sort must preserve input order when tiers are equal.
        let mut items = vec![
            TaggedResource::from_arn("arn:aws:s3:::first").unwrap(),
            TaggedResource::from_arn("arn:aws:s3:::second").unwrap(),
        ];
        sort_for_deletion(&mut items);
        assert_eq!(items[0].name, "first");
        assert_eq!(items[1].name, "second");
    }

    #[test]
    fn cost_hint_rds_is_nonzero_and_usage_dependent() {
        let db = TaggedResource::from_arn("arn:aws:rds:us-east-1:1:db:prod").unwrap();
        let h = estimate_daily_cost(&db).expect("rds has a hint");
        assert!(h.usd_per_day > 0.0);
        assert!(h.usage_dependent);
    }

    #[test]
    fn cost_hint_iam_and_lambda_are_free() {
        let role = TaggedResource::from_arn("arn:aws:iam::1:role/R").unwrap();
        let func = TaggedResource::from_arn("arn:aws:lambda:us-east-1:1:function:f").unwrap();
        assert!(estimate_daily_cost(&role).is_none());
        assert!(estimate_daily_cost(&func).is_none());
    }

    #[test]
    fn cost_hint_kms_is_flat_and_not_usage_dependent() {
        let key = TaggedResource::from_arn("arn:aws:kms:us-east-1:1:key/uuid").unwrap();
        let h = estimate_daily_cost(&key).expect("kms has a hint");
        assert!(h.usd_per_day > 0.0);
        assert!(!h.usage_dependent);
    }

    #[test]
    fn cost_hint_s3_is_usage_dependent_floor_zero() {
        let bucket = TaggedResource::from_arn("arn:aws:s3:::mybk").unwrap();
        let h = estimate_daily_cost(&bucket).expect("s3 has a hint");
        assert_eq!(h.usd_per_day, 0.0);
        assert!(h.usage_dependent);
    }

    #[test]
    fn list_pending_returns_only_states_with_failures() {
        let tmp = tempfile::tempdir().unwrap();
        let store = CleanupStateStore::with_dir(tmp.path()).unwrap();
        // Session A: has failures → should appear.
        let mut a = store.load_or_new("sess-a").unwrap();
        let res = TaggedResource::from_arn("arn:aws:s3:::stuck").unwrap();
        a.record_failed(&res, "throttled");
        store.save(&a).unwrap();
        // Session B: only deleted, no failures → should NOT appear.
        let mut b = store.load_or_new("sess-b").unwrap();
        b.record_deleted("arn:aws:s3:::ok");
        store.save(&b).unwrap();
        // Session C: pristine → load_or_new doesn't persist, skip save.

        let pending = store.list_pending();
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].session_id, "sess-a");
        assert_eq!(pending[0].failed.len(), 1);
    }

    #[test]
    fn list_pending_is_empty_when_no_state_files() {
        let tmp = tempfile::tempdir().unwrap();
        let store = CleanupStateStore::with_dir(tmp.path()).unwrap();
        assert!(store.list_pending().is_empty());
    }

    #[test]
    fn total_cost_sums_hints_and_flags_usage() {
        let items = vec![
            TaggedResource::from_arn("arn:aws:kms:us-east-1:1:key/a").unwrap(),
            TaggedResource::from_arn("arn:aws:secretsmanager:us-east-1:1:secret:s-AbCd").unwrap(),
            TaggedResource::from_arn("arn:aws:iam::1:role/R").unwrap(),
            TaggedResource::from_arn("arn:aws:s3:::bk").unwrap(),
        ];
        let (total, any_usage) = estimate_daily_cost_total(&items);
        // kms ($0.033) + secret ($0.013) + iam (none) + s3 (floor 0)
        assert!((total - 0.046).abs() < 0.0005);
        // s3 flips the usage-dependent flag.
        assert!(any_usage);
    }
}