hyapp-server/services/activity-service/internal/service/growth/service.go

package growth

import (
	"context"
	"encoding/json"
	"strings"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/protobuf/proto"
	roomeventsv1 "hyapp.local/api/proto/events/room/v1"
	walletv1 "hyapp.local/api/proto/wallet/v1"
	"hyapp/pkg/appcode"
	"hyapp/pkg/xerr"
	domain "hyapp/services/activity-service/internal/domain/growth"
)

const levelRewardReason = "growth_level_reward"

// Repository 是等级体系的 activity MySQL 持久化边界。
type Repository interface {
	ListLevelOverview(ctx context.Context, userID int64, nowMS int64) (domain.Overview, error)
	GetLevelTrack(ctx context.Context, userID int64, track string, nowMS int64) (domain.TrackDetail, error)
	ListLevelConfig(ctx context.Context, query domain.ConfigQuery, nowMS int64) (domain.Config, error)
	ListLevelRewards(ctx context.Context, query domain.RewardQuery) ([]domain.RewardJob, int64, error)
	ConsumeLevelEvent(ctx context.Context, event domain.ValueEvent, nowMS int64) (domain.EventResult, error)
	UpsertLevelTrack(ctx context.Context, command domain.TrackCommand, nowMS int64) (domain.Track, bool, error)
	UpsertLevelRule(ctx context.Context, command domain.RuleCommand, nowMS int64) (domain.Rule, bool, error)
	UpsertLevelTier(ctx context.Context, command domain.TierCommand, nowMS int64) (domain.Tier, bool, error)
	ClaimPendingLevelRewardJobs(ctx context.Context, workerID string, nowMS int64, lockTTL time.Duration, batchSize int) ([]domain.RewardJob, error)
	MarkLevelRewardGranted(ctx context.Context, rewardJobID string, walletGrantID string, nowMS int64) error
	MarkLevelRewardFailed(ctx context.Context, rewardJobID string, failureReason string, nextRetryAtMS int64, nowMS int64) error
}

// WalletClient 是等级奖励发放的唯一外部副作用。
type WalletClient interface {
	GrantResourceGroup(ctx context.Context, req *walletv1.GrantResourceGroupRequest, opts ...grpc.CallOption) (*walletv1.ResourceGrantResponse, error)
}

// Service 承载 App 查询、事件消费、后台配置和等级奖励补偿。
type Service struct {
	repository Repository
	wallet     WalletClient
	now        func() time.Time
}

func New(repository Repository, wallet WalletClient) *Service {
	return &Service{repository: repository, wallet: wallet, now: time.Now}
}

func (s *Service) SetClock(now func() time.Time) {
	if now != nil {
		s.now = now
	}
}

func (s *Service) GetMyLevelOverview(ctx context.Context, userID int64) (domain.Overview, error) {
	if err := s.requireRepository(); err != nil {
		return domain.Overview{}, err
	}
	if userID <= 0 {
		return domain.Overview{}, xerr.New(xerr.InvalidArgument, "user_id is required")
	}
	return s.repository.ListLevelOverview(ctx, userID, s.now().UnixMilli())
}

func (s *Service) GetLevelTrack(ctx context.Context, userID int64, track string) (domain.TrackDetail, error) {
	if err := s.requireRepository(); err != nil {
		return domain.TrackDetail{}, err
	}
	track = normalizeTrack(track)
	if userID <= 0 || !validTrack(track) {
		return domain.TrackDetail{}, xerr.New(xerr.InvalidArgument, "user_id and track are required")
	}
	return s.repository.GetLevelTrack(ctx, userID, track, s.now().UnixMilli())
}

func (s *Service) ListLevelConfig(ctx context.Context, query domain.ConfigQuery) (domain.Config, error) {
	if err := s.requireRepository(); err != nil {
		return domain.Config{}, err
	}
	query.Track = normalizeTrack(query.Track)
	query.Status = strings.ToLower(strings.TrimSpace(query.Status))
	if query.Status == "all" {
		query.Status = ""
	}
	if query.Track != "" && !validTrack(query.Track) {
		return domain.Config{}, xerr.New(xerr.InvalidArgument, "track is invalid")
	}
	if query.Status != "" && !validStatus(query.Status) {
		return domain.Config{}, xerr.New(xerr.InvalidArgument, "status is invalid")
	}
	return s.repository.ListLevelConfig(ctx, query, s.now().UnixMilli())
}

