// Copyright 2020-2026 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Adapted from server/jetstream_cluster.go in the NATS server Go source.

using System.Buffers.Binary;
using System.Text;
using System.Text.Json;
using System.Text.Json.Serialization;
using IronSnappy;

namespace ZB.MOM.NatsNet.Server;

// ============================================================================
// JetStreamCluster
// ============================================================================

/// <summary>
/// Holds cluster-level JetStream state on each server: the meta-controller Raft node,
/// stream/consumer assignment maps, and inflight proposal tracking.
/// Mirrors Go <c>jetStreamCluster</c> struct in server/jetstream_cluster.go lines 44-84.
/// </summary>
internal sealed class JetStreamCluster
{
    internal const string JscAllSubj = "$JSC.>";
    private static readonly Exception ErrBadStreamMsg = new("jetstream cluster bad replicated stream msg");
    private const int CompressThreshold = 8192;
    private const ulong MsgFlagFromSourceOrMirror = 1UL;
    private const int ReplySuffixLength = 10;
    private const string Base62 = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";

    /// <summary>The meta-controller Raft node.</summary>
    public IRaftNode? Meta { get; set; }

    /// <summary>
    /// All known stream assignments. Key: account name → stream name → assignment.
    /// </summary>
    public Dictionary<string, Dictionary<string, StreamAssignment>> Streams { get; set; } = new();

    /// <summary>
    /// Inflight stream create/update/delete proposals. Key: account → stream name.
    /// </summary>
    public Dictionary<string, Dictionary<string, InflightStreamInfo>> InflightStreams { get; set; } = new();

    /// <summary>
    /// Inflight consumer create/update/delete proposals. Key: account → stream → consumer.
    /// </summary>
    public Dictionary<string, Dictionary<string, Dictionary<string, InflightConsumerInfo>>> InflightConsumers { get; set; } = new();

    /// <summary>
    /// Peer-remove reply subjects pending quorum. Key: peer ID.
    /// </summary>
    public Dictionary<string, PeerRemoveInfo> PeerRemoveReply { get; set; } = new();

    /// <summary>Signals meta-leader should re-check stream assignments.</summary>
    public bool StreamsCheck { get; set; }

    /// <summary>Reference to the top-level server (object to avoid circular dep).</summary>
    public object? Server { get; set; }

    /// <summary>Internal client for cluster messaging (object to avoid circular dep).</summary>
    public object? Client { get; set; }

    /// <summary>Subscription that receives stream assignment results (object to avoid session dep).</summary>
    public object? StreamResults { get; set; }

    /// <summary>Subscription that receives consumer assignment results (object to avoid session dep).</summary>
    public object? ConsumerResults { get; set; }

    /// <summary>Subscription for meta-leader step-down requests.</summary>
    public object? Stepdown { get; set; }

    /// <summary>Subscription for peer-remove requests.</summary>
    public object? PeerRemove { get; set; }

    /// <summary>Subscription for stream-move requests.</summary>
    public object? PeerStreamMove { get; set; }

    /// <summary>Subscription for stream-move cancellation.</summary>
    public object? PeerStreamCancelMove { get; set; }

    /// <summary>Channel used to pop out monitorCluster before the raft layer.</summary>
    public System.Threading.Channels.Channel<bool>? Qch { get; set; }

    /// <summary>Notifies that monitorCluster has actually stopped.</summary>
    public System.Threading.Channels.Channel<bool>? Stopped { get; set; }

    /// <summary>Last meta-snapshot time (Unix nanoseconds).</summary>
    public long LastMetaSnapTime { get; set; }

    /// <summary>Duration of last meta-snapshot (nanoseconds).</summary>
    public long LastMetaSnapDuration { get; set; }

    internal static UnsupportedStreamAssignment NewUnsupportedStreamAssignment(
        NatsServer server,
        StreamAssignment assignment,
        Exception? error)
    {
        var reason = "stopped";
        if (error != null)
        {
            var msg = error.Message;
            reason = msg.StartsWith("json:", StringComparison.Ordinal)
                ? $"unsupported - config error: {msg["json:".Length..].TrimStart()}"
                : $"stopped - {msg}";
        }
        else if (assignment.Config != null && !JetStreamVersioning.SupportsRequiredApiLevel(assignment.Config.Metadata))
        {
            var required = JetStreamVersioning.GetRequiredApiLevel(assignment.Config.Metadata);
            if (!string.IsNullOrEmpty(required))
                reason = $"unsupported - required API level: {required}, current API level: {JetStreamVersioning.JsApiLevel}";
        }

        var config = assignment.Config ?? new StreamConfig { Name = assignment.Subject ?? string.Empty };
        return new UnsupportedStreamAssignment
        {
            Reason = reason,
            Info = new StreamInfo
            {
                Created = assignment.Created,
                Config = JetStreamVersioning.SetDynamicStreamMetadata(config),
            },
        };
    }

    internal static UnsupportedConsumerAssignment NewUnsupportedConsumerAssignment(
        ConsumerAssignment assignment,
        Exception? error)
    {
        var reason = "stopped";
        if (error != null)
        {
            var msg = error.Message;
            reason = msg.StartsWith("json:", StringComparison.Ordinal)
                ? $"unsupported - config error: {msg["json:".Length..].TrimStart()}"
                : $"stopped - {msg}";
        }
        else if (assignment.Config != null && !JetStreamVersioning.SupportsRequiredApiLevel(assignment.Config.Metadata))
        {
            var required = JetStreamVersioning.GetRequiredApiLevel(assignment.Config.Metadata);
            if (!string.IsNullOrEmpty(required))
                reason = $"unsupported - required API level: {required}, current API level: {JetStreamVersioning.JsApiLevel}";
        }

        var config = assignment.Config ?? new ConsumerConfig { Name = assignment.Name };
        return new UnsupportedConsumerAssignment
        {
            Reason = reason,
            Info = new ConsumerInfo
            {
                Stream = assignment.Stream,
                Name = assignment.Name,
                Created = assignment.Created,
                Config = JetStreamVersioning.SetDynamicConsumerMetadata(config),
                TimeStamp = DateTime.UtcNow,
            },
        };
    }

    internal bool IsLeader()
    {
        if (this == null)
            return true;

        return Meta != null && Meta.Leader();
    }

    internal bool IsStreamCurrent(string account, string stream)
    {
        if (Meta == null)
            return false;

        if (!Streams.TryGetValue(account, out var accountAssignments))
            return false;
        if (!accountAssignments.TryGetValue(stream, out var assignment))
            return false;
        if (assignment.Group == null)
            return false;

        return assignment.Group.Node == null || assignment.Group.Node.Current();
    }

    internal bool IsStreamAssigned(Account account, string stream)
    {
        if (Meta == null)
            return false;

        if (!Streams.TryGetValue(account.Name, out var accountAssignments))
            return false;
        if (!accountAssignments.TryGetValue(stream, out var assignment))
            return false;

        var peers = assignment.Group?.Peers ?? [];
        return peers.Contains(Meta.ID(), StringComparer.Ordinal);
    }

    internal bool IsConsumerAssigned(Account account, string stream, string consumer)
    {
        if (Meta == null)
            return false;

        if (!Streams.TryGetValue(account.Name, out var accountAssignments))
            return false;
        if (!accountAssignments.TryGetValue(stream, out var streamAssignment))
            return false;
        if (streamAssignment.Consumers == null || !streamAssignment.Consumers.TryGetValue(consumer, out var assignment))
            return false;

        var group = assignment.Group;
        if (group == null)
            return false;

        return group.IsMember(Meta.ID());
    }

