AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/Infrastructure/EtlInstanceRegistrar.cs

using Admin.NET.Plugin.AiDOP.Entity.SmartOps;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace Admin.NET.Plugin.AiDOP.Infrastructure;

/// <summary>
/// ETL 执行机实例注册器：把本进程登记进 <c>ado_etl_instance</c> 并定期心跳，
/// 同时把本实例是否为执行机发布到 <see cref="AidopRunnerState"/> 供 <see cref="AidopJobGate"/> 读取。
///
/// <para><b>判定方式：槽位比对</b>。本进程声明自己属于哪个部署槽位
/// （<c>AidopInstanceIdentity.SlotCode</c>），指派记在 <c>ado_etl_runner_designation</c>；
/// 两者相等即为执行机。<b>未声明槽位者永不是执行机</b>。
/// 之所以不拿 <c>instance_id</c> 比对，是因为它含进程ID与启动时间，
/// 指派挂上去每次重启都会丢——2026-09-28 的 28 条积压即由此而来。</para>
///
/// <para><b>只读指派、从不自行指派</b>：本注册器插入新行时 <c>is_runner</c> 恒为 0，
/// 也绝不写指派表。指派是人工动作（超管在 MDP 运行监控页操作，或直接改库）。
/// 禁止在这里实现「抢锁自荐」或「心跳超时自动接管」——
/// 那会让开发机在某次重启后意外变成执行机，正是本次治理要消除的情形。</para>
///
/// <para><b>同槽位撞车则两边都不跑</b>：若同一槽位另有存活实例（滚动重启的重叠窗口，
/// 或有人把两个进程配成同一槽位），本进程发布 false 而不是照常跑。
/// 没有租约就没有仲裁者，此时抢着跑等于双跑全量，那正是 2026-09-24 连接耗尽的成因；
/// 代价只是重启重叠的几十秒空窗，定时作业不受影响。</para>
///
/// <para><b>本服务不受 <see cref="AidopJobGate"/> 管辖</b>，必须常开。
/// 它是闸门的数据来源，被闸门关掉就成了死锁：没人心跳 → 没人发布指派 → 永远不是执行机。</para>
/// </summary>
public sealed class EtlInstanceRegistrar : BackgroundService
{
    /// <summary>心跳周期。<see cref="AidopRunnerState.FreshnessWindow"/> 按其 3 倍取值。</summary>
    public static readonly TimeSpan HeartbeatInterval = TimeSpan.FromSeconds(20);

    /// <summary>启动延时：等宿主配置与建表脚本就绪，避免首拍必然失败刷一条 warn。</summary>
    private static readonly TimeSpan StartupDelay = TimeSpan.FromSeconds(5);

    /// <summary>下线实例行的保留期。超期即清理，防止表随重启无限增长。</summary>
    private static readonly TimeSpan DeadInstanceRetention = TimeSpan.FromDays(7);

    /// <summary>存活判据：连续三拍没心跳即视为下线，与 <see cref="AidopRunnerState.FreshnessWindow"/> 同宽。</summary>
    private static readonly TimeSpan LivenessWindow = TimeSpan.FromSeconds(60);

    private readonly IServiceScopeFactory _scopeFactory;
    private readonly ILogger _logger;
    private int _consecutiveFailures;
    private int _guardrailsLogged;

