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

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

namespace Admin.NET.Plugin.AiDOP.DataPlatform;

/// <summary>
/// 数据中台同步配置中心 API。
/// </summary>
[ApiDescriptionSettings(Order = 323, Description = "数据中台同步配置中心")]
[Route("api/DataPlatform")]
[AllowAnonymous]
[NonUnify]
public class MdpSyncTaskConfigService : IDynamicApiController, ITransient
{
    private static readonly HashSet<string> ProtectedTaskCodes = new(StringComparer.OrdinalIgnoreCase)
    {
        "S1_MDP_SYNC_TRANSFORM",
        "S2_MDP_SYNC_TRANSFORM",
        "S3_MDP_SYNC_TRANSFORM",
        "S4_MDP_SYNC_TRANSFORM"
    };

    private readonly ISqlSugarClient _db;
    private readonly UserManager _userManager;

    public MdpSyncTaskConfigService(ISqlSugarClient db, UserManager userManager)
    {
        _db = db;
        _userManager = userManager;
    }

    [DisplayName("同步任务列表")]
    [HttpGet("sync-tasks")]
    public async Task<object> GetList([FromQuery] MdpSyncTaskListQuery input)
    {
        var tenantId = _userManager.TenantId;
        var page = input.Page <= 0 ? 1 : input.Page;
        var pageSize = input.PageSize <= 0 ? 10 : input.PageSize;
        var offset = (page - 1) * pageSize;
        var where = new List<string> { "(t.tenant_id = @TenantId OR t.tenant_id = 0)", "t.status = 1" };
        var pars = new List<SugarParameter> { new("@TenantId", tenantId) };

        if (!string.IsNullOrWhiteSpace(input.Keyword))
        {
            where.Add("(t.task_code LIKE @Keyword OR t.task_name LIKE @Keyword OR t.business_domain_name LIKE @Keyword)");
            pars.Add(new SugarParameter("@Keyword", $"%{input.Keyword.Trim()}%"));
        }
        if (!string.IsNullOrWhiteSpace(input.BusinessDomainCode))
        {
            where.Add("t.business_domain_code = @BusinessDomainCode");
            pars.Add(new SugarParameter("@BusinessDomainCode", input.BusinessDomainCode.Trim()));
        }
        if (input.Status.HasValue)
        {
            where[1] = "t.status = @Status";
            pars.Add(new SugarParameter("@Status", input.Status.Value));
        }

        var whereSql = string.Join(" AND ", where);
        var total = await _db.Ado.GetIntAsync($"SELECT COUNT(1) FROM mdp_sync_task t WHERE {whereSql}", pars);
        var list = await _db.Ado.SqlQueryAsync<MdpSyncTaskListRow>(
            $"""
            SELECT t.id AS Id, t.tenant_id AS TenantId, t.task_code AS TaskCode, t.task_name AS TaskName,
                   t.task_type AS TaskType, t.business_domain_code AS BusinessDomainCode,
                   t.business_domain_name AS BusinessDomainName, t.consumer_modules AS ConsumerModules,
                   t.source_system_code AS SourceSystemCode, t.service_key AS ServiceKey,
                   t.job_code AS JobCode, t.schedule_job_id AS ScheduleJobId, t.status AS Status,
                   t.config_version AS ConfigVersion, t.description AS Description,
                   sch.schedule_mode AS ScheduleMode,
                   IFNULL(sch.auto_enabled, 1) AS AutoEnabled,
                   IFNULL(sch.manual_enabled, 1) AS ManualEnabled,
                   run.status AS LastRunStatus, run.start_time AS LastRunTime
            FROM mdp_sync_task t
            LEFT JOIN mdp_sync_task_schedule sch
              ON sch.tenant_id = t.tenant_id AND sch.task_code = t.task_code
            LEFT JOIN (
                SELECT job_code, MAX(start_time) AS last_run_time
                FROM mdp_transform_run_log
                WHERE {MdpMonitorService.BuildMdpRunLogTenantWhere(tenantId)}
                GROUP BY job_code
            ) latest ON latest.job_code = IFNULL(t.job_code, t.task_code)
            LEFT JOIN mdp_transform_run_log run
              ON run.job_code = latest.job_code AND run.start_time = latest.last_run_time
            WHERE {whereSql}
            ORDER BY t.task_code
            LIMIT {pageSize} OFFSET {offset}
            """,
            pars);

        return new { total, page, pageSize, list };
    }

