sfn-saga-compensation-orchestrator

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

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

import os, json, time, boto3
from botocore.exceptions import ClientError

ddb = boto3.resource("dynamodb")
inv = ddb.Table("inventory")
idem = ddb.Table("saga-idempotency")


def _idem_key(event):
    return f"{event['idempotency_key']}:reserve"


def handler(event, _ctx):
    key = _idem_key(event)
    try:
        idem.put_item(
            Item={"idempotency_key": key, "ttl": int(time.time()) + 86400, "status": "DONE"},
            ConditionExpression="attribute_not_exists(idempotency_key)",
        )
    except ClientError as e:
        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
            return {**event, "reserve": {"already_done": True}}
        raise

    inv.update_item(
        Key={"sku": event["sku"]},
        UpdateExpression="SET reserved_qty = if_not_exists(reserved_qty, :z) + :q",
        ExpressionAttributeValues={":q": event["qty"], ":z": 0},
    )
    return {**event, "reserve": {"sku": event["sku"], "qty": event["qty"]}}

import os, json, time, boto3
from botocore.exceptions import ClientError

ddb = boto3.resource("dynamodb")
payments = ddb.Table("payments")
idem = ddb.Table("saga-idempotency")
sm = boto3.client("secretsmanager")


def _idem_key(event):
    return f"{event['idempotency_key']}:charge"


def handler(event, _ctx):
    resp = sm.get_secret_value(SecretId="payment/gateway-apikey")
    api_key = resp["SecretString"]
    print(f"charge-payment: loaded gateway api key len={len(api_key)}")

    if event.get("force_failure"):
        raise RuntimeError("forced payment failure")

    key = _idem_key(event)
    try:
        idem.put_item(
            Item={"idempotency_key": key, "ttl": int(time.time()) + 86400, "status": "DONE"},
            ConditionExpression="attribute_not_exists(idempotency_key)",
        )
    except ClientError as e:
        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
            return {**event, "charge": {"already_done": True, "payment_id": f"pay-{event['order_id']}"}}
        raise

    payment_id = f"pay-{event['order_id']}"
    payments.put_item(
        Item={
            "payment_id": payment_id,
            "order_id": event["order_id"],
            "amount": event["amount"],
            "status": "CAPTURED",
        }
    )
    return {**event, "charge": {"payment_id": payment_id, "status": "CAPTURED"}}

import os, json, time, boto3
from botocore.exceptions import ClientError

ddb = boto3.resource("dynamodb")
shipments = ddb.Table("shipments")
idem = ddb.Table("saga-idempotency")


def _idem_key(event):
    return f"{event['idempotency_key']}:ship"


def handler(event, _ctx):
    key = _idem_key(event)
    try:
        idem.put_item(
            Item={"idempotency_key": key, "ttl": int(time.time()) + 86400, "status": "DONE"},
            ConditionExpression="attribute_not_exists(idempotency_key)",
        )
    except ClientError as e:
        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
            return {**event, "ship": {"already_done": True, "shipment_id": f"ship-{event['order_id']}"}}
        raise

    shipment_id = f"ship-{event['order_id']}"
    shipments.put_item(
        Item={
            "shipment_id": shipment_id,
            "order_id": event["order_id"],
            "sku": event["sku"],
            "qty": event["qty"],
            "status": "CREATED",
        }
    )
    return {**event, "ship": {"shipment_id": shipment_id, "status": "CREATED"}}

import os, json, time, boto3
from botocore.exceptions import ClientError

ddb = boto3.resource("dynamodb")
inv = ddb.Table("inventory")
idem = ddb.Table("saga-idempotency")


def _idem_key(event):
    return f"{event['idempotency_key']}:release"


def handler(event, _ctx):
    key = _idem_key(event)
    try:
        idem.put_item(
            Item={"idempotency_key": key, "ttl": int(time.time()) + 86400, "status": "DONE"},
            ConditionExpression="attribute_not_exists(idempotency_key)",
        )
    except ClientError as e:
        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
            return {**event, "release": {"already_done": True}}
        raise

    if not event.get("reserve") or event.get("reserve", {}).get("already_done"):
        pass

    inv.update_item(
        Key={"sku": event["sku"]},
        UpdateExpression="SET reserved_qty = if_not_exists(reserved_qty, :z) - :q",
        ExpressionAttributeValues={":q": event["qty"], ":z": 0},
    )
    return {**event, "release": {"sku": event["sku"], "qty": event["qty"]}}

