// 点对点模型：
// [Producer] → [Queue] → [Consumer 1]  (消息只被一个消费者处理)
//                         [Consumer 2]
//                         [Consumer 3]
//
// 特点：
// 1. 消息被消费后即删除（或标记为已消费）
// 2. 多个消费者竞争消费队列中的消息
// 3. 每条消息只被处理一次（在单个消费者组内）
//
// 适用场景：
// - 任务队列：订单处理、邮件发送、图片转码
// - 工作分发：多个 Worker 竞争处理任务
// - 削峰填谷：突发流量通过队列缓冲

// 发布/订阅模型：
// [Publisher] → [Topic] → [Subscriber A]  (所有订阅者都收到消息)
//                         → [Subscriber B]
//                         → [Subscriber C]
//
// 特点：
// 1. 消息不被删除（保留一段时间）
// 2. 所有订阅者都收到消息的副本
// 3. 订阅者之间互不影响
//
// 适用场景：
// - 事件通知：订单创建通知支付、库存、配送服务
// - 数据同步：用户信息变更同步到多个下游系统
// - 审计日志：所有操作记录到审计日志服务

// Kafka 的核心设计：
// 1. Topic 分为多个 Partition（分区）
// 2. 每个 Partition 是一个有序的、不可变的消息日志
// 3. 消息按 Offset（偏移量）定位
// 4. Consumer Group 内的 Consumer 分配不同的 Partition

// Kafka 的优势：
// - 超高吞吐量（百万级消息/秒）
// - 消息持久化（可回溯历史消息）
// - Consumer 控制消费进度（可重新消费）
// - 分区有序（同一 Key 的消息在同一分区内有序）

// Kafka 的局限：
// - 不支持复杂的路由规则
// - 消息确认机制相对简单
// - 不适合低延迟场景（批量发送优化吞吐）

// 典型应用场景：
// - 日志收集
// - 事件流处理
// - 数据管道
// - 微服务事件总线

// RabbitMQ 的核心设计：
// 1. Producer 发送消息到 Exchange（交换机）
// 2. Exchange 根据路由规则将消息分发到 Queue
// 3. Consumer 从 Queue 消费消息

// Exchange 类型：
// - Direct：精确匹配路由键
// - Fanout：广播到所有绑定的队列
// - Topic：通配符匹配路由键（*.order、#.payment）
// - Headers：基于消息头属性路由

// RabbitMQ 的优势：
// - 灵活的路由规则
// - 丰富的消息确认机制
// - 支持优先级队列、延迟消息
// - 低延迟（单消息处理）

// RabbitMQ 的局限：
// - 吞吐量不如 Kafka
// - 消息持久化性能不如 Kafka
// - 不支持消息回溯

// 典型应用场景：
// - 任务队列
// - 复杂路由场景
// - 需要消息确认的场景
// - 需要延迟消息的场景

// 传统 CRUD 方式：
// 数据库保存订单的当前状态
// UPDATE orders SET status = 'PAID' WHERE id = 123
// → 旧状态被覆盖，无法追溯变更历史

// 事件溯源方式：
// 数据库保存订单的所有事件
// Event 1: OrderCreated(orderId=123, customerId=456, items=[...])
// Event 2: OrderApproved(orderId=123, approvedBy=system)
// Event 3: PaymentProcessed(orderId=123, amount=99.99)
// Event 4: OrderConfirmed(orderId=123)
//
// 当前状态 = 重放所有事件
// OrderCreated → OrderApproved → PaymentProcessed → OrderConfirmed
// 最终状态：CONFIRMED

// 事件溯源的优势：
// 1. 完整的审计日志（每个状态变更都有记录）
// 2. 可以回溯到任意时间点的状态
// 3. 可以派生出新的视图（不需要修改事件）
// 4. 天然支持 CQRS

// 事件溯源的代价：
// 1. 查询当前状态需要重放事件（性能问题 → 需要物化视图）
// 2. 事件 Schema 演化复杂（旧事件需要兼容新格式）
// 3. 学习曲线陡峭
// 4. 不适合所有场景（简单 CRUD 不需要事件溯源）

