ddb-outbox-eventbridge-fanout

import json
import os
import time
import uuid

import boto3

_endpoint = os.environ.get("AWS_ENDPOINT_URL")
ddb = boto3.client("dynamodb", endpoint_url=_endpoint)

ORDERS = os.environ["ORDERS_TABLE"]
OUTBOX = os.environ["OUTBOX_TABLE"]

VALID_KINDS = {"OrderCreated", "OrderCancelled"}


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

    order_id = body["order_id"]
    kind = body["kind"]
    if kind not in VALID_KINDS:
        raise ValueError(f"unknown kind {kind!r}")

    extras = {k: v for k, v in body.items() if k not in ("order_id", "kind")}

    event_id = str(uuid.uuid4())
    now = int(time.time())

    ddb.transact_write_items(
        TransactItems=[
            {
                "Put": {
                    "TableName": ORDERS,
                    "Item": {
                        "order_id": {"S": order_id},
                        "kind": {"S": kind},
                        "updated_at": {"N": str(now)},
                        "data": {"S": json.dumps(extras)},
                    },
                }
            },
            {
                "Put": {
                    "TableName": OUTBOX,
                    "Item": {
                        "event_id": {"S": event_id},
                        "order_id": {"S": order_id},
                        "kind": {"S": kind},
                        "created_at": {"N": str(now)},
                        "payload": {"S": json.dumps(body)},
                    },
                    "ConditionExpression": "attribute_not_exists(event_id)",
                }
            },
        ]
    )

    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")
ddb = boto3.client("dynamodb", endpoint_url=_endpoint)
events = boto3.client("events", endpoint_url=_endpoint)

DEDUP_TABLE = os.environ["DEDUP_TABLE"]
BUS_NAME = os.environ["BUS_NAME"]
TTL_SECONDS = int(os.environ.get("TTL_SECONDS", "604800"))


def _attr(image, key):
    v = image.get(key)
    if v is None:
        return None
    if "S" in v:
        return v["S"]
    if "N" in v:
        return v["N"]
    if "BOOL" in v:
        return v["BOOL"]
    return None


def _already_processed(event_id):
    got = ddb.get_item(
        TableName=DEDUP_TABLE,
        Key={"event_id": {"S": event_id}},
        ConsistentRead=True,
    )
    return "Item" in got


def _mark_processed(event_id, kind):
    expires_at = int(time.time()) + TTL_SECONDS
    try:
        ddb.put_item(
            TableName=DEDUP_TABLE,
            Item={
                "event_id": {"S": event_id},
                "kind": {"S": kind},
                "expires_at": {"N": str(expires_at)},
            },
            ConditionExpression="attribute_not_exists(event_id)",
        )
    except ClientError as e:
        if e.response["Error"]["Code"] != "ConditionalCheckFailedException":
            raise


def handler(event, context):
    failures = []
    for record in event.get("Records", []):
        seq = record.get("dynamodb", {}).get("SequenceNumber")
        try:
            if record.get("eventName") != "INSERT":
                continue

            new_image = record.get("dynamodb", {}).get("NewImage", {})
            event_id = _attr(new_image, "event_id")
            kind = _attr(new_image, "kind")
            order_id = _attr(new_image, "order_id")
            payload = _attr(new_image, "payload")

            if not event_id or not kind:
                print(f"skipping malformed record seq={seq}")
                continue

            if _already_processed(event_id):
                print(f"dedup hit event_id={event_id}")
                continue

            detail = {
                "event_id": event_id,
                "order_id": order_id,
                "kind": kind,
                "payload": payload,
            }
            resp = events.put_events(
                Entries=[
                    {
                        "Source": "outbox.relay",
                        "DetailType": kind,
                        "Detail": json.dumps(detail),
                        "EventBusName": BUS_NAME,
                    }
                ]
            )
            if resp.get("FailedEntryCount", 0) > 0:
                print(f"PutEvents failed for event_id={event_id}: {resp}")
                failures.append({"itemIdentifier": seq})
                continue

            _mark_processed(event_id, kind)

        except Exception as e:
            print(f"record seq={seq} error: {e}")
            if seq:
                failures.append({"itemIdentifier": seq})

    return {"batchItemFailures": failures}

{
  "query": "select:TodoWrite",
  "max_results": 1
}