import os, json, time, boto3
from botocore.exceptions import ClientError

ddb = boto3.resource("dynamodb")
payments = ddb.Table("payments")
idem = ddb.Table("saga-idempotency")


def _idem_key(event):
    return f"{event['idempotency_key']}:refund"


def handler(event, _ctx):
    key = _idem_key(event)
    try:
        idem.put_item(
            Item={"idempotency_key": key, "ttl": int(time.time()) + 86400, "status": "DONE"},
            ConditionExpression="attribute_not_exists(idempotency_key)",
        )
    except ClientError as e:
        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
            return {**event, "refund": {"already_done": True}}
        raise

    payment_id = f"pay-{event['order_id']}"
    try:
        payments.update_item(
            Key={"payment_id": payment_id},
            UpdateExpression="SET #s = :r",
            ConditionExpression="attribute_exists(payment_id)",
            ExpressionAttributeNames={"#s": "status"},
            ExpressionAttributeValues={":r": "REFUNDED"},
        )
        return {**event, "refund": {"payment_id": payment_id, "status": "REFUNDED"}}
    except ClientError as e:
        if e.response["Error"]["Code"] == "ConditionalCheckFailedException":
            return {**event, "refund": {"payment_id": payment_id, "noop": True}}
        raise

import json


def handler(event, _ctx):
    detail_type = event.get("detail-type", "Unknown")
    print(f"saga-terminal:{detail_type} detail={json.dumps(event.get('detail', {}))}")
    return {"ok": True}

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

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "Logs",
      "Effect": "Allow",
      "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"],
      "Resource": [
        "arn:aws:logs:us-east-1:000000000000:log-group:/aws/lambda/reserve-inventory",
        "arn:aws:logs:us-east-1:000000000000:log-group:/aws/lambda/reserve-inventory:*"
      ]
    },
    {
      "Sid": "InventoryTable",
      "Effect": "Allow",
      "Action": ["dynamodb:UpdateItem", "dynamodb:GetItem", "dynamodb:PutItem"],
      "Resource": "arn:aws:dynamodb:us-east-1:000000000000:table/inventory"
    },
    {
      "Sid": "IdempotencyTable",
      "Effect": "Allow",
      "Action": ["dynamodb:PutItem", "dynamodb:GetItem"],
      "Resource": "arn:aws:dynamodb:us-east-1:000000000000:table/saga-idempotency"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "Logs",
      "Effect": "Allow",
      "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"],
      "Resource": [
        "arn:aws:logs:us-east-1:000000000000:log-group:/aws/lambda/charge-payment",
        "arn:aws:logs:us-east-1:000000000000:log-group:/aws/lambda/charge-payment:*"
      ]
    },
    {
      "Sid": "PaymentsTable",
      "Effect": "Allow",
      "Action": ["dynamodb:PutItem", "dynamodb:GetItem", "dynamodb:UpdateItem"],
      "Resource": "arn:aws:dynamodb:us-east-1:000000000000:table/payments"
    },
    {
      "Sid": "IdempotencyTable",
      "Effect": "Allow",
      "Action": ["dynamodb:PutItem", "dynamodb:GetItem"],
      "Resource": "arn:aws:dynamodb:us-east-1:000000000000:table/saga-idempotency"
    },
    {
      "Sid": "ReadGatewaySecret",
      "Effect": "Allow",
      "Action": ["secretsmanager:GetSecretValue", "secretsmanager:DescribeSecret"],
      "Resource": "{{SECRET_ARN}}"
    },
    {
      "Sid": "DecryptSecretWithCmk",
      "Effect": "Allow",
      "Action": ["kms:Decrypt"],
      "Resource": "{{KMS_ARN}}"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "Logs",
      "Effect": "Allow",
      "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"],
      "Resource": [
        "arn:aws:logs:us-east-1:000000000000:log-group:/aws/lambda/create-shipment",
        "arn:aws:logs:us-east-1:000000000000:log-group:/aws/lambda/create-shipment:*"
      ]
    },
    {
      "Sid": "ShipmentsTable",
      "Effect": "Allow",
      "Action": ["dynamodb:PutItem", "dynamodb:GetItem"],
      "Resource": "arn:aws:dynamodb:us-east-1:000000000000:table/shipments"
    },
    {
      "Sid": "IdempotencyTable",
      "Effect": "Allow",
      "Action": ["dynamodb:PutItem", "dynamodb:GetItem"],
      "Resource": "arn:aws:dynamodb:us-east-1:000000000000:table/saga-idempotency"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "Logs",
      "Effect": "Allow",
      "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"],
      "Resource": [
        "arn:aws:logs:us-east-1:000000000000:log-group:/aws/lambda/release-inventory",
        "arn:aws:logs:us-east-1:000000000000:log-group:/aws/lambda/release-inventory:*"
      ]
    },
    {
      "Sid": "InventoryTable",
      "Effect": "Allow",
      "Action": ["dynamodb:UpdateItem", "dynamodb:GetItem"],
      "Resource": "arn:aws:dynamodb:us-east-1:000000000000:table/inventory"
    },
    {
      "Sid": "IdempotencyTable",
      "Effect": "Allow",
      "Action": ["dynamodb:PutItem", "dynamodb:GetItem"],
      "Resource": "arn:aws:dynamodb:us-east-1:000000000000:table/saga-idempotency"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "Logs",
      "Effect": "Allow",
      "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"],
      "Resource": [
        "arn:aws:logs:us-east-1:000000000000:log-group:/aws/lambda/refund-payment",
        "arn:aws:logs:us-east-1:000000000000:log-group:/aws/lambda/refund-payment:*"
      ]
    },
    {
      "Sid": "PaymentsTable",
      "Effect": "Allow",
      "Action": ["dynamodb:UpdateItem", "dynamodb:GetItem"],
      "Resource": "arn:aws:dynamodb:us-east-1:000000000000:table/payments"
    },
    {
      "Sid": "IdempotencyTable",
      "Effect": "Allow",
      "Action": ["dynamodb:PutItem", "dynamodb:GetItem"],
      "Resource": "arn:aws:dynamodb:us-east-1:000000000000:table/saga-idempotency"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "Logs",
      "Effect": "Allow",
      "Action": ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"],
      "Resource": [
        "arn:aws:logs:us-east-1:000000000000:log-group:/aws/lambda/saga-notifier",
        "arn:aws:logs:us-east-1:000000000000:log-group:/aws/lambda/saga-notifier:*"
      ]
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "InvokeSagaLambdas",
      "Effect": "Allow",
      "Action": "lambda:InvokeFunction",
      "Resource": [
        "arn:aws:lambda:us-east-1:000000000000:function:reserve-inventory",
        "arn:aws:lambda:us-east-1:000000000000:function:charge-payment",
        "arn:aws:lambda:us-east-1:000000000000:function:create-shipment",
        "arn:aws:lambda:us-east-1:000000000000:function:release-inventory",
        "arn:aws:lambda:us-east-1:000000000000:function:refund-payment"
      ]
    },
    {
      "Sid": "PublishToSagaBus",
      "Effect": "Allow",
      "Action": "events:PutEvents",
      "Resource": "arn:aws:events:us-east-1:000000000000:event-bus/saga-events"
    }
  ]
}

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

