ddb-outbox-eventbridge-fanout

"""order-api: the only writer to the orders table.

writes the business row and the outbox row in a single TransactWriteItems
so we never have a half-state where the order exists but the event was lost.
"""
import json
import os
import time
import uuid
from datetime import datetime, timezone

import boto3

ENDPOINT = os.environ.get("AWS_ENDPOINT_URL") or os.environ.get("LOCALSTACK_ENDPOINT")
ORDERS_TABLE = os.environ["ORDERS_TABLE"]
OUTBOX_TABLE = os.environ["OUTBOX_TABLE"]

_ddb = boto3.client("dynamodb", endpoint_url=ENDPOINT)

_VALID_KINDS = {"OrderCreated", "OrderCancelled"}


def _now_iso():
    return datetime.now(timezone.utc).isoformat()


def handler(event, _ctx):
    order_id = event.get("order_id")
    kind = event.get("kind")
    if not order_id or kind not in _VALID_KINDS:
        return {"ok": False, "error": "order_id and kind in {OrderCreated,OrderCancelled} required"}

    event_id = str(uuid.uuid4())
    now = _now_iso()
    payload = {k: v for k, v in event.items() if k not in ("order_id", "kind")}
    payload_json = json.dumps(payload, sort_keys=True, default=str)

    order_item = {
        "order_id": {"S": order_id},
        "last_kind": {"S": kind},
        "last_event_id": {"S": event_id},
        "updated_at": {"S": now},
        "payload": {"S": payload_json},
    }
    outbox_item = {
        "event_id": {"S": event_id},
        "order_id": {"S": order_id},
        "kind": {"S": kind},
        "created_at": {"S": now},
        "payload": {"S": payload_json},
    }

    _ddb.transact_write_items(
        TransactItems=[
            {"Put": {"TableName": ORDERS_TABLE, "Item": order_item}},
            {
                "Put": {
                    "TableName": OUTBOX_TABLE,
                    "Item": outbox_item,
                    "ConditionExpression": "attribute_not_exists(event_id)",
                }
            },
        ]
    )

    return {"ok": True, "event_id": event_id, "order_id": order_id, "kind": kind, "ts": now}

"""outbox-relay: ddb stream consumer that republishes onto the eventbridge bus.

per-record contract:
  - only INSERTs into the outbox table produce new domain events
  - dedup by event_id via a conditional put on processed_events; the row carries
    a TTL so the table doesn't grow forever
  - if the put_events call fails, we delete the dedup row and report the record
    as a batch item failure so the stream redelivers without losing the event
"""
import json
import os
import time

import boto3
from botocore.exceptions import ClientError

ENDPOINT = os.environ.get("AWS_ENDPOINT_URL") or os.environ.get("LOCALSTACK_ENDPOINT")
DEDUP_TABLE = os.environ["DEDUP_TABLE"]
EVENT_BUS = os.environ["EVENT_BUS"]
EVENT_SOURCE = os.environ.get("EVENT_SOURCE", "outbox.relay")
DEDUP_TTL_SECONDS = int(os.environ.get("DEDUP_TTL_SECONDS", "604800"))  # 7d default

_ddb = boto3.client("dynamodb", endpoint_url=ENDPOINT)
_events = boto3.client("events", endpoint_url=ENDPOINT)


def _flatten(image):
    """flatten a dynamodb stream image into plain json , covers the types the outbox uses."""
    if not image:
        return {}
    out = {}
    for k, v in image.items():
        if "S" in v:
            out[k] = v["S"]
        elif "N" in v:
            n = v["N"]
            try:
                out[k] = int(n) if "." not in n else float(n)
            except ValueError:
                out[k] = n
        elif "BOOL" in v:
            out[k] = v["BOOL"]
        elif "NULL" in v:
            out[k] = None
        else:
            out[k] = v
    return out


