lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadBoundsTests.cs

using Microsoft.Extensions.Logging.Abstractions;
using Opc.Ua;
using Opc.Ua.Server;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using HistorianRead = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult;
using SdkHistoryReadResult = Opc.Ua.HistoryReadResult;

namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;

/// <summary>Boots ONE real SDK server for the whole bounds suite (the cases swap the historian source per read).</summary>
public sealed class BoundsServerFixture : IAsyncLifetime
{
    private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-bounds-{Guid.NewGuid():N}");
    private OpcUaApplicationHost? _host;

    /// <summary>The node manager under test.</summary>
    public OtOpcUaNodeManager Nm { get; private set; } = null!;

    /// <summary>The historized variable every case reads.</summary>
    public NodeId NodeId { get; private set; } = null!;

    /// <summary>Serialises the cases (they share the node manager's source).</summary>
    public SemaphoreSlim Gate { get; } = new(1, 1);

    public async ValueTask InitializeAsync()
    {
        var started = await OpcUaHostStart.StartAsync(
            port => new OpcUaApplicationHost(
                new OpcUaApplicationHostOptions
                {
                    ApplicationName = "OtOpcUa.BoundsTest",
                    ApplicationUri = $"urn:OtOpcUa.BoundsTest:{Guid.NewGuid():N}",
                    OpcUaPort = port,
                    PublicHostname = "localhost",
                    PkiStoreRoot = _pkiRoot,
                },
                NullLogger<OpcUaApplicationHost>.Instance),
            () => new OtOpcUaSdkServer(), CancellationToken.None);
        _host = started.Host;
        Nm = started.Server.NodeManager!;
        Nm.HistoryContinuationStore = new InMemoryHistoryContinuationStore();
        Nm.EnsureVariable("eq-1/bounds", parentFolderNodeId: null, displayName: "Bounds", dataType: "Double",
            writable: false, historianTagname: "WW.Bounds", realm: AddressSpaceRealm.Uns);
        NodeId = Nm.TryGetVariable("eq-1/bounds")!.NodeId;
    }

    public async ValueTask DisposeAsync()
    {
        if (_host is not null) await _host.DisposeAsync();
        if (Directory.Exists(_pkiRoot))
        {
            try { Directory.Delete(_pkiRoot, recursive: true); }
            catch { /* best-effort */ }
        }
    }
}

/// <summary>
///     Gitea #691: HistoryRead-Raw <c>ReturnBounds</c> (OPC UA Part 11 §4.6 bounding values, §6.5.3.2). The oracle is Part 11
///     Table 1 itself — every row, over its data set (samples at 5:00, 5:02, 5:03, 5:05, 5:06) — run against three backend
///     shapes: one that reads newest-first natively (SqlServer-like), one that reads forward only (Gateway-like, walked back
///     by <see cref="ReverseRawReader"/>) and a microsecond one (QuestDB-like,
///     truncating its window bounds).
/// </summary>
public sealed class NodeManagerHistoryReadBoundsTests(BoundsServerFixture fx) : IClassFixture<BoundsServerFixture>
{
    private static readonly DateTime Day = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
    private static readonly DateTime Unspecified = DateTime.MinValue;

    private static DateTime T(string hhmm)
    {
        var parts = hhmm.Split(':');
        return Day.AddHours(int.Parse(parts[0])).AddMinutes(int.Parse(parts[1]));
    }

    /// <summary>Part 11 Table 1's data: value = minutes past 5:00.</summary>
    private static DataValueSnapshot[] TableData() =>
        new[] { "5:00", "5:02", "5:03", "5:05", "5:06" }
            .Select(s => new DataValueSnapshot((double)(T(s) - T("5:00")).TotalMinutes, StatusCodes.Good, T(s), T(s)))
            .ToArray();

    public enum Shape { Reverse, ForwardOnly, Microsecond }

    private static IHistorianDataSource Source(Shape shape, IReadOnlyList<DataValueSnapshot> series) => shape switch
    {
        Shape.Reverse => new SeriesSource(series, reverse: true, inclusiveForwardEnd: false, resolution: 1),
        Shape.ForwardOnly => new SeriesSource(series, reverse: false, inclusiveForwardEnd: false, resolution: 1),
        _ => new SeriesSource(series, reverse: true, inclusiveForwardEnd: false, resolution: 10),
    };

