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

using System.Text.Json;
using Admin.NET.Core.Service;
using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;

namespace Admin.NET.Plugin.AiDOP.DataPlatform;

[ApiDescriptionSettings(Order = 322, Description = "数据中台数据库同步操作")]
[Route("api/DataPlatform")]
[NonUnify]
public class MdpDbEntityOperationService : IDynamicApiController, ITransient
{
    private readonly ISqlSugarClient _db;
    private readonly UserManager _user;
    private readonly MdpSourceScopeFactory _scopes;
    private readonly MdpSourcePullDispatcher _dispatcher;
    private readonly MdpPullFieldMappingService _mappings;
    private readonly SysCacheService _cache;

    public MdpDbEntityOperationService(
        ISqlSugarClient db,
        UserManager user,
        MdpSourceScopeFactory scopes,
        MdpSourcePullDispatcher dispatcher,
        MdpPullFieldMappingService mappings,
        SysCacheService cache)
    {
        _db = db;
        _user = user;
        _scopes = scopes;
        _dispatcher = dispatcher;
        _mappings = mappings;
        _cache = cache;
    }

    [DisplayName("数据库同步是否正在运行")]
    [HttpGet("entities/{id:long}/run-state")]
    public async Task<object> RunState(long id)
    {
        await EnsureMenuAsync(MdpDbSyncRules.SyncTaskMenuId);
        var entity = await FindEntityAsync(id);
        var hold = _cache.BeginCacheLock(LockKey(entity), 50, 1000, throwOnFailure: false);
        if (hold == null)
            return new { running = true };
        hold.Dispose();
        return new { running = false };
    }

    [DisplayName("校验数据库同步实体")]
    [HttpPost("entities/{id:long}/validate")]
    public async Task<object> Validate(long id) => await ValidateCoreAsync(id, connect: true);

    [DisplayName("预览数据库同步")]
    [HttpPost("entities/{id:long}/preview")]
    public async Task<object> Preview(long id, [FromBody] MdpDbPreviewInput? input)
    {
        await EnsureMenuAsync(MdpDbSyncRules.SyncTaskMenuId);
        var entity = await FindEntityAsync(id);
        var source = await FindSourceAsync(entity);
        var limit = Math.Clamp(input?.Limit ?? 10, 1, 20);
        var beforeCursor = entity.LastCursor;
        var scope = await _scopes.GetScopeAsync(source);
        scope.Ado.CommandTimeOut = 15;
        var isSqlServer = IsSqlServer(source);
        var table = MdpDbSyncRules.RequireSourceTable(entity.SourceTableName);
        var order = string.IsNullOrWhiteSpace(entity.TieBreakerColumn)
            ? ""
            : " ORDER BY " + Quote(MdpDbSyncRules.RequireIdentifier(entity.TieBreakerColumn, "tie_breaker_column")!, isSqlServer);
        var sql = isSqlServer
            ? $"SELECT TOP {limit} * FROM {table}{order}"
            : $"SELECT * FROM {table}{order} LIMIT {limit}";
        var tableData = await scope.Ado.GetDataTableAsync(sql);
        var maps = await _mappings.GetAsync(entity.Id);
        var rows = new List<object>();
        foreach (System.Data.DataRow row in tableData.Rows)
        {
            var raw = new Dictionary<string, object?>(StringComparer.OrdinalIgnoreCase);
            foreach (System.Data.DataColumn col in tableData.Columns)
                raw[col.ColumnName] = RedactCell(col.ColumnName, row[col] == DBNull.Value ? null : row[col]);
            var mapped = MdpDbSyncRules.Apply(raw, maps);
            rows.Add(new
            {
                raw,
                mapped = mapped.Values,
                bizKey = MdpStagingWriter.BuildBizKey(entity.BizKeyExpr, mapped.Values),
                errors = mapped.Errors
            });
            if (JsonSerializer.Serialize(rows).Length > 65536)
                break;
        }
        await WriteAuditAsync(entity, beforeCursor, beforeCursor, "PREVIEW", input?.Reason ?? "预览");
        var after = await _db.Queryable<MdpEntity>().Where(x => x.Id == entity.Id).Select(x => x.LastCursor).FirstAsync();
        return new { rows, cursorUnchanged = string.Equals(beforeCursor, after, StringComparison.Ordinal), limit };
    }

