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

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

namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;

/// <summary>One walked-back reverse events page (<see cref="ReverseEventsReader"/>).</summary>
/// <param name="Events">At most the requested count, newest first by <c>(EventTimeUtc, EventId)</c>.</param>
/// <param name="Hint">What the next page of the same read resumes with.</param>
internal readonly record struct ReverseEventsPage(List<HistoricalEvent> Events, ReverseWalkHint Hint);

/// <summary>
///     Serves a Part 11 reverse events read — <c>(lo, hi]</c> newest-first — on a backend that reads events
///     forward only (<see cref="IHistorianDataSource.SupportsReverseEvents"/> = <c>false</c>: the Gateway) with the
///     <see cref="ReverseWalk"/>. A segment read is cut short when the source flags truncation (its continuation point); ties
///     are ordered by <see cref="EventIdTieOrder"/> because the gateway orders by event time only.
/// </summary>
internal static class ReverseEventsReader
{
    /// <summary>Reads <c>(lo, hi]</c> newest-first from forward event reads.</summary>
    /// <param name="source">A backend that reads events forward.</param>
    /// <param name="sourceName">Event-source filter (null = all).</param>
    /// <param name="hi">The later (inclusive) bound.</param>
    /// <param name="lo">The earlier (exclusive) bound.</param>
    /// <param name="want">How many events; <c>0</c> = all in the window (one forward read).</param>
    /// <param name="maxTieCluster">The tie budget (<c>MaxTieClusterOverfetch</c>) — a ceiling only.</param>
    /// <param name="nowUtc">Now, for the look-ahead clamp.</param>
    /// <param name="hint">What the previous page carried (default = none).</param>
    /// <param name="ct">Cancellation.</param>
    /// <returns>The events, newest first, and the hint for the next page.</returns>
    /// <exception cref="HistoryReadLimitExceededException">One timestamp holds more than the tie budget.</exception>
    public static Task<ReverseEventsPage> ReadAsync(
        IHistorianDataSource source, string? sourceName, DateTime hi, DateTime lo, uint want, int maxTieCluster,
        DateTime nowUtc, ReverseWalkHint hint, CancellationToken ct) =>
        ReadAsync(
            (start, end, cap, token) => source.ReadEventsAsync(sourceName, start, end, cap, reverse: false, token),
            source.TimestampResolutionTicks, sourceName, hi, lo, want, maxTieCluster, nowUtc, hint, ct);

    /// <summary>
    ///     Reads <c>(lo, hi]</c> newest-first from a forward event reader — the form a folder with several event sources
    ///     uses (<see cref="EventSourceFanOut"/>, Gitea #886), whose merged reads stand in for one source's.
    /// </summary>
    /// <param name="readForward">Reads <c>[start, end]</c> forward with a cap (<c>int.MaxValue</c> = none).</param>
    /// <param name="resolutionTicks">The backend's timestamp resolution.</param>
    /// <param name="sourceName">The source (or source set) named in a limit message.</param>
    /// <param name="hi">The later (inclusive) bound.</param>
    /// <param name="lo">The earlier (exclusive) bound.</param>
    /// <param name="want">How many events; <c>0</c> = all in the window (one forward read).</param>
    /// <param name="maxTieCluster">The tie budget (<c>MaxTieClusterOverfetch</c>) — a ceiling only.</param>
    /// <param name="nowUtc">Now, for the look-ahead clamp.</param>
    /// <param name="hint">What the previous page carried (default = none).</param>
    /// <param name="ct">Cancellation.</param>
    /// <returns>The events, newest first, and the hint for the next page.</returns>
    /// <exception cref="HistoryReadLimitExceededException">One timestamp holds more than the tie budget.</exception>
    public static async Task<ReverseEventsPage> ReadAsync(
        Func<DateTime, DateTime, int, CancellationToken, Task<HistoricalEventsResult>> readForward, long resolutionTicks,
        string? sourceName, DateTime hi, DateTime lo, uint want, int maxTieCluster, DateTime nowUtc, ReverseWalkHint hint,
        CancellationToken ct)
    {
        var budget = Math.Max(1, maxTieCluster);
        var (rows, next) = await ReverseWalk.WalkAsync<HistoricalEvent>(
                async (start, end, cap, token) =>
                {
                    // cap 0 = the whole window: int.MaxValue is the events contract's "no cap" (a non-positive value would
                    // mean the backend's DEFAULT cap).
                    var r = await readForward(start, end, cap == 0 ? int.MaxValue : cap, token).ConfigureAwait(false);
                    return new ForwardSegmentRead<HistoricalEvent>(r.Events, r.ContinuationPoint is not null);
                },
                ev => ev.EventTimeUtc.Ticks, EventIdTieOrder.Instance, hi, lo, want, budget, resolutionTicks,
                nowUtc, hint,
                at => new HistoryReadLimitExceededException(budget,
                    $"A reverse events read of '{sourceName}' at {at:O} found more than {budget} events within one millisecond; the " +
                    "backend reads forward only, so the latest of them cannot be selected. Increase " +
                    "ServerHistorian:MaxTieClusterOverfetch."),
                ct)
            .ConfigureAwait(false);
        return new ReverseEventsPage(rows, next);
    }
}