732 lines
30 KiB
Go
732 lines
30 KiB
Go
package main
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"flag"
|
||
"fmt"
|
||
"log"
|
||
"log/slog"
|
||
"net/http"
|
||
"os"
|
||
"os/signal"
|
||
"strings"
|
||
"syscall"
|
||
"time"
|
||
|
||
"hyapp-admin-server/internal/cache"
|
||
"hyapp-admin-server/internal/config"
|
||
"hyapp-admin-server/internal/integration/activityclient"
|
||
dingtalkclient "hyapp-admin-server/internal/integration/dingtalk"
|
||
"hyapp-admin-server/internal/integration/gameclient"
|
||
"hyapp-admin-server/internal/integration/googleorders"
|
||
"hyapp-admin-server/internal/integration/robotclient"
|
||
"hyapp-admin-server/internal/integration/roomclient"
|
||
"hyapp-admin-server/internal/integration/userclient"
|
||
"hyapp-admin-server/internal/integration/walletclient"
|
||
jobrunner "hyapp-admin-server/internal/job"
|
||
"hyapp-admin-server/internal/migration"
|
||
achievementconfigmodule "hyapp-admin-server/internal/modules/achievementconfig"
|
||
adminusermodule "hyapp-admin-server/internal/modules/adminuser"
|
||
agencyopeningmodule "hyapp-admin-server/internal/modules/agencyopening"
|
||
appconfigmodule "hyapp-admin-server/internal/modules/appconfig"
|
||
appregistrymodule "hyapp-admin-server/internal/modules/appregistry"
|
||
appusermodule "hyapp-admin-server/internal/modules/appuser"
|
||
auditmodule "hyapp-admin-server/internal/modules/audit"
|
||
authmodule "hyapp-admin-server/internal/modules/auth"
|
||
coinledgermodule "hyapp-admin-server/internal/modules/coinledger"
|
||
countryregionmodule "hyapp-admin-server/internal/modules/countryregion"
|
||
cprelationmodule "hyapp-admin-server/internal/modules/cprelation"
|
||
cpweeklyrankmodule "hyapp-admin-server/internal/modules/cpweeklyrank"
|
||
cumulativerechargerewardmodule "hyapp-admin-server/internal/modules/cumulativerechargereward"
|
||
dailytaskmodule "hyapp-admin-server/internal/modules/dailytask"
|
||
dashboardmodule "hyapp-admin-server/internal/modules/dashboard"
|
||
databimodule "hyapp-admin-server/internal/modules/databi"
|
||
financeapplicationmodule "hyapp-admin-server/internal/modules/financeapplication"
|
||
financewithdrawalmodule "hyapp-admin-server/internal/modules/financewithdrawal"
|
||
firstrechargerewardmodule "hyapp-admin-server/internal/modules/firstrechargereward"
|
||
fullservernoticemodule "hyapp-admin-server/internal/modules/fullservernotice"
|
||
gamemanagementmodule "hyapp-admin-server/internal/modules/gamemanagement"
|
||
giftdiamondmodule "hyapp-admin-server/internal/modules/giftdiamond"
|
||
healthmodule "hyapp-admin-server/internal/modules/health"
|
||
hostagencypolicymodule "hyapp-admin-server/internal/modules/hostagencypolicy"
|
||
hostorgmodule "hyapp-admin-server/internal/modules/hostorg"
|
||
hostsalarysettlementmodule "hyapp-admin-server/internal/modules/hostsalarysettlement"
|
||
hostwithdrawalmodule "hyapp-admin-server/internal/modules/hostwithdrawal"
|
||
inviteactivityrewardmodule "hyapp-admin-server/internal/modules/inviteactivityreward"
|
||
jobmodule "hyapp-admin-server/internal/modules/job"
|
||
levelconfigmodule "hyapp-admin-server/internal/modules/levelconfig"
|
||
luckygiftmodule "hyapp-admin-server/internal/modules/luckygift"
|
||
menumodule "hyapp-admin-server/internal/modules/menu"
|
||
paymentmodule "hyapp-admin-server/internal/modules/payment"
|
||
prettyidmodule "hyapp-admin-server/internal/modules/prettyid"
|
||
rbacmodule "hyapp-admin-server/internal/modules/rbac"
|
||
redpacketmodule "hyapp-admin-server/internal/modules/redpacket"
|
||
regionblockmodule "hyapp-admin-server/internal/modules/regionblock"
|
||
registrationrewardmodule "hyapp-admin-server/internal/modules/registrationreward"
|
||
reportmodule "hyapp-admin-server/internal/modules/report"
|
||
resourcemodule "hyapp-admin-server/internal/modules/resource"
|
||
riskconfigmodule "hyapp-admin-server/internal/modules/riskconfig"
|
||
roomadminmodule "hyapp-admin-server/internal/modules/roomadmin"
|
||
roomrocketmodule "hyapp-admin-server/internal/modules/roomrocket"
|
||
roomturnoverrewardmodule "hyapp-admin-server/internal/modules/roomturnoverreward"
|
||
searchmodule "hyapp-admin-server/internal/modules/search"
|
||
sevendaycheckinmodule "hyapp-admin-server/internal/modules/sevendaycheckin"
|
||
teammodule "hyapp-admin-server/internal/modules/team"
|
||
teamsalarypolicymodule "hyapp-admin-server/internal/modules/teamsalarypolicy"
|
||
teamsalarysettlementmodule "hyapp-admin-server/internal/modules/teamsalarysettlement"
|
||
uploadmodule "hyapp-admin-server/internal/modules/upload"
|
||
userleaderboardmodule "hyapp-admin-server/internal/modules/userleaderboard"
|
||
vipconfigmodule "hyapp-admin-server/internal/modules/vipconfig"
|
||
weeklystarmodule "hyapp-admin-server/internal/modules/weeklystar"
|
||
wheelmodule "hyapp-admin-server/internal/modules/wheel"
|
||
"hyapp-admin-server/internal/platform/logging"
|
||
"hyapp-admin-server/internal/platform/tencentcos"
|
||
"hyapp-admin-server/internal/repository"
|
||
"hyapp-admin-server/internal/router"
|
||
"hyapp-admin-server/internal/service"
|
||
|
||
mysqlDriver "github.com/go-sql-driver/mysql"
|
||
"go.mongodb.org/mongo-driver/v2/mongo"
|
||
"go.mongodb.org/mongo-driver/v2/mongo/options"
|
||
"go.mongodb.org/mongo-driver/v2/mongo/readpref"
|
||
"google.golang.org/grpc"
|
||
"google.golang.org/grpc/credentials/insecure"
|
||
"google.golang.org/grpc/keepalive"
|
||
"gorm.io/driver/mysql"
|
||
"gorm.io/gorm"
|
||
"gorm.io/gorm/logger"
|
||
walletv1 "hyapp.local/api/proto/wallet/v1"
|
||
)
|
||
|
||
func main() {
|
||
configPath := flag.String("config", "configs/config.yaml", "path to admin-server config")
|
||
runBootstrap := flag.Bool("bootstrap", false, "run configured bootstrap seed before starting")
|
||
flag.Parse()
|
||
|
||
cfg, err := config.Load(*configPath)
|
||
if err != nil {
|
||
log.Fatal(err)
|
||
}
|
||
if err := logging.Init(cfg.ServiceName, cfg.Environment, cfg.NodeID, cfg.Log); err != nil {
|
||
log.Fatal(err)
|
||
}
|
||
|
||
if err := ensureDatabase(cfg.MySQLDSN); err != nil {
|
||
fatalRuntime("ensure_mysql_database_failed", err)
|
||
}
|
||
|
||
dbLogger := logger.New(log.New(logging.StdWriter(), "", 0), logger.Config{
|
||
LogLevel: logger.Warn,
|
||
IgnoreRecordNotFoundError: true,
|
||
})
|
||
db, err := gorm.Open(mysql.Open(cfg.MySQLDSN), &gorm.Config{
|
||
Logger: dbLogger,
|
||
NowFunc: func() time.Time {
|
||
return time.Now().UTC()
|
||
},
|
||
})
|
||
if err != nil {
|
||
fatalRuntime("connect_mysql_failed", err)
|
||
}
|
||
sqlDB, err := db.DB()
|
||
if err != nil {
|
||
fatalRuntime("mysql_db_handle_failed", err)
|
||
}
|
||
if cfg.Migrations.Enabled {
|
||
if err := migration.ApplyWithOptions(context.Background(), sqlDB, cfg.Migrations.Dir, migration.Options{
|
||
AllowChecksumRepair: cfg.Migrations.AllowChecksumRepair,
|
||
}); err != nil {
|
||
fatalRuntime("apply_migrations_failed", err)
|
||
}
|
||
}
|
||
redisClient, err := cache.NewRedis(context.Background(), cfg.Redis)
|
||
if err != nil {
|
||
fatalRuntime("connect_redis_failed", err)
|
||
}
|
||
if redisClient != nil {
|
||
defer redisClient.Close()
|
||
}
|
||
|
||
userDB, err := sql.Open("mysql", cfg.UserMySQLDSN)
|
||
if err != nil {
|
||
fatalRuntime("connect_user_mysql_failed", err)
|
||
}
|
||
if err := userDB.PingContext(context.Background()); err != nil {
|
||
fatalRuntime("ping_user_mysql_failed", err)
|
||
}
|
||
defer userDB.Close()
|
||
|
||
walletDB, err := sql.Open("mysql", cfg.WalletMySQLDSN)
|
||
if err != nil {
|
||
fatalRuntime("connect_wallet_mysql_failed", err)
|
||
}
|
||
if err := walletDB.PingContext(context.Background()); err != nil {
|
||
fatalRuntime("ping_wallet_mysql_failed", err)
|
||
}
|
||
defer walletDB.Close()
|
||
|
||
var robotProfileSource gamemanagementmodule.RobotProfileSource
|
||
if strings.TrimSpace(cfg.RobotProfileSource.MySQLDSN) != "" {
|
||
robotProfileDB, err := sql.Open("mysql", cfg.RobotProfileSource.MySQLDSN)
|
||
if err != nil {
|
||
fatalRuntime("connect_robot_profile_mysql_failed", err)
|
||
}
|
||
robotProfileCtx, robotProfileCancel := context.WithTimeout(context.Background(), cfg.RobotProfileSource.RequestTimeout)
|
||
if err := robotProfileDB.PingContext(robotProfileCtx); err != nil {
|
||
robotProfileCancel()
|
||
fatalRuntime("ping_robot_profile_mysql_failed", err)
|
||
}
|
||
robotProfileCancel()
|
||
defer robotProfileDB.Close()
|
||
robotProfileSource = gamemanagementmodule.NewLikeiRobotProfileSource(robotProfileDB)
|
||
}
|
||
moneyRegionSources, closeMoneyRegionSources, err := connectMoneyRegionSources(context.Background(), cfg.MoneyRegionSources)
|
||
if err != nil {
|
||
fatalRuntime("connect_money_region_sources_failed", err)
|
||
}
|
||
defer closeMoneyRegionSources(context.Background())
|
||
dashboardExternalSources, closeDashboardExternalSources, err := connectDashboardExternalSources(context.Background(), cfg.DashboardExternalSources, dashboardRegionResolvers(moneyRegionSources)...)
|
||
if err != nil {
|
||
fatalRuntime("connect_dashboard_external_sources_failed", err)
|
||
}
|
||
defer closeDashboardExternalSources()
|
||
financeBillSources, closeFinanceBillSources, err := connectFinanceBillSources(context.Background(), cfg.FinanceBillSources, moneyRegionSources, dashboardExternalSourceIndex(dashboardExternalSources))
|
||
if err != nil {
|
||
closeDashboardExternalSources()
|
||
fatalRuntime("connect_finance_bill_sources_failed", err)
|
||
}
|
||
defer closeFinanceBillSources(context.Background())
|
||
|
||
store := repository.New(db)
|
||
if cfg.MySQLAutoMigrate {
|
||
if err := store.AutoMigrate(); err != nil {
|
||
fatalRuntime("auto_migrate_failed", err)
|
||
}
|
||
}
|
||
// legacy 账单源的谷歌实付同步依赖 admin 库缓存表，store 建好后再注入；凭证异常只降级该 App，不阻断启动。
|
||
wireLegacyGooglePaidSync(financeBillSources, cfg.FinanceBillSources, store)
|
||
if *runBootstrap || cfg.Bootstrap.Enabled {
|
||
if err := store.Seed(cfg); err != nil {
|
||
fatalRuntime("seed_database_failed", err)
|
||
}
|
||
}
|
||
|
||
userConn, err := dialBackendGRPC(cfg.UserService.Addr)
|
||
if err != nil {
|
||
fatalRuntime("connect_user_service_failed", err)
|
||
}
|
||
defer userConn.Close()
|
||
|
||
walletConn, err := dialBackendGRPC(cfg.WalletService.Addr)
|
||
if err != nil {
|
||
fatalRuntime("connect_wallet_service_failed", err)
|
||
}
|
||
defer walletConn.Close()
|
||
|
||
roomConn, err := dialBackendGRPC(cfg.RoomService.Addr)
|
||
if err != nil {
|
||
fatalRuntime("connect_room_service_failed", err)
|
||
}
|
||
defer roomConn.Close()
|
||
roomClient := roomclient.NewGRPC(roomConn)
|
||
|
||
robotConn, err := dialBackendGRPC(cfg.RobotService.Addr)
|
||
if err != nil {
|
||
fatalRuntime("connect_robot_service_failed", err)
|
||
}
|
||
defer robotConn.Close()
|
||
robotClient := robotclient.NewGRPC(robotConn)
|
||
|
||
activityConn, err := dialBackendGRPC(cfg.ActivityService.Addr)
|
||
if err != nil {
|
||
fatalRuntime("connect_activity_service_failed", err)
|
||
}
|
||
defer activityConn.Close()
|
||
|
||
gameConn, err := dialBackendGRPC(cfg.GameService.Addr)
|
||
if err != nil {
|
||
fatalRuntime("connect_game_service_failed", err)
|
||
}
|
||
defer gameConn.Close()
|
||
|
||
auth := service.NewAuthService(cfg.JWTSecret, cfg.AccessTokenTTL)
|
||
var runner *jobrunner.Runner
|
||
var jobStatus healthmodule.JobStatusProvider
|
||
if cfg.Jobs.Enabled {
|
||
runner = jobrunner.NewRunner(store, redisClient, cfg)
|
||
runner.Start()
|
||
defer runner.Close()
|
||
jobStatus = runner
|
||
}
|
||
|
||
auditHandler := auditmodule.New(store)
|
||
teamSalarySettlementHandler := teamsalarysettlementmodule.New(sqlDB, walletDB, userDB, auditHandler)
|
||
if cfg.Jobs.Enabled {
|
||
teamSalaryRunner := teamsalarysettlementmodule.NewAutoRunner(teamSalarySettlementHandler.Service(), "lalu", cfg.NodeID)
|
||
teamSalaryRunner.Start()
|
||
defer teamSalaryRunner.Close()
|
||
}
|
||
var objectUploader uploadmodule.ObjectUploader
|
||
if cfg.TencentCOS.Enabled {
|
||
uploader, err := tencentcos.NewUploader(tencentcos.Config{
|
||
SecretID: cfg.TencentCOS.SecretID,
|
||
SecretKey: cfg.TencentCOS.SecretKey,
|
||
Bucket: cfg.TencentCOS.BucketName,
|
||
Region: cfg.TencentCOS.Region,
|
||
AccessURL: cfg.TencentCOS.AccessURL,
|
||
}, nil)
|
||
if err != nil {
|
||
fatalRuntime("create_tencent_cos_uploader_failed", err)
|
||
}
|
||
objectUploader = uploader
|
||
}
|
||
var financeApplicationOptions []financeapplicationmodule.Option
|
||
if cfg.FinanceNotifications.DingTalk.Enabled && strings.TrimSpace(cfg.FinanceNotifications.DingTalk.WebhookURL) != "" {
|
||
client, err := dingtalkclient.New(dingtalkclient.Config{
|
||
WebhookURL: cfg.FinanceNotifications.DingTalk.WebhookURL,
|
||
Secret: cfg.FinanceNotifications.DingTalk.Secret,
|
||
AtMobiles: cfg.FinanceNotifications.DingTalk.AtMobiles,
|
||
AtAll: cfg.FinanceNotifications.DingTalk.AtAll,
|
||
RequestTimeout: cfg.FinanceNotifications.DingTalk.RequestTimeout,
|
||
}, nil)
|
||
if err != nil {
|
||
fatalRuntime("create_finance_dingtalk_client_failed", err)
|
||
}
|
||
financeApplicationOptions = append(financeApplicationOptions, financeapplicationmodule.WithApplicationNotifier(financeapplicationmodule.NewDingTalkNotifier(client)))
|
||
}
|
||
financeApplicationOptions = append(financeApplicationOptions,
|
||
financeapplicationmodule.WithUserClient(userclient.NewGRPC(userConn)),
|
||
financeapplicationmodule.WithWalletClient(walletclient.NewGRPC(walletConn)),
|
||
)
|
||
// 社交 BI 与大屏共享同一个 dashboard 服务实例，保证外部源路由和统计口径一致。
|
||
dashboardService := dashboardmodule.NewService(store, cfg, userclient.NewGRPC(userConn), dashboardmodule.WithExternalDashboardSources(dashboardExternalSources...))
|
||
databiService := databimodule.NewService(
|
||
store,
|
||
dashboardService,
|
||
appregistrymodule.NewService(userDB),
|
||
userclient.NewGRPC(userConn),
|
||
databimodule.WithLegacyApps(databiLegacyApps(cfg.DashboardExternalSources)...),
|
||
databimodule.WithLegacyRegionCatalogs(databiLegacyRegionCatalogs(moneyRegionSources)...),
|
||
)
|
||
handlers := router.Handlers{
|
||
Audit: auditHandler,
|
||
Auth: authmodule.New(store, auth, auditHandler, cfg),
|
||
AdminUser: adminusermodule.New(store, cfg, auditHandler),
|
||
AgencyOpening: agencyopeningmodule.New(activityclient.NewGRPC(activityConn), auditHandler),
|
||
AchievementConfig: achievementconfigmodule.New(activityclient.NewGRPC(activityConn), auditHandler),
|
||
AppConfig: appconfigmodule.New(store, auditHandler),
|
||
AppRegistry: appregistrymodule.New(userDB),
|
||
AppUser: appusermodule.New(userclient.NewGRPC(userConn), activityclient.NewGRPC(activityConn), sqlDB, userDB, walletDB, cfg, auditHandler),
|
||
CoinLedger: coinledgermodule.New(userDB, walletDB, sqlDB, walletclient.NewGRPC(walletConn), auditHandler),
|
||
CountryRegion: countryregionmodule.New(userclient.NewGRPC(userConn), userDB, cfg, auditHandler),
|
||
CPRelation: cprelationmodule.NewWithActivity(userDB, activityclient.NewGRPC(activityConn), auditHandler),
|
||
CPWeeklyRank: cpweeklyrankmodule.New(activityclient.NewGRPC(activityConn), auditHandler),
|
||
CumulativeRecharge: cumulativerechargerewardmodule.New(activityclient.NewGRPC(activityConn), userDB, auditHandler),
|
||
DailyTask: dailytaskmodule.New(activityclient.NewGRPC(activityConn), auditHandler),
|
||
Dashboard: dashboardmodule.NewWithService(dashboardService),
|
||
Databi: databimodule.New(databiService, store, auditHandler),
|
||
FirstRechargeReward: firstrechargerewardmodule.New(activityclient.NewGRPC(activityConn), userDB, auditHandler),
|
||
FinanceApplication: financeapplicationmodule.New(store, auditHandler, financeApplicationOptions...),
|
||
FinanceWithdrawal: financewithdrawalmodule.New(store, walletclient.NewGRPC(walletConn), activityclient.NewGRPC(activityConn), auditHandler),
|
||
FullServerNotice: fullservernoticemodule.New(activityclient.NewGRPC(activityConn), auditHandler),
|
||
Game: gamemanagementmodule.New(
|
||
gameclient.NewGRPC(gameConn),
|
||
userclient.NewGRPC(userConn),
|
||
auditHandler,
|
||
gamemanagementmodule.WithRobotProfileSource(robotProfileSource),
|
||
gamemanagementmodule.WithRobotAppearanceServices(walletclient.NewGRPC(walletConn), activityclient.NewGRPC(activityConn)),
|
||
),
|
||
GiftDiamond: giftdiamondmodule.New(walletDB, auditHandler),
|
||
Health: healthmodule.New(store, redisClient, jobStatus),
|
||
HostAgencyPolicy: hostagencypolicymodule.New(store, walletDB, auditHandler),
|
||
HostOrg: hostorgmodule.New(userclient.NewGRPC(userConn), walletclient.NewGRPC(walletConn), userDB, walletDB, sqlDB, auditHandler),
|
||
HostSalarySettlement: hostsalarysettlementmodule.New(walletDB, userDB, walletv1.NewWalletCronServiceClient(walletConn), cfg.WalletService.RequestTimeout, auditHandler),
|
||
HostWithdrawal: hostwithdrawalmodule.New(sqlDB, walletDB, userDB),
|
||
InviteActivityReward: inviteactivityrewardmodule.New(activityclient.NewGRPC(activityConn), userDB, auditHandler),
|
||
Job: jobmodule.New(store, cfg, auditHandler),
|
||
LevelConfig: levelconfigmodule.New(activityclient.NewGRPC(activityConn), walletclient.NewGRPC(walletConn), auditHandler),
|
||
LuckyGift: luckygiftmodule.New(activityclient.NewGRPC(activityConn), cfg.ActivityService.RequestTimeout, auditHandler),
|
||
Menu: menumodule.New(store, auditHandler),
|
||
Payment: paymentmodule.New(walletclient.NewGRPC(walletConn), userDB, walletDB, store, auditHandler, paymentmodule.WithMoneyRegionSources(moneyRegionSources...), paymentmodule.WithRechargeBillSources(financeBillSources...)),
|
||
PrettyID: prettyidmodule.New(userclient.NewGRPC(userConn), auditHandler),
|
||
RBAC: rbacmodule.New(store, auditHandler),
|
||
RedPacket: redpacketmodule.New(walletclient.NewGRPC(walletConn), auditHandler),
|
||
Report: reportmodule.New(userDB, roomClient),
|
||
RegistrationReward: registrationrewardmodule.New(activityclient.NewGRPC(activityConn), userDB, auditHandler),
|
||
RegionBlock: regionblockmodule.New(userDB, userclient.NewGRPC(userConn), auditHandler),
|
||
Resource: resourcemodule.New(walletclient.NewGRPC(walletConn), store, userDB, cfg.WalletService.RequestTimeout, auditHandler),
|
||
RiskConfig: riskconfigmodule.New(userclient.NewGRPC(userConn), auditHandler),
|
||
RoomAdmin: roomadminmodule.New(userDB, store, roomClient, robotClient, auditHandler),
|
||
RoomRocket: roomrocketmodule.New(roomClient, auditHandler),
|
||
RoomTurnoverReward: roomturnoverrewardmodule.New(activityclient.NewGRPC(activityConn), userDB, auditHandler),
|
||
Search: searchmodule.New(store),
|
||
SevenDayCheckIn: sevendaycheckinmodule.New(activityclient.NewGRPC(activityConn), userDB, auditHandler),
|
||
Team: teammodule.New(store, auditHandler),
|
||
TeamSalaryPolicy: teamsalarypolicymodule.New(store, auditHandler),
|
||
TeamSalarySettlement: teamSalarySettlementHandler,
|
||
Upload: uploadmodule.New(objectUploader, cfg.TencentCOS.ObjectPrefix, auditHandler),
|
||
UserLeaderboard: userleaderboardmodule.New(walletDB, userDB, roomClient),
|
||
VIPConfig: vipconfigmodule.New(walletclient.NewGRPC(walletConn), auditHandler),
|
||
WeeklyStar: weeklystarmodule.New(activityclient.NewGRPC(activityConn), auditHandler),
|
||
Wheel: wheelmodule.New(activityclient.NewGRPC(activityConn), cfg.ActivityService.RequestTimeout, auditHandler),
|
||
}
|
||
engine := router.New(cfg, auth, handlers)
|
||
|
||
server := &http.Server{
|
||
Addr: cfg.HTTPAddr,
|
||
Handler: engine,
|
||
ReadHeaderTimeout: 5 * time.Second,
|
||
}
|
||
|
||
go func() {
|
||
slog.Info("service_listening", "addr", cfg.HTTPAddr)
|
||
if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||
fatalRuntime("listen_failed", err)
|
||
}
|
||
}()
|
||
|
||
quit := make(chan os.Signal, 1)
|
||
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
|
||
<-quit
|
||
|
||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||
defer cancel()
|
||
if err := server.Shutdown(ctx); err != nil {
|
||
slog.Error("shutdown_failed", "error", err.Error())
|
||
}
|
||
}
|
||
|
||
func fatalRuntime(msg string, err error) {
|
||
slog.Error(msg, "error", err.Error())
|
||
os.Exit(1)
|
||
}
|
||
|
||
func dialBackendGRPC(addr string) (*grpc.ClientConn, error) {
|
||
// admin-server 在广州跨区访问 Saudi 内网 CLB；gRPC 长连接如果长时间空闲，
|
||
// TCP 层可能留下半开连接。客户端 keepalive 主动探活并触发重连，避免请求卡到 HTTP 超时。
|
||
return grpc.Dial(addr,
|
||
grpc.WithTransportCredentials(insecure.NewCredentials()),
|
||
grpc.WithKeepaliveParams(keepalive.ClientParameters{
|
||
Time: 30 * time.Second,
|
||
Timeout: 10 * time.Second,
|
||
PermitWithoutStream: true,
|
||
}),
|
||
)
|
||
}
|
||
|
||
func connectMoneyRegionSources(ctx context.Context, configs []config.MoneyRegionSourceConfig) ([]paymentmodule.MoneyRegionSource, func(context.Context), error) {
|
||
sources := []paymentmodule.MoneyRegionSource{}
|
||
clients := []*mongo.Client{}
|
||
cleanup := func(closeCtx context.Context) {
|
||
for _, client := range clients {
|
||
if client != nil {
|
||
_ = client.Disconnect(closeCtx)
|
||
}
|
||
}
|
||
}
|
||
for _, sourceConfig := range configs {
|
||
if !sourceConfig.Enabled {
|
||
continue
|
||
}
|
||
connectCtx, cancel := context.WithTimeout(ctx, sourceConfig.RequestTimeout)
|
||
// 财务范围页面依赖 legacy likei 区域目录；启用后启动阶段先 ping，避免运行时静默漏掉 Yumi/Aslan 区域。
|
||
client, err := mongo.Connect(options.Client().ApplyURI(sourceConfig.MongoURI).SetServerSelectionTimeout(sourceConfig.RequestTimeout))
|
||
if err != nil {
|
||
cancel()
|
||
cleanup(ctx)
|
||
return nil, nil, err
|
||
}
|
||
if err := client.Ping(connectCtx, readpref.Primary()); err != nil {
|
||
cancel()
|
||
_ = client.Disconnect(ctx)
|
||
cleanup(ctx)
|
||
return nil, nil, err
|
||
}
|
||
cancel()
|
||
clients = append(clients, client)
|
||
source := paymentmodule.NewMongoMoneyRegionSource(client.Database(sourceConfig.MongoDatabase), sourceConfig)
|
||
if source != nil {
|
||
sources = append(sources, source)
|
||
}
|
||
}
|
||
return sources, cleanup, nil
|
||
}
|
||
|
||
// connectFinanceBillSources 连接 legacy App 的账单 Mongo；启动阶段 ping 失败直接退出，避免财务明细静默漏掉 Yumi 账单。
|
||
func connectFinanceBillSources(ctx context.Context, configs []config.FinanceBillSourceConfig, regionSources []paymentmodule.MoneyRegionSource, dashboardSources map[string]dashboardmodule.ExternalDashboardSource) ([]paymentmodule.RechargeBillSource, func(context.Context), error) {
|
||
sources := []paymentmodule.RechargeBillSource{}
|
||
clients := []*mongo.Client{}
|
||
cleanup := func(closeCtx context.Context) {
|
||
for _, client := range clients {
|
||
if client != nil {
|
||
_ = client.Disconnect(closeCtx)
|
||
}
|
||
}
|
||
}
|
||
for _, sourceConfig := range configs {
|
||
if !sourceConfig.Enabled {
|
||
continue
|
||
}
|
||
connectCtx, cancel := context.WithTimeout(ctx, sourceConfig.RequestTimeout)
|
||
client, err := mongo.Connect(options.Client().ApplyURI(sourceConfig.MongoURI).SetServerSelectionTimeout(sourceConfig.RequestTimeout))
|
||
if err != nil {
|
||
cancel()
|
||
cleanup(ctx)
|
||
return nil, nil, err
|
||
}
|
||
if err := client.Ping(connectCtx, readpref.Primary()); err != nil {
|
||
cancel()
|
||
_ = client.Disconnect(ctx)
|
||
cleanup(ctx)
|
||
return nil, nil, err
|
||
}
|
||
cancel()
|
||
clients = append(clients, client)
|
||
source := paymentmodule.NewMongoRechargeBillSource(client.Database(sourceConfig.MongoDatabase), sourceConfig, regionSources...)
|
||
if source != nil {
|
||
dashboardSource := dashboardSources[source.AppCode()]
|
||
if dashboardSource == nil {
|
||
cleanup(ctx)
|
||
return nil, nil, fmt.Errorf("finance bill source %s requires matching dashboard_external_sources entry for coin seller stats", source.AppCode())
|
||
}
|
||
source.SetDashboardCoinSellerSource(dashboardSource)
|
||
sources = append(sources, source)
|
||
}
|
||
}
|
||
return sources, cleanup, nil
|
||
}
|
||
|
||
func dashboardExternalSourceIndex(sources []dashboardmodule.ExternalDashboardSource) map[string]dashboardmodule.ExternalDashboardSource {
|
||
index := map[string]dashboardmodule.ExternalDashboardSource{}
|
||
for _, source := range sources {
|
||
if source == nil {
|
||
continue
|
||
}
|
||
appCode := strings.ToLower(strings.TrimSpace(source.AppCode()))
|
||
if appCode == "" {
|
||
continue
|
||
}
|
||
index[appCode] = source
|
||
}
|
||
return index
|
||
}
|
||
|
||
// wireLegacyGooglePaidSync 为配置了 Play Console 包名与服务账号的 legacy 账单源开启谷歌实付同步。
|
||
func wireLegacyGooglePaidSync(sources []paymentmodule.RechargeBillSource, configs []config.FinanceBillSourceConfig, store *repository.Store) {
|
||
configsByApp := map[string]config.FinanceBillSourceConfig{}
|
||
for _, sourceConfig := range configs {
|
||
configsByApp[strings.ToLower(strings.TrimSpace(sourceConfig.AppCode))] = sourceConfig
|
||
}
|
||
for _, source := range sources {
|
||
mongoSource, ok := source.(*paymentmodule.MongoRechargeBillSource)
|
||
if !ok {
|
||
continue
|
||
}
|
||
sourceConfig := configsByApp[source.AppCode()]
|
||
if sourceConfig.GooglePackageName == "" ||
|
||
(sourceConfig.GoogleServiceAccountJSON == "" && sourceConfig.GoogleServiceAccountFile == "") {
|
||
continue
|
||
}
|
||
client, err := googleorders.New(googleorders.Config{
|
||
PackageName: sourceConfig.GooglePackageName,
|
||
ServiceAccountJSON: sourceConfig.GoogleServiceAccountJSON,
|
||
ServiceAccountFile: sourceConfig.GoogleServiceAccountFile,
|
||
})
|
||
if err != nil {
|
||
slog.Error("legacy_google_paid_sync_disabled", "app_code", source.AppCode(), "error", err.Error())
|
||
continue
|
||
}
|
||
mongoSource.SetGooglePaidSync(store, client)
|
||
slog.Info("legacy_google_paid_sync_enabled", "app_code", source.AppCode(), "package_name", sourceConfig.GooglePackageName)
|
||
}
|
||
}
|
||
|
||
// databiLegacyApps 把 dashboard 外部源配置映射成社交 BI 的外接 App 目录；
|
||
// 同一 App 配多个源（如 cdc MySQL + Mongo 补充）时只取一次。
|
||
func databiLegacyApps(configs []config.DashboardExternalSourceConfig) []databimodule.LegacyAppDescriptor {
|
||
out := []databimodule.LegacyAppDescriptor{}
|
||
seen := map[string]struct{}{}
|
||
for _, sourceConfig := range configs {
|
||
if !sourceConfig.Enabled {
|
||
continue
|
||
}
|
||
appCode := strings.ToLower(strings.TrimSpace(sourceConfig.AppCode))
|
||
if appCode == "" {
|
||
continue
|
||
}
|
||
if _, ok := seen[appCode]; ok {
|
||
continue
|
||
}
|
||
seen[appCode] = struct{}{}
|
||
out = append(out, databimodule.LegacyAppDescriptor{AppCode: appCode, AppName: strings.TrimSpace(sourceConfig.AppName)})
|
||
}
|
||
return out
|
||
}
|
||
|
||
func databiLegacyRegionCatalogs(sources []paymentmodule.MoneyRegionSource) []databimodule.LegacyRegionCatalogSource {
|
||
out := []databimodule.LegacyRegionCatalogSource{}
|
||
for _, source := range sources {
|
||
if catalog, ok := source.(databimodule.LegacyRegionCatalogSource); ok && catalog != nil {
|
||
out = append(out, catalog)
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func dashboardRegionResolvers(sources []paymentmodule.MoneyRegionSource) []dashboardmodule.ExternalDashboardRegionResolver {
|
||
resolvers := []dashboardmodule.ExternalDashboardRegionResolver{}
|
||
for _, source := range sources {
|
||
resolver, ok := source.(dashboardmodule.ExternalDashboardRegionResolver)
|
||
if ok && resolver != nil {
|
||
resolvers = append(resolvers, resolver)
|
||
}
|
||
}
|
||
return resolvers
|
||
}
|
||
|
||
func connectDashboardExternalSources(ctx context.Context, configs []config.DashboardExternalSourceConfig, regionResolvers ...dashboardmodule.ExternalDashboardRegionResolver) ([]dashboardmodule.ExternalDashboardSource, func(), error) {
|
||
sources := []dashboardmodule.ExternalDashboardSource{}
|
||
dbs := []*sql.DB{}
|
||
mongoClients := []*mongo.Client{}
|
||
cleanup := func() {
|
||
for _, db := range dbs {
|
||
if db != nil {
|
||
_ = db.Close()
|
||
}
|
||
}
|
||
closeCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer cancel()
|
||
for _, client := range mongoClients {
|
||
if client != nil {
|
||
_ = client.Disconnect(closeCtx)
|
||
}
|
||
}
|
||
}
|
||
for _, sourceConfig := range configs {
|
||
if !sourceConfig.Enabled {
|
||
continue
|
||
}
|
||
switch strings.ToLower(strings.TrimSpace(sourceConfig.SourceType)) {
|
||
case "aslan_http":
|
||
// Aslan 大屏数据由对方 console 免鉴权接口返回；这里不建立数据库连接，避免 admin 绕过 Aslan 服务端口径。
|
||
source, err := dashboardmodule.NewAslanHTTPDashboardSource(sourceConfig, regionResolvers...)
|
||
if err != nil {
|
||
cleanup()
|
||
return nil, nil, err
|
||
}
|
||
if source != nil {
|
||
sources = append(sources, source)
|
||
}
|
||
continue
|
||
case "aslan_mongo":
|
||
// Aslan 当前要求 admin-server 直接读外接项目 Mongo 聚合事实；启动期先 ping，
|
||
// 避免前端选择 Aslan 后才暴露连接或凭证错误。
|
||
client, err := mongo.Connect(options.Client().ApplyURI(sourceConfig.MongoURI).SetServerSelectionTimeout(sourceConfig.RequestTimeout))
|
||
if err != nil {
|
||
cleanup()
|
||
return nil, nil, err
|
||
}
|
||
pingCtx, cancel := context.WithTimeout(ctx, sourceConfig.RequestTimeout)
|
||
if err := client.Ping(pingCtx, readpref.PrimaryPreferred()); err != nil {
|
||
cancel()
|
||
_ = client.Disconnect(ctx)
|
||
cleanup()
|
||
return nil, nil, err
|
||
}
|
||
cancel()
|
||
var legacyDB *sql.DB
|
||
if strings.TrimSpace(sourceConfig.LegacyMySQLDSN) != "" {
|
||
// 金币代理发货充值金额只存在 legacy RDS；它是 Aslan Mongo 大屏的补充事实源，失败应在启动期暴露。
|
||
legacyDB, err = sql.Open("mysql", sourceConfig.LegacyMySQLDSN)
|
||
if err != nil {
|
||
_ = client.Disconnect(ctx)
|
||
cleanup()
|
||
return nil, nil, fmt.Errorf("open dashboard legacy mysql for app %s: %w", sourceConfig.AppCode, err)
|
||
}
|
||
legacyCtx, legacyCancel := context.WithTimeout(ctx, sourceConfig.RequestTimeout)
|
||
if err := legacyDB.PingContext(legacyCtx); err != nil {
|
||
legacyCancel()
|
||
_ = legacyDB.Close()
|
||
_ = client.Disconnect(ctx)
|
||
cleanup()
|
||
return nil, nil, fmt.Errorf("ping dashboard legacy mysql for app %s: %w", sourceConfig.AppCode, err)
|
||
}
|
||
legacyCancel()
|
||
}
|
||
source, err := dashboardmodule.NewAslanMongoDashboardSourceWithLegacyDB(client.Database(sourceConfig.MongoDatabase), legacyDB, sourceConfig, regionResolvers...)
|
||
if err != nil {
|
||
if legacyDB != nil {
|
||
_ = legacyDB.Close()
|
||
}
|
||
_ = client.Disconnect(ctx)
|
||
cleanup()
|
||
return nil, nil, err
|
||
}
|
||
if source == nil {
|
||
if legacyDB != nil {
|
||
_ = legacyDB.Close()
|
||
}
|
||
_ = client.Disconnect(ctx)
|
||
continue
|
||
}
|
||
if legacyDB != nil {
|
||
dbs = append(dbs, legacyDB)
|
||
}
|
||
mongoClients = append(mongoClients, client)
|
||
sources = append(sources, source)
|
||
continue
|
||
case "", "mysql":
|
||
default:
|
||
cleanup()
|
||
return nil, nil, errors.New("unsupported dashboard external source type: " + sourceConfig.SourceType)
|
||
}
|
||
db, err := sql.Open("mysql", sourceConfig.MySQLDSN)
|
||
if err != nil {
|
||
cleanup()
|
||
return nil, nil, err
|
||
}
|
||
pingCtx, cancel := context.WithTimeout(ctx, sourceConfig.RequestTimeout)
|
||
// Yumi 大屏使用外接项目 dashboard-cdc-worker 的只读聚合库；启动期先 ping，避免 admin 正常启动但 Yumi 查询静默回退到错误事实域。
|
||
if err := db.PingContext(pingCtx); err != nil {
|
||
cancel()
|
||
_ = db.Close()
|
||
cleanup()
|
||
return nil, nil, err
|
||
}
|
||
cancel()
|
||
source := dashboardmodule.NewMySQLExternalDashboardSource(db, sourceConfig, regionResolvers...)
|
||
if source == nil {
|
||
_ = db.Close()
|
||
continue
|
||
}
|
||
dbs = append(dbs, db)
|
||
sources = append(sources, source)
|
||
}
|
||
return sources, cleanup, nil
|
||
}
|
||
|
||
func ensureDatabase(dsn string) error {
|
||
parsed, err := mysqlDriver.ParseDSN(dsn)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if parsed.DBName == "" {
|
||
return nil
|
||
}
|
||
|
||
dbName := parsed.DBName
|
||
parsed.DBName = ""
|
||
|
||
db, err := sql.Open("mysql", parsed.FormatDSN())
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer db.Close()
|
||
|
||
escaped := strings.ReplaceAll(dbName, "`", "``")
|
||
_, err = db.Exec("CREATE DATABASE IF NOT EXISTS `" + escaped + "` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci")
|
||
return err
|
||
}
|