// 从事件流重建聚合状态
public class BankAccount {
    private AccountId id;
    private Money balance;
    private AccountStatus status;

    // 从事件流重建
    public static BankAccount reconstitute(List<DomainEvent> events) {
        BankAccount account = new BankAccount();
        for (DomainEvent event : events) {
            account.apply(event);
        }
        return account;
    }

    private void apply(DomainEvent event) {
        if (event instanceof AccountOpenedEvent e) {
            this.id = e.accountId();
            this.balance = Money.ZERO;
            this.status = AccountStatus.ACTIVE;
        } else if (event instanceof MoneyDepositedEvent e) {
            this.balance = this.balance.add(e.amount());
        } else if (event instanceof MoneyWithdrawnEvent e) {
            this.balance = this.balance.subtract(e.amount());
        } else if (event instanceof AccountClosedEvent e) {
            this.status = AccountStatus.CLOSED;
        }
    }
}

-- 事件存储的核心表
CREATE TABLE event_store (
    -- 全局唯一的事件 ID
    event_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),

    -- 事件流标识（通常是聚合类型+聚合ID）
    stream_name VARCHAR(500) NOT NULL,

    -- 事件在流中的版本号（从 1 开始，单调递增）
    stream_version BIGINT NOT NULL,

    -- 事件类型（用于反序列化路由）
    event_type VARCHAR(200) NOT NULL,

    -- 事件负载（JSON 格式）
    payload JSONB NOT NULL,

    -- 事件元数据（因果关系、用户信息等）
    metadata JSONB NOT NULL DEFAULT '{}',

    -- 事件发生时间
    occurred_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),

    -- 事件写入时间（服务器时间，用于排序）
    recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),

    -- 保证同一流内版本号唯一（乐观锁）
    UNIQUE (stream_name, stream_version)
);

-- 按流名称和版本号建索引，加速聚合加载
CREATE INDEX idx_event_store_stream
    ON event_store (stream_name, stream_version);

-- 按事件类型建索引，支持按类型订阅
CREATE INDEX idx_event_store_type
    ON event_store (event_type, recorded_at);

-- 按时间建索引，支持全局事件流订阅
CREATE INDEX idx_event_store_recorded
    ON event_store (recorded_at);

{
  "event_id": "e7a1b2c3-d4e5-6789-abcd-ef0123456789",
  "stream_name": "BankAccount-acc_001",
  "stream_version": 3,
  "event_type": "MoneyDeposited",
  "payload": {
    "account_id": "acc_001",
    "amount": "500.00",
    "currency": "CNY",
    "description": "工资入账"
  },
  "metadata": {
    "user_id": "user_123",
    "correlation_id": "txn_456",
    "causation_id": "cmd_789",
    "ip_address": "192.168.1.100"
  },
  "occurred_at": "2025-04-13T10:30:00Z",
  "recorded_at": "2025-04-13T10:30:00.123Z"
}

type EventStore struct {
    db *sql.DB
}

type StoredEvent struct {
    EventID       string
    StreamName    string
    StreamVersion int64
    EventType     string
    Payload       json.RawMessage
    Metadata      json.RawMessage
    OccurredAt    time.Time
    RecordedAt    time.Time
}

