dynoxide-rs 3.0.0

use crate::actions::helpers;
use crate::errors::{DynoxideError, Result};
use crate::storage_backend::StorageBackend;
use crate::types::{self, AttributeValue, Item};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;

/// Internal deserialization struct that uses Options to detect missing fields.
#[derive(Debug, Default, Deserialize)]
struct PutItemRequestRaw {
    #[serde(rename = "TableName", default)]
    table_name: Option<String>,
    #[serde(rename = "Item", default)]
    item: Option<HashMap<String, AttributeValue>>,
    #[serde(rename = "ReturnValues", default)]
    return_values: Option<String>,
    #[serde(rename = "ConditionExpression", default)]
    condition_expression: Option<String>,
    #[serde(rename = "ExpressionAttributeNames", default)]
    expression_attribute_names: Option<HashMap<String, String>>,
    #[serde(rename = "ExpressionAttributeValues", default)]
    expression_attribute_values: Option<HashMap<String, AttributeValue>>,
    #[serde(rename = "ReturnConsumedCapacity", default)]
    return_consumed_capacity: Option<String>,
    #[serde(rename = "ReturnValuesOnConditionCheckFailure", default)]
    return_values_on_condition_check_failure: Option<String>,
    #[serde(rename = "ReturnItemCollectionMetrics", default)]
    return_item_collection_metrics: Option<String>,
    #[serde(rename = "Expected", default)]
    expected: Option<serde_json::Value>,
    #[serde(rename = "ConditionalOperator", default)]
    conditional_operator: Option<String>,
}

#[derive(Debug, Default)]
pub struct PutItemRequest {
    pub table_name: String,
    pub item: HashMap<String, AttributeValue>,
    pub return_values: Option<String>,
    pub condition_expression: Option<String>,
    pub expression_attribute_names: Option<HashMap<String, String>>,
    pub expression_attribute_values: Option<HashMap<String, AttributeValue>>,
    pub return_consumed_capacity: Option<String>,
    pub return_values_on_condition_check_failure: Option<String>,
    pub return_item_collection_metrics: Option<String>,
    pub expected: Option<serde_json::Value>,
    pub conditional_operator: Option<String>,
}

impl<'de> serde::Deserialize<'de> for PutItemRequest {
    fn deserialize<D: serde::Deserializer<'de>>(
        deserializer: D,
    ) -> std::result::Result<Self, D::Error> {
        let raw = PutItemRequestRaw::deserialize(deserializer)?;

        // A BOOL or NULL in a legacy condition, or its `Exists`, that doesn't
        // read as a boolean makes the whole request malformed, ahead of
        // everything else.
        if let Some(message) = crate::types::malformed_condition_boolean(raw.expected.as_ref()) {
            return Err(serde::de::Error::custom(message));
        }

        use crate::validation::format_validation_errors;

        // An empty name is the newer validation layer's own check and is
        // reported on its own. Any other name error joins the member errors
        // below, one per name (captured 2026-10-09, five of seven regions).
        if raw.table_name.as_deref() == Some("") {
            return Err(serde::de::Error::custom(format!(
                "VALIDATION:{}",
                crate::validation::envelope_message(crate::validation::EMPTY_TABLE_NAME_REJECTION)
            )));
        }
        let mut errors = crate::validation::item_table_name_errors(raw.table_name.as_deref());
        let table_name = raw.table_name.unwrap_or_default();

        // Item constraint validation
        if raw.item.is_none() {
            errors.push(
                "Value null at 'item' failed to satisfy constraint: \
                 Member must not be null"
                    .to_string(),
            );
        }

        // ReturnConsumedCapacity enum validation
        if let Some(msg) = crate::validation::return_consumed_capacity_rejection(
            raw.return_consumed_capacity.as_deref(),
        ) {
            errors.push(msg);
        }

        // ReturnValues enum validation
        if let Some(ref rv) = raw.return_values
            && !["ALL_NEW", "UPDATED_OLD", "ALL_OLD", "NONE", "UPDATED_NEW"].contains(&rv.as_str())
        {
            errors.push(crate::validation::RETURN_VALUES_REJECTION.to_string());
        }

        // ReturnItemCollectionMetrics enum validation
        if let Some(ref ricm) = raw.return_item_collection_metrics
            && !["SIZE", "NONE"].contains(&ricm.as_str())
        {
            errors.push(format!(
                "Value '{}' at 'returnItemCollectionMetrics' failed to satisfy constraint: \
                     Member must satisfy enum value set: [SIZE, NONE]",
                ricm
            ));
        }

        // ReturnValuesOnConditionCheckFailure enum, after the others
        // (captured 2026-10-09).
        if let Some(msg) = crate::validation::return_values_on_condition_check_failure_rejection(
            raw.return_values_on_condition_check_failure.as_deref(),
            "returnValuesOnConditionCheckFailure",
        ) {
            errors.push(msg);
        }

        if let Some(msg) = format_validation_errors(&errors) {
            return Err(serde::de::Error::custom(format!("VALIDATION:{}", msg)));
        }

        Ok(PutItemRequest {
            table_name,
            item: raw.item.unwrap_or_default(),
            return_values: raw.return_values,
            condition_expression: raw.condition_expression,
            expression_attribute_names: raw.expression_attribute_names,
            expression_attribute_values: raw.expression_attribute_values,
            return_consumed_capacity: raw.return_consumed_capacity,
            return_values_on_condition_check_failure: raw.return_values_on_condition_check_failure,
            return_item_collection_metrics: raw.return_item_collection_metrics,
            expected: raw.expected,
            conditional_operator: raw.conditional_operator,
        })
    }
}

