parr_api/PARR.Core/Services/TaskServices/Implementations/TaskReconciliationService.cs

using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using PARR.Core.Common.Interfaces.RabbitServices;
using PARR.Core.Repositories.Interfaces.TaskRepositories;
using PARR.Core.Services.TaskServices.Interfaces;
using PARR.Core.Services.TaskServices.Models;
using PARR.Core.Services.TaskServices.Providers;
using PARR.Domain.Common.Rabbit.Messages;
using PARR.Domain.Entities.TaskEntities;
using PARR.Domain.Enums;
using PARR.Domain.Settings;

namespace PARR.Core.Services.TaskServices.Implementations
{
    internal class TaskReconciliationService : ITaskReconciliationService
    {
        private readonly ILogger<TaskReconciliationService> logger;
        private readonly ITaskTypeRepository taskTypeRepository;
        private readonly ITaskRepository taskRepository;
        private readonly ITaskErrorRepository taskErrorRepository;
        private readonly IReconciliationSettings reconciliationSettings;
        private readonly IRabbitService rabbitService;
        private readonly ITaskMqSettingsProvider taskMqSettingsProvider;

        public TaskReconciliationService(
            ILogger<TaskReconciliationService> logger,
            ITaskTypeRepository taskTypeRepository,
            ITaskRepository taskRepository,
            ITaskErrorRepository taskErrorRepository,
            IReconciliationSettings reconciliationSettings,
            IRabbitService rabbitService,
            ITaskMqSettingsProvider taskMqSettingsProvider
            )
        {
            this.logger = logger;
            this.taskTypeRepository = taskTypeRepository;
            this.taskRepository = taskRepository;
            this.taskErrorRepository = taskErrorRepository;
            this.reconciliationSettings = reconciliationSettings;
            this.rabbitService = rabbitService;
            this.taskMqSettingsProvider = taskMqSettingsProvider;
        }

        public async Task<ReconciliationReport> RunAsync()
        {
            var report = new ReconciliationReport
            {
                StartedAt = DateTimeOffset.UtcNow
            };

            logger.LogInformation("Начало выполнения Reconciliation Job.");

            // Загружаем все типы задач (для получения максимальной длительности)
            var taskTypes = await taskTypeRepository.Get().AsNoTracking().ToDictionaryAsync(t => t.Code, t => t);

            if (taskTypes.Count == 0)
            {
                logger.LogWarning("Типы задач не настроены в БД (таблица {TaskTypes})", nameof(TaskType));
                report.Messages.Add("TaskTypes пустые. Не настроены в БД.");
                return report;
            }

            // Для каждого типа ищем зависшие задачи
            foreach (var taskType in taskTypes.Values)
            {
                var stuckTasks = await FindStuckTasksAsync(taskType);
                report.CheckedCount += stuckTasks.Count;

                foreach (var task in stuckTasks)
                {
                    await ProcessStuckTaskAsync(task, taskType, report);
                }
            }

            report.CompletedAt = DateTimeOffset.UtcNow;

            if (report.HasChanged)
                logger.LogInformation("Reconciliation Job завершен. Проверено: {Checked}, Исправлено: {Fixed}, Retry: {Retried}, Failed: {Failed}. Длительность: {Duration}",
                                         report.CheckedCount, report.FixedCount, report.RetriedCount, report.FailedCount, report.Duration);
            else
                logger.LogDebug("Reconciliation Job завершен. Зависших задач не найдено");

            return report;
        }


