worker/app/app.go

package app

import (
	"context"
	"errors"
	"fmt"
	"github.com/go-redis/redis/v8"
	"github.com/minio/minio-go/v7"
	"github.com/minio/minio-go/v7/pkg/credentials"
	"github.com/skeris/appInit"
	"github.com/themakers/hlog"
	"go.uber.org/zap"
	"penahub.gitlab.yandexcloud.net/backend/quiz/common.git/clients"
	"penahub.gitlab.yandexcloud.net/backend/quiz/common.git/dal"
	"penahub.gitlab.yandexcloud.net/backend/quiz/worker/answerwc"
	"penahub.gitlab.yandexcloud.net/backend/quiz/worker/privilegewc"
	"penahub.gitlab.yandexcloud.net/backend/quiz/worker/senders"
	"penahub.gitlab.yandexcloud.net/backend/quiz/worker/workers/shortstat"
	"penahub.gitlab.yandexcloud.net/backend/quiz/worker/workers/timeout"
	"penahub.gitlab.yandexcloud.net/pena-services/customer/pkg/customer_clients"
	"time"
)

type App struct {
	logger *zap.Logger
	err    chan error
}

func (a App) GetLogger() *zap.Logger {
	return a.logger
}

func (a App) GetErr() chan error {
	return a.err
}

var (
	errInvalidOptions = errors.New("invalid options")
)

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

var _ appInit.CommonApp = (*App)(nil)

type Options struct {
	ServiceName            string `env:"SERVICE_NAME" default:"squiz"`
	KafkaBroker            string `env:"KAFKA_BROKER" default:"localhost:6379"`
	KafkaTopic             string `env:"KAFKA_TOPIC" default:"test-topic"`
	LoggerProdMode         bool   `env:"IS_PROD_LOG" default:"false"`
	IsProd                 bool   `env:"IS_PROD" default:"false"`
	MinioEP                string `env:"MINIO_EP" default:"localhost:3002"`
	MinioAK                string `env:"MINIO_AK" default:"minio"`
	MinioSK                string `env:"MINIO_SK" default:"miniostorage"`
	PostgresCredentials    string `env:"PG_CRED" default:"host=localhost port=35432 user=squiz password=Redalert2 dbname=squiz sslmode=disable"`
	RedisHost              string `env:"REDIS_HOST" default:"localhost:6379"`
	RedisPassword          string `env:"REDIS_PASSWORD" default:"admin"`
	RedisDB                uint64 `env:"REDIS_DB" default:"2"`
	SmtpHost               string `env:"SMTP_HOST" default:"connect.mailclient.bz"`
	SmtpPort               string `env:"SMTP_PORT" default:"587"`
	SmtpSender             string `env:"SMTP_SENDER" default:"noreply@mailing.pena.digital"`
	SmtpUsername           string `env:"SMTP_USERNAME" default:"kotilion.95@gmail.com"`
	SmtpPassword           string `env:"SMTP_PASSWORD" default:"vWwbCSg4bf0p"`
	SmtpApiKey             string `env:"SMTP_API_KEY" default:"P0YsjUB137upXrr1NiJefHmXVKW1hmBWlpev"`
	SmtpApiUrl             string `env:"SMTP_API_URL" default:"https://api.smtp.bz/v1/smtp/send"`
	CustomerServiceAddress string `env:"CUSTOMER_SERVICE_ADDRESS" default:"localhost:9001"`
	TgToken                string `env:"TG_TOKEN"`
}

