DynamoDB Streams Outbox: Domain Events Without Dual-Write Failures

# services/orders.py
import os, time, uuid
import boto3

ddb = boto3.client("dynamodb")
TABLE = os.environ["TABLE"]

def place_order(order_id: str, user_id: str, total_cents: int) -> None:
 event_id = str(uuid.uuid4())
 sk_event = f"OUTBOX#{event_id}"
 ddb.transact_write_items(
 TransactItems=[
 {
 "Put": {
 "TableName": TABLE,
 "Item": {
 "pk": {"S": f"ORDER#{order_id}"},
 "sk": {"S": "META"},
 "user_id": {"S": user_id},
 "total_cents": {"N": str(total_cents)},
 "status": {"S": "PLACED"},
 },
 "ConditionExpression": "attribute_not_exists(pk)",
 }
 },
 {
 "Put": {
 "TableName": TABLE,
 "Item": {
 "pk": {"S": f"ORDER#{order_id}"},
 "sk": {"S": sk_event},
 "type": {"S": "order.placed"},
 "payload": {"S": f'{{"order_id":"{order_id}","total_cents":{total_cents}}}'},
 "ts": {"N": str(int(time.time()))},
 },
 }
 },
 ]
 )

# ❌ Dual write — EventBridge can succeed while DDB fails (or vice versa)
ddb.put_item(...)
events.put_events(...)

# relay/handler.py
import json, os, boto3

eb = boto3.client("events")
BUS = os.environ["BUS"]

def handler(event, context):
 failures = []
 for rec in event["Records"]:
 if rec["eventName"] not in ("INSERT", "MODIFY"):
 continue
 sk = rec["dynamodb"]["NewImage"]["sk"]["S"]
 if not sk.startswith("OUTBOX#"):
 continue
 try:
 img = rec["dynamodb"]["NewImage"]
 eb.put_events(Entries=[{
 "EventBusName": BUS,
 "Source": "orders",
 "DetailType": img["type"]["S"],
 "Detail": img["payload"]["S"],
 "Resources": [img["pk"]["S"]],
 }])
 except Exception:
 failures.append({"itemIdentifier": rec["eventID"]})
 return {"batchItemFailures": failures}