Skip to main content

rusticity_core/
dynamodb.rs

1use crate::config::AwsConfig;
2use anyhow::Result;
3use futures::stream::{self, StreamExt};
4
5/// How many DescribeTable calls to run concurrently while listing tables.
6const DESCRIBE_CONCURRENCY: usize = 10;
7
8/// A DynamoDB table, flattened for display.
9#[derive(Clone, Debug, Default)]
10pub struct DdbTable {
11    pub name: String,
12    pub table_arn: String,
13    pub table_id: String,
14    /// Table status (ACTIVE, CREATING, ...).
15    pub status: String,
16    /// Partition (HASH) key, e.g. "id (S)".
17    pub partition_key: String,
18    /// Sort (RANGE) key, or empty when the table has none.
19    pub sort_key: String,
20    /// "On-demand" or "Provisioned".
21    pub read_capacity_mode: String,
22    pub write_capacity_mode: String,
23    /// Provisioned units (0 for on-demand tables).
24    pub provisioned_read_units: i64,
25    pub provisioned_write_units: i64,
26    /// "Standard" or "Standard-IA".
27    pub table_class: String,
28    /// "Enabled" / "Disabled".
29    pub deletion_protection: String,
30    pub total_size_bytes: i64,
31    pub item_count: i64,
32    pub creation_time: String,
33    /// Server-side encryption type ("Owned by Amazon DynamoDB", "AWS managed key", ...).
34    pub encryption_type: String,
35    pub kms_key_arn: String,
36    /// "Enabled" / "Disabled" for DynamoDB Streams.
37    pub stream_enabled: String,
38    pub stream_view_type: String,
39    pub latest_stream_arn: String,
40    pub global_secondary_indexes: Vec<DdbIndex>,
41    pub local_secondary_indexes: Vec<DdbIndex>,
42    pub replicas: Vec<DdbReplica>,
43}
44
45impl DdbTable {
46    /// Total number of secondary indexes (global + local).
47    pub fn index_count(&self) -> usize {
48        self.global_secondary_indexes.len() + self.local_secondary_indexes.len()
49    }
50
51    /// Comma-separated global-table replica regions ("" when not a global table).
52    pub fn replication_regions(&self) -> String {
53        self.replicas
54            .iter()
55            .map(|r| r.region.as_str())
56            .collect::<Vec<_>>()
57            .join(", ")
58    }
59}
60
61/// A secondary index (global or local) on a table.
62#[derive(Clone, Debug, Default)]
63pub struct DdbIndex {
64    pub name: String,
65    /// "Global" or "Local".
66    pub index_type: String,
67    pub status: String,
68    pub partition_key: String,
69    pub sort_key: String,
70    /// "ALL", "KEYS_ONLY" or "INCLUDE: a, b".
71    pub projection: String,
72    pub read_capacity_mode: String,
73    pub write_capacity_mode: String,
74    pub size_bytes: i64,
75    pub item_count: i64,
76    pub index_arn: String,
77}
78
79/// A global-table replica.
80#[derive(Clone, Debug, Default)]
81pub struct DdbReplica {
82    pub region: String,
83    pub status: String,
84    pub status_description: String,
85    pub table_class: String,
86    pub kms_key_id: String,
87    pub index_count: usize,
88}
89
90/// Point-in-time recovery / TTL settings loaded on demand (Additional settings tab).
91#[derive(Clone, Debug, Default)]
92pub struct DdbTableSettings {
93    /// "Enabled" / "Disabled".
94    pub ttl_status: String,
95    pub ttl_attribute: String,
96    /// "Enabled" / "Disabled".
97    pub pitr_status: String,
98    /// Backup recovery period in days (0 when unknown/disabled).
99    pub recovery_period_days: i32,
100    pub earliest_restorable_time: String,
101    pub latest_restorable_time: String,
102}
103
104/// An on-demand backup of a table (Backups tab).
105#[derive(Clone, Debug, Default)]
106pub struct DdbBackup {
107    pub name: String,
108    pub backup_arn: String,
109    pub status: String,
110    /// USER / SYSTEM / AWS_BACKUP.
111    pub backup_type: String,
112    pub size_bytes: i64,
113    pub creation_time: String,
114    pub expiry_time: String,
115}
116
117pub struct DynamoDbClient {
118    config: AwsConfig,
119}
120
121impl DynamoDbClient {
122    pub fn new(config: AwsConfig) -> Self {
123        Self { config }
124    }
125
126    /// List every table in the region, enriched with DescribeTable details.
127    pub async fn list_tables(&self) -> Result<Vec<DdbTable>> {
128        let client = self.config.dynamodb_client();
129
130        // 1. Collect the table names (paginated).
131        let mut names: Vec<String> = Vec::new();
132        let mut start: Option<String> = None;
133        loop {
134            let mut req = client.list_tables().limit(100);
135            if let Some(s) = start {
136                req = req.exclusive_start_table_name(s);
137            }
138            let resp = req.send().await?;
139            names.extend(resp.table_names().iter().cloned());
140            start = resp.last_evaluated_table_name().map(|s| s.to_string());
141            if start.is_none() {
142                break;
143            }
144        }
145
146        // 2. Describe them concurrently — DynamoDB has no bulk describe API.
147        let mut tables: Vec<DdbTable> = stream::iter(names)
148            .map(|name| {
149                let client = client.clone();
150                async move {
151                    let resp = client
152                        .describe_table()
153                        .table_name(&name)
154                        .send()
155                        .await
156                        .ok()?;
157                    resp.table().map(table_from_description)
158                }
159            })
160            .buffer_unordered(DESCRIBE_CONCURRENCY)
161            .filter_map(|t| async move { t })
162            .collect()
163            .await;
164
165        tables.sort_by(|a, b| a.name.cmp(&b.name));
166        Ok(tables)
167    }
168
169    /// TTL and point-in-time-recovery settings for a table.
170    pub async fn describe_settings(&self, table_name: &str) -> Result<DdbTableSettings> {
171        let client = self.config.dynamodb_client();
172        let mut settings = DdbTableSettings::default();
173
174        if let Ok(resp) = client
175            .describe_time_to_live()
176            .table_name(table_name)
177            .send()
178            .await
179        {
180            if let Some(ttl) = resp.time_to_live_description() {
181                settings.ttl_status = match ttl.time_to_live_status() {
182                    Some(s) => enabled_disabled(s.as_str()),
183                    None => "Disabled".to_string(),
184                };
185                settings.ttl_attribute = ttl.attribute_name().unwrap_or("").to_string();
186            }
187        }
188
189        if let Ok(resp) = client
190            .describe_continuous_backups()
191            .table_name(table_name)
192            .send()
193            .await
194        {
195            if let Some(cb) = resp.continuous_backups_description() {
196                if let Some(pitr) = cb.point_in_time_recovery_description() {
197                    settings.pitr_status = match pitr.point_in_time_recovery_status() {
198                        Some(s) => enabled_disabled(s.as_str()),
199                        None => "Disabled".to_string(),
200                    };
201                    settings.earliest_restorable_time =
202                        fmt_time(pitr.earliest_restorable_date_time());
203                    settings.latest_restorable_time = fmt_time(pitr.latest_restorable_date_time());
204                    settings.recovery_period_days = pitr.recovery_period_in_days().unwrap_or(0);
205                }
206            }
207        }
208
209        Ok(settings)
210    }
211
212    /// Resource-based policy attached to a table or stream ARN, pretty-printed.
213    /// Returns `None` when no policy exists.
214    pub async fn get_resource_policy(&self, resource_arn: &str) -> Result<Option<String>> {
215        let client = self.config.dynamodb_client();
216        match client
217            .get_resource_policy()
218            .resource_arn(resource_arn)
219            .send()
220            .await
221        {
222            Ok(resp) => Ok(resp.policy().map(pretty_json)),
223            // A missing policy is the common case, not an error.
224            Err(_) => Ok(None),
225        }
226    }
227
228    /// Tags on a table (Tags tab).
229    pub async fn list_tags(&self, table_arn: &str) -> Result<Vec<(String, String)>> {
230        let client = self.config.dynamodb_client();
231        let mut out: Vec<(String, String)> = Vec::new();
232        let mut next: Option<String> = None;
233        loop {
234            let mut req = client.list_tags_of_resource().resource_arn(table_arn);
235            if let Some(t) = next {
236                req = req.next_token(t);
237            }
238            let resp = req.send().await?;
239            for tag in resp.tags() {
240                out.push((tag.key().to_string(), tag.value().to_string()));
241            }
242            next = resp.next_token().map(|s| s.to_string());
243            if next.is_none() {
244                break;
245            }
246        }
247        out.sort_by(|a, b| a.0.cmp(&b.0));
248        Ok(out)
249    }
250
251    /// On-demand backups of a table, newest first (Backups tab).
252    pub async fn list_backups(&self, table_name: &str) -> Result<Vec<DdbBackup>> {
253        let client = self.config.dynamodb_client();
254        let mut out: Vec<DdbBackup> = Vec::new();
255        let mut start_arn: Option<String> = None;
256
257        loop {
258            let mut req = client.list_backups().table_name(table_name);
259            if let Some(a) = start_arn {
260                req = req.exclusive_start_backup_arn(a);
261            }
262            let resp = req.send().await?;
263
264            for b in resp.backup_summaries() {
265                out.push(DdbBackup {
266                    name: b.backup_name().unwrap_or("").to_string(),
267                    backup_arn: b.backup_arn().unwrap_or("").to_string(),
268                    status: b
269                        .backup_status()
270                        .map(|s| s.as_str().to_string())
271                        .unwrap_or_default(),
272                    backup_type: b
273                        .backup_type()
274                        .map(|t| t.as_str().to_string())
275                        .unwrap_or_default(),
276                    size_bytes: b.backup_size_bytes().unwrap_or(0),
277                    creation_time: fmt_time(b.backup_creation_date_time()),
278                    expiry_time: fmt_time(b.backup_expiry_date_time()),
279                });
280            }
281
282            start_arn = resp.last_evaluated_backup_arn().map(|s| s.to_string());
283            if start_arn.is_none() {
284                break;
285            }
286        }
287
288        out.sort_by(|a, b| b.creation_time.cmp(&a.creation_time));
289        Ok(out)
290    }
291
292    /// Scan items from a table or one of its indexes (Items tab).
293    pub async fn scan_items(
294        &self,
295        table_name: &str,
296        index_name: Option<&str>,
297        limit: i32,
298        projected_attributes: Option<&[String]>,
299        cursor: Option<DdbItemCursor>,
300    ) -> Result<DdbItemsPage> {
301        let client = self.config.dynamodb_client();
302        let mut req = client
303            .scan()
304            .table_name(table_name)
305            .limit(limit)
306            .set_exclusive_start_key(cursor);
307        if let Some(index) = index_name {
308            req = req.index_name(index);
309        }
310        if let Some(attrs) = projected_attributes {
311            let (expression, names) = projection_expression(attrs);
312            req = req.projection_expression(expression);
313            for (placeholder, name) in names {
314                req = req.expression_attribute_names(placeholder, name);
315            }
316        }
317        let resp = req.send().await?;
318        Ok(DdbItemsPage {
319            items: resp.items().iter().map(item_from_attributes).collect(),
320            next_cursor: resp.last_evaluated_key.filter(|key| !key.is_empty()),
321        })
322    }
323
324    /// Query items by partition-key value from a table or one of its indexes.
325    #[allow(clippy::too_many_arguments)]
326    pub async fn query_items(
327        &self,
328        table_name: &str,
329        index_name: Option<&str>,
330        partition_key: &str,
331        partition_value: &str,
332        partition_type: &str,
333        limit: i32,
334        projected_attributes: Option<&[String]>,
335        cursor: Option<DdbItemCursor>,
336    ) -> Result<DdbItemsPage> {
337        use aws_sdk_dynamodb::types::AttributeValue;
338
339        let client = self.config.dynamodb_client();
340        let value = match partition_type {
341            "N" => AttributeValue::N(partition_value.to_string()),
342            "B" => AttributeValue::S(partition_value.to_string()),
343            _ => AttributeValue::S(partition_value.to_string()),
344        };
345
346        let mut req = client
347            .query()
348            .table_name(table_name)
349            .limit(limit)
350            .set_exclusive_start_key(cursor)
351            .key_condition_expression("#pk = :pk")
352            .expression_attribute_names("#pk", partition_key)
353            .expression_attribute_values(":pk", value);
354        if let Some(index) = index_name {
355            req = req.index_name(index);
356        }
357        if let Some(attrs) = projected_attributes {
358            let (expression, names) = projection_expression(attrs);
359            req = req.projection_expression(expression);
360            for (placeholder, name) in names {
361                req = req.expression_attribute_names(placeholder, name);
362            }
363        }
364        let resp = req.send().await?;
365        Ok(DdbItemsPage {
366            items: resp.items().iter().map(item_from_attributes).collect(),
367            next_cursor: resp.last_evaluated_key.filter(|key| !key.is_empty()),
368        })
369    }
370
371    /// CloudWatch time series for the Monitor tab (last 3 hours, 1-minute period).
372    pub async fn get_monitoring_metrics(&self, table_name: &str) -> Result<DdbMonitoringData> {
373        let now = chrono::Utc::now();
374        let start = now - chrono::Duration::hours(3);
375
376        let mut data = DdbMonitoringData::default();
377
378        // Consumed capacity is emitted as a Sum per period; divide by the period
379        // to get the capacity units per second the console shows.
380        {
381            let queries = vec![
382                ddb_expr_query("e1", "m1/PERIOD(m1)", true),
383                ddb_expr_query("e2", "m2/PERIOD(m2)", true),
384                ddb_metric_query(
385                    "m1",
386                    "ConsumedReadCapacityUnits",
387                    "Sum",
388                    table_name,
389                    None,
390                    false,
391                ),
392                ddb_metric_query(
393                    "m2",
394                    "ConsumedWriteCapacityUnits",
395                    "Sum",
396                    table_name,
397                    None,
398                    false,
399                ),
400            ];
401            let mut res = self.run_get_metric_data(queries, start, now).await?;
402            data.consumed_read_capacity = res.remove("e1").unwrap_or_default();
403            data.consumed_write_capacity = res.remove("e2").unwrap_or_default();
404        }
405
406        // Throttling.
407        {
408            let queries = vec![
409                ddb_metric_query("m1", "ReadThrottleEvents", "Sum", table_name, None, true),
410                ddb_metric_query("m2", "WriteThrottleEvents", "Sum", table_name, None, true),
411                ddb_metric_query("m3", "ThrottledRequests", "Sum", table_name, None, true),
412            ];
413            let mut res = self.run_get_metric_data(queries, start, now).await?;
414            data.read_throttle_events = res.remove("m1").unwrap_or_default();
415            data.write_throttle_events = res.remove("m2").unwrap_or_default();
416            data.throttled_requests = res.remove("m3").unwrap_or_default();
417        }
418
419        // Latency per operation.
420        {
421            let queries = vec![
422                ddb_metric_query(
423                    "m1",
424                    "SuccessfulRequestLatency",
425                    "Average",
426                    table_name,
427                    Some(("Operation", "GetItem")),
428                    true,
429                ),
430                ddb_metric_query(
431                    "m2",
432                    "SuccessfulRequestLatency",
433                    "Average",
434                    table_name,
435                    Some(("Operation", "PutItem")),
436                    true,
437                ),
438                ddb_metric_query(
439                    "m3",
440                    "SuccessfulRequestLatency",
441                    "Average",
442                    table_name,
443                    Some(("Operation", "Query")),
444                    true,
445                ),
446                ddb_metric_query(
447                    "m4",
448                    "SuccessfulRequestLatency",
449                    "Average",
450                    table_name,
451                    Some(("Operation", "Scan")),
452                    true,
453                ),
454            ];
455            let mut res = self.run_get_metric_data(queries, start, now).await?;
456            data.latency_get_item = res.remove("m1").unwrap_or_default();
457            data.latency_put_item = res.remove("m2").unwrap_or_default();
458            data.latency_query = res.remove("m3").unwrap_or_default();
459            data.latency_scan = res.remove("m4").unwrap_or_default();
460        }
461
462        Ok(data)
463    }
464
465    async fn run_get_metric_data(
466        &self,
467        queries: Vec<aws_sdk_cloudwatch::types::MetricDataQuery>,
468        start_time: chrono::DateTime<chrono::Utc>,
469        end_time: chrono::DateTime<chrono::Utc>,
470    ) -> Result<std::collections::HashMap<String, Vec<(i64, f64)>>> {
471        let client = self.config.cloudwatch_client();
472
473        let mut req = client
474            .get_metric_data()
475            .start_time(aws_sdk_cloudwatch::primitives::DateTime::from_millis(
476                start_time.timestamp_millis(),
477            ))
478            .end_time(aws_sdk_cloudwatch::primitives::DateTime::from_millis(
479                end_time.timestamp_millis(),
480            ));
481        for q in queries {
482            req = req.metric_data_queries(q);
483        }
484
485        let resp = req.send().await?;
486
487        let mut out: std::collections::HashMap<String, Vec<(i64, f64)>> =
488            std::collections::HashMap::new();
489        for r in resp.metric_data_results() {
490            let Some(id) = r.id() else { continue };
491            let mut points: Vec<(i64, f64)> = r
492                .timestamps()
493                .iter()
494                .zip(r.values().iter())
495                .map(|(t, v)| (t.secs(), *v))
496                .collect();
497            points.sort_by_key(|(t, _)| *t);
498            out.insert(id.to_string(), points);
499        }
500        Ok(out)
501    }
502}
503
504/// Time series for each DynamoDB Monitor graph. Empty vectors render as "no data".
505#[derive(Default, Clone, Debug)]
506pub struct DdbMonitoringData {
507    /// Consumed capacity units per second.
508    pub consumed_read_capacity: Vec<(i64, f64)>,
509    pub consumed_write_capacity: Vec<(i64, f64)>,
510    /// Throttling counts.
511    pub read_throttle_events: Vec<(i64, f64)>,
512    pub write_throttle_events: Vec<(i64, f64)>,
513    pub throttled_requests: Vec<(i64, f64)>,
514    /// Average successful request latency per operation, in milliseconds.
515    pub latency_get_item: Vec<(i64, f64)>,
516    pub latency_put_item: Vec<(i64, f64)>,
517    pub latency_query: Vec<(i64, f64)>,
518    pub latency_scan: Vec<(i64, f64)>,
519}
520
521/// Resume key kept in the SDK's original types, never reconstructed from display values.
522/// In particular, binary keys and numeric strings must survive pagination unchanged.
523pub type DdbItemCursor = std::collections::HashMap<String, aws_sdk_dynamodb::types::AttributeValue>;
524
525/// Exactly one bounded Scan/Query response, including an optional continuation key.
526#[derive(Debug, Default)]
527pub struct DdbItemsPage {
528    pub items: Vec<DdbItem>,
529    pub next_cursor: Option<DdbItemCursor>,
530}
531
532/// A single item returned by Scan or Query, with values rendered for display.
533#[derive(Clone, Debug, Default)]
534pub struct DdbItem {
535    /// Attribute name -> displayable value, ordered by attribute name.
536    pub attributes: std::collections::BTreeMap<String, String>,
537}
538
539impl DdbItem {
540    /// Displayable value for an attribute, or an empty string when absent.
541    pub fn get(&self, attribute: &str) -> &str {
542        self.attributes
543            .get(attribute)
544            .map(|v| v.as_str())
545            .unwrap_or("")
546    }
547}
548
549fn item_from_attributes(
550    item: &std::collections::HashMap<String, aws_sdk_dynamodb::types::AttributeValue>,
551) -> DdbItem {
552    DdbItem {
553        attributes: item
554            .iter()
555            .map(|(k, v)| (k.clone(), attribute_to_string(v)))
556            .collect(),
557    }
558}
559
560/// Render an `AttributeValue` the way the console item explorer does.
561pub fn attribute_to_string(value: &aws_sdk_dynamodb::types::AttributeValue) -> String {
562    use aws_sdk_dynamodb::types::AttributeValue as Av;
563    match value {
564        Av::S(s) => s.clone(),
565        Av::N(n) => n.clone(),
566        Av::Bool(b) => b.to_string(),
567        Av::Null(_) => "null".to_string(),
568        Av::B(b) => format!("<binary {} bytes>", b.as_ref().len()),
569        Av::Ss(items) => format!("[{}]", items.join(", ")),
570        Av::Ns(items) => format!("[{}]", items.join(", ")),
571        Av::Bs(items) => format!("[{} binary values]", items.len()),
572        Av::L(items) => format!(
573            "[{}]",
574            items
575                .iter()
576                .map(attribute_to_string)
577                .collect::<Vec<_>>()
578                .join(", ")
579        ),
580        Av::M(map) => {
581            let mut pairs: Vec<String> = map
582                .iter()
583                .map(|(k, v)| format!("{}: {}", k, attribute_to_string(v)))
584                .collect();
585            pairs.sort();
586            format!("{{{}}}", pairs.join(", "))
587        }
588        _ => String::new(),
589    }
590}
591
592/// Build a `ProjectionExpression` with placeholders so reserved words are safe.
593fn projection_expression(attributes: &[String]) -> (String, Vec<(String, String)>) {
594    let names: Vec<(String, String)> = attributes
595        .iter()
596        .enumerate()
597        .map(|(i, a)| (format!("#p{}", i), a.clone()))
598        .collect();
599    let expression = names
600        .iter()
601        .map(|(placeholder, _)| placeholder.clone())
602        .collect::<Vec<_>>()
603        .join(", ");
604    (expression, names)
605}
606
607/// Metric period used by every Monitor graph, in seconds.
608const METRIC_PERIOD: i32 = 60;
609
610/// Build a raw metric `MetricDataQuery` for the `AWS/DynamoDB` namespace.
611fn ddb_metric_query(
612    id: &str,
613    metric_name: &str,
614    stat: &str,
615    table_name: &str,
616    extra_dimension: Option<(&str, &str)>,
617    return_data: bool,
618) -> aws_sdk_cloudwatch::types::MetricDataQuery {
619    use aws_sdk_cloudwatch::types::{Dimension, Metric, MetricDataQuery, MetricStat};
620
621    let mut metric_b = Metric::builder()
622        .namespace("AWS/DynamoDB")
623        .metric_name(metric_name)
624        .dimensions(
625            Dimension::builder()
626                .name("TableName")
627                .value(table_name)
628                .build(),
629        );
630    if let Some((k, v)) = extra_dimension {
631        metric_b = metric_b.dimensions(Dimension::builder().name(k).value(v).build());
632    }
633
634    let stat = MetricStat::builder()
635        .metric(metric_b.build())
636        .period(METRIC_PERIOD)
637        .stat(stat)
638        .build();
639
640    MetricDataQuery::builder()
641        .id(id)
642        .metric_stat(stat)
643        .return_data(return_data)
644        .build()
645}
646
647/// Build a metric-math expression `MetricDataQuery`.
648fn ddb_expr_query(
649    id: &str,
650    expression: &str,
651    return_data: bool,
652) -> aws_sdk_cloudwatch::types::MetricDataQuery {
653    aws_sdk_cloudwatch::types::MetricDataQuery::builder()
654        .id(id)
655        .expression(expression)
656        .period(METRIC_PERIOD)
657        .return_data(return_data)
658        .build()
659}
660
661/// Pretty-print a JSON policy document; falls back to the raw string.
662fn pretty_json(raw: &str) -> String {
663    serde_json::from_str::<serde_json::Value>(raw)
664        .ok()
665        .and_then(|v| serde_json::to_string_pretty(&v).ok())
666        .unwrap_or_else(|| raw.to_string())
667}
668
669fn fmt_time(t: Option<&aws_smithy_types::DateTime>) -> String {
670    t.and_then(|t| t.fmt(aws_smithy_types::date_time::Format::DateTime).ok())
671        .unwrap_or_default()
672}
673
674/// Normalize an SDK status debug string to "Enabled"/"Disabled" where possible.
675fn enabled_disabled(raw: &str) -> String {
676    match raw.to_uppercase().as_str() {
677        "ENABLED" | "ENABLING" => "Enabled".to_string(),
678        "DISABLED" | "DISABLING" => "Disabled".to_string(),
679        other => other.to_string(),
680    }
681}
682
683/// Render a key schema as "name (TYPE)" for the partition and sort keys.
684fn key_names(
685    key_schema: &[aws_sdk_dynamodb::types::KeySchemaElement],
686    attribute_types: &std::collections::HashMap<String, String>,
687) -> (String, String) {
688    let mut partition = String::new();
689    let mut sort = String::new();
690    for k in key_schema {
691        let name = k.attribute_name();
692        let label = match attribute_types.get(name) {
693            Some(t) => format!("{} ({})", name, t),
694            None => name.to_string(),
695        };
696        match k.key_type() {
697            aws_sdk_dynamodb::types::KeyType::Hash => partition = label,
698            aws_sdk_dynamodb::types::KeyType::Range => sort = label,
699            _ => {}
700        }
701    }
702    (partition, sort)
703}
704
705/// "ALL", "KEYS_ONLY" or "INCLUDE: a, b".
706fn projection_of(projection: Option<&aws_sdk_dynamodb::types::Projection>) -> String {
707    let Some(p) = projection else {
708        return String::new();
709    };
710    let kind = p
711        .projection_type()
712        .map(|t| t.as_str().to_string())
713        .unwrap_or_default();
714    if kind == "INCLUDE" && !p.non_key_attributes().is_empty() {
715        format!("INCLUDE: {}", p.non_key_attributes().join(", "))
716    } else {
717        kind
718    }
719}
720
721/// "On-demand" when no provisioned units are set, "Provisioned" otherwise.
722fn capacity_mode(units: i64, billing_on_demand: bool) -> String {
723    if billing_on_demand || units == 0 {
724        "On-demand".to_string()
725    } else {
726        "Provisioned".to_string()
727    }
728}
729
730fn table_from_description(t: &aws_sdk_dynamodb::types::TableDescription) -> DdbTable {
731    use std::collections::HashMap;
732
733    // Attribute name -> "S" / "N" / "B", used to label the key columns.
734    let attribute_types: HashMap<String, String> = t
735        .attribute_definitions()
736        .iter()
737        .map(|a| {
738            (
739                a.attribute_name().to_string(),
740                a.attribute_type().as_str().to_string(),
741            )
742        })
743        .collect();
744
745    let (partition_key, sort_key) = key_names(t.key_schema(), &attribute_types);
746
747    let billing_on_demand = t
748        .billing_mode_summary()
749        .and_then(|b| b.billing_mode())
750        .map(|m| m.as_str() == "PAY_PER_REQUEST")
751        .unwrap_or(false);
752
753    let provisioned_read_units = t
754        .provisioned_throughput()
755        .and_then(|p| p.read_capacity_units())
756        .unwrap_or(0);
757    let provisioned_write_units = t
758        .provisioned_throughput()
759        .and_then(|p| p.write_capacity_units())
760        .unwrap_or(0);
761
762    let global_secondary_indexes: Vec<DdbIndex> = t
763        .global_secondary_indexes()
764        .iter()
765        .map(|g| {
766            let (pk, sk) = key_names(g.key_schema(), &attribute_types);
767            let read = g
768                .provisioned_throughput()
769                .and_then(|p| p.read_capacity_units())
770                .unwrap_or(0);
771            let write = g
772                .provisioned_throughput()
773                .and_then(|p| p.write_capacity_units())
774                .unwrap_or(0);
775            DdbIndex {
776                name: g.index_name().unwrap_or("").to_string(),
777                index_type: "Global".to_string(),
778                status: g
779                    .index_status()
780                    .map(|s| s.as_str().to_string())
781                    .unwrap_or_default(),
782                partition_key: pk,
783                sort_key: sk,
784                projection: projection_of(g.projection()),
785                read_capacity_mode: capacity_mode(read, billing_on_demand),
786                write_capacity_mode: capacity_mode(write, billing_on_demand),
787                size_bytes: g.index_size_bytes().unwrap_or(0),
788                item_count: g.item_count().unwrap_or(0),
789                index_arn: g.index_arn().unwrap_or("").to_string(),
790            }
791        })
792        .collect();
793
794    let local_secondary_indexes: Vec<DdbIndex> = t
795        .local_secondary_indexes()
796        .iter()
797        .map(|l| {
798            let (pk, sk) = key_names(l.key_schema(), &attribute_types);
799            DdbIndex {
800                name: l.index_name().unwrap_or("").to_string(),
801                index_type: "Local".to_string(),
802                status: "ACTIVE".to_string(),
803                partition_key: pk,
804                sort_key: sk,
805                projection: projection_of(l.projection()),
806                read_capacity_mode: capacity_mode(provisioned_read_units, billing_on_demand),
807                write_capacity_mode: capacity_mode(provisioned_write_units, billing_on_demand),
808                size_bytes: l.index_size_bytes().unwrap_or(0),
809                item_count: l.item_count().unwrap_or(0),
810                index_arn: l.index_arn().unwrap_or("").to_string(),
811            }
812        })
813        .collect();
814
815    let replicas: Vec<DdbReplica> = t
816        .replicas()
817        .iter()
818        .map(|r| DdbReplica {
819            region: r.region_name().unwrap_or("").to_string(),
820            status: r
821                .replica_status()
822                .map(|s| s.as_str().to_string())
823                .unwrap_or_default(),
824            status_description: r.replica_status_description().unwrap_or("").to_string(),
825            table_class: r
826                .replica_table_class_summary()
827                .and_then(|c| c.table_class())
828                .map(|c| format_table_class(c.as_str()))
829                .unwrap_or_default(),
830            kms_key_id: r.kms_master_key_id().unwrap_or("").to_string(),
831            index_count: r.global_secondary_indexes().len(),
832        })
833        .collect();
834
835    let (stream_enabled, stream_view_type) = match t.stream_specification() {
836        Some(s) if s.stream_enabled() => (
837            "Enabled".to_string(),
838            s.stream_view_type()
839                .map(|v| v.as_str().to_string())
840                .unwrap_or_default(),
841        ),
842        _ => ("Disabled".to_string(), String::new()),
843    };
844
845    let (encryption_type, kms_key_arn) = match t.sse_description() {
846        Some(sse) => (
847            match sse.sse_type().map(|t| t.as_str().to_string()) {
848                Some(ref t) if t.eq_ignore_ascii_case("KMS") => "AWS managed key".to_string(),
849                Some(other) => other,
850                None => "Owned by Amazon DynamoDB".to_string(),
851            },
852            sse.kms_master_key_arn().unwrap_or("").to_string(),
853        ),
854        None => ("Owned by Amazon DynamoDB".to_string(), String::new()),
855    };
856
857    DdbTable {
858        name: t.table_name().unwrap_or("").to_string(),
859        table_arn: t.table_arn().unwrap_or("").to_string(),
860        table_id: t.table_id().unwrap_or("").to_string(),
861        status: t
862            .table_status()
863            .map(|s| s.as_str().to_string())
864            .unwrap_or_default(),
865        partition_key,
866        sort_key,
867        read_capacity_mode: capacity_mode(provisioned_read_units, billing_on_demand),
868        write_capacity_mode: capacity_mode(provisioned_write_units, billing_on_demand),
869        provisioned_read_units,
870        provisioned_write_units,
871        table_class: t
872            .table_class_summary()
873            .and_then(|c| c.table_class())
874            .map(|c| format_table_class(c.as_str()))
875            .unwrap_or_else(|| "Standard".to_string()),
876        deletion_protection: match t.deletion_protection_enabled() {
877            Some(true) => "Enabled".to_string(),
878            _ => "Disabled".to_string(),
879        },
880        total_size_bytes: t.table_size_bytes().unwrap_or(0),
881        item_count: t.item_count().unwrap_or(0),
882        creation_time: fmt_time(t.creation_date_time()),
883        encryption_type,
884        kms_key_arn,
885        stream_enabled,
886        stream_view_type,
887        latest_stream_arn: t.latest_stream_arn().unwrap_or("").to_string(),
888        global_secondary_indexes,
889        local_secondary_indexes,
890        replicas,
891    }
892}
893
894/// "StandardInfrequentAccess" -> "Standard-IA", "Standard" -> "Standard".
895pub fn format_table_class(raw: &str) -> String {
896    match raw.to_uppercase().replace(['_', '-'], "").as_str() {
897        "STANDARDINFREQUENTACCESS" | "DYNAMODBSTANDARDINFREQUENTACCESS" => {
898            "Standard-IA".to_string()
899        }
900        "STANDARD" | "DYNAMODBSTANDARD" => "Standard".to_string(),
901        _ => raw.to_string(),
902    }
903}
904
905#[cfg(test)]
906mod tests {
907    use super::*;
908
909    #[test]
910    fn test_format_table_class() {
911        assert_eq!(format_table_class("Standard"), "Standard");
912        assert_eq!(format_table_class("DynamodbStandard"), "Standard");
913        assert_eq!(
914            format_table_class("StandardInfrequentAccess"),
915            "Standard-IA"
916        );
917        assert_eq!(
918            format_table_class("DYNAMODB_STANDARD_INFREQUENT_ACCESS"),
919            "Standard-IA"
920        );
921    }
922
923    #[test]
924    fn test_capacity_mode() {
925        assert_eq!(capacity_mode(0, true), "On-demand");
926        assert_eq!(capacity_mode(5, true), "On-demand");
927        assert_eq!(capacity_mode(0, false), "On-demand");
928        assert_eq!(capacity_mode(5, false), "Provisioned");
929    }
930
931    #[test]
932    fn test_enabled_disabled() {
933        assert_eq!(enabled_disabled("ENABLED"), "Enabled");
934        assert_eq!(enabled_disabled("Enabling"), "Enabled");
935        assert_eq!(enabled_disabled("DISABLED"), "Disabled");
936        assert_eq!(enabled_disabled("UNKNOWN"), "UNKNOWN");
937    }
938
939    #[test]
940    fn test_index_count_sums_global_and_local() {
941        let table = DdbTable {
942            global_secondary_indexes: vec![DdbIndex::default(), DdbIndex::default()],
943            local_secondary_indexes: vec![DdbIndex::default()],
944            ..Default::default()
945        };
946        assert_eq!(table.index_count(), 3);
947    }
948}