ScadaBridge/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs

using System.Collections.Concurrent;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
using ZB.MOM.WW.ScadaBridge.Commons.Serialization;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.Secrets.Abstractions;

namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;

/// <summary>
/// MxGateway adapter implementing <see cref="IDataConnection"/> + <see cref="IBrowsableDataConnection"/>.
/// Maps IDataConnection concepts onto the MxAccess Gateway session model via the
/// <see cref="IMxGatewayClient"/> seam:
/// <list type="bullet">
///   <item>Connect → OpenSession + Register, then a background event loop.</item>
///   <item>Subscribe → AddItem + Advise; value changes arrive on the event stream.</item>
///   <item>Read/Write → ReadBulk / WriteBulk.</item>
///   <item>Browse → Galaxy repository BrowseChildren.</item>
/// </list>
/// Reconnection is driven by the <c>DataConnectionActor</c>: a stream fault raises
/// <see cref="Disconnected"/>, the actor disposes this adapter, creates a fresh one,
/// reconnects and re-subscribes all tags.
/// </summary>
public class MxGatewayDataConnection : IDataConnection, IBrowsableDataConnection, IAlarmSubscribableConnection
{
    private readonly IMxGatewayClientFactory _clientFactory;
    private readonly ILogger<MxGatewayDataConnection> _logger;
    private readonly ISecretResolver? _secretResolver;
    private IMxGatewayClient? _client;
    private ConnectionHealth _status = ConnectionHealth.Disconnected;
    private CancellationTokenSource? _eventLoopCts;

    // Native alarm feed: the gateway StreamAlarms RPC is session-less and
    // gateway-wide, so one shared feed serves the whole connection. The
    // DataConnectionActor routes transitions to instances by source reference,
    // so a single shared callback (the first registered) suffices; subscriptions
    // are ref-counted so the feed stops when the last one is removed.
    private CancellationTokenSource? _alarmCts;
    private int _alarmSubCount;
    private readonly object _alarmLock = new();

    // subscriptionId → (tagPath, callback) so the event loop can route updates by tag,
    // plus tagPath → subscriptionId for reverse lookup. Concurrent because the event
    // loop reads from a background thread while Subscribe/Unsubscribe mutate.
    private readonly ConcurrentDictionary<string, (string TagPath, SubscriptionCallback Callback)> _subs = new();
    private readonly ConcurrentDictionary<string, string> _tagToSub = new();

    // DataConnectionLayer mirror of OpcUaDataConnection's once-only guard: an int toggled
    // with Interlocked.Exchange so only the first caller raises Disconnected.
    // 0 = not fired, 1 = fired. Reset on (re)connect.
    private int _disconnectFired;

    /// <summary>Initializes a new instance of <see cref="MxGatewayDataConnection"/>.</summary>
    /// <param name="clientFactory">Factory used to create gateway client instances.</param>
    /// <param name="logger">Logger instance.</param>
    /// <param name="secretResolver">
    /// Optional secret resolver used to resolve a <c>secret:&lt;name&gt;</c> ApiKey reference at
    /// connect time. When null, a literal ApiKey still works, but a <c>secret:</c> reference
    /// fails closed (a connection is never established with an unresolved key).
    /// </param>
    public MxGatewayDataConnection(
        IMxGatewayClientFactory clientFactory,
        ILogger<MxGatewayDataConnection> logger,
        ISecretResolver? secretResolver = null)
    {
        _clientFactory = clientFactory;
        _logger = logger;
        _secretResolver = secretResolver;
    }

    /// <inheritdoc />
    public ConnectionHealth Status => _status;

    /// <summary>Raised once when the gateway event stream faults (connection lost).</summary>
    public event Action? Disconnected;