// 追加事件到事件流（带乐观锁）
func (s *EventStore) Append(ctx context.Context, streamName string,
    expectedVersion int64, events []StoredEvent) error {

    tx, err := s.db.BeginTx(ctx, nil)
    if err != nil {
        return fmt.Errorf("开始事务失败: %w", err)
    }
    defer tx.Rollback()

    // 验证当前版本号（乐观锁）
    var currentVersion int64
    err = tx.QueryRowContext(ctx,
        `SELECT COALESCE(MAX(stream_version), 0)
         FROM event_store WHERE stream_name = $1`,
        streamName).Scan(&currentVersion)
    if err != nil {
        return fmt.Errorf("查询当前版本失败: %w", err)
    }

    if currentVersion != expectedVersion {
        return fmt.Errorf(
            "并发冲突: 期望版本 %d, 当前版本 %d",
            expectedVersion, currentVersion)
    }

    // 插入事件
    stmt, err := tx.PrepareContext(ctx,
        `INSERT INTO event_store
         (event_id, stream_name, stream_version, event_type,
          payload, metadata, occurred_at)
         VALUES ($1, $2, $3, $4, $5, $6, $7)`)
    if err != nil {
        return fmt.Errorf("准备语句失败: %w", err)
    }
    defer stmt.Close()

    for i, event := range events {
        version := expectedVersion + int64(i) + 1
        _, err = stmt.ExecContext(ctx,
            event.EventID, streamName, version,
            event.EventType, event.Payload,
            event.Metadata, event.OccurredAt)
        if err != nil {
            return fmt.Errorf("插入事件失败: %w", err)
        }
    }

    return tx.Commit()
}

// 加载事件流
func (s *EventStore) LoadStream(ctx context.Context,
    streamName string) ([]StoredEvent, error) {

    rows, err := s.db.QueryContext(ctx,
        `SELECT event_id, stream_name, stream_version, event_type,
                payload, metadata, occurred_at, recorded_at
         FROM event_store
         WHERE stream_name = $1
         ORDER BY stream_version ASC`,
        streamName)
    if err != nil {
        return nil, fmt.Errorf("查询事件流失败: %w", err)
    }
    defer rows.Close()

    var events []StoredEvent
    for rows.Next() {
        var e StoredEvent
        if err := rows.Scan(&e.EventID, &e.StreamName,
            &e.StreamVersion, &e.EventType,
            &e.Payload, &e.Metadata,
            &e.OccurredAt, &e.RecordedAt); err != nil {
            return nil, fmt.Errorf("扫描事件失败: %w", err)
        }
        events = append(events, e)
    }
    return events, nil
}

// 异步投影器
type OrderListProjector struct {
    db        *sql.DB
    store     *EventStore
    lastPos   int64 // 上次处理的位置
}

func (p *OrderListProjector) Run(ctx context.Context) error {
    ticker := time.NewTicker(100 * time.Millisecond)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-ticker.C:
            if err := p.processNewEvents(ctx); err != nil {
                log.Printf("投影器处理失败: %v", err)
            }
        }
    }
}

func (p *OrderListProjector) processNewEvents(ctx context.Context) error {
    events, err := p.store.LoadSince(ctx, p.lastPos, 100)
    if err != nil {
        return err
    }

    for _, event := range events {
        if err := p.handleEvent(ctx, event); err != nil {
            return fmt.Errorf("处理事件 %s 失败: %w",
                event.EventID, err)
        }
        p.lastPos = event.RecordedAt.UnixNano()
    }
    return nil
}

func (p *OrderListProjector) handleEvent(ctx context.Context,
    event StoredEvent) error {

    switch event.EventType {
    case "OrderPlaced":
        var payload OrderPlacedPayload
        if err := json.Unmarshal(event.Payload, &payload); err != nil {
            return err
        }
        _, err := p.db.ExecContext(ctx,
            `INSERT INTO order_list_view
             (order_id, customer_id, total_amount, status, created_at)
             VALUES ($1, $2, $3, $4, $5)`,
            payload.OrderID, payload.CustomerID,
            payload.TotalAmount, "待支付", event.OccurredAt)
        return err

    case "OrderConfirmed":
        var payload OrderConfirmedPayload
        if err := json.Unmarshal(event.Payload, &payload); err != nil {
            return err
        }
        _, err := p.db.ExecContext(ctx,
            `UPDATE order_list_view SET status = $1
             WHERE order_id = $2`,
            "已确认", payload.OrderID)
        return err

    default:
        return nil // 忽略不关心的事件
    }
}

