feat: merge refactor/topology-node-optimize into develop

refactor: replace topology tree with graph cache
- replace globalTree startup loading with globalTopologyGraph - add topologyGraph adjacency cache with in/out edges, start nodes, and end nodes - remove multiBranchTreeNode and legacy tree-building helpers - make topology queries independent of the all-zero UUID virtual root - keep TOPOLOGY_ANALYSIS on per-task DB querying while clarifying point-to-point reachability results - add coverage for multi-parent topology reachability
2026-07-08 17:02:15 +08:00 · 2026-07-07 10:01:14 +08:00
16 changed files with 478 additions and 845 deletions
--- a/app/app.go
+++ b/app/app.go
@ -1,135 +0,0 @@
-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
@ -1,217 +0,0 @@
-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
@ -1,39 +0,0 @@
-// 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
@ -1,34 +0,0 @@
-//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
@ -1,80 +0,0 @@
-// 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/database/query_topologic.go
+++ b/database/query_topologic.go
@ -3,11 +3,8 @@ package database

 import (
 	"context"
-	"fmt"
 	"time"

-	"modelRT/constants"
-	"modelRT/diagram"
 	"modelRT/logger"
 	"modelRT/orm"
 	"modelRT/sql"
@ -24,16 +21,19 @@ 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).Clauses(clause.Locking{Strength: "UPDATE"}).Raw(sql.RecursiveSQL, constants.UUIDNilStr).Scan(&topologics)
+	result := tx.WithContext(cancelCtx).
+		Clauses(clause.Locking{Strength: "UPDATE"}).
+		Find(&topologics)
 	if result.Error != nil {
-		logger.Error(ctx, "query circuit diagram topologic info by start node uuid failed", "start_node_uuid", constants.UUIDNilStr, "error", result.Error)
+		logger.Error(ctx, "query circuit diagram topologic info failed", "error", result.Error)
 		return nil, result.Error
 	}
 	return topologics, nil
 }

-// QueryTopologicByStartUUID returns all edges reachable from startUUID following
-// directed uuid_from → uuid_to edges in the topologic table.
+// QueryTopologicByStartUUID returns all directed edges reachable from startUUID.
+// It is used by point-to-point topology reachability checks and intentionally
+// 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.RecursiveSQL, startUUID).
+		Raw(sql.RecursiveTopologicByStartSQL, startUUID).
 		Scan(&topologics)
 	if result.Error != nil {
 		logger.Error(ctx, "query topologic by start uuid failed", "start_uuid", startUUID, "error", result.Error)
@ -50,80 +50,3 @@ 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
-}
--- a/deploy/deploy.md
+++ b/deploy/deploy.md
@ -444,109 +444,6 @@ 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
@ -920,9 +817,9 @@ docker inspect coslight/modelrt:latest
 # 验证二进制可执行（无 config 时程序报错退出属预期行为，说明镜像构建正常）
 docker run --rm coslight/modelrt:latest

-# 挂载本地配置做完整启动验证（Ctrl+C 退出）
+# 挂载示例配置做完整启动验证（Ctrl+C 退出）
 docker run --rm \
-  -v "$(pwd)/configs/config.yaml:/app/configs/config.yaml" \
+  -v "$(pwd)/configs/config.example.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:17.5
+          image: postgres:13.16
          imagePullPolicy: IfNotPresent
          ports:
            - name: postgres
--- a/diagram/multi_branch_tree.go
+++ b/diagram/multi_branch_tree.go
@ -1,125 +0,0 @@
-// 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
-}
--- a/diagram/topology_graph.go
+++ b/diagram/topology_graph.go
@ -0,0 +1,138 @@
+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
+}
--- a/diagram/topology_graph_test.go
+++ b/diagram/topology_graph_test.go
@ -0,0 +1,35 @@
+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)
+	}
+}
--- a/go.mod
+++ b/go.mod
@ -10,7 +10,6 @@ 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,8 +80,6 @@ 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,12 +3,44 @@ package main

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

-	"modelRT/app"
+	"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"

+	"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 (
@ -17,6 +49,13 @@ 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 服务
@ -35,24 +74,229 @@ var (
 func main() {
 	flag.Parse()

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

-	cli := app.CLIArgs{
-		ConfigDir:  *modelRTConfigDir,
-		ConfigName: *modelRTConfigName,
-		ConfigType: *modelRTConfigType,
+		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"))
+		}
 	}

-	application, cleanup, err := app.InitializeApp(ctx, cli)
+	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()
 	if err != nil {
-		log.Fatalf("failed to initialize app: %v", err)
+		logger.Error(ctx, "get host name failed", "error", err)
+		panic(err)
 	}
-	defer cleanup()

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

-	// blocks until the server is shut down
-	application.Run(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)
+
+		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)
+		}
+	}
 }
--- a/sql/topologic.go
+++ b/sql/topologic.go
@ -12,3 +12,18 @@ 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;`
--- a/task/handler_factory.go
+++ b/task/handler_factory.go
@ -98,10 +98,10 @@ func NewTopologyAnalysisHandler() *TopologyAnalysisHandler {
 	}
 }

-// Execute processes a topology analysis task.
+// Execute processes a point-to-point topology reachability task.
 // Params (all sourced from the MQ message, no DB lookup needed):
-//   - start_component_uuid (string, required): BFS origin
-//   - end_component_uuid   (string, required): reachability target
+//   - start_component_uuid (string, required): directed traversal origin
+//   - end_component_uuid   (string, required): directed 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,7 +123,8 @@ 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 topology edges from startComponentUUID, build adjacency list
+	// Phase 2: query only edges reachable from startComponentUUID, then build
+	// 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)
@ -159,7 +160,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)
+				checkInService, false, nil, &startComponentUUID, 0)
 		}
 	}

@ -167,7 +168,8 @@ 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: BFS reachability check
+	// Phase 4: point-to-point BFS reachability check. Multiple parents and
+	// 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}
@ -214,7 +216,7 @@ func (h *TopologyAnalysisHandler) Execute(ctx context.Context, taskID uuid.UUID,
 	}

 	return persistTopologyResult(ctx, db, taskID, startComponentUUID, endComponentUUID,
-		checkInService, isReachable, path, blockedBy)
+		checkInService, isReachable, path, blockedBy, len(visited))
 }

 // parseTopologyAnalysisParams extracts and validates the three required fields.
@ -270,6 +272,7 @@ 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 {
@ -281,11 +284,22 @@ 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 {