Go's combination of strong typing, excellent concurrency primitives, and low-overhead runtime makes it an exceptional fit for CQRS and Event Sourcing systems. The language's simplicity forces clarity in aggregate design, and goroutines provide natural building blocks for projection consumers. This guide covers a production-ready CQRS/ES implementation in Go, from event store design through projection management.
Core Architecture
A CQRS/ES system in Go decomposes into four primary components: the command handler (validates and processes commands), the event store (persists domain events), the aggregate (encapsulates domain logic), and the projection engine (builds read models from events).
1┌─────────────┐ ┌──────────────┐ ┌─────────────┐
2│ Command │────▶│ Aggregate │────▶│ Event Store │
3│ Handler │ │ (Domain) │ │ (Append) │
4└─────────────┘ └──────────────┘ └──────┬──────┘
5 │
6 ┌───────────▼───────────┐
7 │ Projection Engine │
8 │ (Async Consumers) │
9 └───────────┬───────────┘
10 │
11 ┌───────────▼───────────┐
12 │ Read Models │
13 │ (Query-Optimized) │
14 └───────────────────────┘
15
Defining Domain Events
Events are immutable records of state changes. In Go, leverage interfaces and struct embedding for a clean event hierarchy.
go
1package domain
2
3import "time"
4
5// Event is the base interface all domain events implement
6type Event interface {
7 EventType() string
8 AggregateID() string
9 OccurredAt() time.Time
10}
11
12// BaseEvent provides common fields for all events
13type BaseEvent struct {
14 ID string `json:"id"`
15 AggregateId string `json:"aggregate_id"`
16 Type string `json:"event_type"`
17 Version int `json:"version"`
18 Timestamp time.Time `json:"timestamp"`
19 CorrelationID string `json:"correlation_id"`
20 CausationID string `json:"causation_id"`
21}
22
23func (e BaseEvent) EventType() string { return e.Type }
24func (e BaseEvent) AggregateID() string { return e.AggregateId }
25func (e BaseEvent) OccurredAt() time.Time { return e.Timestamp }
26
27// Domain-specific events
28type OrderPlaced struct {
29 BaseEvent
30 CustomerID string `json:"customer_id"`
31 LineItems []LineItem `json:"line_items"`
32 TotalAmount int64 `json:"total_amount"` // cents
33 Currency string `json:"currency"`
34 ShippingAddr Address `json:"shipping_address"`
35}
36
37type OrderConfirmed struct {
38 BaseEvent
39 ConfirmedBy string `json:"confirmed_by"`
40 ConfirmedAt time.Time `json:"confirmed_at"`
41}
42
43type OrderCancelled struct {
44 BaseEvent
45 Reason string `json:"reason"`
46 CancelledBy string `json:"cancelled_by"`
47}
48
49type LineItem struct {
50 ProductID string `json:"product_id"`
51 Quantity int `json:"quantity"`
52 UnitPrice int64 `json:"unit_price"`
53}
54
55type Address struct {
56 Street string `json:"street"`
57 City string `json:"city"`
58 Country string `json:"country"`
59 Zip string `json:"zip"`
60}
61
Use integer amounts in cents to avoid floating-point precision issues in financial calculations.
Building the Event Store
The event store provides append-only persistence with optimistic concurrency control. PostgreSQL with JSONB columns is a proven choice for Go-based event stores.
go
1package eventstore
2
3import (
4 "context"
5 "database/sql"
6 "encoding/json"
7 "fmt"
8 "time"
9
10 "github.com/lib/pq"
11)
12
13type PostgresEventStore struct {
14 db *sql.DB
15}
16
17type StoredEvent struct {
18 ID int64 `json:"id"`
19 AggregateID string `json:"aggregate_id"`
20 AggregateType string `json:"aggregate_type"`
21 EventType string `json:"event_type"`
22 Version int `json:"version"`
23 Payload json.RawMessage `json:"payload"`
24 Metadata json.RawMessage `json:"metadata"`
25 CreatedAt time.Time `json:"created_at"`
26}
27
28func NewPostgresEventStore(db *sql.DB) *PostgresEventStore {
29 return &PostgresEventStore{db: db}
30}
31
32func (s *PostgresEventStore) Append(
33 ctx context.Context,
34 aggregateID string,
35 aggregateType string,
36 expectedVersion int,
37 events []StoredEvent,
38) error {
39 tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
40 if err != nil {
41 return fmt.Errorf("begin tx: %w", err)
42 }
43 defer tx.Rollback()
44
45 // Optimistic concurrency check
46 var currentVersion int
47 err = tx.QueryRowContext(ctx,
48 `SELECT COALESCE(MAX(version), 0) FROM events
49 WHERE aggregate_id = $1`,
50 aggregateID,
51 ).Scan(¤tVersion)
52 if err != nil {
53 return fmt.Errorf("check version: %w", err)
54 }
55
56 if currentVersion != expectedVersion {
57 return fmt.Errorf("concurrency conflict: expected version %d, got %d",
58 expectedVersion, currentVersion)
59 }
60
61 stmt, err := tx.PrepareContext(ctx,
62 `INSERT INTO events (aggregate_id, aggregate_type, event_type, version, payload, metadata, created_at)
63 VALUES ($1, $2, $3, $4, $5, $6, $7)`)
64 if err != nil {
65 return fmt.Errorf("prepare: %w", err)
66 }
67 defer stmt.Close()
68
69 for i, event := range events {
70 _, err = stmt.ExecContext(ctx,
71 aggregateID,
72 aggregateType,
73 event.EventType,
74 expectedVersion+i+1,
75 event.Payload,
76 event.Metadata,
77 time.Now().UTC(),
78 )
79 if err != nil {
80 if pqErr, ok := err.(*pq.Error); ok && pqErr.Code == "23505" {
81 return fmt.Errorf("concurrency conflict on insert")
82 }
83 return fmt.Errorf("insert event: %w", err)
84 }
85 }
86
87 return tx.Commit()
88}
89
90func (s *PostgresEventStore) Load(
91 ctx context.Context,
92 aggregateID string,
93 afterVersion int,
94) ([]StoredEvent, error) {
95 rows, err := s.db.QueryContext(ctx,
96 `SELECT id, aggregate_id, aggregate_type, event_type, version, payload, metadata, created_at
97 FROM events
98 WHERE aggregate_id = $1 AND version > $2
99 ORDER BY version ASC`,
100 aggregateID, afterVersion,
101 )
102 if err != nil {
103 return nil, fmt.Errorf("query events: %w", err)
104 }
105 defer rows.Close()
106
107 var events []StoredEvent
108 for rows.Next() {
109 var e StoredEvent
110 err := rows.Scan(&e.ID, &e.AggregateID, &e.AggregateType,
111 &e.EventType, &e.Version, &e.Payload, &e.Metadata, &e.CreatedAt)
112 if err != nil {
113 return nil, fmt.Errorf("scan event: %w", err)
114 }
115 events = append(events, e)
116 }
117
118 return events, rows.Err()
119}
120
121func (s *PostgresEventStore) ReadAll(
122 ctx context.Context,
123 afterPosition int64,
124 batchSize int,
125) ([]StoredEvent, error) {
126 rows, err := s.db.QueryContext(ctx,
127 `SELECT id, aggregate_id, aggregate_type, event_type, version, payload, metadata, created_at
128 FROM events
129 WHERE id > $1
130 ORDER BY id ASC
131 LIMIT $2`,
132 afterPosition, batchSize,
133 )
134 if err != nil {
135 return nil, fmt.Errorf("query all events: %w", err)
136 }
137 defer rows.Close()
138
139 var events []StoredEvent
140 for rows.Next() {
141 var e StoredEvent
142 err := rows.Scan(&e.ID, &e.AggregateID, &e.AggregateType,
143 &e.EventType, &e.Version, &e.Payload, &e.Metadata, &e.CreatedAt)
144 if err != nil {
145 return nil, fmt.Errorf("scan event: %w", err)
146 }
147 events = append(events, e)
148 }
149
150 return events, rows.Err()
151}
152
The PostgreSQL schema:
sql
1CREATE TABLE events (
2 id BIGSERIAL PRIMARY KEY,
3 aggregate_id TEXT NOT NULL,
4 aggregate_type TEXT NOT NULL,
5 event_type TEXT NOT NULL,
6 version INT NOT NULL,
7 payload JSONB NOT NULL,
8 metadata JSONB NOT NULL DEFAULT '{}',
9 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
10 UNIQUE (aggregate_id, version)
11);
12
13CREATE INDEX idx_events_aggregate ON events (aggregate_id, version);
14CREATE INDEX idx_events_type ON events (event_type);
15CREATE INDEX idx_events_created ON events (created_at);
16
Implementing the Aggregate
Aggregates encapsulate domain logic, validate commands, and emit events. The pattern in Go uses method-based command handling and a state evolution function.
go
1package domain
2
3import (
4 "encoding/json"
5 "fmt"
6 "time"
7
8 "github.com/google/uuid"
9 "myapp/eventstore"
10)
11
12type OrderStatus string
13
14const (
15 OrderDraft OrderStatus = "draft"
16 OrderPlaced OrderStatus = "placed"
17 OrderConfirmed OrderStatus = "confirmed"
18 OrderCancelled OrderStatus = "cancelled"
19)
20
21type OrderState struct {
22 ID string
23 Status OrderStatus
24 CustomerID string
25 LineItems []LineItem
26 TotalAmount int64
27 Currency string
28 Version int
29}
30
31type OrderAggregate struct {
32 state OrderState
33 uncommittedEvents []eventstore.StoredEvent
34}
35
36func NewOrderAggregate(id string) *OrderAggregate {
37 return &OrderAggregate{
38 state: OrderState{ID: id, Status: OrderDraft},
39 }
40}
41
42// Rehydrate rebuilds aggregate state from stored events
43func (a *OrderAggregate) Rehydrate(events []eventstore.StoredEvent) error {
44 for _, event := range events {
45 if err := a.apply(event, false); err != nil {
46 return err
47 }
48 }
49 return nil
50}
51
52// PlaceOrder handles the PlaceOrder command
53func (a *OrderAggregate) PlaceOrder(customerID string, items []LineItem, currency string) error {
54 if a.state.Status != OrderDraft {
55 return fmt.Errorf("cannot place order in status %s", a.state.Status)
56 }
57 if len(items) == 0 {
58 return fmt.Errorf("order must contain at least one item")
59 }
60
61 var total int64
62 for _, item := range items {
63 total += item.UnitPrice * int64(item.Quantity)
64 }
65
66 payload, _ := json.Marshal(OrderPlaced{
67 BaseEvent: BaseEvent{
68 AggregateId: a.state.ID,
69 Type: "OrderPlaced",
70 },
71 CustomerID: customerID,
72 LineItems: items,
73 TotalAmount: total,
74 Currency: currency,
75 })
76
77 event := eventstore.StoredEvent{
78 AggregateID: a.state.ID,
79 EventType: "OrderPlaced",
80 Payload: payload,
81 }
82
83 return a.apply(event, true)
84}
85
86// Confirm handles the ConfirmOrder command
87func (a *OrderAggregate) Confirm(confirmedBy string) error {
88 if a.state.Status != OrderPlaced {
89 return fmt.Errorf("cannot confirm order in status %s", a.state.Status)
90 }
91
92 payload, _ := json.Marshal(OrderConfirmed{
93 BaseEvent: BaseEvent{
94 AggregateId: a.state.ID,
95 Type: "OrderConfirmed",
96 },
97 ConfirmedBy: confirmedBy,
98 ConfirmedAt: time.Now().UTC(),
99 })
100
101 event := eventstore.StoredEvent{
102 AggregateID: a.state.ID,
103 EventType: "OrderConfirmed",
104 Payload: payload,
105 }
106
107 return a.apply(event, true)
108}
109
110// Cancel handles the CancelOrder command
111func (a *OrderAggregate) Cancel(reason, cancelledBy string) error {
112 if a.state.Status == OrderCancelled {
113 return fmt.Errorf("order already cancelled")
114 }
115
116 payload, _ := json.Marshal(OrderCancelled{
117 BaseEvent: BaseEvent{
118 AggregateId: a.state.ID,
119 Type: "OrderCancelled",
120 },
121 Reason: reason,
122 CancelledBy: cancelledBy,
123 })
124
125 event := eventstore.StoredEvent{
126 AggregateID: a.state.ID,
127 EventType: "OrderCancelled",
128 Payload: payload,
129 }
130
131 return a.apply(event, true)
132}
133
134func (a *OrderAggregate) apply(event eventstore.StoredEvent, isNew bool) error {
135 switch event.EventType {
136 case "OrderPlaced":
137 var e OrderPlaced
138 if err := json.Unmarshal(event.Payload, &e); err != nil {
139 return err
140 }
141 a.state.Status = OrderPlaced_Status
142 a.state.CustomerID = e.CustomerID
143 a.state.LineItems = e.LineItems
144 a.state.TotalAmount = e.TotalAmount
145 a.state.Currency = e.Currency
146 case "OrderConfirmed":
147 a.state.Status = OrderConfirmed
148 case "OrderCancelled":
149 a.state.Status = OrderCancelled
150 }
151
152 a.state.Version++
153 if isNew {
154 a.uncommittedEvents = append(a.uncommittedEvents, event)
155 }
156 return nil
157}
158
159func (a *OrderAggregate) UncommittedEvents() []eventstore.StoredEvent {
160 return a.uncommittedEvents
161}
162
163func (a *OrderAggregate) Version() int {
164 return a.state.Version
165}
166
167const OrderPlaced_Status OrderStatus = OrderPlaced_s
168
169const OrderPlaced_s OrderStatus = "placed"
170
Command Handler
The command handler orchestrates loading the aggregate, executing the command, and persisting new events.
go
1package application
2
3import (
4 "context"
5 "fmt"
6
7 "myapp/domain"
8 "myapp/eventstore"
9)
10
11type OrderCommandHandler struct {
12 eventStore *eventstore.PostgresEventStore
13}
14
15func NewOrderCommandHandler(es *eventstore.PostgresEventStore) *OrderCommandHandler {
16 return &OrderCommandHandler{eventStore: es}
17}
18
19func (h *OrderCommandHandler) HandlePlaceOrder(
20 ctx context.Context,
21 cmd PlaceOrderCommand,
22) error {
23 aggregate := domain.NewOrderAggregate(cmd.OrderID)
24
25 events, err := h.eventStore.Load(ctx, cmd.OrderID, 0)
26 if err != nil {
27 return fmt.Errorf("load events: %w", err)
28 }
29
30 if err := aggregate.Rehydrate(events); err != nil {
31 return fmt.Errorf("rehydrate: %w", err)
32 }
33
34 if err := aggregate.PlaceOrder(cmd.CustomerID, cmd.LineItems, cmd.Currency); err != nil {
35 return fmt.Errorf("place order: %w", err)
36 }
37
38 if err := h.eventStore.Append(
39 ctx,
40 cmd.OrderID,
41 "Order",
42 aggregate.Version()-len(aggregate.UncommittedEvents()),
43 aggregate.UncommittedEvents(),
44 ); err != nil {
45 return fmt.Errorf("append events: %w", err)
46 }
47
48 return nil
49}
50
51type PlaceOrderCommand struct {
52 OrderID string
53 CustomerID string
54 LineItems []domain.LineItem
55 Currency string
56}
57
Need a second opinion on your system design architecture?
I run free 30-minute strategy calls for engineering teams tackling this exact problem.
Book a Free CallBuilding Projections
Projections consume events asynchronously to build query-optimized read models. Go's goroutines make it natural to run multiple projection consumers concurrently.
go
1package projection
2
3import (
4 "context"
5 "database/sql"
6 "encoding/json"
7 "log"
8 "time"
9
10 "myapp/domain"
11 "myapp/eventstore"
12)
13
14type OrderListProjection struct {
15 eventStore *eventstore.PostgresEventStore
16 readDB *sql.DB
17 checkpoint int64
18 batchSize int
19}
20
21func NewOrderListProjection(es *eventstore.PostgresEventStore, readDB *sql.DB) *OrderListProjection {
22 return &OrderListProjection{
23 eventStore: es,
24 readDB: readDB,
25 batchSize: 100,
26 }
27}
28
29func (p *OrderListProjection) Start(ctx context.Context) error {
30 checkpoint, err := p.loadCheckpoint(ctx)
31 if err != nil {
32 return err
33 }
34 p.checkpoint = checkpoint
35
36 ticker := time.NewTicker(500 * time.Millisecond)
37 defer ticker.Stop()
38
39 for {
40 select {
41 case <-ctx.Done():
42 return ctx.Err()
43 case <-ticker.C:
44 if err := p.poll(ctx); err != nil {
45 log.Printf("projection error: %v", err)
46 }
47 }
48 }
49}
50
51func (p *OrderListProjection) poll(ctx context.Context) error {
52 events, err := p.eventStore.ReadAll(ctx, p.checkpoint, p.batchSize)
53 if err != nil {
54 return err
55 }
56
57 for _, event := range events {
58 if err := p.handle(ctx, event); err != nil {
59 log.Printf("failed to handle event %d: %v", event.ID, err)
60 continue
61 }
62 p.checkpoint = event.ID
63 if err := p.saveCheckpoint(ctx, p.checkpoint); err != nil {
64 return err
65 }
66 }
67
68 return nil
69}
70
71func (p *OrderListProjection) handle(ctx context.Context, event eventstore.StoredEvent) error {
72 switch event.EventType {
73 case "OrderPlaced":
74 var e domain.OrderPlaced
75 if err := json.Unmarshal(event.Payload, &e); err != nil {
76 return err
77 }
78 _, err := p.readDB.ExecContext(ctx,
79 `INSERT INTO order_list_view (order_id, customer_id, status, total_amount, currency, placed_at)
80 VALUES ($1, $2, $3, $4, $5, $6)
81 ON CONFLICT (order_id) DO NOTHING`,
82 e.AggregateId, e.CustomerID, "placed", e.TotalAmount, e.Currency, e.Timestamp,
83 )
84 return err
85
86 case "OrderConfirmed":
87 _, err := p.readDB.ExecContext(ctx,
88 `UPDATE order_list_view SET status = 'confirmed', updated_at = NOW() WHERE order_id = $1`,
89 event.AggregateID,
90 )
91 return err
92
93 case "OrderCancelled":
94 _, err := p.readDB.ExecContext(ctx,
95 `UPDATE order_list_view SET status = 'cancelled', updated_at = NOW() WHERE order_id = $1`,
96 event.AggregateID,
97 )
98 return err
99 }
100
101 return nil
102}
103
104func (p *OrderListProjection) Rebuild(ctx context.Context) error {
105 _, err := p.readDB.ExecContext(ctx, "TRUNCATE order_list_view")
106 if err != nil {
107 return err
108 }
109 p.checkpoint = 0
110 return p.saveCheckpoint(ctx, 0)
111}
112
113func (p *OrderListProjection) loadCheckpoint(ctx context.Context) (int64, error) {
114 var cp int64
115 err := p.readDB.QueryRowContext(ctx,
116 `SELECT position FROM projection_checkpoints WHERE projection_name = 'order_list'`,
117 ).Scan(&cp)
118 if err == sql.ErrNoRows {
119 return 0, nil
120 }
121 return cp, err
122}
123
124func (p *OrderListProjection) saveCheckpoint(ctx context.Context, position int64) error {
125 _, err := p.readDB.ExecContext(ctx,
126 `INSERT INTO projection_checkpoints (projection_name, position, updated_at)
127 VALUES ('order_list', $1, NOW())
128 ON CONFLICT (projection_name) DO UPDATE SET position = $1, updated_at = NOW()`,
129 position,
130 )
131 return err
132}
133
Query Service
The query side reads from projections, completely decoupled from the write side.
go
1package query
2
3import (
4 "context"
5 "database/sql"
6)
7
8type OrderView struct {
9 OrderID string `json:"order_id"`
10 CustomerID string `json:"customer_id"`
11 Status string `json:"status"`
12 TotalAmount int64 `json:"total_amount"`
13 Currency string `json:"currency"`
14 PlacedAt string `json:"placed_at"`
15}
16
17type OrderQueryService struct {
18 db *sql.DB
19}
20
21func (s *OrderQueryService) GetOrder(ctx context.Context, orderID string) (*OrderView, error) {
22 var view OrderView
23 err := s.db.QueryRowContext(ctx,
24 `SELECT order_id, customer_id, status, total_amount, currency, placed_at
25 FROM order_list_view WHERE order_id = $1`,
26 orderID,
27 ).Scan(&view.OrderID, &view.CustomerID, &view.Status,
28 &view.TotalAmount, &view.Currency, &view.PlacedAt)
29 if err != nil {
30 return nil, err
31 }
32 return &view, nil
33}
34
35func (s *OrderQueryService) ListOrders(
36 ctx context.Context,
37 customerID string,
38 limit, offset int,
39) ([]OrderView, error) {
40 rows, err := s.db.QueryContext(ctx,
41 `SELECT order_id, customer_id, status, total_amount, currency, placed_at
42 FROM order_list_view
43 WHERE customer_id = $1
44 ORDER BY placed_at DESC
45 LIMIT $2 OFFSET $3`,
46 customerID, limit, offset,
47 )
48 if err != nil {
49 return nil, err
50 }
51 defer rows.Close()
52
53 var orders []OrderView
54 for rows.Next() {
55 var v OrderView
56 if err := rows.Scan(&v.OrderID, &v.CustomerID, &v.Status,
57 &v.TotalAmount, &v.Currency, &v.PlacedAt); err != nil {
58 return nil, err
59 }
60 orders = append(orders, v)
61 }
62 return orders, rows.Err()
63}
64
Snapshot Support
For aggregates with long event histories, snapshots avoid replaying the entire stream.
go
1package eventstore
2
3import (
4 "context"
5 "database/sql"
6 "encoding/json"
7)
8
9type Snapshot struct {
10 AggregateID string `json:"aggregate_id"`
11 Version int `json:"version"`
12 State json.RawMessage `json:"state"`
13}
14
15type SnapshotStore struct {
16 db *sql.DB
17}
18
19func (s *SnapshotStore) Save(ctx context.Context, snap Snapshot) error {
20 _, err := s.db.ExecContext(ctx,
21 `INSERT INTO snapshots (aggregate_id, version, state, created_at)
22 VALUES ($1, $2, $3, NOW())
23 ON CONFLICT (aggregate_id) DO UPDATE SET version = $2, state = $3, created_at = NOW()`,
24 snap.AggregateID, snap.Version, snap.State,
25 )
26 return err
27}
28
29func (s *SnapshotStore) Load(ctx context.Context, aggregateID string) (*Snapshot, error) {
30 var snap Snapshot
31 err := s.db.QueryRowContext(ctx,
32 `SELECT aggregate_id, version, state FROM snapshots WHERE aggregate_id = $1`,
33 aggregateID,
34 ).Scan(&snap.AggregateID, &snap.Version, &snap.State)
35 if err == sql.ErrNoRows {
36 return nil, nil
37 }
38 if err != nil {
39 return nil, err
40 }
41 return &snap, nil
42}
43
Production Considerations
Concurrency: Use errgroup for running multiple projections concurrently. Each projection goroutine should have its own context for independent cancellation.
Serialization: Stick with encoding/json for event payloads unless you measure serialization as a bottleneck. Protocol Buffers add complexity and reduce debuggability of stored events.
Testing: Test aggregates with given-when-then style tests — given a set of historical events, when a command executes, then specific events should be emitted.
Monitoring: Export projection lag (current event store position minus projection checkpoint) as a Prometheus gauge. Alert when lag exceeds your SLA threshold.
Conclusion
Go provides an excellent foundation for CQRS and Event Sourcing implementations. The explicit error handling prevents silent failures in event processing, the type system catches event schema mismatches at compile time, and goroutines give you a natural concurrency model for projection consumers. Start with PostgreSQL as your event store — it handles surprisingly high throughput — and only move to specialized solutions like EventStoreDB or Kafka when you outgrow it.
The code patterns in this guide are designed to be copied and adapted. Every component — event store, aggregate, projection, query service — follows Go idioms and can be tested independently.