internal/adapter/queue/redis/redis_queue.go - main - AngelCareMe/task_queue

internal/adapter/queue/redis/redis_queue.go
345 строк10 KB
AngelCareMe
done
28 июл 2025, 21:07
28 июл 2025, 21:07c70014b
package redis

import (
	"context"
	"encoding/json"
	"fmt"
	"task-queue/internal/adapter/queue"
	"task-queue/internal/adapter/repository"
	"task-queue/internal/entity"
	"task-queue/pkg/logger"
	"time"

	"github.com/google/uuid"
	"github.com/redis/go-redis/v9"
	"github.com/sirupsen/logrus"
)

const (
	//Redis keys
	TasksQueueKey      = "tasks:queue"
	TasksProcessingKey = "tasks:processing"
	TasksFailedKey     = "tasks:failed"
	TaskTTL            = 24 * time.Hour
)

type redisQueue struct {
	client     *redis.Client
	logger     *logrus.Logger
	serializer *TaskSerializer
	taskRepo   repository.TaskRepository
}

func NewRedisQueue(client *redis.Client, taskRepo repository.TaskRepository) queue.TaskQueue {
	return &redisQueue{
		client:     client,
		logger:     logger.Get(),
		serializer: NewTaskSerializer(),
		taskRepo:   taskRepo,
	}
}

func (rq *redisQueue) Enqueue(ctx context.Context, task *entity.Task) error {
	//serialize task
	data, err := rq.serializer.Serialize(task)
	if err != nil {
		return fmt.Errorf("failed to serialize task: %w", err)
	}

	//add task to queue
	err = rq.client.LPush(ctx, TasksQueueKey, data).Err()
	if err != nil {
		rq.logger.WithError(err).WithField("task_id", task.ID).Error("failed to enqueue task")
		return fmt.Errorf("failed to enqueue tasl: %w", err)
	}

	rq.client.Expire(ctx, TasksQueueKey, TaskTTL)

	rq.logger.WithField("task_id", task.ID).Debug("task enqueued successfuly")
	return nil
}

func (rq *redisQueue) Dequeue(ctx context.Context) (*queue.DequeuedTask, error) {
	// Извлекаем элемент из очереди
	result, err := rq.client.BRPop(ctx, 5*time.Second, TasksQueueKey).Result()
	if err != nil {
		if err == redis.Nil {
			return nil, nil // Нет задач в очереди
		}
		return nil, fmt.Errorf("failed to dequeue task: %w", err)
	}

	if len(result) < 2 {
		return nil, fmt.Errorf("invalid BRPop result")
	}

	// Получаем данные из очереди
	taskData := result[1]

	var task *entity.Task

	// Попробуем сначала десериализовать как полную задачу (старый формат)
	task, deserializeErr := rq.serializer.Deserialize([]byte(taskData))
	if deserializeErr != nil {
		// Если не удалось, предположим, что это ID задачи (новый формат)
		taskID := string(taskData)
		if _, uuidErr := uuid.Parse(taskID); uuidErr == nil {
			// Это валидный UUID, попробуем получить задачу из БД
			fetchedTask, fetchErr := rq.taskRepo.GetByID(ctx, taskID)
			if fetchErr != nil {
				rq.logger.WithError(fetchErr).WithField("task_id", taskID).Error("failed to fetch task from database")
				return nil, fmt.Errorf("failed to fetch task %s from database: %w", taskID, fetchErr)
			}
			task = fetchedTask
			rq.logger.WithField("task_id", taskID).Debug("task fetched from database by ID")
		} else {
			// Не похоже ни на задачу, ни на ID
			rq.logger.WithField("data", taskData).Error("dequeued data is neither a valid task nor a valid task ID")
			return nil, fmt.Errorf("dequeued data is invalid: %w", deserializeErr) // Возвращаем оригинальную ошибку десериализации
		}
	} else {
		rq.logger.WithField("task_id", task.ID).Debug("task deserialized from queue data")
	}

	// Генерируем receipt ID для подтверждения обработки
	receiptID := uuid.New().String()

	// Добавляем задачу в список обрабатываемых (сериализуем полную задачу)
	processingData, _ := rq.serializer.Serialize(task)
	processingKey := fmt.Sprintf("%s:%s", TasksProcessingKey, receiptID)
	err = rq.client.Set(ctx, processingKey, processingData, TaskTTL).Err()
	if err != nil {
		rq.logger.WithError(err).WithField("task_id", task.ID).Error("failed to track processing task")
		// Не критично, продолжаем
	}

	rq.logger.WithFields(logrus.Fields{
		"task_id":    task.ID,
		"receipt_id": receiptID,
	}).Debug("task dequeued successfully")

	return &queue.DequeuedTask{
		Task:      task,
		ReceiptID: receiptID,
	}, nil
}

