feat(M3): 申请审批 + 邮件队列 — 公开申请、管理员审批自动建号、mail_logs 落库重试

2026-08-30 10:01:38 +08:00
parent a5f501dba4
commit 1ea18490e0
13 changed files with 830 additions and 27 deletions
@@ -1,6 +1,6 @@
-// Package mail 提供邮件发送抽象。M1 提供基础直发（net/smtp + STARTTLS），
-// 发送队列/重试/失败记录（mail_logs）在 M3 完善；SMTP 未配置时退化为
-// LogMailer（仅打印，不阻断业务——OTP 邮件失败不阻断 CLI 通道）。
+// Package mail 提供邮件发送抽象。M1 提供基础直发（net/smtp + STARTTLS）；
+// M3 引入 QueuedMailer（mail_logs 队列 + 失败重试，见 queue.go），
+// SMTP 未配置时退化为 LogMailer（仅打印，不阻断业务——OTP 邮件失败不阻断 CLI 通道）。
 package mail

 import (
@@ -0,0 +1,84 @@
+package mail
+
+import (
+	"context"
+	"log/slog"
+
+	"gorm.io/gorm"
+
+	"ws_usernode/internal/model"
+)
+
+// 邮件状态（mail_logs.status）。
+const (
+	StatusPending = "pending" // 已入队，等待直发
+	StatusSent    = "sent"    // 发送成功
+	StatusFailed  = "failed"  // 发送失败，等待重试
+)
+
+// MaxRetries 单封邮件最大重试次数；超出后停止重试，保留失败记录（PLAN F5）。
+const MaxRetries = 5
+
+// Retryable 可由 cron 定时调用的邮件队列重试接口。
+type Retryable interface {
+	Retry(ctx context.Context, limit int) error
+}
+
+// QueuedMailer 邮件队列实现（PLAN F5）：每封邮件先落 mail_logs 再同步尝试直发；
+// 失败记录 error 与 retry_count，由 cron 定时重试；成功标记 sent。
+// Send 恒返回 nil（除 DB 落库失败）：邮件失败不阻断业务
+// （OTP 双通道、审批结果等，见 AuthService.UserOTPSend 的既有约定）。
+type QueuedMailer struct {
+	base Mailer
+	db   *gorm.DB
+	log  *slog.Logger
+}
+
+// NewQueuedMailer 创建邮件队列，包装底层直发实现（SMTP 或 LogMailer）。
+func NewQueuedMailer(base Mailer, db *gorm.DB, log *slog.Logger) *QueuedMailer {
+	return &QueuedMailer{base: base, db: db, log: log}
+}
+
+// Send 入队并尝试直发。返回 nil 表示已受理；实际送达与否见 mail_logs。
+func (m *QueuedMailer) Send(ctx context.Context, to, subject, body string) error {
+	entry := model.MailLog{To: to, Subject: subject, Body: body, Status: StatusPending}
+	if err := m.db.WithContext(ctx).Create(&entry).Error; err != nil {
+		m.log.Error("mail: 记录 mail_log 失败", "err", err)
+		return err
+	}
+	if err := m.base.Send(ctx, to, subject, body); err != nil {
+		_ = m.db.WithContext(ctx).Model(&entry).Updates(map[string]any{
+			"status": StatusFailed, "error": err.Error(), "retry_count": 1,
+		}).Error
+		m.log.Warn("mail: 直发失败，进入重试队列", "to", to, "subject", subject, "err", err)
+		return nil // 非阻断
+	}
+	_ = m.db.WithContext(ctx).Model(&entry).Update("status", StatusSent).Error
+	return nil
+}
+
+// Retry 重试失败邮件（retry_count < limit，按原内容重发）。成功标记 sent，
+// 失败累加 retry_count 并刷新 error；达到上限后不再挑选，保留失败记录待人工处理。
+func (m *QueuedMailer) Retry(ctx context.Context, limit int) error {
+	if limit <= 0 {
+		return nil
+	}
+	var rows []model.MailLog
+	if err := m.db.WithContext(ctx).
+		Where("status = ? AND retry_count < ?", StatusFailed, limit).
+		Find(&rows).Error; err != nil {
+		return err
+	}
+	for _, r := range rows {
+		if err := m.base.Send(ctx, r.To, r.Subject, r.Body); err != nil {
+			_ = m.db.WithContext(ctx).Model(&model.MailLog{}).Where("id = ?", r.ID).Updates(map[string]any{
+				"retry_count": r.RetryCount + 1, "error": err.Error(),
+			}).Error
+			m.log.Warn("mail: 重试仍失败", "id", r.ID, "to", r.To, "retry_count", r.RetryCount+1, "err", err)
+			continue
+		}
+		_ = m.db.WithContext(ctx).Model(&model.MailLog{}).Where("id = ?", r.ID).Update("status", StatusSent).Error
+		m.log.Info("mail: 重试成功", "id", r.ID, "to", r.To)
+	}
+	return nil
+}
@@ -0,0 +1,130 @@
+package mail
+
+import (
+	"context"
+	"errors"
+	"io"
+	"log/slog"
+	"testing"
+
+	"gorm.io/gorm"
+
+	"ws_usernode/internal/model"
+)
+
+// flakyMailer 前 n 次发送失败，之后成功；用于验证重试。
+type flakyMailer struct {
+	failures int
+	calls    int
+}
+
+var errFlaky = errors.New("mail: smtp down")
+
+func (m *flakyMailer) Send(_ context.Context, _, _, _ string) error {
+	m.calls++
+	if m.calls <= m.failures {
+		return errFlaky
+	}
+	return nil
+}
+
+// recordingMailer 记录直发调用次数。
+type recordingMailer struct{ calls int }
+
+func (m *recordingMailer) Send(context.Context, string, string, string) error {
+	m.calls++
+	return nil
+}
+
+func testQueued(t *testing.T, base Mailer) (*QueuedMailer, *gorm.DB) {
+	t.Helper()
+	db, err := model.Open("sqlite", ":memory:", false)
+	if err != nil {
+		t.Fatalf("open test db: %v", err)
+	}
+	if err := db.AutoMigrate(&model.MailLog{}); err != nil {
+		t.Fatalf("migrate mail_logs: %v", err)
+	}
+	log := slog.New(slog.NewTextHandler(io.Discard, nil))
+	return NewQueuedMailer(base, db, log), db
+}
+
+func TestQueuedMailerSendSuccess(t *testing.T) {
+	rec := &recordingMailer{}
+	q, db := testQueued(t, rec)
+	if err := q.Send(context.Background(), "a@example.com", "主题", "正文"); err != nil {
+		t.Fatalf("send: %v", err)
+	}
+	if rec.calls != 1 {
+		t.Fatalf("base send calls = %d, want 1", rec.calls)
+	}
+	var n int64
+	if err := db.Model(&model.MailLog{}).Where("status = ?", StatusSent).Count(&n).Error; err != nil {
+		t.Fatalf("count sent: %v", err)
+	}
+	if n != 1 {
+		t.Fatalf("sent = %d, want 1", n)
+	}
+}
+
+func TestQueuedMailerFailureQueuedAndRetry(t *testing.T) {
+	base := &flakyMailer{failures: 2}
+	q, db := testQueued(t, base)
+
+	// 第一次发送失败：Send 仍返回 nil（不阻断业务），落 failed 记录
+	if err := q.Send(context.Background(), "b@example.com", "s", "body"); err != nil {
+		t.Fatalf("send returned error: %v", err)
+	}
+	if base.calls != 1 {
+		t.Fatalf("base calls = %d, want 1", base.calls)
+	}
+	var row model.MailLog
+	if err := db.First(&row, "status = ?", StatusFailed).Error; err != nil {
+		t.Fatalf("failed row: %v", err)
+	}
+	if row.RetryCount != 1 || row.Error != errFlaky.Error() {
+		t.Fatalf("failed row = %+v", row)
+	}
+
+	// Retry 1：仍失败 → retry_count=2；Retry 2：成功 → sent
+	if err := q.Retry(context.Background(), 5); err != nil {
+		t.Fatalf("retry: %v", err)
+	}
+	if base.calls != 2 {
+		t.Fatalf("calls after retry1 = %d, want 2", base.calls)
+	}
+	if err := q.Retry(context.Background(), 5); err != nil {
+		t.Fatalf("retry2: %v", err)
+	}
+	if base.calls != 3 {
+		t.Fatalf("calls after retry2 = %d, want 3", base.calls)
+	}
+	var n int64
+	if err := db.Model(&model.MailLog{}).Where("status = ?", StatusSent).Count(&n).Error; err != nil {
+		t.Fatalf("count sent: %v", err)
+	}
+	if n != 1 {
+		t.Fatalf("sent = %d, want 1", n)
+	}
+}
+
+func TestQueuedMailerRetryHitsMax(t *testing.T) {
+	base := &flakyMailer{failures: 100} // 永远失败
+	q, db := testQueued(t, base)
+	if err := q.Send(context.Background(), "c@example.com", "s", "body"); err != nil {
+		t.Fatalf("send: %v", err)
+	}
+	// 达到上限后 Retry 不再挑选（retry_count 不再增长）
+	for i := 0; i < 10; i++ {
+		if err := q.Retry(context.Background(), MaxRetries); err != nil {
+			t.Fatalf("retry: %v", err)
+		}
+	}
+	var row model.MailLog
+	if err := db.First(&row, "status = ?", StatusFailed).Error; err != nil {
+		t.Fatalf("failed row: %v", err)
+	}
+	if row.RetryCount > MaxRetries {
+		t.Fatalf("retry_count = %d, want <= %d", row.RetryCount, MaxRetries)
+	}
+}