def _try_dedup_claim(event_id):
    """returns True if we successfully claimed the event_id, False if already processed."""
    now = int(time.time())
    try:
        _ddb.put_item(
            TableName=DEDUP_TABLE,
            Item={
                "event_id": {"S": event_id},
                "processed_at": {"N": str(now)},
                "expires_at": {"N": str(now + DEDUP_TTL_SECONDS)},
            },
            ConditionExpression="attribute_not_exists(event_id)",
        )
        return True
    except ClientError as e:
        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
            return False
        raise


def _release_dedup(event_id):
    try:
        _ddb.delete_item(TableName=DEDUP_TABLE, Key={"event_id": {"S": event_id}})
    except Exception as e:
        print(f"warn: failed to release dedup claim for {event_id}: {e}")


def handler(event, _ctx):
    failures = []
    for record in event.get("Records", []):
        seq = record.get("dynamodb", {}).get("SequenceNumber")
        try:
            if record.get("eventName") != "INSERT":
                continue
            new_image = record["dynamodb"]["NewImage"]
            old_image = record["dynamodb"].get("OldImage") or {}
            event_id = new_image["event_id"]["S"]
            kind = new_image["kind"]["S"]

            if not _try_dedup_claim(event_id):
                print(f"skip: event_id={event_id} already processed")
                continue

            detail = {
                "event_id": event_id,
                "order_id": new_image.get("order_id", {}).get("S"),
                "kind": kind,
                "new_image": _flatten(new_image),
                "old_image": _flatten(old_image),
            }
            try:
                resp = _events.put_events(
                    Entries=[
                        {
                            "Source": EVENT_SOURCE,
                            "DetailType": kind,
                            "Detail": json.dumps(detail, default=str),
                            "EventBusName": EVENT_BUS,
                        }
                    ]
                )
                if resp.get("FailedEntryCount", 0) > 0:
                    raise RuntimeError(f"put_events failed entries: {resp.get('Entries')}")
            except Exception:
                _release_dedup(event_id)
                raise
        except Exception as e:
            print(f"failed record seq={seq} err={e}")
            if seq:
                failures.append({"itemIdentifier": seq})
    return {"batchItemFailures": failures}

"""build the outbox infra on localstack from zero. idempotent: re-running upserts."""
import io
import json
import os
import subprocess
import sys
import time
import zipfile
from pathlib import Path

import boto3
from botocore.exceptions import ClientError

ENDPOINT = os.environ["AWS_ENDPOINT_URL"]
REGION = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")
ACCOUNT = "000000000000"

ROOT = Path(__file__).resolve().parent.parent
LAMBDA_SRC = ROOT / "lambdas"

# names , exactly as the verifier expects them
KMS_ALIAS = "alias/outbox-cmk"
ORDERS_TABLE = "orders"
OUTBOX_TABLE = "outbox"
DEDUP_TABLE = "processed_events"
ORDER_API_FN = "order-api"
ORDER_API_ROLE = "order-api-role"
RELAY_FN = "outbox-relay"
RELAY_ROLE = "outbox-relay-role"
EVENT_BUS = "order-events"
RULE_CREATED = "on-order-created"
RULE_CANCELLED = "on-order-cancelled"
Q_CREATED = "orders-created"
Q_CANCELLED = "orders-cancelled"
DLQ_CREATED = "orders-created-dlq"
DLQ_CANCELLED = "orders-cancelled-dlq"

EVENT_SOURCE = "outbox.relay"

kms = boto3.client("kms", endpoint_url=ENDPOINT)
ddb = boto3.client("dynamodb", endpoint_url=ENDPOINT)
sqs = boto3.client("sqs", endpoint_url=ENDPOINT)
iam = boto3.client("iam", endpoint_url=ENDPOINT)
lam = boto3.client("lambda", endpoint_url=ENDPOINT)
events = boto3.client("events", endpoint_url=ENDPOINT)


