vigilcare-clinical/VigilCareClinicalAPI/BackgroundServices/OutboxRelayService.cs

using System.Text;
using Confluent.Kafka;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.Exceptions;

public class OutboxRelayService : BackgroundService
{
    private readonly IServiceProvider _services;
    private readonly KafkaOptions _options;
    private readonly RabbitMqOptions _rabbitOpts;
    private readonly ClinicalSyncOptions _syncOpts;
    private readonly ILogger<OutboxRelayService> _logger;
    private IProducer<string, string>? _producer;
    private IConnection? _rabbitConnection;
    private IModel? _rabbitChannel;
    private IBasicProperties? _rabbitProps;

    public OutboxRelayService(
        IServiceProvider services,
        IOptions<KafkaOptions> options,
        IOptions<RabbitMqOptions> rabbitOpts,
        IOptions<ClinicalSyncOptions> syncOpts,
        ILogger<OutboxRelayService> logger)
    {
        _services = services;
        _options = options.Value;
        _rabbitOpts = rabbitOpts.Value;
        _syncOpts = syncOpts.Value;
        _logger = logger;
    }

    public override Task StartAsync(CancellationToken cancellationToken)
    {
        _producer = new ProducerBuilder<string, string>(new ProducerConfig
        {
            BootstrapServers = _options.BootstrapServers,
            Acks = Acks.All,
            // Idempotent producer: the broker deduplicates in-flight retries using
            // the producer ID + sequence number. Prevents duplicates from network
            // timeouts that cause the client to retry a message the broker already accepted.
            EnableIdempotence = true,
            MessageSendMaxRetries = 3,
            RetryBackoffMs = 100
        }).Build();

        var factory = new ConnectionFactory
        {
            HostName = _rabbitOpts.Host,
            Port = _rabbitOpts.Port,
            UserName = _rabbitOpts.Username,
            Password = _rabbitOpts.Password,
            DispatchConsumersAsync = true,
        };
        _rabbitConnection = factory.CreateConnection("outbox-relay");
        _rabbitChannel = _rabbitConnection.CreateModel();
        _rabbitProps = _rabbitChannel.CreateBasicProperties();
        _rabbitProps.Persistent = true;

        return base.StartAsync(cancellationToken);
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("Outbox relay started. PollInterval={Interval}ms", _options.OutboxPollIntervalMs);

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                await ProcessBatchAsync(stoppingToken);
            }
            catch (Exception ex) when (ex is not OperationCanceledException)
            {
                _logger.LogError(ex, "Outbox relay error — will retry on next poll cycle");
            }

            await Task.Delay(_options.OutboxPollIntervalMs, stoppingToken);
        }
    }

    private async Task ProcessBatchAsync(CancellationToken ct)
    {
        using var scope = _services.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();

        // FOR UPDATE SKIP LOCKED: if a second relay instance is running concurrently,
        // it skips rows already locked by this instance. Both instances make progress
        // without blocking each other. A plain SELECT without locking would cause both
        // to read the same rows and publish duplicates.
        await using var tx = await db.Database.BeginTransactionAsync(ct);

        var events = await db.OutboxEvents
            .FromSqlRaw("""
                SELECT id, topic, payload, partition_key, created_at, processed_at,
                       retry_count, last_error, failed_at
                FROM outbox_events
                WHERE processed_at IS NULL AND failed_at IS NULL
                ORDER BY created_at ASC
                LIMIT {0}
                FOR UPDATE SKIP LOCKED
                """, _options.OutboxBatchSize)
            .ToListAsync(ct);

        if (events.Count == 0)
        {
            await tx.RollbackAsync(ct);
            return;
        }

        var published = new List<Guid>();
        var hadFailure = false;

        foreach (var ev in events)
        {
            try
            {
                if (ev.Topic == ClinicalSyncOptions.BatchReceivedOutboxTopic)
                {
                    _rabbitChannel!.BasicPublish(
                        exchange: _syncOpts.SyncExchange,
                        routingKey: _syncOpts.SyncBatchReceivedRoutingKey,
                        basicProperties: _rabbitProps,
                        body: Encoding.UTF8.GetBytes(ev.Payload));

                    _logger.LogDebug(
                        "Published sync batch to RabbitMQ exchange={Exchange} routingKey={RoutingKey}",
                        _syncOpts.SyncExchange, _syncOpts.SyncBatchReceivedRoutingKey);
                }
                else
                {
                    var result = await _producer!.ProduceAsync(
                        ev.Topic,
                        new Message<string, string>
                        {
                            Key = ev.PartitionKey ?? string.Empty,
                            Value = ev.Payload
                        },
                        ct);

                    _logger.LogDebug(
                        "Published {Topic} offset={Offset} partition={Partition} key={Key}",
                        ev.Topic, result.Offset.Value, result.Partition.Value, ev.PartitionKey);
                }

                published.Add(ev.Id);
                ev.ProcessedAt = DateTimeOffset.UtcNow;
            }
            catch (Exception ex) when (ex is ProduceException<string, string> or RabbitMQClientException)
            {
                ev.RetryCount++;
                ev.LastError = ex switch
                {
                    ProduceException<string, string> kex => kex.Error.Reason,
                    _ => ex.Message
                };

                if (ev.RetryCount >= _options.OutboxMaxRetries)
                {
                    ev.FailedAt = DateTimeOffset.UtcNow;
                    _logger.LogError(ex,
                        "Outbox event {Id} permanently failed after {Retries} retries — topic={Topic}",
                        ev.Id, ev.RetryCount, ev.Topic);
                }
                else
                {
                    _logger.LogWarning(ex,
                        "Outbox publish failed for event {Id} — retry {Retry}/{Max} topic={Topic}",
                        ev.Id, ev.RetryCount, _options.OutboxMaxRetries, ev.Topic);
                }

                hadFailure = true;
                break;
            }
        }

        await db.SaveChangesAsync(ct);
        await tx.CommitAsync(ct);

        if (published.Count > 0)
            _logger.LogInformation(
                "Outbox relay published {Count} events. Failed={Failed}",
                published.Count, hadFailure);
    }

    public override void Dispose()
    {
        _rabbitChannel?.Dispose();
        _rabbitConnection?.Dispose();
        _producer?.Dispose();
        base.Dispose();
    }
}