// 事件存储的数据库设计：
CREATE TABLE events (
    event_id VARCHAR(64) PRIMARY KEY,
    aggregate_type VARCHAR(64) NOT NULL,   -- 如 'Order'
    aggregate_id VARCHAR(64) NOT NULL,     -- 如 '123'
    event_type VARCHAR(64) NOT NULL,       -- 如 'OrderCreated'
    event_data JSONB NOT NULL,             -- 事件数据
    event_metadata JSONB,                  -- 元数据（用户ID、时间戳等）
    event_version INT NOT NULL,            -- 事件版本号
    created_at TIMESTAMP DEFAULT NOW(),
    UNIQUE(aggregate_type, aggregate_id, event_version)
);

CREATE INDEX idx_aggregate ON events(aggregate_type, aggregate_id);
CREATE INDEX idx_created_at ON events(created_at);

// 查询某个订单的所有事件：
SELECT * FROM events
WHERE aggregate_type = 'Order' AND aggregate_id = '123'
ORDER BY event_version;

// 传统方式：同一个模型处理读写
// [Client] → [Service] → [Database]
// 读和写使用同一个数据模型

// CQRS 方式：读写分离
// 命令端（Write Side）：
// [Client] → [Command Service] → [Event Store]
//   → 处理业务逻辑，产生事件
//
// 查询端（Read Side）：
// [Client] → [Query Service] → [Read Database]
//   → 直接查询物化视图，无需业务逻辑
//
// 同步：
// [Event Store] → [Event Handler] → [Read Database]
//   → 订阅事件，更新物化视图

// CQRS 的优势：
// 1. 读写性能独立优化
// 2. 读模型可以针对查询优化（如反范式化）
// 3. 写端保证业务一致性，读端保证查询效率
// 4. 天然支持事件溯源

// CQRS 的代价：
// 1. 架构复杂度增加
// 2. 读写数据可能暂时不一致
// 3. 需要维护事件处理器

// At-most-once（至多一次）：
// 消息可能丢失，但不会重复
// 实现：发送后不确认
// 适用：可容忍丢失的场景（如日志收集）

// At-least-once（至少一次）：
// 消息不会丢失，但可能重复
// 实现：发送后等待确认，超时重发
// 适用：不能丢失但可去重的场景（如订单处理）
// 消费者需要实现幂等性

// Exactly-once（恰好一次）：
// 消息不丢失且不重复
// 实现：需要分布式事务或幂等机制
// Kafka 的幂等 Producer + 事务可实现
// 适用：金融交易等严格场景

// 全局有序：
// 所有消息按顺序处理
// 需要单分区 → 吞吐量受限
// 适用：银行流水等严格有序场景

// 分区有序：
// 同一 Key 的消息在同一分区内有序
// 不同 Key 的消息可以并行处理
// 例：同一订单的事件有序，不同订单的事件可以并行
//
// Kafka 实现：
// Producer 发送消息时指定 Key
// producer.send({
//   topic: 'order-events',
//   key: 'order-123',    // 同一订单的消息进入同一分区
//   value: '{"type":"OrderCreated",...}'
// });

// Producer：订单服务发布事件
const { Kafka } = require('kafkajs');
const kafka = new Kafka({ brokers: ['kafka:9092'] });
const producer = kafka.producer();

async function createOrder(orderData) {
  await producer.connect();
  
  const order = await orderRepository.save(orderData);
  
  await producer.send({
    topic: 'order-events',
    messages: [{
      key: order.id,
      value: JSON.stringify({
        type: 'OrderCreated',
        orderId: order.id,
        customerId: order.customerId,
        restaurantId: order.restaurantId,
        items: order.items,
        totalAmount: order.totalAmount,
        timestamp: Date.now()
      })
    }]
  });
  
  return order;
}

// Consumer：支付服务消费事件
const consumer = kafka.consumer({ groupId: 'payment-service' });