def log(msg):
    print(f"[build] {msg}", flush=True)


# ---------- KMS ----------
def ensure_cmk():
    # idempotent: look up by alias first
    try:
        a = kms.describe_key(KeyId=KMS_ALIAS)
        key_id = a["KeyMetadata"]["KeyId"]
        key_arn = a["KeyMetadata"]["Arn"]
        log(f"kms cmk already exists: {key_arn}")
        return key_id, key_arn
    except ClientError as e:
        if e.response["Error"]["Code"] not in ("NotFoundException", "NoSuchEntity"):
            raise

    policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "EnableIAM",
                "Effect": "Allow",
                "Principal": {"AWS": f"arn:aws:iam::{ACCOUNT}:root"},
                "Action": "kms:*",
                "Resource": "*",
            },
            {
                "Sid": "AllowEventBridgeService",
                "Effect": "Allow",
                "Principal": {"Service": "events.amazonaws.com"},
                "Action": ["kms:GenerateDataKey", "kms:Decrypt", "kms:DescribeKey"],
                "Resource": "*",
            },
        ],
    }
    out = kms.create_key(
        Description="outbox cmk for orders pipeline",
        KeyUsage="ENCRYPT_DECRYPT",
        KeySpec="SYMMETRIC_DEFAULT",
        Policy=json.dumps(policy),
    )
    key_id = out["KeyMetadata"]["KeyId"]
    key_arn = out["KeyMetadata"]["Arn"]
    kms.create_alias(AliasName=KMS_ALIAS, TargetKeyId=key_id)
    log(f"created cmk {key_arn} aliased {KMS_ALIAS}")
    return key_id, key_arn


# ---------- DynamoDB ----------
def ensure_table(name, key, with_stream=False, with_ttl=False, kms_arn=None):
    try:
        d = ddb.describe_table(TableName=name)["Table"]
        log(f"table {name} already exists")
        return d
    except ClientError as e:
        if e.response["Error"]["Code"] != "ResourceNotFoundException":
            raise

    args = {
        "TableName": name,
        "AttributeDefinitions": [{"AttributeName": key, "AttributeType": "S"}],
        "KeySchema": [{"AttributeName": key, "KeyType": "HASH"}],
        "BillingMode": "PAY_PER_REQUEST",
        "SSESpecification": {
            "Enabled": True,
            "SSEType": "KMS",
            "KMSMasterKeyId": kms_arn,
        },
    }
    if with_stream:
        args["StreamSpecification"] = {
            "StreamEnabled": True,
            "StreamViewType": "NEW_AND_OLD_IMAGES",
        }
    ddb.create_table(**args)
    waiter = ddb.get_waiter("table_exists")
    waiter.wait(TableName=name)
    log(f"created table {name}")

    if with_ttl:
        ddb.update_time_to_live(
            TableName=name,
            TimeToLiveSpecification={"Enabled": True, "AttributeName": "expires_at"},
        )
        log(f"enabled ttl on {name} (expires_at)")
    return ddb.describe_table(TableName=name)["Table"]


# ---------- SQS ----------
def ensure_queue(name, kms_arn, redrive_dlq_arn=None):
    try:
        url = sqs.get_queue_url(QueueName=name)["QueueUrl"]
        log(f"queue {name} already exists")
        return url
    except ClientError as e:
        if "NonExistentQueue" not in str(e) and e.response["Error"]["Code"] != "AWS.SimpleQueueService.NonExistentQueue":
            # fallthrough , check the next message too
            pass

    attrs = {
        "KmsMasterKeyId": kms_arn,
        "KmsDataKeyReusePeriodSeconds": "300",
    }
    if redrive_dlq_arn is not None:
        attrs["RedrivePolicy"] = json.dumps(
            {"deadLetterTargetArn": redrive_dlq_arn, "maxReceiveCount": 5}
        )
    url = sqs.create_queue(QueueName=name, Attributes=attrs)["QueueUrl"]
    log(f"created queue {name}")
    return url


