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

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

namespace Admin.NET.Plugin.AiDOP.DataPlatform;

/// <summary>
/// 同步任务实体与字段映射配置 API。
/// </summary>
[ApiDescriptionSettings(Order = 324, Description = "数据中台同步实体与字段映射")]
[Route("api/DataPlatform")]
[NonUnify]
public class MdpSyncEntityConfigService : IDynamicApiController, ITransient
{
    private static readonly HashSet<string> ValidFieldTypes = new(StringComparer.OrdinalIgnoreCase)
    {
        "DIRECT", "JSONPATH", "SCRIPT", "CONST", "LOOKUP"
    };

    private readonly ISqlSugarClient _db;
    private readonly UserManager _userManager;
    private readonly MdpSourcePullDispatcher _dispatcher;

    public MdpSyncEntityConfigService(
        ISqlSugarClient db,
        UserManager userManager,
        MdpSourcePullDispatcher dispatcher)
    {
        _db = db;
        _userManager = userManager;
        _dispatcher = dispatcher;
    }

    [DisplayName("同步任务实体列表")]
    [HttpGet("sync-tasks/{taskCode}/entities")]
    public async Task<object> GetEntities(string taskCode)
    {
        var tenantId = _userManager.TenantId;
        var task = await FindTaskAsync(taskCode, tenantId);
        var prefix = ResolveEntityCodePrefix(task.TaskCode);
        var jobCode = string.IsNullOrWhiteSpace(task.JobCode) ? task.TaskCode : task.JobCode;

        var where = new List<string> { "(e.tenant_id = @TenantId OR e.tenant_id = 0)", "e.status = 1" };
        var pars = new List<SugarParameter> { new("@TenantId", tenantId) };

        if (!string.IsNullOrWhiteSpace(prefix))
        {
            where.Add("(e.entity_code LIKE @EntityPrefix OR e.job_id = @JobCode)");
            pars.Add(new SugarParameter("@EntityPrefix", prefix + "%"));
            pars.Add(new SugarParameter("@JobCode", jobCode));
        }
        else
        {
            where.Add("e.job_id = @JobCode");
            pars.Add(new SugarParameter("@JobCode", jobCode));
        }

        var whereSql = string.Join(" AND ", where);
        var list = await _db.Ado.SqlQueryAsync<MdpEntityRow>(
            $"""
            SELECT e.id AS Id, e.tenant_id AS TenantId, e.source_id AS SourceId,
                   s.source_code AS SourceCode, s.source_name AS SourceName, s.source_type AS SourceType,
                   e.entity_code AS EntityCode, e.entity_name AS EntityName, e.entity_type AS EntityType,
                   e.source_table_name AS SourceTableName, e.source_api_path AS SourceApiPath,
                   e.target_table_name AS TargetTableName, e.sync_mode AS SyncMode, e.incr_column AS IncrColumn,
                   e.batch_size AS BatchSize, e.job_id AS JobId, e.status AS Status, e.remark AS Remark,
                   IFNULL(m.cnt, 0) AS FieldMappingCount
            FROM mdp_entity e
            LEFT JOIN mdp_source s ON s.id = e.source_id
            LEFT JOIN (
                SELECT entity_id, COUNT(1) AS cnt FROM mdp_field_mapping GROUP BY entity_id
            ) m ON m.entity_id = e.id
            WHERE {whereSql}
            ORDER BY e.entity_code
            """,
            pars);

        return list;
    }

