lmxopcua/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs

using System.Collections;
using System.Collections.Concurrent;
using System.Diagnostics;

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

/// <summary>
///     Shared poll-based subscription engine for drivers whose underlying protocol has no
///     native push model (Modbus, AB CIP, S7, FOCAS). Owns one background Task per subscription
///     that periodically invokes the supplied reader, diffs each snapshot against the last
///     known value, and dispatches a change callback per changed tag. Extracted from
///     <c>ModbusDriver</c> (AB CIP PR 1) so poll-based drivers don't each re-ship the loop,
///     floor logic, and lifecycle plumbing.
/// </summary>
/// <remarks>
///     <para>The engine is read-path agnostic: it calls the supplied <c>reader</c> delegate
///     and trusts the driver to map protocol errors into <see cref="DataValueSnapshot.StatusCode"/>.
///     Callbacks fire on: (a) the first poll after subscribe (initial-data push per the OPC UA
///     Part 4 convention), (b) any subsequent poll where the boxed value or status code differs
///     from the previously-seen snapshot.</para>
///
///     <para>Exceptions thrown by the reader on the initial poll or any subsequent poll are
///     caught — the loop continues on the next tick. When an <c>onError</c> callback is supplied
///     to the constructor the caught exception is routed to it so the driver's health surface
///     can record the failure. Without an <c>onError</c> callback the exception is silently
///     swallowed (preserves the original behaviour for drivers that have not opted in yet).</para>
///
///     <para>An optional <c>onBatch</c> callback delivers the SAME changes collected per poll
///     cycle: it is raised at most once per cycle, only when that cycle produced at least one
///     change, and always AFTER every per-tag <c>onChange</c> callback of that same cycle. Drivers
///     forward it to <see cref="IBatchSubscribable.OnDataChangeBatch"/> so downstream consumers can
///     take one message per cycle instead of one per tag. Both callbacks fire by design — a
///     consumer subscribes to exactly one of the two. Without an <c>onBatch</c> callback the engine
///     behaves byte-identically to before: no batch is collected and no per-cycle allocation is
///     made.</para>
///
///     <para>Pacing is <b>fixed-rate, not fixed-delay</b>: the post-poll wait is
///     <c>max(interval - elapsed, 0)</c>, so a read that takes 80 ms on a 100 ms interval is
///     followed by a 20 ms wait rather than a full 100 ms one, and the observed cadence stays at
///     the configured interval instead of drifting out to (read + interval). When a poll takes
///     LONGER than the interval the next one starts immediately, the overrun is counted in
///     <see cref="OverrunCount"/>, and the optional <c>onOverrun</c> callback is invoked with the
///     measured duration. The failure/backoff branch is deliberately exempt: once
///     <c>consecutiveFailures &gt; 0</c> with a <c>backoffCap</c> configured, the capped-exponential
///     schedule wins, because a failing device must not be hammered at fixed rate.</para>
///
///     <para><b>No <c>Meter</c> lives here, and that is still deliberate.</b>
///     <c>Core.Abstractions</c> references nothing, so <see cref="OverrunCount"/> plus the
///     <c>onOverrun</c> callback ARE the instrumentation surface — a driver that wants a metric
///     wires its own meter to the callback. The driver-level meter this note used to record as a
///     follow-up is <b>delivered</b>: every one of the seven poll drivers now passes an
///     <c>onOverrun</c> closure that bumps <c>OtOpcUaTelemetry.PollOverruns</c>
///     (<c>otopcua.driver.poll.overrun</c>) on the shared OtOpcUa <c>Meter</c>, tagged with the
///     driver's <c>DriverTypeNames</c> constant and its driver-instance id. The counter lives in
///     <c>Commons.Observability</c>, on the consumer side of the reference graph — this type still
///     owns no <c>Meter</c> and takes no metrics dependency.</para>
///
///     <para>Programmer errors and obviously-fatal exceptions (<see cref="OutOfMemoryException"/>,
///     <see cref="ThreadAbortException"/>, <see cref="StackOverflowException"/>,
///     <see cref="AccessViolationException"/>) are NOT caught — they propagate and tear the poll
///     loop down rather than spin a silently-broken subscription.</para>
/// </remarks>
public sealed class PollGroupEngine : IAsyncDisposable
{
    private readonly Func<IReadOnlyList<string>, CancellationToken, Task<IReadOnlyList<DataValueSnapshot>>> _reader;
    private readonly Action<ISubscriptionHandle, string, DataValueSnapshot> _onChange;
    private readonly Action<Exception>? _onError;
    private readonly Action<ISubscriptionHandle, IReadOnlyList<DataChangeItem>>? _onBatch;
    private readonly Action<TimeSpan>? _onOverrun;
    private readonly TimeSpan _minInterval;
    private readonly TimeSpan? _backoffCap;
    private readonly ConcurrentDictionary<long, SubscriptionState> _subscriptions = new();
    private long _nextId;
    private long _overrunCount;

