CQRS (Command Query Responsibility Segregation) на Java

public class Order {
    private OrderId id;
    private CustomerId customerId;
    private List<OrderLine> orderLines;
    private OrderStatus status;
    
    // Конструкторы, геттеры
    
    public void addProduct(Product product, int quantity) {
        // Бизнес-логика добавления товара
        if (status != OrderStatus.DRAFT) {
            throw new IllegalStateException("Can only modify draft orders");
        }
        
        OrderLine existingLine = findOrderLineByProduct(product);
        if (existingLine != null) {
            existingLine.increaseQuantity(quantity);
        } else {
            orderLines.add(new OrderLine(product, quantity));
        }
    }
    
    public void submit() {
        // Валидация и изменение статуса
        if (orderLines.isEmpty()) {
            throw new IllegalStateException("Cannot submit empty order");
        }
        status = OrderStatus.SUBMITTED;
    }
    
    // Другие методы
}

// Обработчик команды
public void handle(SubmitOrderCommand command) {
    Order order = orderRepository.findById(command.getOrderId());
    order.submit();
    orderRepository.save(order);
    
    // Публикация события
    eventBus.publish(new OrderSubmittedEvent(order.getId(), order.getCustomerId()));
}
 
// Обработчик события для обновления модели чтения
public void on(OrderSubmittedEvent event) {
    OrderView orderView = orderViewRepository.findById(event.getOrderId());
    orderView.setStatus("SUBMITTED");
    orderView.setSubmissionDate(LocalDateTime.now());
    orderViewRepository.save(orderView);
}

// Шина команд
public class CommandBus {
    private Map<Class<? extends Command>, CommandHandler> handlers = new HashMap<>();
    
    public <T extends Command> void register(Class<T> commandClass, CommandHandler<T> handler) {
        handlers.put(commandClass, handler);
    }
    
    public void dispatch(Command command) {
        CommandHandler handler = handlers.get(command.getClass());
        if (handler == null) {
            throw new IllegalStateException("No handler for " + command.getClass());
        }
        handler.handle(command);
    }
}

com.example.cqrs
├── command
│   ├── model
│   ├── handler
│   └── repository
├── query
│   ├── model
│   ├── handler
│   └── repository
├── event
│   ├── model
│   └── handler
├── domain
│   └── model
└── infrastructure
    ├── bus
    ├── config
    └── persistence

package com.example.cqrs.command.model;
 
public class CreateProductCommand {
    private final String productId;
    private final String name;
    private final BigDecimal price;
    private final String category;
    
    // Конструктор, геттеры
    
    public CreateProductCommand(String productId, String name, 
                               BigDecimal price, String category) {
        this.productId = productId;
        this.name = name;
        this.price = price;
        this.category = category;
    }
    
    // Геттеры опущены для краткости
}

package com.example.cqrs.command.handler;
 
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
 
@Service
public class CreateProductCommandHandler {
    
    private final ProductRepository productRepository;
    private final EventBus eventBus;
    
    public CreateProductCommandHandler(ProductRepository productRepository, 
                                       EventBus eventBus) {
        this.productRepository = productRepository;
        this.eventBus = eventBus;
    }
    
    @Transactional
    public void handle(CreateProductCommand command) {
        // Проверка бизнес-правил
        if (productRepository.existsByName(command.getName())) {
            throw new DuplicateProductException("Product with name already exists");
        }
        
        // Создание доменной модели
        Product product = new Product(
            new ProductId(command.getProductId()),
            command.getName(),
            command.getPrice(),
            command.getCategory()
        );
        
        // Сохранение
        productRepository.save(product);
        
        // Публикация события
        eventBus.publish(new ProductCreatedEvent(
            product.getId().getValue(),
            product.getName(),
            product.getPrice(),
            product.getCategory()
        ));
    }
}

package com.example.cqrs.query.model;
 
public class GetProductQuery {
    private final String productId;
    
    public GetProductQuery(String productId) {
        this.productId = productId;
    }
    
    public String getProductId() {
        return productId;
    }
}

package com.example.cqrs.query.handler;
 
import org.springframework.stereotype.Service;
 