    [DisplayName("同步任务详情")]
    [HttpGet("sync-tasks/{id:long}")]
    public async Task<object> GetDetail(long id)
    {
        var tenantId = _userManager.TenantId;
        var row = await _db.Ado.SqlQuerySingleAsync<MdpSyncTaskListRow>(
            $"""
            SELECT t.id AS Id, t.tenant_id AS TenantId, t.task_code AS TaskCode, t.task_name AS TaskName,
                   t.task_type AS TaskType, t.business_domain_code AS BusinessDomainCode,
                   t.business_domain_name AS BusinessDomainName, t.consumer_modules AS ConsumerModules,
                   t.source_system_code AS SourceSystemCode, t.service_key AS ServiceKey,
                   t.job_code AS JobCode, t.schedule_job_id AS ScheduleJobId, t.status AS Status,
                   t.config_version AS ConfigVersion, t.description AS Description,
                   sch.schedule_mode AS ScheduleMode,
                   IFNULL(sch.auto_enabled, 1) AS AutoEnabled,
                   IFNULL(sch.manual_enabled, 1) AS ManualEnabled,
                   run.status AS LastRunStatus, run.start_time AS LastRunTime
            FROM mdp_sync_task t
            LEFT JOIN mdp_sync_task_schedule sch
              ON sch.tenant_id = t.tenant_id AND sch.task_code = t.task_code
            LEFT JOIN (
                SELECT job_code, MAX(start_time) AS last_run_time
                FROM mdp_transform_run_log
                WHERE {MdpMonitorService.BuildMdpRunLogTenantWhere(tenantId)}
                GROUP BY job_code
            ) latest ON latest.job_code = IFNULL(t.job_code, t.task_code)
            LEFT JOIN mdp_transform_run_log run
              ON run.job_code = latest.job_code AND run.start_time = latest.last_run_time
            WHERE t.id = @Id AND (t.tenant_id = @TenantId OR t.tenant_id = 0)
            LIMIT 1
            """,
            new SugarParameter("@Id", id),
            new SugarParameter("@TenantId", tenantId));

        if (row == null)
            throw Oops.Oh("同步任务不存在");
        return row;
    }

    [DisplayName("新增同步任务")]
    [HttpPost("sync-tasks")]
    public async Task<object> Create([FromBody] MdpSyncTaskUpsertInput input)
    {
        ValidateUpsert(input);
        var tenantId = _userManager.TenantId;
        var exists = await _db.Queryable<MdpSyncTask>()
            .Where(u => u.TenantId == tenantId && u.TaskCode == input.TaskCode.Trim())
            .AnyAsync();
        if (exists)
            throw Oops.Oh("任务编码已存在");

        var entity = MapUpsert(input, tenantId);
        entity.ConfigVersion = 1;
        entity.CreateTime = DateTime.Now;
        entity.UpdateTime = DateTime.Now;
        var id = await _db.Insertable(entity).ExecuteReturnBigIdentityAsync();
        await EnsureDefaultScheduleAsync(tenantId, entity.TaskCode, entity.ScheduleJobId);
        return new { id };
    }

