lmxopcua/src/ZB.MOM.WW.OtOpcUa.Historian.Aveva/HistorianDataSource.cs

using System;
using System.Collections.Generic;
using StringCollection = System.Collections.Specialized.StringCollection;
using System.Threading;
using System.Threading.Tasks;
using ArchestrA;
using Opc.Ua;
using Serilog;
using ZB.MOM.WW.OtOpcUa.Host.Configuration;
using ZB.MOM.WW.OtOpcUa.Host.Domain;
using ZB.MOM.WW.OtOpcUa.Host.Historian;

namespace ZB.MOM.WW.OtOpcUa.Historian.Aveva
{
    /// <summary>
    ///     Reads historical data from the Wonderware Historian via the aahClientManaged SDK.
    /// </summary>
    public sealed class HistorianDataSource : IHistorianDataSource
    {
        private static readonly ILogger Log = Serilog.Log.ForContext<HistorianDataSource>();

        private readonly HistorianConfiguration _config;
        private readonly object _connectionLock = new object();
        private readonly object _eventConnectionLock = new object();
        private readonly IHistorianConnectionFactory _factory;
        private HistorianAccess? _connection;
        private HistorianAccess? _eventConnection;
        private bool _disposed;

        // Runtime query health state. Guarded by _healthLock — updated on every read
        // method exit (success or failure) so the dashboard can distinguish "plugin
        // loaded but never queried" from "plugin loaded and queries are failing".
        private readonly object _healthLock = new object();
        private long _totalSuccesses;
        private long _totalFailures;
        private int _consecutiveFailures;
        private DateTime? _lastSuccessTime;
        private DateTime? _lastFailureTime;
        private string? _lastError;
        private string? _activeProcessNode;
        private string? _activeEventNode;

        // Cluster endpoint picker — shared across process + event paths so a node that
        // fails on one silo is skipped on the other. Initialized from config at construction.
        private readonly HistorianClusterEndpointPicker _picker;

        /// <summary>
        ///     Initializes a Historian reader that translates OPC UA history requests into Wonderware Historian SDK queries.
        /// </summary>
        /// <param name="config">The Historian SDK connection settings used for runtime history lookups.</param>
        public HistorianDataSource(HistorianConfiguration config)
            : this(config, new SdkHistorianConnectionFactory(), null) { }

        /// <summary>
        ///     Initializes a Historian reader with a custom connection factory for testing. When
        ///     <paramref name="picker"/> is <see langword="null"/> a new picker is built from
        ///     <paramref name="config"/>, preserving backward compatibility with existing tests.
        /// </summary>
        internal HistorianDataSource(
            HistorianConfiguration config,
            IHistorianConnectionFactory factory,
            HistorianClusterEndpointPicker? picker = null)
        {
            _config = config;
            _factory = factory;
            _picker = picker ?? new HistorianClusterEndpointPicker(config);
        }

        /// <summary>
        ///     Iterates the picker's healthy node list, cloning the configuration per attempt and
        ///     handing it to the factory. Marks each tried node as healthy on success or failed on
        ///     exception. Returns the winning connection + node name; throws when no nodes succeed.
        /// </summary>
        private (HistorianAccess Connection, string Node) ConnectToAnyHealthyNode(HistorianConnectionType type)
        {
            var candidates = _picker.GetHealthyNodes();
            if (candidates.Count == 0)
            {
                var total = _picker.NodeCount;
                throw new InvalidOperationException(
                    total == 0
                        ? "No historian nodes configured"
                        : $"All {total} historian nodes are in cooldown — no healthy endpoints to connect to");
            }

            Exception? lastException = null;
            foreach (var node in candidates)
            {
                var attemptConfig = CloneConfigWithServerName(node);
                try
                {
                    var conn = _factory.CreateAndConnect(attemptConfig, type);
                    _picker.MarkHealthy(node);
                    return (conn, node);
                }
                catch (Exception ex)
                {
                    _picker.MarkFailed(node, ex.Message);
                    lastException = ex;
                    Log.Warning(ex,
                        "Historian node {Node} failed during connect attempt; trying next candidate", node);
                }
            }

            var inner = lastException?.Message ?? "(no detail)";
            throw new InvalidOperationException(
                $"All {candidates.Count} healthy historian candidate(s) failed during connect: {inner}",
                lastException);
        }