func (p *OrderListProjector) Rebuild(ctx context.Context) error {
    // 1. 清空现有读模型
    _, err := p.db.ExecContext(ctx, "TRUNCATE TABLE order_list_view")
    if err != nil {
        return fmt.Errorf("清空读模型失败: %w", err)
    }

    // 2. 重置位置
    p.lastPos = 0

    // 3. 重新处理所有事件
    for {
        events, err := p.store.LoadSince(ctx, p.lastPos, 1000)
        if err != nil {
            return err
        }
        if len(events) == 0 {
            break
        }
        for _, event := range events {
            if err := p.handleEvent(ctx, event); err != nil {
                return err
            }
            p.lastPos = event.RecordedAt.UnixNano()
        }
    }

    log.Println("投影重建完成")
    return nil
}

CREATE TABLE snapshots (
    stream_name VARCHAR(500) NOT NULL,
    stream_version BIGINT NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (stream_name, stream_version)
);

-- 每个流只保留最新的快照（可选）
CREATE INDEX idx_snapshots_latest
    ON snapshots (stream_name, stream_version DESC);

func (r *EventSourcedRepository) Load(ctx context.Context,
    streamName string) (*BankAccount, error) {

    // 1. 尝试加载最新快照
    snapshot, snapshotVersion, err := r.loadLatestSnapshot(ctx, streamName)
    if err != nil && !errors.Is(err, ErrSnapshotNotFound) {
        return nil, err
    }

    // 2. 从快照版本之后加载事件
    fromVersion := int64(0)
    if snapshot != nil {
        fromVersion = snapshotVersion
    }

    events, err := r.store.LoadStreamFrom(ctx, streamName, fromVersion)
    if err != nil {
        return nil, err
    }

    // 3. 重建聚合
    var account *BankAccount
    if snapshot != nil {
        account = snapshot // 从快照开始
    } else {
        account = &BankAccount{} // 从空白开始
    }

    for _, event := range events {
        account.apply(event)
    }

    return account, nil
}

func (r *EventSourcedRepository) Save(ctx context.Context,
    account *BankAccount) error {

    // 保存事件
    err := r.store.Append(ctx, account.streamName(),
        account.version(), account.uncommittedEvents())
    if err != nil {
        return err
    }

    // 检查是否需要创建快照
    if account.version() > 0 && account.version()%100 == 0 {
        if err := r.saveSnapshot(ctx, account); err != nil {
            // 快照保存失败不应阻塞主流程
            log.Printf("快照保存失败: %v", err)
        }
    }

    account.markCommitted()
    return nil
}

public interface EventUpcaster {
    boolean canUpcast(String eventType, int fromVersion);
    JsonNode upcast(JsonNode oldPayload, int fromVersion);
}

public class MoneyDepositedUpcaster implements EventUpcaster {

    @Override
    public boolean canUpcast(String eventType, int fromVersion) {
        return "MoneyDeposited".equals(eventType) && fromVersion < 3;
    }

    @Override
    public JsonNode upcast(JsonNode payload, int fromVersion) {
        ObjectNode node = (ObjectNode) payload.deepCopy();

        if (fromVersion < 2) {
            // V1 → V2: 添加默认币种
            node.put("currency", "CNY");
        }
        if (fromVersion < 3) {
            // V2 → V3: 添加默认来源
            node.put("source", "UNKNOWN");
        }

        return node;
    }
}

type Upcaster func(payload json.RawMessage, fromVersion int) (json.RawMessage, error)

var upcasters = map[string]Upcaster{
    "MoneyDeposited": upcastMoneyDeposited,
}

func upcastMoneyDeposited(payload json.RawMessage, fromVersion int) (json.RawMessage, error) {
    var data map[string]interface{}
    if err := json.Unmarshal(payload, &data); err != nil {
        return nil, err
    }

    if fromVersion < 2 {
        if _, ok := data["currency"]; !ok {
            data["currency"] = "CNY"
        }
    }
    if fromVersion < 3 {
        if _, ok := data["source"]; !ok {
            data["source"] = "UNKNOWN"
        }
    }

    return json.Marshal(data)
}