@Service
public class GetProductQueryHandler {
    
    private final ProductViewRepository repository;
    
    public GetProductQueryHandler(ProductViewRepository repository) {
        this.repository = repository;
    }
    
    public ProductView handle(GetProductQuery query) {
        return repository.findById(query.getProductId())
            .orElseThrow(() -> new ProductNotFoundException(query.getProductId()));
    }
}

package com.example.cqrs.infrastructure.bus;
 
import org.springframework.stereotype.Component;
 
@Component
public class CommandBus {
    
    private final Map<Class<? extends Command>, CommandHandler<? extends Command>> handlers = 
        new HashMap<>();
    
    @SuppressWarnings("unchecked")
    public <T extends Command> void register(Class<T> commandType, CommandHandler<T> handler) {
        handlers.put(commandType, handler);
    }
    
    @SuppressWarnings("unchecked")
    public <T extends Command> void dispatch(T command) {
        CommandHandler<T> handler = (CommandHandler<T>) handlers.get(command.getClass());
        if (handler == null) {
            throw new NoHandlerForCommandException(command.getClass());
        }
        handler.handle(command);
    }
}

package com.example.cqrs.event.handler;
 
import org.springframework.stereotype.Service;
 
@Service
public class ProductEventHandler {
    
    private final ProductViewRepository viewRepository;
    
    public ProductEventHandler(ProductViewRepository viewRepository) {
        this.viewRepository = viewRepository;
    }
    
    @EventListener
    public void on(ProductCreatedEvent event) {
        ProductView view = new ProductView(
            event.getProductId(),
            event.getName(),
            event.getPrice(),
            event.getCategory()
        );
        viewRepository.save(view);
    }
    
    @EventListener
    public void on(ProductUpdatedEvent event) {
        ProductView view = viewRepository.findById(event.getProductId())
            .orElseThrow();
        
        view.setName(event.getName());
        view.setPrice(event.getPrice());
        view.setCategory(event.getCategory());
        
        viewRepository.save(view);
    }
}

@Test
public void testCreateProductCommandHandler() {
    // Arrange
    ProductRepository repository = mock(ProductRepository.class);
    EventBus eventBus = mock(EventBus.class);
    
    when(repository.existsByName("Test Product")).thenReturn(false);
    
    CreateProductCommandHandler handler = 
        new CreateProductCommandHandler(repository, eventBus);
    
    CreateProductCommand command = new CreateProductCommand(
        "123", "Test Product", BigDecimal.TEN, "Electronics"
    );
    
    // Act
    handler.handle(command);
    
    // Assert
    verify(repository).save(any(Product.class));
    verify(eventBus).publish(any(ProductCreatedEvent.class));
}

@SpringBootTest
public class ProductIntegrationTest {
 
    @Autowired private CommandBus commandBus;
    @Autowired private QueryBus queryBus;
    
    @Test
    public void createProductShouldUpdateReadModel() {
        // Arrange
        String productId = UUID.randomUUID().toString();
        CreateProductCommand command = new CreateProductCommand(
            productId, "Test Product", BigDecimal.valueOf(99.99), "Test Category"
        );
        
        // Act
        commandBus.dispatch(command);
        
        // Assert with waiting
        await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
            ProductView product = queryBus.dispatch(new GetProductQuery(productId));
            assertNotNull(product);
            assertEquals("Test Product", product.getName());
            assertEquals(BigDecimal.valueOf(99.99), product.getPrice());
        });
    }
}

@Service
public class AsyncCommandBus implements CommandBus {
 
    private final JmsTemplate jmsTemplate;
    
    public AsyncCommandBus(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }
    
    @Override
    public void dispatch(Command command) {
        jmsTemplate.convertAndSend("command-queue", command);
    }
}
 
@Component
@JmsListener(destination = "command-queue")
public class CommandMessageListener {
 
    private final Map<Class<? extends Command>, CommandHandler<? extends Command>> handlers;
    
    // ... инициализация обработчиков ...
    