    /// <summary>
    ///     Part 11 Table 1, every row, verbatim. "FIRST"/"LAST" are <c>Bad_BoundNotFound</c> placeholders (null value) at the
    ///     StartTime/EndTime; a/b footnotes: one second after/before the previous returned value when that side is unspecified.
    /// </summary>
    public static TheoryData<string, string, uint, bool, string> Table1 => new()
    {
        { "5:00", "5:05", 0, true, "5:00, 5:02, 5:03, 5:05" },
        { "5:00", "5:05", 0, false, "5:00, 5:02, 5:03" },
        { "5:01", "5:04", 0, true, "5:00, 5:02, 5:03, 5:05" },
        { "5:01", "5:04", 0, false, "5:02, 5:03" },
        { "5:05", "5:00", 0, true, "5:05, 5:03, 5:02, 5:00" },
        { "5:05", "5:00", 0, false, "5:05, 5:03, 5:02" },
        { "5:04", "5:01", 0, true, "5:05, 5:03, 5:02, 5:00" },
        { "5:04", "5:01", 0, false, "5:03, 5:02" },
        { "4:59", "5:05", 0, true, "FIRST, 5:00, 5:02, 5:03, 5:05" },
        { "4:59", "5:05", 0, false, "5:00, 5:02, 5:03" },
        { "5:01", "5:07", 0, true, "5:00, 5:02, 5:03, 5:05, 5:06, LAST" },
        { "5:01", "5:07", 0, false, "5:02, 5:03, 5:05, 5:06" },
        { "5:00", "5:05", 3, true, "5:00, 5:02, 5:03" },
        { "5:00", "5:05", 3, false, "5:00, 5:02, 5:03" },
        { "5:01", "5:04", 3, true, "5:00, 5:02, 5:03" },
        { "5:01", "5:04", 3, false, "5:02, 5:03" },
        { "5:05", "5:00", 3, true, "5:05, 5:03, 5:02" },
        { "5:05", "5:00", 3, false, "5:05, 5:03, 5:02" },
        { "5:04", "5:01", 3, true, "5:05, 5:03, 5:02" },
        { "5:04", "5:01", 3, false, "5:03, 5:02" },
        { "4:59", "5:05", 3, true, "FIRST, 5:00, 5:02" },
        { "4:59", "5:05", 3, false, "5:00, 5:02, 5:03" },
        { "5:01", "5:07", 3, true, "5:00, 5:02, 5:03" },
        { "5:01", "5:07", 3, false, "5:02, 5:03, 5:05" },
        { "5:00", "", 3, true, "5:00, 5:02, 5:03" },
        { "5:00", "", 3, false, "5:00, 5:02, 5:03" },
        { "5:00", "", 6, true, "5:00, 5:02, 5:03, 5:05, 5:06, LAST+" },
        { "5:00", "", 6, false, "5:00, 5:02, 5:03, 5:05, 5:06" },
        { "5:07", "", 6, true, "5:06, LAST+" },
        { "5:07", "", 6, false, "" },
        { "", "5:06", 3, true, "5:06, 5:05, 5:03" },
        { "", "5:06", 3, false, "5:06, 5:05, 5:03" },
        { "", "5:06", 6, true, "5:06, 5:05, 5:03, 5:02, 5:00, FIRST-" },
        { "", "5:06", 6, false, "5:06, 5:05, 5:03, 5:02, 5:00" },
        { "", "4:48", 6, true, "5:00, FIRST-" },
        { "", "4:48", 6, false, "" },
        { "4:48", "4:48", 0, true, "FIRST, 5:00" },
        { "4:48", "4:48", 0, false, "" },
        { "4:48", "4:48", 1, true, "FIRST" },
        { "4:48", "4:48", 1, false, "" },
        { "4:48", "4:48", 2, true, "FIRST, 5:00" },
        { "5:00", "5:00", 0, true, "5:00, 5:02" },
        { "5:00", "5:00", 0, false, "5:00" },
        { "5:00", "5:00", 1, true, "5:00" },
        { "5:00", "5:00", 1, false, "5:00" },
        { "5:01", "5:01", 0, true, "5:00, 5:02" },
        { "5:01", "5:01", 0, false, "" },
        { "5:01", "5:01", 1, true, "5:00" },
        { "5:01", "5:01", 1, false, "" },
    };

