ScadaBridge/src/ScadaLink.HealthMonitoring/HealthReportSender.cs

using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using ScadaLink.Commons.Messages.Health;
using ScadaLink.StoreAndForward;

namespace ScadaLink.HealthMonitoring;

/// <summary>
/// Periodically collects a SiteHealthReport and sends it to central via Akka remoting.
/// Sequence numbers are monotonic, starting at 1, and reset on service restart.
/// </summary>
public class HealthReportSender : BackgroundService
{
    private readonly ISiteHealthCollector _collector;
    private readonly IHealthReportTransport _transport;
    private readonly HealthMonitoringOptions _options;
    private readonly ILogger<HealthReportSender> _logger;
    private readonly string _siteId;
    private readonly StoreAndForwardStorage? _sfStorage;
    private readonly IClusterNodeProvider? _clusterNodeProvider;
    private long _sequenceNumber;

    public HealthReportSender(
        ISiteHealthCollector collector,
        IHealthReportTransport transport,
        IOptions<HealthMonitoringOptions> options,
        ILogger<HealthReportSender> logger,
        ISiteIdentityProvider siteIdentityProvider,
        StoreAndForwardStorage? sfStorage = null,
        IClusterNodeProvider? clusterNodeProvider = null)
    {
        _collector = collector;
        _transport = transport;
        _options = options.Value;
        _logger = logger;
        _siteId = siteIdentityProvider.SiteId;
        _sfStorage = sfStorage;
        _clusterNodeProvider = clusterNodeProvider;
    }

    /// <summary>
    /// Current sequence number (for testing).
    /// </summary>
    public long CurrentSequenceNumber => Interlocked.Read(ref _sequenceNumber);

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation(
            "Health report sender starting for site {SiteId}, interval {Interval}s",
            _siteId, _options.ReportInterval.TotalSeconds);

        using var timer = new PeriodicTimer(_options.ReportInterval);

        while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false))
        {
            try
            {
                // Only the active node (running the DeploymentManager singleton) sends health reports.
                // The standby node has no instance/connection data and would overwrite the active's report.
                if (!_collector.IsActiveNode)
                    continue;

                if (_clusterNodeProvider != null)
                {
                    try
                    {
                        _collector.SetClusterNodes(_clusterNodeProvider.GetClusterNodes());
                    }
                    catch { /* Non-fatal */ }
                }

                if (_sfStorage != null)
                {
                    try
                    {
                        var parkedCount = await _sfStorage.GetParkedMessageCountAsync();
                        _collector.SetParkedMessageCount(parkedCount);
                    }
                    catch { /* Non-fatal — parked count will be 0 */ }
                }

                var seq = Interlocked.Increment(ref _sequenceNumber);
                var report = _collector.CollectReport(_siteId);

                // Replace the placeholder sequence number with our monotonic one
                var reportWithSeq = report with { SequenceNumber = seq };

                _transport.Send(reportWithSeq);

                _logger.LogInformation("Sent health report #{Seq} for site {SiteId}", seq, _siteId);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Failed to send health report for site {SiteId}", _siteId);
                // Continue sending — don't let a single failure stop reporting
            }
        }
    }
}