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

dynoxide/actions/
mod.rs