    internal bool IsStreamLeader(string account, string stream)
    {
        if (Meta == null)
            return false;

        if (!Streams.TryGetValue(account, out var accountAssignments))
            return false;
        if (!accountAssignments.TryGetValue(stream, out var assignment))
            return false;

        var group = assignment.Group;
        if (group == null)
            return false;

        var ourId = Meta.ID();
        if (!group.Peers.Contains(ourId, StringComparer.Ordinal))
            return false;

        return group.Peers.Length == 1 || (group.Node != null && group.Node.Leader());
    }

    internal bool IsConsumerLeader(string account, string stream, string consumer)
    {
        if (Meta == null)
            return false;

        if (!Streams.TryGetValue(account, out var accountAssignments))
            return false;
        if (!accountAssignments.TryGetValue(stream, out var assignment))
            return false;
        if (assignment.Consumers == null || !assignment.Consumers.TryGetValue(consumer, out var consumerAssignment))
            return false;

        var group = consumerAssignment.Group;
        if (group == null)
            return false;

        var ourId = Meta.ID();
        if (!group.Peers.Contains(ourId, StringComparer.Ordinal))
            return false;

        return group.Peers.Length == 1 || (group.Node != null && group.Node.Leader());
    }

    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 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;
    }

    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 bool RemapStreamAssignment(StreamAssignment assignment, string removePeer)
    {
        if (assignment?.Group == null)
            return false;

        var retain = assignment.Group.Peers.Where(p => !string.Equals(p, removePeer, StringComparison.Ordinal)).ToArray();
        var (newPeers, error) = SelectPeerGroup(
            assignment.Group.Peers.Length,
            assignment.Group.Cluster ?? string.Empty,
            assignment.Config ?? new StreamConfig(),
            retain,
            0,
            [removePeer]);

        if (error == null && newPeers is { Length: > 0 })
        {
            assignment.Group.Peers = newPeers;
            assignment.Group.Preferred = string.Empty;
            return true;
        }

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

        assignment.Group.Peers = retain;
        assignment.Group.Preferred = string.Empty;
        return false;
    }

    internal (string[]? Peers, SelectPeerError? Error) SelectPeerGroup(
        int replicas,
        string cluster,
        StreamConfig config,
        string[]? existing,
        int replaceFirstExisting,
        string[]? ignore)
    {
        _ = config;
        if (replicas <= 0 || string.IsNullOrWhiteSpace(cluster) || Meta == null)
            return (null, new SelectPeerError { Misc = true });

        var selected = new List<string>(replicas);
        if (existing != null)
        {
            foreach (var peer in existing.Skip(Math.Clamp(replaceFirstExisting, 0, existing.Length)))
            {
                if (selected.Count == replicas)
                    break;
                selected.Add(peer);
            }
        }

        var ignored = new HashSet<string>(ignore ?? [], StringComparer.Ordinal);
        foreach (var peer in Meta.Peers())
        {
            if (selected.Count == replicas)
                break;
            if (ignored.Contains(peer.Id))
                continue;
            if (selected.Contains(peer.Id, StringComparer.Ordinal))
                continue;
            selected.Add(peer.Id);
        }

        if (selected.Count < replicas)
            return (null, new SelectPeerError { Offline = true });

        return (selected.Take(replicas).ToArray(), null);
    }

    internal static string GroupNameForStream(string[] peers, StorageType storage) =>
        GroupName("S", peers, storage);

    internal static string GroupNameForConsumer(string[] peers, StorageType storage) =>
        GroupName("C", peers, storage);

    internal static string GroupName(string prefix, string[] peers, StorageType storage)
    {
        var marker = storage == StorageType.MemoryStorage ? "M" : "F";
        var suffix = Guid.NewGuid().ToString("N")[..6];
        return $"{prefix}-R{Math.Max(1, peers.Length)}{marker}-{suffix}";
    }

    internal static (T? Response, Exception? Error) SysRequest<T>(NatsServer server, string subjectFormat, params object[] args)
    {
        _ = server;
        _ = string.Format(subjectFormat, args);
        return (default, null);
    }

    internal static byte[] EncodeStreamPurge(StreamPurge purge)
    {
        var payload = JsonSerializer.SerializeToUtf8Bytes(purge);
        var result = new byte[payload.Length + 1];
        result[0] = (byte)EntryOp.PurgeStreamOp;
        payload.CopyTo(result.AsSpan(1));
        return result;
    }

    internal static (StreamPurge? Purge, Exception? Error) DecodeStreamPurge(byte[] buffer)
    {
        try
        {
            return (JsonSerializer.Deserialize<StreamPurge>(buffer), null);
        }
        catch (Exception ex)
        {
            return (null, ex);
        }
    }

    internal static byte[] EncodeMsgDelete(StreamMsgDelete deleteRequest)
    {
        var payload = JsonSerializer.SerializeToUtf8Bytes(deleteRequest);
        var result = new byte[payload.Length + 1];
        result[0] = (byte)EntryOp.DeleteMsgOp;
        payload.CopyTo(result.AsSpan(1));
        return result;
    }

    internal static (StreamMsgDelete? Delete, Exception? Error) DecodeMsgDelete(byte[] buffer)
    {
        try
        {
            return (JsonSerializer.Deserialize<StreamMsgDelete>(buffer), null);
        }
        catch (Exception ex)
        {
            return (null, ex);
        }
    }

    internal static byte[] EncodeAddStreamAssignment(StreamAssignment assignment) =>
        EncodeStreamAssignmentWithOp(assignment, EntryOp.AssignStreamOp);

    internal static byte[] EncodeUpdateStreamAssignment(StreamAssignment assignment) =>
        EncodeStreamAssignmentWithOp(assignment, EntryOp.UpdateStreamOp);

    internal static byte[] EncodeDeleteStreamAssignment(StreamAssignment assignment) =>
        EncodeStreamAssignmentWithOp(assignment, EntryOp.RemoveStreamOp);

    internal static (StreamAssignment? Assignment, Exception? Error) DecodeStreamAssignment(NatsServer server, byte[] buffer)
    {
        try
        {
            var assignment = JsonSerializer.Deserialize<StreamAssignment>(buffer);
            if (assignment == null)
                return (null, new InvalidOperationException("invalid assignment payload"));

            var error = DecodeStreamAssignmentConfig(server, assignment);
            return error == null ? (assignment, null) : (null, error);
        }
        catch (Exception ex)
        {
            return (null, ex);
        }
    }

    internal static Exception? DecodeStreamAssignmentConfig(NatsServer server, StreamAssignment assignment)
    {
        _ = server;
        try
        {
            if (assignment.ConfigJson.ValueKind == JsonValueKind.Undefined ||
                assignment.ConfigJson.ValueKind == JsonValueKind.Null)
            {
                assignment.Config ??= new StreamConfig();
                return null;
            }

            var cfg = JsonSerializer.Deserialize<StreamConfig>(assignment.ConfigJson.GetRawText());
            assignment.Config = cfg ?? new StreamConfig();
            return null;
        }
        catch (Exception ex)
        {
            assignment.Unsupported = NewUnsupportedStreamAssignment(server, assignment, ex);
            return ex;
        }
    }

    private static byte[] EncodeStreamAssignmentWithOp(StreamAssignment assignment, EntryOp op)
    {
        var copy = assignment.CopyGroup();
        if (copy.Config != null)
            copy.ConfigJson = JsonSerializer.SerializeToElement(copy.Config);

        var payload = JsonSerializer.SerializeToUtf8Bytes(copy);
        var result = new byte[payload.Length + 1];
        result[0] = (byte)op;
        payload.CopyTo(result.AsSpan(1));
        return result;
    }

    internal void TrackInflightStreamProposal(string accountName, StreamAssignment assignment, bool deleted)
    {
        if (!InflightStreams.TryGetValue(accountName, out var streams))
        {
            streams = new Dictionary<string, InflightStreamInfo>(StringComparer.Ordinal);
            InflightStreams[accountName] = streams;
        }

        var streamName = assignment.Config?.Name ?? assignment.Subject ?? string.Empty;
        if (streams.TryGetValue(streamName, out var inflight))
        {
            inflight.Ops++;
            inflight.Deleted = deleted;
            inflight.Assignment = assignment;
            return;
        }

        streams[streamName] = new InflightStreamInfo
        {
            Ops = 1,
            Deleted = deleted,
            Assignment = assignment,
        };
    }

    internal void RemoveInflightStreamProposal(string accountName, string streamName)
    {
        if (!InflightStreams.TryGetValue(accountName, out var streams))
            return;
        if (!streams.TryGetValue(streamName, out var inflight))
            return;

        if (inflight.Ops > 1)
        {
            inflight.Ops--;
            return;
        }

        streams.Remove(streamName);
        if (streams.Count == 0)
            InflightStreams.Remove(accountName);
    }

    internal void TrackInflightConsumerProposal(
        string accountName,
        string streamName,
        ConsumerAssignment assignment,
        bool deleted)
    {
        if (!InflightConsumers.TryGetValue(accountName, out var streams))
        {
            streams = new Dictionary<string, Dictionary<string, InflightConsumerInfo>>(StringComparer.Ordinal);
            InflightConsumers[accountName] = streams;
        }

        if (!streams.TryGetValue(streamName, out var consumers))
        {
            consumers = new Dictionary<string, InflightConsumerInfo>(StringComparer.Ordinal);
            streams[streamName] = consumers;
        }

        if (consumers.TryGetValue(assignment.Name, out var inflight))
        {
            inflight.Ops++;
            inflight.Deleted = deleted;
            inflight.Assignment = assignment;
            return;
        }

        consumers[assignment.Name] = new InflightConsumerInfo
        {
            Ops = 1,
            Deleted = deleted,
            Assignment = assignment,
        };
    }

    internal void RemoveInflightConsumerProposal(string accountName, string streamName, string consumerName)
    {
        if (!InflightConsumers.TryGetValue(accountName, out var streams))
            return;
        if (!streams.TryGetValue(streamName, out var consumers))
            return;
        if (!consumers.TryGetValue(consumerName, out var inflight))
            return;

        if (inflight.Ops > 1)
        {
            inflight.Ops--;
            return;
        }

        consumers.Remove(consumerName);
        if (consumers.Count == 0)
            streams.Remove(streamName);
        if (streams.Count == 0)
            InflightConsumers.Remove(accountName);
    }

    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 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 static byte[] EncodeDeleteRange(DeleteRange deleteRange)
    {
        var payload = JsonSerializer.SerializeToUtf8Bytes(deleteRange);
        var encoded = new byte[payload.Length + 1];
        encoded[0] = (byte)EntryOp.DeleteRangeOp;
        Buffer.BlockCopy(payload, 0, encoded, 1, payload.Length);
        return encoded;
    }

    internal static (DeleteRange? DeleteRange, Exception? Error) DecodeDeleteRange(ReadOnlySpan<byte> payload)
    {
        try
        {
            var deleteRange = JsonSerializer.Deserialize<DeleteRange>(payload);
            return deleteRange == null
                ? (null, new InvalidOperationException("delete range payload did not deserialize"))
                : (deleteRange, null);
        }
        catch (Exception ex)
        {
            return (null, ex);
        }
    }

    internal RaftGroup? CreateGroupForConsumer(ConsumerConfig config, StreamAssignment streamAssignment)
    {
        var group = streamAssignment.Group;
        if (group == null || group.Peers.Length == 0)
            return null;

        var replicas = config.Replicas > 0
            ? config.Replicas
            : Math.Max(1, streamAssignment.Config?.Replicas ?? 1);
        if (replicas > group.Peers.Length)
            return null;

        var peers = group.Peers.ToArray();
        var active = new List<string>(peers.Length);
        if (Server is NatsServer server)
        {
            foreach (var peer in peers)
            {
                if (server.GetNodeInfo(peer) is { Offline: false })
                    active.Add(peer);
            }
        }
        else
        {
            active.AddRange(peers);
        }

        var quorum = replicas / 2 + 1;
        if (active.Count < quorum)
            return null;

        if (replicas > 0 && replicas < peers.Length)
        {
            if (active.Count < replicas)
                return null;

            ShuffleInPlace(active);
            peers = active.Take(replicas).ToArray();
        }

        var storage = config.MemoryStorage ? StorageType.MemoryStorage : streamAssignment.Config?.Storage ?? group.Storage;
        return new RaftGroup
        {
            Name = GroupNameForConsumer(peers, storage),
            Storage = storage,
            Peers = peers,
            Cluster = group.Cluster,
        };
    }

    internal static byte[] EncodeAddConsumerAssignment(ConsumerAssignment assignment)
    {
        var serializable = CloneForProposal(assignment);
        var payload = JsonSerializer.SerializeToUtf8Bytes(serializable);
        return PrependOp((byte)EntryOp.AssignConsumerOp, payload);
    }

    internal static byte[] EncodeDeleteConsumerAssignment(ConsumerAssignment assignment)
    {
        var serializable = CloneForProposal(assignment);
        var payload = JsonSerializer.SerializeToUtf8Bytes(serializable);
        return PrependOp((byte)EntryOp.RemoveConsumerOp, payload);
    }

    internal static (ConsumerAssignment? Assignment, Exception? Error) DecodeConsumerAssignment(ReadOnlySpan<byte> payload)
    {
        try
        {
            var assignment = JsonSerializer.Deserialize<ConsumerAssignment>(payload);
            if (assignment == null)
                return (null, new InvalidOperationException("consumer assignment payload did not deserialize"));

            var decodeError = DecodeConsumerAssignmentConfig(assignment);
            return decodeError != null ? (null, decodeError) : (assignment, null);
        }
        catch (Exception ex)
        {
            return (null, ex);
        }
    }

    internal static Exception? DecodeConsumerAssignmentConfig(ConsumerAssignment assignment)
    {
        if (assignment.ConfigJson.ValueKind is JsonValueKind.Undefined or JsonValueKind.Null)
            return new InvalidOperationException("consumer assignment config payload is missing");

        Exception? strictError = null;
        ConsumerConfig? config;
        try
        {
            var strictOptions = new JsonSerializerOptions { UnmappedMemberHandling = JsonUnmappedMemberHandling.Disallow };
            config = assignment.ConfigJson.Deserialize<ConsumerConfig>(strictOptions);
        }
        catch (Exception ex)
        {
            strictError = ex;
            config = null;
        }

        if (config == null)
        {
            try
            {
                config = assignment.ConfigJson.Deserialize<ConsumerConfig>();
            }
            catch (Exception ex)
            {
                return ex;
            }
        }

        assignment.Config = config ?? new ConsumerConfig();
        if (strictError != null || !JetStreamVersioning.SupportsRequiredApiLevel(assignment.Config.Metadata))
            assignment.Unsupported = NewUnsupportedConsumerAssignment(assignment, strictError);

        return null;
    }

    internal static byte[] EncodeAddConsumerAssignmentCompressed(ConsumerAssignment assignment)
    {
        var serializable = CloneForProposal(assignment);
        var payload = JsonSerializer.SerializeToUtf8Bytes(serializable);
        var compressed = Snappy.Encode(payload);
        return PrependOp((byte)EntryOp.AssignCompressedConsumerOp, compressed);
    }

    internal static (ConsumerAssignment? Assignment, Exception? Error) DecodeConsumerAssignmentCompressed(ReadOnlySpan<byte> payload)
    {
        try
        {
            var decoded = Snappy.Decode(payload);
            return DecodeConsumerAssignment(decoded);
        }
        catch (Exception ex)
        {
            return (null, ex);
        }
    }

    internal static (string Subject, string Reply, byte[]? Header, byte[]? Message, ulong Sequence, long Timestamp, bool Sourced, Exception? Error)
        DecodeStreamMsg(ReadOnlySpan<byte> payload)
    {
        if (payload.Length < 26)
            return (string.Empty, string.Empty, null, null, 0, 0, false, ErrBadStreamMsg);

        var index = 0;
        var sequence = BinaryPrimitives.ReadUInt64LittleEndian(payload.Slice(index, 8));
        index += 8;
        var timestamp = (long)BinaryPrimitives.ReadUInt64LittleEndian(payload.Slice(index, 8));
        index += 8;

        if (!TryReadUInt16(payload, ref index, out var subjectLength) || !TryReadString(payload, ref index, subjectLength, out var subject))
            return (string.Empty, string.Empty, null, null, 0, 0, false, ErrBadStreamMsg);
        if (!TryReadUInt16(payload, ref index, out var replyLength) || !TryReadString(payload, ref index, replyLength, out var reply))
            return (string.Empty, string.Empty, null, null, 0, 0, false, ErrBadStreamMsg);
        if (!TryReadUInt16(payload, ref index, out var headerLength) || !TryReadBytes(payload, ref index, headerLength, out var header))
            return (string.Empty, string.Empty, null, null, 0, 0, false, ErrBadStreamMsg);
        if (!TryReadUInt32(payload, ref index, out var messageLength) || !TryReadBytes(payload, ref index, messageLength, out var message))
            return (string.Empty, string.Empty, null, null, 0, 0, false, ErrBadStreamMsg);

        var sourced = false;
        if (index < payload.Length)
        {
            if (!TryReadUVarInt(payload.Slice(index), out var flags, out _))
                return (string.Empty, string.Empty, null, null, 0, 0, false, ErrBadStreamMsg);
            sourced = (flags & MsgFlagFromSourceOrMirror) != 0;
        }

        return (subject, reply, header.Length == 0 ? null : header, message.Length == 0 ? null : message, sequence, timestamp, sourced, null);
    }

    internal static (string BatchId, ulong BatchSequence, EntryOp Operation, byte[]? Payload, Exception? Error) DecodeBatchMsg(ReadOnlySpan<byte> payload)
    {
        var index = 0;
        if (!TryReadUInt16(payload, ref index, out var batchIdLength) || !TryReadString(payload, ref index, batchIdLength, out var batchId))
            return (string.Empty, 0, default, null, ErrBadStreamMsg);
        if (!TryReadUVarInt(payload.Slice(index), out var sequence, out var consumed))
            return (string.Empty, 0, default, null, ErrBadStreamMsg);

        index += consumed;
        if (index >= payload.Length)
            return (string.Empty, 0, default, null, ErrBadStreamMsg);

        var operation = (EntryOp)payload[index];
        index++;
        var data = payload.Slice(index).ToArray();
        return (batchId, sequence, operation, data, null);
    }

    internal static byte[] EncodeStreamMsg(string subject, string reply, byte[]? header, byte[]? message, ulong sequence, long timestamp, bool sourced)
        => EncodeStreamMsgAllowCompress(subject, reply, header, message, sequence, timestamp, sourced);

    internal static byte[] EncodeStreamMsgAllowCompress(string subject, string reply, byte[]? header, byte[]? message, ulong sequence, long timestamp, bool sourced)
        => EncodeStreamMsgAllowCompressAndBatch(subject, reply, header, message, sequence, timestamp, sourced, string.Empty, 0, false);

    internal static string SyncSubjForStream() => SyncSubject("$JSC.SYNC");

    internal static string SyncReplySubject() => SyncSubject("$JSC.R");

    internal static string InfoReplySubject() => SyncSubject("$JSC.R");

    internal static string SyncAckSubject() => $"{SyncSubject("$JSC.ACK")}.*";

    internal static string SyncSubject(string prefix)
    {
        var suffix = new char[ReplySuffixLength];
        var value = Random.Shared.NextInt64(long.MaxValue);
        for (var i = 0; i < suffix.Length; i++)
        {
            suffix[i] = Base62[(int)(value % Base62.Length)];
            value /= Base62.Length;
        }

        return $"{prefix}.{new string(suffix)}";
    }

    internal static byte[] EncodeStreamMsgAllowCompressAndBatch(
        string subject,
        string reply,
        byte[]? header,
        byte[]? message,
        ulong sequence,
        long timestamp,
        bool sourced,
        string batchId,
        ulong batchSequence,
        bool batchCommit)
    {
        subject ??= string.Empty;
        reply ??= string.Empty;
        header ??= [];
        message ??= [];
        batchId ??= string.Empty;

        var subjectLength = Math.Min(subject.Length, ushort.MaxValue);
        var replyLength = Math.Min(reply.Length, ushort.MaxValue);
        var headerLength = Math.Min(header.Length, ushort.MaxValue);
        var messageLength = Math.Min(message.Length, int.MaxValue);

        var body = new List<byte>(1 + subjectLength + replyLength + headerLength + messageLength + 32);
        var op = EntryOp.StreamMsgOp;
        if (!string.IsNullOrEmpty(batchId))
        {
            body.Add(0); // reserve slot for batch op
            var batchIdLength = Math.Min(batchId.Length, ushort.MaxValue);
            body.AddRange(BitConverter.GetBytes((ushort)batchIdLength));
            body.AddRange(Encoding.ASCII.GetBytes(batchId[..batchIdLength]));
            AppendUVarInt(body, batchSequence);
            op = batchCommit ? EntryOp.BatchCommitMsgOp : EntryOp.BatchMsgOp;
            body.Add((byte)EntryOp.StreamMsgOp);
        }

        body.AddRange(BitConverter.GetBytes(sequence));
        body.AddRange(BitConverter.GetBytes((ulong)timestamp));
        body.AddRange(BitConverter.GetBytes((ushort)subjectLength));
        body.AddRange(Encoding.ASCII.GetBytes(subject[..subjectLength]));
        body.AddRange(BitConverter.GetBytes((ushort)replyLength));
        body.AddRange(Encoding.ASCII.GetBytes(reply[..replyLength]));
        body.AddRange(BitConverter.GetBytes((ushort)headerLength));
        body.AddRange(header.AsSpan(0, headerLength).ToArray());
        body.AddRange(BitConverter.GetBytes((uint)messageLength));
        body.AddRange(message.AsSpan(0, messageLength).ToArray());

        AppendUVarInt(body, sourced ? MsgFlagFromSourceOrMirror : 0);
        var encoded = body.ToArray();

        var shouldCompress = (subjectLength + replyLength + headerLength + messageLength) > CompressThreshold;
        if (shouldCompress)
        {
            var opIndex = !string.IsNullOrEmpty(batchId) ? FindStreamOpOffset(encoded) : 0;
            var compressed = Snappy.Encode(encoded.AsSpan(opIndex + 1));
            if (compressed.Length < encoded.Length - (opIndex + 1))
            {
                var output = new byte[opIndex + 1 + compressed.Length];
                if (opIndex > 0)
                    Buffer.BlockCopy(encoded, 0, output, 0, opIndex);
                output[opIndex] = (byte)EntryOp.CompressedStreamMsgOp;
                Buffer.BlockCopy(compressed, 0, output, opIndex + 1, compressed.Length);
                if (!string.IsNullOrEmpty(batchId))
                    output[0] = (byte)op;
                return output;
            }
        }

        if (!string.IsNullOrEmpty(batchId))
        {
            encoded[0] = (byte)op;
            return encoded;
        }

        return PrependOp((byte)EntryOp.StreamMsgOp, encoded);
    }

    private static byte[] PrependOp(byte op, byte[] payload)
    {
        var encoded = new byte[payload.Length + 1];
        encoded[0] = op;
        Buffer.BlockCopy(payload, 0, encoded, 1, payload.Length);
        return encoded;
    }

    private static ConsumerAssignment CloneForProposal(ConsumerAssignment assignment)
    {
        var clone = assignment.CopyGroup();
        if (assignment.Config != null)
            clone.ConfigJson = SerializeJsonElement(assignment.Config);
        return clone;
    }

    internal static JsonElement SerializeJsonElement<T>(T value)
    {
        var bytes = JsonSerializer.SerializeToUtf8Bytes(value);
        using var doc = JsonDocument.Parse(bytes);
        return doc.RootElement.Clone();
    }

    private static bool TryReadUInt16(ReadOnlySpan<byte> payload, ref int index, out int value)
    {
        if (payload.Length - index < 2)
        {
            value = 0;
            return false;
        }

        value = BinaryPrimitives.ReadUInt16LittleEndian(payload.Slice(index, 2));
        index += 2;
        return true;
    }

    private static bool TryReadUInt32(ReadOnlySpan<byte> payload, ref int index, out int value)
    {
        if (payload.Length - index < 4)
        {
            value = 0;
            return false;
        }

        value = (int)BinaryPrimitives.ReadUInt32LittleEndian(payload.Slice(index, 4));
        index += 4;
        return true;
    }

    private static bool TryReadString(ReadOnlySpan<byte> payload, ref int index, int length, out string value)
    {
        if (payload.Length - index < length)
        {
            value = string.Empty;
            return false;
        }

        value = Encoding.ASCII.GetString(payload.Slice(index, length));
        index += length;
        return true;
    }

    private static bool TryReadBytes(ReadOnlySpan<byte> payload, ref int index, int length, out byte[] value)
    {
        if (payload.Length - index < length)
        {
            value = [];
            return false;
        }

        value = payload.Slice(index, length).ToArray();
        index += length;
        return true;
    }

    private static void AppendUVarInt(List<byte> buffer, ulong value)
    {
        while (value >= 0x80)
        {
            buffer.Add((byte)((value & 0x7F) | 0x80));
            value >>= 7;
        }

        buffer.Add((byte)value);
    }

    private static int FindStreamOpOffset(byte[] encoded)
    {
        var index = 1;
        if (encoded.Length < 3)
            return 0;
        var batchIdLength = BinaryPrimitives.ReadUInt16LittleEndian(encoded.AsSpan(index, 2));
        index += 2 + batchIdLength;
        if (!TryReadUVarInt(encoded.AsSpan(index), out _, out var consumed))
            return 0;
        return index + consumed;
    }

    private static void ShuffleInPlace(List<string> items)
    {
        for (var i = items.Count - 1; i > 0; i--)
        {
            var j = Random.Shared.Next(i + 1);
            (items[i], items[j]) = (items[j], items[i]);
        }
    }
}