func New(ctx context.Context, opts interface{}, ver appInit.Version) (appInit.CommonApp, error) {
	var (
		err, workerErr error
		zapLogger      *zap.Logger
		errChan        = make(chan error)
		options        Options
		ok             bool
	)

	if options, ok = opts.(Options); !ok {
		return App{}, errInvalidOptions
	}

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

	zapLogger = zapLogger.With(
		zap.String("SvcCommit", ver.Commit),
		zap.String("SvcVersion", ver.Release),
		zap.String("SvcBuildTime", ver.BuildTime),
	)

	logger := hlog.New(zapLogger)
	logger.Emit(InfoSvcStarted{})
	zapLogger.Info("config", zap.Any("options", options))

	go func() {
		for {
			select {
			case <-ctx.Done():
				return
			case err := <-errChan:
				zapLogger.Error("Ошибка при работе воркера", zap.Error(err))
			}
		}
	}()

	//init redis
	redisClient := redis.NewClient(&redis.Options{
		Addr:     options.RedisHost,
		Password: options.RedisPassword,
		DB:       int(options.RedisDB),
	})

	mailClent := clients.NewSmtpClient(clients.Deps{
		SmtpHost:   options.SmtpHost,
		SmtpPort:   options.SmtpPort,
		SmtpSender: options.SmtpSender,
		ApiKey:     options.SmtpApiKey,
		SmtpApiUrl: options.SmtpApiUrl,
	})

	// tgSender, err := senders.NewTgSender(options.TgToken)
	// if err != nil {
	// 	fmt.Println(err)
	// 	return nil, err
	// }
	mailSender := senders.NewMailLeadSender(mailClent)
	leadSenders := []senders.LeadSender{mailSender/* , tgSender */}

	customerClient := customer_clients.NewCustomersClient(customer_clients.CustomersClientDeps{
		Logger:              zapLogger,
		CustomerServiceHost: options.CustomerServiceAddress,
	})

	minioClient, err := minio.New(options.MinioEP, &minio.Options{
		Creds:  credentials.NewStaticV4(options.MinioAK, options.MinioSK, ""),
		Secure: options.IsProd,
	})
	if err != nil {
		fmt.Println("MINIOERR", options.MinioEP, err)
		return nil, err
	}

	pgdal, err := dal.New(ctx, options.PostgresCredentials, minioClient)
	if err != nil {
		return nil, err
	}

	kafkaWorker, err := privilegewc.NewKafkaConsumerWorker(privilegewc.Config{
		KafkaBroker:    options.KafkaBroker,
		KafkaTopic:     options.KafkaTopic,
		ServiceKey:     options.ServiceName,
		TickerInterval: time.Second * 10,
		ErrChan:        errChan,
	}, redisClient, pgdal)
	if err != nil {
		logger.Module("Failed start privilege worker")
		return nil, err
	}

	checkWorker := privilegewc.NewCheckWorker(privilegewc.Deps{
		PrivilegeIDsDays:  []string{"quizUnlimTime", "squizHideBadge"},
		PrivilegeIDsCount: []string{"quizCnt", "quizManual"},
		TickerInterval:    time.Minute,
		PrivilegeDAL:      pgdal,
		CustomerClient:    customerClient,
	}, errChan)

	go kafkaWorker.Start(ctx)
	go checkWorker.Start(ctx)

	toClientWorker := answerwc.NewSendToClient(answerwc.DepsSendToClient{
		Redis:           redisClient,
		Dal:             pgdal,
		LeadSenders:     leadSenders,
		CustomerService: customerClient,
	}, errChan)

	toRespWorker := answerwc.NewRespWorker(answerwc.DepsRespWorker{
		Redis:      redisClient,
		Dal:        pgdal,
		MailClient: mailSender,
	}, errChan)

	go toClientWorker.Start(ctx)
	go toRespWorker.Start(ctx)

	tow := timeout.New(pgdal, time.Minute)
	statW := shortstat.New(pgdal, 5*time.Minute)
	tow.ExposeErr(ctx, &workerErr)
	statW.ExposeErr(ctx, &workerErr)
	go tow.Start(ctx)
	go func() {
		//		defer pgdal.CloseWorker()
		statW.Start(ctx)
	}()

	logger.Emit(InfoSvcReady{})
	// todo implement helper func for service app type. such as server preparing, logger preparing, healthchecks and etc.
	return &App{
		logger: zapLogger,
		err:    make(chan error),
	}, err
}