《游戏服务端架构·核心系统与契约》4.3 世界事件与全服广播

每次广播副本数 = 订阅者数量 C
每秒字节数 ≈ C × 单条字节数 B × 频率 f

// file: internal/globalevent/globalevent.go
// tier: L1
// verify: go test ./internal/globalevent/
package globalevent

import (
	"errors"
	"fmt"
)

var (
	ErrBadLog          = errors.New("globalevent: log size must be positive")
	ErrBadTier         = errors.New("globalevent: unknown tier")
	ErrBadPartition    = errors.New("globalevent: partition does not match tier")
	ErrEmptyID         = errors.New("globalevent: empty event id")
	ErrEmptySubscriber = errors.New("globalevent: empty subscriber id")
	ErrDuplicateEvent  = errors.New("globalevent: event id already published")
	ErrOutOfOrder      = errors.New("globalevent: sequence did not advance")
	ErrSequenceGap     = errors.New("globalevent: sequence gap requires backlog replay")
	ErrBadSequence     = errors.New("globalevent: sequence must be positive")
)

// Tier selects the fanout channel. Global reaches gateway-level subscribers
// once; the gateway then copies to its local sessions. Theater, map and room
// events go only to subscribers of that exact partition.
type Tier int

const (
	TierGlobal Tier = iota
	TierTheater
	TierMap
	TierRoom
)

func (t Tier) String() string {
	switch t {
	case TierGlobal:
		return "global"
	case TierTheater:
		return "theater"
	case TierMap:
		return "map"
	case TierRoom:
		return "room"
	default:
		return "unknown"
	}
}

func (t Tier) valid() bool { return t >= TierGlobal && t <= TierRoom }

// Event is one immutable broadcast. Seq is assigned by the bus per partition;
// ID is assigned by the producer and is the dedupe key across retries.
type Event struct {
	ID        string
	Tier      Tier
	Partition string
	Seq       uint64
	Payload   string
}

type partitionKey struct {
	tier      Tier
	partition string
}

func keyFor(tier Tier, partition string) (partitionKey, error) {
	if !tier.valid() {
		return partitionKey{}, fmt.Errorf("%w: %d", ErrBadTier, tier)
	}
	if tier == TierGlobal {
		if partition != "" {
			return partitionKey{}, fmt.Errorf("%w: global must not have a partition", ErrBadPartition)
		}
	} else if partition == "" {
		return partitionKey{}, fmt.Errorf("%w: %s needs a partition", ErrBadPartition, tier)
	}
	return partitionKey{tier: tier, partition: partition}, nil
}

// Bus is the publication side. It keeps a bounded per-partition log so a
// reconnecting subscriber can replay what it missed without a full snapshot.
type Bus struct {
	maxLog int
	subs   map[partitionKey]map[string]struct{}
	seen   map[string]bool
	logs   map[partitionKey][]Event
	next   map[partitionKey]uint64
}

func NewBus(maxLog int) (*Bus, error) {
	if maxLog <= 0 {
		return nil, ErrBadLog
	}
	return &Bus{
		maxLog: maxLog,
		subs:   make(map[partitionKey]map[string]struct{}),
		seen:   make(map[string]bool),
		logs:   make(map[partitionKey][]Event),
		next:   make(map[partitionKey]uint64),
	}, nil
}

// Subscribe registers one gateway, worker or session for a channel. A global
// subscription is a gateway subscription, not one per player.
func (b *Bus) Subscribe(tier Tier, partition, subscriber string) error {
	if subscriber == "" {
		return ErrEmptySubscriber
	}
	k, err := keyFor(tier, partition)
	if err != nil {
		return err
	}
	if b.subs[k] == nil {
		b.subs[k] = make(map[string]struct{})
	}
	b.subs[k][subscriber] = struct{}{}
	return nil
}

// Unsubscribe removes one subscriber; removing an unknown one is a no-op.
func (b *Bus) Unsubscribe(tier Tier, partition, subscriber string) error {
	k, err := keyFor(tier, partition)
	if err != nil {
		return err
	}
	delete(b.subs[k], subscriber)
	return nil
}

