242 lines
7.3 KiB
Go
242 lines
7.3 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"log"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"chatapp-cron/internal/model"
|
||
|
||
"github.com/redis/go-redis/v9"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
const weekStarRewardRetryPendingTTL = 30 * time.Minute
|
||
|
||
// startRetryWorker 启动失败补发任务消费协程，统一处理后台人工补发请求。
|
||
func (s *WeekStarSettlementService) startRetryWorker(ctx context.Context) {
|
||
if strings.TrimSpace(s.cfg.WeekStarRetryStreamKey) == "" {
|
||
log.Printf("week star retry consumer disabled: stream key is blank")
|
||
return
|
||
}
|
||
if err := s.ensureRetryConsumerGroup(ctx); err != nil {
|
||
log.Printf("ensure week star retry consumer group failed before start: %v", err)
|
||
}
|
||
log.Printf(
|
||
"week star retry consumer started. stream=%s group=%s consumer=%s",
|
||
s.cfg.WeekStarRetryStreamKey,
|
||
s.cfg.WeekStarRetryConsumerGroup,
|
||
s.cfg.WeekStarRetryConsumerName,
|
||
)
|
||
go s.consumeRetryLoop(ctx)
|
||
}
|
||
|
||
// consumeRetryLoop 先认领超时 pending 任务，再持续读取新补发任务。
|
||
func (s *WeekStarSettlementService) consumeRetryLoop(ctx context.Context) {
|
||
for {
|
||
if ctx.Err() != nil {
|
||
return
|
||
}
|
||
|
||
reclaimed, err := s.reclaimPendingRetryTasks(ctx)
|
||
if err != nil {
|
||
s.handleRetryLoopError(ctx, "reclaim week star retry tasks", err)
|
||
continue
|
||
}
|
||
if reclaimed > 0 {
|
||
continue
|
||
}
|
||
|
||
err = s.readNewRetryTasks(ctx)
|
||
if err == nil || errors.Is(err, redis.Nil) {
|
||
continue
|
||
}
|
||
s.handleRetryLoopError(ctx, "read week star retry tasks", err)
|
||
}
|
||
}
|
||
|
||
// handleRetryLoopError 统一处理补发消费循环错误，并在必要时重建消费者组。
|
||
func (s *WeekStarSettlementService) handleRetryLoopError(ctx context.Context, action string, err error) {
|
||
if ctx.Err() != nil {
|
||
return
|
||
}
|
||
if isWeekStarRetryNoGroupErr(err) {
|
||
if ensureErr := s.ensureRetryConsumerGroup(ctx); ensureErr != nil {
|
||
log.Printf("%s failed and recreate retry consumer group failed: %v", action, ensureErr)
|
||
sleepWeekStarRetry(ctx, time.Second)
|
||
return
|
||
}
|
||
return
|
||
}
|
||
log.Printf("%s failed: %v", action, err)
|
||
sleepWeekStarRetry(ctx, time.Second)
|
||
}
|
||
|
||
// ensureRetryConsumerGroup 确保失败补发 Redis Stream 消费组存在。
|
||
func (s *WeekStarSettlementService) ensureRetryConsumerGroup(ctx context.Context) error {
|
||
err := s.repo.Redis.XGroupCreateMkStream(
|
||
ctx,
|
||
s.cfg.WeekStarRetryStreamKey,
|
||
s.cfg.WeekStarRetryConsumerGroup,
|
||
"0",
|
||
).Err()
|
||
if err != nil && !strings.Contains(strings.ToUpper(err.Error()), "BUSYGROUP") {
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// reclaimPendingRetryTasks 认领长时间未确认的补发任务，避免 worker 中断后消息遗失。
|
||
func (s *WeekStarSettlementService) reclaimPendingRetryTasks(ctx context.Context) (int, error) {
|
||
messages, _, err := s.repo.Redis.XAutoClaim(ctx, &redis.XAutoClaimArgs{
|
||
Stream: s.cfg.WeekStarRetryStreamKey,
|
||
Group: s.cfg.WeekStarRetryConsumerGroup,
|
||
Consumer: s.cfg.WeekStarRetryConsumerName,
|
||
MinIdle: s.cfg.WeekStarRetryMinIdle,
|
||
Start: "0-0",
|
||
Count: s.cfg.WeekStarRetryBatchSize,
|
||
}).Result()
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
if len(messages) == 0 {
|
||
return 0, nil
|
||
}
|
||
return len(messages), s.processRetryMessages(ctx, messages)
|
||
}
|
||
|
||
// readNewRetryTasks 按消费者组协议读取新的补发任务。
|
||
func (s *WeekStarSettlementService) readNewRetryTasks(ctx context.Context) error {
|
||
streams, err := s.repo.Redis.XReadGroup(ctx, &redis.XReadGroupArgs{
|
||
Group: s.cfg.WeekStarRetryConsumerGroup,
|
||
Consumer: s.cfg.WeekStarRetryConsumerName,
|
||
Streams: []string{s.cfg.WeekStarRetryStreamKey, ">"},
|
||
Count: s.cfg.WeekStarRetryBatchSize,
|
||
Block: s.cfg.WeekStarRetryBlock,
|
||
}).Result()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for _, stream := range streams {
|
||
if err := s.processRetryMessages(ctx, stream.Messages); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// processRetryMessages 顺序处理补发任务，并在完成状态落库后 ACK + DEL。
|
||
func (s *WeekStarSettlementService) processRetryMessages(ctx context.Context, messages []redis.XMessage) error {
|
||
for _, message := range messages {
|
||
ack, recordID, err := s.processRetryMessage(ctx, message)
|
||
if ack {
|
||
if ackErr := s.repo.Redis.XAck(
|
||
context.Background(),
|
||
s.cfg.WeekStarRetryStreamKey,
|
||
s.cfg.WeekStarRetryConsumerGroup,
|
||
message.ID,
|
||
).Err(); ackErr != nil {
|
||
log.Printf("ack week star retry task failed. streamId=%s err=%v", message.ID, ackErr)
|
||
}
|
||
if delErr := s.repo.Redis.XDel(context.Background(), s.cfg.WeekStarRetryStreamKey, message.ID).Err(); delErr != nil {
|
||
log.Printf("delete week star retry task failed. streamId=%s err=%v", message.ID, delErr)
|
||
}
|
||
if recordID > 0 {
|
||
_, _ = s.repo.Redis.Del(context.Background(), s.rewardRetryPendingKey(recordID)).Result()
|
||
}
|
||
}
|
||
if err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// processRetryMessage 处理单条补发任务；只有在状态已经稳定落库后才确认消息。
|
||
func (s *WeekStarSettlementService) processRetryMessage(ctx context.Context, message redis.XMessage) (bool, int64, error) {
|
||
recordID, err := parseWeekStarRetryRecordID(message)
|
||
if err != nil {
|
||
log.Printf("drop malformed week star retry task. streamId=%s err=%v values=%v", message.ID, err, message.Values)
|
||
return true, 0, nil
|
||
}
|
||
|
||
var record model.WeekStarRewardRecord
|
||
err = s.repo.DB.WithContext(ctx).Where("id = ?", recordID).First(&record).Error
|
||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||
log.Printf("skip missing week star retry record. recordId=%d", recordID)
|
||
return true, recordID, nil
|
||
}
|
||
if err != nil {
|
||
return false, recordID, err
|
||
}
|
||
if record.Status != weekStarRewardStatusFailed {
|
||
return true, recordID, nil
|
||
}
|
||
|
||
handled, err := s.dispatchRewardRecord(ctx, &record, true)
|
||
if err != nil {
|
||
if handled {
|
||
log.Printf("week star manual retry finished with failed status. recordId=%d err=%v", recordID, err)
|
||
return true, recordID, nil
|
||
}
|
||
return false, recordID, err
|
||
}
|
||
return true, recordID, nil
|
||
}
|
||
|
||
// parseWeekStarRetryRecordID 把 Redis Stream 消息解析成补发记录主键。
|
||
func parseWeekStarRetryRecordID(message redis.XMessage) (int64, error) {
|
||
return toWeekStarRetryInt64(message.Values["recordId"])
|
||
}
|
||
|
||
// toWeekStarRetryInt64 把 Stream 字段安全转换成 int64。
|
||
func toWeekStarRetryInt64(value any) (int64, error) {
|
||
switch typed := value.(type) {
|
||
case nil:
|
||
return 0, fmt.Errorf("value is nil")
|
||
case int64:
|
||
return typed, nil
|
||
case int32:
|
||
return int64(typed), nil
|
||
case int:
|
||
return int64(typed), nil
|
||
case uint64:
|
||
return int64(typed), nil
|
||
case float64:
|
||
return int64(typed), nil
|
||
case string:
|
||
return strconv.ParseInt(strings.TrimSpace(typed), 10, 64)
|
||
case []byte:
|
||
return strconv.ParseInt(strings.TrimSpace(string(typed)), 10, 64)
|
||
default:
|
||
return strconv.ParseInt(strings.TrimSpace(fmt.Sprint(typed)), 10, 64)
|
||
}
|
||
}
|
||
|
||
// isWeekStarRetryNoGroupErr 判断是否是消费者组不存在错误。
|
||
func isWeekStarRetryNoGroupErr(err error) bool {
|
||
if err == nil {
|
||
return false
|
||
}
|
||
return strings.Contains(strings.ToUpper(err.Error()), "NOGROUP")
|
||
}
|
||
|
||
// rewardRetryPendingKey 返回补发任务去重 key，和 chatapp3-golang 保持一致。
|
||
func (s *WeekStarSettlementService) rewardRetryPendingKey(recordID int64) string {
|
||
return strings.TrimSpace(s.cfg.WeekStarRetryStreamKey) + ":pending:" + strconv.FormatInt(recordID, 10)
|
||
}
|
||
|
||
// sleepWeekStarRetry 支持被上下文中断的短暂退避休眠。
|
||
func sleepWeekStarRetry(ctx context.Context, duration time.Duration) {
|
||
timer := time.NewTimer(duration)
|
||
defer timer.Stop()
|
||
select {
|
||
case <-ctx.Done():
|
||
case <-timer.C:
|
||
}
|
||
}
|