Projections

Projection handlers consume domain events from Kafka and update read models in MongoDB. They are the bridge between the event stream (write side) and the query-optimised read models (read side).

Write side:   Command → Handler → Aggregate → Domain Event → Outbox → Kafka
Read side:    Kafka → Consumer → ProjectionHandler → ReadRepository → MongoDB

Write a Projection Handler

src/User/Infrastructure/Messaging/Projection/UserProjection.php

use Vortos\Cqrs\Attribute\AsProjectionHandler;
use Vortos\Messaging\Contract\EventEnvelope;

#[AsProjectionHandler(consumer: 'user.events', handlerId: 'user.read-model')]
final class UserProjection
{
    public function __construct(
        private readonly UserReadRepository $readRepository,
    ) {}

    public function __invoke(UserRegistered $event, EventEnvelope $envelope): void
    {
        // upsert — never insert (idempotency requirement)
        $this->readRepository->upsert($envelope->aggregateId, [
            '_id'       => $envelope->aggregateId,
            'email'     => $event->email,
            'name'      => $event->name,
            'status'    => 'active',
            'createdAt' => $envelope->occurredAt->format(\DateTimeInterface::ATOM),
        ]);
    }
}

The first parameter is your POPO event payload. EventEnvelope as a second parameter gives you the framework metadata — aggregateId, occurredAt, aggregateVersion, metadata, etc. If you only need the payload, omit EventEnvelope.

#[AsProjectionHandler] Options

#[AsProjectionHandler(
    consumer: 'user.events',        // required — Kafka consumer name
    handlerId: 'user.read-model',   // required — unique ID for idempotency tracking
    priority: 0,                    // optional — higher runs first (default 0)
)]

Parameter	Required	Description
`consumer`	Yes	Must match a registered `KafkaConsumerDefinition` name
`handlerId`	Yes	Unique ID used for idempotency tracking and dead letter identification
`priority`	No	Execution order when multiple handlers process the same event (default 0)

Idempotency — Always Use upsert()

Kafka delivers at-least-once. The same event may arrive twice — on replay, on consumer restart, or after a dead letter retry. Your handler must be idempotent:

// CORRECT — upsert is safe to call twice with same data
$this->readRepository->upsert($envelope->aggregateId, ['_id' => $envelope->aggregateId, ...]);

// WRONG — throws on second delivery
$this->readRepository->insert(['_id' => $envelope->aggregateId, ...]);

MongoStore::upsert() uses replaceOne with upsert: true — inserts if not exists, replaces if exists. Calling it twice with the same data produces the same result.

Multiple Events — One Handler

Handle multiple event types in one projection handler using multiple attributes:

#[AsProjectionHandler(consumer: 'user.events', handlerId: 'user.rm.created')]
#[AsProjectionHandler(consumer: 'user.events', handlerId: 'user.rm.name-updated')]
final class UserReadModelProjection
{
    public function __construct(private readonly UserReadRepository $repo) {}

    public function __invoke(UserRegistered|UserNameUpdated $event, EventEnvelope $envelope): void
    {
        if ($event instanceof UserRegistered) {
            $this->repo->upsert($envelope->aggregateId, [
                '_id'   => $envelope->aggregateId,
                'email' => $event->email,
                'name'  => $event->name,
            ]);
            return;
        }

        $doc         = $this->repo->findById($envelope->aggregateId) ?? [];
        $doc['name'] = $event->newName;
        $this->repo->upsert($envelope->aggregateId, $doc);
    }
}

Projection Handler vs Event Handler

Both use the same infrastructure — #[AsProjectionHandler] is wired exactly like #[AsEventHandler] by HandlerDiscoveryCompilerPass. The distinction is conceptual:

	`#[AsEventHandler]`	`#[AsProjectionHandler]`
Purpose	General event reaction	Update read model
Reads from	Kafka	Kafka
Writes to	Anything	Read repository only
Idempotency	Optional	Required (always upsert)
Return	void	void

Discovery

AsProjectionHandler is registered in CqrsExtension autoconfiguration — handlers tagged vortos.projection_handler are discovered by HandlerDiscoveryCompilerPass in the messaging module and wired as event handlers on the specified consumer. No additional registration needed.

Priority

When multiple projection handlers process the same event on the same consumer, use priority to control execution order:

#[AsProjectionHandler(consumer: 'user.events', handlerId: 'user.primary-rm', priority: 10)]
final class PrimaryUserProjectionHandler { ... }  // runs first

#[AsProjectionHandler(consumer: 'user.events', handlerId: 'user.search-index', priority: 5)]
final class SearchIndexProjectionHandler { ... }  // runs second

#[AsProjectionHandler(consumer: 'user.events', handlerId: 'user.analytics', priority: 0)]
final class AnalyticsProjectionHandler { ... }    // runs last

Higher priority = runs first.

Sync Projections in Dev/Test

In production, projections are updated asynchronously: the event flows through the outbox → Kafka → consumer worker before the read model is updated. In tests and local dev this latency means you cannot assert on read-model state immediately after dispatching a command — you would need arbitrary sleeps or polling.

SyncProjectionEventBusDecorator solves this. In dev and test environments it wraps EventBusInterface and, after the real dispatch() returns, invokes all projection handlers for that event in-process synchronously. The read model is updated before dispatch() returns, so tests can assert on it immediately:

// Test — read model is consistent immediately after dispatch
$this->commandBus->dispatch(new RegisterUser('alice@example.com', 'Alice'));

// No sleep, no polling — the projection already ran synchronously
$user = $this->userReadRepository->findByEmail('alice@example.com');
$this->assertNotNull($user);
$this->assertSame('Alice', $user['name']);

How It Is Registered

MessagingExtension registers SyncProjectionEventBusDecorator automatically in dev and test environments. No configuration is needed.

To opt out in dev when you want to test the real Kafka flow end-to-end, set:

VORTOS_SYNC_PROJECTIONS=false

Error Handling

Projection errors inside the decorator are logged and swallowed — the same way a real Kafka consumer handles transient failures before DLQ. This prevents a broken projection from masking the real domain error when a command fails.

Production Only: Zero Impact

SyncProjectionEventBusDecorator is never active in production. It is gated by kernel.env and VORTOS_SYNC_PROJECTIONS. There is no performance or correctness impact on production deployments.

Projections

On this page