Merge branch 'feature/system-account-types'

Add SYSTEM and ACCOUNT connection types with InternalClient,
InternalEventSystem, system event publishing, request-reply services,
and cross-account import/export support.

src/NATS.Server/NatsServer.cs

@@ -9,6 +9,8 @@ using Microsoft.Extensions.Logging;
 using NATS.NKeys;
 using NATS.Server.Auth;
 using NATS.Server.Configuration;
+using NATS.Server.Events;
+using NATS.Server.Imports;
 using NATS.Server.Monitoring;
 using NATS.Server.Protocol;
 using NATS.Server.Subscriptions;
@@ -36,6 +38,7 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
     private string? _configDigest;
     private readonly Account _globalAccount;
     private readonly Account _systemAccount;
+    private InternalEventSystem? _eventSystem;
     private readonly SslServerAuthenticationOptions? _sslOptions;
     private readonly TlsRateLimiter? _tlsRateLimiter;
     private readonly SubjectTransform[] _subjectTransforms;
@@ -73,6 +76,7 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
     public int Port => _options.Port;
     public Account SystemAccount => _systemAccount;
     public string ServerNKey { get; }
+    public InternalEventSystem? EventSystem => _eventSystem;
     public bool IsShuttingDown => Volatile.Read(ref _shutdown) != 0;
     public bool IsLameDuckMode => Volatile.Read(ref _lameDuck) != 0;
     public Action? ReOpenLogFile { get; set; }
@@ -93,6 +97,21 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
 
         _logger.LogInformation("Initiating Shutdown...");
 
+        // Publish shutdown advisory before tearing down the event system
+        if (_eventSystem != null)
+        {
+            var shutdownSubject = string.Format(EventSubjects.ServerShutdown, _serverInfo.ServerId);
+            _eventSystem.Enqueue(new PublishMessage
+            {
+                Subject = shutdownSubject,
+                Body = new ShutdownEventMsg { Server = BuildEventServerInfo(), Reason = "Server Shutdown" },
+                IsLast = true,
+            });
+            // Give the send loop time to process the shutdown event
+            await Task.Delay(100);
+            await _eventSystem.DisposeAsync();
+        }
+
         // Signal all internal loops to stop
         await _quitCts.CancelAsync();
 
@@ -272,6 +291,14 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
         _systemAccount = new Account("$SYS");
         _accounts["$SYS"] = _systemAccount;
 
+        // Create system internal client and event system
+        var sysClientId = Interlocked.Increment(ref _nextClientId);
+        var sysClient = new InternalClient(sysClientId, ClientKind.System, _systemAccount);
+        _eventSystem = new InternalEventSystem(
+            _systemAccount, sysClient,
+            options.ServerName ?? $"nats-dotnet-{Environment.MachineName}",
+            _loggerFactory.CreateLogger<InternalEventSystem>());
+
         // Generate Ed25519 server NKey identity
         using var serverKeyPair = KeyPair.CreatePair(PrefixByte.Server);
         ServerNKey = serverKeyPair.GetPublicKey();
@@ -416,6 +443,9 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
 
         _listeningStarted.TrySetResult();
 
+        _eventSystem?.Start(this);
+        _eventSystem?.InitEventTracking(this);
+
         var tmpDelay = AcceptMinSleep;
 
         try
@@ -728,6 +758,27 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
             }
         }
 
+        // Check for service imports that match this subject.
+        // When a client in the importer account publishes to a subject
+        // that matches a service import "From" pattern, we forward the
+        // message to the destination (exporter) account's subscribers
+        // using the mapped "To" subject.
+        if (sender.Account != null)
+        {
+            foreach (var kvp in sender.Account.Imports.Services)
+            {
+                foreach (var si in kvp.Value)
+                {
+                    if (si.Invalid) continue;
+                    if (SubjectMatch.MatchLiteral(subject, si.From))
+                    {
+                        ProcessServiceImport(si, subject, replyTo, headers, payload);
+                        delivered = true;
+                    }
+                }
+            }
+        }
+
         // No-responders: if nobody received the message and the publisher
         // opted in, send back a 503 status HMSG on the reply subject.
         if (!delivered && replyTo != null && sender.ClientOpts?.NoResponders == true)
@@ -767,6 +818,153 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
         }
     }
 
