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

using System.Collections.Frozen;
using Grpc.Core;
using Microsoft.Extensions.Logging;
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>
///     Server-side <see cref="IHistorianDataSource"/> backed by the HistorianGateway gRPC surface
///     (via the <see cref="IHistorianGatewayClient"/> seam). Translates OPC UA HistoryRead requests
///     to gateway read calls and maps the wire shapes back to the driver-agnostic
///     <see cref="DataValueSnapshot"/> / <see cref="HistoricalEvent"/> carriers using the pure
///     mappers in <c>Mapping/</c>.
/// </summary>
/// <remarks>
///     The data source owns no historian connection of its own — it delegates to the gateway, which
///     pools and amortizes the underlying historian sessions. A thrown gateway exception is recorded
///     as a health failure and rethrown: the node manager turns it into a Bad HistoryRead result, so
///     a backend fault never crashes the host. An empty time window is a successful (GoodNoData)
///     read, not a fault. Health counters follow the single-<c>_healthLock</c> discipline ported
///     from <c>WonderwareHistorianClient</c> so <c>TotalSuccesses + TotalFailures == TotalQueries</c>
///     holds at every observed snapshot.
/// </remarks>
public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDisposable
{
    /// <summary>
    ///     The processed (aggregate) functions the Gateway backend serves — the single source of truth for this backend's
    ///     aggregate set (#681): the per-sample aggregates evaluated from raw reads (every member of the
    ///     <see cref="AggregateFamily.Simple"/> / <see cref="AggregateFamily.StartEnd"/> /
    ///     <see cref="AggregateFamily.ActualTime"/> families of <see cref="AggregateFamilies"/>: Average, Count, Minimum,
    ///     Maximum, Range, Start, End, Delta, MinimumActualTime, MaximumActualTime) plus the two answered natively by AVEVA
    ///     (TimeAverage, Total — <see cref="AggregateModeMapper"/>). Any other aggregate throws
    ///     <see cref="HistoryAggregateNotSupportedException"/> without calling the gateway. It is what
    ///     <see cref="SupportedAggregates"/> returns, so the advertised <c>AggregateFunctions</c> folders and the server's
    ///     up-front <c>BadAggregateNotSupported</c> read it too (#682).
    /// </summary>
    public static IReadOnlySet<HistoryAggregateType> ServedAggregates { get; } =
        Enum.GetValues<HistoryAggregateType>()
            .Where(a => RawAggregateReader.Serves(a) || AggregateModeMapper.NativeAggregates.Contains(a))
            .ToFrozenSet();

    private readonly IHistorianGatewayClient _client;
    private readonly ILogger<GatewayHistorianDataSource> _logger;
    private readonly int _maxRawRows;
    private readonly TimeProvider _time;

    private readonly object _healthLock = new();
    private DateTime? _lastSuccessUtc;
    private DateTime? _lastFailureUtc;
    private string? _lastError;
    private long _totalQueries;
    private long _totalSuccesses;
    private long _totalFailures;
    private int _consecutiveFailures;

    /// <summary>Creates a gateway-backed historian data source.</summary>
    /// <param name="client">The gateway client seam used for all reads.</param>
    /// <param name="logger">Diagnostic logger; failures are recorded without leaking tag/host detail.</param>
    /// <param name="maxRawRows">
    ///     In-window raw-row cap of one raw-evaluated processed read (<c>ServerHistorian:MaxTimeWeightedRawRows</c>;
    ///     &lt;= 0 = unbounded). Past it the node fails <c>BadTooManyOperations</c>.
    /// </param>
    /// <param name="timeProvider">Clock for result server timestamps and the raw look-ahead horizon; null = system.</param>
    public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger<GatewayHistorianDataSource> logger,
        int maxRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, TimeProvider? timeProvider = null)
    {
        ArgumentNullException.ThrowIfNull(client);
        ArgumentNullException.ThrowIfNull(logger);
        _client = client;
        _logger = logger;
        _maxRawRows = maxRawRows;
        _time = timeProvider ?? TimeProvider.System;
    }

    /// <inheritdoc />
    public async Task<HistoryReadResult> ReadRawAsync(
        string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
        CancellationToken cancellationToken)
    {
        try
        {
            // The gateway seam caps with an int; OPC UA hands us a uint, so clamp to int range.
            var maxValues = (int)Math.Min(maxValuesPerNode, int.MaxValue);
            var samples = new List<HistorianSample>();
            await foreach (var sample in _client
                .ReadRawAsync(fullReference, startUtc, endUtc, maxValues, cancellationToken)
                .ConfigureAwait(false))
            {
                samples.Add(sample);
            }

            var snapshots = SampleMapper.ToSnapshots(samples);
            RecordOutcome(success: true, error: null);
            return new HistoryReadResult(snapshots, ContinuationPoint: null);
        }
        catch (Exception ex)
        {
            RecordReadFailure(ex);
            throw;
        }
    }

    /// <inheritdoc />
    /// <remarks>
    ///     <para>
    ///         <b>Per-sample aggregates (#681)</b> — Average, Count, Minimum, Maximum, Range, Start, End, Delta,
    ///         Minimum/MaximumActualTime — are evaluated from a raw read with the SQL backends' shared evaluator
    ///         (<see cref="RawAggregateReader"/>): identical values, status, Calculated/Partial bits and timestamps to
    ///         SqlServer/QuestDB for the same samples, forward and reverse (#677). The in-window raw rows are capped by
    ///         <c>ServerHistorian:MaxTimeWeightedRawRows</c> (<see cref="HistoryReadLimitExceededException"/>, not counted as
    ///         a backend failure).
    ///     </para>
    ///     <para>
    ///         <b>TimeAverage / Total</b> stay native AVEVA retrieval modes (<see cref="AggregateModeMapper"/>), stamped with
    ///         the interval start and the Calculated bit. Everything else is <see cref="HistoryAggregateNotSupportedException"/>
    ///         (see <see cref="ServedAggregates"/>).
    ///     </para>
    /// </remarks>
    /// <exception cref="ArgumentOutOfRangeException">A raw-evaluated read whose window holds more than
    ///     <see cref="BucketPlan.MaxBuckets"/> intervals.</exception>
    public async Task<HistoryReadResult> ReadProcessedAsync(
        string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
        HistoryAggregateType aggregate, CancellationToken cancellationToken)
    {
        if (RawAggregateReader.Serves(aggregate))
            return await ReadProcessedFromRawAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken)
                .ConfigureAwait(false);

        // Mapped OUTSIDE the failure-recording try: an aggregate the gateway has no mode for is a client
        // request the server answers BadAggregateNotSupported, not a backend fault to count (#285).
        var mode = AggregateModeMapper.ToRetrievalMode(aggregate);
        try
        {
            // TimeAverage / Total are native gateway retrieval modes (TimeWeightedAverage / Integral) — no
            // client-side scaling.
            var buckets = new List<HistorianAggregateSample>();
            await foreach (var bucket in _client
                .ReadAggregateAsync(fullReference, startUtc, endUtc, mode, interval, cancellationToken)
                .ConfigureAwait(false))
            {
                buckets.Add(bucket);
            }

            var snapshots = SampleMapper.ToAggregateSnapshots(buckets);
            RecordOutcome(success: true, error: null);
            return new HistoryReadResult(snapshots, ContinuationPoint: null);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            throw; // caller-requested, not a backend fault
        }
        catch (Exception ex)
        {
            RecordReadFailure(ex);
            throw;
        }
    }

    /// <summary>Serves a per-sample aggregate from a raw read (#681); see <see cref="RawAggregateReader"/>.</summary>
    private async Task<HistoryReadResult> ReadProcessedFromRawAsync(
        string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
        HistoryAggregateType aggregate, CancellationToken cancellationToken)
    {
        // Planned OUTSIDE the failure-recording try (as on the SQL backends): too many intervals is a client request
        // limit, not a backend fault.
        var plan = BucketPlan.Create(startUtc, endUtc, interval);
        try
        {
            var now = _time.GetUtcNow().UtcDateTime;
            var values = await RawAggregateReader
                .ReadAsync(_client, fullReference, aggregate, plan, _maxRawRows, now, cancellationToken)
                .ConfigureAwait(false);
            RecordOutcome(success: true, error: null);
            return new HistoryReadResult(values, ContinuationPoint: null);
        }
        catch (HistoryReadLimitExceededException)
        {
            // A client-request limit (the raw-row cap), not a backend fault: not counted as a failure.
            throw;
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            // The caller (client cancel / HistoryReadDeadline) asked for this — not a backend fault.
            throw;
        }
        catch (Exception ex) when (IsMaxValuesCeilingRejection(ex))
        {
            // The gateway refused the row budget (Historian:ReadLimits:MaxValuesCeiling with
            // AllowStreamingBeyondCeiling=false): the same client-request limit as the cap, not a backend fault.
            throw new HistoryReadLimitExceededException(_maxRawRows,
                $"The HistorianGateway refused the raw row budget of the processed read of '{fullReference}' (its " +
                "Historian:ReadLimits:MaxValuesCeiling); lower ServerHistorian:MaxTimeWeightedRawRows below that ceiling " +
                "or narrow the window.");
        }
        catch (Exception ex)
        {
            RecordReadFailure(ex);
            throw;
        }
    }

    /// <summary>
    ///     True when <paramref name="ex"/> (or an inner exception — the client package wraps the <see cref="RpcException"/>)
    ///     is the gateway's <c>InvalidArgument</c> "max_values exceeds the server ceiling" rejection.
    /// </summary>
    internal static bool IsMaxValuesCeilingRejection(Exception ex)
    {
        for (var e = ex; e is not null; e = e.InnerException)
            if (e is RpcException { StatusCode: StatusCode.InvalidArgument } rpc
                && rpc.Status.Detail.Contains("max_values exceeds the server ceiling", StringComparison.Ordinal))
                return true;
        return false;
    }

    /// <inheritdoc />
    public async Task<HistoryReadResult> ReadAtTimeAsync(
        string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken)
    {
        try
        {
            var samples = await _client
                .ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken)
                .ConfigureAwait(false);
            var aligned = AlignAtTimeSnapshots(timestampsUtc, samples);
            RecordOutcome(success: true, error: null);
            return new HistoryReadResult(aligned, ContinuationPoint: null);
        }
        catch (Exception ex)
        {
            RecordReadFailure(ex);
            throw;
        }
    }

    /// <inheritdoc />
    /// <remarks>
    ///     <c>false</c>: the gateway's <c>ReadEvents</c> streams ascending only (Gitea #675). The server therefore refuses an
    ///     explicit reverse events read without calling, and keeps an end-only capped read forward.
    /// </remarks>
    public bool SupportsReverseEvents => false;

    /// <inheritdoc />
    /// <remarks><see cref="ServedAggregates"/>; every other member answers <c>BadAggregateNotSupported</c> and is not advertised.</remarks>
    public IReadOnlySet<HistoryAggregateType> SupportedAggregates => ServedAggregates;

    /// <inheritdoc />
    /// <remarks>
    ///     A reverse (newest-first) read is refused with <see cref="NotSupportedException"/> — the server reports
    ///     <c>BadHistoryOperationUnsupported</c> (Gitea #675). The gateway's <c>ReadEvents</c> streams ascending only and
    ///     rejects <c>end &lt; start</c>, so the only way to serve "the latest N" would be to drain the whole window and
    ///     keep its tail, an unbounded read the cap exists to prevent. Nothing is sent.
    /// </remarks>
    public Task<HistoricalEventsResult> ReadEventsAsync(
        string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, bool reverse,
        CancellationToken cancellationToken) =>
        reverse
            ? throw new NotSupportedException(
                "The HistorianGateway backend reads events oldest-first only; a reverse (newest-first) events read is not supported.")
            : ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken);

    /// <inheritdoc />
    public async Task<HistoricalEventsResult> ReadEventsAsync(
        string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
        CancellationToken cancellationToken)
    {
        try
        {
            var hasCap = maxEvents > 0;
            var collected = new List<HistoricalEvent>(hasCap ? maxEvents : 0);
            var truncated = false;

            await foreach (var wireEvent in _client
                .ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken)
                .ConfigureAwait(false))
            {
                var mapped = EventMapper.ToHistoricalEvent(wireEvent);

                // Defensive client-side source filter: the gateway's SQL ReadEvents source filter
                // may not be present, so drop any event whose source does not match the request.
                if (sourceName is not null && !string.Equals(mapped.SourceName, sourceName, StringComparison.Ordinal))
                {
                    continue;
                }

                // One more matching event arriving once the cap is full means the result is
                // truncated — stop draining and flag it.
                if (hasCap && collected.Count == maxEvents)
                {
                    truncated = true;
                    break;
                }

                collected.Add(mapped);
            }

            RecordOutcome(success: true, error: null);
            // A non-null, opaque token signals truncation to the caller.
            // The gateway has no resumable cursor, so the token's contents carry no paging state —
            // its presence alone is the "more events exist" signal. A fresh array per call keeps it
            // from being shared/mutated.
            return new HistoricalEventsResult(collected, truncated ? new byte[] { 0x01 } : null);
        }
        catch (Exception ex)
        {
            RecordReadFailure(ex);
            throw;
        }
    }

    /// <inheritdoc />
    public HistorianHealthSnapshot GetHealthSnapshot()
    {
        lock (_healthLock)
        {
            return new HistorianHealthSnapshot(
                TotalQueries: _totalQueries,
                TotalSuccesses: _totalSuccesses,
                TotalFailures: _totalFailures,
                ConsecutiveFailures: _consecutiveFailures,
                LastSuccessTime: _lastSuccessUtc,
                LastFailureTime: _lastFailureUtc,
                LastError: _lastError,
                // No connection probe exists (the old RefreshConnectionStateAsync had no production
                // caller and was deleted, archreview 06/S-6): "open" is derived from observed reads
                // alone -- at least one success and no failure since. The gateway is non-clustered to
                // us, so node fields are null/empty (mirrors the Wonderware client's Finding 010 posture).
                ProcessConnectionOpen: _totalSuccesses > 0 && _consecutiveFailures == 0,
                EventConnectionOpen: _totalSuccesses > 0 && _consecutiveFailures == 0,
                ActiveProcessNode: null,
                ActiveEventNode: null,
                Nodes: []);
        }
    }

    /// <summary>
    ///     Reconciles a gateway at-time reply against the requested timestamps to honour the
    ///     <see cref="IHistorianDataSource.ReadAtTimeAsync"/> contract: exactly one snapshot per
    ///     requested timestamp, in request order. Returned samples are indexed by timestamp ticks;
    ///     any requested timestamp the gateway did not return is filled with a Bad-quality
    ///     (<c>0x80000000</c>) snapshot stamped at the requested time rather than positionally
    ///     misaligning values. The alignment logic was ported from the now-retired Wonderware
    ///     client's at-time snapshot reconciliation.
    /// </summary>
    private static IReadOnlyList<DataValueSnapshot> AlignAtTimeSnapshots(
        IReadOnlyList<DateTime> timestampsUtc, IReadOnlyList<HistorianSample> samples)
    {
        // Index returned samples by timestamp ticks. Duplicate timestamps keep the first.
        var byTicks = new Dictionary<long, HistorianSample>(samples.Count);
        foreach (var sample in samples)
        {
            if (sample.Timestamp is null) continue;
            byTicks.TryAdd(sample.Timestamp.ToDateTime().Ticks, sample);
        }

        var result = new DataValueSnapshot[timestampsUtc.Count];
        for (var i = 0; i < timestampsUtc.Count; i++)
        {
            var requested = DateTime.SpecifyKind(timestampsUtc[i], DateTimeKind.Utc);
            if (byTicks.TryGetValue(requested.Ticks, out var sample))
            {
                // Reuse the shared sample mapper for value + quality, then re-stamp the source
                // timestamp to the requested time per the ReadAtTime contract.
                result[i] = SampleMapper.ToSnapshot(sample) with { SourceTimestampUtc = requested };
            }
            else
            {
                // Gap — gateway returned no sample for this timestamp. Per the contract this is a
                // Bad-quality snapshot stamped at the requested time, not a dropped row.
                result[i] = new DataValueSnapshot(
                    Value: null,
                    StatusCode: 0x80000000u, // Bad
                    SourceTimestampUtc: requested,
                    ServerTimestampUtc: DateTime.UtcNow);
            }
        }

        return result;
    }

    /// <summary>
    ///     Records a failed read: bumps the health counters and logs a generic, redaction-safe
    ///     debug line (no tag, host, or value). The exception itself is rethrown by the caller.
    /// </summary>
    private void RecordReadFailure(Exception ex)
    {
        RecordOutcome(success: false, error: ex.Message);
        _logger.LogDebug("Historian gateway read operation failed and was recorded as a health failure.");
    }

    /// <summary>
    ///     Records the outcome of a single read — increments <c>_totalQueries</c> and exactly one of
    ///     <c>_totalSuccesses</c> / <c>_totalFailures</c> under a single <c>_healthLock</c>
    ///     acquisition so a concurrent <see cref="GetHealthSnapshot"/> never observes a torn state.
    /// </summary>
    private void RecordOutcome(bool success, string? error)
    {
        lock (_healthLock)
        {
            _totalQueries++;
            if (success)
            {
                _totalSuccesses++;
                _consecutiveFailures = 0;
                _lastSuccessUtc = DateTime.UtcNow;
            }
            else
            {
                _totalFailures++;
                _consecutiveFailures++;
                _lastFailureUtc = DateTime.UtcNow;
                _lastError = error;
            }
        }
    }

    /// <summary>Disposes the underlying gateway client. Prefer <see cref="DisposeAsync"/>.</summary>
    public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();

    /// <summary>Asynchronously disposes the underlying gateway client.</summary>
    /// <returns>A task that completes when the client has been disposed.</returns>
    public ValueTask DisposeAsync() => _client.DisposeAsync();
}