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

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

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,
+	}
+}

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

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

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

database/query_topologic.go

@@ -3,8 +3,11 @@ package database
 
 import (
 	"context"
+	"fmt"
 	"time"
 
+	"modelRT/constants"
+	"modelRT/diagram"
 	"modelRT/logger"
 	"modelRT/orm"
 	"modelRT/sql"
@@ -21,19 +24,16 @@ func QueryTopologic(ctx context.Context, tx *gorm.DB) ([]orm.Topologic, error) {
 	cancelCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
 	defer cancel()
 
-	result := tx.WithContext(cancelCtx).
+	result := tx.WithContext(cancelCtx).Clauses(clause.Locking{Strength: "UPDATE"}).Raw(sql.RecursiveSQL, constants.UUIDNilStr).Scan(&topologics)
-		Clauses(clause.Locking{Strength: "UPDATE"}).
-		Find(&topologics)
 	if result.Error != nil {
-		logger.Error(ctx, "query circuit diagram topologic info failed", "error", result.Error)
+		logger.Error(ctx, "query circuit diagram topologic info by start node uuid failed", "start_node_uuid", constants.UUIDNilStr, "error", result.Error)
 		return nil, result.Error
 	}
 	return topologics, nil
 }
 
-// QueryTopologicByStartUUID returns all directed edges reachable from startUUID.
+// QueryTopologicByStartUUID returns all edges reachable from startUUID following
-// It is used by point-to-point topology reachability checks and intentionally
+// directed uuid_from → uuid_to edges in the topologic table.
-// does not depend on the legacy all-zero UUID virtual root.
 func QueryTopologicByStartUUID(ctx context.Context, tx *gorm.DB, startUUID uuid.UUID) ([]orm.Topologic, error) {
 	var topologics []orm.Topologic
 
@@ -42,7 +42,7 @@ func QueryTopologicByStartUUID(ctx context.Context, tx *gorm.DB, startUUID uuid.
 
 	result := tx.WithContext(cancelCtx).
 		Clauses(clause.Locking{Strength: "UPDATE"}).
-		Raw(sql.RecursiveTopologicByStartSQL, startUUID).
+		Raw(sql.RecursiveSQL, startUUID).
 		Scan(&topologics)
 	if result.Error != nil {
 		logger.Error(ctx, "query topologic by start uuid failed", "start_uuid", startUUID, "error", result.Error)
@@ -50,3 +50,80 @@ func QueryTopologicByStartUUID(ctx context.Context, tx *gorm.DB, startUUID uuid.
 	}
 	return topologics, nil
 }
+
+// QueryTopologicFromDB return the result of query topologic info from DB.
+// Returns the root node and a flat nodeMap for O(1) lookup by UUID.
+func QueryTopologicFromDB(ctx context.Context, tx *gorm.DB) (*diagram.MultiBranchTreeNode, map[uuid.UUID]*diagram.MultiBranchTreeNode, error) {
+	topologicInfos, err := QueryTopologic(ctx, tx)
+	if err != nil {
+		logger.Error(ctx, "query topologic info failed", "error", err)
+		return nil, nil, err
+	}
+
+	tree, nodeMap, err := BuildMultiBranchTree(topologicInfos)
+	if err != nil {
+		logger.Error(ctx, "init topologic failed", "error", err)
+		return nil, nil, err
+	}
+	return tree, nodeMap, nil
+}
+
+// BuildMultiBranchTree return the multi branch tree by topologic info.
+// Returns the root node and a flat nodeMap for O(1) lookup by UUID.
+func BuildMultiBranchTree(topologics []orm.Topologic) (*diagram.MultiBranchTreeNode, map[uuid.UUID]*diagram.MultiBranchTreeNode, error) {
+	nodeMap := make(map[uuid.UUID]*diagram.MultiBranchTreeNode, len(topologics)*2)
+
+	for _, topo := range topologics {
+		if _, exists := nodeMap[topo.UUIDFrom]; !exists {
+			// UUIDNil is the virtual root sentinel — skip creating a regular node for it
+			if topo.UUIDFrom != constants.UUIDNil {
+				nodeMap[topo.UUIDFrom] = &diagram.MultiBranchTreeNode{
+					ID:       topo.UUIDFrom,
+					Children: make([]*diagram.MultiBranchTreeNode, 0),
+				}
+			}
+		}
+
+		if _, exists := nodeMap[topo.UUIDTo]; !exists {
+			if topo.UUIDTo != constants.UUIDNil {
+				nodeMap[topo.UUIDTo] = &diagram.MultiBranchTreeNode{
+					ID:       topo.UUIDTo,
+					Children: make([]*diagram.MultiBranchTreeNode, 0),
+				}
+			}
+		}
+	}
+
+	for _, topo := range topologics {
+		var parent *diagram.MultiBranchTreeNode
+		if topo.UUIDFrom == constants.UUIDNil {
+			if _, exists := nodeMap[constants.UUIDNil]; !exists {
+				nodeMap[constants.UUIDNil] = &diagram.MultiBranchTreeNode{
+					ID:       constants.UUIDNil,
+					Children: make([]*diagram.MultiBranchTreeNode, 0),
+				}
+			}
+			parent = nodeMap[constants.UUIDNil]
+		} else {
+			parent = nodeMap[topo.UUIDFrom]
+		}
+
+		var child *diagram.MultiBranchTreeNode
+		if topo.UUIDTo == constants.UUIDNil {
+			child = &diagram.MultiBranchTreeNode{
+				ID: topo.UUIDTo,
+			}
+		} else {
+			child = nodeMap[topo.UUIDTo]
+		}
+		child.Parent = parent
+		parent.Children = append(parent.Children, child)
+	}
+
+	// return root vertex
+	root, exists := nodeMap[constants.UUIDNil]
+	if !exists {
+		return nil, nil, fmt.Errorf("root node not found")
+	}
+	return root, nodeMap, nil
+}

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
 ```

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

diagram/multi_branch_tree.go

@@ -0,0 +1,125 @@
+// Package diagram provide diagram data structure and operation
+package diagram
+
+import (
+	"fmt"
+
+	"github.com/gofrs/uuid"
+)
+
+var GlobalTree *MultiBranchTreeNode
+
+// MultiBranchTreeNode represents a topological structure using an multi branch tree
+type MultiBranchTreeNode struct {
+	ID       uuid.UUID              // 节点唯一标识
+	Parent   *MultiBranchTreeNode   // 指向父节点的指针
+	Children []*MultiBranchTreeNode // 指向所有子节点的指针切片
+}
+
+func NewMultiBranchTree(id uuid.UUID) *MultiBranchTreeNode {
+	return &MultiBranchTreeNode{
+		ID:       id,
+		Children: make([]*MultiBranchTreeNode, 0),
+	}
+}
+
+func (n *MultiBranchTreeNode) AddChild(child *MultiBranchTreeNode) {
+	child.Parent = n
+	n.Children = append(n.Children, child)
+}
+
+func (n *MultiBranchTreeNode) RemoveChild(childID uuid.UUID) bool {
+	for i, child := range n.Children {
+		if child.ID == childID {
+			n.Children = append(n.Children[:i], n.Children[i+1:]...)
+			child.Parent = nil
+			return true
+		}
+	}
+	return false
+}
+
+func (n *MultiBranchTreeNode) FindNodeByID(id uuid.UUID) *MultiBranchTreeNode {
+	if n.ID == id {
+		return n
+	}
+
+	for _, child := range n.Children {
+		if found := child.FindNodeByID(id); found != nil {
+			return found
+		}
+	}
+	return nil
+}
+
+func (n *MultiBranchTreeNode) PrintTree(level int) {
+	for range level {
+		fmt.Print("  ")
+	}
+
+	fmt.Printf("-ID: %s\n", n.ID)
+
+	for _, child := range n.Children {
+		child.PrintTree(level + 1)
+	}
+}
+
+// FindPath returns the ordered node sequence from startID to endID using the
+// supplied nodeMap for O(1) lookup. It walks each node up to the root to find
+// the LCA, then stitches the two half-paths together.
+// Returns nil when either node is absent from nodeMap or no path exists.
+func FindPath(startID, endID uuid.UUID, nodeMap map[uuid.UUID]*MultiBranchTreeNode) []*MultiBranchTreeNode {
+	startNode, ok := nodeMap[startID]
+	if !ok {
+		return nil
+	}
+	endNode, ok := nodeMap[endID]
+	if !ok {
+		return nil
+	}
+
+	// collect ancestors (inclusive) from a node up to the root sentinel
+	ancestors := func(n *MultiBranchTreeNode) []*MultiBranchTreeNode {
+		var chain []*MultiBranchTreeNode
+		for n != nil {
+			chain = append(chain, n)
+			n = n.Parent
+		}
+		return chain
+	}
+
+	startChain := ancestors(startNode) // [start, ..., root]
+	endChain := ancestors(endNode)     // [end,   ..., root]
+
+	// index startChain by ID for fast LCA detection
+	startIdx := make(map[uuid.UUID]int, len(startChain))
+	for i, node := range startChain {
+		startIdx[node.ID] = i
+	}
+
+	// find LCA: first node in endChain that also appears in startChain
+	lcaEndPos := -1
+	lcaStartPos := -1
+	for i, node := range endChain {
+		if j, found := startIdx[node.ID]; found {
+			lcaEndPos = i
+			lcaStartPos = j
+			break
+		}
+	}
+
+	if lcaEndPos < 0 {
+		return nil // disconnected
+	}
+
+	// path = startChain[0..lcaStartPos] reversed + endChain[lcaEndPos..0] reversed
+	path := make([]*MultiBranchTreeNode, 0, lcaStartPos+lcaEndPos+1)
+	for i := 0; i <= lcaStartPos; i++ {
+		path = append(path, startChain[i])
+	}
+	// append end-side (skip LCA to avoid duplication), reversed
+	for i := lcaEndPos - 1; i >= 0; i-- {
+		path = append(path, endChain[i])
+	}
+	return path
+}

diagram/topology_graph.go

@@ -1,138 +0,0 @@
-package diagram
-
-import (
-	"sync"
-
-	"modelRT/orm"
-
-	"github.com/gofrs/uuid"
-)
-
-// TopologyGraph represents directed topologic links with adjacency lists.
-// It preserves multiple parents for one node.
-type TopologyGraph struct {
-	Nodes      map[uuid.UUID]struct{}
-	OutEdges   map[uuid.UUID][]uuid.UUID
-	InEdges    map[uuid.UUID][]uuid.UUID
-	StartNodes []uuid.UUID
-	EndNodes   []uuid.UUID
-}
-
-var (
-	globalTopologyGraphMu sync.RWMutex
-	GlobalTopologyGraph   *TopologyGraph
-)
-
-// NewTopologyGraph builds a directed graph cache from topologic edges.
-func NewTopologyGraph(edges []orm.Topologic) *TopologyGraph {
-	graph := &TopologyGraph{
-		Nodes:    make(map[uuid.UUID]struct{}, len(edges)*2),
-		OutEdges: make(map[uuid.UUID][]uuid.UUID, len(edges)),
-		InEdges:  make(map[uuid.UUID][]uuid.UUID, len(edges)),
-	}
-
-	for _, edge := range edges {
-		from := edge.UUIDFrom
-		to := edge.UUIDTo
-
-		graph.Nodes[from] = struct{}{}
-		graph.Nodes[to] = struct{}{}
-		graph.OutEdges[from] = append(graph.OutEdges[from], to)
-		graph.InEdges[to] = append(graph.InEdges[to], from)
-	}
-
-	graph.StartNodes = graph.findStartNodes()
-	graph.EndNodes = graph.findEndNodes()
-	return graph
-}
-
-// SetGlobalTopologyGraph replaces the process-wide topology graph cache.
-func SetGlobalTopologyGraph(graph *TopologyGraph) {
-	globalTopologyGraphMu.Lock()
-	defer globalTopologyGraphMu.Unlock()
-	GlobalTopologyGraph = graph
-}
-
-// GetGlobalTopologyGraph returns the process-wide topology graph cache.
-func GetGlobalTopologyGraph() *TopologyGraph {
-	globalTopologyGraphMu.RLock()
-	defer globalTopologyGraphMu.RUnlock()
-	return GlobalTopologyGraph
-}
-
-func (g *TopologyGraph) findStartNodes() []uuid.UUID {
-	startNodes := make([]uuid.UUID, 0)
-	for id := range g.Nodes {
-		if len(g.InEdges[id]) == 0 && len(g.OutEdges[id]) > 0 {
-			startNodes = append(startNodes, id)
-		}
-	}
-	return startNodes
-}
-
-func (g *TopologyGraph) findEndNodes() []uuid.UUID {
-	endNodes := make([]uuid.UUID, 0)
-	for id := range g.Nodes {
-		if len(g.InEdges[id]) > 0 && len(g.OutEdges[id]) == 0 {
-			endNodes = append(endNodes, id)
-		}
-	}
-	return endNodes
-}
-
-// IsReachable reports whether end can be reached from start following directed
-// uuid_from -> uuid_to edges.
-func (g *TopologyGraph) IsReachable(start, end uuid.UUID) bool {
-	return len(g.FindPath(start, end)) > 0
-}
-
-// FindPath returns one shortest directed path from start to end, or nil when
-// no directed path exists.
-func (g *TopologyGraph) FindPath(start, end uuid.UUID) []uuid.UUID {
-	if g == nil {
-		return nil
-	}
-	if start == end {
-		if _, exists := g.Nodes[start]; exists {
-			return []uuid.UUID{start}
-		}
-		return nil
-	}
-
-	visited := map[uuid.UUID]struct{}{start: {}}
-	parent := make(map[uuid.UUID]uuid.UUID)
-	queue := []uuid.UUID{start}
-
-	for len(queue) > 0 {
-		cur := queue[0]
-		queue = queue[1:]
-
-		for _, next := range g.OutEdges[cur] {
-			if _, seen := visited[next]; seen {
-				continue
-			}
-			visited[next] = struct{}{}
-			parent[next] = cur
-
-			if next == end {
-				return reconstructTopologyGraphPath(parent, start, end)
-			}
-
-			queue = append(queue, next)
-		}
-	}
-
-	return nil
-}
-
-func reconstructTopologyGraphPath(parent map[uuid.UUID]uuid.UUID, start, end uuid.UUID) []uuid.UUID {
-	path := make([]uuid.UUID, 0)
-	for cur := end; cur != start; cur = parent[cur] {
-		path = append(path, cur)
-	}
-	path = append(path, start)
-	for i, j := 0, len(path)-1; i < j; i, j = i+1, j-1 {
-		path[i], path[j] = path[j], path[i]
-	}
-	return path
-}

diagram/topology_graph_test.go

@@ -1,35 +0,0 @@
-package diagram
-
-import (
-	"testing"
-
-	"modelRT/orm"
-
-	"github.com/gofrs/uuid"
-)
-
-func TestTopologyGraphSupportsMultiParentReachability(t *testing.T) {
-	startA := uuid.Must(uuid.NewV4())
-	startB := uuid.Must(uuid.NewV4())
-	shared := uuid.Must(uuid.NewV4())
-	end := uuid.Must(uuid.NewV4())
-
-	graph := NewTopologyGraph([]orm.Topologic{
-		{UUIDFrom: startA, UUIDTo: shared},
-		{UUIDFrom: startB, UUIDTo: shared},
-		{UUIDFrom: shared, UUIDTo: end},
-	})
-
-	if len(graph.StartNodes) != 2 {
-		t.Fatalf("expected 2 start nodes, got %d", len(graph.StartNodes))
-	}
-	if len(graph.InEdges[shared]) != 2 {
-		t.Fatalf("expected shared node to keep 2 parents, got %d", len(graph.InEdges[shared]))
-	}
-	if !graph.IsReachable(startA, end) {
-		t.Fatalf("expected %s to reach %s", startA, end)
-	}
-	if !graph.IsReachable(startB, end) {
-		t.Fatalf("expected %s to reach %s", startB, end)
-	}
-}

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

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=

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/app"
-	"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"
 
-	"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)
+	ctx := context.Background()
-	if _, err := os.Stat(configPath); os.IsNotExist(err) {
-		log.Println("configuration file not found,checking for example file")
 
-		exampleConfigPath := filepath.Join(*modelRTConfigDir, *modelRTConfigName+".example."+*modelRTConfigType)
+	cli := app.CLIArgs{
-		configDir := filepath.Dir(configPath)
+		ConfigDir:  *modelRTConfigDir,
-		if err := os.MkdirAll(configDir, 0o755); err != nil {
+		ConfigName: *modelRTConfigName,
-			panic(fmt.Errorf("failed to create config directory %s:%w", configDir, err))
+		ConfigType: *modelRTConfigType,
-		}
-		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"))
-		}
 	}
 
-	modelRTConfig = config.ReadAndInitConfig(*modelRTConfigDir, *modelRTConfigName, *modelRTConfigType)
+	application, cleanup, err := app.InitializeApp(ctx, cli)
-
-	// 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()
 	if err != nil {
-		logger.Error(ctx, "get host name failed", "error", err)
+		log.Fatalf("failed to initialize app: %v", err)
-		panic(err)
 	}
+	defer cleanup()
 
-	serviceToken, err := util.GenerateClientToken(hostName, modelRTConfig.ServiceName, modelRTConfig.SecretKey)
+	// warm caches from postgres, then start background publishers/worker
-	if err != nil {
+	application.Warmup(ctx)
-		logger.Error(ctx, "generate client token failed", "error", err)
+	application.StartBackground(ctx)
-		panic(err)
-	}
 
-	// init postgresDBClient
+	// blocks until the server is shut down
-	postgresDBClient = database.InitPostgresDBInstance(ctx, modelRTConfig.PostgresDBURI)
+	application.Run(ctx)
-
-	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)
-
-		topologics, err := database.QueryTopologic(ctx, tx)
-		if err != nil {
-			logger.Error(ctx, "load topologic info from postgres failed", "error", err)
-			panic(err)
-		}
-		diagram.SetGlobalTopologyGraph(diagram.NewTopologyGraph(topologics))
-		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)
-		}
-	}
 }

sql/topologic.go

@@ -12,18 +12,3 @@ var RecursiveSQL = `WITH RECURSIVE recursive_tree as (
 	JOIN recursive_tree rt ON t.uuid_from = rt.uuid_to
 	)
 	SELECT * FROM recursive_tree;`
-
-// RecursiveTopologicByStartSQL returns every directed edge reachable from the
-// supplied start component. It tracks the visited node path inside PostgreSQL
-// so cycles in topologic data cannot recurse forever.
-var RecursiveTopologicByStartSQL = `WITH RECURSIVE recursive_tree as (
-	SELECT uuid_from, uuid_to, flag, ARRAY[uuid_from, uuid_to] AS path
-	FROM "topologic"
-	WHERE uuid_from = ?
-	UNION ALL
-	SELECT t.uuid_from, t.uuid_to, t.flag, rt.path || t.uuid_to
-	FROM "topologic" t
-	JOIN recursive_tree rt ON t.uuid_from = rt.uuid_to
-	WHERE NOT t.uuid_to = ANY(rt.path)
-)
-SELECT uuid_from, uuid_to, flag FROM recursive_tree;`

task/handler_factory.go

@@ -98,10 +98,10 @@ func NewTopologyAnalysisHandler() *TopologyAnalysisHandler {
 	}
 }
 
-// Execute processes a point-to-point topology reachability task.
+// Execute processes a topology analysis task.
 // Params (all sourced from the MQ message, no DB lookup needed):
-//   - start_component_uuid (string, required): directed traversal origin
+//   - start_component_uuid (string, required): BFS origin
-//   - end_component_uuid   (string, required): directed reachability target
+//   - end_component_uuid   (string, required): reachability target
 //   - check_in_service     (bool,   optional, default true): skip out-of-service components
 func (h *TopologyAnalysisHandler) Execute(ctx context.Context, taskID uuid.UUID, params map[string]any, db *gorm.DB) error {
 	logger.Info(ctx, "topology analysis started", "task_id", taskID)
@@ -123,8 +123,7 @@ func (h *TopologyAnalysisHandler) Execute(ctx context.Context, taskID uuid.UUID,
 		logger.Warn(ctx, "update progress failed", "task_id", taskID, "progress", 20, "error", err)
 	}
 
-	// Phase 2: query only edges reachable from startComponentUUID, then build
+	// Phase 2: query topology edges from startComponentUUID, build adjacency list
-	// the adjacency list used for point-to-point directed reachability.
 	topoEdges, err := database.QueryTopologicByStartUUID(ctx, db, startComponentUUID)
 	if err != nil {
 		return fmt.Errorf("query topology from start node: %w", err)
@@ -160,7 +159,7 @@ func (h *TopologyAnalysisHandler) Execute(ctx context.Context, taskID uuid.UUID,
 		// check the start node itself before BFS
 		if !inServiceMap[startComponentUUID] {
 			return persistTopologyResult(ctx, db, taskID, startComponentUUID, endComponentUUID,
-				checkInService, false, nil, &startComponentUUID, 0)
+				checkInService, false, nil, &startComponentUUID)
 		}
 	}
 
@@ -168,8 +167,7 @@ func (h *TopologyAnalysisHandler) Execute(ctx context.Context, taskID uuid.UUID,
 		logger.Warn(ctx, "update progress failed", "task_id", taskID, "progress", 60, "error", err)
 	}
 
-	// Phase 4: point-to-point BFS reachability check. Multiple parents and
+	// Phase 4: BFS reachability check
-	// multiple paths to a node are valid; visited only prevents cycles/rework.
 	visited := make(map[uuid.UUID]struct{})
 	parent := make(map[uuid.UUID]uuid.UUID) // for path reconstruction
 	queue := []uuid.UUID{startComponentUUID}
@@ -216,7 +214,7 @@ func (h *TopologyAnalysisHandler) Execute(ctx context.Context, taskID uuid.UUID,
 	}
 
 	return persistTopologyResult(ctx, db, taskID, startComponentUUID, endComponentUUID,
-		checkInService, isReachable, path, blockedBy, len(visited))
+		checkInService, isReachable, path, blockedBy)
 }
 
 // parseTopologyAnalysisParams extracts and validates the three required fields.
@@ -272,7 +270,6 @@ func persistTopologyResult(
 	ctx context.Context, db *gorm.DB, taskID uuid.UUID,
 	startID, endID uuid.UUID, checkInService, isReachable bool,
 	path []uuid.UUID, blockedBy *uuid.UUID,
-	visitedCount int,
 ) error {
 	pathStrs := make([]string, 0, len(path))
 	for _, id := range path {
@@ -284,22 +281,11 @@ func persistTopologyResult(
 		"end_component_uuid":   endID.String(),
 		"check_in_service":     checkInService,
 		"is_reachable":         isReachable,
-		"analysis_type":        "POINT_TO_POINT_REACHABILITY",
 		"path":                 pathStrs,
-		"path_node_count":      len(pathStrs),
-		"visited_count":        visitedCount,
 		"computed_at":          time.Now().Unix(),
 	}
-	if isReachable {
-		result["hop_count"] = len(pathStrs) - 1
-	}
 	if blockedBy != nil {
 		result["blocked_by"] = blockedBy.String()
-		result["reason"] = "OUT_OF_SERVICE_COMPONENT"
-	} else if isReachable {
-		result["reason"] = "REACHABLE"
-	} else {
-		result["reason"] = "NO_DIRECTED_PATH"
 	}
 
 	if err := database.CreateAsyncTaskResult(ctx, db, taskID, result); err != nil {