def queue_arn(url):
    return sqs.get_queue_attributes(QueueUrl=url, AttributeNames=["QueueArn"])["Attributes"]["QueueArn"]


def set_queue_policy_for_eventbridge(queue_url, q_arn, rule_arn):
    policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "AllowEventBridgeRule",
                "Effect": "Allow",
                "Principal": {"Service": "events.amazonaws.com"},
                "Action": "sqs:SendMessage",
                "Resource": q_arn,
                "Condition": {"ArnEquals": {"aws:SourceArn": rule_arn}},
            }
        ],
    }
    sqs.set_queue_attributes(QueueUrl=queue_url, Attributes={"Policy": json.dumps(policy)})


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


def ensure_role(name, trust):
    try:
        r = iam.get_role(RoleName=name)["Role"]
        log(f"role {name} already exists")
        return r["Arn"]
    except ClientError as e:
        if e.response["Error"]["Code"] != "NoSuchEntity":
            raise
    r = iam.create_role(RoleName=name, AssumeRolePolicyDocument=json.dumps(trust))["Role"]
    log(f"created role {name}")
    return r["Arn"]


def put_inline_policy(role, name, policy):
    iam.put_role_policy(RoleName=role, PolicyName=name, PolicyDocument=json.dumps(policy))
    log(f"attached inline policy {name} on {role}")


# ---------- Lambda ----------
def zip_dir(src_dir):
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as z:
        for p in sorted(Path(src_dir).rglob("*")):
            if p.is_file():
                z.write(p, p.relative_to(src_dir))
    return buf.getvalue()


def ensure_lambda(name, role_arn, src_dir, env, kms_arn, retries=20):
    code = zip_dir(src_dir)
    try:
        lam.get_function(FunctionName=name)
        log(f"lambda {name} exists , updating code+config")
        lam.update_function_code(FunctionName=name, ZipFile=code)
        # wait for active before config update (localstack quirk)
        for _ in range(retries):
            cfg = lam.get_function_configuration(FunctionName=name)
            if cfg.get("LastUpdateStatus") in (None, "Successful"):
                break
            time.sleep(0.5)
        lam.update_function_configuration(
            FunctionName=name,
            Role=role_arn,
            Environment={"Variables": env},
            KMSKeyArn=kms_arn,
            Timeout=30,
            Runtime="python3.11",
            Handler="handler.handler",
        )
        return lam.get_function(FunctionName=name)["Configuration"]["FunctionArn"]
    except ClientError as e:
        if e.response["Error"]["Code"] != "ResourceNotFoundException":
            raise

    # iam-role-not-yet-propagated race on localstack , retry briefly
    last_err = None
    for _ in range(retries):
        try:
            r = lam.create_function(
                FunctionName=name,
                Runtime="python3.11",
                Role=role_arn,
                Handler="handler.handler",
                Code={"ZipFile": code},
                Environment={"Variables": env},
                KMSKeyArn=kms_arn,
                Timeout=30,
                Publish=True,
            )
            log(f"created lambda {name}")
            return r["FunctionArn"]
        except ClientError as ce:
            last_err = ce
            time.sleep(0.5)
    raise last_err


def wait_lambda_active(name):
    for _ in range(40):
        c = lam.get_function_configuration(FunctionName=name)
        if c.get("State") == "Active" and c.get("LastUpdateStatus") in (None, "Successful"):
            return
        time.sleep(0.5)


