fix after merge, now gigachat worker is init and start

This commit is contained in:
pasha1coil 2025-05-16 11:33:04 +03:00
parent 1540deef5d
commit 619a874713
10 changed files with 49 additions and 11 deletions

@ -2,16 +2,20 @@ package app
import (
"context"
"crypto/tls"
"errors"
"gitea.pena/PenaSide/hlog"
"gitea.pena/SQuiz/common/dal"
answerwc2 "gitea.pena/SQuiz/worker/internal/answerwc"
"gitea.pena/SQuiz/worker/internal/answerwc"
"gitea.pena/SQuiz/worker/internal/clients/gigachat"
"gitea.pena/SQuiz/worker/internal/gigachatwc"
"gitea.pena/SQuiz/worker/internal/initialize"
privilegewc2 "gitea.pena/SQuiz/worker/internal/privilegewc"
senders2 "gitea.pena/SQuiz/worker/internal/senders"
"gitea.pena/SQuiz/worker/internal/privilegewc"
"gitea.pena/SQuiz/worker/internal/senders"
"gitea.pena/SQuiz/worker/internal/workers/shortstat"
"gitea.pena/SQuiz/worker/internal/workers/timeout"
"gitea.pena/SQuiz/worker/pkg/closer"
"github.com/go-resty/resty/v2"
"go.uber.org/zap"
"time"
)
@ -87,8 +91,8 @@ func New(ctx context.Context, cfg initialize.Config, build Build) error {
// fmt.Println(err)
// return nil, err
// }
mailSender := senders2.NewMailLeadSender(clients.MailClient)
leadSenders := []senders2.LeadSender{mailSender /* , tgSender */}
mailSender := senders.NewMailLeadSender(clients.MailClient)
leadSenders := []senders.LeadSender{mailSender /* , tgSender */}
pgdal, err := dal.New(ctx, cfg.PostgresURL, minioClient)
if err != nil {
@ -96,7 +100,7 @@ func New(ctx context.Context, cfg initialize.Config, build Build) error {
return err
}
kafkaWorker, err := privilegewc2.NewKafkaConsumerWorker(privilegewc2.Config{
kafkaWorker, err := privilegewc.NewKafkaConsumerWorker(privilegewc.Config{
KafkaBroker: cfg.KafkaBrokers,
KafkaTopic: cfg.KafkaTopicTariff,
ServiceKey: cfg.ServiceName,
@ -108,7 +112,7 @@ func New(ctx context.Context, cfg initialize.Config, build Build) error {
return err
}
checkWorker := privilegewc2.NewCheckWorker(privilegewc2.Deps{
checkWorker := privilegewc.NewCheckWorker(privilegewc.Deps{
PrivilegeIDsDays: []string{"quizUnlimTime", "squizHideBadge"},
PrivilegeIDsCount: []string{"quizCnt", "quizManual"},
TickerInterval: time.Minute,
@ -119,14 +123,14 @@ func New(ctx context.Context, cfg initialize.Config, build Build) error {
go kafkaWorker.Start(ctx)
go checkWorker.Start(ctx)
toClientWorker := answerwc2.NewSendToClient(answerwc2.DepsSendToClient{
toClientWorker := answerwc.NewSendToClient(answerwc.DepsSendToClient{
Redis: redisClient,
Dal: pgdal,
LeadSenders: leadSenders,
CustomerService: clients.CustomerClient,
}, errChan)
toRespWorker := answerwc2.NewRespWorker(answerwc2.DepsRespWorker{
toRespWorker := answerwc.NewRespWorker(answerwc.DepsRespWorker{
Redis: redisClient,
Dal: pgdal,
MailClient: mailSender,
@ -135,6 +139,35 @@ func New(ctx context.Context, cfg initialize.Config, build Build) error {
go toClientWorker.Start(ctx)
go toRespWorker.Start(ctx)
gigaChatClient, err := gigachat.NewGigaChatClient(ctx, gigachat.Deps{
Logger: zapLogger,
Client: resty.New().SetTLSClientConfig(&tls.Config{InsecureSkipVerify: true}),
BaseURL: cfg.GigaChatApiBaseURL,
AuthKey: cfg.GigaChatApiAuthKey,
RedisClient: redisClient,
})
if err != nil {
zapLogger.Error("failed init giga chat client", zap.Error(err))
return err
}
// метод для обновления токенов гигачата
go gigaChatClient.TokenResearch(ctx)
gigaChatWorker, err := gigachatwc.NewGigaChatTaskScheduler(gigachatwc.Deps{
KafkaBrokers: cfg.KafkaBrokers,
KafkaTopic: cfg.KafkaTopicGigaChat,
KafkaGroup: cfg.KafkaGroupGigaChat,
GigaChatClient: gigaChatClient,
Logger: zapLogger,
Dal: pgdal,
})
if err != nil {
zapLogger.Error("failed init giga chat task worker", zap.Error(err))
return err
}
go gigaChatWorker.Start(ctx)
tow := timeout.New(pgdal, time.Minute)
statW := shortstat.New(pgdal, 5*time.Minute)
tow.ExposeErr(ctx, &workerErr)

@ -6,7 +6,7 @@ import (
"errors"
"gitea.pena/SQuiz/common/dal"
"gitea.pena/SQuiz/common/model"
"gitea.pena/SQuiz/worker/clients/gigachat"
"gitea.pena/SQuiz/worker/internal/clients/gigachat"
"github.com/twmb/franz-go/pkg/kgo"
"go.uber.org/zap"
"time"

@ -24,6 +24,11 @@ type Config struct {
ApiUrl string `env:"API_URL" envDefault:"https://api.smtp.bz/v1/smtp/send"`
CustomerMicroserviceRPCURL string `env:"CUSTOMER_MICROSERVICE_RPC_URL" envDefault:"localhost:9001"`
TelegramToken string `env:"TELEGRAM_TOKEN"`
KafkaTopicGigaChat string `env:"KAFKA_TOPIC_GIGA_CHAT"`
KafkaGroupGigaChat string `env:"KAFKA_GROUP_GIGA_CHAT" default:"gigachat"`
GigaChatApiBaseURL string `env:"GIGA_CHAT_API_BASE_URL"`
GigaChatApiAuthKey string `env:"GIGA_CHAT_API_AUTH_KEY"`
}
func LoadConfig() (*Config, error) {

@ -5,7 +5,7 @@ import (
"crypto/tls"
"fmt"
"gitea.pena/SQuiz/common/model"
"gitea.pena/SQuiz/worker/clients/gigachat"
"gitea.pena/SQuiz/worker/internal/clients/gigachat"
"github.com/go-redis/redis/v8"
"github.com/go-resty/resty/v2"
"go.uber.org/zap"