using System.Text;
using System.Text.Json;
using ZB.MOM.NatsNet.Server.Internal;

namespace ZB.MOM.NatsNet.Server;

internal sealed class JetStreamEngine(JetStream state)
{
    private readonly JetStream _state = state;
    private static readonly TimeSpan MinUsageUpdateWindow = TimeSpan.FromMilliseconds(250);
    private const string JsWillExtend = "will_extend";
    private const string JsNoExtend = "no_extend";
    private const string JsDomainApiTemplate = "$JS.{0}.API.>";
    private const string JsApiAccountPrefix = "$JS.API.ACCOUNT.";

    internal void SetStarted()
    {
        _state.Lock.EnterWriteLock();
        try
        {
            _state.Started = DateTime.UtcNow;
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal void ApiDispatch(ClientConnection c, Account acc, string subject, string reply, byte[] rawMessage)
    {
        // These meta/system directives are handled elsewhere.
        if (subject == JsApiSubjects.JsApiLeaderStepDown ||
            subject == JsApiSubjects.JsApiRemoveServer ||
            subject.StartsWith(JsApiAccountPrefix, StringComparison.Ordinal))
        {
            return;
        }

        if (_state.Server is not NatsServer server)
            return;

        var route = server.MatchJetStreamApiSubscription(subject);
        var (header, payload) = c.MsgParts(rawMessage);
        var requestBody = Encoding.UTF8.GetString(payload);

        if (NatsMessageHeaders.GetHeader(AccountEventConstants.ClientInfoHeader, header) is not { Length: > 0 })
        {
            var systemAccount = server.SystemAccount();
            if (!ReferenceEquals(systemAccount, acc))
                return;

            if (subject != JsApiSubjects.JsApiAccountInfo)
            {
                if (c.Kind is ClientKind.Client or ClientKind.Leaf)
                {
                    var response = new ApiResponse
                    {
                        Type = JsApiSubjects.JsApiSystemResponseType,
                        Error = JsApiErrors.NewJSNotEnabledForAccountError(),
                    };
                    server.SendAPIErrResponse(null, acc, subject, reply, requestBody, server.JsonResponse(response));
                }

                return;
            }
        }

        if (route is null)
        {
            if (c.Kind is ClientKind.Client or ClientKind.Leaf)
            {
                var (_, requestAcc, _, _, _) = server.GetRequestInfo(c, rawMessage);
                if (requestAcc is not null && !ReferenceEquals(requestAcc, server.SystemAccount()))
                {
                    var response = new ApiResponse
                    {
                        Type = JsApiSubjects.JsApiSystemResponseType,
                        Error = JsApiErrors.NewJSBadRequestError(),
                    };
                    server.SendAPIErrResponse(null, requestAcc, subject, reply, requestBody, server.JsonResponse(response));
                }
            }

            return;
        }

        Interlocked.Add(ref _state.ApiInflight, 1);

        var request = new JsApiRoutedRequest
        {
            Subscription = route,
            Account = acc,
            Subject = subject,
            Reply = reply,
            Message = rawMessage.ToArray(),
        };

        var (queued, dropped) = server.EnqueueJSApiRequest(_state, request);
        if (!queued)
        {
            if (dropped > 0)
            {
                server.PublishAdvisory(
                    server.SystemAccount(),
                    JsApiSubjects.JsAdvisoryApiLimitReached,
                    new { Type = "io.nats.jetstream.advisory.v1.api_limit_reached", Dropped = dropped, Time = DateTime.UtcNow });
            }

            route.Handler(c, acc, subject, reply, rawMessage);
            Interlocked.Add(ref _state.ApiInflight, -1);
        }
    }

    internal bool IsEnabled() => Interlocked.CompareExchange(ref _state.Disabled, 0, 0) == 0;

    internal void SetJetStreamStandAlone(bool isStandAlone)
    {
        _state.Lock.EnterWriteLock();
        try
        {
            _state.StandAlone = isStandAlone;
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal bool IsShuttingDown()
    {
        _state.Lock.EnterReadLock();
        try
        {
            return _state.ShuttingDown;
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal Exception? DisableJetStream(JsAccount? account)
    {
        if (account?.Account is not Account a)
            return new InvalidOperationException("jetstream not enabled for account");

        _state.Lock.EnterWriteLock();
        try
        {
            _state.Accounts.Remove(a.Name);
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }

        account.Delete();
        return null;
    }

    internal bool WouldExceedLimits(StorageType storageType, int size)
    {
        var total = storageType == StorageType.MemoryStorage
            ? Interlocked.Read(ref _state.MemUsed)
            : Interlocked.Read(ref _state.StoreUsed);
        var max = storageType == StorageType.MemoryStorage
            ? _state.Config.MaxMemory
            : _state.Config.MaxStore;
        return total + size > max;
    }

    internal bool LimitsExceeded(StorageType storageType) => WouldExceedLimits(storageType, 0);

    internal static string TierName(int replicas) => $"R{(replicas <= 0 ? 1 : replicas)}";

    internal static bool IsSameTier(StreamConfig cfgA, StreamConfig cfgB) =>
        cfgA.Replicas == cfgB.Replicas;

    internal static Dictionary<string, JetStreamAccountLimits> DiffCheckedLimits(
        Dictionary<string, JetStreamAccountLimits> a,
        Dictionary<string, JetStreamAccountLimits> b)
    {
        var diff = new Dictionary<string, JetStreamAccountLimits>(StringComparer.Ordinal);

        foreach (var (tier, oldLimit) in a)
        {
            b.TryGetValue(tier, out var newLimit);
            newLimit ??= new JetStreamAccountLimits();
            diff[tier] = new JetStreamAccountLimits
            {
                MaxMemory = newLimit.MaxMemory - oldLimit.MaxMemory,
                MaxStore = newLimit.MaxStore - oldLimit.MaxStore,
            };
        }

        foreach (var (tier, newLimit) in b)
        {
            if (a.ContainsKey(tier))
                continue;

            diff[tier] = new JetStreamAccountLimits
            {
                MaxMemory = newLimit.MaxMemory,
                MaxStore = newLimit.MaxStore,
            };
        }

        return diff;
    }

    internal static (ulong Mem, ulong Store) ReservedStorage(
        Dictionary<string, object?> streamAssignments,
        string tier)
    {
        ulong mem = 0;
        ulong store = 0;

        foreach (var assignment in streamAssignments.Values.OfType<StreamAssignmentView>())
        {
            var cfg = assignment.Config;
            if (!string.IsNullOrEmpty(tier) && !string.Equals(tier, TierName(cfg.Replicas), StringComparison.Ordinal))
                continue;
            if (cfg.MaxBytes <= 0)
                continue;

            if (cfg.Storage == StorageType.FileStorage)
                store += (ulong)cfg.MaxBytes;
            else if (cfg.Storage == StorageType.MemoryStorage)
                mem += (ulong)cfg.MaxBytes;
        }

        return (mem, store);
    }

    internal static bool ShouldSendUsageUpdate(DateTime lastUpdateUtc) =>
        DateTime.UtcNow - lastUpdateUtc >= MinUsageUpdateWindow;

    internal Exception? CheckAccountLimits(
        JetStreamAccountLimits selected,
        StreamConfig config,
        long currentReservation) =>
        CheckLimits(selected, config, checkServer: false, currentReservation, 0);

    internal Exception? CheckAllLimits(
        JetStreamAccountLimits selected,
        StreamConfig config,
        long currentReservation,
        long maxBytesOffset) =>
        CheckLimits(selected, config, checkServer: true, currentReservation, maxBytesOffset);

    internal Exception? CheckLimits(
        JetStreamAccountLimits selected,
        StreamConfig config,
        bool checkServer,
        long currentReservation,
        long maxBytesOffset)
    {
        if (config.MaxConsumers > 0 && selected.MaxConsumers > 0 && config.MaxConsumers > selected.MaxConsumers)
            return new InvalidOperationException("maximum consumers limit exceeded");

        return CheckBytesLimits(selected, config.MaxBytes, config.Storage, checkServer, currentReservation, maxBytesOffset);
    }

    internal Exception? CheckBytesLimits(
        JetStreamAccountLimits selected,
        long addBytes,
        StorageType storage,
        bool checkServer,
        long currentReservation,
        long maxBytesOffset)
    {
        if (addBytes < 0)
            addBytes = 1;

        var totalBytes = addBytes + maxBytesOffset;
        switch (storage)
        {
            case StorageType.MemoryStorage:
                if (selected.MaxMemory >= 0 && currentReservation + totalBytes > selected.MaxMemory)
                    return new InvalidOperationException("insufficient memory resources");
                if (checkServer && Interlocked.Read(ref _state.MemReserved) + totalBytes > _state.Config.MaxMemory)
                    return new InvalidOperationException("insufficient memory resources");
                break;

            case StorageType.FileStorage:
                if (selected.MaxStore >= 0 && currentReservation + totalBytes > selected.MaxStore)
                    return new InvalidOperationException("insufficient storage resources");
                if (checkServer && Interlocked.Read(ref _state.StoreReserved) + totalBytes > _state.Config.MaxStore)
                    return new InvalidOperationException("insufficient storage resources");
                break;
        }

        return null;
    }

    internal JsAccount? LookupAccount(Account? account)
    {
        if (account == null)
            return null;

        _state.Lock.EnterReadLock();
        try
        {
            _state.Accounts.TryGetValue(account.Name, out var jsa);
            return jsa;
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal JetStreamStats UsageStats()
    {
        var stats = new JetStreamStats();
        _state.Lock.EnterReadLock();
        try
        {
            stats.Accounts = _state.Accounts.Count;
            stats.ReservedMemory = (ulong)Math.Max(0, Interlocked.Read(ref _state.MemReserved));
            stats.ReservedStore = (ulong)Math.Max(0, Interlocked.Read(ref _state.StoreReserved));
            stats.Api.Level = JetStreamVersioning.JsApiLevel;
            stats.Api.Total = (ulong)Math.Max(0, Interlocked.Read(ref _state.ApiTotal));
            stats.Api.Errors = (ulong)Math.Max(0, Interlocked.Read(ref _state.ApiErrors));
            stats.Api.Inflight = (ulong)Math.Max(0, Interlocked.Read(ref _state.ApiInflight));
            stats.Memory = (ulong)Math.Max(0, Interlocked.Read(ref _state.MemUsed));
            stats.Store = (ulong)Math.Max(0, Interlocked.Read(ref _state.StoreUsed));
            stats.HAAssets = 0;
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }

        return stats;
    }

    internal Exception? SufficientResources(Dictionary<string, JetStreamAccountLimits>? limits)
    {
        if (limits == null || !_state.StandAlone)
            return null;

        static (long MaxMem, long MaxStore) Totals(Dictionary<string, JetStreamAccountLimits> source)
        {
            long mem = 0;
            long store = 0;
            foreach (var lim in source.Values)
            {
                if (lim.MaxMemory > 0) mem += lim.MaxMemory;
                if (lim.MaxStore > 0) store += lim.MaxStore;
            }

            return (mem, store);
        }

        var (totalMem, totalStore) = Totals(limits);
        if (Interlocked.Read(ref _state.MemReserved) + totalMem > _state.Config.MaxMemory)
            return new InvalidOperationException("insufficient memory resources");
        if (Interlocked.Read(ref _state.StoreReserved) + totalStore > _state.Config.MaxStore)
            return new InvalidOperationException("insufficient storage resources");

        long reservedMem = 0;
        long reservedStore = 0;
        _state.Lock.EnterReadLock();
        try
        {
            foreach (var jsa in _state.Accounts.Values)
            {
                jsa.UsageLock.EnterReadLock();
                try
                {
                    var (m, s) = Totals(jsa.Limits);
                    reservedMem += m;
                    reservedStore += s;
                }
                finally
                {
                    jsa.UsageLock.ExitReadLock();
                }
            }
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }

        if (reservedMem + totalMem > _state.Config.MaxMemory)
            return new InvalidOperationException("insufficient memory resources");
        if (reservedStore + totalStore > _state.Config.MaxStore)
            return new InvalidOperationException("insufficient storage resources");

        return null;
    }

    internal void ReserveStreamResources(StreamConfig? cfg)
    {
        if (cfg == null || cfg.MaxBytes <= 0)
            return;

        if (cfg.Storage == StorageType.MemoryStorage)
            Interlocked.Add(ref _state.MemReserved, cfg.MaxBytes);
        else if (cfg.Storage == StorageType.FileStorage)
            Interlocked.Add(ref _state.StoreReserved, cfg.MaxBytes);
    }

    internal void ReleaseStreamResources(StreamConfig? cfg)
    {
        if (cfg == null || cfg.MaxBytes <= 0)
            return;

        if (cfg.Storage == StorageType.MemoryStorage)
            Interlocked.Add(ref _state.MemReserved, -cfg.MaxBytes);
        else if (cfg.Storage == StorageType.FileStorage)
            Interlocked.Add(ref _state.StoreReserved, -cfg.MaxBytes);
    }

    internal bool IsStreamHealthy(Account? account, StreamAssignment? assignment)
    {
        if (assignment?.Unsupported != null)
            return true;
        if (_state.Cluster is not JetStreamCluster)
            return true;
        if (assignment?.Group == null || assignment.Config == null || account == null)
            return false;

        var (stream, _) = account.LookupStream(assignment.Config.Name);
        if (stream == null)
            return false;

        if (assignment.Config.Replicas <= 1)
            return true;

        var node = assignment.Group.Node;
        return node != null && node.Healthy();
    }

    internal bool IsConsumerHealthy(NatsStream? stream, string consumer, ConsumerAssignment? assignment)
    {
        if (assignment?.Unsupported != null)
            return true;
        if (_state.Cluster is not JetStreamCluster)
            return true;
        if (stream == null || assignment?.Group == null)
            return false;

        if (assignment.Group.Peers.Length <= 1)
            return true;

        return assignment.Group.Node?.Healthy() == true;
    }

    internal bool SubjectsOverlap(string account, IEnumerable<string> subjects, StreamAssignment? ownAssignment = null)
    {
        if (_state.Cluster is not JetStreamCluster cluster)
            return false;

        var targetSubjects = subjects.ToArray();
        if (!cluster.Streams.TryGetValue(account, out var accountAssignments))
            return false;

        foreach (var assignment in accountAssignments.Values)
        {
            if (ownAssignment != null &&
                assignment.Config?.Name != null &&
                assignment.Config.Name == ownAssignment.Config?.Name)
            {
                continue;
            }

            var existingSubjects = assignment.Config?.Subjects ?? [];
            foreach (var existing in existingSubjects)
            {
                foreach (var target in targetSubjects)
                {
                    if (Internal.DataStructures.SubscriptionIndex.SubjectsCollide(target, existing))
                        return true;
                }
            }
        }

        return false;
    }

    internal bool IsClustered() => _state.Cluster is JetStreamCluster;

    internal bool IsClusteredNoLock() => Interlocked.CompareExchange(ref _state.Clustered, 0, 0) == 1;

    internal Exception? SetupMetaGroup()
    {
        var server = Server();
        if (server == null)
            return new InvalidOperationException("jetstream server unavailable");

        _state.Lock.EnterWriteLock();
        try
        {
            if (_state.Cluster is JetStreamCluster)
            {
                Interlocked.Exchange(ref _state.Clustered, 1);
                return null;
            }

            var systemAccount = server.SystemAccount();
            if (systemAccount == null)
                return ServerErrors.ErrNoSysAccount;

            var storeDir = Path.Combine(_state.Config.StoreDir, systemAccount.Name, "_js_", "_meta_");
            Directory.CreateDirectory(storeDir);

            var raftConfig = new RaftConfig
            {
                Name = "_meta_",
                Store = storeDir,
                Recovering = true,
            };

            var (meta, error) = server.StartRaftNode(systemAccount.Name, raftConfig);
            if (error != null)
                return error;

            _state.Cluster = new JetStreamCluster
            {
                Meta = meta,
                Streams = new Dictionary<string, Dictionary<string, StreamAssignment>>(StringComparer.Ordinal),
                Server = server,
                Client = server.CreateInternalJetStreamClient(),
                Qch = System.Threading.Channels.Channel.CreateUnbounded<bool>(),
                Stopped = System.Threading.Channels.Channel.CreateUnbounded<bool>(),
            };

            _state.MetaRecovering = true;
            Interlocked.Exchange(ref _state.Clustered, 1);
            return null;
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal IRaftNode? GetMetaGroup()
    {
        _state.Lock.EnterReadLock();
        try
        {
            return (_state.Cluster as JetStreamCluster)?.Meta;
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal NatsServer? Server() => _state.Server as NatsServer;

    internal bool IsLeaderless()
    {
        var meta = GetMetaGroup();
        if (meta == null)
            return false;
        if (!meta.Leaderless())
            return false;

        return DateTime.UtcNow - meta.Created() > TimeSpan.FromSeconds(10);
    }

    internal bool IsGroupLeaderless(RaftGroup? group)
    {
        if (group?.Node == null)
            return false;

        var meta = GetMetaGroup();
        if (meta == null)
            return false;
        if (!group.Peers.Contains(meta.ID(), StringComparer.Ordinal))
            return false;
        if (!group.Node.Leaderless())
            return false;

        return DateTime.UtcNow - group.Node.Created() > TimeSpan.FromSeconds(10);
    }

    internal System.Threading.Channels.Channel<bool>? ClusterQuitC()
    {
        _state.Lock.EnterReadLock();
        try
        {
            return (_state.Cluster as JetStreamCluster)?.Qch;
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal System.Threading.Channels.Channel<bool>? ClusterStoppedC()
    {
        _state.Lock.EnterReadLock();
        try
        {
            return (_state.Cluster as JetStreamCluster)?.Stopped;
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal void SetMetaRecovering()
    {
        _state.Lock.EnterWriteLock();
        try
        {
            if (_state.Cluster is JetStreamCluster)
                _state.MetaRecovering = true;
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal void ClearMetaRecovering()
    {
        _state.Lock.EnterWriteLock();
        try
        {
            _state.MetaRecovering = false;
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal bool IsMetaRecovering()
    {
        _state.Lock.EnterReadLock();
        try
        {
            return _state.MetaRecovering;
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal void CheckForOrphans()
    {
        var server = Server();
        if (server == null)
            return;

        _state.Lock.EnterReadLock();
        var meta = (_state.Cluster as JetStreamCluster)?.Meta;
        _state.Lock.ExitReadLock();

        if (meta == null || meta.Leaderless() || !meta.Healthy())
            return;

        var (streams, consumers) = GetOrphans();
        foreach (var stream in streams)
            stream.Delete();
        foreach (var consumer in consumers)
            consumer.Delete();
    }

    internal (List<NatsStream> Streams, List<NatsConsumer> Consumers) GetOrphans()
    {
        var orphanStreams = new List<NatsStream>();
        var orphanConsumers = new List<NatsConsumer>();

        _state.Lock.EnterReadLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return (orphanStreams, orphanConsumers);

            foreach (var (accountName, jsa) in _state.Accounts)
            {
                cluster.Streams.TryGetValue(accountName, out var assignments);
                assignments ??= new Dictionary<string, StreamAssignment>(StringComparer.Ordinal);

                jsa.Lock.EnterReadLock();
                try
                {
                    foreach (var (streamName, streamValue) in jsa.Streams)
                    {
                        if (!assignments.ContainsKey(streamName) && streamValue is NatsStream stream)
                            orphanStreams.Add(stream);
                    }
                }
                finally
                {
                    jsa.Lock.ExitReadLock();
                }
            }
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }

        return (orphanStreams, orphanConsumers);
    }

    internal static string FriendlyBytes<T>(T bytes)
        where T : struct, IConvertible
    {
        var value = Convert.ToDouble(bytes);
        const int baseValue = 1024;
        var units = new[] { "K", "M", "G", "T", "P", "E" };

        if (value < baseValue)
            return $"{value} B";

        var exp = (int)(Math.Log(value) / Math.Log(baseValue));
        var index = Math.Clamp(exp - 1, 0, units.Length - 1);
        return $"{value / Math.Pow(baseValue, exp):0.00} {units[index]}B";
    }

    internal static bool IsValidName(string name)
    {
        if (string.IsNullOrWhiteSpace(name))
            return false;

        return name.IndexOfAny([' ', '\t', '\r', '\n', '\f', '.', '*', '>']) < 0;
    }

    internal static Exception? ValidateJetStreamOptions(ServerOptions options)
    {
        foreach (var (account, domain) in options.JsAccDefaultDomain)
        {
            var exists = false;
            if (ServerOptions.IsReservedAccount(account))
            {
                exists = true;
            }
            else if (options.TrustedOperators.Count == 0)
            {
                foreach (var configured in options.Accounts)
                {
                    if (!string.Equals(configured.GetName(), account, StringComparison.Ordinal))
                        continue;

                    if (configured.JetStreamLimits?.Count > 0 && !string.IsNullOrEmpty(domain))
                        return new InvalidOperationException($"default_js_domain contains account name \"{account}\" with enabled JetStream");

                    exists = true;
                    break;
                }
            }
            else
            {
                exists = IsLikelyPublicAccountNkey(account);
            }

            if (!exists)
                return new InvalidOperationException($"in non operator mode, `default_js_domain` references non existing account \"{account}\"");
        }

        foreach (var (account, domain) in options.JsAccDefaultDomain)
        {
            var systemAccount = string.IsNullOrWhiteSpace(options.SystemAccount)
                ? ServerConstants.DefaultSystemAccount
                : options.SystemAccount;

            if (string.Equals(account, systemAccount, StringComparison.Ordinal))
                return new InvalidOperationException($"system account \"{account}\" can not be in default_js_domain");

            if (string.IsNullOrWhiteSpace(domain))
                continue;

            var subject = string.Format(JsDomainApiTemplate, domain);
            if (!Internal.DataStructures.SubscriptionIndex.IsValidSubject(subject))
                return new InvalidOperationException($"default_js_domain contains account \"{account}\" with invalid domain name \"{domain}\"");
        }

        if (!string.IsNullOrWhiteSpace(options.JetStreamDomain))
        {
            var subject = string.Format(JsDomainApiTemplate, options.JetStreamDomain);
            if (!Internal.DataStructures.SubscriptionIndex.IsValidSubject(subject))
                return new InvalidOperationException($"invalid domain name: derived \"{subject}\" is not a valid subject");

            if (!IsValidName(options.JetStreamDomain))
                return new InvalidOperationException("invalid domain name: may not contain ., * or >");
        }

        if (!options.JetStream || options.Cluster.Port == 0)
            return null;
        if (string.IsNullOrWhiteSpace(options.ServerName))
            return new InvalidOperationException("jetstream cluster requires `server_name` to be set");
        if (string.IsNullOrWhiteSpace(options.Cluster.Name))
            return new InvalidOperationException("jetstream cluster requires `cluster.name` to be set");

        var hint = options.JetStreamExtHint.ToLowerInvariant();
        if (hint is not JsWillExtend and not JsNoExtend and not "")
            return new InvalidOperationException($"expected 'no_extend' for string value, got '{hint}'");
        options.JetStreamExtHint = hint;

        if (options.JetStreamMaxCatchup < 0)
            return new InvalidOperationException("jetstream max catchup cannot be negative");

        return null;
    }

    internal static void FixCfgMirrorWithDedupWindow(StreamConfig? config)
    {
        if (config?.Mirror == null)
            return;
        if (config.Duplicates != TimeSpan.Zero)
            config.Duplicates = TimeSpan.Zero;
    }

    private static bool IsLikelyPublicAccountNkey(string value) =>
        !string.IsNullOrWhiteSpace(value) &&
        value.Length >= 10 &&
        value.StartsWith("A", StringComparison.Ordinal);

    // -------------------------------------------------------------------------
    // JetStream cluster stream methods (Batch 33 Group A)
    // -------------------------------------------------------------------------

    internal void MonitorCluster()
    {
        var meta = GetMetaGroup();
        if (meta == null)
            return;

        SetMetaRecovering();
        try
        {
            CheckClusterSize();
            var (_, _, _, error) = MetaSnapshot();
            if (error != null)
                Server()?.Warnf("JetStream meta snapshot failed in MonitorCluster: {0}", error.Message);
        }
        finally
        {
            ClearMetaRecovering();
            ClusterStoppedC()?.Writer.TryWrite(true);
        }
    }

    internal void CheckClusterSize()
    {
        var server = Server();
        var meta = GetMetaGroup();
        if (server == null || meta == null)
            return;

        var activePeers = server.ActivePeers();
        if (activePeers.Count == 0)
            return;

        var jsPeers = 0;
        foreach (var peer in activePeers)
        {
            var info = server.GetNodeInfo(peer);
            if (info?.Js == true)
                jsPeers++;
        }

        if (jsPeers > 0 && jsPeers < meta.ClusterSize())
            meta.AdjustClusterSize(jsPeers);
    }

    internal (StreamConfig Config, bool Ok) ClusterStreamConfig(string accountName, string streamName)
    {
        _state.Lock.EnterReadLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return (new StreamConfig(), false);
            if (!cluster.Streams.TryGetValue(accountName, out var accountStreams))
                return (new StreamConfig(), false);
            if (!accountStreams.TryGetValue(streamName, out var assignment))
                return (new StreamConfig(), false);
            if (assignment.Config == null)
                return (new StreamConfig(), false);

            return (assignment.Config.Clone(), true);
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal ClusterInfo? OfflineClusterInfo(RaftGroup? group)
    {
        if (group == null || _state.Server is not NatsServer server)
            return null;

        var replicas = new List<PeerInfo>(group.Peers.Length);
        foreach (var peer in group.Peers)
        {
            var info = server.GetNodeInfo(peer);
            replicas.Add(new PeerInfo
            {
                Name = info?.Name ?? peer,
                Current = false,
                Offline = true,
                Active = TimeSpan.Zero,
                Lag = 0,
            });
        }

        return new ClusterInfo
        {
            Name = server.ClusterName(),
            Replicas = replicas.ToArray(),
        };
    }

    internal ClusterInfo? ClusterInfo(RaftGroup? group)
    {
        if (_state.Server is not NatsServer server)
            return null;

        _state.Lock.EnterReadLock();
        try
        {
            if (group?.Node == null)
            {
                return new ClusterInfo
                {
                    Name = server.CachedClusterName(),
                    Leader = server.ServerName(),
                };
            }

            var node = group.Node;
            var leader = server.ServerNameForNode(node.GroupLeader());
            var now = DateTime.UtcNow;
            var self = node.ID();
            var replicas = new List<PeerInfo>();

            foreach (var peer in node.Peers())
            {
                if (string.Equals(peer.Id, self, StringComparison.Ordinal))
                    continue;
                if (!group.IsMember(peer.Id))
                    continue;

                var info = server.GetNodeInfo(peer.Id);
                var active = peer.Last == default || now <= peer.Last ? TimeSpan.Zero : now - peer.Last;
                replicas.Add(new PeerInfo
                {
                    Name = info?.Name ?? peer.Id,
                    Current = peer.Current,
                    Offline = info?.Offline ?? true,
                    Active = active,
                    Lag = peer.Lag,
                });
            }

            return new ClusterInfo
            {
                Name = server.CachedClusterName(),
                Leader = leader,
                Replicas = replicas.OrderBy(r => r.Name, StringComparer.Ordinal).ToArray(),
            };
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal string[] StreamAlternates(ClientInfo clientInfo, string streamName)
    {
        if (_state.Server is not NatsServer server)
            return [];

        _state.Lock.EnterReadLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return [];

            var (account, _) = server.LookupAccount(clientInfo.ServiceAccount());
            if (account == null)
                return [];

            if (!cluster.Streams.TryGetValue(account.Name, out var accountStreams))
                return [];

            var weights = new Dictionary<string, int>(StringComparer.Ordinal);
            if (clientInfo.Cluster is { Length: > 0 })
            {
                for (var i = 0; i < clientInfo.Cluster.Length; i++)
                    weights[clientInfo.Cluster[i]] = clientInfo.Cluster.Length - i;
            }

            if (clientInfo.Alternates is { Count: > 0 })
            {
                for (var i = 0; i < clientInfo.Alternates.Count; i++)
                    weights[clientInfo.Alternates[i]] = clientInfo.Alternates.Count - i;
            }

            var candidates = new List<(string Name, string Cluster)>();
            foreach (var assignment in accountStreams.Values)
            {
                if (assignment.Unsupported != null || assignment.Config == null)
                    continue;

                if (string.Equals(assignment.Config.Name, streamName, StringComparison.Ordinal) ||
                    string.Equals(assignment.Config.Mirror?.Name, streamName, StringComparison.Ordinal))
                {
                    candidates.Add((assignment.Config.Name, assignment.Group?.Cluster ?? string.Empty));
                }
            }

            if (candidates.Count <= 1)
                return [];

            return candidates
                .OrderByDescending(c => weights.TryGetValue(c.Cluster, out var weight) ? weight : 0)
                .Select(c => c.Name)
                .Distinct(StringComparer.Ordinal)
                .ToArray();
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal (byte[] Snapshot, int Streams, int Consumers, Exception? Error) MetaSnapshot()
    {
        _state.Lock.EnterReadLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return ([], 0, 0, null);

            return EncodeMetaSnapshot(cluster.Streams);
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal Exception? ApplyMetaSnapshot(byte[] buffer, RecoveryUpdates? updates, bool isRecovering)
    {
        var (decoded, error) = DecodeMetaSnapshot(buffer);
        if (error != null || decoded == null)
            return error;

        _state.Lock.EnterWriteLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return null;

            foreach (var accountStreams in decoded.Values)
            {
                foreach (var assignment in accountStreams.Values)
                {
                    SetStreamAssignmentRecovering(assignment);
                    if (assignment.Consumers == null)
                        continue;

                    foreach (var consumer in assignment.Consumers.Values)
                    {
                        SetConsumerAssignmentRecovering(consumer);
                    }
                }
            }

            cluster.Streams = decoded;
            if (!isRecovering && updates != null)
            {
                updates.AddStreams.Clear();
                updates.UpdateStreams.Clear();
                updates.RemoveStreams.Clear();
                updates.UpdateConsumers.Clear();
                updates.RemoveConsumers.Clear();
            }
            return null;
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal (Dictionary<string, Dictionary<string, StreamAssignment>>? Streams, Exception? Error) DecodeMetaSnapshot(byte[] buffer)
    {
        if (buffer.Length == 0)
        {
            return (
                new Dictionary<string, Dictionary<string, StreamAssignment>>(StringComparer.Ordinal),
                null
            );
        }

        try
        {
            var writeable = System.Text.Json.JsonSerializer.Deserialize<List<WriteableStreamAssignment>>(buffer) ?? [];
            var streams = new Dictionary<string, Dictionary<string, StreamAssignment>>(StringComparer.Ordinal);
            foreach (var item in writeable)
            {
                var accountName = item.Client?.ServiceAccount() ?? string.Empty;
                if (!streams.TryGetValue(accountName, out var accountStreams))
                {
                    accountStreams = new Dictionary<string, StreamAssignment>(StringComparer.Ordinal);
                    streams[accountName] = accountStreams;
                }

                var assignment = new StreamAssignment
                {
                    Client = item.Client,
                    Created = item.Created,
                    ConfigJson = item.ConfigJson,
                    Group = item.Group,
                    Sync = item.Sync,
                    Config = item.ConfigJson.ValueKind == System.Text.Json.JsonValueKind.Undefined
                        ? null
                        : System.Text.Json.JsonSerializer.Deserialize<StreamConfig>(item.ConfigJson.GetRawText()),
                    Consumers = new Dictionary<string, ConsumerAssignment>(StringComparer.Ordinal),
                };

                if (item.Consumers.Count > 0)
                {
                    foreach (var consumer in item.Consumers)
                    {
                        var streamName = !string.IsNullOrWhiteSpace(consumer.Stream)
                            ? consumer.Stream
                            : assignment.Config?.Name ?? string.Empty;
                        var assignmentConsumer = new ConsumerAssignment
                        {
                            Client = consumer.Client,
                            Created = consumer.Created,
                            Name = consumer.Name,
                            Stream = streamName,
                            ConfigJson = consumer.ConfigJson,
                            Group = consumer.Group,
                            State = consumer.State,
                            Config = consumer.ConfigJson.ValueKind == System.Text.Json.JsonValueKind.Undefined
                                ? null
                                : System.Text.Json.JsonSerializer.Deserialize<ConsumerConfig>(consumer.ConfigJson.GetRawText()),
                        };
                        assignment.Consumers[assignmentConsumer.Name] = assignmentConsumer;
                    }
                }

                var streamKey = assignment.Config?.Name ?? string.Empty;
                accountStreams[streamKey] = assignment;
            }

            return (streams, null);
        }
        catch (Exception ex)
        {
            return (null, ex);
        }
    }

    internal (byte[] Snapshot, int Streams, int Consumers, Exception? Error) EncodeMetaSnapshot(
        Dictionary<string, Dictionary<string, StreamAssignment>> streams)
    {
        try
        {
            var streamCount = 0;
            var consumerCount = 0;
            var writeable = new List<WriteableStreamAssignment>();
            foreach (var accountStreams in streams.Values)
            {
                foreach (var assignment in accountStreams.Values)
                {
                    streamCount++;
                    var configJson = assignment.ConfigJson.ValueKind == System.Text.Json.JsonValueKind.Undefined &&
                                     assignment.Config != null
                        ? System.Text.Json.JsonSerializer.SerializeToElement(assignment.Config)
                        : assignment.ConfigJson;
                    var ws = new WriteableStreamAssignment
                    {
                        Client = assignment.Client,
                        Created = assignment.Created,
                        ConfigJson = configJson,
                        Group = assignment.Group,
                        Sync = assignment.Sync,
                    };

                    if (assignment.Consumers != null)
                    {
                        foreach (var consumer in assignment.Consumers.Values)
                        {
                            consumerCount++;
                            var consumerConfigJson =
                                consumer.ConfigJson.ValueKind == System.Text.Json.JsonValueKind.Undefined &&
                                consumer.Config != null
                                    ? System.Text.Json.JsonSerializer.SerializeToElement(consumer.Config)
                                    : consumer.ConfigJson;
                            ws.Consumers.Add(new WriteableConsumerAssignment
                            {
                                Client = consumer.Client,
                                Created = consumer.Created,
                                Name = consumer.Name,
                                Stream = consumer.Stream,
                                ConfigJson = consumerConfigJson,
                                Group = consumer.Group,
                                State = consumer.State,
                            });
                        }
                    }

                    writeable.Add(ws);
                }
            }

            if (writeable.Count == 0)
                return ([], 0, 0, null);

            var snapshot = System.Text.Json.JsonSerializer.SerializeToUtf8Bytes(writeable);
            return (snapshot, streamCount, consumerCount, null);
        }
        catch (Exception ex)
        {
            return ([], 0, 0, ex);
        }
    }

    internal Exception? CollectStreamAndConsumerChanges(
        IRaftNodeCheckpoint checkpoint,
        Dictionary<string, Dictionary<string, StreamAssignment>> streams)
    {
        try
        {
            foreach (var (appendEntry, error) in checkpoint.AppendEntriesSeq())
            {
                if (error != null)
                    return error;

                foreach (var entry in appendEntry.Entries)
                {
                    if (entry.Type != EntryType.EntryNormal || entry.Data.Length == 0)
                        continue;

                    var op = (EntryOp)entry.Data[0];
                    var payload = entry.Data.Length > 1 ? entry.Data.AsSpan(1).ToArray() : [];
                    switch (op)
                    {
                        case EntryOp.AssignStreamOp:
                        case EntryOp.UpdateStreamOp:
                        {
                            var sa = System.Text.Json.JsonSerializer.Deserialize<StreamAssignment>(payload);
                            if (sa?.Client == null || sa.Config == null)
                                break;
                            var account = sa.Client.ServiceAccount();
                            if (!streams.TryGetValue(account, out var accountStreams))
                            {
                                accountStreams = new Dictionary<string, StreamAssignment>(StringComparer.Ordinal);
                                streams[account] = accountStreams;
                            }

                            accountStreams[sa.Config.Name] = sa;
                            break;
                        }
                        case EntryOp.RemoveStreamOp:
                        {
                            var sa = System.Text.Json.JsonSerializer.Deserialize<StreamAssignment>(payload);
                            if (sa?.Client == null || sa.Config == null)
                                break;
                            if (streams.TryGetValue(sa.Client.ServiceAccount(), out var accountStreams))
                            {
                                accountStreams.Remove(sa.Config.Name);
                            }
                            break;
                        }
                        case EntryOp.AssignConsumerOp:
                        case EntryOp.AssignCompressedConsumerOp:
                        {
                            var ca = System.Text.Json.JsonSerializer.Deserialize<ConsumerAssignment>(payload);
                            if (ca?.Client == null || string.IsNullOrWhiteSpace(ca.Stream))
                                break;
                            var account = ca.Client.ServiceAccount();
                            if (!streams.TryGetValue(account, out var accountStreams))
                                break;
                            if (!accountStreams.TryGetValue(ca.Stream, out var streamAssignment))
                                break;
                            streamAssignment.Consumers ??= new Dictionary<string, ConsumerAssignment>(StringComparer.Ordinal);
                            streamAssignment.Consumers[ca.Name] = ca;
                            break;
                        }
                        case EntryOp.RemoveConsumerOp:
                        {
                            var ca = System.Text.Json.JsonSerializer.Deserialize<ConsumerAssignment>(payload);
                            if (ca?.Client == null || string.IsNullOrWhiteSpace(ca.Stream))
                                break;
                            var account = ca.Client.ServiceAccount();
                            if (!streams.TryGetValue(account, out var accountStreams))
                                break;
                            if (!accountStreams.TryGetValue(ca.Stream, out var streamAssignment))
                                break;
                            streamAssignment.Consumers?.Remove(ca.Name);
                            break;
                        }
                    }
                }
            }

            return null;
        }
        catch (Exception ex)
        {
            return ex;
        }
    }

    internal void SetStreamAssignmentRecovering(StreamAssignment assignment)
    {
        assignment.Responded = true;
        assignment.Recovering = true;
        assignment.Restore = null;
        if (assignment.Group != null)
        {
            assignment.Group.Preferred = string.Empty;
            assignment.Group.ScaleUp = false;
        }
    }

    internal void SetConsumerAssignmentRecovering(ConsumerAssignment assignment)
    {
        assignment.Responded = true;
        assignment.Recovering = true;
        if (assignment.Group != null)
        {
            assignment.Group.Preferred = string.Empty;
            assignment.Group.ScaleUp = false;
        }
    }

    internal void ProcessAddPeer(string peer)
    {
        _state.Lock.EnterWriteLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return;
            if (!cluster.IsLeader())
                return;

            foreach (var accountStreams in cluster.Streams.Values)
            {
                foreach (var assignment in accountStreams.Values)
                {
                    if (!assignment.MissingPeers())
                        continue;

                    var copy = assignment.CopyGroup();
                    copy.Group!.Peers = [.. copy.Group.Peers, peer];
                    assignment.Group = copy.Group;

                    if (assignment.Consumers == null)
                        continue;

                    foreach (var consumer in assignment.Consumers.Values)
                    {
                        if (consumer.Config?.Durable is { Length: > 0 } || (consumer.Group?.Peers.Length ?? 0) > 1)
                            consumer.Group!.Peers = assignment.Group.Peers;
                    }
                }
            }
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal void ProcessRemovePeer(string peer)
    {
        _state.Lock.EnterWriteLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return;

            foreach (var accountStreams in cluster.Streams.Values)
            {
                foreach (var assignment in accountStreams.Values)
                {
                    if (assignment.Group == null || !assignment.Group.IsMember(peer))
                        continue;

                    RemovePeerFromStreamLocked(assignment, peer);
                }
            }
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal bool RemovePeerFromStream(StreamAssignment assignment, string peer)
    {
        _state.Lock.EnterWriteLock();
        try
        {
            return RemovePeerFromStreamLocked(assignment, peer);
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal bool RemovePeerFromStreamLocked(StreamAssignment assignment, string peer)
    {
        if (assignment.Group == null || !assignment.Group.IsMember(peer))
            return false;

        assignment.Group.Peers = [.. assignment.Group.Peers.Where(p => !string.Equals(p, peer, StringComparison.Ordinal))];
        assignment.Group.Preferred = string.Empty;

        if (assignment.Consumers == null)
            return true;

        foreach (var consumer in assignment.Consumers.Values)
        {
            if (consumer.Group == null)
                continue;
            consumer.Group.Peers = [.. consumer.Group.Peers.Where(p => !string.Equals(p, peer, StringComparison.Ordinal))];
            consumer.Group.Preferred = string.Empty;
        }

        return true;
    }

    internal bool HasPeerEntries(IEnumerable<Entry> entries)
    {
        foreach (var entry in entries)
        {
            if (entry.Type == EntryType.EntryAddPeer || entry.Type == EntryType.EntryRemovePeer)
                return true;
        }
        return false;
    }

    internal (bool IsRecovering, bool DidSnapshot, Exception? Error) ApplyMetaEntries(
        IReadOnlyList<Entry> entries,
        RecoveryUpdates? updates)
    {
        var recovering = updates != null;
        var didSnapshot = false;

        try
        {
            foreach (var entry in entries)
            {
                if (entry.Type == EntryType.EntryCatchup)
                {
                    recovering = true;
                    break;
                }

                if (entry.Type == EntryType.EntrySnapshot)
                {
                    var error = ApplyMetaSnapshot(entry.Data, updates, recovering);
                    if (error != null)
                        return (recovering, didSnapshot, error);
                    didSnapshot = true;
                    continue;
                }

                if (entry.Type == EntryType.EntryRemovePeer)
                {
                    ProcessRemovePeer(System.Text.Encoding.ASCII.GetString(entry.Data));
                    continue;
                }

                if (entry.Type == EntryType.EntryAddPeer)
                {
                    ProcessAddPeer(System.Text.Encoding.ASCII.GetString(entry.Data));
                    continue;
                }

                if (entry.Type != EntryType.EntryNormal || entry.Data.Length == 0)
                    continue;

                var op = (EntryOp)entry.Data[0];
                var payload = entry.Data.Length > 1 ? entry.Data.AsSpan(1).ToArray() : [];
                switch (op)
                {
                    case EntryOp.AssignStreamOp:
                    {
                        var sa = System.Text.Json.JsonSerializer.Deserialize<StreamAssignment>(payload);
                        if (sa != null)
                            SetStreamAssignmentRecovering(sa);
                        break;
                    }
                    case EntryOp.AssignConsumerOp:
                    case EntryOp.AssignCompressedConsumerOp:
                    {
                        var ca = System.Text.Json.JsonSerializer.Deserialize<ConsumerAssignment>(payload);
                        if (ca != null)
                            SetConsumerAssignmentRecovering(ca);
                        break;
                    }
                }
            }

            return (recovering, didSnapshot, null);
        }
        catch (Exception ex)
        {
            return (recovering, didSnapshot, ex);
        }
    }

    internal (IRaftNode? Node, Exception? Error) CreateRaftGroup(
        string accountName,
        RaftGroup group,
        bool recovering,
        StorageType storage)
    {
        var server = Server();
        if (server == null)
            return (null, new InvalidOperationException("jetstream server unavailable"));
        if (group.Peers.Length <= 1)
            return (null, null);

        var meta = GetMetaGroup();
        if (meta == null || !group.IsMember(meta.ID()))
            return (null, null);

        var existing = server.LookupRaftNode(group.Name);
        if (existing != null && existing.State() != RaftState.Closed)
        {
            group.Node = existing;
            return (existing, null);
        }

        group.SetPreferred(server);
        var config = new RaftConfig
        {
            Name = group.Name,
            Recovering = recovering,
            ScaleUp = group.ScaleUp,
            Store = Path.Combine(_state.Config.StoreDir, accountName, "_js_", group.Name),
            Log = null,
        };

        var (node, error) = server.StartRaftNode(accountName, config);
        if (error != null || node == null)
            return (null, error);

        group.Node = node;
        if (!string.IsNullOrEmpty(group.Preferred) && node.ID() == group.Preferred && node.Term() == 0)
            node.CampaignImmediately();

        _ = storage;
        return (node, null);
    }

    internal static (
        string[] NewPeers,
        string[] OldPeers,
        Dictionary<string, bool> NewPeerSet,
        Dictionary<string, bool> OldPeerSet) GenPeerInfo(string[] peers, int split)
    {
        var safeSplit = Math.Clamp(split, 0, peers.Length);
        var oldPeers = peers.Take(safeSplit).ToArray();
        var newPeers = peers.Skip(safeSplit).ToArray();
        var oldSet = oldPeers.ToDictionary(p => p, _ => true, StringComparer.Ordinal);
        var newSet = newPeers.ToDictionary(p => p, _ => true, StringComparer.Ordinal);
        return (newPeers, oldPeers, newSet, oldSet);
    }

    internal void MonitorStream(NatsStream? stream, StreamAssignment assignment, bool sendSnapshot)
    {
        var node = assignment.Group?.Node;
        if (node == null)
            return;

        var isLeader = node.Leader();
        ProcessStreamLeaderChange(stream, isLeader);
        if (sendSnapshot && stream != null && isLeader)
        {
            var state = stream.State();
            var snap = System.Text.Json.JsonSerializer.SerializeToUtf8Bytes(state);
            node.SendSnapshot(snap);
        }
    }

    internal static bool IsControlHdr(byte[]? headers)
    {
        if (headers == null || headers.Length == 0)
            return false;

        var text = System.Text.Encoding.ASCII.GetString(headers);
        return text.Contains(NatsHeaderConstants.JsResponseType, StringComparison.OrdinalIgnoreCase);
    }

    internal (ulong MaxApplied, Exception? Error) ApplyStreamEntries(
        NatsStream? stream,
        CommittedEntry committed,
        bool isRecovering)
    {
        try
        {
            if (stream == null)
                return (0, null);

            ulong maxApplied = 0;
            foreach (var entry in committed.Entries)
            {
                if (entry.Type != EntryType.EntryNormal || entry.Data.Length == 0)
                    continue;

                var op = (EntryOp)entry.Data[0];
                if (op == EntryOp.StreamMsgOp || op == EntryOp.CompressedStreamMsgOp)
                {
                    var error = ApplyStreamMsgOp(stream, entry, isRecovering);
                    if (error != null)
                        return (maxApplied, error);
                    maxApplied = Math.Max(maxApplied, committed.Index);
                }
            }

            return (maxApplied, null);
        }
        catch (Exception ex)
        {
            return (0, ex);
        }
    }

    internal Exception? ApplyStreamMsgOp(NatsStream stream, Entry entry, bool isRecovering)
    {
        if (isRecovering && stream.SkipBatchIfRecovering())
            return null;

        var payloadLength = Math.Max(0, entry.Data.Length - 1);
        Interlocked.Increment(ref stream.Msgs);
        Interlocked.Add(ref stream.Bytes, payloadLength);
        return null;
    }

    internal void ProcessStreamLeaderChange(NatsStream? stream, bool isLeader)
    {
        if (stream == null)
            return;
        stream.SetLeader(isLeader, term: 0);
        if (!isLeader)
            stream.ResetClusteredState();
    }

    internal StreamAssignment? StreamAssignment(string accountName, string streamName)
    {
        _state.Lock.EnterReadLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return null;
            if (!cluster.Streams.TryGetValue(accountName, out var accountStreams))
                return null;
            return accountStreams.TryGetValue(streamName, out var assignment) ? assignment : null;
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal StreamAssignment? StreamAssignmentOrInflight(string accountName, string streamName)
    {
        _state.Lock.EnterReadLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return null;
            if (cluster.Streams.TryGetValue(accountName, out var accountStreams) &&
                accountStreams.TryGetValue(streamName, out var current))
                return current;

            if (cluster.InflightStreams.TryGetValue(accountName, out var inflight) &&
                inflight.TryGetValue(streamName, out var info))
                return info.Assignment;

            return null;
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal IEnumerable<StreamAssignment> StreamAssignmentsOrInflightSeq(string accountName)
    {
        _state.Lock.EnterReadLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return [];

            var seen = new HashSet<string>(StringComparer.Ordinal);
            var items = new List<StreamAssignment>();
            if (cluster.Streams.TryGetValue(accountName, out var accountStreams))
            {
                foreach (var (streamName, assignment) in accountStreams)
                {
                    seen.Add(streamName);
                    items.Add(assignment);
                }
            }

            if (cluster.InflightStreams.TryGetValue(accountName, out var inflightStreams))
            {
                foreach (var (streamName, inflight) in inflightStreams)
                {
                    if (seen.Contains(streamName) || inflight.Assignment == null)
                        continue;
                    items.Add(inflight.Assignment);
                }
            }

            return items;
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal IEnumerable<(string Account, StreamAssignment Assignment)> StreamAssignmentsOrInflightSeqAllAccounts()
    {
        _state.Lock.EnterReadLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return [];

            var results = new List<(string Account, StreamAssignment Assignment)>();
            foreach (var account in cluster.Streams.Keys.Union(cluster.InflightStreams.Keys, StringComparer.Ordinal))
            {
                var seen = new HashSet<string>(StringComparer.Ordinal);
                if (cluster.Streams.TryGetValue(account, out var accountStreams))
                {
                    foreach (var (streamName, assignment) in accountStreams)
                    {
                        seen.Add(streamName);
                        results.Add((account, assignment));
                    }
                }

                if (cluster.InflightStreams.TryGetValue(account, out var inflightStreams))
                {
                    foreach (var (streamName, inflight) in inflightStreams)
                    {
                        if (seen.Contains(streamName) || inflight.Assignment == null)
                            continue;
                        results.Add((account, inflight.Assignment));
                    }
                }
            }

            return results;
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal void ProcessStreamAssignment(StreamAssignment assignment)
    {
        _state.Lock.EnterWriteLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster || assignment.Client == null)
                return;

            var accountName = assignment.Client.ServiceAccount();
            if (!cluster.Streams.TryGetValue(accountName, out var accountStreams))
            {
                accountStreams = new Dictionary<string, StreamAssignment>(StringComparer.Ordinal);
                cluster.Streams[accountName] = accountStreams;
            }

            var streamName = assignment.Config?.Name ?? assignment.Subject ?? string.Empty;
            assignment.Responded = true;
            accountStreams[streamName] = assignment;
            cluster.RemoveInflightStreamProposal(accountName, streamName);
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal void ProcessUpdateStreamAssignment(StreamAssignment assignment)
    {
        ProcessStreamAssignment(assignment);
    }

    internal void ProcessClusterUpdateStream(StreamAssignment assignment)
    {
        ProcessUpdateStreamAssignment(assignment);
    }

    internal void ProcessClusterCreateStream(StreamAssignment assignment)
    {
        ProcessStreamAssignment(assignment);
    }

    internal void ProcessStreamRemoval(StreamAssignment assignment)
    {
        _state.Lock.EnterWriteLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster || assignment.Client == null)
                return;
            var accountName = assignment.Client.ServiceAccount();
            var streamName = assignment.Config?.Name ?? assignment.Subject ?? string.Empty;
            if (cluster.Streams.TryGetValue(accountName, out var accountStreams))
            {
                accountStreams.Remove(streamName);
                if (accountStreams.Count == 0)
                    cluster.Streams.Remove(accountName);
            }
            cluster.RemoveInflightStreamProposal(accountName, streamName);
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal void ProcessClusterDeleteStream(StreamAssignment assignment)
    {
        ProcessStreamRemoval(assignment);
    }

    internal void ProcessConsumerAssignment(ConsumerAssignment assignment)
    {
        _state.Lock.EnterWriteLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster || assignment.Client == null)
                return;

            var accountName = assignment.Client.ServiceAccount();
            if (!cluster.Streams.TryGetValue(accountName, out var accountStreams))
                return;
            if (!accountStreams.TryGetValue(assignment.Stream, out var streamAssignment))
                return;

            streamAssignment.Consumers ??= new Dictionary<string, ConsumerAssignment>(StringComparer.Ordinal);
            assignment.Responded = true;
            streamAssignment.Consumers[assignment.Name] = assignment;
            cluster.RemoveInflightConsumerProposal(accountName, assignment.Stream, assignment.Name);
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal void ProcessConsumerRemoval(ConsumerAssignment assignment)
    {
        _state.Lock.EnterWriteLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster || assignment.Client == null)
                return;

            var accountName = assignment.Client.ServiceAccount();
            if (cluster.Streams.TryGetValue(accountName, out var accountStreams) &&
                accountStreams.TryGetValue(assignment.Stream, out var streamAssignment))
            {
                streamAssignment.Consumers?.Remove(assignment.Name);
            }
            cluster.RemoveInflightConsumerProposal(accountName, assignment.Stream, assignment.Name);
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal void ProcessClusterCreateConsumer(ConsumerAssignment assignment)
    {
        ProcessConsumerAssignment(assignment);
    }

    internal void ProcessClusterDeleteConsumer(ConsumerAssignment assignment)
    {
        ProcessConsumerRemoval(assignment);
    }

    internal ConsumerAssignment? ConsumerAssignment(string accountName, string streamName, string consumerName)
    {
        _state.Lock.EnterReadLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return null;
            if (!cluster.Streams.TryGetValue(accountName, out var accountStreams))
                return null;
            if (!accountStreams.TryGetValue(streamName, out var streamAssignment))
                return null;
            if (streamAssignment.Consumers == null)
                return null;
            return streamAssignment.Consumers.TryGetValue(consumerName, out var assignment) ? assignment : null;
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal ConsumerAssignment? ConsumerAssignmentOrInflight(string accountName, string streamName, string consumerName)
    {
        _state.Lock.EnterReadLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return null;

            if (cluster.InflightConsumers.TryGetValue(accountName, out var streams) &&
                streams.TryGetValue(streamName, out var consumers) &&
                consumers.TryGetValue(consumerName, out var inflight))
            {
                return inflight.Deleted ? null : inflight.Assignment;
            }

            if (!cluster.Streams.TryGetValue(accountName, out var accountStreams))
                return null;
            if (!accountStreams.TryGetValue(streamName, out var streamAssignment))
                return null;
            if (streamAssignment.Consumers == null)
                return null;

            return streamAssignment.Consumers.TryGetValue(consumerName, out var current) ? current : null;
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal IEnumerable<ConsumerAssignment> ConsumerAssignmentsOrInflightSeq(string accountName, string streamName)
    {
        _state.Lock.EnterReadLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return [];

            var results = new List<ConsumerAssignment>();
            var seen = new HashSet<string>(StringComparer.Ordinal);

            if (cluster.InflightConsumers.TryGetValue(accountName, out var streams) &&
                streams.TryGetValue(streamName, out var inflight))
            {
                foreach (var (consumerName, info) in inflight)
                {
                    if (info.Deleted || info.Assignment == null)
                        continue;

                    seen.Add(consumerName);
                    results.Add(info.Assignment);
                }
            }

            if (cluster.Streams.TryGetValue(accountName, out var accountStreams) &&
                accountStreams.TryGetValue(streamName, out var streamAssignment) &&
                streamAssignment.Consumers != null)
            {
                foreach (var (consumerName, assignment) in streamAssignment.Consumers)
                {
                    if (!seen.Add(consumerName))
                        continue;
                    results.Add(assignment);
                }
            }

            return results;
        }
        finally
        {
            _state.Lock.ExitReadLock();
        }
    }

    internal void MonitorConsumer(NatsConsumer consumer, ConsumerAssignment assignment)
    {
        if (consumer == null || assignment == null)
            return;

        var server = _state.Server as NatsServer;
        if (consumer.RaftNode() == null || GetMetaGroup() == null)
        {
            server?.Warnf(
                "No RAFT group for consumer '{0}>{1}'",
                assignment.Stream,
                assignment.Name);
            return;
        }
    }

    internal Exception? ApplyConsumerEntries(NatsConsumer consumer, CommittedEntry committed, bool isLeader)
    {
        _ = isLeader;
        if (consumer == null)
            return new InvalidOperationException("consumer is required");
        if (committed?.Entries == null)
            return null;

        foreach (var entry in committed.Entries)
        {
            if (entry == null || entry.Data == null || entry.Data.Length == 0)
                continue;
            if (entry.Type == EntryType.EntryCatchup)
                continue;

            var op = (EntryOp)entry.Data[0];
            switch (op)
            {
                case EntryOp.UpdateAcksOp:
                    {
                        var (dseq, sseq, ok) = DecodeAckUpdate(entry.Data[1..]);
                        if (!ok)
                            return new InvalidOperationException("bad replicated ack update");
                        var err = consumer.ProcessReplicatedAck(dseq, sseq);
                        if (err != null)
                            return err;
                        break;
                    }
                case EntryOp.UpdateDeliveredOp:
                    {
                        var (_, sseq, _, _, ok) = DecodeDeliveredUpdate(entry.Data[1..]);
                        if (!ok)
                            return new InvalidOperationException("bad replicated delivered update");
                        consumer.SetDeliveredStreamSequence(sseq);
                        break;
                    }
            }
        }

        return null;
    }

    internal static (ulong Dseq, ulong Sseq, bool Ok) DecodeAckUpdate(byte[] buffer)
    {
        var span = buffer.AsSpan();
        if (!TryReadUVarInt(span, out var dseq, out var consumed))
            return (0, 0, false);
        if (!TryReadUVarInt(span[consumed..], out var sseq, out _))
            return (0, 0, false);
        return (dseq, sseq, true);
    }

    internal static (ulong Dseq, ulong Sseq, ulong DeliveryCount, long Timestamp, bool Ok) DecodeDeliveredUpdate(byte[] buffer)
    {
        var span = buffer.AsSpan();
        if (!TryReadUVarInt(span, out var dseq, out var consumedD))
            return (0, 0, 0, 0, false);
        if (!TryReadUVarInt(span[consumedD..], out var sseq, out var consumedS))
            return (0, 0, 0, 0, false);
        if (!TryReadUVarInt(span[(consumedD + consumedS)..], out var deliveryCount, out var consumedDc))
            return (0, 0, 0, 0, false);
        if (!TryReadVarInt(span[(consumedD + consumedS + consumedDc)..], out var ts, out _))
            return (0, 0, 0, 0, false);

        return (dseq, sseq, deliveryCount, ts, true);
    }

    internal Exception? ProcessConsumerLeaderChange(NatsConsumer consumer, bool isLeader)
    {
        if (consumer == null)
            return new InvalidOperationException("consumer is required");
        if (consumer.IsClosed())
        {
            if (isLeader)
                consumer.StepDownRaftNode();
            return new InvalidOperationException("failed to update consumer leader status");
        }

        consumer.SetLeader(isLeader, term: 0);
        if (!isLeader)
        {
            if (consumer.RaftNode() is { } node && node.LostQuorum())
            {
                (_state.Server as NatsServer)?.SendConsumerLostQuorumAdvisory(consumer);
            }
            return null;
        }

        (_state.Server as NatsServer)?.SendConsumerLeaderElectAdvisory(consumer);
        return null;
    }

    internal static bool IsInsufficientResourcesErr(ApiResponse? response)
    {
        if (response?.Error == null)
            return false;

        var errCode = response.Error.ErrCode;
        return errCode == JsApiErrors.InsufficientResources.ErrCode ||
               errCode == JsApiErrors.MemoryResourcesExceeded.ErrCode ||
               errCode == JsApiErrors.StorageResourcesExceeded.ErrCode;
    }

    internal void ProcessStreamAssignmentResults(object? sub, ClientConnection? client, Account? account, string subject, string reply, byte[] message)
    {
        _ = sub;
        _ = client;
        _ = account;
        _ = subject;
        _ = reply;

        StreamAssignmentResult? result;
        try
        {
            result = JsonSerializer.Deserialize<StreamAssignmentResult>(message);
        }
        catch
        {
            return;
        }

        if (result == null)
            return;

        _state.Lock.EnterWriteLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return;

            var assignment = StreamAssignmentOrInflight(result.Account, result.Stream);
            if (assignment == null || assignment.Reassigning)
                return;

            assignment.Responded = true;
            if (!result.Update && DateTime.UtcNow - assignment.Created < TimeSpan.FromSeconds(5))
            {
                assignment.Error = new InvalidOperationException(JsApiErrors.ClusterNotAssigned.Description ?? "cluster not assigned");
                cluster.TrackInflightStreamProposal(result.Account, assignment, deleted: true);
            }
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal void ProcessConsumerAssignmentResults(object? sub, ClientConnection? client, Account? account, string subject, string reply, byte[] message)
    {
        _ = sub;
        _ = client;
        _ = account;
        _ = subject;
        _ = reply;

        ConsumerAssignmentResult? result;
        try
        {
            result = JsonSerializer.Deserialize<ConsumerAssignmentResult>(message);
        }
        catch
        {
            return;
        }

        if (result == null)
            return;

        _state.Lock.EnterWriteLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return;
            if (!cluster.Streams.TryGetValue(result.Account, out var accountStreams))
                return;
            if (!accountStreams.TryGetValue(result.Stream, out var streamAssignment))
                return;
            if (streamAssignment.Consumers == null || !streamAssignment.Consumers.TryGetValue(result.Consumer, out var consumerAssignment))
                return;

            consumerAssignment.Responded = true;
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal void StartUpdatesSub()
    {
        _state.Lock.EnterWriteLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return;

            cluster.StreamResults ??= new object();
            cluster.ConsumerResults ??= new object();
            cluster.Stepdown ??= new object();
            cluster.PeerRemove ??= new object();
            cluster.PeerStreamMove ??= new object();
            cluster.PeerStreamCancelMove ??= new object();
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal void StopUpdatesSub()
    {
        _state.Lock.EnterWriteLock();
        try
        {
            if (_state.Cluster is not JetStreamCluster cluster)
                return;

            cluster.StreamResults = null;
            cluster.ConsumerResults = null;
            cluster.Stepdown = null;
            cluster.PeerRemove = null;
            cluster.PeerStreamMove = null;
            cluster.PeerStreamCancelMove = null;
        }
        finally
        {
            _state.Lock.ExitWriteLock();
        }
    }

    internal void ProcessLeaderChange(bool isLeader)
    {
        var server = _state.Server as NatsServer;
        if (server == null)
            return;

        if (isLeader)
        {
            server.Noticef("Self is new JetStream cluster metadata leader");
            server.SendDomainLeaderElectAdvisory();
            StartUpdatesSub();
        }
        else
        {
            server.Noticef("JetStream cluster metadata leadership changed");
            StopUpdatesSub();
        }
    }

    internal (int StreamCount, long Reservation) TieredStreamAndReservationCount(string accountName, string tier, StreamConfig config)
    {
        var streamCount = 0;
        long reservation = 0;
        foreach (var assignment in StreamAssignmentsOrInflightSeq(accountName))
        {
            var assignmentConfig = assignment.Config;
            if (assignmentConfig == null)
                continue;
            if (!string.IsNullOrEmpty(tier) && !IsSameTier(assignmentConfig, config))
                continue;
            if (string.Equals(assignmentConfig.Name, config.Name, StringComparison.Ordinal))
                continue;

            streamCount++;
            if (assignmentConfig.MaxBytes > 0 && assignmentConfig.Storage == config.Storage)
                reservation += assignmentConfig.MaxBytes;
        }

        return (streamCount, reservation);
    }

    internal (RaftGroup? Group, SelectPeerError? Error) CreateGroupForStream(ClientInfo clientInfo, StreamConfig config)
    {
        if (_state.Cluster is not JetStreamCluster cluster)
            return (null, new SelectPeerError { Misc = true });

        var replicas = Math.Max(1, config.Replicas);
        var targetCluster = config.Placement?.Cluster;
        if (string.IsNullOrWhiteSpace(targetCluster))
            targetCluster = clientInfo.Cluster?.FirstOrDefault();

        var (peers, error) = cluster.SelectPeerGroup(replicas, targetCluster ?? string.Empty, config, null, 0, null);
        if (peers == null || peers.Length < replicas)
            return (null, error ?? new SelectPeerError { Misc = true });

        var group = new RaftGroup
        {
            Name = JetStreamCluster.GroupNameForStream(peers, config.Storage),
            Storage = config.Storage,
            Cluster = targetCluster,
            Peers = peers,
        };
        group.SetPreferred(_state.Server as NatsServer ?? throw new InvalidOperationException("server not configured"));
        return (group, null);
    }

    internal JsApiError? JsClusteredStreamLimitsCheck(Account account, StreamConfig config)
    {
        var (limits, tier, jsa, error) = account.SelectLimits(config.Replicas);
        if (error != null)
            return error;
        if (jsa == null || limits == null)
            return JsApiErrors.NewJSNoLimitsError();

        var (streamCount, reservation) = TieredStreamAndReservationCount(account.Name, tier, config);
        if (limits.MaxStreams > 0 && streamCount >= limits.MaxStreams)
            return JsApiErrors.NewJSMaximumStreamsLimitError();

        var checkError = CheckAccountLimits(limits, config, reservation);
        return checkError == null ? null : JsApiErrors.NewJSStreamLimitsError(checkError);
    }

    private static bool TryReadUVarInt(ReadOnlySpan<byte> buffer, out ulong value, out int consumed)
    {
        value = 0;
        consumed = 0;
        var shift = 0;
        foreach (var b in buffer)
        {
            var chunk = (ulong)(b & 0x7Fu);
            value |= chunk << shift;
            consumed++;
            if ((b & 0x80) == 0)
                return true;
            shift += 7;
            if (shift >= 64)
                return false;
        }

        return false;
    }

    private static bool TryReadVarInt(ReadOnlySpan<byte> buffer, out long value, out int consumed)
    {
        value = 0;
        if (!TryReadUVarInt(buffer, out var raw, out consumed))
            return false;

        value = (long)((raw >> 1) ^ (~(raw & 1UL) + 1));
        return true;
    }
}

internal sealed class StreamAssignmentView
{
    public required StreamConfig Config { get; init; }
}
