using System.Net; using System.Net.WebSockets; using System.Reflection; using System.Runtime.InteropServices; using System.Text; using System.Text.Json; namespace Zcbot.WindowsNode; internal sealed class NodeConnectionLoop(NodeConfig config, Action? statusChanged = null) { private static readonly TimeSpan[] Backoff = [ TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(2), TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(10), TimeSpan.FromSeconds(30), TimeSpan.FromSeconds(60), ]; internal async Task RunAsync(CancellationToken cancellationToken) { var attempt = 0; while (!cancellationToken.IsCancellationRequested) { try { Report(NodeState.Connecting, "正在连接 zcbot…"); await ConnectOnceAsync(cancellationToken); attempt = 0; } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { throw; } catch (WebSocketException exception) { Report(NodeState.Offline, "连接中断,等待重连"); Console.Error.WriteLine($"[WARN] WebSocket disconnected: {exception.Message}"); } catch (IOException exception) { Report(NodeState.Offline, "网络不可用,等待重连"); Console.Error.WriteLine($"[WARN] Connection I/O failed: {exception.Message}"); } catch (JsonException exception) { Report(NodeState.Offline, "服务端消息无效,等待重连"); Console.Error.WriteLine($"[WARN] Invalid server message: {exception.Message}"); } catch (NodeEndpointException exception) { Report(NodeState.Offline, "WebSocket 握手被拒绝,请检查服务端或反向代理"); Console.Error.WriteLine($"[WARN] WebSocket handshake rejected: {exception.Message}"); } var baseDelay = Backoff[Math.Min(attempt, Backoff.Length - 1)]; attempt++; var jitter = TimeSpan.FromMilliseconds(Random.Shared.Next(0, 750)); var delay = baseDelay + jitter; Console.WriteLine($"[INFO] Reconnecting in {delay.TotalSeconds:F1}s."); await Task.Delay(delay, cancellationToken); } } private async Task ConnectOnceAsync(CancellationToken cancellationToken) { using var socket = new ClientWebSocket(); socket.Options.SetRequestHeader("Authorization", $"Bearer {config.NodeToken}"); socket.Options.SetRequestHeader("X-Node-Id", config.NodeId.ToString()); socket.Options.KeepAliveInterval = TimeSpan.FromSeconds(config.HeartbeatSeconds); socket.Options.CollectHttpResponseDetails = true; var endpoint = NodeUri.WebSocketEndpoint(config.ServerUrl); Console.WriteLine($"[INFO] Connecting to {endpoint.GetLeftPart(UriPartial.Path)}."); try { await socket.ConnectAsync(endpoint, cancellationToken); } catch (WebSocketException exception) when ( socket.HttpStatusCode is HttpStatusCode.Unauthorized or HttpStatusCode.Forbidden) { throw new NodeEndpointException( $"WebSocket handshake returned HTTP {(int?)socket.HttpStatusCode}. " + "Verify that the server includes the current Windows Node routes and that " + $"the reverse proxy forwards WebSocket Upgrade for {endpoint.AbsolutePath}. " + exception.Message); } Console.WriteLine("[OK] Node connected."); Report(NodeState.Online, "已连接"); await SendAsync(socket, "hello", RuntimePayload(), cancellationToken); using var heartbeatStop = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); var heartbeat = HeartbeatLoopAsync(socket, heartbeatStop.Token); try { await ReceiveLoopAsync(socket, cancellationToken); } finally { heartbeatStop.Cancel(); try { await heartbeat; } catch (OperationCanceledException) when (heartbeatStop.IsCancellationRequested) { } } } private async Task HeartbeatLoopAsync(ClientWebSocket socket, CancellationToken cancellationToken) { using var timer = new PeriodicTimer(TimeSpan.FromSeconds(config.HeartbeatSeconds)); while (await timer.WaitForNextTickAsync(cancellationToken)) { await SendAsync(socket, "heartbeat", RuntimePayload(), cancellationToken); } } private static async Task ReceiveLoopAsync( ClientWebSocket socket, CancellationToken cancellationToken) { var buffer = new byte[16 * 1024]; using var message = new MemoryStream(); while (socket.State == WebSocketState.Open) { var result = await socket.ReceiveAsync(buffer, cancellationToken); if (result.MessageType == WebSocketMessageType.Close) { if ((int?)result.CloseStatus == 4003) { throw new NodeConfigurationException( "节点身份已被服务端拒绝。请在管理后台确认该 Node ID 未被禁用或删除;" + "若记录不存在或身份已撤销,请清除本机身份并使用新注册码重新注册。" + $" 服务端信息:{result.CloseStatusDescription}"); } if (result.CloseStatus == WebSocketCloseStatus.PolicyViolation) { throw new NodeConfigurationException( $"节点上报被服务端拒绝:{result.CloseStatusDescription}"); } return; } if (result.MessageType != WebSocketMessageType.Text) { throw new JsonException("Only text WebSocket messages are supported."); } message.Write(buffer, 0, result.Count); if (!result.EndOfMessage) { if (message.Length > 1024 * 1024) { throw new JsonException("Server message exceeded 1 MiB."); } continue; } using var document = JsonDocument.Parse(message.ToArray()); if (document.RootElement.TryGetProperty("type", out var type)) { Console.WriteLine($"[INFO] Server message: {type.GetString()}."); } message.SetLength(0); } } private void Report(NodeState state, string message) => statusChanged?.Invoke(NodeStatus.Create(state, message)); private async Task SendAsync( ClientWebSocket socket, string type, object payload, CancellationToken cancellationToken) { var envelope = JsonSerializer.SerializeToUtf8Bytes(new { protocol_version = 1, message_id = Guid.NewGuid(), type, sent_at = DateTimeOffset.UtcNow, payload, }); await socket.SendAsync( envelope, WebSocketMessageType.Text, endOfMessage: true, cancellationToken); } private object RuntimePayload() { var root = Path.GetPathRoot(Environment.SystemDirectory) ?? "C:\\"; var origin = OriginRuntimeProbe.Detect(); return new { install_id = config.InstallId, node_name = config.NodeName, node_version = Assembly.GetExecutingAssembly().GetName().Version?.ToString(3) ?? "0.1.0", os_version = RuntimeInformation.OSDescription, capabilities = config.Capabilities, available_slots = origin.Health == "ready" ? 1 : 0, disk_free_bytes = new DriveInfo(root).AvailableFreeSpace, desktop_session = Environment.UserInteractive, origin = new { software = origin.Software, software_version = origin.SoftwareVersion, adapter_version = origin.AdapterVersion, health = origin.Health, detail = origin.Detail, }, }; } }