lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/GatewayAtTimeReader.cs

using ZB.MOM.WW.HistorianGateway.Contracts.Grpc;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Mapping;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;

namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway;

/// <summary>
///     Gitea #693: HistoryRead AtTime on the Gateway backend, served from RAW gateway reads and evaluated with the SQL backends'
///     shared <see cref="AtTimeEvaluator"/> (Part 11 §6.5.4) — as #681 did for the per-sample aggregates — so the Gateway
///     answers exactly what SqlServer and QuestDB answer for the same stored samples. AVEVA's own interpolated answer (the
///     gateway's <c>ReadAtTime</c>) is not used: live, it returned the previous value as Good at a stored NULL's own time,
///     null + Bad at a Good sample right after a NULL, a drifted value (99.99999999944) at an exact stored 100, and no
///     Interpolated bit.
/// </summary>
/// <remarks>
///     <para>
///         <b>What each requested time needs.</b> <c>prev</c> = the latest stored sample at or before it, <c>next</c> = the
///         first stored sample after it. <c>next</c> is only read when the evaluator uses it (a usable <c>prev</c> that is not
///         an exact hit).
///     </para>
///     <para>
///         <b>Read plan — O(1) reads for a dense batch, never two round-trips per timestamp by default.</b> The distinct
///         requested times are walked in ascending order with ONE forward window read <c>[t − 1 ms, max + 1 ms]</c> of at most
///         <see cref="ReadRows"/> rows, which yields every in-window neighbour. The sample before the first time comes from a
///         walk-back (below), the sample after the last time from one small forward probe. A window read that fills its budget
///         (dense data, widely spread times) resolves the times before its last row; the next unresolved time either
///         continues from that row (when the rows in between are expected to fit one read) or jumps there with a read that
///         starts a density-sized margin before it (so its <c>prev</c> is normally in the same read). Ten times one second
///         apart on a 1 Hz tag cost 1 read (all exact hits) or 3 (interpolated); ten times an hour apart about one read each.
///     </para>
///     <para>
///         <b>Walk-back</b> (the gateway reads forward only): segments <c>(lo, hi]</c> stepping back from <c>hi</c>, each one
///         forward read; an empty segment grows the next ×4 (from the density the window read showed, else
///         <see cref="InitialWalkWidth"/>), and after two empty segments the whole remainder down to the epoch is asked once
///         (a complete answer ends the walk). A segment that fills its budget is narrowed towards <c>hi</c> to a width sized
///         from the density it showed, never below a row it saw (rows exist there) — so dense history is bisected, not paged.
///         At most <see cref="MaxWalkReads"/> reads.
///     </para>
///     <para>
///         <b>Wire shape.</b> Every request bound is a whole millisecond (AVEVA stores milliseconds and truncates request
///         times); a row counts only strictly INSIDE its request — AVEVA's synthetic boundary row at the query start (the value
///         in effect there, <c>Quality = 133</c>, or a NULL row) can never be taken for a stored sample, nor a possible row at
///         the query end; the epoch and before hold nothing. A Bad row (a stored NULL arrives as value 0, <c>q=1 opc=0</c>,
///         #690) is value-less. Every stream is drained, never abandoned (a cancelled gateway stream evicts its pooled
///         historian session). Nothing is asked past <c>now + </c><see cref="RawAggregateReader.LookAheadHorizon"/> (the
///         historian stores nothing further ahead, and the gateway's AVEVA may refuse a far-future end); a strict gateway's
///         row ceiling (#691.3) is learned and honoured — a short read is ordinary paging here, never re-probed.
///     </para>
///     <para>
///         <b>Values.</b> A gateway value is a Double or a String whatever the node's data type, so numeric tags interpolate
///         linearly (the SQL backends step a Boolean and round an Int64; the gateway wire carries no type).
///     </para>
/// </remarks>
internal static class GatewayAtTimeReader
{
    /// <summary>The row budget of a window read and a walk-back segment.</summary>
    internal const int ReadRows = 1024;

    /// <summary>The row budget of the forward probe for the sample after the last time (boundary row + ties + one).</summary>
    internal const int ProbeRows = 2 + RawAggregateReader.PreWindowAllowance;

    /// <summary>The first walk-back segment's width when the window read showed no density.</summary>
    internal static readonly TimeSpan InitialWalkWidth = TimeSpan.FromMinutes(1);

    /// <summary>The widest walk-back segment sized from density, before ×4 growth.</summary>
    private static readonly TimeSpan MaxSeededWidth = TimeSpan.FromDays(1);

    /// <summary>The most reads one walk-back may take.</summary>
    internal const int MaxWalkReads = 64;

