yumi-golang/internal/service/register_reward.go

package service

import (
	"chatapp3-golang/internal/config"
	"chatapp3-golang/internal/integration"
	"chatapp3-golang/internal/model"
	"chatapp3-golang/internal/repo"
	"chatapp3-golang/internal/util"
	"context"
	"errors"
	"fmt"
	"log"
	"strconv"
	"strings"
	"time"

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

const (
	registerRewardStatusPending = "PENDING"
	registerRewardStatusSuccess = "SUCCESS"
	registerRewardStatusSkipped = "SKIPPED"
	registerRewardStatusFailed  = "FAILED"
	registerRewardGoldOrigin    = "REGISTER_REWARDS"
	registerRewardGoldDesc      = "Register reward"
	registerRewardPropsOrigin   = "ACTIVITY_REWARD"
)

type registerRewardEvent struct {
	EventID   string
	SysOrigin string
	UserID    int64
	CountryID int64
}

type registerRewardResolvedConfig struct {
	GoldAmount    int64
	RewardGroupID *int64
}

type RegisterRewardService struct {
	cfg  config.Config
	repo *repo.Repository
	java *integration.Client
}

func NewRegisterRewardService(cfg config.Config, repository *repo.Repository, javaClient *integration.Client) *RegisterRewardService {
	return &RegisterRewardService{
		cfg:  cfg,
		repo: repository,
		java: javaClient,
	}
}

func (s *RegisterRewardService) Start(ctx context.Context) error {
	if strings.TrimSpace(s.cfg.RegisterRewardStreamKey) == "" {
		log.Printf("register reward consumer disabled: stream key is blank")
		return nil
	}
	if err := s.ensureConsumerGroup(ctx); err != nil {
		return err
	}
	log.Printf(
		"register reward consumer started. stream=%s group=%s consumer=%s",
		s.cfg.RegisterRewardStreamKey,
		s.cfg.RegisterRewardConsumerGroup,
		s.cfg.RegisterRewardConsumerName,
	)
	go s.consumeLoop(ctx)
	return nil
}

func (s *RegisterRewardService) consumeLoop(ctx context.Context) {
	for {
		if ctx.Err() != nil {
			return
		}

		reclaimed, err := s.reclaimPending(ctx)
		if err != nil {
			s.handleLoopError(ctx, "reclaim register reward messages", err)
			continue
		}
		if reclaimed > 0 {
			continue
		}

		err = s.readNewMessages(ctx)
		if err == nil || errors.Is(err, redis.Nil) {
			continue
		}
		s.handleLoopError(ctx, "read register reward messages", err)
	}
}

func (s *RegisterRewardService) handleLoopError(ctx context.Context, action string, err error) {
	if ctx.Err() != nil {
		return
	}
	if isRegisterRewardNoGroupErr(err) {
		if ensureErr := s.ensureConsumerGroup(ctx); ensureErr != nil {
			log.Printf("%s failed and recreate consumer group failed: %v", action, ensureErr)
			sleepWithContext(ctx, time.Second)
			return
		}
		return
	}
	log.Printf("%s failed: %v", action, err)
	sleepWithContext(ctx, time.Second)
}

func (s *RegisterRewardService) ensureConsumerGroup(ctx context.Context) error {
	err := s.repo.Redis.XGroupCreateMkStream(
		ctx,
		s.cfg.RegisterRewardStreamKey,
		s.cfg.RegisterRewardConsumerGroup,
		"0",
	).Err()
	if err != nil && !strings.Contains(strings.ToUpper(err.Error()), "BUSYGROUP") {
		return err
	}
	return nil
}

func (s *RegisterRewardService) reclaimPending(ctx context.Context) (int, error) {
	messages, _, err := s.repo.Redis.XAutoClaim(ctx, &redis.XAutoClaimArgs{
		Stream:   s.cfg.RegisterRewardStreamKey,
		Group:    s.cfg.RegisterRewardConsumerGroup,
		Consumer: s.cfg.RegisterRewardConsumerName,
		MinIdle:  s.cfg.RegisterRewardMinIdle,
		Start:    "0-0",
		Count:    s.cfg.RegisterRewardBatchSize,
	}).Result()
	if err != nil {
		return 0, err
	}
	if len(messages) == 0 {
		return 0, nil
	}
	return len(messages), s.processMessages(ctx, messages)
}

func (s *RegisterRewardService) readNewMessages(ctx context.Context) error {
	streams, err := s.repo.Redis.XReadGroup(ctx, &redis.XReadGroupArgs{
		Group:    s.cfg.RegisterRewardConsumerGroup,
		Consumer: s.cfg.RegisterRewardConsumerName,
		Streams:  []string{s.cfg.RegisterRewardStreamKey, ">"},
		Count:    s.cfg.RegisterRewardBatchSize,
		Block:    s.cfg.RegisterRewardBlock,
	}).Result()
	if err != nil {
		return err
	}
	for _, stream := range streams {
		if err := s.processMessages(ctx, stream.Messages); err != nil {
			return err
		}
	}
	return nil
}

func (s *RegisterRewardService) processMessages(ctx context.Context, messages []redis.XMessage) error {
	for _, message := range messages {
		ack, err := s.processMessage(ctx, message)
		if ack {
			if ackErr := s.repo.Redis.XAck(
				context.Background(),
				s.cfg.RegisterRewardStreamKey,
				s.cfg.RegisterRewardConsumerGroup,
				message.ID,
			).Err(); ackErr != nil {
				log.Printf("ack register reward message failed. streamId=%s err=%v", message.ID, ackErr)
			}
		}
		if err != nil {
			return err
		}
	}
	return nil
}

func (s *RegisterRewardService) processMessage(ctx context.Context, message redis.XMessage) (bool, error) {
	event, err := parseRegisterRewardEvent(message)
	if err != nil {
		log.Printf("drop malformed register reward message. streamId=%s err=%v values=%v", message.ID, err, message.Values)
		return true, nil
	}

	logRecord, terminal, err := s.getOrCreateLog(ctx, event, message.ID)
	if err != nil {
		return false, err
	}
	if terminal {
		return true, nil
	}

	resolvedConfig, ack, err := s.resolveRewardConfig(ctx, logRecord, event)
	if err != nil || ack {
		return ack, err
	}

	if err := s.dispatchGold(ctx, event, resolvedConfig); err != nil {
		_ = s.updateLogStatus(ctx, logRecord.ID, registerRewardStatusFailed, err.Error(), resolvedConfig)
		return false, err
	}
	if err := s.dispatchProps(ctx, logRecord.ID, event, resolvedConfig); err != nil {
		_ = s.updateLogStatus(ctx, logRecord.ID, registerRewardStatusFailed, err.Error(), resolvedConfig)
		return false, err
	}
	if err := s.updateLogStatus(ctx, logRecord.ID, registerRewardStatusSuccess, "", resolvedConfig); err != nil {
		return false, err
	}
	return true, nil
}

func (s *RegisterRewardService) resolveRewardConfig(ctx context.Context, logRecord *model.ResidentRegisterRewardLog, event registerRewardEvent) (registerRewardResolvedConfig, bool, error) {
	if logRecord.Status == registerRewardStatusFailed && (logRecord.GoldAmount > 0 || logRecord.RewardGroupID != nil) {
		return registerRewardResolvedConfig{
			GoldAmount:    logRecord.GoldAmount,
			RewardGroupID: logRecord.RewardGroupID,
		}, false, nil
	}

	cfg, err := s.loadRegisterRewardConfig(ctx, event.SysOrigin)
	if err != nil {
		return registerRewardResolvedConfig{}, false, err
	}
	if cfg == nil || !cfg.Enabled {
		err = s.updateLogStatus(
			ctx,
			logRecord.ID,
			registerRewardStatusSkipped,
			"register reward config missing or disabled",
			registerRewardResolvedConfig{},
		)
		return registerRewardResolvedConfig{}, err == nil, err
	}

	resolved := registerRewardResolvedConfig{
		GoldAmount:    cfg.GoldAmount,
		RewardGroupID: cfg.RewardGroupID,
	}
	if resolved.GoldAmount <= 0 && resolved.RewardGroupID == nil {
		err = s.updateLogStatus(
			ctx,
			logRecord.ID,
			registerRewardStatusSkipped,
			"register reward config has no reward payload",
			resolved,
		)
		return registerRewardResolvedConfig{}, err == nil, err
	}
	if err := s.updateLogStatus(ctx, logRecord.ID, registerRewardStatusPending, "", resolved); err != nil {
		return registerRewardResolvedConfig{}, false, err
	}
	return resolved, false, nil
}

func (s *RegisterRewardService) loadRegisterRewardConfig(ctx context.Context, sysOrigin string) (*model.ResidentRegisterRewardConfig, error) {
	var cfg model.ResidentRegisterRewardConfig
	err := s.repo.DB.WithContext(ctx).
		Where("sys_origin = ?", strings.ToUpper(strings.TrimSpace(sysOrigin))).
		First(&cfg).Error
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &cfg, nil
}

func (s *RegisterRewardService) dispatchGold(ctx context.Context, event registerRewardEvent, resolved registerRewardResolvedConfig) error {
	if resolved.GoldAmount <= 0 {
		return nil
	}
	return s.java.ChangeGoldBalance(ctx, integration.GoldReceiptCommand{
		ReceiptType:         "INCOME",
		UserID:              event.UserID,
		SysOrigin:           event.SysOrigin,
		EventID:             event.EventID + ":gold",
		Remark:              registerRewardGoldDesc,
		Amount:              integration.NewPennyAmountPayloadFromDollar(resolved.GoldAmount),
		CloseDelayAsset:     false,
		OpUserType:          "APP",
		CustomizeOrigin:     registerRewardGoldOrigin,
		CustomizeOriginDesc: registerRewardGoldDesc,
	})
}

func (s *RegisterRewardService) dispatchProps(ctx context.Context, logID int64, event registerRewardEvent, resolved registerRewardResolvedConfig) error {
	if resolved.RewardGroupID == nil {
		return nil
	}
	return s.java.GrantProps(ctx, integration.GrantPropsRequest{
		TrackID:       logID,
		AcceptUserID:  event.UserID,
		SysOrigin:     event.SysOrigin,
		Origin:        registerRewardPropsOrigin,
		SourceGroupID: *resolved.RewardGroupID,
	})
}

func (s *RegisterRewardService) getOrCreateLog(ctx context.Context, event registerRewardEvent, streamMessageID string) (*model.ResidentRegisterRewardLog, bool, error) {
	var record model.ResidentRegisterRewardLog
	err := s.repo.DB.WithContext(ctx).
		Where("event_id = ?", event.EventID).
		First(&record).Error
	if err == nil {
		if record.StreamMessageID != streamMessageID {
			if updateErr := s.repo.DB.WithContext(ctx).
				Model(&model.ResidentRegisterRewardLog{}).
				Where("id = ?", record.ID).
				Updates(map[string]any{
					"stream_message_id": streamMessageID,
					"update_time":       time.Now(),
				}).Error; updateErr == nil {
				record.StreamMessageID = streamMessageID
			}
		}
		return &record, isRegisterRewardTerminalStatus(record.Status), nil
	}
	if !errors.Is(err, gorm.ErrRecordNotFound) {
		return nil, false, err
	}

	id, idErr := util.NextID()
	if idErr != nil {
		return nil, false, idErr
	}
	now := time.Now()
	record = model.ResidentRegisterRewardLog{
		ID:              id,
		EventID:         event.EventID,
		StreamMessageID: streamMessageID,
		SysOrigin:       event.SysOrigin,
		UserID:          event.UserID,
		Status:          registerRewardStatusPending,
		CreateTime:      now,
		UpdateTime:      now,
	}
	if err := s.repo.DB.WithContext(ctx).Create(&record).Error; err != nil {
		if strings.Contains(strings.ToLower(err.Error()), "duplicate") {
			return s.getOrCreateLog(ctx, event, streamMessageID)
		}
		return nil, false, err
	}
	return &record, false, nil
}

func (s *RegisterRewardService) updateLogStatus(ctx context.Context, logID int64, status, errorMessage string, resolved registerRewardResolvedConfig) error {
	return s.repo.DB.WithContext(ctx).
		Model(&model.ResidentRegisterRewardLog{}).
		Where("id = ?", logID).
		Updates(map[string]any{
			"status":          status,
			"error_message":   truncateRegisterRewardMessage(errorMessage),
			"gold_amount":     resolved.GoldAmount,
			"reward_group_id": resolved.RewardGroupID,
			"update_time":     time.Now(),
		}).Error
}

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
}

func toRegisterRewardString(value any) string {
	switch typed := value.(type) {
	case nil:
		return ""
	case string:
		return typed
	case []byte:
		return string(typed)
	default:
		return fmt.Sprint(typed)
	}
}

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(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)
	}
}

func isRegisterRewardTerminalStatus(status string) bool {
	switch strings.ToUpper(strings.TrimSpace(status)) {
	case registerRewardStatusSuccess, registerRewardStatusSkipped:
		return true
	default:
		return false
	}
}

func isRegisterRewardNoGroupErr(err error) bool {
	if err == nil {
		return false
	}
	return strings.Contains(strings.ToUpper(err.Error()), "NOGROUP")
}

func truncateRegisterRewardMessage(message string) string {
	message = strings.TrimSpace(message)
	if len(message) <= 255 {
		return message
	}
	return message[:255]
}

func sleepWithContext(ctx context.Context, duration time.Duration) {
	timer := time.NewTimer(duration)
	defer timer.Stop()
	select {
	case <-ctx.Done():
	case <-timer.C:
	}
}