using ZB.MOM.WW.OtOpcUa.Core.Abstractions; namespace ZB.MOM.WW.OtOpcUa.OpcUaServer; /// /// 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 /// .) /// /// The boundary fields ( + ) encode a /// tie-safe resume cursor: the next page reads from INCLUSIVE and /// then drops the first samples whose SourceTimestamp equals /// . This guarantees that samples sharing the page-boundary timestamp /// are neither re-returned (duplicate) nor skipped — see . /// /// /// This record carries no SDK types so the whole paging decision surface is a pure, allocation- /// cheap value that unit tests can drive directly. /// /// /// The resolved historian tagname (NOT the NodeId) to read from. /// /// Inclusive lower bound for the next page — the boundary timestamp the previous page stopped on. /// /// The original (inclusive) upper bound of the read window; unchanged across pages. /// /// How many samples whose SourceTimestamp equals were already returned on /// prior pages and must be dropped from the head of the next page (tie de-dup). /// /// The client's per-page cap; re-applied to every resumed page. internal sealed record HistoryContinuationState( string Tagname, DateTime NextStartUtc, DateTime EndUtc, int BoundarySkipCount, uint NumValuesPerNode); /// /// 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 /// NumValuesPerNode samples with a null continuation point — so paging is synthesised here, /// time-based: /// /// A page that returns EXACTLY the requested cap (NumValuesPerNode > 0) MAY have /// more behind it ⇒ emit a continuation point. /// A short page (fewer than the cap) is the last page ⇒ no continuation point. /// NumValuesPerNode == 0 means "all values, no limit" (OPC UA Part 11) ⇒ never page; /// return everything in one shot. /// /// All methods are static + pure so they unit-test without a server, a session, or the SDK. /// internal static class HistoryPaging { /// /// 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 ( > 0) and the backend returned exactly that many /// samples. A short page (or an unlimited 0 request) is terminal. /// /// The number of samples the backend returned for this page. /// The client's per-page cap; 0 means "no limit". /// true when a continuation point should be emitted; otherwise false. public static bool IsFullPage(int returnedCount, uint numValuesPerNode) => numValuesPerNode > 0 && returnedCount >= numValuesPerNode; /// /// 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. /// /// The next page resumes from the LAST returned sample's SourceTimestamp inclusive /// (NOT advanced by a tick), and the returned counts how /// many samples in the page already carry that exact boundary timestamp. Resuming inclusively /// + dropping that many head samples guarantees: /// /// no sample is re-returned (the ones already emitted at the boundary are skipped), and /// no sample is skipped (any un-emitted ties at the boundary are still read because we /// start AT the boundary, not after it). /// /// A naive "+1 tick" advance would skip un-emitted ties; this carry-offset strategy does not. /// /// /// The page just returned (chronological, non-empty — guaranteed by the caller, /// which only pages a full page and a full page implies NumValuesPerNode > 0 samples). /// The boundary timestamp the next page resumes from (inclusive). /// How many head samples at the next /// page must drop (samples already emitted at the boundary timestamp). public static void ComputeResumeCursor( IReadOnlyList 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; } /// /// Drop the first samples of a freshly-read resumed page /// whose SourceTimestamp equals the boundary — 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. /// /// The page returned by the resumed backend read (chronological). /// The boundary timestamp the resume read started at (inclusive). /// How many head samples at to drop. /// The page with the already-emitted boundary ties trimmed from the head. public static IReadOnlyList TrimBoundaryDuplicates( IReadOnlyList 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(resumedPage.Count - i); for (var j = i; j < resumedPage.Count; j++) trimmed.Add(resumedPage[j]); return trimmed; } /// /// Page WITHIN a single oversized "tie cluster" — a set of raw samples that all share one /// SourceTimestamp and is larger than the client's per-page cap. /// /// The fixed-(start, end, cap) historian backend cannot skip/offset, so an oversized /// tie cluster defeats the (timestamp, skip) resume cursor that /// builds: every resume re-reads the first cap 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 start == end == T read) and hands it here. We then carve out the /// next ties starting at and compute a cursor /// that advances within the cluster, then steps off it when it is drained. /// /// /// Advance is lossless. When the slice drains the cluster (emitted == clusterCount) /// we resume from T + 1 tick with a fresh skip of 0. This skips no data because no /// value exists strictly between T and T + 1 tick — a tick /// (100 ns) is the type's resolution — so every remaining sample in the window has a timestamp /// of at least T + 1 tick. We do NOT resume inclusively at T here (the way /// does for cross-boundary ties): we have already over-fetched /// and emitted the entire T cluster, so resuming at T would needlessly re-read it. /// /// /// Short-page-with-CP exception. A page that fully drains the cluster but is SHORTER than /// (the cluster had fewer than cap remaining ties) STILL emits a /// continuation point when the window extends past T (T + 1 tick <= endUtc). This /// deliberately violates the usual "short page ⇒ terminal" rule (): /// the page is short only because the cluster ran out, not because the window did, so there may /// still be data after T that the next page must read. /// /// /// Self-heal. If meets or exceeds /// (e.g. a stale cursor against a cluster that shrank between reads), sliceStart clamps to /// and sliceCount is 0; since emitted then /// equals , the cursor advances/terminates rather than looping. /// /// /// The total number of ties at (the over-fetched cluster size). /// How many ties at were already emitted on prior pages. /// The client's per-page cap (NumValuesPerNode); must be > 0 to make progress. /// The single SourceTimestamp every tie in the cluster shares. /// The (inclusive) upper bound of the read window; unchanged across pages. /// The index into the cluster the emitted slice starts at (clamped to the count). /// How many ties the emitted slice contains (may be 0 on a stale-skip self-heal). /// The next page's inclusive start: while the /// cluster still has un-emitted ties, boundaryT + 1 tick once it is drained and the window /// remains, or null when the window is exhausted (terminal — no continuation point). /// The next page's boundary skip count: the running emitted-tie total while the /// cluster drains, else 0 after advancing past it. 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) { sliceStart = Math.Min(skip, clusterCount); sliceCount = Math.Min((int)cap, clusterCount - sliceStart); var emitted = sliceStart + sliceCount; if (emitted < clusterCount) { nextStartUtc = boundaryT; nextSkip = emitted; } else { var next = boundaryT.AddTicks(1); if (next <= endUtc) { nextStartUtc = next; nextSkip = 0; } else { nextStartUtc = null; nextSkip = 0; } } } }