change logic for sendig recovery to mail

internal/controller/recovery/recovery_controller.go

@@ -2,8 +2,10 @@ package recovery
 
 import (
 	"codeword/internal/models"
+	"codeword/internal/repository"
 	"codeword/internal/services"
 	"encoding/base64"
+	"errors"
 	"fmt"
 	"github.com/gofiber/fiber/v2"
 	"go.uber.org/zap"
@@ -80,6 +82,11 @@ func (r *RecoveryController) HandleRecoveryRequest(c *fiber.Ctx) error {
 	err = r.service.RecoveryEmailTask(c.Context(), models.RecEmailDeps{UserID: user.ID.Hex(), Email: req.Email, SignWithID: signWithID, ID: id})
 	if err != nil {
 		r.logger.Error("Failed to send recovery email", zap.Error(err))
+
+		if errors.Is(err, repository.ErrAlreadyReported) {
+			return c.Status(fiber.StatusAlreadyReported).JSON(fiber.Map{"error": "already reported"})
+		}
+
 		return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "Internal Server Error"})
 	}
 

internal/repository/codeword_repository.go

@@ -46,6 +46,17 @@ func (r *CodewordRepository) StoreRecoveryRecord(ctx context.Context, deps model
 
 // добавляем в очередь данные для отправки на почту в редис
 func (r *CodewordRepository) InsertToQueue(ctx context.Context, deps models.RecEmailDeps) error {
+	sendLockKey := "email:sendLock:" + deps.Email
+	ttl := 5 * time.Minute
+
+	lockSuccess, err := r.rdb.SetNX(ctx, sendLockKey, "1", ttl).Result()
+	if err != nil {
+		return err
+	}
+	if !lockSuccess {
+		return ErrAlreadyReported
+	}
+
 	task := models.RecoveryRecord{
 		ID:     deps.ID,
 		UserID: deps.UserID,
@@ -58,11 +69,7 @@ func (r *CodewordRepository) InsertToQueue(ctx context.Context, deps models.RecE
 		return err
 	}
 
-	if err := r.rdb.LPush(ctx, "recoveryQueue", taskBytes).Err(); err != nil {
+	return r.rdb.Set(ctx, "email:task:"+deps.Email, taskBytes, ttl).Err()
-		return err
-	}
-
-	return nil
 }
 
 // получаем данные юзера по подписи

internal/repository/errors.go

@@ -0,0 +1,10 @@
+package repository
+
+import "errors"
+
+var (
+	ErrPromoUserNotFound = errors.New("user not found")
+	ErrAlreadyReported   = errors.New("already reported")
+	ErrDuplicateCodeword = errors.New("duplicate codeword")
+	ErrPromoCodeNotFound = errors.New("promo code not found")
+)

internal/repository/promocode_repository.go

@@ -3,7 +3,6 @@ package repository
 import (
 	"codeword/internal/models"
 	"context"
-	"errors"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.mongodb.org/mongo-driver/mongo"
@@ -11,11 +10,6 @@ import (
 	"time"
 )
 
-var (
-	ErrDuplicateCodeword = errors.New("duplicate codeword")
-	ErrPromoCodeNotFound = errors.New("promo code not found")
-)
-
 // структура для горутины чтобы ошибки не пропускать
 type countResult struct {
 	count int64

internal/repository/user_repository.go

@@ -3,7 +3,6 @@ package repository
 import (
 	"codeword/internal/models"
 	"context"
-	"errors"
 	"github.com/go-redis/redis/v8"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/mongo"
@@ -18,8 +17,6 @@ type UserRepository struct {
 	mdb *mongo.Collection
 }
 
-var ErrPromoUserNotFound = errors.New("user not found")
-
 func NewUserRepository(deps Deps) *UserRepository {
 
 	return &UserRepository{mdb: deps.Mdb}

internal/worker/recovery_worker/recovery_worker.go

@@ -37,14 +37,13 @@ func NewRecoveryWC(deps Deps) *RecoveryWorker {
 }
 
 func (wc *RecoveryWorker) Start(ctx context.Context) {
-	ticker := time.NewTicker(1 * time.Second)
+	ticker := time.NewTicker(10 * time.Second)
 	defer ticker.Stop()
 
 	for {
 		select {
 		case <-ticker.C:
 			wc.processTasks(ctx)
-
 		case <-ctx.Done():
 			return
 		}
@@ -52,29 +51,39 @@ func (wc *RecoveryWorker) Start(ctx context.Context) {
 }
 
 func (wc *RecoveryWorker) processTasks(ctx context.Context) {
-	result, err := wc.redis.BRPop(ctx, 1*time.Second, "recoveryQueue").Result()
+	var cursor uint64
-	if err != nil {
+	for {
-		if err != redis.Nil {
+		var keys []string
-			wc.logger.Error("Failed to BRPop from the recovery queue", zap.Error(err))
+		var err error
+		keys, cursor, err = wc.redis.Scan(ctx, cursor, "email:task:*", 0).Result()
+		if err != nil {
+			wc.logger.Error("Failed to scan for email tasks", zap.Error(err))
+			break
 		}
-		return
-	}
 
-	if len(result) < 2 {
+		for _, key := range keys {
-		wc.logger.Error("Received unexpected number of elements from BRPop", zap.Strings("result", result))
+			taskBytes, err := wc.redis.GetDel(ctx, key).Result()
-		return
+			if err == redis.Nil {
-	}
+				continue
+			} else if err != nil {
+				wc.logger.Error("Failed to getdel recovery task", zap.String("key", key), zap.Error(err))
+				continue
+			}
 
-	var task models.RecoveryRecord
+			var task models.RecoveryRecord
-	if err = json.Unmarshal([]byte(result[1]), &task); err != nil {
+			if json.Unmarshal([]byte(taskBytes), &task) != nil {
-		wc.logger.Error("Failed to unmarshal recovery task", zap.String("key", result[0]), zap.Error(err))
+				wc.logger.Error("Failed to unmarshal recovery task", zap.String("key", key), zap.String("task", taskBytes))
-		return
+				continue
-	}
+			}
 
-	err = wc.sendRecoveryTask(ctx, task)
+			err = wc.sendRecoveryTask(ctx, task)
-	if err != nil {
+			if err != nil {
-		wc.logger.Error("Failed to send recovery task", zap.String("key", result[0]), zap.Error(err))
+				wc.logger.Error("Failed to send recovery task", zap.String("key", key), zap.Error(err))
-		return
+			}
+		}
+		if cursor == 0 {
+			break
+		}
 	}
 }