206 lines
6.2 KiB
Go
206 lines
6.2 KiB
Go
package main
|
|
|
|
import (
|
|
"flag"
|
|
"fmt"
|
|
"os"
|
|
"os/signal"
|
|
"strconv"
|
|
"syscall"
|
|
|
|
"dubbo.apache.org/dubbo-go/v3/client"
|
|
_ "dubbo.apache.org/dubbo-go/v3/imports"
|
|
"dubbo.apache.org/dubbo-go/v3/protocol"
|
|
"dubbo.apache.org/dubbo-go/v3/server"
|
|
"github.com/joho/godotenv"
|
|
|
|
"github.com/topfans/backend/pkg/database"
|
|
"github.com/topfans/backend/pkg/health"
|
|
"github.com/topfans/backend/pkg/logger"
|
|
"github.com/topfans/backend/pkg/models"
|
|
pbAsset "github.com/topfans/backend/pkg/proto/asset"
|
|
pbRanking "github.com/topfans/backend/pkg/proto/ranking"
|
|
pbUser "github.com/topfans/backend/pkg/proto/user"
|
|
assetClient "github.com/topfans/backend/services/assetService/client"
|
|
"github.com/topfans/backend/services/assetService/config"
|
|
"github.com/topfans/backend/services/assetService/provider"
|
|
"github.com/topfans/backend/services/assetService/repository"
|
|
"github.com/topfans/backend/services/assetService/service"
|
|
)
|
|
|
|
var (
|
|
port = flag.Int("port", getEnvInt("PORT", 20003), "Dubbo service port")
|
|
dbHost = flag.String("db-host", getEnv("DB_HOST", "localhost"), "Database host")
|
|
dbPort = flag.Int("db-port", getEnvInt("DB_PORT", 5432), "Database port")
|
|
dbUser = flag.String("db-user", getEnv("DB_USER", "postgres"), "Database user")
|
|
dbPassword = flag.String("db-password", getEnv("DB_PASSWORD", ""), "Database password")
|
|
dbName = flag.String("db-name", getEnv("DB_NAME", "top-fans"), "Database name")
|
|
userServiceURL = flag.String("user-service-url", getEnv("USER_SERVICE_URL", "tri://localhost:20000"), "User service URL")
|
|
healthHandler *health.Handler
|
|
)
|
|
|
|
func getEnv(key, fallback string) string {
|
|
if v := os.Getenv(key); v != "" {
|
|
return v
|
|
}
|
|
return fallback
|
|
}
|
|
|
|
func getEnvInt(key string, fallback int) int {
|
|
if v := os.Getenv(key); v != "" {
|
|
if n, err := strconv.Atoi(v); err == nil {
|
|
return n
|
|
}
|
|
}
|
|
return fallback
|
|
}
|
|
|
|
func main() {
|
|
// 加载 .env 文件中的环境变量
|
|
godotenv.Load()
|
|
|
|
flag.Parse()
|
|
|
|
// 初始化日志(必须在最前面)
|
|
env := os.Getenv("ENV")
|
|
if env == "" {
|
|
env = "development"
|
|
}
|
|
|
|
if err := logger.Init(logger.Config{
|
|
ServiceName: "asset-service",
|
|
Environment: env,
|
|
LogLevel: os.Getenv("LOG_LEVEL"),
|
|
}); err != nil {
|
|
panic(fmt.Sprintf("Failed to initialize logger: %v", err))
|
|
}
|
|
defer logger.Sync()
|
|
|
|
logger.Logger.Info("Starting Asset Service...")
|
|
|
|
// 初始化数据库
|
|
dbConfig := database.Config{
|
|
Host: *dbHost,
|
|
Port: *dbPort,
|
|
User: *dbUser,
|
|
Password: *dbPassword,
|
|
DBName: *dbName,
|
|
SSLMode: "disable",
|
|
TimeZone: "Asia/Shanghai",
|
|
}
|
|
|
|
if err := database.Init(dbConfig); err != nil {
|
|
logger.Logger.Fatal(fmt.Sprintf("Failed to initialize database: %v", err))
|
|
}
|
|
logger.Logger.Info("Database initialized successfully")
|
|
|
|
// 启动健康检查 HTTP 服务器
|
|
healthPort := *port + 1000 // e.g., 20003 -> 21003
|
|
healthHandler = health.NewHandler("asset-service", healthPort)
|
|
healthHandler.Start()
|
|
|
|
// 自动迁移数据库表
|
|
if err := autoMigrate(); err != nil {
|
|
logger.Logger.Fatal(fmt.Sprintf("Failed to migrate database: %v", err))
|
|
}
|
|
|
|
// 创建 Repository 层实例
|
|
assetRepo := repository.NewAssetRepository(database.GetDB())
|
|
mintOrderRepo := repository.NewMintOrderRepository(database.GetDB())
|
|
assetLikeRepo := repository.NewAssetLikeRepository(database.GetDB())
|
|
rankingRepo := repository.NewRankingRepository(database.GetDB())
|
|
logger.Logger.Info("Repository layer initialized")
|
|
|
|
// 创建 Dubbo 客户端
|
|
cli, err := client.NewClient(
|
|
client.WithClientURL(*userServiceURL),
|
|
)
|
|
if err != nil {
|
|
logger.Logger.Fatal(fmt.Sprintf("Failed to create Dubbo client: %v", err))
|
|
}
|
|
|
|
// 获取 User Service RPC 客户端
|
|
userServiceClient, err := pbUser.NewUserSocialService(cli)
|
|
if err != nil {
|
|
logger.Logger.Fatal(fmt.Sprintf("Failed to create User Service RPC client: %v", err))
|
|
}
|
|
userClient := assetClient.NewUserServiceClient(userServiceClient)
|
|
logger.Logger.Info("User Service RPC client initialized")
|
|
|
|
// 创建 Service 层实例
|
|
assetService := service.NewAssetService(assetRepo, mintOrderRepo, assetLikeRepo, userClient, database.GetDB())
|
|
mintService := service.NewMintService(assetRepo, mintOrderRepo, userClient, database.GetDB(), config.GlobalAssetConfig)
|
|
assetLikeService := service.NewAssetLikeService(assetRepo, assetLikeRepo, database.GetDB())
|
|
rankingService := service.NewRankingService(rankingRepo, userClient)
|
|
logger.Logger.Info("Service layer initialized")
|
|
|
|
// 创建 Provider 层实例
|
|
assetProvider := provider.NewAssetProvider(assetService, mintService, assetLikeService)
|
|
rankingProvider := provider.NewRankingProvider(rankingService)
|
|
logger.Logger.Info("Provider layer initialized")
|
|
|
|
// 创建 Dubbo 服务器
|
|
srv, err := server.NewServer(
|
|
server.WithServerProtocol(
|
|
protocol.WithPort(*port),
|
|
protocol.WithTriple(),
|
|
),
|
|
)
|
|
if err != nil {
|
|
logger.Logger.Fatal(fmt.Sprintf("Failed to create Dubbo server: %v", err))
|
|
}
|
|
|
|
// 使用 Triple 协议生成的 RegisterHandler 函数注册服务
|
|
if err := pbAsset.RegisterAssetServiceHandler(srv, assetProvider); err != nil {
|
|
logger.Logger.Fatal(fmt.Sprintf("Failed to register Asset Service: %v", err))
|
|
}
|
|
|
|
// 注册 Ranking Service
|
|
if err := pbRanking.RegisterRankingServiceHandler(srv, rankingProvider); err != nil {
|
|
logger.Logger.Fatal(fmt.Sprintf("Failed to register Ranking Service: %v", err))
|
|
}
|
|
|
|
// 启动服务
|
|
if err := srv.Serve(); err != nil {
|
|
logger.Logger.Fatal(fmt.Sprintf("Failed to start Asset Service: %v", err))
|
|
}
|
|
|
|
logger.Logger.Info(fmt.Sprintf("Asset Service started successfully on port %d", *port))
|
|
|
|
// 等待退出信号
|
|
quit := make(chan os.Signal, 1)
|
|
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
|
|
<-quit
|
|
|
|
logger.Logger.Info("Shutting down Asset Service...")
|
|
|
|
// 关闭健康检查服务器
|
|
if healthHandler != nil {
|
|
healthHandler.Stop()
|
|
}
|
|
}
|
|
|
|
// autoMigrate 自动迁移数据库表
|
|
func autoMigrate() error {
|
|
db := database.GetDB()
|
|
if db == nil {
|
|
return fmt.Errorf("database is not initialized")
|
|
}
|
|
|
|
// 按顺序迁移资产相关表
|
|
tables := []interface{}{
|
|
&models.Asset{},
|
|
&models.MintOrder{},
|
|
&models.AssetLike{},
|
|
}
|
|
|
|
for _, table := range tables {
|
|
if err := db.AutoMigrate(table); err != nil {
|
|
return fmt.Errorf("failed to migrate table: %w", err)
|
|
}
|
|
}
|
|
|
|
logger.Logger.Info("Database migration completed successfully")
|
|
return nil
|
|
}
|