    /// <inheritdoc />
    public async Task ConnectAsync(IDictionary<string, string> connectionDetails, CancellationToken cancellationToken = default)
    {
        var cfg = MxGatewayEndpointConfigSerializer.FromFlatDict(connectionDetails);

        // Resolve a `secret:<name>` ApiKey reference BEFORE any teardown/client creation
        // so a fail-closed throw here leaves adapter state untouched (no half-open client,
        // previous connection — if any — still intact for a later retry). A literal key
        // (no `secret:` prefix) passes through unchanged for back-compat.
        var apiKey = cfg.ApiKey;
        if (!string.IsNullOrEmpty(apiKey) &&
            apiKey.StartsWith("secret:", StringComparison.OrdinalIgnoreCase))
        {
            if (_secretResolver is null)
            {
                throw new InvalidOperationException(
                    "MxGateway ApiKey uses a 'secret:' reference but no ISecretResolver is available to resolve it.");
            }
            var secretName = apiKey["secret:".Length..];
            apiKey = await _secretResolver.GetAsync(new SecretName(secretName), cancellationToken)
                     ?? throw new InvalidOperationException(
                         $"MxGateway ApiKey secret '{secretName}' could not be resolved (not found or tombstoned).");
        }

        // Reconnect reuses this adapter: cancel the previous event loop and dispose
        // the previous client so a flapping gateway cannot accumulate orphaned
        // streams/sockets, and the old loop's fault path cannot signal Disconnected
        // against the new session after the guard resets below. Keep this BEFORE the
        // _disconnectFired reset so a fault raised by the dying loop during teardown
        // is absorbed by the old guard cycle. Dispose is best-effort fire-and-forget.
        if (_eventLoopCts is { } previousLoopCts)
        {
            previousLoopCts.Cancel();
            previousLoopCts.Dispose();
            _eventLoopCts = null;
        }
        if (_client is { } previousClient)
        {
            _ = previousClient.DisposeAsync().AsTask().ContinueWith(
                t => _logger.LogWarning(t.Exception?.GetBaseException(),
                    "Disposing previous MxGateway client failed"),
                TaskContinuationOptions.OnlyOnFaulted);
        }

        Interlocked.Exchange(ref _disconnectFired, 0); // reset guard on (re)connect, like OPC UA
        _client = _clientFactory.Create();

        await _client.ConnectAsync(new MxGatewayConnectionOptions(
            cfg.Endpoint,
            apiKey,
            string.IsNullOrWhiteSpace(cfg.ClientName) ? "scadabridge" : cfg.ClientName,
            cfg.WriteUserId,
            cfg.UseTls,
            string.IsNullOrWhiteSpace(cfg.CaFile) ? null : cfg.CaFile,
            string.IsNullOrWhiteSpace(cfg.ServerName) ? null : cfg.ServerName,
            cfg.ReadTimeoutMs), cancellationToken);

        _status = ConnectionHealth.Connected;

        // Background event loop: route each value change to the matching subscription callback.
        _eventLoopCts = new CancellationTokenSource();
        var loopToken = _eventLoopCts.Token;   // capture NOW: the field may be replaced by a
        var loopClient = _client;              // subsequent reconnect before Task.Run executes (N1)
        _ = Task.Run(() => RunEventLoopAsync(loopClient, loopToken));
    }

