package authPostman import ( "context" "encoding/json" "fmt" "net" "strconv" "git.slaventius.ru/test3k/authPostman/internal/config" kafka "git.slaventius.ru/test3k/authPostman/internal/transport/kafka" // telegram "git.slaventius.ru/test3k/authPostman/internal/transport/telegram" smtp "git.slaventius.ru/test3k/authPostman/internal/transport/smtp" api "git.slaventius.ru/test3k/umate/pkg/kafka" logger "git.slaventius.ru/test3k/umate/pkg/logger" ) type AuthPostmanServer struct { ctx context.Context kafkaReader *kafka.KafkaReader logger *logger.Logger config *config.Config } func NewServer(ctx context.Context, config *config.Config) *AuthPostmanServer { logger := logger.NewLogger("test3k:authPostmanService", config.Sentry.DSN) address := []string{net.JoinHostPort(config.Kafka.Host, strconv.Itoa(config.Kafka.Port))} return &AuthPostmanServer{ ctx: ctx, kafkaReader: kafka.NewReader(ctx, config, api.TopicRegistrations, address...), logger: logger, config: config, } } func (s *AuthPostmanServer) GracefulStop() error { return s.kafkaReader.Close() } func (s *AuthPostmanServer) ReadMessage(offset int64) error { // ... // s.kafkaReader.SetOffset(offset) // for { m, err := s.kafkaReader.ReadMessage() if err != nil { return err } // Декодируем сообщение msg := api.MessageRegistration{} erk := json.Unmarshal(m.Value, &msg) if erk != nil { return erk } // s.logger.Printf("send code %s to %s ...", msg.Code, msg.Email) // text := fmt.Sprintf("Confirmation code %v", msg.Code) // postman := telegram.NewService(s.config.Telegram.ChatToken) // message := postman.NewMessage(s.config.Telegram.ChatID, text) // ers := postman.SendMessage(message) postman := smtp.NewService(s.config.Smtp.Host, s.config.Smtp.Port, s.config.Smtp.Sender, s.config.Smtp.Password) message := smtp.NewMessage("Confirmation code", text) message.AppendRecipient(msg.Email) ers := postman.Send(message) if ers != nil { s.logger.Print(ers) } else { s.logger.Printf("send code %s to %s completed", msg.Code, msg.Email) } } }