// ============================================================================
// InflightStreamInfo
// ============================================================================

/// <summary>
/// Tracks inflight stream create/update/delete proposals.
/// Mirrors Go <c>inflightStreamInfo</c> in server/jetstream_cluster.go lines 87-91.
/// </summary>
internal sealed class InflightStreamInfo
{
    /// <summary>Number of inflight operations.</summary>
    public ulong Ops { get; set; }
    /// <summary>Whether the stream has been deleted.</summary>
    public bool Deleted { get; set; }
    public StreamAssignment? Assignment { get; set; }
}

// ============================================================================
// InflightConsumerInfo
// ============================================================================

/// <summary>
/// Tracks inflight consumer create/update/delete proposals.
/// Mirrors Go <c>inflightConsumerInfo</c> in server/jetstream_cluster.go lines 94-98.
/// </summary>
internal sealed class InflightConsumerInfo
{
    /// <summary>Number of inflight operations.</summary>
    public ulong Ops { get; set; }
    /// <summary>Whether the consumer has been deleted.</summary>
    public bool Deleted { get; set; }
    public ConsumerAssignment? Assignment { get; set; }
}

// ============================================================================
// PeerRemoveInfo
// ============================================================================

/// <summary>
/// Holds the reply information for a peer-remove request pending quorum.
/// Mirrors Go <c>peerRemoveInfo</c> in server/jetstream_cluster.go lines 101-106.
/// </summary>
internal sealed class PeerRemoveInfo
{
    /// <summary>Client info from the request (object to avoid session dep).</summary>
    public ClientInfo? ClientInfo { get; set; }
    public string Subject { get; set; } = string.Empty;
    public string Reply   { get; set; } = string.Empty;
    public string Request { get; set; } = string.Empty;
}

