using System.Net;
using System.Net.Http.Headers;
using System.Text.Json.Nodes;
using Fleet.Core;
using MoShiro.Protocol;
using MoShiro.TestSupport;

namespace MoShiro.Backend.Tests;

[Collection("backend")]
public sealed class IngestAndQueryTests
{
    private const long Minute = 60_000;

    private readonly BackendFixture _backend;

    public IngestAndQueryTests(BackendFixture backend)
    {
        _backend = backend;
    }

    /// <summary>当前分钟的起点：测试数据放在已经结束的分钟里，这样持久路径会立即计算这些窗口。</summary>
    private static long ClosedMinuteStart(int minutesAgo) =>
        ((TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds() / Minute) - minutesAgo) * Minute;

    /// <summary>把 [baseMs, baseMs + chunks × 10 s) 的数据送给后端：每个块 10 s、100 个包；每 <paramref name="lossEvery"/> 个包丢一个回复。</summary>
    private static async Task SendLineAsync(AgentSimulator agent, long baseMs, int chunks, int lossEvery = 0, bool tcp = true, double owdMs = 20)
    {
        for (int k = 0; k < chunks; k++)
        {
            long start = baseMs + (k * 10_000L);
            RawChunk udp = ChunkFactory.Echo(
                agent.NodeId, "udp_echo", start, 100, 100,
                i => new EchoPacket(false, lossEvery > 0 && i % lossEvery == 0, (long)(owdMs * 1000), (long)(owdMs * 1000)));
            Assert.True(await agent.SendChunkAsync(udp));
            if (tcp)
            {
                long probeBase = ChunkFactory.ProbeUs(0);
                RawStream info = ChunkFactory.TcpInfo(Enumerable.Range(0, 10).Select(s => (probeBase + (s * 1_000_000L), (byte)0, 45_000u, (uint)(s * 2), (uint)(s * 400))));
                RawChunk tcpChunk = ChunkFactory.Echo(
                    agent.NodeId, "tcp_echo", start, 200, 50, _ => new EchoPacket(false, false, (long)(owdMs * 1000), (long)(owdMs * 1000)), extraStreams: [info]);
                Assert.True(await agent.SendChunkAsync(tcpChunk));
            }
        }
    }

    private async Task<(AgentSimulator Agent, string Name)> LineAsync(long baseMs, int chunks, int lossEvery = 0, bool tcp = true, double owdMs = 20)
    {
        AgentSimulator agent = await _backend.ConnectAgentAsync();
        await SendLineAsync(agent, baseMs, chunks, lossEvery, tcp, owdMs);
        string name = (await _backend.GetJsonAsync(_backend.Admin, "/api/admin/agents/" + agent.NodeId.Value))["name"]!.GetValue<string>();
        return (agent, name);
    }

    private static string Iso(long ms) => DateTimeOffset.FromUnixTimeMilliseconds(ms).UtcDateTime.ToString("yyyy-MM-dd'T'HH:mm:ss.fff'Z'");

    private async Task<JsonNode> SeriesAsync(string agentId, string query) =>
        await _backend.GetJsonAsync(_backend.Admin, $"/api/public/lines/{_backend.Line(agentId)}/series?{query}");

    [Fact]
    public async Task Line_ReportsWhenItsDataStarted_SoTheDisplayOnlyOffersRangesThatHaveData()
    {
        await _backend.SiteMustBePublicAsync();
        long baseMs = ClosedMinuteStart(30);
        (AgentSimulator agent, _) = await LineAsync(baseMs, 6);
        await using (agent)
        {
            JsonNode before = await _backend.GetJsonAsync(_backend.Admin, $"/api/public/lines/{_backend.Line(agent.NodeId.Value)}");
            DateTimeOffset since = before["data_since"]!.GetValue<DateTimeOffset>();
            // 精确到秒（细的层）；如果细的层已经被保留策略清掉，只剩 5 分钟层，起点最多早 5 分钟
            Assert.InRange(since.ToUnixTimeMilliseconds(), baseMs - 300_000, baseMs + 70_000);

            // 一条没有任何数据的线路：没有 data_since
            await using AgentSimulator empty = await _backend.ConnectAgentAsync();
            JsonNode none = await _backend.GetJsonAsync(_backend.Admin, $"/api/public/lines/{_backend.Line(empty.NodeId.Value)}");
            Assert.Null(none["data_since"]);
        }
    }