    public EtlInstanceRegistrar(IServiceScopeFactory scopeFactory, ILoggerFactory loggerFactory)
    {
        _scopeFactory = scopeFactory;
        _logger = loggerFactory.CreateLogger(nameof(EtlInstanceRegistrar));
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        try { await Task.Delay(StartupDelay, stoppingToken); }
        catch (OperationCanceledException) { return; }

        _logger.LogInformation(
            "[EtlInstanceRegistrar] 本实例 instanceId={InstanceId} machine={Machine} pid={Pid} version={Version} slot={Slot}",
            AidopInstanceIdentity.InstanceId, AidopInstanceIdentity.MachineName,
            AidopInstanceIdentity.ProcessId, AidopInstanceIdentity.AppVersion,
            AidopInstanceIdentity.SlotCode ?? "(未声明)");

        var nextCleanupAt = DateTimeOffset.UtcNow;
        var lastConflictLogged = false;
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                var beat = await HeartbeatAsync(stoppingToken);
                AidopRunnerState.Publish(beat.IsRunner);

                // 撞车是配置事故或滚动重启重叠，两者都必须让人看见；
                // 但每 20 秒一条会刷屏，故只在状态翻转时打。
                if (beat.SlotConflict != lastConflictLogged)
                {
                    if (beat.SlotConflict)
                        _logger.LogWarning(
                            "[EtlInstanceRegistrar] 槽位撞车：slot={Slot} 另有 {Count} 个存活实例，"
                            + "本实例暂不作为执行机（避免双跑全量）",
                            AidopInstanceIdentity.SlotCode, beat.SameSlotLiveOthers);
                    else
                        _logger.LogInformation(
                            "[EtlInstanceRegistrar] 槽位撞车已解除 slot={Slot} isRunner={IsRunner}",
                            AidopInstanceIdentity.SlotCode, beat.IsRunner);
                    lastConflictLogged = beat.SlotConflict;
                }

                if (Interlocked.Exchange(ref _guardrailsLogged, 1) == 0)
                    await LogMySqlGuardrailsAsync(beat.IsRunner, stoppingToken);

                if (_consecutiveFailures > 0)
                {
                    _logger.LogInformation(
                        "[EtlInstanceRegistrar] 心跳恢复，连续失败 {Count} 次后已重新发布指派 isRunner={IsRunner}",
                        _consecutiveFailures, beat.IsRunner);
                    _consecutiveFailures = 0;
                }

                if (DateTimeOffset.UtcNow >= nextCleanupAt)
                {
                    await CleanupDeadAsync(stoppingToken);
                    nextCleanupAt = DateTimeOffset.UtcNow + TimeSpan.FromHours(6);
                }
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception ex)
            {
                // 查库失败必须打回未知，让闸门回退到配置值：
                // 指派可能已被运维改掉，继续沿用旧值会让两台实例同时自认执行机。
                AidopRunnerState.Invalidate();

                // 库不可用时本循环每 20 秒一次，不做节流会刷屏淹没真正的错误。
                _consecutiveFailures++;
                if (_consecutiveFailures == 1 || _consecutiveFailures % 30 == 0)
                    _logger.LogWarning(ex,
                        "[EtlInstanceRegistrar] 心跳失败（连续第 {Count} 次），指派状态已置未知，闸门回退到配置值",
                        _consecutiveFailures);
            }

