using System.Text;

namespace ZB.MOM.NatsNet.Server;

internal sealed partial class NatsConsumer
{
    internal static void ConvertToHeadersOnly(JsPubMsg message)
    {
        ArgumentNullException.ThrowIfNull(message);

        var payloadSize = message.Msg?.Length ?? 0;
        var builder = new StringBuilder();
        builder.Append("NATS/1.0\r\n");
        builder.Append("Nats-Msg-Size: ");
        builder.Append(payloadSize);
        builder.Append("\r\n\r\n");

        message.Hdr = Encoding.ASCII.GetBytes(builder.ToString());
        message.Msg = [];
    }

    internal void DeliverMsg(string deliverySubject, string ackReply, JsPubMsg message, ulong deliveryCount, RetentionPolicy retentionPolicy)
    {
        ArgumentNullException.ThrowIfNull(message);

        _mu.EnterWriteLock();
        try
        {
            var nextDeliverySequence = _state.Delivered.Consumer + 1;
            var streamSequence = ParseAckReplyNum(ackReply.Split('.', StringSplitOptions.RemoveEmptyEntries).LastOrDefault() ?? string.Empty);
            var streamSeq = streamSequence > 0 ? (ulong)streamSequence : _state.Delivered.Stream + 1;

            _state.Delivered.Consumer = nextDeliverySequence;
            _state.Delivered.Stream = streamSeq;

            if (Config.AckPolicy is AckPolicy.AckExplicit or AckPolicy.AckAll)
                TrackPending(streamSeq, nextDeliverySequence);
            else
                _state.AckFloor = new SequencePair { Consumer = nextDeliverySequence, Stream = streamSeq };

            message.Subject = deliverySubject;
            message.Reply = ackReply;
            Interlocked.Add(ref _pendingBytes, message.Size());

            if (NeedFlowControl(message.Size()))
                SendFlowControl();

            if (retentionPolicy != RetentionPolicy.LimitsPolicy && Config.AckPolicy == AckPolicy.AckNone)
                _state.AckFloor = new SequencePair { Consumer = nextDeliverySequence, Stream = streamSeq };

            _ = deliveryCount;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal bool ReplicateDeliveries() =>
        Config.AckPolicy != AckPolicy.AckNone && Config.FlowControl == false && IsLeader();

    internal bool NeedFlowControl(int size)
    {
        if (_flowControlWindow <= 0)
            return false;

        if (string.IsNullOrEmpty(_flowControlReplyId) && _pendingBytes > _flowControlWindow / 2)
            return true;

        if (!string.IsNullOrEmpty(_flowControlReplyId) && _pendingBytes - _flowControlSentBytes >= _flowControlWindow)
            _flowControlSentBytes += size;

        return false;
    }

    internal void ProcessFlowControl(string subject)
    {
        _mu.EnterWriteLock();
        try
        {
            if (!string.Equals(subject, _flowControlReplyId, StringComparison.Ordinal))
                return;

            if (_flowControlWindow > 0 && _flowControlWindow < _maxPendingBytesLimit)
                _flowControlWindow = Math.Min(_flowControlWindow * 2, Math.Max(1, _maxPendingBytesLimit));

            _pendingBytes = Math.Max(0, _pendingBytes - _flowControlSentBytes);
            _flowControlSentBytes = 0;
            _flowControlReplyId = string.Empty;
            SignalNewMessages();
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal string FcReply() =>
        $"$JS.FC.{Stream}.{Name}.{Random.Shared.Next(1000, 9999)}";

    internal bool SendFlowControl()
    {
        _mu.EnterWriteLock();
        try
        {
            if (!IsPushMode())
                return false;

            var reply = FcReply();
            _flowControlReplyId = reply;
            _flowControlSentBytes = (int)Math.Max(0, _pendingBytes);
            return true;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void TrackPending(ulong streamSequence, ulong deliverySequence)
    {
        _state.Pending ??= [];
        if (_state.Pending.TryGetValue(streamSequence, out var pending))
        {
            pending.Timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
            _state.Pending[streamSequence] = pending;
        }
        else
        {
            _state.Pending[streamSequence] = new Pending
            {
                Sequence = deliverySequence,
                Timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
            };
        }
    }

    internal void CreditWaitingRequest(string reply)
    {
        _mu.EnterWriteLock();
        try
        {
            if (_waiting is null)
                return;

            foreach (var waitingRequest in _waiting.Snapshot())
            {
                if (!string.Equals(waitingRequest.Reply, reply, StringComparison.Ordinal))
                    continue;

                waitingRequest.N++;
                waitingRequest.D = Math.Max(0, waitingRequest.D - 1);
                return;
            }
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }
}
