Events & webhooks

One envelope, two transports

Every event the platform produces — whether it stays inside the platform on Kafka or is delivered to your own HTTPS endpoint as a webhook — is serialized as a single canonical shape: a CloudEvents v1.0 (CNCF) envelope. There is exactly one model for it, EventEnvelope, and it uses the official cloudevents SDK (pinned cloudevents==1.12.1) so the platform never hand-rolls the wire format.

That one envelope travels over two transports:

The webhook contract — a per-subscriber HMAC secret and signature-verified delivery — is the one Stripe made standard, so integrators plug in with tooling they already have.

Producers never publish directly to either transport. They write a row to a transactional outbox in the same database transaction as the change that caused the event; a poller drains the outbox, builds the CloudEvents envelope once, and fans it out. This is what keeps an event and the state change that produced it atomic — no event is emitted for a transaction that rolled back, and no committed change silently fails to emit.

The CloudEvents envelope

EventEnvelope is a typed model whose fields map onto CloudEvents core attributes; platform-specific fields (org, instance, correlation) ride along as CloudEvents extension attributes (lowercase, ≤20 chars, per the spec). Distributed tracing is carried via the W3C traceparent extension.

The canonical mapping (typed model → CloudEvents attribute):

On Kafka the message value is the whole envelope as one CloudEvents structured-mode JSON document, with the content-type header application/cloudevents+json; charset=utf-8. There are no parallel value-shadowing X-* headers — the envelope is self-contained.

Because the model lives in egav-outbox (the transport layer) but is re-exported from the app SDK, service code imports it from one place:

from egav_app_sdk.events import EventEnvelope

# EventEnvelope core-attribute mapping (typed model -> CloudEvents v1.0):
#   event_id     -> id
#   event_type   -> type          (manifest emits[].id, e.g. 'support.ticket.created')
#   occurred_at  -> time          (RFC3339)
#   aggregate_id -> subject       (resource id; also the Kafka partition key)
#   payload      -> data
#   dataschema   -> dataschema    (manifest schema_ref)
# datacontenttype = 'application/json'; specversion = '1.0' (set by the SDK)
#
# Kafka value  = envelope.to_json()   (CloudEvents structured JSON)
# Kafka header content-type = 'application/cloudevents+json; charset=utf-8'
# Kafka key    = subject (aggregate id)

Declare your event contract in the manifest

Events are part of your app's contract, declared in app.manifest.yaml — not configured ad hoc. Two blocks matter:

• spec.kafka — the topics your app owns (owned_topic_patterns), its consumer_group_pattern, and its dead_letter_topic.
• spec.events.emits[] — each event you produce, keyed by id, topic, schema_ref, version, audience, webhook_subscribable, and emit_policy.

Validation (run by the SDK locally and re-run by the Control Plane on publish) enforces the cross-field rules the schema alone can't: every emitted topic must match one of your owned_topic_patterns, ids are unique per app, and no two apps may claim overlapping owned_topic_patterns.

Two fields shape delivery behavior:

• audience (required, no fallback) is the routing scope: tenant-private, org-broadcast, or platform-broadcast.
• emit_policy defaults to always. Set it to subscribed to make the event listener-gated — the outbox row is only written when a listener or webhook is actually subscribed, so integration events stay dormant until someone wants them.
• webhook_subscribable: true exposes the event to customer webhook subscriptions.

kafka:
  owned_topic_patterns:
  - support.ticket.*
  - support.comment.*
  consumer_group_pattern: support-{instance_guid}
  dead_letter_topic: support.dlq
events:
  emits:
  - id: support.ticket.created
    topic: support.ticket.created
    transport: outbox                    # REQUIRED: outbox | platform_events
    schema_ref: manifest-schemas/events/support-ticket-created.v1.json
    version: 1
    audience: tenant-private
    webhook_subscribable: true
    emit_policy: subscribed

More on the emit contract

