#事件溯源

// 传统 CRUD - 修改状态
@Service
public class AccountService {

    @Autowired
    private AccountRepository repository;

    public void transfer(Long fromId, Long toId, BigDecimal amount) {
        Account from = repository.findById(fromId);
        Account to = repository.findById(toId);

        from.setBalance(from.getBalance().subtract(amount));
        to.setBalance(to.getBalance().add(amount));

        repository.save(from);
        repository.save(to);
    }
}

// 事件溯源 - 用事件驱动
public class AccountAggregate {

    private Long id;
    private BigDecimal balance;
    private List<DomainEvent> uncommittedEvents = new ArrayList<>();

    // 从事件重建
    public static AccountAggregate fromEvents(List<DomainEvent> events) {
        AccountAggregate aggregate = new AccountAggregate();
        for (DomainEvent event : events) {
            aggregate.apply(event);
        }
        return aggregate;
    }

    // 处理命令
    public void transferTo(AccountAggregate target, BigDecimal amount) {
        if (this.balance.compareTo(amount) < 0) {
            throw new InsufficientBalanceException();
        }

        // 生成事件，而不是直接修改状态
        this.uncommittedEvents.add(new MoneyDebitedEvent(this.id, amount, target.getId()));
        target.uncommittedEvents.add(new MoneyCreditedEvent(target.getId(), amount, this.id));

        // 应用事件
        this.apply(new MoneyDebitedEvent(this.id, amount, target.getId()));
        target.apply(new MoneyCreditedEvent(target.getId(), amount, this.id));
    }

    private void apply(DomainEvent event) {
        if (event instanceof MoneyDebitedEvent) {
            MoneyDebitedEvent e = (MoneyDebitedEvent) event;
            this.balance = this.balance.subtract(e.getAmount());
        } else if (event instanceof MoneyCreditedEvent) {
            MoneyCreditedEvent e = (MoneyCreditedEvent) event;
            this.balance = this.balance.add(e.getAmount());
        }
    }

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

    public void markEventsCommitted() {
        uncommittedEvents.clear();
    }
}

// 事件基类
public abstract class DomainEvent {
    private final String eventId;
    private final LocalDateTime occurredAt;
    private final String aggregateId;

    protected DomainEvent(String aggregateId) {
        this.eventId = UUID.randomUUID().toString();
        this.occurredAt = LocalDateTime.now();
        this.aggregateId = aggregateId;
    }
}

// 具体事件
public class OrderCreatedEvent extends DomainEvent {
    private final String orderId;
    private final String customerId;
    private final List<OrderLineItem> lineItems;
    private final BigDecimal totalAmount;

    public OrderCreatedEvent(String orderId, String customerId,
                             List<OrderLineItem> lineItems, BigDecimal totalAmount) {
        super(orderId);
        this.orderId = orderId;
        this.customerId = customerId;
        this.lineItems = lineItems;
        this.totalAmount = totalAmount;
    }

    // Getter...
}

// 错误：不应该修改历史事件
public void correctMistake(String eventId, String correction) {
    Event event = eventStore.findById(eventId);
    event.setData(correction);  // 绝对不要这样做
    eventStore.save(event);
}

// 正确：通过补偿事件修正
public void reverseOrder(String orderId, String reason) {
    // 发布一个「订单已撤销」事件
    publishEvent(new OrderReversedEvent(orderId, reason, LocalDateTime.now()));
}

public interface EventStore {

    /**
     * 保存事件
     */
    void append(String aggregateId, DomainEvent event);

    /**
     * 批量保存事件
     */
    void append(String aggregateId, List<DomainEvent> events);

    /**
     * 获取聚合根的所有事件
     */
    List<DomainEvent> getEvents(String aggregateId);

    /**
     * 获取聚合根从特定版本之后的事件（用于乐观锁）
     */
    List<DomainEvent> getEvents(String aggregateId, long fromVersion);

    /**
     * 获取所有事件（用于投影重建）
     */
    Stream<DomainEvent> getAllEvents();
}

@Repository
public class JdbcEventStore implements EventStore {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Override
    public void append(String aggregateId, DomainEvent event) {
        String sql = """
            INSERT INTO events (aggregate_id, event_type, event_data, occurred_at, version)
            VALUES (?, ?, ?, ?, ?)
            """;
        jdbcTemplate.update(sql,
            aggregateId,
            event.getClass().getSimpleName(),
            serialize(event),
            event.getOccurredAt(),
            event.getVersion()
        );
    }

    @Override
    public List<DomainEvent> getEvents(String aggregateId) {
        String sql = """
            SELECT * FROM events WHERE aggregate_id = ? ORDER BY version ASC
            """;
        return jdbcTemplate.query(sql, (rs, rowNum) -> {
            String eventType = rs.getString("event_type");
            String eventData = rs.getString("event_data");
            return deserialize(eventType, eventData);
        }, aggregateId);
    }
}

// 投影处理器
@Service
public class OrderProjectionService {

    @Autowired
    private OrderWriteRepository writeRepository;
    @Autowired
    private OrderReadRepository readRepository;

    @KafkaListener(topics = "order-events", groupId = "order-projection")
    public void handleEvent(DomainEvent event) {
        if (event instanceof OrderCreatedEvent) {
            projectCreated((OrderCreatedEvent) event);
        } else if (event instanceof OrderCancelledEvent) {
            projectCancelled((OrderCancelledEvent) event);
        }
    }

    private void projectCreated(OrderCreatedEvent event) {
        OrderReadModel readModel = new OrderReadModel();
        readModel.setOrderId(event.getOrderId());
        readModel.setCustomerId(event.getCustomerId());
        readModel.setTotalAmount(event.getTotalAmount());
        readModel.setStatus("CREATED");
        readModel.setCreatedAt(event.getOccurredAt());
        readRepository.save(readModel);
    }

    private void projectCancelled(OrderCancelledEvent event) {
        readRepository.updateStatus(event.getOrderId(), "CANCELLED");
    }
}

// 快照服务
@Service
public class SnapshotService {

    @Autowired
    private EventStore eventStore;
    @Autowired
    private SnapshotRepository snapshotRepository;

    public AccountAggregate getAggregate(String accountId) {
        // 尝试获取最新的快照
        Optional<Snapshot> latestSnapshot = snapshotRepository.findLatest(accountId);

        if (latestSnapshot.isPresent()) {
            // 从快照恢复
            AccountAggregate aggregate = deserialize(latestSnapshot.get().getState());

            // 从快照之后的事件开始重放
            List<DomainEvent> events = eventStore.getEventsAfter(
                accountId,
                latestSnapshot().getVersion()
            );

            for (DomainEvent event : events) {
                aggregate.apply(event);
            }

            return aggregate;
        } else {
            // 没有快照，从头重建
            List<DomainEvent> events = eventStore.getEvents(accountId);
            return AccountAggregate.fromEvents(events);
        }
    }

    public void takeSnapshot(AccountAggregate aggregate) {
        Snapshot snapshot = new Snapshot();
        snapshot.setAggregateId(aggregate.getId());
        snapshot.setVersion(aggregate.getVersion());
        snapshot.setState(serialize(aggregate));
        snapshot.setTakenAt(LocalDateTime.now());
        snapshotRepository.save(snapshot);
    }
}

@Configuration
public class SnapshotConfiguration {

    @Bean
    public SnapshotTrigger snapshotTrigger() {
        // 每 100 个事件拍一次快照
        return new EventCountSnapshotTrigger(100);
        // 或者基于时间：return new TimeBasedSnapshotTrigger(Duration.ofHours(1));
    }
}