CQRS — Deep Dive

graph LR
    C[Client]
    C -->|Commands| WS[Write Side]
    C -->|Queries| RS[Read Side]
    WS --> WDB[(Write DB)]
    WS -->|Events| EB[Event Bus]
    EB --> P[Projector]
    P --> RDB[(Read DB)]
    RS --> RDB

// Command
public record PlaceOrderCommand(
    String customerId,
    List<OrderLineDto> items
) {}

// Command Handler
@Service
public class OrderCommandHandler {
    private final OrderRepository repository;
    private final EventPublisher publisher;

    @Transactional
    public OrderId handle(PlaceOrderCommand cmd) {
        // Business logic
        Order order = Order.create(cmd.customerId(), cmd.items());
        order.validate();  // Invariants

        // Persist
        repository.save(order);

        // Publish event (or via Outbox)
        publisher.publish(new OrderPlaced(order));

        return order.getId();
    }
}

// Query
public record GetOrderSummaryQuery(String orderId) {}

// Query Handler
@Service
public class OrderQueryHandler {
    private final OrderReadRepository readRepository;

    public OrderSummaryDto handle(GetOrderSummaryQuery query) {
        return readRepository.findSummaryById(query.orderId())
            .orElseThrow(() -> new OrderNotFoundException(query.orderId()));
    }
}

@Component
public class OrderProjector {
    private final OrderReadRepository readRepository;

    @EventListener
    public void on(OrderPlaced event) {
        OrderReadModel readModel = new OrderReadModel(
            event.orderId(),
            event.customerId(),
            event.totalAmount(),
            "PLACED",
            event.occurredAt()
        );
        readRepository.save(readModel);
    }

    @EventListener
    public void on(OrderShipped event) {
        OrderReadModel model = readRepository.findById(event.orderId());
        model.setStatus("SHIPPED");
        model.setShippedAt(event.occurredAt());
        readRepository.save(model);
    }
}

-- Write side: normalized
orders (id, customer_id, status, created_at)
order_lines (id, order_id, product_id, quantity, price)
customers (id, name, email)
products (id, name, sku)

-- Read side: denormalized
order_summaries (
    order_id,
    customer_name,
    customer_email,
    status,
    total_amount,
    item_count,
    created_at,
    shipped_at
)

graph LR
    E[Events] --> P1[Order Summary Projector]
    E --> P2[Order Search Projector]
    E --> P3[Dashboard Projector]
    P1 --> DB1[(PostgreSQL)]
    P2 --> DB2[(Elasticsearch)]
    P3 --> DB3[(Redis)]

CREATE MATERIALIZED VIEW order_dashboard AS
SELECT 
    date_trunc('day', created_at) as day,
    status,
    COUNT(*) as order_count,
    SUM(total_amount) as revenue
FROM orders
GROUP BY 1, 2;

-- Refresh periodically or via trigger
REFRESH MATERIALIZED VIEW order_dashboard;

sequenceDiagram
    participant C as Client
    participant W as Write Side
    participant E as Event Bus
    participant P as Projector
    participant R as Read Side

    C->>W: PlaceOrder command
    W->>W: Save order
    W->>E: Publish OrderPlaced
    W->>C: OrderId (immediate)
    E->>P: OrderPlaced event
    P->>R: Update read model
    Note over C,R: Consistency window
    C->>R: GetOrder query
    R->>C: Order data (eventually)

graph LR
    subgraph Same Database
        WT[orders table]
        RT[order_summaries table]
    end
    W[Write Side] --> WT
    WT -->|Trigger/Event| RT
    RT --> R[Read Side]

graph LR
    W[Write Side] --> WDB[(PostgreSQL)]
    WDB -->|Events| K[Kafka]
    K --> P[Projector]
    P --> RDB[(Elasticsearch)]
    R[Read Side] --> RDB

graph LR
    W[Write Side] --> ES[(Event Store)]
    ES -->|Replay| P1[Projection 1]
    ES -->|Replay| P2[Projection 2]
    P1 --> R1[(Read DB 1)]
    P2 --> R2[(Read DB 2)]

sequenceDiagram
    participant O as Ops
    participant P as Projector
    participant ES as Event Store
    participant R as Read DB

    O->>R: DROP read model
    O->>P: Trigger rebuild
    P->>ES: Fetch all events
    ES->>P: Events stream
    P->>R: Rebuild read model
    P->>O: Rebuild complete

// Application-level sync (simpler, but coupled)
@Transactional
public void placeOrder(PlaceOrderCommand cmd) {
    Order order = Order.create(cmd);
    writeRepository.save(order);

    // Update read model in same transaction
    OrderSummary summary = OrderSummary.from(order);
    readRepository.save(summary);
}

// Command side
@RestController
@RequestMapping("/orders")
public class OrderCommandController {
    private final OrderCommandHandler handler;

    @PostMapping
    public ResponseEntity<OrderIdResponse> placeOrder(@RequestBody PlaceOrderRequest request) {
        OrderId id = handler.handle(request.toCommand());
        return ResponseEntity.created(URI.create("/orders/" + id)).body(new OrderIdResponse(id));
    }
}

// Query side
@RestController
@RequestMapping("/orders")
public class OrderQueryController {
    private final OrderQueryHandler handler;

    @GetMapping("/{id}")
    public OrderSummaryDto getOrder(@PathVariable String id) {
        return handler.handle(new GetOrderSummaryQuery(id));
    }

    @GetMapping("/search")
    public List<OrderSummaryDto> search(@RequestParam String q) {
        return handler.handle(new SearchOrdersQuery(q));
    }
}