core/internal/app/app.go

package app

import (
	"context"
	"errors"
	"gitea.pena/PenaSide/common/privilege"
	"gitea.pena/PenaSide/hlog"
	"gitea.pena/PenaSide/trashlog/wrappers/zaptrashlog"
	"gitea.pena/SQuiz/common/model"
	"gitea.pena/SQuiz/core/internal/brokers"
	"gitea.pena/SQuiz/core/internal/initialize"
	"gitea.pena/SQuiz/core/internal/models"
	server "gitea.pena/SQuiz/core/internal/server/grpc"
	"gitea.pena/SQuiz/core/internal/server/http"
	"gitea.pena/SQuiz/core/internal/tools"
	"gitea.pena/SQuiz/core/internal/workers"
	"gitea.pena/SQuiz/core/pkg/closer"
	"github.com/gofiber/fiber/v2"
	"github.com/gofiber/fiber/v2/log"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"time"
)

type Build struct {
	Commit  string
	Version string
}

var zapOptions = []zap.Option{
	zap.AddCaller(),
	zap.AddCallerSkip(2),
	zap.AddStacktrace(zap.ErrorLevel),
}

func Run(ctx context.Context, cfg initialize.Config, build Build) error {
	var (
		err       error
		zapLogger *zap.Logger
	)

	defer func() {
		if r := recover(); r != nil {
			log.Error("Recovered from a panic", zap.Any("error", r))
		}
	}()

	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	if cfg.LoggerProdMode {
		zapLogger, err = zap.NewProduction(zapOptions...)
		if err != nil {
			return err
		}
	} else {
		zapLogger, err = zap.NewDevelopment(zapOptions...)
		if err != nil {
			return err
		}
	}

	zapLogger = zapLogger.With(
		zap.String("SvcCommit", build.Commit),
		zap.String("SvcVersion", build.Version),
		zap.String("SvcBuildTime", time.Now().String()),
	)

	clickHouseLogger, err := zaptrashlog.NewCore(ctx, zap.InfoLevel, cfg.TrashLogHost, build.Version, build.Commit, time.Now().Unix())
	if err != nil {
		panic(err)
	}

	loggerForHlog := zapLogger.WithOptions(zap.WrapCore(func(core zapcore.Core) zapcore.Core {
		return zapcore.NewTee(core, clickHouseLogger)
	}))

	loggerHlog := hlog.New(loggerForHlog).Module(initialize.ModuleLogger)
	loggerHlog.With(models.AllFields{})
	loggerHlog.Emit(InfoSvcStarted{})

	shutdownGroup := closer.NewCloserGroup()

	dalS, err := initialize.NewDALs(ctx, cfg)
	if err != nil {
		zapLogger.Error("Error initializing dals", zap.Error(err))
		return err
	}

	kafkaClient, err := initialize.KafkaInit(ctx, initialize.KafkaDeps{
		KafkaGroup:   cfg.KafkaGroup,
		KafkaBrokers: cfg.KafkaBrokers,
		KafkaTopic:   cfg.KafkaTopicNotifyer,
	})
	if err != nil {
		zapLogger.Error("Error initializing kafka", zap.Error(err))
		return err
	}

	producer := brokers.NewProducer(brokers.ProducerDeps{
		KafkaClient: kafkaClient,
		Logger:      zapLogger,
	})

	redisClient, err := initialize.Redis(ctx, cfg)
	if err != nil {
		zapLogger.Error("Error initializing redis", zap.Error(err))
		return err
	}

	go tools.PublishPrivilege(privilege.NewPrivilege(privilege.Client{
		URL:         cfg.HubadminMicroserviceURL,
		ServiceName: cfg.ServiceName,
		Privileges:  model.Privileges,
	}, &fiber.Client{}), 10, 5*time.Minute)

	clients, err := initialize.NewClients(ctx, cfg, dalS.PgDAL)
	if err != nil {
		zapLogger.Error("Error initializing clients", zap.Error(err))
		return err
	}

	tgWC := workers.NewTgListenerWC(workers.Deps{
		BotID: int64(6712573453), // todo убрать
		Redis: redisClient,
		Dal:   dalS.PgDAL,
		//TgClient: clients.TgClient,
	})

	go tgWC.Start(ctx)

	controllers := initialize.NewControllers(initialize.ControllerDeps{
		Clients:     clients,
		DALs:        dalS,
		Config:      cfg,
		Producer:    producer,
		RedisClient: redisClient,
	})

	grpc, err := server.NewGRPC(zapLogger)
	if err != nil {
		zapLogger.Error("Error initializing grpc", zap.Error(err))
		return err
	}
	grpc.Register(controllers.GRpcControllers)

	srv := http.NewServer(http.ServerConfig{
		Logger: zapLogger,
		Controllers: []http.Controller{controllers.HttpControllers.Account, controllers.HttpControllers.Telegram, controllers.HttpControllers.Result,
			controllers.HttpControllers.Question, controllers.HttpControllers.Quiz, controllers.HttpControllers.Statistic},
		Hlogger: loggerHlog,
		Dal:     dalS.PgDAL,
	})

	go func() {
		if err := srv.Start(cfg.ClientHttpURL); err != nil {
			zapLogger.Error("HTTP server startup error", zap.Error(err))
			cancel()
		}
	}()

	go grpc.Run(cfg.GrpcURL)

	srv.ListRoutes()

	shutdownGroup.Add(closer.CloserFunc(srv.Shutdown))
	shutdownGroup.Add(closer.CloserFunc(grpc.Stop))
	shutdownGroup.Add(closer.CloserFunc(dalS.PgDAL.Close))
	shutdownGroup.Add(closer.CloserFunc(dalS.ChDAL.Close))

	<-ctx.Done()

	timeoutCtx, timeoutCancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer timeoutCancel()
	if err := shutdownGroup.Call(timeoutCtx); err != nil {
		if errors.Is(err, context.DeadlineExceeded) {
			zapLogger.Error("Shutdown timed out", zap.Error(err))
		} else {
			zapLogger.Error("Failed to shutdown services gracefully", zap.Error(err))
		}
		return err
	}

	zapLogger.Info("Application has stopped")
	return nil
}