第8章：CQRS与Event Sourcing

// 创建订单命令
type CreateOrderCommand struct {
    OrderID  string
    UserID   string
    Items    []OrderItem
    Address  Address
}

// 命令处理器
type CreateOrderCommandHandler struct {
    orderRepo OrderRepository
    eventBus  EventBus
}

func (h *CreateOrderCommandHandler) Handle(cmd CreateOrderCommand) error {
    // 1. 创建聚合
    order, err := NewOrder(cmd.OrderID, cmd.UserID, cmd.Items, cmd.Address)
    if err != nil {
        return err
    }

    // 2. 持久化
    if err := h.orderRepo.Save(order); err != nil {
        return err
    }

    // 3. 发布事件
    for _, event := range order.GetEvents() {
        h.eventBus.Publish(event)
    }

    // 4. 返回简单结果
    return nil
}

// 订单查询DTO
type OrderDTO struct {
    OrderID      string        `json:"order_id"`
    UserName     string        `json:"user_name"`
    Items        []OrderItemDTO `json:"items"`
    TotalPrice   float64       `json:"total_price"`
    Status       string        `json:"status"`
    CreatedAt    time.Time     `json:"created_at"`
}

// 查询处理器
type OrderQueryHandler struct {
    readDB *sql.DB  // 读模型数据库
}

func (h *OrderQueryHandler) GetOrderDetail(orderID string) (*OrderDTO, error) {
    // 直接查询优化后的读模型
    query := `
        SELECT o.id, u.name, o.total_price, o.status, o.created_at
        FROM orders_view o
        JOIN users u ON o.user_id = u.id
        WHERE o.id = ?
    `

    var dto OrderDTO
    err := h.readDB.QueryRow(query, orderID).Scan(
        &dto.OrderID,
        &dto.UserName,
        &dto.TotalPrice,
        &dto.Status,
        &dto.CreatedAt,
    )

    return &dto, err
}

// 列表查询（带分页）
func (h *OrderQueryHandler) ListOrders(userID string, page, pageSize int) ([]OrderDTO, error) {
    query := `
        SELECT id, total_price, status, created_at
        FROM orders_view
        WHERE user_id = ?
        ORDER BY created_at DESC
        LIMIT ? OFFSET ?
    `

    rows, err := h.readDB.Query(query, userID, pageSize, (page-1)*pageSize)
    if err != nil {
        return nil, err
    }
    defer rows.Close()

    var orders []OrderDTO
    for rows.Next() {
        var dto OrderDTO
        rows.Scan(&dto.OrderID, &dto.TotalPrice, &dto.Status, &dto.CreatedAt)
        orders = append(orders, dto)
    }

    return orders, nil
}

// 事件接口
type Event interface {
    AggregateID() string
    EventType() string
    OccurredAt() time.Time
    Version() int
}

// 账户创建事件
type AccountCreatedEvent struct {
    accountID   string
    initialBalance Money
    occurredAt  time.Time
    version     int
}

// 存款事件
type MoneyDepositedEvent struct {
    accountID  string
    amount     Money
    occurredAt time.Time
    version    int
}

// 取款事件
type MoneyWithdrawnEvent struct {
    accountID  string
    amount     Money
    occurredAt time.Time
    version    int
}

type EventStore interface {
    // 保存事件
    SaveEvents(aggregateID string, events []Event, expectedVersion int) error

    // 加载事件
    LoadEvents(aggregateID string) ([]Event, error)

    // 加载指定版本之后的事件
    LoadEventsSince(aggregateID string, version int) ([]Event, error)
}

// 实现（MySQL）
type MySQLEventStore struct {
    db *sql.DB
}

func (es *MySQLEventStore) SaveEvents(aggregateID string, events []Event, expectedVersion int) error {
    tx, _ := es.db.Begin()
    defer tx.Rollback()

    // 1. 检查版本冲突（乐观锁）
    var currentVersion int
    err := tx.QueryRow(
        "SELECT COALESCE(MAX(version), 0) FROM events WHERE aggregate_id = ?",
        aggregateID,
    ).Scan(&currentVersion)

    if currentVersion != expectedVersion {
        return errors.New("version conflict")
    }

    // 2. 保存事件
    for _, event := range events {
        eventData, _ := json.Marshal(event)

        _, err := tx.Exec(`
            INSERT INTO events (aggregate_id, event_type, event_data, version, created_at)
            VALUES (?, ?, ?, ?, ?)
        `, aggregateID, event.EventType(), eventData, event.Version(), event.OccurredAt())

        if err != nil {
            return err
        }
    }

    return tx.Commit()
}

