《游戏服务端编程实践》10.3.2 实现群聊与频道广播

// file: ch10/hub/group.go
// tier: L1
// verify: go test ./ch10/hub/

package hub

import (
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"
	"unicode/utf8"
)

var (
	ErrNoGroup     = errors.New("group chat: empty group id")
	ErrNotMember   = errors.New("group chat: sender is not a member")
	ErrEmptyText   = errors.New("group chat: empty text")
	ErrTextTooLong = errors.New("group chat: text too long")
	ErrRateLimited = errors.New("group chat: rate limited")
)

// Limits 控制单条长度、发言频率、离线队列深度和去重窗口。
type Limits struct {
	MaxTextRunes int
	Burst        int
	Window       time.Duration
	OfflineCap   int
	DedupeTTL    time.Duration
}

func DefaultLimits() Limits {
	return Limits{MaxTextRunes: 200, Burst: 5, Window: 2 * time.Second, OfflineCap: 50, DedupeTTL: 5 * time.Minute}
}

// ChatMessage 的 Seq 由聊天服分配，客户端消息 ID 只用于去重。
type ChatMessage struct {
	GroupID     string    `json:"group"`
	SenderID    uint64    `json:"sender"`
	ClientMsgID string    `json:"client_msg_id"`
	Text        string    `json:"text"`
	Seq         uint64    `json:"seq"`
	At          time.Time `json:"at"`
}

// MemberLink 是网关到客户端的下行通道，Push 必须非阻塞或自带异步队列。
type MemberLink interface {
	Push(ChatMessage)
}

type SendRequest struct {
	ClientMsgID string
	Text        string
}

type DeliveryReport struct {
	Seq       uint64
	Online    int
	Offline   int
	Duplicate bool
}

type rateWindow struct {
	start time.Time
	count int
}

type seenEntry struct {
	expires time.Time
	seq     uint64
}

// GroupChat 组合成员表、在线连接、离线队列和 Broker。
type GroupChat struct {
	broker *Broker
	limits Limits
	now    func() time.Time

	mu      sync.Mutex
	members map[string]map[uint64]struct{}
	online  map[uint64]MemberLink
	offline map[uint64][]ChatMessage
	dropped map[uint64]int
	rates   map[uint64]*rateWindow
	seen    map[string]seenEntry
	seq     uint64
}

func NewGroupChat(b *Broker, limits Limits) *GroupChat {
	return &GroupChat{
		broker:  b,
		limits:  limits,
		now:     time.Now,
		members: make(map[string]map[uint64]struct{}),
		online:  make(map[uint64]MemberLink),
		offline: make(map[uint64][]ChatMessage),
		dropped: make(map[uint64]int),
		rates:   make(map[uint64]*rateWindow),
		seen:    make(map[string]seenEntry),
	}
}

func groupChannel(groupID string) string { return "chat:guild:" + groupID }

// Join 校验组频道名后把玩家加入成员表，重复加入是幂等的。
func (g *GroupChat) Join(groupID string, playerIDs ...uint64) error {
	if groupID == "" {
		return ErrNoGroup
	}
	if err := ValidateChannel(groupChannel(groupID)); err != nil {
		return err
	}
	g.mu.Lock()
	defer g.mu.Unlock()
	if g.members[groupID] == nil {
		g.members[groupID] = make(map[uint64]struct{})
	}
	for _, pid := range playerIDs {
		g.members[groupID][pid] = struct{}{}
	}
	return nil
}

func (g *GroupChat) Leave(groupID string, playerID uint64) {
	g.mu.Lock()
	defer g.mu.Unlock()
	delete(g.members[groupID], playerID)
}

// Bind 绑定在线通道，并按顺序补发积压的离线消息。
func (g *GroupChat) Bind(playerID uint64, link MemberLink) int {
	g.mu.Lock()
	defer g.mu.Unlock()
	g.online[playerID] = link
	pending := g.offline[playerID]
	delete(g.offline, playerID)
	for _, msg := range pending {
		link.Push(msg)
	}
	return len(pending)
}