    [Fact]
    public async Task Series_AnchorLatest_ShowsTheLastWindowOfALineThatHasStoppedReporting()
    {
        long baseMs = ClosedMinuteStart(20);
        (AgentSimulator agent, _) = await LineAsync(baseMs, chunks: 6, tcp: false); // one minute of data, 20 minutes ago
        await using (agent)
        {
            string id = agent.NodeId.Value;

            // Anchored at now: the last 5 minutes are empty
            JsonNode atNow = await SeriesAsync(id, "range=5m&step=10s&metrics=loss.total");
            Assert.Null(atNow["summary"]!["samples"]!["udp_echo"]);

            // Anchored at the latest sample: the window ends where the data ends and carries all of it
            JsonNode anchored = await SeriesAsync(id, "range=5m&step=10s&metrics=loss.total&anchor=latest");
            long to = DateTimeOffset.Parse(anchored["to"]!.GetValue<string>(), System.Globalization.CultureInfo.InvariantCulture).ToUnixTimeMilliseconds();
            Assert.InRange(to, baseMs + (50 * 1000), baseMs + (80 * 1000));
            Assert.Equal(600, anchored["summary"]!["samples"]!["udp_echo"]!.GetValue<long>());
            Assert.False(anchored["live"]!.GetValue<bool>());

            // anchor=now is the default behaviour; anything else is rejected
            JsonNode explicitNow = await SeriesAsync(id, "range=5m&step=10s&metrics=loss.total&anchor=now");
            Assert.Null(explicitNow["summary"]!["samples"]!["udp_echo"]);
            using HttpResponseMessage bad = await _backend.Admin.GetAsync($"/api/public/lines/{_backend.Line(id)}/series?range=5m&anchor=yesterday");
            Assert.Equal(System.Net.HttpStatusCode.BadRequest, bad.StatusCode);
        }
    }

    [Fact]
    public async Task Chunks_AreStoredAndAcknowledged_DuplicatesAndGarbageAreDiscardedWithAReason()
    {
        await using AgentSimulator agent = await _backend.ConnectAgentAsync();
        long start = ClosedMinuteStart(3);
        RawChunk chunk = ChunkFactory.UdpEchoClean(agent.NodeId, start);

        Assert.True(await agent.SendChunkAsync(chunk), "the first delivery is stored");
        Assert.False(await agent.SendChunkAsync(chunk), "the second delivery is acknowledged as a duplicate");

        // 损坏的块：确认掉（否则节点会无限重发）并记入事件
        var garbage = new ChunkId(Guid.CreateVersion7());
        var ack = new TaskCompletionSource<bool>();
        byte[] corrupt = RawChunkCodec.Encode(ChunkFactory.UdpEchoClean(agent.NodeId, start));
        corrupt[^1] ^= 0xFF;
        await agent.SendBinaryAsync(Fleet.Rpc.OutboxFrame.Build(MoShiroChannels.RawChunk, garbage.Value, corrupt));

        // 冒名的块：agent_id 与认证的节点不一致
        RawChunk spoofed = ChunkFactory.UdpEchoClean(new NodeId("agt_someone_else"), start);
        await agent.SendBinaryAsync(Fleet.Rpc.OutboxFrame.Build(MoShiroChannels.RawChunk, spoofed.Header.ChunkId.Value, RawChunkCodec.Encode(spoofed)));

        await BackendFixture.EventuallyAsync(async () =>
        {
            JsonNode events = await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/events?agent_id={agent.NodeId.Value}&kind=chunk_rejected");
            return events["items"]!.AsArray().Count >= 2;
        });
        JsonNode rejected = await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/events?agent_id={agent.NodeId.Value}&kind=chunk_rejected");
        string details = string.Join('|', rejected["items"]!.AsArray().Select(e => e!["detail"]!.GetValue<string>()));
        Assert.Contains("undecodable chunk", details);
        Assert.Contains("does not match the authenticated node", details);

        JsonNode stored = await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/raw/chunks?agent_id={agent.NodeId.Value}");
        Assert.Single(stored["items"]!.AsArray());
    }

