388 lines
14 KiB
Go
388 lines
14 KiB
Go
package achievement
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"strings"
|
||
"time"
|
||
|
||
"google.golang.org/grpc"
|
||
walletv1 "hyapp.local/api/proto/wallet/v1"
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
domain "hyapp/services/activity-service/internal/domain/achievement"
|
||
)
|
||
|
||
const achievementRewardReason = "achievement_reward"
|
||
|
||
// Repository is the persistence boundary for achievements and badge display slots.
|
||
type Repository interface {
|
||
ListAchievements(ctx context.Context, query domain.ListQuery, nowMS int64) ([]domain.UserAchievement, int64, error)
|
||
ListAchievementDefinitions(ctx context.Context, query domain.ListQuery, nowMS int64) ([]domain.Definition, int64, error)
|
||
ConsumeAchievementEvent(ctx context.Context, event domain.Event, nowMS int64) (domain.EventResult, error)
|
||
ConsumeWalletBadgeGrantEvent(ctx context.Context, event domain.Event, nowMS int64) (domain.EventResult, error)
|
||
ListBadgeProfile(ctx context.Context, userID int64, nowMS int64) (domain.BadgeProfile, error)
|
||
SetBadgeDisplay(ctx context.Context, command domain.DisplayCommand, nowMS int64) (domain.BadgeProfile, error)
|
||
UpsertAchievementDefinition(ctx context.Context, command domain.DefinitionCommand, nowMS int64) (domain.Definition, bool, error)
|
||
DeleteAchievementDefinition(ctx context.Context, achievementID string, operatorAdminID int64, nowMS int64) (domain.Definition, error)
|
||
ClaimPendingAchievementRewardJobs(ctx context.Context, workerID string, nowMS int64, lockTTL time.Duration, batchSize int) ([]domain.RewardJob, error)
|
||
MarkAchievementRewardGranted(ctx context.Context, rewardJobID string, walletGrantID string, nowMS int64) error
|
||
MarkAchievementRewardFailed(ctx context.Context, rewardJobID string, failureReason string, nextRetryAtMS int64, nowMS int64) error
|
||
}
|
||
|
||
// WalletClient grants achievement reward resource groups.
|
||
type WalletClient interface {
|
||
GrantResourceGroup(ctx context.Context, req *walletv1.GrantResourceGroupRequest, opts ...grpc.CallOption) (*walletv1.ResourceGrantResponse, error)
|
||
}
|
||
|
||
// Service handles achievement reads, fact consumption, badge display and reward retries.
|
||
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) ListAchievements(ctx context.Context, query domain.ListQuery) ([]domain.UserAchievement, int64, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
query.CollectionType = normalizeCollectionType(query.CollectionType)
|
||
query.CollectionID = strings.TrimSpace(query.CollectionID)
|
||
query.Status = strings.TrimSpace(query.Status)
|
||
if query.UserID <= 0 {
|
||
return nil, 0, xerr.New(xerr.InvalidArgument, "user_id is required")
|
||
}
|
||
if query.CollectionType == "" && query.CollectionID == "" {
|
||
query.CollectionType = domain.CollectionOrdinary
|
||
}
|
||
if query.Page <= 0 {
|
||
query.Page = 1
|
||
}
|
||
if query.PageSize <= 0 {
|
||
query.PageSize = 50
|
||
}
|
||
if query.PageSize > 100 {
|
||
query.PageSize = 100
|
||
}
|
||
return s.repository.ListAchievements(ctx, query, s.now().UnixMilli())
|
||
}
|
||
|
||
func (s *Service) ListAchievementDefinitions(ctx context.Context, query domain.ListQuery) ([]domain.Definition, int64, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
// 后台删除成就只归档定义；默认列表隐藏 archived，避免运营误编辑已经从页面移除的成就。
|
||
query.CollectionType = normalizeCollectionType(query.CollectionType)
|
||
query.CollectionID = strings.TrimSpace(query.CollectionID)
|
||
query.Status = normalizeDefinitionStatusFilter(query.Status)
|
||
if query.Status != "" && !validDefinitionStatus(query.Status) {
|
||
return nil, 0, xerr.New(xerr.InvalidArgument, "achievement definition status is invalid")
|
||
}
|
||
if query.Page <= 0 {
|
||
query.Page = 1
|
||
}
|
||
if query.PageSize <= 0 {
|
||
query.PageSize = 50
|
||
}
|
||
if query.PageSize > 100 {
|
||
query.PageSize = 100
|
||
}
|
||
return s.repository.ListAchievementDefinitions(ctx, query, s.now().UnixMilli())
|
||
}
|
||
|
||
func (s *Service) ConsumeAchievementEvent(ctx context.Context, event domain.Event) (domain.EventResult, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.EventResult{}, err
|
||
}
|
||
event = normalizeEvent(event)
|
||
if event.EventID == "" || event.EventType == "" || event.SourceService == "" || event.UserID <= 0 || event.MetricType == "" || event.Value <= 0 {
|
||
return domain.EventResult{}, xerr.New(xerr.InvalidArgument, "achievement event is incomplete")
|
||
}
|
||
if event.OccurredAtMS <= 0 {
|
||
event.OccurredAtMS = s.now().UnixMilli()
|
||
}
|
||
if isWalletBadgeGrantEvent(event) {
|
||
return s.repository.ConsumeWalletBadgeGrantEvent(ctx, event, s.now().UnixMilli())
|
||
}
|
||
return s.repository.ConsumeAchievementEvent(ctx, event, s.now().UnixMilli())
|
||
}
|
||
|
||
func (s *Service) ListBadgeProfile(ctx context.Context, userID int64) (domain.BadgeProfile, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.BadgeProfile{}, err
|
||
}
|
||
if userID <= 0 {
|
||
return domain.BadgeProfile{}, xerr.New(xerr.InvalidArgument, "user_id is required")
|
||
}
|
||
return s.repository.ListBadgeProfile(ctx, userID, s.now().UnixMilli())
|
||
}
|
||
|
||
func (s *Service) SetBadgeDisplay(ctx context.Context, command domain.DisplayCommand) (domain.BadgeProfile, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.BadgeProfile{}, err
|
||
}
|
||
command.Slot = normalizeSlot(command.Slot)
|
||
command.CommandID = strings.TrimSpace(command.CommandID)
|
||
if command.UserID <= 0 || command.CommandID == "" || command.Slot == "" {
|
||
return domain.BadgeProfile{}, xerr.New(xerr.InvalidArgument, "badge display command is incomplete")
|
||
}
|
||
if command.Slot != domain.BadgeSlotProfileTile {
|
||
return domain.BadgeProfile{}, xerr.New(xerr.InvalidArgument, "only profile_tile can be changed by user")
|
||
}
|
||
if len(command.Items) > 6 {
|
||
return domain.BadgeProfile{}, xerr.New(xerr.InvalidArgument, "too many display badges")
|
||
}
|
||
for index := range command.Items {
|
||
item := &command.Items[index]
|
||
item.Slot = command.Slot
|
||
item.BadgeForm = domain.BadgeFormTile
|
||
item.SourceType = normalizeSourceType(item.SourceType)
|
||
if item.SourceType == "" {
|
||
item.SourceType = domain.SourceAchievement
|
||
}
|
||
item.PinMode = "user"
|
||
if item.Position <= 0 {
|
||
item.Position = int32(index + 1)
|
||
}
|
||
if item.ResourceID <= 0 {
|
||
return domain.BadgeProfile{}, xerr.New(xerr.InvalidArgument, "badge resource_id is required")
|
||
}
|
||
}
|
||
return s.repository.SetBadgeDisplay(ctx, command, s.now().UnixMilli())
|
||
}
|
||
|
||
func (s *Service) UpsertAchievementDefinition(ctx context.Context, command domain.DefinitionCommand) (domain.Definition, bool, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.Definition{}, false, err
|
||
}
|
||
command = normalizeDefinitionCommand(command)
|
||
if err := validateDefinitionCommand(command); err != nil {
|
||
return domain.Definition{}, false, err
|
||
}
|
||
return s.repository.UpsertAchievementDefinition(ctx, command, s.now().UnixMilli())
|
||
}
|
||
|
||
func (s *Service) DeleteAchievementDefinition(ctx context.Context, achievementID string, operatorAdminID int64) (domain.Definition, error) {
|
||
if err := s.requireRepository(); err != nil {
|
||
return domain.Definition{}, err
|
||
}
|
||
achievementID = strings.TrimSpace(achievementID)
|
||
if achievementID == "" || operatorAdminID <= 0 {
|
||
return domain.Definition{}, xerr.New(xerr.InvalidArgument, "achievement delete command is incomplete")
|
||
}
|
||
// 删除使用归档状态承载：历史用户解锁、徽章展示和审计链路都保留，App/后台默认查询会排除归档定义。
|
||
return s.repository.DeleteAchievementDefinition(ctx, achievementID, operatorAdminID, s.now().UnixMilli())
|
||
}
|
||
|
||
func (s *Service) ProcessAchievementRewardBatch(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-achievement"
|
||
}
|
||
jobs, err := s.repository.ClaimPendingAchievementRewardJobs(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: achievementRewardReason,
|
||
OperatorUserId: job.UserID,
|
||
GrantSource: domain.GrantSourceAchievement,
|
||
})
|
||
if grantErr != nil {
|
||
failure++
|
||
nextRetry := s.now().Add(time.Minute).UnixMilli()
|
||
_ = s.repository.MarkAchievementRewardFailed(ctx, job.RewardJobID, xerr.MessageOf(grantErr), nextRetry, s.now().UnixMilli())
|
||
continue
|
||
}
|
||
walletGrantID := ""
|
||
if resp.GetGrant() != nil {
|
||
walletGrantID = resp.GetGrant().GetGrantId()
|
||
}
|
||
if markErr := s.repository.MarkAchievementRewardGranted(ctx, job.RewardJobID, walletGrantID, s.now().UnixMilli()); markErr != nil {
|
||
failure++
|
||
continue
|
||
}
|
||
success++
|
||
}
|
||
return int32(len(jobs)), processed, success, failure, len(jobs) == batchSize, nil
|
||
}
|
||
|
||
func (s *Service) requireRepository() error {
|
||
if s == nil || s.repository == nil {
|
||
return xerr.New(xerr.Unavailable, "achievement repository is not configured")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func normalizeEvent(event domain.Event) domain.Event {
|
||
event.EventID = strings.TrimSpace(event.EventID)
|
||
event.EventType = strings.TrimSpace(event.EventType)
|
||
event.SourceService = strings.TrimSpace(event.SourceService)
|
||
event.MetricType = strings.ToLower(strings.TrimSpace(event.MetricType))
|
||
event.DimensionsJSON = normalizeJSON(event.DimensionsJSON)
|
||
return event
|
||
}
|
||
|
||
func isWalletBadgeGrantEvent(event domain.Event) bool {
|
||
if event.SourceService != "wallet-service" {
|
||
return false
|
||
}
|
||
switch event.EventType {
|
||
case "ResourceGranted", "ResourceGroupGranted":
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
func normalizeDefinitionCommand(command domain.DefinitionCommand) domain.DefinitionCommand {
|
||
command.AchievementID = strings.TrimSpace(command.AchievementID)
|
||
command.CollectionID = strings.TrimSpace(command.CollectionID)
|
||
command.CollectionType = normalizeCollectionType(command.CollectionType)
|
||
command.AchievementType = normalizeAchievementType(command.AchievementType)
|
||
command.Title = strings.TrimSpace(command.Title)
|
||
command.Description = strings.TrimSpace(command.Description)
|
||
command.Status = normalizeDefinitionStatus(command.Status)
|
||
command.AutoPinPolicy = normalizeAutoPinPolicy(command.AutoPinPolicy)
|
||
command.DisplayConfigJSON = normalizeJSON(command.DisplayConfigJSON)
|
||
if command.CollectionID == "" && command.CollectionType == domain.CollectionOrdinary {
|
||
command.CollectionID = domain.CollectionOrdinary
|
||
}
|
||
for index := range command.Conditions {
|
||
condition := &command.Conditions[index]
|
||
condition.MetricType = strings.ToLower(strings.TrimSpace(condition.MetricType))
|
||
condition.TargetUnit = strings.TrimSpace(condition.TargetUnit)
|
||
condition.DimensionFilterJSON = normalizeJSON(condition.DimensionFilterJSON)
|
||
}
|
||
return command
|
||
}
|
||
|
||
func validateDefinitionCommand(command domain.DefinitionCommand) error {
|
||
if command.CollectionType == "" || command.AchievementType == "" || command.Title == "" || command.OperatorAdminID <= 0 {
|
||
return xerr.New(xerr.InvalidArgument, "achievement definition command is incomplete")
|
||
}
|
||
if !validDefinitionStatus(command.Status) || !validAchievementType(command.AchievementType) {
|
||
return xerr.New(xerr.InvalidArgument, "achievement definition status or type is invalid")
|
||
}
|
||
if command.PrimaryBadgeResourceID <= 0 {
|
||
return xerr.New(xerr.InvalidArgument, "achievement badge resource is required")
|
||
}
|
||
if command.RewardResourceGroupID < 0 {
|
||
return xerr.New(xerr.InvalidArgument, "achievement reward resource is invalid")
|
||
}
|
||
if command.EffectiveFromMS > 0 && command.EffectiveToMS > 0 && command.EffectiveToMS <= command.EffectiveFromMS {
|
||
return xerr.New(xerr.InvalidArgument, "achievement effective time is invalid")
|
||
}
|
||
// 新版成就配置不再要求事件条件；历史条件保留兼容，存在时仍可被旧事件流推进。
|
||
for _, condition := range command.Conditions {
|
||
if condition.MetricType == "" || condition.TargetValue <= 0 {
|
||
return xerr.New(xerr.InvalidArgument, "achievement condition is invalid")
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func normalizeAchievementType(value string) string {
|
||
value = strings.ToLower(strings.TrimSpace(value))
|
||
if value == "" {
|
||
return domain.TypeOrdinary
|
||
}
|
||
return value
|
||
}
|
||
|
||
func normalizeCollectionType(value string) string {
|
||
value = strings.ToLower(strings.TrimSpace(value))
|
||
if value == "" {
|
||
return ""
|
||
}
|
||
return value
|
||
}
|
||
|
||
func normalizeDefinitionStatus(value string) string {
|
||
value = strings.ToLower(strings.TrimSpace(value))
|
||
if value == "" {
|
||
return domain.StatusDraft
|
||
}
|
||
return value
|
||
}
|
||
|
||
func normalizeDefinitionStatusFilter(value string) string {
|
||
value = strings.ToLower(strings.TrimSpace(value))
|
||
if value == "" {
|
||
return ""
|
||
}
|
||
return value
|
||
}
|
||
|
||
func normalizeAutoPinPolicy(value string) string {
|
||
value = strings.ToLower(strings.TrimSpace(value))
|
||
if value == "" {
|
||
return domain.AutoPinNone
|
||
}
|
||
return value
|
||
}
|
||
|
||
func normalizeSlot(value string) string {
|
||
return strings.ToLower(strings.TrimSpace(value))
|
||
}
|
||
|
||
func normalizeSourceType(value string) string {
|
||
return strings.ToLower(strings.TrimSpace(value))
|
||
}
|
||
|
||
func validDefinitionStatus(value string) bool {
|
||
switch value {
|
||
case domain.StatusDraft, domain.StatusActive, domain.StatusPaused, domain.StatusArchived:
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
func validAchievementType(value string) bool {
|
||
switch value {
|
||
case domain.TypeOrdinary, domain.TypeActivityLimited:
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
func normalizeJSON(value string) string {
|
||
value = strings.TrimSpace(value)
|
||
if value == "" {
|
||
return "{}"
|
||
}
|
||
if !json.Valid([]byte(value)) {
|
||
return "{}"
|
||
}
|
||
return value
|
||
}
|