    @SuppressWarnings("unchecked")
    public void onMessage(Command command) {
        CommandHandler<Command> handler = 
            (CommandHandler<Command>) handlers.get(command.getClass());
        if (handler != null) {
            handler.handle(command);
        } else {
            throw new NoHandlerForCommandException(command.getClass());
        }
    }
}

@Entity
@Table(name = "category_with_products_view")
public class CategoryWithProductsView {
    @Id
    private String categoryId;
    private String categoryName;
    
    @ElementCollection
    @CollectionTable(name = "category_product_view")
    private List<ProductSummary> products = new ArrayList<>();
    
    // Конструкторы, геттеры, сеттеры
}
 
@Embeddable
public class ProductSummary {
    private String productId;
    private String name;
    private BigDecimal price;
    private String imageUrl;
    
    // Конструкторы, геттеры, сеттеры
}

@Entity
public class Product {
    @Id
    private String id;
    private String name;
    private BigDecimal price;
    private String category;
    
    @Version
    private Long version;
    
    // Методы
}

@Configuration
@EnableJpaRepositories(
    basePackages = "com.example.cqrs.command.repository",
    entityManagerFactoryRef = "commandEntityManagerFactory"
)
public class CommandDbConfig {
    // Конфигурация источника данных для команд
}
 
@Configuration
@EnableMongoRepositories(
    basePackages = "com.example.cqrs.query.repository",
    mongoTemplateRef = "queryMongoTemplate"
)
public class QueryDbConfig {
    // Конфигурация MongoDB для запросов
}

@RestController
@RequestMapping("/products/commands")
public class ProductCommandController {
    
    private final CommandBus commandBus;
    
    @PostMapping
    public ResponseEntity<Void> createProduct(@RequestBody CreateProductDTO dto) {
        String productId = UUID.randomUUID().toString();
        commandBus.dispatch(new CreateProductCommand(
            productId, dto.getName(), dto.getPrice(), dto.getCategory()
        ));
        return ResponseEntity.accepted().build();
    }
}
 
@RestController
@RequestMapping("/products")
public class ProductQueryController {
    
    private final QueryBus queryBus;
    
    @GetMapping("/{id}")
    public ResponseEntity<ProductView> getProduct(@PathVariable String id) {
        try {
            ProductView product = queryBus.dispatch(new GetProductQuery(id));
            return ResponseEntity.ok(product);
        } catch (ProductNotFoundException e) {
            return ResponseEntity.notFound().build();
        }
    }
}

// Агрегат с Event Sourcing
public class ProductAggregate {
private String id;
private String name;
private BigDecimal price;
private List<DomainEvent> changes = new ArrayList<>();
 
// Конструктор загружает агрегат из истории событий
public ProductAggregate(List<DomainEvent> events) {
    // Применяем события к агрегату для воссоздания текущего состояния
    events.forEach(this::apply);
}
 
// Обработка команды
public void createProduct(CreateProductCommand command) {
    // Валидация
    if (name != null) {
        throw new IllegalStateException("Product already exists");
    }
    
    // Создаем и применяем событие
    ProductCreatedEvent event = new ProductCreatedEvent(
        command.getProductId(),
        command.getName(),
        command.getPrice(),
        command.getCategory()
    );
    apply(event);
    changes.add(event);
}
 
// Применение события к состоянию агрегата
private void apply(DomainEvent event) {
    if (event instanceof ProductCreatedEvent) {
        this.id = ((ProductCreatedEvent) event).getProductId();
        this.name = ((ProductCreatedEvent) event).getName();
        this.price = ((ProductCreatedEvent) event).getPrice();
    } else if (event instanceof ProductPriceChangedEvent) {
        this.price = ((ProductPriceChangedEvent) event).getNewPrice();
    }
    // ... другие типы событий
}
 
// Метод для получения непримененных изменений
public List<DomainEvent> getUncommittedChanges() {
    return new ArrayList<>(changes);
}
 
// Метод для фиксации изменений
public void markChangesAsCommitted() {
    changes.clear();
}
}

