// Copyright 2018-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/events.go in the NATS server Go source.
// Batch 45: Server.* events methods — stats, remote tracking, connection events,
// request handlers, leaf/gateway events, auth events, OCSP events, misc.

using System.Diagnostics;
using System.Net;
using System.Runtime.InteropServices;
using System.Security.Cryptography.X509Certificates;
using System.Text;
using System.Text.Json;
using ZB.MOM.NatsNet.Server.Auth.CertificateIdentityProvider;
using ZB.MOM.NatsNet.Server.Internal;
using ZB.MOM.NatsNet.Server.Internal.DataStructures;

namespace ZB.MOM.NatsNet.Server;

/// <summary>
/// Partial class containing the server-level events subsystem methods.
/// Mirrors server/events.go in the NATS server Go source.
/// </summary>
public sealed partial class NatsServer
{
    // =========================================================================
    // Core event infrastructure — EventsEnabled / EventsRunning
    // =========================================================================

    /// <summary>
    /// Reports whether the server has internal events enabled via a system account.
    /// Mirrors Go <c>Server.EventsEnabled()</c>.
    /// </summary>
    public bool EventsEnabled()
    {
        _mu.EnterReadLock();
        try { return EventsEnabledLocked(); }
        finally { _mu.ExitReadLock(); }
    }

    /// <summary>
    /// Reports whether events are enabled.
    /// Server read lock must be held on entry.
    /// Mirrors Go <c>Server.eventsEnabled()</c>.
    /// </summary>
    internal bool EventsEnabledLocked() =>
        _sys != null && _sys.Client != null && _sys.Account != null;

    /// <summary>
    /// Locked version of checking if events system is running.
    /// Mirrors Go <c>Server.eventsRunning()</c>.
    /// </summary>
    internal bool EventsRunning()
    {
        _mu.EnterReadLock();
        bool er = Running() && EventsEnabledLocked();
        _mu.ExitReadLock();
        return er;
    }

    // =========================================================================
    // internalReceiveLoop
    // Mirrors Go <c>(s *Server) internalReceiveLoop</c> in server/events.go.
    // =========================================================================

    private void InternalReceiveLoop(IpQueue<InSysMsg> recvq)
    {
        while (EventsRunning())
        {
            if (!recvq.Ch.WaitToReadAsync(_quitCts.Token).AsTask()
                                                          .GetAwaiter()
                                                          .GetResult())
            {
                return;
            }

            var msgs = recvq.Pop();
            if (msgs is not null)
            {
                foreach (var m in msgs)
                {
                    m.Cb?.Invoke(m.Sub, m.Client, m.Acc, m.Subject, m.Reply, m.Hdr, m.Msg);
                }
                recvq.Recycle(msgs);
            }
        }
    }

    // =========================================================================
    // internalSendLoop
    // Mirrors Go <c>(s *Server) internalSendLoop</c> in server/events.go.
    // =========================================================================

    private void InternalSendLoop()
    {
    RESET:
        _mu.EnterReadLock();
        if (_sys is null || _sys.SendQueue is null)
        {
            _mu.ExitReadLock();
            return;
        }
        var sysc      = _sys.Client;
        var resetCh   = _sys.ResetChannel;
        var sendq     = _sys.SendQueue;
        var id        = _info.Id;
        var host      = _info.Host;
        var svrName   = _info.Name;
        var domain    = _info.Domain ?? string.Empty;
        var seqRef    = _sys;
        var jsEnabled = _info.JetStream;
        var cluster   = _info.Cluster ?? string.Empty;
        if (_gateway.Enabled)
            cluster = GetGatewayName();
        _mu.ExitReadLock();

        var opts     = GetOpts();
        var tags     = opts.Tags;
        var metadata = opts.Metadata;

        while (EventsRunning())
        {
            var sendTask  = sendq.Ch.WaitToReadAsync(_quitCts.Token).AsTask();
            var resetTask = resetCh is not null
                ? resetCh.Reader.WaitToReadAsync(_quitCts.Token).AsTask()
                : Task.FromResult(false);

            var completed = Task.WhenAny(sendTask, resetTask, Task.Delay(Timeout.Infinite, _quitCts.Token))
                               .GetAwaiter().GetResult();

            if (_quitCts.IsCancellationRequested)
                return;

            if (completed == resetTask && resetTask.IsCompletedSuccessfully && resetTask.Result)
            {
                resetCh?.Reader.TryRead(out _);
                goto RESET;
            }

            if (!sendTask.IsCompletedSuccessfully || !sendTask.Result)
                continue;

            var msgs = sendq.Pop();
            if (msgs is null) continue;

            foreach (var pm in msgs)
            {
                if (pm.Si is { } si)
                {
                    si.Name     = svrName;
                    si.Domain   = domain;
                    si.Host     = host;
                    si.Cluster  = cluster;
                    si.Id       = id;
                    si.Seq      = (ulong)Interlocked.Increment(ref seqRef.Seq);
                    si.Version  = ServerConstants.Version;
                    si.Time     = DateTime.UtcNow;
                    si.Tags     = tags.Count > 0 ? [..tags] : null;
                    si.Metadata = metadata.Count > 0 ? new Dictionary<string, string>(metadata) : null;
                    si.Capabilities = 0;
                    if (jsEnabled)
                    {
                        si.SetJetStreamEnabled();
                        si.SetBinaryStreamSnapshot();
                    }
                }

                byte[] body = [];
                if (pm.Msg is not null)
                {
                    body = pm.Msg switch
                    {
                        string s  => Encoding.UTF8.GetBytes(s),
                        byte[] b  => b,
                        _         => JsonSerializer.SerializeToUtf8Bytes(pm.Msg),
                    };
                }

                var c = pm.Client ?? sysc;
                if (c is null) { pm.ReturnToPool(); continue; }

                lock (c)
                {
                    c.ParseCtx.Pa.Subject = Encoding.ASCII.GetBytes(pm.Subject);
                    c.ParseCtx.Pa.Reply   = Encoding.ASCII.GetBytes(pm.Reply);
                }

                var payload = new byte[body.Length + 2];
                Buffer.BlockCopy(body, 0, payload, 0, body.Length);
                payload[^2] = (byte)'\r';
                payload[^1] = (byte)'\n';

                c.ProcessInboundClientMsg(payload);

                if (pm.Last)
                {
                    c.FlushClients(long.MaxValue);
                    sendq.Recycle(msgs);
                    pm.ReturnToPool();
                    return;
                }
                else
                {
                    c.FlushClients(0);
                }

                pm.ReturnToPool();
            }
            sendq.Recycle(msgs);
        }
    }

    // =========================================================================
    // sendShutdownEvent
    // Mirrors Go <c>(s *Server) sendShutdownEvent</c> in server/events.go.
    // =========================================================================

