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

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

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

/// <summary>
///     Production <see cref="IGalaxyAlarmFeed"/> over the gateway's session-less
///     <c>StreamAlarms</c> RPC. The stream opens with one <see cref="ActiveAlarmSnapshot"/>
///     per currently-active alarm (the ConditionRefresh snapshot), then a
///     <c>snapshot_complete</c> sentinel, then a live <see cref="OnAlarmTransitionEvent"/>
///     for every subsequent raise / acknowledge / clear. Each message is decoded into a
///     <see cref="GalaxyAlarmTransition"/> (severity already bucketed via
///     <see cref="MxAccessSeverityMapper"/>) and surfaced on <see cref="OnAlarmTransition"/>.
/// </summary>
/// <remarks>
///     <para>
///     The feed is independent of any worker session — the gateway's always-on central
///     alarm monitor owns the AVEVA subscription. The driver previously decoded alarm
///     transitions off the per-session <c>StreamEvents</c> stream (<see cref="EventPump"/>);
///     that path was retired when the gateway moved to the session-less alarm model.
///     </para>
///     <para>
///     The stream is supplied as a factory delegate (production passes
///     <c>MxGatewayClient.StreamAlarmsAsync</c>) so tests can drive synthetic feeds.
///     Streaming RPCs are not covered by the client's unary retry pipeline, so the feed
///     owns its reconnect: on any non-cancellation stream fault it logs, waits
///     <c>reconnectDelay</c>, and re-opens. The gateway re-sends the active-alarm
///     snapshot on every re-open, so the OPC UA condition layer sees current state
///     after a reconnect.
///     </para>
/// </remarks>
internal sealed class GatewayGalaxyAlarmFeed : IGalaxyAlarmFeed
{
    /// <summary>
    ///     Opens a <c>StreamAlarms</c> feed. Matches the method group
    ///     <c>MxGatewayClient.StreamAlarmsAsync</c>.
    /// </summary>
    /// <param name="request">The stream request parameters.</param>
    /// <param name="cancellationToken">A cancellation token.</param>
    internal delegate IAsyncEnumerable<AlarmFeedMessage> AlarmStreamFactory(
        StreamAlarmsRequest request, CancellationToken cancellationToken);

    private static readonly TimeSpan DefaultReconnectDelay = TimeSpan.FromSeconds(5);

    // Shares the driver meter name so a host-level MeterListener catches feed counters
    // alongside the EventPump's. Distinct Meter instance — same name is intentional.
    private static readonly Meter Meter = new(EventPump.MeterName);
    private static readonly Counter<long> AlarmTransitionsReceived =
        Meter.CreateCounter<long>("galaxy.alarm_feed.transitions.received", unit: "{event}",
            description: "Alarm feed messages decoded and forwarded to driver-level handlers.");
    private static readonly Counter<long> AlarmTransitionsDecodingFailures =
        Meter.CreateCounter<long>("galaxy.alarm_feed.transitions.decoding_failures", unit: "{event}",
            description: "Alarm feed messages dropped for a missing body or unspecified transition kind.");
    private static readonly Counter<long> AlarmFeedReconnects =
        Meter.CreateCounter<long>("galaxy.alarm_feed.reconnects", unit: "{reconnect}",
            description: "Times the alarm feed re-opened its StreamAlarms stream after a transport fault.");

    private readonly AlarmStreamFactory _streamFactory;
    private readonly ILogger _logger;
    private readonly string _alarmFilterPrefix;
    private readonly TimeSpan _reconnectDelay;
    private readonly KeyValuePair<string, object?> _clientTag;
    private readonly CancellationTokenSource _cts = new();

    private Task? _loop;
    private bool _disposed;

    /// <summary>Occurs when an alarm transition (raise, acknowledge, clear) is received from the Galaxy feed.</summary>
    public event EventHandler<GalaxyAlarmTransition>? OnAlarmTransition;

