using Admin.NET.Plugin.AiDOP.DataPlatform;
using Admin.NET.Plugin.AiDOP.DataPlatform.Inbound;
using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
using Furion.Schedule;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
namespace Admin.NET.Plugin.AiDOP.Job;
///
/// 定时探活 mdp_source:DB SELECT 1 / API GET baseUrl(或 /health)。
///
[JobDetail("job_mdp_source_health", Description = "MDP 数据源健康检查", GroupName = "default", Concurrent = false)]
[PeriodSeconds(300, TriggerId = "trigger_mdp_source_health", Description = "每 5 分钟探活数据源", RunOnStart = false)]
public class MdpSourceHealthCheckJob : IJob
{
private readonly IServiceScopeFactory _scopeFactory;
private readonly ILogger _logger;
public MdpSourceHealthCheckJob(IServiceScopeFactory scopeFactory, ILoggerFactory loggerFactory)
{
_scopeFactory = scopeFactory;
_logger = loggerFactory.CreateLogger(nameof(MdpSourceHealthCheckJob));
}
public async Task ExecuteAsync(JobExecutingContext context, CancellationToken stoppingToken)
{
using var scope = _scopeFactory.CreateScope();
var db = scope.ServiceProvider.GetRequiredService();
var scopeFactory = scope.ServiceProvider.GetRequiredService();
using var http = new HttpClient { Timeout = TimeSpan.FromSeconds(15) };
var sources = await db.Queryable().Where(x => x.Status == 1).ToListAsync(stoppingToken);
var ok = 0;
var fail = 0;
foreach (var src in sources)
{
stoppingToken.ThrowIfCancellationRequested();
var now = DateTime.Now;
try
{
if (string.Equals(src.SourceType, "DB", StringComparison.OrdinalIgnoreCase))
{
var scopeDb = await scopeFactory.GetScopeAsync(src.SourceCode, stoppingToken);
await scopeDb.Ado.GetIntAsync("SELECT 1");
src.HealthStatus = 1;
src.HealthMsg = "OK";
ok++;
}
else if (string.Equals(src.SourceType, "API", StringComparison.OrdinalIgnoreCase))
{
if (string.IsNullOrWhiteSpace(src.ApiBaseUrl))
throw new InvalidOperationException("api_base_url 为空");
var url = src.ApiBaseUrl.TrimEnd('/') + "/";
using var resp = await http.GetAsync(url, stoppingToken);
src.HealthStatus = (int)resp.StatusCode is >= 200 and < 500 ? 1 : 0;
src.HealthMsg = $"HTTP {(int)resp.StatusCode}";
if (src.HealthStatus == 1) ok++; else fail++;
}
else if (string.Equals(src.SourceType, "API_INBOUND", StringComparison.OrdinalIgnoreCase))
{
src.HealthStatus = 1;
src.HealthMsg = "OK";
ok++;
}
else
{
src.HealthStatus = 0;
src.HealthMsg = $"未知 source_type={src.SourceType}";
fail++;
}
}
catch (Exception ex)
{
src.HealthStatus = 0;
src.HealthMsg = ex.Message.Length > 480 ? ex.Message[..480] : ex.Message;
fail++;
}
src.LastHealthCheck = now;
src.UpdateTime = now;
await db.Updateable(src)
.UpdateColumns(x => new { x.HealthStatus, x.HealthMsg, x.LastHealthCheck, x.UpdateTime })
.ExecuteCommandAsync(stoppingToken);
}
if (sources.Count > 0)
_logger.LogInformation("[MdpSourceHealthCheckJob] total={Total} ok={Ok} fail={Fail}", sources.Count, ok, fail);
await ExpireInboundSnapshotsAsync(scope, stoppingToken);
await CheckInboundSilenceAsync(db, sources, stoppingToken);
}
/// 顺带把 OPEN 且已过期的快照置 EXPIRED;未 commit 不产生删除语义。
private async Task ExpireInboundSnapshotsAsync(IServiceScope scope, CancellationToken ct)
{
try
{
var snaps = scope.ServiceProvider.GetRequiredService();
await snaps.ExpireStaleOpenAsync(ct);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "[MdpSourceHealthCheckJob] inbound snapshot expire failed");
}
}
///
/// 入站实体沉默:inbound_enabled=1 且配置了 silence_alert_hours,
/// 最近 COMMITTED 超过阈值且当天命中 silence_calendar(空=每天)则写回源健康告警。
/// 从未推送过的实体以 update_time/create_time 为基线。
///
private async Task CheckInboundSilenceAsync(
ISqlSugarClient db, List sources, CancellationToken ct)
{
List entities;
try
{
entities = await db.Queryable()
.Where(e => e.InboundEnabled == 1 && e.SilenceAlertHours != null)
.ToListAsync(ct);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "[MdpSourceHealthCheckJob] inbound silence query entities failed");
return;
}
if (entities.Count == 0)
return;
var today = DateTime.Now;
var isoDow = today.DayOfWeek == DayOfWeek.Sunday ? 7 : (int)today.DayOfWeek;
var alertsBySource = new Dictionary>();
foreach (var entity in entities)
{
ct.ThrowIfCancellationRequested();
if (!HitsSilenceCalendar(entity.SilenceCalendar, isoDow))
continue;
DateTime? lastCommitted = null;
try
{
var last = await db.Queryable()
.Where(r => r.Status == "COMMITTED")
.Where("UPPER(entity_code) = @code", new SugarParameter("@code", entity.EntityCode.ToUpperInvariant()))
.OrderBy(r => r.CreateTime, OrderByType.Desc)
.FirstAsync(ct);
lastCommitted = last?.CreateTime;
}
catch (Exception ex)
{
_logger.LogWarning(ex, "[MdpSourceHealthCheckJob] inbound silence query requests failed entity={Entity}", entity.EntityCode);
continue;
}
var baseline = lastCommitted
?? (entity.UpdateTime != default ? entity.UpdateTime : entity.CreateTime);
var hours = entity.SilenceAlertHours.GetValueOrDefault();
if (hours <= 0)
continue;
if ((today - baseline).TotalHours <= hours)
continue;
var lastText = lastCommitted.HasValue
? lastCommitted.Value.ToString("yyyy-MM-dd HH:mm:ss")
: "never";
var msg = $"INBOUND silence: {entity.EntityCode} last COMMITTED {lastText} (threshold {hours}h)";
_logger.LogWarning("[MdpSourceHealthCheckJob] {Message}", msg);
if (!alertsBySource.TryGetValue(entity.SourceId, out var list))
{
list = [];
alertsBySource[entity.SourceId] = list;
}
list.Add(msg);
}
var sourceById = sources.ToDictionary(s => s.Id);
foreach (var (sourceId, messages) in alertsBySource)
{
if (!sourceById.TryGetValue(sourceId, out var src))
{
src = await db.Queryable().Where(s => s.Id == sourceId).FirstAsync(ct);
if (src == null)
continue;
}
var now = DateTime.Now;
src.HealthStatus = 0;
src.HealthMsg = Truncate(string.Join("; ", messages), 480);
src.LastHealthCheck = now;
src.UpdateTime = now;
await db.Updateable(src)
.UpdateColumns(x => new { x.HealthStatus, x.HealthMsg, x.LastHealthCheck, x.UpdateTime })
.ExecuteCommandAsync(ct);
}
}
private static bool HitsSilenceCalendar(string calendar, int isoDow)
{
if (string.IsNullOrWhiteSpace(calendar))
return true;
var days = calendar.Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries);
return days.Any(d => int.TryParse(d, out var n) && n == isoDow);
}
private static string Truncate(string s, int max) =>
string.IsNullOrEmpty(s) || s.Length <= max ? s : s[..max];
}