lmxopcua/src/Client/ZB.MOM.WW.OtOpcUa.Client.Shared/OpcUaClientService.cs

using System.Text;
using Opc.Ua;
using Serilog;
using ZB.MOM.WW.OtOpcUa.Client.Shared.Adapters;
using ZB.MOM.WW.OtOpcUa.Client.Shared.Helpers;
using ZB.MOM.WW.OtOpcUa.Client.Shared.Models;
using BrowseResult = ZB.MOM.WW.OtOpcUa.Client.Shared.Models.BrowseResult;

namespace ZB.MOM.WW.OtOpcUa.Client.Shared;

/// <summary>
///     Full implementation of <see cref="IOpcUaClientService" /> using adapter abstractions for testability.
/// </summary>
public sealed class OpcUaClientService : IOpcUaClientService
{
    private static readonly ILogger Logger = Log.ForContext<OpcUaClientService>();

    // Guards all access to the subscription state below (_activeDataSubscriptions, _activeAlarmSubscription,
    // _dataSubscription, _alarmSubscription). They are touched from the caller thread, the keep-alive failover path
    // and DisconnectAsync, so every read/write is inside this lock — including failover's null-outs.
    private readonly object _subscriptionLock = new();

    // Serialises every subscription operation that awaits I/O (Subscribe/Unsubscribe, SubscribeAlarms/
    // UnsubscribeAlarms) AND the failover's teardown + reconnect + replay. A subscribe is therefore either
    // wholly before a failover (recorded, then replayed onto the new session) or wholly after it (on the new
    // session) — never half on the dying session. Async-friendly (no thread block). Deliberately never disposed:
    // async-only waits hold no kernel handle, and a failover still running at Dispose releases it afterwards.
    private readonly SemaphoreSlim _subscriptionGate = new(1, 1);

    // Track active data subscriptions for replay after failover
    private readonly Dictionary<string, (NodeId NodeId, int IntervalMs, uint Handle)> _activeDataSubscriptions = new();

    // Re-entry guard for HandleKeepAliveFailureAsync. The OPC UA stack raises
    // KeepAlive repeatedly while a session is down; only one failover loop may
    // run at a time. 0 = idle, 1 = failover in progress (Interlocked-managed).
    private int _failoverInProgress;

    // The connection epoch (review I1): ConnectAsync and DisconnectAsync each start a new one (under
    // _subscriptionLock). A failover runs under the epoch it started in and abandons itself — disposing any session it
    // opened, making no state transition — the moment the epoch moves, so a user Disconnect + Connect during a
    // failover is never undone by it. A keep-alive from a previous connection's session cannot start one either.
    private long _connectionEpoch;

    /// <summary>The most recent keep-alive-triggered failover task (test seam: lets a test await it).</summary>
    internal Task? LastFailover { get; private set; }

    private readonly IApplicationConfigurationFactory _configFactory;
    private readonly IEndpointDiscovery _endpointDiscovery;

    private readonly ISessionFactory _sessionFactory;

    // Track alarm subscription state for replay after failover
    private (NodeId? SourceNodeId, int IntervalMs)? _activeAlarmSubscription;
    private ISubscriptionAdapter? _alarmSubscription;
    private string[]? _allEndpointUrls;
    private int _currentEndpointIndex;
    private ISubscriptionAdapter? _dataSubscription;
    private bool _disposed;

    private ISessionAdapter? _session;
    private ConnectionSettings? _settings;
    private ConnectionState _state = ConnectionState.Disconnected;

    /// <summary>
    ///     Creates a new OpcUaClientService with the specified adapter dependencies.
    /// </summary>
    /// <param name="configFactory">Builds the application configuration and certificate settings for the client session.</param>
    /// <param name="endpointDiscovery">Selects the best matching endpoint for the requested URL and security mode.</param>
    /// <param name="sessionFactory">Creates the underlying OPC UA session used for runtime operations.</param>
    internal OpcUaClientService(
        IApplicationConfigurationFactory configFactory,
        IEndpointDiscovery endpointDiscovery,
        ISessionFactory sessionFactory)
    {
        _configFactory = configFactory;
        _endpointDiscovery = endpointDiscovery;
        _sessionFactory = sessionFactory;
    }

    /// <summary>
    ///     Creates a new OpcUaClientService with default production adapters.
    /// </summary>
    public OpcUaClientService()
        : this(
            new DefaultApplicationConfigurationFactory(),
            new DefaultEndpointDiscovery(),
            new DefaultSessionFactory())
    {
    }

    /// <inheritdoc />
    public event EventHandler<DataChangedEventArgs>? DataChanged;

    /// <inheritdoc />
    public event EventHandler<AlarmEventArgs>? AlarmEvent;

    /// <inheritdoc />
    public event EventHandler<ConnectionStateChangedEventArgs>? ConnectionStateChanged;

    /// <inheritdoc />
    public bool IsConnected => _state == ConnectionState.Connected && _session?.Connected == true;

    /// <inheritdoc />
    public ConnectionInfo? CurrentConnectionInfo { get; private set; }

