《MMO 服务器开发》11.2 敏感词与限速治理

// file: internal/chat/filter.go
// tier: L1
// verify: go test ./internal/chat/

package chat

import "sort"

// Filter 是基于 AC 自动机（Aho–Corasick）的多模式匹配器。
// 一次扫描就能找出文本里命中的所有敏感词，复杂度是 O(文本长 + 命中数)。
type Filter struct {
	root *node
}

type node struct {
	next map[rune]*node
	fail *node
	out  []string // 走到本节点时命中的模式（含沿 fail 链继承的）
}

// Match 描述一次命中：Word 是规范形后的词，[Start, End) 是原文的 rune 区间。
type Match struct {
	Word  string
	Start int
	End   int
}

// normalize 把字符折叠成比较用的规范形：全角转半角、大写转小写。
// 它必须「一个 rune 进、一个 rune 出」，匹配下标才能直接映射回原文。
func normalize(r rune) rune {
	// 先折叠全角，再统一大小写，两步都必须是 1:1 的 rune 映射。
	switch {
	case r >= 0xFF01 && r <= 0xFF5E: // 全角 ASCII
		r -= 0xFEE0
	case r == 0x3000: // 全角空格
		r = ' '
	}
	if r >= 'A' && r <= 'Z' {
		return r + ('a' - 'A')
	}
	return r
}

func normRunes(s string) []rune {
	rs := []rune(s)
	for i, r := range rs {
		rs[i] = normalize(r)
	}
	return rs
}

// NewFilter 用敏感词表构建自动机。空词与重复词会被忽略。
func NewFilter(words []string) *Filter {
	root := &node{next: map[rune]*node{}}
	for _, w := range words {
		nw := normRunes(w)
		if len(nw) == 0 {
			continue
		}
		cur := root
		for _, r := range nw {
			child := cur.next[r]
			if child == nil {
				child = &node{next: map[rune]*node{}}
				cur.next[r] = child
			}
			cur = child
		}
		cur.out = append(cur.out, string(nw))
	}
	// BFS 建失败指针：与 KMP 的 next 数组同源，只是从「一条链」扩到「一棵树」。
	queue := make([]*node, 0, len(root.next))
	for _, child := range root.next {
		child.fail = root
		queue = append(queue, child)
	}
	for len(queue) > 0 {
		cur := queue[0]
		queue = queue[1:]
		for r, child := range cur.next {
			f := cur.fail
			for f != nil && f.next[r] == nil {
				f = f.fail
			}
			if f == nil {
				child.fail = root
			} else {
				child.fail = f.next[r]
			}
			child.out = append(child.out, child.fail.out...)
			queue = append(queue, child)
		}
	}
	return &Filter{root: root}
}

// Match 返回文本里命中的所有敏感词及其在原文中的 rune 下标区间 [Start, End)。
func (f *Filter) Match(text string) []Match {
	rs := normRunes(text)
	var out []Match
	cur := f.root
	for i, r := range rs {
		for cur != f.root && cur.next[r] == nil {
			cur = cur.fail
		}
		if n := cur.next[r]; n != nil {
			cur = n
		} else {
			cur = f.root
		}
		for _, w := range cur.out {
			out = append(out, Match{Word: w, Start: i - len([]rune(w)) + 1, End: i + 1})
		}
	}
	return out
}

// Replace 把命中的区间整段替换成 mask。重叠区间只屏蔽一次。
func (f *Filter) Replace(text string, mask rune) string {
	ms := f.Match(text)
	if len(ms) == 0 {
		return text
	}
	rs := []rune(text)
	for _, sp := range mergeSpans(ms) {
		for i := sp[0]; i < sp[1]; i++ {
			rs[i] = mask
		}
	}
	return string(rs)
}

