Category: Uncategorized

resource "oci_core_vcn" "api_vcn" {
  compartment_id = var.compartment_id
  cidr_block     = "10.0.0.0/16"
  display_name   = "api-gateway-vcn"
  dns_label      = "apivcn"
}
resource "oci_core_subnet" "public_subnet" {
  compartment_id    = var.compartment_id
  vcn_id            = oci_core_vcn.api_vcn.id
  cidr_block        = "10.0.1.0/24"
  display_name      = "gateway-public-subnet"
  dns_label         = "gatewaypub"
  route_table_id    = oci_core_route_table.public_rt.id
  security_list_ids = [oci_core_security_list.gateway_sl.id]
}
resource "oci_core_subnet" "private_subnet" {
  compartment_id             = var.compartment_id
  vcn_id                     = oci_core_vcn.api_vcn.id
  cidr_block                 = "10.0.2.0/24"
  display_name               = "functions-private-subnet"
  dns_label                  = "funcpriv"
  prohibit_public_ip_on_vnic = true
  route_table_id             = oci_core_route_table.private_rt.id
  security_list_ids          = [oci_core_security_list.functions_sl.id]
}

resource "oci_core_service_gateway" "sgw" {
  compartment_id = var.compartment_id
  vcn_id         = oci_core_vcn.api_vcn.id
  display_name   = "functions-service-gateway"
  services {
    service_id = data.oci_core_services.all_services.services[0].id
  }
}
resource "oci_core_route_table" "private_rt" {
  compartment_id = var.compartment_id
  vcn_id         = oci_core_vcn.api_vcn.id
  display_name   = "private-route-table"
  route_rules {
    network_entity_id = oci_core_service_gateway.sgw.id
    destination       = "all-iad-services-in-oracle-services-network"
    destination_type  = "SERVICE_CIDR_BLOCK"
  }
}

resource "oci_identity_dynamic_group" "api_gateway_dg" {
  compartment_id = var.tenancy_ocid
  name           = "api-gateway-dynamic-group"
  description    = "Allows API Gateway to invoke Functions"
  matching_rule  = "ALL {resource.type = 'ApiGateway', resource.compartment.id = '${var.compartment_id}'}"
}
resource "oci_identity_policy" "gateway_invoke_policy" {
  compartment_id = var.compartment_id
  name           = "gateway-invoke-functions-policy"
  description    = "Grants API Gateway permission to invoke Functions"
  statements = [
    "Allow dynamic-group api-gateway-dynamic-group to use functions-family in compartment id ${var.compartment_id}"
  ]
}

schema_version: 20180708
name: order-validator
version: 0.0.1
runtime: python3.11
build_image: fnproject/python:3.11-dev
run_image: fnproject/python:3.11
entrypoint: /python/bin/fdk /function/func.py handler
memory: 256

import io
import json
import logging
from fdk import response
def handler(ctx, data: io.BytesIO = None):
    logger = logging.getLogger()
    
    try:
        body = json.loads(data.getvalue())
    except (Exception, ValueError) as ex:
        logger.error("Failed to parse request body: " + str(ex))
        return response.Response(
            ctx,
            status_code=400,
            response_data=json.dumps({"error": "Invalid JSON in request body"}),
            headers={"Content-Type": "application/json"}
        )
    required_fields = ["order_id", "customer_id", "items", "total_amount"]
    missing = [f for f in required_fields if f not in body]
    if missing:
        return response.Response(
            ctx,
            status_code=422,
            response_data=json.dumps({
                "error": "Missing required fields",
                "fields": missing
            }),
            headers={"Content-Type": "application/json"}
        )
    if not isinstance(body.get("items"), list) or len(body["items"]) == 0:
        return response.Response(
            ctx,
            status_code=422,
            response_data=json.dumps({"error": "Order must contain at least one item"}),
            headers={"Content-Type": "application/json"}
        )
    if body["total_amount"] <= 0:
        return response.Response(
            ctx,
            status_code=422,
            response_data=json.dumps({"error": "total_amount must be greater than zero"}),
            headers={"Content-Type": "application/json"}
        )
    return response.Response(
        ctx,
        status_code=200,
        response_data=json.dumps({
            "status": "valid",
            "order_id": body["order_id"],
            "item_count": len(body["items"]),
            "validated_at": ctx.RequestID()
        }),
        headers={"Content-Type": "application/json"}
    )

# Log in to OCI Container Registry
docker login <region-key>.ocir.io -u '<tenancy-namespace>/<username>'
# Build the function image
fn build --verbose
# Tag and push
docker tag order-validator:0.0.1 <region-key>.ocir.io/<tenancy-namespace>/functions/order-validator:0.0.1
docker push <region-key>.ocir.io/<tenancy-namespace>/functions/order-validator:0.0.1

