1use crate::actions::helpers;
2use crate::actions::index_capacity::{WriteCapacity, aggregate_by_table, per_table_capacity};
3use crate::errors::{CancellationReason, DynoxideError, Result};
4use crate::storage_backend::StorageBackend;
5use crate::types::{self, AttributeValue, Item};
6use serde::{Deserialize, Serialize};
7use std::collections::{HashMap, HashSet};
8
9#[derive(Debug, Clone, Default, Deserialize, Serialize)]
10pub struct TransactWriteItemsRequest {
11 #[serde(rename = "TransactItems")]
12 pub transact_items: Vec<TransactWriteItem>,
13 #[serde(rename = "ClientRequestToken", default)]
14 pub client_request_token: Option<String>,
15 #[serde(rename = "ReturnConsumedCapacity", default)]
16 pub return_consumed_capacity: Option<String>,
17 #[serde(rename = "ReturnItemCollectionMetrics", default)]
18 pub return_item_collection_metrics: Option<String>,
19}
20
21#[derive(Debug, Clone, Default, Deserialize, Serialize)]
22pub struct TransactWriteItem {
23 #[serde(rename = "Put", default)]
24 pub put: Option<TransactPut>,
25 #[serde(rename = "Update", default)]
26 pub update: Option<TransactUpdate>,
27 #[serde(rename = "Delete", default)]
28 pub delete: Option<TransactDelete>,
29 #[serde(rename = "ConditionCheck", default)]
30 pub condition_check: Option<TransactConditionCheck>,
31}
32
33#[derive(Debug, Clone, Default, Deserialize, Serialize)]
34pub struct TransactPut {
35 #[serde(rename = "TableName")]
36 pub table_name: String,
37 #[serde(rename = "Item")]
38 pub item: Item,
39 #[serde(rename = "ConditionExpression", default)]
40 pub condition_expression: Option<String>,
41 #[serde(rename = "ExpressionAttributeNames", default)]
42 pub expression_attribute_names: Option<HashMap<String, String>>,
43 #[serde(
44 rename = "ExpressionAttributeValues",
45 default,
46 deserialize_with = "crate::types::deserialize_older_expression_values"
47 )]
48 pub expression_attribute_values: Option<HashMap<String, AttributeValue>>,
49 #[serde(rename = "ReturnValuesOnConditionCheckFailure", default)]
50 pub return_values_on_condition_check_failure: Option<String>,
51}
52
53#[derive(Debug, Clone, Default, Deserialize, Serialize)]
54pub struct TransactUpdate {
55 #[serde(rename = "TableName")]
56 pub table_name: String,
57 #[serde(rename = "Key")]
58 pub key: HashMap<String, AttributeValue>,
59 #[serde(rename = "UpdateExpression")]
60 pub update_expression: String,
61 #[serde(rename = "ConditionExpression", default)]
62 pub condition_expression: Option<String>,
63 #[serde(rename = "ExpressionAttributeNames", default)]
64 pub expression_attribute_names: Option<HashMap<String, String>>,
65 #[serde(
66 rename = "ExpressionAttributeValues",
67 default,
68 deserialize_with = "crate::types::deserialize_older_expression_values"
69 )]
70 pub expression_attribute_values: Option<HashMap<String, AttributeValue>>,
71 #[serde(rename = "ReturnValuesOnConditionCheckFailure", default)]
72 pub return_values_on_condition_check_failure: Option<String>,
73}
74
75#[derive(Debug, Clone, Default, Deserialize, Serialize)]
76pub struct TransactDelete {
77 #[serde(rename = "TableName")]
78 pub table_name: String,
79 #[serde(rename = "Key")]
80 pub key: HashMap<String, AttributeValue>,
81 #[serde(rename = "ConditionExpression", default)]
82 pub condition_expression: Option<String>,
83 #[serde(rename = "ExpressionAttributeNames", default)]
84 pub expression_attribute_names: Option<HashMap<String, String>>,
85 #[serde(
86 rename = "ExpressionAttributeValues",
87 default,
88 deserialize_with = "crate::types::deserialize_older_expression_values"
89 )]
90 pub expression_attribute_values: Option<HashMap<String, AttributeValue>>,
91 #[serde(rename = "ReturnValuesOnConditionCheckFailure", default)]
92 pub return_values_on_condition_check_failure: Option<String>,
93}
94
95#[derive(Debug, Clone, Default, Deserialize, Serialize)]
96pub struct TransactConditionCheck {
97 #[serde(rename = "TableName")]
98 pub table_name: String,
99 #[serde(rename = "Key")]
100 pub key: HashMap<String, AttributeValue>,
101 #[serde(rename = "ConditionExpression")]
102 pub condition_expression: String,
103 #[serde(rename = "ExpressionAttributeNames", default)]
104 pub expression_attribute_names: Option<HashMap<String, String>>,
105 #[serde(
106 rename = "ExpressionAttributeValues",
107 default,
108 deserialize_with = "crate::types::deserialize_older_expression_values"
109 )]
110 pub expression_attribute_values: Option<HashMap<String, AttributeValue>>,
111 #[serde(rename = "ReturnValuesOnConditionCheckFailure", default)]
112 pub return_values_on_condition_check_failure: Option<String>,
113}
114
115#[derive(Debug, Clone, Default, Serialize)]
116pub struct TransactWriteItemsResponse {
117 #[serde(rename = "ConsumedCapacity", skip_serializing_if = "Option::is_none")]
118 pub consumed_capacity: Option<Vec<crate::types::ConsumedCapacity>>,
119 #[serde(
122 rename = "ItemCollectionMetrics",
123 skip_serializing_if = "Option::is_none"
124 )]
125 pub item_collection_metrics: Option<HashMap<String, Vec<crate::types::ItemCollectionMetrics>>>,
126}
127
128#[derive(Debug, Clone, Default)]
135pub(crate) struct CachedWrite {
136 pub(crate) response: TransactWriteItemsResponse,
137 #[cfg(any(feature = "native-sqlite", feature = "_has-encryption"))]
140 pub(crate) replay_sizes: Vec<(String, usize)>,
141}
142
143pub async fn execute<S: StorageBackend>(
148 storage: &S,
149 request: TransactWriteItemsRequest,
150) -> Result<TransactWriteItemsResponse> {
151 Ok(execute_cached(storage, request).await?.response)
152}
153
154fn empty_name_in_write_action(item: &TransactWriteItem) -> Result<()> {
159 let (attributes, names, values) = if let Some(ref put) = item.put {
160 (
161 &put.item,
162 put.expression_attribute_names.as_ref(),
163 put.expression_attribute_values.as_ref(),
164 )
165 } else if let Some(ref update) = item.update {
166 (
167 &update.key,
168 update.expression_attribute_names.as_ref(),
169 update.expression_attribute_values.as_ref(),
170 )
171 } else if let Some(ref delete) = item.delete {
172 (
173 &delete.key,
174 delete.expression_attribute_names.as_ref(),
175 delete.expression_attribute_values.as_ref(),
176 )
177 } else if let Some(ref check) = item.condition_check {
178 (
179 &check.key,
180 check.expression_attribute_names.as_ref(),
181 check.expression_attribute_values.as_ref(),
182 )
183 } else {
184 return Ok(());
185 };
186 if crate::empty_names::map_has_empty_name(attributes) {
187 return Err(crate::empty_names::bare());
188 }
189 crate::empty_names::check_names(names)?;
190 if let Some(key) = crate::empty_names::value_placeholder(values, None) {
191 return Err(DynoxideError::ValidationException(
192 crate::empty_names::value_placeholder_message(key),
193 ));
194 }
195 Ok(())
196}
197
198fn items_echo(items: &[TransactWriteItem]) -> String {
205 let mut out = crate::actions::batch_write_item::BoundedEcho::new();
206 let _ = write_items(&mut out, items);
207 out.finish()
208}
209
210fn write_items(out: &mut dyn std::fmt::Write, items: &[TransactWriteItem]) -> std::fmt::Result {
211 use crate::actions::batch_write_item::{
212 write_java_item, write_java_names, write_java_text, write_java_values,
213 };
214 let conditions = |out: &mut dyn std::fmt::Write,
216 condition: Option<&str>,
217 names: Option<&HashMap<String, String>>,
218 values: Option<&HashMap<String, AttributeValue>>,
219 return_values: Option<&str>|
220 -> std::fmt::Result {
221 out.write_str(", conditionExpression=")?;
222 write_java_text(out, condition)?;
223 out.write_str(", expressionAttributeNames=")?;
224 write_java_names(out, names)?;
225 out.write_str(", expressionAttributeValues=")?;
226 write_java_values(out, values)?;
227 out.write_str(", returnValuesOnConditionCheckFailure=")?;
228 write_java_text(out, return_values)?;
229 out.write_str(", workloadProfileName=null)")
230 };
231 out.write_str("[")?;
232 for (i, item) in items.iter().enumerate() {
233 if i > 0 {
234 out.write_str(", ")?;
235 }
236 out.write_str("TransactWriteItem(conditionCheck=")?;
237 match item.condition_check {
238 Some(ref check) => {
239 out.write_str("ConditionCheck(key=")?;
240 write_java_item(out, &check.key)?;
241 write!(out, ", tableName={}", check.table_name)?;
242 conditions(
243 out,
244 Some(&check.condition_expression),
245 check.expression_attribute_names.as_ref(),
246 check.expression_attribute_values.as_ref(),
247 check.return_values_on_condition_check_failure.as_deref(),
248 )?;
249 }
250 None => out.write_str("null")?,
251 }
252 out.write_str(", put=")?;
253 match item.put {
254 Some(ref put) => {
255 out.write_str("Put(item=")?;
256 write_java_item(out, &put.item)?;
257 write!(out, ", tableName={}", put.table_name)?;
258 conditions(
259 out,
260 put.condition_expression.as_deref(),
261 put.expression_attribute_names.as_ref(),
262 put.expression_attribute_values.as_ref(),
263 put.return_values_on_condition_check_failure.as_deref(),
264 )?;
265 }
266 None => out.write_str("null")?,
267 }
268 out.write_str(", delete=")?;
269 match item.delete {
270 Some(ref delete) => {
271 out.write_str("Delete(key=")?;
272 write_java_item(out, &delete.key)?;
273 write!(out, ", tableName={}", delete.table_name)?;
274 conditions(
275 out,
276 delete.condition_expression.as_deref(),
277 delete.expression_attribute_names.as_ref(),
278 delete.expression_attribute_values.as_ref(),
279 delete.return_values_on_condition_check_failure.as_deref(),
280 )?;
281 }
282 None => out.write_str("null")?,
283 }
284 out.write_str(", update=")?;
285 match item.update {
286 Some(ref update) => {
287 out.write_str("Update(key=")?;
288 write_java_item(out, &update.key)?;
289 write!(
290 out,
291 ", updateExpression={}, tableName={}",
292 update.update_expression, update.table_name
293 )?;
294 conditions(
295 out,
296 update.condition_expression.as_deref(),
297 update.expression_attribute_names.as_ref(),
298 update.expression_attribute_values.as_ref(),
299 update.return_values_on_condition_check_failure.as_deref(),
300 )?;
301 }
302 None => out.write_str("null")?,
303 }
304 out.write_str(")")?;
305 }
306 out.write_str("]")
307}
308
309pub(crate) fn check_request(request: &TransactWriteItemsRequest) -> Result<()> {
327 let items = &request.transact_items;
328
329 if items.is_empty() {
331 return Err(DynoxideError::ValidationException(
332 "1 validation error detected: Value '[]' at 'transactItems' failed to satisfy constraint: Member must have length greater than or equal to 1".to_string(),
333 ));
334 }
335
336 if items.len() > 100 {
338 return Err(DynoxideError::ValidationException(format!(
339 "1 validation error detected: Value '{}' at 'transactItems' failed to satisfy constraint: Member must have length less than or equal to 100",
340 items_echo(items)
341 )));
342 }
343
344 let mut member_errors: Vec<String> = Vec::new();
350 for (position, item) in items.iter().enumerate() {
351 let actions = [
352 (
353 "put",
354 item.put
355 .as_ref()
356 .map(|a| &a.return_values_on_condition_check_failure),
357 ),
358 (
359 "update",
360 item.update
361 .as_ref()
362 .map(|a| &a.return_values_on_condition_check_failure),
363 ),
364 (
365 "delete",
366 item.delete
367 .as_ref()
368 .map(|a| &a.return_values_on_condition_check_failure),
369 ),
370 (
371 "conditionCheck",
372 item.condition_check
373 .as_ref()
374 .map(|a| &a.return_values_on_condition_check_failure),
375 ),
376 ];
377 for (action, value) in actions {
378 if let Some(msg) = value.and_then(|value| {
379 crate::validation::return_values_on_condition_check_failure_rejection(
380 value.as_deref(),
381 &format!(
382 "transactItems.{}.member.{action}.returnValuesOnConditionCheckFailure",
383 position + 1
384 ),
385 )
386 }) {
387 member_errors.push(msg);
388 }
389 }
390 }
391 member_errors.extend(
392 [
393 crate::token_constraint_error(request.client_request_token.as_deref()),
394 crate::validation::return_consumed_capacity_rejection(
395 request.return_consumed_capacity.as_deref(),
396 ),
397 crate::validation::return_item_collection_metrics_rejection(
398 request.return_item_collection_metrics.as_deref(),
399 ),
400 ]
401 .into_iter()
402 .flatten(),
403 );
404 if let Some(msg) = crate::validation::format_validation_errors(&member_errors) {
405 return Err(DynoxideError::ValidationException(msg));
406 }
407
408 for item in items {
411 let actions = [
412 item.put.is_some(),
413 item.update.is_some(),
414 item.delete.is_some(),
415 item.condition_check.is_some(),
416 ]
417 .into_iter()
418 .filter(|&present| present)
419 .count();
420 match actions {
421 0 => {
422 return Err(DynoxideError::ValidationException(
423 "Invalid Request: TransactWriteRequest should contain Delete or Put or Update request"
424 .to_string(),
425 ));
426 }
427 1 => {}
428 _ => {
429 return Err(DynoxideError::ValidationException(
430 "TransactItems can only contain one of Check, Put, Update or Delete"
431 .to_string(),
432 ));
433 }
434 }
435 }
436
437 let mut name_errors = Vec::new();
440 for (position, item) in items.iter().enumerate() {
441 let (action, table_name) = if let Some(ref put) = item.put {
442 ("put", &put.table_name)
443 } else if let Some(ref update) = item.update {
444 ("update", &update.table_name)
445 } else if let Some(ref delete) = item.delete {
446 ("delete", &delete.table_name)
447 } else if let Some(ref check) = item.condition_check {
448 ("conditionCheck", &check.table_name)
449 } else {
450 continue;
451 };
452 name_errors.extend(crate::validation::member_table_name_errors(
453 table_name,
454 &format!("transactItems.{}.member.{action}.tableName", position + 1),
455 ));
456 }
457 if let Some(msg) = crate::validation::format_validation_errors(&name_errors) {
458 return Err(DynoxideError::ValidationException(msg));
459 }
460
461 Ok(())
462}
463
464pub(crate) async fn execute_cached<S: StorageBackend>(
465 storage: &S,
466 request: TransactWriteItemsRequest,
467) -> Result<CachedWrite> {
468 check_request(&request)?;
469 let items = &request.transact_items;
470
471 for item in items {
475 empty_name_in_write_action(item)?;
476 }
477
478 for item in items {
483 validate_action_request(item)?;
484 }
485
486 for item in items {
489 let key = item
490 .delete
491 .as_ref()
492 .map(|d| &d.key)
493 .or(item.condition_check.as_ref().map(|c| &c.key))
494 .or(item.update.as_ref().map(|u| &u.key));
495 if let Some(key) = key {
496 crate::value_nesting::check_key(key)?;
497 }
498 }
499
500 let mut seen_targets = HashSet::new();
503 for item in items {
504 if let Some(target) = get_item_target(storage, item).await?
505 && !seen_targets.insert(target)
506 {
507 return Err(DynoxideError::ValidationException(
508 "Transaction request cannot include multiple operations on one item".to_string(),
509 ));
510 }
511 }
512
513 let total_size: usize = items.iter().map(|i| get_action_table_and_size(i).1).sum();
515 if total_size > 4 * 1024 * 1024 {
516 return Err(DynoxideError::ValidationException(
517 "Collection size of items exceeded, which can also be caused by the aggregate size of the items in the transaction exceeding the 4MB limit".to_string(),
518 ));
519 }
520
521 let capacity_mode = request.return_consumed_capacity.as_deref();
523 let capacity = helpers::with_write_transaction(
524 storage,
525 execute_within_transaction(storage, items, capacity_mode),
526 )
527 .await?;
528
529 Ok(CachedWrite {
530 response: TransactWriteItemsResponse {
531 consumed_capacity: build_write_capacity(&capacity, &request.return_consumed_capacity),
532 item_collection_metrics: None,
533 },
534 #[cfg(any(feature = "native-sqlite", feature = "_has-encryption"))]
535 replay_sizes: replay_sizes(&capacity),
536 })
537}
538
539#[cfg(any(feature = "native-sqlite", feature = "_has-encryption"))]
545fn replay_sizes(capacity: &[WriteCapacity]) -> Vec<(String, usize)> {
546 capacity
547 .iter()
548 .map(|record| {
549 let larger = record
550 .old_size
551 .unwrap_or(0)
552 .max(record.new_size.unwrap_or(0));
553 (record.table_name.clone(), larger)
554 })
555 .collect()
556}
557
558fn build_write_capacity(
565 capacity: &[WriteCapacity],
566 mode: &Option<String>,
567) -> Option<Vec<crate::types::ConsumedCapacity>> {
568 if !matches!(mode.as_deref(), Some("TOTAL") | Some("INDEXES")) {
572 return None;
573 }
574
575 let by_table = aggregate_by_table(capacity, crate::types::TRANSACTIONAL_CAPACITY_FACTOR);
576 per_table_capacity(
577 &by_table,
578 mode,
579 crate::types::transactional_write_capacity_with_indexes,
580 )
581}
582
583#[cfg(any(feature = "native-sqlite", feature = "_has-encryption"))]
592pub(crate) fn replay_response(cached: &CachedWrite, mode: &Option<String>) -> CachedWrite {
593 CachedWrite {
594 response: TransactWriteItemsResponse {
595 consumed_capacity: crate::types::build_transactional_capacity(
596 &crate::actions::index_capacity::transactional_read_units(&cached.replay_sizes),
597 mode,
598 crate::types::transactional_read_capacity,
599 ),
600 item_collection_metrics: cached.response.item_collection_metrics.clone(),
601 },
602 replay_sizes: cached.replay_sizes.clone(),
603 }
604}
605
606async fn execute_within_transaction<S: StorageBackend>(
618 storage: &S,
619 items: &[TransactWriteItem],
620 capacity_mode: Option<&str>,
621) -> Result<Vec<WriteCapacity>> {
622 let mut cancellation_reasons: Vec<CancellationReason> = Vec::with_capacity(items.len());
623 let mut capacity: Vec<WriteCapacity> = Vec::with_capacity(items.len());
624 let mut has_failure = false;
625
626 for item in items {
627 let reason = execute_single_action(storage, item, capacity_mode).await;
628 match reason {
629 Ok(action_capacity) => {
630 capacity.push(action_capacity);
631 cancellation_reasons.push(CancellationReason {
632 code: "None".to_string(),
633 message: None,
634 item: None,
635 });
636 }
637 Err(e) => {
638 if matches!(e, DynoxideError::KeyEmptyValueValidation(_)) {
642 return Err(e);
643 }
644 let message = Some(e.to_string());
645 let (code, item) = match e {
646 DynoxideError::ConditionalCheckFailedException(_, item) => {
647 ("ConditionalCheckFailed".to_string(), item)
648 }
649 DynoxideError::ValidationException(_) => ("ValidationError".to_string(), None),
650 storage_failure => return Err(storage_failure),
651 };
652 has_failure = true;
653 cancellation_reasons.push(CancellationReason {
654 code,
655 message,
656 item,
657 });
658 }
659 }
660 }
661
662 if has_failure {
663 let codes: Vec<&str> = cancellation_reasons
664 .iter()
665 .map(|r| r.code.as_str())
666 .collect();
667 let message = format!(
668 "Transaction cancelled, please refer cancellation reasons for specific reasons [{}]",
669 codes.join(", ")
670 );
671 return Err(DynoxideError::TransactionCanceledException(
672 message,
673 cancellation_reasons,
674 ));
675 }
676
677 Ok(capacity)
678}
679
680async fn execute_single_action<S: StorageBackend>(
681 storage: &S,
682 item: &TransactWriteItem,
683 capacity_mode: Option<&str>,
684) -> Result<WriteCapacity> {
685 if let Some(ref put) = item.put {
686 execute_put(storage, put, capacity_mode).await
687 } else if let Some(ref update) = item.update {
688 execute_update(storage, update, capacity_mode).await
689 } else if let Some(ref delete) = item.delete {
690 execute_delete(storage, delete, capacity_mode).await
691 } else if let Some(ref check) = item.condition_check {
692 execute_condition_check(storage, check).await
693 } else {
694 Err(DynoxideError::ValidationException(
695 "TransactItem must contain exactly one of Put, Update, Delete, or ConditionCheck"
696 .to_string(),
697 ))
698 }
699}
700
701fn validate_action_request(item: &TransactWriteItem) -> Result<()> {
725 validate_action_sets(item)?;
731 if let Some(ref put) = item.put {
732 validate_expression_value_depth(&put.expression_attribute_values)?;
733 for value in put.item.values() {
734 crate::validation::validate_nesting_depth(value)?;
735 }
736 validate_condition_size(put.condition_expression.as_deref())?;
737 if types::item_size(&put.item) > types::MAX_ITEM_SIZE {
740 return Err(DynoxideError::ValidationException(
741 "Item size has exceeded the maximum allowed size".to_string(),
742 ));
743 }
744 validate_condition_syntax(put.condition_expression.as_deref())?;
745 validate_expression_references(
746 &put.expression_attribute_names,
747 &put.expression_attribute_values,
748 put.condition_expression.as_deref(),
749 None,
750 )?;
751 } else if let Some(ref update) = item.update {
752 validate_expression_value_depth(&update.expression_attribute_values)?;
753 if update.update_expression.len() > crate::expressions::MAX_EXPRESSION_BYTES {
756 return Err(DynoxideError::ValidationException(
757 "Invalid UpdateExpression: Expression size has exceeded the maximum allowed size;"
758 .to_string(),
759 ));
760 }
761 if update.update_expression.is_empty() {
765 return Err(DynoxideError::ValidationException(
766 "Invalid UpdateExpression: The expression can not be empty;".to_string(),
767 ));
768 }
769 let update_expression = crate::expressions::update::parse(&update.update_expression)
770 .map_err(DynoxideError::ValidationException)?;
771 validate_condition_size(update.condition_expression.as_deref())?;
772 validate_condition_syntax(update.condition_expression.as_deref())?;
773 validate_expression_references(
774 &update.expression_attribute_names,
775 &update.expression_attribute_values,
776 update.condition_expression.as_deref(),
777 Some(&update_expression),
778 )?;
779 } else if let Some(ref delete) = item.delete {
780 validate_expression_value_depth(&delete.expression_attribute_values)?;
781 validate_condition_size(delete.condition_expression.as_deref())?;
782 validate_condition_syntax(delete.condition_expression.as_deref())?;
783 validate_expression_references(
784 &delete.expression_attribute_names,
785 &delete.expression_attribute_values,
786 delete.condition_expression.as_deref(),
787 None,
788 )?;
789 } else if let Some(ref check) = item.condition_check {
790 validate_expression_value_depth(&check.expression_attribute_values)?;
791 validate_condition_size(Some(&check.condition_expression))?;
792 validate_condition_syntax(Some(&check.condition_expression))?;
793 validate_expression_references(
794 &check.expression_attribute_names,
795 &check.expression_attribute_values,
796 Some(&check.condition_expression),
797 None,
798 )?;
799 }
800 Ok(())
801}
802
803fn validate_action_sets(item: &TransactWriteItem) -> Result<()> {
806 use crate::validation::{ValueLayer, find_set_error};
807 let (values, values_map) = if let Some(ref put) = item.put {
808 (&put.item, &put.expression_attribute_values)
809 } else if let Some(ref update) = item.update {
810 (&update.key, &update.expression_attribute_values)
811 } else if let Some(ref delete) = item.delete {
812 (&delete.key, &delete.expression_attribute_values)
813 } else if let Some(ref check) = item.condition_check {
814 (&check.key, &check.expression_attribute_values)
815 } else {
816 return Ok(());
817 };
818 let mut names: Vec<&String> = values.keys().collect();
819 names.sort();
820 for name in names {
821 if let Some(message) = find_set_error(&values[name], ValueLayer::Older) {
822 return Err(DynoxideError::ValidationException(message));
823 }
824 }
825 if let Some(eav) = values_map {
826 let mut keys: Vec<&String> = eav.keys().collect();
827 keys.sort();
828 for key in keys {
829 if let Some(message) = find_set_error(&eav[key], ValueLayer::Older) {
830 return Err(DynoxideError::ValidationException(format!(
831 "ExpressionAttributeValues contains invalid value: {message} for key {key}"
832 )));
833 }
834 }
835 }
836 Ok(())
837}
838
839fn validate_expression_references(
854 names: &Option<HashMap<String, String>>,
855 values: &Option<HashMap<String, AttributeValue>>,
856 condition: Option<&str>,
857 update: Option<&crate::expressions::update::UpdateExpr>,
858) -> Result<()> {
859 if condition.is_none() && update.is_none() {
860 return Ok(());
861 }
862 let tracker = crate::expressions::TrackedExpressionAttributes::new(names, values);
863 if let Some(update) = update {
864 tracker.track_update_expr(update);
865 crate::expressions::update::check_target_overlaps(update, &tracker)
868 .map_err(DynoxideError::ValidationException)?;
869 }
870 if let Some(condition) = condition {
871 let parsed = crate::expressions::condition::parse(condition).map_err(|e| {
873 DynoxideError::ValidationException(format!("Invalid ConditionExpression: {e}"))
874 })?;
875 crate::expressions::condition::track_references(&parsed, &tracker).map_err(|e| {
876 DynoxideError::ValidationException(format!("Invalid ConditionExpression: {e}"))
877 })?;
878 }
879 tracker.check_unused()
880}
881
882const MAX_EXPRESSION_VALUE_DEPTH: usize = 32;
887
888fn validate_expression_value_depth(values: &Option<HashMap<String, AttributeValue>>) -> Result<()> {
889 let Some(values) = values else {
890 return Ok(());
891 };
892 let mut keys: Vec<&String> = values.keys().collect();
893 keys.sort();
894 for key in keys {
895 if crate::validation::nesting_depth(&values[key]) > MAX_EXPRESSION_VALUE_DEPTH {
896 return Err(DynoxideError::ValidationException(format!(
897 "ExpressionAttributeValues contains invalid value: Nesting Levels have exceeded \
898 supported limits: Attributes in the item have nested levels beyond supported \
899 limit for key {key}"
900 )));
901 }
902 }
903 Ok(())
904}
905
906fn validate_condition_syntax(expression: Option<&str>) -> Result<()> {
911 let Some(expr) = expression else {
912 return Ok(());
913 };
914 if expr.is_empty() {
915 return Err(DynoxideError::ValidationException(
916 "Invalid ConditionExpression: The expression can not be empty;".to_string(),
917 ));
918 }
919 crate::expressions::condition::parse(expr).map_err(|e| {
920 DynoxideError::ValidationException(format!("Invalid ConditionExpression: {e}"))
921 })?;
922 Ok(())
923}
924
925fn validate_condition_size(expression: Option<&str>) -> Result<()> {
926 if let Some(expr) = expression {
927 crate::expressions::check_expression_size_with_figure(expr).map_err(|e| {
928 DynoxideError::ValidationException(format!("Invalid ConditionExpression: {e}"))
929 })?;
930 }
931 Ok(())
932}
933
934async fn execute_put<S: StorageBackend>(
935 storage: &S,
936 put: &TransactPut,
937 capacity_mode: Option<&str>,
938) -> Result<WriteCapacity> {
939 crate::validation::validate_table_name(&put.table_name)?;
940 let meta = helpers::require_table_for_item_op(storage, &put.table_name).await?;
941 let key_schema = helpers::parse_key_schema(&meta)?;
942
943 helpers::validate_item_keys(&put.item, &key_schema, &meta)?;
944 crate::validation::validate_item_attribute_values(&put.item)?;
945
946 let mut item = put.item.clone();
948 crate::validation::normalize_item_sets(&mut item);
949
950 let vector_defs = super::vector_index::parse_vector_defs(&meta)?;
955 let size = types::billed_item_size(&item, &vector_defs);
956
957 let (pk, sk) = helpers::extract_key_strings(&item, &key_schema)?;
959
960 let tracker = crate::expressions::TrackedExpressionAttributes::new(
961 &put.expression_attribute_names,
962 &put.expression_attribute_values,
963 );
964
965 if let Some(ref cond_expr) = put.condition_expression
967 && let Ok(parsed) = crate::expressions::condition::parse(cond_expr)
968 {
969 tracker.track_condition_expr(&parsed);
970 }
971
972 if let Some(ref cond_expr) = put.condition_expression {
974 let existing_json = storage.get_item(&put.table_name, &pk, &sk).await?;
975 let existing_item: Item = existing_json
976 .as_ref()
977 .and_then(|j| serde_json::from_str(j).ok())
978 .unwrap_or_default();
979
980 let return_item = if put.return_values_on_condition_check_failure.as_deref()
981 == Some("ALL_OLD")
982 && !existing_item.is_empty()
983 {
984 Some(existing_item.clone())
985 } else {
986 None
987 };
988 check_condition_tracked(cond_expr, &existing_item, &tracker, return_item)?;
989 }
990
991 if put.condition_expression.is_some() {
994 tracker.check_unused()?;
995 }
996
997 let item_json = serde_json::to_string(&item)
998 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
999 let hash_prefix = item
1000 .get(&key_schema.partition_key)
1001 .map(crate::storage::compute_hash_prefix)
1002 .unwrap_or_default();
1003 let old_json = storage
1004 .put_item_with_hash(&put.table_name, &pk, &sk, &item_json, size, &hash_prefix)
1005 .await?;
1006
1007 let old_item: Option<Item> = old_json.and_then(|j| serde_json::from_str(&j).ok());
1008
1009 let target = super::gsi::IndexWrite {
1010 table_name: &put.table_name,
1011 pk: &pk,
1012 sk: &sk,
1013 pk_attr: &key_schema.partition_key,
1014 sk_attr: key_schema.sort_key.as_deref(),
1015 old_item: old_item.as_ref(),
1016 capacity_mode,
1017 };
1018
1019 let gsi_units = super::gsi::maintain_gsis_after_write(storage, &meta, &target, &item).await?;
1020
1021 let lsi_units = super::lsi::maintain_lsis_after_write(storage, &meta, &target, &item).await?;
1022
1023 let vector_bytes =
1024 super::vector_index::maintain_vector_indexes_after_write(storage, &meta, &target, &item)
1025 .await?;
1026
1027 crate::streams::record_stream_event(storage, &meta, old_item.as_ref(), Some(&item)).await?;
1029
1030 Ok(WriteCapacity::new(
1031 &put.table_name,
1032 old_item
1033 .as_ref()
1034 .map(|old| types::billed_item_size(old, &vector_defs)),
1035 Some(size),
1036 gsi_units,
1037 lsi_units,
1038 )
1039 .with_vector_bytes(vector_bytes))
1040}
1041
1042async fn execute_update<S: StorageBackend>(
1043 storage: &S,
1044 update: &TransactUpdate,
1045 capacity_mode: Option<&str>,
1046) -> Result<WriteCapacity> {
1047 crate::validation::validate_table_name(&update.table_name)?;
1048 let meta = helpers::require_table_for_item_op(storage, &update.table_name).await?;
1049 let key_schema = helpers::parse_key_schema(&meta)?;
1050
1051 helpers::validate_key_only(&update.key, &key_schema)?;
1052 let (pk, sk) = helpers::extract_key_strings(&update.key, &key_schema)?;
1054
1055 let existing_json = storage.get_item(&update.table_name, &pk, &sk).await?;
1056 let existing_item: Item = existing_json
1057 .as_ref()
1058 .and_then(|j| serde_json::from_str(j).ok())
1059 .unwrap_or_default();
1060
1061 let tracker = crate::expressions::TrackedExpressionAttributes::new(
1062 &update.expression_attribute_names,
1063 &update.expression_attribute_values,
1064 );
1065
1066 if let Some(ref cond_expr) = update.condition_expression
1068 && let Ok(parsed) = crate::expressions::condition::parse(cond_expr)
1069 {
1070 tracker.track_condition_expr(&parsed);
1071 }
1072 if let Ok(parsed) = crate::expressions::update::parse(&update.update_expression) {
1073 tracker.track_update_expr(&parsed);
1074 }
1075
1076 if let Some(ref cond_expr) = update.condition_expression {
1080 let return_item = if update.return_values_on_condition_check_failure.as_deref()
1081 == Some("ALL_OLD")
1082 && existing_json.is_some()
1083 {
1084 Some(existing_item.clone())
1085 } else {
1086 None
1087 };
1088 check_condition_tracked(cond_expr, &existing_item, &tracker, return_item)?;
1089 }
1090
1091 let mut item = existing_item;
1094 if existing_json.is_none() {
1095 for (k, v) in &update.key {
1096 item.insert(k.clone(), v.clone());
1097 }
1098 }
1099 let before_item = item.clone();
1100
1101 let parsed = crate::expressions::update::parse(&update.update_expression)
1103 .map_err(DynoxideError::ValidationException)?;
1104 crate::expressions::update::apply(&mut item, &parsed, &tracker)
1105 .map_err(DynoxideError::ValidationException)?;
1106
1107 tracker.check_unused()?;
1108
1109 crate::validation::validate_updated_item_nesting(&item)?;
1113
1114 crate::validation::validate_item_attribute_values(&item)?;
1116 crate::validation::normalize_item_sets(&mut item);
1117
1118 helpers::validate_updated_index_keys(&before_item, &item, &meta)?;
1120
1121 if types::item_size(&item) > types::MAX_ITEM_SIZE {
1127 return Err(DynoxideError::ValidationException(
1128 "Item size to update has exceeded the maximum allowed size".to_string(),
1129 ));
1130 }
1131 let vector_defs = super::vector_index::parse_vector_defs(&meta)?;
1132 let size = types::billed_item_size(&item, &vector_defs);
1133
1134 let old_for_stream = existing_json.clone();
1136
1137 let item_json = serde_json::to_string(&item)
1138 .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;
1139 let hash_prefix = update
1140 .key
1141 .get(&key_schema.partition_key)
1142 .map(crate::storage::compute_hash_prefix)
1143 .unwrap_or_default();
1144 storage
1145 .put_item_with_hash(&update.table_name, &pk, &sk, &item_json, size, &hash_prefix)
1146 .await?;
1147
1148 let old_item: Option<Item> = old_for_stream.and_then(|j| serde_json::from_str(&j).ok());
1149
1150 let target = super::gsi::IndexWrite {
1151 table_name: &update.table_name,
1152 pk: &pk,
1153 sk: &sk,
1154 pk_attr: &key_schema.partition_key,
1155 sk_attr: key_schema.sort_key.as_deref(),
1156 old_item: old_item.as_ref(),
1157 capacity_mode,
1158 };
1159
1160 let gsi_units = super::gsi::maintain_gsis_after_write(storage, &meta, &target, &item).await?;
1161
1162 let lsi_units = super::lsi::maintain_lsis_after_write(storage, &meta, &target, &item).await?;
1163
1164 let vector_bytes =
1165 super::vector_index::maintain_vector_indexes_after_write(storage, &meta, &target, &item)
1166 .await?;
1167
1168 crate::streams::record_stream_event(storage, &meta, old_item.as_ref(), Some(&item)).await?;
1170
1171 Ok(WriteCapacity::new(
1176 &update.table_name,
1177 old_item
1178 .as_ref()
1179 .map(|old| types::billed_item_size(old, &vector_defs)),
1180 Some(size),
1181 gsi_units,
1182 lsi_units,
1183 )
1184 .with_vector_bytes(vector_bytes))
1185}
1186
1187async fn execute_delete<S: StorageBackend>(
1188 storage: &S,
1189 delete: &TransactDelete,
1190 capacity_mode: Option<&str>,
1191) -> Result<WriteCapacity> {
1192 crate::validation::validate_table_name(&delete.table_name)?;
1193 let meta = helpers::require_table_for_item_op(storage, &delete.table_name).await?;
1194 let key_schema = helpers::parse_key_schema(&meta)?;
1195
1196 helpers::validate_key_only(&delete.key, &key_schema)?;
1197 let (pk, sk) = helpers::extract_key_strings(&delete.key, &key_schema)?;
1199
1200 let tracker = crate::expressions::TrackedExpressionAttributes::new(
1201 &delete.expression_attribute_names,
1202 &delete.expression_attribute_values,
1203 );
1204
1205 if let Some(ref cond_expr) = delete.condition_expression
1207 && let Ok(parsed) = crate::expressions::condition::parse(cond_expr)
1208 {
1209 tracker.track_condition_expr(&parsed);
1210 }
1211
1212 if let Some(ref cond_expr) = delete.condition_expression {
1214 let existing_json = storage.get_item(&delete.table_name, &pk, &sk).await?;
1215 let existing_item: Item = existing_json
1216 .as_ref()
1217 .and_then(|j| serde_json::from_str(j).ok())
1218 .unwrap_or_default();
1219
1220 let return_item = if delete.return_values_on_condition_check_failure.as_deref()
1221 == Some("ALL_OLD")
1222 && !existing_item.is_empty()
1223 {
1224 Some(existing_item.clone())
1225 } else {
1226 None
1227 };
1228 check_condition_tracked(cond_expr, &existing_item, &tracker, return_item)?;
1229 }
1230
1231 if delete.condition_expression.is_some() {
1234 tracker.check_unused()?;
1235 }
1236
1237 let old_json = storage.delete_item(&delete.table_name, &pk, &sk).await?;
1238 let old_item: Option<Item> = old_json.and_then(|j| serde_json::from_str(&j).ok());
1239
1240 let target = super::gsi::IndexWrite {
1241 table_name: &delete.table_name,
1242 pk: &pk,
1243 sk: &sk,
1244 pk_attr: &key_schema.partition_key,
1245 sk_attr: key_schema.sort_key.as_deref(),
1246 old_item: old_item.as_ref(),
1247 capacity_mode,
1248 };
1249
1250 let gsi_units = super::gsi::maintain_gsis_after_delete(storage, &meta, &target).await?;
1251 let lsi_units = super::lsi::maintain_lsis_after_delete(storage, &meta, &target).await?;
1252 let vector_bytes =
1253 super::vector_index::maintain_vector_indexes_after_delete(storage, &meta, &target).await?;
1254
1255 if old_item.is_some() {
1257 crate::streams::record_stream_event(storage, &meta, old_item.as_ref(), None).await?;
1258 }
1259
1260 let vector_defs = super::vector_index::parse_vector_defs(&meta)?;
1261 Ok(WriteCapacity::new(
1262 &delete.table_name,
1263 old_item
1264 .as_ref()
1265 .map(|old| types::billed_item_size(old, &vector_defs)),
1266 None,
1267 gsi_units,
1268 lsi_units,
1269 )
1270 .with_vector_bytes(vector_bytes))
1271}
1272
1273async fn execute_condition_check<S: StorageBackend>(
1274 storage: &S,
1275 check: &TransactConditionCheck,
1276) -> Result<WriteCapacity> {
1277 crate::validation::validate_table_name(&check.table_name)?;
1278 let meta = helpers::require_table_for_item_op(storage, &check.table_name).await?;
1279 let key_schema = helpers::parse_key_schema(&meta)?;
1280
1281 helpers::validate_key_only(&check.key, &key_schema)?;
1282 let (pk, sk) = helpers::extract_key_strings(&check.key, &key_schema)?;
1284
1285 let existing_json = storage.get_item(&check.table_name, &pk, &sk).await?;
1286 let existing_item: Item = existing_json
1287 .as_ref()
1288 .and_then(|j| serde_json::from_str(j).ok())
1289 .unwrap_or_default();
1290
1291 let tracker = crate::expressions::TrackedExpressionAttributes::new(
1292 &check.expression_attribute_names,
1293 &check.expression_attribute_values,
1294 );
1295
1296 if let Ok(parsed) = crate::expressions::condition::parse(&check.condition_expression) {
1298 tracker.track_condition_expr(&parsed);
1299 }
1300
1301 let return_item = if check.return_values_on_condition_check_failure.as_deref()
1302 == Some("ALL_OLD")
1303 && !existing_item.is_empty()
1304 {
1305 Some(existing_item.clone())
1306 } else {
1307 None
1308 };
1309 check_condition_tracked(
1310 &check.condition_expression,
1311 &existing_item,
1312 &tracker,
1313 return_item,
1314 )?;
1315
1316 tracker.check_unused()?;
1317 Ok(WriteCapacity::condition_check(
1320 &check.table_name,
1321 existing_json.is_some().then_some(&existing_item),
1322 &super::vector_index::parse_vector_defs(&meta)?,
1323 ))
1324}
1325
1326fn check_condition_tracked(
1327 expression: &str,
1328 item: &Item,
1329 tracker: &crate::expressions::TrackedExpressionAttributes,
1330 return_item_on_failure: Option<Item>,
1331) -> Result<()> {
1332 let parsed = crate::expressions::condition::parse(expression)
1333 .map_err(DynoxideError::ValidationException)?;
1334 let result = crate::expressions::condition::evaluate(&parsed, item, tracker)
1335 .map_err(DynoxideError::ValidationException)?;
1336 if !result {
1337 return Err(DynoxideError::ConditionalCheckFailedException(
1338 "The conditional request failed".to_string(),
1339 return_item_on_failure,
1340 ));
1341 }
1342 Ok(())
1343}
1344
1345fn get_action_table_and_size(item: &TransactWriteItem) -> (String, usize) {
1351 if let Some(ref put) = item.put {
1352 (put.table_name.clone(), types::item_size(&put.item))
1353 } else if let Some(ref update) = item.update {
1354 let key_size = types::item_size(&update.key);
1355 let eav_size = update
1356 .expression_attribute_values
1357 .as_ref()
1358 .map(|vals| vals.values().map(|v| v.size()).sum::<usize>())
1359 .unwrap_or(0);
1360 (update.table_name.clone(), key_size + eav_size)
1361 } else if let Some(ref delete) = item.delete {
1362 (delete.table_name.clone(), types::item_size(&delete.key))
1363 } else if let Some(ref check) = item.condition_check {
1364 (check.table_name.clone(), types::item_size(&check.key))
1365 } else {
1366 (String::new(), 0)
1367 }
1368}
1369
1370async fn target_for<S: StorageBackend>(
1374 storage: &S,
1375 table_name: &str,
1376 key_source: &HashMap<String, AttributeValue>,
1377) -> Result<Option<String>> {
1378 crate::validation::validate_table_name(table_name)?;
1379 let meta = helpers::require_table_for_item_op(storage, table_name).await?;
1380 let key_schema = helpers::parse_key_schema(&meta)?;
1381 match helpers::extract_key_strings(key_source, &key_schema) {
1382 Ok((pk, sk)) => Ok(Some(format!("{table_name}#{pk}#{sk}"))),
1383 Err(_) => Ok(None),
1384 }
1385}
1386
1387async fn get_item_target<S: StorageBackend>(
1390 storage: &S,
1391 item: &TransactWriteItem,
1392) -> Result<Option<String>> {
1393 if let Some(ref put) = item.put {
1394 target_for(storage, &put.table_name, &put.item).await
1395 } else if let Some(ref update) = item.update {
1396 target_for(storage, &update.table_name, &update.key).await
1397 } else if let Some(ref delete) = item.delete {
1398 target_for(storage, &delete.table_name, &delete.key).await
1399 } else if let Some(ref check) = item.condition_check {
1400 target_for(storage, &check.table_name, &check.key).await
1401 } else {
1402 Err(DynoxideError::ValidationException(
1403 "TransactItem must contain exactly one action".to_string(),
1404 ))
1405 }
1406}