// ============================================================================
// EntryOp enum
// ============================================================================

/// <summary>
/// Operation type encoded in a JetStream cluster Raft entry.
/// Mirrors Go <c>entryOp</c> iota in server/jetstream_cluster.go lines 116-150.
/// ONLY ADD TO THE END — inserting values breaks server interoperability.
/// </summary>
internal enum EntryOp : byte
{
    // Meta ops
    AssignStreamOp             = 0,
    AssignConsumerOp           = 1,
    RemoveStreamOp             = 2,
    RemoveConsumerOp           = 3,
    // Stream ops
    StreamMsgOp                = 4,
    PurgeStreamOp              = 5,
    DeleteMsgOp                = 6,
    // Consumer ops
    UpdateDeliveredOp          = 7,
    UpdateAcksOp               = 8,
    // Compressed consumer assignments
    AssignCompressedConsumerOp = 9,
    // Filtered consumer skip
    UpdateSkipOp               = 10,
    // Update stream
    UpdateStreamOp             = 11,
    // Pending pull requests
    AddPendingRequest          = 12,
    RemovePendingRequest       = 13,
    // Compressed streams (RAFT or catchup)
    CompressedStreamMsgOp      = 14,
    // Deleted gaps on catchups for replicas
    DeleteRangeOp              = 15,
    // Batch stream ops
    BatchMsgOp                 = 16,
    BatchCommitMsgOp           = 17,
    // Consumer reset to specific starting sequence
    ResetSeqOp                 = 18,
}