    [DisplayName("手工执行数据库同步")]
    [HttpPost("entities/{id:long}/run")]
    public async Task<object> Run(long id, [FromBody] MdpDbRunInput input, CancellationToken cancellationToken)
    {
        if (string.IsNullOrWhiteSpace(input.Reason))
            throw Oops.Oh("请填写执行原因");
        var mode = (input.Mode ?? "INCR").Trim().ToUpperInvariant();
        if (mode is not ("FULL" or "INCR" or "ROLLING"))
            throw Oops.Oh("执行模式只能是 FULL、INCR 或 ROLLING");
        return await RunEntityAsync(id, mode, input.Reason.Trim(), "MANUAL", cancellationToken, _user.TenantId);
    }

    public async Task<object> RunEntityAsync(long entityId, string mode, string reason, string trigger, CancellationToken cancellationToken, long tenantId)
    {
        if (string.Equals(trigger, "MANUAL", StringComparison.OrdinalIgnoreCase))
            await EnsureMenuAsync(MdpDbSyncRules.SyncTaskMenuId);
        var entity = await FindEntityAsync(entityId, tenantId);
        var report = await ValidateCoreAsync(entity.Id, connect: true, tenantId, checkMenu: false);
        if (report.Ok == false)
            throw Oops.Oh(string.Join("；", report.Errors.Select(x => x.Message)));
        var hold = _cache.BeginCacheLock(LockKey(entity), 200, 30 * 60 * 1000, throwOnFailure: false);
        if (hold == null)
            throw Oops.Oh("实体正在运行");
        using (hold)
        {
            var oldCursor = entity.LastCursor;
            var ctx = new MdpPullContext
            {
                TenantId = entity.TenantId == 0 ? tenantId : entity.TenantId,
                FullRefresh = mode == "FULL",
                SyncWindowType = mode,
                RequireConfiguredDbSync = true,
                TriggerType = trigger,
                BatchId = $"MDP_{trigger}_{DateTime.Now:yyyyMMddHHmmss}"
            };
            var result = await _dispatcher.PullAllByEntityIdAsync(entity.Id, ctx, cancellationToken);
            await WriteAuditAsync(entity, oldCursor, result.NewCursor, trigger == "MANUAL" ? "RUN" : "SCHEDULE", reason);
            var status = result.Status ?? MdpDbPullExecutor.SyncStatusSuccess;
            if (result.RowsError > 0
                || !string.Equals(status, MdpDbPullExecutor.SyncStatusSuccess, StringComparison.OrdinalIgnoreCase))
            {
                throw new InvalidOperationException(
                    $"实体 {entity.EntityCode} 同步未完整成功：status={status}，错误行={result.RowsError}。本轮不触发中立层投影和模块重算");
            }
            return new
            {
                batchId = ctx.BatchId,
                mode,
                rowsRead = result.RowsPulled,
                rowsWritten = result.RowsWritten,
                rowsError = result.RowsError,
                oldCursor,
                newCursor = result.NewCursor,
                status,
                message = result.Message
            };
        }
    }

    [DisplayName("重置数据库同步水位")]
    [HttpPost("entities/{id:long}/reset-cursor")]
    public async Task<object> ResetCursor(long id, [FromBody] MdpDbResetCursorInput input)
    {
        await EnsureMenuAsync(MdpDbSyncRules.SyncTaskMenuId);
        if (string.IsNullOrWhiteSpace(input.Reason))
            throw Oops.Oh("请填写重置原因");
        var entity = await FindEntityAsync(id);
        var hold = _cache.BeginCacheLock(LockKey(entity), 200, 30 * 60 * 1000, throwOnFailure: false);
        if (hold == null)
            throw Oops.Oh("实体正在运行");
        using (hold)
        {
            var current = await _db.Queryable<MdpEntity>().Where(x => x.Id == entity.Id).Select(x => x.LastCursor).FirstAsync();
            if (!string.Equals(current ?? "", input.ExpectedOldCursor ?? "", StringComparison.Ordinal))
                throw Oops.Oh("水位已变化，请刷新后重试");
            var now = DateTime.Now;
            await _db.Updateable<MdpEntity>()
                .SetColumns(x => new MdpEntity { LastCursor = input.NewCursor, UpdateTime = now })
                .Where(x => x.Id == entity.Id)
                .ExecuteCommandAsync();
            await WriteAuditAsync(entity, current, input.NewCursor, "RESET", input.Reason.Trim());
            return new { oldCursor = current, newCursor = input.NewCursor };
        }
    }

