using System.Text.Json;
using System.Threading.Channels;

namespace ZB.MOM.NatsNet.Server;

public sealed partial class NatsServer
{
    private Channel<bool>? _gcbKickCh;
    private const long DefaultMaxTotalCatchupOutBytes = 64L * 1024 * 1024;

    internal void JsClusteredConsumerRequest(
        ClientInfo clientInfo,
        Account account,
        string subject,
        string reply,
        byte[] requestMessage,
        string stream,
        ConsumerConfig config,
        ConsumerAction action,
        bool pedantic)
    {
        var (js, cluster) = GetJetStreamCluster();
        if (js == null || cluster == null)
            return;

        var response = new JsApiConsumerCreateResponse
        {
            Type = JsApiSubjects.JsApiConsumerCreateResponseType,
        };

        var engine = new JetStreamEngine(js);
        var (streamConfig, streamFound) = engine.ClusterStreamConfig(account.Name, stream);
        if (!streamFound)
        {
            response.Error = JsApiErrors.NewJSStreamNotFoundError();
            SendAPIErrResponse(clientInfo, account, subject, reply, string.Empty, JsonResponse(response));
            return;
        }

        var defaultsError = NatsConsumer.SetConsumerConfigDefaults(config, streamConfig, selectedLimits: null, pedantic);
        if (defaultsError != null)
        {
            response.Error = defaultsError;
            SendAPIErrResponse(clientInfo, account, subject, reply, string.Empty, JsonResponse(response));
            return;
        }

        var cfgError = NatsConsumer.CheckConsumerCfg(config, streamConfig, selectedLimits: null, isRecovering: false);
        if (cfgError != null)
        {
            response.Error = cfgError;
            SendAPIErrResponse(clientInfo, account, subject, reply, string.Empty, JsonResponse(response));
            return;
        }

        var streamAssignment = engine.StreamAssignmentOrInflight(account.Name, stream);
        if (streamAssignment == null)
        {
            response.Error = JsApiErrors.NewJSStreamNotFoundError();
            SendAPIErrResponse(clientInfo, account, subject, reply, string.Empty, JsonResponse(response));
            return;
        }

        var explicitName = !string.IsNullOrWhiteSpace(config.Name)
            ? config.Name
            : config.Durable;

        var existing = !string.IsNullOrWhiteSpace(explicitName)
            ? engine.ConsumerAssignmentOrInflight(account.Name, stream, explicitName)
            : null;

        ConsumerAssignment assignment;
        if (existing != null)
        {
            JetStreamVersioning.CopyConsumerMetadata(config, existing.Config);
            if (action == ConsumerAction.Create)
            {
                var existingConfig = existing.Config ?? new ConsumerConfig();
                if (JsonSerializer.Serialize(existingConfig) != JsonSerializer.Serialize(config))
                {
                    response.Error = JsApiErrors.NewJSConsumerAlreadyExistsError();
                    SendAPIErrResponse(clientInfo, account, subject, reply, string.Empty, JsonResponse(response));
                    return;
                }
            }

            assignment = existing.CopyGroup();
            assignment.Config = config;
            assignment.ConfigJson = JetStreamCluster.SerializeJsonElement(config);
            assignment.Subject = subject;
            assignment.Reply = reply;
            assignment.Client = clientInfo;
        }
        else
        {
            if (action == ConsumerAction.Update)
            {
                response.Error = JsApiErrors.NewJSConsumerDoesNotExistError();
                SendAPIErrResponse(clientInfo, account, subject, reply, string.Empty, JsonResponse(response));
                return;
            }

            JetStreamVersioning.SetStaticConsumerMetadata(config);
            var group = cluster.CreateGroupForConsumer(config, streamAssignment);
            if (group == null)
            {
                response.Error = JsApiErrors.NewJSInsufficientResourcesError();
                SendAPIErrResponse(clientInfo, account, subject, reply, string.Empty, JsonResponse(response));
                return;
            }

            group.SetPreferred(this);
            group.Cluster = streamAssignment.Group?.Cluster;

            var name = explicitName;
            if (!NatsConsumer.IsDurableConsumer(config))
            {
                if (string.IsNullOrWhiteSpace(name))
                {
                    do
                    {
                        name = NatsConsumer.CreateConsumerName();
                    } while (engine.ConsumerAssignmentOrInflight(account.Name, stream, name) != null);
                }
            }

            assignment = new ConsumerAssignment
            {
                Group = group,
                Stream = stream,
                Name = name,
                Config = config,
                ConfigJson = JetStreamCluster.SerializeJsonElement(config),
                Subject = subject,
                Reply = reply,
                Client = clientInfo,
                Created = DateTime.UtcNow,
            };
        }

        if (cluster.Meta != null)
        {
            cluster.Meta.Propose(JetStreamCluster.EncodeAddConsumerAssignment(assignment));
            cluster.TrackInflightConsumerProposal(account.Name, stream, assignment, deleted: false);
        }

        _ = requestMessage;
    }

    internal long GcbTotal()
    {
        _gcbMu.EnterReadLock();
        try
        {
            return _gcbOut;
        }
        finally
        {
            _gcbMu.ExitReadLock();
        }
    }

    internal bool GcbBelowMax()
    {
        _gcbMu.EnterReadLock();
        try
        {
            var limit = _gcbOutMax > 0 ? _gcbOutMax : DefaultMaxTotalCatchupOutBytes;
            return _gcbOut <= limit;
        }
        finally
        {
            _gcbMu.ExitReadLock();
        }
    }

    internal void GcbAdd(ref long localOutstandingBytes, long size)
    {
        _gcbMu.EnterWriteLock();
        try
        {
            localOutstandingBytes += size;
            _gcbOut += size;
            var limit = _gcbOutMax > 0 ? _gcbOutMax : DefaultMaxTotalCatchupOutBytes;
            if (_gcbOut >= limit && _gcbKickCh == null)
                _gcbKickCh = Channel.CreateBounded<bool>(1);
        }
        finally
        {
            _gcbMu.ExitWriteLock();
        }
    }

    internal void GcbSubLocked(ref long localOutstandingBytes, long size)
    {
        if (localOutstandingBytes == 0)
            return;

        localOutstandingBytes -= size;
        if (localOutstandingBytes < 0)
            localOutstandingBytes = 0;
        _gcbOut -= size;
        if (_gcbOut < 0)
            _gcbOut = 0;

        var limit = _gcbOutMax > 0 ? _gcbOutMax : DefaultMaxTotalCatchupOutBytes;
        if (_gcbKickCh != null && _gcbOut < limit)
        {
            _gcbKickCh.Writer.TryWrite(true);
            _gcbKickCh.Writer.TryComplete();
            _gcbKickCh = null;
        }
    }

    internal void GcbSub(ref long localOutstandingBytes, long size)
    {
        _gcbMu.EnterWriteLock();
        try
        {
            GcbSubLocked(ref localOutstandingBytes, size);
        }
        finally
        {
            _gcbMu.ExitWriteLock();
        }
    }

    internal void GcbSubLast(ref long localOutstandingBytes)
    {
        _gcbMu.EnterWriteLock();
        try
        {
            GcbSubLocked(ref localOutstandingBytes, localOutstandingBytes);
            localOutstandingBytes = 0;
        }
        finally
        {
            _gcbMu.ExitWriteLock();
        }
    }

    internal ChannelReader<bool>? CbKickChan()
    {
        _gcbMu.EnterReadLock();
        try
        {
            return _gcbKickCh?.Reader;
        }
        finally
        {
            _gcbMu.ExitReadLock();
        }
    }
}