@Aggregate
public class Product {
@AggregateIdentifier
private String id;
private String name;
private BigDecimal price;
 
@CommandHandler
public Product(CreateProductCommand command) {
    apply(new ProductCreatedEvent(
        command.getProductId(),
        command.getName(),
        command.getPrice(),
        command.getCategory()
    ));
}
 
@CommandHandler
public void changePrice(ChangePriceCommand command) {
    apply(new ProductPriceChangedEvent(id, command.getNewPrice()));
}
 
@EventSourcingHandler
protected void on(ProductCreatedEvent event) {
    this.id = event.getProductId();
    this.name = event.getName();
    this.price = event.getPrice();
}
 
@EventSourcingHandler
protected void on(ProductPriceChangedEvent event) {
    this.price = event.getNewPrice();
}
}

// Компенсирующее событие
public class ProductCorrectionEvent extends DomainEvent {
private final String productId;
private final String correctName;
private final BigDecimal correctPrice;
private final String reason;
 
// ... конструктор, геттеры
}

@Component
@Scheduled(fixedRate = 60000) // Раз в минуту
public class ReadModelIntegrityChecker {
 
private final EventStore eventStore;
private final ProductViewRepository viewRepo;
 
public void checkAndRepair() {
    // Получаем все ID продуктов
    List<String> productIds = eventStore.getAllProductIds();
    
    // Проверяем каждый продукт
    for (String id : productIds) {
        // Проверяем наличие в модели чтения
        if (!viewRepo.existsById(id)) {
            // Восстанавливаем из событий
            List<DomainEvent> events = eventStore.getEventsForProduct(id);
            createOrUpdateProductView(events);
        }
    }
}
}

@Component
public class SnapshotStrategy {
 
private final EventStore eventStore;
private final SnapshotRepository snapshotRepo;
private final int threshhold = 100; // Создавать снимок каждые 100 событий
 
public void handleNewEvents(String aggregateId, long eventCount) {
    if (eventCount % threshhold == 0) {
        // Запрашиваем все события
        List<DomainEvent> events = eventStore.getEventsForAggregate(aggregateId);
        
        // Восстанавливаем состояние
        ProductAggregate aggregate = new ProductAggregate(events);
        
        // Сохраняем снимок
        snapshotRepo.save(new ProductSnapshot(aggregateId, aggregate.getState()));
    }
}
}

@Aspect
@Component
public class EventMonitor {
 
private final Logger logger = LoggerFactory.getLogger(EventMonitor.class);
private final MetricRegistry metrics;
 
@Pointcut("execution(* com.example.cqrs.event.EventStore.save(..))")
public void eventSaved() {}
 
@Before("eventSaved() && args(event)")
public void beforeEventSaved(DomainEvent event) {
    // Инкрементируем счетчик по типу события
    metrics.counter("events." + event.getClass().getSimpleName()).inc();
    
    // Логируем событие
    logger.info("Saving event: {}", event);
}
}

@Component
public class ProductCatalogProjection {
 
private final ProductViewRepository viewRepository;
 
@EventHandler
public void on(ProductCreatedEvent event) {
    ProductView view = new ProductView();
    view.setId(event.getAggregateId());
    view.setName(event.getName());
    view.setPrice(event.getPrice());
    view.setCategory(event.getCategory());
    view.setCreatedAt(event.getTimestamp());
    view.setUpdatedAt(event.getTimestamp());
    
    viewRepository.save(view);
}
 
@EventHandler
public void on(ProductPriceChangedEvent event) {
    ProductView view = viewRepository.findById(event.getAggregateId())
        .orElseThrow();
        
    view.setPrice(event.getNewPrice());
    view.setUpdatedAt(event.getTimestamp());
    
    // Если это специальное предложение, добавляем метку
    if (event.isSpecialOffer()) {
        view.setSpecialOffer(true);
        view.setSpecialOfferEndsAt(event.getSpecialOfferEndsAt());
    }
    
    viewRepository.save(view);
}
 
// Другие обработчики событий
}

