AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/Job/MdpSourceHealthCheckJob.cs

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;

/// <summary>
/// 定时探活 mdp_source：DB SELECT 1 / API GET baseUrl（或 /health）。
/// </summary>
[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)
    {
        if (!AidopJobGate.ShouldRun(nameof(MdpSourceHealthCheckJob), _logger)) return;

        using var scope = _scopeFactory.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>();
        var scopeFactory = scope.ServiceProvider.GetRequiredService<MdpSourceScopeFactory>();
        using var http = new HttpClient { Timeout = TimeSpan.FromSeconds(15) };

        var sources = await db.Queryable<MdpSource>().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);
    }

    /// <summary>顺带把 OPEN 且已过期的快照置 EXPIRED；未 commit 不产生删除语义。</summary>
    private async Task ExpireInboundSnapshotsAsync(IServiceScope scope, CancellationToken ct)
    {
        try
        {
            var snaps = scope.ServiceProvider.GetRequiredService<MdpInboundSnapshotService>();
            await snaps.ExpireStaleOpenAsync(ct);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "[MdpSourceHealthCheckJob] inbound snapshot expire failed");
        }
    }

    /// <summary>
    /// 入站实体沉默：inbound_enabled=1 且配置了 silence_alert_hours，
    /// 最近 COMMITTED 超过阈值且当天命中 silence_calendar（空=每天）则写回源健康告警。
    /// 从未推送过的实体以 update_time/create_time 为基线。
    /// </summary>
    private async Task CheckInboundSilenceAsync(
        ISqlSugarClient db, List<MdpSource> sources, CancellationToken ct)
    {
        List<MdpEntity> entities;
        try
        {
            entities = await db.Queryable<MdpEntity>()
                .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<long, List<string>>();

        foreach (var entity in entities)
        {
            ct.ThrowIfCancellationRequested();
            if (!HitsSilenceCalendar(entity.SilenceCalendar, isoDow))
                continue;

            DateTime? lastCommitted = null;
            try
            {
                var last = await db.Queryable<MdpInboundRequest>()
                    .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<MdpSource>().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];
}