    private const long Ms = TimeSpan.TicksPerMillisecond;

    private static readonly DateTime Floor = RawAggregateReader.ProbeFloor;

    /// <summary>One drained read: the rows strictly inside the request (ascending) and whether the stream ended short.</summary>
    private readonly record struct Read(List<StoredSample> Rows, bool Complete);

    /// <summary>Evaluates AtTime for <paramref name="timestampsUtc"/> (same length and order, duplicates included).</summary>
    /// <param name="client">The gateway client.</param>
    /// <param name="ceiling">The data source's learned gateway row ceiling (#691.3).</param>
    /// <param name="tag">Historian tag name.</param>
    /// <param name="timestampsUtc">Requested times (any kind; treated as UTC, as on the SQL backends).</param>
    /// <param name="now">Server timestamp, the beyond-now rule and the look-ahead horizon.</param>
    /// <param name="ct">Cancellation.</param>
    /// <returns>One value per requested time, stamped with it.</returns>
    /// <exception cref="HistoryReadLimitExceededException">More than <see cref="ReadRows"/> rows within one millisecond.</exception>
    public static async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(IHistorianGatewayClient client, GatewayRowCeiling ceiling,
        string tag, IReadOnlyList<DateTime> timestampsUtc, DateTime now, CancellationToken ct)
    {
        var requested = timestampsUtc.Select(t => DateTime.SpecifyKind(t, DateTimeKind.Utc)).ToArray();
        if (requested.Length == 0) return [];
        var neighbours = await ResolveAsync(new Ctx(client, ceiling, tag, ct), requested.Distinct().Order().ToArray(), now)
            .ConfigureAwait(false);
        var result = new DataValueSnapshot[requested.Length];
        for (var i = 0; i < requested.Length; i++)
        {
            var t = requested[i];
            var (prev, next) = neighbours[t.Ticks];
            result[i] = AtTimeEvaluator.Evaluate(t, prev, next, now, now);
        }
        return result;
    }

    private sealed record Ctx(IHistorianGatewayClient Client, GatewayRowCeiling Ceiling, string Tag, CancellationToken Ct);

    /// <summary>prev / next for every time of <paramref name="times"/> (distinct, ascending).</summary>
    private static async Task<Dictionary<long, (StoredSample? Prev, StoredSample? Next)>> ResolveAsync(
        Ctx ctx, DateTime[] times, DateTime now)
    {
        var result = new Dictionary<long, (StoredSample?, StoredSample?)>(times.Length);
        var horizon = MsFloor(DateTime.MaxValue - now > RawAggregateReader.LookAheadHorizon
            ? DateTime.SpecifyKind(now + RawAggregateReader.LookAheadHorizon, DateTimeKind.Utc)
            : DateTime.MaxValue.AddTicks(-Ms));

        // Nothing is stored at or before the epoch.
        var i = 0;
        for (; i < times.Length && times[i] <= Floor; i++) result[times[i].Ticks] = (null, null);
        if (i == times.Length) return result;

        // Reads are planned on times clamped to the horizon (nothing is stored beyond it).
        var qe = MsFloor(Min(times[^1], horizon)) + TimeSpan.FromMilliseconds(1);
        var qs = RequestStart(Min(times[i], horizon));
        StoredSample? before = null;    // the latest stored sample at or before qs …
        var beforeKnown = false;        // … once known
        DateTime? walkKnown = null;     // a stored sample exists here (at or before qs): the walk-back never goes below it
        var walkWidth = InitialWalkWidth;
        var probeFrom = -1;             // first index whose next lies at or after qe (complete last read)

        while (true)
        {
            var read = await ReadTrustedAsync(ctx, qs, qe, ReadRows).ConfigureAwait(false);
            var rows = read.Rows;
            if (!read.Complete && rows.Count == 0)
                throw TooDense(ctx.Tag, qs);
            if (rows.Count >= 2 && !beforeKnown && walkKnown is null)
                walkWidth = Clamp(TimeSpan.FromTicks(4 * (rows[^1].TimeUtc - rows[0].TimeUtc).Ticks / (rows.Count - 1)));

            // A short stream saw every row up to qe; a full one every row before its last row's time x.
            var x = read.Complete ? qe : rows[^1].TimeUtc;
            var j = -1; // index of the last row at or before the current time
            var k = i;
            for (; k < times.Length; k++)
            {
                var t = times[k];
                var tr = Min(t, horizon);
                if (!read.Complete && tr >= x) break;
                while (j + 1 < rows.Count && rows[j + 1].TimeUtc <= tr) j++;

                StoredSample? prev;
                if (j >= 0) prev = rows[j];
                else
                {
                    if (!beforeKnown)
                    {
                        before = await WalkBackAsync(ctx, qs, walkKnown, walkWidth).ConfigureAwait(false);
                        beforeKnown = true;
                    }
                    prev = before;
                }

                StoredSample? next = null;
                if (NeedsNext(t, prev) && t < horizon)
                {
                    if (j + 1 < rows.Count) next = rows[j + 1];
                    else if (probeFrom < 0) probeFrom = k; // after every row of a complete read: the probe answers it
                }
                result[t.Ticks] = (prev, next);
            }

            if (k == times.Length) break;

            // Truncated before times[k]: continue from the last row, or jump to times[k] with a density-sized margin.
            var tn = Min(times[k], horizon);
            var spanTicks = Math.Max(1, (x - rows[0].TimeUtc).Ticks);
            var density = (double)rows.Count / spanTicks; // rows per tick
            var gapRows = density * (tn - x).Ticks;
            var margin = Clamp(TimeSpan.FromTicks((long)(ReadRows / 4 / Math.Max(density, 1e-18))));
            var resume = MsFloor(x) - TimeSpan.FromMilliseconds(1);
            var nextQs = gapRows < ReadRows / 2.0 ? resume : Max(RequestStart(tn - Min(margin, tn - Floor)), resume);
            if (nextQs <= qs) throw TooDense(ctx.Tag, x);

            if (nextQs < x)
            {
                // Every row before x was seen: the latest at or before nextQs is known.
                var lastIndex = rows.FindLastIndex(r => r.TimeUtc <= nextQs);
                if (lastIndex >= 0)
                {
                    before = rows[lastIndex];
                    beforeKnown = true;
                }
                walkKnown = null;
            }
            else
            {
                beforeKnown = false;
                before = null;
                walkKnown = x;
                walkWidth = margin;
            }
            qs = nextQs;
            i = k;
        }

        if (probeFrom >= 0 && qe - TimeSpan.FromMilliseconds(1) < horizon)
        {
            var after = await FirstAfterAsync(ctx, qe - TimeSpan.FromMilliseconds(1), horizon, Min(times[^1], horizon)).ConfigureAwait(false);
            if (after is not null)
                for (var k = probeFrom; k < times.Length; k++)
                {
                    var t = times[k];
                    var (prev, next) = result[t.Ticks];
                    if (next is null && t < horizon && NeedsNext(t, prev) && after.Value.TimeUtc > t)
                        result[t.Ticks] = (prev, after);
                }
        }
        return result;
    }

