chatwars-spy/cmd/main/main.go

package main

import (
	"context"
	"encoding/json"
	"gitea.russia9.dev/Russia9/chatwars-spy/internal/utils"
	"gitea.russia9.dev/Russia9/chatwars-spy/pkg/domain"
	"gitea.russia9.dev/Russia9/chatwars-spy/pkg/message"
	"gitea.russia9.dev/Russia9/chatwars-spy/pkg/storage/mongodb"
	"github.com/confluentinc/confluent-kafka-go/kafka"
	"github.com/google/uuid"
	"github.com/rs/zerolog"
	"github.com/rs/zerolog/log"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
	"os"
	"strconv"
)

func main() {
	// Log settings
	zerolog.TimeFieldFormat = zerolog.TimeFormatUnix
	pretty, err := strconv.ParseBool(os.Getenv("LOG_PRETTY"))
	if err != nil {
		pretty = false
	}
	if pretty {
		log.Logger = log.Output(zerolog.ConsoleWriter{Out: os.Stderr})
	}
	switch os.Getenv("LOG_LEVEL") {
	case "DISABLED":
		zerolog.SetGlobalLevel(zerolog.Disabled)
	case "PANIC":
		zerolog.SetGlobalLevel(zerolog.PanicLevel)
	case "FATAL":
		zerolog.SetGlobalLevel(zerolog.FatalLevel)
	case "ERROR":
		zerolog.SetGlobalLevel(zerolog.ErrorLevel)
	case "WARN":
		zerolog.SetGlobalLevel(zerolog.WarnLevel)
	case "DEBUG":
		zerolog.SetGlobalLevel(zerolog.DebugLevel)
	case "TRACE":
		zerolog.SetGlobalLevel(zerolog.TraceLevel)
	default:
		zerolog.SetGlobalLevel(zerolog.InfoLevel)
	}

	version := utils.GetEnv("CHATWARS_VERSION", "cw3")

	// Kafka consumer init
	consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
		"bootstrap.servers": utils.GetEnv("KAFKA_ADDRESS", "digest-api.chtwrs.com:9092"),
		"group.id":          version + "-" + uuid.New().String(),
		"auto.offset.reset": "latest",
	})
	if err != nil {
		log.Fatal().Err(err).Send()
	}

	// DB init
	client, err := mongo.Connect(context.Background(), options.Client().ApplyURI(os.Getenv("MONGO_URI")))
	if err != nil {
		log.Fatal().Err(err).Send()
	}
	defer client.Disconnect(context.Background())
	err = client.Ping(context.Background(), nil)
	if err != nil {
		log.Fatal().Err(err).Send()
	}
	db := client.Database(utils.GetEnv("MONGO_DB", "chatwars"))
	repo := mongodb.NewUserRepo(db)

	// Kafka subscribe
	err = consumer.SubscribeTopics([]string{version + "-offers", version + "-deals", version + "-duels"}, nil)
	if err != nil {
		log.Fatal().Err(err).Send()
	}

	log.Info().Msg("Started watching")

	// Get messages
	for {
		// Read message
		raw, err := consumer.ReadMessage(-1)
		if err != nil {
			log.Error().Str("source", "consumer").Err(err).Send()
			continue
		}

		log.Trace().Bytes("message", raw.Value).Msg("received message")

		switch *raw.TopicPartition.Topic {
		case version + "-offers": // Offers
			// Decode message
			var msg message.Offer
			err = json.Unmarshal(raw.Value, &msg)
			if err != nil {
				log.Warn().Err(err).Str("topic", *raw.TopicPartition.Topic).Send()
				continue
			}

			// Store user
			_ = repo.Store(context.Background(),
				&domain.User{
					ID:     msg.SellerID,
					Castle: msg.SellerCastle,
					Name:   msg.SellerName,
					Source: "offers",
				},
			)
		case version + "-duels": // Duels
			// Decode message
			var msg message.Duel
			err = json.Unmarshal(raw.Value, &msg)
			if err != nil {
				log.Warn().Err(err).Str("topic", *raw.TopicPartition.Topic).Send()
				continue
			}

			// Store users
			_ = repo.Store(context.Background(),
				&domain.User{
					ID:     msg.Winner.ID,
					Castle: msg.Winner.Castle,
					Guild:  msg.Winner.Tag,
					Name:   msg.Winner.Name,
					Level:  msg.Winner.Level,
					Source: "duels",
				},
			)
			_ = repo.Store(context.Background(),
				&domain.User{
					ID:     msg.Loser.ID,
					Castle: msg.Loser.Castle,
					Guild:  msg.Loser.Tag,
					Name:   msg.Loser.Name,
					Level:  msg.Loser.Level,
					Source: "duels",
				},
			)
		case version + "-deals": // Deals
			// Decode message
			var msg message.Deal
			err = json.Unmarshal(raw.Value, &msg)
			if err != nil {
				log.Warn().Err(err).Str("topic", *raw.TopicPartition.Topic).Send()
				continue
			}

			// Store users
			_ = repo.Store(context.Background(),
				&domain.User{
					ID:     msg.SellerID,
					Castle: msg.SellerCastle,
					Name:   msg.SellerName,
					Source: "deals",
				},
			)
			_ = repo.Store(context.Background(),
				&domain.User{
					ID:     msg.BuyerID,
					Castle: msg.BuyerCastle,
					Name:   msg.BuyerName,
					Source: "deals",
				},
			)
		}
	}
}