    [DisplayName("更新同步任务")]
    [HttpPut("sync-tasks/{id:long}")]
    public async Task<object> Update(long id, [FromBody] MdpSyncTaskUpsertInput input)
    {
        ValidateUpsert(input);
        var tenantId = _userManager.TenantId;
        var entity = await FindEditableTaskAsync(id, tenantId);
        entity.TaskName = input.TaskName.Trim();
        entity.TaskType = string.IsNullOrWhiteSpace(input.TaskType) ? "SERVICE_SYNC" : input.TaskType.Trim();
        entity.BusinessDomainCode = input.BusinessDomainCode?.Trim();
        entity.BusinessDomainName = input.BusinessDomainName?.Trim();
        entity.ConsumerModules = input.ConsumerModules?.Trim();
        entity.SourceSystemCode = input.SourceSystemCode?.Trim();
        entity.ServiceKey = input.ServiceKey?.Trim();
        entity.JobCode = string.IsNullOrWhiteSpace(input.JobCode) ? input.TaskCode.Trim() : input.JobCode.Trim();
        entity.ScheduleJobId = input.ScheduleJobId?.Trim();
        entity.Status = input.Status <= 0 ? 0 : 1;
        entity.OwnerRole = input.OwnerRole?.Trim();
        entity.Description = input.Description?.Trim();
        entity.ConfigVersion += 1;
        entity.UpdateTime = DateTime.Now;

        if (!ProtectedTaskCodes.Contains(entity.TaskCode))
            entity.TaskCode = input.TaskCode.Trim();

        await _db.Updateable(entity).ExecuteCommandAsync();
        await SyncScheduleJobIdAsync(entity.TenantId, entity.TaskCode, entity.ScheduleJobId);
        return new { id = entity.Id };
    }

    [DisplayName("删除同步任务")]
    [HttpDelete("sync-tasks/{id:long}")]
    public async Task<object> Delete(long id)
    {
        var tenantId = _userManager.TenantId;
        var entity = await FindEditableTaskAsync(id, tenantId);
        if (ProtectedTaskCodes.Contains(entity.TaskCode) && 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 };
    }

    [DisplayName("同步任务步骤列表")]
    [HttpGet("sync-tasks/{taskCode}/steps")]
    public async Task<object> GetSteps(string taskCode)
    {
        var tenantId = _userManager.TenantId;
        await EnsureTaskExistsAsync(taskCode, tenantId);
        var steps = await _db.Queryable<MdpSyncTaskStep>()
            .Where(u => u.TaskCode == taskCode && (u.TenantId == tenantId || u.TenantId == 0))
            .OrderBy(u => u.SortOrder)
            .ToListAsync();

        return steps.Select(MapStepRow).ToList();
    }

    [DisplayName("保存同步任务步骤")]
    [HttpPut("sync-tasks/{taskCode}/steps")]
    public async Task<object> SaveSteps(string taskCode, [FromBody] List<MdpSyncTaskStepRow> input)
    {
        var tenantId = _userManager.TenantId;
        var task = await EnsureTaskExistsAsync(taskCode, tenantId);
        var rows = input ?? new List<MdpSyncTaskStepRow>();
        var now = DateTime.Now;

        foreach (var row in rows)
        {
            if (string.IsNullOrWhiteSpace(row.StepCode) || string.IsNullOrWhiteSpace(row.StepName))
                throw Oops.Oh("步骤编码和名称不能为空");

            var entity = await _db.Queryable<MdpSyncTaskStep>()
                .Where(u => u.TaskCode == taskCode && u.StepCode == row.StepCode && (u.TenantId == tenantId || u.TenantId == 0))
                .FirstAsync();

            if (entity == null)
            {
                entity = new MdpSyncTaskStep
                {
                    TenantId = task.TenantId == 0 ? 0 : tenantId,
                    TaskCode = taskCode,
                    StepCode = row.StepCode.Trim(),
                    StepName = row.StepName.Trim(),
                    StageType = string.IsNullOrWhiteSpace(row.StageType) ? row.StepCode.Trim() : row.StageType.Trim(),
                    ServiceMethodKey = row.ServiceMethodKey?.Trim(),
                    Enabled = row.Enabled ? 1 : 0,
                    SortOrder = row.SortOrder,
                    Description = row.Description?.Trim(),
                    CreateTime = now,
                    UpdateTime = now
                };
                await _db.Insertable(entity).ExecuteCommandAsync();
            }
            else
            {
                entity.StepName = row.StepName.Trim();
                entity.StageType = string.IsNullOrWhiteSpace(row.StageType) ? row.StepCode.Trim() : row.StageType.Trim();
                entity.ServiceMethodKey = row.ServiceMethodKey?.Trim();
                entity.Enabled = row.Enabled ? 1 : 0;
                entity.SortOrder = row.SortOrder;
                entity.Description = row.Description?.Trim();
                entity.UpdateTime = now;
                await _db.Updateable(entity).ExecuteCommandAsync();
            }
        }

        task.ConfigVersion += 1;
        task.UpdateTime = now;
        await _db.Updateable(task).UpdateColumns(u => new { u.ConfigVersion, u.UpdateTime }).ExecuteCommandAsync();
        return new { count = rows.Count };
    }

