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;
    private readonly MdpPullFieldMappingService _fieldMappings;

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

    [DisplayName("同步任务实体列表")]
    [HttpGet("sync-tasks/{taskCode}/entities")]
    public async Task<object> GetEntities(string taskCode, [FromQuery] int? status = null)
    {
        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)" };
        var pars = new List<SugarParameter> { new("@TenantId", tenantId) };
        if (status is 0 or 1)
        {
            where.Add("e.status = @Status");
            pars.Add(new SugarParameter("@Status", status.Value));
        }

        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.transform_source_table AS TransformSourceTable,
                   e.source_api_path AS SourceApiPath,
                   e.target_table_name AS TargetTableName, e.sync_mode AS SyncMode, e.incr_column AS IncrColumn,
                   e.tie_breaker_column AS TieBreakerColumn, e.biz_key_expr AS BizKeyExpr,
                   e.batch_size AS BatchSize, e.job_id AS JobId, e.status AS Status,
                   e.last_cursor AS LastCursor, e.last_sync_to AS LastSyncTo, 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);

        foreach (var row in list)
            ApplyConfigState(row);
        return list;
    }

    [DisplayName("数据库同步实体模板")]
    [HttpGet("sync-tasks/{taskCode}/entity-templates")]
    public async Task<object> GetEntityTemplates(string taskCode, [FromQuery] string? moduleCode = null)
    {
        var tenantId = _userManager.TenantId;
        await FindTaskAsync(taskCode, tenantId);
        var module = (moduleCode ?? "").Trim().ToUpperInvariant();
        var whereModule = module.Length == 2 && module[0] == 'S' && char.IsDigit(module[1])
            ? " AND e.entity_code LIKE @ModulePrefix"
            : "";
        var parameters = string.IsNullOrEmpty(whereModule)
            ? Array.Empty<SugarParameter>()
            : [new SugarParameter("@ModulePrefix", module + "_%")];
        var rows = 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,
                   COALESCE(NULLIF(e.transform_source_table,''), e.source_table_name) AS TransformSourceTable,
                   e.source_api_path AS SourceApiPath, e.target_table_name AS TargetTableName,
                   e.sync_mode AS SyncMode, e.incr_column AS IncrColumn,
                   e.tie_breaker_column AS TieBreakerColumn, e.biz_key_expr AS BizKeyExpr,
                   e.batch_size AS BatchSize, e.job_id AS JobId, e.status AS Status,
                   e.last_cursor AS LastCursor, e.last_sync_to AS LastSyncTo, 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 e.tenant_id=0
              AND e.source_table_name IS NOT NULL AND e.source_table_name<>''
              AND e.target_table_name LIKE 'mdp_stg_%'
              {whereModule}
            ORDER BY e.entity_code
            """,
            parameters);
        return rows;
    }

    [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;
        MdpEntity? template = null;
        List<MdpFieldMapping> templateMappings = [];
        if (input.TemplateEntityId is > 0)
        {
            template = await _db.Queryable<MdpEntity>()
                .Where(x => x.Id == input.TemplateEntityId.Value
                            && x.TenantId == 0
                            && x.SourceTableName != null
                            && x.SourceTableName != ""
                            && x.TargetTableName != null
                            && x.TargetTableName.StartsWith("mdp_stg_"))
                .FirstAsync()
                ?? throw Oops.Oh("内置实体模板不存在或不可复制");
            if (!string.Equals(template.EntityCode, input.EntityCode.Trim(), StringComparison.OrdinalIgnoreCase))
                throw Oops.Oh("实体编码必须与所选内置模板一致");
            templateMappings = await _db.Queryable<MdpFieldMapping>()
                .Where(x => x.EntityId == template.Id)
                .OrderBy(x => x.SortOrder)
                .ToListAsync();
        }

        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 = template?.EntityType
                ?? (string.IsNullOrWhiteSpace(input.EntityType) ? "TABLE" : input.EntityType.Trim()),
            SourceTableName = input.SourceTableName?.Trim(),
            TransformSourceTable = !string.IsNullOrWhiteSpace(input.TransformSourceTable)
                ? input.TransformSourceTable.Trim()
                : template == null
                    ? input.SourceTableName?.Trim()
                    : (string.IsNullOrWhiteSpace(template.TransformSourceTable)
                        ? template.SourceTableName
                        : template.TransformSourceTable),
            SourceApiPath = input.SourceApiPath?.Trim(),
            TargetTableName = template?.TargetTableName ?? input.TargetTableName?.Trim(),
            SyncMode = NormalizeSyncMode(input.SyncMode),
            IncrColumn = input.IncrColumn?.Trim(),
            TieBreakerColumn = input.TieBreakerColumn?.Trim(),
            BatchSize = input.BatchSize is > 0 and <= 10000 ? input.BatchSize.Value : 5000,
            JobId = string.IsNullOrWhiteSpace(input.JobId) ? jobCode : input.JobId.Trim(),
            Status = input.Status <= 0 ? 0 : 1,
            Remark = input.Remark?.Trim(),
            InboundEnabled = await ResolveInboundEnabledAsync(sourceId, input.InboundEnabled, 0, input.BizKeyExpr),
            BizKeyExpr = !string.IsNullOrWhiteSpace(input.BizKeyExpr)
                ? input.BizKeyExpr.Trim()
                : template?.BizKeyExpr,
            CreateTime = now,
            UpdateTime = now
        };
        await EnsureRunnableAsync(entity);

        long id = 0;
        var tran = await _db.Ado.UseTranAsync(async () =>
        {
            id = await _db.Insertable(entity).ExecuteReturnBigIdentityAsync();
            if (templateMappings.Count > 0)
            {
                var copies = templateMappings.Select(x => new MdpFieldMapping
                {
                    EntityId = id,
                    SourceField = x.SourceField,
                    TargetField = x.TargetField,
                    FieldType = x.FieldType,
                    TransformScript = x.TransformScript,
                    ConstValue = x.ConstValue,
                    LookupTable = x.LookupTable,
                    IsRequired = x.IsRequired,
                    DefaultValue = x.DefaultValue,
                    SortOrder = x.SortOrder,
                    CreateTime = now
                }).ToList();
                await _db.Insertable(copies).ExecuteCommandAsync();
            }
            await BumpTaskConfigVersionAsync(task);
        });
        if (!tran.IsSuccess)
            throw tran.ErrorException ?? Oops.Oh("复制内置实体模板失败");
        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))
            .OrderBy(u => u.TenantId, OrderByType.Desc)
            .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.TransformSourceTable = string.IsNullOrWhiteSpace(input.TransformSourceTable)
            ? (string.IsNullOrWhiteSpace(entity.TransformSourceTable)
                ? input.SourceTableName?.Trim()
                : entity.TransformSourceTable)
            : input.TransformSourceTable.Trim();
        entity.SourceApiPath = input.SourceApiPath?.Trim();
        entity.TargetTableName = input.TargetTableName?.Trim();
        entity.SyncMode = string.IsNullOrWhiteSpace(input.SyncMode) ? entity.SyncMode : NormalizeSyncMode(input.SyncMode);
        if (input.IncrColumn != null)
            entity.IncrColumn = string.IsNullOrWhiteSpace(input.IncrColumn) ? null : input.IncrColumn.Trim();
        if (input.TieBreakerColumn != null)
            entity.TieBreakerColumn = string.IsNullOrWhiteSpace(input.TieBreakerColumn) ? null : input.TieBreakerColumn.Trim();
        if (input.BatchSize is > 0 and <= 10000)
            entity.BatchSize = input.BatchSize.Value;
        if (!string.IsNullOrWhiteSpace(input.JobId))
            entity.JobId = input.JobId.Trim();
        entity.Status = input.Status <= 0 ? 0 : 1;
        entity.Remark = input.Remark?.Trim();
        entity.InboundEnabled = await ResolveInboundEnabledAsync(entity.SourceId, input.InboundEnabled, entity.InboundEnabled, input.BizKeyExpr ?? entity.BizKeyExpr);
        if (input.BizKeyExpr != null)
            entity.BizKeyExpr = string.IsNullOrWhiteSpace(input.BizKeyExpr) ? null : input.BizKeyExpr.Trim();
        entity.UpdateTime = DateTime.Now;

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

        await EnsureRunnableAsync(entity);
        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);
        await RejectUnsupportedDbMappingAsync(entityId, input.FieldType);

        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();
        _fieldMappings.Invalidate(entityId);
        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);
        await RejectUnsupportedDbMappingAsync(mapping.EntityId, input.FieldType);

        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();
        _fieldMappings.Invalidate(mapping.EntityId);
        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();
        _fieldMappings.Invalidate(mapping.EntityId);
        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 (!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 (input.SourceId > 0)
            return input.SourceId;

        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, int? requested, int existing, string? bizKeyExpr)
    {
        var sourceType = await _db.Queryable<MdpSource>()
            .Where(s => s.Id == sourceId)
            .Select(s => s.SourceType)
            .FirstAsync();
        if (string.Equals(sourceType, "DB", StringComparison.OrdinalIgnoreCase))
            return existing;
        if (requested == null)
            return existing;
        if (requested != 1)
            return 0;
        if (!string.Equals(sourceType, "API_INBOUND", StringComparison.OrdinalIgnoreCase))
            throw Oops.Oh("只有对方推送来源可以开通入站");
        if (string.IsNullOrWhiteSpace(bizKeyExpr))
            throw Oops.Oh("推送实体必须填写业务键");
        return 1;
    }

    private async Task EnsureRunnableAsync(MdpEntity entity)
    {
        if (entity.Status != 1)
            return;
        var sourceType = await _db.Queryable<MdpSource>()
            .Where(s => s.Id == entity.SourceId)
            .Select(s => s.SourceType)
            .FirstAsync();
        var issues = MdpDbSyncRules.ValidateDbEntity(entity, sourceType);
        if (issues.Count > 0)
            throw Oops.Oh(string.Join("；", issues.Select(x => x.Message)));
    }

    private static void ApplyConfigState(MdpEntityRow row)
    {
        var probe = new MdpEntity
        {
            SourceTableName = row.SourceTableName,
            TargetTableName = row.TargetTableName,
            SyncMode = row.SyncMode,
            BizKeyExpr = row.BizKeyExpr,
            BatchSize = row.BatchSize,
            TieBreakerColumn = row.TieBreakerColumn,
            IncrColumn = row.IncrColumn
        };
        var issues = MdpDbSyncRules.ValidateDbEntity(probe, row.SourceType);
        row.ConfigErrors = issues.Select(x => x.Message).ToList();
        row.ConfigState = row.Status != 1 ? "DISABLED" : issues.Count == 0 ? "COMPLETE" : "INCOMPLETE";
    }

    private static string NormalizeSyncMode(string? syncMode)
    {
        var mode = string.IsNullOrWhiteSpace(syncMode) ? "INCR" : syncMode.Trim().ToUpperInvariant();
        return mode is "FULL" or "INCR" or "ROLLING" ? mode : mode;
    }

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

    private async Task RejectUnsupportedDbMappingAsync(long entityId, string? fieldType)
    {
        var normalized = NormalizeFieldType(fieldType);
        if (normalized is not ("SCRIPT" or "LOOKUP"))
            return;
        var sourceType = await _db.Ado.GetStringAsync(
            """
            SELECT s.source_type FROM mdp_entity e
            INNER JOIN mdp_source s ON s.id = e.source_id
            WHERE e.id = @id
            """,
            new SugarParameter("@id", entityId));
        if (string.Equals(sourceType, "DB", StringComparison.OrdinalIgnoreCase))
            throw Oops.Oh("数据库同步不支持 SCRIPT 或 LOOKUP 映射");
    }

    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
        };
}