+    /// <summary>
+    /// Processes a service import by transforming the subject from the importer's
+    /// subject space to the exporter's subject space, then delivering to matching
+    /// subscribers in the destination account.
+    /// Reference: Go server/accounts.go addServiceImport / processServiceImport.
+    /// </summary>
+    public void ProcessServiceImport(ServiceImport si, string subject, string? replyTo,
+        ReadOnlyMemory<byte> headers, ReadOnlyMemory<byte> payload)
+    {
+        if (si.Invalid) return;
+
+        // Transform subject: map from importer subject space to exporter subject space
+        string targetSubject;
+        if (si.Transform != null)
+        {
+            var transformed = si.Transform.Apply(subject);
+            targetSubject = transformed ?? si.To;
+        }
+        else if (si.UsePub)
+        {
+            targetSubject = subject;
+        }
+        else
+        {
+            // Default: use the "To" subject from the import definition.
+            // For wildcard imports (e.g. "requests.>" -> "api.>"), we need
+            // to map the specific subject tokens from the source pattern to
+            // the destination pattern.
+            targetSubject = MapImportSubject(subject, si.From, si.To);
+        }
+
+        // Match against destination account's SubList
+        var destSubList = si.DestinationAccount.SubList;
+        var result = destSubList.Match(targetSubject);
+
+        // Deliver to plain subscribers in the destination account
+        foreach (var sub in result.PlainSubs)
+        {
+            if (sub.Client == null) continue;
+            DeliverMessage(sub, targetSubject, replyTo, headers, payload);
+        }
+
+        // Deliver to one member of each queue group
+        foreach (var queueGroup in result.QueueSubs)
+        {
+            if (queueGroup.Length == 0) continue;
+            var sub = queueGroup[0]; // Simple selection: first available
+            if (sub.Client != null)
+                DeliverMessage(sub, targetSubject, replyTo, headers, payload);
+        }
+    }
+
+    /// <summary>
+    /// Maps a published subject from the import "From" pattern to the "To" pattern.
+    /// For example, if From="requests.>" and To="api.>" and subject="requests.test",
+    /// this returns "api.test".
+    /// </summary>
+    private static string MapImportSubject(string subject, string fromPattern, string toPattern)
+    {
+        // If "To" doesn't contain wildcards, use it directly
+        if (SubjectMatch.IsLiteral(toPattern))
+            return toPattern;
+
+        // For wildcard patterns, replace matching wildcard segments.
+        // Split into tokens and map from source to destination.
+        var subTokens = subject.Split('.');
+        var fromTokens = fromPattern.Split('.');
+        var toTokens = toPattern.Split('.');
+
+        var result = new string[toTokens.Length];
+        int subIdx = 0;
+
+        // Build a mapping: for each wildcard position in "from",
+        // capture the corresponding subject token(s)
+        var wildcardValues = new List<string>();
+        string? fwcValue = null;
+
+        for (int i = 0; i < fromTokens.Length && subIdx < subTokens.Length; i++)
+        {
+            if (fromTokens[i] == "*")
+            {
+                wildcardValues.Add(subTokens[subIdx]);
+                subIdx++;
+            }
+            else if (fromTokens[i] == ">")
+            {
+                // Capture all remaining tokens
+                fwcValue = string.Join(".", subTokens[subIdx..]);
+                subIdx = subTokens.Length;
+            }
+            else
+            {
+                subIdx++; // Skip literal match
+            }
+        }
+
+        // Now build the output using the "to" pattern
+        int wcIdx = 0;
+        var sb = new StringBuilder();
+        for (int i = 0; i < toTokens.Length; i++)
+        {
+            if (i > 0) sb.Append('.');
+
+            if (toTokens[i] == "*")
+            {
+                sb.Append(wcIdx < wildcardValues.Count ? wildcardValues[wcIdx] : "*");
+                wcIdx++;
+            }
+            else if (toTokens[i] == ">")
+            {
+                sb.Append(fwcValue ?? ">");
+            }
+            else
+            {
+                sb.Append(toTokens[i]);
+            }
+        }
+
+        return sb.ToString();
+    }
+
+    /// <summary>
+    /// Wires service import subscriptions for an account. Creates marker
+    /// subscriptions in the account's SubList so that the import paths
+    /// are tracked. The actual forwarding happens in ProcessMessage when
+    /// it checks the account's Imports.Services.
+    /// Reference: Go server/accounts.go addServiceImportSub.
+    /// </summary>
+    public void WireServiceImports(Account account)
+    {
+        foreach (var kvp in account.Imports.Services)
+        {
+            foreach (var si in kvp.Value)
+            {
+                if (si.Invalid) continue;
+
+                // Create a marker subscription in the importer account.
+                // This subscription doesn't directly deliver messages;
+                // the ProcessMessage method checks service imports after
+                // the regular SubList match.
+                _logger.LogDebug(
+                    "Wired service import for account {Account}: {From} -> {To} (dest: {DestAccount})",
+                    account.Name, si.From, si.To, si.DestinationAccount.Name);
+            }
+        }
+    }
+
     private static void SendNoResponders(NatsClient sender, string replyTo)
     {
         // Find the sid for a subscription matching the reply subject
@@ -812,8 +1010,194 @@ public sealed class NatsServer : IMessageRouter, ISubListAccess, IDisposable
         });
     }
 