        private HistorianConfiguration CloneConfigWithServerName(string serverName)
        {
            return new HistorianConfiguration
            {
                Enabled = _config.Enabled,
                ServerName = serverName,
                ServerNames = _config.ServerNames,
                FailureCooldownSeconds = _config.FailureCooldownSeconds,
                IntegratedSecurity = _config.IntegratedSecurity,
                UserName = _config.UserName,
                Password = _config.Password,
                Port = _config.Port,
                CommandTimeoutSeconds = _config.CommandTimeoutSeconds,
                MaxValuesPerRead = _config.MaxValuesPerRead
            };
        }

        /// <inheritdoc />
        public HistorianHealthSnapshot GetHealthSnapshot()
        {
            var nodeStates = _picker.SnapshotNodeStates();
            var healthyCount = 0;
            foreach (var n in nodeStates)
                if (n.IsHealthy)
                    healthyCount++;

            lock (_healthLock)
            {
                return new HistorianHealthSnapshot
                {
                    TotalQueries = _totalSuccesses + _totalFailures,
                    TotalSuccesses = _totalSuccesses,
                    TotalFailures = _totalFailures,
                    ConsecutiveFailures = _consecutiveFailures,
                    LastSuccessTime = _lastSuccessTime,
                    LastFailureTime = _lastFailureTime,
                    LastError = _lastError,
                    ProcessConnectionOpen = Volatile.Read(ref _connection) != null,
                    EventConnectionOpen = Volatile.Read(ref _eventConnection) != null,
                    ActiveProcessNode = _activeProcessNode,
                    ActiveEventNode = _activeEventNode,
                    NodeCount = nodeStates.Count,
                    HealthyNodeCount = healthyCount,
                    Nodes = nodeStates
                };
            }
        }

        private void RecordSuccess()
        {
            lock (_healthLock)
            {
                _totalSuccesses++;
                _lastSuccessTime = DateTime.UtcNow;
                _consecutiveFailures = 0;
                _lastError = null;
            }
        }

        private void RecordFailure(string error)
        {
            lock (_healthLock)
            {
                _totalFailures++;
                _lastFailureTime = DateTime.UtcNow;
                _consecutiveFailures++;
                _lastError = error;
            }
        }

        private void EnsureConnected()
        {
            if (_disposed)
                throw new ObjectDisposedException(nameof(HistorianDataSource));

            // Fast path: already connected (no lock needed)
            if (Volatile.Read(ref _connection) != null)
                return;

            // Create and wait for connection outside the lock so concurrent history
            // requests are not serialized behind a slow Historian handshake. The cluster
            // picker iterates configured nodes and returns the first that successfully connects.
            var (conn, winningNode) = ConnectToAnyHealthyNode(HistorianConnectionType.Process);

            lock (_connectionLock)
            {
                if (_disposed)
                {
                    conn.CloseConnection(out _);
                    conn.Dispose();
                    throw new ObjectDisposedException(nameof(HistorianDataSource));
                }

                if (_connection != null)
                {
                    // Another thread connected while we were waiting
                    conn.CloseConnection(out _);
                    conn.Dispose();
                    return;
                }

                _connection = conn;
                lock (_healthLock)
                    _activeProcessNode = winningNode;
                Log.Information("Historian SDK connection opened to {Server}:{Port}", winningNode, _config.Port);
            }
        }

        private void HandleConnectionError(Exception? ex = null)
        {
            lock (_connectionLock)
            {
                if (_connection == null)
                    return;

                try
                {
                    _connection.CloseConnection(out _);
                    _connection.Dispose();
                }
                catch (Exception disposeEx)
                {
                    Log.Debug(disposeEx, "Error disposing Historian SDK connection during error recovery");
                }

                _connection = null;
                string? failedNode;
                lock (_healthLock)
                {
                    failedNode = _activeProcessNode;
                    _activeProcessNode = null;
                }

                if (failedNode != null)
                    _picker.MarkFailed(failedNode, ex?.Message ?? "mid-query failure");
                Log.Warning(ex, "Historian SDK connection reset (node={Node}) — will reconnect on next request",
                    failedNode ?? "(unknown)");
            }
        }

