← Back to the section

In CQRS the write side and the read side are physically separated: one database accepts commands, another answers queries. But separating them is only half the job. You need changes from the write side to reliably reach the read side. That's exactly the task of synchronization.

The most popular approach is events through a message broker. This article breaks down the mechanism in practice: how not to lose an event, how to cope with duplicates, and what to do when the read-model has to be rebuilt from scratch.

Why you can't just send an event after saving

The first impulse looks logical: save the aggregate → send an event to Kafka. What could go wrong?

Everything. The save to the DB succeeded, but Kafka was unavailable at that moment — the event is lost, the read-model lags forever. Or the opposite: Kafka received the event, but the DB transaction rolled back — the read side now has data that doesn't exist in the write side.

Both scenarios break consistency between the sides.

Outbox: atomicity with the aggregate

The solution is the outbox pattern. Instead of sending directly to Kafka, the event is written into a separate table (outbox) in the same database and in the same transaction as the aggregate change. Either both changes are applied, or neither.

transaction COMMIT:
  1. UPDATE orders SET status = 'CONFIRMED' WHERE id = $1
  2. INSERT INTO outbox (event_type, payload, aggregate_id)
       VALUES ('OrderConfirmed', '{...}', $1)
  → either both rows, or none

The aggregate registers the event inside its method:

// order.aggregate.ts
export class Order extends AggregateRoot {
  confirm(now: Date): void {
    if (this.status !== OrderStatus.NEW) {
      throw new OrderAlreadyConfirmedError(this.id, this.status);
    }
    this.status = OrderStatus.CONFIRMED;
    this.confirmedAt = now;
    this.registerEvent(new OrderConfirmed(this.id, this.status, now));
  }
}

On save the repository writes the registered events into the outbox table:

// typeorm-order.repository.ts
@Injectable()
export class TypeOrmOrderRepository implements OrderRepository {
  constructor(
    private readonly dataSource: DataSource,
    @Inject(OUTBOX_WRITER) private readonly outbox: OutboxWriter,
  ) {}

  async save(order: Order): Promise<void> {
    const manager = this.dataSource.createEntityManager();
    await manager.save(OrderOrmEntity, toOrmEntity(order));
    for (const event of order.getUncommittedEvents()) {
      await this.outbox.write(manager, event);
    }
    order.commit();
  }
}

A separate component — the outbox-relay — periodically reads from the outbox and publishes messages to Kafka. Until publication succeeds, the row stays in the table and the relay keeps retrying.

// outbox-relay — simplified logic:
// SELECT ... FOR UPDATE SKIP LOCKED LIMIT 100
// → publish to Kafka
// → mark as published

Idempotent consumer: protection against duplicates

Kafka guarantees "at-least-once" delivery. That means a single message can arrive at the consumer twice: for example, the consumer processed it but crashed before committing the offset.

Without protection, repeated processing will break the read-model. So the consumer must be idempotent — able to process a duplicate message safely.

Two approaches:

A table of processed events

Before updating the read-model, we check whether we've already processed this event:

// order-summary.projector.ts
@Injectable()
export class OrderSummaryProjector {
  @EventPattern('order.events')
  async onOrderConfirmed(@Payload() event: OrderConfirmedDto): Promise<void> {
    await this.tx.run(async () => {
      const isDuplicate = await this.processedEvents.exists(
        event.eventId,
        'order-summary-projector',
      );
      if (isDuplicate) return;

      await this.processedEvents.markProcessed(event.eventId, 'order-summary-projector');
      await this.summaries.updateStatus(event.orderId, OrderStatus.CONFIRMED, event.confirmedAt);
    });
  }
}

The table structure:

CREATE TABLE processed_event (
    event_id     UUID PRIMARY KEY,
    consumer     TEXT NOT NULL,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

Pro: an exact guarantee. Con: an extra write on every event.

A conditional UPDATE by version

If every event carries an aggregateVersion, you can use it as protection:

@EventPattern('order.events')
async onOrderConfirmed(@Payload() event: OrderConfirmedDto): Promise<void> {
  await this.summaries.updateStatusIfNewer(
    event.orderId,
    OrderStatus.CONFIRMED,
    event.confirmedAt,
    event.aggregateVersion,
  );
}

UPDATE order_summary
SET    status = $1, confirmed_at = $2, version = $3, updated_at = NOW()
WHERE  order_id = $4 AND version < $3

If the event was already applied or an old one arrived — nothing happens. Pro: no separate table needed. Con: you need aggregateVersion in the event and ordered delivery within a Kafka partition.

Rebuilding the read-model from scratch

Sometimes the read-model has to be rebuilt: at the first launch of new storage, after a failure, during a structural migration. Waiting for events to arrive from Kafka is not an option: the events may be old or their retention may already have expired.

The right approach is batch rebuild from the write-store at application startup:

// order-summary-bootstrap.ts
@Injectable()
export class OrderSummaryBootstrap implements OnApplicationBootstrap {
  private readonly logger = new Logger(OrderSummaryBootstrap.name);

  constructor(
    @Inject(ORDER_SUMMARY_REPOSITORY) private readonly summaries: OrderSummaryRepository,
    @Inject(ORDER_REPOSITORY)         private readonly orders: OrderRepository,
  ) {}

  async onApplicationBootstrap(): Promise<void> {
    const isEmpty = await this.summaries.isEmpty();
    if (!isEmpty) return;

    this.logger.log('order_summary is empty — starting rebuild');
    await this.rebuildAll();
  }

  private async rebuildAll(): Promise<void> {
    let lastId = 0n;
    while (true) {
      const batch = await this.orders.findAllAfter(lastId, 1000);
      if (batch.length === 0) break;
      await this.summaries.upsertBatch(batch.map(o => toSummary(o)));
      lastId = batch[batch.length - 1].id.value;
    }
  }
}

The logic checks whether the read-store is empty and only then starts the walk. On subsequent startups, when the data already exists, initialization is skipped.

Eventual consistency and the API

When write and read are separated through a broker, there's always a small delay between them — usually fractions of a second. The client must be aware of this, otherwise it will treat the delay as an error.

Describe this in the OpenAPI annotation of the endpoint that reads from the projection:

@Get(':id/summary')
@ApiOperation({
  summary: 'Get order summary (read-projection)',
  description:
    'Returns the read-projection of the order.\n\n' +
    'A delay of up to 1 second between the write operation and the projection update is possible.\n\n' +
    'For immediate consistency use GET /orders/:id — it reads from the write-store.',
})
@ApiOkResponse({ type: OrderSummaryDto })
async getSummary(@Param('id', ParseUUIDPipe) id: string): Promise<OrderSummaryDto> {
  const result = await this.handler.execute(new GetOrderSummary(OrderId.of(id)));
  if (!result) throw new NotFoundException();
  return result;
}

Explicit documentation helps in tests and when debugging problems: the delay is a deliberate architectural property of the system, not a bug.

Read-your-writes: when the client wants to see its change immediately

Sometimes the requirement is stricter: the client sent a command and wants to see the result in the read-projection right away. Three options, from simple to complex:

Two endpoints. The cleanest solution: one endpoint reads from the write-store (immediate consistency), the other from the read-projection (with a possible delay). The client picks the one it needs for the situation.

@Get(':id')
@ApiOperation({ summary: 'Get order (from the write-store, immediate consistency)' })
async getOrder(@Param('id', ParseUUIDPipe) id: string): Promise<OrderDto> { ... }

@Get(':id/summary')
@ApiOperation({ summary: 'Get order summary (from the read-projection, with a possible delay)' })
async getSummary(@Param('id', ParseUUIDPipe) id: string): Promise<OrderSummaryDto> { ... }

Waiting in the handler. After the transaction commits, the handler polls the read-store until the record appears or until a timeout. Suitable only when read-your-writes is truly needed and the load is low — the command's p99 latency grows by the waiting time.

Sticky routing at the gateway. Requests from one client are directed to a single pod. Works only if a consumer in the same process updates the projection synchronously, and breaks when scaling to multiple nodes.

In most cases the first option fits.

Common mistakes

Updating the read-model right in the command handler. It looks like a convenient shortcut, but it destroys the meaning of the separation: the handler knows about the read side, they're coupled again.

PG triggers for table synchronization. Invisible when reading the code, they break on bulk operations, and don't work if the read-model is in another database.

The event is a TypeORM entity of the write schema. Any change to the write-side structure immediately breaks the consumer. An event is a separate class with its own versioning.

What the full flow looks like

1. POST /orders
   → CreateOrderHandler
   → order.create(...)       // registers OrderCreated in the aggregate
   → orderRepository.save()  // INSERT into outbox in the same transaction

2. OutboxRelay (every ~200ms)
   → SELECT ... FOR UPDATE SKIP LOCKED LIMIT 100
   → publishes OrderCreated to the Kafka topic order.events

3. OrderSummaryProjector
   → @EventPattern('order.events')
   → idempotent UPDATE order_summary

4. GET /orders/:id/summary
   → reads from order_summary
   → possible delay ~200ms–1s

This template is the same for any aggregate in the system.

In short

  • Sending directly to Kafka after saving is unreliable: you'll either lose the event or get a "phantom" one. The outbox solves both problems — the event is written in the same transaction as the aggregate change.
  • The consumer must be idempotent: Kafka delivers "at least once". The protection is either a table of processed events or a conditional UPDATE by version.
  • With an empty read-model, don't wait for Kafka: run a batch rebuild from the write-store via OnApplicationBootstrap.
  • Eventual consistency is a documented property, not a bug. State it in the OpenAPI description of the endpoint.
  • If you need immediate consistency — set up a separate endpoint that reads from the write-store.