@Component
public class SalesStatisticsProjection {
 
private final StatisticsRepository repository;
private final Clock clock;
 
@EventHandler
public void on(OrderPlacedEvent event) {
    // Обновляем статистику за час
    LocalDateTime hour = event.getTimestamp().truncatedTo(ChronoUnit.HOURS);
    SalesStatistics hourlyStats = repository
        .findByPeriodAndGranularity(hour, Granularity.HOURLY)
        .orElse(new SalesStatistics(hour, Granularity.HOURLY));
    
    hourlyStats.incrementOrdersCount();
    hourlyStats.addToTotalSales(event.getTotalAmount());
    
    // Обновляем статистику товаров
    for (OrderItem item : event.getItems()) {
        hourlyStats.addProductSold(item.getProductId(), item.getQuantity());
    }
    
    repository.save(hourlyStats);
    
    // Аналогичным образом обновляем дневную и месячную статистику
    updateDailyStatistics(event);
    updateMonthlyStatistics(event);
}
 
// ... другие методы
}

@Component
public class ProductSearchProjection {
 
private final ElasticsearchOperations elasticsearchOperations;
 
@EventHandler
public void on(ProductCreatedEvent event) {
    ProductDocument document = new ProductDocument();
    document.setId(event.getAggregateId());
    document.setName(event.getName());
    document.setCategory(event.getCategory());
    document.setPrice(event.getPrice().doubleValue());
    document.setTags(extractTags(event.getName(), event.getCategory()));
    
    elasticsearchOperations.save(document);
}
 
private List<String> extractTags(String name, String category) {
    // Извлекаем ключевые слова из названия и категории
    // для улучшения поиска
    // ...
}
 
// ... другие обработчики
}

@Service
public class EventCompactionService {
 
@Scheduled(cron = "0 0 2 * * *")  // Запуск каждый день в 2 часа ночи
public void compactEventStreams() {
    // Находим агрегаты с чрезмерным количеством событий
    List<String> hotAggregates = eventStore.findHotAggregates(1000);
    
    for (String aggregateId : hotAggregates) {
        compactAggregateEvents(aggregateId);
    }
}
 
private void compactAggregateEvents(String aggregateId) {
    // Получаем все события агрегата
    List<DomainEvent> events = eventStore.getEventsForAggregate(aggregateId);
    
    // Создаем снапшот состояния
    Object aggregate = aggregateService.loadAggregate(aggregateId);
    long latestVersion = events.get(events.size() - 1).getVersion();
    
    // Сохраняем сводное событие
    CompactionEvent compactionEvent = new CompactionEvent(
        aggregateId,
        latestVersion,
        serialize(aggregate)
    );
    
    // Архивируем старые события и сохраняем компактное
    eventStore.archiveEvents(aggregateId, 0, latestVersion - 1);
    eventStore.save(compactionEvent);
}
}

@Component
public class BatchingEventProcessor {
 
private final List<DomainEvent> eventBatch = new ArrayList<>();
private final int BATCH_SIZE = 100;
private final EventHandler eventHandler;
 
@Scheduled(fixedRate = 100)  // Проверяем каждые 100 мс
public void processBatch() {
    List<DomainEvent> currentBatch;
    
    synchronized (eventBatch) {
        if (eventBatch.isEmpty()) {
            return;
        }
        
        currentBatch = new ArrayList<>(eventBatch);
        eventBatch.clear();
    }
    
    // Группируем события по типам для оптимизации обработки
    Map<Class<?>, List<DomainEvent>> eventsByType = currentBatch.stream()
        .collect(Collectors.groupingBy(DomainEvent::getClass));
    
    // Обрабатываем группами
    for (Map.Entry<Class<?>, List<DomainEvent>> entry : eventsByType.entrySet()) {
        eventHandler.handleBatch(entry.getValue());
    }
}
 
public void addEvent(DomainEvent event) {
    synchronized (eventBatch) {
        eventBatch.add(event);
        
        // Если достигли размера пакета, запускаем обработку
        if (eventBatch.size() >= BATCH_SIZE) {
            processBatch();
        }
    }
}
}

@Test
public void shouldCreateProductWhenValidCommand() {
    fixture.given() // Пустая история для нового продукта
        .when(new CreateProductCommand("prod-1", "Test Product", 
                                      BigDecimal.TEN, "Test"))
        .expectEvents(new ProductCreatedEvent("prod-1", "Test Product", 
                                            BigDecimal.TEN, "Test"));
}
 
