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)
                {
                    await MarkAsync(item, 2, null, $"目标源 {item.TargetSourceCode} 未启用", cancellationToken);
                    failed++;
                    continue;
                }

                var executor = Resolve(source);
                _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);
    }

    private IMdpTargetPushExecutor Resolve(MdpSource source)
    {
        var t = (source.SourceType ?? "").Trim().ToUpperInvariant();
        if (t is "DB" or "SQLSERVER" or "MYSQL" or "DATABASE")
            return _dbPush;
        if (t == "API" || t == "HTTP")
            return _api;
        // 未标注类型时：有库连接信息则走 DB，否则走 API（兼容历史 HTTP 源）
        if (!string.IsNullOrWhiteSpace(source.DbHost) && !string.IsNullOrWhiteSpace(source.DbName))
            return _dbPush;
        return _api;
    }

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