456 lines
16 KiB
Go
456 lines
16 KiB
Go
package inviteactivity
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
|
||
"google.golang.org/grpc"
|
||
userv1 "hyapp.local/api/proto/user/v1"
|
||
walletv1 "hyapp.local/api/proto/wallet/v1"
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
domain "hyapp/services/activity-service/internal/domain/inviteactivity"
|
||
)
|
||
|
||
const grantReason = "invite_activity_reward"
|
||
|
||
type Repository interface {
|
||
GetInviteActivityRewardConfig(ctx context.Context) (domain.Config, bool, error)
|
||
UpdateInviteActivityRewardConfig(ctx context.Context, config domain.Config, nowMS int64) (domain.Config, error)
|
||
GetInviteActivityRewardProgress(ctx context.Context, cycleKey string, userID int64) (domain.Progress, bool, error)
|
||
ListInviteActivityRewardClaimsByUserCycle(ctx context.Context, cycleKey string, userID int64) ([]domain.Claim, error)
|
||
ConsumeInviteActivityRechargeEvent(ctx context.Context, event domain.RechargeEvent, cycle domain.Cycle, nowMS int64) (domain.ConsumeResult, error)
|
||
ConsumeInviteActivityValidInviteEvent(ctx context.Context, event domain.ValidInviteEvent, cycle domain.Cycle, nowMS int64) (domain.ConsumeResult, error)
|
||
CreateInviteActivityRewardClaim(ctx context.Context, claim domain.Claim, nowMS int64) (domain.Claim, error)
|
||
MarkInviteActivityRewardClaimGranted(ctx context.Context, claimID string, walletTransactionID string, grantedAtMS int64) (domain.Claim, error)
|
||
MarkInviteActivityRewardClaimFailed(ctx context.Context, claimID string, failureReason string, nowMS int64) error
|
||
ListInviteActivityRewardClaims(ctx context.Context, query domain.ClaimQuery) ([]domain.Claim, int64, error)
|
||
ListInviteActivityLeaderboard(ctx context.Context, query domain.LeaderboardQuery) ([]domain.LeaderboardEntry, int64, error)
|
||
}
|
||
|
||
type WalletClient interface {
|
||
CreditInviteActivityReward(ctx context.Context, req *walletv1.CreditInviteActivityRewardRequest, opts ...grpc.CallOption) (*walletv1.CreditInviteActivityRewardResponse, error)
|
||
}
|
||
|
||
type InviteAttributionSource interface {
|
||
GetInviteAttribution(ctx context.Context, invitedUserID int64) (*userv1.InviteAttribution, error)
|
||
}
|
||
|
||
type Service struct {
|
||
repository Repository
|
||
wallet WalletClient
|
||
invite InviteAttributionSource
|
||
now func() time.Time
|
||
}
|
||
|
||
func New(repository Repository, wallet WalletClient, invite InviteAttributionSource) *Service {
|
||
return &Service{repository: repository, wallet: wallet, invite: invite, now: time.Now}
|
||
}
|
||
|
||
func (s *Service) SetClock(now func() time.Time) {
|
||
if now != nil {
|
||
s.now = now
|
||
}
|
||
}
|
||
|
||
func (s *Service) GetStatus(ctx context.Context, userID int64) (domain.StatusResult, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.StatusResult{}, err
|
||
}
|
||
if userID <= 0 {
|
||
return domain.StatusResult{}, xerr.New(xerr.InvalidArgument, "user_id is required")
|
||
}
|
||
now := s.now().UTC()
|
||
cycle := CycleForTime(now)
|
||
result := domain.StatusResult{Cycle: cycle, ServerTimeMS: now.UnixMilli()}
|
||
|
||
config, exists, err := s.repository.GetInviteActivityRewardConfig(ctx)
|
||
if err != nil {
|
||
return domain.StatusResult{}, err
|
||
}
|
||
if exists {
|
||
config.Tiers = activeTiers(config.Tiers)
|
||
result.Config = config
|
||
} else {
|
||
result.Config = domain.Config{AppCode: appcode.FromContext(ctx), Enabled: true}
|
||
}
|
||
progress, _, err := s.repository.GetInviteActivityRewardProgress(ctx, cycle.Key, userID)
|
||
if err != nil {
|
||
return domain.StatusResult{}, err
|
||
}
|
||
result.Progress = progress
|
||
claims, err := s.repository.ListInviteActivityRewardClaimsByUserCycle(ctx, cycle.Key, userID)
|
||
if err != nil {
|
||
return domain.StatusResult{}, err
|
||
}
|
||
result.Claims = claims
|
||
return result, nil
|
||
}
|
||
|
||
func (s *Service) GetConfig(ctx context.Context) (domain.Config, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.Config{}, err
|
||
}
|
||
config, exists, err := s.repository.GetInviteActivityRewardConfig(ctx)
|
||
if err != nil {
|
||
return domain.Config{}, err
|
||
}
|
||
if !exists {
|
||
return domain.Config{AppCode: appcode.FromContext(ctx), Enabled: true}, nil
|
||
}
|
||
return config, nil
|
||
}
|
||
|
||
func (s *Service) UpdateConfig(ctx context.Context, config domain.Config) (domain.Config, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.Config{}, err
|
||
}
|
||
config.AppCode = appcode.FromContext(ctx)
|
||
config.Tiers = normalizeTiers(config.Tiers)
|
||
if err := validateConfig(config); err != nil {
|
||
return domain.Config{}, err
|
||
}
|
||
return s.repository.UpdateInviteActivityRewardConfig(ctx, config, s.now().UTC().UnixMilli())
|
||
}
|
||
|
||
func (s *Service) ConsumeWalletRechargeEvent(ctx context.Context, event domain.RechargeEvent) error {
|
||
if err := s.requireRepository(); err != nil {
|
||
return err
|
||
}
|
||
if event.EventType != domain.EventTypeWalletRechargeRecorded {
|
||
return nil
|
||
}
|
||
event = normalizeRechargeEvent(event)
|
||
if event.RechargeCoinAmount <= 0 || event.InvitedUserID <= 0 || event.EventID == "" {
|
||
return nil
|
||
}
|
||
if s.invite == nil {
|
||
return xerr.New(xerr.Unavailable, "invite attribution source is not configured")
|
||
}
|
||
// 邀请归因只由 user-service 拥有；activity-service 按充值事实反查邀请人，不复制邀请关系表。
|
||
attribution, err := s.invite.GetInviteAttribution(appcode.WithContext(ctx, event.AppCode), event.InvitedUserID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if attribution == nil || !attribution.GetFound() || attribution.GetInviterUserId() <= 0 {
|
||
return nil
|
||
}
|
||
event.InviterUserID = attribution.GetInviterUserId()
|
||
cycle := CycleForTime(time.UnixMilli(event.OccurredAtMS).UTC())
|
||
_, err = s.repository.ConsumeInviteActivityRechargeEvent(appcode.WithContext(ctx, event.AppCode), event, cycle, s.now().UTC().UnixMilli())
|
||
return err
|
||
}
|
||
|
||
func (s *Service) ConsumeValidInviteEvent(ctx context.Context, event domain.ValidInviteEvent) error {
|
||
if err := s.requireRepository(); err != nil {
|
||
return err
|
||
}
|
||
if event.EventType != domain.EventTypeUserInviteBecameValid || event.EventID == "" || event.InviterUserID <= 0 {
|
||
return nil
|
||
}
|
||
if event.OccurredAtMS <= 0 {
|
||
event.OccurredAtMS = s.now().UTC().UnixMilli()
|
||
}
|
||
cycle := CycleForTime(time.UnixMilli(event.OccurredAtMS).UTC())
|
||
_, err := s.repository.ConsumeInviteActivityValidInviteEvent(appcode.WithContext(ctx, event.AppCode), event, cycle, s.now().UTC().UnixMilli())
|
||
return err
|
||
}
|
||
|
||
func (s *Service) Claim(ctx context.Context, userID int64, rewardType string, tierID int64, commandID string) (domain.Claim, domain.StatusResult, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.Claim{}, domain.StatusResult{}, err
|
||
}
|
||
if s.wallet == nil {
|
||
return domain.Claim{}, domain.StatusResult{}, xerr.New(xerr.Unavailable, "wallet client is not configured")
|
||
}
|
||
rewardType = normalizeRewardType(rewardType)
|
||
commandID = strings.TrimSpace(commandID)
|
||
if userID <= 0 || rewardType == "" || tierID <= 0 || commandID == "" {
|
||
return domain.Claim{}, domain.StatusResult{}, xerr.New(xerr.InvalidArgument, "claim request is incomplete")
|
||
}
|
||
status, err := s.GetStatus(ctx, userID)
|
||
if err != nil {
|
||
return domain.Claim{}, domain.StatusResult{}, err
|
||
}
|
||
if !status.Config.Enabled {
|
||
return domain.Claim{}, domain.StatusResult{}, xerr.New(xerr.Conflict, "invite activity is disabled")
|
||
}
|
||
tier, ok := findTier(status.Config.Tiers, rewardType, tierID)
|
||
if !ok {
|
||
return domain.Claim{}, domain.StatusResult{}, xerr.New(xerr.NotFound, "invite activity tier not found")
|
||
}
|
||
reached := reachedValue(status.Progress, rewardType)
|
||
if !tierReached(tier, reached) {
|
||
return domain.Claim{}, domain.StatusResult{}, xerr.New(xerr.Conflict, "invite activity tier is not reached")
|
||
}
|
||
rewardAmount, err := claimRewardAmount(status.Claims, status.Config.Tiers, tier, reached)
|
||
if err != nil {
|
||
return domain.Claim{}, domain.StatusResult{}, err
|
||
}
|
||
claim := domain.Claim{
|
||
AppCode: appcode.FromContext(ctx),
|
||
ClaimID: fmt.Sprintf("IAR_%d_%s_%s_%d", userID, status.Cycle.Key, rewardType, tier.TierID),
|
||
CycleKey: status.Cycle.Key,
|
||
UserID: userID,
|
||
RewardType: rewardType,
|
||
CommandID: commandID,
|
||
TierID: tier.TierID,
|
||
TierCode: tier.TierCode,
|
||
TierName: tier.TierName,
|
||
ThresholdCoinAmount: tier.ThresholdCoinAmount,
|
||
ThresholdValidInviteCount: tier.ThresholdValidInviteCount,
|
||
ReachedValue: reached,
|
||
RewardCoinAmount: rewardAmount,
|
||
WalletCommandID: commandID,
|
||
}
|
||
created, err := s.repository.CreateInviteActivityRewardClaim(ctx, claim, s.now().UTC().UnixMilli())
|
||
if err != nil {
|
||
return domain.Claim{}, domain.StatusResult{}, err
|
||
}
|
||
if created.Status == domain.ClaimStatusGranted {
|
||
refreshed, refreshErr := s.GetStatus(ctx, userID)
|
||
return created, refreshed, refreshErr
|
||
}
|
||
resp, err := s.wallet.CreditInviteActivityReward(ctx, &walletv1.CreditInviteActivityRewardRequest{
|
||
CommandId: created.WalletCommandID,
|
||
AppCode: appcode.FromContext(ctx),
|
||
TargetUserId: userID,
|
||
Amount: created.RewardCoinAmount,
|
||
ClaimId: created.ClaimID,
|
||
RewardType: created.RewardType,
|
||
TierId: created.TierID,
|
||
TierCode: created.TierCode,
|
||
CycleKey: created.CycleKey,
|
||
ReachedValue: created.ReachedValue,
|
||
Reason: grantReason,
|
||
})
|
||
if err != nil {
|
||
_ = s.repository.MarkInviteActivityRewardClaimFailed(ctx, created.ClaimID, xerr.MessageOf(err), s.now().UTC().UnixMilli())
|
||
return domain.Claim{}, domain.StatusResult{}, err
|
||
}
|
||
updated, err := s.repository.MarkInviteActivityRewardClaimGranted(ctx, created.ClaimID, resp.GetTransactionId(), resp.GetGrantedAtMs())
|
||
if err != nil {
|
||
return domain.Claim{}, domain.StatusResult{}, err
|
||
}
|
||
refreshed, err := s.GetStatus(ctx, userID)
|
||
return updated, refreshed, err
|
||
}
|
||
|
||
func (s *Service) ListClaims(ctx context.Context, query domain.ClaimQuery) ([]domain.Claim, int64, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
query.Status = strings.TrimSpace(query.Status)
|
||
query.RewardType = normalizeRewardType(query.RewardType)
|
||
query.CycleKey = strings.TrimSpace(query.CycleKey)
|
||
if query.Page <= 0 {
|
||
query.Page = 1
|
||
}
|
||
if query.PageSize <= 0 {
|
||
query.PageSize = 20
|
||
}
|
||
if query.PageSize > 100 {
|
||
query.PageSize = 100
|
||
}
|
||
return s.repository.ListInviteActivityRewardClaims(ctx, query)
|
||
}
|
||
|
||
func (s *Service) ListLeaderboard(ctx context.Context, query domain.LeaderboardQuery) (domain.LeaderboardResult, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.LeaderboardResult{}, err
|
||
}
|
||
now := s.now().UTC()
|
||
query.CycleKey = strings.TrimSpace(query.CycleKey)
|
||
if query.CycleKey == "" {
|
||
query.CycleKey = CycleForTime(now).Key
|
||
}
|
||
if query.Page <= 0 {
|
||
query.Page = 1
|
||
}
|
||
if query.PageSize <= 0 {
|
||
query.PageSize = 50
|
||
}
|
||
if query.PageSize > 100 {
|
||
query.PageSize = 100
|
||
}
|
||
// 邀请活动榜单只对外展示“有效邀请人数”榜，累计充值金额只作为同分时的稳定排序依据和扩展字段返回。
|
||
entries, total, err := s.repository.ListInviteActivityLeaderboard(ctx, query)
|
||
if err != nil {
|
||
return domain.LeaderboardResult{}, err
|
||
}
|
||
return domain.LeaderboardResult{
|
||
Entries: entries,
|
||
Total: total,
|
||
CycleKey: query.CycleKey,
|
||
ServerTimeMS: now.UnixMilli(),
|
||
}, nil
|
||
}
|
||
|
||
func CycleForTime(value time.Time) domain.Cycle {
|
||
utc := value.UTC()
|
||
start := time.Date(utc.Year(), utc.Month(), 1, 0, 0, 0, 0, time.UTC)
|
||
end := start.AddDate(0, 1, 0)
|
||
return domain.Cycle{Key: start.Format("2006-01"), StartMS: start.UnixMilli(), EndMS: end.UnixMilli() - 1}
|
||
}
|
||
|
||
func normalizeTiers(tiers []domain.Tier) []domain.Tier {
|
||
items := make([]domain.Tier, 0, len(tiers))
|
||
for _, tier := range tiers {
|
||
tier.RewardType = normalizeRewardType(tier.RewardType)
|
||
tier.TierCode = strings.TrimSpace(tier.TierCode)
|
||
tier.TierName = strings.TrimSpace(tier.TierName)
|
||
tier.Status = strings.ToLower(strings.TrimSpace(tier.Status))
|
||
if tier.Status == "" {
|
||
tier.Status = domain.TierStatusActive
|
||
}
|
||
items = append(items, tier)
|
||
}
|
||
sort.SliceStable(items, func(i, j int) bool {
|
||
if items[i].RewardType == items[j].RewardType {
|
||
if items[i].SortOrder == items[j].SortOrder {
|
||
return thresholdValue(items[i]) < thresholdValue(items[j])
|
||
}
|
||
return items[i].SortOrder < items[j].SortOrder
|
||
}
|
||
return items[i].RewardType < items[j].RewardType
|
||
})
|
||
return items
|
||
}
|
||
|
||
func validateConfig(config domain.Config) error {
|
||
if config.UpdatedByAdminID <= 0 {
|
||
return xerr.New(xerr.InvalidArgument, "operator_admin_id is required")
|
||
}
|
||
if config.Enabled && len(activeTiers(config.Tiers)) == 0 {
|
||
return xerr.New(xerr.InvalidArgument, "enabled invite activity requires active tiers")
|
||
}
|
||
seen := map[string]struct{}{}
|
||
for _, tier := range config.Tiers {
|
||
if tier.RewardType == "" || tier.TierCode == "" || tier.RewardCoinAmount <= 0 {
|
||
return xerr.New(xerr.InvalidArgument, "invite activity tier is incomplete")
|
||
}
|
||
if tier.RewardType == domain.RewardTypeRecharge && tier.ThresholdCoinAmount <= 0 {
|
||
return xerr.New(xerr.InvalidArgument, "recharge tier threshold is required")
|
||
}
|
||
if tier.RewardType == domain.RewardTypeValidInvite && tier.ThresholdValidInviteCount <= 0 {
|
||
return xerr.New(xerr.InvalidArgument, "valid invite tier threshold is required")
|
||
}
|
||
key := fmt.Sprintf("%s:%d", tier.RewardType, thresholdValue(tier))
|
||
if _, ok := seen[key]; ok {
|
||
return xerr.New(xerr.InvalidArgument, "duplicate invite activity tier threshold")
|
||
}
|
||
seen[key] = struct{}{}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func claimRewardAmount(claims []domain.Claim, tiers []domain.Tier, tier domain.Tier, reached int64) (int64, error) {
|
||
if tier.RewardType == domain.RewardTypeValidInvite {
|
||
for _, claim := range claims {
|
||
if claim.RewardType == tier.RewardType && claim.TierID == tier.TierID && claim.Status == domain.ClaimStatusGranted {
|
||
return 0, xerr.New(xerr.Conflict, "invite activity tier already claimed")
|
||
}
|
||
}
|
||
return tier.RewardCoinAmount, nil
|
||
}
|
||
highest, ok := highestReachedTier(tiers, domain.RewardTypeRecharge, reached)
|
||
if !ok || highest.TierID != tier.TierID {
|
||
return 0, xerr.New(xerr.Conflict, "only highest recharge tier can be claimed")
|
||
}
|
||
var granted int64
|
||
for _, claim := range claims {
|
||
if claim.RewardType == domain.RewardTypeRecharge && claim.Status == domain.ClaimStatusGranted {
|
||
granted += claim.RewardCoinAmount
|
||
}
|
||
}
|
||
amount := tier.RewardCoinAmount - granted
|
||
if amount <= 0 {
|
||
return 0, xerr.New(xerr.Conflict, "invite activity reward already claimed")
|
||
}
|
||
return amount, nil
|
||
}
|
||
|
||
func activeTiers(tiers []domain.Tier) []domain.Tier {
|
||
items := make([]domain.Tier, 0, len(tiers))
|
||
for _, tier := range tiers {
|
||
if tier.Status == domain.TierStatusActive {
|
||
items = append(items, tier)
|
||
}
|
||
}
|
||
return items
|
||
}
|
||
|
||
func normalizeRewardType(value string) string {
|
||
switch strings.ToLower(strings.TrimSpace(value)) {
|
||
case domain.RewardTypeRecharge:
|
||
return domain.RewardTypeRecharge
|
||
case domain.RewardTypeValidInvite:
|
||
return domain.RewardTypeValidInvite
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
func normalizeRechargeEvent(event domain.RechargeEvent) domain.RechargeEvent {
|
||
event.AppCode = appcode.Normalize(event.AppCode)
|
||
event.EventID = strings.TrimSpace(event.EventID)
|
||
event.EventType = strings.TrimSpace(event.EventType)
|
||
event.TransactionID = strings.TrimSpace(event.TransactionID)
|
||
event.CommandID = strings.TrimSpace(event.CommandID)
|
||
event.RechargeType = strings.TrimSpace(event.RechargeType)
|
||
if event.OccurredAtMS <= 0 {
|
||
event.OccurredAtMS = time.Now().UTC().UnixMilli()
|
||
}
|
||
return event
|
||
}
|
||
|
||
func reachedValue(progress domain.Progress, rewardType string) int64 {
|
||
if rewardType == domain.RewardTypeValidInvite {
|
||
return progress.ValidInviteCount
|
||
}
|
||
return progress.TotalRechargeCoinAmount
|
||
}
|
||
|
||
func tierReached(tier domain.Tier, reached int64) bool {
|
||
return reached >= thresholdValue(tier)
|
||
}
|
||
|
||
func thresholdValue(tier domain.Tier) int64 {
|
||
if tier.RewardType == domain.RewardTypeValidInvite {
|
||
return tier.ThresholdValidInviteCount
|
||
}
|
||
return tier.ThresholdCoinAmount
|
||
}
|
||
|
||
func findTier(tiers []domain.Tier, rewardType string, tierID int64) (domain.Tier, bool) {
|
||
for _, tier := range tiers {
|
||
if tier.RewardType == rewardType && tier.TierID == tierID {
|
||
return tier, true
|
||
}
|
||
}
|
||
return domain.Tier{}, false
|
||
}
|
||
|
||
func highestReachedTier(tiers []domain.Tier, rewardType string, reached int64) (domain.Tier, bool) {
|
||
var selected domain.Tier
|
||
for _, tier := range tiers {
|
||
if tier.RewardType != rewardType || !tierReached(tier, reached) {
|
||
continue
|
||
}
|
||
if selected.TierID == 0 || thresholdValue(tier) > thresholdValue(selected) {
|
||
selected = tier
|
||
}
|
||
}
|
||
return selected, selected.TierID > 0
|
||
}
|
||
|
||
func (s *Service) requireRepository() error {
|
||
if s == nil || s.repository == nil {
|
||
return xerr.New(xerr.Unavailable, "invite activity repository is not configured")
|
||
}
|
||
return nil
|
||
}
|