《后端架构设计·架构基础与边界》3.2 订单域事件风暴工作坊：产出聚合、事件、边界

// file: internal/eventstorm/model.go
// tier: L1
// verify: go test ./internal/eventstorm/
package eventstorm

import "fmt"

type Trigger string

const (
	TriggerHTTP    Trigger = "HTTP"
	TriggerTimer   Trigger = "TIMER"
	TriggerGateway Trigger = "GATEWAY"
	TriggerPolicy  Trigger = "POLICY"
)

type Command struct {
	Name      string
	Aggregate string
	Triggers  []Trigger
}

type Event struct {
	Name      string
	Aggregate string
	Producer  string
	Observed  bool
}

type Policy struct {
	Name  string
	On    []string
	Emits []string
	Guard string
}

type Model struct {
	Aggregates []string
	Commands   []Command
	Events     []Event
	Policies   []Policy
}

// Validate 把工作坊的四张表变成一组可执行断言。
func (m Model) Validate() []string {
	var out []string
	aggregates := make(map[string]bool, len(m.Aggregates))
	commands := make(map[string]Command, len(m.Commands))
	events := make(map[string]Event, len(m.Events))
	consumed := make(map[string]bool)
	emitted := make(map[string]bool)

	for _, a := range m.Aggregates {
		aggregates[a] = true
	}
	for _, c := range m.Commands {
		if _, dup := commands[c.Name]; dup {
			out = append(out, "duplicate command "+c.Name)
		}
		commands[c.Name] = c
		if !aggregates[c.Aggregate] {
			out = append(out, fmt.Sprintf("command %s has unknown aggregate %s", c.Name, c.Aggregate))
		}
		if len(c.Triggers) == 0 {
			out = append(out, "command without trigger "+c.Name)
		}
	}
	for _, e := range m.Events {
		if _, dup := events[e.Name]; dup {
			out = append(out, "duplicate event "+e.Name)
		}
		events[e.Name] = e
		producer, ok := commands[e.Producer]
		if !ok {
			out = append(out, fmt.Sprintf("event %s has no producer command %s", e.Name, e.Producer))
			continue
		}
		if producer.Aggregate != e.Aggregate {
			out = append(out, fmt.Sprintf("event %s aggregate %s != producer aggregate %s", e.Name, e.Aggregate, producer.Aggregate))
		}
	}
	for _, p := range m.Policies {
		if len(p.On) == 0 || len(p.Emits) == 0 || p.Guard == "" {
			out = append(out, "incomplete policy "+p.Name)
		}
		for _, name := range p.On {
			if _, ok := events[name]; !ok {
				out = append(out, fmt.Sprintf("policy %s references unknown event %s", p.Name, name))
			}
			consumed[name] = true
		}
		for _, name := range p.Emits {
			if _, ok := commands[name]; !ok {
				out = append(out, fmt.Sprintf("policy %s emits unknown command %s", p.Name, name))
			}
			emitted[name] = true
		}
	}
	for _, e := range m.Events {
		if !e.Observed && !consumed[e.Name] {
			out = append(out, "orphan event "+e.Name)
		}
	}
	for _, c := range m.Commands {
		if hasPolicyTrigger(c.Triggers) && !emitted[c.Name] {
			out = append(out, "unreachable policy command "+c.Name)
		}
	}
	return out
}

func hasPolicyTrigger(triggers []Trigger) bool {
	for _, tr := range triggers {
		if tr == TriggerPolicy {
			return true
		}
	}
	return false
}

// file: internal/eventstorm/model_test.go
// tier: L1
// verify: go test ./internal/eventstorm/ -v
package eventstorm

import (
	"strings"
	"testing"
)