// ============================================================================
// RaftGroup
// ============================================================================

/// <summary>
/// Describes a Raft consensus group that houses streams and consumers.
/// Controlled by the meta-group controller.
/// Mirrors Go <c>raftGroup</c> struct in server/jetstream_cluster.go lines 154-163.
/// </summary>
internal sealed class RaftGroup
{
    [JsonPropertyName("name")]      public string      Name      { get; set; } = string.Empty;
    [JsonPropertyName("peers")]     public string[]    Peers     { get; set; } = [];
    [JsonPropertyName("store")]     public StorageType Storage   { get; set; }
    [JsonPropertyName("cluster")]   public string?     Cluster   { get; set; }
    [JsonPropertyName("preferred")] public string?     Preferred { get; set; }
    [JsonPropertyName("scale_up")]  public bool        ScaleUp   { get; set; }
    /// <summary>Internal Raft node — not serialized.</summary>
    [JsonIgnore]
    public IRaftNode? Node { get; set; }

    internal bool IsMember(string id) =>
        Peers.Contains(id, StringComparer.Ordinal);

    internal void SetPreferred(NatsServer server)
    {
        if (Peers.Length == 0)
            return;
        if (Peers.Length == 1)
        {
            Preferred = Peers[0];
            return;
        }

        var online = new List<string>(Peers.Length);
        foreach (var peer in Peers)
        {
            var info = server.GetNodeInfo(peer);
            if (info is { Offline: false })
                online.Add(peer);
        }

        var candidates = online.Count > 0 ? online : [.. Peers];
        Preferred = candidates[Random.Shared.Next(candidates.Count)];
    }
}