func (s *Service) ListLevelRewards(ctx context.Context, query domain.RewardQuery) ([]domain.RewardJob, int64, error) {
	if err := s.requireRepository(); err != nil {
		return nil, 0, err
	}
	query.Track = normalizeTrack(query.Track)
	query.Status = strings.TrimSpace(query.Status)
	if query.UserID <= 0 {
		return nil, 0, xerr.New(xerr.InvalidArgument, "user_id is required")
	}
	if query.Track != "" && !validTrack(query.Track) {
		return nil, 0, xerr.New(xerr.InvalidArgument, "track is invalid")
	}
	if query.Page <= 0 {
		query.Page = 1
	}
	if query.PageSize <= 0 {
		query.PageSize = 20
	}
	if query.PageSize > 100 {
		query.PageSize = 100
	}
	return s.repository.ListLevelRewards(ctx, query)
}

func (s *Service) ConsumeLevelEvent(ctx context.Context, event domain.ValueEvent) (domain.EventResult, error) {
	if err := s.requireRepository(); err != nil {
		return domain.EventResult{}, err
	}
	event = normalizeValueEvent(event)
	if err := validateValueEvent(event); err != nil {
		return domain.EventResult{}, err
	}
	if event.OccurredAtMS <= 0 {
		event.OccurredAtMS = s.now().UnixMilli()
	}
	return s.repository.ConsumeLevelEvent(ctx, event, s.now().UnixMilli())
}

// HandleRoomEvent 把 room-service 已提交送礼事实映射为财富和魅力两条等级增量。
func (s *Service) HandleRoomEvent(ctx context.Context, envelope *roomeventsv1.EventEnvelope) (int32, error) {
	if envelope == nil {
		return 0, xerr.New(xerr.InvalidArgument, "room event envelope is required")
	}
	if envelope.GetEventType() != "RoomGiftSent" {
		return 0, nil
	}
	var gift roomeventsv1.RoomGiftSent
	if err := proto.Unmarshal(envelope.GetBody(), &gift); err != nil {
		return 0, err
	}
	if gift.GetSenderUserId() <= 0 || gift.GetTargetUserId() <= 0 || gift.GetGiftValue() <= 0 {
		return 0, xerr.New(xerr.InvalidArgument, "room gift event is incomplete")
	}
	dimensions, err := roomGiftDimensions(envelope, &gift)
	if err != nil {
		return 0, err
	}
	eventCtx := appcode.WithContext(ctx, envelope.GetAppCode())
	events := []domain.ValueEvent{
		{
			EventID:         envelope.GetEventId() + ":wealth",
			SourceEventID:   envelope.GetEventId(),
			SourceService:   "room-service",
			SourceEventType: envelope.GetEventType(),
			UserID:          gift.GetSenderUserId(),
			Track:           domain.TrackWealth,
			MetricType:      domain.MetricGiftSpendCoin,
			ValueDelta:      gift.GetGiftValue(),
			OccurredAtMS:    envelope.GetOccurredAtMs(),
			DimensionsJSON:  dimensions,
		},
		{
			EventID:         envelope.GetEventId() + ":charm",
			SourceEventID:   envelope.GetEventId(),
			SourceService:   "room-service",
			SourceEventType: envelope.GetEventType(),
			UserID:          gift.GetTargetUserId(),
			Track:           domain.TrackCharm,
			MetricType:      domain.MetricReceivedGiftValue,
			ValueDelta:      gift.GetGiftValue(),
			OccurredAtMS:    envelope.GetOccurredAtMs(),
			DimensionsJSON:  dimensions,
		},
	}
	consumed := int32(0)
	for _, event := range events {
		result, err := s.ConsumeLevelEvent(eventCtx, event)
		if err != nil {
			return consumed, err
		}
		if result.Status == domain.EventStatusConsumed {
			consumed++
		}
	}
	return consumed, nil
}