    [DisplayName("同步任务调度配置")]
    [HttpGet("sync-tasks/{taskCode}/schedule")]
    public async Task<object> GetSchedule(string taskCode)
    {
        var tenantId = _userManager.TenantId;
        await EnsureTaskExistsAsync(taskCode, tenantId);
        var schedule = await FindScheduleAsync(taskCode, tenantId);
        if (schedule == null)
            return new MdpSyncTaskScheduleRow { TaskCode = taskCode, ScheduleMode = "CRON" };
        return MapScheduleRow(schedule);
    }

    [DisplayName("保存同步任务调度配置")]
    [HttpPut("sync-tasks/{taskCode}/schedule")]
    public async Task<object> SaveSchedule(string taskCode, [FromBody] MdpSyncTaskScheduleRow input)
    {
        var tenantId = _userManager.TenantId;
        var task = await EnsureTaskExistsAsync(taskCode, tenantId);
        var schedule = await FindScheduleAsync(taskCode, tenantId);
        var now = DateTime.Now;

        if (schedule == null)
        {
            schedule = new MdpSyncTaskSchedule
            {
                TenantId = task.TenantId == 0 ? 0 : tenantId,
                TaskCode = taskCode,
                CreateTime = now,
                UpdateTime = now
            };
        }

        schedule.ScheduleJobId = input.ScheduleJobId?.Trim() ?? task.ScheduleJobId;
        schedule.ScheduleMode = string.IsNullOrWhiteSpace(input.ScheduleMode) ? "CRON" : input.ScheduleMode.Trim();
        schedule.CronExpr = input.CronExpr?.Trim();
        schedule.CronDesc = input.CronDesc?.Trim();
        schedule.Timezone = string.IsNullOrWhiteSpace(input.Timezone) ? "Asia/Shanghai" : input.Timezone.Trim();
        schedule.AutoEnabled = input.AutoEnabled ? 1 : 0;
        schedule.ManualEnabled = input.ManualEnabled ? 1 : 0;
        schedule.RetryEnabled = input.RetryEnabled ? 1 : 0;
        schedule.MaxRetryCount = input.MaxRetryCount <= 0 ? 3 : input.MaxRetryCount;
        schedule.RetryIntervalSeconds = input.RetryIntervalSeconds <= 0 ? 300 : input.RetryIntervalSeconds;
        schedule.TimeoutSeconds = input.TimeoutSeconds <= 0 ? 3600 : input.TimeoutSeconds;
        schedule.MisfirePolicy = input.MisfirePolicy?.Trim();
        schedule.SyncWindowType = string.IsNullOrWhiteSpace(input.SyncWindowType) ? "FULL" : input.SyncWindowType.Trim();
        schedule.SyncWindowValue = input.SyncWindowValue?.Trim();
        schedule.LastScheduleTime = input.LastScheduleTime;
        schedule.NextScheduleTime = input.NextScheduleTime;
        schedule.AdminJobConfigJson = input.AdminJobConfigJson;
        schedule.Description = input.Description?.Trim();
        schedule.UpdateTime = now;

        if (schedule.Id <= 0)
            await _db.Insertable(schedule).ExecuteCommandAsync();
        else
            await _db.Updateable(schedule).ExecuteCommandAsync();

        task.ScheduleJobId = schedule.ScheduleJobId;
        task.ConfigVersion += 1;
        task.UpdateTime = now;
        await _db.Updateable(task).UpdateColumns(u => new { u.ScheduleJobId, u.ConfigVersion, u.UpdateTime }).ExecuteCommandAsync();
        return MapScheduleRow(schedule);
    }