resource "oci_functions_application" "orders_app" {
  compartment_id = var.compartment_id
  display_name   = "orders-api"
  subnet_ids     = [oci_core_subnet.private_subnet.id]
  config = {
    LOG_LEVEL = "INFO"
    ENV       = "production"
  }
  trace_config {
    is_enabled = true
    domain_id  = oci_apm_apm_domain.tracing.id
  }
}
resource "oci_functions_function" "order_validator" {
  application_id = oci_functions_application.orders_app.id
  display_name   = "order-validator"
  image          = "<region-key>.ocir.io/<tenancy-namespace>/functions/order-validator:0.0.1"
  memory_in_mbs  = 256
  timeout_in_seconds = 30
  provisioned_concurrency_config {
    strategy = "CONSTANT"
    count    = 2
  }
}

resource "oci_apigateway_gateway" "orders_gateway" {
  compartment_id = var.compartment_id
  display_name   = "orders-api-gateway"
  endpoint_type  = "PUBLIC"
  subnet_id      = oci_core_subnet.public_subnet.id
  certificate_id = var.tls_certificate_ocid
}
resource "oci_apigateway_deployment" "orders_deployment" {
  compartment_id = var.compartment_id
  display_name   = "orders-api-v1"
  gateway_id     = oci_apigateway_gateway.orders_gateway.id
  path_prefix    = "/v1"
  specification {
    request_policies {
      authentication {
        type                        = "JWT_AUTHENTICATION"
        token_header                = "Authorization"
        token_auth_scheme           = "Bearer"
        is_anonymous_access_allowed = false
        public_keys {
          type    = "REMOTE_JWKS"
          uri     = "https://your-identity-provider.com/.well-known/jwks.json"
          max_cache_duration_in_hours = 1
        }
        verify_claims {
          key      = "iss"
          values   = ["https://your-identity-provider.com"]
          is_required = true
        }
        verify_claims {
          key      = "aud"
          values   = ["orders-api"]
          is_required = true
        }
      }
      rate_limiting {
        rate_in_requests_per_second = 100
        rate_key                    = "CLIENT_IP"
      }
      cors {
        allowed_origins = ["https://yourdomain.com"]
        allowed_methods = ["GET", "POST", "OPTIONS"]
        allowed_headers = ["Authorization", "Content-Type"]
        max_age_in_seconds = 3600
        is_allow_credentials_enabled = true
      }
    }
    routes {
      path    = "/orders/validate"
      methods = ["POST"]
      backend {
        type        = "ORACLE_FUNCTIONS_BACKEND"
        function_id = oci_functions_function.order_validator.id
      }
      request_policies {
        body_validation {
          required = true
          content {
            media_type      = "application/json"
            validation_type = "NONE"
          }
        }
      }
      response_policies {
        header_transformations {
          set_headers {
            items {
              name   = "X-Request-ID"
              values = ["${request.headers[x-request-id]}"]
            }
            items {
              name   = "Strict-Transport-Security"
              values = ["max-age=31536000; includeSubDomains"]
            }
          }
        }
      }
    }
    routes {
      path    = "/orders/validate"
      methods = ["OPTIONS"]
      backend {
        type = "STOCK_RESPONSE_BACKEND"
        status  = 204
        headers {
          name  = "Access-Control-Allow-Origin"
          value = "https://yourdomain.com"
        }
      }
    }
  }
}

terraform output gateway_endpoint

curl -X POST https://<gateway-id>.apigateway.<region>.oci.customer-oci.com/v1/orders/validate \
  -H "Content-Type: application/json" \
  -d '{"order_id": "ORD-001"}'

TOKEN=$(curl -s -X POST https://your-idp.com/oauth/token \
  -d "grant_type=client_credentials&client_id=...&client_secret=..." | jq -r .access_token)
curl -X POST https://<gateway-id>.apigateway.<region>.oci.customer-oci.com/v1/orders/validate \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "order_id": "ORD-001",
    "customer_id": "CUST-999",
    "items": [{"sku": "ITEM-A", "qty": 2}],
    "total_amount": 49.99
  }'

{
  "status": "valid",
  "order_id": "ORD-001",
  "item_count": 1,
  "validated_at": "ocid1.apigateway.request..."
}

resource "oci_logging_log" "gateway_execution_log" {
  display_name = "gateway-execution-log"
  log_group_id = oci_logging_log_group.api_logs.id
  log_type     = "SERVICE"
  configuration {
    source {
      category    = "execution"
      resource    = oci_apigateway_deployment.orders_deployment.id
      service     = "apigateway"
      source_type = "OCISERVICE"
    }
    compartment_id = var.compartment_id
  }
  retention_duration = 30
  is_enabled         = true
}