func workshopModel() Model {
	return Model{
		Aggregates: []string{"Order", "InventoryReservation", "Payment"},
		Commands: []Command{
			{Name: "PlaceOrder", Aggregate: "Order", Triggers: []Trigger{TriggerHTTP}},
			{Name: "ReserveInventory", Aggregate: "InventoryReservation", Triggers: []Trigger{TriggerPolicy}},
			{Name: "MarkInventoryReserved", Aggregate: "Order", Triggers: []Trigger{TriggerPolicy}},
			{Name: "ConfirmPayment", Aggregate: "Payment", Triggers: []Trigger{TriggerGateway}},
			{Name: "MarkOrderPaid", Aggregate: "Order", Triggers: []Trigger{TriggerPolicy}},
			{Name: "CancelOrder", Aggregate: "Order", Triggers: []Trigger{TriggerHTTP, TriggerPolicy}},
			{Name: "ReleaseInventory", Aggregate: "InventoryReservation", Triggers: []Trigger{TriggerPolicy}},
			{Name: "RefundPayment", Aggregate: "Payment", Triggers: []Trigger{TriggerPolicy}},
			{Name: "FulfillOrder", Aggregate: "Order", Triggers: []Trigger{TriggerHTTP}},
			{Name: "ExpireOrder", Aggregate: "Order", Triggers: []Trigger{TriggerTimer}},
			{Name: "RequestRefund", Aggregate: "Order", Triggers: []Trigger{TriggerHTTP}},
		},
		Events: []Event{
			{Name: "OrderPlaced", Aggregate: "Order", Producer: "PlaceOrder"},
			{Name: "InventoryReserved", Aggregate: "InventoryReservation", Producer: "ReserveInventory"},
			{Name: "InventoryReservationFailed", Aggregate: "InventoryReservation", Producer: "ReserveInventory"},
			{Name: "PaymentConfirmed", Aggregate: "Payment", Producer: "ConfirmPayment"},
			{Name: "PaymentDeclined", Aggregate: "Payment", Producer: "ConfirmPayment"},
			{Name: "OrderPaid", Aggregate: "Order", Producer: "MarkOrderPaid", Observed: true},
			{Name: "OrderCancelled", Aggregate: "Order", Producer: "CancelOrder"},
			{Name: "InventoryReleased", Aggregate: "InventoryReservation", Producer: "ReleaseInventory", Observed: true},
			{Name: "PaymentRefunded", Aggregate: "Payment", Producer: "RefundPayment", Observed: true},
			{Name: "OrderFulfilled", Aggregate: "Order", Producer: "FulfillOrder", Observed: true},
			{Name: "OrderExpired", Aggregate: "Order", Producer: "ExpireOrder"},
			{Name: "RefundRequested", Aggregate: "Order", Producer: "RequestRefund"},
			{Name: "OrderInventoryReserved", Aggregate: "Order", Producer: "MarkInventoryReserved", Observed: true},
		},
		Policies: []Policy{
			{Name: "P1", On: []string{"OrderPlaced"}, Emits: []string{"ReserveInventory"}, Guard: "order.status == CREATED"},
			{Name: "P2", On: []string{"InventoryReserved"}, Emits: []string{"MarkInventoryReserved"}, Guard: "all lines allocated"},
			{Name: "P3", On: []string{"InventoryReservationFailed"}, Emits: []string{"CancelOrder"}, Guard: "order.status == CREATED"},
			{Name: "P4", On: []string{"PaymentConfirmed"}, Emits: []string{"MarkOrderPaid"}, Guard: "order.status == INVENTORY_RESERVED"},
			{Name: "P5", On: []string{"PaymentDeclined"}, Emits: []string{"CancelOrder", "ReleaseInventory"}, Guard: "payment attempt is terminal"},
			{Name: "P6", On: []string{"OrderCancelled"}, Emits: []string{"ReleaseInventory"}, Guard: "order was INVENTORY_RESERVED"},
			{Name: "P7", On: []string{"OrderCancelled"}, Emits: []string{"RefundPayment"}, Guard: "order was PAID"},
			{Name: "P8", On: []string{"OrderExpired"}, Emits: []string{"CancelOrder", "ReleaseInventory"}, Guard: "reservation TTL reached"},
			{Name: "P9", On: []string{"RefundRequested"}, Emits: []string{"RefundPayment"}, Guard: "order.status == PAID"},
		},
	}
}

func has(violations []string, fragment string) bool {
	for _, v := range violations {
		if strings.Contains(v, fragment) {
			return true
		}
	}
	return false
}

func TestWorkshopModelIsValid(t *testing.T) {
	if violations := workshopModel().Validate(); len(violations) != 0 {
		t.Fatalf("workshop model should be valid, got %v", violations)
	}
}

func TestValidateTable(t *testing.T) {
	tests := []struct {
		name   string
		mutate func(*Model)
		want   string
	}{
		{"command without aggregate", func(m *Model) {
			m.Commands = append(m.Commands, Command{Name: "ShipIt", Aggregate: "Shipment", Triggers: []Trigger{TriggerHTTP}})
		}, "unknown aggregate Shipment"},
		{"event without producer", func(m *Model) {
			m.Events = append(m.Events, Event{Name: "GhostEvent", Aggregate: "Order", Producer: "GhostCommand"})
		}, "no producer command GhostCommand"},
		{"event aggregate mismatch", func(m *Model) { m.Events[0].Aggregate = "Payment" }, "aggregate Payment != producer aggregate Order"},
		{"unknown event in policy", func(m *Model) {
			m.Policies = append(m.Policies, Policy{Name: "P10", On: []string{"GhostEvent"}, Emits: []string{"CancelOrder"}, Guard: "never"})
		}, "references unknown event GhostEvent"},
		{"unknown command in policy", func(m *Model) {
			m.Policies = append(m.Policies, Policy{Name: "P10", On: []string{"OrderPlaced"}, Emits: []string{"GhostCommand"}, Guard: "never"})
		}, "emits unknown command GhostCommand"},
		{"orphan event", func(m *Model) {
			m.Events = append(m.Events, Event{Name: "OrderArchived", Aggregate: "Order", Producer: "PlaceOrder"})
		}, "orphan event OrderArchived"},
		{"unreachable policy command", func(m *Model) {
			m.Commands = append(m.Commands, Command{Name: "RecalculateRisk", Aggregate: "Order", Triggers: []Trigger{TriggerPolicy}})
		}, "unreachable policy command RecalculateRisk"},
		{"policy without guard", func(m *Model) { m.Policies[0].Guard = "" }, "incomplete policy P1"},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			m := workshopModel()
			tt.mutate(&m)
			if violations := m.Validate(); !has(violations, tt.want) {
				t.Fatalf("violations = %v, want %q", violations, tt.want)
			}
		})
	}
}

# 输出节选：其余 7 个子用例均为 PASS
=== RUN   TestWorkshopModelIsValid
--- PASS: TestWorkshopModelIsValid (0.00s)
=== RUN   TestValidateTable
=== RUN   TestValidateTable/orphan_event
--- PASS: TestValidateTable (0.00s)
    --- PASS: TestValidateTable/orphan_event (0.00s)
PASS
ok  	archlab/internal/eventstorm	0.457s