ddb-outbox-eventbridge-fanout

import json
import os
import uuid
from datetime import datetime, timezone

import boto3

ENDPOINT = os.environ.get("AWS_ENDPOINT_URL", "http://localstack:4566")
ddb = boto3.client("dynamodb", endpoint_url=ENDPOINT, region_name="us-east-1")


def lambda_handler(event, context):
    payload = event
    if isinstance(event, dict) and "body" in event and isinstance(event["body"], str):
        payload = json.loads(event["body"])

    order_id = payload["order_id"]
    kind = payload["kind"]
    if kind not in ("OrderCreated", "OrderCancelled"):
        raise ValueError(f"invalid kind: {kind}")

    event_id = str(uuid.uuid4())
    now = datetime.now(timezone.utc).isoformat()
    extras = {k: v for k, v in payload.items() if k not in ("order_id", "kind")}

    status = "created" if kind == "OrderCreated" else "cancelled"

    ddb.transact_write_items(
        TransactItems=[
            {
                "Put": {
                    "TableName": "orders",
                    "Item": {
                        "order_id": {"S": order_id},
                        "status": {"S": status},
                        "updated_at": {"S": now},
                        "payload": {"S": json.dumps(extras)},
                    },
                }
            },
            {
                "Put": {
                    "TableName": "outbox",
                    "Item": {
                        "event_id": {"S": event_id},
                        "order_id": {"S": order_id},
                        "kind": {"S": kind},
                        "payload": {"S": json.dumps(extras)},
                        "created_at": {"S": now},
                    },
                }
            },
        ]
    )

    return {"event_id": event_id, "order_id": order_id, "kind": kind}

import json
import os
import time

import boto3
from botocore.exceptions import ClientError

ENDPOINT = os.environ.get("AWS_ENDPOINT_URL", "http://localstack:4566")
BUS_NAME = os.environ.get("EVENT_BUS_NAME", "order-events")
DEDUP_TABLE = os.environ.get("DEDUP_TABLE", "processed_events")
DEDUP_TTL_SECONDS = int(os.environ.get("DEDUP_TTL_SECONDS", str(7 * 24 * 3600)))

ddb = boto3.client("dynamodb", endpoint_url=ENDPOINT, region_name="us-east-1")
events = boto3.client("events", endpoint_url=ENDPOINT, region_name="us-east-1")


def _unwrap(image):
    out = {}
    for k, v in image.items():
        if "S" in v:
            out[k] = v["S"]
        elif "N" in v:
            out[k] = v["N"]
        elif "BOOL" in v:
            out[k] = v["BOOL"]
        else:
            out[k] = v
    return out


def _claim(event_id):
    ttl = int(time.time()) + DEDUP_TTL_SECONDS
    try:
        ddb.put_item(
            TableName=DEDUP_TABLE,
            Item={"event_id": {"S": event_id}, "ttl": {"N": str(ttl)}},
            ConditionExpression="attribute_not_exists(event_id)",
        )
        return True
    except ClientError as e:
        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
            return False
        raise


def _release(event_id):
    try:
        ddb.delete_item(
            TableName=DEDUP_TABLE, Key={"event_id": {"S": event_id}}
        )
    except Exception:
        pass


def _handle(record):
    if record.get("eventName") != "INSERT":
        return
    new_image = record.get("dynamodb", {}).get("NewImage")
    if not new_image:
        return
    item = _unwrap(new_image)
    event_id = item["event_id"]
    kind = item["kind"]
    if kind not in ("OrderCreated", "OrderCancelled"):
        raise ValueError(f"unexpected kind on outbox row: {kind}")

    if not _claim(event_id):
        return  # already processed

    detail = {
        "event_id": event_id,
        "order_id": item.get("order_id"),
        "kind": kind,
        "payload": json.loads(item.get("payload", "{}")),
        "created_at": item.get("created_at"),
    }

    try:
        resp = events.put_events(
            Entries=[
                {
                    "Source": "outbox.relay",
                    "DetailType": kind,
                    "Detail": json.dumps(detail),
                    "EventBusName": BUS_NAME,
                }
            ]
        )
        if resp.get("FailedEntryCount", 0) > 0:
            raise RuntimeError(f"PutEvents failed: {resp}")
    except Exception:
        _release(event_id)
        raise