    [Theory]
    [MemberData(nameof(Table1))]
    public async Task Part11_table1_first_page_on_a_reverse_capable_backend(string start, string end, uint max, bool bounds, string expected) =>
        await AssertFirstPage(Shape.Reverse, start, end, max, bounds, expected);

    [Theory]
    [MemberData(nameof(Table1))]
    public async Task Part11_table1_first_page_on_a_forward_only_backend(string start, string end, uint max, bool bounds, string expected) =>
        await AssertFirstPage(Shape.ForwardOnly, start, end, max, bounds, expected);

    [Theory]
    [MemberData(nameof(Table1))]
    public async Task Part11_table1_first_page_on_a_microsecond_backend(string start, string end, uint max, bool bounds, string expected) =>
        await AssertFirstPage(Shape.Microsecond, start, end, max, bounds, expected);

    private async Task AssertFirstPage(Shape shape, string start, string end, uint max, bool bounds, string expected)
    {
        var s = start.Length == 0 ? Unspecified : T(start);
        var e = end.Length == 0 ? Unspecified : T(end);
        var source = Source(shape, TableData());
        var (values, error, _) = await ReadAsync(source, s, e, max, bounds, cp: null);

        error.StatusCode.Code.ShouldBe(StatusCodes.Good);
        Render(values, s, e).ShouldBe(expected);
        (source as SeriesSource)?.ReverseWindowsOnForwardOnly.ShouldBe(0);
    }

    /// <summary>
    ///     Part 11: bounds count toward NumValuesPerNode, the start bound comes with the first page and the end bound with the
    ///     last — so paging a bounded two-sided read at any page size concatenates to the unpaged (N = 0) answer.
    /// </summary>
    [Theory]
    [InlineData(Shape.Reverse)]
    [InlineData(Shape.ForwardOnly)]
    [InlineData(Shape.Microsecond)]
    public async Task Paging_a_bounded_read_concatenates_to_the_unpaged_answer(Shape shape)
    {
        var windows = new[]
        {
            ("5:00", "5:05"), ("5:01", "5:04"), ("5:05", "5:00"), ("5:04", "5:01"), ("4:59", "5:05"), ("5:01", "5:07"),
            ("4:48", "4:48"), ("5:00", "5:00"), ("5:01", "5:01"), ("5:07", "4:00"), ("5:06", "5:06"),
        };
        foreach (var (start, end) in windows)
        {
            var source = Source(shape, TableData());
            var all = await ReadAsync(source, T(start), T(end), 0, true, cp: null);
            all.Cp.ShouldBeNull();
            var expected = Render(all.Values, T(start), T(end));
            for (uint n = 1; n <= 6; n++)
            {
                var paged = await DrainAsync(source, T(start), T(end), n, bounds: true);
                Render(paged, T(start), T(end)).ShouldBe(expected, $"{start}->{end} paged by {n}");
            }
        }
    }

    /// <summary>One-sided reads page to the open placeholder exactly once (footnotes a/b), never past it.</summary>
    [Theory]
    [InlineData(Shape.Reverse)]
    [InlineData(Shape.ForwardOnly)]
    [InlineData(Shape.Microsecond)]
    public async Task Paging_a_one_sided_bounded_read_ends_with_the_open_placeholder(Shape shape)
    {
        for (uint n = 1; n <= 7; n++)
        {
            var fwd = await DrainAsync(Source(shape, TableData()), T("5:01"), Unspecified, n, bounds: true);
            Render(fwd, T("5:01"), Unspecified).ShouldBe("5:00, 5:02, 5:03, 5:05, 5:06, LAST+", $"forward by {n}");
            var rev = await DrainAsync(Source(shape, TableData()), Unspecified, T("5:04"), n, bounds: true);
            Render(rev, Unspecified, T("5:04")).ShouldBe("5:05, 5:03, 5:02, 5:00, FIRST-", $"reverse by {n}");
        }
    }