// ============================================================================
// StreamAssignment
// ============================================================================

/// <summary>
/// What the meta controller uses to assign streams to peers.
/// Mirrors Go <c>streamAssignment</c> struct in server/jetstream_cluster.go lines 166-184.
/// </summary>
internal sealed class StreamAssignment
{
    [JsonPropertyName("client")]         public ClientInfo?      Client     { get; set; }
    [JsonPropertyName("created")]        public DateTime         Created    { get; set; }
    [JsonPropertyName("stream")]         public JsonElement      ConfigJson { get; set; }
    [JsonIgnore]                         public StreamConfig?    Config     { get; set; }
    [JsonPropertyName("group")]          public RaftGroup?       Group      { get; set; }
    [JsonPropertyName("sync")]           public string           Sync       { get; set; } = string.Empty;
    [JsonPropertyName("subject")]        public string?          Subject    { get; set; }
    [JsonPropertyName("reply")]          public string?          Reply      { get; set; }
    [JsonPropertyName("restore_state")]  public StreamState?     Restore    { get; set; }

    // Internal (not serialized)
    [JsonIgnore] public Dictionary<string, ConsumerAssignment>? Consumers   { get; set; }
    [JsonIgnore] public bool Responded    { get; set; }
    [JsonIgnore] public bool Recovering   { get; set; }
    [JsonIgnore] public bool Reassigning  { get; set; }
    [JsonIgnore] public bool Resetting    { get; set; }
    [JsonIgnore] public Exception? Error  { get; set; }
    [JsonIgnore] public UnsupportedStreamAssignment? Unsupported { get; set; }

    internal StreamAssignment CopyGroup()
    {
        var clone = new StreamAssignment
        {
            Client = Client,
            Created = Created,
            ConfigJson = ConfigJson,
            Config = Config,
            Group = Group == null
                ? null
                : new RaftGroup
                {
                    Name = Group.Name,
                    Peers = [.. Group.Peers],
                    Storage = Group.Storage,
                    Cluster = Group.Cluster,
                    Preferred = Group.Preferred,
                    ScaleUp = Group.ScaleUp,
                    Node = Group.Node,
                },
            Sync = Sync,
            Subject = Subject,
            Reply = Reply,
            Restore = Restore,
            Consumers = Consumers,
            Responded = Responded,
            Recovering = Recovering,
            Reassigning = Reassigning,
            Resetting = Resetting,
            Error = Error,
            Unsupported = Unsupported,
        };
        return clone;
    }

    internal bool MissingPeers() =>
        Group != null &&
        Config != null &&
        Group.Peers.Length < Math.Max(1, Config.Replicas);

    internal string RecoveryKey() =>
        $"{Client?.ServiceAccount() ?? string.Empty}:{Config?.Name ?? Subject ?? string.Empty}";
}

// ============================================================================
// UnsupportedStreamAssignment
// ============================================================================

/// <summary>
/// Holds state for a stream assignment that this server cannot run,
/// so that it can still respond to cluster info requests.
/// Mirrors Go <c>unsupportedStreamAssignment</c> in server/jetstream_cluster.go lines 186-191.
/// </summary>
internal sealed class UnsupportedStreamAssignment
{
    public string  Reason  { get; set; } = string.Empty;
    public StreamInfo Info  { get; set; } = new();
    /// <summary>Internal system client (object to avoid session dep).</summary>
    public object? SysClient   { get; set; }
    /// <summary>Info subscription (object to avoid session dep).</summary>
    public object? InfoSub     { get; set; }

    internal void SetupInfoSub(NatsServer server, StreamAssignment assignment)
    {
        if (InfoSub != null)
            return;

        SysClient = server.CreateInternalJetStreamClient();
        InfoSub = $"{assignment.Client?.ServiceAccount() ?? string.Empty}:{assignment.Config?.Name ?? string.Empty}";
    }

    internal StreamInfoClusterResponse HandleClusterStreamInfoRequest() =>
        new()
        {
            OfflineReason = Reason,
            StreamInfo = Info,
        };

    internal void CloseInfoSub()
    {
        InfoSub = null;
        SysClient = null;
    }
}

// ============================================================================
// ConsumerAssignment
// ============================================================================

/// <summary>
/// What the meta controller uses to assign consumers to streams.
/// Mirrors Go <c>consumerAssignment</c> struct in server/jetstream_cluster.go lines 250-266.
/// </summary>
internal sealed class ConsumerAssignment
{
    [JsonPropertyName("client")]   public ClientInfo?      Client     { get; set; }
    [JsonPropertyName("created")]  public DateTime         Created    { get; set; }
    [JsonPropertyName("name")]     public string           Name       { get; set; } = string.Empty;
    [JsonPropertyName("stream")]   public string           Stream     { get; set; } = string.Empty;
    [JsonPropertyName("consumer")] public JsonElement      ConfigJson { get; set; }
    [JsonIgnore]                   public ConsumerConfig?  Config     { get; set; }
    [JsonPropertyName("group")]    public RaftGroup?       Group      { get; set; }
    [JsonPropertyName("subject")]  public string?          Subject    { get; set; }
    [JsonPropertyName("reply")]    public string?          Reply      { get; set; }
    [JsonPropertyName("state")]    public ConsumerState?   State      { get; set; }

    // Internal (not serialized)
    [JsonIgnore] public bool Responded   { get; set; }
    [JsonIgnore] public bool Recovering  { get; set; }
    [JsonIgnore] public Exception? Error { get; set; }
    [JsonIgnore] public UnsupportedConsumerAssignment? Unsupported { get; set; }

    internal ConsumerAssignment CopyGroup()
    {
        var clone = new ConsumerAssignment
        {
            Client = Client,
            Created = Created,
            Name = Name,
            Stream = Stream,
            ConfigJson = ConfigJson,
            Config = Config,
            Group = Group == null
                ? null
                : new RaftGroup
                {
                    Name = Group.Name,
                    Peers = [.. Group.Peers],
                    Storage = Group.Storage,
                    Cluster = Group.Cluster,
                    Preferred = Group.Preferred,
                    ScaleUp = Group.ScaleUp,
                    Node = Group.Node,
                },
            Subject = Subject,
            Reply = Reply,
            State = State,
            Responded = Responded,
            Recovering = Recovering,
            Error = Error,
            Unsupported = Unsupported,
        };
        return clone;
    }

    internal string StreamRecoveryKey() =>
        $"{Client?.ServiceAccount() ?? string.Empty}:{Stream}";

