Low Level Design: CQRS Service – Tech Interview Dot Org

CQRS Service: Low Level Design

CQRS (Command Query Responsibility Segregation) separates the write model (commands) from the read model (queries), enabling independent scaling, optimization, and consistency strategies for each side.

Architecture Overview

The system splits into two distinct paths: the command side handles all state mutations, while the query side serves all read requests from a denormalized projection.

Command Side

Command Handler

The CommandHandler receives incoming commands, validates business rules against the write model, persists the command, and publishes a domain event to Kafka.

class CommandHandler:
    def handle(self, command: Command) -> CommandResult:
        # Idempotency check
        if self.command_repo.exists(command.idempotency_key):
            return CommandResult(status='already_processed')

        # Business rule validation against write model
        self.validator.validate(command)

        # Persist command record
        cmd_record = self.command_repo.insert(
            type=command.type,
            payload=command.payload,
            status='pending',
            idempotency_key=command.idempotency_key
        )

        # Publish domain event to Kafka
        event = DomainEvent(
            id=uuid4(),
            command_id=cmd_record.id,
            type=command.event_type(),
            payload=command.payload,
            occurred_at=utcnow()
        )
        self.event_bus.publish(topic='domain-events', event=event)

        # Update write model (normalized PostgreSQL)
        self.write_model.apply(command)

        self.command_repo.update_status(cmd_record.id, 'processed')
        return CommandResult(status='accepted', command_id=cmd_record.id)

Command Table Schema

CREATE TABLE commands (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    type            VARCHAR(100) NOT NULL,
    payload         JSONB NOT NULL,
    status          VARCHAR(20) NOT NULL DEFAULT 'pending',
                    -- pending | processed | failed
    idempotency_key VARCHAR(255) UNIQUE NOT NULL,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    processed_at    TIMESTAMPTZ
);

CREATE INDEX idx_commands_status ON commands(status);
CREATE INDEX idx_commands_idempotency ON commands(idempotency_key);

Query Side

Read Model

The read model is a denormalized projection stored in PostgreSQL or Redis, rebuilt from domain events by the ProjectionWorker. Queries bypass the write model entirely.

class ProjectionWorker:
    def __init__(self):
        self.consumer = KafkaConsumer(
            topic='domain-events',
            group_id='projection-worker',
            auto_offset_reset='earliest'
        )

    def run(self):
        for event in self.consumer:
            if self.processed_events.contains(event.id):
                continue  # Idempotent processing
            self.projector.project(event)
            self.processed_events.mark(event.id)
            self.consumer.commit()

class OrderProjector:
    def project(self, event: DomainEvent):
        if event.type == 'OrderCreated':
            self.read_db.upsert('order_views', {
                'order_id': event.payload['order_id'],
                'customer_name': self.lookup_customer(event.payload['customer_id']),
                'total': event.payload['total'],
                'status': 'created',
                'updated_at': event.occurred_at
            })
        elif event.type == 'OrderShipped':
            self.read_db.update('order_views',
                where={'order_id': event.payload['order_id']},
                set={'status': 'shipped', 'updated_at': event.occurred_at}
            )

Idempotent Event Processing

CREATE TABLE processed_events (
    event_id    UUID PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

API Routing

# Command endpoint → command bus
POST /api/orders          → CommandHandler.handle(CreateOrderCommand)
POST /api/orders/:id/ship → CommandHandler.handle(ShipOrderCommand)

# Query endpoint → read model directly
GET /api/orders           → ReadModelRepository.list_orders()
GET /api/orders/:id       → ReadModelRepository.get_order(id)

Eventual Consistency

The read model is typically 50–200ms behind the write model, depending on Kafka consumer lag and projection throughput. This window must be acceptable to the business use case.

Projector Lag Monitoring

# Monitor Kafka consumer group lag
kafka-consumer-groups.sh 
  --bootstrap-server kafka:9092 
  --describe 
  --group projection-worker

# Alert if lag > threshold
class LagMonitor:
    ALERT_THRESHOLD = 1000  # messages

    def check(self):
        lag = self.kafka_admin.consumer_group_lag('projection-worker')
        if lag > self.ALERT_THRESHOLD:
            self.alerting.fire('ProjectionLagHigh', lag=lag)

Compensating Commands

When a command fails mid-way or a business rule is violated post-hoc, compensating commands reverse the effect.

class CancelOrderCommand(Command):
    """Compensating command for CreateOrder"""
    def event_type(self): return 'OrderCancelled'

class RefundPaymentCommand(Command):
    """Compensating command for ChargePayment"""
    def event_type(self): return 'PaymentRefunded'

Scaling Considerations

Command side: scale horizontally; stateless handlers, DB is the consistency boundary
Query side: scale read replicas or Redis cache; reads never touch the write DB
Kafka partitioning: partition by aggregate ID (e.g., order_id) to preserve ordering per aggregate
Multiple projections: same event stream can feed multiple independent read models (e.g., order view, analytics, search index)

Key Design Decisions

Write model uses normalized PostgreSQL for transactional integrity
Read model uses denormalized PostgreSQL or Redis for query performance
Kafka provides durable, replayable event log — projections can be rebuilt from scratch
Idempotency key on commands prevents duplicate processing under retries
Separate processed_events table makes projection idempotent under consumer restarts

{
“@context”: “https://schema.org”,
“@type”: “FAQPage”,
“mainEntity”: [
{
“@type”: “Question”,
“name”: “What is CQRS and why is it used in system design?”,
“acceptedAnswer”: {
“@type”: “Answer”,
“text”: “CQRS (Command Query Responsibility Segregation) separates the write model (commands) from the read model (queries). It's used when read and write workloads have different scaling, performance, or consistency requirements. Commands mutate state and emit events; queries read from a denormalized projection optimized for fast retrieval.”
}
},
{
“@type”: “Question”,
“name”: “How does the read model stay in sync with the write model in CQRS?”,
“acceptedAnswer”: {
“@type”: “Answer”,
“text”: “A ProjectionWorker consumes domain events from Kafka and updates the read model (denormalized PostgreSQL or Redis). The read model is eventually consistent, typically 50'200ms behind the write model depending on consumer lag and projection throughput.”
}
},
{
“@type”: “Question”,
“name”: “How is idempotency handled in a CQRS service?”,
“acceptedAnswer”: {
“@type”: “Answer”,
“text”: “Commands include a unique idempotency_key stored in the commands table with a UNIQUE constraint. The ProjectionWorker tracks processed event IDs in a processed_events table, ensuring each event is applied exactly once even under at-least-once Kafka delivery.”
}
},
{
“@type”: “Question”,
“name”: “How do companies like Atlassian, Amazon, and Uber use CQRS in production?”,
“acceptedAnswer”: {
“@type”: “Answer”,
“text”: “Atlassian uses CQRS-style separation in Jira's issue tracking to scale reads independently from writes. Amazon applies it in order management where order placement (command) and order status queries (read model) are separated for performance. Uber uses event-driven projections to power rider and driver dashboards with low-latency reads from denormalized read models.”
}
}
]
}