PURGE RECYCLEBIN;
DROP DOMAIN IF EXISTS emails;
DROP DOMAIN IF EXISTS cc;
CREATE DOMAIN emails AS VARCHAR2(100)
  CONSTRAINT email_chk CHECK (
    REGEXP_LIKE(emails, '^(\S+)\@(\S+)\.(\S+)$')
  )
  DISPLAY  LOWER(emails)
  ORDER    LOWER(emails)
  ANNOTATIONS (
    Description 'An email address with a check constraint 
    for name @ domain dot (.) something'
  );
CREATE DOMAIN cc AS VARCHAR2(19)
  CONSTRAINT cc_chk CHECK (
    REGEXP_LIKE(cc, '^\d+(\d+)*$')
  )
  ANNOTATIONS (
    Description 'Credit card number with a check constraint 
    no dashes, no spaces!'
  );

DROP TABLE IF EXISTS orders CASCADE CONSTRAINTS PURGE;
DROP TABLE IF EXISTS customers CASCADE CONSTRAINTS PURGE;
CREATE TABLE IF NOT EXISTS orders (
    id             NUMBER,
    product_id     NUMBER,
    order_date     TIMESTAMP,
    customer_id    NUMBER,
    total_value    NUMBER(6,2),
    order_shipped  BOOLEAN,
    warranty       INTERVAL YEAR TO MONTH
);
CREATE TABLE IF NOT EXISTS customers (
    id             NUMBER,
    first_name     VARCHAR2(100),
    last_name      VARCHAR2(100),
    dob            DATE,
    email          emails,
    address        VARCHAR2(200),
    zip            VARCHAR2(10),
    phone_number   VARCHAR2(20),
    credit_card    cc,
    joined_date    TIMESTAMP DEFAULT SYSTIMESTAMP,
    gold_customer  BOOLEAN DEFAULT FALSE,
    CONSTRAINT new_customers_pk PRIMARY KEY (id)
);
ALTER TABLE orders ADD (CONSTRAINT orders_pk PRIMARY KEY (id));
ALTER TABLE orders ADD (
  CONSTRAINT orders_fk FOREIGN KEY (customer_id) 
  REFERENCES customers (id)
);

INSERT INTO customers 
  (id, first_name, last_name, dob, email, address, 
   zip, phone_number, credit_card)
VALUES  
  (1, 'Alice', 'Brown', DATE '1990-01-01', 
   'alice.brown@example.com', '123 Maple Street', 
   '12345', '555-1234', '4111111111110000'),
  (3, 'Bob', 'Brown', DATE '1990-01-01', 
   'email1@example.com', '333 Maple Street', 
   '12345', '555-5678', '4111111111111111'),
  (4, 'Clarice', 'Jones', DATE '1990-01-01', 
   'email8888@example.com', '222 Bourbon Street', 
   '12345', '555-7856', '4111111111111110'),
  (5, 'David', 'Smith', DATE '1990-01-01', 
   'email375@example.com', '111 Walnut Street', 
   '12345', '555-3221', '4111111111111112');
INSERT INTO orders 
  (id, customer_id, product_id, order_date, 
   total_value, order_shipped, warranty)
VALUES
  (100, 1, 101, SYSTIMESTAMP, 300.00, NULL, NULL),
  (101, 4, 101, SYSTIMESTAMP - 30, 129.99, TRUE, 
   INTERVAL '5' YEAR);
COMMIT;

CREATE OR REPLACE FORCE JSON RELATIONAL DUALITY VIEW 
  customers_dv AS 
  customers @insert @update @delete
{
    _id          : id,
    FirstName    : first_name,
    LastName     : last_name,
    DateOfBirth  : dob,
    Email        : email,
    Address      : address,
    Zip          : zip,
    phoneNumber  : phone_number,
    creditCard   : credit_card,
    joinedDate   : joined_date,
    goldStatus   : gold_customer
};

INSERT INTO customers_dv VALUES (
  '{"_id": 2, "FirstName": "Jim", "LastName": "Brown", 
    "Email": "jim.brown@example.com", 
    "Address": "456 Maple Street", "Zip": 12345}'
);
COMMIT;

CREATE OR REPLACE JSON RELATIONAL DUALITY VIEW 
  customer_orders_dv
  ANNOTATIONS (
    Description 'JSON Relational Duality View 
    sourced from CUSTOMERS and ORDERS'
  )
AS SELECT JSON {
    '_id'        : c.ID,
    'FirstName'  : c.FIRST_NAME,
    'LastName'   : c.LAST_NAME,
    'Address'    : c.ADDRESS,
    'Zip'        : c.ZIP,
    'orders'     : 
      [ SELECT JSON {
          'OrderID'      : o.ID WITH NOUPDATE,
          'ProductID'    : o.PRODUCT_ID,
          'OrderDate'    : o.ORDER_DATE,
          'TotalValue'   : o.TOTAL_VALUE,
          'OrderShipped' : o.ORDER_SHIPPED
        }
        FROM ORDERS o WITH INSERT UPDATE DELETE
        WHERE o.CUSTOMER_ID = c.ID
      ]
  }