func (s *Service) UpsertLevelTrack(ctx context.Context, command domain.TrackCommand) (domain.Track, bool, error) {
	if err := s.requireRepository(); err != nil {
		return domain.Track{}, false, err
	}
	command.Track = normalizeTrack(command.Track)
	command.Name = strings.TrimSpace(command.Name)
	command.Status = normalizeStatus(command.Status)
	command.DisplayConfigJSON = normalizeJSON(command.DisplayConfigJSON)
	if !validTrack(command.Track) || command.Name == "" || !validStatus(command.Status) {
		return domain.Track{}, false, xerr.New(xerr.InvalidArgument, "level track command is invalid")
	}
	return s.repository.UpsertLevelTrack(ctx, command, s.now().UnixMilli())
}

func (s *Service) UpsertLevelRule(ctx context.Context, command domain.RuleCommand) (domain.Rule, bool, error) {
	if err := s.requireRepository(); err != nil {
		return domain.Rule{}, false, err
	}
	command.Track = normalizeTrack(command.Track)
	command.Name = strings.TrimSpace(command.Name)
	command.Status = normalizeStatus(command.Status)
	command.DisplayConfigJSON = normalizeJSON(command.DisplayConfigJSON)
	if !validTrack(command.Track) || command.Level <= 0 || command.RequiredValue <= 0 || command.Name == "" || !validStatus(command.Status) || command.OperatorAdminID <= 0 {
		return domain.Rule{}, false, xerr.New(xerr.InvalidArgument, "level rule command is invalid")
	}
	if command.RewardResourceGroupID < 0 {
		return domain.Rule{}, false, xerr.New(xerr.InvalidArgument, "reward_resource_group_id is invalid")
	}
	return s.repository.UpsertLevelRule(ctx, command, s.now().UnixMilli())
}

func (s *Service) UpsertLevelTier(ctx context.Context, command domain.TierCommand) (domain.Tier, bool, error) {
	if err := s.requireRepository(); err != nil {
		return domain.Tier{}, false, err
	}
	command.Track = normalizeTrack(command.Track)
	command.Name = strings.TrimSpace(command.Name)
	command.Status = normalizeStatus(command.Status)
	command.DisplayConfigJSON = normalizeJSON(command.DisplayConfigJSON)
	if !validTrack(command.Track) || command.MinLevel <= 0 || command.MaxLevel < command.MinLevel || command.Name == "" || !validStatus(command.Status) || command.OperatorAdminID <= 0 {
		return domain.Tier{}, false, xerr.New(xerr.InvalidArgument, "level tier command is invalid")
	}
	if command.DisplayAvatarFrameResourceID < 0 || command.DisplayBadgeResourceID < 0 || command.RewardResourceGroupID < 0 {
		return domain.Tier{}, false, xerr.New(xerr.InvalidArgument, "level tier resource id is invalid")
	}
	return s.repository.UpsertLevelTier(ctx, command, s.now().UnixMilli())
}

func (s *Service) ProcessLevelRewardBatch(ctx context.Context, runID string, workerID string, batchSize int, lockTTL time.Duration) (claimed int32, processed int32, success int32, failure int32, hasMore bool, err error) {
	if err := s.requireRepository(); err != nil {
		return 0, 0, 0, 0, false, err
	}
	if s.wallet == nil {
		return 0, 0, 0, 0, false, xerr.New(xerr.Unavailable, "wallet client is not configured")
	}
	if batchSize <= 0 {
		batchSize = 100
	}
	if lockTTL <= 0 {
		lockTTL = 30 * time.Second
	}
	if strings.TrimSpace(workerID) == "" {
		workerID = "activity-growth"
	}
	jobs, err := s.repository.ClaimPendingLevelRewardJobs(ctx, workerID, s.now().UnixMilli(), lockTTL, batchSize)
	if err != nil {
		return 0, 0, 0, 0, false, err
	}
	for _, job := range jobs {
		processed++
		resp, grantErr := s.wallet.GrantResourceGroup(ctx, &walletv1.GrantResourceGroupRequest{
			CommandId:      job.WalletCommandID,
			AppCode:        appcode.FromContext(ctx),
			TargetUserId:   job.UserID,
			GroupId:        job.ResourceGroupID,
			Reason:         levelRewardReason,
			OperatorUserId: job.UserID,
			GrantSource:    domain.GrantSourceGrowthLevel,
		})
		if grantErr != nil {
			failure++
			nextRetry := s.now().Add(time.Minute).UnixMilli()
			_ = s.repository.MarkLevelRewardFailed(ctx, job.RewardJobID, xerr.MessageOf(grantErr), nextRetry, s.now().UnixMilli())
			continue
		}
		walletGrantID := ""
		if resp.GetGrant() != nil {
			walletGrantID = resp.GetGrant().GetGrantId()
		}
		if markErr := s.repository.MarkLevelRewardGranted(ctx, job.RewardJobID, walletGrantID, s.now().UnixMilli()); markErr != nil {
			failure++
			continue
		}
		success++
	}
	return int32(len(jobs)), processed, success, failure, len(jobs) == batchSize, nil
}