    [Fact]
    public async Task Series_HotAndPersistedLayersGiveTheSameExactSummary()
    {
        long baseMs = ClosedMinuteStart(4);
        (AgentSimulator agent, _) = await LineAsync(baseMs, chunks: 18, lossEvery: 10);
        await using (agent)
        {
            long to = baseMs + (3 * Minute);
            string range = $"from={Iso(baseMs)}&to={Iso(to)}&preset=udp,tcp";

            JsonNode hot = await SeriesAsync(agent.NodeId.Value, range + "&step=1s");
            Assert.Equal("hot", hot["source"]!.GetValue<string>());
            Assert.Equal(180, hot["t"]!.AsArray().Count);

            JsonNode minute = null!;
            await BackendFixture.EventuallyAsync(async () =>
            {
                minute = await SeriesAsync(agent.NodeId.Value, range + "&step=1m");
                // 窗口会随着后续的块到达而重算：等到 18 个块的样本都计入之后再比较
                return minute["source"]!.GetValue<string>() == "1m" && minute["summary"]!["samples"]!["udp_echo"]?.GetValue<long>() == 1800
                    && minute["summary"]!["samples"]!["tcp_echo"]?.GetValue<long>() == 3600;
            });

            JsonNode hotSummary = hot["summary"]!["metrics"]!;
            JsonNode minuteSummary = minute["summary"]!["metrics"]!;
            Assert.Equal(10.0, hotSummary["loss.total"]!.GetValue<double>(), 6);
            foreach (string id in new[] { "loss.total", "loss.up", "loss.down", "rtt.p50", "rtt.p95", "rtt.p99", "tcp.rtt.p95", "tcp.retrans", "jitter.down.p95" })
            {
                Assert.True(
                    hotSummary[id] is null == minuteSummary[id] is null && (hotSummary[id] is null || Math.Abs(hotSummary[id]!.GetValue<double>() - minuteSummary[id]!.GetValue<double>()) < 1e-9),
                    $"{id}: hot={hotSummary[id]} persisted={minuteSummary[id]}");
            }

            Assert.Equal(1800, hot["summary"]!["samples"]!["udp_echo"]!.GetValue<long>());
            Assert.Equal(hot["summary"]!["samples"]!["udp_echo"]!.GetValue<long>(), minute["summary"]!["samples"]!["udp_echo"]!.GetValue<long>());
            Assert.Equal(3600, minute["summary"]!["samples"]!["tcp_echo"]!.GetValue<long>());

            // RTT = 2 × 20 ms，直方图的相对误差 ≤ 4.5%
            Assert.InRange(minuteSummary["rtt.p95"]!.GetValue<double>(), 40 * 0.955, 40 * 1.045);
            Assert.Equal(3, minute["t"]!.AsArray().Count);
            Assert.Equal(1.0, minute["summary"]!["coverage"]!.GetValue<double>(), 3);

            // 汇总是精确合并，不是各点的平均：每个分钟桶的丢包率都是 10%，整体也是 10%
            Assert.All(minute["values"]!["loss.total"]!.AsArray(), v => Assert.Equal(10.0, v!.GetValue<double>(), 6));
        }
    }