    /// <summary>A bound is a stored sample of any quality: a Bad sample before StartTime is the start bound, as stored.</summary>
    [Theory]
    [InlineData(Shape.Reverse)]
    [InlineData(Shape.ForwardOnly)]
    [InlineData(Shape.Microsecond)]
    public async Task A_bad_sample_is_returned_as_the_bound_as_stored(Shape shape)
    {
        var series = TableData();
        series[0] = new DataValueSnapshot(null, StatusCodes.BadNoCommunication, T("5:00"), T("5:00"));
        var (values, _, _) = await ReadAsync(Source(shape, series), T("5:01"), T("5:04"), 0, true, cp: null);

        values[0].StatusCode.Code.ShouldBe(StatusCodes.BadNoCommunication);
        values[0].SourceTimestamp.ShouldBe(T("5:00"));
        values.Count.ShouldBe(4);
    }

    /// <summary>
    ///     Ties on a bound: the start bound before a forward window is the LAST of a tie cluster (the sample immediately
    ///     before StartTime in the read's order); the end bound after it is the FIRST of one; and ties ON StartTime are all
    ///     data (the bound is the first of them, so nothing extra is added).
    /// </summary>
    [Theory]
    [InlineData(Shape.Reverse)]
    [InlineData(Shape.ForwardOnly)]
    [InlineData(Shape.Microsecond)]
    public async Task Tie_clusters_on_either_side_yield_one_bound_each(Shape shape)
    {
        var series = new List<DataValueSnapshot>();
        void Add(string t, double v) => series.Add(new DataValueSnapshot(v, StatusCodes.Good, T(t), T(t)));
        Add("5:00", 1); Add("5:00", 2); Add("5:00", 3);
        Add("5:02", 4);
        Add("5:05", 5); Add("5:05", 6);

        var (fwd, _, _) = await ReadAsync(Source(shape, series), T("5:01"), T("5:04"), 0, true, cp: null);
        fwd.Select(v => (double)v.Value).ShouldBe([3d, 4d, 5d]);

        var (rev, _, _) = await ReadAsync(Source(shape, series), T("5:04"), T("5:01"), 0, true, cp: null);
        rev.Select(v => (double)v.Value).ShouldBe([5d, 4d, 3d]);

        var (onStart, _, _) = await ReadAsync(Source(shape, series), T("5:00"), T("5:03"), 0, true, cp: null);
        onStart.Select(v => (double)v.Value).ShouldBe([1d, 2d, 3d, 4d, 5d]);
    }

    /// <summary>
    ///     QuestDB truncates the window bounds it is sent to the microsecond: a sample at ⌊StartTime⌋ is inside a window that
    ///     starts a few ticks later, so it is data — never returned a second time as the start bound.
    /// </summary>
    [Fact]
    public async Task A_microsecond_backend_does_not_return_a_sample_at_the_truncated_start_twice()
    {
        var series = TableData();
        var (values, _, _) = await ReadAsync(Source(Shape.Microsecond, series), T("5:00").AddTicks(7), T("5:04"), 0, true, cp: null);

        values.Select(v => (double)v.Value).ShouldBe([0d, 2d, 3d, 5d]);
    }

    /// <summary>
    ///     A backend whose forward end is inclusive (the gateway's own <c>ReadRaw</c>), or a resumed read that steps onto
    ///     EndTime, returns the sample AT EndTime as data — it must come back once, as the end bound.
    /// </summary>
    [Theory]
    [InlineData(0u)]
    [InlineData(1u)]
    [InlineData(2u)]
    [InlineData(3u)]
    public async Task An_inclusive_forward_end_does_not_return_the_end_bound_twice(uint max)
    {
        var source = new SeriesSource(TableData(), reverse: false, inclusiveForwardEnd: true, resolution: 1);
        var values = await DrainAsync(source, T("5:00"), T("5:05"), max, bounds: true);
        Render(values, T("5:00"), T("5:05")).ShouldBe("5:00, 5:02, 5:03, 5:05");
    }