    [DisplayName("水位操作审计")]
    [HttpGet("entities/{id:long}/cursor-audits")]
    public async Task<object> CursorAudits(long id)
    {
        await EnsureMenuAsync(MdpDbSyncRules.SyncTaskMenuId);
        var entity = await FindEntityAsync(id);
        var rows = await _db.Queryable<MdpEntityCursorAudit>()
            .Where(x => x.EntityId == entity.Id && x.TenantId == entity.TenantId)
            .OrderBy(x => x.CreateTime, OrderByType.Desc)
            .Take(100)
            .ToListAsync();
        return new { list = rows };
    }

    private async Task<MdpDbValidateReport> ValidateCoreAsync(long id, bool connect, long? tenantId = null, bool checkMenu = true)
    {
        if (checkMenu)
            await EnsureMenuAsync(MdpDbSyncRules.SyncTaskMenuId);
        var entity = await FindEntityAsync(id, tenantId ?? _user.TenantId);
        var source = await FindSourceAsync(entity);
        var errors = MdpDbSyncRules.ValidateDbEntity(entity, source.SourceType).ToList();
        var warnings = new List<MdpConfigIssue>();
        if (source.Status != 1)
            errors.Add(new MdpConfigIssue("SOURCE_DISABLED", "sourceId", "数据源未启用"));
        if (!string.Equals(source.SourceType, "DB", StringComparison.OrdinalIgnoreCase))
            errors.Add(new MdpConfigIssue("SOURCE_TYPE", "sourceId", "只有数据库来源可以按方式甲校验"));
        if (connect && errors.All(x => x.Code != "SOURCE_TYPE"))
        {
            try
            {
                var scope = await _scopes.GetScopeAsync(source);
                scope.Ado.CommandTimeOut = 15;
                await scope.Ado.GetScalarAsync("SELECT 1");
                await CheckSourceShapeAsync(scope, entity, source, errors, warnings);
            }
            catch (Exception)
            {
                errors.Add(new MdpConfigIssue("SOURCE_CONNECT", "sourceId", "数据库连接失败"));
            }
        }
        await CheckTargetAsync(entity, errors);
        await CheckTenantGrantAsync(entity, source, errors, tenantId ?? _user.TenantId);
        CheckBizKey(entity, errors);
        return new MdpDbValidateReport(errors.Count == 0, errors, warnings, DateTimeOffset.Now);
    }

