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", }, ) } } }