    /// <summary>Default floor for publishing intervals — matches the Modbus 100 ms cap.</summary>
    public static readonly TimeSpan DefaultMinInterval = TimeSpan.FromMilliseconds(100);

    /// <summary>Constructs the engine with the driver-supplied reader, change callback, and optional interval floor / error sink.</summary>
    /// <param name="reader">Driver-supplied batch reader; snapshots MUST be returned in the same
    ///     order as the input references.</param>
    /// <param name="onChange">Callback invoked per changed tag — the driver forwards to its own
    ///     <see cref="ISubscribable.OnDataChange"/> event.</param>
    /// <param name="minInterval">Interval floor; anything below is clamped. Defaults to 100 ms
    ///     per <see cref="DefaultMinInterval"/>.</param>
    /// <param name="onError">Optional error sink — invoked once per caught reader exception (or
    ///     internal contract-violation throw) so the owning driver can route the failure to its
    ///     health surface. Defensive: an <c>onError</c> handler that
    ///     itself throws is silently absorbed so a buggy forwarder cannot crash the poll loop.</param>
    /// <param name="backoffCap">Optional cap enabling capped-exponential failure backoff
    ///     (05/STAB-8). When supplied, sustained reader failures stretch the retry cadence
    ///     (interval, 2×, 4×, … saturating at this cap) instead of hammering a dead device every
    ///     tick; a successful poll resets to the base interval. <c>null</c> (default) keeps the
    ///     byte-identical fixed-interval cadence for consumers that have not opted in.
    ///     <para><b>Units:</b> a <see cref="TimeSpan"/> ceiling on the post-poll <i>wait</i> — the
    ///     longest gap the loop will ever leave between a failed poll and the next attempt. It is
    ///     not a count, a multiplier, or a total-outage budget: a device that stays down waits
    ///     <c>backoffCap</c> between every retry, forever.</para>
    ///     <para><b>Interaction with <paramref name="minInterval"/> — it REPLACES the paced delay,
    ///     it does not add to it.</b> The loop picks exactly one of two branches per cycle. With no
    ///     failures outstanding (or no cap configured) the wait is the fixed-RATE remainder
    ///     <c>max(interval − pollDuration, 0)</c>. With a cap configured AND
    ///     <c>consecutiveFailures &gt; 0</c>, the whole remainder computation is bypassed and the
    ///     wait becomes <c>ConnectionBackoff.ComputeDelay(interval, failures, cap)</c> — the poll's
    ///     own elapsed time is NOT subtracted, so the failure branch is a fixed DELAY (cycle =
    ///     poll + backoff), deliberately unlike the fixed-rate success branch.</para>
    ///     <para>The <c>interval</c> feeding that schedule is the per-subscription interval
    ///     <i>already clamped</i> to <paramref name="minInterval"/> in <see cref="Subscribe"/>, so
    ///     the first failure's wait equals the clamped interval (shift 0 → 1×) and the floor is
    ///     honoured for free. One asymmetry worth knowing: <c>ComputeDelay</c> saturates at the cap
    ///     unconditionally, so a <paramref name="backoffCap"/> set SMALLER than the clamped
    ///     interval makes failures retry FASTER than the interval floor — the cap outranks
    ///     <paramref name="minInterval"/> on the failure branch. Configure it above the floor
    ///     (every driver here uses 30 s against a 100 ms floor).</para></param>
    /// <param name="onBatch">Optional per-cycle batch callback — invoked once per poll cycle that
    ///     produced at least one change, carrying exactly that cycle's changed items in reader
    ///     order, after all of the cycle's per-tag <paramref name="onChange"/> callbacks. The
    ///     driver forwards it to its own <see cref="IBatchSubscribable.OnDataChangeBatch"/> event.
    ///     <c>null</c> (default) collects nothing and allocates nothing per cycle.</param>
    /// <param name="onOverrun">Optional overrun sink — invoked with the measured poll duration
    ///     every time a poll takes longer than its subscription's interval (the same events counted
    ///     by <see cref="OverrunCount"/>). Since this assembly carries no metrics dependency, this
    ///     callback is how a driver surfaces the condition to its own meter/log. Defensive: a
    ///     handler that throws is absorbed, exactly like <c>onError</c>.</param>
    public PollGroupEngine(
        Func<IReadOnlyList<string>, CancellationToken, Task<IReadOnlyList<DataValueSnapshot>>> reader,
        Action<ISubscriptionHandle, string, DataValueSnapshot> onChange,
        TimeSpan? minInterval = null,
        Action<Exception>? onError = null,
        TimeSpan? backoffCap = null,
        Action<ISubscriptionHandle, IReadOnlyList<DataChangeItem>>? onBatch = null,
        Action<TimeSpan>? onOverrun = null)
    {
        ArgumentNullException.ThrowIfNull(reader);
        ArgumentNullException.ThrowIfNull(onChange);
        _reader = reader;
        _onChange = onChange;
        _onError = onError;
        _minInterval = minInterval ?? DefaultMinInterval;
        _backoffCap = backoffCap;
        _onBatch = onBatch;
        _onOverrun = onOverrun;
    }

