《后端架构设计·架构基础与边界》6.3 订单读模型：CQRS 的引入判据与退出条件

// file: internal/readmodel/projection.go
// tier: L1
// verify: go test ./internal/readmodel/
package readmodel

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"time"
)

var (
	ErrNotCreated   = errors.New("first event must be OrderCreated")
	ErrUnknownEvent = errors.New("unknown event type")
)

type EventType string

const (
	EventCreated       EventType = "OrderCreated"
	EventStatusChanged EventType = "OrderStatusChanged"
)

type OrderEvent struct {
	EventID    string
	OrderID    string
	Version    int
	Type       EventType
	CustomerID string
	Status     string
	TotalCents int64
	ItemCount  int
	OccurredAt time.Time
}

type OrderView struct {
	OrderID    string
	CustomerID string
	Status     string
	TotalCents int64
	ItemCount  int
	Version    int
	UpdatedAt  time.Time
}

type Query struct {
	CustomerID string
	Status     string
	Limit      int
}

type Store interface {
	Get(ctx context.Context, orderID string) (OrderView, bool, error)
	Upsert(ctx context.Context, v OrderView) error
	List(ctx context.Context, q Query) ([]OrderView, error)
	Reset(ctx context.Context) error
}

type Projection struct{ store Store }

func New(store Store) *Projection { return &Projection{store: store} }

// Apply 以聚合版本做幂等：重复投递同版本、或旧版本晚到时直接忽略。
func (p *Projection) Apply(ctx context.Context, e OrderEvent) error {
	current, found, err := p.store.Get(ctx, e.OrderID)
	if err != nil {
		return err
	}
	if found && e.Version <= current.Version {
		return nil
	}
	v := current
	if !found {
		if e.Type != EventCreated {
			return fmt.Errorf("%w: got %s at version %d", ErrNotCreated, e.Type, e.Version)
		}
		v = OrderView{OrderID: e.OrderID}
	}
	switch e.Type {
	case EventCreated:
		v.CustomerID = e.CustomerID
		v.TotalCents = e.TotalCents
		v.ItemCount = e.ItemCount
		v.Status = e.Status
	case EventStatusChanged:
		v.Status = e.Status
	default:
		return fmt.Errorf("%w: %s", ErrUnknownEvent, e.Type)
	}
	v.Version = e.Version
	v.UpdatedAt = e.OccurredAt
	return p.store.Upsert(ctx, v)
}

// Rebuild 先按聚合、再按版本排序，使乱序历史事件也能得到确定结果。
func (p *Projection) Rebuild(ctx context.Context, events []OrderEvent) error {
	if err := p.store.Reset(ctx); err != nil {
		return err
	}
	ordered := append([]OrderEvent(nil), events...)
	sort.SliceStable(ordered, func(i, j int) bool {
		if ordered[i].OrderID != ordered[j].OrderID {
			return ordered[i].OrderID < ordered[j].OrderID
		}
		return ordered[i].Version < ordered[j].Version
	})
	for _, e := range ordered {
		if err := p.Apply(ctx, e); err != nil {
			return err
		}
	}
	return nil
}

// file: internal/readmodel/projection_test.go
// tier: L1
// verify: go test ./internal/readmodel/
package readmodel

import (
	"context"
	"errors"
	"testing"
	"time"
)

type memStore struct {
	views map[string]OrderView
}

func newMemStore() *memStore { return &memStore{views: make(map[string]OrderView)} }

func (s *memStore) Get(ctx context.Context, orderID string) (OrderView, bool, error) {
	v, ok := s.views[orderID]
	return v, ok, nil
}

func (s *memStore) Upsert(ctx context.Context, v OrderView) error {
	s.views[v.OrderID] = v
	return nil
}

func (s *memStore) Reset(ctx context.Context) error {
	s.views = make(map[string]OrderView)
	return nil
}

func (s *memStore) List(ctx context.Context, q Query) ([]OrderView, error) {
	out := make([]OrderView, 0, len(s.views))
	for _, v := range s.views {
		if q.CustomerID != "" && v.CustomerID != q.CustomerID {
			continue
		}
		if q.Status != "" && v.Status != q.Status {
			continue
		}
		out = append(out, v)
	}
	if q.Limit > 0 && len(out) > q.Limit {
		out = out[:q.Limit]
	}
	return out, nil
}