def lambda_handler(event, context):
    failures = []
    for record in event.get("Records", []):
        seq = record.get("dynamodb", {}).get("SequenceNumber") or record.get(
            "eventID"
        )
        try:
            _handle(record)
        except Exception as e:
            print(f"failed record {seq}: {e}")
            if seq:
                failures.append({"itemIdentifier": seq})
    return {"batchItemFailures": failures}

#!/usr/bin/env bash
set -euo pipefail

ENDPOINT="${AWS_ENDPOINT_URL:-http://localstack:4566}"
REGION="${AWS_DEFAULT_REGION:-us-east-1}"
ACCOUNT_ID="000000000000"
AWS="aws --endpoint-url=$ENDPOINT --region $REGION"

cd /app

log() { echo "==> $*"; }

log "1/10 KMS customer-managed key + alias/outbox-cmk"
KEY_POLICY=$(cat <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "EnableRootAccount",
      "Effect": "Allow",
      "Principal": {"AWS": "arn:aws:iam::${ACCOUNT_ID}:root"},
      "Action": "kms:*",
      "Resource": "*"
    },
    {
      "Sid": "AllowEventBridgeToUseKey",
      "Effect": "Allow",
      "Principal": {"Service": "events.amazonaws.com"},
      "Action": ["kms:GenerateDataKey", "kms:Decrypt", "kms:DescribeKey"],
      "Resource": "*"
    },
    {
      "Sid": "AllowSQSToUseKey",
      "Effect": "Allow",
      "Principal": {"Service": "sqs.amazonaws.com"},
      "Action": ["kms:GenerateDataKey", "kms:Decrypt", "kms:DescribeKey"],
      "Resource": "*"
    }
  ]
}
EOF
)

if ! $AWS kms describe-key --key-id alias/outbox-cmk >/dev/null 2>&1; then
  KEY_ID=$($AWS kms create-key \
    --description "outbox cmk" \
    --policy "$KEY_POLICY" \
    --query 'KeyMetadata.KeyId' --output text)
  $AWS kms create-alias --alias-name alias/outbox-cmk --target-key-id "$KEY_ID"
fi
KEY_ARN=$($AWS kms describe-key --key-id alias/outbox-cmk --query 'KeyMetadata.Arn' --output text)
log "    KEY_ARN=$KEY_ARN"

log "2/10 DynamoDB tables (orders, outbox w/stream, processed_events)"
create_or_skip_table() {
  local name=$1; shift
  if ! $AWS dynamodb describe-table --table-name "$name" >/dev/null 2>&1; then
    $AWS dynamodb create-table --table-name "$name" "$@" >/dev/null
  fi
}

create_or_skip_table orders \
  --attribute-definitions AttributeName=order_id,AttributeType=S \
  --key-schema AttributeName=order_id,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST \
  --sse-specification "Enabled=true,SSEType=KMS,KMSMasterKeyId=$KEY_ARN"

create_or_skip_table outbox \
  --attribute-definitions AttributeName=event_id,AttributeType=S \
  --key-schema AttributeName=event_id,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST \
  --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES \
  --sse-specification "Enabled=true,SSEType=KMS,KMSMasterKeyId=$KEY_ARN"

create_or_skip_table processed_events \
  --attribute-definitions AttributeName=event_id,AttributeType=S \
  --key-schema AttributeName=event_id,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST \
  --sse-specification "Enabled=true,SSEType=KMS,KMSMasterKeyId=$KEY_ARN"

for t in orders outbox processed_events; do
  $AWS dynamodb wait table-exists --table-name "$t"
done

$AWS dynamodb update-time-to-live \
  --table-name processed_events \
  --time-to-live-specification "Enabled=true,AttributeName=ttl" >/dev/null || true

OUTBOX_STREAM_ARN=$($AWS dynamodb describe-table --table-name outbox \
  --query 'Table.LatestStreamArn' --output text)