    [Fact]
    public async Task Series_TenSecondLayer_AndTheFivePointFiveSecondStepsDegradeGracefully()
    {
        long baseMs = ClosedMinuteStart(3);
        (AgentSimulator agent, _) = await LineAsync(baseMs, chunks: 6);
        await using (agent)
        {
            string range = $"from={Iso(baseMs)}&to={Iso(baseMs + Minute)}&preset=udp";
            JsonNode tenSeconds = await SeriesAsync(agent.NodeId.Value, range + "&step=10s");
            Assert.Equal("10s", tenSeconds["source"]!.GetValue<string>());
            Assert.Equal(6, tenSeconds["t"]!.AsArray().Count);
            Assert.All(tenSeconds["values"]!["loss.total"]!.AsArray(), v => Assert.Equal(0.0, v!.GetValue<double>()));

            // 15 s 不能整除任何层的宽度（热层以外）：对更早的数据降级到可用的最细层，并如实标记
            long farBase = ClosedMinuteStart(120);
            JsonNode degraded = await SeriesAsync(agent.NodeId.Value, $"from={Iso(farBase)}&to={Iso(farBase + (10 * Minute))}&step=15s&preset=udp");
            Assert.True(degraded["degraded"]!.GetValue<bool>());
            Assert.Equal("30s", degraded["step"]!.GetValue<string>());
        }
    }

    [Fact]
    public async Task Overview_RanksLinesByScore_AndTheLossyLineScoresLower()
    {
        await _backend.SiteMustBePublicAsync();
        long baseMs = ClosedMinuteStart(3);
        (AgentSimulator good, string goodName) = await LineAsync(baseMs, 12);
        (AgentSimulator bad, string badName) = await LineAsync(baseMs, 12, lossEvery: 5, owdMs: 60);
        await using (good)
        await using (bad)
        {
            string range = $"from={Iso(baseMs)}&to={Iso(baseMs + (2 * Minute))}&preset=udp,scores";
            JsonNode overview = await _backend.GetJsonAsync(_backend.Anonymous, "/api/public/overview?" + range);
            JsonNode[] lines = [.. overview["lines"]!.AsArray().Select(l => l!)];
            int goodIndex = Array.FindIndex(lines, l => l["name"]!.GetValue<string>() == goodName);
            int badIndex = Array.FindIndex(lines, l => l["name"]!.GetValue<string>() == badName);
            Assert.True(goodIndex >= 0 && badIndex >= 0);
            Assert.True(goodIndex < badIndex, "the clean line must rank above the lossy one (default sort: first requested score, descending)");

            double goodScore = lines[goodIndex]["summary"]!["metrics"]!["score.base"]!.GetValue<double>();
            double badScore = lines[badIndex]["summary"]!["metrics"]!["score.base"]!.GetValue<double>();
            Assert.True(goodScore > 90, $"clean line scored {goodScore}");
            Assert.True(badScore < goodScore - 30, $"lossy line scored {badScore}");
            Assert.Equal(20.0, lines[badIndex]["summary"]!["metrics"]!["loss.total"]!.GetValue<double>(), 6);
            Assert.Equal(1200, lines[goodIndex]["summary"]!["samples"]!["udp_echo"]!.GetValue<long>());

            // 按指标排序：loss.total 升序时好线路在前
            JsonNode byLoss = await _backend.GetJsonAsync(_backend.Anonymous, "/api/public/overview?" + range + "&sort=loss.total&order=asc");
            string[] names = [.. byLoss["lines"]!.AsArray().Select(l => l!["name"]!.GetValue<string>())];
            Assert.True(Array.IndexOf(names, goodName) < Array.IndexOf(names, badName));
        }
    }

