vigilcare-clinical/VigilCareClinicalAPI/BackgroundServices/OutboxRelayService.cs

using Confluent.Kafka;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Options;

public class OutboxRelayService : BackgroundService
{
    private readonly IServiceProvider _services;
    private readonly KafkaOptions _options;
    private readonly ILogger<OutboxRelayService> _logger;
    private IProducer<string, string>? _producer;

    public OutboxRelayService(
        IServiceProvider services,
        IOptions<KafkaOptions> options,
        ILogger<OutboxRelayService> logger)
    {
        _services = services;
        _options = options.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();

        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
                FROM outbox_events
                WHERE processed_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 failed = false;

        foreach (var ev in events)
        {
            try
            {
                var result = await _producer!.ProduceAsync(
                    ev.Topic,
                    new Message<string, string>
                    {
                        // The partition key ensures all events for one encounter land on
                        // the same partition. The sepsis engine depends on this — if two
                        // observations for the same patient arrive on different partitions,
                        // a single-partition consumer misses one and SIRS evaluation is wrong.
                        Key = ev.PartitionKey ?? string.Empty,
                        Value = ev.Payload
                    },
                    ct);

                published.Add(ev.Id);

                _logger.LogDebug(
                    "Published {Topic} offset={Offset} partition={Partition} key={Key}",
                    ev.Topic, result.Offset.Value, result.Partition.Value, ev.PartitionKey);
            }
            catch (ProduceException<string, string> ex)
            {
                _logger.LogWarning(ex,
                    "Kafka produce failed for outbox event {Id} — leaving unprocessed for next cycle",
                    ev.Id);
                failed = true;
                break;  // stop processing this batch; retry from this row next cycle
            }
        }

        if (published.Count > 0)
        {
            var now = DateTimeOffset.UtcNow;
            foreach (var id in published)
            {
                var ev = events.First(e => e.Id == id);
                ev.ProcessedAt = now;
            }
            await db.SaveChangesAsync(ct);
        }

        await tx.CommitAsync(ct);

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

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