lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs

using System.Diagnostics.Metrics;
using System.Threading.Channels;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;

/// <summary>
///     Long-running consumer of <see cref="IGalaxySubscriber.StreamEventsAsync"/>. Translates
///     each <see cref="MxEvent"/> with family <see cref="MxEventFamily.OnDataChange"/> into
///     <see cref="DataChangeEventArgs"/> and dispatches one event per registered driver
///     subscription that includes the changed item handle (fan-out via
///     <see cref="SubscriptionRegistry.ResolveSubscribers"/>).
/// </summary>
/// <remarks>
///     <para>
///     One pump per connected <see cref="GalaxyMxSession"/>. Reconnect lives in the
///     <see cref="ReconnectSupervisor"/>; on transport failure here we log and invoke
///     the optional <c>onStreamFault</c> callback so the owner (GalaxyDriver) can
///     forward the fault to <see cref="ReconnectSupervisor.ReportTransportFailure"/>
///     and the supervisor can drive reopen → replay.
///     </para>
///     <para>
///     PR 6.2 — the network-read loop and the listener-fanout loop are decoupled by a
///     bounded <see cref="Channel{T}"/>. When a listener is slow enough to fill the
///     channel, new events are dropped (newest-dropped semantics: producer's
///     <c>TryWrite</c> fails) rather than back-pressuring the gw stream. Three counters
///     on the <c>ZB.MOM.WW.OtOpcUa.Driver.Galaxy</c> meter expose received / dispatched
///     / dropped totals so ops sees pressure before it manifests as user-visible loss.
///     </para>
/// </remarks>
internal sealed class EventPump : IAsyncDisposable
{
    public const string MeterName = "ZB.MOM.WW.OtOpcUa.Driver.Galaxy";
    private const int DefaultChannelCapacity = 50_000;

    // Single static meter so a host-level MeterListener catches all pump instances.
    private static readonly Meter Meter = new(MeterName);
    private static readonly Counter<long> EventsReceived =
        Meter.CreateCounter<long>("galaxy.events.received", unit: "{event}",
            description: "MxEvents read from the gateway StreamEvents stream.");
    private static readonly Counter<long> EventsDispatched =
        Meter.CreateCounter<long>("galaxy.events.dispatched", unit: "{event}",
            description: "MxEvents passed through the bounded channel and into OnDataChange.");
    private static readonly Counter<long> EventsDropped =
        Meter.CreateCounter<long>("galaxy.events.dropped", unit: "{event}",
            description: "MxEvents dropped because the bounded channel was full (newest-dropped).");
    private static readonly Counter<long> ReplayGapsObserved =
        Meter.CreateCounter<long>("galaxy.events.replay_gaps", unit: "{gap}",
            description: "Replay-gap sentinels received — the gateway evicted events we can never see.");
    private static readonly Counter<long> ValuesUndecodable =
        Meter.CreateCounter<long>("galaxy.events.undecodable", unit: "{event}",
            description: "Data-change events carrying a value kind this driver cannot represent.");

    private readonly IGalaxySubscriber _subscriber;
    private readonly SubscriptionRegistry _registry;
    private readonly ILogger _logger;
    private readonly Func<long, ISubscriptionHandle> _handleFactory;
    private readonly Action<Exception>? _onStreamFault;
    private readonly Action<ReplayGap>? _onReplayGap;
    private readonly Channel<MxEvent> _channel;
    private readonly KeyValuePair<string, object?> _clientTag;
    private readonly CancellationTokenSource _cts = new();

    private Task? _loop;
    private Task? _dispatchLoop;
    private bool _disposed;

    /// <summary>
    ///     Value kinds already warned about. Bounded by the <c>MxValue</c> oneof enum, so a
    ///     broken producer cannot flood the log, while a SECOND unrepresentable kind appearing
    ///     later still gets its own line — warn-once-per-pump would have swallowed it.
    /// </summary>
    private readonly HashSet<MxValue.KindOneofCase> _undecodableKindsWarned = [];

    /// <summary>Occurs when a data change event is received from the Galaxy subscriber.</summary>
    public event EventHandler<DataChangeEventArgs>? OnDataChange;

