refactor(templateMatcher): Переход на Pipeline-архитектуру для SimpleSync и GroupedSync.
- SimpleTemplateSynchronizer и GroupedTemplateSynchronizer переведены на паттерн Pipeline с разделением на Read/Write этапы - Выделены контракты этапов (ISimpleSyncStage, IGroupedSyncStage) и контексты (SimpleSyncContext, GroupedSyncContext) - Read-этапы безопасны для тестов (не пишут в БД/MQ), Write-этапы изолированы через отдельные интерфейсы - Добавлено [Perf]-логирование каждого этапа с метриками времени выполнения - Логи приведены к человекочитаемому формату 'Имя' (ID) для Job, JobGroup и Unit - Устранено дублирование данных в контекстах (FilteredUnits перезаписывается, TemplateGroups строго типизирован) - Константы неиспользуемых шаблонов вынесены в UnusedTemplateConstants - Структура проекта реорганизована: SimpleSync, GroupedSync, Implementations, Interfaces
This commit is contained in:
Mikhail Kuznetsov
@@ -0,0 +1,272 @@
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using PARR.Core.Common.Interfaces.RabbitServices;
|
||||
using PARR.Core.Repositories.Interfaces;
|
||||
using PARR.Core.Repositories.Interfaces.Unit;
|
||||
using PARR.Domain.Common.Rabbit.Messages.TemplateMatching;
|
||||
using PARR.Domain.Entities;
|
||||
using PARR.Domain.Entities.Base.History;
|
||||
using PARR.Domain.Entities.Job;
|
||||
using PARR.Domain.Enums;
|
||||
using PARR.TemplateMatcher.Models;
|
||||
using PARR.TemplateMatcher.Services.Interfaces;
|
||||
using PARR.TemplateMatcher.Settings;
|
||||
|
||||
namespace PARR.TemplateMatcher.Services.GroupedSync;
|
||||
|
||||
internal class GroupedTemplateProcessor : IGroupedTemplateProcessor
|
||||
{
|
||||
private readonly ILogger<GroupedTemplateProcessor> logger;
|
||||
private readonly ITemplateRepository templateRepository;
|
||||
private readonly IUnitRepository unitRepository;
|
||||
private readonly ITemplateNameNormalizer templateNameNormalizer;
|
||||
private readonly ITemplateAllocationService templateAllocationService;
|
||||
private readonly ITemplateMqPublisher templateMqPublisher;
|
||||
|
||||
public GroupedTemplateProcessor(
|
||||
ILogger<GroupedTemplateProcessor> logger,
|
||||
ITemplateRepository templateRepository,
|
||||
IUnitRepository unitRepository,
|
||||
ITemplateNameNormalizer templateNameNormalizer,
|
||||
ITemplateAllocationService templateAllocationService,
|
||||
ITemplateMqPublisher templateMqPublisher,
|
||||
MqSettings mqSettings,
|
||||
IRabbitService mqService)
|
||||
{
|
||||
this.logger = logger;
|
||||
this.templateRepository = templateRepository;
|
||||
this.unitRepository = unitRepository;
|
||||
this.templateNameNormalizer = templateNameNormalizer;
|
||||
this.templateAllocationService = templateAllocationService;
|
||||
this.templateMqPublisher = templateMqPublisher;
|
||||
}
|
||||
|
||||
public async Task<HashSet<(Guid JobId, Guid UnitId, int Index)>> ProcessAsync(
|
||||
List<GroupedTemplateGroup> groups,
|
||||
List<Job> jobsInGroup,
|
||||
Job maxJob,
|
||||
HistoryInitiator initiator,
|
||||
CancellationToken ct = default)
|
||||
{
|
||||
var expectedTemplateKeys = new HashSet<(Guid JobId, Guid UnitId, int Index)>();
|
||||
|
||||
foreach (var group in groups)
|
||||
{
|
||||
var potentialUnitId = group.PotentialUnitId;
|
||||
|
||||
foreach (var subGroup in group.SubGroups)
|
||||
{
|
||||
var unitsInTemplateSubGroup = subGroup.Entries;
|
||||
var globalIndex = subGroup.GlobalIndex;
|
||||
var originatingInnerGroupName = subGroup.InnerGroupName;
|
||||
|
||||
logger.LogDebug("Обработка подгруппы {Index} ('{GroupingValue}') для UnitId {PotentialUnitId}, размер {Size}.",
|
||||
globalIndex, originatingInnerGroupName, potentialUnitId, unitsInTemplateSubGroup.Count);
|
||||
|
||||
Job targetJob = SelectTargetJob(jobsInGroup, unitsInTemplateSubGroup.Count, maxJob);
|
||||
expectedTemplateKeys.Add((targetJob.Id, potentialUnitId, globalIndex));
|
||||
|
||||
// Поиск существующего шаблона
|
||||
var existingTemplate = await templateRepository.Get()
|
||||
.AsNoTracking()
|
||||
.Include(t => t.Unit)
|
||||
.Include(t => t.Job).ThenInclude(t => t!.Tnk)
|
||||
.Include(t => t.Job).ThenInclude(t => t!.Group).ThenInclude(t => t!.GroupType)
|
||||
.Include(t => t.UnitsInTemplate).ThenInclude(uit => uit.Unit)
|
||||
.Where(t => t.JobId == targetJob.Id &&
|
||||
t.UnitId == potentialUnitId &&
|
||||
t.Index == globalIndex &&
|
||||
t.StatusTypeId == TemplateStatusTypeEnum.Used)
|
||||
.FirstOrDefaultAsync(ct);
|
||||
|
||||
if (existingTemplate != null)
|
||||
{
|
||||
await HandleExistingTemplateAsync(existingTemplate, unitsInTemplateSubGroup, targetJob, globalIndex, initiator, ct);
|
||||
}
|
||||
else
|
||||
{
|
||||
await HandleNewOrReusableTemplateAsync(potentialUnitId, unitsInTemplateSubGroup, targetJob, globalIndex, initiator, ct);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return expectedTemplateKeys;
|
||||
}
|
||||
|
||||
private async Task HandleExistingTemplateAsync(
|
||||
Template existingTemplate,
|
||||
List<(Guid UnitId, Guid UnitFieldValueId)> proposedEntries,
|
||||
Job targetJob,
|
||||
int globalIndex,
|
||||
HistoryInitiator initiator,
|
||||
CancellationToken ct)
|
||||
{
|
||||
var currentEntries = existingTemplate.UnitsInTemplate
|
||||
.Select(uit => (uit.UnitId, uit.UnitFieldValueId))
|
||||
.ToList();
|
||||
|
||||
// Сравнение
|
||||
var allUnitIdsForSort = currentEntries.Select(e => e.UnitId)
|
||||
.Concat(proposedEntries.Select(e => e.UnitId))
|
||||
.Distinct()
|
||||
.ToList();
|
||||
|
||||
var unitNamesForSort = await unitRepository.Get()
|
||||
.AsNoTracking()
|
||||
.Where(u => allUnitIdsForSort.Contains(u.Id))
|
||||
.ToDictionaryAsync(u => u.Id, u => u.Name ?? u.Id.ToString(), ct);
|
||||
|
||||
var sortedCurrent = currentEntries
|
||||
.OrderBy(e => unitNamesForSort.GetValueOrDefault(e.UnitId, e.UnitId.ToString()))
|
||||
.ThenBy(e => e.UnitFieldValueId)
|
||||
.ToList();
|
||||
|
||||
var sortedProposed = proposedEntries
|
||||
.OrderBy(e => unitNamesForSort.GetValueOrDefault(e.UnitId, e.UnitId.ToString()))
|
||||
.ThenBy(e => e.UnitFieldValueId)
|
||||
.ToList();
|
||||
|
||||
bool unitsAreEqual = sortedCurrent.SequenceEqual(sortedProposed);
|
||||
|
||||
if (unitsAreEqual)
|
||||
{
|
||||
logger.LogDebug("Шаблон {TemplateId} актуален по составу.", existingTemplate.Id);
|
||||
|
||||
// Проверка имени
|
||||
var expectedName = await templateNameNormalizer.GetNormalizedTemplateNameAsync(existingTemplate);
|
||||
if (!string.Equals(existingTemplate.Name, expectedName, StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
logger.LogInformation("Шаблон {TemplateId} требует обновления имени.", existingTemplate.Id);
|
||||
var updateRequest = new TemplateUpdaterMessage
|
||||
{
|
||||
TemplateId = existingTemplate.Id,
|
||||
JobId = targetJob.Id,
|
||||
UnitId = existingTemplate.UnitId,
|
||||
Name = expectedName,
|
||||
IsActiveTemplate = existingTemplate.IsActiveTemplate,
|
||||
IsActiveSchedule = existingTemplate.IsActiveSchedule,
|
||||
IsNew = false,
|
||||
Index = globalIndex,
|
||||
StatusTypeId = TemplateStatusTypeEnum.Used,
|
||||
Initiator = initiator,
|
||||
UnitsInTemplate = sortedProposed.Select(t => new UnitInTemplateMessage
|
||||
{
|
||||
UnitId = t.UnitId,
|
||||
UnitFieldValueId = t.UnitFieldValueId
|
||||
}).ToList()
|
||||
};
|
||||
await templateMqPublisher.PublishUpdateAsync(updateRequest);
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
logger.LogInformation("Шаблон {TemplateId} требует обновления состава.", existingTemplate.Id);
|
||||
await UpdateTemplateUnitsAsync(existingTemplate, sortedProposed, targetJob, globalIndex, initiator);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleNewOrReusableTemplateAsync(
|
||||
Guid potentialUnitId,
|
||||
List<(Guid UnitId, Guid UnitFieldValueId)> unitsInTemplateSubGroup,
|
||||
Job targetJob,
|
||||
int globalIndex,
|
||||
HistoryInitiator initiator,
|
||||
CancellationToken ct)
|
||||
{
|
||||
// Вычисляем флаги активности
|
||||
var isActiveTemplate = targetJob.AutoControl?.InitUsedTemplateState ?? false;
|
||||
var isActiveSchedule = targetJob.AutoControl?.InitUsedScheduleState ?? false;
|
||||
|
||||
// Маппим кортежи в сообщения
|
||||
var unitsInTemplateMsg = unitsInTemplateSubGroup
|
||||
.Select(e => new UnitInTemplateMessage
|
||||
{
|
||||
UnitId = e.UnitId,
|
||||
UnitFieldValueId = e.UnitFieldValueId
|
||||
})
|
||||
.ToList();
|
||||
|
||||
var request = new TemplateAllocationRequest(
|
||||
TargetJob: targetJob,
|
||||
TargetUnitId: potentialUnitId,
|
||||
TargetUnit: null,
|
||||
Index: globalIndex,
|
||||
UnitsInTemplate: unitsInTemplateMsg,
|
||||
IsActiveTemplate: isActiveTemplate,
|
||||
IsActiveSchedule: isActiveSchedule,
|
||||
Initiator: initiator);
|
||||
|
||||
await templateAllocationService.AllocateAsync(request, ct);
|
||||
}
|
||||
|
||||
private async Task UpdateTemplateUnitsAsync(
|
||||
Template template,
|
||||
List<(Guid UnitId, Guid UnitFieldValueId)> newUnitEntries,
|
||||
Job targetJob,
|
||||
int newIndex,
|
||||
HistoryInitiator initiator)
|
||||
{
|
||||
// 1. Устанавливаем статус и дату
|
||||
template.StatusTypeId = TemplateStatusTypeEnum.Updating;
|
||||
template.DateModified = DateTimeOffset.UtcNow;
|
||||
|
||||
// 2. КОММИТ В БАЗУ СРАЗУ
|
||||
// Важно зафиксировать изменение статуса до отправки сообщения в очередь
|
||||
if (!await templateRepository.CommitAsync(initiator))
|
||||
{
|
||||
logger.LogError("Не удалось перевести шаблон {TemplateId} в Updating.", template.Id);
|
||||
return;
|
||||
}
|
||||
|
||||
// 3. Формируем временный объект для генерации имени
|
||||
var tempTemplateForName = new Template
|
||||
{
|
||||
Id = template.Id,
|
||||
Name = template.Name,
|
||||
JobId = targetJob.Id,
|
||||
UnitId = template.UnitId,
|
||||
Index = newIndex,
|
||||
Job = targetJob,
|
||||
Unit = template.Unit,
|
||||
UnitsInTemplate = newUnitEntries.Select(e => new UnitsInTemplate { UnitId = e.UnitId, UnitFieldValueId = e.UnitFieldValueId }).ToList()
|
||||
};
|
||||
|
||||
var expectedName = await templateNameNormalizer.GetNormalizedTemplateNameAsync(tempTemplateForName);
|
||||
|
||||
// 4. Отправляем сообщение в очередь
|
||||
var updateRequest = new TemplateUpdaterMessage
|
||||
{
|
||||
TemplateId = template.Id,
|
||||
JobId = targetJob.Id,
|
||||
UnitId = template.UnitId,
|
||||
Name = expectedName,
|
||||
IsActiveTemplate = template.IsActiveTemplate,
|
||||
IsActiveSchedule = template.IsActiveSchedule,
|
||||
IsNew = false,
|
||||
Index = newIndex,
|
||||
StatusTypeId = TemplateStatusTypeEnum.Used,
|
||||
Initiator = initiator,
|
||||
UnitsInTemplate = newUnitEntries.Select(e => new UnitInTemplateMessage { UnitId = e.UnitId, UnitFieldValueId = e.UnitFieldValueId }).ToList()
|
||||
};
|
||||
|
||||
await templateMqPublisher.PublishUpdateAsync(updateRequest);
|
||||
}
|
||||
|
||||
|
||||
private static Job SelectTargetJob(List<Job> jobsInGroup, int subGroupSize, Job maxJob)
|
||||
{
|
||||
Job? targetJob = jobsInGroup
|
||||
.Where(j => j.MaxValueRelationships.HasValue && j.MaxValueRelationships.Value == subGroupSize)
|
||||
.FirstOrDefault();
|
||||
|
||||
if (targetJob == null)
|
||||
{
|
||||
targetJob = jobsInGroup
|
||||
.Where(j => j.MaxValueRelationships.HasValue && j.MaxValueRelationships.Value >= subGroupSize)
|
||||
.OrderBy(j => j.MaxValueRelationships!.Value)
|
||||
.FirstOrDefault();
|
||||
}
|
||||
|
||||
return targetJob ?? maxJob;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user