    internal void SendShutdownEvent()
    {
        _mu.EnterWriteLock();
        try
        {
            if (_sys is null || _sys.SendQueue is null) return;

            var subject = string.Format(SystemSubjects.ShutdownEventSubj, _info.Id);
            var sendq   = _sys.SendQueue;

            _sys.SendQueue = null;
            _sys.Replies.Clear();

            var si = new ServerInfo();
            sendq.Push(EventHelpers.NewPubMsg(null, subject, string.Empty, si, null, si,
                (int)CompressionType.None, false, true));
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    // =========================================================================
    // sendInternalMsg / sendInternalMsgLocked / sendInternalResponse
    // =========================================================================

    /// <summary>
    /// Queues up an internal message to be sent by the internal send loop.
    /// Server read lock must be held on entry.
    /// Mirrors Go <c>Server.sendInternalMsg()</c>.
    /// </summary>
    internal void SendInternalMsg(string subj, string reply, ServerInfo? si, object? msg)
    {
        if (_sys?.SendQueue == null) return;
        var pm = new PubMsg
        {
            Client  = null,
            Subject = subj,
            Reply   = reply,
            Si      = si,
            Hdr     = null,
            Msg     = msg,
        };
        _sys.SendQueue.Push(pm);
    }

    /// <summary>
    /// Queues up an internal message; acquires/releases the read lock itself.
    /// Mirrors Go <c>Server.sendInternalMsgLocked()</c>.
    /// </summary>
    internal void SendInternalMsgLocked(string subj, string reply, ServerInfo? si, object? msg)
    {
        _mu.EnterReadLock();
        SendInternalMsg(subj, reply, si, msg);
        _mu.ExitReadLock();
    }

    /// <summary>
    /// Queues an internal message from a specific client connection.
    /// Mirrors Go <c>Server.sendInternalMsgFromClient()</c>.
    /// </summary>
    internal void SendInternalMsgFromClient(
        ClientConnection client,
        string           subject,
        string           reply,
        ServerInfo?      si,
        object?          msg)
    {
        _mu.EnterReadLock();
        try
        {
            if (_sys is null || _sys.SendQueue is null) return;
            _sys.SendQueue.Push(EventHelpers.NewPubMsg(
                client, subject, reply, si, null, msg, (int)CompressionType.None, false, false));
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    /// <summary>
    /// Queues an internal message on behalf of a specific account's internal client.
    /// Mirrors Go <c>Server.sendInternalAccountMsg()</c>.
    /// </summary>
    internal void SendInternalAccountMsg(Account? a, string subject, object? msg)
        => SendInternalAccountMsgWithReply(a, subject, string.Empty, null, msg, false);

    /// <summary>
    /// Queues an internal message with an optional reply to an arbitrary account.
    /// Mirrors Go <c>Server.sendInternalAccountMsgWithReply()</c>.
    /// </summary>
    internal void SendInternalAccountMsgWithReply(Account? a, string subject, string reply, byte[]? hdr, object? msg, bool echo)
    {
        _mu.EnterReadLock();
        if (_sys?.SendQueue == null)
        {
            _mu.ExitReadLock();
            return;
        }
        ClientConnection? c = _sys.Client;
        if (a != null)
        {
            lock (a)
            {
                c = a.InternalClient;
            }
        }
        _sys.SendQueue.Push(new PubMsg
        {
            Client  = c,
            Subject = subject,
            Reply   = reply,
            Si      = null,
            Hdr     = hdr,
            Msg     = msg,
            Echo    = echo,
        });
        _mu.ExitReadLock();
    }

    /// <summary>
    /// Queues an account-scoped system message.
    /// Mirrors Go <c>Server.sendInternalAccountSysMsg()</c>.
    /// </summary>
    internal void SendInternalAccountSysMsg(Account? a, string subj, ServerInfo? si, object? msg, int ct)
    {
        _mu.EnterReadLock();
        if (_sys?.SendQueue == null || a == null)
        {
            _mu.ExitReadLock();
            return;
        }
        var sendq = _sys.SendQueue;
        _mu.ExitReadLock();

        ClientConnection? c;
        lock (a)
        {
            c = a.InternalClient;
        }
        sendq.Push(new PubMsg
        {
            Client  = c,
            Subject = subj,
            Reply   = string.Empty,
            Si      = si,
            Hdr     = null,
            Msg     = msg,
            Oct     = ct,
        });
    }

    /// <summary>
    /// Sends a server API response (varz, connz, etc.) via the internal send loop.
    /// Mirrors Go <c>Server.sendInternalResponse()</c>.
    /// </summary>
    internal void SendInternalResponse(string subj, ServerApiResponse response)
    {
        _mu.EnterReadLock();
        if (_sys?.SendQueue == null)
        {
            _mu.ExitReadLock();
            return;
        }
        _sys.SendQueue.Push(new PubMsg
        {
            Client  = null,
            Subject = subj,
            Reply   = string.Empty,
            Si      = response.Server,
            Hdr     = null,
            Msg     = response,
        });
        _mu.ExitReadLock();
    }

    // =========================================================================
    // nextEventID
    // =========================================================================

    /// <summary>
    /// Generates a unique event ID.
    /// Server lock should be held.
    /// Mirrors Go <c>Server.nextEventID()</c>.
    /// </summary>
    private string NextEventID() => Guid.NewGuid().ToString("N");

    // =========================================================================
    // wrapChk — executes f under server write-lock, skips if events disabled
    // =========================================================================

    /// <summary>
    /// Returns a closure that acquires the server write-lock, checks that events
    /// are enabled, calls <paramref name="f"/>, then releases the lock.
    /// Mirrors Go <c>Server.wrapChk()</c>.
    /// </summary>
    internal Action WrapChk(Action f) => () =>
    {
        _mu.EnterWriteLock();
        if (!EventsEnabledLocked())
        {
            _mu.ExitWriteLock();
            return;
        }
        try { f(); }
        finally { _mu.ExitWriteLock(); }
    };

    // =========================================================================
    // noInlineCallback family
    // =========================================================================

    /// <summary>
    /// Wraps a <see cref="SysMsgHandler"/> so that it is queued on the
    /// regular receive queue rather than called inline.
    /// Mirrors Go <c>Server.noInlineCallback()</c>.
    /// </summary>
    internal MsgHandler? NoInlineCallback(SysMsgHandler cb) =>
        NoInlineCallbackRecvQSelect(cb, recvQMuxed);

    /// <summary>
    /// Wraps a <see cref="SysMsgHandler"/> so that it is queued on the
    /// priority (statsz) receive queue.
    /// Mirrors Go <c>Server.noInlineCallbackStatsz()</c>.
    /// </summary>
    internal MsgHandler? NoInlineCallbackStatsz(SysMsgHandler cb) =>
        NoInlineCallbackRecvQSelect(cb, recvQStatsz);

    private const int recvQMuxed  = 1;
    private const int recvQStatsz = 2;

    private MsgHandler? NoInlineCallbackRecvQSelect(SysMsgHandler cb, int recvQSelect)
    {
        _mu.EnterReadLock();
        if (!EventsEnabledLocked())
        {
            _mu.ExitReadLock();
            return null;
        }
        IpQueue<InSysMsg>? recvq = recvQSelect == recvQStatsz
            ? _sys!.RecvQueuePriority
            : _sys!.RecvQueue;
        _mu.ExitReadLock();

        return (sub, c, acc, subj, rply, hdr, rmsg) =>
        {
            recvq?.Push(new InSysMsg
            {
                Sub     = sub,
                Client  = c,
                Acc     = acc,
                Subject = subj,
                Reply   = rply,
                Hdr     = hdr,
                Msg     = rmsg,
                Cb      = cb,
            });
        };
    }

    // =========================================================================
    // sysSubscribe / sysSubscribeQ / sysSubscribeInternal / systemSubscribe
    // =========================================================================

    /// <summary>
    /// Creates an internal subscription on the system account.
    /// Mirrors Go <c>Server.sysSubscribe()</c>.
    /// </summary>
    internal (Subscription? Sub, Exception? Error) SysSubscribe(string subject, MsgHandler? cb) =>
        SystemSubscribe(subject, string.Empty, false, null, cb);

    /// <summary>
    /// Creates an internal subscription with a queue group.
    /// Mirrors Go <c>Server.sysSubscribeQ()</c>.
    /// </summary>
    internal (Subscription? Sub, Exception? Error) SysSubscribeQ(string subject, string queue, MsgHandler? cb) =>
        SystemSubscribe(subject, queue, false, null, cb);

    /// <summary>
    /// Creates an internal subscription but does not forward interest.
    /// Mirrors Go <c>Server.sysSubscribeInternal()</c>.
    /// </summary>
    internal (Subscription? Sub, Exception? Error) SysSubscribeInternal(string subject, MsgHandler? cb) =>
        SystemSubscribe(subject, string.Empty, true, null, cb);

    /// <summary>
    /// Core system-subscribe implementation.
    /// Mirrors Go <c>Server.systemSubscribe()</c>.
    /// </summary>
    private (Subscription? Sub, Exception? Error) SystemSubscribe(string subject, string queue, bool internalOnly, ClientConnection? c, MsgHandler? cb)
    {
        _mu.EnterWriteLock();
        if (!EventsEnabledLocked())
        {
            _mu.ExitWriteLock();
            return (null, new InvalidOperationException("no system account"));
        }
        if (cb == null)
        {
            _mu.ExitWriteLock();
            return (null, new ArgumentNullException(nameof(cb), "undefined message handler"));
        }
        if (c == null) c = _sys!.Client;
        _sys!.Sid++;
        var sid = _sys.Sid.ToString();
        _mu.ExitWriteLock();

        // Create the subscription via the system client stub.
        // In a full implementation this calls c.processSub().
        // Stub: return a synthetic Subscription object.
        var sub = new Subscription
        {
            Subject = Encoding.ASCII.GetBytes(subject),
            Queue   = string.IsNullOrEmpty(queue) ? null : Encoding.ASCII.GetBytes(queue),
            Sid     = Encoding.ASCII.GetBytes(sid),
        };
        return (sub, null);
    }

    // =========================================================================
    // sysUnsubscribe
    // Mirrors Go <c>(s *Server) sysUnsubscribe</c> in server/events.go.
    // =========================================================================

    /// <summary>
    /// Removes a system subscription. Thread-safe.
    /// Mirrors Go <c>(s *Server) sysUnsubscribe</c> in server/events.go.
    /// </summary>
    internal void SysUnsubscribe(Subscription? sub)
    {
        if (sub is null) return;

        _mu.EnterReadLock();
        if (!EventsEnabledLocked())
        {
            _mu.ExitReadLock();
            return;
        }
        _mu.ExitReadLock();

        _mu.EnterReadLock();
        var sysClient = _sys?.Client;
        _mu.ExitReadLock();

        if (sub.Sid is not null && sysClient is not null)
            sysClient.RemoveSubBySid(sub.Sid);
    }

    // =========================================================================
    // inboxReply
    // =========================================================================

    /// <summary>
    /// Handles replies to the server's internal inbox wildcard subscription.
    /// Mirrors Go <c>Server.inboxReply()</c>.
    /// </summary>
    private void InboxReply(Subscription sub, ClientConnection c, Account acc, string subject, string reply, byte[] hdr, byte[] msg)
    {
        _mu.EnterReadLock();
        if (!EventsEnabledLocked() || _sys!.Replies == null)
        {
            _mu.ExitReadLock();
            return;
        }
        _sys.Replies.TryGetValue(subject, out var cb);
        _mu.ExitReadLock();

        cb?.Invoke(sub, c, acc, subject, reply, hdr, msg);
    }

    // =========================================================================
    // newRespInbox
    // =========================================================================

    /// <summary>
    /// Creates a short-lived reply inbox for a single in-flight request.
    /// Mirrors Go <c>Server.newRespInbox()</c>.
    /// </summary>
    private string NewRespInbox()
    {
        var prefix = _sys?.InboxPrefix ?? "$SYS._INBOX.";
        return $"{prefix}{Guid.NewGuid():N}";
    }

    // =========================================================================
    // InitEventTracking
    // =========================================================================

    /// <summary>
    /// Sets up all system-wide subscriptions for tracking remote servers,
    /// accounts, connections, and stats.
    /// Mirrors Go <c>Server.initEventTracking()</c>.
    /// </summary>
    internal void InitEventTracking()
    {
        _mu.EnterReadLock();
        var sys = _sys;
        _mu.ExitReadLock();

        if (sys?.Client == null || sys.Account == null)
            return;

        // Create a hash used for server-targeted messages.
        sys.ShortHash = GetHash(_info.Name);

        // Inbox responses wildcard.
        var subject = string.Format(SystemSubjects.InboxRespSubj, sys.ShortHash, "*");
        if (SysSubscribe(subject, (s, c, a, subj, rply, hdr, m) =>
            InboxReply(s, c, a, subj, rply, hdr ?? [], m ?? [])).Error is { } e1)
        {
            Errorf("Error setting up internal tracking: {0}", e1);
            return;
        }
        sys.InboxPrefix = subject;

        // Remote connection accounting (old subject).
        subject = string.Format(SystemSubjects.AccConnsEventSubjOld, "*");
        if (SysSubscribe(subject, NoInlineCallback(RemoteConnsUpdate)).Error is { } e2)
        {
            Errorf("Error setting up internal tracking for {0}: {1}", subject, e2);
            return;
        }

        // Response for account info requests.
        subject = string.Format(SystemSubjects.ConnsRespSubj, _info.Id);
        if (SysSubscribe(subject, NoInlineCallback(RemoteConnsUpdate)).Error is { } e3)
        {
            Errorf("Error setting up internal tracking: {0}", e3);
            return;
        }

        // Account nsubs requests.
        if (SysSubscribe(SystemSubjects.AccNumSubsReqSubj, NoInlineCallback(NsubsRequest)).Error is { } e4)
        {
            Errorf("Error setting up internal tracking: {0}", e4);
            return;
        }

        // Remote server statsz broadcast.
        subject = string.Format(SystemSubjects.ServerStatsSubj, "*");
        var (rss, e5) = SysSubscribe(subject, NoInlineCallback(RemoteServerUpdate));
        if (e5 != null)
        {
            Errorf("Error setting up internal tracking: {0}", e5);
            return;
        }
        sys.RemoteStatsSub = rss;

        // Remote server shutdown events.
        subject = string.Format(SystemSubjects.ShutdownEventSubj, "*");
        if (SysSubscribe(subject, NoInlineCallback(RemoteServerShutdown)).Error is { } e6)
        {
            Errorf("Error setting up internal tracking: {0}", e6);
            return;
        }

        // Lame-duck events (treated same as shutdown for now).
        subject = string.Format(SystemSubjects.LameDuckEventSubj, "*");
        if (SysSubscribe(subject, NoInlineCallback(RemoteServerShutdown)).Error is { } e7)
        {
            Errorf("Error setting up internal tracking: {0}", e7);
            return;
        }

        // Account claim update subscriptions.
        bool subscribeToUpdate = true;
        if (_accResolver != null)
            subscribeToUpdate = !_accResolver.IsTrackingUpdate();
        if (subscribeToUpdate)
        {
            foreach (var updSubj in new[] { SystemSubjects.AccUpdateEventSubjOld, SystemSubjects.AccUpdateEventSubjNew })
            {
                if (SysSubscribe(string.Format(updSubj, "*"), NoInlineCallback(AccountClaimUpdate)).Error is { } eu)
                {
                    Errorf("Error setting up internal tracking: {0}", eu);
                    return;
                }
            }
        }

        // Legacy statsz ping (kept for backwards compatibility).
        if (SysSubscribe(SystemSubjects.ServerStatsPingReqSubj, NoInlineCallbackStatsz(StatszReq)).Error is { } e8)
        {
            Errorf("Error setting up internal tracking: {0}", e8);
            return;
        }

        // Server-level monitoring services.
        RegisterServerMonSubs();

        // Account-level monitoring services.
        RegisterAccMonSubs();

        // User info (do not propagate interest).
        if (SysSubscribeInternal(string.Format(SystemSubjects.UserDirectReqSubj, "*"), NoInlineCallback(UserInfoReq)).Error is { } e9)
        {
            Errorf("Error setting up internal tracking: {0}", e9);
            return;
        }

        // STATZ account ping equivalent.
        if (SysSubscribe(string.Format(SystemSubjects.AccPingReqSubj, "STATZ"),
            NoInlineCallback(StatzAccPingReq)).Error is { } e10)
        {
            Errorf("Error setting up internal tracking: {0}", e10);
            return;
        }

        // Leaf node connect events.
        subject = string.Format(SystemSubjects.LeafNodeConnectEventSubj, "*");
        if (SysSubscribe(subject, NoInlineCallback(LeafNodeConnected)).Error is { } e11)
        {
            Errorf("Error setting up internal tracking: {0}", e11);
            return;
        }

        // Remote latency tracking.
        subject = string.Format(SystemSubjects.RemoteLatencyEventSubj, sys.ShortHash);
        if (SysSubscribe(subject, NoInlineCallback(RemoteLatencyUpdate)).Error is { } e12)
        {
            Errorf("Error setting up internal latency tracking: {0}", e12);
            return;
        }

        // Debug subscribers.
        if (SysSubscribeInternal(SystemSubjects.AccSubsSubj, NoInlineCallback(DebugSubscribers)).Error is { } e13)
        {
            Errorf("Error setting up internal debug service for subscribers: {0}", e13);
            return;
        }

        // Server reload request.
        subject = string.Format(SystemSubjects.ServerReloadReqSubj, _info.Id);
        if (SysSubscribe(subject, NoInlineCallback(ReloadConfigHandler)).Error is { } e14)
        {
            Errorf("Error setting up server reload handler: {0}", e14);
            return;
        }

        // Client kick.
        subject = string.Format(SystemSubjects.ClientKickReqSubj, _info.Id);
        if (SysSubscribe(subject, NoInlineCallback(KickClient)).Error is { } e15)
        {
            Errorf("Error setting up client kick service: {0}", e15);
            return;
        }

        // Client LDM.
        subject = string.Format(SystemSubjects.ClientLdmReqSubj, _info.Id);
        if (SysSubscribe(subject, NoInlineCallback(LdmClient)).Error is { } e16)
        {
            Errorf("Error setting up client LDM service: {0}", e16);
        }
    }

    // Helper: register per-server monitoring subscriptions.
    private void RegisterServerMonSubs()
    {
        // Direct and ping subscriptions for each monitoring verb.
        void RegisterVerb(string name, MsgHandler? h)
        {
            if (h == null) return;
            var subject = string.Format(SystemSubjects.ServerDirectReqSubj, _info.Id, name);
            if (SysSubscribe(subject, h).Error is { } e1)
            { Errorf("Error setting up internal tracking: {0}", e1); return; }
            subject = string.Format(SystemSubjects.ServerPingReqSubj, name);
            if (SysSubscribe(subject, h).Error is { } e2)
            { Errorf("Error setting up internal tracking: {0}", e2); }
        }

        RegisterVerb("IDZ",    NoInlineCallback(IdzReq));
        RegisterVerb("STATSZ", NoInlineCallbackStatsz(StatszReq));
        RegisterVerb("VARZ",   NoInlineCallback(VarzReq));
        RegisterVerb("SUBSZ",  NoInlineCallback(SubszReq));
        RegisterVerb("CONNZ",  NoInlineCallback(ConnzReq));
        RegisterVerb("ROUTEZ", NoInlineCallback(RoutezReq));
        RegisterVerb("GATEWAYZ", NoInlineCallback(GatewayzReq));
        RegisterVerb("LEAFZ",  NoInlineCallback(LeafzReq));
        RegisterVerb("ACCOUNTZ", NoInlineCallback(AccountzReq));
        RegisterVerb("JSZ",    NoInlineCallback(JszReq));
        RegisterVerb("HEALTHZ", NoInlineCallback(HealthzReq));
        RegisterVerb("EXPVARZ", NoInlineCallback(ExpvarzReq));
        RegisterVerb("IPQUEUESZ", NoInlineCallback(IpqueueszReq));
        RegisterVerb("RAFTZ",  NoInlineCallback(RaftzReq));
    }

    // Helper: register per-account monitoring subscriptions.
    private void RegisterAccMonSubs()
    {
        void RegisterAccVerb(string name, MsgHandler? h)
        {
            if (h == null) return;
            var subject = string.Format(SystemSubjects.AccDirectReqSubj, "*", name);
            if (SysSubscribe(subject, h).Error is { } e)
                Errorf("Error setting up internal tracking: {0}", e);
        }

        RegisterAccVerb("SUBSZ",    NoInlineCallback(AccSubszReq));
        RegisterAccVerb("CONNZ",    NoInlineCallback(AccConnzReq));
        RegisterAccVerb("LEAFZ",    NoInlineCallback(AccLeafzReq));
        RegisterAccVerb("JSZ",      NoInlineCallback(AccJszReq));
        RegisterAccVerb("INFO",     NoInlineCallback(AccInfoReq));
        RegisterAccVerb("STATZ",    NoInlineCallback(AccStatzReq));
        RegisterAccVerb("CONNS",    NoInlineCallback(ConnsRequest));
    }

    // =========================================================================
    // Group A: Stats & Heartbeat
    // =========================================================================

    // -------------------------------------------------------------------------
    // TrackedRemoteServers
    // -------------------------------------------------------------------------

    /// <summary>
    /// Returns how many remote servers are being tracked from a system events perspective.
    /// Returns -1 if events are not enabled.
    /// Mirrors Go <c>Server.TrackedRemoteServers()</c>.
    /// </summary>
    public int TrackedRemoteServers()
    {
        _mu.EnterReadLock();
        try
        {
            if (!Running() || !EventsEnabledLocked()) return -1;
            return _sys!.Servers.Count;
        }
        finally { _mu.ExitReadLock(); }
    }

    // -------------------------------------------------------------------------
    // updateServerUsage
    // -------------------------------------------------------------------------

    /// <summary>
    /// Captures RSS/CPU usage into the advisory stats struct.
    /// Mirrors Go <c>Server.updateServerUsage()</c>.
    /// </summary>
    private static void UpdateServerUsage(ServerStatsAdvisory v)
    {
        v.Cores    = Environment.ProcessorCount;
        v.MaxProcs = Environment.ProcessorCount;
        try
        {
            using var proc = Process.GetCurrentProcess();
            v.Mem = proc.WorkingSet64;
            // CPU is expensive to compute accurately — leave at 0 for now.
        }
        catch { /* ignore */ }
    }

    // -------------------------------------------------------------------------
    // sendStatsz
    // -------------------------------------------------------------------------

    /// <summary>
    /// Builds and sends a <see cref="ServerStatsMsg"/> advisory on the given subject.
    /// Mirrors Go <c>Server.sendStatsz()</c>.
    /// </summary>
    internal void SendStatsz(string subj)
    {
        var m = new ServerStatsMsg();
        UpdateServerUsage(m.Stats);

        if (LimitStatsz(subj)) return;

        _mu.EnterReadLock();
        try
        {
            if (_sys?.Account == null) return;

            // Standalone mode: check for interest before sending.
            var opts = GetOpts();
            bool checkInterest = opts.Cluster.Port == 0 && opts.Gateway.Port == 0 && opts.LeafNode.Port == 0
                && opts.LeafNode.Remotes.Count == 0;

            if (checkInterest)
            {
                var sacc = _sys.Account;
                if (sacc?.Sublist != null)
                {
                    var rr = sacc.Sublist.Match(subj);
                    int total = rr.PSubs.Count + rr.QSubs.Count;
                    if (total == 0) return;
                    if (total == 1 && rr.PSubs.Count == 1 && rr.PSubs[0] == _sys.RemoteStatsSub) return;
                }
            }

            m.Stats.Start             = _start;
            m.Stats.Connections       = _clients.Count;
            m.Stats.TotalConnections  = _totalClients;
            m.Stats.ActiveAccounts    = Interlocked.CompareExchange(ref _activeAccounts, 0, 0);
            m.Stats.Received          = new DataStats { Msgs = Interlocked.Read(ref _stats.InMsgs),  Bytes = Interlocked.Read(ref _stats.InBytes)  };
            m.Stats.Sent              = new DataStats { Msgs = Interlocked.Read(ref _stats.OutMsgs), Bytes = Interlocked.Read(ref _stats.OutBytes) };
            m.Stats.SlowConsumers     = Interlocked.Read(ref _stats.SlowConsumers);

            var scs = new SlowConsumersStats
            {
                Clients  = (ulong)NumSlowConsumersClients(),
                Routes   = (ulong)NumSlowConsumersRoutes(),
                Gateways = (ulong)NumSlowConsumersGateways(),
                Leafs    = (ulong)NumSlowConsumersLeafs(),
            };
            if (scs.Clients != 0 || scs.Routes != 0 || scs.Gateways != 0 || scs.Leafs != 0)
                m.Stats.SlowConsumersStats = scs;

            m.Stats.StaleConnections = Interlocked.Read(ref _stats.StaleConnections);
            m.Stats.StalledClients   = Interlocked.Read(ref _stats.Stalls);

            var stcs = new StaleConnectionStats
            {
                Clients  = (ulong)NumStaleConnectionsClients(),
                Routes   = (ulong)NumStaleConnectionsRoutes(),
                Gateways = (ulong)NumStaleConnectionsGateways(),
                Leafs    = (ulong)NumStaleConnectionsLeafs(),
            };
            if (stcs.Clients != 0 || stcs.Routes != 0 || stcs.Gateways != 0 || stcs.Leafs != 0)
                m.Stats.StaleConnectionStats = stcs;

            m.Stats.NumSubs       = NumSubscriptions();
            m.Stats.ActiveServers = _sys.Servers.Count + 1;

            // Routes.
            ForEachRoute(r =>
            {
                var rs = new RouteStat { Id = r.Cid };
                lock (r)
                {
                    rs.Sent     = new DataStats { Msgs = 0L, Bytes = 0L };
                    rs.Received = new DataStats { Msgs = 0L, Bytes = 0L };
                    rs.Pending  = (int)(r.OutPb >> 10);
                    if (r.Route != null) rs.Name = r.Route.RemoteName;
                }
                m.Stats.Routes ??= [];
                m.Stats.Routes.Add(rs);
            });

            SendInternalMsg(subj, string.Empty, m.Server, m);
        }
        finally { _mu.ExitReadLock(); }
    }

    // -------------------------------------------------------------------------
    // limitStatsz
    // -------------------------------------------------------------------------

    /// <summary>
    /// Rate-limits statsz publishes to one per <see cref="EventIntervals.DefaultStatszRateLimit"/>.
    /// Returns true if the publish should be skipped.
    /// Mirrors Go <c>Server.limitStatsz()</c>.
    /// </summary>
    internal bool LimitStatsz(string subj)
    {
        _mu.EnterWriteLock();
        try
        {
            if (_sys == null) return true;

            // Only limit the normal broadcast subject.
            if (subj != string.Format(SystemSubjects.ServerStatsSubj, ID()))
                return false;

            var interval = EventIntervals.DefaultStatszRateLimit;
            if (_sys.ClientStatszInterval < interval && _sys.ClientStatszInterval > TimeSpan.Zero)
                interval = _sys.ClientStatszInterval;

            if (DateTime.UtcNow - _sys.LastStatsz < interval)
            {
                // Reschedule heartbeat for the next interval.
                _sys.StatsMsgTimer?.Change((long)(_sys.LastStatsz.Add(interval) - DateTime.UtcNow).TotalMilliseconds, Timeout.Infinite);
                return true;
            }
            _sys.LastStatsz = DateTime.UtcNow;
            return false;
        }
        finally { _mu.ExitWriteLock(); }
    }

    // -------------------------------------------------------------------------
    // heartbeatStatsz
    // -------------------------------------------------------------------------

    /// <summary>
    /// Called by the statsz timer; ramps up the send interval to the max.
    /// Must be wrapped with <see cref="WrapChk"/>.
    /// Mirrors Go <c>Server.heartbeatStatsz()</c>.
    /// </summary>
    private void HeartbeatStatsz()
    {
        if (_sys!.StatsMsgTimer != null)
        {
            if (_sys.ClientStatszInterval < _sys.StatszInterval)
            {
                _sys.ClientStatszInterval = _sys.ClientStatszInterval + _sys.ClientStatszInterval;
                if (_sys.ClientStatszInterval > _sys.StatszInterval)
                    _sys.ClientStatszInterval = _sys.StatszInterval;
            }
            _sys.StatsMsgTimer.Change((long)_sys.ClientStatszInterval.TotalMilliseconds, Timeout.Infinite);
        }
        Task.Run(SendStatszUpdate);
    }

    // -------------------------------------------------------------------------
    // resetLastStatsz
    // -------------------------------------------------------------------------

    /// <summary>
    /// Resets the last-sent timestamp so the next publish is not rate-limited.
    /// Must be wrapped with <see cref="WrapChk"/>.
    /// Mirrors Go <c>Server.resetLastStatsz()</c>.
    /// </summary>
    private void ResetLastStatsz()
    {
        if (_sys != null) _sys.LastStatsz = default;
    }

    // -------------------------------------------------------------------------
    // sendStatszUpdate
    // -------------------------------------------------------------------------

    /// <summary>
    /// Triggers a statsz advisory publish on the server's broadcast subject.
    /// Mirrors Go <c>Server.sendStatszUpdate()</c>.
    /// </summary>
    private void SendStatszUpdate() =>
        SendStatsz(string.Format(SystemSubjects.ServerStatsSubj, ID()));

    // -------------------------------------------------------------------------
    // startStatszTimer
    // -------------------------------------------------------------------------

    /// <summary>
    /// Starts the statsz heartbeat timer, beginning at 250ms and backing off.
    /// Must be wrapped with <see cref="WrapChk"/>.
    /// Mirrors Go <c>Server.startStatszTimer()</c>.
    /// </summary>
    private void StartStatszTimer()
    {
        _sys!.ClientStatszInterval = TimeSpan.FromMilliseconds(250);
        var delay = (long)_sys.ClientStatszInterval.TotalMilliseconds;
        _sys.StatsMsgTimer = new System.Threading.Timer(_ => WrapChk(HeartbeatStatsz)(), null, delay, Timeout.Infinite);
    }

    // -------------------------------------------------------------------------
    // startRemoteServerSweepTimer
    // -------------------------------------------------------------------------

    /// <summary>
    /// Starts the timer that periodically sweeps for orphaned remote servers.
    /// Must be wrapped with <see cref="WrapChk"/>.
    /// Mirrors Go <c>Server.startRemoteServerSweepTimer()</c>.
    /// </summary>
    private void StartRemoteServerSweepTimer()
    {
        var delay = (long)_sys!.CheckOrphan.TotalMilliseconds;
        _sys.Sweeper = new System.Threading.Timer(_ => WrapChk(CheckRemoteServers)(), null, delay, Timeout.Infinite);
    }

    // -------------------------------------------------------------------------
    // checkRemoteServers
    // -------------------------------------------------------------------------

    /// <summary>
    /// Removes remote server entries that have not sent a heartbeat within the
    /// orphan timeout window.
    /// Must be wrapped with <see cref="WrapChk"/>.
    /// Mirrors Go <c>Server.checkRemoteServers()</c>.
    /// </summary>
    private void CheckRemoteServers()
    {
        var now = DateTime.UtcNow;
        var orphans = new List<string>();
        foreach (var (sid, su) in _sys!.Servers)
        {
            if (now - su.LTime > _sys.OrphanMax)
            {
                Debugf("Detected orphan remote server: {0}", sid);
                orphans.Add(sid);
            }
        }
        foreach (var sid in orphans)
            ProcessRemoteServerShutdown(sid);

        _sys.Sweeper?.Change((long)_sys.CheckOrphan.TotalMilliseconds, Timeout.Infinite);
    }

    // =========================================================================
    // Group B: Remote Server Tracking
    // =========================================================================

    // -------------------------------------------------------------------------
    // accountClaimUpdate
    // -------------------------------------------------------------------------

    /// <summary>
    /// Handles an account JWT claim-update message from another server.
    /// Mirrors Go <c>Server.accountClaimUpdate()</c>.
    /// </summary>
    private void AccountClaimUpdate(Subscription sub, ClientConnection c, Account acc, string subject, string resp, byte[] hdr, byte[] msg)
    {
        if (!EventsEnabled()) return;

        var toks = subject.Split('.');
        string pubKey;
        if (toks.Length == SystemSubjects.AccUpdateTokensNew)
            pubKey = toks[SystemSubjects.AccReqAccIndex];
        else if (toks.Length == SystemSubjects.AccUpdateTokensOld)
            pubKey = toks[SystemSubjects.AccUpdateAccIdxOld];
        else
        {
            Debugf("Received account claims update on bad subject {0}", subject);
            return;
        }

        if (msg.Length == 0)
        {
            EventHelpers.RespondToUpdate(this, resp, pubKey, "jwt update error", new InvalidOperationException("request body is empty"));
        }
        else
        {
            // Try to update the account with the new JWT.
            if (!_accounts.TryGetValue(pubKey, out var account))
            {
                EventHelpers.RespondToUpdate(this, resp, pubKey, "jwt update skipped", null);
                return;
            }
            var err = UpdateAccountWithClaimJwt(account, Encoding.UTF8.GetString(msg));
            if (err != null)
                EventHelpers.RespondToUpdate(this, resp, pubKey, "jwt update resulted in error", err);
            else
                EventHelpers.RespondToUpdate(this, resp, pubKey, "jwt updated", null);
        }
    }

    // -------------------------------------------------------------------------
    // processRemoteServerShutdown
    // -------------------------------------------------------------------------

    /// <summary>
    /// Handles a remote server going away: cleans up account remote-server maps
    /// and removes from nodeToInfo.
    /// Server lock must be held on entry.
    /// Mirrors Go <c>Server.processRemoteServerShutdown()</c>.
    /// </summary>
    private void ProcessRemoteServerShutdown(string sid)
    {
        foreach (var acc in _accounts.Values)
            acc.RemoveRemoteServer(sid);

        foreach (var key in _nodeToInfo.Keys.ToList())
        {
            if (_nodeToInfo.TryGetValue(key, out var v) && v is NodeInfo ni && ni.Id == sid)
            {
                var updated = new NodeInfo { Name = ni.Name, Version = ni.Version, Cluster = ni.Cluster, Domain = ni.Domain, Id = ni.Id, Tags = ni.Tags, Cfg = ni.Cfg, Stats = ni.Stats, Js = ni.Js, BinarySnapshots = ni.BinarySnapshots, AccountNrg = ni.AccountNrg, Offline = true };
                _nodeToInfo.TryUpdate(key, updated, v);
                break;
            }
        }
        _sys!.Servers.Remove(sid);
    }

    // -------------------------------------------------------------------------
    // sameDomain
    // -------------------------------------------------------------------------

    /// <summary>
    /// Returns true when the given domain matches (or is not set).
    /// Mirrors Go <c>Server.sameDomain()</c>.
    /// </summary>
    internal bool SameDomain(string domain) =>
        string.IsNullOrEmpty(domain) || string.IsNullOrEmpty(_info.Domain) || domain == _info.Domain;

    // -------------------------------------------------------------------------
    // remoteServerShutdown
    // -------------------------------------------------------------------------

    /// <summary>
    /// Handles a remote server-shutdown system event.
    /// Mirrors Go <c>Server.remoteServerShutdown()</c>.
    /// </summary>
    private void RemoteServerShutdown(Subscription sub, ClientConnection c, Account acc, string subject, string reply, byte[] hdr, byte[] msg)
    {
        _mu.EnterWriteLock();
        try
        {
            if (!EventsEnabledLocked()) return;

            var toks = subject.Split('.');
            if (toks.Length < SystemSubjects.ShutdownEventTokens)
            {
                Debugf("Received remote server shutdown on bad subject {0}", subject);
                return;
            }

            if (msg.Length == 0)
            {
                Errorf("Remote server sent invalid (empty) shutdown message to {0}", subject);
                return;
            }

            ServerInfo si;
            try { si = JsonSerializer.Deserialize<ServerInfo>(msg)!; }
            catch
            {
                Debugf("Received bad server info for remote server shutdown");
                return;
            }

            // Mark the JetStream node as offline.
            var node = GetHash(si.Name);
            if (_nodeToInfo.TryGetValue(node, out var existing) && existing is NodeInfo oldNi)
            {
                var updatedNi = new NodeInfo { Name = oldNi.Name, Version = oldNi.Version, Cluster = oldNi.Cluster, Domain = oldNi.Domain, Id = oldNi.Id, Tags = oldNi.Tags, Cfg = oldNi.Cfg, Stats = oldNi.Stats, Js = oldNi.Js, BinarySnapshots = oldNi.BinarySnapshots, AccountNrg = oldNi.AccountNrg, Offline = true };
                _nodeToInfo.TryUpdate(node, updatedNi, existing);
            }

            var sid = toks[SystemSubjects.ServerSubjectIndex];
            if (_sys!.Servers.ContainsKey(sid))
                ProcessRemoteServerShutdown(sid);
        }
        finally { _mu.ExitWriteLock(); }
    }

    // -------------------------------------------------------------------------
    // remoteServerUpdate
    // -------------------------------------------------------------------------

    /// <summary>
    /// Handles a statsz heartbeat from a remote server.
    /// Mirrors Go <c>Server.remoteServerUpdate()</c>.
    /// </summary>
    private void RemoteServerUpdate(Subscription sub, ClientConnection c, Account acc, string subject, string reply, byte[] hdr, byte[] msg)
    {
        if (msg.Length == 0)
        {
            Debugf("Received empty server info for remote server update");
            return;
        }

        ServerStatsMsg ssm;
        try { ssm = JsonSerializer.Deserialize<ServerStatsMsg>(msg)!; }
        catch
        {
            Debugf("Received bad server info for remote server update");
            return;
        }

        var si = ssm.Server;

        _mu.EnterWriteLock();
        if (Running() && EventsEnabledLocked() && si.Id != _info.Id)
            UpdateRemoteServer(si);
        _mu.ExitWriteLock();

        // JetStream node updates.
        if (!SameDomain(si.Domain)) return;

        var node = GetHash(si.Name);
        _nodeToInfo[node] = new NodeInfo
        {
            Name    = si.Name,
            Id      = si.Id,
            Cluster = si.Cluster,
            Domain  = si.Domain,
            Version = si.Version,
            Offline = false,
        };
        Task.Run(UpdateNRGAccountStatus);
    }

    // -------------------------------------------------------------------------
    // updateRemoteServer
    // -------------------------------------------------------------------------

    /// <summary>
    /// Tracks or updates a remote server's heartbeat sequence and timestamp.
    /// Server lock must be held on entry.
    /// Mirrors Go <c>Server.updateRemoteServer()</c>.
    /// </summary>
    private void UpdateRemoteServer(ServerInfo si)
    {
        if (!_sys!.Servers.TryGetValue(si.Id, out var su))
        {
            _sys.Servers[si.Id] = new ServerUpdate { Seq = si.Seq, LTime = DateTime.UtcNow };
            ProcessNewServer(si);
        }
        else
        {
            if (si.Seq <= su.Seq)
            {
                Errorf("Received out of order remote server update from: {0}", si.Id);
                return;
            }
            su.Seq   = si.Seq;
            su.LTime = DateTime.UtcNow;
        }
    }

    // -------------------------------------------------------------------------
    // processNewServer
    // -------------------------------------------------------------------------

    /// <summary>
    /// Called the first time a remote server is seen.
    /// Ensures gateways are in interest-only mode for leaf-connected accounts
    /// and announces ourselves to the new server.
    /// Server lock must be held on entry.
    /// Mirrors Go <c>Server.processNewServer()</c>.
    /// </summary>
    private void ProcessNewServer(ServerInfo si)
    {
        EnsureGWsInterestOnlyForLeafNodes();

        if (SameDomain(si.Domain))
        {
            var node = GetHash(si.Name);
            _nodeToInfo.TryAdd(node, new NodeInfo
            {
                Name    = si.Name,
                Id      = si.Id,
                Cluster = si.Cluster,
                Domain  = si.Domain,
                Version = si.Version,
                Offline = false,
            });
        }
        Task.Run(UpdateNRGAccountStatus);
        Task.Run(SendStatszUpdate);
    }

    // -------------------------------------------------------------------------
    // updateNRGAccountStatus
    // -------------------------------------------------------------------------

    /// <summary>
    /// Recreates internal subscriptions on all Raft nodes to reflect updated
    /// account-NRG capability state.
    /// Server lock must NOT be held on entry.
    /// Mirrors Go <c>Server.updateNRGAccountStatus()</c>.
    /// </summary>
    private void UpdateNRGAccountStatus()
    {
        var raftNodes = new List<object>(_raftNodes.Values);
        foreach (var n in raftNodes)
        {
            if (n is IRaftNode rn)
            {
                rn.RecreateInternalSubs();
            }
        }
    }

    // =========================================================================
    // Group C: Connection Events
    // =========================================================================

    // -------------------------------------------------------------------------
    // connsRequest
    // -------------------------------------------------------------------------

    /// <summary>
    /// Handles a request for this server's local connection count for a given account.
    /// Mirrors Go <c>Server.connsRequest()</c>.
    /// </summary>
    private void ConnsRequest(Subscription sub, ClientConnection c, Account acc, string subject, string reply, byte[] hdr, byte[] msg)
    {
        if (!EventsRunning()) return;

        var tk = subject.Split('.');
        if (tk.Length != SystemSubjects.AccReqTokens)
        {
            Errorf("Bad subject account connections request message");
            return;
        }

        var a = tk[SystemSubjects.AccReqAccIndex];
        var m = new AccNumConnsReq { Account = a };
        if (msg.Length > 0)
        {
            try { m = JsonSerializer.Deserialize<AccNumConnsReq>(msg) ?? m; }
            catch (Exception ex)
            {
                Errorf("Error unmarshalling account connections request message: {0}", ex);
                return;
            }
        }
        if (m.Account != a)
        {
            Errorf("Error unmarshalled account does not match subject");
            return;
        }

        if (!_accounts.TryGetValue(m.Account, out var account)) return;
        if (account.NumLocalConnections() <= 0) return;

        _mu.EnterWriteLock();
        SendAccConnsUpdate(account, reply);
        _mu.ExitWriteLock();
    }

    // -------------------------------------------------------------------------
    // leafNodeConnected
    // -------------------------------------------------------------------------

    /// <summary>
    /// Handles a leaf-node-connect event, switching the account to gateway
    /// interest-only mode.
    /// Mirrors Go <c>Server.leafNodeConnected()</c>.
    /// </summary>
    private void LeafNodeConnected(Subscription sub, ClientConnection _, Account acc, string subject, string reply, byte[] hdr, byte[] msg)
    {
        AccNumConnsReq m;
        try { m = JsonSerializer.Deserialize<AccNumConnsReq>(msg) ?? new AccNumConnsReq(); }
        catch (Exception ex)
        {
            Errorf("Error unmarshalling account connections request message: {0}", ex);
            return;
        }

        _mu.EnterReadLock();
        bool na = string.IsNullOrEmpty(m.Account) || !EventsEnabledLocked() || !_gateway.Enabled;
        _mu.ExitReadLock();
        if (na) return;

        var (account, _) = LookupAccount(m.Account);
        if (account != null)
            SwitchAccountToInterestMode(account.Name);
    }

    // -------------------------------------------------------------------------
    // remoteConnsUpdate
    // -------------------------------------------------------------------------

    /// <summary>
    /// Handles a remote account connection count update.
    /// Mirrors Go <c>Server.remoteConnsUpdate()</c>.
    /// </summary>
    private void RemoteConnsUpdate(Subscription sub, ClientConnection c, Account acc, string subject, string reply, byte[] hdr, byte[] msg)
    {
        if (!EventsRunning()) return;

        if (msg.Length == 0) { Errorf("No message body provided"); return; }

        AccountNumConns m;
        try { m = JsonSerializer.Deserialize<AccountNumConns>(msg)!; }
        catch (Exception ex)
        {
            Errorf("Error unmarshalling account connection event message: {0}", ex);
            return;
        }

        if (!_accounts.TryGetValue(m.Account, out var account)) return;

        _mu.EnterWriteLock();
        if (!Running() || !EventsEnabledLocked()) { _mu.ExitWriteLock(); return; }
        if (m.Server.Id == _info.Id)
        {
            _sys!.Client?.Errorf("Processing our own account connection event message: ignored");
            _mu.ExitWriteLock();
            return;
        }
        var clients = account.UpdateRemoteServer(m);
        UpdateRemoteServer(m.Server);
        _mu.ExitWriteLock();

        foreach (var cl in clients)
            cl.MaxAccountConnExceeded();
    }

    // -------------------------------------------------------------------------
    // accConnsUpdate
    // -------------------------------------------------------------------------

    /// <summary>
    /// Sends an account connection count advisory whenever the count changes or
    /// on a heartbeat.
    /// Mirrors Go <c>Server.accConnsUpdate()</c>.
    /// </summary>
    internal void AccConnsUpdate(Account a)
    {
        _mu.EnterWriteLock();
        try
        {
            if (!EventsEnabledLocked() || a == null || a == _gacc) return;
            SendAccConnsUpdate(a,
                string.Format(SystemSubjects.AccConnsEventSubjOld, a.Name),
                string.Format(SystemSubjects.AccConnsEventSubjNew, a.Name));
        }
        finally { _mu.ExitWriteLock(); }
    }

    // -------------------------------------------------------------------------
    // accountConnectEvent
    // -------------------------------------------------------------------------

    /// <summary>
    /// Publishes a client-connect advisory for billing/tracking purposes.
    /// Mirrors Go <c>Server.accountConnectEvent()</c>.
    /// </summary>
    internal void AccountConnectEvent(ClientConnection c)
    {
        _mu.EnterWriteLock();
        if (!EventsEnabledLocked()) { _mu.ExitWriteLock(); return; }
        var eid = NextEventID();
        _mu.ExitWriteLock();

        Account? account;
        string host, name, lang, version, jwt, kind, clientType, accName, nameTag;
        ulong cid;
        DateTime start;
        List<string> tags;
        string issuerKey, user;

        lock (c)
        {
            account = c._account as Account;
            if (account == null) return;

            cid        = c.Cid;
            start      = c.Start;
            host       = c.Host;
            name       = c.Opts.Name;
            lang       = c.Opts.Lang;
            version    = c.Opts.Version;
            jwt        = c.Opts.Jwt;
            kind       = c.KindString();
            clientType = c.ClientTypeString();
            accName    = account.Name;
            nameTag    = account.GetNameTag();
            tags       = [];
            issuerKey  = EventHelpers.IssuerForClient(c);
            user       = c.GetRawAuthUser();
        }

        var m = new ConnectEventMsg
        {
            Type  = EventMsgTypes.ConnectEventMsgType,
            Id    = eid,
            Time  = DateTime.UtcNow,
            Client = new ClientInfo
            {
                Start      = start.ToString("O"),
                Host       = host,
                Id         = cid,
                Account    = EventHelpers.AccForClient(c),
                User       = user,
                Name       = name,
                Lang       = lang,
                Version    = version,
                Jwt        = jwt,
                IssuerKey  = issuerKey,
                Tags       = tags,
                NameTag    = nameTag,
                Kind       = kind,
                ClientType = clientType,
            },
        };

        var subj = string.Format(SystemSubjects.ConnectEventSubj, accName);
        SendInternalMsgLocked(subj, string.Empty, m.Server, m);
    }

    // -------------------------------------------------------------------------
    // sendAccConnsUpdate
    // -------------------------------------------------------------------------

    /// <summary>
    /// Builds and queues an <see cref="AccountNumConns"/> advisory.
    /// Server lock must be held on entry.
    /// Mirrors Go <c>Server.sendAccConnsUpdate()</c>.
    /// </summary>
    internal void SendAccConnsUpdate(Account a, params string[] subjects)
    {
        if (!EventsEnabledLocked() || a == null) return;
        var sendq = _sys!.SendQueue;
        if (sendq == null) return;

        var eid = NextEventID();

        int aConns, aLeafs, aNumSubs;
        lock (a)
        {
            aConns  = a.NumLocalConnections();
            aLeafs  = a.NumLocalLeafNodes();
            aNumSubs = a.TotalSubs();
        }

        var m = new AccountNumConns
        {
            Type        = EventMsgTypes.AccountNumConnsMsgType,
            Id          = eid,
            Time        = DateTime.UtcNow,
            Account     = a.Name,
            Name        = a.GetNameTagLocked(),
            Conns       = aConns,
            LeafNodes   = aLeafs,
            TotalConns  = aConns + aLeafs,
            NumSubs     = (uint)aNumSubs,
            Sent        = new DataStats { Msgs = 0L, Bytes = 0L },
            Received    = new DataStats { Msgs = 0L, Bytes = 0L },
            SlowConsumers = 0,
        };

        // Manage the account heartbeat timer.
        lock (a)
        {
            if (m.TotalConns == 0)
            {
                a.ClearConnectionHeartbeatTimer();
            }
            else
            {
                a.SetConnectionHeartbeatTimer((long)EventIntervals.EventsHbInterval.TotalMilliseconds, () => AccConnsUpdate(a));
            }
        }

        foreach (var subj in subjects)
        {
            sendq.Push(new PubMsg
            {
                Client  = null,
                Subject = subj,
                Reply   = string.Empty,
                Si      = m.Server,
                Hdr     = null,
                Msg     = m,
            });
        }
    }

    // =========================================================================
    // Group D: Request Handlers
    // =========================================================================

    // -------------------------------------------------------------------------
    // userInfoReq
    // -------------------------------------------------------------------------

    /// <summary>
    /// Handles a user-info request, returning the caller's account and permissions.
    /// Mirrors Go <c>Server.userInfoReq()</c>.
    /// </summary>
    private void UserInfoReq(Subscription sub, ClientConnection c, Account acc, string subject, string reply, byte[] hdr, byte[] msg)
    {
        if (!EventsEnabled() || string.IsNullOrEmpty(reply)) return;

        var response = new ServerApiResponse { Server = new ServerInfo() };

        var (ci, _, _, _, err) = GetRequestInfo(c, msg);
        if (err != null)
        {
            response.Error = new ServerApiError { Code = (int)HttpStatusCode.BadRequest };
            SendInternalResponse(reply, response);
            return;
        }

        response.Data = new UserInfo
        {
            UserId      = ci?.User  ?? string.Empty,
            Account     = ci?.Account ?? string.Empty,
            Permissions = c.PublicPermissions(),
            Expires     = c.Expires,
        };
        SendInternalResponse(reply, response);
    }

    // -------------------------------------------------------------------------
    // statszReq
    // -------------------------------------------------------------------------

    /// <summary>
    /// Handles a statsz request from another server.
    /// Mirrors Go <c>Server.statszReq()</c>.
    /// </summary>
    private void StatszReq(Subscription sub, ClientConnection c, Account acc, string subject, string reply, byte[] hdr, byte[] msg)
    {
        if (!EventsEnabled()) return;

        if (string.IsNullOrEmpty(reply))
        {
            reply = string.Format(SystemSubjects.ServerStatsSubj, _info.Id);
            WrapChk(ResetLastStatsz)();
        }

        if (msg.Length != 0)
        {
            StatszEventOptions opts;
            try { opts = JsonSerializer.Deserialize<StatszEventOptions>(msg) ?? new StatszEventOptions(); }
            catch (Exception ex)
            {
                var errResp = new ServerApiResponse
                {
                    Server = new ServerInfo(),
                    Error  = new ServerApiError { Code = (int)HttpStatusCode.BadRequest, Description = ex.Message },
                };
                SendInternalMsgLocked(reply, string.Empty, null, errResp);
                return;
            }
            if (FilterRequest(opts)) return;
        }

        SendStatsz(reply);
    }

    // -------------------------------------------------------------------------
    // idzReq
    // -------------------------------------------------------------------------

    /// <summary>
    /// Handles a request for basic static server identity.
    /// Mirrors Go <c>Server.idzReq()</c>.
    /// </summary>
    private void IdzReq(Subscription sub, ClientConnection c, Account acc, string subject, string reply, byte[] hdr, byte[] msg)
    {
        _mu.EnterReadLock();
        var id = new ServerIdentity { Name = _info.Name, Host = _info.Host, Id = _info.Id };
        SendInternalMsg(reply, string.Empty, null, id);
        _mu.ExitReadLock();
    }

    // -------------------------------------------------------------------------
    // zReq — generic monitoring request handler
    // -------------------------------------------------------------------------

    /// <summary>
    /// Generic z-endpoint request handler that deserialises options, filters by
    /// server, and sends a <see cref="ServerApiResponse"/>.
    /// Mirrors Go <c>Server.zReq()</c>.
    /// </summary>
    internal void ZReq(ClientConnection c, string reply, byte[] hdr, byte[] msg, EventFilterOptions? fOpts, object? optz, Func<(object? data, Exception? err)> respf)
    {
        if (!EventsEnabled() || string.IsNullOrEmpty(reply)) return;

        var response = new ServerApiResponse { Server = new ServerInfo() };
        int status = 0;
        Exception? err = null;

        if (msg.Length != 0 && optz != null)
        {
            try
            {
                var json = JsonSerializer.Deserialize(msg, optz.GetType());
                if (json != null)
                {
                    // Copy values — simplest approach: re-use the deserialized object.
                    optz = json;
                }
            }
            catch (Exception ex)
            {
                err    = ex;
                status = (int)HttpStatusCode.BadRequest;
            }

            if (err == null && fOpts != null && FilterRequest(fOpts)) return;
        }

        if (err == null)
        {
            var (data, callErr) = respf();
            if (callErr?.Message == "filtered response")
                return;
            if (callErr != null)
            {
                err    = callErr;
                status = (int)HttpStatusCode.InternalServerError;
            }
            else
            {
                response.Data = data;
            }
        }

        if (err != null)
            response.Error = new ServerApiError { Code = status, Description = err.Message };

        SendInternalResponse(reply, response);
    }

    // -------------------------------------------------------------------------
    // filterRequest
    // -------------------------------------------------------------------------

    /// <summary>
    /// Returns true if the request does NOT apply to this server (should be ignored).
    /// Mirrors Go <c>Server.filterRequest()</c>.
    /// </summary>
    internal bool FilterRequest(EventFilterOptions? fOpts)
    {
        if (fOpts == null) return false;

        if (fOpts.ExactMatch)
        {
            if ((!string.IsNullOrEmpty(fOpts.Name)    && fOpts.Name    != _info.Name) ||
                (!string.IsNullOrEmpty(fOpts.Host)    && fOpts.Host    != _info.Host) ||
                (!string.IsNullOrEmpty(fOpts.Cluster) && fOpts.Cluster != ClusterName()))
                return true;
        }
        else
        {
            if ((!string.IsNullOrEmpty(fOpts.Name)    && !_info.Name.Contains(fOpts.Name))    ||
                (!string.IsNullOrEmpty(fOpts.Host)    && !_info.Host.Contains(fOpts.Host))    ||
                (!string.IsNullOrEmpty(fOpts.Cluster) && !ClusterName().Contains(fOpts.Cluster)))
                return true;
        }

        if (fOpts.Tags?.Count > 0)
        {
            var opts = GetOpts();
            foreach (var t in fOpts.Tags)
                if (!opts.Tags.Contains(t)) return true;
        }

        if (!string.IsNullOrEmpty(fOpts.Domain) && GetOpts().JetStreamDomain != fOpts.Domain)
            return true;

        return false;
    }

    // -------------------------------------------------------------------------
    // debugSubscribers
    // -------------------------------------------------------------------------

    /// <summary>
    /// Handles a debug request for the count of subscribers matching a given subject.
    /// Mirrors Go <c>Server.debugSubscribers()</c>.
    /// </summary>
    private void DebugSubscribers(Subscription sub, ClientConnection c, Account acc, string subject, string reply, byte[] hdr, byte[] msg)
    {
        if (c.Kind != ClientKind.Client) return;

        // Parse the message body to get the subject and optional queue group.
        var args = Encoding.UTF8.GetString(msg).Trim().Split(' ', StringSplitOptions.RemoveEmptyEntries);
        if (args.Length == 0)
        {
            SendInternalAccountMsg(null, reply, 0);
            return;
        }

        var tsubj  = args[0];
        var qgroup = args.Length > 1 ? args[1] : null;

        // Look up the account from the header or the client.
        Account? lookupAcc = acc ?? _sysAccAtomic ?? _gacc;

        if (lookupAcc == null) return;

        // Count local subscribers.
        var nsubs = CountLocalSubscribers(lookupAcc, tsubj, qgroup);

        int expected = lookupAcc.ExpectedRemoteResponses();
        if (expected == 0)
        {
            SendInternalAccountMsg(null, reply, nsubs);
            return;
        }

        // Solicit remote counts.
        int responses = 0;
        var done = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);

        _mu.EnterWriteLock();
        var replySubj = NewRespInbox();
        _sys!.Replies[replySubj] = (_, __, ___, ____, _____, ______, rMsg) =>
        {
            if (long.TryParse(Encoding.UTF8.GetString(rMsg), out var n))
                Interlocked.Add(ref nsubs, (int)n);
            if (Interlocked.Increment(ref responses) >= expected)
                done.TrySetResult(true);
        };
        var request = new AccNumSubsReq { Account = lookupAcc.Name, Subject = tsubj, Queue = qgroup };
        SendInternalMsg(SystemSubjects.AccNumSubsReqSubj, replySubj, null, request);
        _mu.ExitWriteLock();

        Task.Run(async () =>
        {
            await Task.WhenAny(done.Task, Task.Delay(500)).ConfigureAwait(false);
            bool send = false;
            _mu.EnterWriteLock();
            if (_sys != null && _sys.Replies != null)
            {
                _sys.Replies.Remove(replySubj);
                send = true;
            }
            _mu.ExitWriteLock();
            if (send)
                SendInternalAccountMsg(null, reply, Interlocked.Add(ref nsubs, 0));
        });
    }

    private static int CountLocalSubscribers(Account acc, string subj, string? queue)
    {
        if (acc.Sublist == null) return 0;
        var rr = acc.Sublist.Match(subj);
        int count = 0;
        foreach (var s in rr.PSubs)
        {
            if (queue != null && (s.Queue == null || System.Text.Encoding.ASCII.GetString(s.Queue) != queue)) continue;
            if (s.Client?.Kind == ClientKind.Client) count++;
        }
        foreach (var qsubs in rr.QSubs)
            foreach (var s in qsubs)
            {
                if (queue != null && (s.Queue == null || System.Text.Encoding.ASCII.GetString(s.Queue) != queue)) continue;
                if (s.Client?.Kind == ClientKind.Client) count++;
            }
        return count;
    }

    // -------------------------------------------------------------------------
    // nsubsRequest
    // -------------------------------------------------------------------------

    /// <summary>
    /// Handles a request for local subscription counts from a remote origin server.
    /// Mirrors Go <c>Server.nsubsRequest()</c>.
    /// </summary>
    private void NsubsRequest(Subscription sub, ClientConnection c, Account acc, string subject, string reply, byte[] hdr, byte[] msg)
    {
        if (!EventsRunning()) return;

        if (msg.Length == 0) { Errorf("request requires a body"); return; }

        AccNumSubsReq m;
        try { m = JsonSerializer.Deserialize<AccNumSubsReq>(msg)!; }
        catch (Exception ex) { Errorf("Error unmarshalling account nsubs request message: {0}", ex); return; }

        var (account, _) = LookupAccount(m.Account);
        if (account == null || account.NumLocalAndLeafConnections() == 0) return;

        var nsubs = CountLocalSubscribers(account, m.Subject, m.Queue);
        SendInternalMsgLocked(reply, string.Empty, null, nsubs);
    }

    // -------------------------------------------------------------------------
    // reloadConfig
    // -------------------------------------------------------------------------

    /// <summary>
    /// Handles a server-reload request from the system bus.
    /// Mirrors Go <c>Server.reloadConfig()</c>.
    /// </summary>
    private void ReloadConfigHandler(Subscription sub, ClientConnection c, Account acc, string subject, string reply, byte[] hdr, byte[] msg)
    {
        if (!EventsRunning()) return;
        var fOpts = new EventFilterOptions();
        ZReq(c, reply, hdr, msg, fOpts, fOpts, () => (null, ReloadConfigFromRequest()));
    }

    /// <summary>
    /// Reloads server configuration initiated from a system bus request.
    /// Returns any error.
    /// Mirrors Go <c>Server.Reload()</c> called from the reloadConfig handler.
    /// </summary>
    private Exception? ReloadConfigFromRequest()
    {
        var newOpts = GetOpts().Clone();
        return ReloadOptions(newOpts);
    }

    // =========================================================================
    // Group E: Leaf Node & Gateway Events
    // =========================================================================

    // -------------------------------------------------------------------------
    // ensureGWsInterestOnlyForLeafNodes
    // -------------------------------------------------------------------------

    /// <summary>
    /// Sends a leaf-node connect message for every account that has leaf node
    /// connections, so that gateway peers switch to interest-only mode.
    /// Server lock must be held on entry.
    /// Mirrors Go <c>Server.ensureGWsInterestOnlyForLeafNodes()</c>.
    /// </summary>
    private void EnsureGWsInterestOnlyForLeafNodes()
    {
        if (!_gateway.Enabled || _leafs.Count == 0) return;

        var sent = new HashSet<Account>();
        foreach (var c in _leafs.Values)
        {
            if (c._account is Account a && sent.Add(a))
                SendLeafNodeConnectMsg(a.Name);
        }
    }

    // -------------------------------------------------------------------------
    // sendLeafNodeConnect
    // -------------------------------------------------------------------------

    /// <summary>
    /// Sends a leaf-node connect system event for the given account (lock-free entry).
    /// Mirrors Go <c>Server.sendLeafNodeConnect()</c>.
    /// </summary>
    internal void SendLeafNodeConnect(Account a)
    {
        _mu.EnterWriteLock();
        if (a == null || !EventsEnabledLocked() || !_gateway.Enabled)
        {
            _mu.ExitWriteLock();
            return;
        }
        SendLeafNodeConnectMsg(a.Name);
        _mu.ExitWriteLock();

        SwitchAccountToInterestMode(a.Name);
    }

    // -------------------------------------------------------------------------
    // sendLeafNodeConnectMsg
    // -------------------------------------------------------------------------

    /// <summary>
    /// Queues the internal leaf-node-connect message for the given account name.
    /// Server lock must be held on entry.
    /// Mirrors Go <c>Server.sendLeafNodeConnectMsg()</c>.
    /// </summary>
    internal void SendLeafNodeConnectMsg(string accName)
    {
        var subj = string.Format(SystemSubjects.LeafNodeConnectEventSubj, accName);
        var m    = new AccNumConnsReq { Account = accName };
        SendInternalMsg(subj, string.Empty, m.Server, m);
    }

    // =========================================================================
    // Group F: Auth & Error Events
    // =========================================================================

    // -------------------------------------------------------------------------
    // sendAuthErrorEvent
    // -------------------------------------------------------------------------

    /// <summary>
    /// Sends a system-level auth error advisory to the system account.
    /// Mirrors Go <c>Server.sendAuthErrorEvent()</c>.
    /// </summary>
    internal void SendAuthErrorEvent(ClientConnection c, string reason)
    {
        _mu.EnterWriteLock();
        if (!EventsEnabledLocked()) { _mu.ExitWriteLock(); return; }
        var eid = NextEventID();
        _mu.ExitWriteLock();

        var now = DateTime.UtcNow;
        DisconnectEventMsg m;
        lock (c)
        {
            m = BuildDisconnectMsg(c, eid, now, reason);
        }

        _mu.EnterWriteLock();
        var subj = string.Format(SystemSubjects.AuthErrorEventSubj, _info.Id);
        SendInternalMsg(subj, string.Empty, m.Server, m);
        _mu.ExitWriteLock();
    }

    // -------------------------------------------------------------------------
    // sendAccountAuthErrorEvent
    // -------------------------------------------------------------------------

    /// <summary>
    /// Sends an account-level auth error advisory to the account's system stream.
    /// Mirrors Go <c>Server.sendAccountAuthErrorEvent()</c>.
    /// </summary>
    internal void SendAccountAuthErrorEvent(ClientConnection c, Account? acc, string reason)
    {
        if (acc == null) return;
        _mu.EnterWriteLock();
        if (!EventsEnabledLocked()) { _mu.ExitWriteLock(); return; }
        var eid = NextEventID();
        _mu.ExitWriteLock();

        var now = DateTime.UtcNow;
        DisconnectEventMsg m;
        lock (c)
        {
            m = BuildDisconnectMsg(c, eid, now, reason, acc.Name);
        }

        SendInternalAccountSysMsg(acc, SystemSubjects.AuthErrorAccountEventSubj, m.Server, m, 0);
    }

    private DisconnectEventMsg BuildDisconnectMsg(ClientConnection c, string eid, DateTime now, string reason, string? overrideAccName = null)
    {
        return new DisconnectEventMsg
        {
            Type   = EventMsgTypes.DisconnectEventMsgType,
            Id     = eid,
            Time   = now,
            Client = new ClientInfo
            {
                Start      = c.Start.ToString("O"),
                Host       = c.Host,
                Id         = c.Cid,
                Account    = overrideAccName ?? EventHelpers.AccForClient(c),
                User       = c.GetRawAuthUser(),
                Name       = c.Opts.Name,
                Lang       = c.Opts.Lang,
                Version    = c.Opts.Version,
                Rtt        = c.GetRttValue(),
                Jwt        = c.Opts.Jwt,
                IssuerKey  = EventHelpers.IssuerForClient(c),
                Kind       = c.KindString(),
                ClientType = c.ClientTypeString(),
            },
            Sent     = new DataStats { Msgs = 0L, Bytes = 0L },
            Received = new DataStats { Msgs = 0L, Bytes = 0L },
            Reason   = reason,
        };
    }

    // -------------------------------------------------------------------------
    // registerSystemImportsForExisting
    // -------------------------------------------------------------------------

    /// <summary>
    /// Registers system imports for all existing accounts (excluding the system account).
    /// Mirrors Go <c>Server.registerSystemImportsForExisting()</c>.
    /// </summary>
    internal void RegisterSystemImportsForExisting()
    {
        _mu.EnterReadLock();
        if (_sys == null) { _mu.ExitReadLock(); return; }
        var sacc = _sys.Account;
        var accounts = _accounts.Values.Where(a => a != sacc).ToList();
        _mu.ExitReadLock();

        foreach (var a in accounts)
            RegisterSystemImports(a);
    }

    // =========================================================================
    // Group G: OCSP Events
    // =========================================================================

    // -------------------------------------------------------------------------
    // sendOCSPPeerRejectEvent
    // -------------------------------------------------------------------------

    /// <summary>
    /// Publishes an OCSP peer-reject advisory to the system account.
    /// Mirrors Go <c>Server.sendOCSPPeerRejectEvent()</c>.
    /// </summary>
    internal void SendOcspPeerRejectEvent(string kind, X509Certificate2? peer, string reason)
    {
        _mu.EnterWriteLock();
        try
        {
            if (!EventsEnabledLocked()) return;
            if (peer == null)
            {
                Errorf(OcspMessages.ErrPeerEmptyNoEvent);
                return;
            }
            var eid = NextEventID();
            var now = DateTime.UtcNow;
            var m = new OcspPeerRejectEventMsg
            {
                Type   = EventMsgTypes.OcspPeerRejectEventMsgType,
                Id     = eid,
                Time   = now,
                Kind   = kind,
                Peer   = new CertInfo
                {
                    Subject     = OcspUtilities.GetSubjectDNForm(peer),
                    Issuer      = OcspUtilities.GetIssuerDNForm(peer),
                    Fingerprint = OcspUtilities.GenerateFingerprint(peer),
                    Raw         = peer.RawData,
                },
                Reason = reason,
            };
            var subj = string.Format(SystemSubjects.OcspPeerRejectEventSubj, _info.Id);
            SendInternalMsg(subj, string.Empty, m.Server, m);
        }
        finally { _mu.ExitWriteLock(); }
    }

    // -------------------------------------------------------------------------
    // sendOCSPPeerChainlinkInvalidEvent
    // -------------------------------------------------------------------------

    /// <summary>
    /// Publishes an OCSP chainlink-invalid advisory to the system account.
    /// Mirrors Go <c>Server.sendOCSPPeerChainlinkInvalidEvent()</c>.
    /// </summary>
    internal void SendOcspPeerChainlinkInvalidEvent(X509Certificate2? peer, X509Certificate2? link, string reason)
    {
        _mu.EnterWriteLock();
        try
        {
            if (!EventsEnabledLocked()) return;
            if (peer == null || link == null)
            {
                Errorf(OcspMessages.ErrPeerEmptyNoEvent);
                return;
            }
            var eid = NextEventID();
            var now = DateTime.UtcNow;
            var m = new OcspPeerChainlinkInvalidEventMsg
            {
                Type   = EventMsgTypes.OcspPeerChainlinkInvalidEventMsgType,
                Id     = eid,
                Time   = now,
                Link   = new CertInfo
                {
                    Subject     = OcspUtilities.GetSubjectDNForm(link),
                    Issuer      = OcspUtilities.GetIssuerDNForm(link),
                    Fingerprint = OcspUtilities.GenerateFingerprint(link),
                    Raw         = link.RawData,
                },
                Peer   = new CertInfo
                {
                    Subject     = OcspUtilities.GetSubjectDNForm(peer),
                    Issuer      = OcspUtilities.GetIssuerDNForm(peer),
                    Fingerprint = OcspUtilities.GenerateFingerprint(peer),
                    Raw         = peer.RawData,
                },
                Reason = reason,
            };
            var subj = string.Format(SystemSubjects.OcspPeerChainlinkInvalidEventSubj, _info.Id);
            SendInternalMsg(subj, string.Empty, m.Server, m);
        }
        finally { _mu.ExitWriteLock(); }
    }

    // =========================================================================
    // Group H: Misc
    // =========================================================================

    // -------------------------------------------------------------------------
    // remoteLatencyUpdate
    // -------------------------------------------------------------------------

    /// <summary>
    /// Handles a remote latency measurement event for tracking exported service latency.
    /// Mirrors Go <c>Server.remoteLatencyUpdate()</c>.
    /// </summary>
    private void RemoteLatencyUpdate(Subscription sub, ClientConnection c, Account acc, string subject, string reply, byte[] hdr, byte[] msg)
    {
        if (!EventsRunning()) return;

        RemoteLatency rl;
        try { rl = JsonSerializer.Deserialize<RemoteLatency>(msg)!; }
        catch (Exception ex)
        {
            Errorf("Error unmarshalling remote latency measurement: {0}", ex);
            return;
        }

        var (account, err) = LookupAccount(rl.Account);
        if (err != null)
        {
            Warnf("Could not lookup account {0} for latency measurement", rl.Account);
            return;
        }

        // Look up the service import associated with this response.
        ServiceImportEntry? si = null;
        string lsub = string.Empty;
        lock (account!)
        {
            var exports = account.Exports;
            if (exports.Responses?.TryGetValue(rl.RequestId, out var entry) == true)
            {
                si   = entry;
                lsub = entry.Latency?.Subject ?? string.Empty;
            }
        }
        if (si == null) return;

        ServiceLatency? m1 = null;
        var m2 = rl.M2;

        lock (si.Account!)
        {
            m1 = si.M1;
            if (m1 == null)
                si.M1 = m2;
        }
        if (m1 == null) return;

        // Merge the two halves of the latency measurement.
        // Merge responder (m2) into requestor (m1):
        // m2.ServiceLatencyDuration is correct, m1.TotalLatency is correct.
        var responderRtt = m2.Responder?.Rtt ?? TimeSpan.Zero;
        m1.SystemLatency = m1.ServiceLatencyDuration - (m2.ServiceLatencyDuration + responderRtt);
        if (m1.SystemLatency < TimeSpan.Zero) m1.SystemLatency = TimeSpan.Zero;
        m1.ServiceLatencyDuration = m2.ServiceLatencyDuration;
        if (m1.ServiceLatencyDuration < TimeSpan.Zero) m1.ServiceLatencyDuration = TimeSpan.Zero;
        m1.Responder = m2.Responder;

        lock (account)
        {
            si.RequestingClient = null;
        }

        SendInternalAccountMsg(account, lsub, m1);
    }

    // -------------------------------------------------------------------------
    // kickClient
    // -------------------------------------------------------------------------

    /// <summary>
    /// Handles a request to forcefully disconnect a client by CID.
    /// Mirrors Go <c>Server.kickClient()</c>.
    /// </summary>
    private void KickClient(Subscription sub, ClientConnection c, Account acc, string subject, string reply, byte[] hdr, byte[] msg)
    {
        if (!EventsRunning()) return;

        KickClientReq req;
        try { req = JsonSerializer.Deserialize<KickClientReq>(msg)!; }
        catch (Exception ex) { Errorf("Error unmarshalling kick client request: {0}", ex); return; }

        var fOpts = new EventFilterOptions();
        ZReq(c, reply, hdr, msg, fOpts, fOpts, () => (null, DisconnectClientByID(req.Cid)));
    }

    // -------------------------------------------------------------------------
    // ldmClient
    // -------------------------------------------------------------------------

    /// <summary>
    /// Handles a request to put a specific client into lame-duck mode.
    /// Mirrors Go <c>Server.ldmClient()</c>.
    /// </summary>
    private void LdmClient(Subscription sub, ClientConnection c, Account acc, string subject, string reply, byte[] hdr, byte[] msg)
    {
        if (!EventsRunning()) return;

        LdmClientReq req;
        try { req = JsonSerializer.Deserialize<LdmClientReq>(msg)!; }
        catch (Exception ex) { Errorf("Error unmarshalling LDM client request: {0}", ex); return; }

        var fOpts = new EventFilterOptions();
        ZReq(c, reply, hdr, msg, fOpts, fOpts, () => (null, LDMClientByID(req.Cid)));
    }

    // =========================================================================
    // Monitoring verb stub handlers
    // (Full monitor implementations are in NatsServer.Monitor — these delegate.)
    // =========================================================================

    private void VarzReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void SubszReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void ConnzReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void RoutezReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void GatewayzReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void LeafzReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void AccountzReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void JszReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void HealthzReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void ExpvarzReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void IpqueueszReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void RaftzReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void AccSubszReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void AccConnzReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void AccLeafzReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void AccJszReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void AccInfoReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void AccStatzReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));

    private void StatzAccPingReq(Subscription s, ClientConnection c, Account a, string subj, string rply, byte[] hdr, byte[] msg)
        => ZReq(c, rply, hdr, msg, null, new StatszEventOptions(), () => (null, null));
}
