CQRS with Kafka and Read Models

@Service
@Slf4j
public class OrderReadModelUpdater {

    private final OrderSearchRepository elasticsearchRepo;
    private final OrderCacheRepository redisRepo;

    // Update Elasticsearch read model
    @KafkaListener(topics = {
        "orders.order.created",
        "orders.order.updated",
        "orders.order.cancelled"
    }, groupId = "order-search-updater")
    public void updateSearchIndex(ConsumerRecord<String, Object> record) {
        String eventType = getEventType(record);
        String orderId = record.key();

        switch (eventType) {
            case "OrderCreated", "OrderUpdated" -> {
                OrderSearchDoc doc = buildSearchDoc(record.value());
                elasticsearchRepo.save(doc);  // Upsert by orderId
            }
            case "OrderCancelled" -> {
                OrderSearchDoc doc = elasticsearchRepo.findById(orderId).orElseThrow();
                doc.setStatus("CANCELLED");
                elasticsearchRepo.save(doc);
            }
        }
    }

    // Update Redis cache (hot data)
    @KafkaListener(topics = "orders.order.created", groupId = "order-cache-updater")
    public void updateCache(OrderCreatedEvent event) {
        OrderSummary summary = OrderSummary.from(event);
        redisRepo.save(orderId, summary, Duration.ofHours(24));
    }
}

// Query side — reads from optimized stores, never from Kafka directly
@RestController
@RequestMapping("/api/orders")
public class OrderQueryController {

    @GetMapping("/search")
    public List<OrderSearchResult> search(@RequestParam String query) {
        return elasticsearchRepo.search(query); // Fast full-text search
    }

    @GetMapping("/{id}")
    public OrderSummary getOrder(@PathVariable String id) {
        return redisRepo.get(id) // Cache first
            .orElseGet(() -> orderRepo.findById(id).map(OrderSummary::from).orElseThrow());
    }
}