# ---------- main ----------
def main():
    key_id, key_arn = ensure_cmk()

    orders_d = ensure_table(ORDERS_TABLE, "order_id", kms_arn=key_arn)
    outbox_d = ensure_table(OUTBOX_TABLE, "event_id", with_stream=True, kms_arn=key_arn)
    dedup_d = ensure_table(DEDUP_TABLE, "event_id", with_ttl=True, kms_arn=key_arn)

    orders_arn = orders_d["TableArn"]
    outbox_arn = outbox_d["TableArn"]
    dedup_arn = dedup_d["TableArn"]
    outbox_stream_arn = ddb.describe_table(TableName=OUTBOX_TABLE)["Table"]["LatestStreamArn"]

    # event bus
    try:
        b = events.describe_event_bus(Name=EVENT_BUS)
        bus_arn = b["Arn"]
        log(f"event bus {EVENT_BUS} exists")
    except ClientError:
        b = events.create_event_bus(Name=EVENT_BUS, KmsKeyIdentifier=key_arn)
        bus_arn = b["EventBusArn"]
        log(f"created event bus {EVENT_BUS}")

    # sqs queues , DLQs first so we can wire them as redrive targets
    dlq_created_url = ensure_queue(DLQ_CREATED, key_arn)
    dlq_cancelled_url = ensure_queue(DLQ_CANCELLED, key_arn)
    dlq_created_arn = queue_arn(dlq_created_url)
    dlq_cancelled_arn = queue_arn(dlq_cancelled_url)

    q_created_url = ensure_queue(Q_CREATED, key_arn, redrive_dlq_arn=dlq_created_arn)
    q_cancelled_url = ensure_queue(Q_CANCELLED, key_arn, redrive_dlq_arn=dlq_cancelled_arn)
    q_created_arn = queue_arn(q_created_url)
    q_cancelled_arn = queue_arn(q_cancelled_url)

    # rule arns we need before targets/policies
    rule_created_arn = f"arn:aws:events:{REGION}:{ACCOUNT}:rule/{EVENT_BUS}/{RULE_CREATED}"
    rule_cancelled_arn = f"arn:aws:events:{REGION}:{ACCOUNT}:rule/{EVENT_BUS}/{RULE_CANCELLED}"

    set_queue_policy_for_eventbridge(q_created_url, q_created_arn, rule_created_arn)
    set_queue_policy_for_eventbridge(q_cancelled_url, q_cancelled_arn, rule_cancelled_arn)

    # IAM roles , least privilege, scoped to specific ARNs
    api_role_arn = ensure_role(ORDER_API_ROLE, TRUST_LAMBDA)
    relay_role_arn = ensure_role(RELAY_ROLE, TRUST_LAMBDA)

    api_log_arn = f"arn:aws:logs:{REGION}:{ACCOUNT}:log-group:/aws/lambda/{ORDER_API_FN}:*"
    relay_log_arn = f"arn:aws:logs:{REGION}:{ACCOUNT}:log-group:/aws/lambda/{RELAY_FN}:*"

    api_policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "WriteOrdersAndOutboxAtomically",
                "Effect": "Allow",
                "Action": [
                    "dynamodb:PutItem",
                    "dynamodb:UpdateItem",
                    "dynamodb:GetItem",
                    "dynamodb:TransactWriteItems",
                    "dynamodb:DescribeTable",
                ],
                "Resource": [orders_arn, outbox_arn],
            },
            {
                "Sid": "UseCmkForOrdersAndOutbox",
                "Effect": "Allow",
                "Action": ["kms:Encrypt", "kms:Decrypt", "kms:GenerateDataKey", "kms:DescribeKey"],
                "Resource": key_arn,
                "Condition": {
                    "StringEquals": {
                        "kms:ViaService": [f"dynamodb.{REGION}.amazonaws.com"]
                    }
                },
            },
            {
                "Sid": "Logs",
                "Effect": "Allow",
                "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"],
                "Resource": [api_log_arn, api_log_arn[: -len(":*")]],
            },
        ],
    }
    put_inline_policy(ORDER_API_ROLE, "order-api-inline", api_policy)

    relay_policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "ReadOutboxStreamOnly",
                "Effect": "Allow",
                "Action": [
                    "dynamodb:DescribeStream",
                    "dynamodb:GetRecords",
                    "dynamodb:GetShardIterator",
                    "dynamodb:ListStreams",
                ],
                "Resource": outbox_stream_arn,
            },
            {
                "Sid": "DedupTableWritesOnly",
                "Effect": "Allow",
                "Action": [
                    "dynamodb:PutItem",
                    "dynamodb:DeleteItem",
                    "dynamodb:GetItem",
                    "dynamodb:DescribeTable",
                ],
                "Resource": dedup_arn,
            },
            {
                "Sid": "PublishOnOurBusOnly",
                "Effect": "Allow",
                "Action": "events:PutEvents",
                "Resource": bus_arn,
            },
            {
                "Sid": "UseCmkForDedupAndBus",
                "Effect": "Allow",
                "Action": ["kms:Encrypt", "kms:Decrypt", "kms:GenerateDataKey", "kms:DescribeKey"],
                "Resource": key_arn,
            },
            {
                "Sid": "Logs",
                "Effect": "Allow",
                "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"],
                "Resource": [relay_log_arn, relay_log_arn[: -len(":*")]],
            },
        ],
    }
    put_inline_policy(RELAY_ROLE, "outbox-relay-inline", relay_policy)

    # lambdas
    api_fn_arn = ensure_lambda(
        ORDER_API_FN,
        api_role_arn,
        LAMBDA_SRC / "order_api",
        env={
            "ORDERS_TABLE": ORDERS_TABLE,
            "OUTBOX_TABLE": OUTBOX_TABLE,
            "AWS_ENDPOINT_URL": ENDPOINT,
        },
        kms_arn=key_arn,
    )
    relay_fn_arn = ensure_lambda(
        RELAY_FN,
        relay_role_arn,
        LAMBDA_SRC / "outbox_relay",
        env={
            "DEDUP_TABLE": DEDUP_TABLE,
            "EVENT_BUS": EVENT_BUS,
            "EVENT_SOURCE": EVENT_SOURCE,
            "DEDUP_TTL_SECONDS": "604800",
            "AWS_ENDPOINT_URL": ENDPOINT,
        },
        kms_arn=key_arn,
    )
    wait_lambda_active(ORDER_API_FN)
    wait_lambda_active(RELAY_FN)

    # event source mapping: outbox stream -> outbox-relay with batch item failures
    existing = lam.list_event_source_mappings(FunctionName=RELAY_FN).get("EventSourceMappings", [])
    has_mapping = any(m.get("EventSourceArn") == outbox_stream_arn for m in existing)
    if not has_mapping:
        for _ in range(20):
            try:
                lam.create_event_source_mapping(
                    EventSourceArn=outbox_stream_arn,
                    FunctionName=RELAY_FN,
                    StartingPosition="LATEST",
                    BatchSize=10,
                    MaximumBatchingWindowInSeconds=1,
                    FunctionResponseTypes=["ReportBatchItemFailures"],
                )
                break
            except ClientError as e:
                if "InvalidParameterValueException" in str(e):
                    time.sleep(0.5)
                    continue
                raise
        log(f"created event source mapping {outbox_stream_arn} -> {RELAY_FN}")
    else:
        log("event source mapping already exists")

    # eventbridge rules , match by detail-type only; the outbox.relay source carries them
    events.put_rule(
        Name=RULE_CREATED,
        EventBusName=EVENT_BUS,
        EventPattern=json.dumps(
            {"source": [EVENT_SOURCE], "detail-type": ["OrderCreated"]}
        ),
        State="ENABLED",
    )
    events.put_rule(
        Name=RULE_CANCELLED,
        EventBusName=EVENT_BUS,
        EventPattern=json.dumps(
            {"source": [EVENT_SOURCE], "detail-type": ["OrderCancelled"]}
        ),
        State="ENABLED",
    )
    log("put rules on bus")

    events.put_targets(
        Rule=RULE_CREATED,
        EventBusName=EVENT_BUS,
        Targets=[{"Id": "to-orders-created", "Arn": q_created_arn}],
    )
    events.put_targets(
        Rule=RULE_CANCELLED,
        EventBusName=EVENT_BUS,
        Targets=[{"Id": "to-orders-cancelled", "Arn": q_cancelled_arn}],
    )
    log("targets attached")

    print(json.dumps({
        "kms_key_arn": key_arn,
        "orders_arn": orders_arn,
        "outbox_arn": outbox_arn,
        "outbox_stream_arn": outbox_stream_arn,
        "dedup_arn": dedup_arn,
        "bus_arn": bus_arn,
        "q_created_arn": q_created_arn,
        "q_cancelled_arn": q_cancelled_arn,
        "dlq_created_arn": dlq_created_arn,
        "dlq_cancelled_arn": dlq_cancelled_arn,
        "api_fn_arn": api_fn_arn,
        "relay_fn_arn": relay_fn_arn,
    }, indent=2))