#[derive(Debug, Default, Serialize)]
pub struct PutItemResponse {
    #[serde(rename = "Attributes", skip_serializing_if = "Option::is_none")]
    // Number sets in numeric order here, as DynamoDB returns them;
    // pinned by set_order's every_response_item_field test.
    pub attributes: Option<HashMap<String, AttributeValue>>,
    #[serde(rename = "ConsumedCapacity", skip_serializing_if = "Option::is_none")]
    pub consumed_capacity: Option<types::ConsumedCapacity>,
    #[serde(
        rename = "ItemCollectionMetrics",
        skip_serializing_if = "Option::is_none"
    )]
    pub item_collection_metrics: Option<crate::types::ItemCollectionMetrics>,
}

pub async fn execute<S: StorageBackend>(
    storage: &S,
    request: PutItemRequest,
) -> Result<PutItemResponse> {
    // Apply the request-validation envelope exactly once at the operation
    // boundary, so every early return inside is covered and the tagged
    // variant never escapes. A request read on the older layer never gets it.
    let layer = crate::validation::single_write_layer(request.expected.as_ref());
    execute_inner(storage, request, layer)
        .await
        .map_err(|e| crate::validation::resolve_for_layer(layer, e))
}

async fn execute_inner<S: StorageBackend>(
    storage: &S,
    mut request: PutItemRequest,
    layer: crate::validation::ValueLayer,
) -> Result<PutItemResponse> {
    // Validate table name format before checking existence (DynamoDB validates input first)
    crate::validation::validate_item_table_name(&request.table_name)?;

    // An empty attribute name is refused as the request is read, ahead of every
    // other value check and the table lookup (captured in ten regions).
    crate::empty_names::check_single_item_write(
        &request.item,
        false,
        request.expected.as_ref(),
        request.expression_attribute_names.as_ref(),
        request.expression_attribute_values.as_ref(),
        None,
    )?;

    // Reject {NULL: false} up front. The HTTP path rejects it during request
    // deserialisation; this keeps the in-process API in agreement (eu-west-2).
    helpers::validate_no_null_false(&request.item)?;
    if let Some(ref values) = request.expression_attribute_values {
        helpers::validate_no_null_false(values)?;
    }

    // Validate attribute values (empty strings, empty sets, number precision)
    // DynamoDB validates item values before expression parameter checks for PutItem.
    // The set families are tagged for the request-validation envelope.
    crate::validation::validate_item_attribute_values_classified(&request.item, layer)
        .map_err(crate::validation::ClassifiedValidationError::into_tagged)?;

    // Validate expression/non-expression parameter conflicts
    {
        let mut non_expr = Vec::new();
        let mut expr_params = Vec::new();
        if request.expected.is_some() {
            non_expr.push("Expected");
        }
        if request.conditional_operator.is_some() {
            non_expr.push("ConditionalOperator");
        }
        if request.condition_expression.is_some() {
            expr_params.push("ConditionExpression");
        }
        let ctx = helpers::ExpressionParamContext {
            non_expression_params: non_expr,
            expression_params: expr_params,
            // DynamoDB names no expression after its "can only be specified
            // when using expressions" message on the single-item writes
            // (eu-west-2 and us-east-1, 2026-10-09).
            all_expression_param_names: vec![],
            expression_attribute_names: &request.expression_attribute_names,
            expression_attribute_values: &request.expression_attribute_values,
            expression_attribute_values_raw: &None,
            layer,
        };
        // Raw EAV not used for PutItem HTTP path. Enveloped families are
        // tagged for the request-validation envelope.
        helpers::validate_expression_params(&ctx)
            .map_err(crate::validation::ClassifiedValidationError::into_tagged)?;
    }

    // ConditionalOperator only joins legacy Expected conditions, so it is
    // refused without two of them, whatever its value.
    if let Some(msg) = crate::validation::conditional_operator_misuse(
        request.conditional_operator.as_deref(),
        crate::validation::legacy_condition_count(request.expected.as_ref()),
    ) {
        return Err(DynoxideError::EnvelopedValidation(msg.to_string()));
    }
    // An Expected value nested too deep is refused, inside the envelope.
    crate::value_nesting::check_expected(request.expected.as_ref())?;

    // Validate empty ConditionExpression, inside the envelope (eu-west-2 and
    // us-east-1, 2026-10-09).
    if let Some(ref ce) = request.condition_expression
        && ce.is_empty()
    {
        return Err(DynoxideError::EnvelopedValidation(
            "Invalid ConditionExpression: The expression can not be empty;".to_string(),
        ));
    }

    // Statically validate ConditionExpression (syntax + BETWEEN bounds, etc.) before table lookup.
    // Every parse-time error here is a family DynamoDB wraps in the
    // request-validation envelope, so the whole block is tagged.
    if let Some(ref ce) = request.condition_expression {
        let parsed = crate::expressions::condition::parse(ce).map_err(|e| {
            DynoxideError::EnvelopedValidation(format!("Invalid ConditionExpression: {e}"))
        })?;
        crate::expressions::condition::validate_static(
            &parsed,
            &request.expression_attribute_values,
        )
        .map_err(DynoxideError::EnvelopedValidation)?;
        crate::expressions::condition::validate_operand_semantics(
            &parsed,
            &request.expression_attribute_names,
            &request.expression_attribute_values,
        )
        .map_err(|e| {
            DynoxideError::EnvelopedValidation(format!("Invalid ConditionExpression: {e}"))
        })?;
    }

    // Validate ReturnValues parameter (PutItem only supports NONE and ALL_OLD).
    // Wrapped in the request-validation envelope (eu-west-2).
    if let Some(ref rv) = request.return_values {
        let rv_upper = rv.to_uppercase();
        if rv_upper != "NONE" && rv_upper != "ALL_OLD" {
            return Err(DynoxideError::EnvelopedValidation(
                "ReturnValues can only be ALL_OLD or NONE".to_string(),
            ));
        }
    }

    // Read once and use on both sides of the table lookup: the validation just
    // below, which has to run before the table is checked, and the conversion
    // after it. Nothing between the two touches `expected` or
    // `condition_expression`.
    let expected: Option<HashMap<String, helpers::ExpectedCondition>> =
        match request.expected.as_ref() {
            Some(v) if request.condition_expression.is_none() => helpers::read_expected(v)?,
            _ => None,
        };

    // Validate legacy Expected parameter BEFORE checking table existence
    // (DynamoDB validates request parameters before checking table)
    if let Some(ref expected) = expected {
        // Validate Expected conditions (ComparisonOperator, Value, Exists
        // conflicts). Every family comes back in the envelope (eu-west-2 and
        // us-east-1, 2026-10-09).
        helpers::validate_expected_conditions(expected)
            .map_err(crate::validation::tag_for_envelope)?;
    }

    // Validate item size BEFORE checking table existence
    // (DynamoDB validates item size before checking if table exists)
    let size = types::item_size(&request.item);
    if size > types::MAX_ITEM_SIZE {
        return Err(DynoxideError::ValidationException(
            "Item size has exceeded the maximum allowed size".to_string(),
        ));
    }

    // The condition's names and values, from the request alone: one it names
    // but the request doesn't define, then any the condition doesn't use. Both
    // come back in the envelope, ahead of the table lookup (us-east-1 and
    // eu-west-2, 2026-10-09).
    if let Some(ref ce) = request.condition_expression {
        let parsed = crate::expressions::condition::parse(ce).map_err(|e| {
            DynoxideError::EnvelopedValidation(format!("Invalid ConditionExpression: {e}"))
        })?;
        let tracker = crate::expressions::TrackedExpressionAttributes::new(
            &request.expression_attribute_names,
            &request.expression_attribute_values,
        );
        crate::expressions::condition::track_references(&parsed, &tracker).map_err(|e| {
            DynoxideError::EnvelopedValidation(format!("Invalid ConditionExpression: {e}"))
        })?;
        tracker
            .check_unused()
            .map_err(crate::validation::tag_for_envelope)?;
    }

    let meta = helpers::require_table_for_item_op(storage, &request.table_name).await?;
    let key_schema = helpers::parse_key_schema(&meta)?;

    // Convert legacy Expected parameter to ConditionExpression if no expression is set
    if let Some(ref expected) = expected
        && !expected.is_empty()
    {
        let converted = helpers::convert_expected_to_condition(
            expected,
            request.conditional_operator.as_deref(),
        )?;
        if !converted.expression.is_empty() {
            request.condition_expression = Some(converted.expression);
            let expr_values = request
                .expression_attribute_values
                .get_or_insert_with(HashMap::new);
            expr_values.extend(converted.attribute_values);
            let expr_names = request
                .expression_attribute_names
                .get_or_insert_with(HashMap::new);
            expr_names.extend(converted.attribute_names);
        }
    }

    // Validate key attributes present and correct types
    helpers::validate_item_keys(&request.item, &key_schema, &meta)?;

    // Normalize sets (deduplication)
    crate::validation::normalize_item_sets(&mut request.item);

    // Size again, now over what is actually written: the figure recorded
    // alongside the row and the capacity charged for the write. This is the
    // billed size, which counts a vector attribute at 4 bytes per dimension;
    // the limit check above keeps the number rule, as DynamoDB's does. On a
    // table without vector indexes the two are the same figure, because a
    // number is sized by its significant digits and normalising does not change
    // those.
    let vector_defs = super::vector_index::parse_vector_defs(&meta)?;
    let size = types::billed_item_size(&request.item, &vector_defs);

    // Extract key values
    // TODO: validation must precede this call -- if reaching this line, caller has already validated keys.
    let (pk, sk) = helpers::extract_key_strings(&request.item, &key_schema)?;

    // Check for unused expression attribute names/values
    let tracker = crate::expressions::TrackedExpressionAttributes::new(
        &request.expression_attribute_names,
        &request.expression_attribute_values,
    );

    // Pre-register all expression references statically so check_unused sees
    // every :value and #name, even those in short-circuited AND/OR branches.
    if let Some(ref cond_expr) = request.condition_expression
        && let Ok(parsed) = crate::expressions::condition::parse(cond_expr)
    {
        tracker.track_condition_expr(&parsed);
    }

    // Wrap the condition check, base write and the GSI/LSI index fan-out in a
    // single transaction so a mid-fan-out failure rolls the whole write back,
    // leaving no torn index. The transaction is unconditional (not just for
    // ConditionExpression) because the atomicity guarantee applies to every
    // single-item write.
    let (old_item, gsi_units, lsi_units, vector_bytes) =
        helpers::with_write_transaction(storage, async {
            // Evaluate ConditionExpression against existing item (if any)
            let old_json = if request.condition_expression.is_some() {
                let existing_json = storage.get_item(&request.table_name, &pk, &sk).await?;
                let existing_item: HashMap<String, AttributeValue> = existing_json
                    .as_ref()
                    .and_then(|j| serde_json::from_str(j).ok())
                    .unwrap_or_default();

                if let Some(ref cond_expr) = request.condition_expression {
                    let parsed = crate::expressions::condition::parse(cond_expr)
                        .map_err(DynoxideError::ValidationException)?;
                    let result =
                        crate::expressions::condition::evaluate(&parsed, &existing_item, &tracker)
                            .map_err(DynoxideError::ValidationException)?;
                    if !result {
                        let return_item =
                            if request.return_values_on_condition_check_failure.as_deref()
                                == Some("ALL_OLD")
                                && !existing_item.is_empty()
                            {
                                Some(existing_item.clone())
                            } else {
                                None
                            };
                        return Err(DynoxideError::ConditionalCheckFailedException(
                            "The conditional request failed".to_string(),
                            return_item,
                        ));
                    }
                }
                existing_json
            } else {
                None
            };

            // Check for unused expression attribute names/values
            tracker.check_unused()?;

            // Serialize item
            let item_json = serde_json::to_string(&request.item)
                .map_err(|e| DynoxideError::InternalServerError(e.to_string()))?;

            // Compute hash prefix for parallel scan ordering
            let hash_prefix = request
                .item
                .get(&key_schema.partition_key)
                .map(crate::storage::compute_hash_prefix)
                .unwrap_or_default();

            // Store item (returns old item if it existed)
            // If we already fetched old_json for condition check, use put_item but ignore its return
            let old_json = if old_json.is_some() {
                storage
                    .put_item_with_hash(
                        &request.table_name,
                        &pk,
                        &sk,
                        &item_json,
                        size,
                        &hash_prefix,
                    )
                    .await?;
                old_json
            } else {
                storage
                    .put_item_with_hash(
                        &request.table_name,
                        &pk,
                        &sk,
                        &item_json,
                        size,
                        &hash_prefix,
                    )
                    .await?
            };

            // The displaced item, parsed once here for the index capacity delta and
            // reused for the stream record below.
            let old_item: Option<Item> =
                old_json.as_ref().and_then(|j| serde_json::from_str(j).ok());

            let target = super::gsi::IndexWrite {
                table_name: &request.table_name,
                pk: &pk,
                sk: &sk,
                pk_attr: &key_schema.partition_key,
                sk_attr: key_schema.sort_key.as_deref(),
                old_item: old_item.as_ref(),
                capacity_mode: request.return_consumed_capacity.as_deref(),
            };

            // Maintain GSI tables (inside the transaction)
            let gsi_units =
                super::gsi::maintain_gsis_after_write(storage, &meta, &target, &request.item)
                    .await?;

            // Maintain LSI tables (inside the transaction)
            let lsi_units =
                super::lsi::maintain_lsis_after_write(storage, &meta, &target, &request.item)
                    .await?;

            // Maintain vector index shadow tables (inside the transaction)
            let vector_bytes = super::vector_index::maintain_vector_indexes_after_write(
                storage,
                &meta,
                &target,
                &request.item,
            )
            .await?;

            // Record stream event (inside the transaction)
            crate::streams::record_stream_event(
                storage,
                &meta,
                old_item.as_ref(),
                Some(&request.item),
            )
            .await?;

            Ok((old_item, gsi_units, lsi_units, vector_bytes))
        })
        .await?;

    // Handle ReturnValues
    let return_old = request
        .return_values
        .as_deref()
        .unwrap_or("NONE")
        .eq_ignore_ascii_case("ALL_OLD");

    let attributes = if return_old { old_item.clone() } else { None };

    // Build item collection metrics (only for tables with LSIs)
    let pk_value = request.item.get(&key_schema.partition_key).cloned();
    let item_collection_metrics = helpers::build_item_collection_metrics(
        storage,
        &meta,
        &request.table_name,
        &pk,
        &key_schema.partition_key,
        pk_value
            .as_ref()
            .unwrap_or(&AttributeValue::S(String::new())),
        &request.return_item_collection_metrics,
    )
    .await?;

    let consumed_capacity = types::consumed_capacity_with_vector_indexes(
        &request.table_name,
        types::table_write_capacity_units(
            old_item
                .as_ref()
                .map(|old| types::billed_item_size(old, &vector_defs)),
            Some(size),
        ),
        &gsi_units,
        &lsi_units,
        &vector_bytes,
        &request.return_consumed_capacity,
    );

    Ok(PutItemResponse {
        attributes,
        consumed_capacity,
        item_collection_metrics,
    })
}

