MoShiro/backend/src/MoShiro.Backend.Application/Analysis/AnalysisScheduler.cs

using Fleet.Core;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using MoShiro.Backend.Analysis;

namespace MoShiro.Backend.Application;

/// <summary>
/// 持久路径：分钟窗口就绪（窗口结束 + 宽限，或迟到的块到达）之后从原始块计算 → 写入 metric_1m → 标记降采样脏桶；
/// 然后把脏桶合并成 5 分钟、1 小时层。
/// </summary>
public sealed partial class AnalysisScheduler : BackgroundService
{
    private const int BatchSize = 50;

    private readonly IMetricStore _metrics;
    private readonly WindowComputer _computer;
    private readonly BackendOptions _options;
    private readonly BackendCounters _counters;
    private readonly TimeProvider _time;
    private readonly ILogger<AnalysisScheduler> _logger;

    public AnalysisScheduler(IMetricStore metrics, WindowComputer computer, BackendOptions options, BackendCounters counters, TimeProvider time, ILogger<AnalysisScheduler> logger)
    {
        _metrics = metrics;
        _computer = computer;
        _options = options;
        _counters = counters;
        _time = time;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                int worked;
                do
                {
                    worked = await RunOnceAsync(stoppingToken);
                }
                while (worked >= BatchSize);

                await Task.Delay(_options.AnalysisPollInterval, _time, stoppingToken);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                return;
            }
            catch (Exception ex)
            {
                // 后台循环不能因为一次失败而退出
                LogFailed(ex);
                await Task.Delay(_options.AnalysisPollInterval, _time, stoppingToken);
            }
        }
    }

    /// <summary>处理一批到期的窗口和脏桶，返回处理的窗口数（也供测试直接调用）。</summary>
    public async Task<int> RunOnceAsync(CancellationToken ct)
    {
        long now = _time.GetUtcNow().ToUnixTimeMilliseconds();
        IReadOnlyList<QueueItem> due = await _metrics.DequeueDueAsync(now, BatchSize, ct);
        foreach (QueueItem item in due)
        {
            await ComputeWindowAsync(item, ct);
            _counters.CountWindowAnalyzed();
        }

        if (due.Count > 0)
        {
            await _metrics.CompleteAsync(due, ct);
        }

        await RollupAsync(ct);
        return due.Count;
    }

    private async ValueTask ComputeWindowAsync(QueueItem item, CancellationToken ct)
    {
        long width = MetricLevel.M1.WidthMs();
        WindowComputer.Computed computed = await _computer.ComputeAsync(item.AgentId, item.Series, item.WindowStart, item.WindowStart + width, width, null, ct);
        if (!computed.Buckets.TryGetValue(item.WindowStart, out MetricSet? set))
        {
            return;
        }

        await _metrics.UpsertAsync(MetricLevel.M1, [new MetricRow(item.AgentId, item.Series, item.WindowStart, computed.AnyPartial, AccumulatorCodec.Encode(set))], ct);
        await MarkUpperLevelsAsync(item.AgentId, item.Series, item.WindowStart, ct);
    }

    internal async ValueTask MarkUpperLevelsAsync(NodeId agent, string series, long windowStart, CancellationToken ct)
    {
        await _metrics.MarkRollupDirtyAsync(MetricLevel.M5, agent, series, Floor(windowStart, MetricLevel.M5), ct);
    }

    /// <summary>降采样：读取下一级的累加器状态，精确合并后写回。</summary>
    private async ValueTask RollupAsync(CancellationToken ct)
    {
        foreach ((MetricLevel level, NodeId agent, string series, long bucketStart) in await _metrics.TakeRollupDirtyAsync(100, ct))
        {
            MetricLevel source = level == MetricLevel.M5 ? MetricLevel.M1 : MetricLevel.M5;
            long end = bucketStart + level.WidthMs();
            IReadOnlyList<MetricRow> rows = await _metrics.ReadAsync(source, agent, series, bucketStart, end, ct);
            if (rows.Count == 0)
            {
                continue;
            }

            var merged = new MetricSet();
            bool partial = false;
            foreach (MetricRow row in rows)
            {
                if (AccumulatorCodec.Decode(row.Metrics).TryGet(out MetricSet? set, out _))
                {
                    merged.Merge(set);
                    partial |= row.Partial;
                }
            }

            await _metrics.UpsertAsync(level, [new MetricRow(agent, series, bucketStart, partial, AccumulatorCodec.Encode(merged))], ct);
            if (level == MetricLevel.M5)
            {
                await _metrics.MarkRollupDirtyAsync(MetricLevel.H1, agent, series, Floor(bucketStart, MetricLevel.H1), ct);
            }
        }
    }

    internal static long Floor(long ms, MetricLevel level) => Math.DivRem(ms, level.WidthMs(), out _) * level.WidthMs();

    [LoggerMessage(Level = LogLevel.Error, Message = "analysis iteration failed")]
    private partial void LogFailed(Exception exception);
}