tg-bot-golang

1
// Сервис генерации отчетов.
2
package main
3

4
import (
5
	"context"
6
	"github.com/pkg/errors"
7
	"github.com/ellavs/tg-bot-golang/internal/api"
8
	"github.com/ellavs/tg-bot-golang/internal/config"
9
	"github.com/ellavs/tg-bot-golang/internal/helpers/dbutils"
10
	"github.com/ellavs/tg-bot-golang/internal/helpers/kafka"
11
	"github.com/ellavs/tg-bot-golang/internal/logger"
12
	"github.com/ellavs/tg-bot-golang/internal/model/bottypes"
13
	"github.com/ellavs/tg-bot-golang/internal/model/db"
14
	"google.golang.org/grpc"
15
	"google.golang.org/grpc/credentials/insecure"
16
	"log"
17
	"os/signal"
18
	"strconv"
19
	"syscall"
20
	"time"
21
)
22

23
// Параметры по умолчанию (могут быть изменены через config)
24
var (
25
	mainCurrency       = "RUB"                      // Основная валюта для хранения данных.
26
	connectionStringDB = ""                         // Строка подключения к базе данных.
27
	kafkaTopic         = "tgbot"                    // Наименование топика Kafka.
28
	brokersList        = []string{"localhost:9092"} // Список адресов брокеров сообщений (адрес Kafka).
29
)
30

31
func main() {
32

33
	logger.Info("[Report service] Старт приложения")
34

35
	ctx, cancel := signal.NotifyContext(context.Background(),
36
		syscall.SIGHUP,
37
		syscall.SIGINT,
38
		syscall.SIGTERM,
39
		syscall.SIGQUIT)
40
	defer cancel()
41

42
	config, err := config.New()
43
	if err != nil {
44
		logger.Fatal("[Report service] Ошибка получения файла конфигурации:", "err", err)
45
	}
46

47
	// Изменение параметров по умолчанию из заданной конфигурации.
48
	setConfigSettings(config.GetConfig())
49

50
	// Инициализация хранилищ (подключение к базе данных).
51
	dbconn, err := dbutils.NewDBConnect(connectionStringDB)
52
	if err != nil {
53
		logger.Fatal("[Report service] Ошибка подключения к базе данных:", "err", err)
54
	}
55
	// БД информации пользователей.
56
	userStorage := db.NewUserStorage(dbconn, mainCurrency, 0)
57

58
	// Инициализация кафки для получения сообщений из очереди.
59
	kafkaConsumer, err := kafka.NewConsumer(ctx, brokersList, kafkaTopic)
60
	if err != nil {
61
		logger.Fatal("[Report service] Ошибка инициализации кафки:", "err", err)
62
	}
63

64
	// Назначение функции, которая будет обрабатывать входящие сообщения из кафки.
65
	handlerFunc := func(ctx context.Context, key string, value string) error {
66
		return getKafkaMessage(ctx, userStorage, key, value)
67
	}
68

69
	// Запуск чтения сообщений из очереди.
70
	if err := kafkaConsumer.RunConsume(handlerFunc); err != nil {
71
		logger.Fatal("[Report service] Ошибка чтения кафки:", "err", err)
72
	}
73

74
	<-ctx.Done()
75
	logger.Info("[Report service] Завершение приложения")
76
}
77

78
// setConfigSettings Замена параметров по умолчанию параметрами из конфиг.файла.
79
func setConfigSettings(config config.Config) {
80
	if config.MainCurrency != "" {
81
		mainCurrency = config.MainCurrency
82
	}
83
	if config.ConnectionStringDB != "" {
84
		connectionStringDB = config.ConnectionStringDB
85
	}
86
}
87

88
// getKafkaMessage Получение запроса на построение отчета пользователя из кафки.
89
func getKafkaMessage(ctx context.Context, userStorage *db.UserStorage, key string, value string) error {
90
	if key == "" || value == "" {
91
		logger.Error("[Report service] Сообщение кафка содержит пустой ключ или значение.")
92
		return nil
93
	}
94
	userID, err := strconv.Atoi(key)
95
	if err != nil {
96
		logger.Error("[Report service] Сообщение кафка содержит некорректный ключ.", "err", err)
97
		return nil
98
	}
99
	// Получение данных для отчета из БД.
100
	periodDate := time.Now()
101
	switch value {
102
	case "w":
103
		periodDate = periodDate.AddDate(0, 0, -7)
104
	case "m":
105
		periodDate = periodDate.AddDate(0, -1, 0)
106
	case "y":
107
		periodDate = periodDate.AddDate(-1, 0, 0)
108
	}
109
	dt, err := userStorage.GetUserDataRecord(ctx, int64(userID), periodDate)
110
	if err != nil {
111
		logger.Error("[Report service] Ошибка получения отчета.", "err", err)
112
		return err
113
	}
114
	// Вызов бота для отправки отчета пользователю.
115
	err = sendReportToBot(dt, int64(userID), value)
116
	if err != nil {
117
		logger.Error("[Report service] Ошибка отправки отчета.", "err", err)
118
		return err
119
	}
120
	return nil
121
}
122

123
// sendReportToBot Вызов сервиса бота, принимающего данные для отчета по gRPC
124
func sendReportToBot(dt []bottypes.UserDataReportRecord, userID int64, reportKey string) error {
125

126
	// Соединение с сервисом бота.
127
	conn, err := grpc.Dial("localhost:50051", grpc.WithTransportCredentials(insecure.NewCredentials()))
128
	if err != nil {
129
		log.Fatalf("did not connect: %v", err)
130
	}
131
	defer conn.Close()
132
	c := api.NewUserReportsReciverClient(conn)
133

134
	// Подготовка данных для отправки по gRPC.
135
	items := make([]*api.ReportItem, len(dt))
136
	for ind, r := range dt {
137
		items[ind] = &api.ReportItem{Category: r.Category, Sum: float32(r.Sum)}
138
	}
139

140
	// Отправка данных.
141
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
142
	defer cancel()
143
	r, err := c.PutReport(ctx, &api.ReportRequest{Items: items, UserID: userID, ReportKey: reportKey})
144
	if err != nil {
145
		logger.Error("Ошибка отправки", "err", err)
146
	}
147

148
	if r.Valid {
149
		return nil
150
	}
151
	return errors.New("Отчет не отправлен.")
152
}
153