FROM CUSTOMERS c;

SELECT * FROM customer_orders_dv o 
WHERE o.data."_id" = 1;

UPDATE customer_orders_dv c
SET c.data = json_transform(
    data,
    APPEND '$.orders' = JSON {
      'OrderID': 123, 
      'ProductID': 202, 
      'OrderDate': SYSTIMESTAMP, 
      'TotalValue': 150.00
    }
)
WHERE c.data."_id" = 1;
COMMIT;
SELECT * FROM customer_orders_dv o 
WHERE o.data."_id" = 1;

{
  "version": "0",
  "id": "12345678-1234-1234-1234-123456789012",
  "detail-type": "Order Created",
  "source": "com.mycompany.orders",
  "account": "123456789012",
  "time": "2025-03-05T10:30:00Z",
  "region": "us-east-1",
  "resources": [],
  "detail": {
    "orderId": "ORD-12345",
    "customerId": "CUST-67890",
    "items": [
      {
        "productId": "PROD-111",
        "quantity": 2,
        "price": 29.99
      }
    ],
    "totalAmount": 59.98,
    "currency": "USD",
    "shippingAddress": {
      "country": "US",
      "state": "CA",
      "city": "San Francisco",
      "zipCode": "94102"
    },
    "metadata": {
      "correlationId": "req-abc123",
      "version": "1.0"
    }
  }
}

# Create schema registry
aws schemas create-registry \
  --registry-name my-company-events \
  --description "Event schemas for our microservices"

{
  "openapi": "3.0.0",
  "info": {
    "title": "OrderCreated",
    "version": "1.0.0"
  },
  "paths": {},
  "components": {
    "schemas": {
      "OrderCreated": {
        "type": "object",
        "required": ["orderId", "customerId", "totalAmount"],
        "properties": {
          "orderId": {
            "type": "string",
            "pattern": "^ORD-[0-9]+$"
          },
          "customerId": {
            "type": "string"
          },
          "totalAmount": {
            "type": "number",
            "minimum": 0
          },
          "currency": {
            "type": "string",
            "enum": ["USD", "EUR", "GBP"]
          }
        }
      }
    }
  }
}

# eventbridge.tf
# Create custom event bus (don't use default for production)
resource "aws_cloudwatch_event_bus" "main" {
  name = "mycompany-events"
  
  tags = {
    Environment = "production"
    Team        = "platform"
  }
}
# Event bus policy - allow other accounts to put events
resource "aws_cloudwatch_event_bus_policy" "main" {
  event_bus_name = aws_cloudwatch_event_bus.main.name
  
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Sid       = "AllowAccountsToPutEvents"
        Effect    = "Allow"
        Principal = {
          AWS = [
            "arn:aws:iam::111111111111:root",  # Dev account
            "arn:aws:iam::222222222222:root"   # Staging account
          ]
        }
        Action    = "events:PutEvents"
        Resource  = aws_cloudwatch_event_bus.main.arn
      }
    ]
  })
}
# Archive for event replay (critical for debugging)
resource "aws_cloudwatch_event_archive" "main" {
  name             = "mycompany-events-archive"
  event_source_arn = aws_cloudwatch_event_bus.main.arn
  retention_days   = 30
  
  # Archive all events
  event_pattern = jsonencode({
    source = [{ prefix = "com.mycompany" }]
  })
}

# Order events to inventory service
resource "aws_cloudwatch_event_rule" "order_to_inventory" {
  name           = "order-created-to-inventory"
  event_bus_name = aws_cloudwatch_event_bus.main.name
  
  event_pattern = jsonencode({
    source      = ["com.mycompany.orders"]
    detail-type = ["Order Created"]
  })
  
  tags = {
    Service = "inventory"
  }
}
resource "aws_cloudwatch_event_target" "inventory_lambda" {
  rule           = aws_cloudwatch_event_rule.order_to_inventory.name
  event_bus_name = aws_cloudwatch_event_bus.main.name
  target_id      = "inventory-processor"
  arn            = aws_lambda_function.inventory_processor.arn
  
  # Retry configuration
  retry_policy {
    maximum_event_age_in_seconds = 3600  # 1 hour
    maximum_retry_attempts       = 3
  }
  
  # Dead letter queue for failed events
  dead_letter_config {
    arn = aws_sqs_queue.inventory_dlq.arn
  }
}
# High-value orders get special handling
resource "aws_cloudwatch_event_rule" "high_value_orders" {
  name           = "high-value-orders"
  event_bus_name = aws_cloudwatch_event_bus.main.name
  
  # Content-based filtering - only orders over $1000
  event_pattern = jsonencode({
    source      = ["com.mycompany.orders"]
    detail-type = ["Order Created"]
    detail = {
      totalAmount = [{ numeric = [">=", 1000] }]
    }
  })
}
resource "aws_cloudwatch_event_target" "fraud_check" {
  rule           = aws_cloudwatch_event_rule.high_value_orders.name
  event_bus_name = aws_cloudwatch_event_bus.main.name
  target_id      = "fraud-check"
  arn            = aws_sfn_state_machine.fraud_check.arn
  role_arn       = aws_iam_role.eventbridge_sfn.arn
}

