// Package app 装配 user-service 的 gRPC 入口、领域服务、repository 和健康检查。 package app import ( "context" "encoding/json" "errors" "log/slog" "net" "strings" "sync" "time" "google.golang.org/protobuf/proto" roomeventsv1 "hyapp.local/api/proto/events/room/v1" userv1 "hyapp.local/api/proto/user/v1" "hyapp/pkg/appcode" "hyapp/pkg/grpcclient" "hyapp/pkg/grpchealth" "hyapp/pkg/healthhttp" "hyapp/pkg/idgen" "hyapp/pkg/logx" "hyapp/pkg/rocketmqx" "hyapp/pkg/roommq" serviceapp "hyapp/pkg/servicekit/app" servicegrpc "hyapp/pkg/servicekit/grpcserver" servicehealth "hyapp/pkg/servicekit/health" servicemq "hyapp/pkg/servicekit/mq" "hyapp/pkg/tencentim" "hyapp/pkg/usermq" "hyapp/pkg/walletmq" "hyapp/services/user-service/internal/config" cpdomain "hyapp/services/user-service/internal/domain/cp" invitedomain "hyapp/services/user-service/internal/domain/invite" mictimedomain "hyapp/services/user-service/internal/domain/mictime" "hyapp/services/user-service/internal/integration/activityclient" "hyapp/services/user-service/internal/integration/roomclient" "hyapp/services/user-service/internal/integration/walletclient" authservice "hyapp/services/user-service/internal/service/auth" cpservice "hyapp/services/user-service/internal/service/cp" hostservice "hyapp/services/user-service/internal/service/host" inviteservice "hyapp/services/user-service/internal/service/invite" mictimeservice "hyapp/services/user-service/internal/service/mictime" userservice "hyapp/services/user-service/internal/service/user" mysqlstorage "hyapp/services/user-service/internal/storage/mysql" grpcserver "hyapp/services/user-service/internal/transport/grpc" "google.golang.org/grpc" ) // App 装配 user-service gRPC 入口和底座依赖。 type App struct { // server 承载 AuthService、UserService、UserIdentityService 和 gRPC health。 server *grpc.Server // listener 是 user-service 的唯一网络入口。 listener net.Listener // health 是标准 gRPC health 的 serving/draining 状态来源。 health *grpchealth.ServingChecker // healthHTTP 给 CLB 和发布脚本提供 HTTP readiness，不承载业务请求。 healthHTTP *healthhttp.Server // mysqlRepo 持有用户主数据、认证身份、session 和短号事务的唯一运行时存储。 mysqlRepo *mysqlstorage.Repository // mqConsumers 按 consumer group 消费 wallet/room outbox 事实。 mqConsumers []*rocketmqx.Consumer // userOutboxProducer 把 user_outbox 事实发布到 RocketMQ，供统计等读模型消费。 userOutboxProducer *rocketmqx.Producer // roomConn 是低频平台治理调用 room-service 的 gRPC 连接。 roomConn *grpc.ClientConn // activityConn 是注册完成后触发注册奖励的 gRPC 连接。 activityConn *grpc.ClientConn // walletConn 是读取用户佩戴资源快照的 gRPC 连接。 walletConn *grpc.ClientConn // userSvc 持有用户主数据用例，cron RPC 复用它消费区域重算任务。 userSvc *userservice.Service // authSvc 持有认证、登录 session 和登录 IP 风控用例。 authSvc *authservice.Service // inviteSvc 消费 wallet recharge outbox，维护有效邀请 read model。 inviteSvc *inviteservice.Service // micTimeSvc 消费 room mic outbox，维护用户维度麦上时长 read model。 micTimeSvc *mictimeservice.Service // cpSvc 消费 room gift outbox，维护 CP/兄弟/姐妹申请和亲密值。 cpSvc *cpservice.Service // loginRiskRedisClose 关闭登录风险 Redis cache；cache 不作为启动硬依赖。 loginRiskRedisClose func() error // cpLeaderboardRedisClose 关闭 CP 亲密榜 Redis 读模型连接。 cpLeaderboardRedisClose func() error // cfg 保存 worker 运行参数，避免 Run 阶段重新读配置。 cfg config.Config // workers 统一控制后台 worker 生命周期。 workers *serviceapp.BackgroundGroup // closeOnce 防止信号退出和 Serve 异常同时触发重复关闭。 closeOnce sync.Once } // New 初始化 user-service 应用。 func New(cfg config.Config) (*App, error) { // userRepository 只表达用户主数据用例，authRepository 只表达认证用例；MySQL 实现同时满足两者。 startupCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() // user-service 的登录、注册、短号和 session 都必须落 MySQL；空 DSN 直接启动失败。 mysqlRepo, err := mysqlstorage.Open(startupCtx, cfg.MySQLDSN) if err != nil { return nil, err } // listener 在 New 阶段创建，端口占用会作为启动失败返回。 listener, err := net.Listen("tcp", cfg.GRPCAddr) if err != nil { if mysqlRepo != nil { _ = mysqlRepo.Close() } return nil, err } server := servicegrpc.New("user-service") thirdPartyVerifier := authservice.NewRoutedThirdPartyVerifier( authservice.NewAppFirebaseThirdPartyVerifier(authservice.AppFirebaseVerifierConfig{ Default: authservice.FirebaseVerifierConfig{ ProjectID: cfg.ThirdParty.Firebase.ProjectID, AllowedSignInProviders: cfg.ThirdParty.Firebase.AllowedSignInProviders, CertsURL: cfg.ThirdParty.Firebase.CertsURL, }, Projects: firebaseVerifierProjects(cfg.ThirdParty.Firebase), }), authservice.NewStaticThirdPartyVerifier(cfg.ThirdParty.AllowedProviders), ) authRepo := mysqlRepo.AuthRepository() appRepo := mysqlRepo.AppRepository() userRepo := mysqlRepo.UserRepository() identityRepo := mysqlRepo.IdentityRepository() regionRepo := mysqlRepo.RegionRepository() deviceRepo := mysqlRepo.DeviceRepository() hostRepo := mysqlRepo.HostRepository() inviteRepo := mysqlRepo.InviteRepository() micTimeRepo := mysqlRepo.MicTimeRepository() cpRepo := mysqlRepo.CPRepository() var ipDecisionCache authservice.IPDecisionCache var registerRiskConfigCache authservice.RegisterRiskConfigCache var loginRiskRedisClose func() error if cfg.LoginRisk.Enabled && cfg.LoginRisk.RedisAddr != "" { redisClient, redisErr := authservice.NewLoginRiskRedisClient(startupCtx, cfg.LoginRisk.RedisAddr, cfg.LoginRisk.RedisPassword, cfg.LoginRisk.RedisDB) if redisErr == nil { riskCache := authservice.NewRedisIPDecisionCache(redisClient) ipDecisionCache = riskCache registerRiskConfigCache = riskCache loginRiskRedisClose = redisClient.Close } else { // IP 风控 Redis 缓存按文档 fail-open；启动继续，worker 仍会写 MySQL 审计。 logx.Warn(startupCtx, "login_risk_redis_unavailable") } } var cpLeaderboardStore cpservice.IntimacyLeaderboardStore var cpLeaderboardRedisClose func() error if cfg.CPIntimacyLeaderboard.Enabled { // CP 亲密榜是 App 查询的唯一读模型，启动时必须能连上 Redis；否则服务启动成功但榜单接口必然不可用。 redisClient, redisErr := cpservice.NewRedisIntimacyLeaderboardClient(startupCtx, cfg.CPIntimacyLeaderboard.RedisAddr, cfg.CPIntimacyLeaderboard.RedisPassword, cfg.CPIntimacyLeaderboard.RedisDB) if redisErr != nil { if loginRiskRedisClose != nil { _ = loginRiskRedisClose() } _ = listener.Close() _ = mysqlRepo.Close() return nil, redisErr } // store 只持有 Redis client 和 key 前缀；具体 app_code/relation_type 在 cron 刷新和查询时动态决定。 cpLeaderboardStore = cpservice.NewRedisIntimacyLeaderboardStore(redisClient, cfg.CPIntimacyLeaderboard.KeyPrefix) cpLeaderboardRedisClose = redisClient.Close } var imLoginKicker userservice.IMLoginKicker var imAccountImporter authservice.IMAccountImporter if cfg.TencentIM.Enabled { imClient, err := tencentim.NewRESTClient(cfg.TencentIM.RESTConfig()) if err != nil { if loginRiskRedisClose != nil { _ = loginRiskRedisClose() } if cpLeaderboardRedisClose != nil { _ = cpLeaderboardRedisClose() } _ = listener.Close() _ = mysqlRepo.Close() return nil, err } imLoginKicker = imClient imAccountImporter = tencentIMAccountImporter{client: imClient} } var roomConn *grpc.ClientConn var roomEvictor userservice.RoomEvictor if cfg.RoomService.Enabled { roomConn, err = grpcclient.Dial(cfg.RoomService.Addr, grpcclient.Config{ DefaultTimeout: time.Duration(cfg.RoomService.RequestTimeoutMs) * time.Millisecond, }) if err != nil { if loginRiskRedisClose != nil { _ = loginRiskRedisClose() } if cpLeaderboardRedisClose != nil { _ = cpLeaderboardRedisClose() } _ = listener.Close() _ = mysqlRepo.Close() return nil, err } roomEvictor = roomclient.NewGRPC(roomConn) } var activityConn *grpc.ClientConn var registrationRewardIssuer authservice.RegistrationRewardIssuer var registrationLevelBadgeIssuer authservice.RegistrationLevelBadgeIssuer var levelProfileReader userservice.LevelProfileReader if cfg.ActivityService.Enabled { activityConn, err = grpcclient.Dial(cfg.ActivityService.Addr, grpcclient.Config{ DefaultTimeout: time.Duration(cfg.ActivityService.RequestTimeoutMs) * time.Millisecond, }) if err != nil { if loginRiskRedisClose != nil { _ = loginRiskRedisClose() } if cpLeaderboardRedisClose != nil { _ = cpLeaderboardRedisClose() } if roomConn != nil { _ = roomConn.Close() } _ = listener.Close() _ = mysqlRepo.Close() return nil, err } activityClient := activityclient.NewGRPC(activityConn) registrationRewardIssuer = activityClient registrationLevelBadgeIssuer = activityClient levelProfileReader = activityClient } var walletConn *grpc.ClientConn var avatarFrameReader cpservice.AvatarFrameReader if cfg.WalletService.Enabled { // 头像框佩戴事实归 wallet-service；user-service 刷新榜单前必须能读取该快照，否则榜单会缺少双方头像框。 walletConn, err = grpcclient.Dial(cfg.WalletService.Addr, grpcclient.Config{ DefaultTimeout: time.Duration(cfg.WalletService.RequestTimeoutMs) * time.Millisecond, }) if err != nil { if loginRiskRedisClose != nil { _ = loginRiskRedisClose() } if cpLeaderboardRedisClose != nil { _ = cpLeaderboardRedisClose() } if roomConn != nil { _ = roomConn.Close() } if activityConn != nil { _ = activityConn.Close() } _ = listener.Close() _ = mysqlRepo.Close() return nil, err } // 只把头像框读取能力注入 CP service，避免 CP 领域依赖完整的钱包客户端。 avatarFrameReader = walletclient.NewGRPC(walletConn) } // auth service 负责登录、session 和 token；用户主数据和短号事务仍通过各自领域存储完成。 authSvc := authservice.New(authservice.Config{ Issuer: cfg.JWT.Issuer, AccessTokenTTLSec: cfg.JWT.AccessTokenTTLSec, RefreshTokenTTLSec: cfg.JWT.RefreshTokenTTLSec, SigningAlg: cfg.JWT.SigningAlg, SigningSecret: cfg.JWT.SigningSecret, }, authservice.WithAuthRepository(authRepo), authservice.WithUserRepository(userRepo), authservice.WithIdentityRepository(identityRepo), authservice.WithCountryRegionRepository(regionRepo), authservice.WithIDGenerator(idgen.NewInt64Generator(cfg.IDGenerator.NodeID)), authservice.WithDisplayUserIDAllocateMaxAttempts(cfg.DisplayUserID.AllocateMaxAttempts), authservice.WithThirdPartyVerifier(thirdPartyVerifier), authservice.WithLoginRiskPolicy(authservice.LoginRiskPolicy{ Enabled: cfg.LoginRisk.Enabled, UnsupportedCountries: cfg.LoginRisk.UnsupportedCountries, BlockedTTL: time.Duration(cfg.LoginRisk.BlockedTTLSec) * time.Second, AllowedTTL: time.Duration(cfg.LoginRisk.AllowedTTLSec) * time.Second, UnknownTTL: time.Duration(cfg.LoginRisk.UnknownTTLSec) * time.Second, ProviderTimeout: time.Duration(cfg.LoginRisk.ProviderTimeoutMs) * time.Millisecond, ProviderNames: cfg.LoginRisk.Providers, DenylistTTL: time.Duration(cfg.JWT.AccessTokenTTLSec+60) * time.Second, }), authservice.WithIPDecisionCache(ipDecisionCache), authservice.WithRegisterRiskConfigCache(registerRiskConfigCache), authservice.WithIPGeoProviders(authservice.BuildIPGeoProviders(cfg.LoginRisk.Providers)), authservice.WithRegistrationRewardIssuer(registrationRewardIssuer), authservice.WithRegistrationLevelBadgeIssuer(registrationLevelBadgeIssuer), authservice.WithIMAccountImporter(imAccountImporter), ) // user service 负责用户主状态和 display_user_id/靓号用例，不进入房间高频流程。 userSvc := userservice.New(userRepo, userservice.WithAppRegistryRepository(appRepo), userservice.WithIdentityRepository(identityRepo), userservice.WithLevelProfileReader(levelProfileReader), userservice.WithCountryRegionRepository(regionRepo), userservice.WithCountryAdminRepository(regionRepo), userservice.WithRegionAdminRepository(regionRepo), userservice.WithRegionRebuildRepository(regionRepo), userservice.WithRoleSummaryRepository(hostRepo), userservice.WithDeviceRepository(deviceRepo), userservice.WithModerationRepository(userRepo), userservice.WithSessionDenylist(ipDecisionCache, time.Duration(cfg.JWT.AccessTokenTTLSec+60)*time.Second), userservice.WithIMLoginKicker(imLoginKicker), userservice.WithRoomEvictor(roomEvictor), userservice.WithIDGenerator(idgen.NewInt64Generator(cfg.IDGenerator.NodeID)), userservice.WithDisplayUserIDPolicy(cfg.DisplayUserID.AllocateMaxAttempts, time.Duration(cfg.DisplayUserID.ChangeCooldownSec)*time.Second), ) hostSvc := hostservice.New(hostRepo, hostservice.WithIDGenerator(idgen.NewInt64Generator(cfg.IDGenerator.NodeID)), ) inviteSvc := inviteservice.New(inviteRepo) micTimeSvc := mictimeservice.New(micTimeRepo) cpSvc := cpservice.New(cpRepo, // CP service 同时承担送礼事件消费和榜单读模型刷新；Redis/wallet 依赖只在榜单相关方法里被强校验。 cpservice.WithIntimacyLeaderboardStore(cpLeaderboardStore), cpservice.WithAvatarFrameReader(avatarFrameReader), ) mqConsumers, err := newMQConsumers(cfg, inviteSvc, micTimeSvc, cpSvc) if err != nil { if loginRiskRedisClose != nil { _ = loginRiskRedisClose() } if cpLeaderboardRedisClose != nil { _ = cpLeaderboardRedisClose() } if roomConn != nil { _ = roomConn.Close() } if activityConn != nil { _ = activityConn.Close() } if walletConn != nil { _ = walletConn.Close() } _ = listener.Close() _ = mysqlRepo.Close() return nil, err } var userOutboxProducer *rocketmqx.Producer if cfg.RocketMQ.UserOutbox.Enabled { userOutboxProducer, err = rocketmqx.NewProducer(userOutboxProducerConfig(cfg.RocketMQ)) if err != nil { shutdownMQConsumers(mqConsumers) if loginRiskRedisClose != nil { _ = loginRiskRedisClose() } if cpLeaderboardRedisClose != nil { _ = cpLeaderboardRedisClose() } if roomConn != nil { _ = roomConn.Close() } if activityConn != nil { _ = activityConn.Close() } if walletConn != nil { _ = walletConn.Close() } _ = listener.Close() _ = mysqlRepo.Close() return nil, err } } userServer := grpcserver.NewServer(authSvc, userSvc, hostSvc) userServer.SetMicTimeService(micTimeSvc) userServer.SetCPService(cpSvc) // 多个 protobuf service 共用一个 Server 适配器，领域逻辑仍拆在 auth/user/host service。 userv1.RegisterAuthServiceServer(server, userServer) userv1.RegisterUserServiceServer(server, userServer) userv1.RegisterUserSocialServiceServer(server, userServer) userv1.RegisterUserDeviceServiceServer(server, userServer) userv1.RegisterAppRegistryServiceServer(server, userServer) userv1.RegisterUserIdentityServiceServer(server, userServer) userv1.RegisterUserPrettyDisplayIDAdminServiceServer(server, userServer) userv1.RegisterCountryAdminServiceServer(server, userServer) userv1.RegisterCountryQueryServiceServer(server, userServer) userv1.RegisterRegionAdminServiceServer(server, userServer) userv1.RegisterUserHostServiceServer(server, userServer) userv1.RegisterUserHostAdminServiceServer(server, userServer) userv1.RegisterUserCronServiceServer(server, userServer) userv1.RegisterUserCPServiceServer(server, userServer) userv1.RegisterUserCPInternalServiceServer(server, userServer) healthDependencies := []grpchealth.Dependency{{ Name: "mysql", Check: mysqlRepo.Ping, }} health := grpchealth.NewServingChecker("user-service", healthDependencies...) // user-service 目前使用简单 serving checker，存储探测由 servicekit 统一注册。 healthHTTP, err := servicehealth.Register(server, health, cfg.HealthHTTPAddr, cfg.NodeID) if err != nil { shutdownMQConsumers(mqConsumers) if loginRiskRedisClose != nil { _ = loginRiskRedisClose() } if cpLeaderboardRedisClose != nil { _ = cpLeaderboardRedisClose() } if roomConn != nil { _ = roomConn.Close() } if activityConn != nil { _ = activityConn.Close() } if walletConn != nil { _ = walletConn.Close() } _ = listener.Close() _ = mysqlRepo.Close() return nil, err } return &App{ server: server, listener: listener, health: health, healthHTTP: healthHTTP, mysqlRepo: mysqlRepo, mqConsumers: mqConsumers, userOutboxProducer: userOutboxProducer, roomConn: roomConn, activityConn: activityConn, walletConn: walletConn, authSvc: authSvc, userSvc: userSvc, inviteSvc: inviteSvc, micTimeSvc: micTimeSvc, cpSvc: cpSvc, loginRiskRedisClose: loginRiskRedisClose, cpLeaderboardRedisClose: cpLeaderboardRedisClose, cfg: cfg, workers: serviceapp.NewBackground(context.Background()), }, nil } // Run 启动 gRPC 服务。 func (a *App) Run() error { a.runHealthHTTP() if err := startMQConsumers(a.mqConsumers); err != nil { a.health.MarkStopped() return err } if err := a.startUserOutboxProducer(); err != nil { a.health.MarkStopped() return err } a.runUserOutboxWorker() // 只有 listener 进入 Serve 后才标记 serving，避免健康检查提前放行。 a.health.MarkServing() defer func() { a.workers.StopAndWait() shutdownMQConsumers(a.mqConsumers) a.shutdownUserOutboxProducer() }() err := a.server.Serve(a.listener) a.health.MarkStopped() if errors.Is(err, grpc.ErrServerStopped) { // GracefulStop 触发的退出不是故障。 return nil } return err } // Close 关闭 gRPC 服务和底层持久化连接。 func (a *App) Close() { a.closeOnce.Do(func() { // draining 会让 health 立即失败，避免下线进程继续接新 RPC。 a.health.MarkDraining() a.workers.StopAndWait() shutdownMQConsumers(a.mqConsumers) a.shutdownUserOutboxProducer() // GracefulStop 等待已进入的 RPC 结束，避免 token/session 写入被硬切。 servicegrpc.GracefulStop(a.server, 15*time.Second) a.closeHealthHTTP() if a.mysqlRepo != nil { // MySQL 连接池最后关闭，保证 GracefulStop 期间 repository 仍可用。 _ = a.mysqlRepo.Close() } if a.roomConn != nil { _ = a.roomConn.Close() } if a.activityConn != nil { _ = a.activityConn.Close() } if a.walletConn != nil { _ = a.walletConn.Close() } if a.loginRiskRedisClose != nil { _ = a.loginRiskRedisClose() } if a.cpLeaderboardRedisClose != nil { _ = a.cpLeaderboardRedisClose() } }) } func (a *App) runUserOutboxWorker() { if !a.cfg.OutboxWorker.Enabled || a.userOutboxProducer == nil || a.mysqlRepo == nil { return } workerID := "user-outbox-" + a.cfg.NodeID a.workers.Go(func(ctx context.Context) { ticker := time.NewTicker(a.cfg.OutboxWorker.PollInterval) defer ticker.Stop() for { processed, err := a.processUserOutboxBatch(workerID) if err != nil && !errors.Is(err, context.Canceled) { logx.Error(ctx, "user_outbox_publish_failed", err, slog.String("worker_id", workerID)) } if processed >= a.cfg.OutboxWorker.BatchSize { continue } select { case <-ctx.Done(): return case <-ticker.C: } } }) } func (a *App) processUserOutboxBatch(workerID string) (int, error) { records, err := a.mysqlRepo.ClaimPendingUserOutbox(a.workers.Context(), workerID, a.cfg.OutboxWorker.BatchSize) if err != nil { return 0, err } for _, record := range records { publishCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), a.cfg.OutboxWorker.PublishTimeout) err := a.publishUserOutboxRecord(publishCtx, record) cancel() markCtx, markCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), a.cfg.OutboxWorker.PublishTimeout) if err != nil { nextRetryAtMS := time.Now().UTC().Add(a.cfg.OutboxWorker.PollInterval).UnixMilli() markErr := a.mysqlRepo.MarkUserOutboxRetryable(markCtx, record.EventID, err.Error(), nextRetryAtMS) markCancel() if markErr != nil { return 0, markErr } return len(records), err } markErr := a.mysqlRepo.MarkUserOutboxDelivered(markCtx, record.EventID) markCancel() if markErr != nil { return 0, markErr } } return len(records), nil } func (a *App) publishUserOutboxRecord(ctx context.Context, record mysqlstorage.UserOutboxRecord) error { body, err := usermq.EncodeUserOutboxMessage(usermq.UserOutboxMessage{ AppCode: record.AppCode, EventID: record.EventID, EventType: record.EventType, AggregateType: record.AggregateType, AggregateID: record.AggregateID, PayloadJSON: record.PayloadJSON, OccurredAtMS: record.CreatedAtMS, }) if err != nil { return err } return a.userOutboxProducer.SendSync(ctx, rocketmqx.Message{ Topic: a.cfg.RocketMQ.UserOutbox.Topic, Tag: usermq.TagUserOutboxEvent, Keys: []string{record.EventID, record.AggregateType}, Body: body, }) } func (a *App) startUserOutboxProducer() error { if a.userOutboxProducer == nil { return nil } return servicemq.StartProducers([]*rocketmqx.Producer{a.userOutboxProducer}) } func (a *App) shutdownUserOutboxProducer() { if a.userOutboxProducer == nil { return } servicemq.ShutdownProducers([]*rocketmqx.Producer{a.userOutboxProducer}) } func (a *App) runHealthHTTP() { if a.healthHTTP == nil { return } go func() { if err := a.healthHTTP.Run(); err != nil { logx.Error(context.Background(), "health_http_run_failed", err) } }() } func (a *App) closeHealthHTTP() { if a.healthHTTP == nil { return } ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() _ = a.healthHTTP.Close(ctx) } func firebaseVerifierProjects(cfg config.FirebaseConfig) map[string]authservice.FirebaseVerifierConfig { projects := make(map[string]authservice.FirebaseVerifierConfig, len(cfg.Projects)) for appCode, project := range cfg.Projects { // app_code 已在配置加载阶段归一化；这里仍保留 Normalize，避免测试或手动构造配置绕过 Load。 projects[appcode.Normalize(appCode)] = authservice.FirebaseVerifierConfig{ ProjectID: project.ProjectID, AllowedSignInProviders: project.AllowedSignInProviders, CertsURL: project.CertsURL, } } return projects } type tencentIMAccountImporter struct { client *tencentim.RESTClient } func (i tencentIMAccountImporter) ImportAccount(ctx context.Context, userID int64, nickname string, faceURL string) error { return i.client.ImportAccount(ctx, tencentim.AccountProfile{ UserID: userID, Nick: nickname, FaceURL: faceURL, }) } func newMQConsumers(cfg config.Config, inviteSvc *inviteservice.Service, micTimeSvc *mictimeservice.Service, cpSvc *cpservice.Service) ([]*rocketmqx.Consumer, error) { consumers := make([]*rocketmqx.Consumer, 0, 3) if cfg.InviteRechargeWorker.Enabled && cfg.RocketMQ.WalletOutbox.Enabled { consumer, err := rocketmqx.NewConsumer(walletOutboxConsumerConfig(cfg.RocketMQ)) if err != nil { return nil, err } if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { event, ok, err := inviteRechargeEventFromWalletMessage(message.Body) if err != nil || !ok { return err } _, err = inviteSvc.ConsumeWalletRechargeEvent(appcode.WithContext(ctx, event.AppCode), event) return err }); err != nil { _ = consumer.Shutdown() return nil, err } consumers = append(consumers, consumer) } if cfg.MicTimeWorker.Enabled && cfg.RocketMQ.RoomOutbox.Enabled { consumer, err := rocketmqx.NewConsumer(roomOutboxConsumerConfig(cfg.RocketMQ)) if err != nil { shutdownMQConsumers(consumers) return nil, err } if err := consumer.Subscribe(cfg.RocketMQ.RoomOutbox.Topic, roommq.TagRoomOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { event, ok, err := roomMicEventFromRoomMessage(message.Body) if err != nil || !ok { return err } _, err = micTimeSvc.ConsumeRoomMicEvent(appcode.WithContext(ctx, event.AppCode), event) return err }); err != nil { _ = consumer.Shutdown() shutdownMQConsumers(consumers) return nil, err } consumers = append(consumers, consumer) } if cfg.CPRelationshipWorker.Enabled && cfg.RocketMQ.RoomOutbox.Enabled { // CP 关系消费使用独立 consumer group，保证和麦位时长读模型各自拥有 MQ 位点和重试节奏。 consumer, err := rocketmqx.NewConsumer(cpRoomOutboxConsumerConfig(cfg)) if err != nil { shutdownMQConsumers(consumers) return nil, err } if err := consumer.Subscribe(cfg.RocketMQ.RoomOutbox.Topic, roommq.TagRoomOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error { event, ok, err := roomGiftCPEventFromRoomMessage(message.Body) if err != nil || !ok { return err } _, err = cpSvc.ConsumeGiftEvent(appcode.WithContext(ctx, event.AppCode), event) return err }); err != nil { _ = consumer.Shutdown() shutdownMQConsumers(consumers) return nil, err } consumers = append(consumers, consumer) } return consumers, nil } func inviteRechargeEventFromWalletMessage(body []byte) (invitedomain.RechargeEvent, bool, error) { message, err := walletmq.DecodeWalletOutboxMessage(body) if err != nil { return invitedomain.RechargeEvent{}, false, err } if message.EventType != invitedomain.RechargeEventWalletRecorded { return invitedomain.RechargeEvent{}, false, nil } return invitedomain.RechargeEvent{ AppCode: appcode.Normalize(message.AppCode), EventID: message.EventID, EventType: message.EventType, TransactionID: message.TransactionID, CommandID: message.CommandID, InvitedUserID: message.UserID, AssetType: message.AssetType, RechargeCoinAmount: message.AvailableDelta, RechargeType: rechargeTypeFromWalletPayload(message.PayloadJSON), OccurredAtMs: message.OccurredAtMS, PayloadJSON: message.PayloadJSON, }, true, nil } func roomMicEventFromRoomMessage(body []byte) (mictimedomain.RoomMicEvent, bool, error) { envelope, _, err := roommq.DecodeRoomOutboxMessage(body) if err != nil { return mictimedomain.RoomMicEvent{}, false, err } if envelope.GetEventType() != mictimedomain.RoomMicChangedEventType { return mictimedomain.RoomMicEvent{}, false, nil } var changed roomeventsv1.RoomMicChanged if err := proto.Unmarshal(envelope.GetBody(), &changed); err != nil { return mictimedomain.RoomMicEvent{}, false, err } return mictimedomain.RoomMicEvent{ AppCode: appcode.Normalize(envelope.GetAppCode()), EventID: envelope.GetEventId(), EventType: envelope.GetEventType(), RoomID: envelope.GetRoomId(), RoomVersion: envelope.GetRoomVersion(), OutboxCreatedAtMs: envelope.GetOccurredAtMs(), OccurredAtMs: envelope.GetOccurredAtMs(), ActorUserID: changed.GetActorUserId(), TargetUserID: changed.GetTargetUserId(), FromSeat: changed.GetFromSeat(), ToSeat: changed.GetToSeat(), Action: changed.GetAction(), MicSessionID: changed.GetMicSessionId(), PublishState: changed.GetPublishState(), Reason: changed.GetReason(), PublishDeadlineMs: changed.GetPublishDeadlineMs(), PublishEventTimeMs: changed.GetPublishEventTimeMs(), }, true, nil } func roomGiftCPEventFromRoomMessage(body []byte) (cpdomain.GiftEvent, bool, error) { envelope, _, err := roommq.DecodeRoomOutboxMessage(body) if err != nil { return cpdomain.GiftEvent{}, false, err } if envelope.GetEventType() != "RoomGiftSent" { return cpdomain.GiftEvent{}, false, nil } var sent roomeventsv1.RoomGiftSent if err := proto.Unmarshal(envelope.GetBody(), &sent); err != nil { return cpdomain.GiftEvent{}, false, err } // CP worker 必须接收所有送礼事实：没有 active 关系时仓储层只让 CP/兄弟/姐妹礼物生成申请； // 已有 active 关系时，普通、幸运和超级幸运礼物都按 wallet 已结算的 GiftValue 增加亲密值。 return cpdomain.GiftEvent{ AppCode: appcode.Normalize(envelope.GetAppCode()), EventID: envelope.GetEventId(), RoomID: envelope.GetRoomId(), RoomVersion: envelope.GetRoomVersion(), OccurredAtMS: envelope.GetOccurredAtMs(), SenderUserID: sent.GetSenderUserId(), TargetUserID: sent.GetTargetUserId(), GiftID: sent.GetGiftId(), GiftCount: sent.GetGiftCount(), GiftValue: sent.GetGiftValue(), BillingReceiptID: sent.GetBillingReceiptId(), VisibleRegionID: sent.GetVisibleRegionId(), CommandID: sent.GetCommandId(), GiftTypeCode: sent.GetGiftTypeCode(), CPRelationType: sent.GetCpRelationType(), GiftName: sent.GetGiftName(), GiftIconURL: sent.GetGiftIconUrl(), GiftAnimationURL: sent.GetGiftAnimationUrl(), }, true, nil } func rechargeTypeFromWalletPayload(payload string) string { var decoded map[string]any if err := json.Unmarshal([]byte(payload), &decoded); err == nil { for _, key := range []string{"recharge_type", "provider", "channel"} { if value, ok := decoded[key].(string); ok && strings.TrimSpace(value) != "" { return normalizeInviteRechargeType(value) } } } return invitedomain.RechargeTypeCoinSellerTransfer } func normalizeInviteRechargeType(value string) string { switch strings.ToLower(strings.TrimSpace(value)) { case "coin_seller", "coin_seller_transfer": return invitedomain.RechargeTypeCoinSellerTransfer case "google", "google_play", "googleplay": return invitedomain.RechargeTypeGooglePlay case "mifapay", "mifa_pay": return invitedomain.RechargeTypeMifaPay case "v5pay", "v5_pay": return invitedomain.RechargeTypeV5Pay case "usdt", "usdt_trc20", "trc20": return invitedomain.RechargeTypeUSDTTRC20 default: return strings.ToLower(strings.TrimSpace(value)) } } func startMQConsumers(consumers []*rocketmqx.Consumer) error { return servicemq.StartConsumers(consumers) } func shutdownMQConsumers(consumers []*rocketmqx.Consumer) { servicemq.ShutdownConsumers(consumers) } func walletOutboxConsumerConfig(cfg config.RocketMQConfig) rocketmqx.ConsumerConfig { return rocketmqConsumerConfig(cfg, cfg.WalletOutbox.ConsumerGroup, cfg.WalletOutbox.ConsumerMaxReconsumeTimes) } func roomOutboxConsumerConfig(cfg config.RocketMQConfig) rocketmqx.ConsumerConfig { return rocketmqConsumerConfig(cfg, cfg.RoomOutbox.ConsumerGroup, cfg.RoomOutbox.ConsumerMaxReconsumeTimes) } func cpRoomOutboxConsumerConfig(cfg config.Config) rocketmqx.ConsumerConfig { return rocketmqConsumerConfig(cfg.RocketMQ, cfg.CPRelationshipWorker.ConsumerGroup, cfg.RocketMQ.RoomOutbox.ConsumerMaxReconsumeTimes) } func userOutboxProducerConfig(cfg config.RocketMQConfig) rocketmqx.ProducerConfig { return rocketmqx.ProducerConfig{ EndpointConfig: rocketmqx.EndpointConfig{ NameServers: cfg.NameServers, NameServerDomain: cfg.NameServerDomain, AccessKey: cfg.AccessKey, SecretKey: cfg.SecretKey, SecurityToken: cfg.SecurityToken, Namespace: cfg.Namespace, VIPChannel: cfg.VIPChannel, }, GroupName: cfg.UserOutbox.ProducerGroup, SendTimeout: 3 * time.Second, Retry: 2, } } func rocketmqConsumerConfig(cfg config.RocketMQConfig, group string, maxReconsume int32) rocketmqx.ConsumerConfig { return rocketmqx.ConsumerConfig{ EndpointConfig: rocketmqx.EndpointConfig{ NameServers: cfg.NameServers, NameServerDomain: cfg.NameServerDomain, AccessKey: cfg.AccessKey, SecretKey: cfg.SecretKey, SecurityToken: cfg.SecurityToken, Namespace: cfg.Namespace, VIPChannel: cfg.VIPChannel, }, GroupName: group, MaxReconsumeTimes: maxReconsume, ConsumeRetryDelay: time.Second, ConsumePullBatch: 32, ConsumePullTimeout: 15 * time.Minute, } }