《游戏服务端架构·核心系统与契约》10.1 邮件投递与去重

dedup_key = event_id + 0x1f + template + 0x1f + recipient

全服邮件扇出
  snapshot recipients -> sort -> cursor
       |
       v
  chunk(1000) -> enqueue jobs -> persist cursor
       |
       v
  workers -> per-recipient dedup key -> mail row
       |
       +--> retry with backoff --> dead letter

// file: internal/maildelivery/queue.go
// tier: L1
// verify: go test ./internal/maildelivery/
package maildelivery

import (
	"sort"
	"time"
)

// Queue is an in-memory delivery queue. Production keeps the same state
// machine on a durable store, where the dedupe set and the pending job commit
// in one transaction.
type Queue struct {
	opts     Options
	pending  map[string]Job
	inFlight map[string]Job
	seen     map[string]struct{}
	dead     map[string]DeadLetter
	stats    Stats
}

func NewQueue(opts Options) *Queue {
	def := DefaultOptions()
	if opts.MaxAttempts <= 0 {
		opts.MaxAttempts = def.MaxAttempts
	}
	if opts.BaseBackoff <= 0 {
		opts.BaseBackoff = def.BaseBackoff
	}
	if opts.MaxBackoff < opts.BaseBackoff {
		opts.MaxBackoff = opts.BaseBackoff
	}
	return &Queue{
		opts:     opts,
		pending:  make(map[string]Job),
		inFlight: make(map[string]Job),
		seen:     make(map[string]struct{}),
		dead:     make(map[string]DeadLetter),
	}
}

// DedupKey includes the recipient because one event fans out to many
// mailboxes. The separator is ASCII unit separator, not a legal player ID.
func DedupKey(eventID, template, recipient string) string {
	return eventID + "\x1f" + template + "\x1f" + recipient
}

func validate(j Job) error {
	if j.Key == "" {
		return ErrEmptyKey
	}
	if j.Recipient == "" {
		return ErrEmptyRecipient
	}
	if j.Template == "" {
		return ErrEmptyTemplate
	}
	return nil
}

// Enqueue inserts a job unless its key was seen. The seen set covers pending,
// in-flight and delivered keys, so a retry after Ack is also suppressed.
func (q *Queue) Enqueue(j Job) (Result, error) {
	if err := validate(j); err != nil {
		return 0, err
	}
	if _, ok := q.seen[j.Key]; ok {
		q.stats.Duplicates++
		return ResultDuplicate, nil
	}
	q.pending[j.Key] = j
	q.seen[j.Key] = struct{}{}
	q.stats.Enqueued++
	return ResultEnqueued, nil
}

// Lease picks one due job. Ties break by key so replays are deterministic.
func (q *Queue) Lease(now time.Time) (Job, bool) {
	var bestKey string
	var best Job
	found := false
	for key, job := range q.pending {
		if job.NextAt.After(now) {
			continue
		}
		if !found || job.NextAt.Before(best.NextAt) || (job.NextAt.Equal(best.NextAt) && key < bestKey) {
			bestKey, best, found = key, job, true
		}
	}
	if !found {
		return Job{}, false
	}
	delete(q.pending, bestKey)
	q.inFlight[bestKey] = best
	return best, true
}

// Ack marks an in-flight job delivered. The key stays in the seen set.
func (q *Queue) Ack(key string) error {
	if _, ok := q.inFlight[key]; !ok {
		return ErrNotInFlight
	}
	delete(q.inFlight, key)
	q.stats.Delivered++
	return nil
}

// Nack returns a job to the queue or moves it to the dead-letter set.
func (q *Queue) Nack(key string, now time.Time, cause string) (bool, error) {
	job, ok := q.inFlight[key]
	if !ok {
		return false, ErrNotInFlight
	}
	delete(q.inFlight, key)
	job.Attempts++
	if job.Attempts >= q.opts.MaxAttempts {
		q.dead[key] = DeadLetter{Job: job, Cause: cause, At: now}
		q.stats.DeadLettered++
		return false, nil
	}
	job.NextAt = now.Add(q.backoff(job.Attempts))
	q.pending[key] = job
	q.stats.Retried++
	return true, nil
}

func (q *Queue) backoff(attempts int) time.Duration {
	if attempts <= 1 {
		return q.opts.BaseBackoff
	}
	if attempts > 20 {
		return q.opts.MaxBackoff
	}
	d := q.opts.BaseBackoff << (attempts - 1)
	if d <= 0 || d > q.opts.MaxBackoff {
		return q.opts.MaxBackoff
	}
	return d
}

// DeadLetters returns a stable copy of the dead-letter set.
func (q *Queue) DeadLetters() []DeadLetter {
	out := make([]DeadLetter, 0, len(q.dead))
	for _, d := range q.dead {
		out = append(out, d)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Job.Key < out[j].Job.Key })
	return out
}

func (q *Queue) Stats() Stats {
	return q.stats
}

// file: internal/maildelivery/fanout.go
// tier: L1
// verify: go test ./internal/maildelivery/
package maildelivery

import "sort"

// Fanout is a resumable cursor for global mail. The caller persists the cursor
// after each chunk so a restart resumes instead of resending every recipient.
type Fanout struct {
	Template   string
	Batch      string
	Recipients []string
	cursor     int
}

func NewFanout(template, batch string, recipients []string) (*Fanout, error) {
	if template == "" {
		return nil, ErrEmptyTemplate
	}
	if batch == "" {
		return nil, ErrEmptyKey
	}
	seen := make(map[string]struct{}, len(recipients))
	out := make([]string, 0, len(recipients))
	for _, r := range recipients {
		if r == "" {
			return nil, ErrEmptyRecipient
		}
		if _, ok := seen[r]; ok {
			continue
		}
		seen[r] = struct{}{}
		out = append(out, r)
	}
	sort.Strings(out)
	return &Fanout{Template: template, Batch: batch, Recipients: out}, nil
}

// Next returns at most limit jobs and advances the cursor.
func (f *Fanout) Next(limit int) []Job {
	if limit <= 0 || f.cursor >= len(f.Recipients) {
		return nil
	}
	end := f.cursor + limit
	if end > len(f.Recipients) {
		end = len(f.Recipients)
	}
	jobs := make([]Job, 0, end-f.cursor)
	for _, r := range f.Recipients[f.cursor:end] {
		jobs = append(jobs, Job{
			Key:       DedupKey(f.Batch, f.Template, r),
			Recipient: r,
			Template:  f.Template,
			Batch:     f.Batch,
		})
	}
	f.cursor = end
	return jobs
}

func (f *Fanout) Done() bool {
	return f.cursor >= len(f.Recipients)
}

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