lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/HistoryPaging.cs

using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;

/// <summary>
///     The server-side resume state stored behind an opaque continuation point for a single paged
///     HistoryRead-Raw read. Captures exactly enough to continue the SAME logical read from where the
///     previous page stopped: the tagname, the original (inclusive) end of the window, the next start of
///     the window, and the tie-safe boundary skip. (Only Raw pages server-side — Processed and AtTime
///     carry no client count cap, so they are single-shot and never produce a continuation point; see
///     <see cref="HistoryPaging"/>.)
///     <para>
///         The boundary fields (<see cref="NextStartUtc"/> + <see cref="BoundarySkipCount"/>) encode a
///         tie-safe resume cursor: the next page reads from <see cref="NextStartUtc"/> INCLUSIVE and
///         then drops the first <see cref="BoundarySkipCount"/> samples whose SourceTimestamp equals
///         <see cref="NextStartUtc"/>. This guarantees that samples sharing the page-boundary timestamp
///         are neither re-returned (duplicate) nor skipped — see <see cref="HistoryPaging"/>.
///     </para>
///     <para>
///         This record carries no SDK types so the whole paging decision surface is a pure, allocation-
///         cheap value that unit tests can drive directly.
///     </para>
/// </summary>
/// <param name="Tagname">The resolved historian tagname (NOT the NodeId) to read from.</param>
/// <param name="NextStartUtc">
///     Inclusive lower bound for the next page — the boundary timestamp the previous page stopped on.
/// </param>
/// <param name="EndUtc">The original (inclusive) upper bound of the read window; unchanged across pages.</param>
/// <param name="BoundarySkipCount">
///     How many samples whose SourceTimestamp equals <see cref="NextStartUtc"/> were already returned on
///     prior pages and must be dropped from the head of the next page (tie de-dup).
/// </param>
/// <param name="NumValuesPerNode">The client's per-page cap; re-applied to every resumed page.</param>
/// <param name="Kind">
///     Which read issued the cursor (Gitea #669 / #668). A cursor is only ever resumed by the read kind that issued it, so a
///     raw point presented to an events or modified read (or any other mismatch) is a <c>BadContinuationPointInvalid</c>
///     miss. <see cref="HistoryCursorKind.Events"/>: <paramref name="Tagname"/> carries the registered historian
///     event-source name and the boundary fields are over <c>EventTimeUtc</c>.
/// </param>
/// <param name="LastModifiedId">
///     <see cref="HistoryCursorKind.Modified"/> only: the <c>ModifiedId</c> of the last returned modification. With
///     <paramref name="NextStartUtc"/> (its source time) it is the exclusive resume key; <paramref name="BoundarySkipCount"/>
///     is unused because the key is unique.
/// </param>
/// <param name="WindowStartUtc">
///     <see cref="HistoryCursorKind.Modified"/> only: the normalised window start, so a resumed page keeps the original
///     window and direction (the modified read resumes by key, not by moving the window start).
/// </param>
/// <param name="Reverse">
///     <see cref="HistoryCursorKind.Events"/> only (Gitea #675): the read is newest-first over
///     <c>(EndUtc, NextStartUtc]</c>. A resumed events read takes its direction from here, never from the resume
///     request's details. Raw keeps deriving it from the window (<c>NextStartUtc &gt; EndUtc</c>).
/// </param>
/// <param name="SegmentWidthTicks">
///     <see cref="HistoryCursorKind.Raw"/> reverse reads on a forward-only backend only (Gitea #686): the segment width the
///     walk-back ended on, so the next page starts at the observed density instead of re-discovering it. 0 = none.
/// </param>
/// <param name="EmptyFromTicks">
///     Walked reverse reads only (Gitea #691): with <paramref name="EmptyToTicks"/>, a stretch <c>(from, to)</c> the previous
///     page's walk proved holds no row, so the next page steps over it rather than re-walking it (to the 1970 floor, ~14
///     reads, at the end of a deep page-through). 0/0 = none. The cursor lives in the session's in-memory store (never
///     serialized), so a new field with a "none" default is all the versioning it needs.
/// </param>
/// <param name="EmptyToTicks">The exclusive upper end of that stretch.</param>
/// <param name="PendingEndBound">
///     <see cref="HistoryCursorKind.Raw"/> with <c>ReturnBounds = true</c> only (Gitea #691): the end bound still owed —
///     Part 11 returns it with the last page. <c>null</c> = bounds not requested.
/// </param>
/// <param name="WithinTieCluster">
///     <see cref="HistoryCursorKind.Raw"/> only (Gitea #692): the previous page was a slice of an oversized tie cluster at
///     <paramref name="NextStartUtc"/> that still has un-emitted ties — the next page over-fetches the cluster directly
///     instead of first re-reading <c>N + skip + 1</c> rows that are known to be ties.
/// </param>
internal sealed record HistoryContinuationState(
    string Tagname,
    DateTime NextStartUtc,
    DateTime EndUtc,
    int BoundarySkipCount,
    uint NumValuesPerNode,
    HistoryCursorKind Kind = HistoryCursorKind.Raw,
    long LastModifiedId = 0,
    DateTime WindowStartUtc = default,
    bool Reverse = false,
    long SegmentWidthTicks = 0,
    long EmptyFromTicks = 0,
    long EmptyToTicks = 0,
    HistoryEndBound? PendingEndBound = null,
    bool WithinTieCluster = false)
{
    /// <summary>The walk-back hint this cursor carries (#686/#689/#691).</summary>
    public ReverseWalkHint WalkHint => new(SegmentWidthTicks, EmptyFromTicks, EmptyToTicks);
}