    /// <summary>Initializes a new instance of the EventPump class.</summary>
    /// <param name="subscriber">The Galaxy subscriber to consume events from.</param>
    /// <param name="registry">The subscription registry for resolving subscribers.</param>
    /// <param name="logger">The logger instance; if null, uses NullLogger.</param>
    /// <param name="handleFactory">The factory for creating subscription handles; if null, uses GalaxySubscriptionHandle.</param>
    /// <param name="channelCapacity">The bounded channel capacity for buffering events.</param>
    /// <param name="clientName">The client name for metric tagging; if null, uses "&lt;unknown&gt;".</param>
    /// <param name="onStreamFault">Optional callback invoked when the stream faults.</param>
    /// <param name="onReplayGap">
    ///     Optional callback invoked when the gateway reports a replay gap. The contract
    ///     obliges the client to re-snapshot; see <see cref="HandleReplayGap"/>.
    /// </param>
    public EventPump(
        IGalaxySubscriber subscriber,
        SubscriptionRegistry registry,
        ILogger? logger = null,
        Func<long, ISubscriptionHandle>? handleFactory = null,
        int channelCapacity = DefaultChannelCapacity,
        string? clientName = null,
        Action<Exception>? onStreamFault = null,
        Action<ReplayGap>? onReplayGap = null)
    {
        _subscriber = subscriber ?? throw new ArgumentNullException(nameof(subscriber));
        _registry = registry ?? throw new ArgumentNullException(nameof(registry));
        _logger = logger ?? NullLogger.Instance;
        _handleFactory = handleFactory ?? (id => new GalaxySubscriptionHandle(id));
        _onStreamFault = onStreamFault;
        _onReplayGap = onReplayGap;

        if (channelCapacity < 1)
        {
            throw new ArgumentOutOfRangeException(nameof(channelCapacity),
                "channelCapacity must be >= 1; recommended 50_000 for 50k-tag deployments.");
        }
        _channel = Channel.CreateBounded<MxEvent>(new BoundedChannelOptions(channelCapacity)
        {
            // Newest-dropped semantics: we use FullMode.Wait but never call the
            // awaitable WriteAsync — only the synchronous TryWrite below in
            // RunAsync. With Wait + TryWrite, a full channel makes TryWrite return
            // false immediately, which we account for via the EventsDropped counter.
            // We deliberately do NOT use BoundedChannelFullMode.DropWrite — that
            // would silently discard the new event inside Channel<T> without
            // surfacing the drop on a counter (keep this comment and the FullMode
            // value consistent).
            FullMode = BoundedChannelFullMode.Wait,
            SingleReader = true,
            SingleWriter = true,
        });
        _clientTag = new KeyValuePair<string, object?>("galaxy.client", clientName ?? "<unknown>");
    }

