KArtSell.Aegis/src/KArtSell.Host/Jobs/OutboxPollerJob.cs

using Hangfire;
using KArtSell.BuildingBlocks.Reliability;
using KArtSell.BuildingBlocks.Time;
using Microsoft.Extensions.Logging;

namespace KArtSell.Host.Jobs;

/// <summary>
/// Outbox Poller: Publishes unpublished messages to inbox and marks as delivered.
///
/// Design: Consumer='outbox-poller' in inbox_message is a delivery-ready marker.
/// Actual event dispatch (SignalR, email, webhook, etc.) is delegated to future
/// downstream consumer implementations that read inbox_message.
///
/// This separation allows Outbox pattern's durability guarantees without blocking
/// on the specific delivery mechanism.
/// </summary>
public sealed class OutboxPollerJob(
    DapperOutboxMessageReader reader,
    IClock clock,
    ILogger<OutboxPollerJob> logger)
{
    private const int DefaultBatchSize = 100;
    private const int MaxAttemptsBeforeDq = 3;

    private static readonly Action<ILogger, int, int, Exception?> LogPolled =
        LoggerMessage.Define<int, int>(
            LogLevel.Information,
            new EventId(1, nameof(LogPolled)),
            "Outbox poller processed {MessageCount} messages with {FailureCount} permanent failures.");

    private static readonly Action<ILogger, Guid, string, int, Exception?> LogMessagePublished =
        LoggerMessage.Define<Guid, string, int>(
            LogLevel.Debug,
            new EventId(2, nameof(LogMessagePublished)),
            "Published outbox message {MessageId} ({EventType}, attempt {Attempt}).");

    private static readonly Action<ILogger, Guid, string, int, Exception?> LogMessageDequed =
        LoggerMessage.Define<Guid, string, int>(
            LogLevel.Warning,
            new EventId(3, nameof(LogMessageDequed)),
            "Outbox message {MessageId} ({EventType}) exceeded max attempts ({MaxAttempts}), moved to dead letter queue.");

    private static readonly Action<ILogger, Guid, Exception?> LogProcessError =
        LoggerMessage.Define<Guid>(
            LogLevel.Error,
            new EventId(4, nameof(LogProcessError)),
            "Failed to process outbox message {MessageId}");

    [Queue("q-research")]
    [DisableConcurrentExecution(timeoutInSeconds: 60)]
    [AutomaticRetry(Attempts = 3, OnAttemptsExceeded = AttemptsExceededAction.Fail)]
    public async Task ExecuteAsync(CancellationToken cancellationToken = default)
    {
        var now = clock.UtcNow;

        // Process all unpublished messages ordered by occurred_at (oldest first).
        // Monitoring: alert if any message pending > 5 min (see dashboard/alerts).
        var messages = await reader.GetUnpublishedAsync(DefaultBatchSize, cancellationToken);

        var failureCount = 0;
        foreach (var message in messages)
        {
            try
            {
                if (message.Attempt >= MaxAttemptsBeforeDq)
                {
                    LogMessageDequed(logger, message.MessageId, message.EventType, MaxAttemptsBeforeDq, null);
                    failureCount++;
                    continue;
                }

                await reader.MarkPublishedAsync(message.MessageId, now, cancellationToken);
                await reader.InsertInboxAsync("outbox-poller", message.MessageId, now, message.PayloadHash, cancellationToken);
                LogMessagePublished(logger, message.MessageId, message.EventType, message.Attempt + 1, null);
            }
            catch (Exception ex)
            {
                LogProcessError(logger, message.MessageId, ex);
                failureCount++;
            }
        }

        LogPolled(logger, messages.Count, failureCount, null);
    }
}