    /// <summary>
    ///     Total number of poll cycles, across every subscription of this engine, whose read took
    ///     longer than the subscription's publishing interval. A non-zero and climbing value means
    ///     the device cannot sustain the configured rate — the loop is already running flat out.
    /// </summary>
    public long OverrunCount => Interlocked.Read(ref _overrunCount);

    /// <summary>Register a new polled subscription and start its background loop.</summary>
    /// <param name="fullReferences">The list of tag references to poll.</param>
    /// <param name="publishingInterval">The desired polling interval; will be clamped to the configured minimum.</param>
    /// <returns>A handle identifying the new subscription, for later use with <see cref="Unsubscribe"/>.</returns>
    public ISubscriptionHandle Subscribe(IReadOnlyList<string> fullReferences, TimeSpan publishingInterval)
    {
        ArgumentNullException.ThrowIfNull(fullReferences);
        var id = Interlocked.Increment(ref _nextId);
        var cts = new CancellationTokenSource();
        var interval = publishingInterval < _minInterval ? _minInterval : publishingInterval;
        var handle = new PollSubscriptionHandle(id);
        var state = new SubscriptionState(handle, [.. fullReferences], interval, cts);
        _subscriptions[id] = state;
        state.LoopTask = Task.Run(() => PollLoopAsync(state, cts.Token));
        return handle;
    }

    /// <summary>
    ///     Cancel the background loop for a handle returned by <see cref="Subscribe"/>, BLOCKING the
    ///     calling thread until the in-flight poll drains. Retained for compatibility; prefer
    ///     <see cref="UnsubscribeAsync"/> on any path that runs inside an actor mailbox turn or a
    ///     request pipeline, where blocking a thread for up to 5 s is never acceptable.
    /// </summary>
    /// <returns><c>true</c> when the handle was known to the engine and has been torn down.</returns>
    /// <param name="handle">The subscription handle to cancel.</param>
    public bool Unsubscribe(ISubscriptionHandle handle)
    {
        if (!TryTake(handle, out var state)) return false;
        StopState(state);
        return true;
    }

    /// <summary>
    ///     Non-blocking counterpart of <see cref="Unsubscribe"/>: identical teardown semantics (the
    ///     loop is cancelled and awaited before the CTS is disposed, so no callback can fire after
    ///     this completes) but it awaits the loop task instead of blocking a thread on it.
    /// </summary>
    /// <param name="handle">The subscription handle to cancel.</param>
    /// <returns><c>true</c> when the handle was known to the engine and has been torn down.</returns>
    public async ValueTask<bool> UnsubscribeAsync(ISubscriptionHandle handle)
    {
        if (!TryTake(handle, out var state)) return false;
        await StopStateAsync(state).ConfigureAwait(false);
        return true;
    }

