497 lines
19 KiB
Go
497 lines
19 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)
|
||
DeleteBadgeDisplayItems(ctx context.Context, userID int64, items []domain.BadgeDisplayItem) 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 rewards and validates entitlement-backed badge display slots.
|
||
type WalletClient interface {
|
||
GrantResourceGroup(ctx context.Context, req *walletv1.GrantResourceGroupRequest, opts ...grpc.CallOption) (*walletv1.ResourceGrantResponse, error)
|
||
ListUserResources(ctx context.Context, req *walletv1.ListUserResourcesRequest, opts ...grpc.CallOption) (*walletv1.ListUserResourcesResponse, 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) {
|
||
// 先按 wallet 当前权益回收旧展示槽，再投影本次发放。否则已到期的体验卡徽章仍会占用
|
||
// profile_strip 的 6 个位置，使后续新体验卡的有效徽章被永久跳过。
|
||
if _, err := s.reconciledBadgeProfile(ctx, event.UserID); err != nil {
|
||
return domain.EventResult{}, err
|
||
}
|
||
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.reconciledBadgeProfile(ctx, userID)
|
||
}
|
||
|
||
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")
|
||
}
|
||
// 用户选择前先移除失效 entitlement 对应的旧槽，避免 repository 仅凭历史投影放行已过期徽章。
|
||
availableProfile, err := s.reconciledBadgeProfile(ctx, command.UserID)
|
||
if err != nil {
|
||
return domain.BadgeProfile{}, err
|
||
}
|
||
activeEntitlements := badgeDisplayEntitlementsByResource(availableProfile)
|
||
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")
|
||
}
|
||
// entitlement_id 是 wallet 事实，不能接受客户端回传的旧实例。成就型徽章没有
|
||
// entitlement_id，会继续保持空值并由 achievement 解锁事实授权。
|
||
item.EntitlementID = activeEntitlements[item.ResourceID]
|
||
}
|
||
if _, err := s.repository.SetBadgeDisplay(ctx, command, s.now().UnixMilli()); err != nil {
|
||
return domain.BadgeProfile{}, err
|
||
}
|
||
// 写入和返回之间权益仍可能刚好到期；最终响应继续以 wallet 当前有效事实收口。
|
||
return s.reconciledBadgeProfile(ctx, command.UserID)
|
||
}
|
||
|
||
func (s *Service) reconciledBadgeProfile(ctx context.Context, userID int64) (domain.BadgeProfile, error) {
|
||
profile, err := s.repository.ListBadgeProfile(ctx, userID, s.now().UnixMilli())
|
||
if err != nil {
|
||
return domain.BadgeProfile{}, err
|
||
}
|
||
if !badgeProfileHasEntitlements(profile) {
|
||
// 成就、等级等计算型展示没有 entitlement_id，不需要为纯投影额外访问 wallet。
|
||
return profile, nil
|
||
}
|
||
if s.wallet == nil {
|
||
// 无法确认权益时失败关闭，不能继续把可能已过期的 VIP/系统赠送徽章返回客户端。
|
||
return domain.BadgeProfile{}, xerr.New(xerr.Unavailable, "wallet client is not configured")
|
||
}
|
||
active, err := s.wallet.ListUserResources(ctx, &walletv1.ListUserResourcesRequest{
|
||
AppCode: appcode.FromContext(ctx),
|
||
UserId: userID,
|
||
ResourceType: "badge",
|
||
ActiveOnly: true,
|
||
})
|
||
if err != nil {
|
||
return domain.BadgeProfile{}, err
|
||
}
|
||
// 同一素材续期可能生成新的 entitlement_id；展示资格按当前有效 resource_id 判断，同时把
|
||
// 返回值替换成最新有效实例，避免客户端继续拿到上一张体验卡的过期 entitlement_id。
|
||
activeEntitlementByResource := make(map[int64]string, len(active.GetResources()))
|
||
for _, item := range active.GetResources() {
|
||
if item.GetResourceId() <= 0 || item.GetEntitlementId() == "" {
|
||
continue
|
||
}
|
||
if _, exists := activeEntitlementByResource[item.GetResourceId()]; !exists {
|
||
activeEntitlementByResource[item.GetResourceId()] = item.GetEntitlementId()
|
||
}
|
||
}
|
||
|
||
stale := make([]domain.BadgeDisplayItem, 0)
|
||
profile.StripBadges, stale = filterActiveBadgeDisplayItems(profile.StripBadges, activeEntitlementByResource, stale)
|
||
profile.ProfileTileBadges, stale = filterActiveBadgeDisplayItems(profile.ProfileTileBadges, activeEntitlementByResource, stale)
|
||
profile.HonorBadges, stale = filterActiveBadgeDisplayItems(profile.HonorBadges, activeEntitlementByResource, stale)
|
||
if len(stale) > 0 {
|
||
// 删除条件包含用户、槽位、位置、资源和旧 entitlement，和并发的新投影不相互覆盖。
|
||
if err := s.repository.DeleteBadgeDisplayItems(ctx, userID, stale); err != nil {
|
||
return domain.BadgeProfile{}, err
|
||
}
|
||
}
|
||
return profile, nil
|
||
}
|
||
|
||
func badgeProfileHasEntitlements(profile domain.BadgeProfile) bool {
|
||
for _, items := range [][]domain.BadgeDisplayItem{profile.StripBadges, profile.ProfileTileBadges, profile.HonorBadges} {
|
||
for _, item := range items {
|
||
if item.EntitlementID != "" {
|
||
return true
|
||
}
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
func badgeDisplayEntitlementsByResource(profile domain.BadgeProfile) map[int64]string {
|
||
result := make(map[int64]string)
|
||
for _, items := range [][]domain.BadgeDisplayItem{profile.ProfileTileBadges, profile.HonorBadges} {
|
||
for _, item := range items {
|
||
if item.ResourceID <= 0 || item.EntitlementID == "" {
|
||
continue
|
||
}
|
||
result[item.ResourceID] = item.EntitlementID
|
||
}
|
||
}
|
||
return result
|
||
}
|
||
|
||
func filterActiveBadgeDisplayItems(items []domain.BadgeDisplayItem, active map[int64]string, stale []domain.BadgeDisplayItem) ([]domain.BadgeDisplayItem, []domain.BadgeDisplayItem) {
|
||
filtered := make([]domain.BadgeDisplayItem, 0, len(items))
|
||
for _, item := range items {
|
||
if item.EntitlementID == "" {
|
||
filtered = append(filtered, item)
|
||
continue
|
||
}
|
||
entitlementID, ok := active[item.ResourceID]
|
||
if !ok {
|
||
stale = append(stale, item)
|
||
continue
|
||
}
|
||
item.EntitlementID = entitlementID
|
||
filtered = append(filtered, item)
|
||
}
|
||
return filtered, stale
|
||
}
|
||
|
||
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
|
||
}
|