/// <summary>The HistoryRead kind that issued a <see cref="HistoryContinuationState"/>.</summary>
internal enum HistoryCursorKind
{
    /// <summary>HistoryReadRaw (values).</summary>
    Raw,

    /// <summary>HistoryReadEvents (#669).</summary>
    Events,

    /// <summary>HistoryReadRaw with <c>IsReadModified = true</c> — the Part 11 modification trail (#668).</summary>
    Modified,
}

/// <summary>
///     Pure server-side continuation-point paging decisions for the count-capped variable-history arms
///     (Raw / Processed). The backend (the HistorianGateway read client) does NOT page — it returns up to
///     <c>NumValuesPerNode</c> samples with a null continuation point — so paging is synthesised here,
///     time-based:
///     <list type="bullet">
///       <item>A page that returns EXACTLY the requested cap (<c>NumValuesPerNode &gt; 0</c>) MAY have
///         more behind it ⇒ emit a continuation point.</item>
///       <item>A short page (fewer than the cap) is the last page ⇒ no continuation point.</item>
///       <item><c>NumValuesPerNode == 0</c> means "all values, no limit" (OPC UA Part 11) ⇒ never page;
///         return everything in one shot.</item>
///     </list>
///     All methods are static + pure so they unit-test without a server, a session, or the SDK.
/// </summary>
internal static class HistoryPaging
{
    /// <summary>
    ///     Decide whether a just-returned page is "full" and therefore MAY be followed by more data —
    ///     the signal to emit a continuation point. A page is full when the client asked for a finite
    ///     cap (<paramref name="numValuesPerNode"/> &gt; 0) and the backend returned exactly that many
    ///     samples. A short page (or an unlimited <c>0</c> request) is terminal.
    /// </summary>
    /// <param name="returnedCount">The number of samples the backend returned for this page.</param>
    /// <param name="numValuesPerNode">The client's per-page cap; <c>0</c> means "no limit".</param>
    /// <returns><c>true</c> when a continuation point should be emitted; otherwise <c>false</c>.</returns>
    public static bool IsFullPage(int returnedCount, uint numValuesPerNode) =>
        numValuesPerNode > 0 && returnedCount >= numValuesPerNode;