var base = time.Date(2026, 10, 8, 16, 0, 0, 0, time.UTC)

func created(orderID, customer string) OrderEvent {
	return OrderEvent{
		EventID: "evt-created-" + orderID, OrderID: orderID, Version: 1,
		Type: EventCreated, CustomerID: customer, Status: "CREATED",
		TotalCents: 12800, ItemCount: 3, OccurredAt: base,
	}
}

func changed(orderID string, version int, status string, at time.Time) OrderEvent {
	return OrderEvent{
		EventID: "evt-" + orderID + "-" + status, OrderID: orderID,
		Version: version, Type: EventStatusChanged, Status: status, OccurredAt: at,
	}
}

func TestIncrementalMatchesRebuild(t *testing.T) {
	ctx := context.Background()
	events := []OrderEvent{
		created("o-1", "c-1"),
		changed("o-1", 2, "PAID", base.Add(time.Minute)),
		changed("o-1", 3, "FULFILLED", base.Add(2*time.Minute)),
	}
	incremental := New(newMemStore())
	for _, e := range events {
		if err := incremental.Apply(ctx, e); err != nil {
			t.Fatal(err)
		}
	}
	rebuilt := New(newMemStore())
	if err := rebuilt.Rebuild(ctx, events); err != nil {
		t.Fatal(err)
	}
	a, _, _ := incremental.store.Get(ctx, "o-1")
	b, _, _ := rebuilt.store.Get(ctx, "o-1")
	if a != b || a.Status != "FULFILLED" || a.Version != 3 {
		t.Fatalf("incremental=%+v rebuilt=%+v", a, b)
	}
}

func TestIgnoresDuplicateAndStaleVersions(t *testing.T) {
	ctx := context.Background()
	p := New(newMemStore())
	for _, e := range []OrderEvent{
		created("o-1", "c-1"),
		created("o-1", "c-1"),
		changed("o-1", 2, "PAID", base.Add(time.Minute)),
		changed("o-1", 1, "CANCELLED", base.Add(2*time.Minute)),
	} {
		if err := p.Apply(ctx, e); err != nil {
			t.Fatal(err)
		}
	}
	v, _, _ := p.store.Get(ctx, "o-1")
	if v.Version != 2 || v.Status != "PAID" {
		t.Fatalf("stale event won: %+v", v)
	}
}

func TestRejectsStatusBeforeCreated(t *testing.T) {
	err := New(newMemStore()).Apply(context.Background(), changed("o-1", 2, "PAID", base))
	if !errors.Is(err, ErrNotCreated) {
		t.Fatalf("err = %v, want ErrNotCreated", err)
	}
}

func TestListFiltersAndLimit(t *testing.T) {
	ctx := context.Background()
	p := New(newMemStore())
	events := []OrderEvent{
		created("o-1", "c-1"),
		created("o-2", "c-1"),
		created("o-3", "c-2"),
		changed("o-1", 2, "PAID", base.Add(time.Minute)),
	}
	if err := p.Rebuild(ctx, events); err != nil {
		t.Fatal(err)
	}
	got, err := p.store.List(ctx, Query{CustomerID: "c-1"})
	if err != nil || len(got) != 2 {
		t.Fatalf("customer filter: %+v err=%v", got, err)
	}
	got, err = p.store.List(ctx, Query{Status: "PAID", Limit: 1})
	if err != nil || len(got) != 1 || got[0].OrderID != "o-1" {
		t.Fatalf("status filter: %+v err=%v", got, err)
	}
}

$ gofmt -l .
$ go vet ./...
$ go test ./... -count=1 -v
=== RUN   TestIncrementalMatchesRebuild
--- PASS: TestIncrementalMatchesRebuild (0.00s)
=== RUN   TestIgnoresDuplicateAndStaleVersions
--- PASS: TestIgnoresDuplicateAndStaleVersions (0.00s)
=== RUN   TestRejectsStatusBeforeCreated
--- PASS: TestRejectsStatusBeforeCreated (0.00s)
=== RUN   TestListFiltersAndLimit
--- PASS: TestListFiltersAndLimit (0.00s)
PASS
ok  	archlab/internal/readmodel	0.462s