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

public sealed class KafkaConsumerLagCollector : BackgroundService
{
    private static readonly string[] Groups =
    {
        "es-indexer",
        "sepsis-engine",
        "notification-publisher",
        "data-lake-writer",
    };

    private readonly KafkaOptions _kafkaOptions;
    private readonly ClinicalMetrics _metrics;
    private readonly ILogger<KafkaConsumerLagCollector> _logger;

    public KafkaConsumerLagCollector(
        IOptions<KafkaOptions> kafkaOptions,
        ClinicalMetrics metrics,
        ILogger<KafkaConsumerLagCollector> logger)
    {
        _kafkaOptions = kafkaOptions.Value;
        _metrics      = metrics;
        _logger       = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        // Initial delay so Kafka is reachable before the first collection.
        await Task.Delay(TimeSpan.FromSeconds(20), ct);

        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(30));
        while (await timer.WaitForNextTickAsync(ct))
        {
            foreach (var group in Groups)
            {
                try { await CollectGroupLagAsync(group, ct); }
                catch (Exception ex)
                {
                    _logger.LogDebug(ex, "KafkaConsumerLagCollector: failed for group {Group}", group);
                }
            }
        }
    }

    private async Task CollectGroupLagAsync(string groupId, CancellationToken ct)
    {
        var adminConfig = new AdminClientConfig
            { BootstrapServers = _kafkaOptions.BootstrapServers };

        using var admin = new AdminClientBuilder(adminConfig).Build();

        // List the committed offsets for this consumer group across all its partitions.
        var result = await admin.ListConsumerGroupOffsetsAsync(
            new[] { new ConsumerGroupTopicPartitions(groupId, null) },
            new ListConsumerGroupOffsetsOptions { RequireStableOffsets = false });

        var partitions = result.FirstOrDefault()?.Partitions ?? [];
        if (!partitions.Any())
        {
            _metrics.KafkaConsumerLag.WithLabels(groupId).Set(0);
            return;
        }

        // Query high watermarks using a temporary consumer (does not join the group).
        using var tempConsumer = new ConsumerBuilder<Ignore, Ignore>(new ConsumerConfig
        {
            BootstrapServers = _kafkaOptions.BootstrapServers,
            GroupId          = $"__lag-probe",
        }).Build();

        long totalLag = 0;
        foreach (var tpo in partitions)
        {
            if (tpo.Error.IsError || tpo.Offset == Offset.Unset) continue;

            var watermarks = tempConsumer.QueryWatermarkOffsets(
                tpo.TopicPartition, TimeSpan.FromSeconds(5));

            // Committed offset is the next offset the consumer will read.
            // High watermark is the latest available offset + 1.
            // Lag = messages the consumer has not yet read.
            var lag = watermarks.High.Value - tpo.Offset.Value;
            totalLag += Math.Max(0L, lag);
        }

        _metrics.KafkaConsumerLag.WithLabels(groupId).Set(totalLag);
    }
}