AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/MdpDbSyncNeutralProjection.cs

using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;

namespace Admin.NET.Plugin.AiDOP.DataPlatform;

/// <summary>
/// 方式甲贴源完成后，复用标准 API 已有的通用贴源到中立层投影。
/// 未在 <see cref="InboundNeutralProjectionService"/> 登记的实体为 no-op，
/// 仍由 S1-S7 原有模块转换处理。
/// </summary>
public sealed class MdpDbSyncNeutralProjection : ITransient
{
    private readonly ISqlSugarClient _db;
    private readonly InboundNeutralProjectionService _projection;

    public MdpDbSyncNeutralProjection(
        ISqlSugarClient db,
        InboundNeutralProjectionService projection)
    {
        _db = db;
        _projection = projection;
    }

    public async Task ProjectAsync(
        IEnumerable<MdpEntity> entities,
        long tenantId,
        CancellationToken cancellationToken = default)
    {
        if (tenantId <= 0)
            throw new InvalidOperationException("方式甲中立层投影要求 tenantId 大于 0");

        var list = entities
            .GroupBy(x => x.EntityCode, StringComparer.OrdinalIgnoreCase)
            .Select(x => x.OrderByDescending(e => e.TenantId).First())
            .ToList();
        var sourceIds = list.Select(x => x.SourceId).Distinct().ToList();
        var sources = await _db.Queryable<MdpSource>()
            .Where(x => sourceIds.Contains(x.Id))
            .ToListAsync(cancellationToken);
        var sourceById = sources.ToDictionary(x => x.Id);

        foreach (var entity in list)
        {
            cancellationToken.ThrowIfCancellationRequested();
            if (!sourceById.TryGetValue(entity.SourceId, out var source))
                throw new InvalidOperationException($"实体 {entity.EntityCode} 的数据源不存在");
            var neutral = MdpSourceIdentity.NeutralCode(source.SystemCode, source.SourceCode);
            var routes = new[] { entity.EntityCode, entity.TransformSourceTable }
                .Where(x => !string.IsNullOrWhiteSpace(x))
                .Select(x => x!.Trim())
                .Distinct(StringComparer.OrdinalIgnoreCase);
            foreach (var route in routes)
            {
                await _projection.ProjectAsync(
                    route,
                    tenantId,
                    neutral,
                    MdpSourceIdentity.DbSync,
                    cancellationToken);
            }
        }
    }
}