Add Step Functions state machine for async processing; various fixes in setup

.devcontainer/Dockerfile.github

@@ -17,6 +17,7 @@ ENV EXTRA_CORS_ALLOWED_ORIGINS='*'
 ENV DISABLE_CUSTOM_CORS_APIGATEWAY=1
 ENV LOCALSTACK_APPINSPECTOR_ENABLE=1
 ENV LOCALSTACK_APPINSPECTOR_DEV_ENABLE=1
+ENV LOCALSTACK_DEBUG=1
 
 # Workshop token URL — set by organizer before each event.
 # The setup script fetches the actual token from this endpoint.

01-serverless-app/lambdas/order_handler/handler.py

@@ -1,13 +1,22 @@
 import json
 import os
 import uuid
+from datetime import datetime, timezone
+from decimal import Decimal
 import boto3
 
-dynamodb = boto3.resource("dynamodb", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
-sqs = boto3.client("sqs", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
+dynamodb = boto3.resource("dynamodb")
+sqs = boto3.client("sqs")
+
+TABLE_NAME     = os.environ["ORDERS_TABLE"]
+PRODUCTS_TABLE = os.environ["PRODUCTS_TABLE"]
+QUEUE_URL      = os.environ["ORDERS_QUEUE_URL"]
+DLQ_URL        = os.environ.get("ORDERS_DLQ_URL", "")
+
+class DecimalEncoder(json.JSONEncoder):
+    def default(self, o):
+        return int(o) if isinstance(o, Decimal) else super().default(o)
 
-TABLE_NAME = os.environ["ORDERS_TABLE"]
-QUEUE_URL = os.environ["ORDERS_QUEUE_URL"]
 
 CORS_HEADERS = {
     "Access-Control-Allow-Origin": "*",
@@ -18,10 +27,17 @@
 
 def handler(event, context):
     method = event.get("httpMethod", "")
+    path   = event.get("path", "")
 
     if method == "OPTIONS":
         return {"statusCode": 200, "headers": CORS_HEADERS, "body": ""}
 
+    if method == "POST" and path.endswith("/replay"):
+        return replay_dlq()
+
+    if method == "GET" and "/products" in path:
+        return list_products()
+
     if method == "GET":
         return list_orders()
 
@@ -31,26 +47,50 @@ def handler(event, context):
     return {"statusCode": 405, "headers": CORS_HEADERS, "body": "Method Not Allowed"}
 
 
+def replay_dlq():
+    resp = sqs.receive_message(QueueUrl=DLQ_URL, MaxNumberOfMessages=10)
+    messages = resp.get("Messages", [])
+    for msg in messages:
+        sqs.send_message(QueueUrl=QUEUE_URL, MessageBody=msg["Body"])
+        sqs.delete_message(QueueUrl=DLQ_URL, ReceiptHandle=msg["ReceiptHandle"])
+    return {
+        "statusCode": 200,
+        "headers": {**CORS_HEADERS, "Content-Type": "application/json"},
+        "body": json.dumps({"replayed": len(messages)}),
+    }
+
+
+def list_products():
+    table = dynamodb.Table(PRODUCTS_TABLE)
+    items = sorted(table.scan().get("Items", []), key=lambda x: x.get("name", ""))
+    return {
+        "statusCode": 200,
+        "headers": {**CORS_HEADERS, "Content-Type": "application/json"},
+        "body": json.dumps(items, cls=DecimalEncoder),
+    }
+
+
 def list_orders():
     table = dynamodb.Table(TABLE_NAME)
     result = table.scan()
     items = sorted(result.get("Items", []), key=lambda x: x.get("order_id", ""))
     return {
         "statusCode": 200,
         "headers": {**CORS_HEADERS, "Content-Type": "application/json"},
-        "body": json.dumps(items),
+        "body": json.dumps(items, cls=DecimalEncoder),
     }
 
 
 def create_order(event):
     body = json.loads(event.get("body") or "{}")
-    order_id = str(uuid.uuid4())
+    order_id = uuid.uuid4().hex[:12]
 
     order = {
         "order_id": order_id,
         "item": body.get("item", "unknown"),
         "quantity": int(body.get("quantity", 1)),
         "status": "pending",
+        "created_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
     }
 
     table = dynamodb.Table(TABLE_NAME)

01-serverless-app/lambdas/order_processor/handler.py

@@ -1,38 +1,101 @@
 import json
 import os
+import time
+import uuid
+from datetime import datetime, timezone
 import boto3
 
-dynamodb = boto3.resource("dynamodb", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
-s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
+dynamodb = boto3.resource("dynamodb")
+s3 = boto3.client("s3")
+sfn = boto3.client("stepfunctions")
 
 TABLE_NAME = os.environ["ORDERS_TABLE"]
 RECEIPTS_BUCKET = os.environ["RECEIPTS_BUCKET"]
+STATE_MACHINE_ARN = os.environ["STATE_MACHINE_ARN"]
+
+TERMINAL_STATUSES = {"fulfilled", "failed"}
+
+
+def now():
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def set_status(order_id, status):
+    ts_key = {
+        "validating":          "validating_at",
+        "payment_processing":  "payment_at",
+        "fulfilled":           "fulfilled_at",
+        "failed":              "failed_at",
+    }.get(status)
+
+    expression = "SET #s = :s"
+    names  = {"#s": "status"}
+    values = {":s": status}
+
+    if ts_key:
+        expression += ", #ts = :ts"
+        names["#ts"]  = ts_key
+        values[":ts"] = now()
+
+    dynamodb.Table(TABLE_NAME).update_item(
+        Key={"order_id": order_id},
+        UpdateExpression=expression,
+        ExpressionAttributeNames=names,
+        ExpressionAttributeValues=values,
+    )
 
 
 def handler(event, context):
-    for record in event["Records"]:
-        order = json.loads(record["body"])
-        order_id = order["order_id"]
-
-        # Update order status in DynamoDB
-        table = dynamodb.Table(TABLE_NAME)
-        table.update_item(
-            Key={"order_id": order_id},
-            UpdateExpression="SET #s = :s",
-            ExpressionAttributeNames={"#s": "status"},
-            ExpressionAttributeValues={":s": "processed"},
-        )
-
-        # Store receipt in S3
-        receipt = {
-            "order_id": order_id,
-            "item": order.get("item"),
-            "quantity": order.get("quantity"),
-            "status": "processed",
-        }
-        s3.put_object(
-            Bucket=RECEIPTS_BUCKET,
-            Key=f"receipts/{order_id}.json",
-            Body=json.dumps(receipt),
-            ContentType="application/json",
-        )
+    # Triggered by SQS: start a state machine execution per order
+    if "Records" in event:
+        for record in event["Records"]:
+            order = json.loads(record["body"])
+            set_status(order["order_id"], "validating")  # fails fast if DDB is faulted → SQS retry → DLQ
+            sfn.start_execution(
+                stateMachineArn=STATE_MACHINE_ARN,
+                name=f"order-{order['order_id']}-{uuid.uuid4().hex[:8]}",
+                input=json.dumps({"order": order}),
+            )
+        return
+
+    # Invoked by Step Functions
+    step  = event["step"]
+    order = event["order"]
+
+    if step == "validate":        return validate(order)
+    if step == "process_payment": return process_payment(order)
+    if step == "fulfill":         return fulfill(order)
+    if step == "handle_failure":  return handle_failure(order)
+
+    raise ValueError(f"Unknown step: {step}")
+
+
+def validate(order):
+    time.sleep(2)
+    set_status(order["order_id"], "validating")
+    return order
+
+
+def process_payment(order):
+    time.sleep(3)
+    set_status(order["order_id"], "payment_processing")
+    return order
+
+
+def fulfill(order):
+    time.sleep(2)
+    set_status(order["order_id"], "fulfilled")
+    receipt = {k: order[k] for k in ("order_id", "item", "quantity")}
+    receipt["status"] = "fulfilled"
+    s3.put_object(
+        Bucket=RECEIPTS_BUCKET,
+        Key=f"receipts/{order['order_id']}.json",
+        Body=json.dumps(receipt),
+        ContentType="application/json",
+    )
+    return order
+
+
+def handle_failure(order):
+    set_status(order["order_id"], "failed")
+    return order