501 lines
18 KiB
Go
501 lines
18 KiB
Go
package growth
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strconv"
|
|
"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"
|
|
|
|
const (
|
|
registrationLevelBadgeReason = "growth_level_registration_badge"
|
|
registrationLevelBadgeSourceLevel = int32(1)
|
|
)
|
|
|
|
// 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)
|
|
BatchGetUserLevelDisplayProfiles(ctx context.Context, userIDs []int64, nowMS int64) (domain.LevelDisplayProfiles, 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 {
|
|
GrantResource(ctx context.Context, req *walletv1.GrantResourceRequest, opts ...grpc.CallOption) (*walletv1.ResourceGrantResponse, error)
|
|
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) BatchGetUserLevelDisplayProfiles(ctx context.Context, userIDs []int64) (domain.LevelDisplayProfiles, error) {
|
|
if err := s.requireRepository(); err != nil {
|
|
return domain.LevelDisplayProfiles{}, err
|
|
}
|
|
normalized := make([]int64, 0, len(userIDs))
|
|
seen := make(map[int64]struct{}, len(userIDs))
|
|
for _, userID := range userIDs {
|
|
if userID <= 0 {
|
|
continue
|
|
}
|
|
if _, exists := seen[userID]; exists {
|
|
continue
|
|
}
|
|
seen[userID] = struct{}{}
|
|
normalized = append(normalized, userID)
|
|
}
|
|
if len(normalized) == 0 {
|
|
return domain.LevelDisplayProfiles{Profiles: []domain.UserLevelDisplayProfile{}, ServerTimeMS: s.now().UnixMilli()}, nil
|
|
}
|
|
return s.repository.BatchGetUserLevelDisplayProfiles(ctx, normalized, s.now().UnixMilli())
|
|
}
|
|
|
|
func (s *Service) IssueRegistrationLevelBadges(ctx context.Context, userID int64, commandID string) (domain.RegistrationLevelBadgeGrants, error) {
|
|
if err := s.requireRepository(); err != nil {
|
|
return domain.RegistrationLevelBadgeGrants{}, err
|
|
}
|
|
if s.wallet == nil {
|
|
return domain.RegistrationLevelBadgeGrants{}, xerr.New(xerr.Unavailable, "wallet client is not configured")
|
|
}
|
|
commandID = strings.TrimSpace(commandID)
|
|
if userID <= 0 || commandID == "" {
|
|
return domain.RegistrationLevelBadgeGrants{}, xerr.New(xerr.InvalidArgument, "registration level badge command is incomplete")
|
|
}
|
|
nowMS := s.now().UnixMilli()
|
|
config, err := s.repository.ListLevelConfig(ctx, domain.ConfigQuery{Status: domain.StatusActive}, nowMS)
|
|
if err != nil {
|
|
return domain.RegistrationLevelBadgeGrants{}, err
|
|
}
|
|
grants := make([]domain.RegistrationLevelBadgeGrant, 0, 3)
|
|
for _, track := range []string{domain.TrackWealth, domain.TrackGame, domain.TrackCharm} {
|
|
resourceID := levelRuleBadgeForExactLevel(config.Rules, track, registrationLevelBadgeSourceLevel)
|
|
if resourceID <= 0 {
|
|
// 没有配置 1 级等级徽章时跳过该轨道；注册主流程和其它轨道不受影响。
|
|
continue
|
|
}
|
|
// wallet 的 command_id 是实际发放幂等边界；按轨道拆分后，部分成功的重试只会补齐失败轨道。
|
|
resp, err := s.wallet.GrantResource(ctx, &walletv1.GrantResourceRequest{
|
|
CommandId: fmt.Sprintf("%s:%s", commandID, track),
|
|
AppCode: appcode.FromContext(ctx),
|
|
TargetUserId: userID,
|
|
ResourceId: resourceID,
|
|
Quantity: 1,
|
|
DurationMs: 0,
|
|
Reason: registrationLevelBadgeReason,
|
|
OperatorUserId: userID,
|
|
GrantSource: domain.GrantSourceGrowthLevel,
|
|
})
|
|
if err != nil {
|
|
return domain.RegistrationLevelBadgeGrants{}, err
|
|
}
|
|
grants = append(grants, domain.RegistrationLevelBadgeGrant{
|
|
Track: track,
|
|
Level: registrationLevelBadgeSourceLevel,
|
|
ResourceID: resourceID,
|
|
GrantID: resp.GetGrant().GetGrantId(),
|
|
})
|
|
}
|
|
return domain.RegistrationLevelBadgeGrants{Grants: grants, ServerTimeMS: nowMS}, nil
|
|
}
|
|
|
|
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 levelRuleBadgeForExactLevel(rules []domain.Rule, track string, level int32) int64 {
|
|
for _, rule := range rules {
|
|
if rule.Track != track || rule.Level != level || rule.Status != domain.StatusActive {
|
|
continue
|
|
}
|
|
return longBadgeResourceIDFromDisplayConfig(rule.DisplayConfigJSON)
|
|
}
|
|
return 0
|
|
}
|
|
|
|
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 longBadgeResourceIDFromDisplayConfig(raw string) int64 {
|
|
payload := map[string]any{}
|
|
if err := json.Unmarshal([]byte(strings.TrimSpace(raw)), &payload); err != nil {
|
|
return 0
|
|
}
|
|
value, ok := payload["long_badge_resource_id"]
|
|
if !ok {
|
|
return 0
|
|
}
|
|
switch typed := value.(type) {
|
|
case float64:
|
|
if typed > 0 {
|
|
return int64(typed)
|
|
}
|
|
case string:
|
|
parsed, err := strconv.ParseInt(strings.TrimSpace(typed), 10, 64)
|
|
if err == nil && parsed > 0 {
|
|
return parsed
|
|
}
|
|
}
|
|
return 0
|
|
}
|
|
|
|
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
|
|
}
|
|
}
|