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

package app

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

	"google.golang.org/grpc"
	healthgrpc "google.golang.org/grpc/health/grpc_health_v1"
	walletv1 "hyapp.local/api/proto/wallet/v1"
	"hyapp/pkg/grpchealth"
	"hyapp/pkg/grpcshutdown"
	"hyapp/pkg/healthhttp"
	"hyapp/pkg/logx"
	"hyapp/services/wallet-service/internal/config"
	walletservice "hyapp/services/wallet-service/internal/service/wallet"
	mysqlstorage "hyapp/services/wallet-service/internal/storage/mysql"
	grpcserver "hyapp/services/wallet-service/internal/transport/grpc"
)

// App 装配 wallet-service gRPC 入口。
type App struct {
	server     *grpc.Server
	listener   net.Listener
	health     *grpchealth.ServingChecker
	healthHTTP *healthhttp.Server
	mysqlRepo  *mysqlstorage.Repository
	walletSvc  *walletservice.Service
	nodeID     string
	stopWorker context.CancelFunc
	workers    sync.WaitGroup
	closeOnce  sync.Once
}

// New 初始化 wallet-service。
func New(cfg config.Config) (*App, error) {
	startupCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	// wallet-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
	}

	server := grpc.NewServer(grpc.UnaryInterceptor(logx.UnaryServerInterceptor("wallet-service")))
	svc := walletservice.New(repository)
	walletv1.RegisterWalletServiceServer(server, grpcserver.NewServer(svc))
	health := grpchealth.NewServingChecker("wallet-service", grpchealth.Dependency{
		Name:  "mysql",
		Check: repository.Ping,
	})
	healthgrpc.RegisterHealthServer(server, grpchealth.NewServer(health))
	healthHTTP, err := healthhttp.New(cfg.HealthHTTPAddr, cfg.NodeID, health)
	if err != nil {
		_ = listener.Close()
		_ = repository.Close()
		return nil, err
	}

	return &App{
		server:     server,
		listener:   listener,
		health:     health,
		healthHTTP: healthHTTP,
		mysqlRepo:  repository,
		walletSvc:  svc,
		nodeID:     cfg.NodeID,
	}, nil
}

// Run 启动 gRPC 服务。
func (a *App) Run() error {
	a.runHealthHTTP()
	a.runGiftWallProjectionWorker()
	a.health.MarkServing()
	err := a.server.Serve(a.listener)
	a.health.MarkStopped()
	a.closeGiftWallProjectionWorker()
	if errors.Is(err, grpc.ErrServerStopped) {
		return nil
	}

	return err
}

// Close 关闭 gRPC 服务。
func (a *App) Close() {
	a.closeOnce.Do(func() {
		a.health.MarkDraining()
		a.closeGiftWallProjectionWorker()
		grpcshutdown.GracefulStop(a.server, 15*time.Second)
		a.closeHealthHTTP()
		if a.mysqlRepo != nil {
			// MySQL 连接池最后关闭，避免正在 drain 的扣费请求丢失提交能力。
			_ = a.mysqlRepo.Close()
		}
	})
}

func (a *App) runGiftWallProjectionWorker() {
	if a.walletSvc == nil {
		return
	}
	ctx, cancel := context.WithCancel(context.Background())
	a.stopWorker = cancel
	workerID := "gift-wall-" + a.nodeID
	a.workers.Add(1)
	go func() {
		defer a.workers.Done()
		ticker := time.NewTicker(2 * time.Second)
		defer ticker.Stop()
		for {
			// 礼物墙是展示读模型，worker 常驻在 wallet-service 内消费账务 outbox。
			// 服务重启会回滚未提交的投影事务；已完成事件由 projection done 状态防重。
			processed, err := a.walletSvc.ProjectPendingGiftWallEvents(ctx, workerID, 50, 30*time.Second)
			if err != nil && !errors.Is(err, context.Canceled) {
				logx.Error(ctx, "gift_wall_projection_failed", err, slog.String("worker_id", workerID))
			}
			if processed >= 50 {
				// 批次打满说明还有积压，立即继续 drain，避免固定 tick 放大展示延迟。
				continue
			}
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
			}
		}
	}()
}

func (a *App) closeGiftWallProjectionWorker() {
	if a.stopWorker != nil {
		a.stopWorker()
	}
	a.workers.Wait()
}

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)
}