Every emit also names its transport (required, no default): outbox for an event written through the transactional outbox and published to Kafka on topic, or platform_events for one sent through Control Plane's event-delivery table (no Kafka topic is derived). Beyond id / topic / transport / schema_ref / version / audience / emit_policy, each emits[] entry can carry fields that shape how the platform provisions, orders, prioritizes, and ages the event. All are optional with sensible defaults.

emits:
- id: egav.route.reload
  topic: egav.route.reload
  transport: outbox                    # REQUIRED: outbox | platform_events
  schema_ref: manifest-schemas/events/route-reload.v1.json
  version: 1
  since: 2.1.0
  retention_days: 1
  partition_key: entity_type_code
  visibility: tenant
  audience: tenant-private
  channel: default
  priority: high
  webhook_subscribable: false

visibility and audience are easy to confuse: visibility decides who may subscribe, audience decides how far a produced event fans out. Set both.

Request/response events

Most events are fire-and-forget. An event can instead ask its subscribers for an answer. Set expects_result: true and the Control Plane captures each subscriber's response body into platform_event_results, so the producer can read back per-subscriber results — a synchronous request-response over the bus. The synchronous SDK helper events.request() relies on this flag.

emits:
- id: billing.invoice.finalize_requested
  topic: billing.invoice.finalize_requested
  transport: outbox                    # REQUIRED: outbox | platform_events
  schema_ref: manifest-schemas/events/invoice-finalize.v1.json
  version: 1
  audience: tenant-private
  channel: billing
  priority: high
  expects_result: true
  result_schema_ref: manifest-schemas/events/invoice-finalize-result.v1.json

Payload fields whose shape varies at runtime

When one payload field's shape isn't fixed at author time — for example dynamic record values, whose columns depend on the record type and schema version — declare a dynamic_payload_schema. The static schema_ref still constrains the envelope; this block tells consumers how to fetch the schema for the dynamic field at runtime from one of your declared actions (an entry in spec.apis.actions[], see the manifest reference).

emits:
- id: egav.entity.created
  topic: egav.entity.created
  transport: outbox                    # REQUIRED: outbox | platform_events
  schema_ref: manifest-schemas/events/entity-created.v1.json
  version: 1
  audience: tenant-private
  dynamic_payload_schema:
    payload_field: values
    resolver_action_id: egav.get_entity_values_schema
    path_params_from_payload:
      entity_type_code: entity_type_code
      attribute_set_code: attribute_set_code
      version: attribute_set_version

Emitting events

You never build an event dict and publish it yourself. You insert one outbox row inside the transaction that made the change, and the outbox poller turns it into the CloudEvents envelope through the single EventEnvelope serializer. This guarantees the event's schema mapping, correlation/causation ids, and trace context are all wired the same way, every time.

The low-level primitive is enqueue_event_async(...). For events declared emit_policy: subscribed, use enqueue_event_if_listener(...), which returns None (and writes nothing) when no listener is registered — with the listener-set-unknown case failing open so a Control Plane outage never silently drops an integration signal.

Two things to keep in mind:

• The caller owns the commit. The row is added and flushed inside your route's transaction context; emit on the committed path of a long-lived process. A one-shot script that exits before its transaction commits (or before the poller runs) will appear to lose the event.
• The aggregatetype becomes the CloudEvents type and must be an event declared under spec.events.emits[]; aggregateid becomes the subject and the Kafka key.

event_id = await enqueue_event_async(
    db,
    aggregatetype="support.ticket.created",  # -> CloudEvents 'type' / manifest emits[].id
    aggregateid=str(ticket.guid),             # -> CloudEvents 'subject' / Kafka key
    type="created",
    payload=payload,                          # GUID-only per compliance policy
    org_guid=org_guid,
    app_instance_guid=instance_guid,
)
# Row is added in the caller's transaction; the CALLER commits.
# The outbox poller drains it and publishes the CloudEvents envelope to Kafka.

