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

using System.Net.Http.Headers;
using System.Text;
using System.Text.Json;
using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;

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

/// <summary>
/// 出站回写：扫描 mdp_outbox 待推记录，按 target_source_code 调第三方写接口。
/// </summary>
public sealed class MdpApiPushExecutor : ITransient
{
    private const int MaxRetry = 3;
    private readonly IHttpClientFactory _httpClientFactory;
    private readonly ISqlSugarClient _db;

    public MdpApiPushExecutor(IHttpClientFactory httpClientFactory, ISqlSugarClient db)
    {
        _httpClientFactory = httpClientFactory;
        _db = db;
    }

    /// <summary>处理一批待推记录，返回成功/失败计数。</summary>
    public async Task<(int success, int failed, int skipped)> PushPendingAsync(int take = 50, CancellationToken cancellationToken = default)
    {
        var pending = await _db.Queryable<MdpOutbox>()
            .Where(x => x.Status == 0 && x.RetryCount < MaxRetry)
            .OrderBy(x => x.Id)
            .Take(take)
            .ToListAsync(cancellationToken);

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

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

                if (source == null || string.IsNullOrWhiteSpace(source.ApiBaseUrl))
                {
                    await MarkAsync(item, 2, null, $"目标源 {item.TargetSourceCode} 未配置 API", cancellationToken);
                    failed++;
                    continue;
                }

                // payload 约定：{ "path": "/api/xxx", "method": "POST", "body": {...} }
                using var payloadDoc = JsonDocument.Parse(string.IsNullOrWhiteSpace(item.PayloadJson) ? "{}" : item.PayloadJson!);
                var root = payloadDoc.RootElement;
                var path = root.TryGetProperty("path", out var p) ? p.GetString() : $"/outbox/{item.ActionCode}";
                var method = root.TryGetProperty("method", out var m) ? (m.GetString() ?? "POST") : "POST";
                var bodyEl = root.TryGetProperty("body", out var b) ? b : root;

                var client = _httpClientFactory.CreateClient("MdpApiPush");
                client.Timeout = TimeSpan.FromSeconds(60);
                var url = source.ApiBaseUrl!.TrimEnd('/') + (path!.StartsWith('/') ? path : "/" + path);

                using var request = new HttpRequestMessage(new HttpMethod(method), url);
                ApplyAuth(request, source);
                request.Content = new StringContent(bodyEl.GetRawText(), Encoding.UTF8, "application/json");

                using var response = await client.SendAsync(request, cancellationToken);
                var respBody = await response.Content.ReadAsStringAsync(cancellationToken);
                if (response.IsSuccessStatusCode)
                {
                    await MarkAsync(item, 1, respBody, null, cancellationToken);
                    success++;
                }
                else
                {
                    item.RetryCount++;
                    var status = item.RetryCount >= MaxRetry ? 2 : 0;
                    await MarkAsync(item, status, respBody, $"HTTP {(int)response.StatusCode}", 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++;
            }
        }

        return (success, failed, skipped);
    }

    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 void ApplyAuth(HttpRequestMessage request, MdpSource source)
    {
        var authType = (source.ApiAuthType ?? "NONE").Trim().ToUpperInvariant();
        if (authType is "NONE" or "") return;
        Dictionary<string, string>? cfg = null;
        if (!string.IsNullOrWhiteSpace(source.ApiAuthConfig))
        {
            try { cfg = JsonSerializer.Deserialize<Dictionary<string, string>>(source.ApiAuthConfig!); }
            catch { /* ignore */ }
        }
        cfg ??= new Dictionary<string, string>();
        if ((authType is "BEARER" or "TOKEN") && (cfg.TryGetValue("token", out var token) || cfg.TryGetValue("access_token", out token)))
            request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", token);
        else if (authType == "BASIC" && cfg.TryGetValue("username", out var user) && cfg.TryGetValue("password", out var pwd))
            request.Headers.Authorization = new AuthenticationHeaderValue("Basic", Convert.ToBase64String(Encoding.UTF8.GetBytes($"{user}:{pwd}")));
        else if (authType == "APIKEY")
        {
            var header = cfg.GetValueOrDefault("header") ?? "X-API-Key";
            if (cfg.TryGetValue("apiKey", out var key) || cfg.TryGetValue("key", out key))
                request.Headers.TryAddWithoutValidation(header, key);
        }
    }

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