package weekstar import ( "context" "encoding/json" "log" "strconv" "strings" "time" "chatapp3-golang/internal/model" "chatapp3-golang/internal/service/taskcenter" "chatapp3-golang/internal/utils" ) // ProcessGiftEvent 按送礼事件发生时间归属配置,并把积分累加到对应周期实时榜。 func (s *WeekStarService) ProcessGiftEvent(ctx context.Context, event weekStarGiftEvent, rawPayload string) error { sysOrigin := s.normalizeSysOrigin(event.SysOrigin) trackID := strings.TrimSpace(event.TrackID.String()) if trackID == "" || int64(event.SendUserID) <= 0 { return nil } eventTime := resolveMillisTime(int64(event.CreateTime), time.Now()).In(s.location) s.reportGiftConsumeTaskEvent(ctx, sysOrigin, event, eventTime) configRow, giftRows, _, err := s.loadActiveConfigBundle(ctx, sysOrigin, eventTime) if err != nil { return err } if configRow == nil || !configRow.Enabled { return nil } giftID := int64(event.GiftConfig.ID) if giftID <= 0 || !containsGiftID(giftRows, giftID) { return nil } // 先按活动口径计算本次送礼积分,再写审计日志和实时榜。 scoreGold := calculateGiftScore(event) if scoreGold <= 0 { return nil } acceptUserSize := int64(countDistinctAcceptUsers(event.Accepts)) if acceptUserSize <= 0 { acceptUserSize = 1 } cycleStart, _ := s.cycleBounds(eventTime) cycleKey := s.cycleKey(cycleStart) logID, err := utils.NextID() if err != nil { return err } now := time.Now() record := model.WeekStarGiftLog{ ID: logID, ConfigID: configRow.ID, CycleKey: cycleKey, TrackID: trackID, SendUserID: int64(event.SendUserID), GiftID: giftID, Quantity: int64(event.Quantity), AcceptUserSize: acceptUserSize, ScoreGold: scoreGold, EventTime: eventTime, Status: weekStarGiftLogStatusPending, RawPayload: truncateString(rawPayload, 65535), CreateTime: now, UpdateTime: now, } if err := s.repo.DB.WithContext(ctx).Create(&record).Error; err != nil { if isDuplicateKeyErr(err) { return nil } return err } if err := s.repo.Redis.ZIncrBy(ctx, s.rankRedisKey(configRow.ID, cycleKey), float64(scoreGold), strconv.FormatInt(record.SendUserID, 10)).Err(); err != nil { // Redis 加分失败时把日志标成失败,方便补偿与排查。 _ = s.repo.DB.WithContext(context.Background()). Model(&model.WeekStarGiftLog{}). Where("id = ?", record.ID). Updates(map[string]any{ "status": weekStarGiftLogStatusFailed, "update_time": time.Now(), }).Error return err } return s.repo.DB.WithContext(ctx). Model(&model.WeekStarGiftLog{}). Where("id = ?", record.ID). Updates(map[string]any{ "status": weekStarGiftLogStatusSuccess, "update_time": time.Now(), }).Error } func (s *WeekStarService) reportGiftConsumeTaskEvent(ctx context.Context, sysOrigin string, event weekStarGiftEvent, eventTime time.Time) { if s.taskReporter == nil { return } trackID := strings.TrimSpace(event.TrackID.String()) sendUserID := int64(event.SendUserID) if trackID == "" || sendUserID <= 0 { return } scoreGold := calculateGiftScore(event) if scoreGold <= 0 { return } if err := s.taskReporter.ReportSimpleEventAt(ctx, sysOrigin, "GIFT_CONSUME:"+trackID, taskcenter.EventTypeGiftConsumeGold, sendUserID, scoreGold, eventTime, map[string]any{ "trackId": trackID, "giftId": int64(event.GiftConfig.ID), "giftName": strings.TrimSpace(event.GiftConfig.GiftName), "quantity": int64(event.Quantity), "acceptUserSize": countDistinctAcceptUsers(event.Accepts), "actualAmount": int64(event.GiftValue.ActualAmount), "giftCandy": int64(event.GiftConfig.GiftCandy), }); err != nil { log.Printf("report gift consume task event failed: %v", err) } } // ProcessGiftPayload 解析送礼消息原文并复用周榜统计逻辑,方便本地压测和非 RocketMQ 调用。 func (s *WeekStarService) ProcessGiftPayload(ctx context.Context, payload string) error { payload = strings.TrimSpace(payload) if payload == "" { return nil } event, err := decodeWeekStarGiftEventPayload(payload, s.cfg.WeekStar.RocketMQ.Tag) if err != nil { return err } return s.ProcessGiftEvent(ctx, event, payload) } func decodeWeekStarGiftEventPayload(payload string, expectedTag string) (weekStarGiftEvent, error) { var event weekStarGiftEvent if err := json.Unmarshal([]byte(payload), &event); err != nil { return event, err } if event.hasBusinessFields() { return event, nil } var envelope weekStarMessageEventEnvelope if err := json.Unmarshal([]byte(payload), &envelope); err != nil { return event, err } if strings.TrimSpace(expectedTag) != "" && strings.TrimSpace(envelope.Tag) != "" && strings.TrimSpace(envelope.Tag) != strings.TrimSpace(expectedTag) { return weekStarGiftEvent{}, nil } bodyPayload, ok, err := envelope.bodyPayload() if err != nil || !ok { return event, err } if err := json.Unmarshal([]byte(bodyPayload), &event); err != nil { return event, err } return event, nil } type weekStarMessageEventEnvelope struct { Tag string `json:"tag"` Body json.RawMessage `json:"body"` } func (e weekStarMessageEventEnvelope) bodyPayload() (string, bool, error) { raw := strings.TrimSpace(string(e.Body)) if raw == "" || raw == "null" { return "", false, nil } if strings.HasPrefix(raw, "\"") { var body string if err := json.Unmarshal(e.Body, &body); err != nil { return "", false, err } body = strings.TrimSpace(body) return body, body != "", nil } return raw, true, nil } func (e weekStarGiftEvent) hasBusinessFields() bool { return strings.TrimSpace(e.TrackID.String()) != "" || int64(e.SendUserID) > 0 || int64(e.GiftConfig.ID) > 0 }