Metadata-Version: 2.4
Name: eventcommon
Version: 0.1.1
Summary: Shared event envelope, HMAC signing, and SNS/SQS outbox library for platform services.
Project-URL: Homepage, https://github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon
Project-URL: Repository, https://github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon
Project-URL: Changelog, https://github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon/blob/main/CHANGELOG.md
Author: BCBP Solutions FZC LLC
License: Apache-2.0
Classifier: Framework :: AsyncIO
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Typing :: Typed
Requires-Python: >=3.12
Requires-Dist: aioboto3>=13.0
Requires-Dist: boto3>=1.34
Requires-Dist: opentelemetry-api>=1.24
Requires-Dist: pgcommon<0.3.0,>=0.1.0
Requires-Dist: prometheus-client>=0.20
Requires-Dist: structlog>=24.1
Description-Content-Type: text/markdown

# eventcommon

Shared **event-driven messaging** library for platform services. Consumed as a private Python package — not deployed on its own.

**Features:** typed event envelopes (CloudEvents-aligned) · async SNS publisher · async SQS consumer · transactional outbox (Postgres-backed via `pgcommon`) · HMAC signing/verification · Prometheus metrics · OpenTelemetry tracing · RLS tenant propagation.

See [ARCHITECTURE.md](ARCHITECTURE.md) for design decisions and layer diagrams.

> **Cross-language contract:** `eventcommon`'s `Envelope` wire format and HMAC canonicalisation must stay byte-for-byte compatible with the Go implementation in [`platform-events`](https://github.com/BCBP-SOLUTIONS-FZC-LLC/platform-events) — the two publish to and consume from the same SNS topics/SQS queues in a polyglot deployment. Any change to `envelope.py` or `hmac.py` must be cross-checked against the Go side; the `interop` CI job exists specifically to catch a mismatch before it reaches production. See [CI](#ci).

## Design principles

- **Events are facts, not commands.** An event records something that already happened (`user.created`, `invoice.settled`) — it is an immutable statement of past truth, not an instruction to perform an action. Producers do not direct consumers; consumers decide independently how to react. A consequence: consumers must be resilient to receiving the same fact more than once (`Envelope.id` idempotency) and must handle the absence of a reaction gracefully (a consumer can be added or removed without any producer change).
- **Fail fast on misconfiguration** — `SNSPublisher` raises on an empty or invalid `topic_arn`; `SQSConsumer` raises on an empty `queue_url`. Invalid configuration is rejected at construction time, not at first use.
- **Make safe usage the default** — the SQS consumer automatically propagates `tenant_id` into `pgcommon`'s GUC context for RLS, extends visibility timeouts on slow handlers, and drains in-flight messages on `stop()`. You cannot forget these by accident when the defaults are wired correctly.
- **Keep business logic free from messaging plumbing** — handlers receive a typed `Envelope`; retry backoff, visibility extension, metric recording, and OTel span creation are invisible to the caller.
- **Centralise cross-cutting concerns** — HMAC signing, tenant propagation, and observability hooks live here, not scattered across every service. A single version bump propagates to all consumers.

---

## Contents