from egav_app_sdk.events import enqueue_event_if_listener

# Listener-gated: writes an outbox row only when a listener/webhook is
# subscribed to this event_type (emit_policy: subscribed). Returns the
# event_id, or None when suppressed. The two required collaborators —
# emit_policy_resolver and listener_lookup — are built ONCE at startup.
return await enqueue_event_if_listener(
    db,
    aggregatetype=event_type,                # e.g. "comments.mention.created"
    aggregateid=str(aggregate_guid),         # CloudEvents subject / Kafka key
    type=event_type.rsplit(".", 1)[-1],
    payload=payload,
    org_guid=str(org_guid) if org_guid else None,
    app_instance_guid=str(app_instance_guid) if app_instance_guid else None,
    emit_policy_resolver=_get_resolver().emit_policy_for,   # required kwarg
    listener_lookup=_listener_lookup,                       # required kwarg
)

Consuming events in-platform

To react to another app's event inside the platform, declare it under spec.events.consumes[]. Each entry names the producer, the event id/topic, its schema_ref, and one of two handler styles:

• handler_consumer_group — an in-process Kafka consumer group. This is how, for example, Notifications reacts to comments.mention.created.
• handler_endpoint — an HTTP handler the platform dispatches to; it must live under /v1/<service>/internal/* (service-to-service, x-api-key-gated).

The producer + id must resolve to some app's emits[]. An id may appear in both an app's emits and its consumes — but only when the consume declares a handler_endpoint (the deliberate "emit a fact in one process, durably react in another" pattern). A bare overlap with no handler_endpoint is rejected as a likely typo.

events:
  consumes:
  - id: comments.mention.created
    topic: comments.mention.created
    schema_ref: manifest-schemas/events/comments-mention.v1.json
    version: 1
    producer: comments
    handler_consumer_group: notifications-consumer-group

Consumer delivery options

A consumes[] entry does more than name the event — it tells the Control Plane how you want each message delivered. The delivery_mode picks the transport; when you omit it, CP infers it from which handler field you set:

You can also set delivery_mode explicitly. The full set is kafka, http_async, http_sync, kafka_then_http, and webhook_external.

consumes:
# Kafka in-process consumer group (delivery_mode inferred: kafka)
- id: comments.mention.created
  topic: comments.mention.created
  schema_ref: manifest-schemas/events/comments-mention.v1.json
  version: 1
  producer: comments
  handler_consumer_group: notifications-consumer-group
# Explicit HTTP dispatch to an internal handler
- id: provisioning.requested
  topic: provisioning.requested
  schema_ref: schemas/events/provisioning.requested.v1.json
  version: 1
  producer: control-plane
  delivery_mode: http_async
  handler_endpoint: /v1/control-plane-fleet/internal/handlers/provisioning-requested
  required: false

Set required: true only when your app genuinely cannot start without the upstream producer — it gates your provisioning on theirs.

The events the platform emits

The catalog below is loaded from the Control Plane's currently published manifests. It is not a hand-maintained list, and it exposes only events whose manifests explicitly set visibility: public; tenant and platform contracts stay behind authenticated, scoped discovery. Every entry carries the producer's explanation, audience, webhook eligibility, schema version, and registered JSON Schema. For a dynamic event, the schema covers the stable envelope and the catalog explains that the variable payload portion must be resolved at runtime. A missing schema is shown explicitly as unavailable instead of silently presenting a stale contract.

Reacting to an event in-platform is a spec.events.consumes[] entry naming its producer and id. Getting it out to your own systems is a webhook subscription on an event whose manifest sets webhook_subscribable: true.

Webhooks vs in-platform Kafka

The same envelope reaches your systems through webhooks. The Control Plane's event dispatcher drives two HTTP delivery paths that are deliberately kept distinct — do not conflate their signature headers:

The customer webhook body is a flat JSON document (event_id, event_type, occurred_at, org_guid, app_instance_guid, correlation_id, payload, …), delivered with these headers:

Use Kafka for reactions that stay inside the platform; use webhooks to push events to your own infrastructure.

Signing & verifying webhooks

Customer webhooks are signed HMAC-SHA256 over the string event_id.timestamp.body, using a per-subscriber secret, and the wire signature is persisted on the delivery record before the network call — so even a failed attempt keeps an auditable record of exactly what was signed.

The app SDK ships a reference receiver, verify_webhook(...), so you don't have to reimplement verification. It enforces four checks and raises WebhookVerificationError on any failure:

1. Signature — constant-time compare of sha256=<hex> over event_id.timestamp.body.
2. Freshness — the timestamp must be within a ±300s window (DEFAULT_FRESHNESS_WINDOW_SECONDS = 300), rejecting replays.
3. Idempotency — via your seen_event_id callback (dedupe on the event id).
4. Tenant scope — optional allowed_app_instance_guids restricts which instances you'll accept events for.

# How the dispatcher signs a customer webhook (Stripe-style):
signed_payload = f"{ev.event_id}.{timestamp}.{body.decode('utf-8')}"
sig = hmac.new(secret.encode("utf-8"), signed_payload.encode("utf-8"), hashlib.sha256).hexdigest()
headers["x-signature-256"] = f"sha256={sig}"   # _EXT_WEBHOOK_SIG_HEADER

# Internal app-to-app delivery signs the RAW body under a DIFFERENT header:
#   headers["x-platform-signature-256"] = f"sha256={hmac_sha256(secret, body)}"

from egav_app_sdk.events.webhook_verifier import verify_webhook, WebhookVerificationError

payload = verify_webhook(
    secret=cfg.webhook_secret,
    body=await request.body(),
    signature_header=request.headers.get("x-signature-256"),
    event_id_header=request.headers.get("x-event-id"),
    timestamp_header=request.headers.get("x-event-timestamp"),
    seen_event_id=lambda eid: db.dedupe_seen(eid),        # idempotency
    allowed_app_instance_guids=cfg.our_instance_guids,    # tenant scope
)  # enforces signature + 300s freshness + idempotency + tenant scope; raises -> 401

Retries & dead-lettering

Delivery is at-least-once with a fixed exponential backoff schedule. A failed attempt is rescheduled and the delivery stays pending; after the schedule is exhausted it moves to the dead-letter queue.

• Backoff schedule: [10, 60, 300, 1500, 7500] seconds — 5 attempts, then DLQ.
• Transient failures retry: timeouts, 408, 429, and any 5xx.
• Permanent failures don't: any other 4xx marks the delivery failed immediately (no point retrying a 400 or 404).
• 2xx is success. External deliveries always verify TLS.

Because every event carries a stable x-event-id (echoed as idempotency-key), your receiver can safely dedupe redelivered events — build your handler to be idempotent on that id.

_BACKOFF_SECONDS = [10, 60, 300, 1500, 7500]   # 5 attempts, then dead-letter
_MAX_ATTEMPTS = len(_BACKOFF_SECONDS)

# transient (timeout / 408 / 429 / 5xx) -> reschedule next_attempt_at, stay pending
# other 4xx                              -> permanent 'failed'
# attempts exhausted                     -> 'dlq'
delay = _BACKOFF_SECONDS[min(d.attempts, _MAX_ATTEMPTS) - 1]

Subscribing to webhooks

Webhook subscriptions support four transports. HTTP is the default and the common case; the queue transports let you land events directly in your own broker.

Two security properties to note:

• The HTTP HMAC secret is generated server-side and returned in plaintext exactly once, on create or rotate-secret. Store it then; it is never echoed again.
• Only https:// targets are accepted for http_webhook, and queue-transport credentials are stored encrypted — only display-safe fields surface on read (transport_config_public).

Subscribe only to events whose manifest marks them webhook_subscribable: true; events declared emit_policy: subscribed begin producing as soon as your subscription registers as a listener.