async function startPaymentProcessor() {
  await consumer.connect();
  await consumer.subscribe({ topic: 'order-events', fromBeginning: false });
  
  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      const event = JSON.parse(message.value.toString());
      
      switch (event.type) {
        case 'OrderCreated':
          await processPayment(event);
          break;
        case 'OrderCancelled':
          await refundPayment(event);
          break;
      }
    }
  });
}

// 命令端：订单聚合根
public class Order {
    private String id;
    private String customerId;
    private List<OrderItem> items;
    private OrderStatus status;
    private List<DomainEvent> changes = new ArrayList<>();

    public static Order create(String id, String customerId, List<OrderItem> items) {
        Order order = new Order();
        order.apply(new OrderCreatedEvent(id, customerId, items));
        return order;
    }

    public void approve() {
        if (status != OrderStatus.CREATED) {
            throw new IllegalStateException("Cannot approve order in status: " + status);
        }
        apply(new OrderApprovedEvent(id));
    }

    public void cancel() {
        if (status == OrderStatus.DELIVERED) {
            throw new IllegalStateException("Cannot cancel delivered order");
        }
        apply(new OrderCancelledEvent(id));
    }

    private void apply(DomainEvent event) {
        changes.add(event);
        mutate(event);
    }

    private void mutate(DomainEvent event) {
        if (event instanceof OrderCreatedEvent e) {
            this.id = e.orderId();
            this.customerId = e.customerId();
            this.items = e.items();
            this.status = OrderStatus.CREATED;
        } else if (event instanceof OrderApprovedEvent) {
            this.status = OrderStatus.APPROVED;
        } else if (event instanceof OrderCancelledEvent) {
            this.status = OrderStatus.CANCELLED;
        }
    }

    public List<DomainEvent> getUncommittedEvents() {
        return changes;
    }
}

// 查询端：物化视图
public class OrderView {
    private String orderId;
    private String customerId;
    private String status;
    private BigDecimal totalAmount;
    private LocalDateTime createdAt;
}

// 事件处理器：将事件投影到物化视图
@Component
public class OrderViewProjector {
    @Autowired
    private OrderViewRepository repository;

    @KafkaListener(topics = "order-events", groupId = "view-projector")
    public void handleEvent(String eventJson) {
        DomainEvent event = deserialize(eventJson);
        
        if (event instanceof OrderCreatedEvent e) {
            OrderView view = new OrderView();
            view.setOrderId(e.orderId());
            view.setCustomerId(e.customerId());
            view.setStatus("CREATED");
            view.setTotalAmount(calculateTotal(e.items()));
            view.setCreatedAt(LocalDateTime.now());
            repository.save(view);
        } else if (event instanceof OrderApprovedEvent e) {
            repository.findByOrderId(e.orderId())
                .ifPresent(v -> { v.setStatus("APPROVED"); repository.save(v); });
        }
    }
}

// Outbox 模式：保证业务操作和消息发送的原子性
// 问题：如何保证"保存订单"和"发送 OrderCreated 事件"的原子性？
// 方案：将事件写入同一个数据库的 Outbox 表

@Transactional
public Order createOrder(CreateOrderCommand command) {
    Order order = Order.create(command.getId(), command.getCustomerId(), command.getItems());
    orderRepository.save(order);
    
    // 将事件写入 Outbox 表（同一个事务）
    OutboxEvent event = new OutboxEvent();
    event.setAggregateType("Order");
    event.setAggregateId(order.getId());
    event.setEventType("OrderCreated");
    event.setPayload(serialize(order.getUncommittedEvents()));
    outboxRepository.save(event);
    
    return order;
}

// 独立的 Event Publisher 进程定期扫描 Outbox 表
// 将事件发送到 Kafka，发送成功后删除 Outbox 记录
@Scheduled(fixedDelay = 1000)
public void publishEvents() {
    List<OutboxEvent> events = outboxRepository.findUnpublished();
    for (OutboxEvent event : events) {
        kafkaProducer.send(event.getAggregateType(), event.getPayload());
        outboxRepository.markPublished(event.getId());
    }
}