# Match events from multiple sources
event_pattern = jsonencode({
  source = ["com.mycompany.orders", "com.mycompany.returns"]
})
# Match specific values in nested objects
event_pattern = jsonencode({
  detail = {
    shippingAddress = {
      country = ["US", "CA", "MX"]  # North America only
    }
  }
})
# Prefix matching
event_pattern = jsonencode({
  detail = {
    orderId = [{ prefix = "ORD-PRIORITY-" }]
  }
})
# Exists check
event_pattern = jsonencode({
  detail = {
    promoCode = [{ exists = true }]  # Only orders with promo codes
  }
})
# Combine multiple conditions
event_pattern = jsonencode({
  source      = ["com.mycompany.orders"]
  detail-type = ["Order Created"]
  detail = {
    totalAmount = [{ numeric = [">=", 100] }]
    currency    = ["USD"]
    items = {
      productId = [{ prefix = "DIGITAL-" }]
    }
  }
})

# order_service/handler.py
import json
import boto3
import uuid
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import List
eventbridge = boto3.client('events')
@dataclass
class OrderItem:
    productId: str
    quantity: int
    price: float
@dataclass
class OrderCreatedEvent:
    orderId: str
    customerId: str
    items: List[dict]
    totalAmount: float
    currency: str
    shippingAddress: dict
    metadata: dict
def create_order(event, context):
    """Handle order creation request."""
    body = json.loads(event['body'])
    
    # Generate order ID
    order_id = f"ORD-{uuid.uuid4().hex[:8].upper()}"
    
    # Calculate total
    items = body['items']
    total = sum(item['quantity'] * item['price'] for item in items)
    
    # Save to database (simplified)
    save_order_to_dynamodb(order_id, body)
    
    # Create the event
    order_event = OrderCreatedEvent(
        orderId=order_id,
        customerId=body['customerId'],
        items=items,
        totalAmount=total,
        currency=body.get('currency', 'USD'),
        shippingAddress=body['shippingAddress'],
        metadata={
            'correlationId': event['requestContext']['requestId'],
            'version': '1.0',
            'timestamp': datetime.utcnow().isoformat()
        }
    )
    
    # Publish to EventBridge
    publish_event(
        source='com.mycompany.orders',
        detail_type='Order Created',
        detail=asdict(order_event)
    )
    
    return {
        'statusCode': 201,
        'body': json.dumps({
            'orderId': order_id,
            'status': 'created'
        })
    }
def publish_event(source: str, detail_type: str, detail: dict):
    """Publish event to EventBridge with error handling."""
    try:
        response = eventbridge.put_events(
            Entries=[
                {
                    'Source': source,
                    'DetailType': detail_type,
                    'Detail': json.dumps(detail),
                    'EventBusName': 'mycompany-events'
                }
            ]
        )
        
        # Check for partial failures
        if response['FailedEntryCount'] > 0:
            failed = response['Entries'][0]
            raise Exception(f"Failed to publish event: {failed['ErrorCode']} - {failed['ErrorMessage']}")
            
    except Exception as e:
        # Log the error but don't fail the order
        # Consider sending to a fallback queue
        print(f"Error publishing event: {e}")
        send_to_fallback_queue(source, detail_type, detail)
def send_to_fallback_queue(source, detail_type, detail):
    """Send to SQS as fallback if EventBridge fails."""
    sqs = boto3.client('sqs')
    sqs.send_message(
        QueueUrl=os.environ['FALLBACK_QUEUE_URL'],
        MessageBody=json.dumps({
            'source': source,
            'detailType': detail_type,
            'detail': detail
        })
    )

def publish_events_batch(events: List[dict]):
    """Publish multiple events efficiently."""
    # EventBridge accepts up to 10 events per call
    BATCH_SIZE = 10
    
    entries = []
    for event in events:
        entries.append({
            'Source': event['source'],
            'DetailType': event['detail_type'],
            'Detail': json.dumps(event['detail']),
            'EventBusName': 'mycompany-events'
        })
    
    # Process in batches
    failed_events = []
    for i in range(0, len(entries), BATCH_SIZE):
        batch = entries[i:i + BATCH_SIZE]
        
        response = eventbridge.put_events(Entries=batch)
        
        if response['FailedEntryCount'] > 0:
            for idx, entry in enumerate(response['Entries']):
                if 'ErrorCode' in entry:
                    failed_events.append({
                        'event': batch[idx],
                        'error': entry['ErrorCode']
                    })
    
    return failed_events