- [TL;DR](#tldr)
- [System invariants](#system-invariants)
- [Publishing rules](#publishing-rules)
- [Anti-patterns](#anti-patterns)
- [When NOT to use this](#when-not-to-use-this)
- [Installing](#installing)
- [Quick start](#quick-start)
- [Event envelope](#event-envelope)
- [SNS Publisher](#sns-publisher)
- [SQS Consumer](#sqs-consumer)
  - [Implementing idempotency](#implementing-idempotency)
- [Transactional outbox](#transactional-outbox)
- [HMAC helpers](#hmac-helpers)
  - [When to use HMAC](#when-to-use-hmac)
- [Observability — Prometheus and OTel](#observability--prometheus-and-otel)
  - [Logging correlation](#logging-correlation)
- [Backpressure considerations](#backpressure-considerations)
- [Recommended production defaults](#recommended-production-defaults)
- [Configuration reference](#configuration-reference)
- [Error reference](#error-reference)
- [Testing in consuming services](#testing-in-consuming-services)
- [Testing](#testing)
- [CI](#ci)
- [Versioning and releases](#versioning-and-releases)
- [Service adoption checklist](#service-adoption-checklist)
- [See also](#see-also)

---

## TL;DR

> ⚠️ **Exactly-once delivery is not possible anywhere in this system.** SNS and SQS provide at-least-once delivery — a message may be delivered more than once under normal operating conditions (visibility timeout expiry, consumer crash mid-handler, SQS internal redelivery). FIFO topics reduce the duplicate window to 5 minutes within a `MessageGroupId` but do not eliminate duplicates end-to-end. **Consumer idempotency is not optional — it is required for correctness regardless of whether FIFO is used.** See [Implementing idempotency](#implementing-idempotency) for the concrete pattern.

- The system provides **at-least-once delivery** — consumers must implement idempotency using `Envelope.id`. See [Implementing idempotency](#implementing-idempotency) for the recommended `processed_events` Postgres pattern.
- Use `eventcommon.envelope.new_envelope(...)` to create typed events — never construct `Envelope` directly.
- Always pass `tenant_id=rc.tenant_id` and `trace_id=rc.trace_id` when publishing from an HTTP handler — these fields drive RLS enforcement and trace continuity.
- Use `await eventcommon.outbox.enqueue(conn, env)` inside a `pgcommon.run_in_tx` block to guarantee reliable at-least-once delivery without dual-write risk. See [Publishing rules](#publishing-rules) for the full decision table and crash-window explanation.
- **Never call `publisher.publish` for domain events tied to a DB write** — event loss on process crash is silent and unrecoverable. The outbox is the only safe path.
- Call `eventcommon.metrics.init(service_name, build_version)` once at startup to enable Prometheus metrics.
- OTel tracing is enabled by the consuming service (e.g. `fastapicommon.tracing.init_from_env()`), not by this library.

---

## System invariants

These hold everywhere in the system, always. If your design requires violating any of them, the design needs to change — not the invariant.

| Invariant | What it means for you |
|---|---|
| **Delivery is at-least-once** | Every consumer handler may be called more than once for the same `Envelope.id`. Idempotency is not a nice-to-have. |
| **Ordering is best-effort unless FIFO per group** | Standard SNS/SQS make no ordering promise. FIFO guarantees order only within a single `MessageGroupId`. Cross-group and cross-service ordering is never guaranteed. |
| **Idempotency is required for all consumers** | A handler that is not idempotent is not correct. There is no configuration, queue type, or delivery mode that removes this requirement. |
| **Events are immutable once published** | An `event_type` string and its payload contract are frozen on first production publish. Breaking changes require a new versioned type (`.v2`). |
| **The producer has no knowledge of consumers** | A service must never check "who is listening" before publishing. Consumers come and go; the event type remains. |
| **Tenant context is always explicit** | Every event scoped to a tenant carries `tenant_id`. Using `SYSTEM_TENANT_ID` for a tenant-scoped event is a correctness bug, not a shortcut. |
| **Delivery timing is unbounded** | Consumers must not rely on when an event arrives — only on what it says. A handler that breaks when the event arrives "late" is not correct. |
| **Delivery latency is not SLA-backed** | Event delivery is best-effort within the outbox poll interval plus SNS/SQS propagation time. There is no guaranteed maximum latency. |
| **Idempotency must cover all externally observable side effects** | A DB write, an external HTTP call, a notification, and a cache invalidation are all externally observable. Idempotency that protects the DB write but not the Stripe charge or the email is incomplete. See [Side-effect classification](#side-effect-classification). |

---

## Publishing rules

> **Violating the outbox rule introduces silent data-loss bugs.** A domain event is permanently lost if the process crashes between the DB commit and the SNS call. There is no retry, no error, no log — the event is simply gone.

### Decision table

| Scenario | Correct method | Delivery guarantee |
|---|---|---|
| Domain event tied to a DB write — user created, invoice settled, order placed, state machine transition | `outbox.enqueue` inside `pgcommon.run_in_tx` | **At-least-once** ✅ |
| Background job publishing across tenants — scheduled reconciliation, batch processing | `outbox.enqueue` with `tenant_id=SYSTEM_TENANT_ID` inside `pgcommon.run_in_tx` | **At-least-once** ✅ |
| External notification not tied to a DB write — webhook ping after a read-only operation | `publisher.publish` | At-most-once ⚠️ |
| Fire-and-forget telemetry or audit where loss is explicitly acceptable | `publisher.publish` | At-most-once ⚠️ |
| Service with no PostgreSQL database (document in your service ADR) | `publisher.publish` | At-most-once ⚠️ |

**The rule in one sentence:** if a consumer's state would be wrong or incomplete if it never received this event, use the outbox.

### Why the crash window matters

Without the outbox, there is an unrecoverable gap between the DB commit and the SNS publish:

```
WITHOUT OUTBOX:
  1. BEGIN
  2. INSERT INTO users ...     ← domain write
  3. COMMIT                    ← process crashes here (OOM, deploy, network drop)
  4. await sns.publish(...)    ← never executes — event lost permanently
```

With the outbox, the event is durable the moment the transaction commits:

```
WITH OUTBOX:
  1. BEGIN
  2. INSERT INTO users ...          ← domain write
  3. INSERT INTO outbox_events ...  ← event write in same transaction
  4. COMMIT                         ← both rows durable; crash here is safe
  5. [outbox runner, async]
       await publisher.publish(...) ← retried up to max_attempts on failure
       published_at = NOW()         ← idempotent completion marker
```

If the runner crashes between steps 4 and 5, the row is still in `outbox_events` with `published_at IS NULL`. The next runner instance (or the restarted process) picks it up on the next poll cycle.

### Detecting misuse in code review

Flag any call to `publisher.publish` or `publisher.publish_batch` that appears inside a handler that also writes to the database. The correct pattern is:

```python
# ✅ Correct — both writes in one transaction
async with pgcommon.run_in_tx(pool) as conn:
    await repo.save_user(conn, user)
    await outbox.enqueue(conn, env)

# ❌ Wrong — SNS call outside the transaction; event lost on crash
async with pgcommon.run_in_tx(pool) as conn:
    await repo.save_user(conn, user)
await publisher.publish(env)  # ← data inconsistency risk
```

---

## Anti-patterns

Common mistakes that cause silent failures, data loss, or broken tenant isolation.

| Anti-pattern | Why it's wrong | What to do instead |
|---|---|---|
| `await publisher.publish(env)` directly inside an HTTP handler for a transactional event | If the process dies after the DB commit but before `publish` returns, the event is silently dropped — no retry, no recovery | Use `outbox.enqueue` inside `pgcommon.run_in_tx` alongside the domain write |
| Ignoring `Envelope.id` in the consumer handler | The outbox runner delivers at-least-once; without an idempotency check, a redelivered event causes duplicate side effects | `INSERT INTO processed_events (event_id) VALUES ($1) ON CONFLICT DO NOTHING` inside the same transaction as the side-effect write |
| Branching on `env.data` without checking `env.type` first | A handler subscribed to multiple event types will silently misparse the wrong payload | Always check `env.type` before parsing `env.data`; route to typed handlers by event type |
| FIFO topic with a non-stable `MessageGroupId` (e.g. random UUID per message) | Defeats ordering — every message lands in its own group; SNS treats them as independent and delivers concurrently | Use a deterministic, stable group ID: `tenant_id`, `user_id`, or `aggregate_id` — a value that must be ordered relative to itself |
| `tenant_id=SYSTEM_TENANT_ID` in an HTTP handler context | Bypasses per-tenant RLS on the consumer side; the handler's DB queries run without a tenant GUC, returning wrong row sets silently | Pass `tenant_id=rc.tenant_id` from the request context; only use `SYSTEM_TENANT_ID` for genuine cross-tenant background jobs |
| Parsing event payloads with a schema that rejects unknown fields (`extra="forbid"` in Pydantic, `Model.Config.strict`) | Turns every non-breaking producer schema addition into a consumer runtime error | Use permissive parsing (`extra="ignore"`) on event payloads |
| Calling `outbox.enqueue` outside a transaction (bare connection, no active transaction) | The event is not durably linked to the domain write; partial failures can produce an event with no corresponding domain record | Always call `outbox.enqueue` inside `pgcommon.run_in_tx`; never call it on a connection with no open transaction |

---

## When NOT to use this

This library targets **long-running platform services** that publish or consume domain events. It may be more than you need if:

- **One-off scripts or CLIs** — a plain `boto3`/`aioboto3` SNS call is simpler without the pool setup overhead.
- **Services that only consume a single event type** — if filtering, retries, and concurrency control are handled entirely by the SQS queue configuration, a thin wrapper is sufficient.
- **Non-AWS message brokers** — the library is SNS/SQS-specific. EventBridge, Kafka, and other backends are not supported without a new adapter.
- **Services that do not use PostgreSQL** — `pgcommon` is a hard dependency of this library regardless (the SQS consumer injects tenant GUC context on every message, not just the outbox — see ARCHITECTURE.md's Layer model). If your service has no database, you can still install and use the SNS publisher directly and accept at-most-once delivery; you simply never call `outbox.enqueue`/`outbox.Runner`.

---

## Installing

`eventcommon` publishes to public PyPI (the GitHub repository itself stays private — see [VERSIONING.md § Publishing to PyPI](VERSIONING.md#publishing-to-pypi) for why that split is deliberate):

```bash
uv add eventcommon
```

```bash
pip install eventcommon==0.1.0
```

Pin an exact version in production — see [Versioning and releases](#versioning-and-releases) for the pin-style guidance.

---

## Quick start

```python
import os

from eventcommon import metrics
from eventcommon.config import load_outbox, load_sns, load_sqs, log_warnings, runner_config_from_env
from eventcommon.consumer import SQSConsumer
from eventcommon.outbox import Runner, apply_schema
from eventcommon.publisher import SNSPublisher
from pgcommon import create_pool
from pgcommon.migrate import Runner as MigrateRunner


async def main() -> None:
    # 1. Register Prometheus metrics once.
    metrics.init(os.environ["APP_NAME"], os.environ["BUILD_VERSION"])

    # 2. Open the connection pool for the outbox runner.
    pool = await create_pool(dsn=os.environ["DATABASE_URL"])

    # 3. Apply the outbox schema migration.
    migrate_runner = MigrateRunner(dsn=os.environ["DATABASE_URL"])
    await apply_schema(migrate_runner)

    # 4. Construct the SNS publisher.
    sns_env = load_sns()
    publisher = SNSPublisher(sns_env.to_config(logger=logger))

    # 5. Start the outbox runner (delivers events asynchronously).
    outbox_env = load_outbox()
    log_warnings(outbox_env.warnings, logger=logger)
    runner = Runner(runner_config_from_env(outbox_env, pool, publisher, logger=logger))
    runner_task = asyncio.create_task(runner.start())

    # 6. Construct and start the SQS consumer.
    sqs_env = load_sqs()
    log_warnings(sqs_env.warnings, logger=logger)

    async def handle_event(env): ...  # handle event

    consumer = SQSConsumer(
        sqs_env.to_config(logger=logger), handle_event, **sqs_env.consumer_options()
    )
    consumer_task = asyncio.create_task(consumer.start())

    try:
        await asyncio.gather(runner_task, consumer_task)
    finally:
        await runner.stop()
        await consumer.stop()
        await pool.close()
```

---

## Event envelope

`Envelope[T]` is the canonical wire format for all inter-service events — identical on the wire to the Go implementation's `Envelope[T]`. Terminology used consistently throughout this document:

| Term | Meaning | Example |
|---|---|---|
| **event** | The logical domain occurrence and the `Envelope` that carries it | "publish an event", "the event's `tenant_id`" |
| **message** | The SNS/SQS infrastructure unit — what the broker delivers, retries, and deletes | "SQS message", "delete_message", "`MessageGroupId`" |
| **outbox record** | A row in `outbox_events` or `outbox_dead_letters` — the durable Postgres representation of an event pending publish | "claim outbox records", "failed outbox record" |

```json
{
  "id":             "01926e4f-...",
  "type":           "iam.user.created",
  "source":         "platform-iam",
  "specversion":    "1",
  "tenant_id":      "acme",
  "trace_id":       "4bf92f3577b34da6a3ce929d0e0e4736",
  "correlation_id": "...",
  "subject":        "users/01926e4f-...",
  "actor":          "admin@acme.com",
  "ip_address":     "203.0.113.42",
  "user_agent":     "Mozilla/5.0 (compatible; XPert/1.0)",
  "dataschema":     "550e8400-e29b-41d4-a716-446655440000",
  "time":           "2026-05-27T12:00:00Z",
  "data":           { ... }
}
```

### Creating envelopes

```python
from dataclasses import dataclass

from eventcommon.envelope import new_envelope


@dataclass
class UserCreatedPayload:
    user_id: str
    email: str


# From an HTTP handler — always carry trace and tenant from the request context.
rc = request_context(request)  # raises/returns None for unauthenticated requests
env = new_envelope(
    "iam.user.created",
    "platform-iam",
    UserCreatedPayload(user_id=user.id, email=user.email),
    tenant_id=rc.tenant_id,
    trace_id=rc.trace_id,  # REQUIRED for cross-service trace continuity — see note below
    schema_version="1",
)
```

> ⚠️ **Always pass `trace_id=rc.trace_id` from the request context.** If `trace_id` is omitted:
>
> - The OTel `sqs.receive` span on the consumer side has no parent — it starts a new disconnected trace, severing the link back to the originating HTTP request
> - A production incident that begins as an HTTP 500 in service A and manifests as a data anomaly in service B becomes impossible to correlate in Tempo or Grafana without manual log searching
> - The `trace_id` field in handler log lines will be empty, breaking the log-to-trace pivot that structured logging provides
>
> For background jobs where there is no inbound HTTP trace, generate a new trace ID at job startup and pass it consistently across all events emitted in that job run — this at minimum groups all events from the same run together.

**System-level events** (background jobs, scheduled tasks, cross-tenant operations) that are not scoped to any single tenant should use `SYSTEM_TENANT_ID` instead:

```python
from eventcommon.envelope import SYSTEM_TENANT_ID, new_envelope

# Background job — not associated with any specific tenant.
env = new_envelope(
    "billing.invoices.generated",
    "billing-worker",
    payload,
    tenant_id=SYSTEM_TENANT_ID,
)
```

> ⚠️ **Use `SYSTEM_TENANT_ID` only when the event genuinely has no tenant scope** — scheduled batch jobs, cross-tenant reconciliation, or platform-level operational events. Do not use it to avoid looking up a tenant ID, to simplify a code path, or because the tenant is "unknown at call time" (that last case means the upstream context is missing and should be fixed, not papered over).
> Misuse bypasses the tenant isolation that `pgcommon`'s RLS policies enforce on the consumer side. A consumer handler that receives a `system` tenant ID will execute DB queries without a tenant GUC set, which causes RLS policies to either reject the query or return the wrong row set. The failure is silent at the event level and only surfaces as incorrect data in the consuming service.

**`source` field contract:** identifies the producing service. Rules:

| Rule | Detail |
|---|---|
| Stable and globally unique | Use the canonical service name: `platform-iam`, `billing-service`, `inventory-worker`. It must be unique across all services in the organisation |
| Immutable across deployments | Do not derive it from hostname, pod name, or any runtime variable — it must be the same value in every environment |
| Environment-free | Never append `-dev`, `-staging`, `-prod`, or a region suffix. The environment is implicit in which AWS account/topic the event lands on |
| Set as a constant | Define it once as a module-level constant in the service (`SERVICE_NAME = "platform-iam"`) and pass it to `new_envelope` and `metrics.init` from that single source of truth |

**`event_type` convention:** `<domain>.<entity>.<past-tense-verb>[.v<N>]` — e.g. `iam.user.created`, `billing.invoice.settled`. The `.v<N>` segment is **only added for breaking payload changes** (v1 is implicit). See [EVENT_SCHEMA_GOVERNANCE.md](./EVENT_SCHEMA_GOVERNANCE.md) for the full naming rules, the event type registry, and payload evolution tiers.

**Event types are immutable once published.** Additive fields (optional, defaulted) are allowed in the same type. Removing, renaming, changing the type, or changing the meaning of an existing field requires a new event type (e.g. `iam.user.created.v2`).

**Consumers must ignore unknown payload fields — but unlike Go, this is not automatic in Python.** Always construct typed payloads via `parse_envelope(data, payload_type=YourDataclass)`, never `YourDataclass(**raw_dict)` directly — the former filters unrecognised keys before construction, the latter raises `TypeError` the moment a producer adds a harmless new field. See [EVENT_SCHEMA_GOVERNANCE.md § The Python unknown-field trap](./EVENT_SCHEMA_GOVERNANCE.md#the-python-unknown-field-trap).

**`schema_version`** records the payload contract version in the envelope (wire field `specversion`, CloudEvents-aligned). Set `"1"` at inception. Increment **only** when both conditions are true: (1) a new optional field was added **and** (2) at least one consumer needs to branch logic based on whether that field is present. Consumers that receive an unrecognised `specversion` should log a warning and drop the message rather than silently misparsing it. If `specversion` is absent on the wire, treat it as `"1"` for backward compatibility.

### Serialisation

```python
# Marshal to JSON wire format.
data = env.to_json()

# Unmarshal and validate required fields (raises EnvelopeIDRequiredError etc. on a missing field).
env = parse_envelope(data, payload_type=UserCreatedPayload)
```

### Payload typing

`Envelope[T]` is generic, but two concrete forms appear throughout the codebase with different purposes:

| Form | When to use |
|------|-------------|
| `Envelope[dict[str, Any]]` (raw JSON) | Publishing (`Publisher.publish` accepts this form); handler dispatch; routing layers that inspect headers but not the payload; audit loggers; DLQ handlers that forward without parsing |
| `Envelope[YourDataclass]` | `parse_envelope(data, payload_type=YourDataclass)` when you own the event type and want the payload pre-parsed; mock injection in unit tests |

**The transport boundary always uses raw JSON.** The `Publisher` and `Handler` types are both fixed to an untyped payload — this is intentional. It keeps the routing layer (SQS consumer, outbox runner) free of business-type imports.

Typed payload access happens *inside* the handler after the transport layer has delivered the envelope:

```python
# ✅ Correct pattern — transport receives the raw envelope; handler parses to a typed dataclass
async def handle_user_created(env: Envelope[dict[str, Any]]) -> None:
    payload = UserCreatedPayload(**env.data)
    # payload is now typed; env.id / env.tenant_id / env.trace_id are always available directly
    await repo.create_user(payload)
```

**Routing handlers** that need to inspect the event type and forward without parsing should stay untyped throughout — no parse cost, no coupling to payload dataclasses:

```python
# Fan-out router — dispatches by event type without ever touching the payload
async def route(env: Envelope[dict[str, Any]]) -> None:
    if env.type == "iam.user.created":
        await iam_consumer.publish(env)  # forward as-is
    elif env.type == "billing.invoice.settled":
        await billing_consumer.publish(env)
    # unknown type: discard
```

### Envelope ID

`Envelope.id` is a **UUID v7** (time-ordered, collation-friendly in Postgres B-tree indexes). Use it as an idempotency key on the consumer side. `eventcommon._internal` generates UUIDv7 itself for portability down to Python 3.12 (the stdlib `uuid` module only gained native `uuid7()` support in 3.14).

### Envelope compatibility guarantees

The `id`, `type`, `source`, and `time` fields are **stable** — always present, never removed or renamed, format frozen within `v0.x`/`v1.x`. The remaining fields (`tenant_id`, `trace_id`, `correlation_id`, `specversion`, `subject`, `actor`, `dataschema`) are **contextual** — present when set, never removed. The library may add new optional fields in MINOR releases; existing consumers are unaffected. See [ARCHITECTURE.md § Envelope compatibility guarantees](ARCHITECTURE.md#envelope-compatibility-guarantees) for the full per-field stability class table.

---

## SNS Publisher

```python
from eventcommon.publisher import SNSConfig, SNSPublisher

publisher = SNSPublisher(
    SNSConfig(
        topic_arn=os.environ["SNS_TOPIC_ARN"],  # required — raises if empty or invalid ARN format
        region=os.environ["AWS_REGION"],
        endpoint_url=os.environ.get(
            "AWS_ENDPOINT_URL"
        ),  # set to http://localhost:4566 for LocalStack
        logger=logger,
    )
)
```

### Publishing a single event

```python
env = new_envelope(
    "iam.user.created",
    "platform-iam",
    payload,
    tenant_id=rc.tenant_id,
    trace_id=rc.trace_id,
)
await publisher.publish(env)
```

### Batch publishing

```python
# publish_batch splits automatically at the SNS hard limit of 10.
await publisher.publish_batch(envelopes)
# Partial failures raise a BatchError listing per-message errors.
```

> ⚠️ **`publish_batch` is not atomic and must never be used for critical domain events.** SNS batch publish is a best-effort call: some messages in the batch may succeed while others fail within the same call. A raised error does not mean the entire batch was rejected. For any event tied to a state change or a database write, use `outbox.enqueue` — partial batch failure combined with a process crash leaves no recovery path.

Callers must inspect the error to distinguish partial from total failure:

```python
from eventcommon.errors import BatchError

try:
    await publisher.publish_batch(envelopes)
except BatchError as exc:
    # Partial failure — some messages published, some did not.
    # exc.failures is list[BatchFailure(index: int, error: Exception)].
    # Successfully published messages must NOT be retried.
    failed = [envelopes[f.index] for f in exc.failures]
    for f in exc.failures:
        logger.warning(
            "batch publish failure",
            index=f.index,
            event_id=envelopes[f.index].id,
            error=str(f.error),
        )
    # Retry failed messages individually or hand off to the outbox for durable retry.
    await publisher.publish_batch(failed)
except Exception as exc:
    # Non-BatchError: total failure (network error, auth failure, etc.) — safe to retry the full batch.
    raise RuntimeError(f"publish_batch: {exc}") from exc
```

**Key rules:**

| Rule | Reason |
|---|---|
| Never retry the full batch on a `BatchError` | Messages that succeeded would be published a second time, creating duplicates |
| Always extract failed indices from `exc.failures` | The batch index is the only link between a failure and the original envelope |
| Prefer the outbox for durable retry | If the caller cannot afford to lose messages on process crash, use `outbox.enqueue` instead of `publish_batch` |
| `publish_batch` is appropriate for idempotent or at-most-once scenarios | Background notifications, cache invalidation signals, or any event where a missed or duplicate delivery is acceptable |

### FIFO topics

```python
publisher = SNSPublisher(
    SNSConfig(
        topic_arn="arn:aws:sns:us-east-1:123456789012:my-topic.fifo",
        region="us-east-1",
        logger=logger,
    ),
    message_group_id=lambda env: env.tenant_id,  # group per tenant for ordered delivery
)
```

When `topic_arn` ends in `.fifo`, `message_group_id` is required; `MessageDeduplicationId` defaults to `Envelope.id` (requires content-based deduplication disabled at the topic level).

> ⚠️ **FIFO does not mean exactly-once delivery end-to-end.** This is the most common FIFO misconception:
>
> | Layer | What FIFO guarantees | What it does NOT guarantee |
> |-------|----------------------|---------------------------|
> | SNS | Deduplicates identical `MessageDeduplicationId` values within a **5-minute window** | Deduplication outside that window; delivery to SQS is still at-least-once |
> | SQS | Ordered delivery within a `MessageGroupId`; no duplicate delivery **within a single consumer session** | Protection against redelivery after a visibility timeout expires or a consumer crashes mid-handler |
> | End-to-end | Ordered, deduplicated fan-out from SNS to SQS | That the consumer handler runs exactly once — it will not if the handler crashes after processing but before `delete_message` |
>
> **FIFO gives you ordering. Idempotency still gives you safety.** Use `message_group_id` to enforce processing order within a group (e.g. per-tenant, per-aggregate). Use `Envelope.id` + `INSERT ... ON CONFLICT DO NOTHING` to make the handler safe to run twice.

### Message attributes

`EventType`, `TenantID`, `Source`, `EventID`, and `Subject` (when non-empty) are set as SNS message attributes. This enables SQS subscription filter policies that scope queues to specific event types, tenants, or resource subjects without deserialising the message body. `Actor` is an audit-trail field and is not forwarded as an SNS attribute.

---

## SQS Consumer

```python
from eventcommon.consumer import SQSConfig, SQSConsumer


async def handle_event(env: Envelope[dict[str, Any]]) -> None:
    # eventcommon has already propagated tenant_id into the pgcommon GUC context —
    # pool.with_conn()/run_in_tx() automatically enforce RLS for this tenant.
    await handle(env)


consumer = SQSConsumer(
    SQSConfig(
        queue_url=os.environ["SQS_QUEUE_URL"],  # required
        region=os.environ["AWS_REGION"],
        endpoint_url=os.environ.get("AWS_ENDPOINT_URL"),
        max_messages=10,  # 1–10; defaults to 10
        wait_seconds=20,  # long-poll; defaults to 20
        logger=logger,
    ),
    handle_event,
    concurrency=5,
    visibility_timeout=30,  # seconds
)

task = asyncio.create_task(consumer.start())  # runs until stop() is called
...
await consumer.stop()  # graceful drain — waits up to 30s for in-flight handlers
```

### Handler contract

| Handler outcome | Consumer behaviour |
|---|---|
| Returns normally | SQS message deleted from the queue |
| Raises an exception | SQS message left visible; retried after the visibility timeout |
| Parse failure (`json.loads` on the message body) | Message deleted immediately; counted as `events_consumed_total{status=malformed}` |

**`visibility_timeout` limit:** SQS enforces a hard maximum of 12 hours. `SQSConsumer.__init__` raises if `visibility_timeout > 43200` seconds.

**Task contract:** each handler invocation runs in its own `asyncio.Task`, bounded by an `asyncio.Semaphore(concurrency)`. On `stop()`, the consumer stops receiving new messages and awaits in-flight handler tasks for up to `drain_timeout` — handlers are not cancelled during this window, so a handler that never returns will hold up shutdown. Set your own per-handler timeout (`asyncio.wait_for`) rather than relying on the consumer to enforce one.

### Error classification

The handler's outcome — return normally vs. raise — is the only signal the consumer uses to decide retry-or-delete. **Getting this backwards determines whether a broken message sits in SQS forever or gets silently discarded.**

| Error class | Examples | Handler action | Outcome |
|---|---|---|---|
| **Transient** | DB connection timeout, downstream HTTP 503, network blip, lock contention | `raise` | Message left visible; SQS retries after visibility timeout; moves to DLQ after `MaxReceiveCount` |
| **Permanent** | Payload fails business validation, unknown `event_type` your handler cannot process, schema version too new to parse | Return normally + log at `WARNING`/`ERROR` | Message deleted immediately; no retry; no DLQ pressure |
| **Programming error** | `AttributeError`, `KeyError`, `TypeError` from a bug | `raise` (let it propagate) | Message retried; surfaced in `events_consumed_total{status=error}`; investigate immediately |
| **Unknown / unexpected** | Error from a dependency you haven't classified | `raise` | Retry by default; safe to escalate to DLQ if unresolved |

> ⚠️ **Misclassifying permanent errors as transient is the most common handler mistake.** If a malformed payload causes the handler to raise, SQS retries it `MaxReceiveCount` times, then pushes it to the DLQ — where it will sit indefinitely. Return normally (and log) for any message that cannot succeed on retry regardless of how many times it is delivered.

**Classification pattern:**

```python
async def handle_user_created(env: Envelope[dict[str, Any]]) -> None:
    try:
        payload = UserCreatedPayload(**env.data)
    except (TypeError, KeyError) as exc:
        # Permanent: malformed payload will never parse correctly on retry
        logger.error(
            "failed to parse UserCreatedPayload — discarding", event_id=env.id, error=str(exc)
        )
        return

    if not payload.user_id:
        # Permanent: business validation failure; retrying won't fix a missing user_id
        logger.warning("UserCreatedPayload missing user_id — discarding", event_id=env.id)
        return

    try:
        await repo.create_user(payload)
    except DatabaseError as exc:
        # Transient: DB errors may resolve; let SQS retry
        raise RuntimeError(f"create_user: {exc}") from exc
```

### Poison messages

A **poison message** is structurally valid JSON — it parses without error — but cannot be processed successfully regardless of how many times it is retried. It is distinct from a transient error (which may succeed on retry) and from a malformed message (which fails JSON parsing).

Common causes:

| Cause | Example |
|---|---|
| Invalid business state | `user_id` references a user that was deleted before the event arrived |
| Illegal state transition | An `order.shipped` event arrives for an order already in `cancelled` state |
| Missing precondition | A `payment.settled` event arrives but no corresponding `payment.created` exists |
| Schema version too new | `specversion: "5"` but this consumer only understands up to `"3"` |

**Handling pattern:**

```python
async def handle_order_shipped(env: Envelope[dict[str, Any]]) -> None:
    try:
        payload = OrderShippedPayload(**env.data)
    except (TypeError, KeyError):
        return  # malformed — discard (see Error classification)

    order = await repo.get_order(payload.order_id)
    if order is None:
        # Poison: the order doesn't exist and never will on retry.
        logger.error(
            "poison message: order not found",
            event_id=env.id,
            event_type=env.type,
            order_id=payload.order_id,
            trace_id=env.trace_id,
            tenant_id=env.tenant_id,
        )
        return  # drop — do NOT raise

    if order.status == "cancelled":
        # Poison: invalid state transition; retrying will never change the order status.
        logger.warning(
            "poison message: illegal state transition — order already cancelled",
            event_id=env.id,
            order_id=payload.order_id,
        )
        # Optional: forward to an audit or failure topic for later investigation.
        await audit_publisher.publish(
            new_envelope(
                "platform.event.poison",
                "order-consumer",
                env.data,
                tenant_id=env.tenant_id,
                correlation_id=env.id,
            )
        )
        return  # drop

    await repo.mark_shipped(order.id)
```

**Decision rules:**

1. **Return normally, don't raise** — raising retries the message; a semantically invalid message will never become valid on retry.
2. **Log at `ERROR` or `WARNING` with the full correlation set** (`event_id`, `event_type`, `trace_id`, `tenant_id`) — poison messages are silent data anomalies; the log is the only record that something was discarded.
3. **Optionally forward to an audit or failure topic** if the business impact is high enough to warrant investigation or replay tooling.
4. **Do not forward sensitive payload data** (PII, credentials) to an audit topic without confirming the destination has appropriate access controls and retention policies.

### Implementing idempotency

SQS delivers messages **at least once**. A handler may be called more than once for the same `Envelope.id` due to visibility timeout expiry, a network error between handler completion and `delete_message`, or a consumer restart mid-batch. Use `Envelope.id` as the idempotency key.

#### Pattern 1 — Postgres unique constraint (recommended)

```sql
CREATE TABLE IF NOT EXISTS processed_events (
    event_id     TEXT        PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
```

```python
async def handle_user_created(env: Envelope[dict[str, Any]]) -> None:
    async with pgcommon.run_in_tx(pool) as conn:
        row = await conn.fetchrow(
            "INSERT INTO processed_events (event_id) VALUES ($1) "
            "ON CONFLICT (event_id) DO NOTHING RETURNING event_id",
            env.id,
        )
        if row is None:
            return  # already processed by a previous delivery — safe no-op

        payload = UserCreatedPayload(**env.data)
        await repo.create_user(conn, payload)
```

**Why inside the same transaction?** If the side-effect write succeeds but the transaction rolls back before `processed_events` commits, the next delivery re-inserts the row and re-runs the side effect — correct behaviour. If `processed_events` commits but the side-effect write fails, the transaction rolls back atomically — both are absent, and the next delivery retries both.

#### Pattern 2 — Upsert-based idempotency

Use `ON CONFLICT DO UPDATE` with a sentinel column when the side effect is a row update rather than an insert — see [ARCHITECTURE.md](ARCHITECTURE.md) for the full SQL pattern.

#### Pattern 3 — Redis SETNX (for non-Postgres handlers)

```python
async def handle_welcome_email(env: Envelope[dict[str, Any]]) -> None:
    key = f"processed:{env.id}"
    was_set = await redis.set(
        key, "1", ex=5 * 24 * 3600, nx=True
    )  # 5-day TTL matches SQS max retention
    if not was_set:
        return  # already processed
    await email_svc.send_welcome(env.data)
```

**Caveat:** Redis `SETNX` is not durable across Redis restarts without AOF persistence. For financial or critical events, Pattern 1 is safer.

#### Common mistakes

| Mistake | Consequence | Fix |
|---|---|---|
| Check `processed_events` in a separate query before the transaction | TOCTOU race — two concurrent deliveries both pass the check and both execute | Always check inside the same transaction as the side-effect write |
| No idempotency at all | Duplicate charges, emails, or rows on any redelivery | Use Pattern 1 or 2 |
| Idempotency outside the transaction | DB write succeeds, then `processed_events` insert fails → unguarded on next delivery | Idempotency claim and side-effect write must commit atomically |
| Never pruning `processed_events` | Table grows unboundedly | Nightly job: `DELETE ... WHERE processed_at < NOW() - INTERVAL '5 days'` |

### Side-effect classification

| Side-effect class | Examples | Duplicate risk | Required guard |
|---|---|---|---|
| **DB write** | `INSERT`, `UPDATE`, state machine transition, balance change | High | Wrap in `pgcommon.run_in_tx` with `processed_events ON CONFLICT DO NOTHING` |
| **External HTTP call (mutating)** | Charge a payment, call a third-party API, trigger a webhook | High | Pass `Envelope.id` as the idempotency key in the downstream request |
| **Email / SMS / push notification** | Welcome email, OTP, shipping confirmation | Medium | Gate with `processed_events` before sending |
| **Cache write / invalidation** | Redis `SET`, CDN purge, in-memory state update | Low | No guard needed; if cache drives a downstream decision, guard the underlying DB write instead |
| **Read-only / observability** | Incrementing a counter, logging, emitting a metric | None | No guard needed |

---

## Transactional outbox

The outbox pattern eliminates dual-write risk: the event is written **inside the business transaction** alongside the domain mutation. If the transaction rolls back, the event is never published. The runner delivers asynchronously with at-least-once guarantee. Outbox does not guarantee global ordering — use FIFO topics with a stable `message_group_id` when ordering is required.

### Wiring

```python
from eventcommon.outbox import Runner, RunnerConfig, apply_schema
from pgcommon.migrate import Runner as MigrateRunner

# 1. Apply outbox schema migration (once at startup).
migrate_runner = MigrateRunner(dsn=os.environ["DATABASE_URL"])
await apply_schema(migrate_runner)

# 2. Construct the outbox runner.
runner = Runner(
    RunnerConfig(
        pool=pool,  # pgcommon.Pool — required
        publisher=publisher,  # eventcommon Publisher — required
        logger=logger,
        poll_interval=5.0,  # seconds
        batch_size=50,
        max_attempts=5,
        # Production-hardening options (all have safe defaults):
        publish_concurrency=1,  # default — SNS PublishBatch (10/API call); raise only when SNS latency-bound
        publish_timeout=10.0,  # per-record publish timeout
        drain_timeout=30.0,  # stop() waits up to this for the in-flight batch
        startup_jitter=0.2,  # random delay before first poll; desyncs replicas on rolling restart
        # To load from env: runner_config_from_env(load_outbox(), pool, publisher, logger=logger)
    )
)
runner_task = asyncio.create_task(runner.start())  # runs until stop() is called
...
if (
    err := await runner.stop()
):  # returns an error message if the in-flight batch did not drain in time
    logger.warning("outbox runner drain timeout — records retry after lease expiry", error=err)

# Optional: gate the Kubernetes readiness probe until the first poll succeeds.
await asyncio.wait_for(runner.ready.wait(), timeout=30)
```

**Config defaults:** `poll_interval` 5s · `batch_size` 50 · `max_attempts` 5 · `publish_concurrency` 1 · `publish_timeout` 10s · `drain_timeout` 30s · `claim_lease_duration` 10m · `startup_jitter` 0.

### Enqueueing inside a transaction

```python
async with pgcommon.run_in_tx(pool) as conn:
    await repo.save_user(conn, user)
    env = new_envelope(
        "iam.user.created",
        "platform-iam",
        {"user_id": user.id},
        tenant_id=rc.tenant_id,
        trace_id=rc.trace_id,
    )
    await outbox.enqueue(conn, env)
```

`enqueue` validates the envelope (non-empty `id`/`type`/`source`; non-zero `time`) and rejects payloads whose serialised size exceeds **240 KB** — staying under the SNS 256 KB hard limit so an outbox record that could never publish is never persisted.

### Payload size guidelines

| Tier | Limit | Meaning |
|------|-------|---------|
| **Hard limit** | 256 KB | SNS `Publish` API rejects messages above this — non-negotiable |
| **Enforced limit** | 240 KB | `outbox.enqueue` rejects at this threshold |
| **Recommended** | ≤ 64 KB | Comfortable budget for typical domain events |

When the payload is inherently large (binary content, rendered templates, bulk export rows), store the data in object storage and put only a reference in the envelope — the consumer fetches the object after receiving the event.

### Dead letters

> **"Dead letters" means two different things** — do not confuse them. Outbox dead letters are **publish failures** stored in Postgres. SQS DLQ entries are **consumer processing failures** stored in a separate SQS queue. See [ARCHITECTURE.md § Failure lifecycle](ARCHITECTURE.md#failure-lifecycle) for the complete side-by-side timeline and operational runbook.

Failed outbox records that exhaust `max_attempts` move to `outbox_dead_letters` — a queryable Postgres table.

```python
# Step 1 — Inspect before acting.
records = await runner.list_dead_letters(DLQFilter(tenant_id="acme"), 50)

# Step 2a — Replay after fixing the root cause.
n = await runner.reprocess_dead_letters(100)  # unfiltered: replay all
n = await runner.reprocess_dead_letters_with(
    DLQFilter(event_type="billing.invoice.settled", tenant_id="acme"), 100
)  # filtered

# Step 2b — Discard poison pills that can never succeed.
n = await runner.discard_dead_letters(DLQFilter(event_type="legacy.sync.requested"), 1000)
```

`DLQFilter` fields are all optional (`None` = match all): `event_type: str | None`, `tenant_id: str | None`, `failed_before: datetime | None`.

**Retryable failures** (SNS throttling: `ThrottlingException`, `ServiceUnavailable`, `InternalFailure`, `RequestTimeout`) do not count toward `max_attempts` — the outbox uses `threshold = max_attempts + 1` for these errors. A period of SNS unavailability will not dead-letter records that are otherwise healthy.

### Pruning published records

```python
# Delete published records older than 7 days, up to 1000 per call.
n = await runner.prune_published(older_than=timedelta(days=7), limit=1000)
```

`older_than` must be long enough that all consumers have processed the event before the row is deleted. 7 days covers most SLA windows.

### Replay guarantees

Replay carries no stronger delivery guarantees than the original delivery path. Ordering is not preserved (replayed events re-enter the back of the poll queue), the same `Envelope.id` is presented again (idempotency guards must be in place), and replay runs concurrently with live traffic. A handler that is not idempotent is not replay-safe — and therefore not production-ready.

---

## HMAC helpers

### When to use HMAC

HMAC-SHA256 provides **message authenticity and integrity** — proof that the payload was produced by a party that holds the shared key and has not been tampered with in transit. It does **not** provide confidentiality or replay protection.

| Scenario | HMAC required | Why |
|----------|:---:|-----|
| Receiving webhooks from external systems (Stripe, GitHub, etc.) | **Yes** | The channel is the public internet; the SNS/SQS IAM boundary does not apply |
| Cross-service call over an internal HTTP endpoint (not SNS/SQS) | **Yes** | Service-to-service HTTP has no built-in message-level auth |
| Zero-trust internal network where service identity is not enforced at the infra layer | **Yes** | HMAC adds app-layer authentication even when mTLS or IRSA is absent |
| Events flowing exclusively over SNS → SQS within the same AWS account and IAM boundary | **No** | SNS delivery is authenticated by IAM policies |
| Events signed by the outbox runner and delivered to an SQS queue you own | **No** | The publisher operates under your service's IAM role |

```python
from eventcommon import hmac as event_hmac

key = os.environ["WEBHOOK_SECRET"].encode()  # must be ≥ 32 bytes

# Sign
sig = event_hmac.sign(key, payload)

# Verify — constant-time; returns False on mismatch, never raises for a bad signature
if not event_hmac.verify(key, payload, sig):
    raise ValueError("invalid signature")

# Sign/verify a full envelope (serialises to canonical JSON first)
sig = event_hmac.sign_envelope(key, env)
ok = event_hmac.verify_envelope(key, env, sig)
```

> `verify` uses `hmac.compare_digest` (constant-time) — never replace with string `==`. **Key length:** `sign` raises `KeyTooShortError` for keys < 32 bytes.

**Rules:** always use `verify_envelope` (not `verify`) when checking a full envelope. Rotate keys by accepting both the current and previous key for a short window. Keys must be ≥ 32 bytes and stored in a secrets manager — never in source-controlled environment variables.

---

## Observability — Prometheus and OTel

Both are **optional**. The publisher, consumer, and outbox runner work without calling `metrics.init` or having an OTel provider registered.

### Prometheus metrics

```python
from eventcommon import metrics

metrics.init(os.environ["APP_NAME"], os.environ["BUILD_VERSION"])  # idempotent — first caller wins

# For isolated test registries:
metrics.init_with_registry("test-svc", "v0.0.0", CollectorRegistry())
```

Registered metrics (identical names to the Go implementation, so Python and Go services feeding the same Grafana dashboards use one metric namespace):

| Metric | Type | Labels | Description |
|---|---|---|---|
| `events_published_total` | Counter | `service`, `topic`, `event_type`, `status` | SNS publish attempts |
| `events_publish_duration_seconds` | Histogram | `service`, `topic`, `event_type` | SNS publish latency |
| `events_consumed_total` | Counter | `service`, `queue`, `event_type`, `status` | SQS messages processed (`status` = `success`/`error`/`malformed`) |
| `events_consume_duration_seconds` | Histogram | `service`, `queue`, `event_type` | Handler execution latency |
| `outbox_pending_total` | Gauge | `service` | Unpublished records in `outbox_events` |
| `outbox_leased_total` | Gauge | `service` | Records currently claimed (leased) by a runner |
| `outbox_published_total` | Counter | `service`, `event_type`, `status` | Records published by the runner |
| `outbox_attempts_total` | Counter | `service`, `event_type` | Total publish attempts by the runner |
| `outbox_dead_letters_total` | Counter | `service`, `event_type` | Records moved to `outbox_dead_letters` — alert on `rate() > 0` |
| `outbox_dead_letters_reprocessed_total` | Counter | `service` | Dead-letter records re-queued |
| `outbox_dead_letters_discarded_total` | Counter | `service` | Dead-letter records permanently deleted |
| `sqs_receive_errors_total` | Counter | `service`, `queue` | `receive_message` errors — alert on `rate() > 0` |
| `sqs_delete_errors_total` | Counter | `service`, `queue` | `delete_message` errors — non-zero rate causes duplicate delivery |
| `outbox_poll_errors_total` | Counter | `service` | Outbox poll cycle errors — triggers exponential backoff |
| `outbox_mark_published_errors_total` | Counter | `service` | `mark_published` failures after a successful SNS delivery |
| `events_oversized_event_type_label_total` | Counter | `service` | `event_type` values > 128 bytes, replaced with `"__oversized__"` |

### OpenTelemetry

OTel is **always initialised by the consuming service** (e.g. `fastapicommon.tracing.init_from_env()`). `eventcommon` calls `trace.get_tracer("eventcommon")` and produces no-op spans if no provider is registered.

**SNS publish span:** `sns.publish` with `messaging.system=aws_sns`, `messaging.destination`, `messaging.message_id`.

**SQS receive span:** `sqs.receive` with `messaging.system=aws_sqs`, `messaging.operation=process`, **linked to the publisher's trace** via `Envelope.trace_id`.

### Logging correlation

**Always include these four fields on every log line inside a handler or publisher:** `event_id` (`env.id`), `event_type` (`env.type`), `trace_id` (`trace_id_from_context()`), `tenant_id` (`env.tenant_id`).

```python
async def handle_user_created(env: Envelope[dict[str, Any]]) -> None:
    log = logger.bind(
        event_id=env.id,
        event_type=env.type,
        trace_id=trace_id_from_context(),
        tenant_id=env.tenant_id,
    )
    log.info("handling event")

    payload = UserCreatedPayload(**env.data)
    await repo.create_user(payload)

    log.info("event processed")
```

#### What NOT to log

| Do not log | Reason |
|-----------|--------|
| `env.data` (raw or parsed) | May contain PII, payment data, or credentials — log `event_id` instead and look up the payload if needed |
| Full error chains that include DSNs or URLs | `DATABASE_URL` and `SNS_TOPIC_ARN` may appear in wrapped errors — truncate or sanitise before logging |
| `env.trace_id` as an alert label | Trace IDs are high-cardinality — use them in log lines but not as alert labels |

---

## Backpressure considerations

Under sustained load, the SQS receive buffer and the outbox `outbox_events` table both grow if consumers or the outbox runner can't keep up.

### Consumer-side (SQS)

**Tuning order:** 1) raise `concurrency` — costs only memory and an asyncio task; 2) scale pods horizontally — SQS distributes messages across them naturally; 3) increase `SQS_MAX_MESSAGES` (up to 10) only if handler latency is I/O-bound; 4) increase `SQS_VISIBILITY_TIMEOUT` if handlers legitimately take longer than the current timeout.

### Outbox-side (Postgres → SNS)

**Tuning order:** 1) decrease `OUTBOX_POLL_INTERVAL`; 2) increase `OUTBOX_BATCH_SIZE` (keep below 100 to avoid long-held Postgres locks); 3) scale runner pods horizontally (`SKIP LOCKED` handles distribution with no coordination); 4) check SNS publish errors — a rising `outbox_attempts_total`/`outbox_published_total` gap usually indicates throttling, not a runner configuration problem.

> **Do not increase batch size before scaling pods on either side.** Horizontal scaling is almost always preferable to a larger batch/receive size.

---

## Recommended production defaults

```bash
# SQS Consumer
SQS_MAX_MESSAGES=10          # Always set to max (SQS hard limit)
SQS_WAIT_SECONDS=20          # Long-poll at max
SQS_VISIBILITY_TIMEOUT=60    # seconds — comfortably above your p99 handler latency; 30s default is tight
SQS_CONCURRENCY=5            # Start here; scale up to ~10 before considering horizontal pod scaling

# Outbox runner
OUTBOX_POLL_INTERVAL=2       # seconds — 5s default is conservative
OUTBOX_BATCH_SIZE=50         # 50-100 is a good range
OUTBOX_MAX_ATTEMPTS=5        # Default is fine unless your SNS target has known transient outages
```

> **`SQS_VISIBILITY_TIMEOUT` is the most commonly under-set value.** If your handler calls a slow DB query or downstream HTTP endpoint, the default `30s` may expire before the handler finishes, causing the message to be delivered twice. Set it to 2–3× your p99 handler duration.

### What to monitor on day one

| Metric | Alert threshold | Action |
|---|---|---|
| `outbox_pending_total` | > 500 sustained for 5 min | Lower `OUTBOX_POLL_INTERVAL`; add runner pods |
| `ApproximateNumberOfMessages` (CloudWatch) | > 1000 sustained | Raise `SQS_CONCURRENCY`; add consumer pods |
| `events_consumed_total{status="error"}` | > 1% error rate | Inspect handler errors; check DLQ depth |
| `outbox_published_total{status="failed"}` / total | > 1% | Check SNS reachability; inspect `outbox_dead_letters` |

---

## Configuration reference

| Variable | Default | Notes |
|---|---|---|
| `AWS_REGION` | `us-east-1` | Applies to both SNS and SQS clients |
| `SNS_TOPIC_ARN` | — | Required for SNS publisher |
| `SQS_QUEUE_URL` | — | Required for SQS consumer |
| `SQS_MAX_MESSAGES` | `10` | 1–10; SQS hard limit |
| `SQS_WAIT_SECONDS` | `20` | Long-poll duration |
| `SQS_VISIBILITY_TIMEOUT` | `30` | Seconds |
| `SQS_CONCURRENCY` | `1` | Parallel handler tasks |
| `SQS_MAX_RECEIVE_COUNT` | `0` (unset) | Pairs with a `dead_letter_handler` |
| `OUTBOX_POLL_INTERVAL` | `5` | Seconds |
| `OUTBOX_BATCH_SIZE` | `50` | Records per poll cycle |
| `OUTBOX_MAX_ATTEMPTS` | `5` | Before moving to dead-letter |
| `OUTBOX_CLAIM_LEASE_DURATION` | `600` | Seconds; how long a claimed record is hidden from other runners |
| `OUTBOX_STARTUP_JITTER` | `0` | Seconds; use `5`–`10` with multiple replicas |
| `OUTBOX_PUBLISH_CONCURRENCY` | `1` | Parallel publishes per poll cycle |
| `OUTBOX_PUBLISH_TIMEOUT` | `10` | Seconds; per-record publish timeout |
| `OUTBOX_DRAIN_TIMEOUT` | `30` | Seconds; `runner.stop()` wait bound |
| `DATABASE_URL` | — | Postgres DSN for the outbox runner (via `pgcommon`) |
| `AWS_ENDPOINT_URL` | — | Set to `http://localhost:4566` for LocalStack |
| `OTEL_SERVICE_NAME` | — | OTel resource attribute |
| `OTEL_EXPORTER_OTLP_ENDPOINT` | `localhost:4317` | |
| `APP_ENV` | — | `dev`/`development`/`local` enables OTel insecure mode |

**Wiring helpers (`eventcommon.config`):** `load_sns` / `load_sqs` / `load_outbox` / `load_otel` · `log_warnings` · `runner_config_from_env` · `sqs_consumer_options`

Copy `.env-example` to `.env` via `make setup` for local development.

---

## Error reference

| Error | Raised when |
|---|---|
| `eventcommon.errors.EnvelopeIDRequiredError` | `parse_envelope` or `outbox.enqueue` — missing `id` |
| `eventcommon.errors.EnvelopeTypeRequiredError` | `parse_envelope` or `outbox.enqueue` — missing `type` |
| `eventcommon.errors.EnvelopeSourceRequiredError` | `parse_envelope` or `outbox.enqueue` — missing `source` |
| `eventcommon.errors.KeyTooShortError` | `sign` / `sign_envelope` called with a key < 32 bytes |
| `eventcommon.errors.InvalidSignatureError` | Internal parse failure in `verify` |
| `eventcommon.errors.BatchTooLargeError` | Internal sentinel; `publish_batch` splits automatically at 10 — never raised to callers |
| `eventcommon.errors.RetryableError` | Wraps a transient SNS/AWS error so the outbox runner does not count it toward `max_attempts` |

---

## Testing in consuming services

`eventcommon.mock` ships ready-made, thread-safe test doubles so consuming services never need to stand up LocalStack or SNS just to run a unit test.

### Mock Publisher

```python
from eventcommon.mock import MockPublisher


async def test_place_order_publishes_event():
    pub = MockPublisher()
    svc = OrderService(pub)

    await svc.place_order(Order(id="ord-1", amount=99))

    assert len(pub.published) == 1
    assert pub.published[0].type == "order.placed"
    assert pub.published[0].tenant_id == "acme"


async def test_place_order_publish_error_is_handled():
    pub = MockPublisher()
    pub.set_error(RuntimeError("sns unavailable"))

    svc = OrderService(pub)
    with pytest.raises(RuntimeError):
        await svc.place_order(Order(id="ord-2", amount=50))

    assert pub.published == []  # nothing was recorded
```

| Method / attribute | What it does |
|--------|-------------|
| `await publish(env)` | Records the envelope; raises any error set via `set_error` |
| `await publish_batch(envs)` | Calls `publish` for each envelope |
| `published` | List of all recorded envelopes (thread-safe) |
| `set_error(err)` | Makes all subsequent `publish` calls raise `err` |
| `reset()` | Clears recorded envelopes and any configured error |

### Mock Consumer

```python
from eventcommon.mock import MockConsumer


async def test_notification_worker_handles_user_created():
    consumer = MockConsumer()
    worker = NotificationWorker(consumer)

    await consumer.start()  # no-op — no task needed
    worker.register_handlers()

    env = new_envelope(
        "iam.user.created",
        "platform-iam",
        {"user_id": "u-123", "email": "alice@example.com"},
        tenant_id="acme",
    )
    await consumer.inject(env)  # delivers directly — no SQS, no network

    assert worker.email_was_sent_to("alice@example.com")
```

| Method / attribute | What it does |
|--------|-------------|
| `await start()` | No-op; marks consumer as running |
| `await stop()` | No-op; marks consumer as stopped |
| `set_handler(fn)` | Registers the handler called by `inject` |
| `await inject(env)` | Delivers the envelope synchronously to the registered handler |
| `is_running` | Whether `start` has been called without a matching `stop` |

Both mocks satisfy the same `Publisher`/`Consumer` protocols as the real implementations, so they drop in wherever the real ones are used.

---

## Testing

```bash
make test-unit   # unit tests — no Docker required
make test-int    # integration tests — spins up LocalStack + Postgres (testcontainers)
make test-all    # unit + integration + e2e in one pytest run
make cover       # HTML coverage report
```

End-to-end tests cover the full outbox pipeline (Postgres → runner → SNS → SQS → consumer) and require Docker for LocalStack + Postgres via `testcontainers`. Run a single test by name:

```bash
uv run pytest tests/unit/test_envelope.py -k test_sign -v
uv run pytest tests/integration -m integration -k test_sns_publish_round_trip -v
```

---

## CI

**`ci.yml`** — triggered on every push and pull request to `main`: `validate` (format check, lint, `mypy --strict`, `pip-audit`, full test suite) → `coverage` (Python 3.12/3.13/3.14 matrix) → `build` (wheel + sdist) → `interop` (cross-language compatibility against `platform-events`).

**`release.yml`** — triggered on `v*.*.*` tags: same validate/coverage/build pipeline, then publishes to PyPI (Trusted Publisher, no stored token) and creates a GitHub Release from the tagged `CHANGELOG.md` section.

**`validate.yml`** — reusable workflow fanning out to `quality.yml` (lint/type/format/audit) and `test.yml` (full suite), called by both `ci.yml` and `release.yml`.

---

## Versioning and releases

| Version bump | When |
|---|---|
| **MAJOR** | Breaking change in the public API (any module not under `eventcommon._internal`), or in the `Envelope` wire format / HMAC canonicalisation |
| **MINOR** | New backward-compatible capability |
| **PATCH** | Bug fix, performance improvement, documentation correction |

```bash
git tag -a v0.2.0 -m "v0.2.0"
git push origin v0.2.0   # triggers the release workflow automatically
```

See [VERSIONING.md](./VERSIONING.md) for the full SemVer policy, the envelope wire-format
compatibility rules, PyPI publishing setup, and the maintainer release checklist.

## License / ownership

BCBP Solutions FZC LLC — internal platform shared library.

---

## Service adoption checklist

### Publisher checklist

- [ ] `metrics.init(app_name, build_version)` called once at startup
- [ ] Outbox schema applied via `apply_schema(migrate_runner)` on startup
- [ ] `outbox.Runner` started as a background task and `stop()` awaited on shutdown
- [ ] All domain-event publish call sites use `outbox.enqueue` inside `pgcommon.run_in_tx` — no bare `publisher.publish` for transactional events
- [ ] `tenant_id`, `trace_id`, and `schema_version` passed to every `new_envelope` call in HTTP handler context
- [ ] `OUTBOX_*` env vars wired via `runner_config_from_env`
- [ ] Readiness probe waits for `runner.ready` before marking the pod ready

### Consumer checklist

- [ ] SNS→SQS subscription created with **`RawMessageDelivery=true`** — without it the consumer deletes messages as malformed
- [ ] `SQS_*` env vars wired via `load_sqs` + `sqs_consumer_options`
- [ ] `SQSConsumer` wired with `concurrency` appropriate for handler latency
- [ ] `SQS_VISIBILITY_TIMEOUT` set to ≥ 2× p99 handler duration
- [ ] Every handler implements idempotency via `INSERT INTO processed_events ... ON CONFLICT DO NOTHING`
- [ ] Handlers classify errors as transient (raise) vs. permanent (return + log)
- [ ] Handlers check `env.type` before parsing `env.data`

### Observability checklist

- [ ] Handler entry logs bind `event_id`, `event_type`, `trace_id`, `tenant_id`
- [ ] OTel provider initialised before the first `consumer.start` or `publisher.publish` call
- [ ] Alerts configured on `outbox_pending_total`, `events_consumed_total{status="error"}`, and SQS `ApproximateNumberOfMessages`

### Security checklist

- [ ] HMAC signing enabled on HTTP/webhook ingress that receives events from external systems — **not** on the SNS/SQS messaging path
- [ ] HMAC keys ≥ 32 bytes, stored in a secrets manager — not in source-controlled environment variables
- [ ] `SYSTEM_TENANT_ID` usage reviewed — no HTTP handler context uses it to avoid a tenant lookup

---

## See also

| Document | Description |
|----------|-------------|
| [ARCHITECTURE.md](./ARCHITECTURE.md) | Layer model, sequence diagrams, invariants, performance |
| [EVENT_SCHEMA_GOVERNANCE.md](./EVENT_SCHEMA_GOVERNANCE.md) | Event type naming, payload evolution rules, the event type registry, the Python unknown-field trap |
| [VERSIONING.md](./VERSIONING.md) | SemVer rules, envelope wire-format guarantees, PyPI publishing, release process |
| [CHANGELOG.md](./CHANGELOG.md) | Per-version changes |
| [CONTRIBUTING.md](./CONTRIBUTING.md) | Development setup, adding adapters, PR checklist |
| [SECURITY.md](./SECURITY.md) | Vulnerability reporting, trust model, supported versions |

---

See [Publishing rules](#publishing-rules) for the full decision table. The short version: `publisher.publish` is only valid when event loss is explicitly acceptable. For any event that drives downstream state, use `outbox.enqueue` inside a `pgcommon.run_in_tx` block — no exceptions.

`eventcommon` must be the only entry point for event publishing and consumption to preserve delivery guarantees, tenant isolation, and uniform observability.