ORDERS_TABLE_ARN=$($AWS dynamodb describe-table --table-name orders \
  --query 'Table.TableArn' --output text)
OUTBOX_TABLE_ARN=$($AWS dynamodb describe-table --table-name outbox \
  --query 'Table.TableArn' --output text)
PROCESSED_TABLE_ARN=$($AWS dynamodb describe-table --table-name processed_events \
  --query 'Table.TableArn' --output text)
ORDERS_STREAM_ARN=$($AWS dynamodb describe-table --table-name orders \
  --query 'Table.LatestStreamArn' --output text 2>/dev/null || echo "none")
log "    OUTBOX_STREAM_ARN=$OUTBOX_STREAM_ARN"

log "3/10 EventBridge custom bus order-events (CMK-encrypted)"
if ! $AWS events describe-event-bus --name order-events >/dev/null 2>&1; then
  $AWS events create-event-bus --name order-events --kms-key-identifier "$KEY_ARN" >/dev/null
fi
BUS_ARN=$($AWS events describe-event-bus --name order-events --query 'Arn' --output text)

log "4/10 SQS queues + DLQs (CMK-encrypted)"
create_queue() {
  local name=$1
  local extra=${2:-}
  if ! $AWS sqs get-queue-url --queue-name "$name" >/dev/null 2>&1; then
    if [ -n "$extra" ]; then
      $AWS sqs create-queue --queue-name "$name" --attributes "$extra" >/dev/null
    else
      $AWS sqs create-queue --queue-name "$name" \
        --attributes "{\"KmsMasterKeyId\":\"$KEY_ARN\"}" >/dev/null
    fi
  fi
}

create_queue orders-created-dlq
create_queue orders-cancelled-dlq

DLQ_CREATED_URL=$($AWS sqs get-queue-url --queue-name orders-created-dlq --query 'QueueUrl' --output text)
DLQ_CREATED_ARN=$($AWS sqs get-queue-attributes --queue-url "$DLQ_CREATED_URL" --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)
DLQ_CANCELLED_URL=$($AWS sqs get-queue-url --queue-name orders-cancelled-dlq --query 'QueueUrl' --output text)
DLQ_CANCELLED_ARN=$($AWS sqs get-queue-attributes --queue-url "$DLQ_CANCELLED_URL" --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)

CREATED_ATTRS=$(jq -nc \
  --arg key "$KEY_ARN" \
  --arg dlq "$DLQ_CREATED_ARN" \
  '{KmsMasterKeyId:$key, RedrivePolicy:({deadLetterTargetArn:$dlq, maxReceiveCount:"3"}|tostring)}')
CANCELLED_ATTRS=$(jq -nc \
  --arg key "$KEY_ARN" \
  --arg dlq "$DLQ_CANCELLED_ARN" \
  '{KmsMasterKeyId:$key, RedrivePolicy:({deadLetterTargetArn:$dlq, maxReceiveCount:"3"}|tostring)}')
create_queue orders-created   "$CREATED_ATTRS"
create_queue orders-cancelled "$CANCELLED_ATTRS"

CREATED_URL=$($AWS sqs get-queue-url --queue-name orders-created --query 'QueueUrl' --output text)
CREATED_ARN=$($AWS sqs get-queue-attributes --queue-url "$CREATED_URL" --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)
CANCELLED_URL=$($AWS sqs get-queue-url --queue-name orders-cancelled --query 'QueueUrl' --output text)
CANCELLED_ARN=$($AWS sqs get-queue-attributes --queue-url "$CANCELLED_URL" --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)

log "5/10 SQS queue policies (allow EventBridge rules to send)"
mk_queue_policy() {
  local q_arn=$1
  local rule_arn=$2
  jq -nc \
    --arg q "$q_arn" \
    --arg r "$rule_arn" \
    '{Version:"2012-10-17", Statement:[{
        Sid:"AllowEventBridgeRule",
        Effect:"Allow",
        Principal:{Service:"events.amazonaws.com"},
        Action:"sqs:SendMessage",
        Resource:$q,
        Condition:{ArnEquals:{"aws:SourceArn":$r}}
     }]}'
}