# inventory_service/handler.py
import json
import boto3
from decimal import Decimal
dynamodb = boto3.resource('dynamodb')
inventory_table = dynamodb.Table('inventory')
def process_order_created(event, context):
    """
    Process OrderCreated events to update inventory.
    
    EventBridge invokes this Lambda with the full event envelope.
    """
    # Extract the event detail
    detail = event['detail']
    order_id = detail['orderId']
    items = detail['items']
    correlation_id = detail['metadata']['correlationId']
    
    print(f"Processing order {order_id} (correlation: {correlation_id})")
    
    try:
        # Reserve inventory for each item
        for item in items:
            reserve_inventory(
                product_id=item['productId'],
                quantity=item['quantity'],
                order_id=order_id
            )
        
        # Publish success event
        publish_event(
            source='com.mycompany.inventory',
            detail_type='Inventory Reserved',
            detail={
                'orderId': order_id,
                'status': 'reserved',
                'items': items,
                'metadata': {
                    'correlationId': correlation_id
                }
            }
        )
        
    except InsufficientInventoryError as e:
        # Publish failure event
        publish_event(
            source='com.mycompany.inventory',
            detail_type='Inventory Reservation Failed',
            detail={
                'orderId': order_id,
                'reason': str(e),
                'failedItems': e.failed_items,
                'metadata': {
                    'correlationId': correlation_id
                }
            }
        )
        
        # Don't raise - we've handled it by publishing an event
        return {'status': 'failed', 'reason': str(e)}
    
    return {'status': 'success'}
def reserve_inventory(product_id: str, quantity: int, order_id: str):
    """
    Atomically reserve inventory using DynamoDB conditional writes.
    """
    try:
        inventory_table.update_item(
            Key={'productId': product_id},
            UpdateExpression='''
                SET availableQuantity = availableQuantity - :qty,
                    reservedQuantity = reservedQuantity + :qty,
                    lastUpdated = :now
                ADD reservations :reservation
            ''',
            ConditionExpression='availableQuantity >= :qty',
            ExpressionAttributeValues={
                ':qty': quantity,
                ':now': datetime.utcnow().isoformat(),
                ':reservation': {order_id}
            }
        )
    except dynamodb.meta.client.exceptions.ConditionalCheckFailedException:
        raise InsufficientInventoryError(
            f"Insufficient inventory for {product_id}",
            failed_items=[product_id]
        )

{
  "Comment": "Process order notifications with multiple channels",
  "StartAt": "DetermineNotificationChannels",
  "States": {
    "DetermineNotificationChannels": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.detail.totalAmount",
          "NumericGreaterThanEquals": 500,
          "Next": "HighValueOrderNotifications"
        }
      ],
      "Default": "StandardNotifications"
    },
    
    "HighValueOrderNotifications": {
      "Type": "Parallel",
      "Branches": [
        {
          "StartAt": "SendEmail",
          "States": {
            "SendEmail": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:us-east-1:123456789:function:send-email",
              "End": true
            }
          }
        },
        {
          "StartAt": "SendSMS",
          "States": {
            "SendSMS": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:us-east-1:123456789:function:send-sms",
              "End": true
            }
          }
        },
        {
          "StartAt": "NotifyAccountManager",
          "States": {
            "NotifyAccountManager": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:us-east-1:123456789:function:slack-notify",
              "End": true
            }
          }
        }
      ],
      "Next": "RecordNotificationsSent"
    },
    
    "StandardNotifications": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789:function:send-email",
      "Next": "RecordNotificationsSent"
    },
    
    "RecordNotificationsSent": {
      "Type": "Task",
      "Resource": "arn:aws:states:::dynamodb:putItem",
      "Parameters": {
        "TableName": "notification-log",
        "Item": {
          "orderId": {"S.$": "$.detail.orderId"},
          "notifiedAt": {"S.$": "$$.State.EnteredTime"},
          "channels": {"S": "email,sms"}
        }
      },
      "End": true
    }
  }
}

# DLQ for inventory service
resource "aws_sqs_queue" "inventory_dlq" {
  name                      = "inventory-events-dlq"
  message_retention_seconds = 1209600  # 14 days
  
  tags = {
    Service = "inventory"
    Purpose = "dead-letter-queue"
  }
}
# Alarm when messages hit DLQ
resource "aws_cloudwatch_metric_alarm" "inventory_dlq_alarm" {
  alarm_name          = "inventory-dlq-messages"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 1
  metric_name         = "ApproximateNumberOfMessagesVisible"
  namespace           = "AWS/SQS"
  period              = 300
  statistic           = "Sum"
  threshold           = 0
  alarm_description   = "Messages in inventory DLQ"
  
  dimensions = {
    QueueName = aws_sqs_queue.inventory_dlq.name
  }
  
  alarm_actions = [aws_sns_topic.alerts.arn]
}