            try { await Task.Delay(HeartbeatInterval, stoppingToken); }
            catch (OperationCanceledException) { break; }
        }
    }

    /// <summary>一拍心跳的结果。</summary>
    private readonly record struct HeartbeatResult(bool IsRunner, string DesignatedSlot, int SameSlotLiveOthers)
    {
        public bool SlotConflict => SameSlotLiveOthers > 0;
    }

    /// <summary>
    /// 心跳一拍：更新本实例行（不存在则插入），读取指派槽位并与本进程槽位比对。
    /// </summary>
    private async Task<HeartbeatResult> HeartbeatAsync(CancellationToken ct)
    {
        using var scope = _scopeFactory.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>();
        var instances = scope.ServiceProvider.GetRequiredService<IEtlInstanceStore>();
        var instanceId = AidopInstanceIdentity.InstanceId;
        var slot = AidopInstanceIdentity.SlotCode;
        var now = DateTime.Now;

        // 刻意不更新 is_runner：那已是遗留列，指派在 ado_etl_runner_designation。
        var affected = await db.Updateable<AdoEtlInstance>()
            .SetColumns(x => new AdoEtlInstance
            {
                LastHeartbeatAt = now,
                AppVersion = AidopInstanceIdentity.AppVersion,
                SlotCode = slot,
                UpdateTime = now
            })
            .Where(x => x.InstanceId == instanceId)
            .ExecuteCommandAsync(ct);

        if (affected <= 0)
        {
            await db.Insertable(new AdoEtlInstance
            {
                InstanceId = instanceId,
                MachineName = AidopInstanceIdentity.MachineName,
                ProcessId = AidopInstanceIdentity.ProcessId,
                AppVersion = AidopInstanceIdentity.AppVersion,
                SlotCode = slot,
                IsRunner = false,
                StartedAt = AidopInstanceIdentity.StartedAt,
                LastHeartbeatAt = now,
                CreateTime = now,
                UpdateTime = now
            }).ExecuteCommandAsync(ct);
        }

        var designated = await instances.GetDesignatedSlotAsync(ct);

        // 没声明槽位就不是候选，连撞车都不必查——省一次查询，也省得把「未声明」
        // 误报成撞车（多台未配置的开发机 slot 全是 NULL，NULL 之间不构成同槽位）。
        if (string.IsNullOrEmpty(slot) || !string.Equals(slot, designated, StringComparison.Ordinal))
            return new HeartbeatResult(false, designated, 0);

        var liveCutoff = now - LivenessWindow;
        var others = await db.Queryable<AdoEtlInstance>()
            .Where(x => x.SlotCode == slot && x.InstanceId != instanceId && x.LastHeartbeatAt >= liveCutoff)
            .CountAsync(ct);

        return new HeartbeatResult(others == 0, designated, others);
    }

    /// <summary>
    /// 首次心跳成功后打一条现场参数，全进程只打一次。
    ///
    /// <para><b>为什么要打</b>：2026-09-24 的 1040 事故里，165 的
    /// <c>innodb_buffer_pool_size</c> 一直是默认 128MB、<c>max_connections</c> 是默认 151，
    /// 而没有任何人能从日志看出这一点——只能事后 SSH 上去查。
    /// 把它连同执行机身份打进启动日志，下次排障第一眼就能看到。</para>
    ///
    /// <para>不打连接串、不打任何口令。失败只 warn，不影响心跳。</para>
    /// </summary>
    private async Task LogMySqlGuardrailsAsync(bool assigned, CancellationToken ct)
    {
        try
        {
            using var scope = _scopeFactory.CreateScope();
            var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>();
            var row = await db.Ado.SqlQuerySingleAsync<MySqlGuardrails>(
                "SELECT @@max_connections AS MaxConnections, "
                + "@@innodb_buffer_pool_size AS BufferPoolBytes, "
                + "DATABASE() AS DbName");

            _logger.LogInformation(
                "[EtlInstanceRegistrar] 现场参数 db={Db} maxConnections={MaxConn} bufferPoolMB={PoolMb} "
                + "isRunner={IsRunner} instanceId={InstanceId}",
                row?.DbName, row?.MaxConnections, row?.BufferPoolBytes / 1024 / 1024,
                assigned, AidopInstanceIdentity.InstanceId);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "[EtlInstanceRegistrar] 读取现场 MySQL 参数失败，跳过");
        }
    }

    private sealed class MySqlGuardrails
    {
        public long MaxConnections { get; set; }
        public long BufferPoolBytes { get; set; }
        public string DbName { get; set; }
    }

    /// <summary>
    /// 清理早已下线的实例行，不分槽位一视同仁。
    ///
    /// <para>旧模型必须特意保住被指派实例的行，因为指派就寄生在
    /// <c>ado_etl_instance.is_runner</c> 上，删行等于删指派。现在指派独立存在
    /// <c>ado_etl_runner_designation</c>，这条例外不但没用，还有害：执行机槽位
    /// 每重启一次留一行，恰恰是重启最勤的那个槽位无限堆积，页面上还会摆着一排
    /// 早已死掉的同槽位实例。运维要看的「指派槽位上没有活实例」由监控页的
    /// noLiveRunner 告警给出，不需要靠留尸体来表达。</para>
    /// </summary>
    private async Task CleanupDeadAsync(CancellationToken ct)
    {
        try
        {
            using var scope = _scopeFactory.CreateScope();
            var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>();
            var cutoff = DateTime.Now - DeadInstanceRetention;
            var removed = await db.Deleteable<AdoEtlInstance>()
                .Where(x => x.LastHeartbeatAt < cutoff)
                .ExecuteCommandAsync(ct);
            if (removed > 0)
                _logger.LogInformation("[EtlInstanceRegistrar] 已清理下线实例 {Count} 行", removed);
        }
        catch (Exception ex)
        {
            // 清理失败不影响心跳与指派，吞掉即可
            _logger.LogWarning(ex, "[EtlInstanceRegistrar] 清理下线实例失败");
        }
    }
}