lmxopcua/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverResilienceStatusPublisherService.cs

using Akka.Actor;
using Akka.Cluster.Tools.PublishSubscribe;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
using ZB.MOM.WW.OtOpcUa.Core.Resilience;
using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;

namespace ZB.MOM.WW.OtOpcUa.Host.Drivers;

/// <summary>
///     The operator-facing reader for <see cref="DriverResilienceStatusTracker"/>. Runs on driver-role
///     nodes (registered next to the tracker it reads) and every <see cref="_interval"/> publishes the
///     full tracker snapshot — one <see cref="DriverResilienceStatusChanged"/> per <c>(instance, host)</c>
///     — to the <c>driver-resilience-status</c> DistributedPubSub topic. The AdminUI
///     <c>DriverResilienceStatusBridge</c> subscribes and feeds the resilience section of the driver
///     status panel. Mirrors the driver-health flow; publishing the full set every tick self-heals
///     late-subscribing admin nodes and gives the panel staleness detection for free.
/// </summary>
public sealed class DriverResilienceStatusPublisherService : BackgroundService
{
    private static readonly TimeSpan DefaultInterval = TimeSpan.FromSeconds(5);

    private readonly DriverResilienceStatusTracker _tracker;
    private readonly Func<ActorSystem> _actorSystemAccessor;
    private readonly ILogger<DriverResilienceStatusPublisherService> _logger;
    private readonly ITelemetryLocalHub _hub;
    private readonly TimeSpan _interval;

    /// <summary>Initializes a new instance of <see cref="DriverResilienceStatusPublisherService"/>.</summary>
    /// <param name="tracker">The process-singleton resilience-status tracker to read each tick.</param>
    /// <param name="actorSystemAccessor">Lazy accessor for the Akka <see cref="ActorSystem"/> whose DPS mediator publishes the snapshots.</param>
    /// <param name="logger">Logger for publish diagnostics.</param>
    /// <param name="hub">The node-local live-telemetry hub each published snapshot is also emitted into (Phase 5).</param>
    /// <param name="interval">Publish cadence; defaults to 5 s when null.</param>
    public DriverResilienceStatusPublisherService(
        DriverResilienceStatusTracker tracker,
        Func<ActorSystem> actorSystemAccessor,
        ILogger<DriverResilienceStatusPublisherService> logger,
        ITelemetryLocalHub hub,
        TimeSpan? interval = null)
    {
        _tracker = tracker;
        _actorSystemAccessor = actorSystemAccessor;
        _logger = logger;
        _hub = hub;
        _interval = interval ?? DefaultInterval;
    }

    /// <summary>
    ///     Map the tracker's current snapshot to one wire message per <c>(instance, host)</c> pair.
    ///     Pure + unit-testable; the timer/publish shell in <see cref="ExecuteAsync"/> calls this each tick.
    /// </summary>
    /// <param name="tracker">The tracker to read.</param>
    /// <param name="publishedUtc">The publish timestamp stamped on every emitted message.</param>
    /// <returns>One <see cref="DriverResilienceStatusChanged"/> per tracked pair (empty when the tracker is empty).</returns>
    public static IReadOnlyList<DriverResilienceStatusChanged> BuildMessages(
        DriverResilienceStatusTracker tracker, DateTime publishedUtc) =>
        tracker.Snapshot()
            .Select(entry => new DriverResilienceStatusChanged(
                entry.DriverInstanceId,
                entry.HostName,
                entry.Snapshot.IsBreakerOpen,
                entry.Snapshot.ConsecutiveFailures,
                entry.Snapshot.CurrentInFlight,
                entry.Snapshot.LastBreakerOpenUtc,
                entry.Snapshot.LastSampledUtc,
                publishedUtc))
            .ToList();

    /// <inheritdoc />
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        using var timer = new PeriodicTimer(_interval);
        while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false))
        {
            try
            {
                var messages = BuildMessages(_tracker, DateTime.UtcNow);
                if (messages.Count == 0)
                    continue;

                var mediator = DistributedPubSub.Get(_actorSystemAccessor()).Mediator;
                foreach (var message in messages)
                {
                    mediator.Tell(new Publish(DriverResilienceStatusChanged.TopicName, message));
                    // Phase 5: fan the same snapshot into the node-local live-telemetry hub (no-op until a
                    // gRPC client subscribes). The DPS publish above is unchanged — a strictly additive tap.
                    _hub.Emit(new TelemetryItem.Resilience(message));
                }
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception ex)
            {
                // A transient DPS / actor-system hiccup must never kill the publisher — the next tick retries.
                _logger.LogWarning(ex, "DriverResilienceStatusPublisherService: publish tick failed; retrying next interval");
            }
        }
    }
}