// 在加载事件时应用 upcasting
func loadAndUpcast(stored StoredEvent) (StoredEvent, error) {
    upcaster, exists := upcasters[stored.EventType]
    if !exists {
        return stored, nil
    }

    // 从元数据获取事件版本
    version := extractSchemaVersion(stored.Metadata)
    if version < currentSchemaVersion(stored.EventType) {
        newPayload, err := upcaster(stored.Payload, version)
        if err != nil {
            return stored, fmt.Errorf("upcasting 失败: %w", err)
        }
        stored.Payload = newPayload
    }
    return stored, nil
}

// 策略 1：写后读一致性
func (h *PlaceOrderHandler) Handle(ctx context.Context,
    cmd PlaceOrderCommand) (*OrderResponse, error) {

    // 写入
    order, err := h.orderService.PlaceOrder(ctx, cmd)
    if err != nil {
        return nil, err
    }

    // 直接从写模型构建响应，不查读模型
    return &OrderResponse{
        OrderID:   order.ID().String(),
        Status:    order.Status().String(),
        Total:     order.Total().String(),
        CreatedAt: order.CreatedAt(),
    }, nil
}

// 策略 3：版本号校验
func (h *GetOrderHandler) Handle(ctx context.Context,
    query GetOrderQuery) (*OrderListView, error) {

    view, err := h.readStore.FindByID(ctx, query.OrderID)
    if err != nil {
        return nil, err
    }

    // 如果客户端指定了最小版本号
    if query.MinVersion > 0 && view.Version < query.MinVersion {
        return nil, ErrStaleData // 客户端可以重试
    }

    return view, nil
}

@Test
void shouldWithdrawMoney() {
    // Given: 账户有 1000 元余额
    given(
        new AccountOpenedEvent("acc_001"),
        new MoneyDepositedEvent("acc_001", money(1000, "CNY"))
    )
    // When: 取款 300 元
    .when(new WithdrawMoneyCommand("acc_001", money(300, "CNY")))
    // Then: 产生取款事件
    .then(
        new MoneyWithdrawnEvent("acc_001", money(300, "CNY"))
    );
}

@Test
void shouldRejectWithdrawalWhenInsufficientBalance() {
    given(
        new AccountOpenedEvent("acc_001"),
        new MoneyDepositedEvent("acc_001", money(100, "CNY"))
    )
    .when(new WithdrawMoneyCommand("acc_001", money(200, "CNY")))
    .thenThrows(InsufficientBalanceException.class);
}

func TestOrderListProjector_OrderPlaced(t *testing.T) {
    db := setupTestDB(t)
    projector := NewOrderListProjector(db)

    event := StoredEvent{
        EventType: "OrderPlaced",
        Payload: mustJSON(map[string]interface{}{
            "order_id":     "ord_001",
            "customer_id":  "cust_001",
            "total_amount": "99.99",
        }),
        OccurredAt: time.Now(),
    }

    err := projector.handleEvent(context.Background(), event)
    require.NoError(t, err)

    // 验证读模型
    view, err := queryOrderView(db, "ord_001")
    require.NoError(t, err)
    assert.Equal(t, "待支付", view.Status)
    assert.Equal(t, "99.99", view.TotalAmount)
}

// 使用 EventStoreDB 的 Go 客户端
import "github.com/EventStore/EventStore-Client-Go/v4/esdb"

func appendEvents(client *esdb.Client, streamName string,
    events []esdb.EventData) error {

    _, err := client.AppendToStream(
        context.Background(),
        streamName,
        esdb.AppendToStreamOptions{},
        events...,
    )
    return err
}

func readStream(client *esdb.Client,
    streamName string) ([]*esdb.ResolvedEvent, error) {

    stream, err := client.ReadStream(
        context.Background(),
        streamName,
        esdb.ReadStreamOptions{
            Direction: esdb.Forwards,
            From:      esdb.Start{},
        },
        math.MaxUint64,
    )
    if err != nil {
        return nil, err
    }
    defer stream.Close()

    var events []*esdb.ResolvedEvent
    for {
        event, err := stream.Recv()
        if errors.Is(err, io.EOF) {
            break
        }
        if err != nil {
            return nil, err
        }
        events = append(events, event)
    }
    return events, nil
}