Consuming Events

In the Community Edition the event pipeline ends at RabbitMQ: every domain event flows from the transactional outbox through change data capture onto the kaiten.events exchange, and the API's own subscribers are its first consumers. There is no managed HTTP delivery — instead you attach your consumer to that exchange and route events wherever they need to go. The events, the catalog and the at-least-once guarantee are the same; the transport is under your control, and no public endpoint has to exist.

Two envelopes, one event

This is the one thing to get right before writing a consumer. The event — name, type, data — is identical on both editions. The envelope around it differs.

The Svix-delivered body is the contract. The RabbitMQ body carries the contract's fields as the outbox row's columns, inside a CloudEvents message whose id and type are the event's own. Read the row once, and the code that handles name, type, data, and headers is the same on both paths.

The outbox row inside the message

Debezium captures the outbox_events table, and its Outbox EventRouter publishes each row as an event of its own. The fields under data.payload are the row's columns, with data renamed payload:

For the customer the receive events page creates, on a self-hosted stack — trimmed to the fields a consumer reads:

{
  "specversion": "1.0",
  "id": "5c3bd582-8b5c-40d6-9d13-8e2ab90e5caa",
  "source": "/debezium/postgresql/kaiten",
  "type": "com.kaiten.customer.v1.created",
  "time": "2026-10-16T21:16:33.969Z",
  "datacontenttype": "application/json",
  "traceparent": "00-a6aae6bf0f42842cf70de6fc51b0ee4b-b76239da7c2c862b-01",
  "data": {
    "payload": {
      "id": "5c3bd582-8b5c-40d6-9d13-8e2ab90e5caa",
      "organization_id": "55555555-5555-5555-5555-555555555555",
      "event_name": "CUSTOMER_CREATED",
      "event_type": "com.kaiten.customer.v1.created",
      "occurred_at": "2026-10-16T21:16:33.963821Z",
      "payload": "{\"id\": \"676d7936-4ac2-4d85-8c6d-adbd621f32bc\", \"name\": \"Webhook Test Restaurant\", \"slug\": \"webhook-test-restaurant\", \"createdAt\": \"2026-10-16T21:16:33.964Z\", …}",
      "headers": "{\"traceparent\": \"00-a6aae6bf0f42842cf70de6fc51b0ee4b-b76239da7c2c862b-01\"}"
    }
  }
}

The message also carries Debezium's own attributes, iodebeziumop, iodebeziumtable, iodebeziumlsn and the rest, which describe the change it read: the operation, the table, the position in the database's log. data holds a schema object beside the row. A consumer reads neither.

Heartbeat messages

Not every message on the exchange is an event. Debezium publishes a heartbeat at a fixed interval — every minute on the Compose stack — so that the replication slot keeps advancing while the database is idle. A heartbeat is a CloudEvents message of type cloud.kaiten.cdc.heartbeat with an empty data:

{
  "specversion": "1.0",
  "type": "cloud.kaiten.cdc.heartbeat",
  "source": "/debezium/heartbeat",
  "id": "heartbeat",
  "datacontenttype": "application/json",
  "data": {}
}

It has no payload, so a consumer that reads data.payload without looking first fails on it once a minute. Test the top-level type, acknowledge a heartbeat, and move on.

Attaching a consumer

There are two ways to receive the stream, and both read the same messages.

With a Dapr sidecar

This is how the API consumes its own events, and the pattern the stack is built for: run your service with a Dapr sidecar, point the sidecar at the same RabbitMQ through a pub/sub component like the one in docker/dapr/, and subscribe to the kaiten.events topic. Dapr delivers each message as an HTTP POST to your endpoint — the same CloudEvents body, heartbeats included — and you answer 200 with {"status": "SUCCESS" | "RETRY" | "DROP"}. You never open an AMQP connection yourself.

Two settings decide whether your consumer receives anything, and whether a RETRY means something:

• A RETRY is redelivered only under a Dapr Resiliency policy that covers your app. Without one, the sidecar rejects the message and the broker discards it: one attempt, no redelivery. The policy the stack ships, docker/dapr/resiliency.yaml, is scoped to the API's own app id (kaiten-api); declare the equivalent for yours.
• In a cluster, the pub/sub component is scoped by app id. A sidecar whose app id is not listed does not load the component and never subscribes, and nothing reports it. Add your consumer's app id to dapr.extraScopes in the Helm chart's values.

With an AMQP client

Declare your own durable queue, bind it to the kaiten.events exchange, and consume. The exchange is a fanout: every bound queue receives every event, so filter on the message's type in your consumer rather than through binding keys.

import amqp from 'amqplib';

const conn = await amqp.connect(process.env.RABBITMQ_URL); // amqp://<user>:<password>@localhost:5672
const ch = await conn.createChannel();

const q = await ch.assertQueue('my-app.kaiten-events', { durable: true });
await ch.bindQueue(q.queue, 'kaiten.events', ''); // fanout: the routing key is ignored

