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