@Test
public void shouldUpdatePriceWhenPriceChangeCommand() {
    fixture.given(new ProductCreatedEvent("prod-1", "Test Product", 
                                         BigDecimal.TEN, "Test"))
        .when(new ChangePriceCommand("prod-1", BigDecimal.valueOf(15.0)))
        .expectEvents(new ProductPriceChangedEvent("prod-1", 
                                                 BigDecimal.valueOf(15.0)));
}

@Component
public class ProductCacheManager {
 
private final Cache<String, ProductView> productCache;
 
public ProductCacheManager() {
    // Настраиваем Caffeine с TTL и максимальным размером
    this.productCache = Caffeine.newBuilder()
        .expireAfterWrite(1, TimeUnit.HOURS)
        .maximumSize(10_000)
        .build();
}
 
@EventHandler
public void on(ProductUpdatedEvent event) {
    // Инвалидируем кэш при обновлении товара
    productCache.invalidate(event.getProductId());
}
 
public ProductView getProduct(String productId, Supplier<ProductView> loader) {
    return productCache.get(productId, key -> loader.get());
}
}

@Configuration
public class EventProcessingConfig {
 
@Bean
public Consumer<ProductEvent> processProductEvents() {
    return event -> {
        if (event instanceof ProductCreatedEvent) {
            handleProductCreated((ProductCreatedEvent) event);
        } else if (event instanceof ProductUpdatedEvent) {
            handleProductUpdated((ProductUpdatedEvent) event);
        }
    };
}
 
private void handleProductCreated(ProductCreatedEvent event) {
    // Асинхронное обновление модели чтения
    // ...
}
 
private void handleProductUpdated(ProductUpdatedEvent event) {
    // Асинхронное обновление модели чтения
    // ...
}
}

@Component
@Scheduled(fixedRate = 3600000) // Каждый час
public class DataConsistencyChecker {
 
private final EventStore eventStore;
private final ProductViewRepository viewRepository;
private final ProductEventHandler eventHandler;
 
public void checkConsistency() {
    log.info("Starting consistency check...");
    
    // Получаем все ID продуктов из хранилища событий
    Set<String> eventProductIds = eventStore.getAllProductIds();
    
    // Получаем все ID из модели чтения
    Set<String> viewProductIds = viewRepository.findAllIds();
    
    // Находим отсутствующие в модели чтения
    Set<String> missing = new HashSet<>(eventProductIds);
    missing.removeAll(viewProductIds);
    
    // Перестраиваем отсутствующие проекции
    for (String productId : missing) {
        List<DomainEvent> events = eventStore.getEventsForAggregate(productId);
        for (DomainEvent event : events) {
            eventHandler.handle(event);
        }
    }
    
    // Находим лишние записи в модели чтения
    Set<String> extra = new HashSet<>(viewProductIds);
    extra.removeAll(eventProductIds);
    
    // Удаляем лишние проекции
    for (String productId : extra) {
        viewRepository.deleteById(productId);
    }
    
    log.info("Consistency check completed. Fixed {} missing and removed {} extra records",
            missing.size(), extra.size());
}
}

// В сервисе отзывов
@Service
public class ReviewService {
 
private final ReviewRepository repository;
private final EventPublisher eventPublisher;
 
@Transactional
public void addReview(String productId, String userId, int rating, String comment) {
    Review review = new Review(UUID.randomUUID().toString(), productId, userId, rating, comment);
    repository.save(review);
    
    // Публикуем событие для других сервисов
    eventPublisher.publish(new ReviewAddedEvent(
        review.getId(), 
        review.getProductId(), 
        review.getUserId(), 
        review.getRating(), 
        review.getComment()
    ));
}
}
 
// В сервисе товаров
@Service
@KafkaListener(topics = "reviews")
public class ReviewEventHandler {
 
private final ProductViewRepository repository;
 
public void handleReviewEvent(ReviewAddedEvent event) {
    // Обновляем проекцию товара с информацией об отзыве
    ProductView product = repository.findById(event.getProductId())
        .orElseThrow();
    
    product.getReviews().add(new ReviewSummary(
        event.getReviewId(),
        event.getUserId(),
        event.getRating(),
        event.getComment()
    ));
    
    product.recalculateAverageRating();
    repository.save(product);
}
}

