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