    /// <summary>
    ///     Shared front half of both teardown paths: recognise the handle and remove it from the
    ///     registry exactly once, so a double unsubscribe (or a race between the sync and async
    ///     overloads) tears one subscription down once and reports <c>false</c> for the loser.
    /// </summary>
    private bool TryTake(ISubscriptionHandle handle, out SubscriptionState state)
    {
        if (handle is PollSubscriptionHandle h && _subscriptions.TryRemove(h.Id, out var found))
        {
            state = found;
            return true;
        }
        state = null!;
        return false;
    }

    // Cancel + drain + dispose, shared by both teardown paths. The loop task is awaited (with a
    // generous timeout) before the CTS is disposed so:
    //  (a) no _onChange callback fires after the caller considers the subscription torn down, and
    //  (b) the CTS is not disposed while Task.Delay is still holding a reference to its token,
    //      which can turn OperationCanceledException into ObjectDisposedException.
    private static void StopState(SubscriptionState state)
    {
        try { state.Cts.Cancel(); } catch { }
        var task = state.LoopTask;
        if (task is not null)
        {
            try { task.Wait(DrainTimeout); } catch { }
        }
        state.Cts.Dispose();
    }

    private static async ValueTask StopStateAsync(SubscriptionState state)
    {
        try { state.Cts.Cancel(); } catch { }
        var task = state.LoopTask;
        if (task is not null)
        {
            try { await task.WaitAsync(DrainTimeout).ConfigureAwait(false); } catch { }
        }
        state.Cts.Dispose();
    }

    /// <summary>Ceiling on how long a teardown waits for an in-flight poll to unwind.</summary>
    private static readonly TimeSpan DrainTimeout = TimeSpan.FromSeconds(5);

    /// <summary>Snapshot of active subscription count — exposed for driver diagnostics.</summary>
    public int ActiveSubscriptionCount => _subscriptions.Count;

    private async Task PollLoopAsync(SubscriptionState state, CancellationToken ct)
    {
        var consecutiveFailures = 0;
        // Duration of the poll just performed — the fixed-rate pacing subtracts it from the
        // interval so the cadence is the interval, not (interval + read time).
        TimeSpan lastPollDuration;

        // Initial-data push: every subscribed tag fires once at subscribe time regardless of
        // whether it has changed, satisfying OPC UA Part 4 initial-value semantics.
        var started = Stopwatch.GetTimestamp();
        try
        {
            await PollOnceAsync(state, forceRaise: true, ct).ConfigureAwait(false);
            consecutiveFailures = 0;
        }
        // Only a teardown OCE (the loop's OWN token) exits the loop. A reader that throws OCE with
        // the loop token un-cancelled — a driver-internal timeout CTS (the STAB-14/STAB-15 class) —
        // must NOT tear the loop down: it falls through to the generic handler, is reported, and the
        // loop retries with backoff. This is the guard that makes the S7 fork migration safe.
        catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; }
        catch (Exception ex) when (!IsFatal(ex))
        {
            // first-read error tolerated — loop continues; forward to driver health surface.
            consecutiveFailures++;
            ReportError(ex);
        }
        finally { lastPollDuration = Stopwatch.GetElapsedTime(started); }
        NoteOverrun(state.Interval, lastPollDuration);

