becomer/likedream
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: refactor email handling and introduce task queue for email notifications

Browse Source
This commit is contained in:
RecordingTheSmile
2025-08-22 23:21:14 +08:00
parent de33bdce72
commit 985b2921d7
15 changed files with 187 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -24,6 +24,7 @@ require (
github.com/spf13/viper v1.20.1
github.com/tidwall/gjson v1.18.0
github.com/tidwall/match v1.1.1
github.com/wneessen/go-mail v0.6.2
go.uber.org/automaxprocs v1.6.0
go.uber.org/zap v1.27.0
golang.org/x/crypto v0.37.0
@@ -84,7 +85,6 @@ require (
github.com/subosito/gotenv v1.6.0 // indirect
github.com/tidwall/pretty v1.2.0 // indirect
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
github.com/wneessen/go-mail v0.6.2 // indirect
github.com/zclconf/go-cty v1.14.4 // indirect
github.com/zclconf/go-cty-yaml v1.1.0 // indirect
go.opentelemetry.io/otel v1.32.0 // indirect
go.sum
-4
View File
@@ -108,8 +108,6 @@ github.com/hashicorp/hcl/v2 v2.13.0 h1:0Apadu1w6M11dyGFxWnmhhcMjkbAiKCv7G1r/2QgC
github.com/hashicorp/hcl/v2 v2.13.0/go.mod h1:e4z5nxYlWNPdDSNYX+ph14EvWYMFm3eP0zIUqPc2jr0=
github.com/hibiken/asynq v0.25.1 h1:phj028N0nm15n8O2ims+IvJ2gz4k2auvermngh9JhTw=
github.com/hibiken/asynq v0.25.1/go.mod h1:pazWNOLBu0FEynQRBvHA26qdIKRSmfdIfUm4HdsLmXg=
github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
github.com/joomcode/errorx v1.2.0 h1:7Y/fguon+9r6a/75Rv3nrUwS7nXNEcJjLShjCvz00Og=
github.com/joomcode/errorx v1.2.0/go.mod h1:Mbz68VA9hsQLT50iCQQUZ2Z1XYAKYB4EoFkFCTFyiJM=
github.com/klauspost/cpuid/v2 v2.0.9 h1:lgaqFMSdTdQYdZ04uHyN2d/eKdOMyi2YLSvlQIBFYa4=
@@ -171,8 +169,6 @@ github.com/spf13/afero v1.12.0 h1:UcOPyRBYczmFn6yvphxkn9ZEOY65cpwGKb5mL36mrqs=
github.com/spf13/afero v1.12.0/go.mod h1:ZTlWwG4/ahT8W7T0WQ5uYmjI9duaLQGy3Q2OAl4sk/4=
github.com/spf13/cast v1.9.2 h1:SsGfm7M8QOFtEzumm7UZrZdLLquNdzFYfIbEXntcFbE=
github.com/spf13/cast v1.9.2/go.mod h1:jNfB8QC9IA6ZuY2ZjDp0KtFO2LZZlg4S/7bzP6qqeHo=
github.com/spf13/cobra v1.7.0 h1:hyqWnYt1ZQShIddO5kBpj3vu05/++x6tJ6dg8EC572I=
github.com/spf13/cobra v1.7.0/go.mod h1:uLxZILRyS/50WlhOIKD7W6V5bgeIt+4sICxh6uRMrb0=
github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o=
github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
github.com/spf13/viper v1.20.1 h1:ZMi+z/lvLyPSCoNtFCpqjy0S4kPbirhpTMwl8BkW9X4=
internal/cst/task.go
+2
View File
@@ -2,6 +2,8 @@ package cst
const (
TaskCpdaily = "task:cpdaily"
TaskEmail = "task:email"
TaskQueueCpdaily = "queue:cpdaily"
TaskQueueEmail = "queue:email"
)
internal/model/email/payload.go
+13
View File
@@ -0,0 +1,13 @@
package email
type SendPayload struct {
To string
Subject string
Body string
FromName string
FromEmail string
SMTPHost string
SMTPPort int
SMTPUsername string
SMTPPassword string
}
internal/model/email/status.go
+59
View File
@@ -0,0 +1,59 @@
package email
// TaskStatus 任务状态枚举
type TaskStatus string
const (
TaskStatusSuccess TaskStatus = "success" // 成功
TaskStatusFailed TaskStatus = "failed" // 失败
TaskStatusRunning TaskStatus = "running" // 运行中
TaskStatusPending TaskStatus = "pending" // 等待中
)
// GetStatusColor 获取状态对应的颜色
func (s TaskStatus) GetStatusColor() string {
switch s {
case TaskStatusSuccess:
return "#10b981" // 绿色
case TaskStatusFailed:
return "#ef4444" // 红色
case TaskStatusRunning:
return "#3b82f6" // 蓝色
case TaskStatusPending:
return "#f59e0b" // 橙色
default:
return "#6b7280" // 灰色
}
}
// GetStatusText 获取状态文本
func (s TaskStatus) GetStatusText() string {
switch s {
case TaskStatusSuccess:
return "执行成功"
case TaskStatusFailed:
return "执行失败"
case TaskStatusRunning:
return "执行中"
case TaskStatusPending:
return "等待执行"
default:
return "未知状态"
}
}
// GetStatusIcon 获取状态图标
func (s TaskStatus) GetStatusIcon() string {
switch s {
case TaskStatusSuccess:
return "✅"
case TaskStatusFailed:
return "❌"
case TaskStatusRunning:
return "🔄"
case TaskStatusPending:
return "⏳"
default:
return "📬"
}
}
internal/queue/task.go
+27
View File
@@ -0,0 +1,27 @@
package queue
import (
"context"
"likedream-goframe/internal/redis"
"likedream-goframe/internal/utils"
"github.com/hibiken/asynq"
"github.com/samber/oops"
)
var globalClient *asynq.Client
func AddTask(ctx context.Context, task *asynq.Task, opts ...asynq.Option) *asynq.TaskInfo {
taskInfo, e := globalClient.EnqueueContext(ctx, task, opts...)
utils.OkOrPanic(e, oops.WithContext(ctx).With("taskType", task.Type(), "taskPayload", task.Payload()).Wrap)
return taskInfo
}
func InitClient(_ context.Context) {
globalClient = asynq.NewClientFromRedisClient(redis.GetClient())
}
func GetClient() *asynq.Client {
return globalClient
}
internal/schedule/client.go
-13
View File
@@ -1,13 +0,0 @@
package schedule
import (
"context"
"github.com/hibiken/asynq"
"likedream-goframe/internal/redis"
)
var globalClient *asynq.Client
func initClient(ctx context.Context) {
globalClient = asynq.NewClientFromRedisClient(redis.GetClient())
}
internal/schedule/handler.go → internal/schedule/cpdaily_handler.go
+3 -2
View File
@@ -14,7 +14,8 @@ import (
"likedream-goframe/internal/ent/cpuser"
"likedream-goframe/internal/ent/sysuser"
"likedream-goframe/internal/framework/errors"
"likedream-goframe/internal/service/email"
"likedream-goframe/internal/model/email"
"likedream-goframe/internal/service"
"likedream-goframe/internal/utils"
"time"
@@ -204,5 +205,5 @@ func (c *CpdailyTaskHandler) sendTaskNotificationEmail(ctx context.Context, task
return // 没有邮箱地址，不发送邮件
}
email.NewV1().SendTaskNotificationEmail(ctx, emailAddr, taskName, status, details)
service.Global.Email.SendTaskNotificationEmail(ctx, emailAddr, taskName, status, details)
}
internal/schedule/email_handler.go
+48
View File
@@ -0,0 +1,48 @@
package schedule
import (
"bytes"
"context"
"encoding/gob"
"likedream-goframe/internal/model/email"
"likedream-goframe/internal/utils"
"github.com/hibiken/asynq"
"github.com/samber/oops"
"github.com/wneessen/go-mail"
)
type EmailTaskHandler struct {
}
func (h *EmailTaskHandler) ProcessTask(ctx context.Context, task *asynq.Task) (err error) {
defer utils.Pass(&err)
var payload email.SendPayload
e := gob.NewDecoder(bytes.NewReader(task.Payload())).Decode(&payload)
utils.OkOrPanic(e, oops.WithContext(ctx).Wrap)
message := mail.NewMsg()
e = message.FromFormat(payload.FromName, payload.FromEmail)
utils.OkOrPanic(e, oops.WithContext(ctx).Wrap)
e = message.AddTo(payload.To)
utils.OkOrPanic(e, oops.WithContext(ctx).Wrap)
message.Subject(payload.Subject)
message.SetBodyString(mail.TypeTextHTML, payload.Body)
client, e := mail.NewClient(payload.SMTPHost, mail.WithPort(payload.SMTPPort), mail.WithSMTPAuth(mail.SMTPAuthAutoDiscover),
mail.WithUsername(payload.SMTPUsername), mail.WithPassword(payload.SMTPPassword))
utils.OkOrPanic(e, oops.WithContext(ctx).Wrap)
defer utils.LogError(ctx, "关闭邮件客户端失败", client.Close())
e = client.DialAndSendWithContext(ctx, message)
utils.OkOrPanic(e, oops.WithContext(ctx).Wrap)
return
}
internal/schedule/init.go
+3 -2
View File
@@ -2,11 +2,12 @@ package schedule
import (
"context"
"likedream-goframe/internal/queue"
"likedream-goframe/internal/utils"
)
func Init(ctx context.Context) {
initClient(ctx)
queue.InitClient(ctx)
initServer(ctx)
initInspector(ctx)
initScanner()
@@ -15,7 +16,7 @@ func Init(ctx context.Context) {
func Stop() {
globalScanner.StopScan()
e := globalClient.Close()
e := queue.GetClient().Close()
utils.OkOrPanic(e)
globalServer.Shutdown()
internal/schedule/scanner.go
+2 -1
View File
@@ -10,6 +10,7 @@ import (
"likedream-goframe/internal/db"
"likedream-goframe/internal/ent"
"likedream-goframe/internal/ent/cptask"
"likedream-goframe/internal/queue"
"likedream-goframe/internal/utils"
"slices"
"strconv"
@@ -97,7 +98,7 @@ func (ts *TaskScanner) scanCpdailyTask() {
execTime := time.Date(year, month, day, *task.ExecuteHour, *task.ExecuteMinute, 0, 0, cst.TimeZoneShangHai)
e = utils.Catch(func() {
AddTask(ts.ctx, asynq.NewTask(cst.TaskCpdaily, payloadWriter.Bytes()),
queue.AddTask(ts.ctx, asynq.NewTask(cst.TaskCpdaily, payloadWriter.Bytes()),
asynq.TaskID(fmt.Sprintf("cpdaily:task:%d", task.ID)),
asynq.MaxRetry(5),
asynq.Unique(10*time.Minute),
internal/schedule/server.go
+1
View File
@@ -33,6 +33,7 @@ func initServer(ctx context.Context) {
mux := asynq.NewServeMux()
mux.Handle(cst.TaskCpdaily, &CpdailyTaskHandler{})
mux.Handle(cst.TaskEmail, &EmailTaskHandler{})
e := globalServer.Start(mux)
utils.OkOrPanic(e)
internal/schedule/task.go
-9
View File
@@ -4,18 +4,9 @@ import (
"context"
"likedream-goframe/internal/utils"
"github.com/hibiken/asynq"
"github.com/samber/oops"
)
// AddTask 添加一个任务到任务队列
func AddTask(ctx context.Context, task *asynq.Task, opts ...asynq.Option) *asynq.TaskInfo {
taskInfo, e := globalClient.EnqueueContext(ctx, task, opts...)
utils.OkOrPanic(e, oops.WithContext(ctx).With("taskType", task.Type(), "taskPayload", task.Payload()).Wrap)
return taskInfo
}
// StopTask 让任务队列停止执行某任务
//
// 注意，只有当任务处于等待执行、失败、成功和已排队状态时才能执行成功，否则会返回错误。
internal/service/email/email_service.go
+24 -18
View File
@@ -4,16 +4,20 @@ import (
"bytes"
"context"
"embed"
"encoding/gob"
"fmt"
"html/template"
"likedream-goframe/internal/cst"
"likedream-goframe/internal/model/email"
"likedream-goframe/internal/queue"
"likedream-goframe/internal/utils"
"sync"
"time"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/os/gctx"
"github.com/hibiken/asynq"
"github.com/samber/oops"
"github.com/wneessen/go-mail"
)
//go:embed templates/*.gohtml
@@ -73,7 +77,7 @@ func (s *EmailServiceV1) SendSignupEmail(ctx context.Context, toEmail, username
}
// SendTaskNotificationEmail 发送任务状态通知邮件
func (s *EmailServiceV1) SendTaskNotificationEmail(ctx context.Context, toEmail, taskName string, status TaskStatus, details map[string]string) *utils.Response[any] {
func (s *EmailServiceV1) SendTaskNotificationEmail(ctx context.Context, toEmail, taskName string, status email.TaskStatus, details map[string]string) *utils.Response[any] {
if s.config.SMTPUsername == "" || s.config.SMTPPassword == "" {
return utils.FailBusiness("邮件服务未配置")
}
@@ -94,24 +98,26 @@ func (s *EmailServiceV1) SendTaskNotificationEmail(ctx context.Context, toEmail,
// sendEmail 发送邮件的底层方法
func (s *EmailServiceV1) sendEmail(ctx context.Context, to, subject, body string) error {
message := mail.NewMsg()
e := message.FromFormat(s.config.FromName, s.config.FromEmail)
payload := bytes.NewBuffer(nil)
e := gob.NewEncoder(payload).Encode(&email.SendPayload{
To: to,
Subject: subject,
Body: body,
FromName: s.config.FromName,
FromEmail: s.config.FromEmail,
SMTPHost: s.config.SMTPHost,
SMTPPort: s.config.SMTPPort,
SMTPUsername: s.config.SMTPUsername,
SMTPPassword: s.config.SMTPPassword,
})
utils.OkOrPanic(e, oops.WithContext(ctx).Wrap)
e = message.AddTo(to)
utils.OkOrPanic(e, oops.WithContext(ctx).Wrap)
message.Subject(subject)
message.SetBodyString(mail.TypeTextHTML, body)
client, e := mail.NewClient(s.config.SMTPHost, mail.WithPort(s.config.SMTPPort), mail.WithSMTPAuth(mail.SMTPAuthAutoDiscover),
mail.WithUsername(s.config.SMTPUsername), mail.WithPassword(s.config.SMTPPassword))
utils.OkOrPanic(e, oops.WithContext(ctx).Wrap)
e = client.DialAndSendWithContext(ctx, message)
utils.OkOrPanic(e, oops.WithContext(ctx).Wrap)
queue.AddTask(ctx, asynq.NewTask(cst.TaskEmail, payload.Bytes()), asynq.Queue(cst.TaskQueueEmail),
asynq.MaxRetry(3),
asynq.Timeout(10*time.Minute),
)
return nil
}
@@ -157,7 +163,7 @@ func (s *EmailServiceV1) renderSignupEmailTemplate(username string) (string, err
}
// renderTaskNotificationTemplate 渲染任务通知邮件模板
func (s *EmailServiceV1) renderTaskNotificationTemplate(taskName string, status TaskStatus, details map[string]string) (string, error) {
func (s *EmailServiceV1) renderTaskNotificationTemplate(taskName string, status email.TaskStatus, details map[string]string) (string, error) {
tmpl, err := s.getTemplates()
if err != nil {
return "", err
internal/service/email/template_data.go
+4 -60
View File
@@ -1,5 +1,7 @@
package email
import "likedream-goframe/internal/model/email"
// BaseTemplateData 基础模板数据
type BaseTemplateData struct {
Title string `json:"title"` // 页面标题
@@ -31,75 +33,17 @@ func NewSignupTemplateData(username, appName, copyright string) *SignupTemplateD
}
}
// TaskStatus 任务状态枚举
type TaskStatus string
const (
TaskStatusSuccess TaskStatus = "success" // 成功
TaskStatusFailed TaskStatus = "failed" // 失败
TaskStatusRunning TaskStatus = "running" // 运行中
TaskStatusPending TaskStatus = "pending" // 等待中
)
// GetStatusColor 获取状态对应的颜色
func (s TaskStatus) GetStatusColor() string {
switch s {
case TaskStatusSuccess:
return "#10b981" // 绿色
case TaskStatusFailed:
return "#ef4444" // 红色
case TaskStatusRunning:
return "#3b82f6" // 蓝色
case TaskStatusPending:
return "#f59e0b" // 橙色
default:
return "#6b7280" // 灰色
}
}
// GetStatusText 获取状态文本
func (s TaskStatus) GetStatusText() string {
switch s {
case TaskStatusSuccess:
return "执行成功"
case TaskStatusFailed:
return "执行失败"
case TaskStatusRunning:
return "执行中"
case TaskStatusPending:
return "等待执行"
default:
return "未知状态"
}
}
// GetStatusIcon 获取状态图标
func (s TaskStatus) GetStatusIcon() string {
switch s {
case TaskStatusSuccess:
return "✅"
case TaskStatusFailed:
return "❌"
case TaskStatusRunning:
return "🔄"
case TaskStatusPending:
return "⏳"
default:
return "📬"
}
}
// TaskNotificationTemplateData 任务通知邮件模板数据
type TaskNotificationTemplateData struct {
BaseTemplateData
TaskName string `json:"task_name"` // 任务名称
Status TaskStatus `json:"status"` // 任务状态
Status email.TaskStatus `json:"status"` // 任务状态
StatusText string `json:"status_text"` // 状态文本
Details map[string]string `json:"details"` // 详细信息
}
// NewTaskNotificationTemplateData 创建任务通知邮件模板数据
func NewTaskNotificationTemplateData(taskName string, status TaskStatus, details map[string]string, appName, copyright string) *TaskNotificationTemplateData {
func NewTaskNotificationTemplateData(taskName string, status email.TaskStatus, details map[string]string, appName, copyright string) *TaskNotificationTemplateData {
return &TaskNotificationTemplateData{
BaseTemplateData: BaseTemplateData{
Title: "任务状态通知",