    private async Task CheckSourceShapeAsync(ISqlSugarClient scope, MdpEntity entity, MdpSource source, List<MdpConfigIssue> errors, List<MdpConfigIssue> warnings)
    {
        var isSqlServer = IsSqlServer(source);
        var simple = SimpleName(entity.SourceTableName ?? "");
        var existsSql = isSqlServer
            ? "SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME=@t"
            : "SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=@t";
        var exists = await scope.Ado.GetIntAsync(existsSql, new SugarParameter("@t", simple));
        if (exists <= 0)
        {
            errors.Add(new MdpConfigIssue("SOURCE_TABLE_MISSING", "sourceTableName", "源表或视图不存在"));
            return;
        }
        var columns = await scope.Ado.SqlQueryAsync<SourceColumnRow>(isSqlServer
            ? "SELECT name AS Name, TYPE_NAME(system_type_id) AS DataType FROM sys.columns WHERE object_id=OBJECT_ID(@t)"
            : "SELECT COLUMN_NAME AS Name, DATA_TYPE AS DataType FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=@t",
            new SugarParameter("@t", simple));
        void Need(string? column, string field, string code)
        {
            if (string.IsNullOrWhiteSpace(column)) return;
            var hit = columns.FirstOrDefault(x => string.Equals(x.Name, column, StringComparison.OrdinalIgnoreCase));
            if (hit == null)
                errors.Add(new MdpConfigIssue(code, field, $"{column} 在源表中不存在"));
            else if (!IsSortable(hit.DataType))
                errors.Add(new MdpConfigIssue(code, field, $"{column} 的类型不能用于排序"));
        }
        if (!string.Equals(entity.SyncMode, "FULL", StringComparison.OrdinalIgnoreCase))
            Need(entity.IncrColumn, "incrColumn", "INCR_COLUMN_MISSING");
        Need(entity.TieBreakerColumn, "tieBreakerColumn", "TIE_BREAKER_MISSING");
        if (!string.IsNullOrWhiteSpace(entity.TieBreakerColumn) && columns.Any(x => string.Equals(x.Name, entity.TieBreakerColumn, StringComparison.OrdinalIgnoreCase)))
        {
            var col = Quote(entity.TieBreakerColumn.Trim(), isSqlServer);
            var table = Quote(simple, isSqlServer);
            var sampleSql = isSqlServer
                ? $"SELECT {col} AS V FROM (SELECT TOP 2000 {col} AS {col} FROM {table}) s GROUP BY {col} HAVING COUNT(*)>1"
                : $"SELECT {col} AS V FROM (SELECT {col} FROM {table} LIMIT 2000) s GROUP BY {col} HAVING COUNT(*)>1";
            var dups = await scope.Ado.SqlQueryAsync<dynamic>(sampleSql);
            if (dups.Count > 0)
                errors.Add(new MdpConfigIssue("TIE_BREAKER_DUPLICATE", "tieBreakerColumn", "稳定排序列在抽样中不唯一"));
            else
                warnings.Add(new MdpConfigIssue("TIE_BREAKER_SAMPLED", "tieBreakerColumn", "唯一性只检查了最多 2000 行"));
        }
    }

    private async Task CheckTargetAsync(MdpEntity entity, List<MdpConfigIssue> errors)
    {
        if (string.IsNullOrWhiteSpace(entity.TargetTableName)) return;
        var cols = await _db.Ado.SqlQueryAsync<string>(
            """
            SELECT COLUMN_NAME FROM information_schema.COLUMNS
            WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=@t
            """,
            new SugarParameter("@t", entity.TargetTableName.Trim()));
        if (cols.Count == 0)
        {
            errors.Add(new MdpConfigIssue("TARGET_TABLE_MISSING", "targetTableName", "目标贴源表不存在"));
            return;
        }
        foreach (var required in MdpDbSyncRules.StagingColumns)
        {
            if (!cols.Any(x => string.Equals(x, required, StringComparison.OrdinalIgnoreCase)))
                errors.Add(new MdpConfigIssue("TARGET_COLUMN_MISSING", "targetTableName", $"目标贴源表缺少 {required}"));
        }
    }

    private async Task CheckTenantGrantAsync(MdpEntity entity, MdpSource source, List<MdpConfigIssue> errors, long actingTenantId)
    {
        var tenantId = entity.TenantId == 0 ? actingTenantId : entity.TenantId;
        var system = string.IsNullOrWhiteSpace(source.SystemCode) ? source.SourceCode : source.SystemCode;
        var count = await _db.Ado.GetIntAsync(
            "SELECT COUNT(*) FROM mdp_tenant_std_source WHERE tenant_id=@t AND source_system=@s",
            new SugarParameter("@t", tenantId),
            new SugarParameter("@s", system));
        if (count <= 0)
            errors.Add(new MdpConfigIssue("TENANT_SOURCE_MISSING", "sourceId", "当前租户未登记该来源"));
    }

