275 lines
7.5 KiB
Go
275 lines
7.5 KiB
Go
// Package smsdispatch 负责把业务事务里入队的节点短信真正发出去。
|
|
//
|
|
// 设计要点：
|
|
// - 业务侧只在事务内写 sms_messages（pending），发送完全异步，短信失败不影响业务；
|
|
// - 通道选择与登录验证码一致：分站启用独立短信且配置了该节点模板时走分站通道与分站签名，
|
|
// 否则回退主站通道与主站模板；
|
|
// - 失败按 retryDelay 退避重试，超过 maxRetry 标记 failed，便于运营排查。
|
|
package smsdispatch
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
smsprovider "hfb_sys/backend/internal/integrations/sms"
|
|
"hfb_sys/backend/internal/model"
|
|
"hfb_sys/backend/internal/modules/sms"
|
|
|
|
"github.com/redis/go-redis/v9"
|
|
"go.uber.org/zap"
|
|
"gorm.io/datatypes"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
const lockKey = "hfb:job:smsdispatch:lock"
|
|
|
|
// SiteSenderResolver 解析分销站点的独立短信通道，由 distribution.SMSResolver 实现。
|
|
type SiteSenderResolver interface {
|
|
ResolveBusinessSender(ctx context.Context, siteID uint64) (smsprovider.Provider, string, map[string]string, error)
|
|
}
|
|
|
|
type Job struct {
|
|
db *gorm.DB
|
|
redis *redis.Client
|
|
logger *zap.Logger
|
|
repo *sms.Repository
|
|
configs sms.ConfigReader
|
|
provider smsprovider.Provider
|
|
mainSignName string
|
|
siteResolver SiteSenderResolver
|
|
interval time.Duration
|
|
batchSize int
|
|
maxRetry int
|
|
retryDelay time.Duration
|
|
instanceID string
|
|
}
|
|
|
|
func New(
|
|
db *gorm.DB,
|
|
redisClient *redis.Client,
|
|
logger *zap.Logger,
|
|
provider smsprovider.Provider,
|
|
mainSignName string,
|
|
configs sms.ConfigReader,
|
|
siteResolver SiteSenderResolver,
|
|
) *Job {
|
|
return &Job{
|
|
db: db,
|
|
redis: redisClient,
|
|
logger: logger,
|
|
repo: sms.NewRepository(db),
|
|
configs: configs,
|
|
provider: provider,
|
|
mainSignName: strings.TrimSpace(mainSignName),
|
|
siteResolver: siteResolver,
|
|
interval: time.Minute,
|
|
batchSize: 100,
|
|
maxRetry: 3,
|
|
retryDelay: 5 * time.Minute,
|
|
instanceID: newInstanceID(),
|
|
}
|
|
}
|
|
|
|
func (j *Job) Start(ctx context.Context) {
|
|
if j == nil || j.db == nil {
|
|
return
|
|
}
|
|
go j.loop(ctx)
|
|
}
|
|
|
|
func (j *Job) loop(ctx context.Context) {
|
|
j.run(ctx)
|
|
ticker := time.NewTicker(j.interval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
j.run(ctx)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (j *Job) run(ctx context.Context) {
|
|
release, ok := j.acquireLock(ctx)
|
|
if !ok {
|
|
return
|
|
}
|
|
defer release()
|
|
|
|
sent, err := j.dispatch(ctx)
|
|
if err != nil {
|
|
j.logger.Warn("节点短信派发失败", zap.Error(err))
|
|
return
|
|
}
|
|
if sent > 0 {
|
|
j.logger.Info("节点短信派发完成", zap.Int("handled", sent))
|
|
}
|
|
}
|
|
|
|
func (j *Job) dispatch(ctx context.Context) (int, error) {
|
|
cfg, err := sms.LoadConfig(ctx, j.configs)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
rows, err := j.repo.ListPending(ctx, j.batchSize, time.Now().Add(-j.retryDelay))
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
handled := 0
|
|
for _, row := range rows {
|
|
if err := ctx.Err(); err != nil {
|
|
return handled, err
|
|
}
|
|
if err := j.sendOne(ctx, cfg, row); err != nil {
|
|
j.logger.Warn("节点短信处理失败",
|
|
zap.Uint64("sms_message_id", row.ID),
|
|
zap.String("node", row.Node),
|
|
zap.String("phone", maskPhone(row.Phone)),
|
|
zap.Error(err),
|
|
)
|
|
continue
|
|
}
|
|
handled++
|
|
}
|
|
return handled, nil
|
|
}
|
|
|
|
func (j *Job) sendOne(ctx context.Context, cfg sms.Config, row model.SmsMessage) error {
|
|
node := strings.TrimSpace(row.Node)
|
|
if !sms.IsKnownNode(node) {
|
|
return j.repo.MarkSkipped(ctx, row.ID, "未登记的短信节点")
|
|
}
|
|
if strings.TrimSpace(row.Phone) == "" {
|
|
return j.repo.MarkSkipped(ctx, row.ID, "接收人手机号为空")
|
|
}
|
|
if !cfg.Enabled {
|
|
return j.repo.MarkSkipped(ctx, row.ID, "节点短信总开关未开启")
|
|
}
|
|
|
|
nodeCfg, ok := cfg.Node(node)
|
|
if !ok || !nodeCfg.Enabled {
|
|
return j.repo.MarkSkipped(ctx, row.ID, "该节点短信未启用")
|
|
}
|
|
mainTemplateCode := strings.TrimSpace(nodeCfg.TemplateCode)
|
|
if mainTemplateCode == "" {
|
|
return j.repo.MarkSkipped(ctx, row.ID, "主站未配置该节点模板 CODE")
|
|
}
|
|
|
|
provider, signName, templateCode := j.resolveSender(ctx, row, node, mainTemplateCode)
|
|
if provider == nil {
|
|
return j.repo.MarkSkipped(ctx, row.ID, "短信通道不可用")
|
|
}
|
|
|
|
req := smsprovider.TemplateRequest{
|
|
Phone: strings.TrimSpace(row.Phone),
|
|
SignName: signName,
|
|
TemplateCode: templateCode,
|
|
Params: decodeParams(row.TemplateParams),
|
|
}
|
|
if err := provider.SendTemplate(ctx, req); err != nil {
|
|
return j.handleSendFailure(ctx, row, err)
|
|
}
|
|
return j.repo.MarkSent(ctx, row.ID, templateCode, signName, time.Now())
|
|
}
|
|
|
|
// resolveSender 决定使用分站通道还是主站通道：分站启用独立短信且配置了该节点模板时优先分站。
|
|
func (j *Job) resolveSender(ctx context.Context, row model.SmsMessage, node string, mainTemplateCode string) (smsprovider.Provider, string, string) {
|
|
if row.DistributionSiteID != nil && *row.DistributionSiteID > 0 && j.siteResolver != nil {
|
|
siteProvider, siteSignName, siteTemplates, err := j.siteResolver.ResolveBusinessSender(ctx, *row.DistributionSiteID)
|
|
if err != nil {
|
|
j.logger.Warn("解析分站短信通道失败，回退主站通道",
|
|
zap.Uint64("site_id", *row.DistributionSiteID),
|
|
zap.Error(err),
|
|
)
|
|
} else if siteProvider != nil {
|
|
if code := strings.TrimSpace(siteTemplates[node]); code != "" {
|
|
return siteProvider, siteSignName, code
|
|
}
|
|
}
|
|
}
|
|
|
|
if j.provider == nil {
|
|
return nil, "", ""
|
|
}
|
|
return j.provider, j.mainSignName, mainTemplateCode
|
|
}
|
|
|
|
func (j *Job) handleSendFailure(ctx context.Context, row model.SmsMessage, sendErr error) error {
|
|
retryCount := row.RetryCount + 1
|
|
if retryCount >= j.maxRetry {
|
|
if err := j.repo.MarkFailed(ctx, row.ID, retryCount, sendErr.Error()); err != nil {
|
|
return err
|
|
}
|
|
j.logger.Warn("节点短信重试次数用尽",
|
|
zap.Uint64("sms_message_id", row.ID),
|
|
zap.String("node", row.Node),
|
|
zap.Int("retry_count", retryCount),
|
|
zap.Error(sendErr),
|
|
)
|
|
return nil
|
|
}
|
|
return j.repo.MarkRetry(ctx, row.ID, retryCount, sendErr.Error())
|
|
}
|
|
|
|
func decodeParams(raw datatypes.JSON) map[string]string {
|
|
if len(raw) == 0 {
|
|
return nil
|
|
}
|
|
var params map[string]string
|
|
if err := json.Unmarshal(raw, ¶ms); err != nil {
|
|
return nil
|
|
}
|
|
if len(params) == 0 {
|
|
return nil
|
|
}
|
|
return params
|
|
}
|
|
|
|
// acquireLock 通过 Redis 分布式锁确保同一时刻只有一个实例派发。
|
|
// 未配置 Redis 时直接执行；Redis 出错时降级执行（重复发送由状态与重试次数兜底）。
|
|
func (j *Job) acquireLock(ctx context.Context) (func(), bool) {
|
|
if j.redis == nil {
|
|
return func() {}, true
|
|
}
|
|
ok, err := j.redis.SetNX(ctx, lockKey, j.instanceID, 2*j.interval).Result()
|
|
if err != nil {
|
|
j.logger.Warn("节点短信任务获取锁失败，将以单实例模式执行", zap.Error(err))
|
|
return func() {}, true
|
|
}
|
|
if !ok {
|
|
return nil, false
|
|
}
|
|
return j.releaseLock, true
|
|
}
|
|
|
|
func (j *Job) releaseLock() {
|
|
if j.redis == nil {
|
|
return
|
|
}
|
|
relCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
|
defer cancel()
|
|
script := redis.NewScript(`if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end`)
|
|
if err := script.Run(relCtx, j.redis, []string{lockKey}, j.instanceID).Err(); err != nil {
|
|
j.logger.Warn("节点短信任务释放锁失败", zap.Error(err))
|
|
}
|
|
}
|
|
|
|
func newInstanceID() string {
|
|
return fmt.Sprintf("smsdispatch-%d", time.Now().UnixNano())
|
|
}
|
|
|
|
func maskPhone(phone string) string {
|
|
phone = strings.TrimSpace(phone)
|
|
if len(phone) < 7 {
|
|
return phone
|
|
}
|
|
return phone[:3] + "****" + phone[len(phone)-4:]
|
|
}
|