    /// <summary>
    ///     Build the resume cursor (next-start + boundary skip count) from the last sample of a full
    ///     page, tie-safe against samples that share the boundary SourceTimestamp.
    ///     <para>
    ///         The next page resumes from the LAST returned sample's SourceTimestamp <em>inclusive</em>
    ///         (NOT advanced by a tick), and the returned <paramref name="boundarySkipCount"/> counts how
    ///         many samples in the page already carry that exact boundary timestamp. Resuming inclusively
    ///         + dropping that many head samples guarantees:
    ///         <list type="bullet">
    ///           <item>no sample is re-returned (the ones already emitted at the boundary are skipped), and</item>
    ///           <item>no sample is skipped (any un-emitted ties at the boundary are still read because we
    ///             start AT the boundary, not after it).</item>
    ///         </list>
    ///         A naive "+1 tick" advance would skip un-emitted ties; this carry-offset strategy does not.
    ///     </para>
    /// </summary>
    /// <param name="page">The page just returned (chronological, non-empty — guaranteed by the caller,
    ///     which only pages a full page and a full page implies <c>NumValuesPerNode &gt; 0</c> samples).</param>
    /// <param name="nextStartUtc">The boundary timestamp the next page resumes from (inclusive).</param>
    /// <param name="boundarySkipCount">How many head samples at <paramref name="nextStartUtc"/> the next
    ///     page must drop (samples already emitted at the boundary timestamp).</param>
    public static void ComputeResumeCursor(
        IReadOnlyList<DataValueSnapshot> page,
        out DateTime nextStartUtc,
        out int boundarySkipCount)
    {
        // Enforce the documented precondition at the API boundary rather than relying on the caller's guard
        // (the only caller only pages a full, non-empty page, but this is a public static helper).
        if (page.Count == 0)
            throw new ArgumentOutOfRangeException(nameof(page), "ComputeResumeCursor requires a non-empty page.");

        // The boundary is the last returned sample's SourceTimestamp. A sample whose SourceTimestamp is
        // null (Bad/unset) cannot anchor a time cursor; fall back to MinValue so the next read covers the
        // whole remaining window rather than silently dropping data — duplicates are then de-duped by the
        // skip count below.
        var last = page[^1];
        nextStartUtc = last.SourceTimestampUtc ?? DateTime.MinValue;

        // Count how many trailing samples in THIS page share the boundary timestamp — those are the ties
        // already emitted at the boundary that the next page must drop from its head.
        var skip = 0;
        for (var i = page.Count - 1; i >= 0; i--)
        {
            if ((page[i].SourceTimestampUtc ?? DateTime.MinValue) == nextStartUtc) skip++;
            else break;
        }

        boundarySkipCount = skip;
    }

    /// <summary>
    ///     Drop the first <paramref name="boundarySkipCount"/> samples of a freshly-read resumed page
    ///     whose SourceTimestamp equals the boundary <paramref name="boundaryUtc"/> — the ties already
    ///     emitted on the previous page. Samples past the boundary timestamp are always kept (only the
    ///     exact-boundary head is trimmed), so a backend that returns fewer boundary ties than expected
    ///     (data pruned between pages) still yields a correct, monotonic result.
    /// </summary>
    /// <param name="resumedPage">The page returned by the resumed backend read (chronological).</param>
    /// <param name="boundaryUtc">The boundary timestamp the resume read started at (inclusive).</param>
    /// <param name="boundarySkipCount">How many head samples at <paramref name="boundaryUtc"/> to drop.</param>
    /// <returns>The page with the already-emitted boundary ties trimmed from the head.</returns>
    public static IReadOnlyList<DataValueSnapshot> TrimBoundaryDuplicates(
        IReadOnlyList<DataValueSnapshot> resumedPage,
        DateTime boundaryUtc,
        int boundarySkipCount)
    {
        if (boundarySkipCount <= 0 || resumedPage.Count == 0) return resumedPage;

        var dropped = 0;
        var i = 0;
        while (i < resumedPage.Count
               && dropped < boundarySkipCount
               && (resumedPage[i].SourceTimestampUtc ?? DateTime.MinValue) == boundaryUtc)
        {
            i++;
            dropped++;
        }

        if (i == 0) return resumedPage;
        // Slice off the trimmed head; return a copy so the caller owns a plain list (no SDK coupling).
        var trimmed = new List<DataValueSnapshot>(resumedPage.Count - i);
        for (var j = i; j < resumedPage.Count; j++) trimmed.Add(resumedPage[j]);
        return trimmed;
    }