ch.consume(q.queue, (msg) => {
  const message = JSON.parse(msg.content.toString());
  if (message.type === 'cloud.kaiten.cdc.heartbeat') {            // no event inside
    return ch.ack(msg);
  }
  if (message.type.startsWith('com.kaiten.instance.entitlement.')) { // the event's own type
    const row = message.data.payload;                             // the outbox row
    handleEvent({
      id: message.id,                                             // the event's id
      name: row.event_name,
      type: message.type,
      data: JSON.parse(row.payload),                              // the JSON column arrives as a string
    });
  }
  ch.ack(msg);
});

conn, _ := amqp.Dial(os.Getenv("RABBITMQ_URL")) // amqp://<user>:<password>@localhost:5672
ch, _ := conn.Channel()

q, _ := ch.QueueDeclare("my-app.kaiten-events", true, false, false, false, nil)
ch.QueueBind(q.Name, "", "kaiten.events", false, nil) // fanout: the routing key is ignored

msgs, _ := ch.Consume(q.Name, "", false, false, false, false, nil)
for msg := range msgs {
    var message struct {
        ID   string `json:"id"`   // the event's id
        Type string `json:"type"` // the event's type, or the heartbeat's
        Data struct {
            Payload struct { // the outbox row
                EventName string `json:"event_name"`
                Payload   string `json:"payload"` // the JSON column arrives as a string
            } `json:"payload"`
        } `json:"data"`
    }
    json.Unmarshal(msg.Body, &message)
    if message.Type == "cloud.kaiten.cdc.heartbeat" { // no event inside
        msg.Ack(false)
        continue
    }
    if strings.HasPrefix(message.Type, "com.kaiten.instance.entitlement.") {
        row := message.Data.Payload
        handleEvent(message.ID, row.EventName, message.Type, json.RawMessage(row.Payload))
    }
    msg.Ack(false)
}

The broker's users and permissions are defined in docker/rabbitmq/rabbitmq-defs.json; create a dedicated user for your consumer rather than reusing the stack's.

How the API consumes its own stream

This is the pattern the stack is built on, and the reason your own consumer stays independent of it.

One Dapr delivery arrives on a single route and is then distributed in process to several consumers — the audit trail, and a CRM connector for the organizations that activated one. Each runs in its own transaction with its own deduplication mark, because the mark records a (message, consumer) pair rather than a message alone. Two consequences follow:

• A delivery where one handler succeeds and another fails is redelivered for the failed one only. The one that succeeded is never replayed.
• The mark and the work share one transaction. Marking first is what makes deduplication safe against a concurrent redelivery; a failure afterwards has to take the mark down with it, or the retry would read the mark, call the event a duplicate, and lose the work silently.

The route answers Dapr with HTTP 200 in every case, and puts the outcome in the body — {"status": "SUCCESS" | "RETRY" | "DROP"}. DROP is for an event that will never succeed: an unparseable message, or one belonging to an organization that has since been deleted, which would otherwise fail a foreign key and retry forever.

What retention means for a consumer

The transport tables are pruned by the API itself, on a schedule.

Seven days is a replay and debugging window, not a storage guarantee. The outbox is a transport buffer that change data capture drains from the write-ahead log within seconds; the inbox only has to remember a message id for as long as the pipeline could still redeliver it, which is far longer than any RabbitMQ or Dapr redelivery window. Neither is an event store, and the exchange itself retains nothing.

The buffer is your queue. Declare it durable, and let it hold messages while your consumer is down. If you need history beyond that, write the events you care about into your own store as you receive them — there is no endpoint that replays the stream.

Practical guidance

• Own your queue. Never consume from the queues the API's sidecar declares — bind your own durable queue to the exchange.
• Be idempotent. At-least-once means duplicates, especially after restarts. Deduplicate on the message's id, the event's id, the way the API's own subscribers do with their inbox table — and key your store on (event id, your consumer name) if you fan one delivery out to several handlers, for the reason above.
• Filter on type, in code. The message's type is the event's. The catalog's types are hierarchical (com.kaiten.customer.v1.*, com.kaiten.instance.entitlement.v1.*); a prefix test keeps your consumer from processing the whole stream.
• Continue the trace. The message's traceparent attribute, the same value as the row's headers.traceparent, is the producing request's trace context. Start your span as its child and the consumer shows up in Jaeger under the request that caused it — see observability. The AMQP message's own traceparent header is another context: Debezium's publish to RabbitMQ, which it traces on its own. The API links its consumer spans to it, rather than starting them under it.
• Calling back into the API is fine, loops are not. A handler that writes to Kaiten emits an event your consumer will receive; make sure that path terminates.

When Kaiten Cloud webhooks fit better

Managed retries with backoff, a delivery history in the console, signed HTTPS delivery to third parties — that is what the webhooks section adds on Kaiten Cloud, on top of the same exchange. Consuming RabbitMQ is the right shape when the consumer lives inside your own infrastructure; webhooks are the right shape when the destination is outside it.

Where to go next

Every event, with its payload contract: the event catalog and the payload examples. What produces the stream: the event pipeline. The concept in one screen: events and webhooks.