In CQRS the write side and the read side live separately. A read query goes not to the orders table but to a special projection order_summary — optimized for a specific screen. But how do changes from the write side get into the read side? Who updates the projection, and when?
If you update order_summary right inside the same transaction that changes orders, we're back to the monolithic model. If you do it after the commit, we risk losing the update on a failure.
The solution is the outbox pattern: the event is written in the same transaction as the aggregate change, while delivery to Kafka (and further to the read-model) happens asynchronously.
Outbox: one transaction for the aggregate and the event
The essence of the problem: you confirmed an order and want to publish the order.confirmed event to Kafka. If you first save the order, then send the event — a failure can happen between these two steps. The order is saved, the event is lost, the read-model isn't updated.
The outbox solves this simply: the event is saved to the outbox table inside the same pgx.Tx as the order change. Either both changes went through, or both rolled back.
// adapters/out/outbox/order_outbox_repository.go
package outbox
import (
"context"
"encoding/json"
"fmt"
"github.com/jackc/pgx/v5"
)
type OrderOutboxRepository struct{}
func (r *OrderOutboxRepository) Enqueue(ctx context.Context, tx pgx.Tx, evt OrderConfirmedEvent) error {
payload, err := json.Marshal(evt)
if err != nil {
return fmt.Errorf("marshal OrderConfirmed: %w", err)
}
_, err = tx.Exec(ctx,
`INSERT INTO outbox (event_type, payload, created_at)
VALUES ($1, $2, now())`,
"order.confirmed", payload,
)
return err
}
The outbox table schema:
CREATE TABLE outbox (
id bigserial PRIMARY KEY,
event_type text NOT NULL,
payload jsonb NOT NULL,
created_at timestamptz NOT NULL DEFAULT now(),
published boolean NOT NULL DEFAULT false
);
In the command handler the sequence is like this:
pgx.Tx.Begin()
UPDATE orders SET status = 'confirmed' WHERE id = $1
INSERT INTO outbox (event_type, payload) VALUES ('order.confirmed', $2)
pgx.Tx.Commit()
As long as the row is in the outbox, the event won't disappear anywhere. If Kafka is temporarily unavailable, the relay simply waits and tries again.
The relay goroutine: from outbox to Kafka
A separate goroutine periodically takes unpublished records from the outbox and sends them to Kafka. FOR UPDATE SKIP LOCKED lets you run several relay instances in parallel without conflicts.
// adapters/out/outbox/relay.go
package outbox
import (
"context"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/segmentio/kafka-go"
"log/slog"
)
type Relay struct {
db *pgxpool.Pool
writer *kafka.Writer
interval time.Duration
}
func (r *Relay) Run(ctx context.Context) {
ticker := time.NewTicker(r.interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
if err := r.publishBatch(ctx); err != nil {
slog.ErrorContext(ctx, "outbox relay error", "err", err)
}
}
}
}
func (r *Relay) publishBatch(ctx context.Context) error {
tx, err := r.db.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
rows, err := tx.Query(ctx,
`SELECT id, event_type, payload FROM outbox
WHERE published = false
ORDER BY id
FOR UPDATE SKIP LOCKED
LIMIT 100`,
)
if err != nil {
return err
}
var msgs []kafka.Message
var ids []int64
for rows.Next() {
var id int64
var evtType string
var payload []byte
if err := rows.Scan(&id, &evtType, &payload); err != nil {
return err
}
msgs = append(msgs, kafka.Message{Topic: evtType, Value: payload})
ids = append(ids, id)
}
rows.Close()
if err := rows.Err(); err != nil {
return err
}
if len(msgs) == 0 {
return nil
}
if err := r.writer.WriteMessages(ctx, msgs...); err != nil {
return err
}
_, err = tx.Exec(ctx,
`UPDATE outbox SET published = true WHERE id = ANY($1)`, ids,
)
if err != nil {
return err
}
return tx.Commit(ctx)
}
The event payload — a separate struct
A common mistake is to put into the event payload the same type that sqlc generates from the database schema: db.Order, db.Product. The problem is that any change to the table (ALTER TABLE orders ADD COLUMN ...) will automatically change the event type. A consumer reading the old version of the event will break.
The right way is a separate struct specifically for the event:
// core/order/event/order_confirmed.go
package event
import "time"
type OrderConfirmedEvent struct {
EventID string `json:"event_id"`
OrderID string `json:"order_id"`
CustomerID string `json:"customer_id"`
TotalAmount int64 `json:"total_amount"`
ConfirmedAt time.Time `json:"confirmed_at"`
AggregateVersion int64 `json:"aggregate_version"`
}
If you need to change the event structure — create OrderConfirmedEventV2 and publish in parallel. Existing consumers keep working.
Idempotent consumer: protection against duplicates
Kafka guarantees at-least-once delivery. This means the same message can arrive at the consumer twice — for example, if the consumer crashed after processing but before committing the offset.
Without protection against duplicates, the read-model will get a double update. There are two approaches.
A table of processed events
Before updating the read-model, we check whether we've already processed this event. The check and the update happen in a single transaction:
func (c *OrderSummaryConsumer) handle(ctx context.Context, msg kafka.Message) error {
var evt event.OrderConfirmedEvent
if err := json.Unmarshal(msg.Value, &evt); err != nil {
return fmt.Errorf("unmarshal OrderConfirmed: %w", err)
}
tx, err := c.db.BeginTx(ctx, pgx.TxOptions{})
if err != nil {
return fmt.Errorf("begin tx: %w", err)
}
defer tx.Rollback(ctx)
var alreadyProcessed bool
_ = tx.QueryRow(ctx,
`SELECT EXISTS(SELECT 1 FROM processed_event WHERE event_id = $1 AND consumer = $2)`,
evt.EventID, "order-summary-projector",
).Scan(&alreadyProcessed)
if alreadyProcessed {
slog.DebugContext(ctx, "duplicate event skipped", "event_id", evt.EventID)
return nil
}
if err := c.summaries.Upsert(ctx, tx, toSummary(evt)); err != nil {
return fmt.Errorf("upsert order summary: %w", err)
}
_, err = tx.Exec(ctx,
`INSERT INTO processed_event (event_id, consumer) VALUES ($1, $2)`,
evt.EventID, "order-summary-projector",
)
if err != nil {
return fmt.Errorf("mark processed: %w", err)
}
return tx.Commit(ctx)
}
The table schema:
CREATE TABLE processed_event (
event_id text PRIMARY KEY,
consumer text NOT NULL,
processed_at timestamptz NOT NULL DEFAULT now()
);
A versioned UPDATE
If a single aggregate's events arrive strictly in order (one Kafka partition per product_id), you can do without a separate table. We add a version column to the read-model and update only if the event is fresher:
func (c *ProductSummaryConsumer) handle(ctx context.Context, msg kafka.Message) error {
var evt event.ProductUpdatedEvent
if err := json.Unmarshal(msg.Value, &evt); err != nil {
return fmt.Errorf("unmarshal ProductUpdated: %w", err)
}
tx, err := c.db.BeginTx(ctx, pgx.TxOptions{})
if err != nil {
return fmt.Errorf("begin tx: %w", err)
}
defer tx.Rollback(ctx)
tag, err := tx.Exec(ctx,
`UPDATE product_summary
SET name = $1, price = $2, version = $3, updated_at = now()
WHERE product_id = $4 AND version < $3`,
evt.Name, evt.Price, evt.AggregateVersion, evt.ProductID,
)
if err != nil {
return fmt.Errorf("update product_summary: %w", err)
}
if tag.RowsAffected() == 0 {
slog.DebugContext(ctx, "stale or duplicate event skipped",
"product_id", evt.ProductID,
"version", evt.AggregateVersion,
)
}
return tx.Commit(ctx)
}
The condition WHERE version < $new makes repeated delivery safe: a stale event simply won't overwrite a fresher state.
Rebuilding the read-model from scratch
If the read-model is lost (the database was deleted, a new projection type was added), waiting for events from Kafka is pointless — old events may already have been removed by retention. You need a batch restore from the write store.
This is a separate CLI command, not part of the main server:
// cmd/rebuild-summaries/main.go
func rebuildOrderSummaries(ctx context.Context, db *pgxpool.Pool) error {
var lastID string
for {
rows, err := db.Query(ctx,
`SELECT id, customer_id, customer_name, total_amount, status, item_count, created_at
FROM orders
WHERE id > $1
ORDER BY id
LIMIT 500`,
lastID,
)
if err != nil {
return err
}
var summaries []OrderSummaryDTO
for rows.Next() {
var s OrderSummaryDTO
if err := rows.Scan(
&s.OrderID, &s.CustomerID, &s.CustomerName,
&s.TotalAmount, &s.Status, &s.ItemCount, &s.CreatedAt,
); err != nil {
return err
}
summaries = append(summaries, s)
}
rows.Close()
if err := rows.Err(); err != nil {
return err
}
if len(summaries) == 0 {
return nil
}
if err := upsertBatch(ctx, db, summaries); err != nil {
return err
}
lastID = summaries[len(summaries)-1].OrderID
slog.InfoContext(ctx, "batch rebuilt", "last_id", lastID)
}
}
Run it manually or as an initContainer when deploying a new read store.
Eventual consistency and read-your-writes
The read-model is updated asynchronously — between the write to the write store and the appearance of the data in the read-model there's a delay. This is normal and is called eventual consistency.
It's important to mark this explicitly in the API. In a Go service it's convenient to add a header:
func (h *OrderSummaryHandler) Handle(w http.ResponseWriter, r *http.Request) {
orderID := chi.URLParam(r, "id")
summary, err := h.handler.Handle(r.Context(), query.GetOrderSummary{OrderID: orderID})
if err != nil {
httperr.Write(w, r, err)
return
}
w.Header().Set("X-Data-Freshness", "eventual")
render.JSON(w, r, summary)
}
The client sees the header and knows: right after POST /orders, a GET /orders/{id}/summary request may return the previous state. This is an architectural property, not a bug.
Read-your-writes
Sometimes the user must immediately see their changes — for example, after confirming an order we redirect to the order page. Two practical options:
Two endpoints with an explicit choice. The simplest solution:
GET /orders/{id} — from the write store, data always up to date
GET /orders/{id}/summary — from the read-model, eventual consistency, ≤ 2s
The client chooses the appropriate endpoint depending on the scenario.
A version token. The command returns the aggregate version. The UI passes it into the query and waits until the read-model catches up:
type ConfirmOrderResult struct {
OrderID string
AggregateVersion int64
}
type GetOrderSummary struct {
OrderID string
MinVersion int64
}
The query handler polls with a timeout:
func (h *GetOrderSummaryHandler) Handle(ctx context.Context, q query.GetOrderSummary) (view.OrderSummaryDTO, error) {
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
summary, err := h.views.SummaryByID(ctx, q.OrderID)
if err != nil {
return view.OrderSummaryDTO{}, err
}
if summary.Version >= q.MinVersion {
return summary, nil
}
time.Sleep(100 * time.Millisecond)
}
return view.OrderSummaryDTO{}, &ReadModelNotReadyError{OrderID: q.OrderID, MinVersion: q.MinVersion}
}
This option is more complex, but gives an explicit guarantee for critical scenarios.
Common mistakes
Updating the read-model inside the command transaction. tx.Exec("INSERT INTO order_summary …") in the same pgx.Tx isn't CQRS, it's a single model again. Move the update into a consumer via outbox+Kafka.
A PG trigger instead of a consumer. AFTER UPDATE ON orders → UPDATE order_summary works, but such a trigger isn't traced, isn't tested as a separate component, doesn't scale. A consumer on a Kafka event does the same thing explicitly.
The payload is a sqlc type. payload = db.Order{…} means the database schema has become a public contract. One ALTER TABLE — and all consumers need updating. Always use a separate event struct.
A consumer without protection against duplicates. Kafka delivers at least once. Without a processed_event table or a version guard, the read-model will get double updates.
Rebuilding the read-model by waiting for Kafka. With an empty read store you can't wait for events: old events are already deleted. You need a batch restore from the write store.
In short
- The outbox write and the aggregate change go in one
pgx.Tx— either both went through, or both rolled back. - The relay goroutine reads unpublished rows from the outbox via
FOR UPDATE SKIP LOCKEDand sends them to Kafka. - The event payload is a separate
structwithEventIDandAggregateVersion, not a sqlc type from the write schema. - Kafka at-least-once means duplicates: the consumer must be idempotent — via a
processed_eventtable orUPDATE … WHERE version < $new. - Eventual consistency — declare it explicitly via the
X-Data-Freshness: eventualheader. - On losing the read-model — a batch restore from the write store, not waiting for events from Kafka.
- Read-your-writes: the simplest way is two endpoints with different consistency guarantees.
What to read next
- Command side in CQRS with Go — how the outbox event is registered in the command handler.
- Query side in CQRS with Go — the read-only transaction and the view repository.
- When CQRS is justified — from lightweight to an event-driven read-model.