func (rq *redisQueue) Ack(ctx context.Context, taskID string) error {
	// found n delete task from worker list
	pattern := fmt.Sprintf("%s:*", TasksProcessingKey)
	keys, err := rq.client.Keys(ctx, pattern).Result()
	if err != nil {
		return fmt.Errorf("failed to find processing tasks: %w", err)
	}

	// serach key with new task
	var processingKey string
	for _, key := range keys {
		data, err := rq.client.Get(ctx, key).Result()
		if err != nil {
			continue
		}

		task, err := rq.serializer.Deserialize([]byte(data))
		if err != nil {
			continue
		}

		if task.ID == taskID {
			processingKey = key
			break
		}
	}

	if processingKey == "" {
		return fmt.Errorf("processing task not found: %s", taskID)
	}

	// delete task from worker list
	err = rq.client.Del(ctx, processingKey).Err()
	if err != nil {
		rq.logger.WithError(err).WithField("task_id", taskID).Error("failed to ack task")
		return fmt.Errorf("failed to ack task: %w", err)
	}

	rq.logger.WithField("task_id", taskID).Debug("task acknowledged successfully")
	return nil
}

func (rq *redisQueue) Nack(ctx context.Context, taskID string, reason string) error {
	// search task in workers list
	pattern := fmt.Sprintf("%s:*", TasksProcessingKey)
	keys, err := rq.client.Keys(ctx, pattern).Result()
	if err != nil {
		return fmt.Errorf("failed to find processing tasks: %w", err)
	}

	// seraching key
	var processingKey string
	var task *entity.Task
	for _, key := range keys {
		data, err := rq.client.Get(ctx, key).Result()
		if err != nil {
			continue
		}

		t, err := rq.serializer.Deserialize([]byte(data))
		if err != nil {
			continue
		}

		if t.ID == taskID {
			processingKey = key
			task = t
			break
		}
	}

	if processingKey == "" || task == nil {
		return fmt.Errorf("processing task not found: %s", taskID)
	}

	// delete frim workers
	err = rq.client.Del(ctx, processingKey).Err()
	if err != nil {
		rq.logger.WithError(err).WithField("task_id", taskID).Error("failed to remove from processing")
	}

	// add to failure list
	failedKey := fmt.Sprintf("%s:%s", TasksFailedKey, taskID)
	failedData := map[string]interface{}{
		"task":      task,
		"reason":    reason,
		"failed_at": time.Now(),
	}

	failedDataBytes, _ := json.Marshal(failedData)
	err = rq.client.Set(ctx, failedKey, failedDataBytes, TaskTTL).Err()
	if err != nil {
		rq.logger.WithError(err).WithField("task_id", taskID).Error("failed to track failed task")
	}

	rq.logger.WithFields(logrus.Fields{
		"task_id": taskID,
		"reason":  reason,
	}).Debug("task negatively acknowledged")
	return nil
}

