using System.Collections.Concurrent;
using System.Text;
using System.Text.Json;

namespace ZB.MOM.NatsNet.Server;

internal sealed partial class NatsConsumer
{
    internal static NextMsgReq NewNextMsgReq(string reply, byte[] message) => NextMsgReq.Rent(reply, message);

    internal void ProcessNextMsgReq(string reply, byte[] rawMessage)
    {
        if (string.IsNullOrWhiteSpace(reply))
            return;

        ArgumentNullException.ThrowIfNull(rawMessage);

        _mu.EnterWriteLock();
        try
        {
            if (_nextMsgReqs is null)
                return;

            _nextMsgReqs.Enqueue(NewNextMsgReq(reply, (byte[])rawMessage.Clone()));
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal (ulong Sequence, bool CanRespond, Exception? Error) ProcessResetReq(string reply, byte[] rawMessage)
    {
        if (string.IsNullOrWhiteSpace(reply))
            return (0, false, null);

        ArgumentNullException.ThrowIfNull(rawMessage);

        if (rawMessage.Length == 0)
            return ResetStartingSeq(0, reply);

        try
        {
            var request = JsonSerializer.Deserialize<JsApiConsumerResetRequest>(Encoding.UTF8.GetString(rawMessage));
            return ResetStartingSeq(request?.Seq ?? 0, reply);
        }
        catch (Exception ex)
        {
            return (0, false, ex);
        }
    }

    internal bool ProcessNextMsgRequest(string reply, byte[] message)
    {
        if (string.IsNullOrWhiteSpace(reply))
            return false;

        ArgumentNullException.ThrowIfNull(message);

        _mu.EnterWriteLock();
        try
        {
            if (IsPushMode() || _waiting is null)
                return false;

            var (request, error) = NextReqFromMsg(message);
            if (error is not null || request is null)
                return false;

            var batchSize = Math.Max(1, request.Batch);
            if (Config.MaxRequestBatch > 0 && batchSize > Config.MaxRequestBatch)
                return false;

            if (request.MaxBytes > 0 && Config.MaxRequestMaxBytes > 0 && request.MaxBytes > Config.MaxRequestMaxBytes)
                return false;

            if (request.Expires > TimeSpan.Zero && Config.MaxRequestExpires > TimeSpan.Zero && request.Expires > Config.MaxRequestExpires)
                return false;

            var expires = request.Expires > TimeSpan.Zero ? DateTime.UtcNow.Add(request.Expires) : (DateTime?)null;
            if (_waiting.IsFull(Config.MaxWaiting))
                return false;

            var waitingRequest = new WaitingRequest
            {
                Subject = reply,
                Reply = reply,
                N = batchSize,
                D = 0,
                NoWait = request.NoWait ? 1 : 0,
                Expires = expires,
                MaxBytes = Math.Max(0, request.MaxBytes),
                B = 0,
                PriorityGroup = request.Priority,
            };

            if (Config.PriorityPolicy == PriorityPolicy.PriorityPrioritized)
            {
                if (!_waiting.AddPrioritized(waitingRequest))
                    return false;
            }
            else
            {
                _waiting.Add(waitingRequest);
            }

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

internal sealed class NextMsgReq
{
    private static readonly ConcurrentBag<NextMsgReq> Pool = [];

    internal string Reply { get; private set; } = string.Empty;
    internal byte[] Message { get; private set; } = [];

    internal static NextMsgReq Rent(string reply, byte[] message)
    {
        ArgumentNullException.ThrowIfNull(message);
        if (!Pool.TryTake(out var request))
            request = new NextMsgReq();

        request.Reply = reply;
        request.Message = message;
        return request;
    }

    internal void ReturnToPool()
    {
        Reply = string.Empty;
        Message = [];
        Pool.Add(this);
    }
}
