common/validate/common.go

package validate

import (
	"bytes"
	"context"
	"fmt"
	"gitea.pena/PenaSide/common/encrypt"
	"gitea.pena/PenaSide/common/minio_initialize"
	"gitea.pena/PenaSide/common/mongo"
	"github.com/go-redis/redis/v8"
	"github.com/gofiber/fiber/v2"
	"github.com/minio/minio-go/v7"
	"github.com/pioz/faker"
	"github.com/twmb/franz-go/pkg/kgo"
	"github.com/twmb/franz-go/pkg/kmsg"
	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"log"
	"regexp"
	"time"
)

type t struct {
	ID string `bson:"_id,omitempty"`
	I  int    `bson:"i"`
}

// todo в будущем в монге будут запрещены некоторые операции, надо будет обновлять
func ValidateMongo(cfg mongo.Configuration) error {
	if cfg.URL == "" {
		return fmt.Errorf("mongo URL is empty")
	}
	if cfg.DatabaseName == "" {
		return fmt.Errorf("mongo database name is empty")
	}

	cfg.DatabaseName = "testDBName"
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	database, err := mongo.Connect(ctx, &mongo.ConnectDeps{
		Configuration: &cfg,
		Timeout:       10 * time.Second,
	})
	if err != nil {
		return err
	}

	defer database.Drop(ctx)

	testCollection := database.Collection(cfg.DatabaseName)

	receivedChannel := make(chan string, 10)
	errorChannel := make(chan error, 1)

	go func() {
		defer close(receivedChannel)
		defer close(errorChannel)
		for i := 0; i <= 100; i++ {
			d := t{
				ID: primitive.NewObjectID().Hex(),
				I:  i,
			}

			_, err = testCollection.InsertOne(ctx, d)
			if err != nil {
				errorChannel <- err
			}

			receivedChannel <- d.ID
		}
	}()
	timeout := time.After(30 * time.Second)
	for {
		select {
		case err = <-errorChannel:
			if err != nil {
				return fmt.Errorf("error document insert: %w", err)
			}
		case id := <-receivedChannel:
			result := t{}
			err := testCollection.FindOne(ctx, bson.M{"_id": id}).Decode(&result)
			if err != nil {
				return fmt.Errorf("mongo error finding document: %v", err)
			}
			if id != result.ID {
				return fmt.Errorf("invalid id received")
			}
			if result.I == 100 {
				return nil
			}
		case <-timeout:
			return fmt.Errorf("timeout")
		}
	}
}

func ValidateTgToken(token string) error {
	if token == "" {
		return fmt.Errorf("tg token is empty")
	}

	// todo обдумать еще регулярку
	pattern := `^\d+:.+$`
	ok, err := regexp.MatchString(pattern, token)
	if err != nil {
		return fmt.Errorf("error validating tg token - %s: %w", token, err)
	}
	if !ok {
		return fmt.Errorf("invalid tg token format: %s", token)
	}

	return nil
}

func ValidateEncryptKeys(e *encrypt.Encrypt) error {
	codeWord := faker.String()
	shifr, err := e.EncryptStr(codeWord)
	if err != nil {
		return err
	}

	deShifr, err := e.DecryptStr(shifr)
	if err != nil {
		return err
	}

	if deShifr != codeWord {
		return fmt.Errorf("invalid encrypt key")
	}
	return nil
}

type ValidateS3Deps struct {
	S3Endpoint  string
	S3AccessKey string
	S3SecretKey string
	S3Token     string
	BucketName  string
	IsProd      bool
}

func ValidateS3(cfg ValidateS3Deps) error {
	if cfg.S3Endpoint == "" {
		return fmt.Errorf("s3 endpoint is empty")
	}
	if cfg.S3AccessKey == "" {
		return fmt.Errorf("s3 access key is empty")
	}
	if cfg.S3SecretKey == "" {
		return fmt.Errorf("s3 secret key is empty")
	}

	ctx := context.TODO()

	client, err := minio_initialize.Minio(minio_initialize.MinioInitialize{
		S3Endpoint:  cfg.S3Endpoint,
		S3AccessKey: cfg.S3AccessKey,
		S3SecretKey: cfg.S3SecretKey,
		S3Token:     cfg.S3Token,
		IsProd:      cfg.IsProd,
	})
	if err != nil {
		return fmt.Errorf("error connect s3 in validate: %w", err)
	}

	objectName := faker.String()
	content := []byte(faker.String())
	contentType := "text/plain"

	exist, err := client.BucketExists(ctx, cfg.BucketName)
	if err != nil {
		return fmt.Errorf("error find s3 bucket in validate: %w", err)
	}

	// не создаем бакет так как бакет могут не дать создать
	if !exist {
		log.Println("bucket does not exist")
		return nil
	}

	_, err = client.PutObject(ctx, cfg.BucketName, objectName,
		bytes.NewReader(content), int64(len(content)),
		minio.PutObjectOptions{ContentType: contentType})
	if err != nil {
		return fmt.Errorf("error create test object: %w", err)
	}

	err = client.RemoveObject(ctx, cfg.BucketName, objectName, minio.RemoveObjectOptions{})
	if err != nil {
		return fmt.Errorf("error remove test object: %w", err)
	}

	return nil
}

func ValidateSmtp(apiKey string) error {
	client := fiber.AcquireClient()
	url := "https://api.smtp.bz/v1/user"
	agent := client.Get(url)
	agent.Set("Authorization", apiKey)

	code, _, errs := agent.Bytes()
	if errs != nil {
		return errs[0]
	}

	if code != 200 {
		return fmt.Errorf("invalid SMTP code: %d", code)
	}
	return nil
}

func ValidateKafka(brokers []string, topic string) error {
	if len(brokers) == 0 {
		return fmt.Errorf("kafka brokers is empty")
	}
	if topic == "" {
		return fmt.Errorf("kafka topic is empty")
	}

	for _, addr := range brokers {
		if addr == "" {
			return fmt.Errorf("empty kafka broker")
		}
	}

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	kafkaClient, err := kgo.NewClient(
		kgo.SeedBrokers(brokers...),
		kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()),
	)
	if err != nil {
		return err
	}

	defer kafkaClient.Close()

	err = kafkaClient.Ping(ctx)
	if err != nil {
		return err
	}

	req := kmsg.NewMetadataRequest()
	req.Topics = []kmsg.MetadataRequestTopic{
		{Topic: kmsg.StringPtr(topic)},
	}

	res, err := req.RequestWith(ctx, kafkaClient)
	if err != nil {
		return fmt.Errorf("metadata request failed: %w", err)
	}

	if len(res.Topics) == 0 {
		return fmt.Errorf("no metadata returned for topic %q", topic)
	}

	// если == 0 то существует, https://kafka.apache.org/protocol.html#The_Messages_Metadata
	if res.Topics[0].ErrorCode != 0 {
		return fmt.Errorf("topic %q does not exist (error code %d)", topic, res.Topics[0].ErrorCode)
	}

	return nil
}

func ValidateRedis(redisHost, redisPassword string, redisDB int) error {
	if redisHost == "" {
		return fmt.Errorf("redis host is empty")
	}

	f := func() (*redis.Client, error) {
		rdb := redis.NewClient(&redis.Options{
			Addr:     redisHost,
			Password: redisPassword,
			DB:       redisDB,
		})

		status := rdb.Ping(context.TODO())
		if err := status.Err(); err != nil {
			return nil, err
		}

		return rdb, nil
	}

	_, err := f()
	if err != nil {
		return err
	}
	return nil
}