Skip to main content

fakecloud_eks/
service.rs

1//! AWS EKS REST-JSON service handler.
2//!
3//! Implements the cluster control plane (Create/Describe/List/DeleteCluster,
4//! UpdateClusterConfig, UpdateClusterVersion, Describe/ListUpdates,
5//! Tag/Untag/ListTagsForResource) plus managed node groups
6//! (Create/Describe/List/DeleteNodegroup, UpdateNodegroupConfig/Version) and
7//! Fargate profiles (Create/Describe/List/DeleteFargateProfile), add-ons,
8//! access entries and their access-policy associations, OIDC identity-provider
9//! configs, and Pod Identity associations, all modeled as cluster
10//! sub-resources.
11
12use async_trait::async_trait;
13use chrono::Utc;
14use http::{Method, StatusCode};
15use serde_json::{json, Value};
16use std::sync::Arc;
17use tokio::sync::Mutex as AsyncMutex;
18
19use fakecloud_core::pagination::paginate_checked;
20use fakecloud_core::service::{AwsRequest, AwsResponse, AwsService, AwsServiceError};
21use fakecloud_persistence::SnapshotStore;
22
23use crate::eks_helpers::*;
24
25use crate::state::{
26    access_entry_arn, addon_arn, capability_arn, cluster_arn, eks_anywhere_subscription_arn,
27    fargate_profile_arn, identity_provider_config_arn, nodegroup_arn, pod_identity_association_arn,
28    AccessEntry, Addon, AssociatedPolicy, Capability, Cluster, EksAnywhereSubscription,
29    EksSnapshot, FargateProfile, IdentityProviderConfig, InsightsRefresh, Nodegroup,
30    PodIdentityAssociation, SharedEksState, DEFAULT_K8S_VERSION, EKS_SNAPSHOT_SCHEMA_VERSION,
31};
32
33/// The set of control-plane log types EKS reports on every cluster.
34pub(crate) const LOG_TYPES: &[&str] = &[
35    "api",
36    "audit",
37    "authenticator",
38    "controllerManager",
39    "scheduler",
40];
41
42pub const EKS_ACTIONS: &[&str] = &[
43    "CreateCluster",
44    "DescribeCluster",
45    "ListClusters",
46    "DeleteCluster",
47    "UpdateClusterConfig",
48    "UpdateClusterVersion",
49    "DescribeUpdate",
50    "ListUpdates",
51    "TagResource",
52    "UntagResource",
53    "ListTagsForResource",
54    "CreateNodegroup",
55    "DescribeNodegroup",
56    "ListNodegroups",
57    "DeleteNodegroup",
58    "UpdateNodegroupConfig",
59    "UpdateNodegroupVersion",
60    "CreateFargateProfile",
61    "DescribeFargateProfile",
62    "ListFargateProfiles",
63    "DeleteFargateProfile",
64    "CreateAddon",
65    "DescribeAddon",
66    "ListAddons",
67    "DeleteAddon",
68    "UpdateAddon",
69    "DescribeAddonVersions",
70    "DescribeAddonConfiguration",
71    "CreateAccessEntry",
72    "DescribeAccessEntry",
73    "ListAccessEntries",
74    "DeleteAccessEntry",
75    "UpdateAccessEntry",
76    "AssociateAccessPolicy",
77    "DisassociateAccessPolicy",
78    "ListAssociatedAccessPolicies",
79    "ListAccessPolicies",
80    "AssociateIdentityProviderConfig",
81    "DisassociateIdentityProviderConfig",
82    "DescribeIdentityProviderConfig",
83    "ListIdentityProviderConfigs",
84    "CreatePodIdentityAssociation",
85    "DeletePodIdentityAssociation",
86    "DescribePodIdentityAssociation",
87    "ListPodIdentityAssociations",
88    "UpdatePodIdentityAssociation",
89    "DescribeInsight",
90    "ListInsights",
91    "DescribeInsightsRefresh",
92    "StartInsightsRefresh",
93    "AssociateEncryptionConfig",
94    "CancelUpdate",
95    "DeregisterCluster",
96    "RegisterCluster",
97    "DescribeClusterVersions",
98    "CreateCapability",
99    "DeleteCapability",
100    "DescribeCapability",
101    "ListCapabilities",
102    "UpdateCapability",
103    "CreateEksAnywhereSubscription",
104    "DeleteEksAnywhereSubscription",
105    "DescribeEksAnywhereSubscription",
106    "ListEksAnywhereSubscriptions",
107    "UpdateEksAnywhereSubscription",
108];
109
110pub struct EksService {
111    state: SharedEksState,
112    snapshot_store: Option<Arc<dyn SnapshotStore>>,
113    snapshot_lock: Arc<AsyncMutex<()>>,
114}
115
116enum PathArgs {
117    None,
118    Name(String),
119    Update {
120        name: String,
121        update_id: String,
122    },
123    Arn(String),
124    /// A sub-resource collection scoped to a cluster (create/list).
125    Cluster(String),
126    /// A named sub-resource of a cluster (describe/delete/update).
127    ClusterChild {
128        cluster: String,
129        name: String,
130    },
131    /// A named access policy association on an access entry
132    /// (`DisassociateAccessPolicy`).
133    AccessPolicyChild {
134        cluster: String,
135        principal: String,
136        policy_arn: String,
137    },
138}
139
140impl EksService {
141    pub fn new(state: SharedEksState) -> Self {
142        Self {
143            state,
144            snapshot_store: None,
145            snapshot_lock: Arc::new(AsyncMutex::new(())),
146        }
147    }
148
149    pub fn with_snapshot_store(mut self, store: Arc<dyn SnapshotStore>) -> Self {
150        self.snapshot_store = Some(store);
151        self
152    }
153
154    async fn save_snapshot(&self) {
155        save_eks_snapshot(
156            &self.state,
157            self.snapshot_store.clone(),
158            &self.snapshot_lock,
159        )
160        .await;
161    }
162
163    /// Persist hook for the CloudFormation provisioner; `None` in memory mode.
164    pub fn snapshot_hook(&self) -> Option<fakecloud_persistence::SnapshotHook> {
165        let store = self.snapshot_store.clone()?;
166        let state = self.state.clone();
167        let lock = self.snapshot_lock.clone();
168        Some(Arc::new(move || {
169            let state = state.clone();
170            let store = store.clone();
171            let lock = lock.clone();
172            Box::pin(async move {
173                save_eks_snapshot(&state, Some(store), &lock).await;
174            })
175        }))
176    }
177
178    fn resolve_action(req: &AwsRequest) -> Option<(&'static str, PathArgs)> {
179        // Derive segments from the raw path, preserving INTERNAL empty segments
180        // (only the leading empty from the leading `/` and a single trailing
181        // empty are dropped). `req.path_segments` strips every empty part, which
182        // would collapse a real 3-segment path with an empty middle label
183        // (e.g. `/clusters//addons`, ListAddons with an empty clusterName) down
184        // to `["clusters","addons"]` and mis-route it to DescribeCluster. Keeping
185        // internal empties makes the exact-slice matches below precise, so a
186        // not-yet-implemented sub-resource path falls through to None.
187        let raw = req.raw_path.split('?').next().unwrap_or(&req.raw_path);
188        let trimmed = raw.strip_prefix('/').unwrap_or(raw);
189        let trimmed = trimmed.strip_suffix('/').unwrap_or(trimmed);
190        let segs: Vec<&str> = if trimmed.is_empty() {
191            Vec::new()
192        } else {
193            trimmed.split('/').collect()
194        };
195        match (&req.method, segs.as_slice()) {
196            (&Method::POST, ["clusters"]) => Some(("CreateCluster", PathArgs::None)),
197            (&Method::GET, ["clusters"]) => Some(("ListClusters", PathArgs::None)),
198            (&Method::GET, ["clusters", name]) => {
199                Some(("DescribeCluster", PathArgs::Name(decode(name))))
200            }
201            (&Method::DELETE, ["clusters", name]) => {
202                Some(("DeleteCluster", PathArgs::Name(decode(name))))
203            }
204            (&Method::POST, ["clusters", name, "update-config"]) => {
205                Some(("UpdateClusterConfig", PathArgs::Name(decode(name))))
206            }
207            (&Method::POST, ["clusters", name, "updates"]) => {
208                Some(("UpdateClusterVersion", PathArgs::Name(decode(name))))
209            }
210            (&Method::GET, ["clusters", name, "updates"]) => {
211                Some(("ListUpdates", PathArgs::Name(decode(name))))
212            }
213            (&Method::GET, ["clusters", name, "updates", update_id]) => Some((
214                "DescribeUpdate",
215                PathArgs::Update {
216                    name: decode(name),
217                    update_id: decode(update_id),
218                },
219            )),
220            // Node groups (sub-resources of a cluster).
221            (&Method::POST, ["clusters", c, "node-groups"]) => {
222                Some(("CreateNodegroup", PathArgs::Cluster(decode(c))))
223            }
224            (&Method::GET, ["clusters", c, "node-groups"]) => {
225                Some(("ListNodegroups", PathArgs::Cluster(decode(c))))
226            }
227            (&Method::GET, ["clusters", c, "node-groups", n]) => Some((
228                "DescribeNodegroup",
229                PathArgs::ClusterChild {
230                    cluster: decode(c),
231                    name: decode(n),
232                },
233            )),
234            (&Method::DELETE, ["clusters", c, "node-groups", n]) => Some((
235                "DeleteNodegroup",
236                PathArgs::ClusterChild {
237                    cluster: decode(c),
238                    name: decode(n),
239                },
240            )),
241            (&Method::POST, ["clusters", c, "node-groups", n, "update-config"]) => Some((
242                "UpdateNodegroupConfig",
243                PathArgs::ClusterChild {
244                    cluster: decode(c),
245                    name: decode(n),
246                },
247            )),
248            (&Method::POST, ["clusters", c, "node-groups", n, "update-version"]) => Some((
249                "UpdateNodegroupVersion",
250                PathArgs::ClusterChild {
251                    cluster: decode(c),
252                    name: decode(n),
253                },
254            )),
255            // Fargate profiles (sub-resources of a cluster).
256            (&Method::POST, ["clusters", c, "fargate-profiles"]) => {
257                Some(("CreateFargateProfile", PathArgs::Cluster(decode(c))))
258            }
259            (&Method::GET, ["clusters", c, "fargate-profiles"]) => {
260                Some(("ListFargateProfiles", PathArgs::Cluster(decode(c))))
261            }
262            (&Method::GET, ["clusters", c, "fargate-profiles", n]) => Some((
263                "DescribeFargateProfile",
264                PathArgs::ClusterChild {
265                    cluster: decode(c),
266                    name: decode(n),
267                },
268            )),
269            (&Method::DELETE, ["clusters", c, "fargate-profiles", n]) => Some((
270                "DeleteFargateProfile",
271                PathArgs::ClusterChild {
272                    cluster: decode(c),
273                    name: decode(n),
274                },
275            )),
276            // Add-ons (sub-resources of a cluster).
277            (&Method::POST, ["clusters", c, "addons"]) => {
278                Some(("CreateAddon", PathArgs::Cluster(decode(c))))
279            }
280            (&Method::GET, ["clusters", c, "addons"]) => {
281                Some(("ListAddons", PathArgs::Cluster(decode(c))))
282            }
283            (&Method::GET, ["clusters", c, "addons", n]) => Some((
284                "DescribeAddon",
285                PathArgs::ClusterChild {
286                    cluster: decode(c),
287                    name: decode(n),
288                },
289            )),
290            (&Method::DELETE, ["clusters", c, "addons", n]) => Some((
291                "DeleteAddon",
292                PathArgs::ClusterChild {
293                    cluster: decode(c),
294                    name: decode(n),
295                },
296            )),
297            (&Method::POST, ["clusters", c, "addons", n, "update"]) => Some((
298                "UpdateAddon",
299                PathArgs::ClusterChild {
300                    cluster: decode(c),
301                    name: decode(n),
302                },
303            )),
304            // Add-on catalogue ops (not scoped to a cluster).
305            (&Method::GET, ["addons", "supported-versions"]) => {
306                Some(("DescribeAddonVersions", PathArgs::None))
307            }
308            (&Method::GET, ["addons", "configuration-schemas"]) => {
309                Some(("DescribeAddonConfiguration", PathArgs::None))
310            }
311            // Access entries (sub-resources of a cluster, keyed by principalArn).
312            (&Method::POST, ["clusters", c, "access-entries"]) => {
313                Some(("CreateAccessEntry", PathArgs::Cluster(decode(c))))
314            }
315            (&Method::GET, ["clusters", c, "access-entries"]) => {
316                Some(("ListAccessEntries", PathArgs::Cluster(decode(c))))
317            }
318            (&Method::GET, ["clusters", c, "access-entries", p]) => Some((
319                "DescribeAccessEntry",
320                PathArgs::ClusterChild {
321                    cluster: decode(c),
322                    name: decode(p),
323                },
324            )),
325            (&Method::DELETE, ["clusters", c, "access-entries", p]) => Some((
326                "DeleteAccessEntry",
327                PathArgs::ClusterChild {
328                    cluster: decode(c),
329                    name: decode(p),
330                },
331            )),
332            (&Method::POST, ["clusters", c, "access-entries", p]) => Some((
333                "UpdateAccessEntry",
334                PathArgs::ClusterChild {
335                    cluster: decode(c),
336                    name: decode(p),
337                },
338            )),
339            // Access-policy associations on an access entry.
340            (&Method::POST, ["clusters", c, "access-entries", p, "access-policies"]) => Some((
341                "AssociateAccessPolicy",
342                PathArgs::ClusterChild {
343                    cluster: decode(c),
344                    name: decode(p),
345                },
346            )),
347            (&Method::GET, ["clusters", c, "access-entries", p, "access-policies"]) => Some((
348                "ListAssociatedAccessPolicies",
349                PathArgs::ClusterChild {
350                    cluster: decode(c),
351                    name: decode(p),
352                },
353            )),
354            (&Method::DELETE, ["clusters", c, "access-entries", p, "access-policies", policy]) => {
355                Some((
356                    "DisassociateAccessPolicy",
357                    PathArgs::AccessPolicyChild {
358                        cluster: decode(c),
359                        principal: decode(p),
360                        policy_arn: decode(policy),
361                    },
362                ))
363            }
364            // Access-policy catalogue (not scoped to a cluster).
365            (&Method::GET, ["access-policies"]) => Some(("ListAccessPolicies", PathArgs::None)),
366            // Identity-provider configs (sub-resources of a cluster). The
367            // associate/disassociate/describe forms are POSTs to fixed action
368            // sub-paths, carrying the config identity in the body.
369            (&Method::POST, ["clusters", c, "identity-provider-configs", "associate"]) => Some((
370                "AssociateIdentityProviderConfig",
371                PathArgs::Cluster(decode(c)),
372            )),
373            (&Method::POST, ["clusters", c, "identity-provider-configs", "disassociate"]) => {
374                Some((
375                    "DisassociateIdentityProviderConfig",
376                    PathArgs::Cluster(decode(c)),
377                ))
378            }
379            (&Method::POST, ["clusters", c, "identity-provider-configs", "describe"]) => Some((
380                "DescribeIdentityProviderConfig",
381                PathArgs::Cluster(decode(c)),
382            )),
383            (&Method::GET, ["clusters", c, "identity-provider-configs"]) => {
384                Some(("ListIdentityProviderConfigs", PathArgs::Cluster(decode(c))))
385            }
386            // Pod identity associations (sub-resources of a cluster, keyed by id).
387            (&Method::POST, ["clusters", c, "pod-identity-associations"]) => {
388                Some(("CreatePodIdentityAssociation", PathArgs::Cluster(decode(c))))
389            }
390            (&Method::GET, ["clusters", c, "pod-identity-associations"]) => {
391                Some(("ListPodIdentityAssociations", PathArgs::Cluster(decode(c))))
392            }
393            (&Method::GET, ["clusters", c, "pod-identity-associations", id]) => Some((
394                "DescribePodIdentityAssociation",
395                PathArgs::ClusterChild {
396                    cluster: decode(c),
397                    name: decode(id),
398                },
399            )),
400            (&Method::DELETE, ["clusters", c, "pod-identity-associations", id]) => Some((
401                "DeletePodIdentityAssociation",
402                PathArgs::ClusterChild {
403                    cluster: decode(c),
404                    name: decode(id),
405                },
406            )),
407            (&Method::POST, ["clusters", c, "pod-identity-associations", id]) => Some((
408                "UpdatePodIdentityAssociation",
409                PathArgs::ClusterChild {
410                    cluster: decode(c),
411                    name: decode(id),
412                },
413            )),
414            // Insights (sub-resources of a cluster). List is a POST (it carries
415            // a filter body); Describe is a GET keyed by insight id.
416            (&Method::POST, ["clusters", c, "insights"]) => {
417                Some(("ListInsights", PathArgs::Cluster(decode(c))))
418            }
419            (&Method::GET, ["clusters", c, "insights", id]) => Some((
420                "DescribeInsight",
421                PathArgs::ClusterChild {
422                    cluster: decode(c),
423                    name: decode(id),
424                },
425            )),
426            (&Method::GET, ["clusters", c, "insights-refresh"]) => {
427                Some(("DescribeInsightsRefresh", PathArgs::Cluster(decode(c))))
428            }
429            (&Method::POST, ["clusters", c, "insights-refresh"]) => {
430                Some(("StartInsightsRefresh", PathArgs::Cluster(decode(c))))
431            }
432            // Encryption config association (mints a tracked cluster Update).
433            (&Method::POST, ["clusters", c, "encryption-config", "associate"]) => {
434                Some(("AssociateEncryptionConfig", PathArgs::Cluster(decode(c))))
435            }
436            // Cancel an in-progress update.
437            (&Method::POST, ["clusters", name, "updates", update_id, "cancel-update"]) => Some((
438                "CancelUpdate",
439                PathArgs::Update {
440                    name: decode(name),
441                    update_id: decode(update_id),
442                },
443            )),
444            // Connected (registered) clusters.
445            (&Method::POST, ["cluster-registrations"]) => Some(("RegisterCluster", PathArgs::None)),
446            (&Method::DELETE, ["cluster-registrations", name]) => {
447                Some(("DeregisterCluster", PathArgs::Name(decode(name))))
448            }
449            // Kubernetes version catalogue (account/region-scoped).
450            (&Method::GET, ["cluster-versions"]) => {
451                Some(("DescribeClusterVersions", PathArgs::None))
452            }
453            // Capabilities (sub-resources of a cluster).
454            (&Method::POST, ["clusters", c, "capabilities"]) => {
455                Some(("CreateCapability", PathArgs::Cluster(decode(c))))
456            }
457            (&Method::GET, ["clusters", c, "capabilities"]) => {
458                Some(("ListCapabilities", PathArgs::Cluster(decode(c))))
459            }
460            (&Method::GET, ["clusters", c, "capabilities", n]) => Some((
461                "DescribeCapability",
462                PathArgs::ClusterChild {
463                    cluster: decode(c),
464                    name: decode(n),
465                },
466            )),
467            (&Method::DELETE, ["clusters", c, "capabilities", n]) => Some((
468                "DeleteCapability",
469                PathArgs::ClusterChild {
470                    cluster: decode(c),
471                    name: decode(n),
472                },
473            )),
474            (&Method::POST, ["clusters", c, "capabilities", n]) => Some((
475                "UpdateCapability",
476                PathArgs::ClusterChild {
477                    cluster: decode(c),
478                    name: decode(n),
479                },
480            )),
481            // EKS Anywhere subscriptions (account-scoped).
482            (&Method::POST, ["eks-anywhere-subscriptions"]) => {
483                Some(("CreateEksAnywhereSubscription", PathArgs::None))
484            }
485            (&Method::GET, ["eks-anywhere-subscriptions"]) => {
486                Some(("ListEksAnywhereSubscriptions", PathArgs::None))
487            }
488            (&Method::GET, ["eks-anywhere-subscriptions", id]) => Some((
489                "DescribeEksAnywhereSubscription",
490                PathArgs::Name(decode(id)),
491            )),
492            (&Method::DELETE, ["eks-anywhere-subscriptions", id]) => {
493                Some(("DeleteEksAnywhereSubscription", PathArgs::Name(decode(id))))
494            }
495            (&Method::POST, ["eks-anywhere-subscriptions", id]) => {
496                Some(("UpdateEksAnywhereSubscription", PathArgs::Name(decode(id))))
497            }
498            (&Method::POST, ["tags", arn]) => Some(("TagResource", PathArgs::Arn(decode(arn)))),
499            (&Method::DELETE, ["tags", arn]) => Some(("UntagResource", PathArgs::Arn(decode(arn)))),
500            (&Method::GET, ["tags", arn]) => {
501                Some(("ListTagsForResource", PathArgs::Arn(decode(arn))))
502            }
503            _ => None,
504        }
505    }
506
507    fn create_cluster(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
508        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
509
510        let name = body
511            .get("name")
512            .and_then(|v| v.as_str())
513            .ok_or_else(|| invalid_parameter("name is required"))?
514            .to_string();
515        validate_cluster_name(&name)?;
516
517        let role_arn = body
518            .get("roleArn")
519            .and_then(|v| v.as_str())
520            .ok_or_else(|| invalid_parameter("roleArn is required"))?
521            .to_string();
522
523        let vpc_req = body
524            .get("resourcesVpcConfig")
525            .filter(|v| v.is_object())
526            .ok_or_else(|| invalid_parameter("resourcesVpcConfig is required"))?;
527
528        let version = body
529            .get("version")
530            .and_then(|v| v.as_str())
531            .unwrap_or(DEFAULT_K8S_VERSION)
532            .to_string();
533
534        let mut accounts = self.state.write();
535        let state = accounts.get_or_create(&req.account_id);
536        if state.clusters.contains_key(&name) {
537            return Err(AwsServiceError::aws_error(
538                StatusCode::CONFLICT,
539                "ResourceInUseException",
540                format!("Cluster already exists with name: {name}"),
541            ));
542        }
543
544        let region = req.region.clone();
545        let account_id = req.account_id.clone();
546        let arn = cluster_arn(&region, &account_id, &name);
547        let id = uuid::Uuid::new_v4().to_string();
548
549        let tags = parse_tag_map(body.get("tags"));
550
551        let cluster = Cluster {
552            name: name.clone(),
553            arn: arn.clone(),
554            version,
555            role_arn,
556            status: "CREATING".to_string(),
557            created_at: Utc::now(),
558            endpoint: format!(
559                "https://{}.gr7.{region}.eks.amazonaws.com",
560                id.replace('-', "").to_uppercase()
561            ),
562            platform_version: "eks.1".to_string(),
563            certificate_authority_data: default_ca_data(),
564            resources_vpc_config: build_vpc_config_response(vpc_req, &id),
565            kubernetes_network_config: build_k8s_network_config(
566                body.get("kubernetesNetworkConfig"),
567            ),
568            logging: build_logging(body.get("logging")),
569            tags,
570            updates: Default::default(),
571            connector_config: None,
572            encryption_config: body
573                .get("encryptionConfig")
574                .filter(|v| v.is_array())
575                .cloned(),
576            access_config: build_access_config(body.get("accessConfig")),
577            upgrade_policy: build_upgrade_policy(body.get("upgradePolicy")),
578            compute_config: body.get("computeConfig").cloned(),
579            storage_config: body.get("storageConfig").cloned(),
580            zonal_shift_config: body.get("zonalShiftConfig").cloned(),
581            remote_network_config: body.get("remoteNetworkConfig").cloned(),
582            control_plane_scaling_config: body.get("controlPlaneScalingConfig").cloned(),
583            deletion_protection: body.get("deletionProtection").and_then(|v| v.as_bool()),
584        };
585
586        let out = cluster_json(&cluster, &id);
587        state.clusters.insert(name, cluster);
588        Ok(AwsResponse::json(
589            StatusCode::OK,
590            json!({ "cluster": out }).to_string(),
591        ))
592    }
593
594    fn describe_cluster(
595        &self,
596        req: &AwsRequest,
597        name: &str,
598    ) -> Result<AwsResponse, AwsServiceError> {
599        let mut accounts = self.state.write();
600        let state = accounts.get_or_create(&req.account_id);
601        let cluster = state
602            .clusters
603            .get_mut(name)
604            .ok_or_else(not_found_cluster(name))?;
605        // Lazily settle CREATING -> ACTIVE on first describe, mirroring the
606        // eventual-consistency of the real control plane.
607        if cluster.status == "CREATING" {
608            cluster.status = "ACTIVE".to_string();
609        }
610        let id = arn_cluster_id(&cluster.endpoint);
611        Ok(AwsResponse::json(
612            StatusCode::OK,
613            json!({ "cluster": cluster_json(cluster, &id) }).to_string(),
614        ))
615    }
616
617    fn list_clusters(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
618        let max_results = validate_max_results(req)?;
619        let next_token = req.query_params.get("nextToken").cloned();
620
621        let accounts = self.state.read();
622        let Some(state) = accounts.get(&req.account_id) else {
623            return Ok(AwsResponse::json(
624                StatusCode::OK,
625                json!({ "clusters": [] }).to_string(),
626            ));
627        };
628        let names: Vec<String> = state.clusters.keys().cloned().collect();
629        let (page, token) = paginate_checked(&names, next_token.as_deref(), max_results)
630            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
631        let mut out = json!({ "clusters": page });
632        if let Some(t) = token {
633            out["nextToken"] = Value::String(t);
634        }
635        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
636    }
637
638    fn delete_cluster(&self, req: &AwsRequest, name: &str) -> Result<AwsResponse, AwsServiceError> {
639        let mut accounts = self.state.write();
640        let state = accounts.get_or_create(&req.account_id);
641        if !state.clusters.contains_key(name) {
642            return Err(not_found_cluster(name)());
643        }
644        // Real EKS refuses to delete a cluster that still has managed node
645        // groups or Fargate profiles attached; the caller must delete those
646        // first.
647        let has_nodegroups = state.nodegroups.get(name).is_some_and(|m| !m.is_empty());
648        let has_fargate = state
649            .fargate_profiles
650            .get(name)
651            .is_some_and(|m| !m.is_empty());
652        if has_nodegroups || has_fargate {
653            return Err(AwsServiceError::aws_error(
654                StatusCode::CONFLICT,
655                "ResourceInUseException",
656                format!(
657                    "Cluster has {} attached that must be deleted first",
658                    if has_nodegroups {
659                        "nodegroups"
660                    } else {
661                        "Fargate profiles"
662                    }
663                ),
664            ));
665        }
666        // Cascade-remove every remaining cluster sub-resource so a same-name
667        // recreate does not inherit the previous cluster's orphaned state.
668        state.nodegroups.remove(name);
669        state.fargate_profiles.remove(name);
670        state.addons.remove(name);
671        state.access_entries.remove(name);
672        state.identity_provider_configs.remove(name);
673        state.pod_identity_associations.remove(name);
674        state.insights.remove(name);
675        state.insights_refresh.remove(name);
676        state.capabilities.remove(name);
677        let mut cluster = state
678            .clusters
679            .remove(name)
680            .expect("existence checked above");
681        cluster.status = "DELETING".to_string();
682        let id = arn_cluster_id(&cluster.endpoint);
683        Ok(AwsResponse::json(
684            StatusCode::OK,
685            json!({ "cluster": cluster_json(&cluster, &id) }).to_string(),
686        ))
687    }
688
689    fn update_cluster_config(
690        &self,
691        req: &AwsRequest,
692        name: &str,
693    ) -> Result<AwsResponse, AwsServiceError> {
694        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
695        let mut accounts = self.state.write();
696        let state = accounts.get_or_create(&req.account_id);
697        let cluster = state
698            .clusters
699            .get_mut(name)
700            .ok_or_else(not_found_cluster(name))?;
701
702        // A single UpdateClusterConfig call may carry more than one config
703        // aspect; apply every field present (previously all but logging and
704        // resourcesVpcConfig were silently dropped) and report the first-matched
705        // aspect as the Update's type for a faithful DescribeUpdate round-trip.
706        let mut update_type = "ConfigUpdate";
707        let mut matched = false;
708        let mut params: Vec<(String, String)> = Vec::new();
709        macro_rules! mark {
710            ($ty:expr) => {
711                if !matched {
712                    update_type = $ty;
713                    matched = true;
714                }
715            };
716        }
717
718        if let Some(logging) = body.get("logging") {
719            cluster.logging = build_logging(Some(logging));
720            mark!("LoggingUpdate");
721            params.push(("ClusterLogging".to_string(), logging.to_string()));
722        }
723        if let Some(vpc) = body.get("resourcesVpcConfig") {
724            let id = arn_cluster_id(&cluster.endpoint);
725            cluster.resources_vpc_config = build_vpc_config_response(vpc, &id);
726            mark!("VpcConfigUpdate");
727            params.push(("ResourcesVpcConfig".to_string(), vpc.to_string()));
728        }
729        if let Some(ac) = body.get("accessConfig") {
730            cluster.access_config = build_access_config(Some(ac));
731            mark!("AccessConfigUpdate");
732            params.push((
733                "AuthenticationMode".to_string(),
734                cluster.access_config["authenticationMode"].to_string(),
735            ));
736        }
737        if let Some(up) = body.get("upgradePolicy") {
738            cluster.upgrade_policy = build_upgrade_policy(Some(up));
739            mark!("UpgradePolicyUpdate");
740            params.push((
741                "SupportType".to_string(),
742                cluster.upgrade_policy["supportType"].to_string(),
743            ));
744        }
745        if let Some(kn) = body.get("kubernetesNetworkConfig") {
746            cluster.kubernetes_network_config = build_k8s_network_config(Some(kn));
747            mark!("VpcConfigUpdate");
748            params.push(("KubernetesNetworkConfig".to_string(), kn.to_string()));
749        }
750        if let Some(cc) = body.get("computeConfig") {
751            cluster.compute_config = Some(cc.clone());
752            mark!("AutoModeUpdate");
753            params.push(("ComputeConfig".to_string(), cc.to_string()));
754        }
755        if let Some(sc) = body.get("storageConfig") {
756            cluster.storage_config = Some(sc.clone());
757            mark!("AutoModeUpdate");
758            params.push(("StorageConfig".to_string(), sc.to_string()));
759        }
760        if let Some(z) = body.get("zonalShiftConfig") {
761            cluster.zonal_shift_config = Some(z.clone());
762            mark!("ZonalShiftConfigUpdate");
763            params.push(("ZonalShiftConfig".to_string(), z.to_string()));
764        }
765        if let Some(rn) = body.get("remoteNetworkConfig") {
766            cluster.remote_network_config = Some(rn.clone());
767            mark!("RemoteNetworkConfigUpdate");
768            params.push(("RemoteNetworkConfig".to_string(), rn.to_string()));
769        }
770        if let Some(sp) = body.get("controlPlaneScalingConfig") {
771            cluster.control_plane_scaling_config = Some(sp.clone());
772            mark!("ControlPlaneScalingConfigUpdate");
773            params.push(("ControlPlaneScalingConfig".to_string(), sp.to_string()));
774        }
775        if let Some(dp) = body.get("deletionProtection").and_then(|v| v.as_bool()) {
776            cluster.deletion_protection = Some(dp);
777            mark!("DeletionProtectionUpdate");
778            params.push(("DeletionProtection".to_string(), dp.to_string()));
779        }
780        let _ = matched;
781
782        let update = new_update(update_type, params);
783        let out = update_json(&update);
784        cluster.updates.insert(update.id.clone(), update);
785        Ok(AwsResponse::json(
786            StatusCode::OK,
787            json!({ "update": out }).to_string(),
788        ))
789    }
790
791    fn update_cluster_version(
792        &self,
793        req: &AwsRequest,
794        name: &str,
795    ) -> Result<AwsResponse, AwsServiceError> {
796        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
797        let version = body
798            .get("version")
799            .and_then(|v| v.as_str())
800            .ok_or_else(|| invalid_parameter("version is required"))?
801            .to_string();
802
803        let mut accounts = self.state.write();
804        let state = accounts.get_or_create(&req.account_id);
805        let cluster = state
806            .clusters
807            .get_mut(name)
808            .ok_or_else(not_found_cluster(name))?;
809        cluster.version = version.clone();
810
811        let update = new_update(
812            "VersionUpdate",
813            vec![
814                ("Version".to_string(), version),
815                (
816                    "PlatformVersion".to_string(),
817                    cluster.platform_version.clone(),
818                ),
819            ],
820        );
821        let out = update_json(&update);
822        cluster.updates.insert(update.id.clone(), update);
823        Ok(AwsResponse::json(
824            StatusCode::OK,
825            json!({ "update": out }).to_string(),
826        ))
827    }
828
829    fn describe_update(
830        &self,
831        req: &AwsRequest,
832        name: &str,
833        update_id: &str,
834    ) -> Result<AwsResponse, AwsServiceError> {
835        let nodegroup_name = req.query_params.get("nodegroupName").cloned();
836        let addon_name = req.query_params.get("addonName").cloned();
837        let mut accounts = self.state.write();
838        let state = accounts.get_or_create(&req.account_id);
839        if !state.clusters.contains_key(name) {
840            return Err(not_found_cluster(name)());
841        }
842        // `DescribeUpdate` disambiguates cluster updates from node-group and
843        // add-on updates via the optional `nodegroupName` / `addonName` query
844        // params, mirroring the real API. When present, the update lives on
845        // that sub-resource's own update history.
846        let update = if let Some(ng_name) = nodegroup_name.as_deref() {
847            let ng = state
848                .nodegroups
849                .get_mut(name)
850                .and_then(|m| m.get_mut(ng_name))
851                .ok_or_else(not_found_nodegroup(ng_name))?;
852            ng.updates
853                .get_mut(update_id)
854                .ok_or_else(not_found_update(update_id))?
855        } else if let Some(a_name) = addon_name.as_deref() {
856            let addon = state
857                .addons
858                .get_mut(name)
859                .and_then(|m| m.get_mut(a_name))
860                .ok_or_else(not_found_addon(a_name))?;
861            addon
862                .updates
863                .get_mut(update_id)
864                .ok_or_else(not_found_update(update_id))?
865        } else {
866            let cluster = state
867                .clusters
868                .get_mut(name)
869                .ok_or_else(not_found_cluster(name))?;
870            cluster
871                .updates
872                .get_mut(update_id)
873                .ok_or_else(not_found_update(update_id))?
874        };
875        // Updates settle to Successful on first describe.
876        if update.status == "InProgress" {
877            update.status = "Successful".to_string();
878        }
879        Ok(AwsResponse::json(
880            StatusCode::OK,
881            json!({ "update": update_json(update) }).to_string(),
882        ))
883    }
884
885    fn list_updates(&self, req: &AwsRequest, name: &str) -> Result<AwsResponse, AwsServiceError> {
886        let max_results = validate_max_results(req)?;
887        let next_token = req.query_params.get("nextToken").cloned();
888
889        let nodegroup_name = req.query_params.get("nodegroupName").cloned();
890        let addon_name = req.query_params.get("addonName").cloned();
891        let accounts = self.state.read();
892        let state = accounts
893            .get(&req.account_id)
894            .ok_or_else(not_found_cluster(name))?;
895        if !state.clusters.contains_key(name) {
896            return Err(not_found_cluster(name)());
897        }
898        let ids: Vec<String> = if let Some(ng_name) = nodegroup_name.as_deref() {
899            let ng = state
900                .nodegroups
901                .get(name)
902                .and_then(|m| m.get(ng_name))
903                .ok_or_else(not_found_nodegroup(ng_name))?;
904            ng.updates.keys().cloned().collect()
905        } else if let Some(a_name) = addon_name.as_deref() {
906            let addon = state
907                .addons
908                .get(name)
909                .and_then(|m| m.get(a_name))
910                .ok_or_else(not_found_addon(a_name))?;
911            addon.updates.keys().cloned().collect()
912        } else {
913            let cluster = state
914                .clusters
915                .get(name)
916                .ok_or_else(not_found_cluster(name))?;
917            cluster.updates.keys().cloned().collect()
918        };
919        let (page, token) = paginate_checked(&ids, next_token.as_deref(), max_results)
920            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
921        let mut out = json!({ "updateIds": page });
922        if let Some(t) = token {
923            out["nextToken"] = Value::String(t);
924        }
925        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
926    }
927
928    fn tag_resource(&self, req: &AwsRequest, arn: &str) -> Result<AwsResponse, AwsServiceError> {
929        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
930        let tags = body
931            .get("tags")
932            .filter(|v| v.is_object())
933            .ok_or_else(|| bad_request("tags is required"))?;
934        validate_eks_arn(arn)?;
935        let mut accounts = self.state.write();
936        let state = accounts.get_or_create(&req.account_id);
937        // Resolve the ARN to whichever resource owns it (cluster or any
938        // sub-resource) and tag that resource. Previously every non-cluster ARN
939        // was rejected with a 400, so tagging a node group / add-on failed.
940        let target = locate_tag_target(state, arn);
941        let dest = tags_mut(state, &target);
942        for (k, v) in tags.as_object().unwrap() {
943            if let Some(v) = v.as_str() {
944                dest.insert(k.clone(), v.to_string());
945            }
946        }
947        Ok(AwsResponse::json(StatusCode::OK, "{}"))
948    }
949
950    fn untag_resource(&self, req: &AwsRequest, arn: &str) -> Result<AwsResponse, AwsServiceError> {
951        let keys = parse_multi_query(&req.raw_query, "tagKeys");
952        validate_eks_arn(arn)?;
953        let mut accounts = self.state.write();
954        let state = accounts.get_or_create(&req.account_id);
955        let target = locate_tag_target(state, arn);
956        let dest = tags_mut(state, &target);
957        for k in keys {
958            dest.remove(&k);
959        }
960        Ok(AwsResponse::json(StatusCode::OK, "{}"))
961    }
962
963    fn list_tags_for_resource(
964        &self,
965        req: &AwsRequest,
966        arn: &str,
967    ) -> Result<AwsResponse, AwsServiceError> {
968        validate_eks_arn(arn)?;
969        let accounts = self.state.read();
970        let tags: serde_json::Map<String, Value> = accounts
971            .get(&req.account_id)
972            .map(|state| {
973                let target = locate_tag_target(state, arn);
974                tags_ref(state, &target)
975                    .map(|t| {
976                        t.iter()
977                            .map(|(k, v)| (k.clone(), Value::String(v.clone())))
978                            .collect()
979                    })
980                    .unwrap_or_default()
981            })
982            .unwrap_or_default();
983        Ok(AwsResponse::json(
984            StatusCode::OK,
985            json!({ "tags": tags }).to_string(),
986        ))
987    }
988
989    // -----------------------------------------------------------------------
990    // Node groups
991    // -----------------------------------------------------------------------
992
993    fn create_nodegroup(
994        &self,
995        req: &AwsRequest,
996        cluster_name: &str,
997    ) -> Result<AwsResponse, AwsServiceError> {
998        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
999        let name = body
1000            .get("nodegroupName")
1001            .and_then(|v| v.as_str())
1002            .ok_or_else(|| invalid_parameter("nodegroupName is required"))?
1003            .to_string();
1004        let node_role = body
1005            .get("nodeRole")
1006            .and_then(|v| v.as_str())
1007            .ok_or_else(|| invalid_parameter("nodeRole is required"))?
1008            .to_string();
1009        let subnets = body
1010            .get("subnets")
1011            .filter(|v| v.is_array())
1012            .cloned()
1013            .ok_or_else(|| invalid_parameter("subnets is required"))?;
1014
1015        let region = req.region.clone();
1016        let account_id = req.account_id.clone();
1017
1018        let mut accounts = self.state.write();
1019        let state = accounts.get_or_create(&req.account_id);
1020        // The parent cluster must exist. AWS returns ResourceNotFoundException
1021        // here even though the Smithy model omits it from CreateNodegroup's
1022        // declared errors.
1023        let cluster_version = state
1024            .clusters
1025            .get(cluster_name)
1026            .ok_or_else(not_found_cluster(cluster_name))?
1027            .version
1028            .clone();
1029        if state
1030            .nodegroups
1031            .get(cluster_name)
1032            .is_some_and(|m| m.contains_key(&name))
1033        {
1034            return Err(AwsServiceError::aws_error(
1035                StatusCode::CONFLICT,
1036                "ResourceInUseException",
1037                format!(
1038                    "NodeGroup already exists with name {name} and cluster name {cluster_name}"
1039                ),
1040            ));
1041        }
1042
1043        let id = uuid::Uuid::new_v4().to_string();
1044        let arn = nodegroup_arn(&region, &account_id, cluster_name, &name, &id);
1045        let now = Utc::now();
1046        let version = body
1047            .get("version")
1048            .and_then(|v| v.as_str())
1049            .map(|s| s.to_string())
1050            .unwrap_or(cluster_version);
1051        let release_version = body
1052            .get("releaseVersion")
1053            .and_then(|v| v.as_str())
1054            .map(|s| s.to_string())
1055            .unwrap_or_else(|| format!("{version}-20240000"));
1056        let capacity_type = body
1057            .get("capacityType")
1058            .and_then(|v| v.as_str())
1059            .unwrap_or("ON_DEMAND")
1060            .to_string();
1061        let ami_type = body
1062            .get("amiType")
1063            .and_then(|v| v.as_str())
1064            .unwrap_or("AL2023_x86_64_STANDARD")
1065            .to_string();
1066        let disk_size = body.get("diskSize").and_then(|v| v.as_i64()).unwrap_or(20);
1067
1068        let ng = Nodegroup {
1069            name: name.clone(),
1070            arn,
1071            cluster_name: cluster_name.to_string(),
1072            version,
1073            release_version,
1074            status: "CREATING".to_string(),
1075            capacity_type,
1076            ami_type,
1077            node_role,
1078            created_at: now,
1079            modified_at: now,
1080            disk_size,
1081            scaling_config: build_scaling_config(body.get("scalingConfig")),
1082            update_config: build_nodegroup_update_config(body.get("updateConfig")),
1083            instance_types: body
1084                .get("instanceTypes")
1085                .cloned()
1086                .unwrap_or_else(|| json!(["t3.medium"])),
1087            subnets,
1088            labels: body.get("labels").cloned().unwrap_or_else(|| json!({})),
1089            taints: body.get("taints").cloned().unwrap_or_else(|| json!([])),
1090            remote_access: body.get("remoteAccess").cloned(),
1091            launch_template: body.get("launchTemplate").cloned(),
1092            asg_name: format!("eks-{name}-{}", &id[..8]),
1093            tags: parse_tag_map(body.get("tags")),
1094            updates: Default::default(),
1095        };
1096
1097        let out = nodegroup_json(&ng);
1098        state
1099            .nodegroups
1100            .entry(cluster_name.to_string())
1101            .or_default()
1102            .insert(name, ng);
1103        Ok(AwsResponse::json(
1104            StatusCode::OK,
1105            json!({ "nodegroup": out }).to_string(),
1106        ))
1107    }
1108
1109    fn describe_nodegroup(
1110        &self,
1111        req: &AwsRequest,
1112        cluster_name: &str,
1113        name: &str,
1114    ) -> Result<AwsResponse, AwsServiceError> {
1115        let mut accounts = self.state.write();
1116        let state = accounts.get_or_create(&req.account_id);
1117        if !state.clusters.contains_key(cluster_name) {
1118            return Err(not_found_cluster(cluster_name)());
1119        }
1120        let ng = state
1121            .nodegroups
1122            .get_mut(cluster_name)
1123            .and_then(|m| m.get_mut(name))
1124            .ok_or_else(not_found_nodegroup(name))?;
1125        if ng.status == "CREATING" {
1126            ng.status = "ACTIVE".to_string();
1127        }
1128        Ok(AwsResponse::json(
1129            StatusCode::OK,
1130            json!({ "nodegroup": nodegroup_json(ng) }).to_string(),
1131        ))
1132    }
1133
1134    fn list_nodegroups(
1135        &self,
1136        req: &AwsRequest,
1137        cluster_name: &str,
1138    ) -> Result<AwsResponse, AwsServiceError> {
1139        let max_results = validate_max_results(req)?;
1140        let next_token = req.query_params.get("nextToken").cloned();
1141        let accounts = self.state.read();
1142        let state = accounts
1143            .get(&req.account_id)
1144            .ok_or_else(not_found_cluster(cluster_name))?;
1145        if !state.clusters.contains_key(cluster_name) {
1146            return Err(not_found_cluster(cluster_name)());
1147        }
1148        let names: Vec<String> = state
1149            .nodegroups
1150            .get(cluster_name)
1151            .map(|m| m.keys().cloned().collect())
1152            .unwrap_or_default();
1153        let (page, token) = paginate_checked(&names, next_token.as_deref(), max_results)
1154            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
1155        let mut out = json!({ "nodegroups": page });
1156        if let Some(t) = token {
1157            out["nextToken"] = Value::String(t);
1158        }
1159        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
1160    }
1161
1162    fn delete_nodegroup(
1163        &self,
1164        req: &AwsRequest,
1165        cluster_name: &str,
1166        name: &str,
1167    ) -> Result<AwsResponse, AwsServiceError> {
1168        let mut accounts = self.state.write();
1169        let state = accounts.get_or_create(&req.account_id);
1170        if !state.clusters.contains_key(cluster_name) {
1171            return Err(not_found_cluster(cluster_name)());
1172        }
1173        let mut ng = state
1174            .nodegroups
1175            .get_mut(cluster_name)
1176            .and_then(|m| m.remove(name))
1177            .ok_or_else(not_found_nodegroup(name))?;
1178        ng.status = "DELETING".to_string();
1179        Ok(AwsResponse::json(
1180            StatusCode::OK,
1181            json!({ "nodegroup": nodegroup_json(&ng) }).to_string(),
1182        ))
1183    }
1184
1185    fn update_nodegroup_config(
1186        &self,
1187        req: &AwsRequest,
1188        cluster_name: &str,
1189        name: &str,
1190    ) -> Result<AwsResponse, AwsServiceError> {
1191        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
1192        let mut accounts = self.state.write();
1193        let state = accounts.get_or_create(&req.account_id);
1194        if !state.clusters.contains_key(cluster_name) {
1195            return Err(not_found_cluster(cluster_name)());
1196        }
1197        let ng = state
1198            .nodegroups
1199            .get_mut(cluster_name)
1200            .and_then(|m| m.get_mut(name))
1201            .ok_or_else(not_found_nodegroup(name))?;
1202
1203        let mut params = Vec::new();
1204        if let Some(scaling) = body.get("scalingConfig") {
1205            ng.scaling_config = build_scaling_config(Some(scaling));
1206            params.push(("ScalingConfig".to_string(), scaling.to_string()));
1207        }
1208        if let Some(labels) = body.get("labels") {
1209            if !ng.labels.is_object() {
1210                ng.labels = json!({});
1211            }
1212            if let Some(map) = ng.labels.as_object_mut() {
1213                if let Some(add) = labels.get("addOrUpdateLabels").and_then(|v| v.as_object()) {
1214                    for (k, v) in add {
1215                        map.insert(k.clone(), v.clone());
1216                    }
1217                }
1218                if let Some(remove) = labels.get("removeLabels").and_then(|v| v.as_array()) {
1219                    for k in remove.iter().filter_map(|v| v.as_str()) {
1220                        map.remove(k);
1221                    }
1222                }
1223            }
1224            params.push(("LabelsToAdd".to_string(), labels.to_string()));
1225        }
1226        if let Some(taints) = body.get("taints") {
1227            if !ng.taints.is_array() {
1228                ng.taints = json!([]);
1229            }
1230            if let Some(arr) = ng.taints.as_array_mut() {
1231                // Remove first so an add+remove of the same key in one call keeps the add.
1232                if let Some(remove) = taints.get("removeTaints").and_then(|v| v.as_array()) {
1233                    for rt in remove {
1234                        let id = taint_identity(rt);
1235                        arr.retain(|t| taint_identity(t) != id);
1236                    }
1237                }
1238                if let Some(add) = taints.get("addOrUpdateTaints").and_then(|v| v.as_array()) {
1239                    for nt in add {
1240                        let id = taint_identity(nt);
1241                        if let Some(existing) = arr.iter_mut().find(|t| taint_identity(t) == id) {
1242                            *existing = nt.clone();
1243                        } else {
1244                            arr.push(nt.clone());
1245                        }
1246                    }
1247                }
1248            }
1249            params.push(("Taints".to_string(), taints.to_string()));
1250        }
1251        if let Some(update_config) = body.get("updateConfig") {
1252            ng.update_config = build_nodegroup_update_config(Some(update_config));
1253            params.push(("MaxUnavailable".to_string(), update_config.to_string()));
1254        }
1255        ng.modified_at = Utc::now();
1256
1257        let update = new_update("ConfigUpdate", params);
1258        let out = update_json(&update);
1259        ng.updates.insert(update.id.clone(), update);
1260        Ok(AwsResponse::json(
1261            StatusCode::OK,
1262            json!({ "update": out }).to_string(),
1263        ))
1264    }
1265
1266    fn update_nodegroup_version(
1267        &self,
1268        req: &AwsRequest,
1269        cluster_name: &str,
1270        name: &str,
1271    ) -> Result<AwsResponse, AwsServiceError> {
1272        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
1273        let mut accounts = self.state.write();
1274        let state = accounts.get_or_create(&req.account_id);
1275        if !state.clusters.contains_key(cluster_name) {
1276            return Err(not_found_cluster(cluster_name)());
1277        }
1278        let ng = state
1279            .nodegroups
1280            .get_mut(cluster_name)
1281            .and_then(|m| m.get_mut(name))
1282            .ok_or_else(not_found_nodegroup(name))?;
1283
1284        let mut params = Vec::new();
1285        if let Some(version) = body.get("version").and_then(|v| v.as_str()) {
1286            ng.version = version.to_string();
1287            params.push(("Version".to_string(), version.to_string()));
1288        }
1289        if let Some(release) = body.get("releaseVersion").and_then(|v| v.as_str()) {
1290            ng.release_version = release.to_string();
1291            params.push(("ReleaseVersion".to_string(), release.to_string()));
1292        }
1293        ng.modified_at = Utc::now();
1294
1295        let update = new_update("VersionUpdate", params);
1296        let out = update_json(&update);
1297        ng.updates.insert(update.id.clone(), update);
1298        Ok(AwsResponse::json(
1299            StatusCode::OK,
1300            json!({ "update": out }).to_string(),
1301        ))
1302    }
1303
1304    // -----------------------------------------------------------------------
1305    // Fargate profiles
1306    // -----------------------------------------------------------------------
1307
1308    fn create_fargate_profile(
1309        &self,
1310        req: &AwsRequest,
1311        cluster_name: &str,
1312    ) -> Result<AwsResponse, AwsServiceError> {
1313        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
1314        let name = body
1315            .get("fargateProfileName")
1316            .and_then(|v| v.as_str())
1317            .ok_or_else(|| invalid_parameter("fargateProfileName is required"))?
1318            .to_string();
1319        let pod_execution_role_arn = body
1320            .get("podExecutionRoleArn")
1321            .and_then(|v| v.as_str())
1322            .ok_or_else(|| invalid_parameter("podExecutionRoleArn is required"))?
1323            .to_string();
1324
1325        let region = req.region.clone();
1326        let account_id = req.account_id.clone();
1327
1328        let mut accounts = self.state.write();
1329        let state = accounts.get_or_create(&req.account_id);
1330        if !state.clusters.contains_key(cluster_name) {
1331            return Err(not_found_cluster(cluster_name)());
1332        }
1333        if state
1334            .fargate_profiles
1335            .get(cluster_name)
1336            .is_some_and(|m| m.contains_key(&name))
1337        {
1338            return Err(AwsServiceError::aws_error(
1339                StatusCode::CONFLICT,
1340                "ResourceInUseException",
1341                format!(
1342                    "FargateProfile already exists with name {name} and cluster name {cluster_name}"
1343                ),
1344            ));
1345        }
1346
1347        let id = uuid::Uuid::new_v4().to_string();
1348        let arn = fargate_profile_arn(&region, &account_id, cluster_name, &name, &id);
1349        let profile = FargateProfile {
1350            name: name.clone(),
1351            arn,
1352            cluster_name: cluster_name.to_string(),
1353            pod_execution_role_arn,
1354            status: "CREATING".to_string(),
1355            created_at: Utc::now(),
1356            subnets: body.get("subnets").cloned().unwrap_or_else(|| json!([])),
1357            selectors: body.get("selectors").cloned().unwrap_or_else(|| json!([])),
1358            tags: parse_tag_map(body.get("tags")),
1359        };
1360
1361        let out = fargate_profile_json(&profile);
1362        state
1363            .fargate_profiles
1364            .entry(cluster_name.to_string())
1365            .or_default()
1366            .insert(name, profile);
1367        Ok(AwsResponse::json(
1368            StatusCode::OK,
1369            json!({ "fargateProfile": out }).to_string(),
1370        ))
1371    }
1372
1373    fn describe_fargate_profile(
1374        &self,
1375        req: &AwsRequest,
1376        cluster_name: &str,
1377        name: &str,
1378    ) -> Result<AwsResponse, AwsServiceError> {
1379        let mut accounts = self.state.write();
1380        let state = accounts.get_or_create(&req.account_id);
1381        if !state.clusters.contains_key(cluster_name) {
1382            return Err(not_found_cluster(cluster_name)());
1383        }
1384        let profile = state
1385            .fargate_profiles
1386            .get_mut(cluster_name)
1387            .and_then(|m| m.get_mut(name))
1388            .ok_or_else(not_found_fargate_profile(name))?;
1389        if profile.status == "CREATING" {
1390            profile.status = "ACTIVE".to_string();
1391        }
1392        Ok(AwsResponse::json(
1393            StatusCode::OK,
1394            json!({ "fargateProfile": fargate_profile_json(profile) }).to_string(),
1395        ))
1396    }
1397
1398    fn list_fargate_profiles(
1399        &self,
1400        req: &AwsRequest,
1401        cluster_name: &str,
1402    ) -> Result<AwsResponse, AwsServiceError> {
1403        let max_results = validate_max_results(req)?;
1404        let next_token = req.query_params.get("nextToken").cloned();
1405        let accounts = self.state.read();
1406        let state = accounts
1407            .get(&req.account_id)
1408            .ok_or_else(not_found_cluster(cluster_name))?;
1409        if !state.clusters.contains_key(cluster_name) {
1410            return Err(not_found_cluster(cluster_name)());
1411        }
1412        let names: Vec<String> = state
1413            .fargate_profiles
1414            .get(cluster_name)
1415            .map(|m| m.keys().cloned().collect())
1416            .unwrap_or_default();
1417        let (page, token) = paginate_checked(&names, next_token.as_deref(), max_results)
1418            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
1419        let mut out = json!({ "fargateProfileNames": page });
1420        if let Some(t) = token {
1421            out["nextToken"] = Value::String(t);
1422        }
1423        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
1424    }
1425
1426    fn delete_fargate_profile(
1427        &self,
1428        req: &AwsRequest,
1429        cluster_name: &str,
1430        name: &str,
1431    ) -> Result<AwsResponse, AwsServiceError> {
1432        let mut accounts = self.state.write();
1433        let state = accounts.get_or_create(&req.account_id);
1434        if !state.clusters.contains_key(cluster_name) {
1435            return Err(not_found_cluster(cluster_name)());
1436        }
1437        let mut profile = state
1438            .fargate_profiles
1439            .get_mut(cluster_name)
1440            .and_then(|m| m.remove(name))
1441            .ok_or_else(not_found_fargate_profile(name))?;
1442        profile.status = "DELETING".to_string();
1443        Ok(AwsResponse::json(
1444            StatusCode::OK,
1445            json!({ "fargateProfile": fargate_profile_json(&profile) }).to_string(),
1446        ))
1447    }
1448
1449    // -----------------------------------------------------------------------
1450    // Add-ons
1451    // -----------------------------------------------------------------------
1452
1453    fn create_addon(
1454        &self,
1455        req: &AwsRequest,
1456        cluster_name: &str,
1457    ) -> Result<AwsResponse, AwsServiceError> {
1458        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
1459        let name = body
1460            .get("addonName")
1461            .and_then(|v| v.as_str())
1462            .ok_or_else(|| invalid_parameter("addonName is required"))?
1463            .to_string();
1464
1465        let region = req.region.clone();
1466        let account_id = req.account_id.clone();
1467
1468        let mut accounts = self.state.write();
1469        let state = accounts.get_or_create(&req.account_id);
1470        // The parent cluster must exist; AWS returns ResourceNotFoundException.
1471        let cluster_version = state
1472            .clusters
1473            .get(cluster_name)
1474            .ok_or_else(not_found_cluster(cluster_name))?
1475            .version
1476            .clone();
1477        if state
1478            .addons
1479            .get(cluster_name)
1480            .is_some_and(|m| m.contains_key(&name))
1481        {
1482            return Err(AwsServiceError::aws_error(
1483                StatusCode::CONFLICT,
1484                "ResourceInUseException",
1485                format!("Addon already exists with name {name} and cluster name {cluster_name}"),
1486            ));
1487        }
1488
1489        let id = uuid::Uuid::new_v4().to_string();
1490        let arn = addon_arn(&region, &account_id, cluster_name, &name, &id);
1491        let now = Utc::now();
1492        let addon_version = body
1493            .get("addonVersion")
1494            .and_then(|v| v.as_str())
1495            .map(|s| s.to_string())
1496            .unwrap_or_else(|| default_addon_version(&name, &cluster_version));
1497        let namespace = body
1498            .get("namespaceConfig")
1499            .and_then(|v| v.get("namespace"))
1500            .and_then(|v| v.as_str())
1501            .map(|s| s.to_string());
1502        let pod_identity_associations = build_pod_identity_association_arns(
1503            &region,
1504            &account_id,
1505            cluster_name,
1506            body.get("podIdentityAssociations"),
1507        );
1508
1509        let addon = Addon {
1510            name: name.clone(),
1511            arn,
1512            cluster_name: cluster_name.to_string(),
1513            addon_version,
1514            status: "CREATING".to_string(),
1515            created_at: now,
1516            modified_at: now,
1517            service_account_role_arn: body
1518                .get("serviceAccountRoleArn")
1519                .and_then(|v| v.as_str())
1520                .map(|s| s.to_string()),
1521            configuration_values: body
1522                .get("configurationValues")
1523                .and_then(|v| v.as_str())
1524                .map(|s| s.to_string()),
1525            namespace,
1526            pod_identity_associations,
1527            tags: parse_tag_map(body.get("tags")),
1528            updates: Default::default(),
1529        };
1530
1531        let out = addon_json(&addon);
1532        state
1533            .addons
1534            .entry(cluster_name.to_string())
1535            .or_default()
1536            .insert(name, addon);
1537        Ok(AwsResponse::json(
1538            StatusCode::OK,
1539            json!({ "addon": out }).to_string(),
1540        ))
1541    }
1542
1543    fn describe_addon(
1544        &self,
1545        req: &AwsRequest,
1546        cluster_name: &str,
1547        name: &str,
1548    ) -> Result<AwsResponse, AwsServiceError> {
1549        let mut accounts = self.state.write();
1550        let state = accounts.get_or_create(&req.account_id);
1551        if !state.clusters.contains_key(cluster_name) {
1552            return Err(not_found_cluster(cluster_name)());
1553        }
1554        let addon = state
1555            .addons
1556            .get_mut(cluster_name)
1557            .and_then(|m| m.get_mut(name))
1558            .ok_or_else(not_found_addon(name))?;
1559        // Settle CREATING -> ACTIVE on first describe.
1560        if addon.status == "CREATING" {
1561            addon.status = "ACTIVE".to_string();
1562        }
1563        Ok(AwsResponse::json(
1564            StatusCode::OK,
1565            json!({ "addon": addon_json(addon) }).to_string(),
1566        ))
1567    }
1568
1569    fn list_addons(
1570        &self,
1571        req: &AwsRequest,
1572        cluster_name: &str,
1573    ) -> Result<AwsResponse, AwsServiceError> {
1574        let max_results = validate_max_results(req)?;
1575        let next_token = req.query_params.get("nextToken").cloned();
1576        let accounts = self.state.read();
1577        let state = accounts
1578            .get(&req.account_id)
1579            .ok_or_else(not_found_cluster(cluster_name))?;
1580        if !state.clusters.contains_key(cluster_name) {
1581            return Err(not_found_cluster(cluster_name)());
1582        }
1583        let names: Vec<String> = state
1584            .addons
1585            .get(cluster_name)
1586            .map(|m| m.keys().cloned().collect())
1587            .unwrap_or_default();
1588        let (page, token) = paginate_checked(&names, next_token.as_deref(), max_results)
1589            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
1590        let mut out = json!({ "addons": page });
1591        if let Some(t) = token {
1592            out["nextToken"] = Value::String(t);
1593        }
1594        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
1595    }
1596
1597    fn delete_addon(
1598        &self,
1599        req: &AwsRequest,
1600        cluster_name: &str,
1601        name: &str,
1602    ) -> Result<AwsResponse, AwsServiceError> {
1603        let mut accounts = self.state.write();
1604        let state = accounts.get_or_create(&req.account_id);
1605        if !state.clusters.contains_key(cluster_name) {
1606            return Err(not_found_cluster(cluster_name)());
1607        }
1608        let mut addon = state
1609            .addons
1610            .get_mut(cluster_name)
1611            .and_then(|m| m.remove(name))
1612            .ok_or_else(not_found_addon(name))?;
1613        addon.status = "DELETING".to_string();
1614        Ok(AwsResponse::json(
1615            StatusCode::OK,
1616            json!({ "addon": addon_json(&addon) }).to_string(),
1617        ))
1618    }
1619
1620    fn update_addon(
1621        &self,
1622        req: &AwsRequest,
1623        cluster_name: &str,
1624        name: &str,
1625    ) -> Result<AwsResponse, AwsServiceError> {
1626        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
1627        let region = req.region.clone();
1628        let account_id = req.account_id.clone();
1629        let mut accounts = self.state.write();
1630        let state = accounts.get_or_create(&req.account_id);
1631        if !state.clusters.contains_key(cluster_name) {
1632            return Err(not_found_cluster(cluster_name)());
1633        }
1634        let addon = state
1635            .addons
1636            .get_mut(cluster_name)
1637            .and_then(|m| m.get_mut(name))
1638            .ok_or_else(not_found_addon(name))?;
1639
1640        let mut params = Vec::new();
1641        if let Some(version) = body.get("addonVersion").and_then(|v| v.as_str()) {
1642            addon.addon_version = version.to_string();
1643            params.push(("AddonVersion".to_string(), version.to_string()));
1644        }
1645        if let Some(role) = body.get("serviceAccountRoleArn").and_then(|v| v.as_str()) {
1646            addon.service_account_role_arn = Some(role.to_string());
1647            params.push(("ServiceAccountRoleArn".to_string(), role.to_string()));
1648        }
1649        if let Some(cfg) = body.get("configurationValues").and_then(|v| v.as_str()) {
1650            addon.configuration_values = Some(cfg.to_string());
1651            params.push(("ConfigurationValues".to_string(), cfg.to_string()));
1652        }
1653        if let Some(resolve) = body.get("resolveConflicts").and_then(|v| v.as_str()) {
1654            params.push(("ResolveConflicts".to_string(), resolve.to_string()));
1655        }
1656        if let Some(assocs) = body.get("podIdentityAssociations") {
1657            addon.pod_identity_associations = build_pod_identity_association_arns(
1658                &region,
1659                &account_id,
1660                cluster_name,
1661                Some(assocs),
1662            );
1663        }
1664        addon.modified_at = Utc::now();
1665
1666        let update = new_update("AddonUpdate", params);
1667        let out = update_json(&update);
1668        addon.updates.insert(update.id.clone(), update);
1669        Ok(AwsResponse::json(
1670            StatusCode::OK,
1671            json!({ "update": out }).to_string(),
1672        ))
1673    }
1674
1675    fn describe_addon_versions(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
1676        let max_results = validate_max_results(req)?;
1677        let next_token = req.query_params.get("nextToken").cloned();
1678        let addon_filter = req.query_params.get("addonName").cloned();
1679        let k8s_version = req
1680            .query_params
1681            .get("kubernetesVersion")
1682            .cloned()
1683            .unwrap_or_else(|| DEFAULT_K8S_VERSION.to_string());
1684
1685        let catalog = addon_catalog(&k8s_version);
1686        let filtered: Vec<Value> = catalog
1687            .into_iter()
1688            .filter(|a| addon_filter.as_deref().is_none_or(|f| a["addonName"] == f))
1689            .collect();
1690
1691        let (page, token) = paginate_checked(&filtered, next_token.as_deref(), max_results)
1692            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
1693        let mut out = json!({ "addons": page });
1694        if let Some(t) = token {
1695            out["nextToken"] = Value::String(t);
1696        }
1697        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
1698    }
1699
1700    fn describe_addon_configuration(
1701        &self,
1702        req: &AwsRequest,
1703    ) -> Result<AwsResponse, AwsServiceError> {
1704        let addon_name = req
1705            .query_params
1706            .get("addonName")
1707            .cloned()
1708            .ok_or_else(|| invalid_parameter("addonName is required"))?;
1709        let addon_version = req
1710            .query_params
1711            .get("addonVersion")
1712            .cloned()
1713            .ok_or_else(|| invalid_parameter("addonVersion is required"))?;
1714
1715        Ok(AwsResponse::json(
1716            StatusCode::OK,
1717            json!({
1718                "addonName": addon_name,
1719                "addonVersion": addon_version,
1720                "configurationSchema": addon_configuration_schema(&addon_name),
1721                "podIdentityConfiguration": pod_identity_configuration(&addon_name),
1722            })
1723            .to_string(),
1724        ))
1725    }
1726
1727    // -----------------------------------------------------------------------
1728    // Access entries / access policies
1729    // -----------------------------------------------------------------------
1730
1731    fn create_access_entry(
1732        &self,
1733        req: &AwsRequest,
1734        cluster_name: &str,
1735    ) -> Result<AwsResponse, AwsServiceError> {
1736        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
1737        let principal_arn = body
1738            .get("principalArn")
1739            .and_then(|v| v.as_str())
1740            .ok_or_else(|| invalid_parameter("principalArn is required"))?
1741            .to_string();
1742
1743        let region = req.region.clone();
1744        let account_id = req.account_id.clone();
1745
1746        let mut accounts = self.state.write();
1747        let state = accounts.get_or_create(&req.account_id);
1748        // The parent cluster must exist; AWS returns ResourceNotFoundException.
1749        if !state.clusters.contains_key(cluster_name) {
1750            return Err(not_found_cluster(cluster_name)());
1751        }
1752        if state
1753            .access_entries
1754            .get(cluster_name)
1755            .is_some_and(|m| m.contains_key(&principal_arn))
1756        {
1757            return Err(AwsServiceError::aws_error(
1758                StatusCode::CONFLICT,
1759                "ResourceInUseException",
1760                format!(
1761                    "The specified access entry resource is already in use on this cluster: {principal_arn}"
1762                ),
1763            ));
1764        }
1765
1766        let (principal_type, principal_name) = principal_parts(&principal_arn);
1767        let id = uuid::Uuid::new_v4().to_string();
1768        let arn = access_entry_arn(
1769            &region,
1770            &account_id,
1771            cluster_name,
1772            &principal_type,
1773            &principal_name,
1774            &id,
1775        );
1776        let now = Utc::now();
1777        let username = body
1778            .get("username")
1779            .and_then(|v| v.as_str())
1780            .map(|s| s.to_string())
1781            .unwrap_or_else(|| default_username(&principal_arn));
1782        let entry_type = body
1783            .get("type")
1784            .and_then(|v| v.as_str())
1785            .unwrap_or("STANDARD")
1786            .to_string();
1787        let kubernetes_groups = string_list(body.get("kubernetesGroups"));
1788
1789        let entry = AccessEntry {
1790            principal_arn: principal_arn.clone(),
1791            cluster_name: cluster_name.to_string(),
1792            arn,
1793            kubernetes_groups,
1794            username,
1795            type_: entry_type,
1796            created_at: now,
1797            modified_at: now,
1798            tags: parse_tag_map(body.get("tags")),
1799            associated_policies: Vec::new(),
1800        };
1801
1802        let out = access_entry_json(&entry);
1803        state
1804            .access_entries
1805            .entry(cluster_name.to_string())
1806            .or_default()
1807            .insert(principal_arn, entry);
1808        Ok(AwsResponse::json(
1809            StatusCode::OK,
1810            json!({ "accessEntry": out }).to_string(),
1811        ))
1812    }
1813
1814    fn describe_access_entry(
1815        &self,
1816        req: &AwsRequest,
1817        cluster_name: &str,
1818        principal_arn: &str,
1819    ) -> Result<AwsResponse, AwsServiceError> {
1820        let accounts = self.state.read();
1821        let state = accounts
1822            .get(&req.account_id)
1823            .ok_or_else(not_found_cluster(cluster_name))?;
1824        if !state.clusters.contains_key(cluster_name) {
1825            return Err(not_found_cluster(cluster_name)());
1826        }
1827        let entry = state
1828            .access_entries
1829            .get(cluster_name)
1830            .and_then(|m| m.get(principal_arn))
1831            .ok_or_else(not_found_access_entry(principal_arn))?;
1832        Ok(AwsResponse::json(
1833            StatusCode::OK,
1834            json!({ "accessEntry": access_entry_json(entry) }).to_string(),
1835        ))
1836    }
1837
1838    fn list_access_entries(
1839        &self,
1840        req: &AwsRequest,
1841        cluster_name: &str,
1842    ) -> Result<AwsResponse, AwsServiceError> {
1843        let max_results = validate_max_results(req)?;
1844        let next_token = req.query_params.get("nextToken").cloned();
1845        let associated_policy = req.query_params.get("associatedPolicyArn").cloned();
1846        let accounts = self.state.read();
1847        let state = accounts
1848            .get(&req.account_id)
1849            .ok_or_else(not_found_cluster(cluster_name))?;
1850        if !state.clusters.contains_key(cluster_name) {
1851            return Err(not_found_cluster(cluster_name)());
1852        }
1853        // Optional `associatedPolicyArn` filter narrows the result to entries
1854        // that have that policy associated, mirroring the real API.
1855        let names: Vec<String> = state
1856            .access_entries
1857            .get(cluster_name)
1858            .map(|m| {
1859                m.values()
1860                    .filter(|e| {
1861                        associated_policy.as_deref().is_none_or(|p| {
1862                            e.associated_policies.iter().any(|ap| ap.policy_arn == p)
1863                        })
1864                    })
1865                    .map(|e| e.principal_arn.clone())
1866                    .collect()
1867            })
1868            .unwrap_or_default();
1869        let (page, token) = paginate_checked(&names, next_token.as_deref(), max_results)
1870            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
1871        let mut out = json!({ "accessEntries": page });
1872        if let Some(t) = token {
1873            out["nextToken"] = Value::String(t);
1874        }
1875        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
1876    }
1877
1878    fn delete_access_entry(
1879        &self,
1880        req: &AwsRequest,
1881        cluster_name: &str,
1882        principal_arn: &str,
1883    ) -> Result<AwsResponse, AwsServiceError> {
1884        let mut accounts = self.state.write();
1885        let state = accounts.get_or_create(&req.account_id);
1886        if !state.clusters.contains_key(cluster_name) {
1887            return Err(not_found_cluster(cluster_name)());
1888        }
1889        state
1890            .access_entries
1891            .get_mut(cluster_name)
1892            .and_then(|m| m.remove(principal_arn))
1893            .ok_or_else(not_found_access_entry(principal_arn))?;
1894        Ok(AwsResponse::json(StatusCode::OK, "{}"))
1895    }
1896
1897    fn update_access_entry(
1898        &self,
1899        req: &AwsRequest,
1900        cluster_name: &str,
1901        principal_arn: &str,
1902    ) -> Result<AwsResponse, AwsServiceError> {
1903        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
1904        let mut accounts = self.state.write();
1905        let state = accounts.get_or_create(&req.account_id);
1906        if !state.clusters.contains_key(cluster_name) {
1907            return Err(not_found_cluster(cluster_name)());
1908        }
1909        let entry = state
1910            .access_entries
1911            .get_mut(cluster_name)
1912            .and_then(|m| m.get_mut(principal_arn))
1913            .ok_or_else(not_found_access_entry(principal_arn))?;
1914
1915        if let Some(groups) = body.get("kubernetesGroups") {
1916            entry.kubernetes_groups = string_list(Some(groups));
1917        }
1918        if let Some(username) = body.get("username").and_then(|v| v.as_str()) {
1919            entry.username = username.to_string();
1920        }
1921        entry.modified_at = Utc::now();
1922        Ok(AwsResponse::json(
1923            StatusCode::OK,
1924            json!({ "accessEntry": access_entry_json(entry) }).to_string(),
1925        ))
1926    }
1927
1928    fn associate_access_policy(
1929        &self,
1930        req: &AwsRequest,
1931        cluster_name: &str,
1932        principal_arn: &str,
1933    ) -> Result<AwsResponse, AwsServiceError> {
1934        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
1935        let policy_arn = body
1936            .get("policyArn")
1937            .and_then(|v| v.as_str())
1938            .ok_or_else(|| invalid_parameter("policyArn is required"))?
1939            .to_string();
1940        let access_scope = build_access_scope(body.get("accessScope"));
1941
1942        let mut accounts = self.state.write();
1943        let state = accounts.get_or_create(&req.account_id);
1944        if !state.clusters.contains_key(cluster_name) {
1945            return Err(not_found_cluster(cluster_name)());
1946        }
1947        let entry = state
1948            .access_entries
1949            .get_mut(cluster_name)
1950            .and_then(|m| m.get_mut(principal_arn))
1951            .ok_or_else(not_found_access_entry(principal_arn))?;
1952
1953        let now = Utc::now();
1954        // Re-associating the same policy updates its scope in place, matching
1955        // the real API's idempotent upsert semantics.
1956        let associated = if let Some(existing) = entry
1957            .associated_policies
1958            .iter_mut()
1959            .find(|ap| ap.policy_arn == policy_arn)
1960        {
1961            existing.access_scope = access_scope;
1962            existing.modified_at = now;
1963            existing.clone()
1964        } else {
1965            let ap = AssociatedPolicy {
1966                policy_arn: policy_arn.clone(),
1967                access_scope,
1968                associated_at: now,
1969                modified_at: now,
1970            };
1971            entry.associated_policies.push(ap.clone());
1972            ap
1973        };
1974        Ok(AwsResponse::json(
1975            StatusCode::OK,
1976            json!({
1977                "clusterName": cluster_name,
1978                "principalArn": entry.principal_arn,
1979                "associatedAccessPolicy": associated_policy_json(&associated),
1980            })
1981            .to_string(),
1982        ))
1983    }
1984
1985    fn disassociate_access_policy(
1986        &self,
1987        req: &AwsRequest,
1988        cluster_name: &str,
1989        principal_arn: &str,
1990        policy_arn: &str,
1991    ) -> Result<AwsResponse, AwsServiceError> {
1992        let mut accounts = self.state.write();
1993        let state = accounts.get_or_create(&req.account_id);
1994        if !state.clusters.contains_key(cluster_name) {
1995            return Err(not_found_cluster(cluster_name)());
1996        }
1997        let entry = state
1998            .access_entries
1999            .get_mut(cluster_name)
2000            .and_then(|m| m.get_mut(principal_arn))
2001            .ok_or_else(not_found_access_entry(principal_arn))?;
2002        entry
2003            .associated_policies
2004            .retain(|ap| ap.policy_arn != policy_arn);
2005        Ok(AwsResponse::json(StatusCode::OK, "{}"))
2006    }
2007
2008    fn list_associated_access_policies(
2009        &self,
2010        req: &AwsRequest,
2011        cluster_name: &str,
2012        principal_arn: &str,
2013    ) -> Result<AwsResponse, AwsServiceError> {
2014        let max_results = validate_max_results(req)?;
2015        let next_token = req.query_params.get("nextToken").cloned();
2016        let accounts = self.state.read();
2017        let state = accounts
2018            .get(&req.account_id)
2019            .ok_or_else(not_found_cluster(cluster_name))?;
2020        if !state.clusters.contains_key(cluster_name) {
2021            return Err(not_found_cluster(cluster_name)());
2022        }
2023        let entry = state
2024            .access_entries
2025            .get(cluster_name)
2026            .and_then(|m| m.get(principal_arn))
2027            .ok_or_else(not_found_access_entry(principal_arn))?;
2028        let policies: Vec<Value> = entry
2029            .associated_policies
2030            .iter()
2031            .map(associated_policy_json)
2032            .collect();
2033        let (page, token) = paginate_checked(&policies, next_token.as_deref(), max_results)
2034            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
2035        let mut out = json!({
2036            "clusterName": cluster_name,
2037            "principalArn": entry.principal_arn,
2038            "associatedAccessPolicies": page,
2039        });
2040        if let Some(t) = token {
2041            out["nextToken"] = Value::String(t);
2042        }
2043        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
2044    }
2045
2046    fn list_access_policies(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
2047        // ListAccessPolicies is a read-only catalogue. Its `maxResults` carries
2048        // an `@range` (1-100) constraint the probe exercises, so an out-of-range
2049        // value is rejected. The nextToken, however, is tolerated (an
2050        // unparseable one restarts from the beginning) since the op's Smithy
2051        // contract declares no client error for it.
2052        let max_results = validate_max_results(req)?;
2053        let next_token = req.query_params.get("nextToken").cloned();
2054        let catalog = access_policy_catalog();
2055        let (page, token) = paginate_checked(&catalog, next_token.as_deref(), max_results)
2056            .unwrap_or_else(|_| (catalog.clone(), None));
2057        let mut out = json!({ "accessPolicies": page });
2058        if let Some(t) = token {
2059            out["nextToken"] = Value::String(t);
2060        }
2061        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
2062    }
2063
2064    // -----------------------------------------------------------------------
2065    // Identity-provider configs
2066    // -----------------------------------------------------------------------
2067
2068    fn associate_identity_provider_config(
2069        &self,
2070        req: &AwsRequest,
2071        cluster_name: &str,
2072    ) -> Result<AwsResponse, AwsServiceError> {
2073        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2074        let oidc = body
2075            .get("oidc")
2076            .filter(|v| v.is_object())
2077            .ok_or_else(|| invalid_parameter("oidc is required"))?;
2078        let name = oidc
2079            .get("identityProviderConfigName")
2080            .and_then(|v| v.as_str())
2081            .ok_or_else(|| invalid_parameter("oidc.identityProviderConfigName is required"))?
2082            .to_string();
2083        let issuer_url = oidc
2084            .get("issuerUrl")
2085            .and_then(|v| v.as_str())
2086            .ok_or_else(|| invalid_parameter("oidc.issuerUrl is required"))?
2087            .to_string();
2088        let client_id = oidc
2089            .get("clientId")
2090            .and_then(|v| v.as_str())
2091            .ok_or_else(|| invalid_parameter("oidc.clientId is required"))?
2092            .to_string();
2093
2094        let region = req.region.clone();
2095        let account_id = req.account_id.clone();
2096
2097        let mut accounts = self.state.write();
2098        let state = accounts.get_or_create(&req.account_id);
2099        // The parent cluster must exist; AWS returns ResourceNotFoundException.
2100        if !state.clusters.contains_key(cluster_name) {
2101            return Err(not_found_cluster(cluster_name)());
2102        }
2103        if state
2104            .identity_provider_configs
2105            .get(cluster_name)
2106            .is_some_and(|m| m.contains_key(&name))
2107        {
2108            return Err(AwsServiceError::aws_error(
2109                StatusCode::CONFLICT,
2110                "ResourceInUseException",
2111                format!(
2112                    "Identity provider config already exists with name {name} and cluster name {cluster_name}"
2113                ),
2114            ));
2115        }
2116
2117        let id = uuid::Uuid::new_v4().to_string();
2118        let arn = identity_provider_config_arn(&region, &account_id, cluster_name, &name, &id);
2119        let config = IdentityProviderConfig {
2120            name: name.clone(),
2121            arn,
2122            cluster_name: cluster_name.to_string(),
2123            issuer_url,
2124            client_id,
2125            username_claim: str_field(oidc, "usernameClaim"),
2126            username_prefix: str_field(oidc, "usernamePrefix"),
2127            groups_claim: str_field(oidc, "groupsClaim"),
2128            groups_prefix: str_field(oidc, "groupsPrefix"),
2129            required_claims: oidc
2130                .get("requiredClaims")
2131                .cloned()
2132                .unwrap_or_else(|| json!({})),
2133            status: "CREATING".to_string(),
2134            tags: parse_tag_map(body.get("tags")),
2135        };
2136        state
2137            .identity_provider_configs
2138            .entry(cluster_name.to_string())
2139            .or_default()
2140            .insert(name, config.clone());
2141
2142        // Associating an IdP config mints a tracked cluster-scoped Update, read
2143        // back through DescribeUpdate/ListUpdates like other config changes.
2144        let cluster = state.clusters.get_mut(cluster_name).unwrap();
2145        let update = new_update(
2146            "AssociateIdentityProviderConfig",
2147            vec![("IdentityProviderConfig".to_string(), oidc.to_string())],
2148        );
2149        let update_out = update_json(&update);
2150        cluster.updates.insert(update.id.clone(), update);
2151        Ok(AwsResponse::json(
2152            StatusCode::OK,
2153            json!({ "update": update_out, "tags": config.tags }).to_string(),
2154        ))
2155    }
2156
2157    fn disassociate_identity_provider_config(
2158        &self,
2159        req: &AwsRequest,
2160        cluster_name: &str,
2161    ) -> Result<AwsResponse, AwsServiceError> {
2162        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2163        let name = body
2164            .get("identityProviderConfig")
2165            .and_then(|v| v.get("name"))
2166            .and_then(|v| v.as_str())
2167            .ok_or_else(|| invalid_parameter("identityProviderConfig.name is required"))?
2168            .to_string();
2169
2170        let mut accounts = self.state.write();
2171        let state = accounts.get_or_create(&req.account_id);
2172        if !state.clusters.contains_key(cluster_name) {
2173            return Err(not_found_cluster(cluster_name)());
2174        }
2175        state
2176            .identity_provider_configs
2177            .get_mut(cluster_name)
2178            .and_then(|m| m.remove(&name))
2179            .ok_or_else(not_found_identity_provider_config(&name))?;
2180
2181        let cluster = state.clusters.get_mut(cluster_name).unwrap();
2182        let update = new_update(
2183            "DisassociateIdentityProviderConfig",
2184            vec![("IdentityProviderConfig".to_string(), name)],
2185        );
2186        let update_out = update_json(&update);
2187        cluster.updates.insert(update.id.clone(), update);
2188        Ok(AwsResponse::json(
2189            StatusCode::OK,
2190            json!({ "update": update_out }).to_string(),
2191        ))
2192    }
2193
2194    fn describe_identity_provider_config(
2195        &self,
2196        req: &AwsRequest,
2197        cluster_name: &str,
2198    ) -> Result<AwsResponse, AwsServiceError> {
2199        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2200        let name = body
2201            .get("identityProviderConfig")
2202            .and_then(|v| v.get("name"))
2203            .and_then(|v| v.as_str())
2204            .ok_or_else(|| invalid_parameter("identityProviderConfig.name is required"))?
2205            .to_string();
2206
2207        let mut accounts = self.state.write();
2208        let state = accounts.get_or_create(&req.account_id);
2209        if !state.clusters.contains_key(cluster_name) {
2210            return Err(not_found_cluster(cluster_name)());
2211        }
2212        let config = state
2213            .identity_provider_configs
2214            .get_mut(cluster_name)
2215            .and_then(|m| m.get_mut(&name))
2216            .ok_or_else(not_found_identity_provider_config(&name))?;
2217        // Settle CREATING -> ACTIVE on first describe.
2218        if config.status == "CREATING" {
2219            config.status = "ACTIVE".to_string();
2220        }
2221        Ok(AwsResponse::json(
2222            StatusCode::OK,
2223            json!({ "identityProviderConfig": identity_provider_config_json(config) }).to_string(),
2224        ))
2225    }
2226
2227    fn list_identity_provider_configs(
2228        &self,
2229        req: &AwsRequest,
2230        cluster_name: &str,
2231    ) -> Result<AwsResponse, AwsServiceError> {
2232        let max_results = validate_max_results(req)?;
2233        let next_token = req.query_params.get("nextToken").cloned();
2234        let accounts = self.state.read();
2235        let state = accounts
2236            .get(&req.account_id)
2237            .ok_or_else(not_found_cluster(cluster_name))?;
2238        if !state.clusters.contains_key(cluster_name) {
2239            return Err(not_found_cluster(cluster_name)());
2240        }
2241        let configs: Vec<Value> = state
2242            .identity_provider_configs
2243            .get(cluster_name)
2244            .map(|m| {
2245                m.keys()
2246                    .map(|n| json!({ "type": "oidc", "name": n }))
2247                    .collect()
2248            })
2249            .unwrap_or_default();
2250        let (page, token) = paginate_checked(&configs, next_token.as_deref(), max_results)
2251            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
2252        let mut out = json!({ "identityProviderConfigs": page });
2253        if let Some(t) = token {
2254            out["nextToken"] = Value::String(t);
2255        }
2256        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
2257    }
2258
2259    // -----------------------------------------------------------------------
2260    // Pod identity associations
2261    // -----------------------------------------------------------------------
2262
2263    fn create_pod_identity_association(
2264        &self,
2265        req: &AwsRequest,
2266        cluster_name: &str,
2267    ) -> Result<AwsResponse, AwsServiceError> {
2268        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2269        let namespace = body
2270            .get("namespace")
2271            .and_then(|v| v.as_str())
2272            .ok_or_else(|| invalid_parameter("namespace is required"))?
2273            .to_string();
2274        let service_account = body
2275            .get("serviceAccount")
2276            .and_then(|v| v.as_str())
2277            .ok_or_else(|| invalid_parameter("serviceAccount is required"))?
2278            .to_string();
2279        let role_arn = body
2280            .get("roleArn")
2281            .and_then(|v| v.as_str())
2282            .ok_or_else(|| invalid_parameter("roleArn is required"))?
2283            .to_string();
2284
2285        let region = req.region.clone();
2286        let account_id = req.account_id.clone();
2287
2288        let mut accounts = self.state.write();
2289        let state = accounts.get_or_create(&req.account_id);
2290        // The parent cluster must exist; AWS returns ResourceNotFoundException.
2291        if !state.clusters.contains_key(cluster_name) {
2292            return Err(not_found_cluster(cluster_name)());
2293        }
2294        // A given (namespace, serviceAccount) can only be associated once per
2295        // cluster; AWS returns ResourceInUseException for a duplicate.
2296        if state
2297            .pod_identity_associations
2298            .get(cluster_name)
2299            .is_some_and(|m| {
2300                m.values()
2301                    .any(|a| a.namespace == namespace && a.service_account == service_account)
2302            })
2303        {
2304            return Err(AwsServiceError::aws_error(
2305                StatusCode::CONFLICT,
2306                "ResourceInUseException",
2307                format!(
2308                    "Association already exists for namespace {namespace} and service account {service_account}"
2309                ),
2310            ));
2311        }
2312
2313        let suffix = uuid::Uuid::new_v4().to_string().replace('-', "");
2314        let suffix = &suffix[..17.min(suffix.len())];
2315        let association_id = format!("a-{suffix}");
2316        let association_arn =
2317            pod_identity_association_arn(&region, &account_id, cluster_name, suffix);
2318        let target_role_arn = str_field(&body, "targetRoleArn");
2319        // AWS returns an externalId (for the target role's trust policy) only
2320        // when a cross-account targetRoleArn is supplied.
2321        let external_id = target_role_arn
2322            .as_ref()
2323            .map(|_| uuid::Uuid::new_v4().to_string().replace('-', ""));
2324        let now = Utc::now();
2325        let assoc = PodIdentityAssociation {
2326            cluster_name: cluster_name.to_string(),
2327            namespace,
2328            service_account,
2329            role_arn,
2330            association_arn,
2331            association_id: association_id.clone(),
2332            created_at: now,
2333            modified_at: now,
2334            disable_session_tags: body
2335                .get("disableSessionTags")
2336                .and_then(|v| v.as_bool())
2337                .unwrap_or(false),
2338            target_role_arn,
2339            external_id,
2340            tags: parse_tag_map(body.get("tags")),
2341        };
2342        let out = pod_identity_association_json(&assoc);
2343        state
2344            .pod_identity_associations
2345            .entry(cluster_name.to_string())
2346            .or_default()
2347            .insert(association_id, assoc);
2348        Ok(AwsResponse::json(
2349            StatusCode::OK,
2350            json!({ "association": out }).to_string(),
2351        ))
2352    }
2353
2354    fn describe_pod_identity_association(
2355        &self,
2356        req: &AwsRequest,
2357        cluster_name: &str,
2358        association_id: &str,
2359    ) -> Result<AwsResponse, AwsServiceError> {
2360        let accounts = self.state.read();
2361        let state = accounts
2362            .get(&req.account_id)
2363            .ok_or_else(not_found_cluster(cluster_name))?;
2364        if !state.clusters.contains_key(cluster_name) {
2365            return Err(not_found_cluster(cluster_name)());
2366        }
2367        let assoc = state
2368            .pod_identity_associations
2369            .get(cluster_name)
2370            .and_then(|m| m.get(association_id))
2371            .ok_or_else(not_found_pod_identity_association(association_id))?;
2372        Ok(AwsResponse::json(
2373            StatusCode::OK,
2374            json!({ "association": pod_identity_association_json(assoc) }).to_string(),
2375        ))
2376    }
2377
2378    fn list_pod_identity_associations(
2379        &self,
2380        req: &AwsRequest,
2381        cluster_name: &str,
2382    ) -> Result<AwsResponse, AwsServiceError> {
2383        let max_results = validate_max_results(req)?;
2384        let next_token = req.query_params.get("nextToken").cloned();
2385        let namespace = req.query_params.get("namespace").cloned();
2386        let service_account = req.query_params.get("serviceAccount").cloned();
2387        let accounts = self.state.read();
2388        let state = accounts
2389            .get(&req.account_id)
2390            .ok_or_else(not_found_cluster(cluster_name))?;
2391        if !state.clusters.contains_key(cluster_name) {
2392            return Err(not_found_cluster(cluster_name)());
2393        }
2394        let summaries: Vec<Value> = state
2395            .pod_identity_associations
2396            .get(cluster_name)
2397            .map(|m| {
2398                m.values()
2399                    .filter(|a| namespace.as_deref().is_none_or(|n| a.namespace == n))
2400                    .filter(|a| {
2401                        service_account
2402                            .as_deref()
2403                            .is_none_or(|s| a.service_account == s)
2404                    })
2405                    .map(pod_identity_association_summary_json)
2406                    .collect()
2407            })
2408            .unwrap_or_default();
2409        let (page, token) = paginate_checked(&summaries, next_token.as_deref(), max_results)
2410            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
2411        let mut out = json!({ "associations": page });
2412        if let Some(t) = token {
2413            out["nextToken"] = Value::String(t);
2414        }
2415        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
2416    }
2417
2418    fn delete_pod_identity_association(
2419        &self,
2420        req: &AwsRequest,
2421        cluster_name: &str,
2422        association_id: &str,
2423    ) -> Result<AwsResponse, AwsServiceError> {
2424        let mut accounts = self.state.write();
2425        let state = accounts.get_or_create(&req.account_id);
2426        if !state.clusters.contains_key(cluster_name) {
2427            return Err(not_found_cluster(cluster_name)());
2428        }
2429        let assoc = state
2430            .pod_identity_associations
2431            .get_mut(cluster_name)
2432            .and_then(|m| m.remove(association_id))
2433            .ok_or_else(not_found_pod_identity_association(association_id))?;
2434        Ok(AwsResponse::json(
2435            StatusCode::OK,
2436            json!({ "association": pod_identity_association_json(&assoc) }).to_string(),
2437        ))
2438    }
2439
2440    fn update_pod_identity_association(
2441        &self,
2442        req: &AwsRequest,
2443        cluster_name: &str,
2444        association_id: &str,
2445    ) -> Result<AwsResponse, AwsServiceError> {
2446        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2447        let mut accounts = self.state.write();
2448        let state = accounts.get_or_create(&req.account_id);
2449        if !state.clusters.contains_key(cluster_name) {
2450            return Err(not_found_cluster(cluster_name)());
2451        }
2452        let assoc = state
2453            .pod_identity_associations
2454            .get_mut(cluster_name)
2455            .and_then(|m| m.get_mut(association_id))
2456            .ok_or_else(not_found_pod_identity_association(association_id))?;
2457
2458        if let Some(role) = body.get("roleArn").and_then(|v| v.as_str()) {
2459            assoc.role_arn = role.to_string();
2460        }
2461        if let Some(target) = body.get("targetRoleArn").and_then(|v| v.as_str()) {
2462            assoc.target_role_arn = Some(target.to_string());
2463            // Mint an externalId for the target trust policy if we don't have one.
2464            if assoc.external_id.is_none() {
2465                assoc.external_id = Some(uuid::Uuid::new_v4().to_string().replace('-', ""));
2466            }
2467        }
2468        if let Some(disable) = body.get("disableSessionTags").and_then(|v| v.as_bool()) {
2469            assoc.disable_session_tags = disable;
2470        }
2471        assoc.modified_at = Utc::now();
2472        Ok(AwsResponse::json(
2473            StatusCode::OK,
2474            json!({ "association": pod_identity_association_json(assoc) }).to_string(),
2475        ))
2476    }
2477
2478    // -----------------------------------------------------------------------
2479    // Insights
2480    // -----------------------------------------------------------------------
2481
2482    fn list_insights(
2483        &self,
2484        req: &AwsRequest,
2485        cluster_name: &str,
2486    ) -> Result<AwsResponse, AwsServiceError> {
2487        // ListInsights is a POST carrying the filter/pagination in its body.
2488        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2489        // maxResults is constrained to 1..=100 by the model; reject out-of-range
2490        // values rather than silently clamping them.
2491        let max_results = match body.get("maxResults") {
2492            Some(v) => {
2493                let n = v
2494                    .as_i64()
2495                    .ok_or_else(|| invalid_parameter("maxResults must be an integer"))?;
2496                if !(1..=100).contains(&n) {
2497                    return Err(invalid_parameter("maxResults must be between 1 and 100"));
2498                }
2499                n as usize
2500            }
2501            None => 100,
2502        };
2503        let next_token = body
2504            .get("nextToken")
2505            .and_then(|v| v.as_str())
2506            .map(|s| s.to_string());
2507        // Optional category / status filters.
2508        let categories = string_list(body.get("filter").and_then(|f| f.get("categories")));
2509        let statuses = string_list(body.get("filter").and_then(|f| f.get("statuses")));
2510
2511        let mut accounts = self.state.write();
2512        let state = accounts.get_or_create(&req.account_id);
2513        let version = state
2514            .clusters
2515            .get(cluster_name)
2516            .ok_or_else(not_found_cluster(cluster_name))?
2517            .version
2518            .clone();
2519        let insights = state
2520            .insights
2521            .entry(cluster_name.to_string())
2522            .or_insert_with(|| {
2523                default_insights(&version)
2524                    .into_iter()
2525                    .map(|i| (i.id.clone(), i))
2526                    .collect()
2527            });
2528        let summaries: Vec<Value> = insights
2529            .values()
2530            .filter(|i| categories.is_empty() || categories.iter().any(|c| c == &i.category))
2531            .filter(|i| statuses.is_empty() || statuses.iter().any(|s| s == &i.status))
2532            .map(insight_summary_json)
2533            .collect();
2534        let (page, token) = paginate_checked(&summaries, next_token.as_deref(), max_results)
2535            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
2536        let mut out = json!({ "insights": page });
2537        if let Some(t) = token {
2538            out["nextToken"] = Value::String(t);
2539        }
2540        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
2541    }
2542
2543    fn describe_insight(
2544        &self,
2545        req: &AwsRequest,
2546        cluster_name: &str,
2547        id: &str,
2548    ) -> Result<AwsResponse, AwsServiceError> {
2549        let mut accounts = self.state.write();
2550        let state = accounts.get_or_create(&req.account_id);
2551        let version = state
2552            .clusters
2553            .get(cluster_name)
2554            .ok_or_else(not_found_cluster(cluster_name))?
2555            .version
2556            .clone();
2557        let insights = state
2558            .insights
2559            .entry(cluster_name.to_string())
2560            .or_insert_with(|| {
2561                default_insights(&version)
2562                    .into_iter()
2563                    .map(|i| (i.id.clone(), i))
2564                    .collect()
2565            });
2566        let insight = insights.get(id).ok_or_else(not_found_insight(id))?;
2567        Ok(AwsResponse::json(
2568            StatusCode::OK,
2569            json!({ "insight": insight_json(insight) }).to_string(),
2570        ))
2571    }
2572
2573    fn describe_insights_refresh(
2574        &self,
2575        req: &AwsRequest,
2576        cluster_name: &str,
2577    ) -> Result<AwsResponse, AwsServiceError> {
2578        let mut accounts = self.state.write();
2579        let state = accounts.get_or_create(&req.account_id);
2580        if !state.clusters.contains_key(cluster_name) {
2581            return Err(not_found_cluster(cluster_name)());
2582        }
2583        let refresh = state
2584            .insights_refresh
2585            .entry(cluster_name.to_string())
2586            .or_insert_with(|| InsightsRefresh {
2587                status: "COMPLETED".to_string(),
2588                started_at: Utc::now(),
2589                ended_at: Some(Utc::now()),
2590            });
2591        // An in-progress refresh settles to COMPLETED on first describe.
2592        if refresh.status == "IN_PROGRESS" {
2593            refresh.status = "COMPLETED".to_string();
2594            refresh.ended_at = Some(Utc::now());
2595        }
2596        Ok(AwsResponse::json(
2597            StatusCode::OK,
2598            insights_refresh_json(refresh).to_string(),
2599        ))
2600    }
2601
2602    fn start_insights_refresh(
2603        &self,
2604        req: &AwsRequest,
2605        cluster_name: &str,
2606    ) -> Result<AwsResponse, AwsServiceError> {
2607        let mut accounts = self.state.write();
2608        let state = accounts.get_or_create(&req.account_id);
2609        let version = state
2610            .clusters
2611            .get(cluster_name)
2612            .ok_or_else(not_found_cluster(cluster_name))?
2613            .version
2614            .clone();
2615        // A refresh regenerates the cluster's insights.
2616        state.insights.insert(
2617            cluster_name.to_string(),
2618            default_insights(&version)
2619                .into_iter()
2620                .map(|i| (i.id.clone(), i))
2621                .collect(),
2622        );
2623        let refresh = InsightsRefresh {
2624            status: "IN_PROGRESS".to_string(),
2625            started_at: Utc::now(),
2626            ended_at: None,
2627        };
2628        // StartInsightsRefresh's response carries only `message` + `status`;
2629        // the timestamps are read back through DescribeInsightsRefresh.
2630        let out = json!({
2631            "message": "Insights refresh started for the cluster.",
2632            "status": refresh.status,
2633        });
2634        state
2635            .insights_refresh
2636            .insert(cluster_name.to_string(), refresh);
2637        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
2638    }
2639
2640    // -----------------------------------------------------------------------
2641    // Cluster misc (encryption config, cancel update, register/deregister,
2642    // cluster versions)
2643    // -----------------------------------------------------------------------
2644
2645    fn associate_encryption_config(
2646        &self,
2647        req: &AwsRequest,
2648        cluster_name: &str,
2649    ) -> Result<AwsResponse, AwsServiceError> {
2650        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2651        let encryption_config = body
2652            .get("encryptionConfig")
2653            .filter(|v| v.is_array())
2654            .cloned()
2655            .ok_or_else(|| invalid_parameter("encryptionConfig is required"))?;
2656
2657        let mut accounts = self.state.write();
2658        let state = accounts.get_or_create(&req.account_id);
2659        let cluster = state
2660            .clusters
2661            .get_mut(cluster_name)
2662            .ok_or_else(not_found_cluster(cluster_name))?;
2663        cluster.encryption_config = Some(encryption_config.clone());
2664
2665        let update = new_update(
2666            "AssociateEncryptionConfig",
2667            vec![(
2668                "EncryptionConfig".to_string(),
2669                encryption_config.to_string(),
2670            )],
2671        );
2672        let out = update_json(&update);
2673        cluster.updates.insert(update.id.clone(), update);
2674        Ok(AwsResponse::json(
2675            StatusCode::OK,
2676            json!({ "update": out }).to_string(),
2677        ))
2678    }
2679
2680    fn cancel_update(
2681        &self,
2682        req: &AwsRequest,
2683        name: &str,
2684        update_id: &str,
2685    ) -> Result<AwsResponse, AwsServiceError> {
2686        let mut accounts = self.state.write();
2687        let state = accounts.get_or_create(&req.account_id);
2688        let cluster = state
2689            .clusters
2690            .get_mut(name)
2691            .ok_or_else(not_found_cluster(name))?;
2692        let update = cluster
2693            .updates
2694            .get_mut(update_id)
2695            .ok_or_else(not_found_update(update_id))?;
2696        update.status = "Cancelled".to_string();
2697        Ok(AwsResponse::json(
2698            StatusCode::OK,
2699            json!({ "update": update_json(update) }).to_string(),
2700        ))
2701    }
2702
2703    fn register_cluster(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
2704        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2705        let name = body
2706            .get("name")
2707            .and_then(|v| v.as_str())
2708            .ok_or_else(|| invalid_parameter("name is required"))?
2709            .to_string();
2710        validate_cluster_name(&name)?;
2711        let connector = body
2712            .get("connectorConfig")
2713            .filter(|v| v.is_object())
2714            .ok_or_else(|| invalid_parameter("connectorConfig is required"))?;
2715        let role_arn = connector
2716            .get("roleArn")
2717            .and_then(|v| v.as_str())
2718            .ok_or_else(|| invalid_parameter("connectorConfig.roleArn is required"))?
2719            .to_string();
2720        let provider = connector
2721            .get("provider")
2722            .and_then(|v| v.as_str())
2723            .ok_or_else(|| invalid_parameter("connectorConfig.provider is required"))?
2724            .to_string();
2725
2726        let region = req.region.clone();
2727        let account_id = req.account_id.clone();
2728
2729        let mut accounts = self.state.write();
2730        let state = accounts.get_or_create(&req.account_id);
2731        if state.clusters.contains_key(&name) {
2732            return Err(AwsServiceError::aws_error(
2733                StatusCode::CONFLICT,
2734                "ResourceInUseException",
2735                format!("Cluster already exists with name: {name}"),
2736            ));
2737        }
2738
2739        let arn = cluster_arn(&region, &account_id, &name);
2740        let id = uuid::Uuid::new_v4().to_string();
2741        let activation_id = uuid::Uuid::new_v4().to_string();
2742        let activation_code = uuid::Uuid::new_v4().to_string().replace('-', "");
2743        let connector_config = json!({
2744            "activationId": activation_id,
2745            "activationCode": activation_code,
2746            "activationExpiry": timestamp_to_number(Utc::now() + chrono::Duration::hours(72)),
2747            "provider": provider,
2748            "roleArn": role_arn,
2749        });
2750
2751        let cluster = Cluster {
2752            name: name.clone(),
2753            arn,
2754            version: String::new(),
2755            role_arn: String::new(),
2756            status: "PENDING".to_string(),
2757            created_at: Utc::now(),
2758            endpoint: String::new(),
2759            platform_version: "eks.1".to_string(),
2760            certificate_authority_data: String::new(),
2761            resources_vpc_config: json!({}),
2762            kubernetes_network_config: json!({}),
2763            logging: build_logging(None),
2764            tags: parse_tag_map(body.get("tags")),
2765            updates: Default::default(),
2766            connector_config: Some(connector_config),
2767            encryption_config: None,
2768            access_config: build_access_config(None),
2769            upgrade_policy: build_upgrade_policy(None),
2770            compute_config: None,
2771            storage_config: None,
2772            zonal_shift_config: None,
2773            remote_network_config: None,
2774            control_plane_scaling_config: None,
2775            deletion_protection: None,
2776        };
2777        let out = connected_cluster_json(&cluster, &id);
2778        state.clusters.insert(name, cluster);
2779        Ok(AwsResponse::json(
2780            StatusCode::OK,
2781            json!({ "cluster": out }).to_string(),
2782        ))
2783    }
2784
2785    fn deregister_cluster(
2786        &self,
2787        req: &AwsRequest,
2788        name: &str,
2789    ) -> Result<AwsResponse, AwsServiceError> {
2790        let mut accounts = self.state.write();
2791        let state = accounts.get_or_create(&req.account_id);
2792        // Deregister only applies to connected (registered) clusters.
2793        let is_connected = state
2794            .clusters
2795            .get(name)
2796            .map(|c| c.connector_config.is_some())
2797            .unwrap_or(false);
2798        if !is_connected {
2799            return Err(not_found_cluster(name)());
2800        }
2801        let mut cluster = state.clusters.remove(name).unwrap();
2802        cluster.status = "DELETING".to_string();
2803        let id = uuid::Uuid::new_v4().to_string();
2804        Ok(AwsResponse::json(
2805            StatusCode::OK,
2806            json!({ "cluster": connected_cluster_json(&cluster, &id) }).to_string(),
2807        ))
2808    }
2809
2810    fn describe_cluster_versions(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
2811        let next_token = req.query_params.get("nextToken").cloned();
2812        let max_results = match req.query_params.get("maxResults") {
2813            Some(raw) => {
2814                let n: i64 = raw
2815                    .parse()
2816                    .map_err(|_| invalid_parameter("maxResults must be an integer"))?;
2817                if !(1..=100).contains(&n) {
2818                    return Err(invalid_parameter("maxResults must be between 1 and 100"));
2819                }
2820                n as usize
2821            }
2822            None => 100,
2823        };
2824        // Enum query params reject values outside the model's declared members.
2825        if let Some(status) = req.query_params.get("status") {
2826            if !["unsupported", "standard_support", "extended_support"].contains(&status.as_str()) {
2827                return Err(invalid_parameter(format!("Invalid status: {status}")));
2828            }
2829        }
2830        if let Some(vs) = req.query_params.get("versionStatus") {
2831            if !["UNSUPPORTED", "STANDARD_SUPPORT", "EXTENDED_SUPPORT"].contains(&vs.as_str()) {
2832                return Err(invalid_parameter(format!("Invalid versionStatus: {vs}")));
2833            }
2834        }
2835        let default_only = req
2836            .query_params
2837            .get("defaultOnly")
2838            .map(|v| v == "true")
2839            .unwrap_or(false);
2840        // `includeAll=false` (the default) hides versions no longer in any
2841        // support window (UNSUPPORTED); `includeAll=true` returns the full set.
2842        let include_all = req
2843            .query_params
2844            .get("includeAll")
2845            .map(|v| v == "true")
2846            .unwrap_or(false);
2847        let status_filter = req.query_params.get("status").cloned();
2848        let version_status_filter = req.query_params.get("versionStatus").cloned();
2849        let version_filter = parse_multi_query(&req.raw_query, "clusterVersions");
2850        let cluster_type = req
2851            .query_params
2852            .get("clusterType")
2853            .cloned()
2854            .unwrap_or_else(|| "eks".to_string());
2855
2856        let mut catalog: Vec<Value> = cluster_version_catalog(&cluster_type)
2857            .into_iter()
2858            .filter(|v| {
2859                version_filter.is_empty()
2860                    || version_filter
2861                        .iter()
2862                        .any(|f| v["clusterVersion"] == f.as_str())
2863            })
2864            .filter(|v| !default_only || v["defaultVersion"] == true)
2865            // The `status` (deprecated) and `versionStatus` params narrow to a
2866            // specific support tier; previously validated then ignored.
2867            .filter(|v| status_filter.as_deref().is_none_or(|s| v["status"] == s))
2868            .filter(|v| {
2869                version_status_filter
2870                    .as_deref()
2871                    .is_none_or(|s| v["versionStatus"] == s)
2872            })
2873            // Drop UNSUPPORTED rows unless the caller asked to include all, or
2874            // explicitly filtered to an unsupported tier.
2875            .filter(|v| {
2876                include_all
2877                    || status_filter.is_some()
2878                    || version_status_filter.is_some()
2879                    || v["versionStatus"] != "UNSUPPORTED"
2880            })
2881            .collect();
2882        // AWS reports the default version first; a stable sort keying non-default
2883        // rows after default ones preserves the ascending version order otherwise.
2884        catalog.sort_by_key(|v| v["defaultVersion"] != true);
2885
2886        let (page, token) = paginate_checked(&catalog, next_token.as_deref(), max_results)
2887            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
2888        let mut out = json!({ "clusterVersions": page });
2889        if let Some(t) = token {
2890            out["nextToken"] = Value::String(t);
2891        }
2892        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
2893    }
2894
2895    // -----------------------------------------------------------------------
2896    // Capabilities
2897    // -----------------------------------------------------------------------
2898
2899    fn create_capability(
2900        &self,
2901        req: &AwsRequest,
2902        cluster_name: &str,
2903    ) -> Result<AwsResponse, AwsServiceError> {
2904        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
2905        let name = body
2906            .get("capabilityName")
2907            .and_then(|v| v.as_str())
2908            .ok_or_else(|| invalid_parameter("capabilityName is required"))?
2909            .to_string();
2910        let type_ = body
2911            .get("type")
2912            .and_then(|v| v.as_str())
2913            .ok_or_else(|| invalid_parameter("type is required"))?
2914            .to_string();
2915        let role_arn = body
2916            .get("roleArn")
2917            .and_then(|v| v.as_str())
2918            .ok_or_else(|| invalid_parameter("roleArn is required"))?
2919            .to_string();
2920
2921        let region = req.region.clone();
2922        let account_id = req.account_id.clone();
2923
2924        let mut accounts = self.state.write();
2925        let state = accounts.get_or_create(&req.account_id);
2926        if !state.clusters.contains_key(cluster_name) {
2927            return Err(not_found_cluster(cluster_name)());
2928        }
2929        if state
2930            .capabilities
2931            .get(cluster_name)
2932            .is_some_and(|m| m.contains_key(&name))
2933        {
2934            return Err(AwsServiceError::aws_error(
2935                StatusCode::CONFLICT,
2936                "ResourceInUseException",
2937                format!(
2938                    "Capability already exists with name {name} and cluster name {cluster_name}"
2939                ),
2940            ));
2941        }
2942
2943        let id = uuid::Uuid::new_v4().to_string();
2944        let arn = capability_arn(&region, &account_id, cluster_name, &name, &id);
2945        let now = Utc::now();
2946        let capability = Capability {
2947            name: name.clone(),
2948            arn,
2949            cluster_name: cluster_name.to_string(),
2950            type_,
2951            role_arn,
2952            status: "CREATING".to_string(),
2953            version: "v1".to_string(),
2954            configuration: normalize_capability_configuration(body.get("configuration")),
2955            tags: parse_tag_map(body.get("tags")),
2956            created_at: now,
2957            modified_at: now,
2958            delete_propagation_policy: str_field(&body, "deletePropagationPolicy"),
2959        };
2960        let out = capability_json(&capability);
2961        state
2962            .capabilities
2963            .entry(cluster_name.to_string())
2964            .or_default()
2965            .insert(name, capability);
2966        Ok(AwsResponse::json(
2967            StatusCode::OK,
2968            json!({ "capability": out }).to_string(),
2969        ))
2970    }
2971
2972    fn describe_capability(
2973        &self,
2974        req: &AwsRequest,
2975        cluster_name: &str,
2976        name: &str,
2977    ) -> Result<AwsResponse, AwsServiceError> {
2978        let mut accounts = self.state.write();
2979        let state = accounts.get_or_create(&req.account_id);
2980        if !state.clusters.contains_key(cluster_name) {
2981            return Err(not_found_cluster(cluster_name)());
2982        }
2983        let capability = state
2984            .capabilities
2985            .get_mut(cluster_name)
2986            .and_then(|m| m.get_mut(name))
2987            .ok_or_else(not_found_capability(name))?;
2988        if capability.status == "CREATING" {
2989            capability.status = "ACTIVE".to_string();
2990        }
2991        Ok(AwsResponse::json(
2992            StatusCode::OK,
2993            json!({ "capability": capability_json(capability) }).to_string(),
2994        ))
2995    }
2996
2997    fn list_capabilities(
2998        &self,
2999        req: &AwsRequest,
3000        cluster_name: &str,
3001    ) -> Result<AwsResponse, AwsServiceError> {
3002        let max_results = validate_max_results(req)?;
3003        let next_token = req.query_params.get("nextToken").cloned();
3004        let accounts = self.state.read();
3005        let state = accounts
3006            .get(&req.account_id)
3007            .ok_or_else(not_found_cluster(cluster_name))?;
3008        if !state.clusters.contains_key(cluster_name) {
3009            return Err(not_found_cluster(cluster_name)());
3010        }
3011        let summaries: Vec<Value> = state
3012            .capabilities
3013            .get(cluster_name)
3014            .map(|m| m.values().map(capability_summary_json).collect())
3015            .unwrap_or_default();
3016        let (page, token) = paginate_checked(&summaries, next_token.as_deref(), max_results)
3017            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
3018        let mut out = json!({ "capabilities": page });
3019        if let Some(t) = token {
3020            out["nextToken"] = Value::String(t);
3021        }
3022        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
3023    }
3024
3025    fn delete_capability(
3026        &self,
3027        req: &AwsRequest,
3028        cluster_name: &str,
3029        name: &str,
3030    ) -> Result<AwsResponse, AwsServiceError> {
3031        let mut accounts = self.state.write();
3032        let state = accounts.get_or_create(&req.account_id);
3033        if !state.clusters.contains_key(cluster_name) {
3034            return Err(not_found_cluster(cluster_name)());
3035        }
3036        let mut capability = state
3037            .capabilities
3038            .get_mut(cluster_name)
3039            .and_then(|m| m.remove(name))
3040            .ok_or_else(not_found_capability(name))?;
3041        capability.status = "DELETING".to_string();
3042        Ok(AwsResponse::json(
3043            StatusCode::OK,
3044            json!({ "capability": capability_json(&capability) }).to_string(),
3045        ))
3046    }
3047
3048    fn update_capability(
3049        &self,
3050        req: &AwsRequest,
3051        cluster_name: &str,
3052        name: &str,
3053    ) -> Result<AwsResponse, AwsServiceError> {
3054        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
3055        let mut accounts = self.state.write();
3056        let state = accounts.get_or_create(&req.account_id);
3057        if !state.clusters.contains_key(cluster_name) {
3058            return Err(not_found_cluster(cluster_name)());
3059        }
3060        let capability = state
3061            .capabilities
3062            .get_mut(cluster_name)
3063            .and_then(|m| m.get_mut(name))
3064            .ok_or_else(not_found_capability(name))?;
3065
3066        let mut params = Vec::new();
3067        if let Some(role) = body.get("roleArn").and_then(|v| v.as_str()) {
3068            capability.role_arn = role.to_string();
3069            params.push(("RoleArn".to_string(), role.to_string()));
3070        }
3071        if let Some(cfg) = normalize_capability_configuration(body.get("configuration")) {
3072            params.push(("Configuration".to_string(), cfg.to_string()));
3073            capability.configuration = Some(cfg);
3074        }
3075        capability.modified_at = Utc::now();
3076
3077        // UpdateCapability returns a tracked cluster-scoped Update.
3078        let update = new_update("CapabilityUpdate", params);
3079        let out = update_json(&update);
3080        let cluster = state.clusters.get_mut(cluster_name).unwrap();
3081        cluster.updates.insert(update.id.clone(), update);
3082        Ok(AwsResponse::json(
3083            StatusCode::OK,
3084            json!({ "update": out }).to_string(),
3085        ))
3086    }
3087
3088    // -----------------------------------------------------------------------
3089    // EKS Anywhere subscriptions
3090    // -----------------------------------------------------------------------
3091
3092    fn create_eks_anywhere_subscription(
3093        &self,
3094        req: &AwsRequest,
3095    ) -> Result<AwsResponse, AwsServiceError> {
3096        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
3097        let name = body
3098            .get("name")
3099            .and_then(|v| v.as_str())
3100            .ok_or_else(|| invalid_parameter("name is required"))?
3101            .to_string();
3102        // EksAnywhereSubscriptionName: length 1-100, ^[0-9A-Za-z][A-Za-z0-9\-_]*$.
3103        if name.is_empty() || name.len() > 100 {
3104            return Err(invalid_parameter("name must be 1-100 characters"));
3105        }
3106        if !name.starts_with(|c: char| c.is_ascii_alphanumeric())
3107            || !name
3108                .chars()
3109                .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_'))
3110        {
3111            return Err(invalid_parameter(
3112                "name must match ^[0-9A-Za-z][A-Za-z0-9\\-_]*$",
3113            ));
3114        }
3115        if let Some(lt) = body.get("licenseType").and_then(|v| v.as_str()) {
3116            if lt != "Cluster" {
3117                return Err(invalid_parameter(format!("Invalid licenseType: {lt}")));
3118            }
3119        }
3120        let term = body
3121            .get("term")
3122            .filter(|v| v.is_object())
3123            .ok_or_else(|| invalid_parameter("term is required"))?;
3124        let term_duration = term.get("duration").and_then(|v| v.as_i64()).unwrap_or(12);
3125        // AWS EKS Anywhere subscriptions use month-based terms (12 or 36); bound
3126        // the value so a hostile `duration` can't overflow the date arithmetic
3127        // below (`DateTime + Duration` panics on overflow in all build profiles).
3128        if !(1..=120).contains(&term_duration) {
3129            return Err(invalid_parameter(format!(
3130                "term.duration must be between 1 and 120 months, got {term_duration}"
3131            )));
3132        }
3133        let term_unit = term
3134            .get("unit")
3135            .and_then(|v| v.as_str())
3136            .unwrap_or("MONTHS")
3137            .to_string();
3138        let license_quantity = body
3139            .get("licenseQuantity")
3140            .and_then(|v| v.as_i64())
3141            .unwrap_or(0);
3142        let license_type = body
3143            .get("licenseType")
3144            .and_then(|v| v.as_str())
3145            .unwrap_or("Cluster")
3146            .to_string();
3147        let auto_renew = body
3148            .get("autoRenew")
3149            .and_then(|v| v.as_bool())
3150            .unwrap_or(false);
3151
3152        let region = req.region.clone();
3153        let account_id = req.account_id.clone();
3154
3155        let mut accounts = self.state.write();
3156        let state = accounts.get_or_create(&req.account_id);
3157
3158        let raw = uuid::Uuid::new_v4().to_string().replace('-', "");
3159        let id = raw[..17.min(raw.len())].to_string();
3160        let arn = eks_anywhere_subscription_arn(&region, &account_id, &id);
3161        let now = Utc::now();
3162        let expiration = now
3163            .checked_add_signed(chrono::Duration::days(term_duration * 30))
3164            .ok_or_else(|| {
3165                invalid_parameter("term.duration produces an out-of-range expiration date")
3166            })?;
3167        let subscription = EksAnywhereSubscription {
3168            id: id.clone(),
3169            arn,
3170            name,
3171            created_at: now,
3172            effective_date: now,
3173            expiration_date: expiration,
3174            license_quantity,
3175            license_type,
3176            term_duration,
3177            term_unit,
3178            status: "ACTIVE".to_string(),
3179            auto_renew,
3180            tags: parse_tag_map(body.get("tags")),
3181        };
3182        let out = subscription_json(&subscription);
3183        state.eks_anywhere_subscriptions.insert(id, subscription);
3184        Ok(AwsResponse::json(
3185            StatusCode::OK,
3186            json!({ "subscription": out }).to_string(),
3187        ))
3188    }
3189
3190    fn list_eks_anywhere_subscriptions(
3191        &self,
3192        req: &AwsRequest,
3193    ) -> Result<AwsResponse, AwsServiceError> {
3194        let max_results = validate_max_results(req)?;
3195        let next_token = req.query_params.get("nextToken").cloned();
3196        let accounts = self.state.read();
3197        let Some(state) = accounts.get(&req.account_id) else {
3198            return Ok(AwsResponse::json(
3199                StatusCode::OK,
3200                json!({ "subscriptions": [] }).to_string(),
3201            ));
3202        };
3203        let subs: Vec<Value> = state
3204            .eks_anywhere_subscriptions
3205            .values()
3206            .map(subscription_json)
3207            .collect();
3208        let (page, token) = paginate_checked(&subs, next_token.as_deref(), max_results)
3209            .map_err(|_| invalid_parameter("Invalid nextToken"))?;
3210        let mut out = json!({ "subscriptions": page });
3211        if let Some(t) = token {
3212            out["nextToken"] = Value::String(t);
3213        }
3214        Ok(AwsResponse::json(StatusCode::OK, out.to_string()))
3215    }
3216
3217    fn describe_eks_anywhere_subscription(
3218        &self,
3219        req: &AwsRequest,
3220        id: &str,
3221    ) -> Result<AwsResponse, AwsServiceError> {
3222        let accounts = self.state.read();
3223        let state = accounts
3224            .get(&req.account_id)
3225            .ok_or_else(not_found_subscription(id))?;
3226        let subscription = state
3227            .eks_anywhere_subscriptions
3228            .get(id)
3229            .ok_or_else(not_found_subscription(id))?;
3230        Ok(AwsResponse::json(
3231            StatusCode::OK,
3232            json!({ "subscription": subscription_json(subscription) }).to_string(),
3233        ))
3234    }
3235
3236    fn delete_eks_anywhere_subscription(
3237        &self,
3238        req: &AwsRequest,
3239        id: &str,
3240    ) -> Result<AwsResponse, AwsServiceError> {
3241        let mut accounts = self.state.write();
3242        let state = accounts.get_or_create(&req.account_id);
3243        let mut subscription = state
3244            .eks_anywhere_subscriptions
3245            .remove(id)
3246            .ok_or_else(not_found_subscription(id))?;
3247        subscription.status = "DELETING".to_string();
3248        Ok(AwsResponse::json(
3249            StatusCode::OK,
3250            json!({ "subscription": subscription_json(&subscription) }).to_string(),
3251        ))
3252    }
3253
3254    fn update_eks_anywhere_subscription(
3255        &self,
3256        req: &AwsRequest,
3257        id: &str,
3258    ) -> Result<AwsResponse, AwsServiceError> {
3259        let body: Value = serde_json::from_slice(&req.body).unwrap_or_default();
3260        let auto_renew = body
3261            .get("autoRenew")
3262            .and_then(|v| v.as_bool())
3263            .ok_or_else(|| invalid_parameter("autoRenew is required"))?;
3264        let mut accounts = self.state.write();
3265        let state = accounts.get_or_create(&req.account_id);
3266        let subscription = state
3267            .eks_anywhere_subscriptions
3268            .get_mut(id)
3269            .ok_or_else(not_found_subscription(id))?;
3270        subscription.auto_renew = auto_renew;
3271        Ok(AwsResponse::json(
3272            StatusCode::OK,
3273            json!({ "subscription": subscription_json(subscription) }).to_string(),
3274        ))
3275    }
3276}
3277
3278/// Persist the current EKS state as a snapshot. Shared by
3279/// `EksService::save_snapshot` and the CFN provisioner's persist hook.
3280pub async fn save_eks_snapshot(
3281    state: &SharedEksState,
3282    store: Option<Arc<dyn SnapshotStore>>,
3283    lock: &AsyncMutex<()>,
3284) {
3285    let Some(store) = store else {
3286        return;
3287    };
3288    let _guard = lock.lock().await;
3289    let snapshot = EksSnapshot {
3290        schema_version: EKS_SNAPSHOT_SCHEMA_VERSION,
3291        accounts: state.read().clone(),
3292    };
3293    let join = tokio::task::spawn_blocking(move || -> std::io::Result<()> {
3294        let bytes = serde_json::to_vec(&snapshot)
3295            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?;
3296        store.save(&bytes)
3297    })
3298    .await;
3299    match join {
3300        Ok(Ok(())) => {}
3301        Ok(Err(err)) => tracing::error!(%err, "failed to write eks snapshot"),
3302        Err(err) => tracing::error!(%err, "eks snapshot task panicked"),
3303    }
3304}
3305
3306#[async_trait]
3307impl AwsService for EksService {
3308    fn service_name(&self) -> &str {
3309        "eks"
3310    }
3311
3312    async fn handle(&self, req: AwsRequest) -> Result<AwsResponse, AwsServiceError> {
3313        let (action, args) = Self::resolve_action(&req).ok_or_else(|| {
3314            AwsServiceError::aws_error(
3315                StatusCode::NOT_FOUND,
3316                "UnknownOperationException",
3317                format!("Unknown operation: {} {}", req.method, req.raw_path),
3318            )
3319        })?;
3320
3321        let mutates = matches!(
3322            action,
3323            "CreateCluster"
3324                | "DeleteCluster"
3325                | "UpdateClusterConfig"
3326                | "UpdateClusterVersion"
3327                | "TagResource"
3328                | "UntagResource"
3329                // Describe settles CREATING -> ACTIVE and IN_PROGRESS updates,
3330                // which mutates persisted state.
3331                | "DescribeCluster"
3332                | "DescribeUpdate"
3333                | "CreateNodegroup"
3334                | "DeleteNodegroup"
3335                | "UpdateNodegroupConfig"
3336                | "UpdateNodegroupVersion"
3337                | "DescribeNodegroup"
3338                | "CreateFargateProfile"
3339                | "DeleteFargateProfile"
3340                | "DescribeFargateProfile"
3341                | "CreateAddon"
3342                | "DeleteAddon"
3343                | "UpdateAddon"
3344                | "DescribeAddon"
3345                | "CreateAccessEntry"
3346                | "DeleteAccessEntry"
3347                | "UpdateAccessEntry"
3348                | "AssociateAccessPolicy"
3349                | "DisassociateAccessPolicy"
3350                | "AssociateIdentityProviderConfig"
3351                | "DisassociateIdentityProviderConfig"
3352                // Describe settles the config's CREATING -> ACTIVE transition.
3353                | "DescribeIdentityProviderConfig"
3354                | "CreatePodIdentityAssociation"
3355                | "DeletePodIdentityAssociation"
3356                | "UpdatePodIdentityAssociation"
3357                // Insights are seeded/settled on read; refresh settles on describe.
3358                | "ListInsights"
3359                | "DescribeInsight"
3360                | "DescribeInsightsRefresh"
3361                | "StartInsightsRefresh"
3362                | "AssociateEncryptionConfig"
3363                | "CancelUpdate"
3364                | "RegisterCluster"
3365                | "DeregisterCluster"
3366                | "CreateCapability"
3367                | "DeleteCapability"
3368                | "UpdateCapability"
3369                | "DescribeCapability"
3370                | "CreateEksAnywhereSubscription"
3371                | "DeleteEksAnywhereSubscription"
3372                | "UpdateEksAnywhereSubscription"
3373                | "DescribeEksAnywhereSubscription"
3374        );
3375
3376        let result = match (action, &args) {
3377            ("CreateCluster", _) => self.create_cluster(&req),
3378            ("ListClusters", _) => self.list_clusters(&req),
3379            ("DescribeCluster", PathArgs::Name(n)) => self.describe_cluster(&req, n),
3380            ("DeleteCluster", PathArgs::Name(n)) => self.delete_cluster(&req, n),
3381            ("UpdateClusterConfig", PathArgs::Name(n)) => self.update_cluster_config(&req, n),
3382            ("UpdateClusterVersion", PathArgs::Name(n)) => self.update_cluster_version(&req, n),
3383            ("ListUpdates", PathArgs::Name(n)) => self.list_updates(&req, n),
3384            ("DescribeUpdate", PathArgs::Update { name, update_id }) => {
3385                self.describe_update(&req, name, update_id)
3386            }
3387            ("TagResource", PathArgs::Arn(a)) => self.tag_resource(&req, a),
3388            ("UntagResource", PathArgs::Arn(a)) => self.untag_resource(&req, a),
3389            ("ListTagsForResource", PathArgs::Arn(a)) => self.list_tags_for_resource(&req, a),
3390            ("CreateNodegroup", PathArgs::Cluster(c)) => self.create_nodegroup(&req, c),
3391            ("ListNodegroups", PathArgs::Cluster(c)) => self.list_nodegroups(&req, c),
3392            ("DescribeNodegroup", PathArgs::ClusterChild { cluster, name }) => {
3393                self.describe_nodegroup(&req, cluster, name)
3394            }
3395            ("DeleteNodegroup", PathArgs::ClusterChild { cluster, name }) => {
3396                self.delete_nodegroup(&req, cluster, name)
3397            }
3398            ("UpdateNodegroupConfig", PathArgs::ClusterChild { cluster, name }) => {
3399                self.update_nodegroup_config(&req, cluster, name)
3400            }
3401            ("UpdateNodegroupVersion", PathArgs::ClusterChild { cluster, name }) => {
3402                self.update_nodegroup_version(&req, cluster, name)
3403            }
3404            ("CreateFargateProfile", PathArgs::Cluster(c)) => self.create_fargate_profile(&req, c),
3405            ("ListFargateProfiles", PathArgs::Cluster(c)) => self.list_fargate_profiles(&req, c),
3406            ("DescribeFargateProfile", PathArgs::ClusterChild { cluster, name }) => {
3407                self.describe_fargate_profile(&req, cluster, name)
3408            }
3409            ("DeleteFargateProfile", PathArgs::ClusterChild { cluster, name }) => {
3410                self.delete_fargate_profile(&req, cluster, name)
3411            }
3412            ("CreateAddon", PathArgs::Cluster(c)) => self.create_addon(&req, c),
3413            ("ListAddons", PathArgs::Cluster(c)) => self.list_addons(&req, c),
3414            ("DescribeAddon", PathArgs::ClusterChild { cluster, name }) => {
3415                self.describe_addon(&req, cluster, name)
3416            }
3417            ("DeleteAddon", PathArgs::ClusterChild { cluster, name }) => {
3418                self.delete_addon(&req, cluster, name)
3419            }
3420            ("UpdateAddon", PathArgs::ClusterChild { cluster, name }) => {
3421                self.update_addon(&req, cluster, name)
3422            }
3423            ("DescribeAddonVersions", _) => self.describe_addon_versions(&req),
3424            ("DescribeAddonConfiguration", _) => self.describe_addon_configuration(&req),
3425            ("CreateAccessEntry", PathArgs::Cluster(c)) => self.create_access_entry(&req, c),
3426            ("ListAccessEntries", PathArgs::Cluster(c)) => self.list_access_entries(&req, c),
3427            ("DescribeAccessEntry", PathArgs::ClusterChild { cluster, name }) => {
3428                self.describe_access_entry(&req, cluster, name)
3429            }
3430            ("DeleteAccessEntry", PathArgs::ClusterChild { cluster, name }) => {
3431                self.delete_access_entry(&req, cluster, name)
3432            }
3433            ("UpdateAccessEntry", PathArgs::ClusterChild { cluster, name }) => {
3434                self.update_access_entry(&req, cluster, name)
3435            }
3436            ("AssociateAccessPolicy", PathArgs::ClusterChild { cluster, name }) => {
3437                self.associate_access_policy(&req, cluster, name)
3438            }
3439            ("ListAssociatedAccessPolicies", PathArgs::ClusterChild { cluster, name }) => {
3440                self.list_associated_access_policies(&req, cluster, name)
3441            }
3442            (
3443                "DisassociateAccessPolicy",
3444                PathArgs::AccessPolicyChild {
3445                    cluster,
3446                    principal,
3447                    policy_arn,
3448                },
3449            ) => self.disassociate_access_policy(&req, cluster, principal, policy_arn),
3450            ("ListAccessPolicies", _) => self.list_access_policies(&req),
3451            ("AssociateIdentityProviderConfig", PathArgs::Cluster(c)) => {
3452                self.associate_identity_provider_config(&req, c)
3453            }
3454            ("DisassociateIdentityProviderConfig", PathArgs::Cluster(c)) => {
3455                self.disassociate_identity_provider_config(&req, c)
3456            }
3457            ("DescribeIdentityProviderConfig", PathArgs::Cluster(c)) => {
3458                self.describe_identity_provider_config(&req, c)
3459            }
3460            ("ListIdentityProviderConfigs", PathArgs::Cluster(c)) => {
3461                self.list_identity_provider_configs(&req, c)
3462            }
3463            ("CreatePodIdentityAssociation", PathArgs::Cluster(c)) => {
3464                self.create_pod_identity_association(&req, c)
3465            }
3466            ("ListPodIdentityAssociations", PathArgs::Cluster(c)) => {
3467                self.list_pod_identity_associations(&req, c)
3468            }
3469            ("DescribePodIdentityAssociation", PathArgs::ClusterChild { cluster, name }) => {
3470                self.describe_pod_identity_association(&req, cluster, name)
3471            }
3472            ("DeletePodIdentityAssociation", PathArgs::ClusterChild { cluster, name }) => {
3473                self.delete_pod_identity_association(&req, cluster, name)
3474            }
3475            ("UpdatePodIdentityAssociation", PathArgs::ClusterChild { cluster, name }) => {
3476                self.update_pod_identity_association(&req, cluster, name)
3477            }
3478            ("ListInsights", PathArgs::Cluster(c)) => self.list_insights(&req, c),
3479            ("DescribeInsight", PathArgs::ClusterChild { cluster, name }) => {
3480                self.describe_insight(&req, cluster, name)
3481            }
3482            ("DescribeInsightsRefresh", PathArgs::Cluster(c)) => {
3483                self.describe_insights_refresh(&req, c)
3484            }
3485            ("StartInsightsRefresh", PathArgs::Cluster(c)) => self.start_insights_refresh(&req, c),
3486            ("AssociateEncryptionConfig", PathArgs::Cluster(c)) => {
3487                self.associate_encryption_config(&req, c)
3488            }
3489            ("CancelUpdate", PathArgs::Update { name, update_id }) => {
3490                self.cancel_update(&req, name, update_id)
3491            }
3492            ("RegisterCluster", _) => self.register_cluster(&req),
3493            ("DeregisterCluster", PathArgs::Name(n)) => self.deregister_cluster(&req, n),
3494            ("DescribeClusterVersions", _) => self.describe_cluster_versions(&req),
3495            ("CreateCapability", PathArgs::Cluster(c)) => self.create_capability(&req, c),
3496            ("ListCapabilities", PathArgs::Cluster(c)) => self.list_capabilities(&req, c),
3497            ("DescribeCapability", PathArgs::ClusterChild { cluster, name }) => {
3498                self.describe_capability(&req, cluster, name)
3499            }
3500            ("DeleteCapability", PathArgs::ClusterChild { cluster, name }) => {
3501                self.delete_capability(&req, cluster, name)
3502            }
3503            ("UpdateCapability", PathArgs::ClusterChild { cluster, name }) => {
3504                self.update_capability(&req, cluster, name)
3505            }
3506            ("CreateEksAnywhereSubscription", _) => self.create_eks_anywhere_subscription(&req),
3507            ("ListEksAnywhereSubscriptions", _) => self.list_eks_anywhere_subscriptions(&req),
3508            ("DescribeEksAnywhereSubscription", PathArgs::Name(id)) => {
3509                self.describe_eks_anywhere_subscription(&req, id)
3510            }
3511            ("DeleteEksAnywhereSubscription", PathArgs::Name(id)) => {
3512                self.delete_eks_anywhere_subscription(&req, id)
3513            }
3514            ("UpdateEksAnywhereSubscription", PathArgs::Name(id)) => {
3515                self.update_eks_anywhere_subscription(&req, id)
3516            }
3517            _ => Err(AwsServiceError::action_not_implemented("eks", action)),
3518        };
3519
3520        if mutates && matches!(result.as_ref(), Ok(resp) if resp.status.is_success()) {
3521            self.save_snapshot().await;
3522        }
3523        result
3524    }
3525
3526    fn supported_actions(&self) -> &[&str] {
3527        EKS_ACTIONS
3528    }
3529}
3530
3531// ---------------------------------------------------------------------------
3532// Helpers
3533// ---------------------------------------------------------------------------
3534
3535#[cfg(test)]
3536#[path = "service_tests.rs"]
3537mod tests;