697 lines
25 KiB
Go
697 lines
25 KiB
Go
// 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/grpcshutdown"
|
||
"hyapp/pkg/healthhttp"
|
||
"hyapp/pkg/idgen"
|
||
"hyapp/pkg/logx"
|
||
"hyapp/pkg/rocketmqx"
|
||
"hyapp/pkg/roommq"
|
||
"hyapp/pkg/tencentim"
|
||
"hyapp/pkg/usermq"
|
||
"hyapp/pkg/walletmq"
|
||
"hyapp/services/user-service/internal/config"
|
||
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"
|
||
authservice "hyapp/services/user-service/internal/service/auth"
|
||
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"
|
||
healthgrpc "google.golang.org/grpc/health/grpc_health_v1"
|
||
)
|
||
|
||
// 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
|
||
// 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
|
||
// loginRiskRedisClose 关闭登录风险 Redis cache；cache 不作为启动硬依赖。
|
||
loginRiskRedisClose func() error
|
||
// cfg 保存 worker 运行参数，避免 Run 阶段重新读配置。
|
||
cfg config.Config
|
||
// workerCtx 统一控制后台 worker 生命周期。
|
||
workerCtx context.Context
|
||
// workerCancel 停止补偿 worker。
|
||
workerCancel context.CancelFunc
|
||
// workerWG 等待后台 worker 当前批次安全退出。
|
||
workerWG sync.WaitGroup
|
||
// 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 := grpc.NewServer(grpc.UnaryInterceptor(logx.UnaryServerInterceptor("user-service")))
|
||
thirdPartyVerifier := authservice.NewRoutedThirdPartyVerifier(
|
||
authservice.NewFirebaseThirdPartyVerifier(authservice.FirebaseVerifierConfig{
|
||
ProjectID: cfg.ThirdParty.Firebase.ProjectID,
|
||
AllowedSignInProviders: cfg.ThirdParty.Firebase.AllowedSignInProviders,
|
||
CertsURL: cfg.ThirdParty.Firebase.CertsURL,
|
||
}),
|
||
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()
|
||
var ipDecisionCache authservice.IPDecisionCache
|
||
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 {
|
||
ipDecisionCache = authservice.NewRedisIPDecisionCache(redisClient)
|
||
loginRiskRedisClose = redisClient.Close
|
||
} else {
|
||
// IP 风控 Redis 缓存按文档 fail-open；启动继续，worker 仍会写 MySQL 审计。
|
||
logx.Warn(startupCtx, "login_risk_redis_unavailable")
|
||
}
|
||
}
|
||
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()
|
||
}
|
||
_ = 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()
|
||
}
|
||
_ = listener.Close()
|
||
_ = mysqlRepo.Close()
|
||
return nil, err
|
||
}
|
||
roomEvictor = roomclient.NewGRPC(roomConn)
|
||
}
|
||
var activityConn *grpc.ClientConn
|
||
var registrationRewardIssuer authservice.RegistrationRewardIssuer
|
||
var registrationLevelBadgeIssuer authservice.RegistrationLevelBadgeIssuer
|
||
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 roomConn != nil {
|
||
_ = roomConn.Close()
|
||
}
|
||
_ = listener.Close()
|
||
_ = mysqlRepo.Close()
|
||
return nil, err
|
||
}
|
||
activityClient := activityclient.NewGRPC(activityConn)
|
||
registrationRewardIssuer = activityClient
|
||
registrationLevelBadgeIssuer = activityClient
|
||
}
|
||
|
||
// 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.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.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)
|
||
mqConsumers, err := newMQConsumers(cfg, inviteSvc, micTimeSvc)
|
||
if err != nil {
|
||
if loginRiskRedisClose != nil {
|
||
_ = loginRiskRedisClose()
|
||
}
|
||
if roomConn != nil {
|
||
_ = roomConn.Close()
|
||
}
|
||
if activityConn != nil {
|
||
_ = activityConn.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 roomConn != nil {
|
||
_ = roomConn.Close()
|
||
}
|
||
if activityConn != nil {
|
||
_ = activityConn.Close()
|
||
}
|
||
_ = listener.Close()
|
||
_ = mysqlRepo.Close()
|
||
return nil, err
|
||
}
|
||
}
|
||
userServer := grpcserver.NewServer(authSvc, userSvc, hostSvc)
|
||
userServer.SetMicTimeService(micTimeSvc)
|
||
// 多个 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.RegisterCountryAdminServiceServer(server, userServer)
|
||
userv1.RegisterCountryQueryServiceServer(server, userServer)
|
||
userv1.RegisterRegionAdminServiceServer(server, userServer)
|
||
userv1.RegisterUserHostServiceServer(server, userServer)
|
||
userv1.RegisterUserHostAdminServiceServer(server, userServer)
|
||
userv1.RegisterUserCronServiceServer(server, userServer)
|
||
|
||
healthDependencies := []grpchealth.Dependency{{
|
||
Name: "mysql",
|
||
Check: mysqlRepo.Ping,
|
||
}}
|
||
health := grpchealth.NewServingChecker("user-service", healthDependencies...)
|
||
// user-service 目前使用简单 serving checker，存储探测由后续专用 health 可扩展。
|
||
healthgrpc.RegisterHealthServer(server, grpchealth.NewServer(health))
|
||
healthHTTP, err := healthhttp.New(cfg.HealthHTTPAddr, cfg.NodeID, health)
|
||
if err != nil {
|
||
shutdownMQConsumers(mqConsumers)
|
||
if loginRiskRedisClose != nil {
|
||
_ = loginRiskRedisClose()
|
||
}
|
||
if roomConn != nil {
|
||
_ = roomConn.Close()
|
||
}
|
||
if activityConn != nil {
|
||
_ = activityConn.Close()
|
||
}
|
||
_ = listener.Close()
|
||
_ = mysqlRepo.Close()
|
||
return nil, err
|
||
}
|
||
workerCtx, workerCancel := context.WithCancel(context.Background())
|
||
|
||
return &App{
|
||
server: server,
|
||
listener: listener,
|
||
health: health,
|
||
healthHTTP: healthHTTP,
|
||
mysqlRepo: mysqlRepo,
|
||
mqConsumers: mqConsumers,
|
||
userOutboxProducer: userOutboxProducer,
|
||
roomConn: roomConn,
|
||
activityConn: activityConn,
|
||
authSvc: authSvc,
|
||
userSvc: userSvc,
|
||
inviteSvc: inviteSvc,
|
||
micTimeSvc: micTimeSvc,
|
||
loginRiskRedisClose: loginRiskRedisClose,
|
||
cfg: cfg,
|
||
workerCtx: workerCtx,
|
||
workerCancel: workerCancel,
|
||
}, 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() {
|
||
if a.workerCancel != nil {
|
||
a.workerCancel()
|
||
}
|
||
a.workerWG.Wait()
|
||
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()
|
||
if a.workerCancel != nil {
|
||
a.workerCancel()
|
||
}
|
||
a.workerWG.Wait()
|
||
shutdownMQConsumers(a.mqConsumers)
|
||
a.shutdownUserOutboxProducer()
|
||
// GracefulStop 等待已进入的 RPC 结束，避免 token/session 写入被硬切。
|
||
grpcshutdown.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.loginRiskRedisClose != nil {
|
||
_ = a.loginRiskRedisClose()
|
||
}
|
||
})
|
||
}
|
||
|
||
func (a *App) runUserOutboxWorker() {
|
||
if !a.cfg.OutboxWorker.Enabled || a.userOutboxProducer == nil || a.mysqlRepo == nil {
|
||
return
|
||
}
|
||
workerID := "user-outbox-" + a.cfg.NodeID
|
||
a.workerWG.Add(1)
|
||
go func() {
|
||
defer a.workerWG.Done()
|
||
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(a.workerCtx, "user_outbox_publish_failed", err, slog.String("worker_id", workerID))
|
||
}
|
||
if processed >= a.cfg.OutboxWorker.BatchSize {
|
||
continue
|
||
}
|
||
select {
|
||
case <-a.workerCtx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
}
|
||
}
|
||
}()
|
||
}
|
||
|
||
func (a *App) processUserOutboxBatch(workerID string) (int, error) {
|
||
records, err := a.mysqlRepo.ClaimPendingUserOutbox(a.workerCtx, 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 a.userOutboxProducer.Start()
|
||
}
|
||
|
||
func (a *App) shutdownUserOutboxProducer() {
|
||
if a.userOutboxProducer == nil {
|
||
return
|
||
}
|
||
if err := a.userOutboxProducer.Shutdown(); err != nil {
|
||
logx.Warn(context.Background(), "rocketmq_producer_shutdown_failed", slog.String("error", err.Error()))
|
||
}
|
||
}
|
||
|
||
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)
|
||
}
|
||
|
||
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) ([]*rocketmqx.Consumer, error) {
|
||
consumers := make([]*rocketmqx.Consumer, 0, 2)
|
||
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)
|
||
}
|
||
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 rechargeTypeFromWalletPayload(payload string) string {
|
||
var decoded map[string]any
|
||
if err := json.Unmarshal([]byte(payload), &decoded); err == nil {
|
||
if value, ok := decoded["recharge_type"].(string); ok && strings.TrimSpace(value) != "" {
|
||
return strings.TrimSpace(value)
|
||
}
|
||
}
|
||
return invitedomain.RechargeTypeCoinSellerTransfer
|
||
}
|
||
|
||
func startMQConsumers(consumers []*rocketmqx.Consumer) error {
|
||
for _, consumer := range consumers {
|
||
if err := consumer.Start(); err != nil {
|
||
shutdownMQConsumers(consumers)
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func shutdownMQConsumers(consumers []*rocketmqx.Consumer) {
|
||
for _, consumer := range consumers {
|
||
if err := consumer.Shutdown(); err != nil {
|
||
logx.Warn(context.Background(), "rocketmq_consumer_shutdown_failed", slog.String("error", err.Error()))
|
||
}
|
||
}
|
||
}
|
||
|
||
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 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,
|
||
}
|
||
}
|