chore: remove runtime config bootstrap and update local deploy docs

- stop copying config.example.yaml to config.yaml during app initialization - document the local config.yaml example directly in deploy.md - update Docker smoke test docs to mount configs/config.yaml - bump the K8s Postgres image from 13.16 to 17.5
refactor: move ModelRT startup assembly into Wire app package
2026-06-26 16:01:32 +08:00 · 2026-06-24 15:18:27 +08:00
10 changed files with 628 additions and 261 deletions
--- a/app/app.go
+++ b/app/app.go
@ -0,0 +1,135 @@
+package app
+
+import (
+	"context"
+	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"modelRT/config"
+	"modelRT/database"
+	"modelRT/diagram"
+	"modelRT/logger"
+	"modelRT/model"
+	"modelRT/mq"
+	"modelRT/real-time-data/alert"
+	"modelRT/task"
+
+	"github.com/gomodule/redigo/redis"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	"gorm.io/gorm"
+
+	realtimedata "modelRT/real-time-data"
+)
+
+// App owns the assembled service: the HTTP server plus every long-lived resource
+// constructed during startup. Resource handles are retained so their lifecycles
+// (and the Wire-aggregated cleanup) are tied to the App.
+type App struct {
+	cfg    config.ModelRTConfig
+	server *http.Server
+	db     *gorm.DB
+
+	tp         *sdktrace.TracerProvider
+	parsePool  ParsePool
+	searchPool *redis.Pool
+	storage    StorageRedis
+	locker     LockerRedis
+	anchorPool AnchorPool
+	rabbit     *mq.RabbitMQProxy
+	taskWorker *task.TaskWorker
+	alertMgr   *alert.EventManager
+}
+
+// Warmup loads circuit-diagram, measurement, attribute and topology data from
+// Postgres into the in-memory/redis caches. It mirrors the bootstrap transaction
+// that used to run inline in main(); failures panic, as before.
+func (a *App) Warmup(ctx context.Context) {
+	a.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		cacheMap, err := model.GetNSpathToIsLocalMap(ctx, a.db)
+		if err != nil {
+			logger.Error(ctx, "get nspath to is_local map failed", "error", err)
+			panic(err)
+		}
+		model.NSPathToIsLocalMap = cacheMap
+
+		if err := model.CleanupRecommendRedisCache(ctx); err != nil {
+			logger.Error(ctx, "clean up component measurement and attribute group failed", "error", err)
+			panic(err)
+		}
+
+		measurementSet, err := database.GetFullMeasurementSet(ctx, a.db)
+		if err != nil {
+			logger.Error(ctx, "generate component measurement group failed", "error", err)
+			panic(err)
+		}
+		fullParentPath, isLocalParentPath, err := model.TraverseMeasurementGroupTables(ctx, *measurementSet)
+		if err != nil {
+			logger.Error(ctx, "store component measurement group into redis failed", "error", err)
+			panic(err)
+		}
+
+		compAttrSet, err := database.GenAllAttributeMap(tx)
+		if err != nil {
+			logger.Error(ctx, "generate component attribute group failed", "error", err)
+			panic(err)
+		}
+
+		if err := model.TraverseAttributeGroupTables(ctx, tx, fullParentPath, isLocalParentPath, compAttrSet); err != nil {
+			logger.Error(ctx, "store component attribute group into redis failed", "error", err)
+			panic(err)
+		}
+
+		allMeasurement, err := database.GetAllMeasurements(ctx, tx)
+		if err != nil {
+			logger.Error(ctx, "load topologic info from postgres failed", "error", err)
+			panic(err)
+		}
+		go realtimedata.StartComputingRealTimeDataLimit(ctx, allMeasurement)
+
+		tree, _, err := database.QueryTopologicFromDB(ctx, tx)
+		if err != nil {
+			logger.Error(ctx, "load topologic info from postgres failed", "error", err)
+			panic(err)
+		}
+		diagram.GlobalTree = tree
+		return nil
+	})
+}
+
+// StartBackground launches the async publishers and the task worker. Each runs in
+// its own goroutine, matching the original main() startup.
+func (a *App) StartBackground(ctx context.Context) {
+	if a.taskWorker != nil {
+		go a.taskWorker.Start()
+	}
+	go mq.PushUpDownLimitEventToRabbitMQ(ctx, mq.MsgChan)
+	go mq.PushMessageToRabbitMQ(ctx, mq.MessageMsgChan)
+	go task.PushTaskToRabbitMQ(ctx, a.cfg.RabbitMQConfig, task.TaskMsgChan)
+}
+
+// Run starts the HTTP server and blocks until it is shut down. A received
+// interrupt/terminate signal triggers a graceful shutdown.
+func (a *App) Run(ctx context.Context) {
+	done := make(chan os.Signal, 10)
+	signal.Notify(done, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
+	go func() {
+		<-done
+		logger.Info(ctx, "shutdown signal received, cleaning up...")
+		if err := a.server.Shutdown(context.Background()); err != nil {
+			logger.Error(ctx, "shutdown serverError", "err", err)
+		}
+		logger.Info(ctx, "resources cleaned up, exiting")
+	}()
+
+	logger.Info(ctx, "starting ModelRT server")
+	err := a.server.ListenAndServe()
+	if err != nil {
+		if err == http.ErrServerClosed {
+			logger.Info(ctx, "server closed under request")
+		} else {
+			logger.Error(ctx, "server closed unexpected", "err", err)
+		}
+	}
+}
--- a/app/providers.go
+++ b/app/providers.go
@ -0,0 +1,217 @@
+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,
+	}
+}
--- a/app/types.go
+++ b/app/types.go
@ -0,0 +1,39 @@
+// Package app holds the dependency-injection wiring for the ModelRT service.
+//
+// Phase 1 of the Wire migration centralizes the resource assembly that used to
+// live inline in main(). Package-level singletons and their Get... accessors are
+// intentionally left in place; only the startup assembly is moved here.
+package app
+
+import (
+	"github.com/panjf2000/ants/v2"
+	"github.com/redis/go-redis/v9"
+)
+
+// CLIArgs carries the command-line configuration locators into the injector.
+type CLIArgs struct {
+	ConfigDir  string
+	ConfigName string
+	ConfigType string
+}
+
+// The following named types disambiguate providers that would otherwise share
+// the same underlying type. Wire resolves dependencies by type, so the two
+// *redis.Client instances (storage vs locker) and the two *ants.PoolWithFunc
+// instances (parse vs anchor) must be distinct named types.
+type (
+	// ServiceToken is the signed client token used to authenticate routes.
+	ServiceToken string
+
+	// StorageRedis is the go-redis client backing diagram/storage data.
+	StorageRedis *redis.Client
+
+	// LockerRedis is the go-redis client backing the distributed lock.
+	LockerRedis *redis.Client
+
+	// ParsePool is the ants pool running model parse tasks.
+	ParsePool *ants.PoolWithFunc
+
+	// AnchorPool is the ants pool running real-time anchor param tasks.
+	AnchorPool *ants.PoolWithFunc
+)
--- a/app/wire.go
+++ b/app/wire.go
@ -0,0 +1,34 @@
+//go:build wireinject
+// +build wireinject
+
+package app
+
+import (
+	"context"
+
+	"github.com/google/wire"
+)
+
+// InitializeApp builds the fully assembled App and an aggregated cleanup function
+// from the command-line config locators. The body is a Wire injector declaration;
+// the real implementation is generated into wire_gen.go by `go generate`.
+func InitializeApp(ctx context.Context, cli CLIArgs) (*App, func(), error) {
+	wire.Build(
+		provideConfig,
+		provideTracerProvider,
+		provideServiceToken,
+		providePostgres,
+		provideParsePool,
+		provideSearchPool,
+		provideStorageRedis,
+		provideLockerRedis,
+		provideAnchorPool,
+		provideRabbit,
+		provideTaskWorker,
+		provideAlertManager,
+		provideEngine,
+		provideServer,
+		provideApp,
+	)
+	return nil, nil, nil
+}
--- a/app/wire_gen.go
+++ b/app/wire_gen.go
@ -0,0 +1,80 @@
+// Code generated by Wire. DO NOT EDIT.
+
+//go:generate go run -mod=mod github.com/google/wire/cmd/wire
+//go:build !wireinject
+// +build !wireinject
+
+package app
+
+import (
+	"context"
+)
+
+// Injectors from wire.go:
+
+// InitializeApp builds the fully assembled App and an aggregated cleanup function
+// from the command-line config locators. The body is a Wire injector declaration;
+// the real implementation is generated into wire_gen.go by `go generate`.
+func InitializeApp(ctx context.Context, cli CLIArgs) (*App, func(), error) {
+	modelRTConfig, cleanup, err := provideConfig(cli)
+	if err != nil {
+		return nil, nil, err
+	}
+	serviceToken, err := provideServiceToken(modelRTConfig)
+	if err != nil {
+		cleanup()
+		return nil, nil, err
+	}
+	engine := provideEngine(modelRTConfig, serviceToken)
+	server := provideServer(modelRTConfig, engine)
+	db, cleanup2, err := providePostgres(ctx, modelRTConfig)
+	if err != nil {
+		cleanup()
+		return nil, nil, err
+	}
+	tracerProvider, cleanup3 := provideTracerProvider(ctx, modelRTConfig)
+	parsePool, cleanup4, err := provideParsePool(modelRTConfig)
+	if err != nil {
+		cleanup3()
+		cleanup2()
+		cleanup()
+		return nil, nil, err
+	}
+	pool, cleanup5, err := provideSearchPool(modelRTConfig)
+	if err != nil {
+		cleanup4()
+		cleanup3()
+		cleanup2()
+		cleanup()
+		return nil, nil, err
+	}
+	storageRedis, cleanup6 := provideStorageRedis(modelRTConfig)
+	lockerRedis, cleanup7 := provideLockerRedis(modelRTConfig)
+	anchorPool, cleanup8, err := provideAnchorPool(modelRTConfig)
+	if err != nil {
+		cleanup7()
+		cleanup6()
+		cleanup5()
+		cleanup4()
+		cleanup3()
+		cleanup2()
+		cleanup()
+		return nil, nil, err
+	}
+	rabbitMQProxy, cleanup9 := provideRabbit(ctx, modelRTConfig)
+	taskWorker, cleanup10 := provideTaskWorker(ctx, modelRTConfig, db)
+	eventManager := provideAlertManager()
+	app := provideApp(modelRTConfig, server, db, tracerProvider, parsePool, pool, storageRedis, lockerRedis, anchorPool, rabbitMQProxy, taskWorker, eventManager)
+	return app, func() {
+		cleanup10()
+		cleanup9()
+		cleanup8()
+		cleanup7()
+		cleanup6()
+		cleanup5()
+		cleanup4()
+		cleanup3()
+		cleanup2()
+		cleanup()
+	}, nil
+}
--- a/deploy/deploy.md
+++ b/deploy/deploy.md
@ -444,6 +444,109 @@ go run deploy/redis-test-data/measurments-recommend/measurement_injection.go
 | | `polling_api` | 轮询数据的 `API` 路径。 | `"datart/getPointData"` |
 | | `polling_api_method` | 调用该 `API` 使用的 `HTTP` 方法。 | `"GET"` |

