MoShiro/backend/src/MoShiro.Backend.Application/Live/HotLayer.cs

using System.Collections.Concurrent;
using Fleet.Core;
using MoShiro.Backend.Analysis;
using MoShiro.Protocol;

namespace MoShiro.Backend.Application;

/// <summary>
/// 1 秒精度的内存热层：最近 1 小时的累加器。桶以 <see cref="AccumulatorCodec"/> 编码后的字节保存，
/// 15 条线路约十几 MB；它同时是 SSE 实时推送的数据来源。重启后为空，由后台任务从原始块回填。
/// </summary>
public sealed class HotLayer
{
    public const long BucketMs = 1000;
    public static readonly TimeSpan Retention = TimeSpan.FromHours(1);

    private readonly ConcurrentDictionary<(NodeId Agent, string Series), Ring> _rings = new();
    private readonly TimeProvider _time;
    private volatile bool _ready;

    public HotLayer(TimeProvider time)
    {
        _time = time;
    }

    /// <summary>回填完成之前热层不完整，查询规划器不会选用它。</summary>
    public bool Ready => _ready;

    public void MarkReady() => _ready = true;

    /// <summary>评估一个块并合并进热层。同一个块重复合并（例如回填与实时竞争）会被忽略。</summary>
    public void Merge(NodeId agent, string series, RawChunk chunk)
    {
        Ring ring = _rings.GetOrAdd((agent, series), _ => new Ring());
        var buckets = new SortedDictionary<long, MetricSet>();
        WindowEvaluator.Default.Evaluate(chunk, BucketMs, buckets);
        long cutoff = _time.GetUtcNow().ToUnixTimeMilliseconds() - (long)Retention.TotalMilliseconds;
        lock (ring)
        {
            if (!ring.Chunks.Add(chunk.Header.ChunkId))
            {
                return;
            }

            foreach ((long start, MetricSet set) in buckets)
            {
                if (start < cutoff)
                {
                    continue;
                }

                if (ring.Buckets.TryGetValue(start, out byte[]? existing) && AccumulatorCodec.Decode(existing).TryGet(out MetricSet? merged, out _))
                {
                    merged.Merge(set);
                    ring.Buckets[start] = AccumulatorCodec.Encode(merged);
                }
                else
                {
                    ring.Buckets[start] = AccumulatorCodec.Encode(set);
                }
            }

            Prune(ring, cutoff);
        }
    }

    /// <summary>[from, to) 内的 1 秒桶（解码后的副本，按时间升序）。</summary>
    public IReadOnlyList<KeyValuePair<long, MetricSet>> GetRange(NodeId agent, string series, long fromMs, long toMs)
    {
        if (!_rings.TryGetValue((agent, series), out Ring? ring))
        {
            return [];
        }

        var result = new List<KeyValuePair<long, MetricSet>>();
        lock (ring)
        {
            foreach ((long start, byte[] bytes) in ring.Buckets)
            {
                if (start >= toMs)
                {
                    break;
                }

                if (start >= fromMs && AccumulatorCodec.Decode(bytes).TryGet(out MetricSet? set, out _))
                {
                    result.Add(new KeyValuePair<long, MetricSet>(start, set));
                }
            }
        }

        return result;
    }

    private static void Prune(Ring ring, long cutoff)
    {
        while (ring.Buckets.Count > 0)
        {
            long first = ring.Buckets.Keys.First();
            if (first >= cutoff)
            {
                break;
            }

            ring.Buckets.Remove(first);
        }

        // 块 ID 集合只需要覆盖热层的时间窗，限制大小即可
        if (ring.Chunks.Count > 2000)
        {
            ring.Chunks.Clear();
        }
    }

    private sealed class Ring
    {
        public SortedDictionary<long, byte[]> Buckets { get; } = [];

        public HashSet<ChunkId> Chunks { get; } = [];
    }
}