        private void EnsureEventConnected()
        {
            if (_disposed)
                throw new ObjectDisposedException(nameof(HistorianDataSource));

            if (Volatile.Read(ref _eventConnection) != null)
                return;

            var (conn, winningNode) = ConnectToAnyHealthyNode(HistorianConnectionType.Event);

            lock (_eventConnectionLock)
            {
                if (_disposed)
                {
                    conn.CloseConnection(out _);
                    conn.Dispose();
                    throw new ObjectDisposedException(nameof(HistorianDataSource));
                }

                if (_eventConnection != null)
                {
                    conn.CloseConnection(out _);
                    conn.Dispose();
                    return;
                }

                _eventConnection = conn;
                lock (_healthLock)
                    _activeEventNode = winningNode;
                Log.Information("Historian SDK event connection opened to {Server}:{Port}",
                    winningNode, _config.Port);
            }
        }

        private void HandleEventConnectionError(Exception? ex = null)
        {
            lock (_eventConnectionLock)
            {
                if (_eventConnection == null)
                    return;

                try
                {
                    _eventConnection.CloseConnection(out _);
                    _eventConnection.Dispose();
                }
                catch (Exception disposeEx)
                {
                    Log.Debug(disposeEx, "Error disposing Historian SDK event connection during error recovery");
                }

                _eventConnection = null;
                string? failedNode;
                lock (_healthLock)
                {
                    failedNode = _activeEventNode;
                    _activeEventNode = null;
                }

                if (failedNode != null)
                    _picker.MarkFailed(failedNode, ex?.Message ?? "mid-query failure");
                Log.Warning(ex, "Historian SDK event connection reset (node={Node}) — will reconnect on next request",
                    failedNode ?? "(unknown)");
            }
        }


        /// <inheritdoc />
        public Task<List<DataValue>> ReadRawAsync(
            string tagName, DateTime startTime, DateTime endTime, int maxValues,
            CancellationToken ct = default)
        {
            var results = new List<DataValue>();

            try
            {
                EnsureConnected();

                using var query = _connection!.CreateHistoryQuery();
                var args = new HistoryQueryArgs
                {
                    TagNames = new StringCollection { tagName },
                    StartDateTime = startTime,
                    EndDateTime = endTime,
                    RetrievalMode = HistorianRetrievalMode.Full
                };

                if (maxValues > 0)
                    args.BatchSize = (uint)maxValues;
                else if (_config.MaxValuesPerRead > 0)
                    args.BatchSize = (uint)_config.MaxValuesPerRead;

                if (!query.StartQuery(args, out var error))
                {
                    Log.Warning("Historian SDK raw query start failed for {Tag}: {Error}", tagName, error.ErrorCode);
                    RecordFailure($"raw StartQuery: {error.ErrorCode}");
                    HandleConnectionError();
                    return Task.FromResult(results);
                }

                var count = 0;
                var limit = maxValues > 0 ? maxValues : _config.MaxValuesPerRead;

                while (query.MoveNext(out error))
                {
                    ct.ThrowIfCancellationRequested();

                    var result = query.QueryResult;
                    var timestamp = DateTime.SpecifyKind(result.StartDateTime, DateTimeKind.Utc);

                    object? value;
                    if (!string.IsNullOrEmpty(result.StringValue) && result.Value == 0)
                        value = result.StringValue;
                    else
                        value = result.Value;

                    var quality = (byte)(result.OpcQuality & 0xFF);

                    results.Add(new DataValue
                    {
                        Value = new Variant(value),
                        SourceTimestamp = timestamp,
                        ServerTimestamp = timestamp,
                        StatusCode = QualityMapper.MapToOpcUaStatusCode(QualityMapper.MapFromMxAccessQuality(quality))
                    });

                    count++;
                    if (limit > 0 && count >= limit)
                        break;
                }

                query.EndQuery(out _);
                RecordSuccess();
            }
            catch (OperationCanceledException)
            {
                throw;
            }
            catch (ObjectDisposedException)
            {
                throw;
            }
            catch (Exception ex)
            {
                Log.Warning(ex, "HistoryRead raw failed for {Tag}", tagName);
                RecordFailure($"raw: {ex.Message}");
                HandleConnectionError(ex);
            }

            Log.Debug("HistoryRead raw: {Tag} returned {Count} values ({Start} to {End})",
                tagName, results.Count, startTime, endTime);

            return Task.FromResult(results);
        }