, output, 
{"type": "tool_reference", "tool_name": "TodoWrite"}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": { "Service": "lambda.amazonaws.com" },
      "Action": "sts:AssumeRole"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Id": "outbox-cmk-policy",
  "Statement": [
    {
      "Sid": "EnableAccountRootIAM",
      "Effect": "Allow",
      "Principal": { "AWS": "arn:aws:iam::${ACCOUNT_ID}:root" },
      "Action": [
        "kms:Create*",
        "kms:Describe*",
        "kms:Enable*",
        "kms:List*",
        "kms:Put*",
        "kms:Update*",
        "kms:Revoke*",
        "kms:Disable*",
        "kms:Get*",
        "kms:Delete*",
        "kms:TagResource",
        "kms:UntagResource",
        "kms:ScheduleKeyDeletion",
        "kms:CancelKeyDeletion",
        "kms:Encrypt",
        "kms:Decrypt",
        "kms:ReEncrypt*",
        "kms:GenerateDataKey*"
      ],
      "Resource": "*"
    },
    {
      "Sid": "AllowEventBridgeUseForSqsEncryption",
      "Effect": "Allow",
      "Principal": { "Service": "events.amazonaws.com" },
      "Action": [
        "kms:GenerateDataKey",
        "kms:Decrypt",
        "kms:DescribeKey"
      ],
      "Resource": "*"
    },
    {
      "Sid": "AllowDynamoDBService",
      "Effect": "Allow",
      "Principal": { "Service": "dynamodb.amazonaws.com" },
      "Action": [
        "kms:GenerateDataKey",
        "kms:Decrypt",
        "kms:DescribeKey"
      ],
      "Resource": "*"
    },
    {
      "Sid": "AllowSqsService",
      "Effect": "Allow",
      "Principal": { "Service": "sqs.amazonaws.com" },
      "Action": [
        "kms:GenerateDataKey",
        "kms:Decrypt",
        "kms:DescribeKey"
      ],
      "Resource": "*"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "Logs",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:${REGION}:${ACCOUNT_ID}:log-group:/aws/lambda/order-api*"
    },
    {
      "Sid": "WriteOrdersAndOutboxAtomically",
      "Effect": "Allow",
      "Action": [
        "dynamodb:PutItem",
        "dynamodb:UpdateItem",
        "dynamodb:GetItem",
        "dynamodb:DescribeTable",
        "dynamodb:TransactWriteItems"
      ],
      "Resource": [
        "arn:aws:dynamodb:${REGION}:${ACCOUNT_ID}:table/orders",
        "arn:aws:dynamodb:${REGION}:${ACCOUNT_ID}:table/outbox"
      ]
    },
    {
      "Sid": "UseCmkForTableEncryption",
      "Effect": "Allow",
      "Action": [
        "kms:Encrypt",
        "kms:Decrypt",
        "kms:ReEncrypt*",
        "kms:GenerateDataKey*",
        "kms:DescribeKey"
      ],
      "Resource": "${CMK_ARN}"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "Logs",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:${REGION}:${ACCOUNT_ID}:log-group:/aws/lambda/outbox-relay*"
    },
    {
      "Sid": "ReadOutboxStream",
      "Effect": "Allow",
      "Action": [
        "dynamodb:DescribeStream",
        "dynamodb:GetRecords",
        "dynamodb:GetShardIterator"
      ],
      "Resource": "${OUTBOX_STREAM_ARN}"
    },
    {
      "Sid": "ListStreams",
      "Effect": "Allow",
      "Action": ["dynamodb:ListStreams"],
      "Resource": "*"
    },
    {
      "Sid": "DedupTable",
      "Effect": "Allow",
      "Action": [
        "dynamodb:GetItem",
        "dynamodb:PutItem",
        "dynamodb:DescribeTable"
      ],
      "Resource": "arn:aws:dynamodb:${REGION}:${ACCOUNT_ID}:table/processed_events"
    },
    {
      "Sid": "PublishToOrderEventsBus",
      "Effect": "Allow",
      "Action": ["events:PutEvents"],
      "Resource": "arn:aws:events:${REGION}:${ACCOUNT_ID}:event-bus/order-events"
    },
    {
      "Sid": "UseCmkForStreamAndDedup",
      "Effect": "Allow",
      "Action": [
        "kms:Decrypt",
        "kms:GenerateDataKey",
        "kms:DescribeKey"
      ],
      "Resource": "${CMK_ARN}"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowEventBridgeRule",
      "Effect": "Allow",
      "Principal": { "Service": "events.amazonaws.com" },
      "Action": "sqs:SendMessage",
      "Resource": "${QUEUE_ARN}",
      "Condition": {
        "ArnEquals": { "aws:SourceArn": "${RULE_ARN}" }
      }
    }
  ]
}

