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

using Admin.NET.Plugin.AiDOP.DataPlatform.HotWatch;
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 执行器。
/// WP10 S4b / D12：退避 1m/5m/15m/30m/60m/120m，MaxRetry=6；配置类失败不烧重试。
/// P-029：MarkAsync 按 UpdateTime 乐观锁，避免与 TryEnqueueOrRefreshAsync 并发整列覆盖。
/// </summary>
public sealed class MdpTargetPushDispatcher : ITransient
{
    public const int DefaultTake = 500;
    private const int MaxRetry = 6;

    /// <summary>D12：按 retry_count 下标取退避分钟数（clamp 到末项）。</summary>
    private static readonly int[] BackoffMinutes = [1, 5, 15, 30, 60, 120];

    private static readonly string[] PermanentErrorPrefixes =
    [
        "PAYLOAD_INVALID",
        "CONSTRAINT_VIOLATION",
        "SQL_CONSTRAINT",
        "UNSUPPORTED_OP",
    ];

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

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

    public async Task<(int success, int failed, int skipped)> PushPendingAsync(
        int take = DefaultTake, CancellationToken cancellationToken = default)
    {
        var now = DateTime.Now;
        var pending = await _db.Queryable<MdpOutbox>()
            .Where(x => x.Status == 0
                        && x.RetryCount < MaxRetry
                        && (x.NextRetryTime == null || x.NextRetryTime <= now))
            .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();
            // 乐观锁基准：仅当 UpdateTime 未变时写入（期间被 refresh 则放弃）
            var loadedUpdateTime = item.UpdateTime;
            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,
                        "SOURCE_DISABLED",
                        $"目标源 {item.TargetSourceCode} 未启用",
                        cancellationToken);
                    skipped++;
                    continue;
                }

                if (!TryResolve(source, out var executor, out var routeError))
                {
                    await MarkConfigIssueAsync(item, "SOURCE_ROUTE_UNRESOLVED", 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)
                {
                    var marked = await MarkAsync(
                        item, 1, result.ResponseJson, null, null, null, loadedUpdateTime, cancellationToken);
                    if (!marked)
                    {
                        item.Status = 0;
                        skipped++;
                        continue;
                    }

                    try
                    {
                        await _hotWatch.TryEnrollFromOutboxSuccessAsync(item, cancellationToken);
                    }
                    catch (Exception enrollEx)
                    {
                        _logger.LogWarning(
                            enrollEx,
                            "[MdpTargetPushDispatcher] hot-watch enroll failed outbox id={Id} action={Action}",
                            item.Id, item.ActionCode);
                    }
                    success++;
                }
                else
                {
                    await ApplyFailureAsync(
                        item, result.ResponseJson, result.ErrorMessage, loadedUpdateTime, cancellationToken);
                    if (item.Status == 2) failed++; else skipped++;
                }
            }
            catch (Exception ex)
            {
                await ApplyFailureAsync(item, null, ex.Message, loadedUpdateTime, cancellationToken);
                if (item.Status == 2) failed++; else skipped++;
                _logger.LogWarning(ex, "[MdpTargetPushDispatcher] outbox id={Id} failed", item.Id);
            }
        }

        return (success, failed, skipped);
    }

    private async Task ApplyFailureAsync(
        MdpOutbox item,
        string? responseJson,
        string? errorMessage,
        DateTime loadedUpdateTime,
        CancellationToken ct)
    {
        var msg = Truncate(errorMessage, 900);
        var (errorCode, permanent) = ClassifyFailure(msg);

        if (permanent)
        {
            // 立即死信，不递增 retry_count
            item.NextRetryTime = null;
            item.LastErrorCode = errorCode;
            var marked = await MarkAsync(item, 2, responseJson, msg, null, errorCode, loadedUpdateTime, ct);
            if (!marked) item.Status = 0;
            return;
        }

        item.RetryCount++;
        item.LastErrorCode = errorCode;
        if (item.RetryCount >= MaxRetry)
        {
            item.NextRetryTime = null;
            var marked = await MarkAsync(item, 2, responseJson, msg, null, errorCode, loadedUpdateTime, ct);
            if (!marked) item.Status = 0;
        }
        else
        {
            item.NextRetryTime = DateTime.Now.AddMinutes(GetBackoffMinutes(item.RetryCount));
            var marked = await MarkAsync(
                item, 0, responseJson, msg, item.NextRetryTime, errorCode, loadedUpdateTime, ct);
            if (!marked) item.Status = 0;
        }
    }

    /// <summary>
    /// 可重试 vs 不可重试。配置类走 <see cref="MarkConfigIssueAsync"/>，不进本方法。
    /// </summary>
    private static (string? errorCode, bool permanent) ClassifyFailure(string? error)
    {
        if (string.IsNullOrWhiteSpace(error))
            return ("UNKNOWN", false);

        var trimmed = error.Trim();
        foreach (var prefix in PermanentErrorPrefixes)
        {
            if (trimmed.StartsWith(prefix, StringComparison.OrdinalIgnoreCase))
                return (prefix, true);
        }

        // payload 反序列化 / 解析失败 → 永久
        if (trimmed.Contains("payload 解析失败", StringComparison.OrdinalIgnoreCase)
            || trimmed.Contains("payload parse", StringComparison.OrdinalIgnoreCase)
            || trimmed.Contains("JsonException", StringComparison.OrdinalIgnoreCase)
            || trimmed.Contains("反序列化", StringComparison.OrdinalIgnoreCase))
            return ("PAYLOAD_INVALID", true);

        // SQL 约束 / 唯一冲突 → 永久
        if (ContainsAny(trimmed,
                "unique constraint", "unique key", "duplicate key", "duplicate entry",
                "primary key", "violation of unique", "violation of primary",
                "cannot insert duplicate", "约束", "唯一索引", "主键冲突"))
            return ("CONSTRAINT_VIOLATION", true);

        if (trimmed.Contains("不支持的 op", StringComparison.OrdinalIgnoreCase))
            return ("UNSUPPORTED_OP", true);

        // 连接 / 超时 → 可重试
        if (ContainsAny(trimmed,
                "timeout", "timed out", "connection", "network", "could not open",
                "unable to connect", "transport-level", "broken pipe", "socket",
                "连接", "超时", "无法连接"))
            return ("TRANSIENT", false);

        return ("RETRYABLE", false);
    }

    private static bool ContainsAny(string haystack, params string[] needles)
    {
        foreach (var n in needles)
        {
            if (haystack.Contains(n, StringComparison.OrdinalIgnoreCase))
                return true;
        }
        return false;
    }

    /// <summary>retry_count 从 1 起对应 Backoff[0]；越界 clamp 到末项。</summary>
    private static int GetBackoffMinutes(int retryCount)
    {
        var idx = Math.Max(0, retryCount - 1);
        if (idx >= BackoffMinutes.Length)
            idx = BackoffMinutes.Length - 1;
        return BackoffMinutes[idx];
    }

    /// <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/LastErrorCode/UpdateTime，status 保持 0，retry_count 不变。</summary>
    private async Task MarkConfigIssueAsync(MdpOutbox item, string errorCode, string error, CancellationToken ct)
    {
        item.ErrorMsg = Truncate(error, 900);
        item.LastErrorCode = Truncate(errorCode, 64);
        item.UpdateTime = DateTime.Now;
        await _db.Updateable(item)
            .UpdateColumns(x => new { x.ErrorMsg, x.LastErrorCode, x.UpdateTime })
            .ExecuteCommandAsync(ct);
    }

    /// <summary>
    /// 条件更新：仅当 UpdateTime 与加载时一致才写入，避免与入队 refresh 丢失更新。
    /// </summary>
    private async Task<bool> MarkAsync(
        MdpOutbox item,
        int status,
        string? responseJson,
        string? error,
        DateTime? nextRetryTime,
        string? lastErrorCode,
        DateTime loadedUpdateTime,
        CancellationToken ct)
    {
        var now = DateTime.Now;
        var resp = Truncate(responseJson, 4000);
        var code = Truncate(lastErrorCode, 64);
        var affected = await _db.Updateable<MdpOutbox>()
            .SetColumns(x => x.Status == status)
            .SetColumns(x => x.RetryCount == item.RetryCount)
            .SetColumns(x => x.ResponseJson == resp)
            .SetColumns(x => x.ErrorMsg == error)
            .SetColumns(x => x.NextRetryTime == nextRetryTime)
            .SetColumns(x => x.LastErrorCode == code)
            .SetColumns(x => x.UpdateTime == now)
            .Where(x => x.Id == item.Id && x.UpdateTime == loadedUpdateTime)
            .ExecuteCommandAsync(ct);

        if (affected <= 0)
        {
            _logger.LogInformation(
                "[MdpTargetPushDispatcher] 标记被跳过（期间已被重新入队）outbox id={Id} idem={Idem} wantStatus={Status}",
                item.Id, item.IdemKey, status);
            return false;
        }

        item.Status = status;
        item.ResponseJson = resp;
        item.ErrorMsg = error;
        item.NextRetryTime = nextRetryTime;
        item.LastErrorCode = code;
        item.UpdateTime = now;
        return true;
    }

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