func (es *MySQLEventStore) LoadEvents(aggregateID string) ([]Event, error) {
    rows, err := es.db.Query(`
        SELECT event_type, event_data, version, created_at
        FROM events
        WHERE aggregate_id = ?
        ORDER BY version ASC
    `, aggregateID)

    if err != nil {
        return nil, err
    }
    defer rows.Close()

    var events []Event
    for rows.Next() {
        var eventType string
        var eventData []byte
        var version int
        var createdAt time.Time

        rows.Scan(&eventType, &eventData, &version, &createdAt)

        // 反序列化事件
        event := es.deserializeEvent(eventType, eventData)
        events = append(events, event)
    }

    return events, nil
}

// 账户聚合
type Account struct {
    id              string
    balance         Money
    status          AccountStatus
    version         int
    uncommittedEvents []Event
}

// 从事件重建聚合
func LoadAccount(eventStore EventStore, accountID string) (*Account, error) {
    events, err := eventStore.LoadEvents(accountID)
    if err != nil {
        return nil, err
    }

    account := &Account{id: accountID}

    // 重放事件
    for _, event := range events {
        account.Apply(event)
    }

    return account, nil
}

// 应用事件（更新状态）
func (a *Account) Apply(event Event) {
    switch e := event.(type) {
    case *AccountCreatedEvent:
        a.balance = e.initialBalance
        a.status = AccountStatusActive

    case *MoneyDepositedEvent:
        a.balance = a.balance.Add(e.amount)

    case *MoneyWithdrawnEvent:
        a.balance = a.balance.Subtract(e.amount)

    case *AccountClosedEvent:
        a.status = AccountStatusClosed
    }

    a.version = event.Version()
}

// 业务方法（产生事件）
func (a *Account) Deposit(amount Money) error {
    if a.status != AccountStatusActive {
        return errors.New("account is not active")
    }

    event := &MoneyDepositedEvent{
        accountID:  a.id,
        amount:     amount,
        occurredAt: time.Now(),
        version:    a.version + 1,
    }

    // 1. 应用事件（更新状态）
    a.Apply(event)

    // 2. 记录未提交事件
    a.uncommittedEvents = append(a.uncommittedEvents, event)

    return nil
}

func (a *Account) Withdraw(amount Money) error {
    if a.status != AccountStatusActive {
        return errors.New("account is not active")
    }

    if a.balance.LessThan(amount) {
        return errors.New("insufficient balance")
    }

    event := &MoneyWithdrawnEvent{
        accountID:  a.id,
        amount:     amount,
        occurredAt: time.Now(),
        version:    a.version + 1,
    }

    a.Apply(event)
    a.uncommittedEvents = append(a.uncommittedEvents, event)

    return nil
}

// 获取未提交事件
func (a *Account) GetUncommittedEvents() []Event {
    return a.uncommittedEvents
}

func (a *Account) ClearUncommittedEvents() {
    a.uncommittedEvents = nil
}

type Snapshot struct {
    AggregateID string
    Version     int
    State       interface{}
    CreatedAt   time.Time
}

type SnapshotStore interface {
    SaveSnapshot(snapshot Snapshot) error
    LoadSnapshot(aggregateID string) (*Snapshot, error)
}

// 从快照加载聚合
func LoadAccountWithSnapshot(eventStore EventStore, snapshotStore SnapshotStore, accountID string) (*Account, error) {
    // 1. 加载快照
    snapshot, err := snapshotStore.LoadSnapshot(accountID)

    var account *Account
    var fromVersion int

    if err == nil && snapshot != nil {
        // 从快照恢复
        account = snapshot.State.(*Account)
        fromVersion = snapshot.Version
    } else {
        // 从头开始
        account = &Account{id: accountID}
        fromVersion = 0
    }

    // 2. 加载快照之后的事件
    events, _ := eventStore.LoadEventsSince(accountID, fromVersion)

    // 3. 重放事件
    for _, event := range events {
        account.Apply(event)
    }

    return account, nil
}

