1use std::sync::Arc;
4
5use async_trait::async_trait;
6use parking_lot::RwLock;
7use tokio::sync::Mutex as AsyncMutex;
8
9use fakecloud_core::service::{AwsRequest, AwsResponse, AwsService, AwsServiceError};
10use fakecloud_persistence::SnapshotStore;
11
12use crate::state::{
13 RedshiftAccounts, RedshiftSnapshot, SharedRedshiftState, REDSHIFT_SNAPSHOT_SCHEMA_VERSION,
14};
15
16mod access;
17mod clusters;
18mod groups;
19pub(crate) mod helpers;
20mod resources;
21mod snapshots;
22mod subscriptions;
23
24#[cfg(test)]
25mod tests;
26
27const SUPPORTED_ACTIONS: &[&str] = &[
28 "AcceptReservedNodeExchange",
29 "AddPartner",
30 "AssociateDataShareConsumer",
31 "AuthorizeClusterSecurityGroupIngress",
32 "AuthorizeDataShare",
33 "AuthorizeEndpointAccess",
34 "AuthorizeSnapshotAccess",
35 "BatchDeleteClusterSnapshots",
36 "BatchModifyClusterSnapshots",
37 "CancelResize",
38 "CopyClusterSnapshot",
39 "CreateAuthenticationProfile",
40 "CreateCluster",
41 "CreateClusterParameterGroup",
42 "CreateClusterSecurityGroup",
43 "CreateClusterSnapshot",
44 "CreateClusterSubnetGroup",
45 "CreateCustomDomainAssociation",
46 "CreateEndpointAccess",
47 "CreateEventSubscription",
48 "CreateHsmClientCertificate",
49 "CreateHsmConfiguration",
50 "CreateIntegration",
51 "CreateRedshiftIdcApplication",
52 "CreateScheduledAction",
53 "CreateSnapshotCopyGrant",
54 "CreateSnapshotSchedule",
55 "CreateTags",
56 "CreateUsageLimit",
57 "DeauthorizeDataShare",
58 "DeleteAuthenticationProfile",
59 "DeleteCluster",
60 "DeleteClusterParameterGroup",
61 "DeleteClusterSecurityGroup",
62 "DeleteClusterSnapshot",
63 "DeleteClusterSubnetGroup",
64 "DeleteCustomDomainAssociation",
65 "DeleteEndpointAccess",
66 "DeleteEventSubscription",
67 "DeleteHsmClientCertificate",
68 "DeleteHsmConfiguration",
69 "DeleteIntegration",
70 "DeletePartner",
71 "DeleteRedshiftIdcApplication",
72 "DeleteResourcePolicy",
73 "DeleteScheduledAction",
74 "DeleteSnapshotCopyGrant",
75 "DeleteSnapshotSchedule",
76 "DeleteTags",
77 "DeleteUsageLimit",
78 "DeregisterNamespace",
79 "DescribeAccountAttributes",
80 "DescribeAuthenticationProfiles",
81 "DescribeClusterDbRevisions",
82 "DescribeClusterParameterGroups",
83 "DescribeClusterParameters",
84 "DescribeClusterSecurityGroups",
85 "DescribeClusterSnapshots",
86 "DescribeClusterSubnetGroups",
87 "DescribeClusterTracks",
88 "DescribeClusterVersions",
89 "DescribeClusters",
90 "DescribeCustomDomainAssociations",
91 "DescribeDataShares",
92 "DescribeDataSharesForConsumer",
93 "DescribeDataSharesForProducer",
94 "DescribeDefaultClusterParameters",
95 "DescribeEndpointAccess",
96 "DescribeEndpointAuthorization",
97 "DescribeEventCategories",
98 "DescribeEventSubscriptions",
99 "DescribeEvents",
100 "DescribeHsmClientCertificates",
101 "DescribeHsmConfigurations",
102 "DescribeInboundIntegrations",
103 "DescribeIntegrations",
104 "DescribeLoggingStatus",
105 "DescribeNodeConfigurationOptions",
106 "DescribeOrderableClusterOptions",
107 "DescribePartners",
108 "DescribeRedshiftIdcApplications",
109 "DescribeReservedNodeExchangeStatus",
110 "DescribeReservedNodeOfferings",
111 "DescribeReservedNodes",
112 "DescribeResize",
113 "DescribeScheduledActions",
114 "DescribeSnapshotCopyGrants",
115 "DescribeSnapshotSchedules",
116 "DescribeStorage",
117 "DescribeTableRestoreStatus",
118 "DescribeTags",
119 "DescribeUsageLimits",
120 "DisableLogging",
121 "DisableSnapshotCopy",
122 "DisassociateDataShareConsumer",
123 "EnableLogging",
124 "EnableSnapshotCopy",
125 "FailoverPrimaryCompute",
126 "GetClusterCredentials",
127 "GetClusterCredentialsWithIAM",
128 "GetIdentityCenterAuthToken",
129 "GetReservedNodeExchangeConfigurationOptions",
130 "GetReservedNodeExchangeOfferings",
131 "GetResourcePolicy",
132 "ListRecommendations",
133 "ModifyAquaConfiguration",
134 "ModifyAuthenticationProfile",
135 "ModifyCluster",
136 "ModifyClusterDbRevision",
137 "ModifyClusterIamRoles",
138 "ModifyClusterMaintenance",
139 "ModifyClusterParameterGroup",
140 "ModifyClusterSnapshot",
141 "ModifyClusterSnapshotSchedule",
142 "ModifyClusterSubnetGroup",
143 "ModifyCustomDomainAssociation",
144 "ModifyEndpointAccess",
145 "ModifyEventSubscription",
146 "ModifyIntegration",
147 "ModifyLakehouseConfiguration",
148 "ModifyRedshiftIdcApplication",
149 "ModifyScheduledAction",
150 "ModifySnapshotCopyRetentionPeriod",
151 "ModifySnapshotSchedule",
152 "ModifyUsageLimit",
153 "PauseCluster",
154 "PurchaseReservedNodeOffering",
155 "PutResourcePolicy",
156 "RebootCluster",
157 "RegisterNamespace",
158 "RejectDataShare",
159 "ResetClusterParameterGroup",
160 "ResizeCluster",
161 "RestoreFromClusterSnapshot",
162 "RestoreTableFromClusterSnapshot",
163 "ResumeCluster",
164 "RevokeClusterSecurityGroupIngress",
165 "RevokeEndpointAccess",
166 "RevokeSnapshotAccess",
167 "RotateEncryptionKey",
168 "UpdatePartnerStatus",
169];
170
171pub struct RedshiftService {
174 state: SharedRedshiftState,
175 snapshot_store: Option<Arc<dyn SnapshotStore>>,
176 snapshot_lock: Arc<AsyncMutex<()>>,
177}
178
179impl RedshiftService {
180 pub fn new(state: SharedRedshiftState) -> Self {
181 Self {
182 state,
183 snapshot_store: None,
184 snapshot_lock: Arc::new(AsyncMutex::new(())),
185 }
186 }
187
188 pub fn with_snapshot_store(mut self, store: Arc<dyn SnapshotStore>) -> Self {
189 self.snapshot_store = Some(store);
190 self
191 }
192
193 pub fn shared_state(&self) -> SharedRedshiftState {
194 Arc::clone(&self.state)
195 }
196
197 async fn save_snapshot(&self) {
198 save_redshift_snapshot(
199 &self.state,
200 self.snapshot_store.clone(),
201 &self.snapshot_lock,
202 )
203 .await;
204 }
205
206 pub fn snapshot_hook(&self) -> Option<fakecloud_persistence::SnapshotHook> {
210 let store = self.snapshot_store.clone()?;
211 let state = self.state.clone();
212 let lock = self.snapshot_lock.clone();
213 Some(Arc::new(move || {
214 let state = state.clone();
215 let store = store.clone();
216 let lock = lock.clone();
217 Box::pin(async move {
218 save_redshift_snapshot(&state, Some(store), &lock).await;
219 })
220 }))
221 }
222
223 fn is_mutating(action: &str) -> bool {
226 !(action.starts_with("Describe") || action.starts_with("Get") || action.starts_with("List"))
227 }
228
229 fn dispatch(&self, req: &AwsRequest) -> Result<AwsResponse, AwsServiceError> {
230 crate::validation::validate_request(&req.action, req)?;
231 match req.action.as_str() {
232 "AcceptReservedNodeExchange" => self.accept_reserved_node_exchange(req),
233 "AddPartner" => self.add_partner(req),
234 "AssociateDataShareConsumer" => self.associate_data_share_consumer(req),
235 "AuthorizeClusterSecurityGroupIngress" => {
236 self.authorize_cluster_security_group_ingress(req)
237 }
238 "AuthorizeDataShare" => self.authorize_data_share(req),
239 "AuthorizeEndpointAccess" => self.authorize_endpoint_access(req),
240 "AuthorizeSnapshotAccess" => self.authorize_snapshot_access(req),
241 "BatchDeleteClusterSnapshots" => self.batch_delete_cluster_snapshots(req),
242 "BatchModifyClusterSnapshots" => self.batch_modify_cluster_snapshots(req),
243 "CancelResize" => self.cancel_resize(req),
244 "CopyClusterSnapshot" => self.copy_cluster_snapshot(req),
245 "CreateAuthenticationProfile" => self.create_authentication_profile(req),
246 "CreateCluster" => self.create_cluster(req),
247 "CreateClusterParameterGroup" => self.create_cluster_parameter_group(req),
248 "CreateClusterSecurityGroup" => self.create_cluster_security_group(req),
249 "CreateClusterSnapshot" => self.create_cluster_snapshot(req),
250 "CreateClusterSubnetGroup" => self.create_cluster_subnet_group(req),
251 "CreateCustomDomainAssociation" => self.create_custom_domain_association(req),
252 "CreateEndpointAccess" => self.create_endpoint_access(req),
253 "CreateEventSubscription" => self.create_event_subscription(req),
254 "CreateHsmClientCertificate" => self.create_hsm_client_certificate(req),
255 "CreateHsmConfiguration" => self.create_hsm_configuration(req),
256 "CreateIntegration" => self.create_integration(req),
257 "CreateRedshiftIdcApplication" => self.create_redshift_idc_application(req),
258 "CreateScheduledAction" => self.create_scheduled_action(req),
259 "CreateSnapshotCopyGrant" => self.create_snapshot_copy_grant(req),
260 "CreateSnapshotSchedule" => self.create_snapshot_schedule(req),
261 "CreateTags" => self.create_tags(req),
262 "CreateUsageLimit" => self.create_usage_limit(req),
263 "DeauthorizeDataShare" => self.deauthorize_data_share(req),
264 "DeleteAuthenticationProfile" => self.delete_authentication_profile(req),
265 "DeleteCluster" => self.delete_cluster(req),
266 "DeleteClusterParameterGroup" => self.delete_cluster_parameter_group(req),
267 "DeleteClusterSecurityGroup" => self.delete_cluster_security_group(req),
268 "DeleteClusterSnapshot" => self.delete_cluster_snapshot(req),
269 "DeleteClusterSubnetGroup" => self.delete_cluster_subnet_group(req),
270 "DeleteCustomDomainAssociation" => self.delete_custom_domain_association(req),
271 "DeleteEndpointAccess" => self.delete_endpoint_access(req),
272 "DeleteEventSubscription" => self.delete_event_subscription(req),
273 "DeleteHsmClientCertificate" => self.delete_hsm_client_certificate(req),
274 "DeleteHsmConfiguration" => self.delete_hsm_configuration(req),
275 "DeleteIntegration" => self.delete_integration(req),
276 "DeletePartner" => self.delete_partner(req),
277 "DeleteRedshiftIdcApplication" => self.delete_redshift_idc_application(req),
278 "DeleteResourcePolicy" => self.delete_resource_policy(req),
279 "DeleteScheduledAction" => self.delete_scheduled_action(req),
280 "DeleteSnapshotCopyGrant" => self.delete_snapshot_copy_grant(req),
281 "DeleteSnapshotSchedule" => self.delete_snapshot_schedule(req),
282 "DeleteTags" => self.delete_tags(req),
283 "DeleteUsageLimit" => self.delete_usage_limit(req),
284 "DeregisterNamespace" => self.deregister_namespace(req),
285 "DescribeAccountAttributes" => self.describe_account_attributes(req),
286 "DescribeAuthenticationProfiles" => self.describe_authentication_profiles(req),
287 "DescribeClusterDbRevisions" => self.describe_cluster_db_revisions(req),
288 "DescribeClusterParameterGroups" => self.describe_cluster_parameter_groups(req),
289 "DescribeClusterParameters" => self.describe_cluster_parameters(req),
290 "DescribeClusterSecurityGroups" => self.describe_cluster_security_groups(req),
291 "DescribeClusterSnapshots" => self.describe_cluster_snapshots(req),
292 "DescribeClusterSubnetGroups" => self.describe_cluster_subnet_groups(req),
293 "DescribeClusterTracks" => self.describe_cluster_tracks(req),
294 "DescribeClusterVersions" => self.describe_cluster_versions(req),
295 "DescribeClusters" => self.describe_clusters(req),
296 "DescribeCustomDomainAssociations" => self.describe_custom_domain_associations(req),
297 "DescribeDataShares" => self.describe_data_shares(req),
298 "DescribeDataSharesForConsumer" => self.describe_data_shares_for_consumer(req),
299 "DescribeDataSharesForProducer" => self.describe_data_shares_for_producer(req),
300 "DescribeDefaultClusterParameters" => self.describe_default_cluster_parameters(req),
301 "DescribeEndpointAccess" => self.describe_endpoint_access(req),
302 "DescribeEndpointAuthorization" => self.describe_endpoint_authorization(req),
303 "DescribeEventCategories" => self.describe_event_categories(req),
304 "DescribeEventSubscriptions" => self.describe_event_subscriptions(req),
305 "DescribeEvents" => self.describe_events(req),
306 "DescribeHsmClientCertificates" => self.describe_hsm_client_certificates(req),
307 "DescribeHsmConfigurations" => self.describe_hsm_configurations(req),
308 "DescribeInboundIntegrations" => self.describe_inbound_integrations(req),
309 "DescribeIntegrations" => self.describe_integrations(req),
310 "DescribeLoggingStatus" => self.describe_logging_status(req),
311 "DescribeNodeConfigurationOptions" => self.describe_node_configuration_options(req),
312 "DescribeOrderableClusterOptions" => self.describe_orderable_cluster_options(req),
313 "DescribePartners" => self.describe_partners(req),
314 "DescribeRedshiftIdcApplications" => self.describe_redshift_idc_applications(req),
315 "DescribeReservedNodeExchangeStatus" => {
316 self.describe_reserved_node_exchange_status(req)
317 }
318 "DescribeReservedNodeOfferings" => self.describe_reserved_node_offerings(req),
319 "DescribeReservedNodes" => self.describe_reserved_nodes(req),
320 "DescribeResize" => self.describe_resize(req),
321 "DescribeScheduledActions" => self.describe_scheduled_actions(req),
322 "DescribeSnapshotCopyGrants" => self.describe_snapshot_copy_grants(req),
323 "DescribeSnapshotSchedules" => self.describe_snapshot_schedules(req),
324 "DescribeStorage" => self.describe_storage(req),
325 "DescribeTableRestoreStatus" => self.describe_table_restore_status(req),
326 "DescribeTags" => self.describe_tags(req),
327 "DescribeUsageLimits" => self.describe_usage_limits(req),
328 "DisableLogging" => self.disable_logging(req),
329 "DisableSnapshotCopy" => self.disable_snapshot_copy(req),
330 "DisassociateDataShareConsumer" => self.disassociate_data_share_consumer(req),
331 "EnableLogging" => self.enable_logging(req),
332 "EnableSnapshotCopy" => self.enable_snapshot_copy(req),
333 "FailoverPrimaryCompute" => self.failover_primary_compute(req),
334 "GetClusterCredentials" => self.get_cluster_credentials(req),
335 "GetClusterCredentialsWithIAM" => self.get_cluster_credentials_with_iam(req),
336 "GetIdentityCenterAuthToken" => self.get_identity_center_auth_token(req),
337 "GetReservedNodeExchangeConfigurationOptions" => {
338 self.get_reserved_node_exchange_configuration_options(req)
339 }
340 "GetReservedNodeExchangeOfferings" => self.get_reserved_node_exchange_offerings(req),
341 "GetResourcePolicy" => self.get_resource_policy(req),
342 "ListRecommendations" => self.list_recommendations(req),
343 "ModifyAquaConfiguration" => self.modify_aqua_configuration(req),
344 "ModifyAuthenticationProfile" => self.modify_authentication_profile(req),
345 "ModifyCluster" => self.modify_cluster(req),
346 "ModifyClusterDbRevision" => self.modify_cluster_db_revision(req),
347 "ModifyClusterIamRoles" => self.modify_cluster_iam_roles(req),
348 "ModifyClusterMaintenance" => self.modify_cluster_maintenance(req),
349 "ModifyClusterParameterGroup" => self.modify_cluster_parameter_group(req),
350 "ModifyClusterSnapshot" => self.modify_cluster_snapshot(req),
351 "ModifyClusterSnapshotSchedule" => self.modify_cluster_snapshot_schedule(req),
352 "ModifyClusterSubnetGroup" => self.modify_cluster_subnet_group(req),
353 "ModifyCustomDomainAssociation" => self.modify_custom_domain_association(req),
354 "ModifyEndpointAccess" => self.modify_endpoint_access(req),
355 "ModifyEventSubscription" => self.modify_event_subscription(req),
356 "ModifyIntegration" => self.modify_integration(req),
357 "ModifyLakehouseConfiguration" => self.modify_lakehouse_configuration(req),
358 "ModifyRedshiftIdcApplication" => self.modify_redshift_idc_application(req),
359 "ModifyScheduledAction" => self.modify_scheduled_action(req),
360 "ModifySnapshotCopyRetentionPeriod" => self.modify_snapshot_copy_retention_period(req),
361 "ModifySnapshotSchedule" => self.modify_snapshot_schedule(req),
362 "ModifyUsageLimit" => self.modify_usage_limit(req),
363 "PauseCluster" => self.pause_cluster(req),
364 "PurchaseReservedNodeOffering" => self.purchase_reserved_node_offering(req),
365 "PutResourcePolicy" => self.put_resource_policy(req),
366 "RebootCluster" => self.reboot_cluster(req),
367 "RegisterNamespace" => self.register_namespace(req),
368 "RejectDataShare" => self.reject_data_share(req),
369 "ResetClusterParameterGroup" => self.reset_cluster_parameter_group(req),
370 "ResizeCluster" => self.resize_cluster(req),
371 "RestoreFromClusterSnapshot" => self.restore_from_cluster_snapshot(req),
372 "RestoreTableFromClusterSnapshot" => self.restore_table_from_cluster_snapshot(req),
373 "ResumeCluster" => self.resume_cluster(req),
374 "RevokeClusterSecurityGroupIngress" => self.revoke_cluster_security_group_ingress(req),
375 "RevokeEndpointAccess" => self.revoke_endpoint_access(req),
376 "RevokeSnapshotAccess" => self.revoke_snapshot_access(req),
377 "RotateEncryptionKey" => self.rotate_encryption_key(req),
378 "UpdatePartnerStatus" => self.update_partner_status(req),
379 other => Err(AwsServiceError::action_not_implemented("redshift", other)),
380 }
381 }
382}
383
384pub async fn save_redshift_snapshot(
388 state: &SharedRedshiftState,
389 store: Option<Arc<dyn SnapshotStore>>,
390 lock: &AsyncMutex<()>,
391) {
392 let Some(store) = store else {
393 return;
394 };
395 let _guard = lock.lock().await;
396 let snapshot = RedshiftSnapshot {
397 schema_version: REDSHIFT_SNAPSHOT_SCHEMA_VERSION,
398 accounts: Some(state.read().clone()),
399 };
400 let join = tokio::task::spawn_blocking(move || -> std::io::Result<()> {
401 let bytes = serde_json::to_vec(&snapshot)
402 .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?;
403 store.save(&bytes)
404 })
405 .await;
406 match join {
407 Ok(Ok(())) => {}
408 Ok(Err(err)) => tracing::error!(%err, "failed to write redshift snapshot"),
409 Err(err) => tracing::error!(%err, "redshift snapshot task panicked"),
410 }
411}
412
413impl Default for RedshiftService {
414 fn default() -> Self {
415 Self::new(Arc::new(RwLock::new(RedshiftAccounts::new())))
416 }
417}
418
419#[async_trait]
420impl AwsService for RedshiftService {
421 fn service_name(&self) -> &str {
422 "redshift"
423 }
424
425 fn supported_actions(&self) -> &[&str] {
426 SUPPORTED_ACTIONS
427 }
428
429 async fn handle(&self, req: AwsRequest) -> Result<AwsResponse, AwsServiceError> {
430 let mutates = Self::is_mutating(&req.action);
431 let result = self.dispatch(&req);
432 if mutates && matches!(result.as_ref(), Ok(resp) if resp.status.is_success()) {
433 self.save_snapshot().await;
434 }
435 result
436 }
437}