ScadaBridge/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/SiteHealthCollector.cs

using System.Collections.Concurrent;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;

namespace ZB.MOM.WW.ScadaBridge.HealthMonitoring;

/// <summary>
/// Collects health metrics from all site subsystems.
/// Thread-safe: counters use Interlocked operations, connection/tag data uses ConcurrentDictionary.
/// </summary>
public class SiteHealthCollector : ISiteHealthCollector
{
    private int _scriptErrorCount;
    private int _alarmErrorCount;
    private int _deadLetterCount;
    private int _siteAuditWriteFailures;
    private int _auditRedactionFailures;
    private volatile SiteAuditBacklogSnapshot? _siteAuditBacklog;
    private long _siteEventLogWriteFailures;
    private readonly ConcurrentDictionary<string, ConnectionHealth> _connectionStatuses = new();
    private readonly ConcurrentDictionary<string, TagResolutionStatus> _tagResolutionCounts = new();
    private readonly ConcurrentDictionary<string, string> _connectionEndpoints = new();
    private readonly ConcurrentDictionary<string, TagQualityCounts> _tagQualityCounts = new();
    private IReadOnlyDictionary<string, int> _sfBufferDepths = new Dictionary<string, int>();
    private int _deployedInstanceCount, _enabledInstanceCount, _disabledInstanceCount;
    private int _parkedMessageCount;
    private volatile string _nodeHostname = "";
    private volatile IReadOnlyList<Commons.Messages.Health.NodeStatus>? _clusterNodes;
    private volatile bool _isActiveNode;
    private readonly TimeProvider _timeProvider;

    /// <summary>
    /// Creates a collector. The <paramref name="timeProvider"/> stamps each
    /// report's timestamp; it defaults to <see cref="TimeProvider.System"/> and
    /// is injectable so the report timestamp is deterministically testable —
    /// consistent with the rest of the module's time-dependent classes.
    /// </summary>
    /// <param name="timeProvider">Optional custom time provider; defaults to system time.</param>
    public SiteHealthCollector(TimeProvider? timeProvider = null)
    {
        _timeProvider = timeProvider ?? TimeProvider.System;
    }

    /// <inheritdoc />
    public void IncrementScriptError()
    {
        Interlocked.Increment(ref _scriptErrorCount);
    }

    /// <inheritdoc />
    public void IncrementAlarmError()
    {
        Interlocked.Increment(ref _alarmErrorCount);
    }

    /// <inheritdoc />
    public void IncrementDeadLetter()
    {
        Interlocked.Increment(ref _deadLetterCount);
    }

    /// <inheritdoc />
    public void IncrementSiteAuditWriteFailures()
    {
        Interlocked.Increment(ref _siteAuditWriteFailures);
    }

    /// <inheritdoc />
    public void IncrementAuditRedactionFailure()
    {
        Interlocked.Increment(ref _auditRedactionFailures);
    }

    /// <inheritdoc />
    public void UpdateSiteAuditBacklog(SiteAuditBacklogSnapshot snapshot)
    {
        _siteAuditBacklog = snapshot ?? throw new ArgumentNullException(nameof(snapshot));
    }

    /// <inheritdoc />
    public void SetSiteEventLogWriteFailures(long count)
    {
        Interlocked.Exchange(ref _siteEventLogWriteFailures, count);
    }

    /// <inheritdoc />
    public void UpdateConnectionHealth(string connectionName, ConnectionHealth health)
    {
        _connectionStatuses[connectionName] = health;
    }

    /// <inheritdoc />
    public void RemoveConnection(string connectionName)
    {
        _connectionStatuses.TryRemove(connectionName, out _);
        _tagResolutionCounts.TryRemove(connectionName, out _);
        _connectionEndpoints.TryRemove(connectionName, out _);
        _tagQualityCounts.TryRemove(connectionName, out _);
    }

    /// <inheritdoc />
    public void UpdateTagResolution(string connectionName, int totalSubscribed, int successfullyResolved)
    {
        _tagResolutionCounts[connectionName] = new TagResolutionStatus(totalSubscribed, successfullyResolved);
    }

    /// <inheritdoc />
    public void UpdateConnectionEndpoint(string connectionName, string endpoint)
    {
        _connectionEndpoints[connectionName] = endpoint;
    }

    /// <inheritdoc />
    public void UpdateTagQuality(string connectionName, int good, int bad, int uncertain)
    {
        _tagQualityCounts[connectionName] = new TagQualityCounts(good, bad, uncertain);
    }

    /// <inheritdoc />
    public void SetParkedMessageCount(int count)
    {
        Interlocked.Exchange(ref _parkedMessageCount, count);
    }

