// transact_handlers.odin — HTTP handlers for TransactWriteItems and TransactGetItems // // Also contains the UPDATED_NEW / UPDATED_OLD filtering helper for UpdateItem. package main import "core:encoding/json" import "core:fmt" import "core:strings" import "dynamodb" // ============================================================================ // TransactWriteItems Handler // // Request format: // { // "TransactItems": [ // { // "Put": { // "TableName": "...", // "Item": { ... }, // "ConditionExpression": "...", // optional // "ExpressionAttributeNames": { ... }, // optional // "ExpressionAttributeValues": { ... } // optional // } // }, // { // "Delete": { // "TableName": "...", // "Key": { ... }, // "ConditionExpression": "...", // optional // ... // } // }, // { // "Update": { // "TableName": "...", // "Key": { ... }, // "UpdateExpression": "...", // "ConditionExpression": "...", // optional // "ExpressionAttributeNames": { ... }, // optional // "ExpressionAttributeValues": { ... } // optional // } // }, // { // "ConditionCheck": { // "TableName": "...", // "Key": { ... }, // "ConditionExpression": "...", // "ExpressionAttributeNames": { ... }, // optional // "ExpressionAttributeValues": { ... } // optional // } // } // ] // } // ============================================================================ handle_transact_write_items :: proc( engine: ^dynamodb.Storage_Engine, request: ^HTTP_Request, response: ^HTTP_Response, ) { data, parse_err := json.parse(request.body, allocator = context.allocator) if parse_err != nil { make_error_response(response, .SerializationException, "Invalid JSON") return } defer json.destroy_value(data) root, root_ok := data.(json.Object) if !root_ok { make_error_response(response, .SerializationException, "Request must be an object") return } transact_items_val, found := root["TransactItems"] if !found { make_error_response(response, .ValidationException, "Missing TransactItems") return } transact_items, ti_ok := transact_items_val.(json.Array) if !ti_ok { make_error_response(response, .ValidationException, "TransactItems must be an array") return } if len(transact_items) == 0 { make_error_response(response, .ValidationException, "TransactItems must contain at least one item") return } if len(transact_items) > 100 { make_error_response(response, .ValidationException, "Member must have length less than or equal to 100") return } // Parse each action actions := make([dynamic]dynamodb.Transact_Write_Action) defer { for &action in actions { dynamodb.transact_write_action_destroy(&action) } delete(actions) } for elem in transact_items { elem_obj, elem_ok := elem.(json.Object) if !elem_ok { make_error_response(response, .ValidationException, "Each TransactItem must be an object") return } action, action_ok := parse_transact_write_action(elem_obj) if !action_ok { make_error_response(response, .ValidationException, "Invalid TransactItem action") return } append(&actions, action) } // Execute transaction result, tx_err := dynamodb.transact_write_items(engine, actions[:]) defer dynamodb.transact_write_result_destroy(&result) switch tx_err { case .None: response_set_body(response, transmute([]byte)string("{}")) case .Cancelled: // Build TransactionCanceledException response builder := strings.builder_make() strings.write_string(&builder, `{"__type":"com.amazonaws.dynamodb.v20120810#TransactionCanceledException","message":"Transaction cancelled, please refer cancellation reasons for specific reasons [`) for reason, i in result.cancellation_reasons { if i > 0 { strings.write_string(&builder, ", ") } strings.write_string(&builder, reason.code) } strings.write_string(&builder, `]","CancellationReasons":[`) for reason, i in result.cancellation_reasons { if i > 0 { strings.write_string(&builder, ",") } fmt.sbprintf(&builder, `{{"Code":"%s","Message":"%s"}}`, reason.code, reason.message) } strings.write_string(&builder, "]}") response_set_status(response, .Bad_Request) resp_body := strings.to_string(builder) response_set_body(response, transmute([]byte)resp_body) case .Validation_Error: make_error_response(response, .ValidationException, "Transaction validation failed") case .Internal_Error: make_error_response(response, .InternalServerError, "Internal error during transaction") } } // Parse a single TransactItem action from JSON @(private = "file") parse_transact_write_action :: proc(obj: json.Object) -> (dynamodb.Transact_Write_Action, bool) { action: dynamodb.Transact_Write_Action action.expr_attr_values = make(map[string]dynamodb.Attribute_Value) // Try Put if put_val, has_put := obj["Put"]; has_put { put_obj, put_ok := put_val.(json.Object) if !put_ok { return {}, false } action.type = .Put return parse_transact_put_action(put_obj, &action) } // Try Delete if del_val, has_del := obj["Delete"]; has_del { del_obj, del_ok := del_val.(json.Object) if !del_ok { return {}, false } action.type = .Delete return parse_transact_key_action(del_obj, &action) } // Try Update if upd_val, has_upd := obj["Update"]; has_upd { upd_obj, upd_ok := upd_val.(json.Object) if !upd_ok { return {}, false } action.type = .Update return parse_transact_update_action(upd_obj, &action) } // Try ConditionCheck if cc_val, has_cc := obj["ConditionCheck"]; has_cc { cc_obj, cc_ok := cc_val.(json.Object) if !cc_ok { return {}, false } action.type = .Condition_Check return parse_transact_key_action(cc_obj, &action) } return {}, false } // Parse common expression fields from a transact action object @(private = "file") parse_transact_expression_fields :: proc(obj: json.Object, action: ^dynamodb.Transact_Write_Action) { // ConditionExpression if ce_val, found := obj["ConditionExpression"]; found { if ce_str, str_ok := ce_val.(json.String); str_ok { action.condition_expr = strings.clone(string(ce_str)) } } // ExpressionAttributeNames if ean_val, found := obj["ExpressionAttributeNames"]; found { if ean_obj, ean_ok := ean_val.(json.Object); ean_ok { names := make(map[string]string) for key, val in ean_obj { if str, str_ok := val.(json.String); str_ok { names[strings.clone(key)] = strings.clone(string(str)) } } action.expr_attr_names = names } } // ExpressionAttributeValues if eav_val, found := obj["ExpressionAttributeValues"]; found { if eav_obj, eav_ok := eav_val.(json.Object); eav_ok { for key, val in eav_obj { attr, attr_ok := dynamodb.parse_attribute_value(val) if attr_ok { action.expr_attr_values[strings.clone(key)] = attr } } } } } // Parse a Put transact action @(private = "file") parse_transact_put_action :: proc( obj: json.Object, action: ^dynamodb.Transact_Write_Action, ) -> (dynamodb.Transact_Write_Action, bool) { // TableName tn_val, tn_found := obj["TableName"] if !tn_found { return {}, false } tn_str, tn_ok := tn_val.(json.String) if !tn_ok { return {}, false } action.table_name = strings.clone(string(tn_str)) // Item item_val, item_found := obj["Item"] if !item_found { return {}, false } item, item_ok := dynamodb.parse_item_from_value(item_val) if !item_ok { return {}, false } action.item = item // Expression fields parse_transact_expression_fields(obj, action) return action^, true } // Parse a Delete or ConditionCheck transact action (both use Key) @(private = "file") parse_transact_key_action :: proc( obj: json.Object, action: ^dynamodb.Transact_Write_Action, ) -> (dynamodb.Transact_Write_Action, bool) { // TableName tn_val, tn_found := obj["TableName"] if !tn_found { return {}, false } tn_str, tn_ok := tn_val.(json.String) if !tn_ok { return {}, false } action.table_name = strings.clone(string(tn_str)) // Key key_val, key_found := obj["Key"] if !key_found { return {}, false } key, key_ok := dynamodb.parse_item_from_value(key_val) if !key_ok { return {}, false } action.key = key // Expression fields parse_transact_expression_fields(obj, action) return action^, true } // Parse an Update transact action @(private = "file") parse_transact_update_action :: proc( obj: json.Object, action: ^dynamodb.Transact_Write_Action, ) -> (dynamodb.Transact_Write_Action, bool) { // TableName tn_val, tn_found := obj["TableName"] if !tn_found { return {}, false } tn_str, tn_ok := tn_val.(json.String) if !tn_ok { return {}, false } action.table_name = strings.clone(string(tn_str)) // Key key_val, key_found := obj["Key"] if !key_found { return {}, false } key, key_ok := dynamodb.parse_item_from_value(key_val) if !key_ok { return {}, false } action.key = key // Expression fields (must be parsed before UpdateExpression so attr values are available) parse_transact_expression_fields(obj, action) // UpdateExpression ue_val, ue_found := obj["UpdateExpression"] if !ue_found { return {}, false } ue_str, ue_ok := ue_val.(json.String) if !ue_ok { return {}, false } plan, plan_ok := dynamodb.parse_update_expression( string(ue_str), action.expr_attr_names, action.expr_attr_values, ) if !plan_ok { return {}, false } action.update_plan = plan return action^, true } // ============================================================================ // TransactGetItems Handler // // Request format: // { // "TransactItems": [ // { // "Get": { // "TableName": "...", // "Key": { ... }, // "ProjectionExpression": "...", // optional // "ExpressionAttributeNames": { ... } // optional // } // } // ] // } // ============================================================================ handle_transact_get_items :: proc( engine: ^dynamodb.Storage_Engine, request: ^HTTP_Request, response: ^HTTP_Response, ) { data, parse_err := json.parse(request.body, allocator = context.allocator) if parse_err != nil { make_error_response(response, .SerializationException, "Invalid JSON") return } defer json.destroy_value(data) root, root_ok := data.(json.Object) if !root_ok { make_error_response(response, .SerializationException, "Request must be an object") return } transact_items_val, found := root["TransactItems"] if !found { make_error_response(response, .ValidationException, "Missing TransactItems") return } transact_items, ti_ok := transact_items_val.(json.Array) if !ti_ok { make_error_response(response, .ValidationException, "TransactItems must be an array") return } if len(transact_items) == 0 { make_error_response(response, .ValidationException, "TransactItems must contain at least one item") return } if len(transact_items) > 100 { make_error_response(response, .ValidationException, "Member must have length less than or equal to 100") return } // Parse each get action actions := make([dynamic]dynamodb.Transact_Get_Action) defer { for &action in actions { dynamodb.transact_get_action_destroy(&action) } delete(actions) } for elem in transact_items { elem_obj, elem_ok := elem.(json.Object) if !elem_ok { make_error_response(response, .ValidationException, "Each TransactItem must be an object") return } get_val, has_get := elem_obj["Get"] if !has_get { make_error_response(response, .ValidationException, "TransactGetItems only supports Get actions") return } get_obj, get_ok := get_val.(json.Object) if !get_ok { make_error_response(response, .ValidationException, "Get action must be an object") return } action, action_ok := parse_transact_get_action(get_obj) if !action_ok { make_error_response(response, .ValidationException, "Invalid Get action") return } append(&actions, action) } // Execute transaction get result, tx_err := dynamodb.transact_get_items(engine, actions[:]) defer dynamodb.transact_get_result_destroy(&result) if tx_err != .None { make_error_response(response, .InternalServerError, "Transaction get failed") return } // Build response builder := strings.builder_make(context.allocator) defer strings.builder_destroy(&builder) strings.write_string(&builder, `{"Responses":[`) for maybe_item, i in result.items { if i > 0 { strings.write_string(&builder, ",") } if item, has_item := maybe_item.?; has_item { strings.write_string(&builder, `{"Item":`) dynamodb.serialize_item_to_builder(&builder, item) strings.write_string(&builder, `}`) } else { strings.write_string(&builder, "{}") } } strings.write_string(&builder, "]}") // Clone the string or we gonna have issues again resp_body := strings.clone(strings.to_string(builder)) response_set_body(response, transmute([]byte)resp_body) } // Parse a single TransactGetItems Get action @(private = "file") parse_transact_get_action :: proc(obj: json.Object) -> (dynamodb.Transact_Get_Action, bool) { action: dynamodb.Transact_Get_Action // TableName tn_val, tn_found := obj["TableName"] if !tn_found { return {}, false } tn_str, tn_ok := tn_val.(json.String) if !tn_ok { return {}, false } action.table_name = strings.clone(string(tn_str)) // Key key_val, key_found := obj["Key"] if !key_found { return {}, false } key, key_ok := dynamodb.parse_item_from_value(key_val) if !key_ok { return {}, false } action.key = key // ProjectionExpression (optional) if pe_val, pe_found := obj["ProjectionExpression"]; pe_found { if pe_str, pe_ok := pe_val.(json.String); pe_ok { // Parse ExpressionAttributeNames for projection attr_names: Maybe(map[string]string) = nil if ean_val, ean_found := obj["ExpressionAttributeNames"]; ean_found { if ean_obj, ean_ok := ean_val.(json.Object); ean_ok { names := make(map[string]string, allocator = context.temp_allocator) for key_str, val in ean_obj { if str, str_ok := val.(json.String); str_ok { names[key_str] = string(str) } } attr_names = names } } parts := strings.split(string(pe_str), ",") paths := make([dynamic]string) for part in parts { trimmed := strings.trim_space(part) if len(trimmed) == 0 { continue } resolved, res_ok := dynamodb.resolve_attribute_name(trimmed, attr_names) if !res_ok { delete(paths) dynamodb.item_destroy(&action.key) return {}, false } append(&paths, strings.clone(resolved)) } action.projection = paths[:] } } return action, true } // ============================================================================ // UPDATED_NEW / UPDATED_OLD Filtering Helper // // DynamoDB ReturnValues semantics: // ALL_NEW → all attributes of the item after the update // ALL_OLD → all attributes of the item before the update // UPDATED_NEW → only the attributes that were modified, with new values // UPDATED_OLD → only the attributes that were modified, with old values // // This filters an item to only include the attributes touched by the // UpdateExpression (the "modified paths"). // ============================================================================ filter_updated_attributes :: proc( item: dynamodb.Item, plan: ^dynamodb.Update_Plan, ) -> dynamodb.Item { modified_paths := dynamodb.get_update_plan_modified_paths(plan) defer delete(modified_paths) return dynamodb.filter_item_to_paths(item, modified_paths) }