403 lines
15 KiB
Go
403 lines
15 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"log/slog"
|
|
"net"
|
|
"sync"
|
|
"time"
|
|
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/credentials/insecure"
|
|
healthgrpc "google.golang.org/grpc/health/grpc_health_v1"
|
|
activityv1 "hyapp.local/api/proto/activity/v1"
|
|
walletv1 "hyapp.local/api/proto/wallet/v1"
|
|
"hyapp/pkg/appcode"
|
|
"hyapp/pkg/grpchealth"
|
|
"hyapp/pkg/grpcshutdown"
|
|
"hyapp/pkg/healthhttp"
|
|
"hyapp/pkg/logx"
|
|
"hyapp/pkg/rocketmqx"
|
|
"hyapp/pkg/roommq"
|
|
"hyapp/pkg/tencentim"
|
|
"hyapp/services/activity-service/internal/client"
|
|
"hyapp/services/activity-service/internal/config"
|
|
achievementservice "hyapp/services/activity-service/internal/service/achievement"
|
|
activityservice "hyapp/services/activity-service/internal/service/activity"
|
|
broadcastservice "hyapp/services/activity-service/internal/service/broadcast"
|
|
firstrechargeservice "hyapp/services/activity-service/internal/service/firstrecharge"
|
|
growthservice "hyapp/services/activity-service/internal/service/growth"
|
|
luckygiftservice "hyapp/services/activity-service/internal/service/luckygift"
|
|
messageservice "hyapp/services/activity-service/internal/service/message"
|
|
registrationrewardservice "hyapp/services/activity-service/internal/service/registrationreward"
|
|
sevendaycheckinservice "hyapp/services/activity-service/internal/service/sevendaycheckin"
|
|
taskservice "hyapp/services/activity-service/internal/service/task"
|
|
mysqlstorage "hyapp/services/activity-service/internal/storage/mysql"
|
|
grpcserver "hyapp/services/activity-service/internal/transport/grpc"
|
|
)
|
|
|
|
// App 装配 activity-service gRPC 入口和底座依赖。
|
|
type App struct {
|
|
server *grpc.Server
|
|
listener net.Listener
|
|
health *grpchealth.ServingChecker
|
|
healthHTTP *healthhttp.Server
|
|
mysqlRepo *mysqlstorage.Repository
|
|
walletOutboxDB *sql.DB
|
|
broadcast *broadcastservice.Service
|
|
firstRechargeReward *firstrechargeservice.Service
|
|
redPacketReader broadcastservice.RedPacketWalletOutboxReader
|
|
broadcastWorkerEnabled bool
|
|
firstRechargeWorkerCfg config.FirstRechargeRewardWorkerConfig
|
|
redPacketWorkerCfg config.RedPacketBroadcastWorkerConfig
|
|
workerNodeID string
|
|
mqConsumers []*rocketmqx.Consumer
|
|
userConn *grpc.ClientConn
|
|
walletConn *grpc.ClientConn
|
|
workerCtx context.Context
|
|
workerStop context.CancelFunc
|
|
workerWG sync.WaitGroup
|
|
closeOnce sync.Once
|
|
}
|
|
|
|
// New 初始化 activity-service 应用。
|
|
func New(cfg config.Config) (*App, error) {
|
|
startupCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
// activity-service 的活动状态、消费幂等和 outbox 都必须使用 MySQL。
|
|
repository, err := mysqlstorage.Open(startupCtx, cfg.MySQLDSN)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
listener, err := net.Listen("tcp", cfg.GRPCAddr)
|
|
if err != nil {
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
userConn, err := grpc.Dial(cfg.UserServiceAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
|
|
if err != nil {
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
walletConn, err := grpc.Dial(cfg.WalletServiceAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
|
|
if err != nil {
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
|
|
server := grpc.NewServer(grpc.UnaryInterceptor(logx.UnaryServerInterceptor("activity-service")))
|
|
svc := activityservice.New(activityservice.Config{NodeID: cfg.NodeID}, repository)
|
|
messageSvc := messageservice.New(messageservice.Config{NodeID: cfg.NodeID}, repository, messageservice.WithTargetSource(client.NewGRPCUserTargetSource(userConn)))
|
|
taskSvc := taskservice.New(repository, walletv1.NewWalletServiceClient(walletConn))
|
|
registrationRewardSvc := registrationrewardservice.New(repository, walletv1.NewWalletServiceClient(walletConn))
|
|
sevenDayCheckInSvc := sevendaycheckinservice.New(repository, walletv1.NewWalletServiceClient(walletConn))
|
|
growthSvc := growthservice.New(repository, walletv1.NewWalletServiceClient(walletConn))
|
|
achievementSvc := achievementservice.New(repository, walletv1.NewWalletServiceClient(walletConn))
|
|
luckyGiftSvc := luckygiftservice.New(repository)
|
|
var broadcastPublisher broadcastservice.Publisher
|
|
if cfg.TencentIM.Enabled {
|
|
tencentClient, err := tencentim.NewRESTClient(cfg.TencentIM.RESTConfig())
|
|
if err != nil {
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
broadcastPublisher = tencentClient
|
|
}
|
|
if cfg.Broadcast.Enabled && broadcastPublisher == nil {
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, errors.New("broadcast worker requires tencent_im.enabled")
|
|
}
|
|
broadcastSvc := broadcastservice.New(broadcastservice.Config{
|
|
NodeID: cfg.NodeID,
|
|
SuperGiftMinValue: cfg.Broadcast.SuperGiftMinValue,
|
|
WorkerBatchSize: cfg.Broadcast.WorkerBatchSize,
|
|
WorkerLockTTL: cfg.Broadcast.WorkerLockTTL,
|
|
WorkerMaxRetry: cfg.Broadcast.WorkerMaxRetry,
|
|
WorkerPollInterval: cfg.Broadcast.WorkerPollInterval,
|
|
EnsureGroupsOnStartup: cfg.Broadcast.EnsureGroupsOnStartup,
|
|
}, repository, broadcastPublisher, client.NewGRPCRegionSource(userConn))
|
|
activityv1.RegisterActivityServiceServer(server, grpcserver.NewServer(svc))
|
|
activityv1.RegisterMessageInboxServiceServer(server, grpcserver.NewMessageServer(messageSvc))
|
|
activityv1.RegisterActivityCronServiceServer(server, grpcserver.NewCronServer(messageSvc, growthSvc, achievementSvc))
|
|
activityv1.RegisterTaskServiceServer(server, grpcserver.NewTaskServer(taskSvc))
|
|
activityv1.RegisterAdminTaskServiceServer(server, grpcserver.NewAdminTaskServer(taskSvc))
|
|
activityv1.RegisterGrowthLevelServiceServer(server, grpcserver.NewGrowthLevelServer(growthSvc))
|
|
activityv1.RegisterAdminGrowthLevelServiceServer(server, grpcserver.NewAdminGrowthLevelServer(growthSvc))
|
|
activityv1.RegisterAchievementServiceServer(server, grpcserver.NewAchievementServer(achievementSvc))
|
|
activityv1.RegisterAdminAchievementServiceServer(server, grpcserver.NewAdminAchievementServer(achievementSvc))
|
|
activityv1.RegisterLuckyGiftServiceServer(server, grpcserver.NewLuckyGiftServer(luckyGiftSvc))
|
|
activityv1.RegisterAdminLuckyGiftServiceServer(server, grpcserver.NewAdminLuckyGiftServer(luckyGiftSvc))
|
|
activityv1.RegisterRegistrationRewardServiceServer(server, grpcserver.NewRegistrationRewardServer(registrationRewardSvc))
|
|
activityv1.RegisterAdminRegistrationRewardServiceServer(server, grpcserver.NewAdminRegistrationRewardServer(registrationRewardSvc))
|
|
activityv1.RegisterSevenDayCheckInServiceServer(server, grpcserver.NewSevenDayCheckInServer(sevenDayCheckInSvc))
|
|
activityv1.RegisterAdminSevenDayCheckInServiceServer(server, grpcserver.NewAdminSevenDayCheckInServer(sevenDayCheckInSvc))
|
|
broadcastServer := grpcserver.NewBroadcastServer(broadcastSvc, growthSvc)
|
|
activityv1.RegisterBroadcastServiceServer(server, broadcastServer)
|
|
activityv1.RegisterRoomEventConsumerServiceServer(server, broadcastServer)
|
|
var walletOutboxDB *sql.DB
|
|
var firstRechargeReader firstrechargeservice.WalletOutboxReader
|
|
var redPacketReader broadcastservice.RedPacketWalletOutboxReader
|
|
needsWalletOutboxDB := cfg.FirstRechargeRewardWorker.Enabled || cfg.RedPacketBroadcastWorker.Enabled
|
|
walletOutboxDSN := cfg.FirstRechargeRewardWorker.WalletMySQLDSN
|
|
if cfg.RedPacketBroadcastWorker.Enabled {
|
|
if cfg.FirstRechargeRewardWorker.Enabled && cfg.RedPacketBroadcastWorker.WalletMySQLDSN != cfg.FirstRechargeRewardWorker.WalletMySQLDSN {
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, errors.New("wallet outbox workers must use the same wallet_mysql_dsn")
|
|
}
|
|
walletOutboxDSN = cfg.RedPacketBroadcastWorker.WalletMySQLDSN
|
|
}
|
|
if needsWalletOutboxDB {
|
|
walletOutboxDB, err = sql.Open("mysql", walletOutboxDSN)
|
|
if err != nil {
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
if err := walletOutboxDB.PingContext(startupCtx); err != nil {
|
|
_ = walletOutboxDB.Close()
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
if cfg.FirstRechargeRewardWorker.Enabled {
|
|
firstRechargeReader = mysqlstorage.NewWalletRechargeOutboxReader(walletOutboxDB)
|
|
}
|
|
if cfg.RedPacketBroadcastWorker.Enabled {
|
|
redPacketReader = mysqlstorage.NewRedPacketWalletOutboxReader(walletOutboxDB)
|
|
}
|
|
}
|
|
firstRechargeRewardSvc := firstrechargeservice.New(repository, walletv1.NewWalletServiceClient(walletConn), firstRechargeReader)
|
|
activityv1.RegisterFirstRechargeRewardServiceServer(server, grpcserver.NewFirstRechargeRewardServer(firstRechargeRewardSvc))
|
|
activityv1.RegisterAdminFirstRechargeRewardServiceServer(server, grpcserver.NewAdminFirstRechargeRewardServer(firstRechargeRewardSvc))
|
|
mqConsumers := make([]*rocketmqx.Consumer, 0, 1)
|
|
if cfg.RocketMQ.RoomOutbox.Enabled {
|
|
consumer, err := rocketmqx.NewConsumer(rocketMQConsumerConfig(cfg.RocketMQ))
|
|
if err != nil {
|
|
if walletOutboxDB != nil {
|
|
_ = walletOutboxDB.Close()
|
|
}
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
if err := consumer.Subscribe(cfg.RocketMQ.RoomOutbox.Topic, roommq.TagRoomOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
|
|
envelope, _, err := roommq.DecodeRoomOutboxMessage(message.Body)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
eventCtx := appcode.WithContext(ctx, envelope.GetAppCode())
|
|
if _, err := broadcastSvc.HandleRoomEvent(eventCtx, envelope); err != nil {
|
|
return err
|
|
}
|
|
_, err = growthSvc.HandleRoomEvent(eventCtx, envelope)
|
|
return err
|
|
}); err != nil {
|
|
if walletOutboxDB != nil {
|
|
_ = walletOutboxDB.Close()
|
|
}
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
mqConsumers = append(mqConsumers, consumer)
|
|
}
|
|
healthDependencies := []grpchealth.Dependency{{
|
|
Name: "mysql",
|
|
Check: repository.Ping,
|
|
}}
|
|
if walletOutboxDB != nil {
|
|
healthDependencies = append(healthDependencies, grpchealth.Dependency{
|
|
Name: "wallet_mysql",
|
|
Check: walletOutboxDB.PingContext,
|
|
})
|
|
}
|
|
health := grpchealth.NewServingChecker("activity-service", healthDependencies...)
|
|
healthgrpc.RegisterHealthServer(server, grpchealth.NewServer(health))
|
|
healthHTTP, err := healthhttp.New(cfg.HealthHTTPAddr, cfg.NodeID, health)
|
|
if err != nil {
|
|
if walletOutboxDB != nil {
|
|
_ = walletOutboxDB.Close()
|
|
}
|
|
_ = walletConn.Close()
|
|
_ = userConn.Close()
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
workerCtx, workerStop := context.WithCancel(context.Background())
|
|
|
|
return &App{
|
|
server: server,
|
|
listener: listener,
|
|
health: health,
|
|
healthHTTP: healthHTTP,
|
|
mysqlRepo: repository,
|
|
walletOutboxDB: walletOutboxDB,
|
|
broadcast: broadcastSvc,
|
|
firstRechargeReward: firstRechargeRewardSvc,
|
|
redPacketReader: redPacketReader,
|
|
broadcastWorkerEnabled: cfg.Broadcast.Enabled,
|
|
firstRechargeWorkerCfg: cfg.FirstRechargeRewardWorker,
|
|
redPacketWorkerCfg: cfg.RedPacketBroadcastWorker,
|
|
workerNodeID: cfg.NodeID,
|
|
mqConsumers: mqConsumers,
|
|
userConn: userConn,
|
|
walletConn: walletConn,
|
|
workerCtx: workerCtx,
|
|
workerStop: workerStop,
|
|
}, nil
|
|
}
|
|
|
|
// Run 启动 gRPC 服务。
|
|
func (a *App) Run() error {
|
|
a.runHealthHTTP()
|
|
a.health.MarkServing()
|
|
if err := a.startMQ(); err != nil {
|
|
a.health.MarkStopped()
|
|
return err
|
|
}
|
|
if a.broadcastWorkerEnabled && a.broadcast != nil && a.workerCtx != nil {
|
|
a.workerWG.Go(func() {
|
|
a.broadcast.RunWorker(a.workerCtx, broadcastservice.WorkerOptions{})
|
|
})
|
|
}
|
|
if a.firstRechargeWorkerCfg.Enabled && a.firstRechargeReward != nil && a.workerCtx != nil {
|
|
a.workerWG.Go(func() {
|
|
a.firstRechargeReward.RunWalletRechargeWorker(a.workerCtx, firstrechargeservice.WorkerOptions{
|
|
WorkerID: a.workerNodeID,
|
|
PollInterval: time.Duration(a.firstRechargeWorkerCfg.PollIntervalSec) * time.Second,
|
|
BatchSize: a.firstRechargeWorkerCfg.BatchSize,
|
|
})
|
|
})
|
|
}
|
|
if a.redPacketWorkerCfg.Enabled && a.broadcast != nil && a.redPacketReader != nil && a.workerCtx != nil {
|
|
a.workerWG.Go(func() {
|
|
a.broadcast.RunRedPacketBroadcastWorker(a.workerCtx, a.redPacketReader, broadcastservice.RedPacketBroadcastWorkerOptions{
|
|
WorkerID: a.workerNodeID,
|
|
PollInterval: time.Duration(a.redPacketWorkerCfg.PollIntervalSec) * time.Second,
|
|
BatchSize: a.redPacketWorkerCfg.BatchSize,
|
|
})
|
|
})
|
|
}
|
|
err := a.server.Serve(a.listener)
|
|
if a.workerStop != nil {
|
|
a.workerStop()
|
|
}
|
|
a.workerWG.Wait()
|
|
a.shutdownMQ()
|
|
a.health.MarkStopped()
|
|
if errors.Is(err, grpc.ErrServerStopped) {
|
|
return nil
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
// Close 关闭 gRPC 服务。
|
|
func (a *App) Close() {
|
|
a.closeOnce.Do(func() {
|
|
a.health.MarkDraining()
|
|
if a.workerStop != nil {
|
|
a.workerStop()
|
|
}
|
|
a.workerWG.Wait()
|
|
a.shutdownMQ()
|
|
grpcshutdown.GracefulStop(a.server, 15*time.Second)
|
|
a.closeHealthHTTP()
|
|
if a.userConn != nil {
|
|
_ = a.userConn.Close()
|
|
}
|
|
if a.walletConn != nil {
|
|
_ = a.walletConn.Close()
|
|
}
|
|
if a.walletOutboxDB != nil {
|
|
_ = a.walletOutboxDB.Close()
|
|
}
|
|
if a.mysqlRepo != nil {
|
|
// MySQL 连接池最后关闭，保证 drain 中的查询或消费提交能完成。
|
|
_ = a.mysqlRepo.Close()
|
|
}
|
|
})
|
|
}
|
|
|
|
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 (a *App) startMQ() error {
|
|
for _, consumer := range a.mqConsumers {
|
|
if err := consumer.Start(); err != nil {
|
|
a.shutdownMQ()
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (a *App) shutdownMQ() {
|
|
for _, consumer := range a.mqConsumers {
|
|
if err := consumer.Shutdown(); err != nil {
|
|
logx.Warn(context.Background(), "rocketmq_consumer_shutdown_failed", slog.String("error", err.Error()))
|
|
}
|
|
}
|
|
}
|
|
|
|
func rocketMQConsumerConfig(cfg config.RocketMQConfig) 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: cfg.RoomOutbox.ConsumerGroup,
|
|
MaxReconsumeTimes: cfg.RoomOutbox.ConsumerMaxReconsumeTimes,
|
|
ConsumeRetryDelay: time.Second,
|
|
ConsumePullBatch: 32,
|
|
ConsumePullTimeout: 15 * time.Minute,
|
|
}
|
|
}
|