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;

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 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>
    public MxGatewayDataConnection(IMxGatewayClientFactory clientFactory, ILogger<MxGatewayDataConnection> logger)
    {
        _clientFactory = clientFactory;
        _logger = logger;
    }

    /// <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);
        Interlocked.Exchange(ref _disconnectFired, 0); // reset guard on (re)connect, like OPC UA
        _client = _clientFactory.Create();

        await _client.ConnectAsync(new MxGatewayConnectionOptions(
            cfg.Endpoint,
            cfg.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();
        _ = Task.Run(() => RunEventLoopAsync(_eventLoopCts.Token));
    }

    private async Task RunEventLoopAsync(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)
        {
            _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 actor filters per source reference.
                _ = 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, CancellationToken cancellationToken = default)
    {
        if (_status != ConnectionHealth.Connected || _client is null)
            throw new ConnectionNotConnectedException($"MxGateway connection is not connected (status: {_status}).");

        var (children, truncated) = await _client.BrowseChildrenAsync(parentNodeId, cancellationToken);
        var nodes = children
            .Select(c => new BrowseNode(c.NodeId, c.DisplayName, c.NodeClass, c.HasChildren))
            .ToList();
        return new BrowseChildrenResult(nodes, truncated);
    }

    /// <inheritdoc />
    public async ValueTask DisposeAsync()
    {
        _eventLoopCts?.Cancel();
        if (_client is not null)
            await _client.DisposeAsync();
        GC.SuppressFinalize(this);
    }
}