    [DisplayName("新增同步任务实体")]
    [HttpPost("sync-tasks/{taskCode}/entities")]
    public async Task<object> CreateEntity(string taskCode, [FromBody] MdpEntityUpsertInput input)
    {
        ValidateEntityUpsert(input);
        var tenantId = _userManager.TenantId;
        var task = await FindTaskAsync(taskCode, tenantId);
        var sourceId = await ResolveSourceIdAsync(input, tenantId);
        var entityTenantId = task.TenantId == 0 ? 0L : tenantId;
        var jobCode = string.IsNullOrWhiteSpace(task.JobCode) ? task.TaskCode : task.JobCode;

        var exists = await _db.Queryable<MdpEntity>()
            .Where(u => u.TenantId == entityTenantId && u.EntityCode == input.EntityCode.Trim())
            .AnyAsync();
        if (exists)
            throw Oops.Oh("实体编码已存在");

        var now = DateTime.Now;
        var entity = new MdpEntity
        {
            TenantId = entityTenantId,
            SourceId = sourceId,
            EntityCode = input.EntityCode.Trim(),
            EntityName = input.EntityName.Trim(),
            EntityType = string.IsNullOrWhiteSpace(input.EntityType) ? "TABLE" : input.EntityType.Trim(),
            SourceTableName = input.SourceTableName?.Trim(),
            SourceApiPath = input.SourceApiPath?.Trim(),
            TargetTableName = input.TargetTableName?.Trim(),
            SyncMode = string.IsNullOrWhiteSpace(input.SyncMode) ? "INCR" : input.SyncMode.Trim(),
            IncrColumn = input.IncrColumn?.Trim(),
            BatchSize = input.BatchSize <= 0 ? 5000 : input.BatchSize,
            JobId = string.IsNullOrWhiteSpace(input.JobId) ? jobCode : input.JobId.Trim(),
            Status = input.Status <= 0 ? 0 : 1,
            Remark = input.Remark?.Trim(),
            InboundEnabled = await ResolveInboundEnabledAsync(sourceId, input),
            BizKeyExpr = input.BizKeyExpr?.Trim(),
            CreateTime = now,
            UpdateTime = now
        };

        var id = await _db.Insertable(entity).ExecuteReturnBigIdentityAsync();
        await BumpTaskConfigVersionAsync(task);
        return new { id };
    }

    /// <summary>
    /// 按实体立即拉取一次（源 → 贴源）。
    ///
    /// <para><b>为什么需要它</b>：<see cref="Executors.MdpModuleStagingPuller"/> 的实体清单全部由调用方
    /// 显式传入，各模块只传自己硬编码的那一份（<c>S1MdpEntityConfig.All</c> 等），配置层新增的
    /// 自定义实体没有任何运行时入口，登记完也拉不动。这里补一个受控的通用触发口。</para>
    ///
    /// <para><b>越权面收敛</b>：不把 entityCode 直接透传给 dispatcher，而是先按「归属本任务 +
    /// 当前租户可见 + 已启用」三条件校验，避免借该端点拉取任意实体。</para>
    /// </summary>
    [DisplayName("按实体立即拉取一次")]
    [HttpPost("sync-tasks/{taskCode}/entities/{entityCode}/pull")]
    public async Task<object> PullEntityNow(
        string taskCode,
        string entityCode,
        [FromQuery] bool fullRefresh = false,
        CancellationToken cancellationToken = default)
    {
        var tenantId = _userManager.TenantId;
        // tenantId<=0 时贴源写入的租户解析会回落到 null，整行被静默跳过（只打 Warning），
        // 对外表现为「拉取成功但 0 行」。提前挡住，不给静默失败留口子。
        if (tenantId <= 0)
            throw Oops.Oh("无法解析当前租户，请在具体租户下执行拉取");

        var task = await FindTaskAsync(taskCode, tenantId);
        var jobCode = string.IsNullOrWhiteSpace(task.JobCode) ? task.TaskCode : task.JobCode;
        var code = entityCode.Trim();

        var target = await _db.Queryable<MdpEntity>()
            .Where(u => u.EntityCode == code
                        && u.Status == 1
                        && u.JobId == jobCode
                        && (u.TenantId == tenantId || u.TenantId == 0))
            .FirstAsync(cancellationToken)
            ?? throw Oops.Oh($"实体 {code} 不存在、未启用，或不归属任务 {taskCode}");

        // sync_batch_id 是 varchar(64)：固定段 22 字符 + 实体编码最多截 42 字符，
        // 截尾只落在实体编码上，不动时间戳，避免批次号互相撞。
        var codePart = code.Length > 42 ? code[..42] : code;
        var ctx = new MdpPullContext
        {
            TenantId = tenantId,
            FullRefresh = fullRefresh,
            BatchId = $"MANUAL_{DateTime.Now:yyyyMMddHHmmss}_{codePart}",
            TaskCode = task.TaskCode
        };

        try
        {
            var result = await _dispatcher.PullAllByEntityCodeAsync(code, ctx, cancellationToken);
            return new
            {
                entityCode = code,
                targetTable = target.TargetTableName,
                batchId = ctx.BatchId,
                requestedFullRefresh = fullRefresh,
                // 调度窗口可能把 FULL 落成 ctx.FullRefresh=true，回传实际生效值，不做黑盒
                effectiveWindowType = ctx.SyncWindowType,
                effectiveFullRefresh = ctx.FullRefresh,
                rowsPulled = result.RowsPulled,
                rowsWritten = result.RowsWritten,
                newCursor = result.NewCursor,
                message = result.Message
            };
        }
        catch (InvalidOperationException ex)
        {
            // 贴源契约缺列、源表标识符非法、源不可达等都走这里；转业务异常把原因如实带回前端
            throw Oops.Oh(ex.Message);
        }
    }

