《互联网产品的服务端架构·产品、流量与基础能力》9.2 事件、通知与最终一致

// file: internal/async/event.go
// tier: L1
// verify: go test ./internal/async/

package async

import (
	"fmt"
	"strings"
	"time"
)

// Event 是已经发生的事实。业务系统只描述发生了什么，不描述要通知谁。
type Event struct {
	ID         string
	Type       string
	Version    int
	OccurredAt time.Time
}

// Consumer 是一个消费者的幂等账本。
// Seen 以事件 ID 为键，记录首次处理时间；Types 记录首次看到的类型，
// 用来发现同一个 ID 被不同事件复用的生产端错误。
type Consumer struct {
	Name       string
	Seen       map[string]time.Time
	Types      map[string]string
	MinVersion int
}

// Apply 处理一个事件。返回 true 表示本次生效，false 表示重复投递。
// 版本过低、ID 与 Type 冲突、账本为空都会返回错误，且不改变状态。
func Apply(c *Consumer, e Event, now time.Time) (bool, error) {
	if c == nil {
		return false, fmt.Errorf("async: nil consumer")
	}
	if strings.TrimSpace(e.ID) == "" || strings.TrimSpace(e.Type) == "" {
		return false, fmt.Errorf("async: event id and type are required")
	}
	if e.Version < c.MinVersion {
		return false, fmt.Errorf("async: %s: event %s version %d below minimum %d", c.Name, e.ID, e.Version, c.MinVersion)
	}
	if _, ok := c.Seen[e.ID]; ok {
		if first, typed := c.Types[e.ID]; typed && first != e.Type {
			return false, fmt.Errorf("async: %s: event %s seen as %q, now %q", c.Name, e.ID, first, e.Type)
		}
		return false, nil
	}
	if c.Seen == nil {
		c.Seen = make(map[string]time.Time)
	}
	if c.Types == nil {
		c.Types = make(map[string]string)
	}
	c.Seen[e.ID] = now
	c.Types[e.ID] = e.Type
	return true, nil
}

// file: internal/async/event_test.go
// tier: L1
// verify: go test ./internal/async/

package async

import (
	"testing"
	"time"
)

func TestApplyDeduplicatesByID(t *testing.T) {
	first := time.Date(2026, 10, 28, 13, 40, 0, 0, time.UTC)
	later := first.Add(time.Minute)
	c := &Consumer{Name: "like-counter", MinVersion: 1}
	e := Event{ID: "evt-1", Type: "post.liked", Version: 2, OccurredAt: first}

	applied, err := Apply(c, e, first)
	if err != nil {
		t.Fatal(err)
	}
	if !applied {
		t.Fatal("first delivery must apply")
	}
	if got := c.Seen["evt-1"]; !got.Equal(first) {
		t.Fatalf("seen at %s, want %s", got, first)
	}

	applied, err = Apply(c, e, later)
	if err != nil {
		t.Fatalf("duplicate must not error: %v", err)
	}
	if applied {
		t.Fatal("duplicate must not apply twice")
	}
	if got := c.Seen["evt-1"]; !got.Equal(first) {
		t.Fatalf("duplicate moved first-seen time to %s", got)
	}
}

func TestApplyRejectsOldVersion(t *testing.T) {
	c := &Consumer{Name: "order-view", MinVersion: 3}
	applied, err := Apply(c, Event{ID: "evt-2", Type: "order.paid", Version: 2}, time.Now())
	if err == nil {
		t.Fatal("want error for version below minimum")
	}
	if applied {
		t.Fatal("old version must not apply")
	}
	if len(c.Seen) != 0 {
		t.Fatalf("old version was recorded: %+v", c.Seen)
	}

	applied, err = Apply(c, Event{ID: "evt-3", Type: "order.paid", Version: 3}, time.Now())
	if err != nil || !applied {
		t.Fatalf("version equal to minimum should apply: applied=%v err=%v", applied, err)
	}
}

func TestApplyRejectsIDTypeConflict(t *testing.T) {
	now := time.Date(2026, 10, 28, 13, 40, 0, 0, time.UTC)
	c := &Consumer{Name: "like-counter"}
	if _, err := Apply(c, Event{ID: "evt-4", Type: "post.liked", Version: 1}, now); err != nil {
		t.Fatal(err)
	}

	applied, err := Apply(c, Event{ID: "evt-4", Type: "post.unliked", Version: 1}, now.Add(time.Second))
	if err == nil {
		t.Fatal("want conflict error for reused ID")
	}
	if applied {
		t.Fatal("conflicting event must not apply")
	}
	if got := c.Seen["evt-4"]; !got.Equal(now) {
		t.Fatalf("conflict changed first-seen time to %s", got)
	}
	if got := c.Types["evt-4"]; got != "post.liked" {
		t.Fatalf("first type = %q, want post.liked", got)
	}
}

func TestApplyRejectsInvalidInput(t *testing.T) {
	now := time.Now()
	if _, err := Apply(nil, Event{ID: "e", Type: "t"}, now); err == nil {
		t.Fatal("want error for nil consumer")
	}
	c := &Consumer{Name: "x"}
	if _, err := Apply(c, Event{ID: " ", Type: "t"}, now); err == nil {
		t.Fatal("want error for blank id")
	}
	if _, err := Apply(c, Event{ID: "e", Type: ""}, now); err == nil {
		t.Fatal("want error for empty type")
	}
	if len(c.Seen) != 0 {
		t.Fatalf("invalid events were recorded: %+v", c.Seen)
	}
}

cd .local/checks/webapp-lab && go test ./internal/async/ -count=1 -run 'TestApply' -v

=== RUN   TestApplyDeduplicatesByID
--- PASS: TestApplyDeduplicatesByID (0.00s)
=== RUN   TestApplyRejectsOldVersion
--- PASS: TestApplyRejectsOldVersion (0.00s)
=== RUN   TestApplyRejectsIDTypeConflict
--- PASS: TestApplyRejectsIDTypeConflict (0.00s)
=== RUN   TestApplyRejectsInvalidInput
--- PASS: TestApplyRejectsInvalidInput (0.00s)
PASS
ok  	webapp/internal/async	0.237s