scadalink-design/deprecated/LmxProxyDataConnection.cs

using Microsoft.Extensions.Logging;
using ScadaLink.Commons.Interfaces.Protocol;
using ScadaLink.Commons.Types.Enums;
using ZB.MOM.WW.LmxProxy.Client.Domain;
using ScadaLink.Commons.Types;
using QualityCode = ScadaLink.Commons.Interfaces.Protocol.QualityCode;
using WriteResult = ScadaLink.Commons.Interfaces.Protocol.WriteResult;

namespace ScadaLink.DataConnectionLayer.Adapters;

/// <summary>
/// LmxProxy adapter implementing IDataConnection.
/// Maps IDataConnection operations to the real LmxProxy SDK client
/// via the <see cref="ILmxProxyClient"/> abstraction.
///
/// LmxProxy-specific behavior:
/// - Session-based connection with automatic 30s keep-alive (managed by SDK)
/// - gRPC streaming for subscriptions via ILmxSubscription handles
/// - API key authentication via x-api-key gRPC metadata header
/// - Native TypedValue writes (v2 protocol)
/// </summary>
public class LmxProxyDataConnection : IDataConnection
{
    private readonly ILmxProxyClientFactory _clientFactory;
    private readonly ILogger<LmxProxyDataConnection> _logger;
    private ILmxProxyClient? _client;
    private string _host = "localhost";
    private int _port = 50051;
    private ConnectionHealth _status = ConnectionHealth.Disconnected;

    private readonly Dictionary<string, ILmxSubscription> _subscriptions = new();
    private volatile bool _disconnectFired;
    private StaleTagMonitor? _staleMonitor;
    private string? _heartbeatSubscriptionId;

    public LmxProxyDataConnection(ILmxProxyClientFactory clientFactory, ILogger<LmxProxyDataConnection> logger)
    {
        _clientFactory = clientFactory;
        _logger = logger;
    }

    public ConnectionHealth Status => _status;
    public event Action? Disconnected;

    public async Task ConnectAsync(IDictionary<string, string> connectionDetails, CancellationToken cancellationToken = default)
    {
        _host = connectionDetails.TryGetValue("Host", out var host) ? host : "localhost";
        if (connectionDetails.TryGetValue("Port", out var portStr) && int.TryParse(portStr, out var port))
            _port = port;
        connectionDetails.TryGetValue("ApiKey", out var apiKey);

        var useTls = connectionDetails.TryGetValue("UseTls", out var tlsStr) && bool.TryParse(tlsStr, out var tls) && tls;

        _status = ConnectionHealth.Connecting;
        _client = _clientFactory.Create(_host, _port, apiKey, useTls);

        await _client.ConnectAsync(cancellationToken);
        _status = ConnectionHealth.Connected;
        _disconnectFired = false;

        _logger.LogInformation("LmxProxy connected to {Host}:{Port}", _host, _port);

        // Heartbeat stale tag monitoring (optional)
        await StartHeartbeatMonitorAsync(connectionDetails, cancellationToken);
    }