#!/usr/bin/env bash
# Deploy outbox pattern stack into LocalStack from zero.
set -euo pipefail

ENDPOINT="${AWS_ENDPOINT_URL:-http://localstack:4566}"
REGION="${AWS_DEFAULT_REGION:-us-east-1}"
ACCOUNT_ID="$(aws --endpoint-url "$ENDPOINT" sts get-caller-identity --query Account --output text)"
export ENDPOINT REGION ACCOUNT_ID

ROOT="$(cd "$(dirname "$0")" && pwd)"
POLICIES="$ROOT/policies"
BUILD="$ROOT/build"
mkdir -p "$BUILD"

awsl() { aws --endpoint-url "$ENDPOINT" --region "$REGION" "$@"; }

render() {
  local tpl="$1"
  python3 -c '
import os, sys, string
with open(sys.argv[1]) as f:
    print(string.Template(f.read()).substitute(os.environ), end="")
' "$tpl"
}

log() { printf "\n==> %s\n" "$*"; }

# ---- 1. KMS CMK + alias --------------------------------------------------
log "KMS: customer managed key + alias/outbox-cmk"

EXISTING_KEY_ID="$(awsl kms list-aliases --query "Aliases[?AliasName=='alias/outbox-cmk'].TargetKeyId | [0]" --output text 2>/dev/null || echo None)"
if [[ -z "$EXISTING_KEY_ID" || "$EXISTING_KEY_ID" == "None" ]]; then
  KEY_ID="$(awsl kms create-key \
      --description "outbox pattern CMK" \
      --key-usage ENCRYPT_DECRYPT \
      --query 'KeyMetadata.KeyId' --output text)"
  awsl kms create-alias --alias-name alias/outbox-cmk --target-key-id "$KEY_ID"
else
  KEY_ID="$EXISTING_KEY_ID"
fi
CMK_ARN="arn:aws:kms:${REGION}:${ACCOUNT_ID}:key/${KEY_ID}"
export CMK_ARN

render "$POLICIES/kms_key_policy.json.tpl" > "$BUILD/kms_key_policy.json"
awsl kms put-key-policy \
  --key-id "$KEY_ID" \
  --policy-name default \
  --policy "file://$BUILD/kms_key_policy.json"

echo "  CMK_ARN=$CMK_ARN"

# ---- 2. DynamoDB tables --------------------------------------------------
create_table() {
  local name="$1"; local pk="$2"; shift 2
  if awsl dynamodb describe-table --table-name "$name" >/dev/null 2>&1; then
    echo "  table $name already exists, skipping create"
    return
  fi
  awsl dynamodb create-table \
    --table-name "$name" \
    --attribute-definitions AttributeName="$pk",AttributeType=S \
    --key-schema AttributeName="$pk",KeyType=HASH \
    --billing-mode PAY_PER_REQUEST \
    --sse-specification "Enabled=true,SSEType=KMS,KMSMasterKeyId=${KEY_ID}" \
    "$@" >/dev/null
  awsl dynamodb wait table-exists --table-name "$name"
}

log "DynamoDB: orders / outbox (with stream) / processed_events"
create_table orders order_id
create_table outbox event_id \
  --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES
create_table processed_events event_id

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

OUTBOX_STREAM_ARN="$(awsl dynamodb describe-table --table-name outbox \
  --query 'Table.LatestStreamArn' --output text)"
export OUTBOX_STREAM_ARN
echo "  OUTBOX_STREAM_ARN=$OUTBOX_STREAM_ARN"

# ---- 3. EventBridge bus + SQS (DLQs first) -------------------------------
log "EventBridge bus order-events"
awsl events create-event-bus --name order-events 2>/dev/null || echo "  bus exists"
BUS_ARN="arn:aws:events:${REGION}:${ACCOUNT_ID}:event-bus/order-events"