    [DisplayName("更新同步实体")]
    [HttpPut("entities/{id:long}")]
    public async Task<object> UpdateEntity(long id, [FromBody] MdpEntityUpsertInput input)
    {
        ValidateEntityUpsert(input);
        var tenantId = _userManager.TenantId;
        var entity = await FindEditableEntityAsync(id, tenantId);

        entity.EntityName = input.EntityName.Trim();
        entity.EntityType = string.IsNullOrWhiteSpace(input.EntityType) ? "TABLE" : input.EntityType.Trim();
        entity.SourceId = await ResolveSourceIdAsync(input, tenantId, entity.SourceId);
        entity.SourceTableName = input.SourceTableName?.Trim();
        entity.SourceApiPath = input.SourceApiPath?.Trim();
        entity.TargetTableName = input.TargetTableName?.Trim();
        entity.SyncMode = string.IsNullOrWhiteSpace(input.SyncMode) ? "INCR" : input.SyncMode.Trim();
        entity.IncrColumn = input.IncrColumn?.Trim();
        entity.BatchSize = input.BatchSize <= 0 ? 5000 : input.BatchSize;
        entity.JobId = input.JobId?.Trim();
        entity.Status = input.Status <= 0 ? 0 : 1;
        entity.Remark = input.Remark?.Trim();
        entity.InboundEnabled = await ResolveInboundEnabledAsync(entity.SourceId, input);
        entity.BizKeyExpr = input.BizKeyExpr?.Trim();
        entity.UpdateTime = DateTime.Now;

        if (!IsBuiltInEntity(entity))
            entity.EntityCode = input.EntityCode.Trim();

        await _db.Updateable(entity).ExecuteCommandAsync();
        return new { id = entity.Id };
    }

    [DisplayName("删除同步实体")]
    [HttpDelete("entities/{id:long}")]
    public async Task<object> DeleteEntity(long id)
    {
        var tenantId = _userManager.TenantId;
        var entity = await FindEditableEntityAsync(id, tenantId);
        if (IsBuiltInEntity(entity) && entity.TenantId == 0)
            throw Oops.Oh("内置 MDP 实体不允许删除");

        entity.Status = 0;
        entity.UpdateTime = DateTime.Now;
        await _db.Updateable(entity).UpdateColumns(u => new { u.Status, u.UpdateTime }).ExecuteCommandAsync();
        return new { id = entity.Id };
    }

    [DisplayName("实体字段映射列表")]
    [HttpGet("entities/{entityId:long}/field-mappings")]
    public async Task<object> GetFieldMappings(long entityId)
    {
        var tenantId = _userManager.TenantId;
        await FindEditableEntityAsync(entityId, tenantId);
        var list = await _db.Queryable<MdpFieldMapping>()
            .Where(u => u.EntityId == entityId)
            .OrderBy(u => u.SortOrder)
            .OrderBy(u => u.TargetField)
            .ToListAsync();
        return list.Select(MapFieldMappingRow).ToList();
    }

    [DisplayName("新增实体字段映射")]
    [HttpPost("entities/{entityId:long}/field-mappings")]
    public async Task<object> CreateFieldMapping(long entityId, [FromBody] MdpFieldMappingUpsertInput input)
    {
        ValidateFieldMappingUpsert(input);
        var tenantId = _userManager.TenantId;
        await FindEditableEntityAsync(entityId, tenantId);

        var exists = await _db.Queryable<MdpFieldMapping>()
            .Where(u => u.EntityId == entityId && u.TargetField == input.TargetField.Trim())
            .AnyAsync();
        if (exists)
            throw Oops.Oh("目标字段映射已存在");

        var entity = MapFieldMappingInsert(entityId, input);
        var id = await _db.Insertable(entity).ExecuteReturnBigIdentityAsync();
        return new { id };
    }

