Metadata-Version: 2.5
Name: eventcommon
Version: 0.3.0
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-Expression: Apache-2.0
License-File: LICENSE
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>=15.5.0
Requires-Dist: boto3>=1.40.61
Requires-Dist: opentelemetry-api>=1.45.1
Requires-Dist: pgcommon<0.5.0,>=0.4.0
Requires-Dist: prometheus-client>=0.26.0
Requires-Dist: structlog>=26.1.0
Description-Content-Type: text/markdown

# eventcommon

> Opinionated async SNS/SQS messaging layer that enforces at-least-once delivery, tenant propagation and observability by making the safe path the default, and the unsafe path visibly wrong.

This is the platform's **event-driven messaging library for Python services**, an `asyncio` library on top of aioboto3 and pgcommon. Every Python platform service imports it to publish and consume domain events.

It is the Python port of [`platform-events`](https://github.com/BCBP-SOLUTIONS-FZC-LLC/platform-events) (Go) with full feature parity, currently tracking Go **v2.1.0**, reimplemented idiomatically for `asyncio`. It owns the messaging boundary:
- the `Envelope` wire format, byte-identical to Go's, and its HMAC canonicalisation;
- the SNS publisher and the SQS consumer;
- dead-letter forwarding;
- the transactional outbox and the idempotency inbox;
- tenant propagation into pgcommon's RLS context;
- the messaging telemetry every service emits: Tier 1 `platform_*` metrics, OTel spans and Go's log messages.

The few places where Python cannot match Go are listed in [ARCHITECTURE.md § Divergences from Go](ARCHITECTURE.md#divergences-from-go).

**Repository:** `github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon` (private)

**Package:** `eventcommon` 0.3.0 (on [PyPI](https://pypi.org/project/eventcommon/); [release status](#versioning-and-releases)) · Apache-2.0 · Python ≥ 3.12 (CI runs 3.12, 3.13 and 3.14) · consumed as a dependency, never deployed on its own. Each release also ships the **example outbox worker** as a container image (`ghcr.io/bcbp-solutions-fzc-llc/platform-eventcommon`): a reference deployment of the runner, not a service to deploy.

**Design:** one `src/eventcommon` package of public modules on top of a private `_internal/` layer, with the layer rules enforced by import-linter in CI. See [ARCHITECTURE.md](ARCHITECTURE.md) for design decisions, layer diagrams, the failure model and how the `asyncio`/`contextvars` model differs from Go's `context.Context`. This README is how to **use** it.

**Features:** typed event envelopes (CloudEvents-aligned) · async SNS publisher (single, batch, FIFO) · async SQS consumer (concurrency, FIFO groups, visibility extension, retry backoff, handler timeouts) · DLQ forwarding · transactional outbox (Postgres via `pgcommon`, per-key ordering, dead letters) · idempotency inbox · HMAC signing · schema-registry codec hook · Prometheus metrics · OpenTelemetry tracing · RLS tenant propagation · test doubles.

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

> **Reading this on PyPI?** Links to other files (ARCHITECTURE.md, CHANGELOG.md, docs/, …) are relative to the source repository, which is private, so they do not resolve here. Every linked document ships in the source distribution: `pip download --no-binary :all: --no-deps eventcommon` and unpack the `.tar.gz`.

---

## Contents

- [Mental model](#mental-model) — [Event flow](#event-flow) · [Design principles](#design-principles) · [TL;DR](#tldr)
- [Why this library exists](#why-this-library-exists) — [Why eventcommon instead of raw aioboto3?](#why-eventcommon-instead-of-raw-aioboto3) · [When NOT to use this](#when-not-to-use-this)
- [API overview](#api-overview)
- [Input validation](#input-validation) — [Error reference](#error-reference)
- [Architecture](#architecture)
- [Integrating with a consumer service](#integrating-with-a-consumer-service)
  - [Installing](#installing) · [Quick start](#quick-start) · [Integration checklist](#integration-checklist) · [Production checklist](#production-checklist)
  - [System invariants](#system-invariants) · [Publishing rules](#publishing-rules) · [Event envelope](#event-envelope) · [SNS Publisher](#sns-publisher) · [SQS Consumer](#sqs-consumer) · [Implementing idempotency](#implementing-idempotency)
  - [Transactional outbox](#transactional-outbox) · [HMAC helpers](#hmac-helpers) · [Health, readiness and shutdown](#health-readiness-and-shutdown) · [Failure model](#failure-model)
  - [Backpressure considerations](#backpressure-considerations) · [Performance considerations](#performance-considerations) · [Common pitfalls](#common-pitfalls) · [Configuration reference](#configuration-reference)
- [Upgrading from 0.2.0](#upgrading-from-020)
- [Local development](#local-development)
- [Testing SNS/SQS locally](#testing-snssqs-locally)
- [Testing](#testing) — [Testing in consuming services](#testing-in-consuming-services)
- [Environment variables](#environment-variables)
- [Security](#security)
- [Observability — Prometheus and OTel](#observability--prometheus-and-otel)
- [Deployment](#deployment) — [Versioning and releases](#versioning-and-releases)
- [CI](#ci)
- [Docker](#docker)
- [Parity with platform-events](#parity-with-platform-events)
- [Dependencies](#dependencies)
- [Out of scope](#out-of-scope)
- [Contributing](#contributing)
- [Glossary](#glossary)
- [License / ownership](#license--ownership)
- [See also](#see-also)

---

## Mental model

| This library owns | It does NOT own |
|---|---|
| The **`Envelope`** wire format (`new_envelope`, `parse_envelope`, `to_json`), byte-identical to Go's, and its HMAC canonicalisation | The **payload schema**. `data` is yours; [EVENT_SCHEMA_GOVERNANCE.md](EVENT_SCHEMA_GOVERNANCE.md) has the rules |
| **Publishing**: `SNSPublisher` (single, batch, FIFO), message attributes for filter policies, trace-context propagation, error classification, `call_timeout` | SNS topics, SQS queues, subscriptions, filter policies and RedrivePolicies (infrastructure) |
| **Consuming**: `SQSConsumer` (long poll, concurrency, FIFO groups, visibility extension, retry backoff, handler and drain timeouts, malformed-body handling, dead-lettering) and `SQSDLQPublisher` | **What your handler does**, and its idempotency. The library delivers at least once; your side effects must tolerate it |
| **Tenant propagation**: the consumer binds each message's `tenant_id` into pgcommon's GUC context, so every pool call in the handler runs under RLS | Your tables and **RLS policies**, and the pool (`pgcommon`). eventcommon supplies the tenant; your policies enforce it |
| The **transactional outbox** (`enqueue`, `Runner`, `outbox_events` / `outbox_dead_letters`, golang-migrate-compatible schema shared with Go) and the **inbox** (`processed_events`) | Your business transaction. `enqueue` joins it; it never opens one |
| Messaging metrics, spans and logs, registered on **your** `CollectorRegistry`, tracer provider and logger | Your Prometheus exposition, OTel SDK and logging setup (`fastapicommon`) |
| The `Codec` hook and a decode-only `GlueDecodeCodec` | A schema-registry client. You implement `Codec` against yours |
| Test doubles (`eventcommon.mock`) | — |
| — | Metric ratification and alert routing (the vendored `monitoring/` bundle is Go's, unchanged) |

```
HTTP request (producer)
  → your handler ── await pool.run_in_tx(tx)
      → tx(conn): domain write + await outbox.enqueue(conn, new_envelope(type, source, data,
                                                     tenant_id=rc.tenant_id, trace_id=rc.trace_id))
      → COMMIT ── the event is durable with the write, or neither is
  → outbox.Runner (background) ── claim (SKIP LOCKED + lease) → SNS publish → mark published
  → SNS → SQS (RawMessageDelivery=true)
  → SQSConsumer ── receive → malformed check → decode → bind tenant / trace / source message
      → your handler(env) ── pool.with_conn / run_in_tx now scoped to env.tenant_id by RLS
      → returns: delete · raises: retry (backoff) · past max_receive_count: dead-letter + DLQ
```

### Event flow

This diagram shows the life of an event from the domain write that produces it to the handler that consumes it. Every box is either owned by the library or governed by its configuration.

> Think of `eventcommon` as the only door between a service and the bus: every event passes through it to gain durability, tenancy and observability guarantees.

```mermaid
flowchart TD
    A[HTTP request — producer service]
    B["pgcommon.run_in_tx\ndomain write + outbox.enqueue(conn, env)"]
    C[("outbox_events\nPostgreSQL")]
    D["outbox.Runner\nclaim · publish · settle"]
    E[SNSPublisher]
    F[SNS topic → SQS queue\nexternal infrastructure]
    G["SQSConsumer\nreceive · validate · decode"]
    H["handler context\ntenant GUC · trace ID · source message"]
    I["your handler(env)\nidempotent on env.id"]
    J[(PostgreSQL + RLS\nconsumer's database)]

    A --> B --> C --> D --> E --> F --> G --> H --> I --> J

    style B fill:#fff3cd,stroke:#856404
    style D fill:#d1ecf1,stroke:#0c5460
    style E fill:#d1ecf1,stroke:#0c5460
    style G fill:#d1ecf1,stroke:#0c5460
    style H fill:#d1ecf1,stroke:#0c5460
    style F fill:#e2e3e5,stroke:#6c757d
    style J fill:#d4edda,stroke:#155724
```

> Read top → bottom: each layer adds a guarantee (durability → delivery → tenancy → observability) before the event reaches your handler.

**Detailed annotated view:**

```
HTTP request (producer)
    │
    ▼
pgcommon.run_in_tx(pool, tx)             ← caller's responsibility: one business transaction
    │  tx(conn): your domain write
    │            await outbox.enqueue(conn, env)   ← validated, ≤ 240 KB, no SNS call
    │  COMMIT: both rows durable, or neither
    │
    ▼
outbox_events                            ← PostgreSQL (platform-events' schema, shared with Go)
    │
    ▼
outbox.Runner                            ← background task: await runner.start()
    │  claim: FOR UPDATE SKIP LOCKED + a claim lease (several runners, Go or Python)
    │  publish: publish_batch (10 per request) or publish per record, publish_timeout
    │  settle: published · transient → released, no attempt · permanent → backoff
    │          · max_attempts → outbox_dead_letters
    │
    ▼
SNSPublisher                             ← attributes EventType · TenantID · Source · EventID · Subject
    │                                       + traceparent; span sns.publish(.batch)
    ▼
SNS topic → SQS queue                    (external infrastructure; RawMessageDelivery=true)
    │
    ▼
SQSConsumer                              ← await consumer.start()
    │  long poll · concurrency · visibility extension · FIFO group order
    │  malformed body → DLQ or delete (never retried) · codec decode
    │
    ▼
handler context                          ← automatic, per message, reset afterwards
    │  pgcommon.with_guc_set(GUCSet(tenant_id=env.tenant_id))
    │  trace ID · source message · span sqs.receive (linked to the producer)
    │
    ▼
your handler(env)                        ← idempotent on env.id (inbox.handler / inbox.Store)
    │  returns → message deleted · raises → retried · past max_receive_count → DLQ
    │
    ▼
    ├── Prometheus metrics               ← Tier 1 platform_* only
    ├── OTel spans                       ← no-op without the service's TracerProvider
    └── logs                             ← Go's messages, through the logger you pass
```

> **Invariant:** every domain event tied to a database write goes through `outbox.enqueue` inside the business transaction. A direct `publisher.publish` after the commit is a silent data-loss bug, not a shortcut.

**Key invariants:**
- Delivery is at least once on both sides, so every handler must be idempotent on `Envelope.id`.
- Tenant context is bound by the consumer before your handler runs, and reset after it; your handler never sets a GUC.
- Transient failures (throttling, outages, timeouts, shutdown) never count an outbox attempt, so an SNS outage cannot dead-letter healthy events.
- Observability is wired once (`metrics.init_metrics`, the service's tracer provider and logger); there is no per-call code in services.

> **Ownership summary:**
> - **Service:** the business transaction, handlers and their idempotency, RLS policies
> - **eventcommon:** envelope, delivery, retries, dead-lettering, tenant binding, observability
> - **Infra:** SNS, SQS, IAM, PostgreSQL

### 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. Producers do not direct consumers; consumers decide independently how to react. So consumers must be resilient to receiving the same fact more than once (`Envelope.id` idempotency), and 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` on an empty `queue_url` or `visibility_timeout > 12h`, and all three AWS components when no region resolves. Invalid configuration is rejected at construction, not at first use. A consumer with DLQ forwarding refuses to start against a queue without a usable RedrivePolicy.
- **Make safe usage the default.** The SQS consumer propagates `tenant_id` into pgcommon's GUC context for RLS, deletes only after a handler returns, and drains in-flight messages on `stop()`. With `visibility_timeout` set, it extends slow handlers' messages. The default of 0 uses the queue's own timeout and extends nothing, as in Go.
- **Keep business logic free from messaging plumbing.** Handlers receive an `Envelope`. Retry backoff, visibility extension, metric recording and OTel spans are invisible to them.
- **Centralise cross-cutting concerns.** HMAC signing, tenant propagation, dead-lettering and observability live here, not in every service. A single version bump propagates to all consumers.

### TL;DR

> ⚠️ **Exactly-once delivery is not possible anywhere in this system.** SNS and SQS deliver at least once: a message may arrive more than once under normal operation, through visibility timeout expiry, a consumer crash mid-handler, or 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 whether or not FIFO is used.** See [Implementing idempotency](#implementing-idempotency).

- Use `eventcommon.envelope.new_envelope(...)` to create 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 on the consumer and trace continuity.
- Use `await outbox.enqueue(conn, env)` inside a `pgcommon.run_in_tx` block for every event tied to a database write. See [Publishing rules](#publishing-rules).
- **Never call `publisher.publish` for domain events tied to a DB write.** Event loss on a process crash is silent and unrecoverable.
- Make every handler idempotent on `Envelope.id`: wrap it with `inbox.handler(store, handle)`.
- Call `eventcommon.metrics.init_metrics(MetricsIdentity(domain=..., service=...))` once at startup. The identity is mandatory, and the library emits only Tier 1 `platform_*` metrics.
- OTel tracing is enabled by the consuming service (e.g. `fastapicommon.tracing.init_from_env()`), not by this library.

---

## Why this library exists

Every platform service is multi-tenant, talks over SNS/SQS and keeps its state in Postgres. Each needs the same half-dozen messaging things done exactly right, and done separately in each service, each is a production incident waiting to happen:

- **No lost events.** A domain write that commits while its event is lost leaves every downstream service wrong, with no error anywhere. The outbox makes the event part of the transaction and redelivers it until SNS accepts it. A full SNS outage only delays events.
- **No poison loops.** A malformed message or a handler bug must not loop forever, block a FIFO group or burn capacity. Malformed bodies are dead-lettered on first receipt, failing ones after `max_receive_count`, and outbox records after `max_attempts`, each counted with its reason.
- **Tenant isolation in the consumer.** A handler must run under the tenant of the event it is handling, never the previous message's. The consumer binds it into pgcommon's RLS context per message and resets it afterwards, so a handler cannot forget it.
- **One wire format across languages.** Go and Python services exchange envelopes, signatures and outbox rows. A field order, time format or JSON escape that differs breaks signatures and parsing in production. eventcommon is checked byte for byte against Go in CI.
- **Uniform telemetry.** Every service exports the same messaging metrics with the same names, labels and semantics as the Go library. One dashboard and one alert set cover both languages (Tier 1 of the Enterprise Platform Observability Standard).

A fix lands in every service with a version bump. This is enforced structurally: `eventcommon._internal` is private (a leading underscore, never re-exported, outside the SemVer surface), and the import-linter contracts in `pyproject.toml` encode the dependency-direction rules CI checks on every PR.

### Why eventcommon instead of raw aioboto3?

| Concern | Raw `aioboto3` SNS/SQS | `eventcommon` |
|---------|------------------------|---------------|
| Event loss after a DB commit | A publish after `COMMIT` is lost on a crash; nobody notices | `outbox.enqueue` in the transaction; the runner redelivers until SNS accepts |
| Envelope format | Each service invents one; Go and Python drift | One `Envelope`, byte-identical to Go's, checked against golden vectors |
| Retry classification | Every error retried or none; throttling dead-letters healthy events | Go's classification: transient errors never count an attempt; permanent ones back off and dead-letter |
| Batch publishing | Manual 10-entry / 256 KiB splitting, partial-failure bookkeeping | `publish_batch` splits, attempts every request, raises one `BatchError` keyed by envelope ID |
| Visibility | Manual `ChangeMessageVisibility` loops; slow handlers redelivered while running | `visibility_timeout` extends every message from receipt until settled |
| Poison messages | Loop until SQS's redrive, without a reason | Malformed → DLQ at once; `max_receive_count` → dead-letter handler + DLQ with `DLQReason`, `OriginalQueue`, `FailedAt` |
| Tenant context in handlers | Each handler must set the RLS GUC itself, and reset it | Bound per message from `env.tenant_id` and reset afterwards |
| Shutdown | In-flight handlers killed, or waited on forever | `stop()` drains up to `drain_timeout`, then cancels; unsent outbox records released at once |
| Metrics and traces | Hand-written counters; no producer → consumer trace link | Tier 1 `platform_*` metrics and linked spans, identical to Go's |
| Idempotency | Every team writes its own table | `inbox.handler` / `inbox.Store.process` on a shared schema |
| Testing | An AWS emulator for every unit test | `MockPublisher`, `MockConsumer`, `MockDLQPublisher` with the real dispatch rules |

**The single-sentence justification:** every row above is a bug we have seen in services that used raw SNS/SQS calls: lost events, duplicate side effects, poison loops, tenant leaks, invisible failures. `eventcommon` makes the correct path the default path.

> This is not an abstraction over aioboto3. It is a guardrail layer that enforces delivery and tenancy guarantees by construction.

### 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 for:

- **One-off scripts or CLIs.** A plain `aioboto3` SNS call is simpler, without the pool setup.
- **Services that consume a single event type**, where filtering, retries and concurrency are handled entirely by the queue configuration. A thin wrapper is enough.
- **Non-AWS message brokers.** The library is SNS/SQS-specific; EventBridge, Kafka and other backends are not supported without a new adapter ([CONTRIBUTING.md](CONTRIBUTING.md)).
- **Services without PostgreSQL.** `pgcommon` is a hard dependency regardless, because the SQS consumer binds the tenant GUC context for every message, not just for the outbox. A service with no database can still use the publisher and consumer and accept at-most-once publishing; it simply never calls `outbox.enqueue` or `outbox.Runner`.

---

## API overview

The public modules are listed in [VERSIONING.md § What counts as public API](VERSIONING.md#what-counts-as-public-api). Import only those, never `eventcommon._internal`. The source of truth is each module's `__all__`, docstrings and type hints; the package ships `py.typed` and is checked with `mypy --strict`. [ARCHITECTURE.md § Public API reference](ARCHITECTURE.md#public-api-reference) has the full tables.

### `eventcommon.envelope` — the wire format

| Symbol | Purpose |
|---|---|
| `new_envelope(event_type, source, payload, *, tenant_id, system_tenant, trace_id, correlation_id, schema_version, subject, actor, ip_address, user_agent, schema_id)` | Build an envelope: UUID v7 `id`, `time = now(UTC)` |
| `parse_envelope(data, payload_type=…, require_time=True)` | Parse and validate, as Go's `ParseEnvelope`; `payload_type` builds a typed payload, ignoring unknown fields |
| `Envelope.to_json()` · `Envelope.data_json_bytes()` | What Go's `json.Marshal` writes, byte for byte · `data`'s bytes as received |
| `decode_payload(data, YourDataclass)` | Typed payload from `env.data` inside a handler, dropping undeclared fields |
| `SYSTEM_TENANT_ID` · `format_time` · `parse_time` · `ZERO_TIME` · `LongInt` | `"system"` · Go-compatible time helpers · integers beyond Python's 4,300-digit limit |
| `trace_id_from_context()` | The envelope's trace ID inside a handler |

### `eventcommon.publisher` — SNS

| Symbol | Purpose |
|---|---|
| `SNSPublisher(SNSConfig(topic_arn, region, endpoint_url, logger, call_timeout), *options, message_group_id, message_deduplication_id, attributes, codec, client)` | The publisher. Validates at construction; one client per event loop |
| `await publisher.publish(env)` · `await publisher.publish_batch(envs)` | One event · many, split at 10 entries / 256 KiB, one `BatchError` for any failure |
| `await publisher.close()` · `async with publisher:` | Close every loop's client (never an injected `client=`) |
| `Publisher` (protocol) · `with_codec`, `with_attributes`, `with_message_group_id`, `with_message_deduplication_id` | What the outbox runner accepts · Go's functional options |

### `eventcommon.consumer` — SQS and DLQ

| Symbol | Purpose |
|---|---|
| `SQSConsumer(SQSConfig(queue_url, region, endpoint_url, max_messages, wait_seconds, logger), handler, *options, concurrency, visibility_timeout, handler_timeout, retry_backoff, max_retry_backoff, max_receive_count, dead_letter_handler, dlq_publisher, drain_timeout, codec, malformed_body_logging, queue_depth_interval, client)` | The consumer |
| `await consumer.start()` (as a task) · `await consumer.stop()` | Long-poll until stopped · drain, then cancel at `drain_timeout` |
| `Handler` · `AcceptedHandler` · `DeadLetterHandler` · `EventData` | Handler types: middleware and routers use `Handler`; what the consumer accepts is `AcceptedHandler` |
| `SQSDLQPublisher(DLQConfig(region, endpoint_url, consumer_name, logger, cache_ttl, strict_attributes))` | Forwards to the source queue's RedrivePolicy DLQ: `send_to_dlq`, `resolve_dlq`, `close` |
| `forward_source_message(dlq, reason)` | Dead-letter the message the current handler is processing |
| `handler_context` · `validate_dlq_message` · `parse_redrive_policy` · `parse_sqs_queue_arn` · `dlq_event_type` · `DLQ_ATTR_*` | Helpers, as Go exports them |
| `with_concurrency`, `with_visibility_timeout`, `with_handler_timeout`, `with_retry_backoff`, `with_max_receive_count`, `with_dead_letter_handler`, `with_dlq_forwarding`, `with_drain_timeout`, `with_consumer_codec`, `with_malformed_body_logging`, `with_queue_depth_metrics` | Go's functional options |

### `eventcommon.outbox` and `eventcommon.inbox` — Postgres

| Symbol | Purpose |
|---|---|
| `await outbox.enqueue(conn, env)` · `await outbox.enqueue_ordered(conn, env, key)` | Insert inside the caller's transaction · with per-key publish order |
| `outbox.Runner(RunnerConfig(...))` · `start()` · `stop()` · `ready` · `healthy(max_age)` | The poll runner; see [Transactional outbox](#transactional-outbox) |
| `runner.list_dead_letters` · `reprocess_dead_letters(_with)` · `discard_dead_letters` · `prune_published` | Operator tools (`DLQFilter`, `DeadLetterRecord`) |
| `await outbox.apply_schema(pool)` · `await inbox.apply_schema(pool)` | Apply platform-events' migrations, golang-migrate-compatible |
| `outbox.OutboxStore` · `PostgresStore` · `OutboxRecord` | The store port and its adapter |
| `inbox.Store(pool, consumer)` · `inbox.handler(ledger, handle)` | `is_processed`, `mark_processed`, `process(env, fn)`, `prune(retention)` · dedup wrapper |

### `eventcommon.metrics`, `hmac`, `codec`, `context`, `mock`, `config`

| Symbol | Purpose |
|---|---|
| `metrics.init_metrics(MetricsIdentity(domain, service, environment), registry=None, *, event_type_limit, event_types)` · `metrics_identity_from_env(domain)` | **The only** init: registers the Tier 1 `platform_*` metrics; returns `list[RegistrationWarning]` |
| `hmac.sign` · `verify` · `sign_envelope` · `verify_envelope` | HMAC-SHA256 over Go's canonical JSON |
| `codec.Codec` (protocol) · `NoopCodec` · `GlueDecodeCodec` · `wrap_codec_payload` · `unwrap_codec_payload` | Schema-registry hook |
| `context.source_message_from_context()` · `record_batch_published(ids)` | The raw SQS message in a handler · progress reporting for custom publishers |
| `mock.MockPublisher` · `MockConsumer` · `MockDLQPublisher` | Test doubles |
| `config.load_sns` · `load_sqs` · `load_outbox` · `log_warnings` · `sns_config_from_env` · `sqs_config_from_env` · `sqs_consumer_options` · `runner_config_from_env` · `mask_dsn` | Environment wiring with Go's names, defaults and warning texts |

### `eventcommon-config` CLI

| Command | Purpose |
|---|---|
| `eventcommon-config` | Print the effective configuration (credentials masked), configuration warnings and problems, and production wiring reminders (Go: `cmd/platform-events`) |
| `eventcommon-config --strict` | The same; exits 1 when a required variable is missing — an init-container or CI smoke check |

---

## Input validation

eventcommon validates everything it can **before** a network call or a database write, so misconfiguration fails at startup and a bad envelope fails at the call site, not in a consumer three services away. Environment variables are the exception to "fail fast": the `load_*` loaders never raise for a bad value. They apply the default and add a warning (Go's exact text) to `.warnings`, which you must log.

| Input | Rule | Checked by | Raises |
|---|---|---|---|
| `SNSConfig.topic_arn` | Non-empty; starts with `arn:aws:sns:`, `arn:aws-cn:sns:` or `arn:aws-us-gov:sns:`. A `.fifo` topic needs `message_group_id` | `SNSPublisher(...)` | `ValueError` |
| `SQSConfig.queue_url` | Non-empty; `visibility_timeout` ≤ 12 h. (`max_messages` outside 1–10 and `wait_seconds` outside 1–20 are not errors: a warning, then 10 / 20, as Go) | `SQSConsumer(...)` | `ValueError` |
| AWS region | `region=`, else `AWS_REGION`, else boto3's chain (`AWS_DEFAULT_REGION`, the profile) — the Go SDK's order | `SNSPublisher`, `SQSConsumer`, `SQSDLQPublisher` | `ValueError("…: no AWS region configured — set AWS_REGION (or the config's region)")` |
| Envelope to publish | Non-empty `id` / `type` / `source`; valid JSON (`NaN`, `±Infinity`, naive `datetime` rejected); ≤ 10 message attributes after trace context | `publish`, `publish_batch` (per entry), `MockPublisher` | `ValueError`; per entry `BatchFailure` (`InvalidEnvelope`, `MarshalError`, `TooManyAttributes`) |
| Envelope to enqueue | As above, plus non-zero `time`, a canonical lowercase UUID `id`, no NUL anywhere, serialised ≤ 240 KB, an open transaction | `outbox.enqueue` / `enqueue_ordered` | `ValueError`, `Envelope*RequiredError` |
| Ordering key | Non-empty, ≤ 256 bytes (`MAX_ORDERING_KEY_LEN`) | `enqueue_ordered` | `ValueError` |
| Received body | An object with `id`, `type`, `source`; no NUL in `tenant_id` / `trace_id`; parsed as Go's decoder parses it | `SQSConsumer`, `MockConsumer.inject` | Not raised: the message is **malformed** (DLQ or delete) · `MalformedEnvelopeError` from the mock |
| Envelope to parse | As Go's `ParseEnvelope`; `time` required unless `require_time=False`. A `null` body is an envelope with no fields (Go's no-op unmarshal), so it is missing its ID | `parse_envelope` | `EnvelopeParseError`, `Envelope{ID,Type,Source,Time}RequiredError` |
| `apply_schema(lock_timeout=…)` | Finite, > 0, at most 2,147,483,647 ms; rounded **up** to whole milliseconds (`'0ms'` would mean no timeout in PostgreSQL) | `outbox.apply_schema`, `inbox.apply_schema` | `ValueError`, before any SQL |
| HMAC key | ≥ 32 bytes | `sign`, `sign_envelope`, `verify_envelope` | `KeyTooShortError` |
| Metric identity | `domain`, `service`, `environment` required and well-formed; the message names `APP_ENV` / `ENVIRONMENT` when the bad value came from there | `init_metrics` (nothing changes on failure) | `InvalidMetricsIdentityError` |
| `RunnerConfig` | `publisher`, and `pool` or `store`, required; `claim_lease_duration ≥ batch_size × publish_timeout + 60s` (30 s when the timeout is disabled). Zero values mean the default | `Runner(...)` | `ValueError` |
| DLQ forward | Source URL, body, reason, SQS-legal characters, attribute names, ≤ 1 MiB, ≤ 10 attributes with `strict_attributes` | `send_to_dlq` | `DLQInvalidMessageError`, before any AWS call |
| Inbox event ID | A UUID | `inbox.Store` | `InvalidEventIDError` |
| Environment variables | Go's `time.ParseDuration` and integer rules: a bare `30` is not a duration | `load_sns` / `load_sqs` / `load_outbox` | Never raises: default + warning; `.validate()` raises `ConfigValidationError` for a missing required variable |

### Notable validation rules

| Rule | Enforcement |
|---|---|
| An event that could never publish is never persisted | `enqueue` rejects envelopes over 240 KB (under SNS's 256 KB), NUL characters (`jsonb` cannot store `\u0000`, so the insert would roll back the business transaction on every retry) and non-UUID IDs |
| A malformed message never reaches a handler, and is never retried | The consumer forwards it to the DLQ or deletes it on first receipt; its log line carries size + SHA-256, never the body |
| A NUL never reaches a GUC | A NUL in `tenant_id` / `trace_id` makes a received message malformed |
| A bare number is not a duration | `SQS_VISIBILITY_TIMEOUT=60` is rejected (warning, default applied), exactly as Go rejects it — write `60s` |
| Metric labels are bounded | `event_type` capped at 200 values (`__other__`) and 128 bytes (`__oversized__`); `queue` / `topic` are names, never URLs or ARNs |
| A misconfigured DLQ fails at startup | With `dlq_publisher`, `start()` raises when the queue has no usable RedrivePolicy, instead of looping poison messages |

### Error reference

| Error | Raised by | When |
|---|---|---|
| `EnvelopeIDRequiredError` · `EnvelopeTypeRequiredError` · `EnvelopeSourceRequiredError` | `parse_envelope`, `outbox.enqueue` | A missing `id` / `type` / `source` |
| `EnvelopeTimeRequiredError` | `parse_envelope` | A missing or zero `time` (unless `require_time=False`) |
| `EnvelopeParseError` (a `ValueError`) | `parse_envelope` | Invalid JSON, a body that is not an object (an array, number, string or boolean; `null` is missing its ID instead, as in Go), a field of the wrong type, a non-RFC 3339 time |
| `KeyTooShortError` | `sign`, `sign_envelope`, `verify_envelope` | A key under 32 bytes |
| `InvalidSignatureError` | — | Exported for parity; never raised (verification returns `False`) |
| `CodecEncodeError` | `SNSPublisher` | The codec failed to encode (wrapped in `RetryableError` when the codec's error was retryable) |
| `BatchError` | `publish_batch` | One or more entries failed; `.failures` lists `BatchFailure(id, code, message, retryable)` |
| `RetryableError` | `SNSPublisher` | Wraps a transient SNS/AWS error so the outbox does not count an attempt; `str()` is the cause's text. Test with `errors.is_retryable(exc)`, which follows `__cause__` |
| `DLQNotConfiguredError` · `DLQInvalidRedrivePolicyError` | `SQSDLQPublisher`, `SQSConsumer.start()` | The source queue has no RedrivePolicy · it is unparseable or not an SQS ARN |
| `DLQUnresolvedError` · `DLQSendFailedError` | `send_to_dlq` | The DLQ lookup failed or timed out (30 s) · the send failed (`.retryable` for throttling, 5xx and timeouts) |
| `DLQInvalidMessageError` | `send_to_dlq` | The message is not valid for SQS (before any AWS call) |
| `InvalidMetricsIdentityError` (a `ValueError`) | `init_metrics` | An invalid metric identity |
| `InvalidEventIDError` · `TxEndedInCallbackError` · `TxCommitRollbackError` (`eventcommon.inbox`) | `inbox.Store` | The envelope ID is not a UUID · `process`'s callback ended the transaction · it swallowed a failed statement, so the transaction rolled back. The message is retried |
| `MigrationDirtyError` · `SchemaVersionNotInSourceError` (`eventcommon.outbox`) | `apply_schema` | A Go migration failed half-way · the database was migrated by a newer release |
| `ConfigValidationError` (`eventcommon.config`) | `.validate()` | A required environment variable is missing |

All errors in `eventcommon.errors` subclass `EventCommonError`; check with `isinstance(exc, EventCommonError)`, the Python equivalent of Go's `errors.Is` against the package's sentinels.

---

## Architecture

The Go module's Clean Architecture layers become one flat package in four layers: composition roots (`cli`, `config`, `mock`), transports (`publisher`, `consumer`, `outbox`, `inbox`), a shared core (`envelope`, `hmac`, `codec`, `context`, `metrics`) and two sinks (`errors`, `_internal`). Full layer diagrams, flows, invariants, the failure model and the asyncio translation notes are in **[ARCHITECTURE.md](ARCHITECTURE.md)**; the standalone Mermaid sources are in **[docs/architecture/](docs/architecture/README.md)**.

```
platform-eventcommon/
├── src/eventcommon/                 # The package (the wheel ships only this)
│   ├── __init__.py                  # Re-exports of the shared core
│   ├── envelope.py · errors.py      # Envelope, new_envelope, parse_envelope · EventCommonError and every sentinel
│   ├── hmac.py · codec.py           # sign / verify over Go's canonical JSON · Codec, NoopCodec, GlueDecodeCodec
│   ├── context.py                   # trace ID, source message, batch progress and DLQ attribution (contextvars)
│   ├── metrics.py · metrics_registry.py  # init_metrics, MetricsIdentity · the governed Tier 1 platform_* inventory
│   ├── publisher/sns.py             # SNSPublisher, SNSConfig, options
│   ├── consumer/sqs.py · dlq.py     # SQSConsumer, SQSConfig, options · SQSDLQPublisher, DLQConfig
│   ├── outbox/                      # enqueue, Runner, PostgresStore, dead-letter tools, apply_schema, migrations/ (001–011)
│   ├── inbox/                       # Store, handler, apply_schema, migrations/ (001)
│   ├── config/ · cli.py             # Environment loaders and wiring · eventcommon-config
│   ├── mock/                        # MockPublisher, MockConsumer, MockDLQPublisher
│   └── _internal/                   # PRIVATE: AWS clients, canonical JSON, Go durations and formatting, logging, migrations, UUID v7, …
├── examples/outbox_worker/          # The example worker (the container image)
├── tests/                           # unit (no Docker) · integration (Postgres + floci) · e2e (full pipeline); fixtures/
├── monitoring/                      # Go's Prometheus rules + tests, Grafana dashboard, KEDA example (vendored, byte-identical)
├── docs/                            # architecture/ (Mermaid sources), guides/, observability/
├── docker/ · docker-compose.yaml    # floci (SNS/SQS), Postgres and the observability stack for local work
├── .github/workflows/ · scripts/    # ci, validate-test, validate-quality, release, docs, changelog-check · their job logic
├── Dockerfile · build-constraints.txt  # The example-worker image (digest-pinned bases) · the hash-locked build backend
└── pyproject.toml · uv.lock · Makefile · .env-example
```

### Dependency rules (enforced by import-linter in CI)

The contracts live in `pyproject.toml` (`[tool.importlinter]`); `make arch-lint` runs them, as validate-test's `test` job does.

| Component | May depend on |
|---|---|
| `eventcommon.errors` | Nothing in the package (a leaf) |
| `eventcommon._internal.*` | `errors` only (one function-level exception: `canonical_json` → `envelope.format_time`) |
| The shared core: `envelope`, `hmac`, `codec`, `context`, `metrics`, `metrics_registry` | Each other and the sinks — **never** a transport, `config`, `cli` or `mock` |
| The transports: `publisher`, `consumer`, `outbox`, `inbox` | The core and the sinks — **never** each other, `config`, `cli` or `mock` |
| `config`, `cli`, `mock` | Anything |
| The package root (`eventcommon/__init__.py`) | The shared core only |

Plus the **RLS guard**, a grep in validate-quality: no session-level `SET app.tenant_id|user_id|tenant_roles` in `src/`. The tenant is bound only through pgcommon (`with_guc_set`), which the pool sets with `set_config` and resets.

### Runtime dependencies (what consumers inherit)

| Library | Range (`pyproject.toml`) | Locked in `uv.lock` | Purpose |
|---|---|---|---|
| `pgcommon` | `>=0.4.0,<0.5.0` | 0.4.0 | Pool and transactions for the outbox / inbox; GUC context for tenant binding |
| `aioboto3` | `>=15.5.0` | 15.5.0 | Async SNS / SQS clients |
| `boto3` | `>=1.40.61` | 1.40.61 | Error classes and the credential / region chain |
| `prometheus-client` | `>=0.26.0` | 0.26.0 | Metrics |
| `opentelemetry-api` | `>=1.45.1` | 1.45.1 | Span API (the SDK is the service's) |
| `structlog` | `>=26.1.0` | 26.1.0 | Structured logging in the example worker and the default logger adapter |

No OTel SDK, no web framework and no test tool: those live in the `dev` dependency group, so they never enter a consumer's environment. CI's `lowest direct dependencies` job runs every suite with each runtime dependency at its floor.

---

## Integrating with a consumer service

Everything a consuming service does with eventcommon, in roughly the order it does it:
1. Install.
2. Wire metrics, the pool and the schema once at startup.
3. Enqueue events inside business transactions.
4. Run the outbox runner and the consumer as background tasks.
5. Make handlers idempotent.
6. Expose readiness and liveness.
7. Drain at shutdown.

The two checklists are the short version; the subsections after them are the reference.

### Installing

Published on [PyPI](https://pypi.org/project/eventcommon/). The source repository stays private, but the released package is public; [VERSIONING.md § Publishing to PyPI](VERSIONING.md#publishing-to-pypi) explains why.

```bash
pip install eventcommon
# or with uv:
uv add eventcommon
```

Pin an exact version in production; [VERSIONING.md § Consume a release](VERSIONING.md#consume-a-release) has the recommended pin styles.

> Import only the public modules (`eventcommon.envelope`, `eventcommon.publisher`, `eventcommon.consumer`, `eventcommon.outbox`, …). Never import `eventcommon._internal`: it is private by lint rule and not part of the supported API.

### Quick start

```python
import asyncio
from typing import Any

import pgcommon
import structlog

from eventcommon import inbox, metrics, outbox
from eventcommon.config import (
    load_outbox,
    load_sns,
    load_sqs,
    log_warnings,
    runner_config_from_env,
    sns_config_from_env,
    sqs_config_from_env,
    sqs_consumer_options,
)
from eventcommon.consumer import DLQConfig, SQSConsumer, SQSDLQPublisher
from eventcommon.envelope import Envelope
from eventcommon.publisher import SNSPublisher

logger = structlog.get_logger()


async def main() -> None:
    # 1. Tier 1 platform_* metrics, once — same registry and identity as pgcommon's.
    #    environment ← APP_ENV / ENVIRONMENT / "dev"; service ← APP_NAME.
    for warning in metrics.init_metrics(metrics.metrics_identity_from_env("iam")):
        logger.warning("metric_disabled", warning=str(warning))  # one metric off, not fatal

    # 2. Load and validate configuration (warnings: invalid values, defaults applied).
    sns_env, sqs_env, outbox_env = load_sns(), load_sqs(), load_outbox()
    log_warnings(sqs_env.warnings, logger=logger)
    log_warnings(outbox_env.warnings, logger=logger)
    for env in (sns_env, sqs_env, outbox_env):
        env.validate()  # ConfigValidationError when a required variable is missing

    # 3. Pool (DATABASE_URL or PG_*, via pgcommon.config_from_env) and schemas.
    pool = await pgcommon.Pool.create(outbox_env.pool_config())  # validated, non-optional
    await outbox.apply_schema(pool)
    await inbox.apply_schema(pool)

    # 4. Publisher + outbox runner (delivers enqueued events asynchronously). The
    #    publisher keeps one SNS client: close it at shutdown.
    publisher = SNSPublisher(sns_config_from_env(sns_env, logger=logger))
    runner = outbox.Runner(runner_config_from_env(outbox_env, pool, publisher, logger=logger))

    # 5. Consumer: inbox dedup + DLQ forwarding for poison messages.
    ledger = inbox.Store(pool, "iam-user-created")
    dlq = SQSDLQPublisher(DLQConfig(consumer_name="iam", logger=logger))

    async def handle(env: Envelope[dict[str, Any]]) -> None: ...  # idempotent business logic

    consumer = SQSConsumer(
        sqs_config_from_env(sqs_env, logger=logger),
        inbox.handler(ledger, handle),
        dlq_publisher=dlq,
        **sqs_consumer_options(sqs_env),
    )

    tasks = [asyncio.create_task(runner.start()), asyncio.create_task(consumer.start())]
    await asyncio.wait_for(runner.ready.wait(), timeout=30)  # gate the readiness probe
    try:
        # Returns when either stops; a crashed runner or consumer ends the service
        # (the orchestrator restarts it) instead of leaving it half alive.
        done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
        for task in done:
            task.result()
    finally:
        # Stop both at once: their drains then fit one terminationGracePeriodSeconds.
        await asyncio.gather(consumer.stop(), runner.stop(), return_exceptions=True)
        await publisher.close()
        await dlq.close()
        await pool.drain_and_close()


asyncio.run(main())
```

`examples/outbox_worker/main.py` is the same wiring, production-shaped: startup retries, signals, a heartbeat for liveness, `/metrics` and bounded shutdown.

### Integration checklist

Run through this before wiring the library into a new service. Each item links to the relevant section.

- [ ] **`metrics.init_metrics`**: call once at startup, before building any publisher, consumer or runner, with the service's `domain` / `service` / `environment`; log the returned warnings ([Observability](#observability--prometheus-and-otel))
- [ ] **Outbox for transactional events**: every event tied to a database write goes through `outbox.enqueue` inside `pgcommon.run_in_tx`; no bare `publisher.publish` for them ([Publishing rules](#publishing-rules))
- [ ] **Request context on every envelope**: `tenant_id=rc.tenant_id`, `trace_id=rc.trace_id` from the HTTP request; `system_tenant=True` only for genuine cross-tenant jobs ([Event envelope](#event-envelope))
- [ ] **Idempotent handlers**: `inbox.handler(store, handle)` or `store.process(env, fn)`; external calls carry `Envelope.id` as their idempotency key ([Implementing idempotency](#implementing-idempotency))
- [ ] **Error classification in handlers**: raise for transient errors, return and log for permanent ones ([Error classification](#error-classification))
- [ ] **Configuration from the environment**: `load_sns` / `load_sqs` / `load_outbox`, `log_warnings`, `.validate()`; durations carry a unit ([Configuration reference](#configuration-reference))
- [ ] **Dead-lettering**: `dlq_publisher` (and/or a dead-letter handler) with `max_receive_count`, and the queue's `maxReceiveCount` higher than it ([Dead-letter handling and DLQ forwarding](#dead-letter-handling-and-dlq-forwarding))
- [ ] **Shutdown**: `consumer.stop()`, `runner.stop()`, `publisher.close()`, `dlq.close()`, then `pool.drain_and_close()` ([Health, readiness and shutdown](#health-readiness-and-shutdown))
- [ ] **Tests**: unit tests use `eventcommon.mock`; nothing calls AWS ([Testing in consuming services](#testing-in-consuming-services))

### Production checklist

The deployment-time complement of the [Integration checklist](#integration-checklist): what to have in place before a service takes production traffic.

- [ ] **`RawMessageDelivery=true`** on every SNS → SQS subscription. Without it the consumer treats every message as malformed
- [ ] **AWS region and IAM**: `AWS_REGION` set for the `config` loaders, or `region=` / `AWS_DEFAULT_REGION` for configs built in code (construction fails without one); the role may `Publish` / `PublishBatch` to the topic, `ReceiveMessage` / `DeleteMessage` / `ChangeMessageVisibility` / `GetQueueAttributes` on the queue, and `GetQueueUrl` / `SendMessage` on the DLQ
- [ ] **Visibility and timeouts**: `SQS_VISIBILITY_TIMEOUT` ≥ 2× p99 handler duration; `SQS_HANDLER_TIMEOUT` set when handlers call slow dependencies; `SQS_RETRY_BACKOFF` set so a dependency outage does not dead-letter messages ([Suggested defaults](#suggested-defaults-production))
- [ ] **Schema as a deploy step**: `outbox.apply_schema` / `inbox.apply_schema` once per deploy with a DDL-privileged role (`MIGRATION_DATABASE_URL`) ([Migration safety](#migration-safety))
- [ ] **Probes**: readiness on `runner.ready`; liveness on `runner.healthy(max_age)` (or a heartbeat driven by it, as the example worker does), not on a database query ([Health, readiness and shutdown](#health-readiness-and-shutdown))
- [ ] **Drain budget**: `SQS_DRAIN_TIMEOUT` and `OUTBOX_DRAIN_TIMEOUT` (run concurrently) below `terminationGracePeriodSeconds`
- [ ] **Retention**: `prune_published` and `inbox.Store.prune` scheduled; dead-letter retention decided ([Pruning published records](#pruning-published-records))
- [ ] **Alerts**: the vendored `monitoring/` rules loaded, routed to the owning team ([Alerts, rules and dashboard](#alerts-rules-and-dashboard))
- [ ] **HMAC** only where the channel needs it (webhook / HTTP ingress), with keys ≥ 32 bytes from a secrets manager ([HMAC helpers](#hmac-helpers))

### 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 until published or dead-lettered
       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`, and its claim lease expires. The next runner instance (or the restarted process) picks it up.

#### Detecting misuse in code review

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

```python
# ✅ Correct — both writes in one transaction
async def save_and_announce(conn: Any) -> None:
    await repo.save_user(conn, user)
    await outbox.enqueue(conn, env)


await pool.run_in_tx(save_and_announce)


# ❌ Wrong — SNS call outside the transaction; event lost on crash
async def save(conn: Any) -> None:
    await repo.save_user(conn, user)


await pool.run_in_tx(save)
await publisher.publish(env)  # ← data inconsistency risk
```

### 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]`. This document uses these terms consistently:

| 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",            // omitted when empty (new_envelope sets none by default)
  "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:00.1234Z",  // RFC 3339, Go RFC3339Nano form (trailing zeros trimmed)
  "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 handler's `trace_id_from_context()` and its log lines carry no trace ID, which breaks the log-to-trace pivot structured logging provides;
> - an incident that starts as an HTTP 500 in service A and shows up as a data anomaly in service B can no longer be correlated in Tempo or Grafana without manual log searching. (The `sqs.receive` span is still linked to the producer's span through the `traceparent` attribute when a publish ran inside an active span.)
>
> For background jobs with no inbound HTTP trace, generate a trace ID at job startup and pass it to every event the job run emits, so at least the events of one run are grouped together.

**System-level events** (background jobs, scheduled tasks, cross-tenant operations) that are not scoped to any single tenant use `SYSTEM_TENANT_ID` (or `new_envelope(..., system_tenant=True)`, Go's `WithSystemTenant()`):

```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 it should be fixed, not papered over.
>
> `"system"` does **not** disable RLS: the consumer sets `app.tenant_id = 'system'` like any other tenant, so a global-event handler needs an RLS-bypassing pool or role. Misuse on a tenant-scoped event runs the handler's queries under the wrong tenant, so RLS policies either reject them or return the wrong rows. 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.MetricsIdentity` 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 the 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 build typed payloads with `parse_envelope(data, payload_type=YourDataclass)` or `decode_payload(env.data, YourDataclass)`, never `YourDataclass(**raw_dict)`. The former filter 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). `new_envelope` sets none, as Go's `NewEnvelope` does, so `specversion` is omitted from the wire and parses as `""`; pass `schema_version="1"` to start versioning a type. Increment it **only** when both conditions hold: (1) a new optional field was added, **and** (2) at least one consumer needs to branch on whether that field is present. A consumer that receives an unrecognised `specversion` should log a warning and drop the message rather than misparse it. Treat an empty `specversion` as `"1"` in consumer logic, but never rewrite it on the envelope: the HMAC covers the field as sent.

**Wire compatibility with Go.** Fields are serialised in Go's order, and `time` uses Go's RFC3339Nano form (up to 9 fractional digits, trailing zeros trimmed, the original UTC offset kept).
- **What parsing keeps.** It keeps the nanoseconds (`Envelope.time_ns_remainder`) and `data`'s received bytes. While `data` is unchanged, `to_json()` emits those bytes (compacted and HTML-escaped, as Go's `json.Marshal` emits a `json.RawMessage`) and HMAC signs them, so a forwarded or re-signed envelope verifies in both languages (`Envelope.data_json_bytes()` returns them). Changing `data` (any value, key order, or the sign of a zero) re-encodes it.
- **What `to_json()` writes.** What Go's `json.Marshal` of the envelope writes, byte for byte: compact, with `&`, `<`, `>` and U+2028 / U+2029 escaped in every string. The one exception is invalid UTF-8 in `data`, forwarded as U+FFFD.
- **How parsing reads.** It follows Go's decoder. Keys match case-insensitively. A duplicate member resolves as `encoding/json` resolves it (a later `null` keeps the earlier value of a string or `time` field). Nesting is capped at 10,000 levels, and integers of any length are kept.
- **Very long integers.** An integer longer than 4,300 digits (Python's int conversion limit) arrives as `eventcommon.envelope.LongInt`, a `Decimal`. Compare it, or convert it with `int(value)` (exact) or `str(value)`. Arithmetic on it raises rather than rounding, and `json.dumps` cannot encode it until it is converted.

`tests/unit/test_interop.py` checks all of this against golden vectors generated by platform-events v2.1.0.

#### Serialisation

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

# Unmarshal and validate (Go: ParseEnvelope). Raises EnvelopeParseError (a ValueError) for
# invalid JSON, a non-object body, a field of the wrong type or a non-RFC 3339 time;
# EnvelopeIDRequiredError / EnvelopeTypeRequiredError / EnvelopeSourceRequiredError for a
# missing field; EnvelopeTimeRequiredError for a missing or zero time.
env = parse_envelope(data, payload_type=UserCreatedPayload)
```

The SQS consumer is deliberately less strict: it accepts an envelope without `time` (`parse_envelope(..., require_time=False)`), as Go's consumer does.

#### 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. Use `decode_payload(env.data, YourDataclass)` (from `eventcommon.envelope`), never `YourDataclass(**env.data)`. It drops fields your dataclass does not declare, so a producer's new optional field cannot break the consumer, and it raises `TypeError` for a payload that cannot be your type. Treat that as permanent: log and return, or dead-letter.

```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 = decode_payload(env.data, UserCreatedPayload)
    # payload is now typed; env.id / env.tenant_id / env.trace_id are always available directly
    await repo.create_user(payload)
```

**Routing handlers** that inspect the event type and forward without parsing 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, and friendly to Postgres B-tree indexes. Use it as the idempotency key on the consumer side. `eventcommon._internal` generates UUID v7 itself, for portability down to Python 3.12; the stdlib `uuid` module only gained `uuid7()` 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 without affecting existing consumers. See [ARCHITECTURE.md § Envelope compatibility guarantees](ARCHITECTURE.md#envelope-compatibility-guarantees) for the per-field stability classes.

### 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=None,  # default: AWS_REGION, then AWS_DEFAULT_REGION / profile; none → ValueError
        endpoint_url=os.environ.get(
            "AWS_ENDPOINT_URL"
        ),  # set to http://localhost:4575 for the local floci (make docker-up)
        logger=logger,
    )
)
```

**Region:** `SNSPublisher`, `SQSConsumer` and `SQSDLQPublisher` resolve the region at construction: `region=`, else `AWS_REGION`, else boto3's chain (`AWS_DEFAULT_REGION`, the profile) — the Go SDK's order. (botocore alone does not read `AWS_REGION`; the library does.) When there is none they raise `ValueError("…: no AWS region configured — set AWS_REGION (or the config's region)")` instead of failing on the first call.

**Deadline:** `SNSConfig(call_timeout=…)` bounds each `Publish` / `PublishBatch` request, retries included (default 120 s; 0 disables). aiobotocore does not bound DNS resolution or the wait for a pooled connection. A timeout is a retryable error (per entry in `publish_batch`).

**Client lifecycle:**
- **One client per event loop.** The publisher opens one SNS client on first use (one per event loop) and reuses it, so a publish does not pay a TCP + TLS handshake. Publishers may be used from event loops on several threads.
- **Close it on shutdown.** Call `await publisher.close()`, or use `async with publisher:`, after the outbox runner has stopped, on the event loop that published. A client left open when its loop ends is reported with a warning. `SQSDLQPublisher` works the same way (`close()` / `async with`).
- **Injected clients.** A client passed as `client=` (Go: `NewWithClient`) belongs to the caller: it is used as-is and never closed.
- **Credentials that were not ready.** A client opened without credentials (`NoCredentialsError` / `PartialCredentialsError`, e.g. at pod start, before IRSA or the container-credentials endpoint is ready) is replaced, at most once per 5 s per event loop, so the publisher recovers without a restart. It is closed at once: those errors are raised while signing, so it never has a request in flight. The failure is retryable, so the outbox releases the record without counting an attempt.

#### 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 into requests of at most 10 entries and 256 KiB, and attempts
# every request even when one fails. Any failed entry raises one BatchError.
await publisher.publish_batch(envelopes)
```

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

Callers must inspect the error to tell a partial failure from a total one:

```python
from eventcommon.errors import BatchError

try:
    await publisher.publish_batch(envelopes)
except BatchError as exc:
    # str(exc): "events: N message(s) failed in batch (first: <id> <code>: <message>)".
    # exc.failures is list[BatchFailure(id, code, message, retryable)] — id is the
    # envelope ID. Successfully published messages must NOT be retried.
    by_id = {env.id: env for env in envelopes}
    retry = [by_id[f.id] for f in exc.failures if f.is_transient]
    for f in exc.failures:
        logger.warning("batch publish failure", event_id=f.id, code=f.code, error=f.message)
    # Retry transient failures, or hand off to the outbox for durable retry.
    await publisher.publish_batch(retry)
```

Failure codes (Go parity) are `InvalidEnvelope`, `CodecEncodeError`, `MarshalError`, `TooManyAttributes`, `TransportError` (the whole request failed, retryable), `PublishError`, or the AWS error code SNS returned. `retryable` is true for throttling / 5xx codes, or when SNS reports no sender fault; `is_transient` also covers `TransportError`.

**Key rules:**

| Rule | Reason |
|---|---|
| Never retry the full batch on a `BatchError` | Messages that succeeded would be published a second time, creating duplicates |
| Always map failures by `BatchFailure.id` | The envelope ID is the 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`, which requires content-based deduplication to be disabled on the topic.

> ⚠️ **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 (per tenant, per aggregate), and `Envelope.id` with `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. SQS subscription filter policies can then scope queues to event types, tenants or subjects without deserialising the body. `Actor` is an audit-trail field and is not forwarded as an attribute.

- **Static attributes.** `SNSPublisher(attributes={...})` adds them. A key that collides with one of the five routing attributes is ignored with a warning, and 5 or more extra attributes log a warning at construction.
- **Trace context.** The OTel propagator's `traceparent`, `tracestate` and `baggage` are injected as attributes, so the consumer can link its span to the producer's.
- **The 10-attribute limit.** SNS allows 10 attributes. `baggage`, then `tracestate`, are dropped first (logged once per publisher); if the message still does not fit, `publish` raises before calling SNS.

#### Schema registry codec

`eventcommon` adds no registry SDK dependency. For consumers of Glue-framed payloads, it ships the decode-only `eventcommon.codec.GlueDecodeCodec` (Go: `events.GlueDecodeCodec`). It strips Glue's 18-byte header and returns the JSON payload, with no registry client or AWS permission needed, and it rejects compressed and non-JSON (e.g. Avro) payloads. To encode, implement `eventcommon.codec.Codec` against your own registry client and pass it to `SNSPublisher` / `SQSConsumer`:

```python
from eventcommon.codec import Codec


class GlueCodec:
    async def encode(
        self, event_type: str, payload: bytes
    ) -> tuple[bytes, str]: ...  # returns (registry-encoded bytes, schema ID)

    async def decode(
        self, schema_id: str, encoded: bytes
    ) -> bytes: ...  # JSON → parsed data; else raw bytes


publisher = SNSPublisher(SNSConfig(topic_arn=...), codec=GlueCodec())
consumer = SQSConsumer(SQSConfig(queue_url=...), handle_event, codec=GlueCodec())
```

- **No codec (the default, `None`).** `data` stays plain JSON and `schema_id` stays empty.
- **Publishing with a codec.** `Publisher.publish` base64-wraps the codec's encoded bytes into `data` and stamps `schema_id`. An encode failure raises `CodecEncodeError`, wrapped in `RetryableError` when the codec's error was retryable.
- **Consuming with a codec.** `SQSConsumer` decodes a message whose `schema_id` is non-empty **and** whose `data` is a JSON string (the codec wire format); a `dataschema` on plain-JSON `data` is passed through undecoded. A decode failure is logged and the message is left visible for redelivery; the handler is never called with an undecoded payload.

See `codec.py`'s docstrings for the full contract, and `eventcommon.codec.NoopCodec` for a reference identity implementation.

### 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=60,  # seconds; 0 (default) = the queue's timeout, no extension
    # Optional (Go option in brackets; all off by default):
    handler_timeout=300,  # [WithHandlerTimeout] bounds decode + DLH + handler
    retry_backoff=60,  # [WithRetryBackoff] failed delivery n hidden 60s·2^(n-1)…
    max_retry_backoff=900,  # …capped at 15m (+ up to 10% jitter)
    dlq_publisher=SQSDLQPublisher(DLQConfig(consumer_name="billing")),  # [WithDLQForwarding]
    max_receive_count=5,  # with a DLQ publisher / dead-letter handler (default 5)
    queue_depth_interval=60,  # [WithQueueDepthMetrics] platform_queue_depth / _dlq_depth
    # To load from env: **sqs_consumer_options(load_sqs())
)

task = asyncio.create_task(consumer.start())  # runs until stop() is called
...
await consumer.stop()  # drain — waits up to drain_timeout (30s), then cancels handlers
```

Behaviour worth knowing (all Go v2.1.0 parity):

- **Concurrency** is enforced at dispatch: a received message waits for a free worker, and its visibility is extended from receipt while it waits. With `visibility_timeout=0`, a receive asks only for as many messages as there are free workers.
- **FIFO queues** (URL ending in `.fifo`) are processed per `MessageGroupId`, in receive order, one message per group at a time. A message that is not settled hands the group's later messages of that batch back unprocessed, so order holds. Groups run in parallel up to `concurrency`, and each receive carries a fresh `ReceiveRequestAttemptId`.
- **Retry backoff** applies to every unsettled message: a handler, decode or dead-letter-handler error, a `handler_panic`, or a failed DLQ forward. The message is hidden for `retry_backoff·2^(n-1)` after its n-th delivery, through `ChangeMessageVisibility`. Without it, a failed message reappears after the visibility timeout, so a dependency outage longer than about `maxReceiveCount × VisibilityTimeout` dead-letters messages that would have succeeded.
- **Handler timeout** cancels the handler's task, counts `platform_message_timeouts_total{operation}` and stops extending visibility, so the message is redelivered. A FIFO group's waiting messages share the running message's budget.
- **Timeouts cancel; the handler decides** (Go: the deadline or drain timeout cancels the handler's `ctx`). At `handler_timeout` or `drain_timeout`, the handler task is cancelled, which interrupts the `await` it is in, as Go's ctx-aware I/O returns `ctx.Err()`. The outcome is then the handler's:
  - If it raises, the message is a `handler_error` (plus `platform_message_timeouts_total` past the deadline) and is retried.
  - If it catches `CancelledError` (Python's way of ignoring the cancellation) and returns normally, the message is deleted and counted processed, like a Go handler returning `nil` after its deadline.
  - The same holds for codec decode and the dead-letter handler. After the drain timeout `start()` returns, but the SQS client stays open until such a handler finishes, so its late success still deletes the message.
- **Lifecycle.**
  - `start()` is a plain method that returns the coroutine (Go: `Start(ctx)`). A `stop()` that runs after `create_task(consumer.start())` but before the task first runs is therefore not lost: that start returns at once. A `start()` that was cancelled or never ran leaves no trace.
  - `start()` raises `RuntimeError` if the consumer is already running, and can be called again after `stop()`. It opens its own SQS client (unless one was injected with `client=`) and closes it when it returns, or, after a drain timeout, once a handler that ignored the cancellation has finished.
  - `stop()` is safe before `start()` and more than once. It waits until `start()` has returned, and raises `TimeoutError` if that takes longer than `drain_timeout` + 5 s.
  - Called from a handler (or the `start()` task), `stop()` only signals, because waiting there would deadlock; `start()` returns once that handler has finished. Any other task waits, including one a handler spawned (e.g. a shutdown coroutine that then closes the DB pool), as Go's `Stop` does, so a handler must not await such a task.
  - Cancelling the `start()` task drains, then raises `CancelledError` as usual.
- **Handler typing.** Where a handler is accepted (`SQSConsumer`, `dead_letter_handler`, `with_dead_letter_handler`, `MockConsumer.set_handler`, `inbox.handler`), it is typed `AcceptedHandler`: an `Envelope` of what the consumer delivers, which is parsed JSON (`Envelope[dict[str, Any]]`, `Envelope[Any]`) or `bytes` from a non-JSON codec. A handler annotated `Envelope[YourModel]` is a mypy error there, because `env.data` is never `YourModel` at runtime: parse it in the handler (`parse_envelope(…, payload_type=YourModel)` is typed `Envelope[YourModel]`). Annotate middleware and route tables with `Handler` (`Callable[[Envelope[Any]], Awaitable[None]]`), which can be called with any envelope.
- **Synchronous handlers** (`def`, not `async def`) are supported but block the event loop while they run; the consumer logs `sqs: handler is synchronous — it blocks the event loop; use async def` once. Their return value is final, exactly as for an `async def` handler: a plain return deletes the message and a raise retries it. The same applies to a synchronous dead-letter handler.
- **Tracing:** the `sqs.receive` span (tracer `platform-events`, CONSUMER kind) starts a new trace **linked** to the producer's span through the message's `traceparent`. It is rooted in an empty context, never in the span that was current when `start()` was called, and it carries the producer's baggage into the handler.
- **Context:** inside a handler, `eventcommon.context.source_message_from_context()` returns the raw SQS message (queue URL, message ID, body, attributes, receive count), and `trace_id_from_context()` returns the envelope's trace ID.

#### Handler contract

| Handler outcome | Consumer behaviour |
|---|---|
| Returns normally | SQS message deleted from the queue |
| Raises an exception, or times out | SQS message left visible; retried after `retry_backoff`, or the visibility timeout |
| Raises a "panic" (a `BaseException` that is not an `Exception`, other than `CancelledError`, `KeyboardInterrupt` and `SystemExit`) | Recovered, logged with its stack (`sqs: handler panic recovered`), counted as `reason="handler_panic"`; retried |
| Malformed body (not an envelope: invalid JSON, missing `id` / `type` / `source`, …) | Forwarded to the DLQ (with `dlq_publisher`) or deleted immediately; counted as `platform_messages_failed_total{reason="malformed"}` — never retried |

**`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 `concurrency`. On `stop()`, the consumer stops receiving new messages and awaits in-flight handler tasks for up to `drain_timeout`, then cancels them. Use `handler_timeout` to bound a single handler call.

#### Error classification

The handler's outcome (return normally, or raise) is the only signal the consumer uses to decide between retrying and deleting. **Getting this backwards decides whether a broken message sits in SQS forever or is 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 `platform_messages_failed_total{reason="handler_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 makes the handler raise, SQS retries it `MaxReceiveCount` times and then pushes it to the DLQ, where it sits indefinitely. Return normally (and log) for any message that cannot succeed on retry, however many times it is delivered.

**Classification pattern:**

```python
async def handle_user_created(env: Envelope[dict[str, Any]]) -> None:
    try:
        payload = decode_payload(env.data, UserCreatedPayload)
    except TypeError as exc:
        # Permanent: a payload that cannot be this type will never parse on retry.
        # (decode_payload drops fields the dataclass does not declare, so a producer's
        # new optional field is not an error.)
        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
```

#### Dead-letter handling and DLQ forwarding

Messages that can never succeed reach dead-letter storage three ways:

| Path | Trigger | What happens |
|---|---|---|
| `max_receive_count` | `ApproximateReceiveCount > max_receive_count`, with a `dead_letter_handler` and/or `dlq_publisher` set | The dead-letter handler runs (it receives the parsed, decoded `Envelope`; the raw message is in `source_message_from_context()`). Then the message is forwarded to the DLQ (`DLQReason` `receive count N exceeded consumer max receive count M`) and deleted |
| Malformed body / decode failure | With `dlq_publisher`, a malformed body is forwarded at once, and a decode failure past the threshold is forwarded | Without a DLQ publisher, a malformed body is logged (size + SHA-256; `malformed_body_logging=True` adds a 512-byte excerpt) and deleted |
| Explicit | The handler calls `await forward_source_message(dlq, reason)` and returns normally | Counted as `platform_dlq_messages_total{reason="explicit"}`, not as processed |

`SQSDLQPublisher` (Go: `NewSQSDLQPublisher`) works like this:
- **Which DLQ.** It resolves the DLQ from the source queue's `RedrivePolicy`, cached for `DLQConfig.cache_ttl` (default 15m). Concurrent forwards share one lookup per queue, bounded by 30 s. With forwarding enabled, `start()` fails when the queue has none.
- **What it sends.** Each forwarded message keeps the original body and String/Number attributes, and gains `EventType`, `DLQReason`, `OriginalQueue`, `FailedAt` and (when configured) `ConsumerName`. Lower-priority caller attributes are dropped to fit SQS's 10-attribute limit (`strict_attributes=True` raises instead).
- **FIFO DLQs.** `MessageGroupId` is the envelope ID, and `MessageDeduplicationId` is unique per forward.
- **Failures.** They raise a `DLQError` subclass: `DLQNotConfiguredError`, `DLQInvalidRedrivePolicyError`, `DLQUnresolvedError`, `DLQSendFailedError` or `DLQInvalidMessageError`, with `.retryable` set for throttling and 5xx. The original is deleted only after the forward succeeded.

A dead-letter handler that raises (or panics) leaves the message visible, and it is not forwarded.

> **Breaking change in 0.3.0:** `dead_letter_handler` used to receive the raw SQS message dict before parsing; it now receives the `Envelope`, as Go's does.

#### Poison messages

A **poison message** is structurally valid JSON (it parses without error) but cannot be processed successfully however many times it is retried. It is distinct from a transient error (which may succeed on retry) and from a malformed message (which fails 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 = decode_payload(env.data, OrderShippedPayload)
    except TypeError:
        return  # cannot be an OrderShippedPayload — 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: dead-letter it for later investigation instead of dropping it.
        await forward_source_message(dlq, "illegal state transition: order cancelled")
        return

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

**Decision rules:**

1. **Return normally; don't raise.** Raising retries the message, and a semantically invalid message never becomes 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, and the log is the only record that something was discarded.
3. **Optionally dead-letter it** with `forward_source_message(dlq, reason)` when the business impact warrants investigation or replay.
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`, because of 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 0 — `eventcommon.inbox` (recommended)

The inbox (Go: `pkg/inbox`) records each handled envelope ID per consumer in `processed_events`:

```python
from eventcommon import inbox

await inbox.apply_schema(pool)  # own tracking table: inbox_migrations
store = inbox.Store(pool, "billing-invoice-settled")  # stable consumer name

# Either wrap the handler — a redelivery is acknowledged without calling it
# (counted in platform_duplicate_messages_total):
consumer = SQSConsumer(config, inbox.handler(store, handle_invoice_settled))


# …or, when the side effects are Postgres writes, claim the ID in the same transaction
# (exactly-once writes; concurrent copies serialise on the claim):
async def handle_invoice_settled(env: Envelope[dict[str, Any]]) -> None:
    async def write(conn: Any) -> None:
        await repo.mark_settled(conn, env.data["invoice_id"])

    await store.process(env, write)


# Periodically: keep at least the queue's retention plus any replay window.
await store.prune(timedelta(days=7))
```

The envelope ID must be a UUID: otherwise `InvalidEventIDError` is raised, so the message reaches the DLQ instead of being processed unguarded.

These are retried, because the transaction is aborted and its writes and the claim roll back:
- a `write` that ends its own transaction (`TxEndedInCallbackError`), including one that ends it and begins another (`ROLLBACK; BEGIN`);
- a `write` that catches a failed statement and returns normally (`TxCommitRollbackError`). On pgcommon 0.2, that `COMMIT` would have reported success and the message would have been deleted unprocessed.

To tolerate an expected failure (e.g. a unique violation), run that statement in a savepoint (`conn.transaction()`). A message the handler dead-lettered is not recorded, so redriving the DLQ after a fix processes it.

#### Pattern 1 — Postgres unique constraint (hand-rolled)

```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 def tx(conn: Any) -> None:
        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 = decode_payload(env.data, UserCreatedPayload)
        await repo.create_user(conn, payload)

    await pool.run_in_tx(tx)
```

**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, which is correct. 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

When the side effect is a row update rather than an insert, record the last applied event on the row itself and make the update conditional on it:

```sql
UPDATE invoices
   SET status = 'settled', last_event_id = $2
 WHERE id = $1 AND last_event_id IS DISTINCT FROM $2;
-- 0 rows: this event was already applied (or the invoice does not exist — check separately)
```

This only works for updates that are safe to apply once per event regardless of order. For anything else, use Pattern 0 or 1.

#### 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 0 or 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 0 or 1 |
| 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 | `inbox.Store.prune(retention)` on a schedule |

#### 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 the inbox claim (`store.process`) or `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, at least once. The outbox does not guarantee global ordering; use `enqueue_ordered`, and a FIFO topic with a stable `message_group_id`, when ordering is required.

#### Wiring

```python
from eventcommon.outbox import Runner, RunnerConfig, apply_schema

# 1. Apply outbox schema migrations 001-011 — platform-events' files and tracking, so Go
#    and Python services can share the schema. See Migration safety below.
await apply_schema(pool)  # or apply_schema(pool, lock_timeout=60.0); eventcommon.inbox too

# 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; negative disables it
        retry_backoff=1.0,  # a record's n-th failed publish waits retry_backoff·2^(n-1)…
        max_retry_backoff=300.0,  # …capped at 5m (equal jitter)
        gauge_interval=15.0,  # backlog gauges + ordered-record promotion sweep
        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)
```

**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 · `retry_backoff` 1s · `max_retry_backoff` 5m · `gauge_interval` 15s.

`claim_lease_duration` must be at least `batch_size × publish_timeout + 60s` (30s when the timeout is disabled), else `RunnerConfig` raises `ValueError`. The runner re-polls immediately while batches keep publishing (for at most `poll_interval`). On shutdown, claimed records that were not yet sent are released at once instead of staying hidden for the lease.

**A stalled database** (a hung primary, a half-open connection after a failover) makes each poll fail within the store's 5 s query timeout. The failure is counted in `platform_outbox_errors_total{operation="poll"}` and logged, and the runner recovers by itself. This relies on pgcommon ≥ 0.3, whose timed-out queries clean up without waiting on the frozen server. Drive a liveness probe from `runner.healthy(...)`, not from whether the task is alive; see [Health, readiness and shutdown](#health-readiness-and-shutdown).

#### Enqueueing inside a transaction

```python
async def tx(conn: Any) -> None:  # pgcommon runs it in one transaction (Go: RunInTx)
    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)


await pool.run_in_tx(tx)  # or pgcommon.run_in_tx(pool, tx)
```

`enqueue` requires an open transaction and validates the envelope as Go does:
- non-empty `id` / `type` / `source`, a non-zero `time`, and a canonical lowercase UUID `id`;
- no NUL character in `id` / `type` / `source` / `tenant_id` / `trace_id` or anywhere in the JSON. Postgres `jsonb` cannot store `\u0000`, so the INSERT would roll back the business transaction on every retry; strip NULs from user input first;
- a serialised size of at most **240 KB**, under the SNS 256 KB hard limit, so an outbox record that could never publish is never persisted.

#### Ordered outbox

```python
async def tx(conn: Any) -> None:
    await conn.execute(
        "SELECT 1 FROM orders WHERE id = $1 FOR UPDATE", order.id
    )  # serialise per key
    await outbox.enqueue_ordered(conn, env, f"order/{order.id}")


await pool.run_in_tx(tx)
```

Records with the same ordering key (≤ 256 bytes) are published one at a time, in enqueue order, across all runner replicas.
- A record enqueued while an earlier one of its key is unpublished waits (`scheduled_at = 'infinity'`), and is promoted when that one is published or dead-lettered.
- A failing head record holds its key until it is dead-lettered; `platform_outbox_ordering_blocked_events` counts the waiting records.
- A replayed dead letter joins the back of its key.

On a FIFO topic, derive `message_group_id` from the same key. Requires migration 0010.

#### 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. A row written around `enqueue` that SNS rejects is dead-lettered after `max_attempts`; one over 1 MiB as stored is not even loaded by the runner (a memory guard) |
| **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 side-by-side timeline and the 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`.

**Transient failures** never count toward `max_attempts`. They are throttling and 5xx codes, network errors, timeouts, and `BatchFailure`s that are retryable or `TransportError`. The record's lease is released with a shared backoff that advances at most once per poll and resets on the next successful publish, so a period of SNS unavailability does not dead-letter healthy records. Permanent failures count an attempt and wait `retry_backoff·2^(n-1)` (capped at `max_retry_backoff`). A publisher "panic" (a `BaseException` that is not an `Exception`) is recovered, logged with its stack, and counts an attempt, so a poison batch cannot hold its lease.

Replay inserts into `outbox_events` first and deletes only the dead letters it inserted. A dead letter whose ID is already back in `outbox_events` is left in place (logged). List, replay and discard order by `(failed_at, id)`, and `limit` must be positive.

#### 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 every consumer has processed the event before the row is deleted; 7 days covers most SLA windows. `outbox_dead_letters` has no automatic TTL; schedule your own cleanup (see [ARCHITECTURE.md § Dead-letter table retention](ARCHITECTURE.md#dead-letter-table-retention)).

#### 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, or of their ordering key;
- the same `Envelope.id` is presented again, so idempotency guards must be in place;
- 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 holding the shared key, and was not tampered with in transit. It does **not** provide confidentiality or replay protection: a captured signed message verifies again. Where replay matters (e.g. webhook ingress), the receiver must reject it: check that `time` is recent (allowing for clock skew), and that `Envelope.id` has not been seen within that window (the inbox, or any ID store).

| 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 it with string `==`. **Key length:** `sign` raises `KeyTooShortError` for keys under 32 bytes.

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

### Health, readiness and shutdown

| Signal | Use | How |
|---|---|---|
| **Readiness** | Take traffic only once the database is reachable and the schema exists | `await asyncio.wait_for(runner.ready.wait(), timeout=…)` — `ready` is set after the first successful poll |
| **Liveness** | Restart a worker whose database work has stalled | `runner.healthy(max_age)`, e.g. `3 * poll_interval + 30`: True while a publish bounded by `publish_timeout` is in flight, or a store round trip succeeded within `max_age`. It tracks the database, not SNS speed (Python-only; Go has `Ready()` only). `runner.last_poll_succeeded` is the monotonic time behind it |
| **Consumer health** | Detect a stuck or saturated consumer | `platform_messages_in_flight` against `concurrency`, `platform_queue_depth` (with `queue_depth_interval`), and the `start()` task: it only returns on `stop()` or a fatal error |
| **Shutdown** | Fit inside `terminationGracePeriodSeconds` | `await asyncio.gather(consumer.stop(), runner.stop())`, then `publisher.close()`, `dlq.close()`, `pool.drain_and_close()` |

The example worker drives a heartbeat file from `runner.healthy(...)` (the image's `HEALTHCHECK` reads it), retries the first database connection until `STARTUP_TIMEOUT`, and bounds the whole SIGTERM shutdown by `SHUTDOWN_TIMEOUT`. Use it as the template for a service's entrypoint.

### Failure model

| Failure | Behaviour |
|---|---|
| Invalid configuration (topic ARN, queue URL, region, `RunnerConfig`, metric identity) | Raised at construction; the process should not start |
| A queue with DLQ forwarding but no usable RedrivePolicy | `consumer.start()` raises `DLQNotConfiguredError` / `DLQInvalidRedrivePolicyError` |
| SNS throttling, 5xx, network or DNS failure, `call_timeout` | `RetryableError`; the outbox releases the record without counting an attempt (shared backoff) |
| Permanent SNS error (authorization, not found, invalid parameter) | The outbox counts an attempt and backs off; dead letter at `max_attempts` |
| Credentials not yet available (pod start) | The client is replaced (≤ 1 per 5 s per loop); the error is retryable; recovers without a restart |
| SQS receive failing | Logged; backoff 1 s → 30 s with jitter; the loop continues |
| Malformed body | DLQ (with `dlq_publisher`) or deleted on first receipt; never retried; logged as size + SHA-256 |
| Codec decode failure | `decode_error`; left visible and retried; DLQ past the threshold |
| Handler raises / times out / panics | Left visible; `retry_backoff` when set; dead-letter handler + DLQ past `max_receive_count` |
| Dead-letter handler raises | The message stays visible; it is not forwarded |
| DLQ lookup or send failure | The original stays on the source queue (retryable for throttling, 5xx and timeouts) |
| Postgres unreachable or stalled | Each poll fails within 5 s, counted; backoff 1 s → 30 s; `healthy(max_age)` turns False; recovers |
| `mark_published` fails after SNS accepted the event | The record is published again after its lease: a duplicate, absorbed by idempotent consumers |
| Process killed mid-batch | Claimed records reappear after `claim_lease_duration`; in-flight SQS messages after their visibility timeout |
| `stop()` while handlers run | Waits `drain_timeout`, then cancels; cancelled messages are redelivered without delay |
| A payload over 1 MiB in `outbox_events` (written around `enqueue`) | Not loaded; counts an attempt; dead-lettered at `max_attempts` |
| A metric name already registered with another shape | That metric is disabled; a `RegistrationWarning` is returned; everything else works |

Errors are never swallowed: every path either propagates the original exception or chains it (`raise … from exc`). ARCHITECTURE.md has the full [degradation matrix](ARCHITECTURE.md#failure-domains).

### Backpressure considerations

Under sustained load, the SQS backlog and the `outbox_events` table both grow if consumers or the outbox runner cannot keep up.

#### Consumer-side (SQS)

**Tuning order:**
1. Raise `concurrency`; it 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.

`monitoring/kubernetes/keda-scaledobject.example.yaml` scales a consumer on `platform_queue_depth`.

#### Outbox-side (Postgres → SNS)

**Tuning order:**
1. Decrease `OUTBOX_POLL_INTERVAL`.
2. Increase `OUTBOX_BATCH_SIZE`, keeping it below 100 to avoid long-held Postgres locks.
3. Scale runner pods horizontally; `SKIP LOCKED` distributes the work with no coordination.
4. Check SNS publish errors: a rising `platform_outbox_publish_attempts_total{outcome="error"}` rate usually means 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 or receive size.

### Performance considerations

#### Overhead

eventcommon is a thin layer over aioboto3 and pgcommon. Network latency to SNS, SQS and Postgres dominates every path:

| Mechanism | When it runs | Approximate cost | Notes |
|-----------|-------------|------------------|-------|
| `new_envelope` | Every event | Microseconds | UUID v7, `datetime.now(UTC)`, object construction |
| `outbox.enqueue` | Every transactional event | One `INSERT` in the caller's transaction | Plus serialisation and validation; no AWS call |
| Runner poll, empty | Every `poll_interval` | One claim query | Gauge queries every `gauge_interval` (15 s), capped at `MAX_COUNTED_ROWS` |
| Runner poll, full batch | While there is a backlog | `ceil(batch_size / 10)` `PublishBatch` calls (`publish_concurrency=1`), or one `publish` per record | Successes settled with one `mark_published_many` write |
| Consumer, per message | Every delivery | One delete, plus one visibility extension every `max(visibility_timeout/2, 1 s)` while it runs | Dispatch is a semaphore acquire and a task |
| Metrics, spans | Every operation | Microseconds, in-process | Export is the OTel SDK's, off the path |
| `verify` (HMAC) | Per call | Microseconds | Canonical JSON + HMAC-SHA256 + constant-time compare |

See [ARCHITECTURE.md § Performance characteristics](ARCHITECTURE.md#performance-characteristics).

#### Tuning

- **Keep handlers short and async.** A worker slot is held for the whole handler; a synchronous or CPU-bound handler blocks every other message on the loop. Use `asyncio.to_thread` for blocking work.
- **Keep external I/O out of the outbox transaction.** `enqueue` adds one insert; anything slow you add to the business transaction holds its locks longer.
- **Use `publish_concurrency=1` (batch publishing) unless SNS latency is the bottleneck.** It needs a tenth of the API calls.
- **Raise `concurrency` before `max_messages`.** More workers use the long poll better than larger receives.

### Common pitfalls

---

#### ❌ Publishing directly after a database write

If the process dies after the commit but before `publish` returns, the event is silently dropped: no retry, no recovery.

```python
# ❌ WRONG: the event is lost if the process dies between the commit and the publish
await pool.run_in_tx(save_order)
await publisher.publish(order_placed)


# ✅ CORRECT: the event commits with the order, or neither does
async def tx(conn):
    await save_order(conn)
    await outbox.enqueue(conn, order_placed)


await pool.run_in_tx(tx)
```

#### ❌ Ignoring `Envelope.id` in the handler

The outbox and SQS deliver at least once; without an idempotency check, a redelivered event causes duplicate side effects.

```python
# ❌ WRONG: a redelivery charges the customer twice
consumer = SQSConsumer(config, charge_customer)

# ✅ CORRECT: redeliveries of the same Envelope.id are acknowledged without running it again
consumer = SQSConsumer(config, inbox.handler(inbox.Store(pool, "billing-charge"), charge_customer))
```

#### ❌ Raising for a message that can never succeed

Raising retries the message `max_receive_count` times and then dead-letters it. A validation failure never fixes itself.

```python
# ❌ WRONG: a permanently invalid payload is retried, then sits in the DLQ
payload = decode_payload(env.data, Order)  # TypeError propagates

# ✅ CORRECT: log and return, or dead-letter it explicitly
try:
    payload = decode_payload(env.data, Order)
except TypeError:
    await forward_source_message(dlq, "payload is not an Order")
    return
```

#### ❌ Branching on `env.data` without checking `env.type`

A handler subscribed to several event types silently misparses the wrong payload. Check `env.type` first, and route to typed handlers by event type.

#### ❌ A non-stable FIFO `MessageGroupId`

A random group ID per message puts every message in its own group, so SNS delivers them concurrently and ordering is lost. Use a deterministic, stable ID (`tenant_id`, `user_id`, an aggregate ID): a value that must be ordered relative to itself.

#### ❌ `SYSTEM_TENANT_ID` in an HTTP handler

The consumer's queries then run as `app.tenant_id = 'system'`, which RLS policies reject or answer with the wrong rows, silently. Pass `tenant_id=rc.tenant_id`; use `SYSTEM_TENANT_ID` only for genuine cross-tenant jobs.

#### ❌ Strict payload schemas

Parsing payloads with a schema that rejects unknown fields (`extra="forbid"` in Pydantic, a strict model) turns every non-breaking producer addition into a consumer runtime error. Parse permissively (`extra="ignore"`, `decode_payload`).

#### ❌ `enqueue` outside a transaction

The event is then not durably linked to the domain write. `enqueue` raises on a connection with no open transaction; always call it inside `pgcommon.run_in_tx`.

#### ❌ Swallowing `asyncio.CancelledError` in a handler

A handler that catches the cancellation keeps running past the drain deadline and the handler timeout. Let it propagate (clean up in `finally`).

#### ❌ Bare numbers in duration variables

`SQS_VISIBILITY_TIMEOUT=60` is rejected with a warning and the default (30 s) is used, exactly as Go does. Write `60s`, and log `.warnings` at startup so the mistake is visible.

#### ❌ Forgetting `RawMessageDelivery=true`

Without it SNS wraps the envelope in its own JSON notification, and the consumer treats every message as malformed and dead-letters it.

### Configuration reference

The components take dataclass configs plus keyword options. Every field but `topic_arn` / `queue_url` / `pool` + `publisher` is optional. Each can also come from the environment through `eventcommon.config` (see [Environment variables](#environment-variables)).

| Field | Env var | Default | Notes |
|---|---|---|---|
| `SNSConfig.topic_arn` | `SNS_TOPIC_ARN` | — | Required |
| `SNSConfig.region` / `SQSConfig.region` / `DLQConfig.region` | `AWS_REGION` | `None`: `AWS_REGION`, then boto3's chain (`AWS_DEFAULT_REGION`, profile) | No region → `ValueError` at construction |
| `SNSConfig.endpoint_url` (and SQS / DLQ) | `AWS_ENDPOINT_URL` | `None` | `http://localhost:4575` for the local floci |
| `SNSConfig.call_timeout` | — | `120` s | Bounds each SNS request, retries included; 0 disables (Python-only) |
| `SQSConfig.queue_url` | `SQS_QUEUE_URL` | — | Required |
| `SQSConfig.max_messages` / `wait_seconds` | `SQS_MAX_MESSAGES` / `SQS_WAIT_SECONDS` | `10` / `20` | 1–10 / the long poll |
| `concurrency` | `SQS_CONCURRENCY` | `1` | Parallel handlers (FIFO: message groups) |
| `visibility_timeout` | `SQS_VISIBILITY_TIMEOUT` | `0` (env loader: `30s`) | 0: the queue's timeout, nothing extended; ≤ 12 h |
| `handler_timeout` | `SQS_HANDLER_TIMEOUT` | `0` (off) | Bounds decode + dead-letter handler + handler |
| `retry_backoff` / `max_retry_backoff` | `SQS_RETRY_BACKOFF` / `SQS_MAX_RETRY_BACKOFF` | off / `15m` | Failed delivery n hidden `base·2^(n-1)`, +≤10% jitter |
| `max_receive_count` | `SQS_MAX_RECEIVE_COUNT` | `0` (5 with a DLQ publisher or dead-letter handler) | Above it: dead-letter handler + DLQ |
| `dead_letter_handler` / `dlq_publisher` | — | `None` | See [Dead-letter handling](#dead-letter-handling-and-dlq-forwarding) |
| `drain_timeout` | `SQS_DRAIN_TIMEOUT` | `30` s | `stop()` wait before cancelling handlers |
| `queue_depth_interval` | `SQS_QUEUE_DEPTH_INTERVAL` | `0` (off) | Samples `platform_queue_depth` / `platform_dlq_depth`; minimum 10 s |
| `malformed_body_logging` | — | `False` | Adds a 512-byte excerpt of malformed bodies to their log line |
| `codec` | — | `None` | See [Schema registry codec](#schema-registry-codec) |
| `DLQConfig.consumer_name` / `cache_ttl` / `strict_attributes` | — | `""` / `0` (15m) / `False` | `ConsumerName` attribute · DLQ lookup cache (negative: forever) · raise instead of dropping attributes |
| `RunnerConfig.poll_interval` | `OUTBOX_POLL_INTERVAL` | `5s` | |
| `RunnerConfig.batch_size` / `max_attempts` | `OUTBOX_BATCH_SIZE` / `OUTBOX_MAX_ATTEMPTS` | `50` / `5` | |
| `RunnerConfig.claim_lease_duration` | `OUTBOX_CLAIM_LEASE_DURATION` | `10m` | ≥ `batch_size × publish_timeout + 60s` |
| `RunnerConfig.publish_concurrency` / `publish_timeout` | `OUTBOX_PUBLISH_CONCURRENCY` / `OUTBOX_PUBLISH_TIMEOUT` | `1` / `10s` | 1 uses `PublishBatch`; negative timeout disables it |
| `RunnerConfig.retry_backoff` / `max_retry_backoff` | `OUTBOX_RETRY_BACKOFF` / `OUTBOX_MAX_RETRY_BACKOFF` | `1s` / `5m` | Equal jitter |
| `RunnerConfig.drain_timeout` / `startup_jitter` / `gauge_interval` | `OUTBOX_DRAIN_TIMEOUT` / `OUTBOX_STARTUP_JITTER` / `OUTBOX_GAUGE_INTERVAL` | `30s` / `0` / `15s` | |
| `RunnerConfig.logger` (and every `logger`) | — | `None` | structlog or a stdlib `logging.Logger`; `None` logs nothing |

#### Suggested defaults (production)

```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=60s   # comfortably above your p99 handler latency; 30s default is tight
SQS_HANDLER_TIMEOUT=5m       # release hung handlers for redelivery
SQS_RETRY_BACKOFF=60s        # ride out dependency outages instead of dead-lettering
SQS_CONCURRENCY=5            # Start here; scale up to ~10 before considering horizontal pod scaling

# Outbox runner
OUTBOX_POLL_INTERVAL=2s      # 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
OUTBOX_STARTUP_JITTER=5s     # desync replicas on a rolling restart
```

> **`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, and the message is delivered twice. Set it to 2–3× your p99 handler duration.

**What to monitor on day one:**

| Metric | Alert threshold | Action |
|---|---|---|
| `platform_outbox_oldest_pending_age` | > 5 min | Delivery is stalled: check SNS reachability and the runner |
| `platform_outbox_pending_events` | > 500 sustained for 5 min | Lower `OUTBOX_POLL_INTERVAL`; add runner pods |
| `platform_queue_depth` (or CloudWatch `ApproximateNumberOfMessages`) | > 1000 sustained | Raise `SQS_CONCURRENCY`; add consumer pods |
| `platform_messages_failed_total` / `platform_messages_received_total` | > 1% error rate | Inspect handler errors; check DLQ depth |
| `platform_dlq_messages_total` | rate > 0 for 15 min | Something is being dead-lettered; see the runbook |

The vendored `monitoring/prometheus/platform-events.rules.yml` encodes these and more.

---

## Upgrading from 0.2.0

0.3.0 brings the port in line with Go `platform-events` **v2.1.0**, and is breaking in several places (while the package is `0.y.z`, a MINOR bump may break the public API). The authoritative list is [CHANGELOG § Upgrade notes](CHANGELOG.md#upgrade-notes); this is the checklist.

| Change | What to check |
|---|---|
| The outbox schema cannot be upgraded in place (platform-events' migrations 001–011, golang-migrate tracking) | Drain the 0.2.0 outbox, drop `outbox_dead_letters`, `outbox_events`, `outbox_migrations`, then `apply_schema` with 0.3.0 |
| Only Tier 1 `platform_*` metrics remain; `metrics.init` / `init_with_registry` are gone | Call `init_metrics(MetricsIdentity(domain, service, environment))`; update dashboards to the vendored `monitoring/` bundle |
| `BatchFailure` is `(id, code, message, retryable)` | `envs[f.index]` → map by `f.id`; custom publishers raise `BatchError` and call `record_batch_published` |
| `dead_letter_handler` receives the `Envelope`; a DLQ handler/publisher without `max_receive_count` uses 5 | Handlers that read the raw SQS dict now use `source_message_from_context()` |
| Env durations need a unit; integers must be clean | `SQS_VISIBILITY_TIMEOUT=60` → `60s`; log `load_*().warnings` |
| `SQSConsumer(visibility_timeout=)` defaults to 0 | Pass it explicitly when constructing in code (`sqs_consumer_options` is unaffected) |
| A missing AWS region fails at construction | Pass `region=` (the `config` loaders pass `AWS_REGION`), or set `AWS_DEFAULT_REGION` |
| `parse_envelope` requires `time`; `specversion` is no longer sent by default | `require_time=False` for leniency; `schema_version="1"` to keep sending it |
| Payloads must be valid JSON (no `NaN`, no naive `datetime`) | Use aware datetimes; check for float edge cases |
| Publishers hold a long-lived client per loop | `await publisher.close()` / `async with` at shutdown |
| The drain timeout cancels handlers; `stop()` waits for `start()` and can raise | Handlers tolerate `CancelledError`; a handler must not await a task that calls `stop()` |
| `start()` returns the coroutine; sync handlers' results are final | `inspect.iscoroutinefunction(x.start)` is `False` |
| `pgcommon>=0.4.0,<0.5.0` | Your lock file; pgcommon's own 0.3 upgrade notes (0.4.0 changes only its FastAPI integration's error bodies) |
| The outbox backs off failed records by default; transient failures count no attempt | Dead-lettering takes ~8–15 s of retries instead of 5 polls |
| Tracer name `platform-events`; `sqs.receive` rooted in an empty context | Span filters and sampling rules |
| `Handler` / `AcceptedHandler` typing | Handlers annotated `Envelope[YourModel]` parse inside the handler instead |

---

## Local development

### Prerequisites

- Python ≥ 3.12 (CI runs 3.12, 3.13 and 3.14; `uv python install 3.12` if you have none)
- [uv](https://docs.astral.sh/uv/): the environment is `uv sync` from `uv.lock`
- Docker: testcontainers for the integration and e2e suites, and `make docker-up` for the local stack
- Optional: `hadolint` and `trivy` for the image gates (`brew install hadolint trivy`); `promtool` for `make rules-check` (else it uses Docker); `gh` and `cosign` ≥ 2.4 to verify a release; a sibling `platform-events` checkout for `make go-parity`

### Setup

```bash
git clone https://github.com/BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon
cd platform-eventcommon
make setup       # uv sync + copy .env-example → .env + git hooks
make docker-up   # floci (demo topic + queue + DLQ), Postgres on :5539, the observability stack
make ci          # everything CI's validate jobs run, in the same order (Docker required for the test suites)
```

### Common commands

| Command | Description |
|---|---|
| `make setup` / `make install` | `uv sync`, copy `.env-example` to `.env` if missing (never overwrites) and `make install-hooks` / `uv sync` only |
| `make install-hooks` | Point git at `.githooks/` (`core.hooksPath`): the pre-commit hook runs lock-check, format-check, lint and typecheck (`git commit --no-verify` skips it) |
| `make pin-base-images` / `make docs-serve` | Re-resolve the Dockerfile's base images to their current `@sha256:` digests / browse the API docs with `pydoc` |
| `make format` / `make format-check` | `ruff format` + `ruff check --fix` (rewrites files) / `ruff format --check` (CI). Aliases `fmt` / `fmt-check` |
| `make lint` / `make typecheck` | `ruff check` / `mypy --strict` |
| `make lock-check` / `make pip-check` / `make vulncheck` | `uv lock --check` / `uv pip check` (alias `deps-check`) / `pip-audit` (alias `vuln-check`) |
| `make arch-lint` | The six import-linter contracts (the same script CI runs) |
| `make metrics-lint` / `make metrics-doc` / `make rules-check` | Observability-standard conformance + inventory + `monitoring/` assets / regenerate `docs/observability/metrics-registry.md` / promtool rules and alert tests |
| `make go-parity` | Vendored migrations, `monitoring/` and log texts against a platform-events checkout (`PLATFORM_EVENTS_DIR=…`, else a sibling checkout) |
| `make docs-check` | Every `docs/architecture/mermaid/*.mmd` embedded byte-identically in `ARCHITECTURE.md` |
| `make ci-scripts-test` | Regression tests for the CI shell scripts (`detect-changes.sh`) |
| `make test-unit` · `make test-int` · `make test-e2e` · `make test` | Unit (no Docker) · integration (Postgres + floci; alias `test-integration`) · e2e (the full pipeline) · all three in one run (alias `test-all`) |
| `make test-ci` / `make cover-check` / `make cover` / `make cover-func` / `make test-lowest` | Every suite with coverage, no threshold (CI's test gate) / the same + the ≥ 95% gate / HTML report in `htmlcov/` / per-file missing lines / every suite at the dependency floors (replaces `.venv`) |
| `make build` / `make build-constraints` | Wheel + sdist into `dist/`, `twine check --strict`, an install check / re-lock the build backend with hashes |
| `make api-compat` / `make release-guard TAG=vX.Y.Z` | griffe public-API diff against the last release tag / pre-tag check (package version = tag, a dated CHANGELOG section) |
| `make docker-build` / `make lint-docker` / `make scan` / `make smoke-test` | Build the example-worker image / hadolint / Trivy, CI's severities / build + CI's `smoke-tests.sh` |
| `make docker-up` / `make run` / `make docker-down` / `make docker-reset` / `make docker-logs` / `make docker-ps` | The local stack / + the example worker / stop and remove it, volumes included / down + up / tail logs / list containers |
| `make psql` / `make show-config` | psql in the local Postgres / `eventcommon-config` with `.env` loaded (credentials masked) |
| `make ci` | lock-check + pip-check + format-check + lint + typecheck + metrics-lint + rules-check + docs-check + ci-scripts-test + vulncheck → test-ci → coverage gate → arch-lint → build |
| `make clean` / `make help` | Remove `.venv`, caches, build artefacts and coverage output / list every target |

Always run uv commands from a shell without another project's virtualenv active (`env -u VIRTUAL_ENV make …` if one is), or uv warns and may use the wrong environment.

### Running a single test

```bash
uv run pytest tests/unit/test_envelope.py -k test_sign -v --no-cov
uv run pytest tests/integration/test_sns_publisher_floci.py -k test_publish_delivers_a_real_message_with_attributes -v --no-cov
uv run pytest tests/e2e/test_full_pipeline.py -v --no-cov
```

`--no-cov` matters: `pyproject.toml`'s pytest `addopts` collect coverage, and the 95% `fail_under` would fail a subset of the suite on coverage even when its tests pass. Do not pass `-q`; it clashes with the configured `addopts`.

### Developer tools

```bash
make docker-up
uv run eventcommon-config --strict   # what the loaders resolve from .env, with problems and reminders
make run                             # the example worker against the local stack (floci + Postgres)
```

`uv run python` gives a REPL with the package installed in editable mode, for trying the API against the same stack. The floci UI at http://localhost:4506 shows topics, queues and messages.

---

## Testing SNS/SQS locally

The fastest way to see the delivery contract working, or to debug a consumer, is to drive the local stack by hand. `make docker-up` provisions what the library expects (`docker/floci/init-floci.sh`): topic `eventcommon-demo`, queue `eventcommon-demo-q` subscribed with `RawMessageDelivery=true`, and its DLQ `eventcommon-demo-q-dlq` through a RedrivePolicy with `maxReceiveCount` 5.

### How it works

```
publish  ── SNSPublisher → floci SNS (eventcommon-demo) → SQS eventcommon-demo-q (raw delivery)
consume  ── SQSConsumer receives → handler returns → deleted
fail     ── handler raises → visible again → … → past max_receive_count → DLQ eventcommon-demo-q-dlq
```

### Step 1 — Start the stack

```bash
make docker-up
set -a; . ./.env; set +a      # AWS_ENDPOINT_URL=http://localhost:4575, SNS_TOPIC_ARN, SQS_QUEUE_URL, test credentials
```

### Step 2 — Publish an event

Save as `publish_demo.py` and run `uv run python publish_demo.py`:

```python
import asyncio
import os

from eventcommon.envelope import new_envelope
from eventcommon.publisher import SNSConfig, SNSPublisher


async def main() -> None:
    config = SNSConfig(
        topic_arn=os.environ["SNS_TOPIC_ARN"],
        region=os.environ["AWS_REGION"],
        endpoint_url=os.environ["AWS_ENDPOINT_URL"],
    )
    async with SNSPublisher(config) as publisher:
        await publisher.publish(
            new_envelope("demo.thing.happened", "demo", {"n": 1}, tenant_id="acme")
        )


asyncio.run(main())
```

### Step 3 — Look at the raw message

```bash
aws --endpoint-url "$AWS_ENDPOINT_URL" sqs receive-message --queue-url "$SQS_QUEUE_URL" \
  --message-attribute-names All --visibility-timeout 0
```

The body is the envelope itself (raw delivery), and the attributes include `EventType`, `TenantID`, `Source` and `EventID`.

### Step 4 — Consume it, then fail it into the DLQ

```python
import asyncio
import os

from eventcommon.consumer import DLQConfig, SQSConfig, SQSConsumer, SQSDLQPublisher


async def handle(env):
    print("got", env.type, env.tenant_id)
    raise RuntimeError("simulated failure")  # remove to see the message deleted


async def main() -> None:
    region, endpoint = os.environ["AWS_REGION"], os.environ["AWS_ENDPOINT_URL"]
    async with SQSDLQPublisher(DLQConfig(region=region, endpoint_url=endpoint)) as dlq:
        consumer = SQSConsumer(
            SQSConfig(
                queue_url=os.environ["SQS_QUEUE_URL"],
                region=region,
                endpoint_url=endpoint,
                wait_seconds=1,
            ),
            handle,
            visibility_timeout=1,
            max_receive_count=2,
            dlq_publisher=dlq,
        )
        task = asyncio.create_task(consumer.start())
        await asyncio.sleep(10)
        await consumer.stop()
        await task


asyncio.run(main())
```

After the third delivery the message is forwarded to `eventcommon-demo-q-dlq` with `DLQReason`, `OriginalQueue` and `FailedAt` attributes:

```bash
aws --endpoint-url "$AWS_ENDPOINT_URL" sqs receive-message \
  --queue-url "${SQS_QUEUE_URL}-dlq" --message-attribute-names All
```

### Troubleshooting

| Symptom | Likely cause | Fix |
|---|---|---|
| Every message counted `malformed` | The subscription lacks `RawMessageDelivery=true`, so the body is SNS's notification JSON | Set the subscription attribute |
| `ValueError: … no AWS region configured` | No `region=`, no `AWS_REGION` / `AWS_DEFAULT_REGION`, and no profile region | Export `AWS_REGION` (the `.env` sets `us-east-1`) or pass `region=` |
| `start()` raises `DLQNotConfiguredError` | `dlq_publisher` set on a queue without a RedrivePolicy | Add the RedrivePolicy, or drop DLQ forwarding |
| Messages go to the DLQ without the library's attributes | The queue's `maxReceiveCount` is at or below `max_receive_count`, so SQS moved them itself | Keep the queue's `maxReceiveCount` higher |
| A handler runs twice for one message | The handler outlived the visibility timeout with `visibility_timeout=0` | Set `visibility_timeout` (it is then extended) |
| A handler's queries return no rows | The pool lacks `guc_provider=guc_set_from_context`, or the policy does not handle the tenant | Wire the provider; check the policy against `app.tenant_id` |
| `connection refused` on 4575 | The stack is down, or another project holds the port | `make docker-up`; set `FLOCI_PORT` and `AWS_ENDPOINT_URL` together |

---

## Testing

```bash
make test-unit   # unit tests — no Docker required
make test-int    # integration tests — floci + Postgres (testcontainers)
make test-e2e    # the full pipeline: Postgres → runner → SNS → SQS → consumer
make cover       # full suite + HTML coverage report (threshold: ≥ 95%)
```

Skip Docker-dependent tests:

```bash
uv run pytest -m "not integration and not e2e"
```

Integration and e2e tests start their own containers (testcontainers; Docker must be running) and never use the `docker-compose.yaml` stack. Every test has a 300 s timeout (`pytest-timeout`), and faulthandler dumps every stack at 240 s.

**Status (0.3.0 branch):** 1700 tests pass (plus 1 expected failure pinning the year-0000 divergence from Go), with **99.97% branch coverage** of `src/eventcommon`. That covers 58 unit files (the `test_cov_*.py` suites close every coverage gap with behaviour tests, Go-confirmed where they assert Go parity), 6 integration files and the e2e pipeline. The cross-language interop harness passes 53/53 checks against platform-events v2.1.0.

### Canonical tests (do not break)

- **`tests/e2e/test_full_pipeline.py`**: no dual write, at-least-once delivery and idempotent redelivery together, against real Postgres and real SNS/SQS semantics.
- **`tests/unit/test_interop.py`, `test_go_wire_corpus.py`, `test_wire_go_parity.py`**: the wire format and HMAC against golden vectors and messages generated by platform-events v2.1.0, including the Go consumer's decisions on malformed and unusual bodies.
- **`tests/unit/test_schema_parity.py`, `test_monitoring.py`, `test_log_parity.py`** (`make go-parity`): the vendored migrations, `monitoring/` assets and every log message identical to platform-events.
- **`tests/unit/test_metrics_standard.py`, `test_metrics_inventory.py`** (`make metrics-lint`): every metric conforms to the registry and to Go's, and `docs/observability/metrics-registry.md` is current.
- **`tests/integration/test_outbox_schema_postgres.py`**: concurrent and Go-compatible migrations on real Postgres.
- **`tests/integration/test_outbox_db_stall.py`**: a database that stalls mid-poll fails each poll within 5 s, turns `healthy()` False, and recovers.
- **`tests/unit/test_sqs_consumer_lifecycle.py`, `test_outbox_runner_execution.py`, `test_cov_sqs_consumer.py`, `test_cov_outbox_runner.py`**: drain, cancellation, panic recovery and settlement rules.
- **`tests/unit/test_migrations_lock_timeout.py`**: `lock_timeout` never weakens to `'0ms'` and is validated before any SQL.
- **`tests/unit/test_null_body_parity.py`, `test_cov_envelope.py`**: Go's decoder behaviour on edge-case bodies (including `null`), with the one remaining divergence (year 0000) pinned as a strict expected failure.

### Coverage

`pyproject.toml` sets `fail_under = 95` (branch coverage of `src/eventcommon`). `make test-ci` runs every suite once with coverage, and `.github/scripts/coverage-gate.sh` checks the threshold separately, so a failing test and a coverage drop are told apart. `make cover-check` runs both, and `make cover-func` lists missing lines per file from the last run. On Python 3.14, coverage's default `sys.monitoring` core does not record the entry / exit arcs of a plain `with` block, so two arcs in `outbox/runner.py` show as partial locally; `COVERAGE_CORE=ctrace` (and CI's 3.12 gate) measure them as covered. No reachable code is excluded: the two `# pragma: no cover` lines are the CLI's `__main__` guard and the uninstalled-package version fallback.

### Testing in consuming services

`eventcommon.mock` ships ready-made, thread-safe test doubles, so consuming services never need an AWS emulator or SNS 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)` | Validates (`id`/`type`/`source`) and records the envelope; raises any error set via `set_error` |
| `await publish_batch(envs)` | Records the valid envelopes; raises one `BatchError` for invalid ones (`InvalidEnvelope`) and any IDs in `set_batch_error` |
| `published` | List of all recorded envelopes (thread-safe) |
| `set_error(err)` | Makes all subsequent `publish` calls raise `err` |
| `set_batch_error(err)` | Makes `publish_batch` fail the `BatchError`'s IDs as given |
| `reset()` | Clears recorded envelopes and any configured errors |

#### 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 as the SQS consumer does: raises `MalformedEnvelopeError` (without calling the handler) for a missing `id` / `type` / `source` or a NUL in `tenant_id` / `trace_id`; round-trips it through JSON; decodes a codec payload with `MockConsumer(codec=...)`; and binds the tenant GUC, trace ID, `source_message_from_context()` (queue `MOCK_QUEUE_URL` unless `queue_url=` is given) and DLQ attribution. Returns without a handler |
| `is_running` | Whether `start` has been called without a matching `stop` |

`MockDLQPublisher(dlq_url="mock://dlq")` records `send_to_dlq` calls in `.sent` (validated like SQS, counted as dead-lettered), with `set_error` / `reset`.

The mocks satisfy the same `Publisher` / `Consumer` protocols as the real implementations, so they drop in wherever the real ones are used. For metrics in tests, pass a fresh `CollectorRegistry` to `init_metrics` and call `metrics.reset()` between tests.

---

## Environment variables

Read only when the service calls a `load_*` loader, `metrics_identity_from_env`, or runs `eventcommon-config`. Nothing is read at import time, and the library reads no `OTEL_*` or log-level variable itself (the deprecated `load_otel` aside). `.env-example` documents the local defaults (`make setup` copies it to `.env`).

Durations use Go's syntax (`5s`, `250ms`, `1m30s`), so a shared ConfigMap means the same in both languages. Anything Go rejects is rejected too: a bare number (`300`; write `300s`), surrounding whitespace, `nan` / `inf`, exponents, `_` separators, non-ASCII digits, and integers outside int64. An invalid integer or duration never raises: the default is applied and a warning (Go's exact text) is added to `.warnings`. A missing required variable is reported by `.validate()` (`ConfigValidationError`).

### AWS, SNS and SQS (`load_sns`, `load_sqs`)

| Variable | Default | Notes |
|---|---|---|
| `AWS_REGION` | `AWS_DEFAULT_REGION`, else `us-east-1` | Read by the loaders, and by the components themselves when `region=None` (before boto3's own chain) |
| `AWS_ENDPOINT_URL` | — | `http://localhost:4575` for the local floci (`FLOCI_PORT` moves it) |
| `SNS_TOPIC_ARN` | — | Required for the publisher |
| `SQS_QUEUE_URL` | — | Required for the consumer |
| `SQS_MAX_MESSAGES` | `10` | 1–10; SQS hard limit |
| `SQS_WAIT_SECONDS` | `20` | Long-poll duration |
| `SQS_VISIBILITY_TIMEOUT` | `30s` | Duration |
| `SQS_CONCURRENCY` | `1` | Parallel handler tasks (FIFO: message groups) |
| `SQS_MAX_RECEIVE_COUNT` | `0` (unset) | Applied only when > 0; pairs with `dead_letter_handler` / `dlq_publisher` |
| `SQS_HANDLER_TIMEOUT` | `0` (off) | Bounds each handler call |
| `SQS_DRAIN_TIMEOUT` | `0` (consumer default 30s) | `stop()` wait before cancelling handlers; keep below `terminationGracePeriodSeconds` |
| `SQS_RETRY_BACKOFF` | `0` (off) | Failed delivery n is hidden `base·2^(n-1)` |
| `SQS_MAX_RETRY_BACKOFF` | `0` (15m) | Cap for `SQS_RETRY_BACKOFF`; at most 12h |
| `SQS_QUEUE_DEPTH_INTERVAL` | `0` (off) | Samples `platform_queue_depth` / `platform_dlq_depth`; minimum 10s |

### Outbox and database (`load_outbox`)

| Variable | Default | Notes |
|---|---|---|
| `OUTBOX_POLL_INTERVAL` | `5s` | |
| `OUTBOX_BATCH_SIZE` | `50` | Records per poll cycle |
| `OUTBOX_MAX_ATTEMPTS` | `5` | Before moving to dead-letter |
| `OUTBOX_CLAIM_LEASE_DURATION` | `0` (runner default 10m) | How long a claimed record is hidden from other runners |
| `OUTBOX_STARTUP_JITTER` | `0` | Use `5s`–`10s` with multiple replicas |
| `OUTBOX_PUBLISH_CONCURRENCY` | `1` | Parallel publishes per poll cycle |
| `OUTBOX_PUBLISH_TIMEOUT` | `10s` | Per-record publish timeout |
| `OUTBOX_DRAIN_TIMEOUT` | `30s` | `runner.stop()` wait bound |
| `OUTBOX_RETRY_BACKOFF` | `1s` | A record's n-th failed publish waits `base·2^(n-1)` |
| `OUTBOX_MAX_RETRY_BACKOFF` | `5m` | Cap for `OUTBOX_RETRY_BACKOFF` |
| `OUTBOX_GAUGE_INTERVAL` | `15s` | Backlog gauge refresh |
| `DATABASE_URL` / `PG_*` | — | Postgres pool config via `pgcommon.config_from_env` (`OutboxEnv.db`); its warnings are merged into `OutboxEnv.warnings` |
| `MIGRATION_DATABASE_URL` | `DATABASE_URL` | DDL-privileged DSN for `apply_schema` (`OutboxEnv.migration_database_url`) |

`str(OutboxEnv)` / `repr(OutboxEnv)` mask credentials (URL userinfo, and `password` / `passwd` / `sslpassword` in URL and libpq forms); never log `.db` / `.database_url` directly.

### Metrics identity

| Variable | Default | Notes |
|---|---|---|
| `APP_ENV` (then `ENVIRONMENT`) | `dev` | The `environment` label when `MetricsIdentity.environment` is empty (trimmed, lower-cased) |
| `APP_NAME` | — (`eventcommon-config`: `eventcommon`) | The `service` label in `metrics.metrics_identity_from_env` (trimmed, lower-cased) |
| `BUILD_VERSION` | the installed eventcommon version | Shown by `eventcommon-config`; not a metric label (`platform_library_info` reports the installed library) |
| `OTEL_SERVICE_NAME`, `OTEL_EXPORTER_OTLP_ENDPOINT`, `OTEL_EXPORTER_OTLP_INSECURE` | — | Read only by the deprecated `load_otel`; tracing is configured by the service |

### Example worker only

Read only by `examples/outbox_worker` (the image CI builds), a template for a service's own entrypoint, not library settings:

| Variable | Default | Notes |
|---|---|---|
| `METRICS_PORT` | `9464` | `/metrics` (the bundled Prometheus scrapes `worker:9464`) |
| `HEARTBEAT_PATH` | `/tmp/heartbeat` | Touched while `Runner.healthy(...)` holds; the image's `HEALTHCHECK` reads it |
| `LOG_LEVEL` / `LOG_FORMAT` | `info` / `json` (`console` on a terminal) | structlog output |
| `STARTUP_TIMEOUT` | `60s` | How long the database is retried at start (1→10 s backoff) before exiting 1 with one line |
| `SHUTDOWN_TIMEOUT` | `25s` | Bound on the whole SIGTERM shutdown; keep it below `terminationGracePeriodSeconds` |

---

## Security

| Topic | Guidance |
|---|---|
| **Who may publish** | SNS topic policies and IAM decide; the library does not authenticate producers. Within one account and IAM boundary that is the control. Across trust boundaries (webhooks, HTTP ingress), sign with `sign_envelope` and verify in the receiver ([HMAC helpers](#hmac-helpers)) |
| **Tenant isolation in handlers** | The consumer binds each message's `tenant_id` into pgcommon's GUC context and resets it afterwards; each message runs in its own task with its own context copy. RLS policies (yours) do the enforcement. A NUL in `tenant_id` / `trace_id` makes the message malformed, so it never reaches a GUC |
| **`"system"` events** | A real GUC value, not an RLS bypass. Handle them with a pool or policy built for cross-tenant work, never by widening the app role |
| **Message bodies and PII** | Malformed bodies are logged as size + SHA-256 only (`malformed_body_logging=True` opts into a 512-byte excerpt). Payloads are never metric labels or span attributes. Do not log `env.data` in handlers |
| **Credentials** | AWS credentials come from boto3's chain (IRSA in production); nothing is logged. `OutboxEnv`, `eventcommon-config` and `make show-config` mask database passwords; pgcommon ≥ 0.3.1 masks the DSN in reprs and error chains |
| **HMAC keys** | ≥ 32 bytes (`KeyTooShortError`), constant-time verification, keys from a secrets manager, rotation by accepting two keys for a window |
| **Poison and oversized input** | Malformed bodies dead-lettered on first receipt; `max_receive_count` and `max_attempts` bound retries; the claim does not load payloads over 1 MiB; `event_type` labels are capped |
| **Example-worker image** | `python:3.14-slim`, digest-pinned, numeric non-root `USER 10001:10001`, `pip` stripped, dependencies from `uv.lock`. Trivy-gated (CRITICAL / HIGH / UNKNOWN), with a CycloneDX SBOM, SLSA provenance and a Cosign keyless signature |
| **Supply chain** | PyPI uploads only through Trusted Publishing (OIDC, no stored token) from `release.yml`; the release's `SHA256SUMS` is Cosign-signed ([Deployment](#deployment)). The build backend is hash-locked (`build-constraints.txt`), `pip-audit` runs in CI, and every third-party action is pinned by commit SHA |

[SECURITY.md](.github/SECURITY.md) has the trust model, the scope of vulnerability reports and how to report one privately. [ARCHITECTURE.md § Threat model](ARCHITECTURE.md#threat-model) has the STRIDE analysis.

---

## Observability — Prometheus and OTel

Both are **optional**. The publisher, consumer and outbox runner work without `metrics.init_metrics` or a registered OTel provider.

### Prometheus metrics

eventcommon implements the **Enterprise Platform Observability Standard** exactly as `platform-events` v2.1.0 does. It emits **only** Tier 1 `platform_*` metrics, carrying the required `domain` / `service` / `environment` labels from a mandatory `MetricsIdentity`. Full details (tiers, semantics, CI enforcement and every difference from Go) are in [docs/observability/README.md](docs/observability/README.md); the generated inventory is [docs/observability/metrics-registry.md](docs/observability/metrics-registry.md).

```python
from eventcommon import metrics

# Once at startup. environment empty → APP_ENV, then ENVIRONMENT, else "dev".
identity = metrics.MetricsIdentity(domain="iam", service="event-consumer")
# Or: metrics.metrics_identity_from_env("iam")  # service from APP_NAME
for warning in metrics.init_metrics(identity):  # registry=None → prometheus_client.REGISTRY
    logger.warning(str(warning))  # a platform_* metric the registry refused — disabled, not fatal
# Raises InvalidMetricsIdentityError (a ValueError) on an invalid identity, changing nothing.
# Options: event_type_limit=200, event_types=[...] (pre-register the event types you handle).

# For isolated test registries:
metrics.init_metrics(identity, CollectorRegistry())
```

- **Required labels are bound centrally.** `domain`, `service` and `environment` come from one identity, so no recording site can omit or misspell them. Pass the same values (and registry) pgcommon's `init_with_identity` gets.
- **Fail-soft registration.** A `platform_*` metric whose name is already registered with another shape is disabled and returned as a `RegistrationWarning`; everything else keeps recording.
- **Go's exposition.** No `*_created` series, and a family appears once it has a series. The interop job compares the exposition with Go's.
- **Removed metrics.** The pre-standard metrics, the legacy init path (`metrics.init` / `metrics.init_with_registry`), the `legacy=` toggle and `MetricsIdentity.version` were removed without a compatibility period, because nothing had been deployed. See [docs/observability/README.md § Removed pre-standard metrics](docs/observability/README.md#removed-pre-standard-metrics).

**Tier 1 metrics.** Every one also carries `domain`, `service` and `environment`. Canonical ones may back alerts and SLOs; Proposed ones are shadow-emitted until ratified.

| Metric | Type | Labels | Status | Description |
|---|---|---|---|---|
| `platform_messages_received_total` | Counter | `queue` | canonical | SQS deliveries, redeliveries included |
| `platform_messages_processed_total` | Counter | `queue`, `event_type` | canonical | Handler succeeded, message deleted |
| `platform_messages_failed_total` | Counter | `queue`, `event_type`, `reason` | canonical | `malformed` / `decode_error` / `handler_error` / `handler_panic` / `dead_letter_error` |
| `platform_retry_total` | Counter | `operation`, `event_type` | canonical | Failed attempt left for retry (`consume`, `outbox_publish`) |
| `platform_dlq_messages_total` | Counter | `operation`, `event_type`, `reason` | canonical | Dead-lettered, counted once (`consume`: `max_receive_count` / `malformed` / `decode_error` / `explicit`; `outbox_publish`: `max_attempts`) |
| `platform_dependency_request_seconds` | Histogram | `dependency`, `operation`, `outcome` | proposed | SNS / SQS / codec call latency and outcome |
| `platform_event_propagation_seconds` | Histogram | `queue`, `event_type` | proposed | Envelope `time` → first receipt |
| `platform_messages_in_flight` | Gauge | `queue` | proposed | Messages being processed, per replica |
| `platform_messages_published_total` | Counter | `topic`, `event_type`, `outcome` | proposed | Publish attempts per event |
| `platform_message_processing_duration_seconds` | Histogram | `queue`, `event_type` | proposed | Handler / dead-letter handler wall time |
| `platform_outbox_pending_events` | Gauge | — | proposed | Due unpublished outbox rows (capped at 100000) |
| `platform_outbox_leased_events` | Gauge | — | proposed | Leased / backing-off outbox rows (capped at 100000) |
| `platform_outbox_oldest_pending_age` | Gauge | — | proposed | Age (s) of the oldest unpublished outbox row — the delivery-stall signal |
| `platform_outbox_publish_attempts_total` | Counter | `event_type`, `outcome` | proposed | Outbox runner publish attempts |
| `platform_outbox_errors_total` | Counter | `operation` | proposed | `poll` / `unmarshal` / `mark_published` / gauge-query failures |
| `platform_outbox_dead_letter_operations_total` | Counter | `operation` | proposed | Dead-letter records reprocessed / discarded |
| `platform_telemetry_label_overflow_total` | Counter | `label` | proposed | `event_type` values replaced (`__oversized__`, `__other__`) |
| `platform_library_info` | Gauge | `library`, `library_version` | proposed | Always 1; `library="platform-eventcommon"` |
| `platform_duplicate_messages_total` | Counter | `queue`, `event_type` | proposed | Inbox duplicates skipped (`inbox.handler` / `Store.process`) |
| `platform_queue_depth` | Gauge | `queue` | proposed | Visible messages, sampled with `queue_depth_interval` (opt-in) |
| `platform_dlq_depth` | Gauge | `queue` | proposed | Visible messages in the RedrivePolicy DLQ, sampled likewise |
| `platform_message_timeouts_total` | Counter | `queue`, `event_type`, `operation` | proposed | `handler_timeout` expiries by stage (`decode`, `dead_letter_handler`, `handler`) |
| `platform_outbox_ordering_blocked_events` | Gauge | — | proposed | Ordered records waiting behind their key's head |

`queue` / `topic` are the queue / topic **name**, never the URL / ARN. `event_type` is bounded in-process: at most 200 distinct values (`__other__` beyond), 128 bytes (`__oversized__` beyond), and `unknown` for an empty or unparseable type.

### OpenTelemetry

OTel is **always initialised by the consuming service** (e.g. `fastapicommon.tracing.init_from_env()`). `eventcommon` calls `trace.get_tracer("platform-events")`, the same instrumentation scope as Go, and produces no-op spans when no provider is registered.

| Span | Kind | Attributes |
|---|---|---|
| `sns.publish` | PRODUCER | `messaging.system=aws_sns`, `messaging.destination` (topic ARN), `messaging.message_id`, `events.event_type`, `events.event_id` |
| `sns.publish.batch` | PRODUCER | `messaging.system=aws_sns`, `messaging.destination`, `messaging.batch.message_count` |
| `sqs.receive` | CONSUMER | `messaging.system=aws_sqs`, `messaging.destination` (queue URL), `messaging.message_id`, `messaging.operation=process`, `events.event_type`, `events.event_id` |
| `sqs.dlq_forward` | PRODUCER | `messaging.system=aws_sqs`, `messaging.operation=publish`, `messaging.destination` (DLQ URL), `messaging.message_id`, `events.event_type`, `events.dlq.source_queue` |

The publisher injects `traceparent` / `tracestate` / `baggage` as message attributes. `sqs.receive` is rooted in an empty context and **linked** to the producer's span through them, and the producer's baggage is carried into the handler.

### Logging correlation

**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()`) and `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 = decode_payload(env.data, UserCreatedPayload)
    await repo.create_user(payload)

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

The library's own log lines use Go's messages verbatim (`tests/unit/test_log_parity.py`), so one log query matches both languages. They go through the logger you pass (`logger=` on each config): structlog, or a stdlib `logging.Logger` with the fields appended as ` key=value`.

#### 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 |

### Alerts, rules and dashboard

`monitoring/` vendors platform-events' reference assets, byte-identical (`make go-parity`):
- `prometheus/platform-events.rules.yml`: recording rules and alerts, with promtool unit tests in `platform-events.rules.test.yml` (`make rules-check`);
- `grafana/platform-events.json`: the dashboard;
- `kubernetes/keda-scaledobject.example.yaml`: queue-depth autoscaling.

Because the metric families are identical, the same assets serve Go and Python services, and a single query aggregates both. Group every query `by (domain, service, environment)`, never by `service` alone. The runbook is [docs/observability/runbook.md](docs/observability/runbook.md).

---

## Deployment

eventcommon itself is released to PyPI, and services adopt it by bumping their pin; it is never deployed on its own. The example-worker image is a reference, built and scanned by every release.

### Installing a release

Pin an exact version in production ([VERSIONING.md § Consume a release](VERSIONING.md#consume-a-release) has the pin styles):

```bash
pip install "eventcommon==0.3.0"
uv add "eventcommon==0.3.0"
```

The files on PyPI are byte-identical to the GitHub Release's wheel and sdist: the release builds them once and uploads the same files to both.

### Container image — the example worker

| Executable | In the image | Purpose |
|---|---|---|
| `python -m examples.outbox_worker.main` | `ENTRYPOINT`, `EXPOSE 9464`, a heartbeat `HEALTHCHECK` | Applies the outbox schema, runs a `Runner` and an `SQSConsumer`, serves `/metrics`, shuts down within `SHUTDOWN_TIMEOUT` |

The `Dockerfile` has a build and a runtime stage:
- **Build stage.** `ghcr.io/astral-sh/uv` and `python:3.14-slim`, both digest-pinned. It installs the runtime dependencies from `uv.lock` and the wheel built with the hash-locked backend (`build-constraints.txt`).
- **Runtime stage.** `python:3.14-slim`, digest-pinned, with pinned Debian fixes. It removes `pip` and runs as the numeric `USER 10001:10001`, so `runAsNonRoot` is enforceable.
- **Size.** The smoke gate fails above 250 MB.
- **Tags.** They have no `v`: `X.Y.Z` always, plus `X.Y` and `X` only when the release is the newest stable version of that line, and **never `latest`**, because it is a demonstration, not a deployable service. A pre-release gets only its own tag.

### Verifying a release

Needs [cosign](https://docs.sigstore.dev/cosign/system_config/installation/) ≥ 2.4. A tag-triggered release is signed by `release.yml@refs/tags/<tag>`, and a re-run dispatched from `main` by `release.yml@refs/heads/main`; the regular expression below accepts both.

```bash
TAG=vX.Y.Z   # the release to verify
REPO=BCBP-SOLUTIONS-FZC-LLC/platform-eventcommon
IDENTITY="^https://github\.com/${REPO}/\.github/workflows/release\.yml@refs/(tags/${TAG}|heads/main)$"

# Wheel + sdist: verify the signed checksums, then the files against them
gh release download "$TAG" -R "$REPO" -p SHA256SUMS -p SHA256SUMS.sigstore.json -p '*.whl' -p '*.tar.gz'
cosign verify-blob --bundle SHA256SUMS.sigstore.json \
  --certificate-identity-regexp "$IDENTITY" \
  --certificate-oidc-issuer https://token.actions.githubusercontent.com \
  SHA256SUMS
sha256sum --check SHA256SUMS   # macOS: shasum -a 256 -c SHA256SUMS

# Example-worker image (tag without the v)
cosign verify --certificate-identity-regexp "$IDENTITY" \
  --certificate-oidc-issuer https://token.actions.githubusercontent.com \
  "ghcr.io/bcbp-solutions-fzc-llc/platform-eventcommon:${TAG#v}"
```

Each GitHub Release carries the wheel, the sdist, `SHA256SUMS`, `SHA256SUMS.sigstore.json`, the image's SLSA provenance and its CycloneDX SBOM; PyPI also shows the Trusted Publishing attestations. Releases up to `v0.2.0` predate signing and `SHA256SUMS`. [VERSIONING.md § Verifying a release](VERSIONING.md#verifying-a-release) is the canonical copy of these commands.

### Migration safety

`outbox.apply_schema(pool)` and `inbox.apply_schema(pool)` apply platform-events' own migration files, tracked as golang-migrate tracks them, so Go and Python services can share and migrate one database:
- **Locking.** The run takes every advisory-lock key Go's runner may use (`/db`, `db`, `/`) as transaction-level locks, so concurrent Go and Python runners serialise. It runs in one transaction with `lock_timeout` (30 s by default) set first, so a migration blocked behind a long transaction on the outbox tables fails with `asyncpg.LockNotAvailableError` instead of queueing every query behind it. Because the locks are transaction-level, it also works through PgBouncer.
- **Refused states.** A dirty version (a Go run that failed half-way) raises `MigrationDirtyError`. A version this release does not ship raises `SchemaVersionNotInSourceError`; migrate down with the newer release before rolling back.

Run it once per deploy, at startup or as a pre-deploy job, with a DDL-privileged role (`MIGRATION_DATABASE_URL`). Concurrent runs are safe, and the loser finds nothing to do. Go and Python services sharing a database must run releases that ship the same migration set. See [ARCHITECTURE.md § Schema migration flow](ARCHITECTURE.md#schema-migration-flow).

### Versioning and releases

**Release status:** the newest release is **0.3.0** (2026-10-16). It brings platform-events v2.1.0 parity and production hardening, and is breaking against 0.2.0; read [Upgrading from 0.2.0](#upgrading-from-020) first.

Earlier releases were **0.2.0** (2026-08-07), and **0.1.1** and **0.1.0** (2026-07-27).

**Go parity:** this port tracks `platform-events` **v2.1.0**: the same wire format, HMAC, schema, metrics, configuration, log messages and behaviour, except for the documented [divergences](ARCHITECTURE.md#divergences-from-go). The Python package keeps its own version line (`0.x`); the CHANGELOG names the Go version each release corresponds to.

See [VERSIONING.md](VERSIONING.md) for the full release process, the public-API surface SemVer applies to, the envelope wire-format rules, and how Trusted Publishing to PyPI is set up. Summary:

| 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 |

While the package is `0.y.z`, a MINOR bump may break the public API (SemVer §4) and a PATCH may not. The release's `api-compat` gate (griffe) enforces exactly that.

```bash
# Tag a commit on main; pushing the tag triggers the release workflow.
make release-guard TAG=vX.Y.Z          # pyproject version == tag, a dated CHANGELOG [X.Y.Z]
git tag -a vX.Y.Z -m "vX.Y.Z"
git push origin vX.Y.Z
```

`release.yml` runs these stages in order:
1. **`verify`** checks the dispatch ref, that the tag is at HEAD and on `origin/main`, that the version equals the tag, that the CHANGELOG section is dated, and which floating image tags may move.
2. **The CI gates run again at the tag:** tests, quality (Go parity required), `api-compat` (blocking), Hadolint, Trivy, smoke and interop.
3. **`dist`** builds the wheel and sdist **once**, reproducibly.
4. **`push`**, only after every gate: publishes the fingerprint-checked image with SBOM, SLSA provenance and a Cosign signature.
5. **`pypi-publish`** uploads the same files via Trusted Publishing.
6. **`publish`** creates the GitHub Release.

Interop gates `push`, so a red cross-language check stops the whole release.

---

## CI

The pipeline is platform-pgcommon-py's (itself the Go repos', translated to Python tooling): the same jobs, gates and scripts (`.github/scripts/`), adapted to this library and its example-worker image. Every third-party action is pinned by commit SHA.

- **`ci.yml`** runs on every push and pull request to `main`. `changes` (`detect-changes.sh`) decides whether the change is documentation only; if so, the build and test jobs are skipped through `if:`, so the required checks still report. Otherwise `validate-test` and `validate-quality` run in parallel with `build-image`. The image then goes to `trivy` and `smoke`, then `interop` runs, and `pr-summary` comments on the PR. On a push to `main`, `push` publishes the image (`main` / `sha-<short>`), but only one whose content fingerprint matches the image Trivy and the smoke tests judged, and Cosign signs it.
- **`validate-test.yml`** (reusable) runs `make test-ci` (unit + integration + e2e on floci + Postgres, with coverage), the coverage gate (**≥ 95%**) and `make arch-lint` on Python 3.12. It also runs the suite on 3.13 / 3.14 and at the lowest direct dependencies.
- **`validate-quality.yml`** (reusable) runs:
  - `uv lock --check` and `uv pip check`;
  - the RLS session-`SET app.*` guard;
  - ruff format / lint and `mypy --strict`;
  - `make metrics-lint`, `make rules-check` (promtool), `make docs-check` and `make ci-scripts-test`;
  - **Go parity** against a platform-events checkout: required on `main` and for releases, a warning on PRs without access;
  - `pip-audit`, digest-pinned `FROM` lines, and `make build`.
- **`release.yml`** runs on `vX.Y.Z` / `vX.Y.Z-pre.N` tags; see [Versioning and releases](#versioning-and-releases).
- **`docs.yml`** checks the diagram sync on documentation-only changes (not a required check).
- **`changelog-check.yml`** fails a PR that changes `src/`, `pyproject.toml`, `uv.lock`, `build-constraints.txt` or the `Dockerfile` without updating `CHANGELOG.md`.

| Job (`ci.yml`) | Check name | What it gates | Required |
|---|---|---|---|
| `changes` | `Detect changes` | Docs-only decision + the shared `SOURCE_DATE_EPOCH` | — |
| `validate-test` → `test` | **`Validate / Test / test`** | Full suite on Python 3.12 + coverage ≥ 95% + import-linter | ✅ |
| `validate-test` → `python` / `lowest-deps` | `Validate / Test / python 3.13`, `… / python 3.14`, `… / lowest direct dependencies` | The suite on the other Pythons and at the dependency floors | — |
| `validate-quality` → `quality` | **`Validate / Quality / quality`** | Lock, format, lint, types, RLS guard, metrics standard, rules, docs, CI-script tests, Go parity, pip-audit, digest pins, packaging | ✅ |
| `api-compat` | `API compatibility` | `griffe` vs the last release tag; **warning only** here, blocking in `release.yml` | — |
| `build-image` | `Build image (cache)` | Hadolint + `.dockerignore`, then a cached build | — |
| `trivy` | `Trivy CVE scan` | CRITICAL / HIGH / UNKNOWN → fail; SARIF to the Security tab; CycloneDX SBOM artifact | — |
| `smoke` | `Smoke tests` | `smoke-tests.sh`: size ≤ 250 MB; a bare run exits 1 naming `SNS_TOPIC_ARN`; the installed version; `eventcommon-config --strict`; `HEALTHCHECK`; no `pip`; non-root | — |
| `interop` | `Cross-language compatibility` | platform-interop-tests (SHA-pinned). Not required in CI; **gates the release's `push`** | — |
| `pr-summary` | `PR summary` | PRs only: one status comment | — |
| `push` | `Push image → GHCR` | Pushes only: fingerprint-checked push, SBOM + provenance, Cosign sign + verify | — |

**Private repository:** checkouts use `CI_REPO_READ_TOKEN || github.token` without persisting credentials. `CI_REPO_READ_TOKEN` also checks out platform-events for the Go parity step, and reusable workflows get explicit `secrets:` maps.

Run the same gates locally before pushing:

```bash
make ci            # lock-check + pip-check + format-check + lint + typecheck + metrics-lint + rules-check
                   # + docs-check + ci-scripts-test + vulncheck + test-ci + coverage gate + arch-lint + build
make go-parity     # against a sibling platform-events checkout (or PLATFORM_EVENTS_DIR=…)
make api-compat    # griffe vs the last release tag
make lint-docker   # hadolint (CI: first step of build-image)
make scan          # Trivy, same severities as CI
make smoke-test    # build the image + .github/scripts/smoke-tests.sh
```

`make docs-check` checks that every `docs/architecture/mermaid/*.mmd` is embedded byte-identically in `ARCHITECTURE.md`; edit a `.mmd` file and its embedded block together.

---

## Docker

### What the bundled `docker-compose.yaml` starts

Every port is bound to `127.0.0.1` and can be moved with the variable in brackets, so the stack can run beside other projects' stacks.

| Container | Image | Host port | Purpose |
|---|---|---|---|
| `floci` | `floci/floci` (digest-pinned) | `4575` (`FLOCI_PORT`) | AWS emulator (SNS, SQS); provisions the demo topic, queue and DLQ on start |
| `floci-ui` | `floci/floci-ui` | `4506` (`FLOCI_UI_PORT`) | Web console for floci's resources and messages |
| `postgres` | `postgres:16-alpine` | `5539` (`POSTGRES_PORT`) | Database `eventcommon` (user and password `eventcommon`) |
| `otel-collector` | `otel/opentelemetry-collector-contrib` | `4317` / `4318` | OTLP receiver → Tempo |
| `tempo` · `loki` · `promtail` | Grafana Tempo 2.7 · Loki · Promtail | `3200` · `3100` · — | Traces · logs · log shipping from the worker |
| `prometheus` · `grafana` | Prometheus · Grafana | `9090` · `3000` | Scrapes `worker:9464` · provisions the vendored `monitoring/grafana` dashboard |
| `worker` (profile `demo`) | built from the `Dockerfile` | — | The example worker; `make run` starts it |

`make docker-up` starts everything but the worker and waits for floci; `make run` adds the worker; `make docker-down` removes the stack **with its volumes**. The integration and e2e suites don't use it: they start their own containers.

### Building the example-worker image

```bash
make docker-build   # the example worker image, tagged :local
make smoke-test     # build + CI's image gates (.github/scripts/smoke-tests.sh)
make lint-docker    # hadolint, as the first step of CI's build-image job
make scan           # Trivy, CI's severities
```

### Running the worker with the image

```bash
docker run --rm -p 9464:9464 \
  -e AWS_REGION=eu-west-1 -e SNS_TOPIC_ARN=arn:aws:sns:eu-west-1:123456789012:orders \
  -e SQS_QUEUE_URL=https://sqs.eu-west-1.amazonaws.com/123456789012/orders-q \
  -e DATABASE_URL='postgres://app:…@db:5432/app?sslmode=verify-full' \
  -e APP_NAME=orders-worker -e APP_ENV=production \
  ghcr.io/bcbp-solutions-fzc-llc/platform-eventcommon:<version>
```

A bare `docker run` exits 1 naming the first missing variable, with no traceback. The worker exits 1 when the runner or the consumer dies, so the orchestrator restarts it.

### Minimum required environment variables

```bash
# Publisher + outbox runner
AWS_REGION=eu-west-1
SNS_TOPIC_ARN=arn:aws:sns:eu-west-1:123456789012:orders
DATABASE_URL=postgres://app_user:…@pgbouncer:6432/app?sslmode=verify-full
MIGRATION_DATABASE_URL=postgres://migrator:…@postgres:5432/app?sslmode=verify-full   # apply_schema
APP_ENV=production              # the environment label — use the platform's agreed spelling

# Consumer
SQS_QUEUE_URL=https://sqs.eu-west-1.amazonaws.com/123456789012/orders-q
SQS_VISIBILITY_TIMEOUT=60s
```

---

## Parity with platform-events

eventcommon is the Python port of [platform-events](https://github.com/BCBP-SOLUTIONS-FZC-LLC/platform-events) v2.1.0 and has its features:
- the envelope, HMAC and the SNS publisher;
- the codec hook (`GlueDecodeCodec`);
- the SQS consumer: FIFO group ordering, handler and drain timeouts, retry backoff, queue-depth sampling and DLQ forwarding;
- the DLQ publisher;
- the transactional outbox: per-key ordering, retry backoff and dead-letter replay;
- the inbox, the metrics registry and the reference monitoring.

Go is the source of truth. The wire format, metric names, labels, buckets, help text, log and warning texts, environment variables and defaults are the same. So are the outbox and inbox schemas, down to the migration files and golang-migrate tracking, so Go and Python services can share one outbox or inbox database.

Wire interop is verified end to end: every envelope and payload shape, single and batch publishing, FIFO group order, the Glue codec, HMAC signatures, trace links, baggage, tenant / trace context and a shared outbox, published by either language over SNS/SQS and consumed by the other. `tests/unit/test_go_wire_corpus.py` replays messages platform-events published, and the Go consumer's decisions on malformed and unusual bodies, so any drift fails this repository's CI.

The behavioural differences left are ones Python cannot express. An envelope `time` in year 0000 is rejected (Python's `datetime` starts at year 1; Go accepts it), and `data` containing invalid UTF-8 is forwarded with U+FFFD in its place (Go forwards the raw bytes). The full list, including the Python-only additions, is [ARCHITECTURE.md § Divergences from Go](ARCHITECTURE.md#divergences-from-go). Where the two languages differ in style, both forms are available:

- **Options.** Every Go functional option has a Python counterpart with the same name, usable positionally (`SQSConsumer(cfg, handler, with_concurrency(4), with_retry_backoff(timedelta(minutes=1), timedelta(minutes=30)))`, `SNSPublisher(cfg, with_codec(codec))`) or as the constructor keyword it stands for (`concurrency=4`, `codec=codec`). Options apply in order, after the keywords. Durations take seconds or a `timedelta` (Go's `time.Duration`).
- **Clients.** `SNSPublisher(cfg, client=…)` and `SQSConsumer(cfg, handler, client=…)` take a caller-owned aioboto3 client (Go: `NewWithClient` / `NewSQSConsumerWithClient`), as `SQSDLQPublisher(client=…)` does. The library never closes it; without one, they open their own.
- **Durations** use Go's `time.ParseDuration` syntax and nothing else: `SQS_HANDLER_TIMEOUT=300` is rejected (a warning, default applied) by both. Write `300s`.

Metric-level differences are listed in [docs/observability/README.md § Parity with platform-events](docs/observability/README.md#parity-with-platform-events).

---

## Dependencies

### 1. Consumers (projects that import this library)

| Consumer | eventcommon pin | Uses |
|---|---|---|
| Python platform services (FastAPI, on `fastapicommon` + `pgcommon`) | their own | Publisher, consumer, outbox, inbox, metrics, mocks |
| `platform-interop-tests` (Python probe) | editable path to this checkout | Cross-language compatibility with platform-events (CI's `interop` job) |

### 2. Upstream libraries

[Runtime dependencies](#runtime-dependencies-what-consumers-inherit) lists the ranges. `pgcommon` is the load-bearing one: the outbox and inbox use its `Pool` and `run_in_tx`, the consumer binds tenants through its `with_guc_set`, and its floor is the latest release, 0.4.0 (0.3.1 brought the bounded queries on a stalled database and the DSN masking). aioboto3 and boto3 provide the clients and the error classes the retry classification reads. CI's `lowest direct dependencies` job runs every suite at each floor.

### 3. Platform repos

| Repo | Relationship |
|---|---|
| `platform-events` (Go) | The source of this port: the parity target (v2.1.0), and the origin of the vendored migrations and `monitoring/` assets |
| `platform-pgcommon-py` | Runtime dependency (above); the model for this repository's pipeline, Makefile and documentation |
| `platform-fastapicommon` | No dependency either way. A service using both passes the same `CollectorRegistry` and identity values, and sets up tracing there |
| `platform-interop-tests` | Cross-language compatibility suite run by CI's `interop` job (pinned to a commit SHA) |

### 4. Infrastructure

| System | Role |
|---|---|
| **AWS SNS / SQS** | Topics, queues, subscriptions (`RawMessageDelivery=true`) and RedrivePolicies, provisioned by infrastructure |
| **PostgreSQL** | The outbox and inbox tables, in each service's own database; RLS policies are the service's |
| **Prometheus / OTel collector** | Scrape and collect through the service's own endpoints; the library opens no listener |
| **PyPI** · **GHCR** | Host the package (Trusted Publishing) · the signed example-worker image |

### 5. Development tooling

The `dev` dependency group (`uv sync` installs it by default) has `pytest`, `pytest-asyncio`, `pytest-cov`, `pytest-timeout`, `testcontainers[postgres]`, `opentelemetry-sdk`, `ruff`, `mypy`, `types-aioboto3`, `pip-audit`, `pyyaml`, `griffe` (`make api-compat`) and `import-linter` (`make arch-lint`). The build backend is pinned and hash-locked in `build-constraints.txt`.

---

## Out of scope

| Concern | Where it lives |
|---|---|
| Topics, queues, subscriptions, filter policies, RedrivePolicies, IAM, KMS | Platform infrastructure |
| Authenticating producers and deciding their tenant | Each service's auth middleware (`fastapicommon`) and IAM |
| RLS policies, database roles, the pool | Each consuming service · `platform-pgcommon-py` |
| Payload schemas and their registry | The producing service · [EVENT_SCHEMA_GOVERNANCE.md](EVENT_SCHEMA_GOVERNANCE.md) · your `Codec` |
| HTTP app factory, request middleware, logging and tracing setup | `platform-fastapicommon` / the service |
| Alert routing, dashboard hosting | The platform's monitoring stack (the rules and dashboard are vendored here) |
| Metric naming ratification | Observability governance |
| Brokers other than SNS / SQS | Not supported without a new adapter ([CONTRIBUTING.md](CONTRIBUTING.md)) |
| Synchronous (non-`asyncio`) services | Not supported |

---

## Contributing

See [CONTRIBUTING.md](CONTRIBUTING.md) for the development setup, project layout, adding event types and adapters, changing the outbox or inbox schema, adding a metric, testing requirements, the PR checklist and the release process. Before pushing:
- run `make ci`;
- update `CHANGELOG.md` for any change to `src/`, `pyproject.toml`, `uv.lock`, `build-constraints.txt` or the `Dockerfile` (`changelog-check.yml` enforces it);
- keep this README's API, configuration and error tables in step with the code;
- edit a `docs/architecture/mermaid/*.mmd` file together with its embedded copy in `ARCHITECTURE.md` (`make docs-check`);
- change wire-format, schema, metric or log behaviour in platform-events first, then here (`make go-parity`).

---

## Glossary

| Term | Meaning in this library |
|------|------------------------|
| **Envelope** | The typed event wrapper (`Envelope[T]`) and its JSON wire format, identical to Go's. `id`, `type`, `source`, `time` are always present |
| **Event / message / outbox record** | The domain occurrence and its envelope · the SNS/SQS unit the broker delivers · the Postgres row in `outbox_events` / `outbox_dead_letters` |
| **Outbox** | The `outbox_events` table and the `Runner` that publishes from it; `enqueue` writes to it inside the business transaction |
| **Claim lease** | The time a runner hides a claimed record from other runners (`claim_lease_duration`, 10m), by moving `scheduled_at` forward |
| **Inbox** | The `processed_events` table that records which envelope IDs a consumer has handled (`inbox.Store`) |
| **DLQ** | The SQS dead-letter queue named by the source queue's RedrivePolicy. Not the same as outbox dead letters (Postgres) |
| **Malformed** | A body that is not an envelope by Go's rules; dead-lettered or deleted on first receipt, never retried |
| **Poison message** | A valid envelope that can never be processed successfully; return (and log) or dead-letter it, never raise |
| **Transient / permanent failure** | A failure that may succeed on retry (throttling, 5xx, timeouts; no outbox attempt counted) · one that will not (counts an attempt, backs off, dead-letters) |
| **GUC** | A PostgreSQL runtime parameter. The consumer binds `app.tenant_id` through pgcommon for RLS policies to read |
| **`SYSTEM_TENANT_ID`** | `"system"`: the tenant of genuinely cross-tenant events. A real GUC value, not an RLS bypass |
| **Panic** | Go's term; here a `BaseException` that is not an `Exception` (other than `CancelledError`, `KeyboardInterrupt`, `SystemExit`), recovered as Go recovers a panic |

---

## License / ownership

BCBP Solutions FZC LLC — internal platform shared library. Licensed Apache-2.0 ([LICENSE](LICENSE)).

---

## See also

| Document | Description |
|----------|-------------|
| [ARCHITECTURE.md](./ARCHITECTURE.md) | Layer model, flows, failure model, invariants, threat model, divergences from Go, design decisions |
| [docs/lld/platform-lld-eventcommon-py.md](./docs/lld/platform-lld-eventcommon-py.md) | Low-level design: API contract, algorithms, failure handling, invariant IDs (`EVPY-*`, mapped to the Go LLD's), sign-off register |
| [docs/architecture/README.md](./docs/architecture/README.md) | The standalone Mermaid sources of every ARCHITECTURE.md diagram |
| [docs/guides/](./docs/guides/) | Task guides: quick start, publishing, consuming, outbox, envelope, codec, HMAC, observability, operations, testing in services |
| [docs/observability/README.md](./docs/observability/README.md) · [metrics-registry.md](./docs/observability/metrics-registry.md) · [runbook.md](./docs/observability/runbook.md) | The observability standard as implemented here · the generated metric inventory · the alert runbook |
| [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, verifying a release |
| [CHANGELOG.md](./CHANGELOG.md) | Per-version changes, with upgrade notes |
| [CONTRIBUTING.md](./CONTRIBUTING.md) | Development setup, adding event types and adapters, PR checklist |
| [SECURITY.md](.github/SECURITY.md) | Vulnerability reporting, trust model, supported versions (the root `SECURITY.md` points here) |
| [.claude/CLAUDE.md](./.claude/CLAUDE.md) | Agent reference set for Claude Code (with `architecture.md`, `api-reference.md`, `database-schema.md`, `request-flows.md`, `operations.md`, `settings.json`): commands, critical rules and condensed contracts |

---

`eventcommon` must be the only entry point for event publishing and consumption, to preserve delivery guarantees, tenant isolation and uniform observability. For any event that drives downstream state, use `outbox.enqueue` inside a `pgcommon.run_in_tx` block, with no exceptions.