{
  "Comment": "Order saga: reserve -> charge -> ship. On failure, compensate in reverse.",
  "StartAt": "ReserveInventory",
  "States": {
    "ReserveInventory": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:000000000000:function:reserve-inventory",
      "ResultPath": "$.reserve_result",
      "OutputPath": "$.reserve_result",
      "Retry": [
        {
          "ErrorEquals": ["Lambda.ServiceException", "Lambda.AWSLambdaException", "Lambda.SdkClientException"],
          "IntervalSeconds": 1,
          "MaxAttempts": 2,
          "BackoffRate": 2.0
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "ResultPath": "$.error",
          "Next": "PublishFailed"
        }
      ],
      "Next": "ChargePayment"
    },
    "ChargePayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:000000000000:function:charge-payment",
      "ResultPath": "$.charge_result",
      "OutputPath": "$.charge_result",
      "Retry": [
        {
          "ErrorEquals": ["Lambda.ServiceException", "Lambda.AWSLambdaException", "Lambda.SdkClientException"],
          "IntervalSeconds": 1,
          "MaxAttempts": 2,
          "BackoffRate": 2.0
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "ResultPath": "$.error",
          "Next": "ReleaseInventory"
        }
      ],
      "Next": "CreateShipment"
    },
    "CreateShipment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:000000000000:function:create-shipment",
      "ResultPath": "$.ship_result",
      "OutputPath": "$.ship_result",
      "Retry": [
        {
          "ErrorEquals": ["Lambda.ServiceException", "Lambda.AWSLambdaException", "Lambda.SdkClientException"],
          "IntervalSeconds": 1,
          "MaxAttempts": 2,
          "BackoffRate": 2.0
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "ResultPath": "$.error",
          "Next": "RefundPayment"
        }
      ],
      "Next": "PublishCompleted"
    },
    "RefundPayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:000000000000:function:refund-payment",
      "ResultPath": "$.refund_result",
      "OutputPath": "$.refund_result",
      "Next": "ReleaseInventory"
    },
    "ReleaseInventory": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:000000000000:function:release-inventory",
      "ResultPath": "$.release_result",
      "OutputPath": "$.release_result",
      "Next": "PublishFailed"
    },
    "PublishCompleted": {
      "Type": "Task",
      "Resource": "arn:aws:states:::events:putEvents",
      "Parameters": {
        "Entries": [
          {
            "EventBusName": "saga-events",
            "Source": "order.saga",
            "DetailType": "OrderCompleted",
            "Detail": {
              "order_id.$": "$.order_id",
              "outcome": "completed"
            }
          }
        ]
      },
      "End": true
    },
    "PublishFailed": {
      "Type": "Task",
      "Resource": "arn:aws:states:::events:putEvents",
      "Parameters": {
        "Entries": [
          {
            "EventBusName": "saga-events",
            "Source": "order.saga",
            "DetailType": "OrderFailed",
            "Detail": {
              "order_id.$": "$.order_id",
              "outcome": "failed-compensated"
            }
          }
        ]
      },
      "End": true
    }
  }
}

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