    [DisplayName("更新字段映射")]
    [HttpPut("field-mappings/{id:long}")]
    public async Task<object> UpdateFieldMapping(long id, [FromBody] MdpFieldMappingUpsertInput input)
    {
        ValidateFieldMappingUpsert(input);
        var tenantId = _userManager.TenantId;
        var mapping = await _db.Queryable<MdpFieldMapping>().Where(u => u.Id == id).FirstAsync();
        if (mapping == null)
            throw Oops.Oh("字段映射不存在");

        await FindEditableEntityAsync(mapping.EntityId, tenantId);

        var duplicate = await _db.Queryable<MdpFieldMapping>()
            .Where(u => u.EntityId == mapping.EntityId && u.TargetField == input.TargetField.Trim() && u.Id != id)
            .AnyAsync();
        if (duplicate)
            throw Oops.Oh("目标字段映射已存在");

        mapping.SourceField = input.SourceField.Trim();
        mapping.TargetField = input.TargetField.Trim();
        mapping.FieldType = NormalizeFieldType(input.FieldType);
        mapping.TransformScript = input.TransformScript?.Trim();
        mapping.ConstValue = input.ConstValue?.Trim();
        mapping.LookupTable = input.LookupTable?.Trim();
        mapping.IsRequired = input.IsRequired ? 1 : 0;
        mapping.DefaultValue = input.DefaultValue?.Trim();
        mapping.SortOrder = input.SortOrder;
        await _db.Updateable(mapping).ExecuteCommandAsync();
        return new { id = mapping.Id };
    }

    [DisplayName("删除字段映射")]
    [HttpDelete("field-mappings/{id:long}")]
    public async Task<object> DeleteFieldMapping(long id)
    {
        var tenantId = _userManager.TenantId;
        var mapping = await _db.Queryable<MdpFieldMapping>().Where(u => u.Id == id).FirstAsync();
        if (mapping == null)
            throw Oops.Oh("字段映射不存在");

        await FindEditableEntityAsync(mapping.EntityId, tenantId);
        await _db.Deleteable<MdpFieldMapping>().Where(u => u.Id == id).ExecuteCommandAsync();
        return new { id };
    }

    private async Task<MdpSyncTask> FindTaskAsync(string taskCode, long tenantId)
    {
        if (string.IsNullOrWhiteSpace(taskCode))
            throw Oops.Oh("任务编码不能为空");

        var task = await _db.Queryable<MdpSyncTask>()
            .Where(u => u.TaskCode == taskCode && (u.TenantId == tenantId || u.TenantId == 0) && u.Status == 1)
            .OrderBy(u => u.TenantId, OrderByType.Desc)
            .FirstAsync();
        if (task == null)
            throw Oops.Oh("同步任务不存在");
        return task;
    }

    private async Task<MdpEntity> FindEditableEntityAsync(long id, long tenantId)
    {
        var entity = await _db.Queryable<MdpEntity>()
            .Where(u => u.Id == id && (u.TenantId == tenantId || u.TenantId == 0))
            .FirstAsync();
        if (entity == null)
            throw Oops.Oh("同步实体不存在");
        return entity;
    }

    private async Task<long> ResolveSourceIdAsync(MdpEntityUpsertInput input, long tenantId, long? fallbackSourceId = null)
    {
        if (input.SourceId > 0)
            return input.SourceId;

        if (!string.IsNullOrWhiteSpace(input.SourceCode))
        {
            var sourceId = await _db.Ado.GetLongAsync(
                """
                SELECT id FROM mdp_source
                WHERE source_code = @SourceCode AND (tenant_id = @TenantId OR tenant_id = 0)
                ORDER BY tenant_id DESC
                LIMIT 1
                """,
                new List<SugarParameter>
                {
                    new("@SourceCode", input.SourceCode.Trim()),
                    new("@TenantId", tenantId)
                });
            if (sourceId > 0)
                return sourceId;
            throw Oops.Oh($"数据源 {input.SourceCode} 不存在");
        }

        if (fallbackSourceId is > 0)
            return fallbackSourceId.Value;

        throw Oops.Oh("请指定 sourceId 或 sourceCode");
    }