    [Fact]
    public async Task Compare_Distribution_AndHeatmap_AreConsistentWithTheSeries()
    {
        long baseMs = ClosedMinuteStart(3);
        (AgentSimulator a, _) = await LineAsync(baseMs, 12, owdMs: 20);
        (AgentSimulator b, _) = await LineAsync(baseMs, 12, owdMs: 40);
        await using (a)
        await using (b)
        {
            string window = $"from={Iso(baseMs)}&to={Iso(baseMs + (2 * Minute))}";
            JsonNode compare = await _backend.GetJsonAsync(_backend.Admin, $"/api/public/compare?lines={_backend.Line(a.NodeId.Value)},{_backend.Line(b.NodeId.Value)}&metric=rtt.p50&{window}&step=10s");
            Assert.Equal(12, compare["t"]!.AsArray().Count);
            double rttA = compare["summary"]![_backend.Line(a.NodeId.Value)]!.GetValue<double>();
            double rttB = compare["summary"]![_backend.Line(b.NodeId.Value)]!.GetValue<double>();
            Assert.InRange(rttA, 40 * 0.955, 40 * 1.045);
            Assert.InRange(rttB, 80 * 0.955, 80 * 1.045);

            JsonNode distribution = await _backend.GetJsonAsync(_backend.Admin, $"/api/public/lines/{_backend.Line(b.NodeId.Value)}/distribution?metric=rtt&{window}");
            Assert.Equal(1200, distribution["count"]!.GetValue<long>());
            Assert.Equal(distribution["count"]!.GetValue<long>(), distribution["buckets"]!.AsArray().Sum(x => x!["count"]!.GetValue<long>()));
            Assert.InRange(distribution["p50"]!.GetValue<double>(), 80 * 0.955, 80 * 1.045);

            // step=1m 读的是持久层，由后台分析异步算出：等它完整
            JsonNode heatmap = null!;
            await BackendFixture.EventuallyAsync(async () =>
            {
                heatmap = await _backend.GetJsonAsync(_backend.Admin, $"/api/public/lines/{_backend.Line(b.NodeId.Value)}/heatmap?metric=rtt&{window}&step=1m");
                return heatmap["counts"]!.AsArray().Sum(r => r!.AsArray().Sum(c => c!.GetValue<long>())) == 1200;
            });
            Assert.Equal(2, heatmap["t"]!.AsArray().Count);
            Assert.Equal(2, heatmap["counts"]!.AsArray().Count);
            Assert.All(heatmap["counts"]!.AsArray(), row => Assert.Equal(heatmap["edges"]!.AsArray().Count, row!.AsArray().Count));
            Assert.Equal(1200, heatmap["counts"]!.AsArray().Sum(r => r!.AsArray().Sum(c => c!.GetValue<long>())));

            using HttpResponseMessage notHistogram = await _backend.Admin.GetAsync($"/api/public/lines/{_backend.Line(b.NodeId.Value)}/distribution?metric=loss.up&{window}");
            Assert.Equal(HttpStatusCode.UnprocessableEntity, notHistogram.StatusCode);
        }
    }