    /// <summary>Initializes a new instance of the <see cref="GatewayGalaxyAlarmFeed"/> class.</summary>
    /// <param name="streamFactory">A factory delegate that opens the alarm stream.</param>
    /// <param name="logger">An optional logger for diagnostic output.</param>
    /// <param name="clientName">An optional client name for tagging log entries.</param>
    /// <param name="alarmFilterPrefix">An optional prefix to filter alarms in the stream.</param>
    /// <param name="reconnectDelay">An optional delay before reconnecting after a stream fault.</param>
    public GatewayGalaxyAlarmFeed(
        AlarmStreamFactory streamFactory,
        ILogger? logger = null,
        string? clientName = null,
        string? alarmFilterPrefix = null,
        TimeSpan? reconnectDelay = null)
    {
        _streamFactory = streamFactory ?? throw new ArgumentNullException(nameof(streamFactory));
        _logger = logger ?? NullLogger.Instance;
        _alarmFilterPrefix = alarmFilterPrefix ?? string.Empty;
        _reconnectDelay = reconnectDelay ?? DefaultReconnectDelay;
        _clientTag = new KeyValuePair<string, object?>("galaxy.client", clientName ?? "<unknown>");
    }

    /// <summary>Starts the alarm feed by opening the stream and processing messages in a background task.</summary>
    public void Start()
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
        if (_loop is not null) return;
        _loop = Task.Run(() => RunAsync(_cts.Token));
    }

    private async Task RunAsync(CancellationToken ct)
    {
        var firstAttempt = true;
        while (!ct.IsCancellationRequested)
        {
            if (!firstAttempt)
            {
                AlarmFeedReconnects.Add(1, _clientTag);
            }
            firstAttempt = false;

            try
            {
                var request = new StreamAlarmsRequest
                {
                    ClientCorrelationId = Guid.NewGuid().ToString("N"),
                    AlarmFilterPrefix = _alarmFilterPrefix,
                };

                await foreach (var message in _streamFactory(request, ct)
                    .WithCancellation(ct).ConfigureAwait(false))
                {
                    if (ct.IsCancellationRequested) break;
                    Dispatch(message);
                }
            }
            catch (OperationCanceledException) when (ct.IsCancellationRequested)
            {
                return; // clean shutdown
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex,
                    "Galaxy alarm feed stream faulted — reopening in {DelaySeconds}s.",
                    _reconnectDelay.TotalSeconds);
            }

            try
            {
                await Task.Delay(_reconnectDelay, ct).ConfigureAwait(false);
            }
            catch (OperationCanceledException) when (ct.IsCancellationRequested)
            {
                return;
            }
        }
    }

    private void Dispatch(AlarmFeedMessage message)
    {
        switch (message.PayloadCase)
        {
            case AlarmFeedMessage.PayloadOneofCase.ActiveAlarm:
                DispatchSnapshotEntry(message.ActiveAlarm);
                break;
            case AlarmFeedMessage.PayloadOneofCase.Transition:
                DispatchTransition(message.Transition);
                break;
            case AlarmFeedMessage.PayloadOneofCase.SnapshotComplete:
                _logger.LogDebug("Galaxy alarm feed active-alarm snapshot complete.");
                break;
            default:
                // Empty oneof — worker / gateway version skew. Count and drop.
                AlarmTransitionsDecodingFailures.Add(1, _clientTag);
                break;
        }
    }

    /// <summary>
    ///     Decode one entry of the initial active-alarm snapshot. Each currently-active
    ///     alarm is surfaced as a transition so the OPC UA Part 9 condition layer sees
    ///     the alarm's present state on (re)connect: an unacknowledged active alarm as
    ///     a <see cref="GalaxyAlarmTransitionKind.Raise"/>, an acknowledged one as a
    ///     <see cref="GalaxyAlarmTransitionKind.Acknowledge"/>.
    /// </summary>
    private void DispatchSnapshotEntry(ActiveAlarmSnapshot snapshot)
    {
        var kind = snapshot.CurrentState switch
        {
            AlarmConditionState.Active => GalaxyAlarmTransitionKind.Raise,
            AlarmConditionState.ActiveAcked => GalaxyAlarmTransitionKind.Acknowledge,
            AlarmConditionState.Inactive => GalaxyAlarmTransitionKind.Clear,
            _ => GalaxyAlarmTransitionKind.Unspecified,
        };
        if (kind == GalaxyAlarmTransitionKind.Unspecified)
        {
            AlarmTransitionsDecodingFailures.Add(1, _clientTag);
            _logger.LogDebug(
                "Galaxy alarm feed snapshot entry for {AlarmRef} has unspecified condition state; ignoring.",
                snapshot.AlarmFullReference);
            return;
        }

        var (bucket, opcUaSeverity) = MxAccessSeverityMapper.Map(snapshot.Severity);
        Raise(new GalaxyAlarmTransition(
            AlarmFullReference: snapshot.AlarmFullReference,
            SourceObjectReference: snapshot.SourceObjectReference,
            AlarmTypeName: snapshot.AlarmTypeName,
            TransitionKind: kind,
            SeverityBucket: bucket,
            OpcUaSeverity: opcUaSeverity,
            RawMxAccessSeverity: snapshot.Severity,
            OriginalRaiseTimestampUtc: snapshot.OriginalRaiseTimestamp?.ToDateTime(),
            TransitionTimestampUtc: snapshot.LastTransitionTimestamp?.ToDateTime() ?? DateTime.UtcNow,
            OperatorUser: snapshot.OperatorUser,
            OperatorComment: snapshot.OperatorComment,
            Category: snapshot.Category,
            Description: snapshot.Description));
    }

    private void DispatchTransition(OnAlarmTransitionEvent body)
    {
        if (body.TransitionKind == AlarmTransitionKind.Unspecified)
        {
            AlarmTransitionsDecodingFailures.Add(1, _clientTag);
            _logger.LogDebug(
                "Galaxy alarm feed transition for {AlarmRef} has unspecified transition kind; ignoring.",
                body.AlarmFullReference);
            return;
        }

        var (bucket, opcUaSeverity) = MxAccessSeverityMapper.Map(body.Severity);
        Raise(new GalaxyAlarmTransition(
            AlarmFullReference: body.AlarmFullReference,
            SourceObjectReference: body.SourceObjectReference,
            AlarmTypeName: body.AlarmTypeName,
            TransitionKind: MapTransitionKind(body.TransitionKind),
            SeverityBucket: bucket,
            OpcUaSeverity: opcUaSeverity,
            RawMxAccessSeverity: body.Severity,
            OriginalRaiseTimestampUtc: body.OriginalRaiseTimestamp?.ToDateTime(),
            TransitionTimestampUtc: body.TransitionTimestamp?.ToDateTime() ?? DateTime.UtcNow,
            OperatorUser: body.OperatorUser,
            OperatorComment: body.OperatorComment,
            Category: body.Category,
            Description: body.Description));
    }

    private void Raise(GalaxyAlarmTransition transition)
    {
        AlarmTransitionsReceived.Add(1, _clientTag);
        try
        {
            OnAlarmTransition?.Invoke(this, transition);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex,
                "Galaxy alarm feed OnAlarmTransition handler threw for {AlarmRef} — continuing.",
                transition.AlarmFullReference);
        }
    }

    private static GalaxyAlarmTransitionKind MapTransitionKind(AlarmTransitionKind kind) => kind switch
    {
        AlarmTransitionKind.Raise => GalaxyAlarmTransitionKind.Raise,
        AlarmTransitionKind.Acknowledge => GalaxyAlarmTransitionKind.Acknowledge,
        AlarmTransitionKind.Clear => GalaxyAlarmTransitionKind.Clear,
        AlarmTransitionKind.Retrigger => GalaxyAlarmTransitionKind.Retrigger,
        _ => GalaxyAlarmTransitionKind.Unspecified,
    };

    /// <summary>Releases the alarm feed resources and stops the background stream task.</summary>
    public async ValueTask DisposeAsync()
    {
        if (_disposed) return;
        _disposed = true;
        _cts.Cancel();
        if (_loop is not null)
        {
            try { await _loop.ConfigureAwait(false); } catch { /* shutdown */ }
        }
        _cts.Dispose();
    }
}