    private async Task BumpTaskConfigVersionAsync(MdpSyncTask task)
    {
        task.ConfigVersion += 1;
        task.UpdateTime = DateTime.Now;
        await _db.Updateable(task).UpdateColumns(u => new { u.ConfigVersion, u.UpdateTime }).ExecuteCommandAsync();
    }

    private static string? ResolveEntityCodePrefix(string taskCode)
    {
        if (string.IsNullOrWhiteSpace(taskCode))
            return null;

        var first = taskCode.Split('_', StringSplitOptions.RemoveEmptyEntries).FirstOrDefault();
        if (string.IsNullOrWhiteSpace(first) || first.Length < 2 || first[0] != 'S' || !char.IsDigit(first[1]))
            return null;

        return first + "_";
    }

    private static bool IsBuiltInEntity(MdpEntity entity) =>
        entity.EntityCode.StartsWith("S1_", StringComparison.OrdinalIgnoreCase)
        || entity.EntityCode.StartsWith("S2_", StringComparison.OrdinalIgnoreCase)
        || entity.EntityCode.StartsWith("S3_", StringComparison.OrdinalIgnoreCase)
        || entity.EntityCode.StartsWith("S4_", StringComparison.OrdinalIgnoreCase);

    private async Task<int> ResolveInboundEnabledAsync(long sourceId, MdpEntityUpsertInput input)
    {
        if (input.InboundEnabled != 1)
            return 0;
        var sourceType = await _db.Queryable<MdpSource>()
            .Where(s => s.Id == sourceId)
            .Select(s => s.SourceType)
            .FirstAsync();
        if (!string.Equals(sourceType, "API_INBOUND", StringComparison.OrdinalIgnoreCase))
            throw Oops.Oh("只有对方推送来源可以开通入站");
        if (string.IsNullOrWhiteSpace(input.BizKeyExpr))
            throw Oops.Oh("推送实体必须填写业务键");
        return 1;
    }

    private static void ValidateEntityUpsert(MdpEntityUpsertInput input)
    {
        if (string.IsNullOrWhiteSpace(input.EntityCode))
            throw Oops.Oh("实体编码不能为空");
        if (string.IsNullOrWhiteSpace(input.EntityName))
            throw Oops.Oh("实体名称不能为空");
    }

    private static void ValidateFieldMappingUpsert(MdpFieldMappingUpsertInput input)
    {
        if (string.IsNullOrWhiteSpace(input.SourceField))
            throw Oops.Oh("源字段不能为空");
        if (string.IsNullOrWhiteSpace(input.TargetField))
            throw Oops.Oh("目标字段不能为空");
    }

    private static string NormalizeFieldType(string? fieldType)
    {
        var normalized = string.IsNullOrWhiteSpace(fieldType) ? "DIRECT" : fieldType.Trim().ToUpperInvariant();
        return ValidFieldTypes.Contains(normalized) ? normalized : "DIRECT";
    }

    private static MdpFieldMapping MapFieldMappingInsert(long entityId, MdpFieldMappingUpsertInput input) =>
        new()
        {
            EntityId = entityId,
            SourceField = input.SourceField.Trim(),
            TargetField = input.TargetField.Trim(),
            FieldType = NormalizeFieldType(input.FieldType),
            TransformScript = input.TransformScript?.Trim(),
            ConstValue = input.ConstValue?.Trim(),
            LookupTable = input.LookupTable?.Trim(),
            IsRequired = input.IsRequired ? 1 : 0,
            DefaultValue = input.DefaultValue?.Trim(),
            SortOrder = input.SortOrder,
            CreateTime = DateTime.Now
        };

    private static MdpFieldMappingRow MapFieldMappingRow(MdpFieldMapping row) =>
        new()
        {
            Id = row.Id,
            EntityId = row.EntityId,
            SourceField = row.SourceField,
            TargetField = row.TargetField,
            FieldType = row.FieldType,
            TransformScript = row.TransformScript,
            ConstValue = row.ConstValue,
            LookupTable = row.LookupTable,
            IsRequired = row.IsRequired != 0,
            DefaultValue = row.DefaultValue,
            SortOrder = row.SortOrder
        };
}