package wallet import ( "context" activityv1 "hyapp.local/api/proto/activity/v1" "hyapp/pkg/appcode" "hyapp/pkg/walletmq" "hyapp/pkg/xerr" resourcedomain "hyapp/services/wallet-service/internal/domain/resource" "strings" "time" ) // ProjectPendingGiftWallEvents 消费钱包 outbox 中的送礼成功事件,异步维护用户礼物墙投影。 func (s *Service) ProjectPendingGiftWallEvents(ctx context.Context, workerID string, limit int, lockTTL time.Duration) (int, error) { workerID = strings.TrimSpace(workerID) if workerID == "" { return 0, xerr.New(xerr.InvalidArgument, "worker_id is required") } if s.repository == nil { return 0, xerr.New(xerr.Unavailable, "wallet repository is not configured") } return s.repository.ProjectPendingGiftWallEvents(ctx, workerID, limit, lockTTL) } // ProcessWalletProjectionMessage consumes one wallet_outbox MQ fact for wallet-owned projections. func (s *Service) ProcessWalletProjectionMessage(ctx context.Context, workerID string, message walletmq.WalletOutboxMessage, lockTTL time.Duration) (bool, error) { workerID = strings.TrimSpace(workerID) if workerID == "" { return false, xerr.New(xerr.InvalidArgument, "worker_id is required") } if s.repository == nil { return false, xerr.New(xerr.Unavailable, "wallet repository is not configured") } ctx = appcode.WithContext(ctx, message.AppCode) handled := false projected, err := s.repository.ProjectGiftWallMessage(ctx, workerID+"-gift-wall", message, lockTTL) if err != nil { return handled, err } handled = handled || projected projected, err = s.projectBadgeGrantMessage(ctx, workerID+"-badge-grant", message, lockTTL) if err != nil { return handled, err } handled = handled || projected return handled, nil } // ProjectPendingBadgeGrantEvents relays wallet resource grant outbox facts to activity badge display projection. func (s *Service) ProjectPendingBadgeGrantEvents(ctx context.Context, workerID string, limit int, lockTTL time.Duration) (int, error) { workerID = strings.TrimSpace(workerID) if workerID == "" { return 0, xerr.New(xerr.InvalidArgument, "worker_id is required") } if s.repository == nil { return 0, xerr.New(xerr.Unavailable, "wallet repository is not configured") } if s.activity == nil { return 0, xerr.New(xerr.Unavailable, "activity badge client is not configured") } if limit <= 0 { limit = 50 } if lockTTL <= 0 { lockTTL = 30 * time.Second } nowMS := s.now().UnixMilli() events, err := s.repository.ClaimPendingBadgeGrantEvents(ctx, workerID, nowMS, lockTTL, limit) if err != nil { return 0, err } processed := 0 for _, event := range events { processed++ if err := s.deliverBadgeGrantProjection(ctx, event); err != nil { return processed, err } } return processed, nil } func (s *Service) projectBadgeGrantMessage(ctx context.Context, workerID string, message walletmq.WalletOutboxMessage, lockTTL time.Duration) (bool, error) { if !isBadgeGrantWalletProjectionMessage(message) { return false, nil } if s.activity == nil { return false, xerr.New(xerr.Unavailable, "activity badge client is not configured") } nowMS := s.now().UnixMilli() event, claimed, err := s.repository.ClaimBadgeGrantMessage(ctx, workerID, message, nowMS, lockTTL) if err != nil || !claimed { return claimed, err } if err := s.deliverBadgeGrantProjection(ctx, event); err != nil { return true, err } return true, nil } func (s *Service) deliverBadgeGrantProjection(ctx context.Context, event resourcedomain.BadgeGrantOutbox) error { reqCtx := appcode.WithContext(ctx, event.AppCode) resp, consumeErr := s.activity.ConsumeAchievementEvent(reqCtx, &activityv1.ConsumeAchievementEventRequest{ Meta: &activityv1.RequestMeta{ Caller: "wallet-service", AppCode: event.AppCode, SentAtMs: s.now().UnixMilli(), }, EventId: event.EventID, EventType: event.EventType, SourceService: "wallet-service", UserId: event.UserID, MetricType: walletBadgeGrantMetric, Value: 1, OccurredAtMs: event.CreatedAtMS, DimensionsJson: event.PayloadJSON, }) if consumeErr != nil { return s.repository.MarkBadgeGrantEventFailed(ctx, event, xerr.MessageOf(consumeErr), s.now().UnixMilli()) } if !isBadgeGrantProjectionStatusTerminal(resp.GetStatus()) { return s.repository.MarkBadgeGrantEventFailed(ctx, event, "unexpected activity status: "+resp.GetStatus(), s.now().UnixMilli()) } return s.repository.MarkBadgeGrantEventDelivered(ctx, event, s.now().UnixMilli()) } func isBadgeGrantWalletProjectionMessage(message walletmq.WalletOutboxMessage) bool { if strings.TrimSpace(message.AssetType) != "RESOURCE" { return false } switch message.EventType { case "ResourceGranted", "ResourceGroupGranted": return true default: return false } } func isBadgeGrantProjectionStatusTerminal(status string) bool { switch status { case "consumed", "skipped", "duplicate": return true default: return false } }