// 定期保存快照
func (repo *AccountRepository) SaveWithSnapshot(account *Account) error {
    // 1. 保存事件
    err := repo.eventStore.SaveEvents(
        account.id,
        account.GetUncommittedEvents(),
        account.version,
    )
    if err != nil {
        return err
    }

    // 2. 每100个事件保存一次快照
    if account.version%100 == 0 {
        snapshot := Snapshot{
            AggregateID: account.id,
            Version:     account.version,
            State:       account,
            CreatedAt:   time.Now(),
        }
        repo.snapshotStore.SaveSnapshot(snapshot)
    }

    account.ClearUncommittedEvents()
    return nil
}

// 应用服务
type AccountApplicationService struct {
    eventStore EventStore
    eventBus   EventBus
}

func (app *AccountApplicationService) CreateAccount(cmd CreateAccountCommand) error {
    // 1. 创建聚合
    account := NewAccount(cmd.AccountID, cmd.InitialBalance)

    // 2. 保存事件
    events := account.GetUncommittedEvents()
    err := app.eventStore.SaveEvents(account.id, events, 0)
    if err != nil {
        return err
    }

    // 3. 发布事件（更新读模型）
    for _, event := range events {
        app.eventBus.Publish(event)
    }

    return nil
}

func (app *AccountApplicationService) Deposit(cmd DepositCommand) error {
    // 1. 加载聚合（重放事件）
    account, err := LoadAccount(app.eventStore, cmd.AccountID)
    if err != nil {
        return err
    }

    // 2. 执行业务逻辑
    if err := account.Deposit(cmd.Amount); err != nil {
        return err
    }

    // 3. 保存事件
    events := account.GetUncommittedEvents()
    err = app.eventStore.SaveEvents(account.id, events, account.version-len(events))
    if err != nil {
        return err
    }

    // 4. 发布事件
    for _, event := range events {
        app.eventBus.Publish(event)
    }

    return nil
}

// 读模型
type AccountReadModel struct {
    ID             string
    Balance        float64
    Status         string
    TransactionCount int
    LastTransaction time.Time
}

// 投影（Projection）
type AccountProjection struct {
    db *sql.DB
}

// 监听事件，更新读模型
func (proj *AccountProjection) Handle(event Event) error {
    switch e := event.(type) {
    case *AccountCreatedEvent:
        return proj.handleAccountCreated(e)

    case *MoneyDepositedEvent:
        return proj.handleMoneyDeposited(e)

    case *MoneyWithdrawnEvent:
        return proj.handleMoneyWithdrawn(e)
    }

    return nil
}

func (proj *AccountProjection) handleAccountCreated(event *AccountCreatedEvent) error {
    _, err := proj.db.Exec(`
        INSERT INTO account_read_model (id, balance, status, transaction_count, last_transaction)
        VALUES (?, ?, ?, ?, ?)
    `, event.accountID, event.initialBalance.Amount, "ACTIVE", 0, event.occurredAt)

    return err
}

func (proj *AccountProjection) handleMoneyDeposited(event *MoneyDepositedEvent) error {
    _, err := proj.db.Exec(`
        UPDATE account_read_model
        SET balance = balance + ?,
            transaction_count = transaction_count + 1,
            last_transaction = ?
        WHERE id = ?
    `, event.amount.Amount, event.occurredAt, event.accountID)

    return err
}

// 查询服务
type AccountQueryService struct {
    db *sql.DB
}

func (qs *AccountQueryService) GetAccount(accountID string) (*AccountReadModel, error) {
    var model AccountReadModel

    err := qs.db.QueryRow(`
        SELECT id, balance, status, transaction_count, last_transaction
        FROM account_read_model
        WHERE id = ?
    `, accountID).Scan(
        &model.ID,
        &model.Balance,
        &model.Status,
        &model.TransactionCount,
        &model.LastTransaction,
    )

    return &model, err
}

func (qs *AccountQueryService) GetTopAccounts(limit int) ([]AccountReadModel, error) {
    rows, err := qs.db.Query(`
        SELECT id, balance, status, transaction_count
        FROM account_read_model
        ORDER BY balance DESC
        LIMIT ?
    `, limit)

    if err != nil {
        return nil, err
    }
    defer rows.Close()

    var accounts []AccountReadModel
    for rows.Next() {
        var model AccountReadModel
        rows.Scan(&model.ID, &model.Balance, &model.Status, &model.TransactionCount)
        accounts = append(accounts, model)
    }

    return accounts, nil
}