    /// <inheritdoc />
    public void SetNodeHostname(string hostname) => _nodeHostname = hostname;

    /// <inheritdoc />
    public void SetClusterNodes(IReadOnlyList<Commons.Messages.Health.NodeStatus> nodes) => _clusterNodes = nodes;

    /// <inheritdoc />
    public void SetStoreAndForwardDepths(IReadOnlyDictionary<string, int> depths)
    {
        _sfBufferDepths = depths;
    }

    /// <inheritdoc />
    public void SetInstanceCounts(int deployed, int enabled, int disabled)
    {
        Interlocked.Exchange(ref _deployedInstanceCount, deployed);
        Interlocked.Exchange(ref _enabledInstanceCount, enabled);
        Interlocked.Exchange(ref _disabledInstanceCount, disabled);
    }

    /// <inheritdoc />
    public void SetActiveNode(bool isActive) => _isActiveNode = isActive;

    /// <inheritdoc />
    public bool IsActiveNode => _isActiveNode;

    /// <inheritdoc />
    public void AddIntervalCounters(
        int scriptErrors,
        int alarmErrors,
        int deadLetters,
        int siteAuditWriteFailures,
        int auditRedactionFailures)
    {
        // HealthMonitoring-017: each counter is restored atomically via
        // Interlocked.Add so an increment that arrived during the failed Send
        // (and therefore accumulated against the zero left by CollectReport's
        // Exchange) is correctly summed with the values being put back. No
        // ordering between the five Adds is required — they target independent
        // fields.
        if (scriptErrors != 0) Interlocked.Add(ref _scriptErrorCount, scriptErrors);
        if (alarmErrors != 0) Interlocked.Add(ref _alarmErrorCount, alarmErrors);
        if (deadLetters != 0) Interlocked.Add(ref _deadLetterCount, deadLetters);
        if (siteAuditWriteFailures != 0) Interlocked.Add(ref _siteAuditWriteFailures, siteAuditWriteFailures);
        if (auditRedactionFailures != 0) Interlocked.Add(ref _auditRedactionFailures, auditRedactionFailures);
    }

    /// <inheritdoc />
    public SiteHealthReport CollectReport(string siteId)
    {
        // Atomically read and reset the counters
        var scriptErrors = Interlocked.Exchange(ref _scriptErrorCount, 0);
        var alarmErrors = Interlocked.Exchange(ref _alarmErrorCount, 0);
        var deadLetters = Interlocked.Exchange(ref _deadLetterCount, 0);
        var siteAuditWriteFailures = Interlocked.Exchange(ref _siteAuditWriteFailures, 0);
        var auditRedactionFailures = Interlocked.Exchange(ref _auditRedactionFailures, 0);

        // Snapshot current connection and tag resolution state
        var connectionStatuses = new Dictionary<string, ConnectionHealth>(_connectionStatuses);
        var tagResolution = new Dictionary<string, TagResolutionStatus>(_tagResolutionCounts);
        var connectionEndpoints = new Dictionary<string, string>(_connectionEndpoints);
        var tagQuality = new Dictionary<string, TagQualityCounts>(_tagQualityCounts);

        // Snapshot current S&F buffer depths
        var sfBufferDepths = new Dictionary<string, int>(_sfBufferDepths);

        // Determine node role from active/standby state
        var nodeRole = _isActiveNode ? "Active" : "Standby";

        return new SiteHealthReport(
            SiteId: siteId,
            SequenceNumber: 0, // Caller (HealthReportSender) assigns the sequence number
            ReportTimestamp: _timeProvider.GetUtcNow(),
            DataConnectionStatuses: connectionStatuses,
            TagResolutionCounts: tagResolution,
            ScriptErrorCount: scriptErrors,
            AlarmEvaluationErrorCount: alarmErrors,
            StoreAndForwardBufferDepths: sfBufferDepths,
            DeadLetterCount: deadLetters,
            DeployedInstanceCount: _deployedInstanceCount,
            EnabledInstanceCount: _enabledInstanceCount,
            DisabledInstanceCount: _disabledInstanceCount,
            NodeRole: nodeRole,
            NodeHostname: _nodeHostname,
            DataConnectionEndpoints: connectionEndpoints,
            DataConnectionTagQuality: tagQuality,
            ParkedMessageCount: Interlocked.CompareExchange(ref _parkedMessageCount, 0, 0),
            ClusterNodes: _clusterNodes?.ToList(),
            SiteAuditWriteFailures: siteAuditWriteFailures,
            AuditRedactionFailure: auditRedactionFailures,
            SiteAuditBacklog: _siteAuditBacklog,
            SiteEventLogWriteFailures: Interlocked.Read(ref _siteEventLogWriteFailures));
    }
}