package app import ( "context" "fmt" "log" "net/http" "os" "time" "modelRT/config" "modelRT/constants" "modelRT/database" "modelRT/diagram" locker "modelRT/distributedlock" "modelRT/logger" "modelRT/middleware" "modelRT/model" "modelRT/mq" "modelRT/pool" "modelRT/real-time-data/alert" "modelRT/router" "modelRT/task" "modelRT/util" "github.com/gin-contrib/cors" "github.com/gin-gonic/gin" "github.com/gomodule/redigo/redis" "github.com/panjf2000/ants/v2" swaggerFiles "github.com/swaggo/files" ginSwagger "github.com/swaggo/gin-swagger" sdktrace "go.opentelemetry.io/otel/sdk/trace" "gorm.io/gorm" ) // provideConfig loads the service configuration, then initializes the global // logger. Because every other provider depends transitively on the config, // initializing the logger here guarantees it is ready before any provider that // logs. The returned cleanup flushes the logger on shutdown. func provideConfig(cli CLIArgs) (config.ModelRTConfig, func(), error) { cfg := config.ReadAndInitConfig(cli.ConfigDir, cli.ConfigName, cli.ConfigType) logger.InitLoggerInstance(cfg.LoggerConfig) cleanup := func() { _ = logger.GetLoggerInstance().Sync() } return cfg, cleanup, nil } // provideTracerProvider initializes the OTel tracer provider. A failure here is // non-fatal: tracing is simply disabled, matching the original startup behavior. func provideTracerProvider(ctx context.Context, cfg config.ModelRTConfig) (*sdktrace.TracerProvider, func()) { tp, err := middleware.InitTracerProvider(ctx, cfg) if err != nil { log.Printf("warn: OTLP tracer init failed, tracing disabled: %v", err) return nil, func() {} } cleanup := func() { shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() _ = tp.Shutdown(shutdownCtx) } return tp, cleanup } // provideServiceToken generates the signed client token bound to this host. func provideServiceToken(cfg config.ModelRTConfig) (ServiceToken, error) { hostName, err := os.Hostname() if err != nil { return "", fmt.Errorf("get host name failed: %w", err) } tok, err := util.GenerateClientToken(hostName, cfg.ServiceName, cfg.SecretKey) if err != nil { return "", fmt.Errorf("generate client token failed: %w", err) } return ServiceToken(tok), nil } // providePostgres opens the global Postgres client and closes the underlying // sql.DB on shutdown. func providePostgres(ctx context.Context, cfg config.ModelRTConfig) (*gorm.DB, func(), error) { db := database.InitPostgresDBInstance(ctx, cfg.PostgresDBURI) cleanup := func() { if sqlDB, err := db.DB(); err == nil { _ = sqlDB.Close() } } return db, cleanup, nil } // provideParsePool builds the ants pool that runs model parse tasks. func provideParsePool(cfg config.ModelRTConfig) (ParsePool, func(), error) { p, err := ants.NewPoolWithFunc(cfg.ParseConcurrentQuantity, pool.ParseFunc) if err != nil { return nil, nil, fmt.Errorf("init concurrent parse task pool failed: %w", err) } return ParsePool(p), func() { p.Release() }, nil } // provideSearchPool builds the redigo pool used for autocomplete/search and // wires it into the global autocompleter. func provideSearchPool(cfg config.ModelRTConfig) (*redis.Pool, func(), error) { sp, err := util.NewRedigoPool(cfg.StorageRedisConfig) if err != nil { return nil, nil, fmt.Errorf("init redigo pool failed: %w", err) } model.InitAutocompleterWithPool(sp) return sp, func() { _ = sp.Close() }, nil } // provideStorageRedis opens the go-redis client backing diagram/storage data. func provideStorageRedis(cfg config.ModelRTConfig) (StorageRedis, func()) { c := diagram.InitRedisClientInstance(cfg.StorageRedisConfig, cfg.DeployEnv) return StorageRedis(c), func() { _ = c.Close() } } // provideLockerRedis opens the go-redis client backing the distributed lock. func provideLockerRedis(cfg config.ModelRTConfig) (LockerRedis, func()) { c := locker.InitClientInstance(cfg.LockerRedisConfig, cfg.DeployEnv) return LockerRedis(c), func() { _ = c.Close() } } // provideAnchorPool builds the ants pool that runs real-time anchor param tasks. func provideAnchorPool(cfg config.ModelRTConfig) (AnchorPool, func(), error) { p, err := pool.AnchorPoolInit(cfg.RTDReceiveConcurrentQuantity) if err != nil { return nil, nil, fmt.Errorf("init concurrent anchor param task pool failed: %w", err) } return AnchorPool(p), func() { p.Release() }, nil } // provideRabbit opens the global RabbitMQ proxy. func provideRabbit(ctx context.Context, cfg config.ModelRTConfig) (*mq.RabbitMQProxy, func()) { proxy := mq.InitRabbitProxy(ctx, cfg.RabbitMQConfig) return proxy, func() { mq.CloseRabbitProxy() } } // provideTaskWorker initializes the async task worker. A failure is non-fatal: // the service continues without a worker, matching the original startup behavior. // The returned worker may be nil; App.StartBackground guards against that. func provideTaskWorker(ctx context.Context, cfg config.ModelRTConfig, db *gorm.DB) (*task.TaskWorker, func()) { tw, err := task.InitTaskWorker(ctx, cfg, db) if err != nil { logger.Error(ctx, "failed to initialize task worker", "error", err) return nil, func() {} } return tw, func() { _ = tw.Stop() } } // provideAlertManager initializes the global alert event manager. Its value is // held by App purely so Wire keeps it in the dependency graph. func provideAlertManager() *alert.EventManager { return alert.InitAlertEventManager() } // provideEngine builds the gin engine with CORS, routes and (outside production) // the swagger UI. func provideEngine(cfg config.ModelRTConfig, token ServiceToken) *gin.Engine { if cfg.DeployEnv == constants.ProductionDeployMode { gin.SetMode(gin.ReleaseMode) } engine := gin.New() engine.Use(cors.New(cors.Config{ AllowOrigins: []string{"*"}, AllowMethods: []string{"GET", "POST", "PUT", "DELETE", "OPTIONS"}, AllowHeaders: []string{"Origin", "Content-Type", "Authorization"}, ExposeHeaders: []string{"Content-Length"}, AllowCredentials: true, MaxAge: 12 * time.Hour, })) router.RegisterRoutes(engine, string(token)) if cfg.DeployEnv != constants.ProductionDeployMode { engine.GET("/swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler)) } return engine } // provideServer builds the HTTP server bound to the configured address. func provideServer(cfg config.ModelRTConfig, engine *gin.Engine) *http.Server { return &http.Server{ Addr: cfg.ServiceAddr, Handler: engine, } } // provideApp aggregates every constructed resource into the App. App holds the // resource handles (pools, redis clients, etc.) so that Wire keeps their // providers — and therefore their cleanups — in the dependency graph. func provideApp( cfg config.ModelRTConfig, server *http.Server, db *gorm.DB, tp *sdktrace.TracerProvider, parsePool ParsePool, searchPool *redis.Pool, storage StorageRedis, lockerClient LockerRedis, anchorPool AnchorPool, rabbit *mq.RabbitMQProxy, taskWorker *task.TaskWorker, alertMgr *alert.EventManager, ) *App { return &App{ cfg: cfg, server: server, db: db, tp: tp, parsePool: parsePool, searchPool: searchPool, storage: storage, locker: lockerClient, anchorPool: anchorPool, rabbit: rabbit, taskWorker: taskWorker, alertMgr: alertMgr, } }