    [DisplayName("同步 Admin.NET 调度快照")]
    [HttpPost("sync-tasks/{taskCode}/schedule/sync-admin-job")]
    public async Task<object> SyncAdminJob(string taskCode)
    {
        var tenantId = _userManager.TenantId;
        var task = await EnsureTaskExistsAsync(taskCode, tenantId);
        var schedule = await FindScheduleAsync(taskCode, tenantId) ?? new MdpSyncTaskSchedule
        {
            TenantId = task.TenantId == 0 ? 0 : tenantId,
            TaskCode = taskCode,
            ScheduleJobId = task.ScheduleJobId,
            ScheduleMode = "CRON",
            CreateTime = DateTime.Now,
            UpdateTime = DateTime.Now
        };

        schedule.AdminJobConfigJson = System.Text.Json.JsonSerializer.Serialize(new
        {
            scheduleJobId = schedule.ScheduleJobId ?? task.ScheduleJobId,
            message = "第一阶段仅保存 Admin.NET Job 绑定快照，不写回底层 Cron 配置。"
        });
        schedule.UpdateTime = DateTime.Now;

        if (schedule.Id <= 0)
            await _db.Insertable(schedule).ExecuteCommandAsync();
        else
            await _db.Updateable(schedule).UpdateColumns(u => new { u.AdminJobConfigJson, u.UpdateTime }).ExecuteCommandAsync();

        return new
        {
            taskCode,
            scheduleJobId = schedule.ScheduleJobId ?? task.ScheduleJobId,
            adminJobConfigJson = schedule.AdminJobConfigJson
        };
    }