    /// <inheritdoc />
    public async Task<ConnectionInfo> ConnectAsync(ConnectionSettings settings, CancellationToken ct = default)
    {
        ThrowIfDisposed();
        settings.Validate();

        // Overwriting a live (or reconnecting) session would leak it and orphan its subscriptions and keep-alive
        // handler. The caller must disconnect first.
        if (_state != ConnectionState.Disconnected || _session != null)
            throw new InvalidOperationException(
                $"Already {_state.ToString().ToLowerInvariant()}; call DisconnectAsync before connecting again.");

        long epoch;
        lock (_subscriptionLock) epoch = ++_connectionEpoch;

        _settings = settings;
        _allEndpointUrls = FailoverUrlParser.Parse(settings.EndpointUrl, settings.FailoverUrls);
        _currentEndpointIndex = 0;

        TransitionState(ConnectionState.Connecting, settings.EndpointUrl);

        try
        {
            var session = await ConnectToEndpointAsync(settings, _allEndpointUrls[0], ct);
            lock (_subscriptionLock) _session = session;

            session.RegisterKeepAliveHandler(isGood =>
            {
                if (!isGood) OnKeepAliveFailed(epoch);
            });

            CurrentConnectionInfo = BuildConnectionInfo(session);
            TransitionState(ConnectionState.Connected, session.EndpointUrl);
            Logger.Information("Connected to {EndpointUrl}", session.EndpointUrl);
            return CurrentConnectionInfo;
        }
        catch
        {
            TransitionState(ConnectionState.Disconnected, settings.EndpointUrl);
            throw;
        }
    }

    /// <inheritdoc />
    public async Task DisconnectAsync(CancellationToken ct = default)
    {
        if (_state == ConnectionState.Disconnected)
            return;

        // A new epoch first: a failover still connecting abandons itself instead of resurrecting the connection.
        lock (_subscriptionLock) _connectionEpoch++;

        var endpointUrl = _session?.EndpointUrl ?? _settings?.EndpointUrl ?? string.Empty;

        // Snapshot the subscription adapter references under the lock so a concurrent
        // RunFailoverAsync that nulls _dataSubscription / _alarmSubscription
        // cannot invalidate the references between the null-check and the DeleteAsync call.
        ISubscriptionAdapter? dataSubscription;
        ISubscriptionAdapter? alarmSubscription;
        lock (_subscriptionLock)
        {
            dataSubscription = _dataSubscription;
            alarmSubscription = _alarmSubscription;
            _dataSubscription = null;
            _alarmSubscription = null;
        }

        try
        {
            if (dataSubscription != null)
                await dataSubscription.DeleteAsync(ct);

            if (alarmSubscription != null)
                await alarmSubscription.DeleteAsync(ct);

            if (_session != null)
            {
                await _session.CloseAsync(ct);
                _session.Dispose();
                _session = null;
            }
        }
        catch (Exception ex)
        {
            Logger.Warning(ex, "Error during disconnect");
        }
        finally
        {
            lock (_subscriptionLock)
            {
                _activeDataSubscriptions.Clear();
                _activeAlarmSubscription = null;
            }

            CurrentConnectionInfo = null;
            TransitionState(ConnectionState.Disconnected, endpointUrl);
        }
    }