func roomGiftDimensions(envelope *roomeventsv1.EventEnvelope, gift *roomeventsv1.RoomGiftSent) (string, error) {
	payload := map[string]any{
		"room_id":            envelope.GetRoomId(),
		"room_version":       envelope.GetRoomVersion(),
		"gift_id":            gift.GetGiftId(),
		"gift_count":         gift.GetGiftCount(),
		"billing_receipt_id": gift.GetBillingReceiptId(),
		"command_id":         gift.GetCommandId(),
		"visible_region_id":  gift.GetVisibleRegionId(),
	}
	data, err := json.Marshal(payload)
	if err != nil {
		return "", err
	}
	return string(data), nil
}

func (s *Service) requireRepository() error {
	if s == nil || s.repository == nil {
		return xerr.New(xerr.Unavailable, "growth repository is not configured")
	}
	return nil
}

func normalizeValueEvent(event domain.ValueEvent) domain.ValueEvent {
	event.EventID = strings.TrimSpace(event.EventID)
	event.SourceEventID = strings.TrimSpace(event.SourceEventID)
	event.SourceService = strings.TrimSpace(event.SourceService)
	event.SourceEventType = strings.TrimSpace(event.SourceEventType)
	event.Track = normalizeTrack(event.Track)
	event.MetricType = strings.ToLower(strings.TrimSpace(event.MetricType))
	event.DimensionsJSON = normalizeJSON(event.DimensionsJSON)
	return event
}

func validateValueEvent(event domain.ValueEvent) error {
	if event.EventID == "" || event.SourceEventID == "" || event.SourceService == "" || event.SourceEventType == "" || event.UserID <= 0 || event.ValueDelta <= 0 {
		return xerr.New(xerr.InvalidArgument, "level event is incomplete")
	}
	if !validTrack(event.Track) {
		return xerr.New(xerr.InvalidArgument, "track is invalid")
	}
	if !validMetricForTrack(event.Track, event.MetricType) {
		return xerr.New(xerr.InvalidArgument, "metric_type is invalid for track")
	}
	return nil
}

func normalizeTrack(value string) string {
	return strings.ToLower(strings.TrimSpace(value))
}

func normalizeStatus(value string) string {
	value = strings.ToLower(strings.TrimSpace(value))
	if value == "" {
		return domain.StatusActive
	}
	return value
}

func normalizeJSON(value string) string {
	value = strings.TrimSpace(value)
	if value == "" {
		return "{}"
	}
	return value
}

func validTrack(value string) bool {
	switch value {
	case domain.TrackWealth, domain.TrackGame, domain.TrackCharm:
		return true
	default:
		return false
	}
}

func validStatus(value string) bool {
	switch value {
	case domain.StatusActive, domain.StatusDisabled:
		return true
	default:
		return false
	}
}

func validMetricForTrack(track string, metric string) bool {
	switch track {
	case domain.TrackWealth:
		return metric == domain.MetricGiftSpendCoin
	case domain.TrackGame:
		return metric == domain.MetricGameSpendCoin
	case domain.TrackCharm:
		return metric == domain.MetricReceivedGiftValue
	default:
		return false
	}
}