# dlq_processor/handler.py
import json
import boto3
eventbridge = boto3.client('events')
sqs = boto3.client('sqs')
def process_dlq(event, context):
    """
    Process messages from DLQ.
    Attempt to republish or escalate.
    """
    for record in event['Records']:
        message = json.loads(record['body'])
        
        # Parse the original event
        original_event = json.loads(message.get('detail', '{}'))
        failure_reason = message.get('errorMessage', 'Unknown')
        receipt_handle = record['receiptHandle']
        
        # Get retry count from message attributes
        retry_count = int(
            record.get('messageAttributes', {})
            .get('RetryCount', {})
            .get('stringValue', '0')
        )
        
        if retry_count < 3:
            # Try to republish with delay
            try:
                reprocess_event(original_event, retry_count + 1)
                delete_from_dlq(record['eventSourceARN'], receipt_handle)
            except Exception as e:
                print(f"Retry failed: {e}")
                
        else:
            # Max retries exceeded - escalate
            escalate_to_operations(original_event, failure_reason)
            move_to_permanent_failure_queue(record)
def escalate_to_operations(event, reason):
    """Alert operations team about permanent failure."""
    sns = boto3.client('sns')
    sns.publish(
        TopicArn=os.environ['OPS_ALERT_TOPIC'],
        Subject='Event Processing Failure - Manual Intervention Required',
        Message=json.dumps({
            'event': event,
            'reason': reason,
            'action_required': 'Manual review and potential data reconciliation'
        }, indent=2)
    )

import hashlib
def process_order_created(event, context):
    """Idempotent event processor."""
    detail = event['detail']
    
    # Create idempotency key from event ID
    event_id = event['id']
    
    # Check if we've already processed this event
    if is_already_processed(event_id):
        print(f"Event {event_id} already processed, skipping")
        return {'status': 'duplicate'}
    
    try:
        # Process the event
        result = do_actual_processing(detail)
        
        # Mark as processed
        mark_as_processed(event_id, result)
        
        return result
        
    except Exception as e:
        # Don't mark as processed on failure - allow retry
        raise
def is_already_processed(event_id: str) -> bool:
    """Check DynamoDB for processed event."""
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('processed-events')
    
    response = table.get_item(Key={'eventId': event_id})
    return 'Item' in response
def mark_as_processed(event_id: str, result: dict):
    """Record that we processed this event."""
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('processed-events')
    
    table.put_item(
        Item={
            'eventId': event_id,
            'processedAt': datetime.utcnow().isoformat(),
            'result': result,
            'ttl': int((datetime.utcnow() + timedelta(days=7)).timestamp())
        }
    )

import boto3
cloudwatch = boto3.client('cloudwatch')
def publish_business_metrics(event_type: str, properties: dict):
    """Publish custom business metrics."""
    cloudwatch.put_metric_data(
        Namespace='MyCompany/Events',
        MetricData=[
            {
                'MetricName': 'EventsProcessed',
                'Dimensions': [
                    {'Name': 'EventType', 'Value': event_type},
                    {'Name': 'Service', 'Value': 'inventory'}
                ],
                'Value': 1,
                'Unit': 'Count'
            },
            {
                'MetricName': 'OrderValue',
                'Dimensions': [
                    {'Name': 'Currency', 'Value': properties.get('currency', 'USD')}
                ],
                'Value': properties.get('totalAmount', 0),
                'Unit': 'None'
            }
        ]
    )

from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.core import patch_all
# Patch all supported libraries
patch_all()
@xray_recorder.capture('process_order_created')
def process_order_created(event, context):
    # Add correlation ID as annotation
    correlation_id = event['detail']['metadata']['correlationId']
    xray_recorder.current_subsegment().put_annotation('correlationId', correlation_id)
    
    # Your processing logic
    with xray_recorder.in_subsegment('reserve_inventory'):
        reserve_inventory(event['detail']['items'])
    
    with xray_recorder.in_subsegment('publish_event'):
        publish_event(...)