+本地开发可参考以下示例创建 `configs/config.yaml`：
+
+```yaml
+postgres:
+  host: "192.168.1.101"
+  port: 5432
+  database: "demo"
+  user: "postgres"
+  password: "coslight"
+
+kafka:
+  servers: "127.0.0.1:9092"
+  port: 9092
+  group_id: "modelRT"
+  topic: ""
+  auto_offset_reset: "earliest"
+  enable_auto_commit: "false"
+  read_message_time_duration: 0.5
+
+rabbitmq:
+  ca_cert_path: "./configs/certs/ca_certificate.pem"
+  client_key_path: "./configs/certs/modelrt_client_key.pem"
+  client_key_password: ""
+  client_cert_path: "./configs/certs/modelrt_client_cert.pem"
+  insecure_skip_verify: false
+  server_name: "rabbitmq-server"
+  user: ""
+  password: ""
+  host: "127.0.0.1"
+  port: 5671
+
+# zap logger config
+logger:
+  mode: "development"
+  level: "debug"
+  filepath: ""
+  maxsize: 100
+  maxbackups: 5
+  maxage: 30
+  compress: false
+  loki:
+    endpoint: "http://127.0.0.1:3100"
+    labels:
+      app: "model-rt"
+
+otel:
+  endpoint: "127.0.0.1:4318"
+  insecure: true # Jaeger all-in-one 不启用 TLS，生产环境改为 false
+
+# ants config
+ants:
+  parse_concurrent_quantity: 10
+  rtd_receive_concurrent_quantity: 10
+
+# async task config
+async_task:
+  worker_pool_size: 10
+  queue_consumer_count: 2
+  max_retry_count: 3
+  retry_initial_delay: 1s
+  retry_max_delay: 5m
+  health_check_interval: 30s
+
+# redis config
+locker_redis:
+  addr: "127.0.0.1:6379"
+  password: ""
+  db: 1
+  poolsize: 50
+  dial_timeout: 10
+  read_timeout: 10
+  write_timeout: 10
+
+storage_redis:
+  addr: "127.0.0.1:6379"
+  password: ""
+  db: 0
+  poolsize: 50
+  dial_timeout: 10
+  read_timeout: 10
+  write_timeout: 10
+
+# modelRT base config
+base:
+  grid_id: 1
+  zone_id: 1
+  station_id: 1
+
+# modelRT service config
+service:
+  service_addr: ":8080"
+  service_name: "modelRT"
+  secret_key: "modelrt_key"
+  deploy_env: "development"
+
+# dataRT api config
+dataRT:
+  host: "http://127.0.0.1"
+  port: 8888
+  polling_api: "ws/points"
+  polling_api_method: "GET"
+```
+
 #### 3.2 编译 ModelRT 服务

 ```bash
@ -817,9 +920,9 @@ docker inspect coslight/modelrt:latest
 # 验证二进制可执行（无 config 时程序报错退出属预期行为，说明镜像构建正常）
 docker run --rm coslight/modelrt:latest

-# 挂载示例配置做完整启动验证（Ctrl+C 退出）
+# 挂载本地配置做完整启动验证（Ctrl+C 退出）
 docker run --rm \
-  -v "$(pwd)/configs/config.example.yaml:/app/configs/config.yaml" \
+  -v "$(pwd)/configs/config.yaml:/app/configs/config.yaml" \
  -p 8080:8080 \
  coslight/modelrt:latest
 ```