    /// <summary>The evaluator reads <c>next</c> only for a usable <c>prev</c> that is not an exact hit.</summary>
    private static bool NeedsNext(DateTime t, StoredSample? prev) => prev is { IsUsable: true } p && p.TimeUtc != t;

    /// <summary>
    ///     The latest stored sample with time in <c>(Floor, hi]</c>, walking back in forward-read segments (see the type
    ///     remarks). <paramref name="known"/>: a stored sample exists at that time (≤ <paramref name="hi"/>).
    /// </summary>
    private static async Task<StoredSample?> WalkBackAsync(Ctx ctx, DateTime hi, DateTime? known, TimeSpan width)
    {
        var empties = 0;
        var triedRest = false;
        for (var reads = 0; reads < MaxWalkReads; reads++)
        {
            if (hi <= Floor) return null;
            var rest = known is null && !triedRest && empties >= 2;
            DateTime lo;
            if (rest)
            {
                lo = Floor;
                triedRest = true;
            }
            else
            {
                lo = (hi - Floor).Ticks > width.Ticks ? hi - width : Floor;
                if (known is { } kn && lo < kn) lo = kn.AddTicks(-1); // never below a row known to exist (it is the floor)
            }

            var read = await ReadTrustedAsync(ctx, MsFloor(lo), MsFloor(hi) + TimeSpan.FromMilliseconds(1), ReadRows)
                .ConfigureAwait(false);
            StoredSample? first = null, last = null;
            DateTime? lastAny = null;
            foreach (var r in read.Rows)
            {
                if (r.TimeUtc > lo && r.TimeUtc <= hi)
                {
                    first ??= r;
                    last = r;
                }
                lastAny = r.TimeUtc;
            }

            if (read.Complete)
            {
                if (last is not null) return last;
                if (lo <= Floor) return null;
                hi = lo;
                empties++;
                width = Grow(width);
                continue;
            }

            // Truncated. A whole-remainder read of deep history proves nothing: carry on with segments.
            if (rest) continue;
            if (last is not { } seen || (known is { } k0 && seen.TimeUtc <= k0))
                throw TooDense(ctx.Tag, lastAny ?? lo);
            var spanTicks = Math.Max(1, (seen.TimeUtc - first!.Value.TimeUtc).Ticks); // the density of the rows seen
            known = seen.TimeUtc;
            // Aim at about a quarter of a read from the density just seen, and never wider than half the open stretch.
            var aimed = TimeSpan.FromTicks(Math.Max(Ms, spanTicks / 4));
            var half = TimeSpan.FromTicks(Math.Max(Ms, (hi - seen.TimeUtc).Ticks / 2));
            width = aimed < half ? aimed : half;
            empties = 0;
        }
        throw new HistoryReadLimitExceededException(MaxWalkReads,
            $"The AtTime read of '{ctx.Tag}' could not find the stored sample before a requested time within {MaxWalkReads} raw reads.");
    }

