ScadaBridge/src/ZB.MOM.WW.ScadaBridge.Host/Actors/DeadLetterMonitorActor.cs

using Akka.Actor;
using Akka.Event;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;

namespace ZB.MOM.WW.ScadaBridge.Host.Actors;

/// <summary>
/// Subscribes to Akka.NET dead letter events, logs them, and tracks count
/// for health monitoring integration.
///
/// Warning logging is rate-limited to <see cref="MaxWarningsPerWindow"/> per
/// <see cref="WarningWindow"/>: a failover/undeploy storm can produce thousands
/// of dead letters, and one Warning per letter is an unbounded log flood
/// (review 01 [Low]). Once the per-window cap is hit, per-letter warnings are
/// suppressed and a single summary of the suppressed count is logged when the
/// window rolls over. The health-metric count is <b>never</b> throttled — every
/// dead letter is counted.
/// </summary>
public class DeadLetterMonitorActor : ReceiveActor, IWithTimers
{
    /// <summary>Maximum per-letter Warning logs emitted within one <see cref="WarningWindow"/>.</summary>
    public const int MaxWarningsPerWindow = 10;

    /// <summary>Rolling window over which <see cref="MaxWarningsPerWindow"/> applies.</summary>
    internal static readonly TimeSpan WarningWindow = TimeSpan.FromMinutes(1);

    private long _deadLetterCount;
    private int _warningsThisWindow;
    private long _suppressedThisWindow;
    private readonly ISiteHealthCollector? _healthCollector;

    /// <summary>Akka timer scheduler (assigned by the runtime via <see cref="IWithTimers"/>).</summary>
    public ITimerScheduler Timers { get; set; } = null!;

    /// <summary>
    /// Initializes the actor and registers dead-letter message handlers.
    /// </summary>
    /// <param name="logger">Logger for dead-letter events.</param>
    /// <param name="healthCollector">Optional health collector used to increment the dead-letter metric.</param>
    public DeadLetterMonitorActor(ILogger<DeadLetterMonitorActor> logger, ISiteHealthCollector? healthCollector = null)
    {
        _healthCollector = healthCollector;

        Receive<DeadLetter>(dl =>
        {
            // Metric counting is never throttled.
            _deadLetterCount++;
            _healthCollector?.IncrementDeadLetter();

            if (_warningsThisWindow < MaxWarningsPerWindow)
            {
                _warningsThisWindow++;
                logger.LogWarning(
                    "Dead letter: {MessageType} from {Sender} to {Recipient}",
                    dl.Message.GetType().Name,
                    dl.Sender,
                    dl.Recipient);

                if (_warningsThisWindow == MaxWarningsPerWindow)
                    logger.LogWarning(
                        "Dead-letter warning limit ({Limit}) reached — suppressing per-letter warnings for {Window}",
                        MaxWarningsPerWindow, WarningWindow);
            }
            else
            {
                _suppressedThisWindow++;
            }
        });

        Receive<WindowTick>(_ =>
        {
            if (_suppressedThisWindow > 0)
                logger.LogWarning(
                    "Suppressed {Count} dead-letter warnings in the last {Window}",
                    _suppressedThisWindow, WarningWindow);

            _warningsThisWindow = 0;
            _suppressedThisWindow = 0;
        });

        Receive<GetDeadLetterCount>(_ => Sender.Tell(new DeadLetterCountResponse(_deadLetterCount)));
    }

    /// <inheritdoc />
    protected override void PreStart()
    {
        Context.System.EventStream.Subscribe(Self, typeof(DeadLetter));
        Timers.StartPeriodicTimer("dl-window", WindowTick.Instance, WarningWindow);
    }

    /// <inheritdoc />
    protected override void PostStop()
    {
        Context.System.EventStream.Unsubscribe(Self, typeof(DeadLetter));
    }

    /// <summary>Periodic self-message that rolls the warning-rate-limit window.</summary>
    private sealed class WindowTick
    {
        public static readonly WindowTick Instance = new();
        private WindowTick() { }
    }
}

/// <summary>
/// Message to request the current dead letter count.
/// </summary>
public sealed class GetDeadLetterCount
{
    public static readonly GetDeadLetterCount Instance = new();
    private GetDeadLetterCount() { }
}

/// <summary>
/// Response containing the current dead letter count.
/// </summary>
public sealed record DeadLetterCountResponse(long Count);