diff --git a/internal/service/registerreward/stream.go b/internal/service/registerreward/stream.go index 2e3cab8..cb7640f 100644 --- a/internal/service/registerreward/stream.go +++ b/internal/service/registerreward/stream.go @@ -2,6 +2,7 @@ package registerreward import ( "context" + "encoding/binary" "fmt" "strconv" "strings" @@ -42,14 +43,54 @@ func toRegisterRewardString(value any) string { case nil: return "" case string: - return typed + return decodeJavaSerializedString([]byte(typed)) case []byte: - return string(typed) + 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) { @@ -66,9 +107,9 @@ func toRegisterRewardInt64(value any) (int64, error) { case float64: return int64(typed), nil case string: - return strconv.ParseInt(strings.TrimSpace(typed), 10, 64) + return strconv.ParseInt(strings.TrimSpace(decodeJavaSerializedString([]byte(typed))), 10, 64) case []byte: - return strconv.ParseInt(strings.TrimSpace(string(typed)), 10, 64) + return strconv.ParseInt(strings.TrimSpace(decodeJavaSerializedString(typed)), 10, 64) default: return strconv.ParseInt(strings.TrimSpace(fmt.Sprint(typed)), 10, 64) } diff --git a/internal/service/registerreward/stream_test.go b/internal/service/registerreward/stream_test.go new file mode 100644 index 0000000..791953c --- /dev/null +++ b/internal/service/registerreward/stream_test.go @@ -0,0 +1,67 @@ +package registerreward + +import ( + "encoding/binary" + "testing" + + "github.com/redis/go-redis/v9" +) + +func TestParseRegisterRewardEventSupportsJavaSerializedStringValues(t *testing.T) { + message := redis.XMessage{ + ID: "1777262756108-0", + Values: map[string]any{ + "eventId": javaSerializedStringForTest("REGISTER_REWARD:LIKEI:2048614597363036161"), + "sysOrigin": javaSerializedStringForTest("LIKEI"), + "userId": javaSerializedStringForTest("2048614597363036161"), + "countryId": javaSerializedStringForTest("1231833351745187842"), + "publishedAt": javaSerializedStringForTest("1777262756107"), + }, + } + + event, err := parseRegisterRewardEvent(message) + if err != nil { + t.Fatalf("parseRegisterRewardEvent() error = %v", err) + } + if event.EventID != "REGISTER_REWARD:LIKEI:2048614597363036161" { + t.Fatalf("eventID = %q", event.EventID) + } + if event.SysOrigin != "LIKEI" { + t.Fatalf("sysOrigin = %q", event.SysOrigin) + } + if event.UserID != 2048614597363036161 { + t.Fatalf("userID = %d", event.UserID) + } + if event.CountryID != 1231833351745187842 { + t.Fatalf("countryID = %d", event.CountryID) + } +} + +func TestParseRegisterRewardEventStillSupportsPlainStringValues(t *testing.T) { + message := redis.XMessage{ + ID: "1-0", + Values: map[string]any{ + "eventId": "REGISTER_REWARD:LIKEI:2048614597363036161", + "sysOrigin": "LIKEI", + "userId": "2048614597363036161", + "countryId": "1231833351745187842", + }, + } + + event, err := parseRegisterRewardEvent(message) + if err != nil { + t.Fatalf("parseRegisterRewardEvent() error = %v", err) + } + if event.UserID != 2048614597363036161 || event.CountryID != 1231833351745187842 { + t.Fatalf("unexpected event: %+v", event) + } +} + +func javaSerializedStringForTest(value string) string { + payload := []byte(value) + result := make([]byte, 7+len(payload)) + copy(result[:5], []byte{0xac, 0xed, 0x00, 0x05, 0x74}) + binary.BigEndian.PutUint16(result[5:7], uint16(len(payload))) + copy(result[7:], payload) + return string(result) +}