    [Fact]
    public async Task Queries_ValidateTheirParameters_WithActionableErrors()
    {
        await using AgentSimulator agent = await _backend.ConnectAgentAsync();
        string id = agent.NodeId.Value;

        async Task<(HttpStatusCode Status, string Code, string Detail)> Fail(string url)
        {
            using HttpResponseMessage response = await _backend.Admin.GetAsync(url);
            JsonNode problem = JsonNode.Parse(await response.Content.ReadAsStringAsync())!;
            return (response.StatusCode, problem["code"]!.GetValue<string>(), problem["detail"]?.GetValue<string>() ?? "");
        }

        Assert.Equal((HttpStatusCode.UnprocessableEntity, "invalid_range"), Trim(await Fail($"/api/public/lines/{_backend.Line(id)}/series?range=abc")));
        Assert.Equal((HttpStatusCode.UnprocessableEntity, "invalid_range"), Trim(await Fail($"/api/public/lines/{_backend.Line(id)}/series?from=2026-01-02T00:00:00Z&to=2026-01-01T00:00:00Z")));
        Assert.Equal((HttpStatusCode.UnprocessableEntity, "invalid_step"), Trim(await Fail($"/api/public/lines/{_backend.Line(id)}/series?range=1h&step=7s")));
        Assert.Equal((HttpStatusCode.UnprocessableEntity, "unknown_metric"), Trim(await Fail($"/api/public/lines/{_backend.Line(id)}/series?range=1h&metrics=bogus.metric")));
        Assert.Equal((HttpStatusCode.UnprocessableEntity, "unknown_preset"), Trim(await Fail($"/api/public/lines/{_backend.Line(id)}/series?range=1h&preset=zzz")));
        Assert.Equal((HttpStatusCode.BadRequest, "validation_failed"), Trim(await Fail($"/api/public/lines/{_backend.Line(id)}/series?range=1h&from=2026-01-01T00:00:00Z")));
        Assert.Equal((HttpStatusCode.NotFound, "agent_not_found"), Trim(await Fail("/api/public/lines/agt_nope/series?range=1h")));

        (HttpStatusCode, string, string Detail) tooLarge = await Fail($"/api/public/lines/{_backend.Line(id)}/series?range=30d&step=1m");
        Assert.Equal(HttpStatusCode.UnprocessableEntity, tooLarge.Item1);
        Assert.Equal("range_too_large", tooLarge.Item2);
        Assert.Contains("step=", tooLarge.Detail); // 给出可行的替代方案，而不是悄悄截断

        // 重复的指标 id 只算一个；真正超过 32 个不同的指标才报错
        string repeated = string.Join(',', Enumerable.Range(0, 40).Select(_ => "rtt.p95"));
        using HttpResponseMessage deduplicated = await _backend.Admin.GetAsync($"/api/public/lines/{_backend.Line(id)}/series?range=5m&metrics={repeated}");
        Assert.Equal(HttpStatusCode.OK, deduplicated.StatusCode);
        JsonNode catalog = await _backend.GetJsonAsync(_backend.Admin, "/api/public/catalog/metrics");
        string distinct = string.Join(',', catalog["items"]!.AsArray().Select(m => m!["id"]!.GetValue<string>()).Take(40));
        Assert.Equal((HttpStatusCode.UnprocessableEntity, "too_many_metrics"), Trim(await Fail($"/api/public/lines/{_backend.Line(id)}/series?range=5m&metrics={distinct}")));

        static (HttpStatusCode, string) Trim((HttpStatusCode Status, string Code, string Detail) f) => (f.Status, f.Code);
    }

    [Fact]
    public async Task ClosedRanges_AreImmutableAndConditional_RelativeRangesAreNeverCached()
    {
        long baseMs = ClosedMinuteStart(10);
        (AgentSimulator agent, _) = await LineAsync(baseMs, 6, tcp: false);
        await using (agent)
        {
            string url = $"/api/public/lines/{_backend.Line(agent.NodeId.Value)}/series?from={Iso(baseMs)}&to={Iso(baseMs + Minute)}&step=10s&preset=udp";
            using HttpResponseMessage closed = await _backend.Admin.GetAsync(url);
            Assert.Equal("public, max-age=31536000, immutable", closed.Headers.CacheControl!.ToString());
            string etag = closed.Headers.ETag!.Tag;

            using var conditional = new HttpRequestMessage(HttpMethod.Get, url);
            conditional.Headers.IfNoneMatch.Add(new EntityTagHeaderValue(etag));
            using HttpResponseMessage notModified = await _backend.Admin.SendAsync(conditional);
            Assert.Equal(HttpStatusCode.NotModified, notModified.StatusCode);

            using HttpResponseMessage relative = await _backend.Admin.GetAsync($"/api/public/lines/{_backend.Line(agent.NodeId.Value)}/series?range=5m&preset=udp");
            Assert.Equal("no-store", relative.Headers.CacheControl!.ToString());
            Assert.Null(relative.Headers.ETag);
        }
    }