+    public void SendInternalMsg(string subject, string? reply, object? msg)
+    {
+        _eventSystem?.Enqueue(new PublishMessage { Subject = subject, Reply = reply, Body = msg });
+    }
+
+    public void SendInternalAccountMsg(Account account, string subject, object? msg)
+    {
+        _eventSystem?.Enqueue(new PublishMessage { Subject = subject, Body = msg });
+    }
+
+    /// <summary>
+    /// Handles $SYS.REQ.SERVER.{id}.VARZ requests.
+    /// Returns core server information including stats counters.
+    /// </summary>
+    public void HandleVarzRequest(string subject, string? reply)
+    {
+        if (reply == null) return;
+        var varz = new
+        {
+            server_id = _serverInfo.ServerId,
+            server_name = _serverInfo.ServerName,
+            version = NatsProtocol.Version,
+            host = _options.Host,
+            port = _options.Port,
+            max_payload = _options.MaxPayload,
+            connections = ClientCount,
+            total_connections = Interlocked.Read(ref _stats.TotalConnections),
+            in_msgs = Interlocked.Read(ref _stats.InMsgs),
+            out_msgs = Interlocked.Read(ref _stats.OutMsgs),
+            in_bytes = Interlocked.Read(ref _stats.InBytes),
+            out_bytes = Interlocked.Read(ref _stats.OutBytes),
+        };
+        SendInternalMsg(reply, null, varz);
+    }
+
+    /// <summary>
+    /// Handles $SYS.REQ.SERVER.{id}.HEALTHZ requests.
+    /// Returns a simple health status response.
+    /// </summary>
+    public void HandleHealthzRequest(string subject, string? reply)
+    {
+        if (reply == null) return;
+        SendInternalMsg(reply, null, new { status = "ok" });
+    }
+
+    /// <summary>
+    /// Handles $SYS.REQ.SERVER.{id}.SUBSZ requests.
+    /// Returns the current subscription count.
+    /// </summary>
+    public void HandleSubszRequest(string subject, string? reply)
+    {
+        if (reply == null) return;
+        SendInternalMsg(reply, null, new { num_subscriptions = SubList.Count });
+    }
+
+    /// <summary>
+    /// Handles $SYS.REQ.SERVER.{id}.STATSZ requests.
+    /// Publishes current server statistics through the event system.
+    /// </summary>
+    public void HandleStatszRequest(string subject, string? reply)
+    {
+        if (reply == null) return;
+        var process = System.Diagnostics.Process.GetCurrentProcess();
+        var statsMsg = new Events.ServerStatsMsg
+        {
+            Server = BuildEventServerInfo(),
+            Stats = new Events.ServerStatsData
+            {
+                Start = StartTime,
+                Mem = process.WorkingSet64,
+                Cores = Environment.ProcessorCount,
+                Connections = ClientCount,
+                TotalConnections = Interlocked.Read(ref _stats.TotalConnections),
+                Subscriptions = SubList.Count,
+                InMsgs = Interlocked.Read(ref _stats.InMsgs),
+                OutMsgs = Interlocked.Read(ref _stats.OutMsgs),
+                InBytes = Interlocked.Read(ref _stats.InBytes),
+                OutBytes = Interlocked.Read(ref _stats.OutBytes),
+                SlowConsumers = Interlocked.Read(ref _stats.SlowConsumers),
+            },
+        };
+        SendInternalMsg(reply, null, statsMsg);
+    }
+
+    /// <summary>
+    /// Handles $SYS.REQ.SERVER.{id}.IDZ requests.
+    /// Returns basic server identity information.
+    /// </summary>
+    public void HandleIdzRequest(string subject, string? reply)
+    {
+        if (reply == null) return;
+        var idz = new
+        {
+            server_id = _serverInfo.ServerId,
+            server_name = _serverInfo.ServerName,
+            version = NatsProtocol.Version,
+            host = _options.Host,
+            port = _options.Port,
+        };
+        SendInternalMsg(reply, null, idz);
+    }
+
+    /// <summary>
+    /// Builds an EventServerInfo block for embedding in system event messages.
+    /// Maps to Go's serverInfo() helper used in events.go advisory publishing.
+    /// </summary>
+    public EventServerInfo BuildEventServerInfo()
+    {
+        var seq = _eventSystem?.NextSequence() ?? 0;
+        return new EventServerInfo
+        {
+            Name = _serverInfo.ServerName,
+            Host = _options.Host,
+            Id = _serverInfo.ServerId,
+            Version = NatsProtocol.Version,
+            Seq = seq,
+        };
+    }
+
+    private static EventClientInfo BuildEventClientInfo(NatsClient client)
+    {
+        return new EventClientInfo
+        {
+            Id = client.Id,
+            Host = client.RemoteIp,
+            Account = client.Account?.Name,
+            Name = client.ClientOpts?.Name,
+            Lang = client.ClientOpts?.Lang,
+            Version = client.ClientOpts?.Version,
+            Start = client.StartTime,
+        };
+    }
+
+    /// <summary>
+    /// Publishes a $SYS.ACCOUNT.{account}.CONNECT advisory when a client
+    /// completes authentication. Maps to Go's sendConnectEvent in events.go.
+    /// </summary>
+    public void PublishConnectEvent(NatsClient client)
+    {
+        if (_eventSystem == null) return;
+        var accountName = client.Account?.Name ?? Account.GlobalAccountName;
+        var subject = string.Format(EventSubjects.ConnectEvent, accountName);
+        var evt = new ConnectEventMsg
+        {
+            Id = Guid.NewGuid().ToString("N"),
+            Time = DateTime.UtcNow,
+            Server = BuildEventServerInfo(),
+            Client = BuildEventClientInfo(client),
+        };
+        SendInternalMsg(subject, null, evt);
+    }
+
+    /// <summary>
+    /// Publishes a $SYS.ACCOUNT.{account}.DISCONNECT advisory when a client
+    /// disconnects. Maps to Go's sendDisconnectEvent in events.go.
+    /// </summary>
+    public void PublishDisconnectEvent(NatsClient client)
+    {
+        if (_eventSystem == null) return;
+        var accountName = client.Account?.Name ?? Account.GlobalAccountName;
+        var subject = string.Format(EventSubjects.DisconnectEvent, accountName);
+        var evt = new DisconnectEventMsg
+        {
+            Id = Guid.NewGuid().ToString("N"),
+            Time = DateTime.UtcNow,
+            Server = BuildEventServerInfo(),
+            Client = BuildEventClientInfo(client),
+            Sent = new DataStats
+            {
+                Msgs = Interlocked.Read(ref client.OutMsgs),
+                Bytes = Interlocked.Read(ref client.OutBytes),
+            },
+            Received = new DataStats
+            {
+                Msgs = Interlocked.Read(ref client.InMsgs),
+                Bytes = Interlocked.Read(ref client.InBytes),
+            },
+            Reason = client.CloseReason.ToReasonString(),
+        };
+        SendInternalMsg(subject, null, evt);
+    }
+
     public void RemoveClient(NatsClient client)
     {
+        // Publish disconnect advisory before removing client state
+        if (client.ConnectReceived)
+            PublishDisconnectEvent(client);
+
         _clients.TryRemove(client.Id, out _);
         _logger.LogDebug("Removed client {ClientId}", client.Id);