if __name__ == "__main__":
    main()

"""end-to-end probe , drives the same scenarios the verifier will run."""
import json
import os
import time

import boto3

ENDPOINT = os.environ["AWS_ENDPOINT_URL"]
lam = boto3.client("lambda", endpoint_url=ENDPOINT)
sqs = boto3.client("sqs", endpoint_url=ENDPOINT)
ddb = boto3.client("dynamodb", endpoint_url=ENDPOINT)


def invoke(payload):
    r = lam.invoke(FunctionName="order-api", Payload=json.dumps(payload).encode())
    body = json.loads(r["Payload"].read())
    print(f"invoke {payload['kind']} {payload['order_id']} -> {body}")
    return body


def queue_url(name):
    return sqs.get_queue_url(QueueName=name)["QueueUrl"]


def queue_depth(name):
    a = sqs.get_queue_attributes(
        QueueUrl=queue_url(name),
        AttributeNames=["ApproximateNumberOfMessages", "ApproximateNumberOfMessagesNotVisible"],
    )["Attributes"]
    return int(a["ApproximateNumberOfMessages"]) + int(a["ApproximateNumberOfMessagesNotVisible"])


def wait_for(predicate, timeout=30, interval=1):
    deadline = time.time() + timeout
    while time.time() < deadline:
        if predicate():
            return True
        time.sleep(interval)
    return False