    private async Task<MdpSyncTask> EnsureTaskExistsAsync(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<MdpSyncTask> FindEditableTaskAsync(long id, long tenantId)
    {
        var entity = await _db.Queryable<MdpSyncTask>()
            .Where(u => u.Id == id && (u.TenantId == tenantId || u.TenantId == 0))
            .FirstAsync();
        if (entity == null)
            throw Oops.Oh("同步任务不存在");
        return entity;
    }

    private async Task<MdpSyncTaskSchedule?> FindScheduleAsync(string taskCode, long tenantId) =>
        await _db.Queryable<MdpSyncTaskSchedule>()
            .Where(u => u.TaskCode == taskCode && (u.TenantId == tenantId || u.TenantId == 0))
            .OrderBy(u => u.TenantId, OrderByType.Desc)
            .FirstAsync();

    private async Task EnsureDefaultScheduleAsync(long tenantId, string taskCode, string? scheduleJobId)
    {
        var exists = await _db.Queryable<MdpSyncTaskSchedule>()
            .Where(u => u.TaskCode == taskCode && (u.TenantId == tenantId || u.TenantId == 0))
            .AnyAsync();
        if (exists)
            return;

        await _db.Insertable(new MdpSyncTaskSchedule
        {
            TenantId = tenantId,
            TaskCode = taskCode,
            ScheduleJobId = scheduleJobId,
            ScheduleMode = "CRON",
            CronDesc = "按 Admin.NET 任务调度执行",
            AutoEnabled = 1,
            ManualEnabled = 1,
            RetryEnabled = 1,
            MaxRetryCount = 3,
            RetryIntervalSeconds = 300,
            TimeoutSeconds = 3600,
            SyncWindowType = "FULL",
            CreateTime = DateTime.Now,
            UpdateTime = DateTime.Now
        }).ExecuteCommandAsync();
    }

    private async Task SyncScheduleJobIdAsync(long tenantId, string taskCode, string? scheduleJobId)
    {
        var schedule = await FindScheduleAsync(taskCode, tenantId);
        if (schedule == null)
        {
            await EnsureDefaultScheduleAsync(tenantId, taskCode, scheduleJobId);
            return;
        }

        schedule.ScheduleJobId = scheduleJobId;
        schedule.UpdateTime = DateTime.Now;
        await _db.Updateable(schedule).UpdateColumns(u => new { u.ScheduleJobId, u.UpdateTime }).ExecuteCommandAsync();
    }

    private static void ValidateUpsert(MdpSyncTaskUpsertInput input)
    {
        if (string.IsNullOrWhiteSpace(input.TaskCode))
            throw Oops.Oh("任务编码不能为空");
        if (string.IsNullOrWhiteSpace(input.TaskName))
            throw Oops.Oh("任务名称不能为空");
    }

    private static MdpSyncTask MapUpsert(MdpSyncTaskUpsertInput input, long tenantId) =>
        new()
        {
            TenantId = tenantId,
            TaskCode = input.TaskCode.Trim(),
            TaskName = input.TaskName.Trim(),
            TaskType = string.IsNullOrWhiteSpace(input.TaskType) ? "SERVICE_SYNC" : input.TaskType.Trim(),
            BusinessDomainCode = input.BusinessDomainCode?.Trim(),
            BusinessDomainName = input.BusinessDomainName?.Trim(),
            ConsumerModules = input.ConsumerModules?.Trim(),
            SourceSystemCode = input.SourceSystemCode?.Trim(),
            ServiceKey = input.ServiceKey?.Trim(),
            JobCode = string.IsNullOrWhiteSpace(input.JobCode) ? input.TaskCode.Trim() : input.JobCode.Trim(),
            ScheduleJobId = input.ScheduleJobId?.Trim(),
            Status = input.Status <= 0 ? 0 : 1,
            OwnerRole = input.OwnerRole?.Trim(),
            Description = input.Description?.Trim()
        };

    private static MdpSyncTaskStepRow MapStepRow(MdpSyncTaskStep step) =>
        new()
        {
            Id = step.Id,
            StepCode = step.StepCode,
            StepName = step.StepName,
            StageType = step.StageType,
            ServiceMethodKey = step.ServiceMethodKey,
            Enabled = step.Enabled != 0,
            SortOrder = step.SortOrder,
            Description = step.Description
        };

    private static MdpSyncTaskScheduleRow MapScheduleRow(MdpSyncTaskSchedule schedule) =>
        new()
        {
            TaskCode = schedule.TaskCode,
            ScheduleJobId = schedule.ScheduleJobId,
            ScheduleMode = schedule.ScheduleMode,
            CronExpr = schedule.CronExpr,
            CronDesc = schedule.CronDesc,
            Timezone = schedule.Timezone,
            AutoEnabled = schedule.AutoEnabled != 0,
            ManualEnabled = schedule.ManualEnabled != 0,
            RetryEnabled = schedule.RetryEnabled != 0,
            MaxRetryCount = schedule.MaxRetryCount,
            RetryIntervalSeconds = schedule.RetryIntervalSeconds,
            TimeoutSeconds = schedule.TimeoutSeconds,
            MisfirePolicy = schedule.MisfirePolicy,
            SyncWindowType = schedule.SyncWindowType,
            SyncWindowValue = schedule.SyncWindowValue,
            LastScheduleTime = schedule.LastScheduleTime,
            NextScheduleTime = schedule.NextScheduleTime,
            AdminJobConfigJson = schedule.AdminJobConfigJson,
            Description = schedule.Description
        };
}