#[cfg(test)]
mod tests {
    use crate::actions::{create_table, put_item};
    use crate::storage::Storage;
    use crate::storage_backend::StorageBackend;

    #[test]
    fn empty_table_name_short_circuits_before_return_values() {
        // eu-west-2: an empty TableName is reported alone, before the invalid
        // ReturnValues enum.
        let err = serde_json::from_value::<super::PutItemRequest>(serde_json::json!({
            "TableName": "",
            "ReturnValues": "INVALID",
            "Item": {}
        }))
        .unwrap_err()
        .to_string();
        assert!(err.contains("1 validation error detected"), "got: {err}");
        assert!(err.to_lowercase().contains("tablename"), "got: {err}");
        assert!(!err.to_lowercase().contains("returnvalues"), "got: {err}");
    }

    #[test]
    fn valid_table_name_aggregates_all_invalid_enums() {
        // With a valid table name, PutItem aggregates every invalid Return* enum
        // (unlike UpdateItem, which stops at the first).
        let err = serde_json::from_value::<super::PutItemRequest>(serde_json::json!({
            "TableName": "_conformance_valid_table_name",
            "Item": {"pk": {"S": "test"}},
            "ReturnConsumedCapacity": "INVALID",
            "ReturnItemCollectionMetrics": "INVALID",
            "ReturnValues": "INVALID"
        }))
        .unwrap_err()
        .to_string();
        assert!(err.contains("3 validation errors detected"), "got: {err}");
        assert!(
            err.to_lowercase().contains("returnconsumedcapacity"),
            "got: {err}"
        );
        assert!(
            err.to_lowercase().contains("returnitemcollectionmetrics"),
            "got: {err}"
        );
        // ReturnValues in the generic form, which names neither the value nor
        // the member (captured 2026-10-08).
        assert!(
            err.contains("[ALL_OLD, UPDATED_OLD, ALL_NEW, UPDATE_NEW, NONE]"),
            "got: {err}"
        );
    }