        while (!ct.IsCancellationRequested)
        {
            // Fixed RATE, not fixed delay: the wait is what remains of the interval after the poll
            // that just ran, so a slow-but-succeeding device keeps the configured cadence and a
            // poll that overran starts the next one immediately.
            //
            // The failure branch is deliberately exempt: once a backoff cap is configured AND the
            // device is failing, the capped-exponential schedule wins outright — a dead device must
            // not be hammered at fixed rate. With no failures ComputeDelay returns the base interval
            // anyway, so it is only consulted when it can actually change the answer.
            TimeSpan delay;
            if (_backoffCap is not null && consecutiveFailures > 0)
            {
                delay = ConnectionBackoff.ComputeDelay(state.Interval, consecutiveFailures, _backoffCap.Value);
            }
            else
            {
                var remaining = state.Interval - lastPollDuration;
                delay = remaining > TimeSpan.Zero ? remaining : TimeSpan.Zero;
            }

            if (delay > TimeSpan.Zero)
            {
                try { await Task.Delay(delay, ct).ConfigureAwait(false); }
                catch (OperationCanceledException) { return; }
                // Defensive: the CTS may be disposed by Unsubscribe/DisposeAsync between the
                // cancellation check above and the Task.Delay touching the token. Treat that race
                // as a normal cancellation rather than a fatal exception.
                catch (ObjectDisposedException) { return; }
            }
            else
            {
                // Overrun: no wait is owed. Still yield, so a reader that completes synchronously
                // can never turn a zero delay into a thread-pool-starving tight loop, and so the
                // cancellation check below is reached.
                await Task.Yield();
                if (ct.IsCancellationRequested) return;
            }

            started = Stopwatch.GetTimestamp();
            try
            {
                await PollOnceAsync(state, forceRaise: false, ct).ConfigureAwait(false);
                consecutiveFailures = 0;
            }
            // Only teardown (the loop token) exits; a reader OCE with ct un-cancelled backs off
            // like any other failure instead of killing the loop (STAB-14 engine guard).
            catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; }
            catch (Exception ex) when (!IsFatal(ex))
            {
                // transient poll error — loop continues (with backoff when configured); driver
                // health surface logs it via the supplied onError callback.
                consecutiveFailures++;
                ReportError(ex);
            }
            finally { lastPollDuration = Stopwatch.GetElapsedTime(started); }
            NoteOverrun(state.Interval, lastPollDuration);
        }
    }

    /// <summary>
    ///     Record a poll that took longer than its interval: bump <see cref="OverrunCount"/> and
    ///     hand the measured duration to the optional <c>onOverrun</c> sink. Defensive against a
    ///     sink that throws, for the same reason <see cref="ReportError"/> is.
    ///     <para><b>Caveat — this CONFLATES a slow success with a slow failure.</b> It is called
    ///     from the <c>finally</c> of both poll sites, so the only input is the measured elapsed
    ///     time; a device that answers in 300 ms on a 200 ms interval and one that throws (or times
    ///     out) after 300 ms both increment the counter and both raise <c>onOverrun</c> with the
    ///     same duration. A consumer reading <c>otopcua.driver.poll.overrun</c> therefore learns
    ///     "the loop cannot keep its cadence", NOT "the device is healthy but slow" — pair it with
    ///     the driver's health surface (which <c>onError</c> already feeds) to tell the two apart.
    ///     Disambiguating here would mean threading the cycle's outcome into this method and
    ///     splitting the counter/callback by it; no consumer needs that yet, so it is deliberately
    ///     not done.</para>
    /// </summary>
    private void NoteOverrun(TimeSpan interval, TimeSpan elapsed)
    {
        if (elapsed <= interval) return;
        Interlocked.Increment(ref _overrunCount);
        if (_onOverrun is null) return;
        try { _onOverrun(elapsed); }
        catch { /* never let a buggy overrun sink stop the poll loop */ }
    }

    /// <summary>
    ///     Programmer-error / process-fatal exception classification: anything that cannot be
    ///     safely "swallowed and retry on the next tick" must escape the poll loop instead.
    /// </summary>
    private static bool IsFatal(Exception ex)
        => ex is OutOfMemoryException
            or StackOverflowException
            or AccessViolationException
            or ThreadAbortException;

    /// <summary>
    ///     Forward a caught exception to the optional <c>onError</c> callback. Defensive
    ///     against an <c>onError</c> implementation that itself throws — that would crash the
    ///     poll loop and re-introduce the silent-stall failure mode this method exists to prevent.
    /// </summary>
    private void ReportError(Exception ex)
    {
        if (_onError is null) return;
        try { _onError(ex); }
        catch { /* never let a buggy error sink stop the poll loop */ }
    }

    private async Task PollOnceAsync(SubscriptionState state, bool forceRaise, CancellationToken ct)
    {
        var snapshots = await _reader(state.TagReferences, ct).ConfigureAwait(false);

        // Validate the reader contract before indexing. A reader that
        // returns fewer snapshots than references would silently stall the subscription; surface
        // the violation immediately with a descriptive exception instead.
        if (snapshots.Count != state.TagReferences.Count)
            throw new InvalidOperationException(
                $"Reader contract violation: expected {state.TagReferences.Count} snapshots but received {snapshots.Count}. " +
                "The reader delegate must return one snapshot per input reference in input order.");

        // Collected lazily: nothing is allocated for a cycle with no changes, and nothing at all
        // when no batch callback is wired.
        List<DataChangeItem>? changed = null;

        for (var i = 0; i < state.TagReferences.Count; i++)
        {
            var tagRef = state.TagReferences[i];
            var current = snapshots[i];
            var lastSeen = state.LastValues.TryGetValue(tagRef, out var prev) ? prev : default;

            if (forceRaise || ValuesAreDifferent(lastSeen?.Value, current.Value) || lastSeen?.StatusCode != current.StatusCode)
            {
                state.LastValues[tagRef] = current;
                _onChange(state.Handle, tagRef, current);
                // Capacity hint: a cycle can produce at most one item per subscribed reference, so
                // sizing the list once avoids the doubling re-allocations a large group would incur.
                if (_onBatch is not null)
                    (changed ??= new List<DataChangeItem>(state.TagReferences.Count)).Add(new DataChangeItem(tagRef, current));
            }
        }

        // One raise per cycle, after every per-tag callback of that cycle, never empty.
        if (changed is { Count: > 0 }) _onBatch!(state.Handle, changed);
    }

    /// <summary>
    ///     Returns <c>true</c> when <paramref name="previous"/> and <paramref name="current"/>
    ///     represent different values. Array values are compared structurally
    ///     (element-by-element) so that a driver producing a fresh array instance on every poll
    ///     does not trigger spurious change events when the contents are identical.
    /// </summary>
    private static bool ValuesAreDifferent(object? previous, object? current)
    {
        if (previous is Array prevArr && current is Array currArr)
            return !StructuralComparisons.StructuralEqualityComparer.Equals(prevArr, currArr);

        return !Equals(previous, current);
    }

    /// <summary>Cancel every active subscription and await all loop tasks. Idempotent.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    public async ValueTask DisposeAsync()
    {
        // Cancel all loops first so they can all start winding down in parallel.
        foreach (var state in _subscriptions.Values)
        {
            try { state.Cts.Cancel(); } catch { }
        }

        // Await every loop task before disposing CTSs, ensuring no callback fires after disposal.
        var waitTasks = _subscriptions.Values
            .Select(s => s.LoopTask ?? Task.CompletedTask)
            .ToArray();
        if (waitTasks.Length > 0)
        {
            try { await Task.WhenAll(waitTasks).WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); }
            catch { }
        }

        foreach (var state in _subscriptions.Values)
        {
            state.Cts.Dispose();
        }
        _subscriptions.Clear();
    }

    private sealed record SubscriptionState(
        PollSubscriptionHandle Handle,
        IReadOnlyList<string> TagReferences,
        TimeSpan Interval,
        CancellationTokenSource Cts)
    {
        /// <summary>Gets the cache of last-seen values per tag reference.</summary>
        public ConcurrentDictionary<string, DataValueSnapshot> LastValues { get; }
            = new(StringComparer.OrdinalIgnoreCase);

        /// <summary>
        ///     The background poll-loop task. Assigned immediately after creation in
        ///     <see cref="Subscribe"/>; awaited during <see cref="Unsubscribe"/> /
        ///     <see cref="DisposeAsync"/> so disposal is deterministic and no
        ///     <c>_onChange</c> callback can fire after the caller tears down the subscription.
        /// </summary>
        public Task? LoopTask { get; set; }
    }

    private sealed record PollSubscriptionHandle(long Id) : ISubscriptionHandle
    {
        /// <inheritdoc />
        public string DiagnosticId => $"poll-sub-{Id}";
    }
}