    private void CheckBizKey(MdpEntity entity, List<MdpConfigIssue> errors)
    {
        if (string.IsNullOrWhiteSpace(entity.BizKeyExpr)) return;
        foreach (var field in entity.BizKeyExpr.Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries))
        {
            if (!System.Text.RegularExpressions.Regex.IsMatch(field, "^[A-Za-z_][A-Za-z0-9_]*$"))
                errors.Add(new MdpConfigIssue("BIZ_KEY_INVALID", "bizKeyExpr", $"业务键字段 {field} 不合法"));
        }
    }

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

    private async Task<MdpSource> FindSourceAsync(MdpEntity entity)
    {
        return await _db.Queryable<MdpSource>().Where(x => x.Id == entity.SourceId).FirstAsync()
            ?? throw Oops.Oh("数据源不存在");
    }

    private async Task WriteAuditAsync(MdpEntity entity, string? oldCursor, string? newCursor, string action, string reason)
    {
        await _db.Insertable(new MdpEntityCursorAudit
        {
            TenantId = entity.TenantId == 0 ? _user.TenantId : entity.TenantId,
            EntityId = entity.Id,
            EntityCode = entity.EntityCode,
            OldCursor = oldCursor,
            NewCursor = newCursor,
            ActionType = action,
            Reason = reason.Length > 500 ? reason[..500] : reason,
            OperatorId = _user.UserId,
            OperatorName = _user.RealName,
            CreateTime = DateTime.Now
        }).ExecuteCommandAsync();
    }

    private async Task EnsureMenuAsync(long menuId)
    {
        if (_user.SuperAdmin || _user.SysAdmin)
            return;
        var roleIds = await _db.Queryable<SysUserRole>().Where(x => x.UserId == _user.UserId).Select(x => x.RoleId).ToListAsync();
        var allowed = roleIds.Count > 0 && await _db.Queryable<SysRoleMenu>().Where(x => roleIds.Contains(x.RoleId) && x.MenuId == menuId).AnyAsync();
        if (!allowed)
            throw Oops.Oh("没有同步任务权限");
    }

    private static string LockKey(MdpEntity entity) =>
        $"lock:mdp:dbsync:{(entity.TenantId == 0 ? 0 : entity.TenantId)}:{entity.Id}";

    private static bool IsSqlServer(MdpSource source) =>
        string.Equals(source.DbType, "SQLSERVER", StringComparison.OrdinalIgnoreCase)
        || string.Equals(source.DbType, "MSSQL", StringComparison.OrdinalIgnoreCase);

    private static string SimpleName(string table)
    {
        var parts = table.Trim().Trim('[', ']').Split('.', StringSplitOptions.RemoveEmptyEntries);
        return parts.Length == 0 ? table : parts[^1].Trim('[', ']');
    }

    private static string Quote(string name, bool isSqlServer) => isSqlServer ? name : $"`{name}`";

    private static bool IsSortable(string? dataType)
    {
        var type = (dataType ?? "").ToLowerInvariant();
        return type is not ("text" or "blob" or "json" or "image" or "ntext" or "xml" or "geometry");
    }

    private static object? RedactCell(string column, object? value)
    {
        var name = column.ToLowerInvariant();
        if (name.Contains("password") || name.Contains("secret") || name.Contains("token")
            || name.Contains("phone") || name.Contains("mobile") || name.Contains("email") || name.Contains("idcard"))
            return value == null ? null : "***";
        return value is string text ? MdpDbSyncRules.Redact(text) : value;
    }

    private sealed class SourceColumnRow
    {
        public string Name { get; set; } = "";
        public string? DataType { get; set; }
    }
}

public sealed record MdpDbValidateReport(bool Ok, List<MdpConfigIssue> Errors, List<MdpConfigIssue> Warnings, DateTimeOffset CheckedAt);

public sealed class MdpDbPreviewInput
{
    public int? Limit { get; set; }
    public string? Reason { get; set; }
}

public sealed class MdpDbRunInput
{
    public string? Mode { get; set; }
    public string? Reason { get; set; }
}

public sealed class MdpDbResetCursorInput
{
    public string? ExpectedOldCursor { get; set; }
    public string? NewCursor { get; set; }
    public string? Reason { get; set; }
}