        /// <inheritdoc />
        public Task<List<DataValue>> ReadAggregateAsync(
            string tagName, DateTime startTime, DateTime endTime,
            double intervalMs, string aggregateColumn,
            CancellationToken ct = default)
        {
            var results = new List<DataValue>();

            try
            {
                EnsureConnected();

                using var query = _connection!.CreateAnalogSummaryQuery();
                var args = new AnalogSummaryQueryArgs
                {
                    TagNames = new StringCollection { tagName },
                    StartDateTime = startTime,
                    EndDateTime = endTime,
                    Resolution = (ulong)intervalMs
                };

                if (!query.StartQuery(args, out var error))
                {
                    Log.Warning("Historian SDK aggregate query start failed for {Tag}: {Error}", tagName,
                        error.ErrorCode);
                    RecordFailure($"aggregate StartQuery: {error.ErrorCode}");
                    HandleConnectionError();
                    return Task.FromResult(results);
                }

                while (query.MoveNext(out error))
                {
                    ct.ThrowIfCancellationRequested();

                    var result = query.QueryResult;
                    var timestamp = DateTime.SpecifyKind(result.StartDateTime, DateTimeKind.Utc);
                    var value = ExtractAggregateValue(result, aggregateColumn);

                    results.Add(new DataValue
                    {
                        Value = new Variant(value),
                        SourceTimestamp = timestamp,
                        ServerTimestamp = timestamp,
                        StatusCode = value != null ? StatusCodes.Good : StatusCodes.BadNoData
                    });
                }

                query.EndQuery(out _);
                RecordSuccess();
            }
            catch (OperationCanceledException)
            {
                throw;
            }
            catch (ObjectDisposedException)
            {
                throw;
            }
            catch (Exception ex)
            {
                Log.Warning(ex, "HistoryRead aggregate failed for {Tag}", tagName);
                RecordFailure($"aggregate: {ex.Message}");
                HandleConnectionError(ex);
            }

            Log.Debug("HistoryRead aggregate ({Aggregate}): {Tag} returned {Count} values",
                aggregateColumn, tagName, results.Count);

            return Task.FromResult(results);
        }

        /// <inheritdoc />
        public Task<List<DataValue>> ReadAtTimeAsync(
            string tagName, DateTime[] timestamps,
            CancellationToken ct = default)
        {
            var results = new List<DataValue>();

            if (timestamps == null || timestamps.Length == 0)
                return Task.FromResult(results);

            try
            {
                EnsureConnected();

                foreach (var timestamp in timestamps)
                {
                    ct.ThrowIfCancellationRequested();

                    using var query = _connection!.CreateHistoryQuery();
                    var args = new HistoryQueryArgs
                    {
                        TagNames = new StringCollection { tagName },
                        StartDateTime = timestamp,
                        EndDateTime = timestamp,
                        RetrievalMode = HistorianRetrievalMode.Interpolated,
                        BatchSize = 1
                    };

                    if (!query.StartQuery(args, out var error))
                    {
                        results.Add(new DataValue
                        {
                            Value = Variant.Null,
                            SourceTimestamp = timestamp,
                            ServerTimestamp = timestamp,
                            StatusCode = StatusCodes.BadNoData
                        });
                        continue;
                    }

                    if (query.MoveNext(out error))
                    {
                        var result = query.QueryResult;
                        object? value;
                        if (!string.IsNullOrEmpty(result.StringValue) && result.Value == 0)
                            value = result.StringValue;
                        else
                            value = result.Value;

                        var quality = (byte)(result.OpcQuality & 0xFF);
                        results.Add(new DataValue
                        {
                            Value = new Variant(value),
                            SourceTimestamp = timestamp,
                            ServerTimestamp = timestamp,
                            StatusCode = QualityMapper.MapToOpcUaStatusCode(
                                QualityMapper.MapFromMxAccessQuality(quality))
                        });
                    }
                    else
                    {
                        results.Add(new DataValue
                        {
                            Value = Variant.Null,
                            SourceTimestamp = timestamp,
                            ServerTimestamp = timestamp,
                            StatusCode = StatusCodes.BadNoData
                        });
                    }

                    query.EndQuery(out _);
                }
                RecordSuccess();
            }
            catch (OperationCanceledException)
            {
                throw;
            }
            catch (ObjectDisposedException)
            {
                throw;
            }
            catch (Exception ex)
            {
                Log.Warning(ex, "HistoryRead at-time failed for {Tag}", tagName);
                RecordFailure($"at-time: {ex.Message}");
                HandleConnectionError(ex);
            }

            Log.Debug("HistoryRead at-time: {Tag} returned {Count} values for {Timestamps} timestamps",
                tagName, results.Count, timestamps.Length);

            return Task.FromResult(results);
        }

