using System.Text.Json;
using Confluent.Kafka;
using Prometheus;

/// <summary>
/// Prevents a single un-processable Kafka message from blocking a consumer
/// partition forever.  Permanent errors (malformed JSON, bad format) are
/// skipped immediately; transient errors are retried up to
/// <paramref name="maxRetries"/> times before the offset is committed and
/// the message is abandoned.
/// </summary>
public sealed class PoisonPillGuard
{
    private static readonly Counter PoisonPillsSkipped = Metrics.CreateCounter(
        "kafka_poison_pills_skipped_total",
        "Messages skipped as poison pills, labeled by consumer group and topic.",
        labelNames: new[] { "consumer_group", "topic" });

    private readonly string _consumerGroup;
    private readonly int _maxRetries;
    private readonly ILogger _logger;

    private (string Topic, int Partition, long Offset)? _lastFailedKey;
    private int _retryCount;

    public PoisonPillGuard(string consumerGroup, int maxRetries, ILogger logger)
    {
        _consumerGroup = consumerGroup;
        _maxRetries = maxRetries;
        _logger = logger;
    }

    public bool ShouldSkip(ConsumeResult<string, string> result, Exception ex)
    {
        if (IsPermanent(ex))
        {
            LogSkip(result, ex, "permanent");
            return true;
        }

        var key = (result.Topic, result.Partition.Value, result.Offset.Value);

        if (_lastFailedKey == key)
        {
            _retryCount++;
        }
        else
        {
            _lastFailedKey = key;
            _retryCount = 1;
        }

        if (_retryCount >= _maxRetries)
        {
            LogSkip(result, ex, $"transient after {_retryCount} retries");
            return true;
        }

        return false;
    }

    public void OnSuccess()
    {
        _lastFailedKey = null;
        _retryCount = 0;
    }

    private void LogSkip(ConsumeResult<string, string> result, Exception ex, string reason)
    {
        PoisonPillsSkipped.WithLabels(_consumerGroup, result.Topic).Inc();

        _logger.LogCritical(ex,
            "Poison pill skipped ({Reason}): consumer_group={Group} topic={Topic} " +
            "partition={Partition} offset={Offset} payload={Payload}",
            reason, _consumerGroup, result.Topic,
            result.Partition.Value, result.Offset.Value,
            Truncate(result.Message.Value, 2000));
    }

    private static bool IsPermanent(Exception ex) =>
        GetRoot(ex) is JsonException or FormatException or ArgumentNullException;

    private static Exception GetRoot(Exception ex)
    {
        while (ex.InnerException is not null)
            ex = ex.InnerException;
        return ex;
    }

    private static string? Truncate(string? value, int max) =>
        value is null ? null
        : value.Length <= max ? value
        : value[..max] + "…[truncated]";
}