AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/Executors/MdpTargetPushDispatcher.cs

using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
using Microsoft.Extensions.Logging;

namespace Admin.NET.Plugin.AiDOP.DataPlatform.Executors;

/// <summary>
/// 出站分发：扫描 mdp_outbox，按 source_type 路由到 API / DB 执行器。
/// </summary>
public sealed class MdpTargetPushDispatcher : ITransient
{
    public const int DefaultTake = 500;
    private const int MaxRetry = 3;

    private readonly ISqlSugarClient _db;
    private readonly MdpApiPushExecutor _api;
    private readonly MdpDbPushExecutor _dbPush;
    private readonly ILogger _logger;

    public MdpTargetPushDispatcher(
        ISqlSugarClient db,
        MdpApiPushExecutor api,
        MdpDbPushExecutor dbPush,
        ILoggerFactory loggerFactory)
    {
        _db = db;
        _api = api;
        _dbPush = dbPush;
        _logger = loggerFactory.CreateLogger(nameof(MdpTargetPushDispatcher));
    }

    public async Task<(int success, int failed, int skipped)> PushPendingAsync(
        int take = DefaultTake, CancellationToken cancellationToken = default)
    {
        var pending = await _db.Queryable<MdpOutbox>()
            .Where(x => x.Status == 0 && x.RetryCount < MaxRetry)
            .OrderBy(x => x.Id)
            .Take(take > 0 ? take : DefaultTake)
            .ToListAsync(cancellationToken);

        var success = 0;
        var failed = 0;
        var skipped = 0;

        foreach (var item in pending)
        {
            cancellationToken.ThrowIfCancellationRequested();
            try
            {
                var sources = await _db.Queryable<MdpSource>()
                    .Where(x => x.SourceCode == item.TargetSourceCode && x.Status == 1)
                    .Take(1)
                    .ToListAsync(cancellationToken);
                var source = sources.FirstOrDefault();

                if (source == null)
                {
                    // 配置类失败：保持 pending，不烧 retry_count，便于配置修复后被 60s 兜底作业捡起
                    await MarkConfigIssueAsync(
                        item,
                        $"目标源 {item.TargetSourceCode} 未启用",
                        cancellationToken);
                    skipped++;
                    continue;
                }

                if (!TryResolve(source, out var executor, out var routeError))
                {
                    await MarkConfigIssueAsync(item, routeError!, cancellationToken);
                    skipped++;
                    continue;
                }

                _logger.LogInformation(
                    "[MdpTargetPushDispatcher] outbox id={Id} source={Source} type={SourceType} via={Executor}",
                    item.Id, source.SourceCode, source.SourceType, executor!.SupportedType);
                var result = await executor.PushAsync(source, item, cancellationToken);
                if (result.Success)
                {
                    await MarkAsync(item, 1, result.ResponseJson, null, cancellationToken);
                    success++;
                }
                else
                {
                    item.RetryCount++;
                    var status = item.RetryCount >= MaxRetry ? 2 : 0;
                    await MarkAsync(item, status, result.ResponseJson, Truncate(result.ErrorMessage, 900), cancellationToken);
                    if (status == 2) failed++; else skipped++;
                }
            }
            catch (Exception ex)
            {
                item.RetryCount++;
                var status = item.RetryCount >= MaxRetry ? 2 : 0;
                await MarkAsync(item, status, null, Truncate(ex.Message, 900), cancellationToken);
                if (status == 2) failed++; else skipped++;
                _logger.LogWarning(ex, "[MdpTargetPushDispatcher] outbox id={Id} failed", item.Id);
            }
        }

        return (success, failed, skipped);
    }

    /// <summary>
    /// 按 source_type / 连接信息解析执行器；无法解析时返回可诊断错误（不再静默落到 API）。
    /// </summary>
    private bool TryResolve(MdpSource source, out IMdpTargetPushExecutor? executor, out string? errorMessage)
    {
        var t = (source.SourceType ?? "").Trim().ToUpperInvariant();
        if (t is "DB" or "SQLSERVER" or "MYSQL" or "DATABASE")
        {
            executor = _dbPush;
            errorMessage = null;
            return true;
        }
        if (t == "API" || t == "HTTP")
        {
            executor = _api;
            errorMessage = null;
            return true;
        }
        // 未标注类型时：有库连接信息则走 DB，否则不再猜 API
        if (!string.IsNullOrWhiteSpace(source.DbHost) && !string.IsNullOrWhiteSpace(source.DbName))
        {
            executor = _dbPush;
            errorMessage = null;
            return true;
        }

        _logger.LogWarning(
            "[MdpTargetPushDispatcher] 路由未解析 source={Code} source_type='{Type}' DbHost='{Host}' DbName='{Name}' ApiBaseUrl='{Api}'",
            source.SourceCode, source.SourceType, source.DbHost, source.DbName, source.ApiBaseUrl);
        executor = null;
        errorMessage =
            $"SOURCE_ROUTE_UNRESOLVED: 源 {source.SourceCode} 既未标注 source_type，也无 DbHost/DbName";
        return false;
    }

    /// <summary>配置类问题：仅写 ErrorMsg/UpdateTime，status 保持 0，retry_count 不变。</summary>
    private async Task MarkConfigIssueAsync(MdpOutbox item, string error, CancellationToken ct)
    {
        item.ErrorMsg = Truncate(error, 900);
        item.UpdateTime = DateTime.Now;
        await _db.Updateable(item)
            .UpdateColumns(x => new { x.ErrorMsg, x.UpdateTime })
            .ExecuteCommandAsync(ct);
    }

    private async Task MarkAsync(MdpOutbox item, int status, string? responseJson, string? error, CancellationToken ct)
    {
        item.Status = status;
        item.ResponseJson = Truncate(responseJson, 4000);
        item.ErrorMsg = error;
        item.UpdateTime = DateTime.Now;
        await _db.Updateable(item)
            .UpdateColumns(x => new { x.Status, x.RetryCount, x.ResponseJson, x.ErrorMsg, x.UpdateTime })
            .ExecuteCommandAsync(ct);
    }

    private static string? Truncate(string? s, int max) =>
        string.IsNullOrEmpty(s) ? s : (s.Length <= max ? s : s[..max]);
}