    private async Task StartHeartbeatMonitorAsync(IDictionary<string, string> connectionDetails, CancellationToken cancellationToken)
    {
        if (!connectionDetails.TryGetValue("HeartbeatTagPath", out var heartbeatTag) || string.IsNullOrWhiteSpace(heartbeatTag))
            return;

        var maxSilenceSeconds = connectionDetails.TryGetValue("HeartbeatMaxSilence", out var silenceStr)
            && int.TryParse(silenceStr, out var sec) ? sec : 30;

        _staleMonitor?.Dispose();
        _staleMonitor = new StaleTagMonitor(TimeSpan.FromSeconds(maxSilenceSeconds));
        _staleMonitor.Stale += () =>
        {
            _logger.LogWarning("LmxProxy heartbeat tag '{Tag}' stale — no update in {Seconds}s", heartbeatTag, maxSilenceSeconds);
            RaiseDisconnected();
        };

        try
        {
            _heartbeatSubscriptionId = await SubscribeAsync(heartbeatTag, (tag, value) =>
            {
                _logger.LogDebug("LmxProxy heartbeat received: {Tag} = {Value} (quality={Quality})", tag, value.Value, value.Quality);
                _staleMonitor.OnValueReceived();
            }, cancellationToken);
            _staleMonitor.Start();
            _logger.LogInformation("LmxProxy heartbeat monitor started for '{Tag}' with {Seconds}s max silence", heartbeatTag, maxSilenceSeconds);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "Failed to subscribe to heartbeat tag '{Tag}' — stale monitor not active", heartbeatTag);
            _staleMonitor.Dispose();
            _staleMonitor = null;
        }
    }

    public async Task DisconnectAsync(CancellationToken cancellationToken = default)
    {
        StopHeartbeatMonitor();
        if (_client != null)
        {
            await _client.DisconnectAsync();
            _status = ConnectionHealth.Disconnected;
            _logger.LogInformation("LmxProxy disconnected from {Host}:{Port}", _host, _port);
        }
    }

    public async Task<ReadResult> ReadAsync(string tagPath, CancellationToken cancellationToken = default)
    {
        EnsureConnected();

        try
        {
            var vtq = await _client!.ReadAsync(tagPath, cancellationToken);
            var quality = MapQuality(vtq.Quality);
            var tagValue = new TagValue(NormalizeValue(vtq.Value), quality, new DateTimeOffset(vtq.Timestamp, TimeSpan.Zero));

            return vtq.Quality.IsBad()
                ? new ReadResult(false, tagValue, "LmxProxy read returned bad quality")
                : new ReadResult(true, tagValue, null);
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            _logger.LogWarning(ex, "LmxProxy read failed for {TagPath} — connection may be lost", tagPath);
            RaiseDisconnected();
            throw;
        }
    }

    public async Task<IReadOnlyDictionary<string, ReadResult>> ReadBatchAsync(IEnumerable<string> tagPaths, CancellationToken cancellationToken = default)
    {
        EnsureConnected();

        var vtqs = await _client!.ReadBatchAsync(tagPaths, cancellationToken);
        var results = new Dictionary<string, ReadResult>();

        foreach (var (tag, vtq) in vtqs)
        {
            var quality = MapQuality(vtq.Quality);
            var tagValue = new TagValue(NormalizeValue(vtq.Value), quality, new DateTimeOffset(vtq.Timestamp, TimeSpan.Zero));
            results[tag] = vtq.Quality.IsBad()
                ? new ReadResult(false, tagValue, "LmxProxy read returned bad quality")
                : new ReadResult(true, tagValue, null);
        }

        return results;
    }

    public async Task<WriteResult> WriteAsync(string tagPath, object? value, CancellationToken cancellationToken = default)
    {
        EnsureConnected();

        try
        {
            await _client!.WriteAsync(tagPath, ToTypedValue(value), cancellationToken);
            return new WriteResult(true, null);
        }
        catch (Exception ex)
        {
            return new WriteResult(false, ex.Message);
        }
    }

    public async Task<IReadOnlyDictionary<string, WriteResult>> WriteBatchAsync(IDictionary<string, object?> values, CancellationToken cancellationToken = default)
    {
        EnsureConnected();

        try
        {
            var typedValues = values.ToDictionary(kv => kv.Key, kv => ToTypedValue(kv.Value));
            await _client!.WriteBatchAsync(typedValues, cancellationToken);

            return values.Keys.ToDictionary(k => k, _ => new WriteResult(true, null))
                as IReadOnlyDictionary<string, WriteResult>;
        }
        catch (Exception ex)
        {
            return values.Keys.ToDictionary(k => k, _ => new WriteResult(false, ex.Message))
                as IReadOnlyDictionary<string, WriteResult>;
        }
    }

    public async Task<bool> WriteBatchAndWaitAsync(
        IDictionary<string, object?> values, string flagPath, object? flagValue,
        string responsePath, object? responseValue, TimeSpan timeout,
        CancellationToken cancellationToken = default)
    {
        var allValues = new Dictionary<string, object?>(values) { [flagPath] = flagValue };
        var writeResults = await WriteBatchAsync(allValues, cancellationToken);

        if (writeResults.Values.Any(r => !r.Success))
            return false;

        var deadline = DateTimeOffset.UtcNow + timeout;
        while (DateTimeOffset.UtcNow < deadline)
        {
            cancellationToken.ThrowIfCancellationRequested();
            var readResult = await ReadAsync(responsePath, cancellationToken);
            if (readResult.Success && readResult.Value != null && Equals(readResult.Value.Value, responseValue))
                return true;

            await Task.Delay(100, cancellationToken);
        }

        return false;
    }

    public async Task<string> SubscribeAsync(string tagPath, SubscriptionCallback callback, CancellationToken cancellationToken = default)
    {
        EnsureConnected();

        var subscription = await _client!.SubscribeAsync(
            [tagPath],
            (path, vtq) =>
            {
                var quality = MapQuality(vtq.Quality);
                callback(path, new TagValue(NormalizeValue(vtq.Value), quality, new DateTimeOffset(vtq.Timestamp, TimeSpan.Zero)));
            },
            onStreamError: ex =>
            {
                _logger.LogWarning(ex, "LmxProxy subscription stream ended unexpectedly for {TagPath}", tagPath);
                RaiseDisconnected();
            },
            cancellationToken);

        var subscriptionId = Guid.NewGuid().ToString("N");
        _subscriptions[subscriptionId] = subscription;
        return subscriptionId;
    }

    public async Task UnsubscribeAsync(string subscriptionId, CancellationToken cancellationToken = default)
    {
        if (_subscriptions.Remove(subscriptionId, out var subscription))
        {
            await subscription.DisposeAsync();
        }
    }

    private void StopHeartbeatMonitor()
    {
        _staleMonitor?.Dispose();
        _staleMonitor = null;
        _heartbeatSubscriptionId = null;
    }

    public async ValueTask DisposeAsync()
    {
        StopHeartbeatMonitor();
        foreach (var subscription in _subscriptions.Values)
        {
            try { await subscription.DisposeAsync(); }
            catch { /* best-effort cleanup */ }
        }
        _subscriptions.Clear();

        if (_client != null)
        {
            await _client.DisposeAsync();
            _client = null;
        }
        _status = ConnectionHealth.Disconnected;
    }

    private void EnsureConnected()
    {
        if (_client == null || !_client.IsConnected)
            throw new InvalidOperationException("LmxProxy client is not connected.");
    }

    private void RaiseDisconnected()
    {
        if (_disconnectFired) return;
        _disconnectFired = true;
        _status = ConnectionHealth.Disconnected;
        _logger.LogWarning("LmxProxy connection to {Host}:{Port} lost", _host, _port);
        Disconnected?.Invoke();
    }

    /// <summary>
    /// Normalizes a Vtq value for consumption by the rest of the system.
    /// Converts .NET arrays (bool[], int[], DateTime[], etc.) to comma-separated
    /// display strings so downstream code sees simple string representations.
    /// </summary>
    private static object? NormalizeValue(object? value) => value switch
    {
        null or string => value,
        IFormattable => value,
        _ => ValueFormatter.FormatDisplayValue(value)
    };

    private static QualityCode MapQuality(Quality quality)
    {
        if (quality.IsGood()) return QualityCode.Good;
        if (quality.IsUncertain()) return QualityCode.Uncertain;
        return QualityCode.Bad;
    }

    private static TypedValue ToTypedValue(object? value) => value switch
    {
        bool b => new TypedValue { BoolValue = b },
        int i => new TypedValue { Int32Value = i },
        long l => new TypedValue { Int64Value = l },
        float f => new TypedValue { FloatValue = f },
        double d => new TypedValue { DoubleValue = d },
        string s => new TypedValue { StringValue = s },
        DateTime dt => new TypedValue { DatetimeValue = dt.ToUniversalTime().Ticks },
        null => new TypedValue { StringValue = string.Empty },
        _ => new TypedValue { StringValue = value.ToString() ?? string.Empty }
    };
}