package main

import (
    "encoding/json"
    "errors"
    "time"
)

// ========== 事件定义 ==========

type Event interface {
    AggregateID() string
    Version() int
    OccurredAt() time.Time
}

type BaseEvent struct {
    aggregateID string
    version     int
    occurredAt  time.Time
}

func (e BaseEvent) AggregateID() string { return e.aggregateID }
func (e BaseEvent) Version() int        { return e.version }
func (e BaseEvent) OccurredAt() time.Time { return e.occurredAt }

type AccountCreatedEvent struct {
    BaseEvent
    InitialBalance float64
}

type MoneyDepositedEvent struct {
    BaseEvent
    Amount float64
}

type MoneyWithdrawnEvent struct {
    BaseEvent
    Amount float64
}

// ========== 聚合根 ==========

type AccountStatus int

const (
    AccountStatusActive AccountStatus = iota
    AccountStatusClosed
)

type Account struct {
    id               string
    balance          float64
    status           AccountStatus
    version          int
    uncommittedEvents []Event
}

func NewAccount(id string, initialBalance float64) *Account {
    account := &Account{
        id:      id,
        balance: 0,
        status:  AccountStatusActive,
        version: 0,
    }

    event := &AccountCreatedEvent{
        BaseEvent: BaseEvent{
            aggregateID: id,
            version:     1,
            occurredAt:  time.Now(),
        },
        InitialBalance: initialBalance,
    }

    account.Apply(event)
    account.uncommittedEvents = append(account.uncommittedEvents, event)

    return account
}

func (a *Account) Apply(event Event) {
    switch e := event.(type) {
    case *AccountCreatedEvent:
        a.balance = e.InitialBalance
        a.status = AccountStatusActive

    case *MoneyDepositedEvent:
        a.balance += e.Amount

    case *MoneyWithdrawnEvent:
        a.balance -= e.Amount
    }

    a.version = event.Version()
}

func (a *Account) Deposit(amount float64) error {
    if amount <= 0 {
        return errors.New("amount must be positive")
    }

    event := &MoneyDepositedEvent{
        BaseEvent: BaseEvent{
            aggregateID: a.id,
            version:     a.version + 1,
            occurredAt:  time.Now(),
        },
        Amount: amount,
    }

    a.Apply(event)
    a.uncommittedEvents = append(a.uncommittedEvents, event)

    return nil
}

func (a *Account) Withdraw(amount float64) error {
    if amount <= 0 {
        return errors.New("amount must be positive")
    }

    if a.balance < amount {
        return errors.New("insufficient balance")
    }

    event := &MoneyWithdrawnEvent{
        BaseEvent: BaseEvent{
            aggregateID: a.id,
            version:     a.version + 1,
            occurredAt:  time.Now(),
        },
        Amount: amount,
    }

    a.Apply(event)
    a.uncommittedEvents = append(a.uncommittedEvents, event)

    return nil
}

func (a *Account) GetUncommittedEvents() []Event {
    return a.uncommittedEvents
}

func (a *Account) ClearEvents() {
    a.uncommittedEvents = nil
}

// ========== 完整应用 ==========

func main() {
    // 1. 创建账户
    account := NewAccount("ACC-001", 1000.0)
    fmt.Printf("Account created: balance=%.2f\n", account.balance)

    // 2. 存款
    account.Deposit(500)
    fmt.Printf("After deposit: balance=%.2f\n", account.balance)

    // 3. 取款
    account.Withdraw(200)
    fmt.Printf("After withdrawal: balance=%.2f\n", account.balance)

    // 4. 查看事件
    fmt.Println("\nEvents:")
    for _, event := range account.GetUncommittedEvents() {
        eventJSON, _ := json.MarshalIndent(event, "", "  ")
        fmt.Println(string(eventJSON))
    }

    // 5. 重放事件（重建状态）
    fmt.Println("\nRebuilding from events:")
    rebuiltAccount := &Account{id: "ACC-001"}
    for _, event := range account.GetUncommittedEvents() {
        rebuiltAccount.Apply(event)
    }
    fmt.Printf("Rebuilt balance: %.2f\n", rebuiltAccount.balance)
}

type CachedAccountRepository struct {
    cache      *Cache
    eventStore EventStore
}