CREATED_RULE_ARN="arn:aws:events:${REGION}:${ACCOUNT_ID}:rule/order-events/on-order-created"
CANCELLED_RULE_ARN="arn:aws:events:${REGION}:${ACCOUNT_ID}:rule/order-events/on-order-cancelled"

CREATED_QPOL=$(mk_queue_policy "$CREATED_ARN" "$CREATED_RULE_ARN")
CANCELLED_QPOL=$(mk_queue_policy "$CANCELLED_ARN" "$CANCELLED_RULE_ARN")

$AWS sqs set-queue-attributes --queue-url "$CREATED_URL" \
  --attributes "$(jq -nc --arg p "$CREATED_QPOL" '{Policy:$p}')" >/dev/null
$AWS sqs set-queue-attributes --queue-url "$CANCELLED_URL" \
  --attributes "$(jq -nc --arg p "$CANCELLED_QPOL" '{Policy:$p}')" >/dev/null

log "6/10 EventBridge rules + targets"
$AWS events put-rule --name on-order-created --event-bus-name order-events \
  --event-pattern '{"detail-type":["OrderCreated"]}' >/dev/null
$AWS events put-rule --name on-order-cancelled --event-bus-name order-events \
  --event-pattern '{"detail-type":["OrderCancelled"]}' >/dev/null

$AWS events put-targets --event-bus-name order-events --rule on-order-created \
  --targets "Id=created-q,Arn=$CREATED_ARN" >/dev/null
$AWS events put-targets --event-bus-name order-events --rule on-order-cancelled \
  --targets "Id=cancelled-q,Arn=$CANCELLED_ARN" >/dev/null

log "7/10 IAM roles (order-api-role, outbox-relay-role)"
ASSUME_LAMBDA='{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"lambda.amazonaws.com"},"Action":"sts:AssumeRole"}]}'

create_or_skip_role() {
  local name=$1
  if ! $AWS iam get-role --role-name "$name" >/dev/null 2>&1; then
    $AWS iam create-role --role-name "$name" \
      --assume-role-policy-document "$ASSUME_LAMBDA" >/dev/null
  fi
}
create_or_skip_role order-api-role
create_or_skip_role outbox-relay-role

API_POLICY=$(jq -nc \
  --arg orders "$ORDERS_TABLE_ARN" \
  --arg outbox "$OUTBOX_TABLE_ARN" \
  --arg key "$KEY_ARN" \
  '{Version:"2012-10-17", Statement:[
    {Sid:"WriteOrdersAndOutboxAtomically",
     Effect:"Allow",
     Action:["dynamodb:PutItem","dynamodb:UpdateItem","dynamodb:GetItem","dynamodb:TransactWriteItems"],
     Resource:[$orders, $outbox]},
    {Sid:"UseCMKForDdbEncryption",
     Effect:"Allow",
     Action:["kms:Encrypt","kms:Decrypt","kms:GenerateDataKey","kms:DescribeKey"],
     Resource:$key},
    {Sid:"Logs",
     Effect:"Allow",
     Action:["logs:CreateLogGroup","logs:CreateLogStream","logs:PutLogEvents"],
     Resource:"arn:aws:logs:*:*:*"}
  ]}')

RELAY_POLICY=$(jq -nc \
  --arg stream "$OUTBOX_STREAM_ARN" \
  --arg processed "$PROCESSED_TABLE_ARN" \
  --arg bus "$BUS_ARN" \
  --arg key "$KEY_ARN" \
  '{Version:"2012-10-17", Statement:[
    {Sid:"ReadOutboxStreamOnly",
     Effect:"Allow",
     Action:["dynamodb:DescribeStream","dynamodb:GetRecords","dynamodb:GetShardIterator","dynamodb:ListStreams"],
     Resource:$stream},
    {Sid:"DedupTableWrites",
     Effect:"Allow",
     Action:["dynamodb:PutItem","dynamodb:GetItem","dynamodb:DeleteItem","dynamodb:UpdateItem"],
     Resource:$processed},
    {Sid:"PutEventsOnOrderEventsBus",
     Effect:"Allow",
     Action:["events:PutEvents"],
     Resource:$bus},
    {Sid:"UseCMK",
     Effect:"Allow",
     Action:["kms:Decrypt","kms:GenerateDataKey","kms:DescribeKey"],
     Resource:$key},
    {Sid:"Logs",
     Effect:"Allow",
     Action:["logs:CreateLogGroup","logs:CreateLogStream","logs:PutLogEvents"],
     Resource:"arn:aws:logs:*:*:*"}
  ]}')

