1
package eventsourcing
2

3
import (
4
  "fmt"
5
  "time"
6
)
7

8
// ---- 事件定义 ----
9

10
type EventType string
11

12
const (
13
  EventAccountOpened  EventType = "AccountOpened"
14
  EventMoneyDeposited EventType = "MoneyDeposited"
15
  EventMoneyWithdrawn EventType = "MoneyWithdrawn"
16
)
17

18
type Event struct {
19
  Type      EventType
20
  Timestamp time.Time
21
  Data      map[string]interface{}
22
}
23

24
// ---- 聚合：银行账户 ----
25

26
type BankAccount struct {
27
  ID      string
28
  Owner   string
29
  Balance float64
30
  Version int // 用于乐观并发控制
31
}
32

33
// ApplyEvent 将单个事件应用到聚合上
34
func (a *BankAccount) ApplyEvent(event Event) {
35
  switch event.Type {
36
  case EventAccountOpened:
37
    a.ID = event.Data["id"].(string)
38
    a.Owner = event.Data["owner"].(string)
39
    a.Balance = 0
40
  case EventMoneyDeposited:
41
    a.Balance += event.Data["amount"].(float64)
42
  case EventMoneyWithdrawn:
43
    a.Balance -= event.Data["amount"].(float64)
44
  }
45
  a.Version++
46
}
47

48
// ReplayFromEvents 从事件流重建聚合状态
49
func ReplayFromEvents(events []Event) *BankAccount {
50
  account := &BankAccount{}
51
  for _, event := range events {
52
    account.ApplyEvent(event)
53
  }
54
  return account
55
}
56

57
// ---- 事件存储 ----
58

59
type EventStore struct {
60
  streams map[string][]Event // 聚合 ID → 事件流
61
}
62

63
func NewEventStore() *EventStore {
64
  return &EventStore{
65
    streams: make(map[string][]Event),
66
  }
67
}
68

69
// Append 向指定聚合的事件流追加事件
70
func (s *EventStore) Append(aggregateID string, events []Event) error {
71
  // 实际实现应检查乐观并发版本号
72
  s.streams[aggregateID] = append(s.streams[aggregateID], events...)
73
  return nil
74
}
75

76
// LoadEvents 加载指定聚合的全部事件
77
func (s *EventStore) LoadEvents(aggregateID string) ([]Event, error) {
78
  events, ok := s.streams[aggregateID]
79
  if !ok {
80
    return nil, fmt.Errorf("aggregate %s not found", aggregateID)
81
  }
82
  return events, nil
83
}
84

85
// ---- 快照机制 ----
86

87
type Snapshot struct {
88
  AggregateID string
89
  Version     int
90
  State       *BankAccount
91
}
92

93
type SnapshotStore struct {
94
  snapshots map[string]Snapshot
95
}
96

97
func NewSnapshotStore() *SnapshotStore {
98
  return &SnapshotStore{
99
    snapshots: make(map[string]Snapshot),
100
  }
101
}
102

103
// SaveSnapshot 保存聚合在某个版本的快照
104
func (ss *SnapshotStore) SaveSnapshot(snap Snapshot) {
105
  ss.snapshots[snap.AggregateID] = snap
106
}
107

108
// LoadSnapshot 加载最近的快照
109
func (ss *SnapshotStore) LoadSnapshot(aggregateID string) (Snapshot, bool) {
110
  snap, ok := ss.snapshots[aggregateID]
111
  return snap, ok
112
}
113

114
// LoadWithSnapshot 用快照 + 增量事件重建状态
115
func LoadWithSnapshot(
116
  snapStore *SnapshotStore,
117
  eventStore *EventStore,
118
  aggregateID string,
119
) (*BankAccount, error) {
120
  // 先尝试加载快照
121
  if snap, ok := snapStore.LoadSnapshot(aggregateID); ok {
122
    account := snap.State
123
    // 只回放快照之后的事件
124
    allEvents, err := eventStore.LoadEvents(aggregateID)
125
    if err != nil {
126
      return nil, err
127
    }
128
    for _, event := range allEvents[snap.Version:] {
129
      account.ApplyEvent(event)
130
    }
131
    return account, nil
132
  }
133
  // 没有快照，从头回放
134
  events, err := eventStore.LoadEvents(aggregateID)
135
  if err != nil {
136
    return nil, err
137
  }
138
  return ReplayFromEvents(events), nil
139
}

1
// ---- 事件定义 ----
2

3
type EventType = "TaskCreated" | "TaskCompleted" | "TaskReopened";
4

5
interface Event {
6
  type: EventType;
7
  timestamp: Date;
8
  data: Record<string, unknown>;
9
}
10

11
// ---- 聚合：任务 ----
12

13
class Task {
14
  id = "";
15
  title = "";
16
  completed = false;
17
  version = 0;
18

19
  // 将事件应用到自身状态
20
  applyEvent(event: Event): void {
21
    switch (event.type) {
22
      case "TaskCreated":
23
        this.id = event.data.id as string;
24
        this.title = event.data.title as string;
25
        this.completed = false;
26
        break;
27
      case "TaskCompleted":
28
        this.completed = true;
29
        break;
30
      case "TaskReopened":
31
        this.completed = false;
32
        break;
33
    }
34
    this.version++;
35
  }
36
}
37

38
// ---- 事件存储 ----
39

40
class EventStore {
41
  private streams = new Map<string, Event[]>();
42

43
  // 追加事件到指定聚合的流
44
  append(aggregateId: string, events: Event[]): void {
45
    const existing = this.streams.get(aggregateId) ?? [];
46
    this.streams.set(aggregateId, [...existing, ...events]);
47
  }
48

49
  // 加载指定聚合的全部事件
50
  loadEvents(aggregateId: string): Event[] {
51
    return this.streams.get(aggregateId) ?? [];
52
  }
53
}
54

55
// ---- 事件回放与状态推导 ----
56

57
function replayFromEvents(events: Event[]): Task {
58
  const task = new Task();
59
  for (const event of events) {
60
    task.applyEvent(event);
61
  }
62
  return task;
63
}
64

65
// ---- 使用示例 ----
66

67
const store = new EventStore();
68
const taskId = "task-001";
69

70
// 记录任务生命周期的事件
71
store.append(taskId, [
72
  {
73
    type: "TaskCreated",
74
    timestamp: new Date("2026-01-10"),
75
    data: { id: taskId, title: "实现事件溯源" },
76
  },
77
  {
78
    type: "TaskCompleted",
79
    timestamp: new Date("2026-01-15"),
80
    data: {},
81
  },
82
  {
83
    type: "TaskReopened",
84
    timestamp: new Date("2026-01-16"),
85
    data: { reason: "发现回归 bug" },
86
  },
87
]);
88

89
// 从事件流重建状态——任何时候都可以得到一致的状态
90
const task = replayFromEvents(store.loadEvents(taskId));
91
console.log(task.title);     // "实现事件溯源"
92
console.log(task.completed); // false（被重新打开了）
93
console.log(task.version);   // 3（经历了 3 个事件）