    /// <summary>
    ///     Page WITHIN a single oversized "tie cluster" — a set of raw samples that all share one
    ///     SourceTimestamp <paramref name="boundaryT"/> and is larger than the client's per-page cap.
    ///     <para>
    ///         The fixed-<c>(start, end, cap)</c> historian backend cannot skip/offset, so an oversized
    ///         tie cluster defeats the (timestamp, skip) resume cursor that <see cref="ComputeResumeCursor"/>
    ///         builds: every resume re-reads the first <c>cap</c> ties, the boundary-tie trim empties the
    ///         page, and the cursor never advances. To page past it, the caller over-fetches the WHOLE
    ///         cluster (a bounded <c>start == end == T</c> read) and hands it here. We then carve out the
    ///         next <paramref name="cap"/> ties starting at <paramref name="skip"/> and compute a cursor
    ///         that advances within the cluster, then steps off it when it is drained.
    ///     </para>
    ///     <para>
    ///         <b>Advance is lossless.</b> When the slice drains the cluster (<c>emitted == clusterCount</c>)
    ///         we resume from <c>T + 1 tick</c> with a fresh skip of <c>0</c>. This skips no data because no
    ///         <see cref="DateTime"/> value exists strictly between <c>T</c> and <c>T + 1 tick</c> — a tick
    ///         (100 ns) is the type's resolution — so every remaining sample in the window has a timestamp
    ///         of at least <c>T + 1 tick</c>. We do NOT resume inclusively at <c>T</c> here (the way
    ///         <see cref="ComputeResumeCursor"/> does for cross-boundary ties): we have already over-fetched
    ///         and emitted the entire <c>T</c> cluster, so resuming at <c>T</c> would needlessly re-read it.
    ///     </para>
    ///     <para>
    ///         <b>Short-page-with-CP exception.</b> A page that fully drains the cluster but is SHORTER than
    ///         <paramref name="cap"/> (the cluster had fewer than <c>cap</c> remaining ties) STILL emits a
    ///         continuation point when the window extends past <c>T</c> (<c>T + 1 tick &lt;= endUtc</c>). This
    ///         deliberately violates the usual "short page ⇒ terminal" rule (<see cref="IsFullPage"/>):
    ///         the page is short only because the cluster ran out, not because the window did, so there may
    ///         still be data after <c>T</c> that the next page must read.
    ///     </para>
    ///     <para>
    ///         <b>Self-heal.</b> If <paramref name="skip"/> meets or exceeds <paramref name="clusterCount"/>
    ///         (e.g. a stale cursor against a cluster that shrank between reads), <c>sliceStart</c> clamps to
    ///         <paramref name="clusterCount"/> and <c>sliceCount</c> is <c>0</c>; since <c>emitted</c> then
    ///         equals <paramref name="clusterCount"/>, the cursor advances/terminates rather than looping.
    ///     </para>
    /// </summary>
    /// <param name="clusterCount">The total number of ties at <paramref name="boundaryT"/> (the over-fetched cluster size).</param>
    /// <param name="skip">How many ties at <paramref name="boundaryT"/> were already emitted on prior pages.</param>
    /// <param name="cap">The client's per-page cap (<c>NumValuesPerNode</c>); must be &gt; 0 to make progress.</param>
    /// <param name="boundaryT">The single SourceTimestamp every tie in the cluster shares.</param>
    /// <param name="endUtc">The (inclusive) upper bound of the read window; unchanged across pages.
    ///     The forward window is end-EXCLUSIVE (Part 11; the <c>IHistorianDataSource.ReadRawAsync</c> contract), so a drained
    ///     cluster continues only while <c>next</c> is strictly before <paramref name="endUtc"/> truncated to
    ///     <paramref name="resolutionTicks"/>. (OpcUaServer-003 had kept <c>next &lt;= endUtc</c> as "the conservative side",
    ///     but it was not: resuming AT EndTime makes the next read a <c>start == end</c> point read, which returns the samples
    ///     at EndTime as data — #692 review F1; on a microsecond backend EndTime is truncated first.)</param>
    /// <param name="sliceStart">The index into the cluster the emitted slice starts at (clamped to the count).</param>
    /// <param name="sliceCount">How many ties the emitted slice contains (may be <c>0</c> on a stale-skip self-heal).</param>
    /// <param name="nextStartUtc">The next page's inclusive start: <paramref name="boundaryT"/> while the
    ///     cluster still has un-emitted ties, <c>boundaryT + 1 tick</c> once it is drained and the window
    ///     remains, or <c>null</c> when the window is exhausted (terminal — no continuation point).</param>
    /// <param name="nextSkip">The next page's boundary skip count: the running emitted-tie total while the
    ///     cluster drains, else <c>0</c> after advancing past it.</param>
    /// <param name="reverse">
    ///     <c>true</c> for a reverse read (Part 11: <c>start &gt; end</c>, samples in descending time, window
    ///     <c>(end, start]</c>). A drained cluster then steps DOWN to <c>T − 1 tick</c>, which is lossless for the
    ///     same reason <c>T + 1 tick</c> is going forward, and the read continues only while that is still strictly
    ///     after <paramref name="endUtc"/> (the reverse window excludes <c>end</c>; a resume at exactly <c>end</c>
    ///     would turn into a <c>start == end</c> point read and return the excluded sample). Forward is unchanged.
    /// </param>
    /// <param name="resolutionTicks">
    ///     The backend's storage timestamp resolution (<c>IHistorianDataSource.TimestampResolutionTicks</c>, &gt;= 1). A
    ///     drained cluster steps off <paramref name="boundaryT"/> by this many ticks rather than by one: a backend that
    ///     truncates its inputs to its resolution (QuestDB, 10 ticks) would otherwise truncate <c>T + 1 tick</c> back
    ///     to <c>T</c> and return the drained sample again, forever. Lossless for the same reason one tick is at tick
    ///     precision: no stored timestamp lies strictly between <c>T</c> and <c>T ± resolution</c>. The default
    ///     <c>1</c> keeps the tick-precision behaviour exactly.
    /// </param>
    public static void SliceTieCluster(
        int clusterCount, int skip, uint cap, DateTime boundaryT, DateTime endUtc,
        out int sliceStart, out int sliceCount, out DateTime? nextStartUtc, out int nextSkip,
        bool reverse = false, long resolutionTicks = 1)
    {
        ArgumentOutOfRangeException.ThrowIfLessThan(resolutionTicks, 1L);
        sliceStart = Math.Min(skip, clusterCount);
        sliceCount = Math.Min((int)cap, clusterCount - sliceStart);
        var emitted = sliceStart + sliceCount;
        if (emitted < clusterCount) { nextStartUtc = boundaryT; nextSkip = emitted; }
        else if (reverse)
        {
            // Nothing can lie below DateTime.MinValue, so a boundary within one resolution step of it cannot step down.
            var more = boundaryT.Ticks >= resolutionTicks && boundaryT.AddTicks(-resolutionTicks) > endUtc;
            nextStartUtc = more ? boundaryT.AddTicks(-resolutionTicks) : null;
            nextSkip = 0;
        }
        else
        {
            var next = boundaryT.AddTicks(resolutionTicks);
            // The forward window is end-EXCLUSIVE, at the backend's resolution: resuming AT EndTime would turn the next read
            // into a start == end point read, which returns the samples at EndTime as data (#692 review F1).
            if (next.Ticks < endUtc.Ticks - endUtc.Ticks % resolutionTicks) { nextStartUtc = next; nextSkip = 0; }
            else { nextStartUtc = null; nextSkip = 0; }
        }
    }
}