internal/usecase/task_usecase.go - main - AngelCareMe/task_queue

internal/usecase/task_usecase.go
315 строк10 KB
AngelCareMe
done
28 июл 2025, 21:07
28 июл 2025, 21:07c70014b
package usecase

import (
	"context"
	"fmt"
	"math/rand"
	"strings"
	"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/sirupsen/logrus"
)

type taskUseCase struct {
	taskRepo  TaskRepository
	taskQueue TaskQueue
	logger    *logrus.Logger
}

func NewTaskUseCase(taskRepo interface {
	repository.TaskRepository
	repository.TaskStatistics
	repository.RepositoryHealthChecker
}, taskQueue interface {
	queue.TaskQueue
	queue.QueueMetrics
	queue.QueueHealthChecker
}) TaskUseCase {
	return &taskUseCase{
		taskRepo:  taskRepo,
		taskQueue: taskQueue,
		logger:    logger.Get(),
	}
}

// createTask creates new task and puts it to queue
func (uc *taskUseCase) CreateTask(ctx context.Context, name, payload string, maxRetry, delay int) (*entity.Task, error) {
	// create task entity
	task := &entity.Task{
		ID:        uuid.New().String(),
		Name:      name,
		Payload:   payload,
		Status:    entity.TaskStatusPending,
		Retry:     0,
		MaxRetry:  maxRetry,
		Delay:     delay,
		CreatedAt: time.Now(),
		UpdatedAt: time.Now(),
	}

	// save to database
	if err := uc.taskRepo.Create(ctx, task); err != nil {
		uc.logger.WithError(err).WithField("task_name", name).Error("failed to create task in repository")
		return nil, fmt.Errorf("failed to create task in repository: %w", err)
	}

	// put to queue (considering delay)
	if delay > 0 {
		// for delayed tasks use retry mechanism with delay
		if err := uc.taskQueue.Retry(ctx, task.ID, time.Duration(delay)*time.Second); err != nil {
			uc.logger.WithError(err).WithField("task_id", task.ID).Error("failed to enqueue delayed task")
			// don't return error since task is already created in db
		}
	} else {
		// put to regular queue
		if err := uc.taskQueue.Enqueue(ctx, task); err != nil {
			uc.logger.WithError(err).WithField("task_id", task.ID).Error("failed to enqueue task")
			// don't return error since task is already created in db
		}
	}

	uc.logger.WithField("task_id", task.ID).Info("task created successfully")
	return task, nil
}

// getTask gets task by id
func (uc *taskUseCase) GetTask(ctx context.Context, id string) (*entity.Task, error) {
	task, err := uc.taskRepo.GetByID(ctx, id)
	if err != nil {
		uc.logger.WithError(err).WithField("task_id", id).Error("failed to get task")
		return nil, fmt.Errorf("failed to get task: %w", err)
	}

	return task, nil
}

// listTasks returns list of tasks with filtering
func (uc *taskUseCase) ListTasks(ctx context.Context, filter repository.TaskFilter, limit, offset int) ([]*entity.Task, error) {
	tasks, err := uc.taskRepo.List(ctx, filter, limit, offset)
	if err != nil {
		uc.logger.WithError(err).Error("failed to list tasks")
		return nil, fmt.Errorf("failed to list tasks: %w", err)
	}

	return tasks, nil
}

// retryTask retries task by putting it back to queue
func (uc *taskUseCase) RetryTask(ctx context.Context, id string) error {
	// get task from database first
	task, err := uc.taskRepo.GetByID(ctx, id)
	if err != nil {
		return fmt.Errorf("failed to get task for retry: %w", err)
	}

	// check if retry is allowed
	if task.Retry >= task.MaxRetry {
		return fmt.Errorf("max retry attempts exceeded for task %s", id)
	}

	// update task status to pending for reprocessing
	task.Status = entity.TaskStatusPending
	task.UpdatedAt = time.Now()

	// update task in database
	if err := uc.taskRepo.Update(ctx, task); err != nil {
		uc.logger.WithError(err).WithField("task_id", id).Error("failed to update task status for retry")
		return fmt.Errorf("failed to update task for retry: %w", err)
	}

	// put task to queue for reprocessing
	// for retry we put it directly to queue without delay by default
	// but we can add some delay if needed
	if err := uc.taskQueue.Retry(ctx, id, 0); err != nil {
		uc.logger.WithError(err).WithField("task_id", id).Error("failed to retry task in queue")
		// Don't return error here, task is updated in DB
		// It's up to the implementation how to handle queue errors
		// For now we'll log it and continue
		uc.logger.WithField("task_id", id).Warn("task updated in DB but failed to enqueue for retry")
	}

	uc.logger.WithField("task_id", id).Info("task scheduled for retry")
	return nil
}

// deleteTask deletes task
func (uc *taskUseCase) DeleteTask(ctx context.Context, id string) error {
	// todo: maybe check task status before deletion

	if err := uc.taskRepo.Delete(ctx, id); err != nil {
		uc.logger.WithError(err).WithField("task_id", id).Error("failed to delete task")
		return fmt.Errorf("failed to delete task: %w", err)
	}

	uc.logger.WithField("task_id", id).Info("task deleted successfully")
	return nil
}

// getStats returns task statistics
func (uc *taskUseCase) GetStats(ctx context.Context) (*repository.TaskStats, error) {
	stats, err := uc.taskRepo.GetStats(ctx)
	if err != nil {
		uc.logger.WithError(err).Error("failed to get task stats")
		return nil, fmt.Errorf("failed to get task stats: %w", err)
	}

	return stats, nil
}