    internal string RecoveryKey() =>
        $"{Client?.ServiceAccount() ?? string.Empty}:{Stream}:{Name}";
}

// ============================================================================
// UnsupportedConsumerAssignment
// ============================================================================

/// <summary>
/// Holds state for a consumer assignment that this server cannot run.
/// Mirrors Go <c>unsupportedConsumerAssignment</c> in server/jetstream_cluster.go lines 268-273.
/// </summary>
internal sealed class UnsupportedConsumerAssignment
{
    public string       Reason    { get; set; } = string.Empty;
    public ConsumerInfo Info      { get; set; } = new();
    /// <summary>Internal system client (object to avoid session dep).</summary>
    public object? SysClient      { get; set; }
    /// <summary>Info subscription (object to avoid session dep).</summary>
    public object? InfoSub        { get; set; }

    internal void SetupInfoSub(NatsServer server, ConsumerAssignment assignment)
    {
        if (InfoSub != null)
            return;

        SysClient = server.CreateInternalJetStreamClient();
        InfoSub = $"{assignment.Client?.ServiceAccount() ?? string.Empty}:{assignment.Stream}:{assignment.Name}";
    }

    internal ConsumerInfoClusterResponse HandleClusterConsumerInfoRequest() =>
        new()
        {
            OfflineReason = Reason,
            ConsumerInfo = Info,
        };

    internal void CloseInfoSub()
    {
        InfoSub = null;
        SysClient = null;
    }
}

// ============================================================================
// WriteableConsumerAssignment
// ============================================================================

/// <summary>
/// Serialisable form of a consumer assignment used in meta-snapshots.
/// Mirrors Go <c>writeableConsumerAssignment</c> in server/jetstream_cluster.go lines 332-340.
/// </summary>
internal sealed class WriteableConsumerAssignment
{
    [JsonPropertyName("client")]   public ClientInfo?      Client     { get; set; }
    [JsonPropertyName("created")]  public DateTime         Created    { get; set; }
    [JsonPropertyName("name")]     public string           Name       { get; set; } = string.Empty;
    [JsonPropertyName("stream")]   public string           Stream     { get; set; } = string.Empty;
    [JsonPropertyName("consumer")] public JsonElement      ConfigJson { get; set; }
    [JsonPropertyName("group")]    public RaftGroup?       Group      { get; set; }
    [JsonPropertyName("state")]    public ConsumerState?   State      { get; set; }
}

// ============================================================================
// StreamPurge
// ============================================================================

/// <summary>
/// What the stream leader replicates when purging a stream.
/// Mirrors Go <c>streamPurge</c> struct in server/jetstream_cluster.go lines 343-350.
/// </summary>
internal sealed class StreamPurge
{
    [JsonPropertyName("client")]   public ClientInfo?               Client  { get; set; }
    [JsonPropertyName("stream")]   public string                    Stream  { get; set; } = string.Empty;
    [JsonPropertyName("last_seq")] public ulong                     LastSeq { get; set; }
    [JsonPropertyName("subject")]  public string                    Subject { get; set; } = string.Empty;
    [JsonPropertyName("reply")]    public string                    Reply   { get; set; } = string.Empty;
    [JsonPropertyName("request")]  public JSApiStreamPurgeRequest?  Request { get; set; }
}

// ============================================================================
// StreamMsgDelete
// ============================================================================

/// <summary>
/// What the stream leader replicates when deleting a message.
/// Mirrors Go <c>streamMsgDelete</c> struct in server/jetstream_cluster.go lines 353-360.
/// </summary>
internal sealed class StreamMsgDelete
{
    [JsonPropertyName("client")]    public ClientInfo? Client   { get; set; }
    [JsonPropertyName("stream")]    public string      Stream   { get; set; } = string.Empty;
    [JsonPropertyName("seq")]       public ulong       Seq      { get; set; }
    [JsonPropertyName("no_erase")]  public bool        NoErase  { get; set; }
    [JsonPropertyName("subject")]   public string      Subject  { get; set; } = string.Empty;
    [JsonPropertyName("reply")]     public string      Reply    { get; set; } = string.Empty;
}

// ============================================================================
// RecoveryUpdates
// ============================================================================

/// <summary>
/// Accumulates stream/consumer changes discovered during meta-recovery so they
/// can be applied after the full snapshot has been processed.
/// Mirrors Go <c>recoveryUpdates</c> struct in server/jetstream_cluster.go lines 1327-1333.
/// </summary>
internal sealed class RecoveryUpdates
{
    public Dictionary<string, StreamAssignment>                          RemoveStreams   { get; set; } = new();
    public Dictionary<string, Dictionary<string, ConsumerAssignment>>   RemoveConsumers { get; set; } = new();
    public Dictionary<string, StreamAssignment>                          AddStreams      { get; set; } = new();
    public Dictionary<string, StreamAssignment>                          UpdateStreams   { get; set; } = new();
    public Dictionary<string, Dictionary<string, ConsumerAssignment>>   UpdateConsumers { get; set; } = new();

    internal void RemoveStream(StreamAssignment assignment)
    {
        var key = StreamRecoveryKey(assignment);
        RemoveStreams[key] = assignment;
        AddStreams.Remove(key);
        UpdateStreams.Remove(key);
        UpdateConsumers.Remove(key);
        RemoveConsumers.Remove(key);
    }

    internal void AddStream(StreamAssignment assignment)
    {
        var key = StreamRecoveryKey(assignment);
        AddStreams[key] = assignment;
    }

    internal void UpdateStream(StreamAssignment assignment)
    {
        var key = StreamRecoveryKey(assignment);
        UpdateStreams[key] = assignment;
    }

    internal void RemoveConsumer(ConsumerAssignment assignment)
    {
        var streamKey = ConsumerStreamRecoveryKey(assignment);
        var consumerKey = ConsumerRecoveryKey(assignment);

        if (!RemoveConsumers.TryGetValue(streamKey, out var consumers))
        {
            consumers = new Dictionary<string, ConsumerAssignment>(StringComparer.Ordinal);
            RemoveConsumers[streamKey] = consumers;
        }

        consumers[consumerKey] = assignment;
        if (UpdateConsumers.TryGetValue(streamKey, out var updates))
            updates.Remove(consumerKey);
    }

    internal void AddOrUpdateConsumer(ConsumerAssignment assignment)
    {
        var streamKey = ConsumerStreamRecoveryKey(assignment);
        var consumerKey = ConsumerRecoveryKey(assignment);

        if (!UpdateConsumers.TryGetValue(streamKey, out var consumers))
        {
            consumers = new Dictionary<string, ConsumerAssignment>(StringComparer.Ordinal);
            UpdateConsumers[streamKey] = consumers;
        }

        consumers[consumerKey] = assignment;
    }

    private static string StreamRecoveryKey(StreamAssignment assignment) =>
        $"{assignment.Client?.ServiceAccount() ?? string.Empty}:{assignment.Config?.Name ?? assignment.Subject ?? string.Empty}";

    private static string ConsumerStreamRecoveryKey(ConsumerAssignment assignment) =>
        $"{assignment.Client?.ServiceAccount() ?? string.Empty}:{assignment.Stream}";

    private static string ConsumerRecoveryKey(ConsumerAssignment assignment) =>
        $"{assignment.Stream}:{assignment.Name}";
}

internal sealed class StreamInfoClusterResponse
{
    [JsonPropertyName("offline_reason")]
    public string? OfflineReason { get; set; }

    [JsonPropertyName("stream_info")]
    public StreamInfo? StreamInfo { get; set; }
}