    /// <summary>
    ///     #736 (H1 fuzz) — a backend row ceiling sizes the page BELOW NumValuesPerNode (RawPageLimit: ceiling 4 ⇒ 3 data
    ///     rows per page). When the data exactly fills that page, the end bound must go to the NEXT page through the
    ///     continuation point, not be appended to a page that is already at its effective cap: the fit used to be
    ///     judged against NumValuesPerNode (10 here), so the page came back with cap + 1 values.
    /// </summary>
    [Fact]
    public async Task A_full_ceiling_sized_page_carries_the_end_bound_on_the_next_page()
    {
        var source = new SeriesSource(TableData(), reverse: true, inclusiveForwardEnd: false, resolution: 1, ceiling: 4);
        var (first, error, cp) = await ReadAsync(source, T("5:01"), T("5:06"), max: 10, bounds: true, cp: null);
        error.StatusCode.Code.ShouldBe(StatusCodes.Good);

        // Start bound 5:00 + the three data rows the ceiling allows = the page's effective cap, exactly.
        Render(first, T("5:01"), T("5:06")).ShouldBe("5:00, 5:02, 5:03, 5:05");
        cp.ShouldNotBeNull("the page is full, so the end bound is owed on the next page");

        var (second, error2, cp2) = await ReadAsync(source, T("5:01"), T("5:06"), max: 10, bounds: true, cp);
        error2.StatusCode.Code.ShouldBe(StatusCodes.Good);
        Render(second, T("5:01"), T("5:06")).ShouldBe("5:06");
        cp2.ShouldBeNull();

        // Control: without a ceiling the same read fits on ONE page of NumValuesPerNode — the fit keys on the
        // effective cap, not on a smaller constant.
        var (whole, _, cpWhole) = await ReadAsync(Source(Shape.Reverse, TableData()), T("5:01"), T("5:06"), max: 10, bounds: true, cp: null);
        Render(whole, T("5:01"), T("5:06")).ShouldBe("5:00, 5:02, 5:03, 5:05, 5:06");
        cpWhole.ShouldBeNull();
    }

    /// <summary>
    ///     #691 review: no data AND no bound found ⇒ <c>GoodNoData</c> with no values — the answer a tag with no history (and a
    ///     disabled historian) gave before bounds (§6.5.3.2: bounds make the result Good only "if one or both exist"). Only
    ///     when the WHOLE read is empty: a single bound found keeps the placeholders (Table 1: "4:48 4:48 1 ⇒ FIRST").
    /// </summary>
    [Theory]
    [InlineData(Shape.Reverse)]
    [InlineData(Shape.ForwardOnly)]
    [InlineData(Shape.Microsecond)]
    public async Task No_data_and_no_bound_found_is_GoodNoData_with_no_values(Shape shape)
    {
        var windows = new (string S, string E)[] { ("5:00", "5:05"), ("5:05", "5:00"), ("5:00", "5:00"), ("5:00", ""), ("", "5:05") };
        foreach (var (start, end) in windows)
        {
            var s = start.Length == 0 ? Unspecified : T(start);
            var e = end.Length == 0 ? Unspecified : T(end);
            foreach (var n in new uint[] { 0, 1, 2, 3 })
            {
                if (n == 0 && (s == Unspecified || e == Unspecified)) continue; // a one-sided read needs a cap
                var (values, error, cp) = await ReadResultAsync(Source(shape, []), s, e, n);
                values.ShouldBeEmpty($"{start}->{end} by {n}");
                cp.ShouldBeNull();
                error.StatusCode.Code.ShouldBe(StatusCodes.Good);
                LastNodeStatus.Code.ShouldBe(StatusCodes.GoodNoData, $"{start}->{end} by {n}");
            }
        }

        // Positive control: one bound found keeps the missing one's placeholder.
        var single = await DrainAsync(Source(shape, TableData()), T("4:48"), T("4:48"), 1, bounds: true);
        Render(single, T("4:48"), T("4:48")).ShouldBe("FIRST, 5:00");
    }