AWSX="aws --endpoint-url=${AWS_ENDPOINT_URL}"
REGION="us-east-1"
ACCOUNT="000000000000"
ARN_PREFIX_LAMBDA="arn:aws:lambda:${REGION}:${ACCOUNT}:function"
BUS_NAME="saga-events"
BUS_ARN="arn:aws:events:${REGION}:${ACCOUNT}:event-bus/${BUS_NAME}"

cd /app
mkdir -p build/policies build/zips

echo "== KMS key + alias =="
if ! $AWSX kms describe-key --key-id alias/saga-cmk >/dev/null 2>&1; then
  KMS_KEY_ID=$($AWSX kms create-key --description "saga customer managed key" --query 'KeyMetadata.KeyId' --output text)
  $AWSX kms create-alias --alias-name alias/saga-cmk --target-key-id "$KMS_KEY_ID"
else
  KMS_KEY_ID=$($AWSX kms describe-key --key-id alias/saga-cmk --query 'KeyMetadata.KeyId' --output text)
fi
KMS_ARN="arn:aws:kms:${REGION}:${ACCOUNT}:key/${KMS_KEY_ID}"
echo "kms key: $KMS_ARN"

echo "== Secret =="
if ! $AWSX secretsmanager describe-secret --secret-id payment/gateway-apikey >/dev/null 2>&1; then
  $AWSX secretsmanager create-secret \
    --name payment/gateway-apikey \
    --kms-key-id "$KMS_ARN" \
    --secret-string "sk_test_FAKEGATEWAYKEY_ABC123" >/dev/null
fi
SECRET_ARN=$($AWSX secretsmanager describe-secret --secret-id payment/gateway-apikey --query 'ARN' --output text)
echo "secret: $SECRET_ARN"

echo "== DynamoDB tables =="
create_table() {
  local name="$1" pk="$2"
  if ! $AWSX dynamodb describe-table --table-name "$name" >/dev/null 2>&1; then
    $AWSX dynamodb create-table \
      --table-name "$name" \
      --attribute-definitions AttributeName=${pk},AttributeType=S \
      --key-schema AttributeName=${pk},KeyType=HASH \
      --billing-mode PAY_PER_REQUEST >/dev/null
  fi
}
create_table inventory sku
create_table payments payment_id
create_table shipments shipment_id
create_table saga-idempotency idempotency_key
$AWSX dynamodb update-time-to-live \
  --table-name saga-idempotency \
  --time-to-live-specification "Enabled=true, AttributeName=ttl" >/dev/null 2>&1 || true

