modelRT/app/app.go

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