    /// <inheritdoc />
    public async Task<DataValue> ReadValueAsync(NodeId nodeId, CancellationToken ct = default)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();
        return await _session!.ReadValueAsync(nodeId, ct);
    }

    /// <inheritdoc />
    public async Task<StatusCode> WriteValueAsync(NodeId nodeId, object value, CancellationToken ct = default)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();

        // Read current value for type coercion when value is a string
        var typedValue = value;
        if (value is string rawString)
        {
            var currentDataValue = await _session!.ReadValueAsync(nodeId, ct);
            if (!StatusCode.IsGood(currentDataValue.StatusCode) || ValueConverter.ToClrObject(currentDataValue.WrappedValue) == null)
                throw new InvalidOperationException(
                    $"Cannot infer target type for node {nodeId}: read returned status {currentDataValue.StatusCode} with no value. Provide a typed value instead of a string.");
            typedValue = ValueConverter.ConvertValue(rawString, ValueConverter.ToClrObject(currentDataValue.WrappedValue));
        }

        var dataValue = new DataValue(ValueConverter.ToVariant(typedValue));
        return await _session!.WriteValueAsync(nodeId, dataValue, ct);
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<BrowseResult>> BrowseAsync(NodeId? parentNodeId = null,
        CancellationToken ct = default)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();

        var startNode = parentNodeId ?? ObjectIds.ObjectsFolder;
        var nodeClassMask = (uint)NodeClass.Object | (uint)NodeClass.Variable | (uint)NodeClass.Method;
        var results = new List<BrowseResult>();

        // A cursor belongs to the session that issued it, including while failover swaps _session.
        var session = _session!;
        byte[]? continuationPoint = null;
        try
        {
            ArrayOf<ReferenceDescription> references;
            (continuationPoint, references) = await session.BrowseAsync(startNode, nodeClassMask, ct);
            while (references.Count > 0)
            {
                foreach (var reference in references)
                    results.Add(new BrowseResult(
                        reference.NodeId.ToString(),
                        reference.DisplayName.Text ?? string.Empty,
                        reference.NodeClass.ToString(),
                        reference.NodeClass == NodeClass.Object));

                if (continuationPoint is not { Length: > 0 } pending) break;
                ct.ThrowIfCancellationRequested();
                (continuationPoint, references) = await session.BrowseNextAsync(pending, ct);
            }
        }
        finally
        {
            if (continuationPoint is { Length: > 0 })
            {
                try
                {
                    using var release = new CancellationTokenSource(HistoryReadPaging.ReleaseTimeout);
                    await session.ReleaseBrowseContinuationPointAsync(continuationPoint, release.Token)
                        .WaitAsync(release.Token).ConfigureAwait(false);
                }
                catch (Exception ex) { Logger.Warning(ex, "Releasing browse continuation point failed"); }
            }
        }

        return results;
    }

    /// <inheritdoc />
    public async Task SubscribeAsync(NodeId nodeId, int intervalMs = 1000, CancellationToken ct = default)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();

        var nodeIdStr = nodeId.ToString();
        await _subscriptionGate.WaitAsync(ct);
        try
        {
            // A failover may have run (and failed) while this call waited for the gate.
            ThrowIfNotConnected();

            ISubscriptionAdapter? subscription;
            ISessionAdapter session;
            lock (_subscriptionLock)
            {
                if (_activeDataSubscriptions.ContainsKey(nodeIdStr))
                    return; // Already subscribed
                subscription = _dataSubscription;
                session = _session!;
            }

            if (subscription == null)
            {
                subscription = await session.CreateSubscriptionAsync(intervalMs, ct);
                lock (_subscriptionLock) _dataSubscription = subscription;
            }

            var handle = await subscription.AddDataChangeMonitoredItemAsync(
                nodeId, intervalMs, DataCallback(session, subscription), ct);

            lock (_subscriptionLock)
            {
                _activeDataSubscriptions[nodeIdStr] = (nodeId, intervalMs, handle);
            }
        }
        finally
        {
            _subscriptionGate.Release();
        }

        Logger.Debug("Subscribed to data changes on {NodeId}", nodeId);
    }

    /// <inheritdoc />
    public async Task UnsubscribeAsync(NodeId nodeId, CancellationToken ct = default)
    {
        ThrowIfDisposed();

        var nodeIdStr = nodeId.ToString();
        await _subscriptionGate.WaitAsync(ct);
        try
        {
            (NodeId NodeId, int IntervalMs, uint Handle) sub;
            ISubscriptionAdapter? subscription;
            lock (_subscriptionLock)
            {
                if (!_activeDataSubscriptions.TryGetValue(nodeIdStr, out sub))
                    return; // Not subscribed, safe to ignore
                subscription = _dataSubscription;
            }

            try
            {
                if (subscription != null) await subscription.RemoveMonitoredItemAsync(sub.Handle, ct);
            }
            finally
            {
                // The caller asked to stop: drop the bookkeeping even when the server-side remove failed (a session
                // that died under us), or a later failover would replay the item onto the new session (review m5).
                lock (_subscriptionLock)
                {
                    _activeDataSubscriptions.Remove(nodeIdStr);
                }
            }
        }
        finally
        {
            _subscriptionGate.Release();
        }

        Logger.Debug("Unsubscribed from data changes on {NodeId}", nodeId);
    }

    /// <inheritdoc />
    public async Task SubscribeAlarmsAsync(NodeId? sourceNodeId = null, int intervalMs = 1000,
        CancellationToken ct = default)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();

        // Serialise check-and-create under the semaphore: concurrent
        // callers cannot both pass the null-check and create duplicate alarm subscriptions.
        // The semaphore is async-friendly; the subscription I/O runs inside the critical section.
        await _subscriptionGate.WaitAsync(ct);
        try
        {
            ThrowIfNotConnected();

            ISessionAdapter session;
            lock (_subscriptionLock)
            {
                if (_alarmSubscription != null)
                    return; // Already subscribed to alarms
                session = _session!;
            }

            var monitorNode = sourceNodeId ?? ObjectIds.Server;
            var subscription = await session.CreateSubscriptionAsync(intervalMs, ct);
            lock (_subscriptionLock) _alarmSubscription = subscription;

            var filter = CreateAlarmEventFilter();
            await subscription.AddEventMonitoredItemAsync(
                monitorNode, intervalMs, filter, AlarmCallback(session, subscription), ct);

            lock (_subscriptionLock)
            {
                _activeAlarmSubscription = (sourceNodeId, intervalMs);
            }

            Logger.Debug("Subscribed to alarm events on {NodeId}", monitorNode);
        }
        finally
        {
            _subscriptionGate.Release();
        }
    }

    /// <inheritdoc />
    public async Task UnsubscribeAlarmsAsync(CancellationToken ct = default)
    {
        ThrowIfDisposed();

        await _subscriptionGate.WaitAsync(ct);
        try
        {
            ISubscriptionAdapter? subscription;
            lock (_subscriptionLock)
            {
                subscription = _alarmSubscription;
                if (subscription == null) return;
            }

            await subscription.DeleteAsync(ct);

            lock (_subscriptionLock)
            {
                _alarmSubscription = null;
                _activeAlarmSubscription = null;
            }
        }
        finally
        {
            _subscriptionGate.Release();
        }

        Logger.Debug("Unsubscribed from alarm events");
    }

    /// <inheritdoc />
    public async Task RequestConditionRefreshAsync(CancellationToken ct = default)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();

        ISubscriptionAdapter? subscription;
        lock (_subscriptionLock) subscription = _alarmSubscription;
        if (subscription == null)
            throw new InvalidOperationException("No alarm subscription is active.");

        await subscription.ConditionRefreshAsync(ct);
        Logger.Debug("Condition refresh requested");
    }

    /// <inheritdoc />
    public async Task<StatusCode> AcknowledgeAlarmAsync(string conditionNodeId, byte[] eventId, string comment,
        CancellationToken ct = default)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();

        // OtOpcUa conditions are their own event source (SourceNode == ConditionId), so the Part 9 methods are
        // invoked on that node directly. There is no ".Condition" child to append.
        var conditionObjId = NodeId.Parse(conditionNodeId);
        var acknowledgeMethodId = MethodIds.AcknowledgeableConditionType_Acknowledge;

        // CallMethodAsync throws ServiceResultException on a bad call result;
        // surface that as the returned StatusCode so callers using the documented
        // `Task<StatusCode>` contract (e.g. `if (StatusCode.IsBad(result))`) see
        // the failure instead of an uncaught exception they did not anticipate.
        try
        {
            await _session!.CallMethodAsync(
                conditionObjId,
                acknowledgeMethodId,
                [eventId, new LocalizedText(comment)],
                ct);
        }
        catch (ServiceResultException ex)
        {
            Logger.Warning(ex, "Failed to acknowledge alarm on {ConditionId} (status {Status})",
                conditionNodeId, ex.StatusCode);
            return ex.StatusCode;
        }

        Logger.Debug("Acknowledged alarm on {ConditionId}", conditionNodeId);
        return StatusCodes.Good;
    }

    /// <inheritdoc />
    public async Task<StatusCode> ConfirmAlarmAsync(string conditionNodeId, byte[] eventId, string comment,
        CancellationToken ct = default)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();

        // Confirm is invoked on the condition node itself, like Acknowledge.
        var conditionObjId = NodeId.Parse(conditionNodeId);
        var confirmMethodId = MethodIds.AcknowledgeableConditionType_Confirm;

        try
        {
            await _session!.CallMethodAsync(
                conditionObjId,
                confirmMethodId,
                [eventId, new LocalizedText(comment)],
                ct);
        }
        catch (ServiceResultException ex)
        {
            Logger.Warning(ex, "Failed to confirm alarm on {ConditionId} (status {Status})",
                conditionNodeId, ex.StatusCode);
            return ex.StatusCode;
        }

        Logger.Debug("Confirmed alarm on {ConditionId}", conditionNodeId);
        return StatusCodes.Good;
    }

    /// <inheritdoc />
    public async Task<StatusCode> AddCommentAsync(string conditionNodeId, byte[] eventId, string comment,
        CancellationToken ct = default)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();

        // AddComment is a ConditionType method, invoked on the condition node itself.
        var conditionObjId = NodeId.Parse(conditionNodeId);

        try
        {
            await _session!.CallMethodAsync(
                conditionObjId,
                MethodIds.ConditionType_AddComment,
                [eventId, new LocalizedText(comment)],
                ct);
        }
        catch (ServiceResultException ex)
        {
            Logger.Warning(ex, "Failed to add comment on {ConditionId} (status {Status})",
                conditionNodeId, ex.StatusCode);
            return ex.StatusCode;
        }

        Logger.Debug("Added comment on {ConditionId}", conditionNodeId);
        return StatusCodes.Good;
    }

    /// <inheritdoc />
    public async Task<StatusCode> ShelveAlarmAsync(string conditionNodeId, ShelveKind kind,
        double shelvingTimeSeconds = 0, CancellationToken ct = default)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();

        // The server's ShelvingState child is SDK-created with a numeric id in namespace 0, so a synthesized
        // "<condition>.ShelvingState" node id never exists. The shelve methods are called on the condition
        // object itself with the AlarmConditionType_ShelvingState_* method ids, which the server dispatches
        // to the condition's ShelvingState methods.
        var shelvingStateNodeId = NodeId.Parse(conditionNodeId);

        NodeId methodId;
        object[] inputArgs;

        switch (kind)
        {
            case ShelveKind.OneShot:
                methodId = MethodIds.AlarmConditionType_ShelvingState_OneShotShelve;
                inputArgs = [];
                break;
            case ShelveKind.Timed:
                if (shelvingTimeSeconds <= 0)
                    throw new ArgumentOutOfRangeException(nameof(shelvingTimeSeconds),
                        "Timed shelving requires a positive shelvingTimeSeconds value.");
                methodId = MethodIds.AlarmConditionType_ShelvingState_TimedShelve;
                // OPC UA Part 9 TimedShelve ShelvingTime input is a Duration (Double) in milliseconds.
                inputArgs = [shelvingTimeSeconds * 1000.0];
                break;
            case ShelveKind.Unshelve:
                methodId = MethodIds.AlarmConditionType_ShelvingState_Unshelve;
                inputArgs = [];
                break;
            default:
                throw new ArgumentOutOfRangeException(nameof(kind), kind, "Unknown ShelveKind value.");
        }

        try
        {
            await _session!.CallMethodAsync(shelvingStateNodeId, methodId, inputArgs, ct);
        }
        catch (ServiceResultException ex)
        {
            Logger.Warning(ex, "Failed to shelve alarm on {ConditionId} kind={Kind} (status {Status})",
                conditionNodeId, kind, ex.StatusCode);
            return ex.StatusCode;
        }

        Logger.Debug("Shelved alarm on {ConditionId} kind={Kind}", conditionNodeId, kind);
        return StatusCodes.Good;
    }

    /// <inheritdoc />
    public Task<StatusCode> EnableAsync(string conditionNodeId, CancellationToken ct = default) =>
        CallEnableDisableAsync(conditionNodeId, enabling: true, ct);

    /// <inheritdoc />
    public Task<StatusCode> DisableAsync(string conditionNodeId, CancellationToken ct = default) =>
        CallEnableDisableAsync(conditionNodeId, enabling: false, ct);

    /// <summary>
    /// Shared body for the H4 Enable/Disable client path. Invokes the OPC UA Part 9 ConditionType
    /// <c>Enable</c>/<c>Disable</c> method (no input arguments) on the condition object node, mirroring
    /// the <see cref="AcknowledgeAlarmAsync"/> convention: the methods are invoked on the condition
    /// node itself, and a bad call result surfaces as the returned <see cref="StatusCode"/> rather than
    /// escaping as an uncaught <see cref="ServiceResultException"/>.
    /// </summary>
    /// <param name="conditionNodeId">The condition node associated with the alarm.</param>
    /// <param name="enabling"><c>true</c> to invoke Enable; <c>false</c> to invoke Disable.</param>
    /// <param name="ct">The cancellation token that aborts the request.</param>
    private async Task<StatusCode> CallEnableDisableAsync(string conditionNodeId, bool enabling, CancellationToken ct)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();

        // Enable/Disable are invoked on the condition node itself, like Acknowledge/Confirm.
        var conditionObjId = NodeId.Parse(conditionNodeId);
        var methodId = enabling ? MethodIds.ConditionType_Enable : MethodIds.ConditionType_Disable;

        try
        {
            await _session!.CallMethodAsync(conditionObjId, methodId, [], ct);
        }
        catch (ServiceResultException ex)
        {
            Logger.Warning(ex, "Failed to {Operation} alarm on {ConditionId} (status {Status})",
                enabling ? "enable" : "disable", conditionNodeId, ex.StatusCode);
            return ex.StatusCode;
        }

        Logger.Debug("{Operation} alarm on {ConditionId}", enabling ? "Enabled" : "Disabled", conditionNodeId);
        return StatusCodes.Good;
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<DataValue>> HistoryReadRawAsync(
        NodeId nodeId, DateTime startTime, DateTime endTime, int maxValues = 1000, CancellationToken ct = default)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();
        return await _session!.HistoryReadRawAsync(nodeId, startTime, endTime, maxValues, ct);
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<HistoryModifiedEntry>> HistoryReadModifiedAsync(
        NodeId nodeId, DateTime startTime, DateTime endTime, int maxValues = 1000, CancellationToken ct = default)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();
        return await _session!.HistoryReadModifiedAsync(nodeId, startTime, endTime, maxValues, ct);
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<HistoryEventEntry>> HistoryReadEventsAsync(
        NodeId nodeId, DateTime startTime, DateTime endTime, EventFilter filter, int maxEvents = 1000,
        CancellationToken ct = default)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();
        return await _session!.HistoryReadEventsAsync(nodeId, startTime, endTime, maxEvents, filter, ct);
    }

    /// <inheritdoc />
    public async Task<HistoryUpdateOutcome> HistoryUpdateAsync(HistoryUpdateDetails details, CancellationToken ct = default)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();
        return await _session!.HistoryUpdateAsync(details, ct);
    }

    /// <inheritdoc />
    public async Task<IReadOnlyList<DataValue>> HistoryReadAggregateAsync(
        NodeId nodeId, DateTime startTime, DateTime endTime, AggregateType aggregate,
        double intervalMs = 3600000, CancellationToken ct = default)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();
        var aggregateNodeId = AggregateTypeMapper.ToNodeId(aggregate);
        return await _session!.HistoryReadAggregateAsync(nodeId, startTime, endTime, aggregateNodeId, intervalMs, ct);
    }

    /// <inheritdoc />
    public async Task<RedundancyInfo> GetRedundancyInfoAsync(CancellationToken ct = default)
    {
        ThrowIfDisposed();
        ThrowIfNotConnected();

        var redundancySupportValue =
            await _session!.ReadValueAsync(VariableIds.Server_ServerRedundancy_RedundancySupport, ct);
        RedundancySupport redundancySupport;
        if (StatusCode.IsGood(redundancySupportValue.StatusCode) && ValueConverter.ToClrObject(redundancySupportValue.WrappedValue) != null)
            redundancySupport = (RedundancySupport)Convert.ToInt32(ValueConverter.ToClrObject(redundancySupportValue.WrappedValue));
        else
            redundancySupport = RedundancySupport.None;
        var redundancyMode = redundancySupport.ToString();

        var serviceLevelValue = await _session.ReadValueAsync(VariableIds.Server_ServiceLevel, ct);
        var serviceLevel = StatusCode.IsGood(serviceLevelValue.StatusCode) && ValueConverter.ToClrObject(serviceLevelValue.WrappedValue) != null
            ? Convert.ToByte(ValueConverter.ToClrObject(serviceLevelValue.WrappedValue))
            : (byte)0;

        string[] serverUris = [];
        try
        {
            var serverUriArrayValue =
                await _session.ReadValueAsync(VariableIds.Server_ServerRedundancy_ServerUriArray, ct);
            if (ValueConverter.ToClrObject(serverUriArrayValue.WrappedValue) is string[] uris)
                serverUris = uris;
        }
        catch
        {
            // ServerUriArray may not be present when RedundancySupport is None
        }

        var applicationUri = string.Empty;
        try
        {
            var serverArrayValue = await _session.ReadValueAsync(VariableIds.Server_ServerArray, ct);
            if (ValueConverter.ToClrObject(serverArrayValue.WrappedValue) is string[] serverArray && serverArray.Length > 0)
                applicationUri = serverArray[0];
        }
        catch
        {
            // Informational only
        }

        return new RedundancyInfo(redundancyMode, serviceLevel, serverUris, applicationUri);
    }

    /// <summary>
    /// Releases the current session and any active monitored-item subscriptions held by the client service.
    /// </summary>
    public void Dispose()
    {
        lock (_subscriptionLock)
        {
            if (_disposed) return;
            _disposed = true;
            _connectionEpoch++;   // a failover still running abandons itself
        }

        _dataSubscription?.Dispose();
        _alarmSubscription?.Dispose();
        _session?.Dispose();

        lock (_subscriptionLock)
        {
            _activeDataSubscriptions.Clear();
            _activeAlarmSubscription = null;
        }

        CurrentConnectionInfo = null;
        _state = ConnectionState.Disconnected;
    }

    // --- Private helpers ---

    private async Task<ISessionAdapter> ConnectToEndpointAsync(ConnectionSettings settings, string endpointUrl,
        CancellationToken ct)
    {
        // Create a settings copy with the current endpoint URL
        var effectiveSettings = new ConnectionSettings
        {
            EndpointUrl = endpointUrl,
            SecurityMode = settings.SecurityMode,
            SessionTimeoutSeconds = settings.SessionTimeoutSeconds,
            AutoAcceptCertificates = settings.AutoAcceptCertificates,
            ValidateServerHostName = settings.ValidateServerHostName,
            CertificateStorePath = settings.CertificateStorePath,
            Username = settings.Username,
            Password = settings.Password
        };

        var config = await _configFactory.CreateAsync(effectiveSettings, ct);
        try
        {
            var requestedMode = SecurityModeMapper.ToMessageSecurityMode(settings.SecurityMode);
            var endpoint = await _endpointDiscovery.SelectEndpointAsync(config, endpointUrl, requestedMode, ct);

            var identity = settings.Username != null
                ? new UserIdentity(settings.Username, Encoding.UTF8.GetBytes(settings.Password ?? ""))
                : new UserIdentity();

            var sessionTimeoutMs = (uint)(settings.SessionTimeoutSeconds * 1000);
            // #816: never on a None endpoint — SDK 2.0.1 also checks the host at channel creation against the
            // GetEndpoints certificate, before any security-mode test, and on an unauthenticated channel that proves
            // nothing (same rule as OpcUaEndpointSelection.CheckDomainFor for the driver and the AdminUI browser).
            var checkDomain = settings.ValidateServerHostName && endpoint.SecurityMode != MessageSecurityMode.None;
            return await _sessionFactory.CreateSessionAsync(config, endpoint, "OtOpcUaClient", sessionTimeoutMs, identity,
                checkDomain, ct);
        }
        catch
        {
            // The session adapter owns it after successful creation; a failed attempt has no adapter owner.
            try { (config.CertificateManager as IDisposable)?.Dispose(); }
            catch (Exception ex) { Logger.Warning(ex, "Disposing certificate manager after connection failure"); }
            throw;
        }
    }

    /// <summary>A bad keep-alive from the session of connection <paramref name="epoch"/>.</summary>
    private void OnKeepAliveFailed(long epoch)
    {
        // A late keep-alive from a previous connection's session is not about the current one.
        if (!IsCurrentEpoch(epoch)) return;

        // Serialize failover: the OPC UA stack raises KeepAlive repeatedly while a session is down, so several bad
        // keep-alives can fire before the first failover loop finishes. CompareExchange claims the failover slot; a
        // re-entrant call sees 1 and returns, so exactly one loop runs — and only the winner sets the test seam
        // (review m8: a re-entrant call used to overwrite it with an already-completed no-op).
        if (Interlocked.CompareExchange(ref _failoverInProgress, 1, 0) != 0)
            return;

        LastFailover = HandleKeepAliveFailureAsync(epoch);
    }

    private async Task HandleKeepAliveFailureAsync(long epoch)
    {
        try
        {
            await RunFailoverAsync(epoch);
        }
        finally
        {
            Interlocked.Exchange(ref _failoverInProgress, 0);
        }
    }

    private bool IsCurrentEpoch(long epoch)
    {
        lock (_subscriptionLock) return _connectionEpoch == epoch && !_disposed;
    }

    /// <summary>
    ///     A state transition made only while <paramref name="epoch"/> is still current (checked and applied under
    ///     <see cref="_subscriptionLock"/>, the event raised outside it).
    /// </summary>
    /// <returns><see langword="false"/> when the epoch has moved on (the caller must abandon).</returns>
    private bool TransitionStateIfCurrent(long epoch, ConnectionState newState, string endpointUrl)
    {
        ConnectionState oldState;
        lock (_subscriptionLock)
        {
            if (_connectionEpoch != epoch || _disposed) return false;
            oldState = _state;
            if (oldState == newState) return true;
            _state = newState;
        }

        ConnectionStateChanged?.Invoke(this, new ConnectionStateChangedEventArgs(oldState, newState, endpointUrl));
        return true;
    }

    private async Task RunFailoverAsync(long epoch)
    {
        if (_state == ConnectionState.Reconnecting || _state == ConnectionState.Disconnected)
            return;

        var oldSession = _session;
        var oldEndpoint = oldSession?.EndpointUrl ?? string.Empty;
        if (!TransitionStateIfCurrent(epoch, ConnectionState.Reconnecting, oldEndpoint)) return;
        Logger.Warning("Session lost on {EndpointUrl}. Attempting failover...", oldEndpoint);

        // Dispose the dead session BEFORE taking the gate: an in-flight subscribe holding the gate is blocked on
        // that session, and disposing it is what makes the call fail fast and release the gate.
        if (oldSession != null)
        {
            try
            {
                oldSession.Dispose();
            }
            catch
            {
            }
        }

        // The rest runs under the subscription gate (and its null-outs under _subscriptionLock): no subscribe can
        // interleave with the teardown, the reconnect or the replay.
        await _subscriptionGate.WaitAsync();
        try
        {
            // The user's settings / endpoint list as of THIS epoch: a Connect during the failover replaces the fields.
            ConnectionSettings? settings;
            string[]? endpoints;
            lock (_subscriptionLock)
            {
                if (_connectionEpoch != epoch || _disposed) return;   // the user disconnected / reconnected meanwhile
                if (ReferenceEquals(_session, oldSession)) _session = null;
                _dataSubscription = null;
                _alarmSubscription = null;
                settings = _settings;
                endpoints = _allEndpointUrls;
            }

            if (settings == null || endpoints == null)
            {
                TransitionStateIfCurrent(epoch, ConnectionState.Disconnected, oldEndpoint);
                return;
            }

            // Try each endpoint
            for (var attempt = 0; attempt < endpoints.Length; attempt++)
            {
                if (!IsCurrentEpoch(epoch)) return;               // stop trying once the user has moved on (m5)
                _currentEndpointIndex = (_currentEndpointIndex + 1) % endpoints.Length;
                var url = endpoints[_currentEndpointIndex];

                try
                {
                    Logger.Information("Failover attempt to {EndpointUrl}", url);
                    var session = await ConnectToEndpointAsync(settings, url, CancellationToken.None);

                    // A Disconnect / Connect / Dispose that ran while this attempt was connecting wins: never
                    // resurrect the old connection or overwrite the user's new session. Checked and installed
                    // atomically against the epoch.
                    bool installed;
                    lock (_subscriptionLock)
                    {
                        installed = _connectionEpoch == epoch && !_disposed;
                        if (installed) _session = session;
                    }

                    if (!installed)
                    {
                        session.Dispose();
                        return;
                    }

                    session.RegisterKeepAliveHandler(isGood =>
                    {
                        if (!isGood) OnKeepAliveFailed(epoch);
                    });

                    CurrentConnectionInfo = BuildConnectionInfo(session);
                    if (!TransitionStateIfCurrent(epoch, ConnectionState.Connected, url)) return;
                    Logger.Information("Failover succeeded to {EndpointUrl}", url);

                    // Replay subscriptions
                    await ReplaySubscriptionsAsync(session);
                    return;
                }
                catch (Exception ex)
                {
                    Logger.Warning(ex, "Failover to {EndpointUrl} failed", url);
                }
            }

            if (TransitionStateIfCurrent(epoch, ConnectionState.Disconnected, oldEndpoint))
                Logger.Error("All failover endpoints unreachable");
        }
        finally
        {
            _subscriptionGate.Release();
        }
    }

    /// <summary>Re-creates the recorded subscriptions on <paramref name="session"/>. Runs under the subscription gate.</summary>
    private async Task ReplaySubscriptionsAsync(ISessionAdapter session)
    {
        // Snapshot the bookkeeping state under the lock, then clear it so the
        // replayed handles can be recorded fresh as each monitored item is
        // re-created. Awaited calls run outside the lock.
        List<KeyValuePair<string, (NodeId NodeId, int IntervalMs, uint Handle)>> subscriptions;
        (NodeId? SourceNodeId, int IntervalMs)? alarmSubscription;
        lock (_subscriptionLock)
        {
            subscriptions = _activeDataSubscriptions.ToList();
            alarmSubscription = _activeAlarmSubscription;
            _activeDataSubscriptions.Clear();
            _activeAlarmSubscription = null;
        }

        // Replay data subscriptions
        ISubscriptionAdapter? dataSubscription = null;
        foreach (var (nodeIdStr, (nodeId, intervalMs, _)) in subscriptions)
            try
            {
                if (dataSubscription == null)
                {
                    dataSubscription = await session.CreateSubscriptionAsync(intervalMs, CancellationToken.None);
                    lock (_subscriptionLock) _dataSubscription = dataSubscription;
                }

                var handle = await dataSubscription.AddDataChangeMonitoredItemAsync(
                    nodeId, intervalMs, DataCallback(session, dataSubscription), CancellationToken.None);

                lock (_subscriptionLock)
                {
                    _activeDataSubscriptions[nodeIdStr] = (nodeId, intervalMs, handle);
                }
            }
            catch (Exception ex)
            {
                Logger.Warning(ex, "Failed to replay data subscription for {NodeId}", nodeIdStr);
            }

        // Replay alarm subscription
        if (alarmSubscription.HasValue)
        {
            var (sourceNodeId, intervalMs) = alarmSubscription.Value;
            try
            {
                var monitorNode = sourceNodeId ?? ObjectIds.Server;
                var replayedAlarms = await session.CreateSubscriptionAsync(intervalMs, CancellationToken.None);
                lock (_subscriptionLock) _alarmSubscription = replayedAlarms;
                var filter = CreateAlarmEventFilter();
                await replayedAlarms.AddEventMonitoredItemAsync(
                    monitorNode, intervalMs, filter, AlarmCallback(session, replayedAlarms), CancellationToken.None);

                lock (_subscriptionLock)
                {
                    _activeAlarmSubscription = (sourceNodeId, intervalMs);
                }
            }
            catch (Exception ex)
            {
                Logger.Warning(ex, "Failed to replay alarm subscription");
            }
        }
    }

    private bool IsCurrentNotification(ISessionAdapter session, ISubscriptionAdapter subscription, bool alarm)
    {
        lock (_subscriptionLock)
            return !_disposed && _state == ConnectionState.Connected && ReferenceEquals(_session, session)
                && ReferenceEquals(alarm ? _alarmSubscription : _dataSubscription, subscription);
    }

    private Action<string, DataValue> DataCallback(ISessionAdapter session, ISubscriptionAdapter subscription) =>
        (nodeId, value) =>
        {
            if (IsCurrentNotification(session, subscription, false)) OnDataChangeNotification(nodeId, value);
        };

    private Action<EventFieldList> AlarmCallback(ISessionAdapter session, ISubscriptionAdapter subscription) =>
        fields =>
        {
            if (IsCurrentNotification(session, subscription, true)) OnAlarmEventNotification(fields);
        };

    private void OnDataChangeNotification(string nodeId, DataValue value)
    {
        DataChanged?.Invoke(this, new DataChangedEventArgs(nodeId, value));
    }

    private void OnAlarmEventNotification(EventFieldList eventFields)
    {
        var fields = eventFields.EventFields;
        if (fields.Count < 1)
            return;

        var eventId = fields.Count > 0 ? ValueConverter.ToClrObject(fields[0]) as byte[] : null;
        var sourceName = fields.Count > 2 ? ValueConverter.ToClrObject(fields[2]) as string ?? string.Empty : string.Empty;
        var time = DateTime.MinValue;
        if (fields.Count > 3 && ValueConverter.ToClrObject(fields[3]) != null)
        {
            if (ValueConverter.ToClrObject(fields[3]) is DateTime dt)
                time = dt;
            else if (DateTime.TryParse(ValueConverter.ToClrObject(fields[3])?.ToString(), out var parsed))
                time = parsed;
        }
        var message = fields.Count > 4 ? (fields[4].TryGetValue(out LocalizedText text) ? text.Text : null) ?? string.Empty : string.Empty;
        var severity = fields.Count > 5 ? Convert.ToUInt16(ValueConverter.ToClrObject(fields[5])) : (ushort)0;
        var conditionName = fields.Count > 6 ? ValueConverter.ToClrObject(fields[6]) as string ?? string.Empty : string.Empty;
        var retain = fields.Count > 7 && ParseBool(ValueConverter.ToClrObject(fields[7]));
        var conditionNodeId = fields.Count > 12 ? ValueConverter.ToClrObject(fields[12])?.ToString() : null;

        // Try standard OPC UA ActiveState/AckedState fields first
        bool? ackedField = fields.Count > 8 && ValueConverter.ToClrObject(fields[8]) != null ? ParseBool(ValueConverter.ToClrObject(fields[8])) : null;
        bool? activeField = fields.Count > 9 && ValueConverter.ToClrObject(fields[9]) != null ? ParseBool(ValueConverter.ToClrObject(fields[9])) : null;

        var ackedState = ackedField ?? false;
        var activeState = activeField ?? false;

        // No Galaxy-era supplemental reads (<condition>.InAlarm/.Acked/...): the server publishes the
        // standard Part 9 fields. A foreign server that leaves both null reports Active=false/Acked=false.

        AlarmEvent?.Invoke(this, new AlarmEventArgs(
            sourceName, conditionName, severity, message, retain, activeState, ackedState, time,
            eventId, conditionNodeId));
    }

    private static bool ParseBool(object? value)
    {
        if (value == null) return false;
        if (value is bool b) return b;
        try { return Convert.ToBoolean(value); }
        catch { return false; }
    }

    /// <summary>Builds the event filter (select clauses for the alarm and condition fields) used by alarm subscriptions.</summary>
    /// <returns>The populated <see cref="EventFilter"/>.</returns>
    internal static EventFilter CreateAlarmEventFilter()
    {
        var filter = new EventFilter
        {
            SelectClauses =
            [
                new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.BaseEventType, AttributeId = Attributes.Value, BrowsePath = [new QualifiedName("EventId")] },
                new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.BaseEventType, AttributeId = Attributes.Value, BrowsePath = [new QualifiedName("EventType")] },
                new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.BaseEventType, AttributeId = Attributes.Value, BrowsePath = [new QualifiedName("SourceName")] },
                new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.BaseEventType, AttributeId = Attributes.Value, BrowsePath = [new QualifiedName("Time")] },
                new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.BaseEventType, AttributeId = Attributes.Value, BrowsePath = [new QualifiedName("Message")] },
                new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.BaseEventType, AttributeId = Attributes.Value, BrowsePath = [new QualifiedName("Severity")] },
                new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.ConditionType, AttributeId = Attributes.Value, BrowsePath = [new QualifiedName("ConditionName")] },
                new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.ConditionType, AttributeId = Attributes.Value, BrowsePath = [new QualifiedName("Retain")] },
                new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.AcknowledgeableConditionType, AttributeId = Attributes.Value, BrowsePath = [new QualifiedName("AckedState"), new QualifiedName("Id")] },
                new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.AlarmConditionType, AttributeId = Attributes.Value, BrowsePath = [new QualifiedName("ActiveState"), new QualifiedName("Id")] },
                new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.ConditionType, AttributeId = Attributes.Value, BrowsePath = [new QualifiedName("EnabledState"), new QualifiedName("Id")] },
                new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.AlarmConditionType, AttributeId = Attributes.Value, BrowsePath = [new QualifiedName("SuppressedOrShelved")] },
                new SimpleAttributeOperand { TypeDefinitionId = ObjectTypeIds.BaseEventType, AttributeId = Attributes.Value, BrowsePath = [new QualifiedName("SourceNode")] },
            ]
        };
        return filter;
    }

    private static ConnectionInfo BuildConnectionInfo(ISessionAdapter session)
    {
        return new ConnectionInfo(
            session.EndpointUrl,
            session.ServerName,
            session.SecurityMode,
            session.SecurityPolicyUri,
            session.SessionId,
            session.SessionName);
    }

    private void TransitionState(ConnectionState newState, string endpointUrl)
    {
        ConnectionState oldState;
        lock (_subscriptionLock)
        {
            oldState = _state;
            if (oldState == newState) return;
            _state = newState;
        }
        ConnectionStateChanged?.Invoke(this, new ConnectionStateChangedEventArgs(oldState, newState, endpointUrl));
    }

    private void ThrowIfDisposed()
    {
        if (_disposed) throw new ObjectDisposedException(nameof(OpcUaClientService));
    }

    private void ThrowIfNotConnected()
    {
        if (_state != ConnectionState.Connected || _session == null)
            throw new InvalidOperationException("Not connected to an OPC UA server.");
    }
}