create_queue() {
  local name="$1"; shift
  awsl sqs create-queue \
    --queue-name "$name" \
    --attributes "$@" >/dev/null 2>&1 || true
  awsl sqs get-queue-url --queue-name "$name" --query QueueUrl --output text
}

log "SQS: DLQs"
DLQ_CREATED_URL="$(create_queue orders-created-dlq KmsMasterKeyId=$KEY_ID)"
DLQ_CANCELLED_URL="$(create_queue orders-cancelled-dlq KmsMasterKeyId=$KEY_ID)"
DLQ_CREATED_ARN="$(awsl sqs get-queue-attributes --queue-url "$DLQ_CREATED_URL" --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)"
DLQ_CANCELLED_ARN="$(awsl sqs get-queue-attributes --queue-url "$DLQ_CANCELLED_URL" --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)"

log "SQS: main queues with redrive policies"
REDRIVE_CREATED="$(jq -nc --arg arn "$DLQ_CREATED_ARN" '{deadLetterTargetArn:$arn, maxReceiveCount:"5"}')"
REDRIVE_CANCELLED="$(jq -nc --arg arn "$DLQ_CANCELLED_ARN" '{deadLetterTargetArn:$arn, maxReceiveCount:"5"}')"

CREATED_ATTRS="$(jq -nc --arg kms "$KEY_ID" --arg rd "$REDRIVE_CREATED" '{KmsMasterKeyId:$kms, RedrivePolicy:$rd}')"
CANCELLED_ATTRS="$(jq -nc --arg kms "$KEY_ID" --arg rd "$REDRIVE_CANCELLED" '{KmsMasterKeyId:$kms, RedrivePolicy:$rd}')"

awsl sqs create-queue --queue-name orders-created --attributes "$CREATED_ATTRS" >/dev/null 2>&1 || \
  awsl sqs set-queue-attributes \
    --queue-url "$(awsl sqs get-queue-url --queue-name orders-created --query QueueUrl --output text)" \
    --attributes "$CREATED_ATTRS" >/dev/null

awsl sqs create-queue --queue-name orders-cancelled --attributes "$CANCELLED_ATTRS" >/dev/null 2>&1 || \
  awsl sqs set-queue-attributes \
    --queue-url "$(awsl sqs get-queue-url --queue-name orders-cancelled --query QueueUrl --output text)" \
    --attributes "$CANCELLED_ATTRS" >/dev/null

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

# ---- 4. EventBridge rules + targets --------------------------------------
log "EventBridge rules"
RULE_CREATED_PATTERN='{"source":["outbox.relay"],"detail-type":["OrderCreated"]}'
RULE_CANCELLED_PATTERN='{"source":["outbox.relay"],"detail-type":["OrderCancelled"]}'

awsl events put-rule \
  --name on-order-created \
  --event-bus-name order-events \
  --event-pattern "$RULE_CREATED_PATTERN" \
  --state ENABLED >/dev/null

awsl events put-rule \
  --name on-order-cancelled \
  --event-bus-name order-events \
  --event-pattern "$RULE_CANCELLED_PATTERN" \
  --state ENABLED >/dev/null

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

# Set SQS queue policies allowing the rules to send messages.
QUEUE_ARN="$CREATED_ARN" RULE_ARN="$RULE_CREATED_ARN" \
  render "$POLICIES/sqs_queue_policy.json.tpl" > "$BUILD/policy_orders_created.json"
QUEUE_ARN="$CANCELLED_ARN" RULE_ARN="$RULE_CANCELLED_ARN" \
  render "$POLICIES/sqs_queue_policy.json.tpl" > "$BUILD/policy_orders_cancelled.json"

awsl sqs set-queue-attributes \
  --queue-url "$CREATED_URL" \
  --attributes "Policy=$(cat $BUILD/policy_orders_created.json)" >/dev/null
awsl sqs set-queue-attributes \
  --queue-url "$CANCELLED_URL" \
  --attributes "Policy=$(cat $BUILD/policy_orders_cancelled.json)" >/dev/null

awsl events put-targets \
  --rule on-order-created \
  --event-bus-name order-events \
  --targets "Id=q1,Arn=$CREATED_ARN" >/dev/null

awsl events put-targets \
  --rule on-order-cancelled \
  --event-bus-name order-events \
  --targets "Id=q1,Arn=$CANCELLED_ARN" >/dev/null