        /// <summary>
        /// Поиск зависших задач для конкретного типа.
        /// </summary>
        /// <param name="taskType"></param>
        /// <returns></returns>
        private async Task<List<TaskItem>> FindStuckTasksAsync(TaskType taskType)
        {
            // Вычисляем порог: задача зависла, если выполняется дольше чем MaxExecutionTimeMinutes
            var timeoutThreshold = DateTimeOffset.UtcNow.AddMinutes(-taskType.MaxExecutionTimeMinutes);

            logger.LogDebug("Поиск зависших задач типа {Type}. Порог: {Threshold}", taskType.Code, timeoutThreshold);

            return await taskRepository.Get()
                            .Where(t =>
                                        //t.TypeCode == taskType.Code && t.DateModified < timeoutThreshold &&
                                        t.TypeCode == taskType.Code
                                        // если это вновь созданная задача и она упала, то у нее может не быть DateModified, тогда сравним с DateCreated
                                        && (t.DateModified ?? t.DateCreated) < timeoutThreshold 
                                        &&  (
                                                // Зависла в Processing
                                                t.StatusCode == TaskItemStatusEnum.Processing
                                                //Зависли в Pending
                                                || (t.StatusCode == TaskItemStatusEnum.Pending /*&& t.RetryCount > 0*/)
                                            )
                                        )
                            .OrderBy(t => t.DateModified) // Сначала самые старые
                            .Take(reconciliationSettings.MaxTaskPerRun) // Защита от перегрузки
                            .ToListAsync();
        }

        /// <summary>
        /// Обработка одной зависшей задачи.
        /// </summary>
        /// <param name="task"></param>
        /// <param name="taskType"></param>
        /// <param name="report"></param>
        /// <returns></returns>
        private async System.Threading.Tasks.Task ProcessStuckTaskAsync(TaskItem task, TaskType taskType, ReconciliationReport report)
        {
            logger.LogWarning("Обнаружена одна зависшая задача {TaskId} (тип {Type}). Последнее обновление: {LastModified}", task.Id, task.TypeCode, task.DateModified);

            // Записываем ошибку в историю
            var error = new TaskError
            {
                Id = Guid.NewGuid(),
                TaskId = task.Id,
                AttemptNumber = task.RetryCount + 1, // показываем планируемую попытку
                ErrorMessage = $"Превышен порог выполнения задачи. LastDateModified: {task.DateModified}"
            };

            await taskErrorRepository.CreateAsync(error);
            await taskErrorRepository.CommitAsync();

            // Если лимит исчерпан - ставим Failed
            if (task.RetryCount >= taskType.MaxRetries)
            {
                task.StatusCode = TaskItemStatusEnum.Failed;
                task.ProcessedAt = DateTimeOffset.UtcNow;
                task.DateModified = DateTimeOffset.UtcNow;

                await taskRepository.CommitAsync();

                report.FailedCount++;
                report.Messages.Add($"Задача {task.Id}: закончились попытки.");

                return;
            }

            // Сначала пытаемся отправить в очередь
            var mqSettings = taskMqSettingsProvider.GetSettings(taskType.Code);

            var message = new TaskMessage { TaskId = task.Id, TypeCode = task.TypeCode };
            var sendResult = await rabbitService.SendAsync(mqSettings, new List<object> { message });

            if (!sendResult.IsSuccess)
            {
                // Очередь недоступна — НЕ меняем БД, НЕ тратим RetryCount
                // Задача останется в текущем статусе и будет найдена в следующий раз

                logger.LogError("Не удалось отправить задачу {TaskId} в очередь. Повторим в следующей итерации.", task.Id);
                report.Messages.Add($"Задача {task.Id}: не смогла отправиться в очередь, обработается в следующей итерации.");
                return;  // Выходим без сохранения изменений
            }

            // Очередь успешна - обновляем БД
            // Ставим задачу в ожидание
            task.StatusCode = TaskItemStatusEnum.Pending;

            // Может получиться так, что воркер возьмет задачу быстрее, чем в БД у нее установится статус Pending.
            // Если это произойдет, ничего страшного, воркер это не отработает. Задача опять попадет в ProcessStuckTaskAsync и повторно отправится в очередь. 

            task.RetryCount++;
            task.DateModified = DateTimeOffset.UtcNow;

            await taskRepository.CommitAsync();

            report.RetriedCount++;
            report.Messages.Add($"Задача {task.Id}: опубликована в очередь, попытка #{task.RetryCount}");
            report.FixedCount++;
        }


    }
}