WebSockets Serverless em Hiperescala: Comunicação Bidirecional com Amazon API Gateway e EventBridge

resource "aws_apigatewayv2_api" "websocket_api" {
name = "enterprise-realtime-api"
protocol_type = "WEBSOCKET"
route_selection_expression = "$request.body.action"
}

resource "aws_dynamodb_table" "connection_state" {
name = "WebSocketConnections"
billing_mode = "PAY_PER_REQUEST"
hash_key = "UserId"
range_key = "ConnectionId"

attribute {
name = "UserId"
type = "S"
}

attribute {
name = "ConnectionId"
type = "S"
}

tags = {
Domain = "EdgeConnectivity"
}
}

import json
import boto3
from dataclasses import dataclass
from aws_lambda_powertools import Logger

logger = Logger(service="BillingDomain")
eventbridge = boto3.client("events")

@dataclass(frozen=True)
class InvoicePaidEvent:
user_id: str
invoice_id: str
amount: float

class BillingDomainService:
def process_payment(self, event: InvoicePaidEvent) -> None:
logger.info(f"Pagamento processado e validado para a fatura {event.invoice_id}")

# O payload contém exclusivamente semântica de negócios
payload = {
"user_id": event.user_id,
"type": "INVOICE_PAID",
"data": {"amount": event.amount, "status": "CLEARED"}
}

eventbridge.put_events(
Entries=[{
"Source": "com.enterprise.billing",
"DetailType": "RealTimeNotification",
"Detail": json.dumps(payload),
"EventBusName": "enterprise-bus"
}]
)

import os
import json
import boto3
from botocore.exceptions import ClientError
from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.data_classes import SQSEvent

logger = Logger(service="WebSocketDelivery")
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.getenv("CONNECTIONS_TABLE"))

# O endpoint requer o protocolo HTTPS e o estágio de implantação da API
management_api = boto3.client(
"apigatewaymanagementapi",
endpoint_url=os.getenv("APIGW_MANAGEMENT_ENDPOINT")
)

class NotificationDeliveryAdapter:
def push_to_active_devices(self, user_id: str, payload: dict) -> None:
response = table.query(
KeyConditionExpression=boto3.dynamodb.conditions.Key('UserId').eq(user_id)
)

for item in response.get('Items', []):
connection_id = item['ConnectionId']
try:
management_api.post_to_connection(
ConnectionId=connection_id,
Data=json.dumps(payload).encode('utf-8')
)
logger.info(f"Dados transmitidos com sucesso para o soquete {connection_id}")

except ClientError as e:
# Interceptação vitalícia da conexão silenciosamente perdida
if e.response['Error']['Code'] == 'GoneException':
logger.warning(f"Soquete obsoleto {connection_id}. Executando expurgo.")
table.delete_item(
Key={'UserId': user_id, 'ConnectionId': connection_id}
)
else:
raise e

delivery_adapter = NotificationDeliveryAdapter()

@logger.inject_lambda_context
def lambda_handler(event: dict, context: dict) -> None:
sqs_event = SQSEvent(event)
for record in sqs_event.records:
body = json.loads(record.body)
detail = body.get("detail", {})

target_user = detail.get("user_id")
if target_user:
delivery_adapter.push_to_active_devices(target_user, detail)