// Retry puts task back to queue for reprocessing
func (rq *redisQueue) Retry(ctx context.Context, taskID string, delay time.Duration) error {
	// For retry, we simply add the task ID back to queue
	// The worker will fetch full task data from database
	// This is simpler and more reliable

	taskData := []byte(taskID) // Just the ID for now

	if delay > 0 {
		// Use Redis delayed queue through sorted set
		score := float64(time.Now().Add(delay).Unix())
		err := rq.client.ZAdd(ctx, "tasks:delayed", redis.Z{
			Score:  score,
			Member: taskData,
		}).Err()
		if err != nil {
			rq.logger.WithError(err).WithField("task_id", taskID).Error("failed to add delayed task")
			return fmt.Errorf("failed to add delayed task: %w", err)
		}
	} else {
		// Add to regular queue
		err := rq.client.LPush(ctx, TasksQueueKey, taskData).Err()
		if err != nil {
			rq.logger.WithError(err).WithField("task_id", taskID).Error("failed to add task to queue")
			return fmt.Errorf("failed to add task to queue: %w", err)
		}
	}

	rq.logger.WithFields(logrus.Fields{
		"task_id": taskID,
		"delay":   delay,
	}).Debug("task ID scheduled for retry")
	return nil
}

func (rq *redisQueue) GetStatus(ctx context.Context, taskID string) (queue.TaskStatus, error) {
	// check in queue
	queueLen, err := rq.client.LLen(ctx, TasksQueueKey).Result()
	if err == nil && queueLen > 0 {
		queueTasks, err := rq.client.LRange(ctx, TasksQueueKey, 0, queueLen-1).Result()
		if err == nil {
			for _, taskData := range queueTasks {
				task, err := rq.serializer.Deserialize([]byte(taskData))
				if err == nil && task.ID == taskID {
					return queue.TaskStatus{
						Status: entity.TaskStatusPending,
					}, nil
				}
			}
		}
	}

	// check in worker list
	pattern := fmt.Sprintf("%s:*", TasksProcessingKey)
	keys, err := rq.client.Keys(ctx, pattern).Result()
	if err == nil {
		for _, key := range keys {
			data, err := rq.client.Get(ctx, key).Result()
			if err != nil {
				continue
			}

			task, err := rq.serializer.Deserialize([]byte(data))
			if err != nil {
				continue
			}

			if task.ID == taskID {
				return queue.TaskStatus{
					Status: entity.TaskStatusProcessing,
				}, nil
			}
		}
	}

	// check in failure list
	failedKey := fmt.Sprintf("%s:%s", TasksFailedKey, taskID)
	exists, err := rq.client.Exists(ctx, failedKey).Result()
	if err == nil && exists > 0 {
		return queue.TaskStatus{
			Status: entity.TaskStatusFailed,
		}, nil
	}

	return queue.TaskStatus{}, fmt.Errorf("task status not found: %s", taskID)
}

func (rq *redisQueue) GetQueueSize(ctx context.Context) (int64, error) {
	return rq.client.LLen(ctx, TasksQueueKey).Result()
}

func (rq *redisQueue) GetProcessingCount(ctx context.Context) (int64, error) {
	pattern := fmt.Sprintf("%s:*", TasksProcessingKey)
	keys, err := rq.client.Keys(ctx, pattern).Result()
	if err != nil {
		return 0, err
	}
	return int64(len(keys)), nil
}

func (rq *redisQueue) GetFailedCount(ctx context.Context) (int64, error) {
	pattern := fmt.Sprintf("%s:*", TasksFailedKey)
	keys, err := rq.client.Keys(ctx, pattern).Result()
	if err != nil {
		return 0, err
	}
	return int64(len(keys)), nil
}

func (rq *redisQueue) HealthCheck(ctx context.Context) error {
	return rq.client.Ping(ctx).Err()
}

func (rq *redisQueue) Close() error {
	// Connection closing by connection layer
	return nil
}