hyapp-server/services/activity-service/internal/app/app.go

package app

import (
	"context"
	"errors"
	"log/slog"
	"net"
	"sync"
	"time"

	"google.golang.org/grpc"
	"hyapp/pkg/grpchealth"
	"hyapp/pkg/grpcshutdown"
	"hyapp/pkg/healthhttp"
	"hyapp/pkg/logx"
	"hyapp/pkg/rocketmqx"
	"hyapp/services/activity-service/internal/config"
	broadcastservice "hyapp/services/activity-service/internal/service/broadcast"
	cumulativerechargeservice "hyapp/services/activity-service/internal/service/cumulativerecharge"
	firstrechargeservice "hyapp/services/activity-service/internal/service/firstrecharge"
	luckygiftservice "hyapp/services/activity-service/internal/service/luckygift"
	mysqlstorage "hyapp/services/activity-service/internal/storage/mysql"
)

// App 装配 activity-service 的 gRPC 入口、健康检查、后台 worker 和外部连接。
// 具体业务模块的创建和注册都放在独立文件里，避免启动入口继续膨胀成一个难维护的大函数。
type App struct {
	server                 *grpc.Server
	listener               net.Listener
	health                 *grpchealth.ServingChecker
	healthHTTP             *healthhttp.Server
	mysqlRepo              *mysqlstorage.Repository
	broadcast              *broadcastservice.Service
	luckyGift              *luckygiftservice.Service
	firstRechargeReward    *firstrechargeservice.Service
	cumulativeRecharge     *cumulativerechargeservice.Service
	broadcastWorkerEnabled bool
	luckyGiftWorkerEnabled bool
	workerNodeID           string
	luckyGiftWorkerOptions luckygiftservice.WorkerOptions
	mqConsumers            []*rocketmqx.Consumer
	userConn               *grpc.ClientConn
	walletConn             *grpc.ClientConn
	roomConn               *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()

	repository, err := openRepository(startupCtx, cfg)
	if err != nil {
		return nil, err
	}
	var listener net.Listener
	var clients externalClients
	var mqConsumers []*rocketmqx.Consumer
	cleanup := func() {
		shutdownConsumers(mqConsumers)
		clients.Close()
		if listener != nil {
			_ = listener.Close()
		}
		_ = repository.Close()
	}

	listener, err = listenGRPC(cfg.GRPCAddr)
	if err != nil {
		cleanup()
		return nil, err
	}
	clients, err = dialExternalClients(cfg)
	if err != nil {
		cleanup()
		return nil, err
	}

	server := grpc.NewServer(grpc.UnaryInterceptor(logx.UnaryServerInterceptor("activity-service")))
	services, err := buildServiceBundle(cfg, repository, clients)
	if err != nil {
		cleanup()
		return nil, err
	}
	registerGRPCServers(server, services)

	mqConsumers, err = buildMQConsumers(cfg, services)
	if err != nil {
		cleanup()
		return nil, err
	}
	health, healthHTTP, err := newHealthServers(cfg, server, repository)
	if err != nil {
		cleanup()
		return nil, err
	}

	workerCtx, workerStop := context.WithCancel(context.Background())
	return &App{
		server:                 server,
		listener:               listener,
		health:                 health,
		healthHTTP:             healthHTTP,
		mysqlRepo:              repository,
		broadcast:              services.broadcast,
		luckyGift:              services.luckyGift,
		firstRechargeReward:    services.firstRechargeReward,
		cumulativeRecharge:     services.cumulativeRecharge,
		broadcastWorkerEnabled: cfg.Broadcast.Enabled,
		luckyGiftWorkerEnabled: cfg.LuckyGiftWorker.Enabled,
		workerNodeID:           cfg.NodeID,
		luckyGiftWorkerOptions: luckyGiftWorkerOptions(cfg.NodeID, cfg.LuckyGiftWorker),
		mqConsumers:            mqConsumers,
		userConn:               clients.userConn,
		walletConn:             clients.walletConn,
		roomConn:               clients.roomConn,
		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.luckyGiftWorkerEnabled && a.luckyGift != nil && a.workerCtx != nil {
		a.workerWG.Go(func() {
			a.luckyGift.RunWorker(a.workerCtx, a.luckyGiftWorkerOptions)
		})
	}
	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.roomConn != nil {
			_ = a.roomConn.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()))
		}
	}
}