// イベント定義
public class OrderCreatedEvent {
    private final Long orderId;
    private final Instant occurredAt;

    public OrderCreatedEvent(Long orderId) {
        this.orderId = orderId;
        this.occurredAt = Instant.now();
    }

    // getters...
}

// イベント発行者
@Service
public class OrderService {
    private final ApplicationEventPublisher eventPublisher;

    public void createOrder(CreateOrderCommand command) {
        Order order = new Order(command);
        orderRepository.save(order);

        // イベント発行
        eventPublisher.publishEvent(new OrderCreatedEvent(order.getId()));
    }
}

// イベント購読者
@Component
public class NotificationService {
    @EventListener
    public void handleOrderCreated(OrderCreatedEvent event) {
        // 注文IDをもとに詳細データを取得
        Order order = orderRepository.findById(event.getOrderId()).orElseThrow();
        sendNotification(order);
    }
}

// イベントストア
public interface EventStore {
    void save(DomainEvent event);
    List<DomainEvent> getEvents(UUID aggregateId);
}

// 集約ルート
public class Order {
    private UUID id;
    private OrderStatus status;
    private List<DomainEvent> uncommittedEvents = new ArrayList<>();

    // イベントを適用
    public void apply(OrderCreatedEvent event) {
        this.id = event.getOrderId();
        this.status = OrderStatus.CREATED;
    }

    public void apply(OrderConfirmedEvent event) {
        this.status = OrderStatus.CONFIRMED;
    }

    // イベントを生成
    public void confirm() {
        if (this.status != OrderStatus.CREATED) {
            throw new IllegalStateException("Cannot confirm order");
        }
        var event = new OrderConfirmedEvent(this.id);
        apply(event);
        uncommittedEvents.add(event);
    }

    // イベントから復元
    public static Order fromEvents(List<DomainEvent> events) {
        Order order = new Order();
        events.forEach(order::apply);
        return order;
    }
}

@Component
public class OrderReadModelUpdater {
    private final OrderReadRepository readRepository;

    @EventListener
    public void handleOrderCreated(OrderCreatedEvent event) {
        OrderReadModel readModel = new OrderReadModel(
            event.getOrderId(),
            event.getCustomerId(),
            "CREATED"
        );
        readRepository.save(readModel);
    }

    @EventListener
    public void handleOrderConfirmed(OrderConfirmedEvent event) {
        readRepository.updateStatus(event.getOrderId(), "CONFIRMED");
    }
}

@Service
public class OrderEventPublisher {
    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
    private static final String TOPIC = "order-events";

    public void publishOrderCreated(Order order) {
        OrderCreatedEvent event = new OrderCreatedEvent(
            order.getId(),
            order.getCustomerId(),
            order.getTotalAmount(),
            Instant.now()
        );

        kafkaTemplate.send(TOPIC, order.getId().toString(), event)
            .addCallback(
                result -> log.info("Event published: {}", event),
                ex -> log.error("Failed to publish event", ex)
            );
    }
}

@Component
public class PaymentEventConsumer {

    @KafkaListener(topics = "order-events", groupId = "payment-service-group")
    public void handleOrderEvent(OrderCreatedEvent event) {
        log.info("Received OrderCreatedEvent: {}", event.getOrderId());

        try {
            // 支払い処理
            processPayment(event);
        } catch (Exception e) {
            // エラーハンドリング（リトライ、DLQへ送信等）
            log.error("Payment processing failed", e);
            throw e; // Kafkaがリトライ
        }
    }

    private void processPayment(OrderCreatedEvent event) {
        // 支払いロジック実装
    }
}

@Component
public class IdempotentEventConsumer {
    private final ProcessedEventRepository processedEventRepo;

    @KafkaListener(topics = "order-events")
    @Transactional
    public void handleEvent(OrderCreatedEvent event) {
        String eventId = event.getEventId();

        // 既に処理済みかチェック
        if (processedEventRepo.existsById(eventId)) {
            log.info("Event already processed: {}", eventId);
            return;
        }

        // ビジネスロジック実行
        processOrder(event);

        // 処理済みマーク
        processedEventRepo.save(new ProcessedEvent(eventId, Instant.now()));
    }
}

@Component
public class OrderSagaOrchestrator {

    @KafkaListener(topics = "payment-events")
    public void handlePaymentCompleted(PaymentCompletedEvent event) {
        // 次のステップ：在庫確保
        inventoryService.reserveStock(event.getOrderId());
    }

    @KafkaListener(topics = "inventory-events")
    public void handleStockReservationFailed(StockReservationFailedEvent event) {
        // 補償トランザクション：支払い取り消し
        paymentService.refund(event.getOrderId());
    }
}

// Kafkaのパーティションキーを使用
kafkaTemplate.send(
    "order-events",
    order.getId().toString(), // ← 同じorderIdは同じパーティションへ
    event
);

// 分散トレーシングID（Correlation ID）の付与
public class OrderCreatedEvent {
    private String eventId;
    private String correlationId; // ← リクエスト全体を追跡
    private Long orderId;
    private Instant timestamp;
}

// ログ出力時に常にcorrelationIdを含める
MDC.put("correlationId", event.getCorrelationId());
log.info("Processing event: {}", event);

public class OrderCreatedEvent {
    private String eventType = "OrderCreated";
    private int version = 2; // ← イベントスキーマのバージョン
    private Long orderId;
    private String customerId;
    private BigDecimal amount; // v2で追加

    // 古いバージョンからの変換
    public static OrderCreatedEvent fromV1(OrderCreatedEventV1 oldEvent) {
        return new OrderCreatedEvent(
            oldEvent.getOrderId(),
            oldEvent.getCustomerId(),
            BigDecimal.ZERO // デフォルト値
        );
    }
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, OrderEvent> kafkaListenerContainerFactory() {
    factory.setCommonErrorHandler(new DefaultErrorHandler(
        new DeadLetterPublishingRecoverer(kafkaTemplate),
        new FixedBackOff(1000L, 3L) // 1秒間隔で3回リトライ
    ));
    return factory;
}