resource "aws_cloudwatch_dashboard" "events" {
  dashboard_name = "event-driven-system"
  
  dashboard_body = jsonencode({
    widgets = [
      {
        type   = "metric"
        x      = 0
        y      = 0
        width  = 12
        height = 6
        properties = {
          title  = "Events Published"
          region = "us-east-1"
          metrics = [
            ["AWS/Events", "Invocations", "EventBusName", "mycompany-events"]
          ]
          period = 60
          stat   = "Sum"
        }
      },
      {
        type   = "metric"
        x      = 12
        y      = 0
        width  = 12
        height = 6
        properties = {
          title  = "Failed Invocations"
          region = "us-east-1"
          metrics = [
            ["AWS/Events", "FailedInvocations", "EventBusName", "mycompany-events"]
          ]
          period = 60
          stat   = "Sum"
        }
      },
      {
        type   = "metric"
        x      = 0
        y      = 6
        width  = 24
        height = 6
        properties = {
          title  = "Event Processing Latency by Service"
          region = "us-east-1"
          metrics = [
            ["AWS/Lambda", "Duration", "FunctionName", "inventory-processor"],
            ["AWS/Lambda", "Duration", "FunctionName", "notification-processor"],
            ["AWS/Lambda", "Duration", "FunctionName", "analytics-processor"]
          ]
          period = 60
          stat   = "Average"
        }
      }
    ]
  })
}

# test_inventory_handler.py
import pytest
from unittest.mock import patch, MagicMock
from inventory_service.handler import process_order_created
@pytest.fixture
def order_created_event():
    return {
        'id': 'test-event-123',
        'source': 'com.mycompany.orders',
        'detail-type': 'Order Created',
        'detail': {
            'orderId': 'ORD-TEST',
            'customerId': 'CUST-123',
            'items': [
                {'productId': 'PROD-1', 'quantity': 2, 'price': 29.99}
            ],
            'totalAmount': 59.98,
            'metadata': {
                'correlationId': 'req-test'
            }
        }
    }
@patch('inventory_service.handler.reserve_inventory')
@patch('inventory_service.handler.publish_event')
def test_process_order_reserves_inventory(mock_publish, mock_reserve, order_created_event):
    result = process_order_created(order_created_event, None)
    
    assert result['status'] == 'success'
    mock_reserve.assert_called_once_with(
        product_id='PROD-1',
        quantity=2,
        order_id='ORD-TEST'
    )
    mock_publish.assert_called_once()
@patch('inventory_service.handler.reserve_inventory')
@patch('inventory_service.handler.publish_event')
def test_insufficient_inventory_publishes_failure(mock_publish, mock_reserve, order_created_event):
    mock_reserve.side_effect = InsufficientInventoryError("Out of stock", ['PROD-1'])
    
    result = process_order_created(order_created_event, None)
    
    assert result['status'] == 'failed'
    
    # Verify failure event was published
    call_args = mock_publish.call_args
    assert call_args[1]['detail_type'] == 'Inventory Reservation Failed'

# test_integration.py
import boto3
import pytest
import json
@pytest.fixture(scope='session')
def localstack_eventbridge():
    """Set up LocalStack EventBridge for testing."""
    client = boto3.client(
        'events',
        endpoint_url='http://localhost:4566',
        region_name='us-east-1'
    )
    
    # Create test event bus
    client.create_event_bus(Name='test-events')
    
    yield client
    
    # Cleanup
    client.delete_event_bus(Name='test-events')
def test_event_routing(localstack_eventbridge):
    """Test that events are routed correctly."""
    # Create a rule that sends to SQS for testing
    localstack_eventbridge.put_rule(
        Name='test-rule',
        EventBusName='test-events',
        EventPattern=json.dumps({
            'source': ['com.mycompany.orders'],
            'detail-type': ['Order Created']
        })
    )
    
    # Publish test event
    localstack_eventbridge.put_events(
        Entries=[{
            'Source': 'com.mycompany.orders',
            'DetailType': 'Order Created',
            'Detail': json.dumps({'orderId': 'TEST-123'}),
            'EventBusName': 'test-events'
        }]
    )
    
    # Verify event was received (check target queue)
    # ...

def create_order(order_data):
    order_id = generate_order_id()
    
    # Save state
    save_to_database(order_id, order_data)
    
    # Also save the event
    save_event({
        'eventType': 'OrderCreated',
        'entityId': order_id,
        'data': order_data,
        'timestamp': datetime.utcnow()
    })
    
    # Publish to EventBridge
    publish_event(...)
```
### Pattern: Saga for Distributed Transactions
When you need coordination across services:
```
Order Created
  └─> Inventory Reserved (success)
        └─> Payment Processed (success)
              └─> Order Confirmed
              
        └─> Payment Failed
              └─> Release Inventory (compensation)
              └─> Order Cancelled
```
### Anti-Pattern: Event Chains
Avoid long chains where each service publishes an event that triggers the next:
```
# BAD: Long chain creates debugging nightmare
A -> B -> C -> D -> E
# BETTER: Use orchestration (Step Functions) for complex workflows
A -> Step Functions orchestrates B, C, D, E

// BAD
{
  "customer": {
    "id": "123",
    "name": "...",
    "address": "...",
    "creditHistory": [...],  // 50KB of data
    "orderHistory": [...]     // Another 100KB
  }
}
// GOOD
{
  "customerId": "123",
  "customerName": "John Doe"  // Only what consumers need
}