yumi-golang/internal/service/registerreward/stream.go

package registerreward

import (
	"context"
	"encoding/binary"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/redis/go-redis/v9"
)

// parseRegisterRewardEvent 把 Redis Stream 消息解析成统一事件对象。
func parseRegisterRewardEvent(message redis.XMessage) (registerRewardEvent, error) {
	eventID := strings.TrimSpace(toRegisterRewardString(message.Values["eventId"]))
	if eventID == "" {
		return registerRewardEvent{}, fmt.Errorf("eventId is required")
	}
	sysOrigin := strings.ToUpper(strings.TrimSpace(toRegisterRewardString(message.Values["sysOrigin"])))
	if sysOrigin == "" {
		return registerRewardEvent{}, fmt.Errorf("sysOrigin is required")
	}
	userID, err := toRegisterRewardInt64(message.Values["userId"])
	if err != nil {
		return registerRewardEvent{}, fmt.Errorf("invalid userId: %w", err)
	}
	if userID <= 0 {
		return registerRewardEvent{}, fmt.Errorf("invalid userId")
	}
	countryID, _ := toRegisterRewardInt64(message.Values["countryId"])
	return registerRewardEvent{
		EventID:   eventID,
		SysOrigin: sysOrigin,
		UserID:    userID,
		CountryID: countryID,
	}, nil
}

// toRegisterRewardString 把 Stream 字段安全转换成字符串。
func toRegisterRewardString(value any) string {
	switch typed := value.(type) {
	case nil:
		return ""
	case string:
		return decodeJavaSerializedString([]byte(typed))
	case []byte:
		return decodeJavaSerializedString(typed)
	default:
		return fmt.Sprint(typed)
	}
}

// decodeJavaSerializedString 兼容 Java RedisTemplate 默认 hash value 序列化写入的 String。
// 正常路径仍然是纯字符串；只有命中 Java Object Serialization 头时才解包。
func decodeJavaSerializedString(value []byte) string {
	const (
		javaStreamMagic0 = 0xac
		javaStreamMagic1 = 0xed
		javaStreamVer0   = 0x00
		javaStreamVer1   = 0x05
		javaTCString     = 0x74
		javaTCLongString = 0x7c
	)
	if len(value) < 7 ||
		value[0] != javaStreamMagic0 ||
		value[1] != javaStreamMagic1 ||
		value[2] != javaStreamVer0 ||
		value[3] != javaStreamVer1 {
		return string(value)
	}

	switch value[4] {
	case javaTCString:
		length := int(binary.BigEndian.Uint16(value[5:7]))
		if length < 0 || len(value) < 7+length {
			return string(value)
		}
		return string(value[7 : 7+length])
	case javaTCLongString:
		if len(value) < 13 {
			return string(value)
		}
		length := binary.BigEndian.Uint64(value[5:13])
		if length > uint64(len(value)-13) {
			return string(value)
		}
		return string(value[13 : 13+int(length)])
	default:
		return string(value)
	}
}

// toRegisterRewardInt64 把 Stream 字段安全转换成 int64。
func toRegisterRewardInt64(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(decodeJavaSerializedString([]byte(typed))), 10, 64)
	case []byte:
		return strconv.ParseInt(strings.TrimSpace(decodeJavaSerializedString(typed)), 10, 64)
	default:
		return strconv.ParseInt(strings.TrimSpace(fmt.Sprint(typed)), 10, 64)
	}
}

// isRegisterRewardTerminalStatus 判断该日志状态是否已经终态。
func isRegisterRewardTerminalStatus(status string) bool {
	switch strings.ToUpper(strings.TrimSpace(status)) {
	case registerRewardStatusSuccess, registerRewardStatusSkipped:
		return true
	default:
		return false
	}
}

// isRegisterRewardNoGroupErr 判断是否是消费者组不存在错误。
func isRegisterRewardNoGroupErr(err error) bool {
	if err == nil {
		return false
	}
	return strings.Contains(strings.ToUpper(err.Error()), "NOGROUP")
}

// truncateRegisterRewardMessage 截断错误信息，避免超出数据库字段长度。
func truncateRegisterRewardMessage(message string) string {
	message = strings.TrimSpace(message)
	if len(message) <= 255 {
		return message
	}
	return message[:255]
}

// sleepWithContext 支持被上下文中断的休眠。
func sleepWithContext(ctx context.Context, duration time.Duration) {
	timer := time.NewTimer(duration)
	defer timer.Stop()
	select {
	case <-ctx.Done():
	case <-timer.C:
	}
}