// Publish assigns the next sequence for the partition and returns the number of
// subscribers that must receive it. A repeated event ID is rejected before it
// can consume a sequence number.
func (b *Bus) Publish(e Event) (Event, int, error) {
	if e.ID == "" {
		return Event{}, 0, ErrEmptyID
	}
	k, err := keyFor(e.Tier, e.Partition)
	if err != nil {
		return Event{}, 0, err
	}
	if b.seen[e.ID] {
		return Event{}, 0, fmt.Errorf("%w: %s", ErrDuplicateEvent, e.ID)
	}
	b.next[k]++
	e.Seq = b.next[k]
	b.seen[e.ID] = true
	b.logs[k] = append(b.logs[k], e)
	if len(b.logs[k]) > b.maxLog {
		b.logs[k] = b.logs[k][len(b.logs[k])-b.maxLog:]
	}
	return e, len(b.subs[k]), nil
}

// Backlog returns events with Seq greater than after, oldest first. The caller
// then applies them through a Reader to close the reconnect window.
func (b *Bus) Backlog(tier Tier, partition string, after uint64) ([]Event, error) {
	k, err := keyFor(tier, partition)
	if err != nil {
		return nil, err
	}
	out := make([]Event, 0, len(b.logs[k]))
	for _, e := range b.logs[k] {
		if e.Seq > after {
			out = append(out, e)
		}
	}
	return out, nil
}

// Fanout is the pre-send cost estimate: subscriber count times bytes per copy.
// Compare it with the gateway's per-second budget before adding a channel.
type Fanout struct {
	Subscribers int
	Bytes       int64
}

func (f Fanout) TotalBytes() int64 { return int64(f.Subscribers) * f.Bytes }

func (b *Bus) Estimate(tier Tier, partition string, bytesPerCopy int64) (Fanout, error) {
	k, err := keyFor(tier, partition)
	if err != nil {
		return Fanout{}, err
	}
	if bytesPerCopy < 0 {
		return Fanout{}, fmt.Errorf("globalevent: negative payload size")
	}
	return Fanout{Subscribers: len(b.subs[k]), Bytes: bytesPerCopy}, nil
}

// Reader applies events for one subscriber. It rejects duplicates, gaps and
// stale sequences; a gap means the caller should fetch Backlog(last) first.
type Reader struct {
	seen map[string]bool
	last map[partitionKey]uint64
}

func NewReader() *Reader {
	return &Reader{seen: make(map[string]bool), last: make(map[partitionKey]uint64)}
}

// Apply advances the reader by exactly one sequence on one partition.
func (r *Reader) Apply(e Event) error {
	if e.ID == "" {
		return ErrEmptyID
	}
	if e.Seq == 0 {
		return ErrBadSequence
	}
	k, err := keyFor(e.Tier, e.Partition)
	if err != nil {
		return err
	}
	if r.seen[e.ID] {
		return fmt.Errorf("%w: %s", ErrDuplicateEvent, e.ID)
	}
	last := r.last[k]
	switch {
	case last == 0:
		// First event on a partition may start anywhere: the reader may have
		// restored a checkpoint before subscribing.
	case e.Seq == last+1:
	case e.Seq <= last:
		return fmt.Errorf("%w: got %d after %d", ErrOutOfOrder, e.Seq, last)
	default:
		return fmt.Errorf("%w: got %d after %d", ErrSequenceGap, e.Seq, last)
	}
	r.seen[e.ID] = true
	r.last[k] = e.Seq
	return nil
}

// Last returns the last applied sequence on a partition.
func (r *Reader) Last(tier Tier, partition string) (uint64, error) {
	k, err := keyFor(tier, partition)
	if err != nil {
		return 0, err
	}
	return r.last[k], nil
}

$ cd .local/checks/gsarch-lab
$ gofmt -l internal/globalevent
$ go vet ./internal/globalevent/...
$ go test -count=1 ./internal/globalevent/
ok  	gsarchlab/internal/globalevent