echo "== Event bus =="
$AWSX events create-event-bus --name "$BUS_NAME" >/dev/null 2>&1 || true

echo "== Substitute policy templates =="
for f in policies/*.json; do
  out="build/$(basename "$f" .json).json"
  # skip trust policies
  case "$(basename "$f")" in
    lambda-trust.json|states-trust.json)
      cp "$f" "build/$(basename "$f")"; continue;;
  esac
  sed -e "s|{{KMS_ARN}}|${KMS_ARN}|g" -e "s|{{SECRET_ARN}}|${SECRET_ARN}|g" "$f" > "build/policies/$(basename "$f")"
done
cp policies/lambda-trust.json build/lambda-trust.json
cp policies/states-trust.json build/states-trust.json

echo "== IAM roles =="
create_role() {
  local role="$1" trust="$2" policy_doc="$3" policy_name="${role}-policy"
  if ! $AWSX iam get-role --role-name "$role" >/dev/null 2>&1; then
    $AWSX iam create-role --role-name "$role" --assume-role-policy-document "file://$trust" >/dev/null
  fi
  $AWSX iam put-role-policy --role-name "$role" --policy-name "$policy_name" --policy-document "file://$policy_doc" >/dev/null
}
create_role reserve-inventory-role  build/lambda-trust.json build/policies/reserve-inventory-role.json
create_role charge-payment-role     build/lambda-trust.json build/policies/charge-payment-role.json
create_role create-shipment-role    build/lambda-trust.json build/policies/create-shipment-role.json
create_role release-inventory-role  build/lambda-trust.json build/policies/release-inventory-role.json
create_role refund-payment-role     build/lambda-trust.json build/policies/refund-payment-role.json
create_role saga-notifier-role      build/lambda-trust.json build/policies/saga-notifier-role.json
create_role order-saga-role         build/states-trust.json build/policies/order-saga-role.json

role_arn() { echo "arn:aws:iam::${ACCOUNT}:role/$1"; }

echo "== Package lambdas =="
zip_lambda() {
  local src="$1" zipname="$2"
  (cd lambdas && zip -q -j "/app/build/zips/${zipname}.zip" "${src}")
}
zip_lambda reserve_inventory.py reserve-inventory
zip_lambda charge_payment.py    charge-payment
zip_lambda create_shipment.py   create-shipment
zip_lambda release_inventory.py release-inventory
zip_lambda refund_payment.py    refund-payment
zip_lambda saga_notifier.py     saga-notifier

echo "== Create lambdas =="
upsert_lambda() {
  local name="$1" handler="$2" role="$3" zip="$4"
  if $AWSX lambda get-function --function-name "$name" >/dev/null 2>&1; then
    $AWSX lambda update-function-code --function-name "$name" --zip-file "fileb://$zip" >/dev/null
  else
    $AWSX lambda create-function \
      --function-name "$name" \
      --runtime python3.11 \
      --role "$(role_arn "$role")" \
      --handler "$handler" \
      --zip-file "fileb://$zip" \
      --timeout 30 >/dev/null
  fi
  # wait until Active
  for i in 1 2 3 4 5 6 7 8 9 10; do
    state=$($AWSX lambda get-function --function-name "$name" --query 'Configuration.State' --output text 2>/dev/null || echo Pending)
    [ "$state" = "Active" ] && break
    sleep 1
  done
}
upsert_lambda reserve-inventory  reserve_inventory.handler reserve-inventory-role  build/zips/reserve-inventory.zip
upsert_lambda charge-payment     charge_payment.handler    charge-payment-role     build/zips/charge-payment.zip
upsert_lambda create-shipment    create_shipment.handler   create-shipment-role    build/zips/create-shipment.zip
upsert_lambda release-inventory  release_inventory.handler release-inventory-role  build/zips/release-inventory.zip
upsert_lambda refund-payment     refund_payment.handler    refund-payment-role     build/zips/refund-payment.zip
upsert_lambda saga-notifier      saga_notifier.handler     saga-notifier-role      build/zips/saga-notifier.zip

echo "== EventBridge rule -> notifier =="
$AWSX events put-rule \
  --name on-saga-terminal \
  --event-bus-name "$BUS_NAME" \
  --event-pattern '{"source":["order.saga"],"detail-type":["OrderCompleted","OrderFailed"]}' >/dev/null

$AWSX lambda add-permission \
  --function-name saga-notifier \
  --statement-id allow-eventbridge-invoke \
  --action lambda:InvokeFunction \
  --principal events.amazonaws.com \
  --source-arn "arn:aws:events:${REGION}:${ACCOUNT}:rule/${BUS_NAME}/on-saga-terminal" >/dev/null 2>&1 || true

$AWSX events put-targets \
  --rule on-saga-terminal \
  --event-bus-name "$BUS_NAME" \
  --targets "Id=notifier,Arn=${ARN_PREFIX_LAMBDA}:saga-notifier" >/dev/null

echo "== State machine =="
SM_NAME="order-saga"
SM_ROLE_ARN=$(role_arn order-saga-role)
SM_DEF=$(cat sm/order-saga.json)
EXISTING_SM_ARN=$($AWSX stepfunctions list-state-machines --query "stateMachines[?name=='${SM_NAME}'].stateMachineArn | [0]" --output text)
if [ "$EXISTING_SM_ARN" = "None" ] || [ -z "$EXISTING_SM_ARN" ]; then
  SM_ARN=$($AWSX stepfunctions create-state-machine \
    --name "$SM_NAME" \
    --role-arn "$SM_ROLE_ARN" \
    --definition "$SM_DEF" \
    --query 'stateMachineArn' --output text)
else
  SM_ARN="$EXISTING_SM_ARN"
  $AWSX stepfunctions update-state-machine \
    --state-machine-arn "$SM_ARN" \
    --definition "$SM_DEF" \
    --role-arn "$SM_ROLE_ARN" >/dev/null
fi
echo "state machine: $SM_ARN"

printf '%s\n' "$SM_ARN" > build/state_machine_arn.txt
printf '%s\n' "$KMS_ARN" > build/kms_arn.txt
printf '%s\n' "$SECRET_ARN" > build/secret_arn.txt
echo
echo "OK , provisioned"

#!/usr/bin/env bash
set -euo pipefail
AWSX="aws --endpoint-url=${AWS_ENDPOINT_URL}"
SM_ARN=$(cat /app/build/state_machine_arn.txt)

pass() { echo "PASS: $1"; }
fail() { echo "FAIL: $1"; exit 1; }

reset_tables() {
  for t in inventory payments shipments saga-idempotency; do
    pk=$($AWSX dynamodb describe-table --table-name "$t" --query 'Table.KeySchema[0].AttributeName' --output text)
    items=$($AWSX dynamodb scan --table-name "$t" --projection-expression "#k" \
      --expression-attribute-names "{\"#k\":\"$pk\"}" --query 'Items[].'"$pk"'.S' --output text 2>/dev/null || true)
    for v in $items; do
      $AWSX dynamodb delete-item --table-name "$t" --key "{\"$pk\":{\"S\":\"$v\"}}" >/dev/null
    done
  done
}

run_sm() {
  local name="$1" input="$2"
  local arn
  arn=$($AWSX stepfunctions start-execution --state-machine-arn "$SM_ARN" --name "$name" --input "$input" --query 'executionArn' --output text)
  for _ in $(seq 1 60); do
    st=$($AWSX stepfunctions describe-execution --execution-arn "$arn" --query 'status' --output text)
    [ "$st" != "RUNNING" ] && break
    sleep 1
  done
  echo "$st"
}

notifier_has() {
  local needle="$1"
  local lg="/aws/lambda/saga-notifier"
  local streams
  streams=$($AWSX logs describe-log-streams --log-group-name "$lg" --order-by LastEventTime --descending --max-items 10 --query 'logStreams[].logStreamName' --output text 2>/dev/null || true)
  for s in $streams; do
    if $AWSX logs get-log-events --log-group-name "$lg" --log-stream-name "$s" --query 'events[].message' --output text 2>/dev/null | tr '\t' '\n' | grep -qa -- "$needle"; then
      return 0
    fi
  done
  return 1
}

echo "== Policy docs: no wildcards on KMS/Lambda invoke =="
# charge-payment kms:Decrypt must not be on "*"
if jq -e '.Statement[] | select(.Action | tostring | contains("kms:")) | select(.Resource=="*")' \
    /app/build/policies/charge-payment-role.json >/dev/null 2>&1; then
  fail "charge-payment kms has wildcard resource"
fi
pass "charge-payment kms resource is not *"

if jq -e '.Statement[] | select(.Action=="lambda:InvokeFunction") | select(.Resource=="*")' \
    /app/build/policies/order-saga-role.json >/dev/null 2>&1; then
  fail "saga role has wildcard on lambda invoke"
fi
pass "saga role lambda:InvokeFunction is per-function, not *"

# Only payment role has secrets read
for p in reserve-inventory release-inventory create-shipment refund-payment saga-notifier; do
  if grep -q 'secretsmanager:GetSecretValue' /app/build/policies/${p}-role.json; then
    fail "$p role has secret read permission"
  fi
done
pass "only charge-payment role reads the secret"

echo
echo "== Happy path =="
reset_tables
st=$(run_sm "v-happy-$(date +%s)" '{"order_id":"o-1","sku":"x","qty":2,"amount":100,"idempotency_key":"k-1"}')
[ "$st" = "SUCCEEDED" ] || fail "happy path status=$st"
pass "state machine SUCCEEDED"

rq=$($AWSX dynamodb get-item --table-name inventory --key '{"sku":{"S":"x"}}' --query 'Item.reserved_qty.N' --output text)
[ "$rq" = "2" ] || fail "expected reserved_qty=2, got $rq"
pass "inventory reserved_qty=2"

ps=$($AWSX dynamodb get-item --table-name payments --key '{"payment_id":{"S":"pay-o-1"}}' --query 'Item.status.S' --output text)
[ "$ps" = "CAPTURED" ] || fail "expected payment CAPTURED, got $ps"
pass "payment row CAPTURED"

sid=$($AWSX dynamodb get-item --table-name shipments --key '{"shipment_id":{"S":"ship-o-1"}}' --query 'Item.shipment_id.S' --output text)
[ "$sid" = "ship-o-1" ] || fail "shipment row missing"
pass "shipment row exists"

sleep 2
notifier_has "saga-terminal:OrderCompleted" || fail "no saga-terminal:OrderCompleted log"
pass "notifier logged saga-terminal:OrderCompleted"

echo
echo "== Failure path =="
reset_tables
st=$(run_sm "v-fail-$(date +%s)" '{"order_id":"o-2","sku":"x","qty":2,"amount":100,"idempotency_key":"k-2","force_failure":true}')
[ "$st" = "SUCCEEDED" ] || fail "failure path status=$st"
pass "state machine SUCCEEDED (saga handled gracefully)"

rq=$($AWSX dynamodb get-item --table-name inventory --key '{"sku":{"S":"x"}}' --query 'Item.reserved_qty.N' --output text)
[ "$rq" = "0" ] || fail "expected reserved_qty=0 after release, got $rq"
pass "inventory reserved_qty=0 (release ran)"

pexists=$($AWSX dynamodb get-item --table-name payments --key '{"payment_id":{"S":"pay-o-2"}}' --query 'Item.payment_id.S' --output text 2>/dev/null)
if [ -n "$pexists" ] && [ "$pexists" != "None" ]; then
  pstatus=$($AWSX dynamodb get-item --table-name payments --key '{"payment_id":{"S":"pay-o-2"}}' --query 'Item.status.S' --output text)
  [ "$pstatus" = "REFUNDED" ] || fail "payment row present with status=$pstatus (want absent or REFUNDED)"
  pass "payment row REFUNDED"
else
  pass "payment row absent (nothing to refund)"
fi

sexists=$($AWSX dynamodb get-item --table-name shipments --key '{"shipment_id":{"S":"ship-o-2"}}' --query 'Item.shipment_id.S' --output text 2>/dev/null)
if [ -n "$sexists" ] && [ "$sexists" != "None" ]; then
  fail "shipment row should not exist"
fi
pass "shipment row absent"

sleep 2
notifier_has "saga-terminal:OrderFailed" || fail "no saga-terminal:OrderFailed log"
pass "notifier logged saga-terminal:OrderFailed"

echo
echo "ALL CHECKS PASSED"