MoShiro/backend/src/MoShiro.Backend.Application/Query/AdaptiveAudit.cs

using System.Text.Json;
using Fleet.Core;
using MoShiro.Protocol;

namespace MoShiro.Backend.Application;

/// <summary>一个时间桶里一种采集方法的自适应探测频率：各状态各占多少秒、时间加权的平均频率、状态变化次数、变化时的最大丢包率。</summary>
public sealed record AdaptiveBucket(long StartMs, double QuietSeconds, double WatchSeconds, double ActiveSeconds, double DeadSeconds, double? AverageHz, int Changes, double MaxLossPct);

public sealed record AdaptiveKind(string Kind, IReadOnlyList<AdaptiveBucket> Buckets);

public sealed record AdaptiveResult(long FromMs, long ToMs, long StepMs, IReadOnlyList<AdaptiveKind> Kinds, IReadOnlyList<RateChange> Changes);

/// <summary>
/// 自适应探测频率的审计：节点把每一次状态变化（Quiet / Watch / Active / Dead、频率、原因、丢包）随心跳上报，Backend 存成事件（保留期和别的事件一样），
/// 这里把它们还原成「某个时段里各状态占多久、平均频率多少」，所以几天之后也能回答「晚高峰的时候自适应是什么状态」。
/// </summary>
public sealed class AdaptiveAuditService
{
    private readonly IEventStore _events;

    public AdaptiveAuditService(IEventStore events)
    {
        _events = events;
    }

    public async ValueTask<AdaptiveResult> GetAsync(LineKey line, long fromMs, long toMs, long stepMs, CancellationToken ct)
    {
        // 往前多取一段：区间开头的状态要由这之前最近的一次变化决定
        IReadOnlyList<EventRecord> records = await _events.ListAsync(line.Listener, AdaptiveAudit.EventKind, fromMs - AdaptiveAudit.LookBackMs, toMs, null, 20_000, ct);
        List<RateChange> changes = [];
        foreach (EventRecord record in records.OrderBy(r => r.AtMs).ThenBy(r => r.Id))
        {
            RateChange? change = AdaptiveAudit.Parse(record);
            if (change is not null && change.Dialer == line.Dialer.Value && (change.IpVersion ?? "ipv4") == line.IpVersion && change.Listener == line.Listener.Value)
            {
                changes.Add(change);
            }
        }

        return AdaptiveAudit.Summarize(changes, fromMs, toMs, stepMs);
    }
}

public static class AdaptiveAudit
{
    public const string EventKind = "rate_change";

    /// <summary>区间开头之前最多往回看多久，找那时的状态。</summary>
    public const long LookBackMs = 24 * 3_600_000L;

    public static RateChange? Parse(EventRecord record)
    {
        if (record.Detail is null)
        {
            return null;
        }

        try
        {
            RateChange? change = JsonSerializer.Deserialize(record.Detail, MoShiroJsonContext.Default.RateChange);
            if (change is not null)
            {
                change.AtMs = record.AtMs;
            }

            return change;
        }
        catch (JsonException)
        {
            return null;
        }
    }

    /// <summary>
    /// 把按时间排好的状态变化还原成每个时间桶里各状态的秒数。一次变化之后到下一次变化之前，是变化后的状态和频率；
    /// 第一次记录之前什么都不知道，不计入任何状态（节点启动时会记一条「start → watch」，所以每个自适应的任务从启动起就有记录；
    /// 没有这条记录的时段是节点还没升级到会上报的版本）。一种采集方法一条时间线（同一条线路上每种方法各自变化）。
    /// </summary>
    public static AdaptiveResult Summarize(IReadOnlyList<RateChange> changes, long fromMs, long toMs, long stepMs)
    {
        int count = (int)Math.Max(1, Math.Ceiling((toMs - fromMs) / (double)stepMs));
        var kinds = new List<AdaptiveKind>();
        foreach (IGrouping<string, RateChange> group in changes.GroupBy(c => c.Kind).OrderBy(g => g.Key, StringComparer.Ordinal))
        {
            var quiet = new double[count];
            var watch = new double[count];
            var active = new double[count];
            var dead = new double[count];
            var hzWeighted = new double[count];
            var hzSeconds = new double[count];
            var changeCounts = new int[count];
            var maxLoss = new double[count];
            RateChange[] ordered = [.. group.OrderBy(c => c.AtMs)];

            // 每一段：[这次变化, 下一次变化) 内是变化后的状态和频率；第一次记录之前不知道，不计
            string state = ordered[0].To;
            double? hz = ordered[0].Hz;
            long cursor = ordered[0].AtMs;
            foreach (RateChange change in ordered)
            {
                AddSegment(cursor, change.AtMs, state, hz);
                int slot = (int)((change.AtMs - fromMs) / stepMs);
                if (change.AtMs >= fromMs && change.AtMs < toMs && slot >= 0 && slot < count)
                {
                    changeCounts[slot]++;
                    maxLoss[slot] = Math.Max(maxLoss[slot], change.LossPct);
                }

                cursor = change.AtMs;
                state = change.To;
                hz = change.Hz;
            }

            AddSegment(cursor, toMs, state, hz);
            var buckets = new List<AdaptiveBucket>(count);
            for (int i = 0; i < count; i++)
            {
                buckets.Add(new AdaptiveBucket(fromMs + (i * stepMs), Math.Round(quiet[i], 1), Math.Round(watch[i], 1), Math.Round(active[i], 1), Math.Round(dead[i], 1),
                    hzSeconds[i] > 0 ? Math.Round(hzWeighted[i] / hzSeconds[i], 3) : null, changeCounts[i], maxLoss[i]));
            }

            kinds.Add(new AdaptiveKind(group.Key, buckets));

            void AddSegment(long start, long end, string segmentState, double? segmentHz)
            {
                start = Math.Max(start, fromMs);
                end = Math.Min(end, toMs);
                for (long t = start; t < end;)
                {
                    int slot = (int)((t - fromMs) / stepMs);
                    long slotEnd = Math.Min(end, fromMs + ((slot + 1) * stepMs));
                    double seconds = (slotEnd - t) / 1000.0;
                    switch (segmentState)
                    {
                        case "quiet": quiet[slot] += seconds; break;
                        case "watch": watch[slot] += seconds; break;
                        case "active": active[slot] += seconds; break;
                        case "dead": dead[slot] += seconds; break;
                    }

                    if (segmentHz is { } known)
                    {
                        hzWeighted[slot] += known * seconds;
                        hzSeconds[slot] += seconds;
                    }

                    t = slotEnd;
                }
            }
        }

        RateChange[] inRange = [.. changes.Where(c => c.AtMs >= fromMs && c.AtMs < toMs).OrderByDescending(c => c.AtMs).Take(200)];
        return new AdaptiveResult(fromMs, toMs, stepMs, kinds, inRange);
    }
}