def dedup_has(event_id):
    r = ddb.get_item(TableName="processed_events", Key={"event_id": {"S": event_id}})
    return "Item" in r


def main():
    res1 = invoke({"order_id": "o-1", "kind": "OrderCreated", "total": 4200, "currency": "USD"})
    eid1 = res1["event_id"]

    ok = wait_for(lambda: queue_depth("orders-created") >= 1 and dedup_has(eid1), timeout=30)
    print(f"after OrderCreated: created={queue_depth('orders-created')} cancelled={queue_depth('orders-cancelled')} dedup_for_eid1={dedup_has(eid1)} pass={ok}")
    assert ok, "OrderCreated did not propagate within 30s"
    assert queue_depth("orders-cancelled") == 0, "OrderCreated leaked into cancelled queue"

    res2 = invoke({"order_id": "o-2", "kind": "OrderCancelled", "reason": "user-cancel"})
    eid2 = res2["event_id"]

    ok = wait_for(lambda: queue_depth("orders-cancelled") >= 1 and dedup_has(eid2), timeout=30)
    print(f"after OrderCancelled: created={queue_depth('orders-created')} cancelled={queue_depth('orders-cancelled')} dedup_for_eid2={dedup_has(eid2)} pass={ok}")
    assert ok, "OrderCancelled did not propagate within 30s"
    assert queue_depth("orders-created") == 1, "OrderCancelled leaked into created queue"

    print("\nALL CHECKS PASSED")


if __name__ == "__main__":
    main()