feat(core): TaskManagementService - отправка сообщений

PARR.Core/Services/Task/Implementations/TaskManagementService.cs

@@ -0,0 +1,134 @@
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Logging;
+using PARR.Core.Common.Interfaces.RabbitServices;
+using PARR.Core.Repositories.Interfaces.TaskRepositories;
+using PARR.Core.Services.Task.Interfaces;
+using PARR.Domain.Common.Rabbit.Messages;
+using PARR.Domain.Entities.Base.History;
+using PARR.Domain.Entities.TaskEntities;
+using PARR.Domain.Enums;
+using PARR.Domain.Settings;
+using System.Text.Encodings.Web;
+using System.Text.Json;
+
+namespace PARR.Core.Services.Task.Implementations
+{
+    internal class TaskManagementService : ITaskManagementService
+    {
+        private readonly ILogger<TaskManagementService> logger;
+        private readonly ITaskTypeRepository taskTypeRepository;
+        private readonly ITaskRepository taskRepository;
+        private readonly IRabbitService rabbitService;
+
+        public TaskManagementService(
+                ILogger<TaskManagementService> logger,
+                ITaskTypeRepository taskTypeRepository,
+                ITaskRepository taskRepository,
+                IRabbitService rabbitService
+            )
+        {
+            this.logger = logger;
+            this.taskTypeRepository = taskTypeRepository;
+            this.taskRepository = taskRepository;
+            this.rabbitService = rabbitService;
+        }
+
+
+        public async Task<Guid> CreateTaskAsync<T>(TaskTypeEnum typeCode, T payload, IHistoryInitiator initiator, IMqSettings mqSettings)
+        {
+            var taskType = await taskTypeRepository.Get().AsNoTracking().FirstOrDefaultAsync(t => t.Code == typeCode);
+
+            if (taskType == null)
+            {
+                logger.LogError("Тип задачи {TypeCode} не найден в БД", typeCode);
+                throw new InvalidOperationException($"Тип задачи typeCode не найден в БД");
+            }
+
+            // Проверка IsSingleton: если задача уже активна — возвращаем её
+            if (taskType.IsSingleton)
+            {
+                var existingTask = await GetActiveSingletonTaskAsync(typeCode);
+
+                if (existingTask != null)
+                {
+                    logger.LogInformation("Задача типа {TypeCode} уже активна (id: {ExistingId}). Возвращаем существующую.", typeCode, existingTask.Id);
+                    return existingTask.Id;
+                }
+            }
+
+            var payloadStr = PayloadToString(payload);
+
+            // Создаем новую запись задачи
+            var task = new TaskItem
+            {
+                Id = Guid.NewGuid(),
+                TypeCode = typeCode,
+                StatusCode = TaskItemStatusEnum.Pending,
+                Payload = payloadStr,
+                RetryCount = 0,
+                ProcessedAt = null
+            };
+
+            if (!await taskRepository.CreateAsync(task) || !await taskRepository.CommitAsync(initiator))
+            {
+                logger.LogError("Ошибка при сохранении задачи в БД");
+                throw new DbUpdateException("Ошибка при сохранении задачи в БД");
+            }
+
+            logger.LogInformation("Задача {TaskId} типа {TypeCode} сохранена в БД со статусом Pending", task.Id, typeCode);
+
+            var message = new TaskMessage { TaskId = task.Id, TypeCode = task.TypeCode };
+
+            var sendResult = await rabbitService.SendAsync(mqSettings, new List<object> { message });
+            if (sendResult.IsSuccess)
+            {
+                logger.LogDebug("Задача {TaskId} типа {TypeCode} отправлена в очередь {QueueName}", task.Id, typeCode, mqSettings.QueueName);
+            }
+            else
+            {
+                // Не пробрасываем исключение дальше — задача создана, просто не в очереди.
+                // задача останется в БД со статусом pending, и Reconciliation Task позже ее возмет в работу
+                logger.LogError("Не удалось опубликовать задачу {TaskId} в очередь. Задача осталась в БД со статусом Pending.", task.Id);
+            }
+
+            return task.Id;
+        }
+
+
+        /// <summary>
+        /// Получить активную singleton задачу указанного типа
+        /// </summary>
+        /// <param name="typeCode"></param>
+        /// <returns></returns>
+        private async Task<TaskItem?> GetActiveSingletonTaskAsync(TaskTypeEnum typeCode)
+        {
+            return await taskRepository.Get().AsNoTracking()
+                                        .FirstOrDefaultAsync(t => t.TypeCode == typeCode && (
+                                            t.StatusCode == TaskItemStatusEnum.Pending
+                                            || t.StatusCode == TaskItemStatusEnum.Processing
+                                        ));
+        }
+
+
+        /// <summary>
+        /// Payload конвертировать в string
+        /// </summary>
+        /// <typeparam name="T"></typeparam>
+        /// <param name="payload"></param>
+        /// <returns></returns>
+        private string? PayloadToString<T>(T payload)
+        {
+            var jsonOptions = new JsonSerializerOptions
+            {
+                Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping
+            };
+
+            if (payload == null)
+                return null;
+
+            return JsonSerializer.Serialize(payload, jsonOptions);
+        }
+
+
+    }
+}

PARR.Core/Services/Task/Interfaces/ITaskManagementService.cs

@@ -0,0 +1,23 @@
+using PARR.Domain.Entities.Base.History;
+using PARR.Domain.Enums;
+using PARR.Domain.Settings;
+
+namespace PARR.Core.Services.Task.Interfaces
+{
+    /// <summary>
+    /// Управления задачами (очередями)
+    /// </summary>
+    public interface ITaskManagementService
+    {
+        /// <summary>
+        /// Создать новую задачу в БД и опубликовать в очередь
+        /// </summary>
+        /// <typeparam name="T"></typeparam>
+        /// <param name="typeCode">Тип задачи</param>
+        /// <param name="payload">Payload</param>
+        /// <param name="initiator">Инициатор</param>
+        /// <param name="mqSettings">Настройки очереди</param>
+        /// <returns></returns>
+        Task<Guid> CreateTaskAsync<T>(TaskTypeEnum typeCode, T payload, IHistoryInitiator initiator, IMqSettings mqSettings);
+    }
+}