--- a/deploy/k8s/pg-statefulset.yaml
+++ b/deploy/k8s/pg-statefulset.yaml
@ -17,7 +17,7 @@ spec:
    spec:
      containers:
        - name: postgres
-          image: postgres:13.16
+          image: postgres:17.5
          imagePullPolicy: IfNotPresent
          ports:
            - name: postgres
--- a/go.mod
+++ b/go.mod
@ -10,6 +10,7 @@ require (
 	github.com/gin-gonic/gin v1.10.1
 	github.com/gofrs/uuid v4.4.0+incompatible
 	github.com/gomodule/redigo v1.8.9
+	github.com/google/wire v0.7.0
 	github.com/gorilla/websocket v1.5.3
 	github.com/json-iterator/go v1.1.12
 	github.com/natefinch/lumberjack v2.0.0+incompatible
--- a/go.sum
+++ b/go.sum
@ -80,6 +80,8 @@ github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/wire v0.7.0 h1:JxUKI6+CVBgCO2WToKy/nQk0sS+amI9z9EjVmdaocj4=
+github.com/google/wire v0.7.0/go.mod h1:n6YbUQD9cPKTnHXEBN2DXlOp/mVADhVErcMFb0v3J18=
 github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs=
--- a/main.go
+++ b/main.go
@ -3,44 +3,12 @@ package main

 import (
 	"context"
-	"errors"
 	"flag"
-	"fmt"
 	"log"
-	"net/http"
-	"os"
-	"os/signal"
-	"path/filepath"
-	"syscall"
-	"time"

-	"modelRT/config"
-	"modelRT/constants"
-	"modelRT/database"
-	"modelRT/diagram"
-	"modelRT/logger"
-	"modelRT/middleware"
-	"modelRT/model"
-	"modelRT/mq"
-	"modelRT/pool"
-	"modelRT/real-time-data/alert"
-	"modelRT/router"
-	"modelRT/task"
-	"modelRT/util"
+	"modelRT/app"

-	"github.com/gin-contrib/cors"
-
-	locker "modelRT/distributedlock"
 	_ "modelRT/docs"
-
-	realtimedata "modelRT/real-time-data"
-
-	"github.com/gin-gonic/gin"
-	"github.com/panjf2000/ants/v2"
-	swaggerFiles "github.com/swaggo/files"
-	ginSwagger "github.com/swaggo/gin-swagger"
-	"go.opentelemetry.io/otel"
-	"gorm.io/gorm"
 )

 var (
@ -49,13 +17,6 @@ var (
 	modelRTConfigType = flag.String("modelRT_config_type", "yaml", "config file type of model runtime service")
 )

-var (
-	modelRTConfig    config.ModelRTConfig
-	postgresDBClient *gorm.DB
-	// alertManager     *alert.EventManager
-)
-
-// TODO 使用 wire 依赖注入管理 DVIE 面板注册的 panel
 // @title ModelRT 实时模型服务 API 文档
 // @version 1.0
 // @description 实时数据计算和模型运行服务的 API 服务
@ -74,229 +35,24 @@ var (
 func main() {
 	flag.Parse()

-	configPath := filepath.Join(*modelRTConfigDir, *modelRTConfigName+"."+*modelRTConfigType)
-	if _, err := os.Stat(configPath); os.IsNotExist(err) {
-		log.Println("configuration file not found,checking for example file")
+	ctx := context.Background()

-		exampleConfigPath := filepath.Join(*modelRTConfigDir, *modelRTConfigName+".example."+*modelRTConfigType)
-		configDir := filepath.Dir(configPath)
-		if err := os.MkdirAll(configDir, 0o755); err != nil {
-			panic(fmt.Errorf("failed to create config directory %s:%w", configDir, err))
-		}
-		if _, err := os.Stat(exampleConfigPath); err == nil {
-			if err := util.CopyFile(exampleConfigPath, configPath); err != nil {
-				panic(fmt.Errorf("failed to copy example config file:%w", err))
-			}
-		} else {
-			panic(errors.New("no config file and no config example file found"))
-		}
+	cli := app.CLIArgs{
+		ConfigDir:  *modelRTConfigDir,
+		ConfigName: *modelRTConfigName,
+		ConfigType: *modelRTConfigType,
 	}

-	modelRTConfig = config.ReadAndInitConfig(*modelRTConfigDir, *modelRTConfigName, *modelRTConfigType)
-
-	// init logger
-	logger.InitLoggerInstance(modelRTConfig.LoggerConfig)
-	defer logger.GetLoggerInstance().Sync()
-
-	// init OTel TracerProvider
-	tp, tpErr := middleware.InitTracerProvider(context.Background(), modelRTConfig)
-	if tpErr != nil {
-		log.Printf("warn: OTLP tracer init failed, tracing disabled: %v", tpErr)
-	}
-	if tp != nil {
-		defer func() {
-			shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-			defer cancel()
-			tp.Shutdown(shutdownCtx)
-		}()
-	}
-
-	ctx, startupSpan := otel.Tracer("modelRT/main").Start(context.Background(), "startup")
-	defer startupSpan.End()
-
-	hostName, err := os.Hostname()
+	application, cleanup, err := app.InitializeApp(ctx, cli)
 	if err != nil {
-		logger.Error(ctx, "get host name failed", "error", err)
-		panic(err)
+		log.Fatalf("failed to initialize app: %v", err)
 	}
+	defer cleanup()

-	serviceToken, err := util.GenerateClientToken(hostName, modelRTConfig.ServiceName, modelRTConfig.SecretKey)
-	if err != nil {
-		logger.Error(ctx, "generate client token failed", "error", err)
-		panic(err)
-	}
+	// warm caches from postgres, then start background publishers/worker
+	application.Warmup(ctx)
+	application.StartBackground(ctx)

-	// init postgresDBClient
-	postgresDBClient = database.InitPostgresDBInstance(ctx, modelRTConfig.PostgresDBURI)
-
-	defer func() {
-		sqlDB, err := postgresDBClient.DB()
-		if err != nil {
-			panic(err)
-		}
-		sqlDB.Close()
-	}()
-
-	// init alert manager
-	_ = alert.InitAlertEventManager()
-
-	// init model parse ants pool
-	parsePool, err := ants.NewPoolWithFunc(modelRTConfig.ParseConcurrentQuantity, pool.ParseFunc)
-	if err != nil {
-		logger.Error(ctx, "init concurrent parse task pool failed", "error", err)
-		panic(err)
-	}
-	defer parsePool.Release()
-
-	searchPool, err := util.NewRedigoPool(modelRTConfig.StorageRedisConfig)
-	if err != nil {
-		logger.Error(ctx, "init redigo pool failed", "error", err)
-		panic(err)
-	}
-	defer searchPool.Close()
-	model.InitAutocompleterWithPool(searchPool)
-
-	storageClient := diagram.InitRedisClientInstance(modelRTConfig.StorageRedisConfig, modelRTConfig.DeployEnv)
-	defer storageClient.Close()
-
-	lockerClient := locker.InitClientInstance(modelRTConfig.LockerRedisConfig, modelRTConfig.DeployEnv)
-	defer lockerClient.Close()
-
-	// init anchor param ants pool
-	anchorRealTimePool, err := pool.AnchorPoolInit(modelRTConfig.RTDReceiveConcurrentQuantity)
-	if err != nil {
-		logger.Error(ctx, "init concurrent anchor param task pool failed", "error", err)
-		panic(err)
-	}
-	defer anchorRealTimePool.Release()
-
-	// init rabbitmq connection
-	mq.InitRabbitProxy(ctx, modelRTConfig.RabbitMQConfig)
-	defer mq.CloseRabbitProxy()
-
-	// init async task worker
-	taskWorker, err := task.InitTaskWorker(ctx, modelRTConfig, postgresDBClient)
-	if err != nil {
-		logger.Error(ctx, "failed to initialize task worker", "error", err)
-		// continue without task worker, but log warning
-	} else {
-		go taskWorker.Start()
-		defer taskWorker.Stop()
-	}
-
-	// async push event to rabbitMQ
-	go mq.PushUpDownLimitEventToRabbitMQ(ctx, mq.MsgChan)
-	// async push message (task state changes etc.) to rabbitMQ
-	go mq.PushMessageToRabbitMQ(ctx, mq.MessageMsgChan)
-	// async push task message to rabbitMQ
-	go task.PushTaskToRabbitMQ(ctx, modelRTConfig.RabbitMQConfig, task.TaskMsgChan)
-
-	postgresDBClient.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
-		// load circuit diagram from postgres
-		// componentTypeMap, err := database.QueryCircuitDiagramComponentFromDB(cancelCtx, tx, parsePool)
-		// if err != nil {
-		// 	logger.Error(ctx, "load circuit diagrams from postgres failed", "error", err)
-		// 	panic(err)
-		// }
-
-		cacheMap, err := model.GetNSpathToIsLocalMap(ctx, postgresDBClient)
-		if err != nil {
-			logger.Error(ctx, "get nspath to is_local map failed", "error", err)
-			panic(err)
-		}
-		model.NSPathToIsLocalMap = cacheMap
-
-		err = model.CleanupRecommendRedisCache(ctx)
-		if err != nil {
-			logger.Error(ctx, "clean up component measurement and attribute group failed", "error", err)
-			panic(err)
-		}
-
-		measurementSet, err := database.GetFullMeasurementSet(ctx, postgresDBClient)
-		if err != nil {
-			logger.Error(ctx, "generate component measurement group failed", "error", err)
-			panic(err)
-		}
-		fullParentPath, isLocalParentPath, err := model.TraverseMeasurementGroupTables(ctx, *measurementSet)
-		if err != nil {
-			logger.Error(ctx, "store component measurement group into redis failed", "error", err)
-			panic(err)
-		}
-
-		compAttrSet, err := database.GenAllAttributeMap(tx)
-		if err != nil {
-			logger.Error(ctx, "generate component attribute group failed", "error", err)
-			panic(err)
-		}
-
-		err = model.TraverseAttributeGroupTables(ctx, tx, fullParentPath, isLocalParentPath, compAttrSet)
-		if err != nil {
-			logger.Error(ctx, "store component attribute group into redis failed", "error", err)
-			panic(err)
-		}
-
-		allMeasurement, err := database.GetAllMeasurements(ctx, tx)
-		if err != nil {
-			logger.Error(ctx, "load topologic info from postgres failed", "error", err)
-			panic(err)
-		}
-		go realtimedata.StartComputingRealTimeDataLimit(ctx, allMeasurement)
-
-		tree, _, err := database.QueryTopologicFromDB(ctx, tx)
-		if err != nil {
-			logger.Error(ctx, "load topologic info from postgres failed", "error", err)
-			panic(err)
-		}
-		diagram.GlobalTree = tree
-		return nil
-	})
-
-	// use release mode in production
-	if modelRTConfig.DeployEnv == constants.ProductionDeployMode {
-		gin.SetMode(gin.ReleaseMode)
-	}
-	engine := gin.New()
-	// add CORS middleware
-	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, serviceToken)
-
-	if modelRTConfig.DeployEnv != constants.ProductionDeployMode {
-		engine.GET("/swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler))
-	}
-
-	server := http.Server{
-		Addr:    modelRTConfig.ServiceAddr,
-		Handler: engine,
-	}
-
-	// creating a system signal receiver
-	done := make(chan os.Signal, 10)
-	signal.Notify(done, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
-	go func() {
-		<-done
-		logger.Info(ctx, "shutdown signal received, cleaning up...")
-		if err := server.Shutdown(context.Background()); err != nil {
-			logger.Error(ctx, "shutdown serverError", "err", err)
-		}
-		logger.Info(ctx, "resources cleaned up, exiting")
-	}()
-
-	logger.Info(ctx, "starting ModelRT server")
-	err = server.ListenAndServe()
-	if err != nil {
-		if err == http.ErrServerClosed {
-			// the service receives the shutdown signal normally and then closes
-			logger.Info(ctx, "server closed under request")
-		} else {
-			// abnormal shutdown of service
-			logger.Error(ctx, "server closed unexpected", "err", err)
-		}
-	}
+	// blocks until the server is shut down
+	application.Run(ctx)
 }