namespace ZB.MOM.NatsNet.Server;

internal sealed partial class NatsConsumer
{
    internal void DidNotDeliver(ulong sequence, string subject)
    {
        _mu.EnterWriteLock();
        try
        {
            DecDeliveryCount(sequence);
            if (IsPushMode())
                _hasLocalDeliveryInterest = false;
            else
                CreditWaitingRequest(subject);

            if (_state.Pending is { } pending && pending.ContainsKey(sequence) && !OnRedeliverQueue(sequence))
            {
                AddToRedeliverQueue(sequence);
                if (_waiting is { } waiting && !waiting.IsEmpty())
                    SignalNewMessages();
            }
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void AddToRedeliverQueue(params ulong[] sequences)
    {
        _mu.EnterWriteLock();
        try
        {
            foreach (var sequence in sequences)
            {
                _redeliveryQueue.Enqueue(sequence);
                _redeliveryIndex.Add(sequence);
            }
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal bool HasRedeliveries() => _redeliveryQueue.Count > 0;

    internal ulong GetNextToRedeliver()
    {
        _mu.EnterWriteLock();
        try
        {
            if (_redeliveryQueue.Count == 0)
                return 0;

            var sequence = _redeliveryQueue.Dequeue();
            _redeliveryIndex.Remove(sequence);
            return sequence;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal bool OnRedeliverQueue(ulong sequence) => _redeliveryIndex.Contains(sequence);

    internal bool RemoveFromRedeliverQueue(ulong sequence)
    {
        _mu.EnterWriteLock();
        try
        {
            if (!_redeliveryIndex.Remove(sequence))
                return false;

            if (_redeliveryQueue.Count == 0)
                return true;

            var retained = _redeliveryQueue.Where(s => s != sequence).ToArray();
            _redeliveryQueue.Clear();
            foreach (var s in retained)
                _redeliveryQueue.Enqueue(s);

            return true;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal int CheckPending()
    {
        _mu.EnterWriteLock();
        try
        {
            if (_state.Pending is not { Count: > 0 })
                return 0;

            var expired = 0;
            var cutoff = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() - (long)Math.Max(1, Config.AckWait.TotalMilliseconds);
            var toRedeliver = new List<ulong>();

            foreach (var (sequence, pending) in _state.Pending.ToArray())
            {
                if (pending.Timestamp < cutoff)
                {
                    toRedeliver.Add(sequence);
                    expired++;
                }
            }

            if (toRedeliver.Count > 0)
            {
                toRedeliver.Sort();
                AddToRedeliverQueue(toRedeliver.ToArray());
                SignalNewMessages();
            }

            return expired;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal ulong SeqFromReply(string reply)
    {
        var (_, deliverySequence, _) = AckReplyInfo(reply);
        return deliverySequence;
    }

    internal ulong StreamSeqFromReply(string reply)
    {
        var (streamSequence, _, _) = AckReplyInfo(reply);
        return streamSequence;
    }

    internal static long ParseAckReplyNum(string token)
    {
        if (string.IsNullOrWhiteSpace(token))
            return -1;

        long number = 0;
        foreach (var character in token)
        {
            if (!char.IsAsciiDigit(character))
                return -1;

            number = (number * 10) + (character - '0');
        }

        return number;
    }
}