// processTask processes task (called by worker)
func (uc *taskUseCase) ProcessTask(ctx context.Context, task *entity.Task) error {
	uc.logger.WithField("task_id", task.ID).Info("processing task")

	// Simulate real work based on task payload
	result, err := uc.executeTaskLogic(ctx, task)
	if err != nil {
		// Handle task processing error
		uc.logger.WithFields(logrus.Fields{
			"task_id": task.ID,
			"error":   err.Error(),
		}).Error("task processing failed")

		// Update task status to failed
		task.Status = entity.TaskStatusFailed
		task.Retry++
		task.UpdatedAt = time.Now()

		// Update task in database
		if updateErr := uc.taskRepo.Update(ctx, task); updateErr != nil {
			uc.logger.WithError(updateErr).WithField("task_id", task.ID).Error("failed to update failed task")
			return fmt.Errorf("failed to update task and process error: %w, update error: %v", err, updateErr)
		}

		// Log the failure for monitoring
		uc.logger.WithFields(logrus.Fields{
			"task_id":   task.ID,
			"retry":     task.Retry,
			"max_retry": task.MaxRetry,
		}).Warn("task marked as failed")

		// If max retries not exceeded, we could automatically retry
		// But this should be handled by worker or separate retry mechanism
		return fmt.Errorf("task processing failed: %w", err)
	}

	// Task processed successfully
	task.Status = entity.TaskStatusSuccess
	task.UpdatedAt = time.Now()

	if err := uc.taskRepo.Update(ctx, task); err != nil {
		uc.logger.WithError(err).WithField("task_id", task.ID).Error("failed to update task after processing")
		return fmt.Errorf("failed to update task: %w", err)
	}

	uc.logger.WithFields(logrus.Fields{
		"task_id": task.ID,
		"result":  result,
	}).Info("task processed successfully")
	return nil
}

// executeTaskLogic simulates actual task processing logic
// This is where real business logic would go
func (uc *taskUseCase) executeTaskLogic(ctx context.Context, task *entity.Task) (string, error) {
	// Simulate different types of work based on task name or payload
	switch {
	case strings.Contains(strings.ToLower(task.Name), "http"):
		return uc.processHTTPTask(ctx, task)
	case strings.Contains(strings.ToLower(task.Name), "email"):
		return uc.processEmailTask(ctx, task)
	case strings.Contains(strings.ToLower(task.Name), "calc"):
		return uc.processCalculationTask(ctx, task)
	default:
		return uc.processGenericTask(ctx, task)
	}
}

// processHTTPTask simulates HTTP request task
func (uc *taskUseCase) processHTTPTask(ctx context.Context, task *entity.Task) (string, error) {
	uc.logger.WithField("task_id", task.ID).Debug("processing HTTP task")

	// Simulate HTTP request delay
	select {
	case <-ctx.Done():
		return "", ctx.Err()
	case <-time.After(time.Duration(100+rand.Intn(200)) * time.Millisecond):
		// Simulate 10% failure rate for HTTP tasks
		if rand.Float32() < 0.1 {
			return "", fmt.Errorf("HTTP request failed: timeout")
		}
		return fmt.Sprintf("HTTP request completed for %s", task.Payload), nil
	}
}

// processEmailTask simulates email sending task
func (uc *taskUseCase) processEmailTask(ctx context.Context, task *entity.Task) (string, error) {
	uc.logger.WithField("task_id", task.ID).Debug("processing email task")

	// Simulate email sending delay
	select {
	case <-ctx.Done():
		return "", ctx.Err()
	case <-time.After(time.Duration(50+rand.Intn(100)) * time.Millisecond):
		// Simulate 5% failure rate for email tasks
		if rand.Float32() < 0.05 {
			return "", fmt.Errorf("email sending failed: SMTP error")
		}
		return fmt.Sprintf("Email sent to %s", task.Payload), nil
	}
}

// processCalculationTask simulates calculation task
func (uc *taskUseCase) processCalculationTask(ctx context.Context, task *entity.Task) (string, error) {
	uc.logger.WithField("task_id", task.ID).Debug("processing calculation task")

	// Simulate calculation work
	select {
	case <-ctx.Done():
		return "", ctx.Err()
	case <-time.After(time.Duration(200+rand.Intn(300)) * time.Millisecond):
		// Simulate 2% failure rate for calculation tasks
		if rand.Float32() < 0.02 {
			return "", fmt.Errorf("calculation failed: division by zero")
		}
		return fmt.Sprintf("Calculation result: %d", rand.Intn(10000)), nil
	}
}

// processGenericTask simulates generic task
func (uc *taskUseCase) processGenericTask(ctx context.Context, task *entity.Task) (string, error) {
	uc.logger.WithField("task_id", task.ID).Debug("processing generic task")

	// Simulate generic work
	select {
	case <-ctx.Done():
		return "", ctx.Err()
	case <-time.After(time.Duration(50+rand.Intn(150)) * time.Millisecond):
		// Simulate 3% failure rate for generic tasks
		if rand.Float32() < 0.03 {
			return "", fmt.Errorf("generic task failed: unknown error")
		}
		return fmt.Sprintf("Processed payload: %s", task.Payload), nil
	}
}

// healthCheck checks health of all components
func (uc *taskUseCase) HealthCheck(ctx context.Context) error {
	// check repository health
	if err := uc.taskRepo.HealthCheck(ctx); err != nil {
		return fmt.Errorf("repository health check failed: %w", err)
	}

	// check queue health (if queue has health check method)
	if healthChecker, ok := uc.taskQueue.(interface{ HealthCheck(context.Context) error }); ok {
		if err := healthChecker.HealthCheck(ctx); err != nil {
			return fmt.Errorf("queue health check failed: %w", err)
		}
	}

	return nil
}