# ---- 5. IAM roles --------------------------------------------------------
log "IAM: roles + scoped inline policies"

create_role() {
  local role="$1"; local trust="$2"
  if awsl iam get-role --role-name "$role" >/dev/null 2>&1; then
    awsl iam update-assume-role-policy --role-name "$role" --policy-document "file://$trust" >/dev/null
  else
    awsl iam create-role --role-name "$role" --assume-role-policy-document "file://$trust" >/dev/null
  fi
}

create_role order-api-role     "$POLICIES/lambda_trust.json"
create_role outbox-relay-role  "$POLICIES/lambda_trust.json"

render "$POLICIES/order_api_role.json.tpl"    > "$BUILD/order_api_role.json"
render "$POLICIES/outbox_relay_role.json.tpl" > "$BUILD/outbox_relay_role.json"

awsl iam put-role-policy --role-name order-api-role \
  --policy-name order-api-inline \
  --policy-document "file://$BUILD/order_api_role.json"

awsl iam put-role-policy --role-name outbox-relay-role \
  --policy-name outbox-relay-inline \
  --policy-document "file://$BUILD/outbox_relay_role.json"

ORDER_API_ROLE_ARN="arn:aws:iam::${ACCOUNT_ID}:role/order-api-role"
OUTBOX_RELAY_ROLE_ARN="arn:aws:iam::${ACCOUNT_ID}:role/outbox-relay-role"

# ---- 6. Lambda packaging + deploy ----------------------------------------
log "Lambda: package and deploy"

pkg_lambda() {
  local src_dir="$1"; local zip_path="$2"
  rm -f "$zip_path"
  ( cd "$src_dir" && zip -qr "$zip_path" . )
}

pkg_lambda "$ROOT/lambdas/order_api"     "$BUILD/order_api.zip"
pkg_lambda "$ROOT/lambdas/outbox_relay"  "$BUILD/outbox_relay.zip"

deploy_lambda() {
  local name="$1"; local role_arn="$2"; local zip="$3"; local env="$4"
  if awsl lambda get-function --function-name "$name" >/dev/null 2>&1; then
    awsl lambda update-function-code --function-name "$name" --zip-file "fileb://$zip" >/dev/null
    awsl lambda wait function-updated --function-name "$name"
    awsl lambda update-function-configuration \
      --function-name "$name" \
      --role "$role_arn" \
      --environment "$env" \
      --timeout 30 >/dev/null
    awsl lambda wait function-updated --function-name "$name"
  else
    awsl lambda create-function \
      --function-name "$name" \
      --runtime python3.11 \
      --role "$role_arn" \
      --handler handler.handler \
      --timeout 30 \
      --environment "$env" \
      --zip-file "fileb://$zip" >/dev/null
    awsl lambda wait function-active-v2 --function-name "$name"
  fi
}

ORDER_API_ENV="Variables={AWS_ENDPOINT_URL=$ENDPOINT,ORDERS_TABLE=orders,OUTBOX_TABLE=outbox}"
RELAY_ENV="Variables={AWS_ENDPOINT_URL=$ENDPOINT,DEDUP_TABLE=processed_events,BUS_NAME=order-events,TTL_SECONDS=604800}"

deploy_lambda order-api    "$ORDER_API_ROLE_ARN"    "$BUILD/order_api.zip"    "$ORDER_API_ENV"
deploy_lambda outbox-relay "$OUTBOX_RELAY_ROLE_ARN" "$BUILD/outbox_relay.zip" "$RELAY_ENV"

# ---- 7. Event source mapping (DDB stream -> relay) -----------------------
log "Event source mapping outbox stream -> outbox-relay (ReportBatchItemFailures)"

EXISTING_ESM_UUID="$(awsl 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 [[ -z "$EXISTING_ESM_UUID" || "$EXISTING_ESM_UUID" == "None" ]]; then
  awsl lambda create-event-source-mapping \
    --function-name outbox-relay \
    --event-source-arn "$OUTBOX_STREAM_ARN" \
    --starting-position TRIM_HORIZON \
    --batch-size 10 \
    --maximum-batching-window-in-seconds 1 \
    --function-response-types ReportBatchItemFailures >/dev/null
else
  awsl lambda update-event-source-mapping \
    --uuid "$EXISTING_ESM_UUID" \
    --function-response-types ReportBatchItemFailures >/dev/null
