Event-Driven Development with Amazon EventBridge

What you'll learn

How to build event-driven services on Amazon EventBridge — the pattern JunctionNet and CargoNaut use so services can react to each other without ever calling each other directly. A producer publishes a fact; any number of consumers subscribe by pattern and run in their own Lambda. The producer never knows they exist.

The mental model

Four moving parts, and only four. Learn these and the rest is detail:

producer  ──put_events──▶  event bus  ──rule (pattern match)──▶  target (consumer)
                              │                                    ├─▶ Lambda  (react)
                              │                                    ├─▶ SQS     (buffer + retry)
                              │                                    ├─▶ Firehose ▶ OpenSearch (analytics)
                              │                                    └─▶ another bus / API
  ▲ one bus per environment (e.g. Platform-Apps-events-dev)
  ▲ a producer knows the bus — never the consumers

What an event actually looks like

You control four fields when you publish; EventBridge stamps the rest (id, time, account, region). Rules match on the top-level envelope — mainly source and detail-type:

{
  // --- you set these ---
  "EventBusName": "Platform-Apps-events-dev",
  "Source":       "cargonaut.shipment.router",   // who emitted it
  "DetailType":   "ShipmentDelivered",           // what happened (a fact)
  "Detail": {                                    // the payload (your JSON)
    "shipment_id": "shp_123",
    "org_id": "org_abc",
    "occurred_at": "2026-07-12T10:04:00Z"
  },
  // --- EventBridge adds: id, time, region, account, resources ---
}

Naming the source

source answers "who emitted this?" Use a stable, dotted, lowercase namespace — never a display name or something that changes. The platform convention:

# {platform}.{service}.{module}   — dotted, lowercase, stable
cargonaut.shipment.router
cargonaut.document.extractor
blueprint.events_management.publisher
audit.import

• Namespaced & owned — the prefix (cargonaut, blueprint) makes it obvious which platform/team owns the event, and lets a rule subscribe to a whole domain with a prefix filter.
• Stable forever — consumers pattern-match on it. Renaming a source silently breaks every subscriber. Treat it as a public contract.
• Not the AWS aws.* space — that prefix is reserved for AWS's own events (S3, etc.). Yours is your namespace.

Naming the detail-type

detail-type answers "what happened?" The rule that makes an event-driven system sane: events are facts, in the past tense — {Entity}{PastTenseVerb}, PascalCase.

Versioning

Payloads evolve. Add fields freely (consumers ignore unknowns); for a breaking change, publish a new type (ShipmentDeliveredV2) and retire the old one once no rule matches it — never silently repurpose an existing type.

When & where to publish

The single most common bug in event-driven code is publishing at the wrong moment. The rule: publish a fact only after the state change is durably committed, from the service layer, off the request's critical path.

# application/services.py — publish AFTER the write succeeds
def deliver_shipment(self, shipment_id: str) -> Shipment:
    shipment = self.repository.mark_delivered(shipment_id)   # 1. commit the fact
    self.event_bus.publish(                                  # 2. then announce it
        "ShipmentDelivered",
        {"shipment_id": shipment.id, "org_id": shipment.org_id},
    )
    return shipment

// application/services.js — publish AFTER the write succeeds
import { EventBridgeClient, PutEventsCommand } from "@aws-sdk/client-eventbridge";
const eb = new EventBridgeClient({});

export async function deliverShipment(shipmentId) {
  const shipment = await repository.markDelivered(shipmentId);   // 1. commit the fact
  await eb.send(new PutEventsCommand({                           // 2. then announce it
    Entries: [{
      EventBusName: process.env.EVENT_BUS,
      Source: "cargonaut.shipment.router",
      DetailType: "ShipmentDelivered",
      Detail: JSON.stringify({ shipmentId: shipment.id, orgId: shipment.orgId }),
    }],
  }));
  return shipment;
}

