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;
///
/// 出站回写:扫描 mdp_outbox 待推记录,按 target_source_code 调第三方写接口。
///
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;
}
/// 处理一批待推记录,返回成功/失败计数。
public async Task<(int success, int failed, int skipped)> PushPendingAsync(int take = 50, CancellationToken cancellationToken = default)
{
var pending = await _db.Queryable()
.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()
.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? cfg = null;
if (!string.IsNullOrWhiteSpace(source.ApiAuthConfig))
{
try { cfg = JsonSerializer.Deserialize>(source.ApiAuthConfig!); }
catch { /* ignore */ }
}
cfg ??= new Dictionary();
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]);
}