    /// <summary>
    ///     #691 review: <c>ServerHistorian:HonorReturnBounds = false</c> restores the pre-#691 answer — ReturnBounds is ignored
    ///     (Table 1's "No" rows for every "Yes" row) and a modified read with ReturnBounds is not refused for it.
    /// </summary>
    [Fact]
    public async Task The_kill_switch_ignores_return_bounds()
    {
        await fx.Gate.WaitAsync(TestContext.Current.CancellationToken);
        try
        {
            fx.Nm.HonorReturnBounds = false;
            fx.Nm.HistorianDataSource = Source(Shape.Reverse, TableData());
            var (values, _, _) = Read(T("5:01"), T("5:07"), 0, bounds: true, cp: null, modified: false);
            Render(values, T("5:01"), T("5:07")).ShouldBe("5:02, 5:03, 5:05, 5:06");

            var (_, modErr, _) = Read(T("5:00"), T("5:05"), 0, bounds: true, cp: null, modified: true);
            modErr.StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported, "no trail on this fake — not InvalidArgument");
        }
        finally
        {
            fx.Nm.HonorReturnBounds = true;
            fx.Gate.Release();
        }
    }

    [Fact]
    public async Task Bounds_are_honoured_by_default()
    {
        await fx.Gate.WaitAsync(TestContext.Current.CancellationToken);
        try { fx.Nm.HonorReturnBounds.ShouldBeTrue(); }
        finally { fx.Gate.Release(); }
    }

    /// <summary>Part 11 §6.5.3.3: a modified read with ReturnBounds is <c>Bad_InvalidArgument</c>.</summary>
    [Fact]
    public async Task A_modified_read_with_return_bounds_is_an_invalid_argument()
    {
        await fx.Gate.WaitAsync(TestContext.Current.CancellationToken);
        try
        {
            fx.Nm.HistorianDataSource = Source(Shape.Reverse, TableData());
            var (_, error, cp) = Read(T("5:00"), T("5:05"), 0, bounds: true, cp: null, modified: true);
            error.StatusCode.Code.ShouldBe(StatusCodes.BadInvalidArgument);
            cp.ShouldBeNull();
        }
        finally { fx.Gate.Release(); }
    }

    // --- helpers --------------------------------------------------------------------------------

    private static string Render(IReadOnlyList<DataValue> values, DateTime start, DateTime end)
    {
        var parts = new List<string>();
        for (var i = 0; i < values.Count; i++)
        {
            var v = values[i];
            if (v.StatusCode.Code != StatusCodes.BadBoundNotFound)
            {
                v.Value.ShouldNotBeNull();
                parts.Add($"{v.SourceTimestamp.Hour}:{v.SourceTimestamp.Minute:00}");
                continue;
            }

            v.Value.ShouldBeNull("a missing bound carries a null value");
            var isFirst = i == 0 && start != Unspecified && v.SourceTimestamp == start;
            var isLast = end != Unspecified && v.SourceTimestamp == end;
            if (isFirst) { parts.Add("FIRST"); continue; }
            if (isLast) { parts.Add("LAST"); continue; }
            // Open side (footnotes a/b): one second from the previous returned value.
            i.ShouldBeGreaterThan(0);
            var prev = values[i - 1].SourceTimestamp;
            if (v.SourceTimestamp == prev.AddSeconds(1)) parts.Add("LAST+");
            else if (v.SourceTimestamp == prev.AddSeconds(-1)) parts.Add("FIRST-");
            else parts.Add($"?{v.SourceTimestamp:O}");
        }

        return string.Join(", ", parts);
    }

    private async Task<List<DataValue>> DrainAsync(IHistorianDataSource source, DateTime start, DateTime end, uint max, bool bounds)
    {
        var all = new List<DataValue>();
        byte[]? cp = null;
        var pages = 0;
        do
        {
            var (values, error, next) = await ReadAsync(source, start, end, max, bounds, cp);
            error.StatusCode.Code.ShouldBe(StatusCodes.Good);
            if (max > 0) values.Count.ShouldBeLessThanOrEqualTo((int)max, "bounds count toward NumValuesPerNode");
            all.AddRange(values);
            cp = next;
            (++pages).ShouldBeLessThan(50, "paging must terminate");
        }
        while (cp is not null);

        return all;
    }

    /// <summary>The node-level StatusCode of the last <see cref="Read"/> (GoodNoData vs Good).</summary>
    private StatusCode LastNodeStatus;

    private async Task<(IReadOnlyList<DataValue> Values, ServiceResult Error, byte[]? Cp)> ReadResultAsync(
        IHistorianDataSource source, DateTime start, DateTime end, uint max) =>
        await ReadAsync(source, start, end, max, bounds: true, cp: null);

    private async Task<(IReadOnlyList<DataValue> Values, ServiceResult Error, byte[]? Cp)> ReadAsync(
        IHistorianDataSource source, DateTime start, DateTime end, uint max, bool bounds, byte[]? cp)
    {
        await fx.Gate.WaitAsync(TestContext.Current.CancellationToken);
        try
        {
            fx.Nm.HistorianDataSource = source;
            return Read(start, end, max, bounds, cp, modified: false);
        }
        finally { fx.Gate.Release(); }
    }

    private (IReadOnlyList<DataValue> Values, ServiceResult Error, byte[]? Cp) Read(
        DateTime start, DateTime end, uint max, bool bounds, byte[]? cp, bool modified)
    {
        var context = new OperationContext(new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null);
        var details = new ReadRawModifiedDetails
        {
            StartTime = start, EndTime = end, NumValuesPerNode = max, ReturnBounds = bounds, IsReadModified = modified,
        };
        var nodesToRead = new List<HistoryReadValueId> { new() { NodeId = fx.NodeId, ContinuationPoint = cp } };
        var results = new List<SdkHistoryReadResult> { null! };
        var errors = new List<ServiceResult> { null! };
        fx.Nm.HistoryRead(context, details, TimestampsToReturn.Both, releaseContinuationPoints: false, nodesToRead, results, errors);
        LastNodeStatus = results[0]?.StatusCode ?? StatusCodes.Bad;

        var values = new List<DataValue>();
        if (results[0]?.HistoryData is { } ho && !ExtensionObject.IsNull(ho))
            values.AddRange(((HistoryData)ExtensionObject.ToEncodeable(ho)).DataValues);
        return (values, errors[0], results[0]?.ContinuationPoint);
    }

    /// <summary>
    ///     A series-backed historian. <paramref name="reverse"/>: serves <c>start &gt; end</c> as <c>(end, start]</c>
    ///     newest-first (SqlServer/QuestDB); otherwise rejects it (the Gateway) and counts that. The forward read is
    ///     <c>[start, end)</c>, or <c>[start, end]</c> when <paramref name="inclusiveForwardEnd"/> (the gateway's own end);
    ///     a point read is the instant. <paramref name="resolution"/> &gt; 1 truncates every window bound to it first (QuestDB
    ///     µs); the series must already be stored at that resolution.
    /// </summary>
    private sealed class SeriesSource(IReadOnlyList<DataValueSnapshot> series, bool reverse, bool inclusiveForwardEnd, long resolution, int ceiling = 0)
        : IHistorianDataSource
    {
        public int ReverseWindowsOnForwardOnly;

        public bool SupportsReverseRaw => reverse;

        public long TimestampResolutionTicks => resolution;

        /// <summary>A strict row ceiling (0 = none), so a page is sized below NumValuesPerNode (RawPageLimit).</summary>
        public int MaxRawValuesPerRead => ceiling;

        private DateTime Trunc(DateTime t) => new(t.Ticks - t.Ticks % resolution, DateTimeKind.Utc);

        public Task<HistorianRead> ReadRawAsync(
            string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, CancellationToken cancellationToken)
        {
            static DateTime Ts(DataValueSnapshot s) => s.SourceTimestampUtc ?? DateTime.MinValue;
            startUtc = Trunc(startUtc);
            endUtc = Trunc(endUtc);
            List<DataValueSnapshot> window;
            if (startUtc > endUtc)
            {
                if (!reverse)
                {
                    Interlocked.Increment(ref ReverseWindowsOnForwardOnly);
                    throw new InvalidOperationException("start_time must be less than or equal to end_time. (InvalidArgument)");
                }

                window = series.Where(s => Ts(s) > endUtc && Ts(s) <= startUtc).Reverse().ToList();
            }
            else if (startUtc == endUtc) window = series.Where(s => Ts(s) == startUtc).ToList();
            else window = series.Where(s => Ts(s) >= startUtc && (inclusiveForwardEnd ? Ts(s) <= endUtc : Ts(s) < endUtc)).ToList();

            var page = maxValuesPerNode == 0 ? window : window.Take((int)maxValuesPerNode).ToList();
            return Task.FromResult(new HistorianRead(page, null));
        }

        public Task<HistorianRead> ReadProcessedAsync(
            string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
            HistoryAggregateType aggregate, CancellationToken cancellationToken) =>
            Task.FromResult(new HistorianRead(Array.Empty<DataValueSnapshot>(), null));

        public Task<HistorianRead> ReadAtTimeAsync(
            string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
            Task.FromResult(new HistorianRead(Array.Empty<DataValueSnapshot>(), null));

        public Task<HistoricalEventsResult> ReadEventsAsync(
            string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, CancellationToken cancellationToken) =>
            Task.FromResult(new HistoricalEventsResult(Array.Empty<HistoricalEvent>(), null));

        public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();

        public void Dispose() { }
    }
}