    private async Task RunEventLoopAsync(IMxGatewayClient client, CancellationToken ct)
    {
        try
        {
            await client.RunEventLoopAsync(update =>
            {
                if (_tagToSub.TryGetValue(update.TagPath, out var subId) && _subs.TryGetValue(subId, out var s))
                    s.Callback(update.TagPath, new TagValue(update.Value, update.Quality, update.Timestamp));
            }, ct);
        }
        catch (OperationCanceledException)
        {
            // Normal shutdown (DisconnectAsync / DisposeAsync cancelled the loop).
        }
        catch (Exception ex) when (ct.IsCancellationRequested)
        {
            // Teardown fault of a superseded loop (N1): Cancel() is not synchronous with
            // the loop's exit — a cancelled/disposed gRPC stream commonly surfaces as
            // RpcException(StatusCode.Cancelled) (Grpc.Net default without
            // ThrowOperationCanceledOnCancellation) or ObjectDisposedException from the
            // concurrent DisposeAsync, milliseconds AFTER ConnectAsync has already reset
            // _disconnectFired. Any fault on a cancelled token is normal shutdown;
            // raising Disconnected here would flap the NEW healthy session.
            _logger.LogDebug(ex,
                "MxGateway event loop faulted after cancellation — superseded loop teardown; not signalling disconnect");
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "MxGateway event stream faulted; signalling disconnect");
            RaiseDisconnected();
        }
    }

    private void RaiseDisconnected()
    {
        if (Interlocked.Exchange(ref _disconnectFired, 1) == 0)
        {
            _status = ConnectionHealth.Disconnected;
            Disconnected?.Invoke();
        }
    }

    /// <inheritdoc />
    public async Task DisconnectAsync(CancellationToken cancellationToken = default)
    {
        _eventLoopCts?.Cancel();
        lock (_alarmLock)
        {
            _alarmCts?.Cancel();
            _alarmCts?.Dispose();
            _alarmCts = null;
            _alarmSubCount = 0;
        }
        if (_client is not null)
            await _client.DisconnectAsync(cancellationToken);
        _status = ConnectionHealth.Disconnected;
    }

    /// <inheritdoc />
    public async Task<string> SubscribeAsync(string tagPath, SubscriptionCallback callback, CancellationToken cancellationToken = default)
    {
        var subId = await _client!.SubscribeAsync(tagPath, cancellationToken);
        _subs[subId] = (tagPath, callback);
        _tagToSub[tagPath] = subId;
        return subId;
    }

    /// <inheritdoc />
    public async Task UnsubscribeAsync(string subscriptionId, CancellationToken cancellationToken = default)
    {
        if (_subs.TryRemove(subscriptionId, out var s))
            _tagToSub.TryRemove(s.TagPath, out _);
        await _client!.UnsubscribeAsync(subscriptionId, cancellationToken);
    }

    /// <inheritdoc />
    public Task<string> SubscribeAlarmsAsync(
        string sourceReference, string? conditionFilter,
        AlarmTransitionCallback callback, CancellationToken cancellationToken = default)
    {
        lock (_alarmLock)
        {
            _alarmSubCount++;
            if (_alarmCts == null)
            {
                _alarmCts = new CancellationTokenSource();
                var token = _alarmCts.Token;
                var client = _client!;
                // Gateway-wide feed (null prefix). The MxGateway has no server-side
                // condition filter, so conditionFilter is intentionally NOT forwarded
                // here: the DataConnectionActor applies it as the authoritative
                // client-side gate per source reference AND per condition type
                // (AlarmConditionFilter), uniform with the OPC UA path.
                _ = Task.Run(() => client.RunAlarmStreamAsync(null, t => callback(t), token), token);
            }
        }
        return Task.FromResult(Guid.NewGuid().ToString());
    }

    /// <inheritdoc />
    public Task UnsubscribeAlarmsAsync(string subscriptionId, CancellationToken cancellationToken = default)
    {
        lock (_alarmLock)
        {
            if (_alarmSubCount > 0)
                _alarmSubCount--;
            if (_alarmSubCount == 0)
            {
                _alarmCts?.Cancel();
                _alarmCts?.Dispose();
                _alarmCts = null;
            }
        }
        return Task.CompletedTask;
    }

    /// <inheritdoc />
    public async Task<ReadResult> ReadAsync(string tagPath, CancellationToken cancellationToken = default)
    {
        var r = (await _client!.ReadAsync(new[] { tagPath }, cancellationToken)).Single();
        return ToReadResult(r);
    }

    /// <inheritdoc />
    public async Task<IReadOnlyDictionary<string, ReadResult>> ReadBatchAsync(IEnumerable<string> tagPaths, CancellationToken cancellationToken = default)
    {
        var list = tagPaths.ToList();
        var results = await _client!.ReadAsync(list, cancellationToken);
        return results.ToDictionary(r => r.TagPath, ToReadResult);
    }

    private static ReadResult ToReadResult(MxReadOutcome r) => r.Success
        ? new ReadResult(true, new TagValue(r.Value, r.Quality, r.Timestamp), null)
        : new ReadResult(false, null, r.Error);

    /// <inheritdoc />
    public async Task<WriteResult> WriteAsync(string tagPath, object? value, CancellationToken cancellationToken = default)
    {
        var w = (await _client!.WriteAsync(new[] { (tagPath, value) }, cancellationToken)).Single();
        return new WriteResult(w.Success, w.Error);
    }

    /// <inheritdoc />
    public async Task<IReadOnlyDictionary<string, WriteResult>> WriteBatchAsync(IDictionary<string, object?> values, CancellationToken cancellationToken = default)
    {
        var results = await _client!.WriteAsync(values.Select(kv => (kv.Key, kv.Value)).ToList(), cancellationToken);
        return results.ToDictionary(w => w.TagPath, w => new WriteResult(w.Success, w.Error));
    }

    /// <inheritdoc />
    public async Task<bool> WriteBatchAndWaitAsync(
        IDictionary<string, object?> values, string flagPath, object? flagValue,
        string responsePath, object? responseValue, TimeSpan timeout, CancellationToken cancellationToken = default)
    {
        await WriteBatchAsync(values, cancellationToken);
        await WriteAsync(flagPath, flagValue, cancellationToken);

        using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        timeoutCts.CancelAfter(timeout);
        try
        {
            while (!timeoutCts.IsCancellationRequested)
            {
                var r = await ReadAsync(responsePath, timeoutCts.Token);
                // r.Value is a TagValue wrapper; compare its underlying scalar. String
                // projection tolerates numeric type differences across the gRPC boundary.
                if (r.Success && string.Equals(r.Value?.Value?.ToString(), responseValue?.ToString(), StringComparison.Ordinal))
                    return true;
                await Task.Delay(TimeSpan.FromMilliseconds(200), timeoutCts.Token);
            }
        }
        catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
        {
            // Timeout elapsed (the linked CTS, not the caller's token) — fall through to false.
        }
        return false;
    }

    /// <inheritdoc />
    public async Task<BrowseChildrenResult> BrowseChildrenAsync(
        string? parentNodeId,
        string? continuationToken = null,
        CancellationToken cancellationToken = default)
    {
        if (_status != ConnectionHealth.Connected || _client is null)
            throw new ConnectionNotConnectedException($"MxGateway connection is not connected (status: {_status}).");

        // MxGateway browse is not pageable — the gateway returns a single,
        // already-bounded child set per node. Any continuation token is ignored
        // and no continuation is ever surfaced (ContinuationToken stays null).
        var (children, truncated) = await _client.BrowseChildrenAsync(parentNodeId, cancellationToken);
        var nodes = children
            .Select(c => new BrowseNode(c.NodeId, c.DisplayName, c.NodeClass, c.HasChildren, DataType: c.DataType))
            .ToList();
        return new BrowseChildrenResult(nodes, truncated);
    }

    /// <summary>Cancels the event loop and disposes the underlying MxGateway client.</summary>
    /// <returns>A <see cref="ValueTask"/> that completes when disposal is finished.</returns>
    public async ValueTask DisposeAsync()
    {
        _eventLoopCts?.Cancel();
        // The DataConnectionActor disposes adapters
        // fire-and-forget on failover/stop without necessarily calling
        // DisconnectAsync first, so tear down the alarm stream here too — otherwise
        // the long-running RunAlarmStreamAsync task and its CTS leak on every
        // MxGateway failover/teardown that goes through DisposeAsync. Mirror the
        // lock-guarded block already in DisconnectAsync.
        lock (_alarmLock)
        {
            _alarmCts?.Cancel();
            _alarmCts?.Dispose();
            _alarmCts = null;
            _alarmSubCount = 0;
        }
        if (_client is not null)
            await _client.DisposeAsync();
        GC.SuppressFinalize(this);
    }
}