    /// <summary>
    ///     Start consuming the event stream on a background task. Idempotent — second
    ///     calls are no-ops while the loop is running.
    /// </summary>
    public void Start()
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
        if (_loop is not null) return;
        _loop = Task.Run(() => RunAsync(_cts.Token));
        _dispatchLoop = Task.Run(() => DispatchLoopAsync(_cts.Token));
    }

    private async Task RunAsync(CancellationToken ct)
    {
        try
        {
            await foreach (var ev in _subscriber.StreamEventsAsync(ct).WithCancellation(ct).ConfigureAwait(false))
            {
                if (ct.IsCancellationRequested) break;
                EventsReceived.Add(1, _clientTag);

                // The replay-gap sentinel is a control signal, not data, and is handled HERE
                // rather than in the dispatch loop for one reason: everything past this point
                // is subject to newest-dropped channel pressure, and a dropped "you have
                // missed events" signal is the one loss that cannot be recovered from later.
                // It carries no item payload, so there is nothing to fan out downstream.
                if (ev.ReplayGap is { } gap)
                {
                    HandleReplayGap(gap);
                    continue;
                }

                // Newest-dropped: TryWrite fast-paths the common case (channel has room).
                // When full we count the drop and continue reading the gw stream so
                // back-pressure doesn't propagate upstream.
                if (!_channel.Writer.TryWrite(ev))
                {
                    EventsDropped.Add(1, _clientTag);
                }
            }

            // Falling out of the loop without an exception means the SERVER completed the
            // stream — a closed or evicted gateway session, not a transport failure. That is
            // just as fatal to this driver as a fault (the pump is now dead and no further
            // data-change events can ever arrive) but it raises nothing, so without this the
            // pump exits silently, the supervisor is never told, GetHealth() keeps reporting
            // Healthy, and the comms-loss watchdog never trips either — it requires a
            // non-Healthy state. The result is a driver that is permanently, invisibly deaf.
            if (!ct.IsCancellationRequested)
            {
                var cause = new InvalidOperationException(
                    "The gateway completed the StreamEvents stream without an error. The session was " +
                    "closed or evicted server-side; the event pump cannot receive further events.");
                _logger.LogWarning(
                    "Galaxy EventPump stream was completed by the gateway — notifying reconnect supervisor.");
                NotifyStreamFault(cause);
            }
        }
        catch (Exception) when (ct.IsCancellationRequested)
        {
            // Clean shutdown — no log. A cancelled gRPC stream may surface as RpcException(Cancelled)
            // rather than OperationCanceledException; either way the pump was told to stop, so this
            // is not a transport fault and must not wake the reconnect supervisor mid-teardown.
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex,
                "Galaxy EventPump loop ended with an exception — notifying reconnect supervisor.");

            // The gw StreamEvents stream faulted. Signal the reconnect supervisor so it
            // drives reopen → replay. Without this the stream silently dies and a
            // transient gateway drop permanently stops data-change notifications.
            NotifyStreamFault(ex);
        }
        finally
        {
            // Tell the dispatch loop the producer is done so it drains and exits.
            _channel.Writer.TryComplete();
        }
    }

    /// <summary>
    ///     Logs the first occurrence of each unrepresentable value kind. The dispatch loop is
    ///     single-reader, so the set needs no lock — but it is only ever touched from there.
    /// </summary>
    private void WarnUndecodableOnce(MxEvent ev)
    {
        var kind = ev.Value?.KindCase ?? MxValue.KindOneofCase.None;
        if (!_undecodableKindsWarned.Add(kind)) return;

        _logger.LogWarning(
            "Galaxy event for item handle {ItemHandle} carried a value kind this driver cannot " +
            "represent ({KindCase}) — publishing BadTypeMismatch. This means the gateway is sending " +
            "a wire type MxValueDecoder has no mapping for; readings of this kind are being lost. " +
            "Further occurrences of {KindCase} are counted on galaxy.events.undecodable, not logged.",
            ev.ItemHandle, kind, kind);
    }

    private void NotifyStreamFault(Exception cause)
    {
        if (_onStreamFault is null) return;
        try { _onStreamFault(cause); }
        catch (Exception cbEx)
        {
            _logger.LogWarning(cbEx,
                "Galaxy EventPump stream-fault callback threw — supervisor may not have been notified.");
        }
    }

    /// <summary>
    ///     Handles the gateway's replay-gap sentinel: events between the sequence we asked to
    ///     resume after and the oldest the gateway still retains were evicted and can never be
    ///     delivered. The contract is explicit that a client "MUST treat a present replay_gap as
    ///     'you missed events — discard local state and re-snapshot'".
    /// </summary>
    /// <remarks>
    ///     ⚠️ <b>This is unreachable in the current configuration and that is deliberate.</b>
    ///     The gateway emits the sentinel only on a stream opened with
    ///     <c>after_worker_sequence &gt; 0</c>; <c>GatewayGalaxySubscriber</c> passes 0, which the
    ///     gateway treats as "fresh stream, not a resume: plain attach, no replay, no sentinel".
    ///     It is implemented anyway because the alternative — the sentinel falling through the
    ///     family switch into the default drop, which is what happened before, since a sentinel
    ///     carries <c>family = UNSPECIFIED</c> — means silently ignoring a documented MUST if
    ///     resume is ever switched on. Enabling resume is a larger change than it looks: the
    ///     sequence is only meaningful within one gateway session, and our reconnect path
    ///     recreates the session, so a remembered sequence replayed against a fresh session
    ///     would set the server's live watermark above every event it is about to send.
    /// </remarks>
    private void HandleReplayGap(ReplayGap gap)
    {
        ReplayGapsObserved.Add(1, _clientTag);
        _logger.LogWarning(
            "Galaxy event stream reported a replay gap: events after worker_sequence {RequestedAfter} " +
            "up to {OldestAvailable} were evicted from the gateway's replay ring and cannot be " +
            "recovered. Forcing a re-snapshot of every active subscription.",
            gap.RequestedAfterSequence, gap.OldestAvailableSequence);

        if (_onReplayGap is null) return;
        try { _onReplayGap(gap); }
        catch (Exception ex)
        {
            _logger.LogWarning(ex,
                "Galaxy EventPump replay-gap callback threw — the re-snapshot may not have run.");
        }
    }

    private async Task DispatchLoopAsync(CancellationToken ct)
    {
        try
        {
            await foreach (var ev in _channel.Reader.ReadAllAsync(ct).ConfigureAwait(false))
            {
                Dispatch(ev);
                EventsDispatched.Add(1, _clientTag);
            }
        }
        catch (OperationCanceledException) when (ct.IsCancellationRequested)
        {
            // Clean shutdown.
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex,
                "Galaxy EventPump dispatch loop ended with an exception — events past this point will be lost until restart.");
        }
    }

    private void Dispatch(MxEvent ev)
    {
        switch (ev.Family)
        {
            case MxEventFamily.OnDataChange:
                DispatchDataChange(ev);
                break;
            default:
                // OnAlarmTransition is no longer carried on the per-session event stream
                // — alarms come from the gateway's session-less StreamAlarms feed
                // (GatewayGalaxyAlarmFeed). OnWriteComplete / OperationComplete /
                // OnBufferedDataChange are filtered out: write callers get their reply
                // via the InvokeAsync round-trip, not via the event stream.
                return;
        }
    }

    private void DispatchDataChange(MxEvent ev)
    {
        var subscribers = _registry.ResolveSubscribers(ev.ItemHandle);
        if (subscribers.Count == 0) return; // stale event after unsubscribe — drop quietly

        var snapshot = ToSnapshot(ev);
        foreach (var (subscriptionId, fullReference) in subscribers)
        {
            var handle = _handleFactory(subscriptionId);
            try
            {
                OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, fullReference, snapshot));
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex,
                    "Galaxy OnDataChange handler threw for {FullRef} subscription {SubscriptionId} — continuing fan-out.",
                    fullReference, subscriptionId);
            }
        }
    }

    private DataValueSnapshot ToSnapshot(MxEvent ev)
    {
        if (!MxValueDecoder.TryDecode(ev.Value, out var value))
        {
            // A value we cannot represent is published as Bad, NOT as a null carrying the
            // event's own (typically Good) quality. A good null is indistinguishable from a
            // real null reading, so a client would treat a value it never received as a
            // genuine one — the failure would be invisible on the wire and survive only in a
            // log nobody is reading. BadTypeMismatch says what actually happened: the payload
            // arrived, and this driver has no CLR type to carry it in.
            ValuesUndecodable.Add(1, _clientTag);
            WarnUndecodableOnce(ev);
            return new DataValueSnapshot(
                Value: null,
                StatusCode: StatusCodeMap.BadTypeMismatch,
                SourceTimestampUtc: ev.SourceTimestamp is { } undecodableTs ? undecodableTs.ToDateTime() : null,
                ServerTimestampUtc: DateTime.UtcNow);
        }

        var statusCode = ev.Statuses.Count > 0
            ? StatusCodeMap.FromMxStatus(ev.Statuses[0], _logger)
            : StatusCodeMap.FromQualityByte((byte)(ev.Quality & 0xFF), _logger);

        DateTime? sourceTimestamp = ev.SourceTimestamp is { } ts ? ts.ToDateTime() : null;
        return new DataValueSnapshot(
            Value: value,
            StatusCode: statusCode,
            SourceTimestampUtc: sourceTimestamp,
            ServerTimestampUtc: DateTime.UtcNow);
    }

    /// <summary>Disposes the event pump and cancels all running tasks.</summary>
    /// <returns>A value task representing the asynchronous disposal operation.</returns>
    public async ValueTask DisposeAsync()
    {
        if (_disposed) return;
        _disposed = true;
        _cts.Cancel();
        _channel.Writer.TryComplete();
        if (_loop is not null)
        {
            try { await _loop.ConfigureAwait(false); } catch { /* shutdown */ }
        }
        if (_dispatchLoop is not null)
        {
            try { await _dispatchLoop.ConfigureAwait(false); } catch { /* shutdown */ }
        }
        _cts.Dispose();
    }
}