internal sealed class ConsumerInfoClusterResponse
{
    [JsonPropertyName("offline_reason")]
    public string? OfflineReason { get; set; }

    [JsonPropertyName("consumer_info")]
    public ConsumerInfo? ConsumerInfo { get; set; }
}

// ============================================================================
// WriteableStreamAssignment
// ============================================================================

/// <summary>
/// Serialisable form of a stream assignment (with its consumers) used in meta-snapshots.
/// Mirrors Go <c>writeableStreamAssignment</c> in server/jetstream_cluster.go lines 1872-1879.
/// </summary>
internal sealed class WriteableStreamAssignment
{
    [JsonPropertyName("client")]    public ClientInfo?                        Client     { get; set; }
    [JsonPropertyName("created")]   public DateTime                           Created    { get; set; }
    [JsonPropertyName("stream")]    public JsonElement                        ConfigJson { get; set; }
    [JsonPropertyName("group")]     public RaftGroup?                         Group      { get; set; }
    [JsonPropertyName("sync")]      public string                             Sync       { get; set; } = string.Empty;
    public List<WriteableConsumerAssignment>                                  Consumers  { get; set; } = new();
}

// ============================================================================
// ConsumerAssignmentResult
// ============================================================================

/// <summary>
/// Result sent by a member after processing a consumer assignment.
/// Mirrors Go <c>consumerAssignmentResult</c> in server/jetstream_cluster.go lines 5592-5597.
/// </summary>
internal sealed class ConsumerAssignmentResult
{
    [JsonPropertyName("account")]  public string Account  { get; set; } = string.Empty;
    [JsonPropertyName("stream")]   public string Stream   { get; set; } = string.Empty;
    [JsonPropertyName("consumer")] public string Consumer { get; set; } = string.Empty;
    /// <summary>Stub: JSApiConsumerCreateResponse — full type in session 20+.</summary>
    [JsonPropertyName("response")] public object? Response { get; set; }
}

// ============================================================================
// StreamAssignmentResult
// ============================================================================

/// <summary>
/// Result sent by a member after processing a stream assignment.
/// Mirrors Go <c>streamAssignmentResult</c> in server/jetstream_cluster.go lines 6779-6785.
/// </summary>
internal sealed class StreamAssignmentResult
{
    [JsonPropertyName("account")]          public string Account  { get; set; } = string.Empty;
    [JsonPropertyName("stream")]           public string Stream   { get; set; } = string.Empty;
    /// <summary>Stub: JSApiStreamCreateResponse — full type in session 20+.</summary>
    [JsonPropertyName("create_response")]  public object? Response { get; set; }
    /// <summary>Stub: JSApiStreamRestoreResponse — full type in session 20+.</summary>
    [JsonPropertyName("restore_response")] public object? Restore  { get; set; }
    [JsonPropertyName("is_update")]        public bool    Update   { get; set; }
}

// ============================================================================
// SelectPeerError
// ============================================================================

/// <summary>
/// Collects the reasons why no suitable peer could be found for a placement.
/// Mirrors Go <c>selectPeerError</c> struct in server/jetstream_cluster.go lines 7113-7121.
/// </summary>
internal sealed class SelectPeerError : Exception
{
    public bool ExcludeTag  { get; set; }
    public bool Offline     { get; set; }
    public bool NoStorage   { get; set; }
    public bool UniqueTag   { get; set; }
    public bool Misc        { get; set; }
    public bool NoJsClust   { get; set; }
    public HashSet<string>? NoMatchTags  { get; set; }
    public HashSet<string>? ExcludeTags  { get; set; }

    public override string Message => BuildMessage();

    private string BuildMessage()
    {
        var b = new System.Text.StringBuilder("no suitable peers for placement");
        if (Offline)    b.Append(", peer offline");
        if (ExcludeTag) b.Append(", exclude tag set");
        if (NoStorage)  b.Append(", insufficient storage");
        if (UniqueTag)  b.Append(", server tag not unique");
        if (Misc)       b.Append(", miscellaneous issue");
        if (NoJsClust)  b.Append(", jetstream not enabled in cluster");
        if (NoMatchTags is { Count: > 0 })
        {
            b.Append(", tags not matched [");
            b.Append(string.Join(", ", NoMatchTags));
            b.Append(']');
        }
        if (ExcludeTags is { Count: > 0 })
        {
            b.Append(", tags excluded [");
            b.Append(string.Join(", ", ExcludeTags));
            b.Append(']');
        }
        return b.ToString();
    }

    internal string Error() => Message;

    internal void AddMissingTag(string tag)
    {
        NoMatchTags ??= new HashSet<string>(StringComparer.Ordinal);
        NoMatchTags.Add(tag);
    }

    internal void AddExcludeTag(string tag)
    {
        ExcludeTags ??= new HashSet<string>(StringComparer.Ordinal);
        ExcludeTags.Add(tag);
    }

    internal void Accumulate(SelectPeerError? other)
    {
        if (other == null)
            return;

        ExcludeTag |= other.ExcludeTag;
        Offline |= other.Offline;
        NoStorage |= other.NoStorage;
        UniqueTag |= other.UniqueTag;
        Misc |= other.Misc;
        NoJsClust |= other.NoJsClust;

        if (other.NoMatchTags != null)
        {
            foreach (var tag in other.NoMatchTags)
                AddMissingTag(tag);
        }

        if (other.ExcludeTags != null)
        {
            foreach (var tag in other.ExcludeTags)
                AddExcludeTag(tag);
        }
    }
}

// ============================================================================
// StreamSnapshot
// ============================================================================

/// <summary>
/// JSON-serialisable snapshot of a stream's state for cluster catchup.
/// Mirrors Go <c>streamSnapshot</c> struct in server/jetstream_cluster.go lines 9454-9461.
/// </summary>
internal sealed class StreamSnapshot
{
    [JsonPropertyName("messages")]  public ulong    Msgs     { get; set; }
    [JsonPropertyName("bytes")]     public ulong    Bytes    { get; set; }
    [JsonPropertyName("first_seq")] public ulong    FirstSeq { get; set; }
    [JsonPropertyName("last_seq")]  public ulong    LastSeq  { get; set; }
    [JsonPropertyName("clfs")]      public ulong    Failed   { get; set; }
    [JsonPropertyName("deleted")]   public ulong[]? Deleted  { get; set; }
}

// ============================================================================
// StreamSyncRequest
// ============================================================================

/// <summary>
/// Request sent by a lagging follower to trigger stream catch-up from the leader.
/// Mirrors Go <c>streamSyncRequest</c> struct in server/jetstream_cluster.go lines 9707-9713.
/// </summary>
internal sealed class StreamSyncRequest
{
    [JsonPropertyName("peer")]           public string? Peer           { get; set; }
    [JsonPropertyName("first_seq")]      public ulong   FirstSeq       { get; set; }
    [JsonPropertyName("last_seq")]       public ulong   LastSeq        { get; set; }
    [JsonPropertyName("delete_ranges")]  public bool    DeleteRangesOk { get; set; }
    [JsonPropertyName("min_applied")]    public ulong   MinApplied     { get; set; }
}

// ============================================================================
// Stub API request/response types (referenced by cluster types; full impl later)
// ============================================================================

/// <summary>Stub: full definition in session 21 (jetstream_api.go).</summary>
internal sealed class JSApiStreamPurgeRequest
{
    [JsonPropertyName("filter")]   public string? Filter  { get; set; }
    [JsonPropertyName("seq")]      public ulong?  Seq     { get; set; }
    [JsonPropertyName("keep")]     public ulong?  Keep    { get; set; }
}