func (repo *CachedAccountRepository) Load(id string) (*Account, error) {
    // 先查缓存
    if cached, ok := repo.cache.Get(id); ok {
        return cached.(*Account), nil
    }

    // 缓存未命中，从事件重建
    account, _ := LoadAccount(repo.eventStore, id)

    // 放入缓存
    repo.cache.Set(id, account, 10*time.Minute)

    return account, nil
}

func (app *AccountApp) Deposit(cmd DepositCommand) error {
    // 1. 保存事件
    account, _ := LoadAccount(app.eventStore, cmd.AccountID)
    account.Deposit(cmd.Amount)
    app.eventStore.SaveEvents(account.id, account.GetUncommittedEvents(), account.version-1)

    // 2. 同步更新读模型
    for _, event := range account.GetUncommittedEvents() {
        app.projection.Handle(event)  // 同步处理
    }

    return nil
}

func (qs *AccountQueryService) GetAccount(accountID string, minVersion int) (*AccountReadModel, error) {
    for i := 0; i < 10; i++ {
        model, _ := qs.loadAccount(accountID)

        if model.Version >= minVersion {
            return model, nil
        }

        time.Sleep(100 * time.Millisecond)  // 等待投影完成
    }

    return nil, errors.New("timeout waiting for projection")
}

// V1事件
type MoneyDepositedEventV1 struct {
    Amount float64
}

// V2事件（增加货币字段）
type MoneyDepositedEventV2 struct {
    Amount   float64
    Currency string
}

// 事件升级器
type EventUpgrader struct{}

func (eu *EventUpgrader) Upgrade(event Event) Event {
    switch e := event.(type) {
    case *MoneyDepositedEventV1:
        return &MoneyDepositedEventV2{
            Amount:   e.Amount,
            Currency: "CNY",  // 默认值
        }
    }

    return event
}

// 加载事件时自动升级
func (es *EventStore) LoadEvents(aggregateID string) ([]Event, error) {
    rawEvents, _ := es.loadRawEvents(aggregateID)

    upgrader := &EventUpgrader{}
    upgradedEvents := make([]Event, 0)

    for _, event := range rawEvents {
        upgraded := upgrader.Upgrade(event)
        upgradedEvents = append(upgradedEvents, upgraded)
    }

    return upgradedEvents, nil
}

type CachedRepository struct {
    cache      *LRUCache
    eventStore EventStore
}

func (repo *CachedRepository) Load(id string) (*Account, error) {
    // 先查缓存
    if cached, ok := repo.cache.Get(id); ok {
        return cached.(*Account), nil
    }

    // 从事件重建
    account, _ := LoadAccount(repo.eventStore, id)

    // 缓存
    repo.cache.Set(id, account)

    return account, nil
}

type TransferSaga struct {
    eventBus EventBus
}

// 1. 扣款
func (saga *TransferSaga) DebitAccount(fromAccountID string, amount float64) error {
    // 发布扣款命令
    saga.eventBus.Publish(DebitAccountCommand{
        AccountID: fromAccountID,
        Amount:    amount,
    })

    return nil
}

// 2. 监听扣款成功事件，执行入账
func (saga *TransferSaga) OnMoneyDebited(event MoneyDebitedEvent) {
    // 发布入账命令
    saga.eventBus.Publish(CreditAccountCommand{
        AccountID: event.ToAccountID,
        Amount:    event.Amount,
    })
}

// 3. 监听入账失败事件，回滚扣款
func (saga *TransferSaga) OnCreditFailed(event CreditFailedEvent) {
    // 发布补偿命令（回滚）
    saga.eventBus.Publish(RefundAccountCommand{
        AccountID: event.FromAccountID,
        Amount:    event.Amount,
    })
}

type TransferProcessManager struct {
    state  TransferState
    eventBus EventBus
}

func (pm *TransferProcessManager) Handle(event Event) {
    switch pm.state {
    case StateInitiated:
        if _, ok := event.(*MoneyDebitedEvent); ok {
            pm.state = StateDebited
            pm.creditToAccount()
        }

    case StateDebited:
        if _, ok := event.(*MoneyCreditedEvent); ok {
            pm.state = StateCompleted
        } else if _, ok := event.(*CreditFailedEvent); ok {
            pm.state = StateFailed
            pm.refundFromAccount()
        }
    }
}