    #[test]
    fn deserialiser_constraint_error_carries_exactly_one_envelope() {
        // Constraint errors raised during deserialisation envelope themselves;
        // they never reach the execute boundary, so no second envelope appears.
        let err = serde_json::from_value::<super::PutItemRequest>(serde_json::json!({
            "TableName": "Tbl",
            "Item": {"pk": {"S": "test"}},
            "ReturnValues": "INVALID"
        }))
        .unwrap_err()
        .to_string();
        assert_eq!(
            err.matches("validation error detected").count(),
            1,
            "expected exactly one envelope, got: {err}"
        );
    }

    /// A single-item write and its GSI fan-out succeed or fail as one unit:
    /// when a later GSI write fails mid-fan-out, the base write and the GSI
    /// entry already written earlier in the same transaction both roll back.
    #[test]
    fn put_item_rolls_back_base_write_when_gsi_fan_out_fails() {
        let storage = Storage::memory().unwrap();

        let create = serde_json::from_value(serde_json::json!({
            "TableName": "Orders",
            "KeySchema": [
                {"AttributeName": "UserId", "KeyType": "HASH"},
                {"AttributeName": "Timestamp", "KeyType": "RANGE"}
            ],
            "AttributeDefinitions": [
                {"AttributeName": "UserId", "AttributeType": "S"},
                {"AttributeName": "Timestamp", "AttributeType": "S"},
                {"AttributeName": "Status", "AttributeType": "S"},
                {"AttributeName": "Priority", "AttributeType": "S"}
            ],
            "GlobalSecondaryIndexes": [
                {
                    "IndexName": "StatusIndex",
                    "KeySchema": [{"AttributeName": "Status", "KeyType": "HASH"}],
                    "Projection": {"ProjectionType": "ALL"}
                },
                {
                    "IndexName": "PriorityIndex",
                    "KeySchema": [{"AttributeName": "Priority", "KeyType": "HASH"}],
                    "Projection": {"ProjectionType": "ALL"}
                }
            ]
        }))
        .unwrap();
        pollster::block_on(create_table::execute(&storage, create)).unwrap();

        // Inject a fan-out failure: drop the second GSI's physical table while
        // its metadata stays. The fan-out maintains StatusIndex first (which
        // succeeds inside the transaction), then hits the now-missing
        // PriorityIndex table and errors.
        storage.drop_gsi_table("Orders", "PriorityIndex").unwrap();

        let put = serde_json::from_value(serde_json::json!({
            "TableName": "Orders",
            "Item": {
                "UserId": {"S": "user1"},
                "Timestamp": {"S": "2024-01-01"},
                "Status": {"S": "SHIPPED"},
                "Priority": {"S": "HIGH"}
            }
        }))
        .unwrap();
        let res = pollster::block_on(put_item::execute(&storage, put));
        assert!(
            res.is_err(),
            "a mid-fan-out failure must surface as an error"
        );

        // The base write must roll back entirely.
        let count =
            pollster::block_on(<Storage as StorageBackend>::count_items(&storage, "Orders"))
                .unwrap();
        assert_eq!(count, 0, "base write must roll back when fan-out fails");

        // The first GSI's entry, written before the failure, must roll back too
        // (no torn index).
        let g1 = pollster::block_on(<Storage as StorageBackend>::query_gsi_items(
            &storage,
            "Orders",
            "StatusIndex",
            "SHIPPED",
            &Default::default(),
        ))
        .unwrap();
        assert!(
            g1.is_empty(),
            "first GSI entry must roll back with the base write"
        );
    }
}