• After commit, not before — publish ShipmentDelivered before the DB write and a crash leaves consumers acting on something that never happened.
• From the service, not the handler or repository — the handler is HTTP glue; the repository only touches data. The service owns the business transaction, so it owns the event.
• Off the critical path — the caller's response shouldn't wait on downstream work. Publish and return; consumers run asynchronously.
• One event per real business fact — not one per function call. If a method does two things, that may be two events (or one — model the domain, not the code).

Subscribing: rules & patterns

A consumer declares an event pattern — EventBridge only invokes it for matching events. In SAM this is an EventBridgeRule event source; SAM creates the rule and the invoke permission:

# template.yaml — a consumer that only wakes for two shipment facts
OnShipmentEventFn:
  Type: AWS::Serverless::Function
  Properties:
    Handler: shipment.handlers.event_bus.on_shipment.handler
    Events:
      ShipmentFacts:
        Type: EventBridgeRule
        Properties:
          EventBusName: !Ref PlatformBus
          Pattern:
            source: [ cargonaut.shipment.router ]
            detail-type: [ ShipmentDelivered, ShipmentCancelled ]

• One consumer per concern — a "notify customer" Lambda and an "update analytics" Lambda are two rules, not one function with an if.
• Match narrowly — subscribe to the exact detail-types you handle, so you aren't invoked (and billed) for events you'll ignore.
• Content filtering — patterns can match inside detail too (prefix, anything-but, numeric ranges), so you filter in the bus, not in code.

Make it reliable

EventBridge delivers at least once — occasionally twice, occasionally out of order. Design consumers for that:

Wiring event → SQS → Lambda

A rule can target a Lambda directly (as above), but for anything bursty or rate-sensitive you put an SQS queue in between: the rule delivers to the queue, and the Lambda drains it in batches. The queue becomes a durable buffer — it absorbs spikes, gives you real retry control, and hands poison messages to a dead-letter queue.

event ──rule──▶ SQS queue ──batch──▶ Lambda
                   └──(after N failed receives)──▶ dead-letter queue

The SAM template — four pieces

# template.yaml
Resources:
  # 1. the buffer queue + a dead-letter queue for poison messages
  ShipmentDLQ:
    Type: AWS::SQS::Queue
    Properties: { MessageRetentionPeriod: 1209600 }   # 14 days
  ShipmentQueue:
    Type: AWS::SQS::Queue
    Properties:
      VisibilityTimeout: 180                # ≥ ~6× the Lambda timeout
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt ShipmentDLQ.Arn
        maxReceiveCount: 5                  # give up after 5 tries → DLQ

  # 2. an EventBridge rule that delivers matching events TO the queue
  ShipmentRule:
    Type: AWS::Events::Rule
    Properties:
      EventBusName: !Ref PlatformBus
      EventPattern:
        source: [ cargonaut.shipment.router ]
        detail-type: [ ShipmentDelivered, ShipmentCancelled ]
      Targets:
        - Id: shipment-queue
          Arn: !GetAtt ShipmentQueue.Arn

  # 3. let EventBridge write to the queue (scoped to THIS rule)
  ShipmentQueuePolicy:
    Type: AWS::SQS::QueuePolicy
    Properties:
      Queues: [ !Ref ShipmentQueue ]
      PolicyDocument:
        Statement:
          - Effect: Allow
            Principal: { Service: events.amazonaws.com }
            Action: sqs:SendMessage
            Resource: !GetAtt ShipmentQueue.Arn
            Condition:
              ArnEquals: { aws:SourceArn: !GetAtt ShipmentRule.Arn }

  # 4. the Lambda drains the queue in batches (SAM makes the event-source mapping)
  OnShipmentEventFn:
    Type: AWS::Serverless::Function
    Properties:
      Handler: shipment.handlers.sqs.on_shipment.handler
      Runtime: python3.13
      Events:
        FromQueue:
          Type: SQS
          Properties:
            Queue: !GetAtt ShipmentQueue.Arn
            BatchSize: 10
            FunctionResponseTypes: [ ReportBatchItemFailures ]   # partial-batch retries

The handler — one record at a time, retry only what failed