fi

# Wait for ESM to become enabled
for i in $(seq 1 30); do
  state="$(awsl lambda list-event-source-mappings \
    --function-name outbox-relay \
    --event-source-arn "$OUTBOX_STREAM_ARN" \
    --query 'EventSourceMappings[0].State' --output text)"
  echo "  ESM state: $state"
  [[ "$state" == "Enabled" ]] && break
  sleep 1
done

log "Done."

#!/usr/bin/env bash
# End-to-end check matching the spec's "done looks like this".
set -euo pipefail

awsl() { aws "$@"; }

CREATED_URL="$(awsl sqs get-queue-url --queue-name orders-created   --query QueueUrl --output text)"
CANCELLED_URL="$(awsl sqs get-queue-url --queue-name orders-cancelled --query QueueUrl --output text)"

count() {
  awsl sqs get-queue-attributes --queue-url "$1" \
    --attribute-names ApproximateNumberOfMessages \
    --query 'Attributes.ApproximateNumberOfMessages' --output text
}

purge() { awsl sqs purge-queue --queue-url "$1" >/dev/null 2>&1 || true; }

wait_count() {
  local url="$1" want="$2" timeout="${3:-30}"
  for i in $(seq 1 "$timeout"); do
    [[ "$(count "$url")" == "$want" ]] && return 0
    sleep 1
  done
  return 1
}

echo "== purge queues + reset dedup =="
purge "$CREATED_URL"; purge "$CANCELLED_URL"
# Wipe processed_events so this run is fresh
for ev in $(awsl dynamodb scan --table-name processed_events --query 'Items[*].event_id.S' --output text); do
  awsl dynamodb delete-item --table-name processed_events --key "{\"event_id\":{\"S\":\"$ev\"}}"
done

echo
echo "== invoke order-api { o-1, OrderCreated } =="
awsl lambda invoke --function-name order-api \
  --cli-binary-format raw-in-base64-out \
  --payload '{"order_id":"o-1","kind":"OrderCreated","sku":"abc","qty":2}' \
  /tmp/r1.json >/dev/null
cat /tmp/r1.json; echo
EVENT_ID_1=$(jq -r .event_id /tmp/r1.json)

echo "  waiting for orders-created to reach 1..."
wait_count "$CREATED_URL" 1 30 || { echo "TIMEOUT: orders-created count != 1 (got $(count "$CREATED_URL"))"; exit 1; }
sleep 2  # give cancelled queue a chance to (incorrectly) receive
C_CREATED="$(count "$CREATED_URL")"
C_CANCELLED="$(count "$CANCELLED_URL")"
echo "  orders-created=$C_CREATED orders-cancelled=$C_CANCELLED"
[[ "$C_CREATED" == "1" && "$C_CANCELLED" == "0" ]] || { echo "FAIL"; exit 1; }

echo "  processed_events row for $EVENT_ID_1:"
awsl dynamodb get-item --table-name processed_events \
  --key "{\"event_id\":{\"S\":\"$EVENT_ID_1\"}}" --query 'Item' --output json
[[ "$(awsl dynamodb get-item --table-name processed_events \
        --key "{\"event_id\":{\"S\":\"$EVENT_ID_1\"}}" \
        --query 'Item.event_id.S' --output text)" == "$EVENT_ID_1" ]] \
  || { echo "FAIL: missing dedup row"; exit 1; }

echo
echo "== invoke order-api { o-2, OrderCancelled } =="
awsl lambda invoke --function-name order-api \
  --cli-binary-format raw-in-base64-out \
  --payload '{"order_id":"o-2","kind":"OrderCancelled","reason":"customer-request"}' \
  /tmp/r2.json >/dev/null
cat /tmp/r2.json; echo

echo "  waiting for orders-cancelled to reach 1..."
wait_count "$CANCELLED_URL" 1 30 || { echo "TIMEOUT: cancelled queue not 1"; exit 1; }
sleep 2
C_CREATED="$(count "$CREATED_URL")"
C_CANCELLED="$(count "$CANCELLED_URL")"
echo "  orders-created=$C_CREATED orders-cancelled=$C_CANCELLED"
[[ "$C_CREATED" == "1" && "$C_CANCELLED" == "1" ]] || { echo "FAIL"; exit 1; }

echo
echo "ALL CHECKS PASSED"