    /// <summary>The first stored sample after <paramref name="after"/> in <c>(qs, end)</c>, from a small forward read.</summary>
    private static async Task<StoredSample?> FirstAfterAsync(Ctx ctx, DateTime qs, DateTime end, DateTime after)
    {
        foreach (var budget in new[] { ProbeRows, ReadRows })
        {
            var read = await ReadTrustedAsync(ctx, qs, end, budget).ConfigureAwait(false);
            foreach (var r in read.Rows)
                if (r.TimeUtc > after) return r;
            if (read.Complete) return null;
        }
        throw TooDense(ctx.Tag, qs);
    }

    /// <summary>
    ///     One forward read <c>[qs, qe]</c> (whole milliseconds), drained, keeping only rows strictly inside it and after the
    ///     epoch. A strict gateway's ceiling is learned on refusal and then asked for directly; a full read is not re-probed
    ///     (the caller pages on, it does not need the rest of this read).
    /// </summary>
    private static async Task<Read> ReadTrustedAsync(Ctx ctx, DateTime qs, DateTime qe, int budget)
    {
        var (read, _) = await ctx.Ceiling.ReadAsync(budget, b => PassAsync(ctx, qs, qe, b), (_, _) => false)
            .ConfigureAwait(false);
        return read;
    }

    private static async Task<Read> PassAsync(Ctx ctx, DateTime qs, DateTime qe, int budget)
    {
        var rows = new List<StoredSample>();
        var received = 0;
        await foreach (var raw in ctx.Client.ReadRawAsync(ctx.Tag, qs, qe, budget, ctx.Ct).ConfigureAwait(false))
        {
            received++; // drained to completion, never broken out of
            if (raw.Timestamp is null) continue;
            var t = raw.Timestamp.ToDateTime();
            if (t > qs && t < qe && t > Floor) rows.Add(ToStored(raw, t));
        }
        return new Read(rows, received < budget);
    }

    /// <summary>A gateway row as a stored sample; a Bad row carries no value (#690: a stored NULL arrives as value 0).</summary>
    private static StoredSample ToStored(HistorianSample raw, DateTime t)
    {
        var status = SampleMapper.ToStatus(raw);
        var value = StatusBits.IsBad(status) ? HistorianValue.Empty
            : raw.HasNumericValue ? HistorianValue.FromDouble(raw.NumericValue)
            : raw.HasStringValue ? HistorianValue.FromString(raw.StringValue)
            : HistorianValue.Empty;
        return StoredSample.FromStored(t, value, status);
    }

    /// <summary>The window read's start for a time: a millisecond before its millisecond (never before the epoch).</summary>
    private static DateTime RequestStart(DateTime t)
    {
        var s = MsFloor(t) - TimeSpan.FromMilliseconds(1);
        return s < Floor ? Floor : s;
    }

    private static DateTime MsFloor(DateTime t) => new(t.Ticks - t.Ticks % Ms, DateTimeKind.Utc);

    private static DateTime Min(DateTime a, DateTime b) => a <= b ? a : b;

    private static DateTime Max(DateTime a, DateTime b) => a >= b ? a : b;

    private static TimeSpan Min(TimeSpan a, TimeSpan b) => a <= b ? a : b;

    private static TimeSpan Clamp(TimeSpan w) =>
        w < TimeSpan.FromMilliseconds(1) ? TimeSpan.FromMilliseconds(1) : w > MaxSeededWidth ? MaxSeededWidth : w;

    private static TimeSpan Grow(TimeSpan w) => w.Ticks > TimeSpan.MaxValue.Ticks / 4 ? TimeSpan.MaxValue : w * 4;

    private static HistoryReadLimitExceededException TooDense(string tag, DateTime at) =>
        new(ReadRows,
            $"The AtTime read of '{tag}' met more than {ReadRows} stored samples within about one millisecond of " +
            $"{at:O}; the Gateway backend serves AtTime from raw reads.");
}