// mergeSpans 把可能重叠的命中区间合并成互不相交的区间。
func mergeSpans(ms []Match) [][2]int {
	spans := make([][2]int, 0, len(ms))
	for _, m := range ms {
		spans = append(spans, [2]int{m.Start, m.End})
	}
	sort.Slice(spans, func(i, j int) bool {
		if spans[i][0] != spans[j][0] {
			return spans[i][0] < spans[j][0]
		}
		return spans[i][1] < spans[j][1]
	})
	var merged [][2]int
	for _, s := range spans {
		if n := len(merged); n > 0 && s[0] <= merged[n-1][1] {
			if s[1] > merged[n-1][1] {
				merged[n-1][1] = s[1]
			}
			continue
		}
		merged = append(merged, s)
	}
	return merged
}

// file: internal/chat/ratelimit.go
// tier: L1
// verify: go test ./internal/chat/

package chat

import (
	"math"
	"sync"
	"time"
)

type bucketKey struct {
	Player PlayerID
	Kind   ChannelKind
}

type bucket struct {
	tokens float64
	last   time.Time
}

// RateLimiter 是「每玩家 × 每频道」的令牌桶限速器。
// rate 是每秒补充的令牌数，burst 是桶容量（允许的瞬时突发）。
type RateLimiter struct {
	mu      sync.Mutex
	rate    float64
	burst   float64
	buckets map[bucketKey]*bucket
	now     func() time.Time
}

func NewRateLimiter(rate, burst float64) *RateLimiter {
	return &RateLimiter{
		rate:    rate,
		burst:   burst,
		buckets: map[bucketKey]*bucket{},
		now:     time.Now,
	}
}

// Allow 尝试从桶里取一个令牌。失败时返回还需等待的时长。
func (l *RateLimiter) Allow(p PlayerID, k ChannelKind) (bool, time.Duration) {
	l.mu.Lock()
	defer l.mu.Unlock()
	now := l.now()
	key := bucketKey{p, k}
	b := l.buckets[key]
	if b == nil {
		b = &bucket{tokens: l.burst, last: now}
		l.buckets[key] = b
	}
	if elapsed := now.Sub(b.last).Seconds(); elapsed > 0 {
		b.tokens = math.Min(l.burst, b.tokens+elapsed*l.rate)
		b.last = now
	}
	if b.tokens >= 1 {
		b.tokens--
		return true, 0
	}
	wait := time.Duration((1 - b.tokens) / l.rate * float64(time.Second))
	return false, wait
}

// Sweep 丢弃空闲超过 idle 的桶，避免长期运行后 map 无限增长。
func (l *RateLimiter) Sweep(idle time.Duration) int {
	l.mu.Lock()
	defer l.mu.Unlock()
	now := l.now()
	n := 0
	for k, b := range l.buckets {
		if now.Sub(b.last) > idle {
			delete(l.buckets, k)
			n++
		}
	}
	return n
}

// snippet: internal/chat/moderator.go（三级治理管线，节选）

// Verdict 是一条消息经过治理管线后的处置结论。
type Verdict uint8

const (
	VerdictPass     Verdict = iota // 原样放行
	VerdictMasked                  // 命中敏感词，替换后放行
	VerdictRejected                // 直接拒绝，不投递
)

// Moderate 返回处置结论、处理后的正文与拒绝原因。
func (m *Moderator) Moderate(p PlayerID, k ChannelKind, text string) (Verdict, string, string) {
	m.mu.Lock()
	until, muted := m.muted[p]
	m.mu.Unlock()
	// 第一级：禁言。最便宜，且与频道无关。
	if muted && m.now().Before(until) {
		return VerdictRejected, "", "muted until " + until.Format(time.RFC3339)
	}
	// 第二级：限速。桶空时直接拒绝，并告诉客户端还要等多久。
	if ok, wait := m.Limiter.Allow(p, k); !ok {
		return VerdictRejected, "", "rate limited, retry in " + wait.Round(time.Millisecond).String()
	}
	// 第三级：过滤。命中就替换后放行，而不是整条丢弃——正常玩家不该被误伤。
	if ms := m.Filter.Match(text); len(ms) > 0 {
		return VerdictMasked, m.Filter.Replace(text, '*'), "masked " + strconv.Itoa(len(ms)) + " words"
	}
	return VerdictPass, text, ""
}