dynoxide/actions/
mod.rs

1pub mod batch_execute_statement;
2pub mod batch_get_item;
3pub mod batch_write_item;
4pub mod create_table;
5pub mod delete_item;
6pub mod delete_table;
7pub mod describe_stream;
8pub mod describe_table;
9pub mod describe_time_to_live;
10pub mod execute_statement;
11pub mod execute_transaction;
12pub mod get_item;
13pub mod get_records;
14pub mod get_shard_iterator;
15pub(crate) mod gsi;
16pub(crate) mod helpers;
17pub mod import_items;
18// Public because `partiql::executor::StatementPage` carries a `WriteCapacity`
19// on a public field, and a consumer reading it has to be able to name the type.
20pub mod index_capacity;
21pub mod list_streams;
22pub mod list_tables;
23pub mod list_tags_of_resource;
24pub(crate) mod lsi;
25pub mod put_item;
26pub mod query;
27pub mod scan;
28pub mod search_vectors;
29pub mod tag_resource;
30pub mod transact_get_items;
31pub mod transact_write_items;
32pub mod untag_resource;
33pub mod update_item;
34pub mod update_table;
35pub mod update_time_to_live;
36pub(crate) mod vector_index;
37// Public because the four actions that take a `VectorIndexLifecycle` are
38// themselves public, the way `TokenCaches` is public for the transactional
39// ones. Only the handle is exposed; the phase derivation stays internal.
40pub mod vector_lifecycle;
41
42use crate::types::{
43    AttributeDefinition, GlobalSecondaryIndex, KeySchemaElement, LocalSecondaryIndex, Projection,
44    SearchSchemaElement, VectorAttributeDefinition, VectorIndex,
45};
46use serde::{Deserialize, Serialize};
47
48/// Full table description returned by DescribeTable and CreateTable.
49#[derive(Debug, Clone, Default, Serialize, Deserialize)]
50pub struct TableDescription {
51    #[serde(rename = "TableName")]
52    pub table_name: String,
53
54    #[serde(rename = "TableId", skip_serializing_if = "Option::is_none")]
55    pub table_id: Option<String>,
56
57    #[serde(rename = "TableArn")]
58    pub table_arn: String,
59
60    #[serde(rename = "TableStatus")]
61    pub table_status: String,
62
63    #[serde(rename = "KeySchema")]
64    pub key_schema: Vec<KeySchemaElement>,
65
66    #[serde(rename = "AttributeDefinitions")]
67    pub attribute_definitions: Vec<AttributeDefinition>,
68
69    #[serde(rename = "CreationDateTime", skip_serializing_if = "Option::is_none")]
70    pub creation_date_time: Option<f64>,
71
72    #[serde(rename = "ItemCount", skip_serializing_if = "Option::is_none")]
73    pub item_count: Option<i64>,
74
75    #[serde(rename = "TableSizeBytes", skip_serializing_if = "Option::is_none")]
76    pub table_size_bytes: Option<i64>,
77
78    #[serde(
79        rename = "ProvisionedThroughput",
80        skip_serializing_if = "Option::is_none"
81    )]
82    pub provisioned_throughput: Option<TableProvisionedThroughputDescription>,
83
84    #[serde(rename = "BillingModeSummary", skip_serializing_if = "Option::is_none")]
85    pub billing_mode_summary: Option<BillingModeSummary>,
86
87    #[serde(
88        rename = "TableThroughputModeSummary",
89        skip_serializing_if = "Option::is_none"
90    )]
91    pub table_throughput_mode_summary: Option<TableThroughputModeSummary>,
92
93    #[serde(
94        rename = "GlobalSecondaryIndexes",
95        skip_serializing_if = "Option::is_none"
96    )]
97    pub global_secondary_indexes: Option<Vec<GlobalSecondaryIndexDescription>>,
98
99    #[serde(
100        rename = "LocalSecondaryIndexes",
101        skip_serializing_if = "Option::is_none"
102    )]
103    pub local_secondary_indexes: Option<Vec<LocalSecondaryIndexDescription>>,
104
105    #[serde(rename = "VectorIndexes", skip_serializing_if = "Option::is_none")]
106    pub vector_indexes: Option<Vec<VectorIndexDescription>>,
107
108    #[serde(
109        rename = "StreamSpecification",
110        skip_serializing_if = "Option::is_none"
111    )]
112    pub stream_specification: Option<StreamSpecificationDescription>,
113
114    #[serde(rename = "LatestStreamArn", skip_serializing_if = "Option::is_none")]
115    pub latest_stream_arn: Option<String>,
116
117    #[serde(rename = "LatestStreamLabel", skip_serializing_if = "Option::is_none")]
118    pub latest_stream_label: Option<String>,
119
120    #[serde(rename = "SSEDescription", skip_serializing_if = "Option::is_none")]
121    pub sse_description: Option<SseDescription>,
122
123    #[serde(rename = "TableClassSummary", skip_serializing_if = "Option::is_none")]
124    pub table_class_summary: Option<TableClassSummary>,
125
126    #[serde(
127        rename = "DeletionProtectionEnabled",
128        skip_serializing_if = "Option::is_none"
129    )]
130    pub deletion_protection_enabled: Option<bool>,
131
132    #[serde(rename = "OnDemandThroughput", skip_serializing_if = "Option::is_none")]
133    pub on_demand_throughput: Option<crate::types::OnDemandThroughput>,
134}
135
136/// SSE description returned in TableDescription.
137#[derive(Debug, Clone, Default, Serialize, Deserialize)]
138pub struct SseDescription {
139    #[serde(rename = "Status")]
140    pub status: String,
141    #[serde(rename = "SSEType", skip_serializing_if = "Option::is_none")]
142    pub sse_type: Option<String>,
143    #[serde(rename = "KMSMasterKeyArn", skip_serializing_if = "Option::is_none")]
144    pub kms_master_key_arn: Option<String>,
145}
146
147/// Stream specification description returned in TableDescription.
148#[derive(Debug, Clone, Default, Serialize, Deserialize)]
149pub struct StreamSpecificationDescription {
150    #[serde(rename = "StreamEnabled")]
151    pub stream_enabled: bool,
152    #[serde(rename = "StreamViewType", skip_serializing_if = "Option::is_none")]
153    pub stream_view_type: Option<String>,
154}
155
156/// Table class summary returned in TableDescription.
157#[derive(Debug, Clone, Default, Serialize, Deserialize)]
158pub struct TableClassSummary {
159    #[serde(rename = "TableClass")]
160    pub table_class: String,
161}
162
163/// Provisioned throughput description (with additional metadata fields).
164#[derive(Debug, Clone, Default, Serialize, Deserialize)]
165pub struct TableProvisionedThroughputDescription {
166    #[serde(rename = "ReadCapacityUnits")]
167    pub read_capacity_units: u64,
168    #[serde(rename = "WriteCapacityUnits")]
169    pub write_capacity_units: u64,
170    #[serde(rename = "NumberOfDecreasesToday")]
171    pub number_of_decreases_today: u64,
172    #[serde(
173        rename = "LastIncreaseDateTime",
174        skip_serializing_if = "Option::is_none"
175    )]
176    pub last_increase_date_time: Option<f64>,
177    #[serde(
178        rename = "LastDecreaseDateTime",
179        skip_serializing_if = "Option::is_none"
180    )]
181    pub last_decrease_date_time: Option<f64>,
182}
183
184/// Billing mode summary.
185#[derive(Debug, Clone, Default, Serialize, Deserialize)]
186pub struct BillingModeSummary {
187    #[serde(rename = "BillingMode")]
188    pub billing_mode: String,
189    #[serde(
190        rename = "LastUpdateToPayPerRequestDateTime",
191        skip_serializing_if = "Option::is_none"
192    )]
193    pub last_update_to_pay_per_request_date_time: Option<f64>,
194}
195
196/// Table throughput mode summary.
197#[derive(Debug, Clone, Default, Serialize, Deserialize)]
198pub struct TableThroughputModeSummary {
199    #[serde(rename = "TableThroughputMode")]
200    pub table_throughput_mode: String,
201    #[serde(
202        rename = "LastUpdateToPayPerRequestDateTime",
203        skip_serializing_if = "Option::is_none"
204    )]
205    pub last_update_to_pay_per_request_date_time: Option<f64>,
206}
207
208/// GSI description (returned in TableDescription).
209#[derive(Debug, Clone, Default, Serialize, Deserialize)]
210pub struct GlobalSecondaryIndexDescription {
211    #[serde(rename = "IndexName")]
212    pub index_name: String,
213    #[serde(rename = "IndexArn")]
214    pub index_arn: String,
215    #[serde(rename = "KeySchema")]
216    pub key_schema: Vec<KeySchemaElement>,
217    #[serde(rename = "Projection")]
218    pub projection: Projection,
219    #[serde(rename = "IndexStatus")]
220    pub index_status: String,
221    #[serde(
222        rename = "ProvisionedThroughput",
223        skip_serializing_if = "Option::is_none"
224    )]
225    pub provisioned_throughput: Option<TableProvisionedThroughputDescription>,
226    #[serde(rename = "ItemCount", skip_serializing_if = "Option::is_none")]
227    pub item_count: Option<i64>,
228    #[serde(rename = "IndexSizeBytes", skip_serializing_if = "Option::is_none")]
229    pub index_size_bytes: Option<i64>,
230}
231
232/// LSI description (returned in TableDescription).
233#[derive(Debug, Clone, Default, Serialize, Deserialize)]
234pub struct LocalSecondaryIndexDescription {
235    #[serde(rename = "IndexName")]
236    pub index_name: String,
237    #[serde(rename = "IndexArn")]
238    pub index_arn: String,
239    #[serde(rename = "KeySchema")]
240    pub key_schema: Vec<KeySchemaElement>,
241    #[serde(rename = "Projection")]
242    pub projection: Projection,
243    #[serde(rename = "ItemCount", skip_serializing_if = "Option::is_none")]
244    pub item_count: Option<i64>,
245    #[serde(rename = "IndexSizeBytes", skip_serializing_if = "Option::is_none")]
246    pub index_size_bytes: Option<i64>,
247}
248
249/// Vector index description (returned in TableDescription).
250///
251/// Field set per the AWS response shape: IndexName, SearchSchema (when
252/// declared), Projection, VectorAttribute, Dimensions, DistanceFunction,
253/// IndexStatus, IndexSizeBytes, ItemCount, IndexArn. `Backfilling` is never
254/// present on the CreateTable path, so it stays `None` here and is only
255/// populated by paths that observe a backfill.
256#[derive(Debug, Clone, Default, Serialize, Deserialize)]
257pub struct VectorIndexDescription {
258    #[serde(rename = "IndexName")]
259    pub index_name: String,
260    #[serde(rename = "SearchSchema", skip_serializing_if = "Option::is_none")]
261    pub search_schema: Option<Vec<SearchSchemaElement>>,
262    #[serde(rename = "Projection")]
263    pub projection: Projection,
264    #[serde(rename = "VectorAttribute")]
265    pub vector_attribute: VectorAttributeDefinition,
266    #[serde(rename = "Dimensions")]
267    pub dimensions: u32,
268    #[serde(rename = "DistanceFunction")]
269    pub distance_function: String,
270    #[serde(rename = "IndexStatus")]
271    pub index_status: String,
272    #[serde(rename = "Backfilling", skip_serializing_if = "Option::is_none")]
273    pub backfilling: Option<bool>,
274    #[serde(rename = "IndexSizeBytes", skip_serializing_if = "Option::is_none")]
275    pub index_size_bytes: Option<i64>,
276    #[serde(rename = "ItemCount", skip_serializing_if = "Option::is_none")]
277    pub item_count: Option<i64>,
278    #[serde(rename = "IndexArn")]
279    pub index_arn: String,
280}
281
282/// Deterministic fallback `TableId` for a table that has no persisted id.
283///
284/// The primary scheme persists a random v4 UUID assigned at create time (see
285/// `Storage::insert_table_metadata` and the v8 migration's backfill), matching
286/// AWS: stable across reads, and a new id for a dropped-and-recreated table.
287/// This fallback only applies to a row whose stored `table_id` is `None`: for
288/// example one written by an older binary into a newer database. It derives a
289/// stable UUID (v5) from the table name and creation timestamp so such a row
290/// still reports a consistent id across reads, rather than the per-call random
291/// value that #55 fixed.
292fn table_id_for(table_name: &str, created_at: i64) -> String {
293    uuid::Uuid::new_v5(
294        &uuid::Uuid::NAMESPACE_OID,
295        format!("dynoxide:table:{table_name}:{created_at}").as_bytes(),
296    )
297    .to_string()
298}
299
300/// Helper: Build a TableDescription from stored metadata.
301///
302/// `vector_phases` carries where each vector index sits in its creation
303/// lifecycle, resolved once by the caller. Taking it as an input rather than
304/// reading it here is what stops two responses describing the same table at
305/// the same instant from disagreeing: an `UpdateTable` touching a GSI while a
306/// vector index is creating returns a full table description too.
307pub(crate) fn build_table_description(
308    meta: &crate::storage::TableMetadata,
309    item_count: Option<i64>,
310    table_size_bytes: Option<i64>,
311    vector_phases: &vector_lifecycle::VectorIndexPhases,
312) -> TableDescription {
313    use crate::streams;
314
315    let key_schema: Vec<KeySchemaElement> =
316        serde_json::from_str(&meta.key_schema).unwrap_or_default();
317    let attribute_definitions: Vec<AttributeDefinition> =
318        serde_json::from_str(&meta.attribute_definitions).unwrap_or_default();
319
320    let gsi_definitions: Option<Vec<GlobalSecondaryIndex>> = meta
321        .gsi_definitions
322        .as_ref()
323        .and_then(|s| serde_json::from_str(s).ok());
324
325    let table_name = &meta.table_name;
326
327    let global_secondary_indexes = gsi_definitions.map(|gsis| {
328        gsis.into_iter()
329            .map(|gsi| {
330                let idx_arn = streams::index_arn(table_name, &gsi.index_name);
331                GlobalSecondaryIndexDescription {
332                    index_name: gsi.index_name,
333                    index_arn: idx_arn,
334                    key_schema: gsi.key_schema,
335                    projection: gsi.projection,
336                    index_status: "ACTIVE".to_string(),
337                    provisioned_throughput: Some(if let Some(pt) = gsi.provisioned_throughput {
338                        TableProvisionedThroughputDescription {
339                            read_capacity_units: pt.read_capacity_units.unwrap_or(0) as u64,
340                            write_capacity_units: pt.write_capacity_units.unwrap_or(0) as u64,
341                            number_of_decreases_today: 0,
342                            last_increase_date_time: None,
343                            last_decrease_date_time: None,
344                        }
345                    } else {
346                        // PAY_PER_REQUEST or no PT specified
347                        TableProvisionedThroughputDescription {
348                            read_capacity_units: 0,
349                            write_capacity_units: 0,
350                            number_of_decreases_today: 0,
351                            last_increase_date_time: None,
352                            last_decrease_date_time: None,
353                        }
354                    }),
355                    item_count: Some(0),
356                    index_size_bytes: Some(0),
357                }
358            })
359            .collect()
360    });
361
362    let lsi_definitions: Option<Vec<LocalSecondaryIndex>> = meta
363        .lsi_definitions
364        .as_ref()
365        .and_then(|s| serde_json::from_str(s).ok());
366
367    let local_secondary_indexes = lsi_definitions.map(|lsis| {
368        lsis.into_iter()
369            .map(|lsi| {
370                let idx_arn = streams::index_arn(table_name, &lsi.index_name);
371                LocalSecondaryIndexDescription {
372                    index_name: lsi.index_name,
373                    index_arn: idx_arn,
374                    key_schema: lsi.key_schema,
375                    projection: lsi.projection,
376                    item_count: Some(0),
377                    index_size_bytes: Some(0),
378                }
379            })
380            .collect()
381    });
382
383    let vector_index_definitions: Option<Vec<VectorIndex>> = meta
384        .vector_index_definitions
385        .as_ref()
386        .and_then(|s| serde_json::from_str(s).ok());
387
388    // Status and Backfilling both come from the index's lifecycle phase, so an
389    // index created with its table reports ACTIVE with the field absent (the
390    // CreateTable path, captured from real DynamoDB, eu-west-2, 2026-08-11)
391    // while one added to a live table walks CREATING first.
392    let vector_indexes = vector_index_definitions.map(|vixs| {
393        vixs.into_iter()
394            .map(|vix| {
395                let idx_arn = streams::index_arn(table_name, &vix.index_name);
396                let phase = vector_phases.get(&vix.index_name);
397                VectorIndexDescription {
398                    index_name: vix.index_name,
399                    search_schema: vix.search_schema,
400                    projection: vix.projection,
401                    vector_attribute: vix.vector_attribute,
402                    dimensions: vix.dimensions,
403                    distance_function: vix.distance_function,
404                    index_status: phase.index_status().to_string(),
405                    backfilling: phase.backfilling(),
406                    index_size_bytes: Some(0),
407                    item_count: Some(0),
408                    index_arn: idx_arn,
409                }
410            })
411            .collect()
412    });
413
414    // A vector index allocating resources holds the base table at UPDATING;
415    // it returns to ACTIVE for the backfill, which is the far longer half and
416    // the state a readiness check should be written against (captured
417    // eu-west-2, 2026-08-23).
418    let table_status = if vector_phases.any_holding_table_updating() {
419        "UPDATING".to_string()
420    } else {
421        meta.table_status.clone()
422    };
423
424    let billing_mode = meta.billing_mode.clone();
425
426    let provisioned_throughput = if let Some(pt_json) = &meta.provisioned_throughput {
427        // Try parsing extended format (with timestamps) first, fall back to basic
428        serde_json::from_str::<serde_json::Value>(pt_json)
429            .ok()
430            .map(|v| {
431                let rcu = v
432                    .get("ReadCapacityUnits")
433                    .and_then(|v| v.as_i64())
434                    .unwrap_or(0) as u64;
435                let wcu = v
436                    .get("WriteCapacityUnits")
437                    .and_then(|v| v.as_i64())
438                    .unwrap_or(0) as u64;
439                let last_inc = v.get("LastIncreaseDateTime").and_then(|v| v.as_f64());
440                let last_dec = v.get("LastDecreaseDateTime").and_then(|v| v.as_f64());
441                let num_dec = v
442                    .get("NumberOfDecreasesToday")
443                    .and_then(|v| v.as_u64())
444                    .unwrap_or(0);
445                TableProvisionedThroughputDescription {
446                    read_capacity_units: rcu,
447                    write_capacity_units: wcu,
448                    number_of_decreases_today: num_dec,
449                    last_increase_date_time: last_inc,
450                    last_decrease_date_time: last_dec,
451                }
452            })
453    } else if billing_mode.as_deref() == Some("PAY_PER_REQUEST") {
454        // PAY_PER_REQUEST tables have zero provisioned throughput
455        Some(TableProvisionedThroughputDescription {
456            read_capacity_units: 0,
457            write_capacity_units: 0,
458            number_of_decreases_today: 0,
459            last_increase_date_time: None,
460            last_decrease_date_time: None,
461        })
462    } else {
463        None
464    };
465
466    let stream_specification = if meta.stream_enabled {
467        Some(StreamSpecificationDescription {
468            stream_enabled: true,
469            stream_view_type: meta.stream_view_type.clone(),
470        })
471    } else {
472        None
473    };
474
475    let latest_stream_arn = if meta.stream_enabled {
476        meta.stream_label
477            .as_ref()
478            .map(|label| streams::stream_arn(table_name, label))
479    } else {
480        None
481    };
482
483    let latest_stream_label = if meta.stream_enabled {
484        meta.stream_label.clone()
485    } else {
486        None
487    };
488
489    // Build SSE description from stored specification. When encryption is
490    // enabled, AWS reports SSEType=KMS and a KMS key ARN even if the request
491    // only set Enabled=true, so default those here rather than dropping them.
492    let sse_description = meta.sse_specification.as_ref().and_then(|json| {
493        serde_json::from_str::<crate::types::SseSpecification>(json)
494            .ok()
495            .map(|spec| {
496                let enabled = spec.enabled.unwrap_or(false);
497                SseDescription {
498                    status: if enabled { "ENABLED" } else { "DISABLED" }.to_string(),
499                    sse_type: if enabled {
500                        spec.sse_type.or_else(|| Some("KMS".to_string()))
501                    } else {
502                        spec.sse_type
503                    },
504                    kms_master_key_arn: if enabled {
505                        // New tables persist a synthesised key id at create time;
506                        // fall back to a table-derived key for any older row that
507                        // enabled SSE without one, so the ARN stays stable across
508                        // repeated DescribeTable calls rather than changing each read.
509                        Some(
510                            spec.kms_master_key_id
511                                .map(|id| streams::kms_key_arn(&id))
512                                .unwrap_or_else(|| streams::kms_key_arn(table_name)),
513                        )
514                    } else {
515                        None
516                    },
517                }
518            })
519    });
520
521    let table_class_summary = meta.table_class.as_ref().map(|tc| TableClassSummary {
522        table_class: tc.clone(),
523    });
524
525    let on_demand_throughput = meta
526        .on_demand_throughput
527        .as_ref()
528        .and_then(|json| serde_json::from_str::<crate::types::OnDemandThroughput>(json).ok());
529
530    let deletion_protection_enabled = Some(meta.deletion_protection_enabled);
531
532    TableDescription {
533        table_name: meta.table_name.clone(),
534        table_id: Some(
535            meta.table_id
536                .clone()
537                .unwrap_or_else(|| table_id_for(&meta.table_name, meta.created_at)),
538        ),
539        table_arn: streams::table_arn(table_name),
540        table_status,
541        key_schema,
542        attribute_definitions,
543        creation_date_time: Some(meta.created_at as f64),
544        item_count,
545        table_size_bytes,
546        provisioned_throughput,
547        billing_mode_summary: match billing_mode.as_deref() {
548            Some("PAY_PER_REQUEST") => Some(BillingModeSummary {
549                billing_mode: "PAY_PER_REQUEST".to_string(),
550                last_update_to_pay_per_request_date_time: None,
551            }),
552            _ => None,
553        },
554        table_throughput_mode_summary: match billing_mode.as_deref() {
555            Some("PAY_PER_REQUEST") => Some(TableThroughputModeSummary {
556                table_throughput_mode: "PAY_PER_REQUEST".to_string(),
557                last_update_to_pay_per_request_date_time: None,
558            }),
559            _ => None,
560        },
561        global_secondary_indexes,
562        local_secondary_indexes,
563        vector_indexes,
564        stream_specification,
565        latest_stream_arn,
566        latest_stream_label,
567        sse_description,
568        table_class_summary,
569        deletion_protection_enabled,
570        on_demand_throughput,
571    }
572}
dynoxide/actions/mod.rs

dynoxide/actions/
mod.rs