    [Fact]
    public async Task HiddenLines_AreInvisibleToAnonymousVisitors_ButNotToSignedInOnes()
    {
        await _backend.SiteMustBePublicAsync();
        long baseMs = ClosedMinuteStart(3);
        (AgentSimulator agent, string name) = await LineAsync(baseMs, 3, tcp: false);
        await using (agent)
        {
            using HttpResponseMessage hide = await _backend.Admin.PatchAsync(
                "/api/admin/agents/" + agent.NodeId.Value, BackendFixture.Json(new JsonObject { ["public_visible"] = false }, "application/merge-patch+json"));
            Assert.Equal(HttpStatusCode.OK, hide.StatusCode);

            JsonNode anonymousOverview = await _backend.GetJsonAsync(_backend.Anonymous, "/api/public/overview?range=15m");
            Assert.DoesNotContain(anonymousOverview["lines"]!.AsArray(), l => l!["name"]!.GetValue<string>() == name);
            using HttpResponseMessage anonymousSeries = await _backend.Anonymous.GetAsync($"/api/public/lines/{_backend.Line(agent.NodeId.Value)}/series?range=5m");
            Assert.Equal(HttpStatusCode.NotFound, anonymousSeries.StatusCode); // 404，不是 403：不暴露线路是否存在

            JsonNode adminOverview = await _backend.GetJsonAsync(_backend.Admin, "/api/public/overview?range=15m");
            Assert.Contains(adminOverview["lines"]!.AsArray(), l => l!["name"]!.GetValue<string>() == name);
        }
    }

    [Fact]
    public async Task LateChunks_UpdateAlreadyComputedWindows()
    {
        long baseMs = ClosedMinuteStart(5);
        (AgentSimulator agent, _) = await LineAsync(baseMs, 6, tcp: false);
        await using (agent)
        {
            string url = $"/api/public/lines/{_backend.Line(agent.NodeId.Value)}/series?from={Iso(baseMs)}&to={Iso(baseMs + Minute)}&step=1m&preset=udp";
            await BackendFixture.EventuallyAsync(async () => (await _backend.GetJsonAsync(_backend.Admin, url))["summary"]!["samples"]!["udp_echo"]?.GetValue<long>() == 600);

            // Agent 重连后补传一个迟到的块（落在同一个分钟窗口里）
            RawChunk late = ChunkFactory.Echo(agent.NodeId, "udp_echo", baseMs + 5_000, 100, 10, _ => new EchoPacket(false, false, 20_000, 20_000));
            Assert.True(await agent.SendChunkAsync(late));
            await BackendFixture.EventuallyAsync(async () => (await _backend.GetJsonAsync(_backend.Admin, url))["summary"]!["samples"]!["udp_echo"]?.GetValue<long>() == 700, because: "the late chunk must re-enter the persistent path");
        }
    }

    [Fact]
    public async Task RolledUpLayers_AgreeWithTheMinuteLayer()
    {
        long baseMs = ClosedMinuteStart(7) / (5 * Minute) * (5 * Minute); // 对齐到 5 分钟，整段都在已经结束的时间里
        (AgentSimulator agent, _) = await LineAsync(baseMs, 30, lossEvery: 20, tcp: false);
        await using (agent)
        {
            long to = baseMs + (5 * Minute);
            string range = $"from={Iso(baseMs)}&to={Iso(to)}&preset=udp";
            JsonNode oneMinute = null!;
            await BackendFixture.EventuallyAsync(async () =>
            {
                oneMinute = await SeriesAsync(agent.NodeId.Value, range + "&step=1m");
                return oneMinute["summary"]!["samples"]!["udp_echo"]?.GetValue<long>() == 3000;
            });

            // 5 分钟层由降采样把 5 个 1 分钟窗口精确合并而来
            JsonNode fiveMinutes = null!;
            await BackendFixture.EventuallyAsync(async () =>
            {
                fiveMinutes = await SeriesAsync(agent.NodeId.Value, range + "&step=5m");
                return fiveMinutes["source"]!.GetValue<string>() == "5m" && fiveMinutes["summary"]!["samples"]!["udp_echo"]?.GetValue<long>() == 3000;
            });
            Assert.Equal(oneMinute["summary"]!["metrics"]!["loss.total"]!.GetValue<double>(), fiveMinutes["summary"]!["metrics"]!["loss.total"]!.GetValue<double>(), 9);
            Assert.Equal(oneMinute["summary"]!["metrics"]!["rtt.p99"]!.GetValue<double>(), fiveMinutes["summary"]!["metrics"]!["rtt.p99"]!.GetValue<double>(), 9);
        }
    }
}