public abstract class VersionedEvent {
private final String aggregateId;
private final long version;
private final LocalDateTime timestamp;
 
// ... геттеры, конструкторы
 
public boolean isNewerThan(VersionedEvent other) {
    if (!aggregateId.equals(other.aggregateId)) {
        throw new IllegalArgumentException("Cannot compare events of different aggregates");
    }
    
    return this.version > other.version;
}
}
 
@Service
public class VersionAwareEventHandler {
 
private final Map<String, Long> processedVersions = new ConcurrentHashMap<>();
 
public void handle(VersionedEvent event) {
    // Проверяем, не обрабатывали ли мы уже более новую версию
    Long currentVersion = processedVersions.get(event.getAggregateId());
    if (currentVersion != null && currentVersion >= event.getVersion()) {
        log.info("Skipping obsolete event version {} for aggregate {}, current version is {}",
                event.getVersion(), event.getAggregateId(), currentVersion);
        return;
    }
    
    // Обработка события...
    
    // Запоминаем версию
    processedVersions.put(event.getAggregateId(), event.getVersion());
}
}

@Service
public class ProductCommandService {
 
private final CommandBus commandBus;
private final ProjectionUpdateTracker updateTracker;
 
@Transactional
public void updateProduct(UpdateProductCommand command) {
    // Отправляем команду
    commandBus.dispatch(command);
    
    // Помечаем проекции как ожидающие обновления
    updateTracker.markPending("product-details-" + command.getProductId());
    updateTracker.markPending("product-catalog");
    
    // Возвращаем клиенту идентификатор обновления
    return updateTracker.getCurrentUpdateId();
}
}
 
// На клиенте
async function updateProduct(product) {
    const updateId = await api.updateProduct(product);
    
    // Ожидаем обновления проекций
    await api.waitForProjectionUpdate(updateId);
    
    // Теперь можно загружать актуальные данные
    const updatedProduct = await api.getProduct(product.id);
    
    // Обновляем UI
    renderProduct(updatedProduct);
}

@Configuration
public class ShardingConfig {
 
@Bean
@Profile("command-node")
public DataSource commandDataSource() {
    ShardedDataSource dataSource = new ShardedDataSource();
    
    // Настраиваем шардирование по customer_id
    dataSource.setShardingStrategy(new CustomerShardingStrategy());
    
    // Добавляем шарды
    dataSource.addShard("shard1", createDataSource("jdbc:mysql://shard1:3306/orders"));
    dataSource.addShard("shard2", createDataSource("jdbc:mysql://shard2:3306/orders"));
    
    return dataSource;
}
 
@Bean
@Profile("query-node")
public DataSource queryDataSource() {
    ShardedDataSource dataSource = new ShardedDataSource();
    
    // Настраиваем шардирование по product_category
    dataSource.setShardingStrategy(new CategoryShardingStrategy());
    
    // Добавляем шарды
    dataSource.addShard("electronics", createDataSource("jdbc:mysql://read1:3306/products"));
    dataSource.addShard("clothing", createDataSource("jdbc:mysql://read2:3306/products"));
    
    return dataSource;
}
 
// ... вспомогательные методы
}

@Aspect
@Component
public class CommandTracingAspect {
 
private final Tracer tracer;
 
@Around("execution(* com.example.cqrs.command.handler.*.*(..))")
public Object traceCommand(ProceedingJoinPoint pjp) throws Throwable {
    String commandName = pjp.getArgs()[0].getClass().getSimpleName();
    
    // Создаем спан трассировки
    Span span = tracer.buildSpan("CommandHandler: " + commandName)
        .withTag("command.type", commandName)
        .start();
    
    try (Scope scope = tracer.activateSpan(span)) {
        return pjp.proceed();
    } catch (Exception e) {
        span.setTag("error", true);
        span.log(Map.of(
            "event", "error",
            "error.kind", e.getClass().getName(),
            "message", e.getMessage()
        ));
        throw e;
    } finally {
        span.finish();
    }
}
}