545 lines
17 KiB
Go
545 lines
17 KiB
Go
package voiceroomrocket
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"chatapp3-golang/internal/model"
|
|
"chatapp3-golang/internal/utils"
|
|
|
|
"github.com/redis/go-redis/v9"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
type launchResult struct {
|
|
Launched bool
|
|
Launch model.VoiceRoomRocketLaunch
|
|
}
|
|
|
|
// ProcessGiftPayload 解析送礼消息原文并按火箭规则加能量。
|
|
func (s *Service) ProcessGiftPayload(ctx context.Context, payload string) (*ProcessGiftResponse, error) {
|
|
payload = strings.TrimSpace(payload)
|
|
if payload == "" {
|
|
return &ProcessGiftResponse{Processed: false, Reason: "empty_payload"}, nil
|
|
}
|
|
event, err := decodeGiftEventPayload(payload, s.cfg.VoiceRoomRocket.MQ.Tag)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return s.ProcessGiftEvent(ctx, event, payload)
|
|
}
|
|
|
|
// ProcessGiftEvent 处理一条已经解析的送礼事件。
|
|
func (s *Service) ProcessGiftEvent(ctx context.Context, event giftEvent, rawPayload string) (*ProcessGiftResponse, error) {
|
|
trackID := strings.TrimSpace(event.TrackID.String())
|
|
sysOrigin := s.normalizeSysOrigin(event.SysOrigin)
|
|
roomID := event.RoomID.Int64()
|
|
sendUserID := event.SendUserID.Int64()
|
|
if trackID == "" || roomID <= 0 || sendUserID <= 0 {
|
|
return &ProcessGiftResponse{Processed: false, Reason: "missing_required_fields"}, nil
|
|
}
|
|
eventTime := resolveMillisTime(event.CreateTime.Int64(), time.Now())
|
|
snapshot, err := s.loadConfig(ctx, sysOrigin)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !snapshot.Enabled {
|
|
return &ProcessGiftResponse{Processed: false, Reason: "rocket_disabled"}, nil
|
|
}
|
|
rawEnergy, ignoreReason := resolveGiftEnergy(event, snapshot)
|
|
if rawEnergy <= 0 {
|
|
if ignoreReason == "" {
|
|
ignoreReason = "zero_energy"
|
|
}
|
|
if err := s.createIgnoredGiftLog(ctx, sysOrigin, trackID, roomID, sendUserID, event, rawPayload, eventTime, ignoreReason); err != nil {
|
|
return nil, err
|
|
}
|
|
return &ProcessGiftResponse{Processed: false, Reason: ignoreReason}, nil
|
|
}
|
|
|
|
unlock, err := s.acquireRoomLock(ctx, sysOrigin, roomID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if unlock != nil {
|
|
defer unlock()
|
|
}
|
|
|
|
local := resolveLocation(snapshot.Timezone)
|
|
day := dayKey(eventTime, local)
|
|
resp := &ProcessGiftResponse{
|
|
Processed: false,
|
|
RoomID: roomID,
|
|
DayKey: day,
|
|
}
|
|
var result launchResult
|
|
err = s.repo.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
statusRow, err := s.loadOrInitStatus(ctx, tx, sysOrigin, roomID, day, eventTime.In(local))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
resp.RoundNo = statusRow.RoundNo
|
|
resp.Level = statusRow.CurrentLevel
|
|
|
|
logID, err := utils.NextID()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
giftID := event.GiftConfig.ID.Int64()
|
|
var giftIDPtr *int64
|
|
if giftID > 0 {
|
|
giftIDPtr = &giftID
|
|
}
|
|
remain := statusRow.NeedEnergy - statusRow.CurrentEnergy
|
|
if remain < 0 {
|
|
remain = 0
|
|
}
|
|
accepted := rawEnergy
|
|
if accepted > remain {
|
|
accepted = remain
|
|
}
|
|
if accepted <= 0 {
|
|
record := model.VoiceRoomRocketGiftEventLog{
|
|
ID: logID,
|
|
SysOrigin: sysOrigin,
|
|
TrackID: trackID,
|
|
RoomID: roomID,
|
|
SendUserID: sendUserID,
|
|
GiftID: giftIDPtr,
|
|
GiftType: strings.TrimSpace(event.GiftConfig.Type),
|
|
GiftTab: strings.TrimSpace(event.GiftConfig.GiftTab),
|
|
RawEnergy: rawEnergy,
|
|
AcceptedEnergy: 0,
|
|
DayKey: statusRow.DayKey,
|
|
RoundNo: statusRow.RoundNo,
|
|
Level: statusRow.CurrentLevel,
|
|
Status: eventStatusIgnored,
|
|
RawPayload: truncateString(rawPayload, 16*1024*1024),
|
|
ErrorMessage: "level_already_full",
|
|
EventTime: eventTime,
|
|
CreateTime: time.Now(),
|
|
UpdateTime: time.Now(),
|
|
}
|
|
if err := tx.Create(&record).Error; err != nil {
|
|
if isDuplicateKeyErr(err) {
|
|
resp.Reason = "duplicate_event"
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
resp.Reason = "level_already_full"
|
|
return nil
|
|
}
|
|
|
|
now := time.Now()
|
|
record := model.VoiceRoomRocketGiftEventLog{
|
|
ID: logID,
|
|
SysOrigin: sysOrigin,
|
|
TrackID: trackID,
|
|
RoomID: roomID,
|
|
SendUserID: sendUserID,
|
|
GiftID: giftIDPtr,
|
|
GiftType: strings.TrimSpace(event.GiftConfig.Type),
|
|
GiftTab: strings.TrimSpace(event.GiftConfig.GiftTab),
|
|
RawEnergy: rawEnergy,
|
|
AcceptedEnergy: accepted,
|
|
DayKey: statusRow.DayKey,
|
|
RoundNo: statusRow.RoundNo,
|
|
Level: statusRow.CurrentLevel,
|
|
Status: eventStatusSuccess,
|
|
RawPayload: truncateString(rawPayload, 16*1024*1024),
|
|
EventTime: eventTime,
|
|
CreateTime: now,
|
|
UpdateTime: now,
|
|
}
|
|
if err := tx.Create(&record).Error; err != nil {
|
|
if isDuplicateKeyErr(err) {
|
|
resp.Reason = "duplicate_event"
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
|
|
if err := s.addContribution(ctx, tx, statusRow, sendUserID, accepted, eventTime); err != nil {
|
|
return err
|
|
}
|
|
|
|
newEnergy := statusRow.CurrentEnergy + accepted
|
|
statusUpdates := map[string]any{
|
|
"current_energy": newEnergy,
|
|
"last_event_time": eventTime,
|
|
"update_time": now,
|
|
}
|
|
if newEnergy >= statusRow.NeedEnergy {
|
|
statusUpdates["status"] = statusLaunching
|
|
}
|
|
if err := tx.Model(&model.VoiceRoomRocketStatus{}).
|
|
Where("id = ?", statusRow.ID).
|
|
Updates(statusUpdates).Error; err != nil {
|
|
return err
|
|
}
|
|
statusRow.CurrentEnergy = newEnergy
|
|
statusRow.LastEventTime = &eventTime
|
|
resp.Processed = true
|
|
resp.AcceptedEnergy = accepted
|
|
|
|
if err := s.updateRedisRank(ctx, statusRow, sendUserID, accepted); err != nil {
|
|
return err
|
|
}
|
|
if newEnergy >= statusRow.NeedEnergy {
|
|
if err := s.markLevelReachedLocked(ctx, tx, statusRow, sendUserID, now); err != nil {
|
|
return err
|
|
}
|
|
launch, err := s.scheduleLaunchLocked(ctx, tx, statusRow, sendUserID, now)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
result = launch
|
|
resp.Launched = launch.Launched
|
|
resp.LaunchNo = launch.Launch.LaunchNo
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if result.Launched {
|
|
_ = s.notifyLaunch(context.Background(), result.Launch)
|
|
} else if resp.Processed {
|
|
_ = s.notifyStatusUpdate(context.Background(), sysOrigin, roomID, sendUserID)
|
|
}
|
|
return resp, nil
|
|
}
|
|
|
|
func (s *Service) createIgnoredGiftLog(ctx context.Context, sysOrigin, trackID string, roomID, sendUserID int64, event giftEvent, rawPayload string, eventTime time.Time, reason string) error {
|
|
id, err := utils.NextID()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
giftID := event.GiftConfig.ID.Int64()
|
|
var giftIDPtr *int64
|
|
if giftID > 0 {
|
|
giftIDPtr = &giftID
|
|
}
|
|
now := time.Now()
|
|
record := model.VoiceRoomRocketGiftEventLog{
|
|
ID: id,
|
|
SysOrigin: sysOrigin,
|
|
TrackID: trackID,
|
|
RoomID: roomID,
|
|
SendUserID: sendUserID,
|
|
GiftID: giftIDPtr,
|
|
GiftType: strings.TrimSpace(event.GiftConfig.Type),
|
|
GiftTab: strings.TrimSpace(event.GiftConfig.GiftTab),
|
|
Status: eventStatusIgnored,
|
|
RawPayload: truncateString(rawPayload, 16*1024*1024),
|
|
ErrorMessage: reason,
|
|
EventTime: eventTime,
|
|
CreateTime: now,
|
|
UpdateTime: now,
|
|
}
|
|
err = s.repo.DB.WithContext(ctx).Create(&record).Error
|
|
if err != nil && !isDuplicateKeyErr(err) {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) addContribution(ctx context.Context, tx *gorm.DB, statusRow model.VoiceRoomRocketStatus, userID int64, accepted int64, eventTime time.Time) error {
|
|
var row model.VoiceRoomRocketContribution
|
|
err := tx.WithContext(ctx).
|
|
Where(
|
|
"sys_origin = ? AND room_id = ? AND day_key = ? AND round_no = ? AND level = ? AND user_id = ?",
|
|
statusRow.SysOrigin, statusRow.RoomID, statusRow.DayKey, statusRow.RoundNo, statusRow.CurrentLevel, userID,
|
|
).
|
|
First(&row).Error
|
|
now := time.Now()
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
id, idErr := utils.NextID()
|
|
if idErr != nil {
|
|
return idErr
|
|
}
|
|
row = model.VoiceRoomRocketContribution{
|
|
ID: id,
|
|
SysOrigin: statusRow.SysOrigin,
|
|
RoomID: statusRow.RoomID,
|
|
DayKey: statusRow.DayKey,
|
|
RoundNo: statusRow.RoundNo,
|
|
Level: statusRow.CurrentLevel,
|
|
UserID: userID,
|
|
ScoreEnergy: accepted,
|
|
GiftCount: 1,
|
|
FirstContributeTime: eventTime,
|
|
LastContributeTime: eventTime,
|
|
CreateTime: now,
|
|
UpdateTime: now,
|
|
}
|
|
return tx.Create(&row).Error
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return tx.Model(&model.VoiceRoomRocketContribution{}).
|
|
Where("id = ?", row.ID).
|
|
Updates(map[string]any{
|
|
"score_energy": gorm.Expr("score_energy + ?", accepted),
|
|
"gift_count": gorm.Expr("gift_count + 1"),
|
|
"last_contribute_time": eventTime,
|
|
"update_time": now,
|
|
}).Error
|
|
}
|
|
|
|
func (s *Service) markLevelReachedLocked(ctx context.Context, tx *gorm.DB, statusRow model.VoiceRoomRocketStatus, igniteUserID int64, now time.Time) error {
|
|
return tx.WithContext(ctx).Model(&model.VoiceRoomRocketLevelRound{}).
|
|
Where(
|
|
"sys_origin = ? AND room_id = ? AND day_key = ? AND round_no = ? AND level = ?",
|
|
statusRow.SysOrigin, statusRow.RoomID, statusRow.DayKey, statusRow.RoundNo, statusRow.CurrentLevel,
|
|
).
|
|
Updates(map[string]any{
|
|
"final_energy": statusRow.NeedEnergy,
|
|
"status": levelRoundStatusReached,
|
|
"ignite_user_id": igniteUserID,
|
|
"reach_time": now,
|
|
"update_time": now,
|
|
}).Error
|
|
}
|
|
|
|
func (s *Service) scheduleLaunchLocked(ctx context.Context, tx *gorm.DB, statusRow model.VoiceRoomRocketStatus, igniteUserID int64, now time.Time) (launchResult, error) {
|
|
if s.repo.Redis == nil {
|
|
return s.launchLocked(ctx, tx, statusRow, igniteUserID, now)
|
|
}
|
|
due := now.Add(time.Duration(defaultLaunchDelaySeconds) * time.Second)
|
|
if err := s.repo.Redis.ZAdd(ctx, s.launchDueZSetKey(), redis.Z{
|
|
Score: float64(due.Unix()),
|
|
Member: strconv.FormatInt(statusRow.ID, 10),
|
|
}).Err(); err != nil {
|
|
return launchResult{}, err
|
|
}
|
|
return launchResult{Launched: false}, nil
|
|
}
|
|
|
|
func (s *Service) launchLocked(ctx context.Context, tx *gorm.DB, statusRow model.VoiceRoomRocketStatus, igniteUserID int64, now time.Time) (launchResult, error) {
|
|
top1UserID, err := s.resolveTop1UserID(ctx, tx, statusRow)
|
|
if err != nil {
|
|
return launchResult{}, err
|
|
}
|
|
launchID, err := utils.NextID()
|
|
if err != nil {
|
|
return launchResult{}, err
|
|
}
|
|
launchNo := fmt.Sprintf("VRR%d", launchID)
|
|
roomAccount := s.resolveRoomAccount(ctx, statusRow.RoomID)
|
|
launch := model.VoiceRoomRocketLaunch{
|
|
ID: launchID,
|
|
LaunchNo: launchNo,
|
|
SysOrigin: statusRow.SysOrigin,
|
|
RoomID: statusRow.RoomID,
|
|
RoomAccount: roomAccount,
|
|
DayKey: statusRow.DayKey,
|
|
RoundNo: statusRow.RoundNo,
|
|
Level: statusRow.CurrentLevel,
|
|
FinalEnergy: statusRow.NeedEnergy,
|
|
Top1UserID: top1UserID,
|
|
IgniteUserID: &igniteUserID,
|
|
InRoomSnapshotStatus: inRoomSnapshotPending,
|
|
LaunchTime: now,
|
|
CreateTime: now,
|
|
UpdateTime: now,
|
|
}
|
|
if err := tx.Create(&launch).Error; err != nil {
|
|
if isDuplicateKeyErr(err) {
|
|
return launchResult{Launched: false}, nil
|
|
}
|
|
return launchResult{}, err
|
|
}
|
|
if err := tx.Model(&model.VoiceRoomRocketLevelRound{}).
|
|
Where(
|
|
"sys_origin = ? AND room_id = ? AND day_key = ? AND round_no = ? AND level = ?",
|
|
statusRow.SysOrigin, statusRow.RoomID, statusRow.DayKey, statusRow.RoundNo, statusRow.CurrentLevel,
|
|
).
|
|
Updates(map[string]any{
|
|
"final_energy": statusRow.NeedEnergy,
|
|
"status": levelRoundStatusLaunched,
|
|
"top1_user_id": top1UserID,
|
|
"ignite_user_id": igniteUserID,
|
|
"reach_time": gorm.Expr("COALESCE(reach_time, ?)", now),
|
|
"launch_time": now,
|
|
"update_time": now,
|
|
}).Error; err != nil {
|
|
return launchResult{}, err
|
|
}
|
|
if err := s.createImmediateRewardRecords(ctx, tx, launch, top1UserID, igniteUserID, now); err != nil {
|
|
return launchResult{}, err
|
|
}
|
|
if err := s.enqueueInRoomReward(ctx, launch, now); err != nil {
|
|
return launchResult{}, err
|
|
}
|
|
|
|
nextRound := statusRow.RoundNo
|
|
nextLevel := statusRow.CurrentLevel + 1
|
|
if nextLevel > defaultMaxLevel {
|
|
nextRound++
|
|
nextLevel = 1
|
|
}
|
|
nextLevelConfig, err := s.loadLevelConfig(ctx, statusRow.SysOrigin, nextLevel)
|
|
if err != nil {
|
|
return launchResult{}, err
|
|
}
|
|
if err := tx.Model(&model.VoiceRoomRocketStatus{}).
|
|
Where("id = ?", statusRow.ID).
|
|
Updates(map[string]any{
|
|
"round_no": nextRound,
|
|
"current_level": nextLevel,
|
|
"current_energy": 0,
|
|
"need_energy": nextLevelConfig.NeedEnergy,
|
|
"status": statusCharging,
|
|
"level_start_time": now,
|
|
"update_time": now,
|
|
}).Error; err != nil {
|
|
return launchResult{}, err
|
|
}
|
|
if err := s.ensureLevelRound(ctx, tx, statusRow.SysOrigin, statusRow.RoomID, statusRow.DayKey, nextRound, nextLevel, nextLevelConfig.NeedEnergy, now); err != nil {
|
|
return launchResult{}, err
|
|
}
|
|
return launchResult{Launched: true, Launch: launch}, nil
|
|
}
|
|
|
|
func (s *Service) resolveTop1UserID(ctx context.Context, tx *gorm.DB, statusRow model.VoiceRoomRocketStatus) (*int64, error) {
|
|
var row model.VoiceRoomRocketContribution
|
|
err := tx.WithContext(ctx).
|
|
Where(
|
|
"sys_origin = ? AND room_id = ? AND day_key = ? AND round_no = ? AND level = ?",
|
|
statusRow.SysOrigin, statusRow.RoomID, statusRow.DayKey, statusRow.RoundNo, statusRow.CurrentLevel,
|
|
).
|
|
Order("score_energy DESC, last_contribute_time ASC, user_id ASC").
|
|
First(&row).Error
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &row.UserID, nil
|
|
}
|
|
|
|
func (s *Service) acquireRoomLock(ctx context.Context, sysOrigin string, roomID int64) (func(), error) {
|
|
if s.repo.Redis == nil {
|
|
return nil, nil
|
|
}
|
|
ttl := time.Duration(s.cfg.VoiceRoomRocket.RoomLockTTLSeconds) * time.Second
|
|
if ttl <= 0 {
|
|
ttl = defaultRoomLockTTL
|
|
}
|
|
key := roomLockKey(sysOrigin, roomID)
|
|
token := strconv.FormatInt(time.Now().UnixNano(), 10)
|
|
for i := 0; i < 20; i++ {
|
|
ok, err := s.repo.Redis.SetNX(ctx, key, token, ttl).Result()
|
|
if err != nil {
|
|
return nil, NewAppError(http.StatusServiceUnavailable, "rocket_lock_failed", err.Error())
|
|
}
|
|
if ok {
|
|
return func() { _ = s.releaseRoomLock(context.Background(), key, token) }, nil
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
case <-time.After(50 * time.Millisecond):
|
|
}
|
|
}
|
|
return nil, NewAppError(http.StatusConflict, "rocket_room_busy", "room rocket is busy")
|
|
}
|
|
|
|
func (s *Service) releaseRoomLock(ctx context.Context, key string, token string) error {
|
|
if s.repo.Redis == nil {
|
|
return nil
|
|
}
|
|
const script = `
|
|
if redis.call("get", KEYS[1]) == ARGV[1] then
|
|
return redis.call("del", KEYS[1])
|
|
end
|
|
return 0`
|
|
return s.repo.Redis.Eval(ctx, script, []string{key}, token).Err()
|
|
}
|
|
|
|
func (s *Service) updateRedisRank(ctx context.Context, statusRow model.VoiceRoomRocketStatus, userID int64, accepted int64) error {
|
|
if s.repo.Redis == nil || accepted <= 0 {
|
|
return nil
|
|
}
|
|
return s.repo.Redis.ZIncrBy(ctx, rankRedisKey(statusRow.SysOrigin, statusRow.RoomID, statusRow.DayKey, statusRow.RoundNo, statusRow.CurrentLevel), float64(accepted), strconv.FormatInt(userID, 10)).Err()
|
|
}
|
|
|
|
func (s *Service) enqueueInRoomReward(ctx context.Context, launch model.VoiceRoomRocketLaunch, now time.Time) error {
|
|
if s.repo.Redis == nil {
|
|
return nil
|
|
}
|
|
delaySeconds := defaultInRoomRewardDelaySeconds
|
|
if snapshot, err := s.loadConfig(ctx, launch.SysOrigin); err == nil && snapshot.InRoomRewardDelaySeconds > 0 {
|
|
delaySeconds = snapshot.InRoomRewardDelaySeconds
|
|
}
|
|
due := now.Add(time.Duration(delaySeconds) * time.Second)
|
|
key := strings.TrimSpace(s.cfg.VoiceRoomRocket.InRoomRewardDueZSetKey)
|
|
if key == "" {
|
|
key = "voice_room:rocket:in_room_due"
|
|
}
|
|
return s.repo.Redis.ZAdd(ctx, key, redis.Z{
|
|
Score: float64(due.Unix()),
|
|
Member: launch.LaunchNo,
|
|
}).Err()
|
|
}
|
|
|
|
func decodeGiftEventPayload(payload string, expectedTag string) (giftEvent, error) {
|
|
var event giftEvent
|
|
if err := json.Unmarshal([]byte(payload), &event); err != nil {
|
|
return event, err
|
|
}
|
|
if event.hasBusinessFields() {
|
|
return event, nil
|
|
}
|
|
var envelope messageEventEnvelope
|
|
if err := json.Unmarshal([]byte(payload), &envelope); err != nil {
|
|
return event, err
|
|
}
|
|
if strings.TrimSpace(expectedTag) != "" &&
|
|
strings.TrimSpace(envelope.Tag) != "" &&
|
|
strings.TrimSpace(expectedTag) != strings.TrimSpace(envelope.Tag) {
|
|
return giftEvent{}, 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
|
|
}
|
|
|
|
func (e messageEventEnvelope) 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 giftEvent) hasBusinessFields() bool {
|
|
return strings.TrimSpace(e.TrackID.String()) != "" ||
|
|
e.SendUserID.Int64() > 0 ||
|
|
e.RoomID.Int64() > 0 ||
|
|
e.GiftConfig.ID.Int64() > 0
|
|
}
|