$AWS iam put-role-policy --role-name order-api-role \
  --policy-name order-api-policy --policy-document "$API_POLICY"
$AWS iam put-role-policy --role-name outbox-relay-role \
  --policy-name outbox-relay-policy --policy-document "$RELAY_POLICY"

API_ROLE_ARN=$($AWS iam get-role --role-name order-api-role --query 'Role.Arn' --output text)
RELAY_ROLE_ARN=$($AWS iam get-role --role-name outbox-relay-role --query 'Role.Arn' --output text)

log "8/10 Package + deploy lambdas"
rm -f /tmp/order-api.zip /tmp/outbox-relay.zip
( cd /app && zip -q /tmp/order-api.zip order_api.py )
( cd /app && zip -q /tmp/outbox-relay.zip outbox_relay.py )

deploy_lambda() {
  local name=$1 role=$2 handler=$3 zipfile=$4
  if $AWS lambda get-function --function-name "$name" >/dev/null 2>&1; then
    $AWS lambda update-function-code --function-name "$name" \
      --zip-file "fileb://$zipfile" >/dev/null
    $AWS lambda wait function-updated --function-name "$name"
    $AWS lambda update-function-configuration --function-name "$name" \
      --role "$role" --handler "$handler" --runtime python3.11 \
      --timeout 30 \
      --environment "Variables={AWS_ENDPOINT_URL=$ENDPOINT,EVENT_BUS_NAME=order-events,DEDUP_TABLE=processed_events}" >/dev/null
  else
    $AWS lambda create-function --function-name "$name" \
      --runtime python3.11 \
      --role "$role" \
      --handler "$handler" \
      --zip-file "fileb://$zipfile" \
      --timeout 30 \
      --environment "Variables={AWS_ENDPOINT_URL=$ENDPOINT,EVENT_BUS_NAME=order-events,DEDUP_TABLE=processed_events}" >/dev/null
  fi
  $AWS lambda wait function-active --function-name "$name"
  $AWS lambda wait function-updated --function-name "$name"
}

deploy_lambda order-api    "$API_ROLE_ARN"   order_api.lambda_handler    /tmp/order-api.zip
deploy_lambda outbox-relay "$RELAY_ROLE_ARN" outbox_relay.lambda_handler /tmp/outbox-relay.zip

log "9/10 Event source mapping (outbox stream -> outbox-relay) with ReportBatchItemFailures"
EXISTING_ESM=$($AWS lambda list-event-source-mappings \
  --function-name outbox-relay \
  --event-source-arn "$OUTBOX_STREAM_ARN" \
  --query 'EventSourceMappings[0].UUID' --output text 2>/dev/null || echo "None")
if [ "$EXISTING_ESM" = "None" ] || [ -z "$EXISTING_ESM" ]; then
  $AWS lambda create-event-source-mapping \
    --function-name outbox-relay \
    --event-source-arn "$OUTBOX_STREAM_ARN" \
    --starting-position LATEST \
    --batch-size 10 \
    --maximum-batching-window-in-seconds 1 \
    --function-response-types ReportBatchItemFailures >/dev/null
fi

log "10/10 Done."
echo "KEY_ARN=$KEY_ARN"
echo "BUS_ARN=$BUS_ARN"
echo "OUTBOX_STREAM_ARN=$OUTBOX_STREAM_ARN"
echo "CREATED_QUEUE_URL=$CREATED_URL"
echo "CANCELLED_QUEUE_URL=$CANCELLED_URL"