func (g *GroupChat) Unbind(playerID uint64) {
	g.mu.Lock()
	defer g.mu.Unlock()
	delete(g.online, playerID)
}

// Send 校验成员、长度、去重和限频后发布到 Broker，并投递给在线成员。
func (g *GroupChat) Send(groupID string, senderID uint64, req SendRequest) (DeliveryReport, error) {
	if groupID == "" {
		return DeliveryReport{}, ErrNoGroup
	}
	if err := ValidateChannel(groupChannel(groupID)); err != nil {
		return DeliveryReport{}, err
	}
	text := strings.TrimSpace(req.Text)
	if text == "" {
		return DeliveryReport{}, ErrEmptyText
	}
	if n := utf8.RuneCountInString(text); n > g.limits.MaxTextRunes {
		return DeliveryReport{}, fmt.Errorf("%w: %d > %d", ErrTextTooLong, n, g.limits.MaxTextRunes)
	}

	g.mu.Lock()
	defer g.mu.Unlock()
	now := g.now()
	if _, ok := g.members[groupID][senderID]; !ok {
		return DeliveryReport{}, ErrNotMember
	}
	key := ""
	if req.ClientMsgID != "" {
		key = fmt.Sprintf("%d:%s", senderID, req.ClientMsgID)
		if entry, ok := g.seen[key]; ok && now.Before(entry.expires) {
			return DeliveryReport{Seq: entry.seq, Duplicate: true}, nil
		}
	}
	if !g.allow(senderID, now) {
		return DeliveryReport{}, ErrRateLimited
	}

	g.seq++
	msg := ChatMessage{
		GroupID:     groupID,
		SenderID:    senderID,
		ClientMsgID: req.ClientMsgID,
		Text:        text,
		Seq:         g.seq,
		At:          now,
	}
	payload, err := json.Marshal(msg)
	if err != nil {
		return DeliveryReport{}, err
	}
	if _, err := g.broker.Publish(groupChannel(groupID), payload); err != nil {
		return DeliveryReport{}, err
	}
	if key != "" {
		g.seen[key] = seenEntry{expires: now.Add(g.limits.DedupeTTL), seq: msg.Seq}
	}

	report := DeliveryReport{Seq: msg.Seq}
	for pid := range g.members[groupID] {
		if link, ok := g.online[pid]; ok {
			link.Push(msg)
			report.Online++
			continue
		}
		g.enqueueOffline(pid, msg)
		report.Offline++
	}
	return report, nil
}

func (g *GroupChat) allow(playerID uint64, now time.Time) bool {
	w := g.rates[playerID]
	if w == nil || now.Sub(w.start) >= g.limits.Window {
		g.rates[playerID] = &rateWindow{start: now, count: 1}
		return true
	}
	if w.count >= g.limits.Burst {
		return false
	}
	w.count++
	return true
}

func (g *GroupChat) enqueueOffline(playerID uint64, msg ChatMessage) {
	q := append(g.offline[playerID], msg)
	if len(q) > g.limits.OfflineCap {
		drop := len(q) - g.limits.OfflineCap
		q = append([]ChatMessage(nil), q[drop:]...)
		g.dropped[playerID] += drop
	}
	g.offline[playerID] = q
}

// Offline 返回玩家当前的离线消息副本。
func (g *GroupChat) Offline(playerID uint64) []ChatMessage {
	g.mu.Lock()
	defer g.mu.Unlock()
	return append([]ChatMessage(nil), g.offline[playerID]...)
}

// DroppedOffline 返回因超出 OfflineCap 被丢弃的消息数。
func (g *GroupChat) DroppedOffline(playerID uint64) int {
	g.mu.Lock()
	defer g.mu.Unlock()
	return g.dropped[playerID]
}

cd .local/checks/gs-practice-lab && gofmt -l ch10/hub && go vet ./ch10/hub/ && go test ./ch10/hub/ -count=1

ok  	gspractice/ch10/hub	0.275s