《游戏服务端架构·核心系统与契约》8.3 聊天频道、审核与限速

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

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

// Kind is a channel scope. Each scope has a different fanout and isolation
// boundary; the router never derives one scope from another.
type Kind int

const (
	World Kind = iota + 1
	Zone
	Guild
	Private
)

// Channel identifies one scope. World normally uses the fixed ID "global";
// zone, guild and private use their own identifier.
type Channel struct {
	Kind Kind
	ID   string
}

func (c Channel) key() string { return fmt.Sprintf("%d/%s", c.Kind, c.ID) }

// Message is an inbound chat line.
type Message struct {
	Sender  string
	Channel Channel
	Text    string
	At      time.Time
}

// Decision is the synchronous moderation result.
type Decision int

const (
	Allow Decision = iota
	Review
	Reject
)

// Moderator runs before fanout. Reject drops the line; Review delivers it and
// queues it for delayed inspection; Allow delivers normally.
type Moderator interface {
	Inspect(Message) Decision
}

// Config holds the two rate buckets and the duplicate window. Burst applies per
// sender globally and per sender/channel; refill is in tokens per second.
type Config struct {
	Burst           int
	RefillPerSecond float64
	DuplicateWindow time.Duration
	MaxTextRunes    int
}

// Delivery is the fanout decision returned to the caller.
type Delivery struct {
	Message    Message
	Recipients []string
	Review     bool
}

var (
	ErrConfig      = errors.New("chatmod: invalid config")
	ErrEmpty       = errors.New("chatmod: empty sender, text or channel id")
	ErrChannel     = errors.New("chatmod: unknown channel kind")
	ErrTooLong     = errors.New("chatmod: text exceeds limit")
	ErrNotMember   = errors.New("chatmod: sender is not a member")
	ErrNoRecipient = errors.New("chatmod: no recipient in channel")
	ErrDuplicate   = errors.New("chatmod: duplicate message")
	ErrRateLimited = errors.New("chatmod: rate limited")
	ErrRejected    = errors.New("chatmod: rejected by moderation")
)

type bucket struct {
	tokens float64
	last   time.Time
}

type rateSet struct {
	burst   float64
	rate    float64
	buckets map[string]*bucket
}

func newRateSet(burst int, rate float64) *rateSet {
	return &rateSet{burst: float64(burst), rate: rate, buckets: make(map[string]*bucket)}
}

func (r *rateSet) allow(key string, now time.Time) bool {
	b := r.buckets[key]
	if b == nil {
		b = &bucket{tokens: r.burst, last: now}
		r.buckets[key] = b
	}
	if elapsed := now.Sub(b.last).Seconds(); elapsed > 0 {
		b.tokens += elapsed * r.rate
		if b.tokens > r.burst {
			b.tokens = r.burst
		}
		b.last = now
	}
	if b.tokens < 1 {
		return false
	}
	b.tokens--
	return true
}

type stamp struct {
	text string
	at   time.Time
}

// Router owns membership, rate state and the moderation hook. The lock covers
// one Publish call; it is not held across network sends.
type Router struct {
	mu      sync.Mutex
	cfg     Config
	mod     Moderator
	members map[string]map[string]struct{}
	global  *rateSet
	scoped  *rateSet
	last    map[string]stamp
	pending int
}

type allowAll struct{}

func (allowAll) Inspect(Message) Decision { return Allow }

// New validates the config and returns a router. A nil moderator means allow.
func New(cfg Config, mod Moderator) (*Router, error) {
	if cfg.Burst < 1 || cfg.RefillPerSecond <= 0 || cfg.DuplicateWindow <= 0 || cfg.MaxTextRunes < 1 {
		return nil, ErrConfig
	}
	if mod == nil {
		mod = allowAll{}
	}
	return &Router{
		cfg:     cfg,
		mod:     mod,
		members: make(map[string]map[string]struct{}),
		global:  newRateSet(cfg.Burst, cfg.RefillPerSecond),
		scoped:  newRateSet(cfg.Burst, cfg.RefillPerSecond),
		last:    make(map[string]stamp),
	}, nil
}

// Join adds role to a channel. Joining is idempotent.
func (r *Router) Join(role string, ch Channel) error {
	if role == "" || ch.ID == "" {
		return ErrEmpty
	}
	if ch.Kind < World || ch.Kind > Private {
		return ErrChannel
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	key := ch.key()
	if r.members[key] == nil {
		r.members[key] = make(map[string]struct{})
	}
	r.members[key][role] = struct{}{}
	return nil
}

// Publish checks membership, duplicates, both rate buckets and moderation in
// that order. The global bucket is spent before the scoped one, so switching
// channels cannot buy extra messages.
func (r *Router) Publish(m Message) (Delivery, error) {
	if m.Sender == "" || m.Text == "" || m.Channel.ID == "" {
		return Delivery{}, ErrEmpty
	}
	if m.Channel.Kind < World || m.Channel.Kind > Private {
		return Delivery{}, ErrChannel
	}
	if len([]rune(m.Text)) > r.cfg.MaxTextRunes {
		return Delivery{}, ErrTooLong
	}
	if m.At.IsZero() {
		m.At = time.Now()
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	members := r.members[m.Channel.key()]
	if _, ok := members[m.Sender]; !ok {
		return Delivery{}, ErrNotMember
	}
	recipients := make([]string, 0, len(members))
	for role := range members {
		if role != m.Sender {
			recipients = append(recipients, role)
		}
	}
	if len(recipients) == 0 {
		return Delivery{}, ErrNoRecipient
	}
	sort.Strings(recipients)
	if prev, ok := r.last[m.Sender]; ok && prev.text == m.Text && m.At.Sub(prev.at) <= r.cfg.DuplicateWindow {
		return Delivery{}, ErrDuplicate
	}
	if !r.global.allow(m.Sender, m.At) {
		return Delivery{}, ErrRateLimited
	}
	if !r.scoped.allow(m.Sender+"|"+m.Channel.key(), m.At) {
		return Delivery{}, ErrRateLimited
	}
	switch r.mod.Inspect(m) {
	case Reject:
		return Delivery{}, ErrRejected
	case Review:
		r.pending++
		r.last[m.Sender] = stamp{text: m.Text, at: m.At}
		return Delivery{Message: m, Recipients: recipients, Review: true}, nil
	default:
		r.last[m.Sender] = stamp{text: m.Text, at: m.At}
		return Delivery{Message: m, Recipients: recipients}, nil
	}
}

// PendingReviews is the number of delivered lines waiting for async review.
func (r *Router) PendingReviews() int {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.pending
}

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