Each SQS message body is the full EventBridge envelope, so parse record.body to get back detail-type and detail. Use Powertools' batch processor so a single bad message is retried on its own instead of re-running the whole batch:

# src/shipment/handlers/sqs/on_shipment.py
import json
from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.batch import (
    BatchProcessor, EventType, process_partial_response)
from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord

logger = Logger(service="ShipmentEvents")
processor = BatchProcessor(event_type=EventType.SQS)

def record_handler(record: SQSRecord):
    event = json.loads(record.body)              # the EventBridge envelope
    detail = event["detail"]
    if event["detail-type"] == "ShipmentDelivered":
        logger.info("delivered", extra={"id": detail["shipment_id"]})
        # …idempotent work; raise to retry ONLY this message

def handler(event, context):
    return process_partial_response(event, record_handler, processor, context)

// src/shipment/handlers/sqs/onShipment.js
export const handler = async (event) => {
  const batchItemFailures = [];
  for (const record of event.Records) {
    try {
      const evt = JSON.parse(record.body);          // the EventBridge envelope
      if (evt["detail-type"] === "ShipmentDelivered") {
        // …idempotent work; throw to retry ONLY this message
      }
    } catch (err) {
      batchItemFailures.push({ itemIdentifier: record.messageId });
    }
  }
  return { batchItemFailures };                       // ReportBatchItemFailures
};

• Backpressure & batching — the queue absorbs a spike; the Lambda pulls BatchSize messages per invoke (fewer, cheaper invocations). Cap the function's reserved concurrency to protect a fragile downstream (a DB, a third-party API).
• Real retry control — a failed message returns to the queue and is retried up to maxReceiveCount, then lands in the DLQ. Set VisibilityTimeout to at least ~6× the function timeout so a slow message isn't delivered twice at once.
• Partial-batch failure — ReportBatchItemFailures + the Powertools processor return only the failed message ids, so a single poison record doesn't force the whole batch to reprocess.
• Need ordering? — use an SQS FIFO queue with a MessageGroupId (e.g. per shipment_id) so events for one entity process in order, while different entities still run in parallel.

Extending development with events

This is why the ceremony pays off. Because producers don't know their consumers, new behaviour is a new subscriber — you add a feature without opening, testing, or redeploying the service that owns the data.

# Day 1: shipment-service publishes ShipmentDelivered.
# Later, with ZERO changes to shipment-service, add consumers:
ShipmentDelivered
   ├─▶ notification-service   # email the customer
   ├─▶ usage-service          # meter a billable event
   ├─▶ cockpit-service        # close the action card
   └─▶ analytics (OpenSearch) # dashboards + audit trail

• Choreography over orchestration — no central coordinator; each service reacts to facts it cares about. The system's behaviour is the sum of subscriptions.
• Safe to add, safe to remove — deleting a consumer removes a behaviour with no risk to the producer. Try that with a direct function call.
• Parallel teams — the team that owns notifications ships a new rule against an existing event without a PR to the shipment team.

Three things every event platform needs

Once services talk through a bus, the bus becomes the place you debug from — and an unobserved bus is worse than no bus at all, because failures stop being exceptions in a caller's stack trace and start being events that simply never arrived. Three pieces cover it, and they deploy together:

producers ──put_events──► EventBridge bus ──┬─► rule (Error/*) ─► SQS ─► digest λ ─► SNS ─► you
                                            └─► rule (all)  ──────────► forwarder λ ─► OpenSearch

          + CloudWatch alarms on the bus itself (failed invocations, throttles, DLQ depth)
          + a daily report for what the stream can't see (cost, quotas, capacity)

The bus, with its plumbing attached

The bus stack creates more than an AWS::Events::EventBus. It also publishes the bus to SSM so no service ever hardcodes its name, and it creates the error path that the digest Lambda will drain:

PlatformEventBus:
  Type: AWS::Events::EventBus
  Properties:
    Name: CargoNaut-Apps-events-${self:provider.stage}

# the contract: services discover the bus here, never by string literal
EventBusNameParameter:
  Type: AWS::SSM::Parameter
  Properties:
    Name: /platform/event_bus_name
    Value: !Ref PlatformEventBus

# one rule selects the failures — everything else flows to the forwarder
ErrorAlertsRule:
  Type: AWS::Events::Rule
  Properties:
    EventBusName: !Ref PlatformEventBus
    EventPattern:
      detail-type: [ "Error", "ProcessingError", "IntegrationError" ]
    Targets:
      - Arn: !GetAtt ErrorAlertsQueue.Arn
        Id: ErrorAlertsQueue

Throttled error alerts: rule → SQS → digest → SNS

The obvious wiring — rule straight to SNS — works beautifully until the first real incident, when a retry loop emits four hundred IntegrationError events in ninety seconds and your team gets four hundred emails. After that, people filter the alert address, and you have built an outage detector nobody reads.

Buffer in SQS and batch. Two settings do the throttling, and neither is code:

AlertDigestFunction:
  Type: AWS::Serverless::Function
  Properties:
    ReservedConcurrentExecutions: 1        # only ever one digest in flight
    Events:
      ErrorQueue:
        Type: SQS
        Properties:
          Queue: !GetAtt ErrorAlertsQueue.Arn
          BatchSize: 100
          MaximumBatchingWindowInSeconds: 60   # wait up to a minute, then send

Concurrency 1 plus a 60-second batching window gives a hard ceiling of one notification per minute, no matter how loud the failure is. The Lambda's only job is to make that one message worth reading — group by source, de-duplicate, and cap the detail:

def lambda_handler(event, context):
    errors_by_source = {}
    for record in event.get("Records", []):                  # up to 100 per batch
        body   = json.loads(record.get("body", "{}"))
        detail = body.get("detail", {})
        errors_by_source.setdefault(body.get("source", "unknown"), []).append({
            "type":  body.get("detail-type", "Error"),
            "org":   detail.get("organization", "unknown"),
            # an alert is a pointer, not a log
            "error": str(detail.get("error") or detail.get("message", ""))[:200],
        })
    # collapse "the same error 300 times" to one line, 5 distinct per source
    blocks, total = [], sum(len(v) for v in errors_by_source.values())
    for source, errs in errors_by_source.items():
        seen = {f"{e['type']}:{e['error'][:80]}": e for e in errs}
        blocks.append(f"--- {source} ({len(errs)}) ---")
        blocks += [f"  [{e['type']}] {e['org']}: {e['error']}" for e in list(seen.values())[:5]]
    sns.publish(TopicArn=TOPIC, Subject=f"[{STAGE}] {total} error(s)",
                Message="\n".join(blocks))

export const handler = async (event) => {
  const bySource = {};
  for (const record of event.Records ?? []) {          // up to 100 per batch
    const body = JSON.parse(record.body ?? "{}"), detail = body.detail ?? {};
    (bySource[body.source ?? "unknown"] ??= []).push({
      type: body["detail-type"] ?? "Error", org: detail.organization ?? "unknown",
      // an alert is a pointer, not a log
      error: String(detail.error ?? detail.message ?? "").slice(0, 200),
    });
  }
  // collapse "the same error 300 times" to one line, 5 distinct per source
  const blocks = [], total = Object.values(bySource).flat().length;
  for (const [source, errs] of Object.entries(bySource)) {
    const seen = new Map(errs.map((e) => [`${e.type}:${e.error.slice(0, 80)}`, e]));
    blocks.push(`--- ${source} (${errs.length}) ---`, ...[...seen.values()].slice(0, 5).map((e) => `  [${e.type}] ${e.org}: ${e.error}`));
  }
  await sns.send(new PublishCommand({ TopicArn: TOPIC, Subject: `[${STAGE}] ${total} error(s)`, Message: blocks.join("\n") }));
};

Forward every event to OpenSearch — the stream becomes memory

The second rule is a catch-all: every event on the bus goes to a forwarder Lambda that indexes it. The same stream that drives your services becomes a searchable, durable record you can build dashboards on.

Events:
  AllPlatformEvents:
    Type: EventBridgeRule
    Properties:
      EventBusName: !Ref PlatformEventBus
      Pattern:
        account: [ !Ref 'AWS::AccountId' ]     # i.e. everything on this bus

The forwarder's real work is shaping. An EventBridge envelope is not a good document: the useful fields are buried in detail, and every producer nests them differently. Flatten to one schema at the boundary, once, so every dashboard can rely on it:

# domain.py — one document shape for every event, whatever produced it
class AppEventModel(BaseModel):
    source:       str                 # cargonaut.shipment
    event:        str                 # ShipmentDelivered  (the detail-type)
    payload:      Dict[str, Any]      # the whole detail, kept raw
    environment:  str
    application:  str
    organization: str
    username:     str
    timestamp:    str = Field(default_factory=lambda: datetime.utcnow().isoformat() + "Z")

    @field_serializer('payload')
    def serialize_payload(self, payload, _info) -> str:
        # stringify: producers put arbitrary shapes in here, and a mapping
        # explosion in OpenSearch is a painful way to find that out.
        return json.dumps(payload, ensure_ascii=False)

// domain.js — one document shape for every event, whatever produced it
export const toAppEvent = (event) => {
  const detail = event.detail ?? {};
  return {
    source:       event.source,                  // cargonaut.shipment
    event:        event["detail-type"],          // ShipmentDelivered
    // stringify: producers put arbitrary shapes in here, and a mapping
    // explosion in OpenSearch is a painful way to find that out.
    payload:      JSON.stringify(detail),
    environment:  detail.environment ?? "unknown",
    application:  detail.application ?? "unknown",
    organization: detail.organization ?? "default_org",
    username:     detail.username ?? "default_user",
    timestamp:    new Date().toISOString(),
  };
};

The handler itself stays trivial — shape it, save it, and skip your own noise:

# handler.py
EXCLUDED_DETAIL_TYPES = {"WarmupLambda"}      # don't index your own noise

@logger.inject_lambda_context(log_event=True)
@event_source(data_class=EventBridgeEvent)
def lambda_handler(event: EventBridgeEvent, context):
    if event.detail_type in EXCLUDED_DETAIL_TYPES:
        return True
    app_event = ServiceFactory.create_app_event_entity(event)
    ServiceFactory.create_app_events_forwarder_service().save(app_event)
    return True

// handler.js
const EXCLUDED_DETAIL_TYPES = new Set(["WarmupLambda"]);   // don't index your own noise

export const handler = async (event) => {
  if (EXCLUDED_DETAIL_TYPES.has(event["detail-type"])) return true;
  await repository.saveEvent(toAppEvent(event));
  return true;
};

Authentication to OpenSearch is SigV4 with the Lambda's own execution role — no password anywhere, the same keyless principle as everything else on the platform. One parameter decides which service you're signing for, and getting it wrong produces a 403 that looks like a network problem:

Why the mirror earns its keep:

• A durable audit trail, for free — every fact the business emitted, timestamped and immutable. "What happened to shipment X, and when?" becomes one query instead of eleven log groups.
• Analytics without touching producers — funnels, volumes, latency between events (created → delivered), per-org activity. Services stay lean; analytics is a read-side concern.
• Dashboards & alerting — OpenSearch Dashboards over the index gives ops real-time throughput; alert when ExtractionFailed spikes.
• Decoupled & safe — the forwarder is just another subscriber. If OpenSearch is down your services don't notice; only the mirror lags.

Alarms on the bus itself

The failures above are your code's. These are the platform's, and they are the ones that produce the worst kind of incident — the one where nothing errors anywhere and events simply don't arrive:

The daily report: what the stream can't tell you

Events tell you what your application did. They say nothing about whether the account underneath it is healthy — cost drift, a table filling up, Lambdas being throttled, a certificate quietly approaching expiry. That needs a scheduled collector, and it is the simplest EventBridge consumer in the platform: a cron rule and one Lambda.

Events:
  DailySchedule:
    Type: Schedule
    Properties:
      # 10:00 UTC = 4am Mexico City. Pick a time in YOUR team's morning.
      Schedule: cron(0 10 * * ? *)
      Enabled: !Ref ScheduleEnabled       # parameterised: off in ephemeral stages

The report sweeps the account and emails an HTML digest:

Two details make it useful rather than another ignored email. First, it emits the same data as a JSON snapshot to S3, which a portal renders as a live infra dashboard — one collector, two consumers. Second, the snapshot write happens before the email:

# Publish before sending: a snapshot failure must never block the email.
snapshot = build_snapshot(now, alarms, costs, lambda_health, ...)
snapshot_written = write_snapshot(snapshot)

html = html.replace("</body>", snapshot_section(snapshot) + "</body>")
send_email(report_date, html, snapshot)

Packaging it: from a platform folder to a one-click app

Everything above is generic. A bus, an error digest, a forwarder and some alarms are what any EventBridge platform needs — nothing in them is about freight. So the last step was to lift the whole thing out of the platform repo into marketplace/events-observability: one SAM template, parameterised, published to the AWS Serverless Application Repository.

Metadata:
  AWS::ServerlessRepo::Application:
    Name: serverless-events-observability
    Author: CargoNaut
    SpdxLicenseId: Apache-2.0
    LicenseUrl: LICENSE.txt
    ReadmeUrl: README.md         # this becomes the SAR listing page
    Labels: ['eventbridge', 'observability', 'opensearch', 'serverless']
    SemanticVersion: 1.0.0        # bump before every publish — SAR rejects re-use

What turned an internal stack into a publishable one was mostly removing assumptions, and that exercise is worth doing even if you never publish:

Publishing is four make targets, and the whole pipeline is one:

$ make create-bucket S3_BUCKET=my-sar-artifacts   # once: artifact bucket + serverlessrepo read policy
$ make release      S3_BUCKET=my-sar-artifacts   # validate → build → package → publish

# then, for a consumer, the entire install is:
$ sam deploy --template-url <SAR-url> --stack-name events-observability \
    --capabilities CAPABILITY_IAM \
    --parameter-overrides OpenSearchEndpoint=… OpenSearchIndex=platform-events \
                          AlertEmail=team@example.com

A CLAUDE.md for event-driven work

# CLAUDE.md — Events

## Conventions
- source: {platform}.{service}.{module}  (dotted, lowercase, STABLE — it's a contract)
- detail-type: {Entity}{PastTenseVerb}   (PascalCase, a fact — never a command)
- Detail carries ids + org_id + occurred_at; never secrets; add fields, don't repurpose.

## Publishing
- Publish from the SERVICE layer, AFTER the state change commits. Fire-and-forget.
- Never publish from handlers or repositories. One event per real business fact.

## Consuming
- One consumer per concern; match narrowly on source + detail-type.
- Idempotent (key on event id); attach a DLQ; tolerate re-delivery and out-of-order.
- If a consumer publishes, carry a hop-count to prevent loops.

## Analytics
- A catch-all rule mirrors the bus to OpenSearch (Firehose) for audit + dashboards.
  Producers are unaware of it. Never make a service depend on the analytics mirror.

Ask for a feature the event-driven way

State the fact, its name, when it fires, and let the consumer be new — don't touch the producer's logic:

> When a shipment is delivered, email the customer — event-driven:
  Producer: shipment-service already commits the delivery in
  ShipmentService.deliver_shipment. Add a publish of "ShipmentDelivered"
  (source cargonaut.shipment.router) AFTER the repository write, detail =
  { shipment_id, org_id, occurred_at }. Don't change the delivery logic.
  Consumer: a NEW Lambda in notification-service, EventBridgeRule on
  source=cargonaut.shipment.router, detail-type=[ShipmentDelivered].
  Make it idempotent on the event id and give it a DLQ.
  Show me the template.yaml rule + the handler, and a put-events test.