        /// <inheritdoc />
        public Task<List<HistorianEventDto>> ReadEventsAsync(
            string? sourceName, DateTime startTime, DateTime endTime, int maxEvents,
            CancellationToken ct = default)
        {
            var results = new List<HistorianEventDto>();

            try
            {
                EnsureEventConnected();

                using var query = _eventConnection!.CreateEventQuery();
                var args = new EventQueryArgs
                {
                    StartDateTime = startTime,
                    EndDateTime = endTime,
                    EventCount = maxEvents > 0 ? (uint)maxEvents : (uint)_config.MaxValuesPerRead,
                    QueryType = HistorianEventQueryType.Events,
                    EventOrder = HistorianEventOrder.Ascending
                };

                if (!string.IsNullOrEmpty(sourceName))
                {
                    query.AddEventFilter("Source", HistorianComparisionType.Equal, sourceName, out _);
                }

                if (!query.StartQuery(args, out var error))
                {
                    Log.Warning("Historian SDK event query start failed: {Error}", error.ErrorCode);
                    RecordFailure($"events StartQuery: {error.ErrorCode}");
                    HandleEventConnectionError();
                    return Task.FromResult(results);
                }

                var count = 0;
                while (query.MoveNext(out error))
                {
                    ct.ThrowIfCancellationRequested();
                    results.Add(ToDto(query.QueryResult));
                    count++;
                    if (maxEvents > 0 && count >= maxEvents)
                        break;
                }

                query.EndQuery(out _);
                RecordSuccess();
            }
            catch (OperationCanceledException)
            {
                throw;
            }
            catch (ObjectDisposedException)
            {
                throw;
            }
            catch (Exception ex)
            {
                Log.Warning(ex, "HistoryRead events failed for source {Source}", sourceName ?? "(all)");
                RecordFailure($"events: {ex.Message}");
                HandleEventConnectionError(ex);
            }

            Log.Debug("HistoryRead events: source={Source} returned {Count} events ({Start} to {End})",
                sourceName ?? "(all)", results.Count, startTime, endTime);

            return Task.FromResult(results);
        }

        private static HistorianEventDto ToDto(HistorianEvent evt)
        {
            return new HistorianEventDto
            {
                Id = evt.Id,
                Source = evt.Source,
                EventTime = evt.EventTime,
                ReceivedTime = evt.ReceivedTime,
                DisplayText = evt.DisplayText,
                Severity = (ushort)evt.Severity
            };
        }

        /// <summary>
        ///     Extracts the requested aggregate value from an <see cref="AnalogSummaryQueryResult"/> by column name.
        /// </summary>
        internal static double? ExtractAggregateValue(AnalogSummaryQueryResult result, string column)
        {
            switch (column)
            {
                case "Average": return result.Average;
                case "Minimum": return result.Minimum;
                case "Maximum": return result.Maximum;
                case "ValueCount": return result.ValueCount;
                case "First": return result.First;
                case "Last": return result.Last;
                case "StdDev": return result.StdDev;
                default: return null;
            }
        }

        /// <summary>
        ///     Closes the Historian SDK connection and releases resources.
        /// </summary>
        public void Dispose()
        {
            if (_disposed)
                return;
            _disposed = true;

            try
            {
                _connection?.CloseConnection(out _);
                _connection?.Dispose();
            }
            catch (Exception ex)
            {
                Log.Warning(ex, "Error closing Historian SDK connection");
            }

            try
            {
                _eventConnection?.CloseConnection(out _);
                _eventConnection?.Dispose();
            }
            catch (Exception ex)
            {
                Log.Warning(ex, "Error closing Historian SDK event connection");
            }

            _connection = null;
            _eventConnection = null;
        }
    }
}