using System.Net.Http.Headers; using System.Net.WebSockets; using System.Text; using System.Text.Json; using Microsoft.Extensions.Options; namespace Nexus.Api.Services; /// /// BackgroundService that maintains a persistent WebSocket connection to the /// OpenClaw Gateway for real-time event streaming and health monitoring. /// /// Uses exponential backoff for reconnects, never logs the gateway token, /// and reports connection state via . /// public sealed class GatewayConnector : BackgroundService, IGatewayConnector { private readonly IHttpClientFactory _httpClientFactory; private readonly IConfiguration _configuration; private readonly ILogger _logger; private readonly GatewayConnectorOptions _options; // ── Connection state (lock-protected) ── private readonly object _lock = new(); private GatewayConnectionState _state = GatewayConnectionState.Initializing; private string? _gatewayVersion; private string? _requiredVersion; private DateTimeOffset? _lastConnectedAt; private int _reconnectAttempts; private string? _statusMessage; public GatewayConnector( IHttpClientFactory httpClientFactory, IConfiguration configuration, IOptions options, ILogger logger) { _httpClientFactory = httpClientFactory; _configuration = configuration; _logger = logger; _options = options.Value; } // ── IGatewayConnector ── public GatewayConnectionState ConnectionState { get { lock (_lock) return _state; } } public string? GatewayVersion { get { lock (_lock) return _gatewayVersion; } } public string? RequiredVersion { get { lock (_lock) return _requiredVersion; } } public DateTimeOffset? LastConnectedAt { get { lock (_lock) return _lastConnectedAt; } } public int ReconnectAttempts { get { lock (_lock) return _reconnectAttempts; } } public string? StatusMessage { get { lock (_lock) return _statusMessage; } } // ── BackgroundService ── protected override async Task ExecuteAsync(CancellationToken stoppingToken) { // Load version pin from configuration once at startup var pinnedVersion = _configuration["Integrations:OpenClaw:RequiredVersion"]; lock (_lock) { _requiredVersion = string.IsNullOrWhiteSpace(pinnedVersion) ? null : pinnedVersion.Trim(); } if (_requiredVersion is null) { _logger.LogWarning( "Gateway version is UNPINNED (Integrations:OpenClaw:RequiredVersion is empty). " + "Drift detection is disabled; set a required version to enable fail-fast on mismatch."); } while (!stoppingToken.IsCancellationRequested) { try { await ConnectAndReceiveAsync(stoppingToken); } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { break; } catch (Exception ex) { _logger.LogError(ex, "GatewayConnector connection loop exception (will retry)"); } // Exponential backoff before next reconnect attempt if (!stoppingToken.IsCancellationRequested) { var delay = ComputeBackoff(); _logger.LogInformation( "GatewayConnector reconnecting in {DelayMs}ms (attempt {Attempt})", (int)delay.TotalMilliseconds, GetReconnectAttempts() + 1); SetState(GatewayConnectionState.Reconnecting); try { await Task.Delay(delay, stoppingToken); } catch (OperationCanceledException) { break; } } } _logger.LogInformation("GatewayConnector background service stopped"); } private async Task ConnectAndReceiveAsync(CancellationToken stoppingToken) { using var ws = new ClientWebSocket(); ConfigureWebSocketWithAuth(ws); var baseUrl = _configuration["Integrations:OpenClaw:BaseUrl"] ?? "http://127.0.0.1:18789"; var wsBase = ConvertToWebSocketUrl(baseUrl); // ── Step 1: Pre-flight version check via HTTP ── string? detectedVersion = null; try { detectedVersion = await FetchGatewayVersionAsync(stoppingToken); } catch (Exception ex) { _logger.LogWarning(ex, "Pre-flight version check failed (non-fatal)"); } // Version check against the pin var (versionOk, versionWarning) = EvaluateVersion(detectedVersion); if (!versionOk) { if (_options.FailFastOnVersionMismatch) { _logger.LogError( "Gateway version mismatch — fail-fast enabled. Detected: {Detected}, Required: {Required}", detectedVersion ?? "none", _requiredVersion); IncrementReconnectAttempts(); SetState(GatewayConnectionState.Failed, $"Version mismatch — fail-fast: detected '{detectedVersion}', required '{_requiredVersion}'."); return; } _logger.LogWarning("Gateway version mismatch (non-fatal): {Warning}", versionWarning); } // ── Step 2: Establish WebSocket connection ── var wsUri = new Uri(new Uri(wsBase), _options.WebSocketPath); _logger.LogInformation("GatewayConnector connecting to {Uri}", wsUri); try { await ws.ConnectAsync(wsUri, stoppingToken); } catch (WebSocketException ex) { _logger.LogWarning(ex, "WebSocket connection failed to {Uri} — gateway may not expose a WS endpoint", wsUri); IncrementReconnectAttempts(); SetState(GatewayConnectionState.Disconnected, $"Connection refused — WebSocket endpoint may not be available at {wsUri}. Error: {ex.Message}"); return; } catch (HttpRequestException ex) { _logger.LogWarning(ex, "WebSocket HTTP upgrade failed to {Uri}", wsUri); IncrementReconnectAttempts(); SetState(GatewayConnectionState.Disconnected, $"HTTP upgrade failed: {ex.Message}"); return; } // ── Step 3: Submit authentication token (if configured) ── var token = ResolveToken(); if (token is not null) { try { var authFrame = Encoding.UTF8.GetBytes( JsonSerializer.Serialize(new { type = "auth", token })); await ws.SendAsync( new ArraySegment(authFrame), WebSocketMessageType.Text, endOfMessage: true, cancellationToken: stoppingToken); _logger.LogDebug("WebSocket auth token sent"); } catch (Exception ex) { _logger.LogWarning(ex, "Failed to send WebSocket auth frame"); } } // ── After successful connection ── lock (_lock) { _state = GatewayConnectionState.Connected; _gatewayVersion = detectedVersion; _lastConnectedAt = DateTimeOffset.UtcNow; _reconnectAttempts = 0; _statusMessage = versionWarning; } _logger.LogInformation( "GatewayConnector connected. Version: {Version}, Required: {Required}", detectedVersion ?? "unknown", _requiredVersion ?? "unpinned"); // ── Step 4: Receive loop (heartbeat / event processing) ── await ReceiveLoopAsync(ws, stoppingToken); } private async Task ReceiveLoopAsync(ClientWebSocket ws, CancellationToken stoppingToken) { var buffer = new byte[4096]; try { while (ws.State == WebSocketState.Open && !stoppingToken.IsCancellationRequested) { var result = await ws.ReceiveAsync(new ArraySegment(buffer), stoppingToken); if (result.MessageType == WebSocketMessageType.Close) { _logger.LogInformation( "GatewayConnector received close frame: {Description}", result.CloseStatusDescription ?? "no description"); break; } // Process text frames (events, heartbeats, status updates) if (result.MessageType == WebSocketMessageType.Text) { var message = Encoding.UTF8.GetString(buffer, 0, result.Count); ProcessGatewayMessage(message); } } } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { // Expected on shutdown } catch (WebSocketException ex) { _logger.LogWarning(ex, "WebSocket connection lost"); } finally { if (ws.State == WebSocketState.Open || ws.State == WebSocketState.CloseReceived) { try { await ws.CloseAsync( WebSocketCloseStatus.NormalClosure, "Connector shutdown", CancellationToken.None); } catch { /* Best effort */ } } SetState(GatewayConnectionState.Disconnected, "Connection closed"); } } private void ProcessGatewayMessage(string message) { try { using var doc = JsonDocument.Parse(message); var root = doc.RootElement; var type = root.TryGetProperty("type", out var typeEl) ? typeEl.GetString() : null; switch (type) { case "heartbeat": case "pong": _logger.LogDebug("Gateway heartbeat received"); break; case "event": var eventName = root.TryGetProperty("name", out var nameEl) ? nameEl.GetString() : "unknown"; _logger.LogDebug("Gateway event: {EventName}", eventName); // Future: fan-out to other services via a channel/event bus break; case "error": var errorMsg = root.TryGetProperty("message", out var msgEl) ? msgEl.GetString() : "unknown error"; _logger.LogWarning("Gateway error frame received: {Error}", errorMsg); break; default: _logger.LogDebug("Gateway message (type={Type}): {Preview}", type ?? "none", Truncate(message, 120)); break; } } catch (JsonException) { _logger.LogDebug("Non-JSON gateway message: {Preview}", Truncate(message, 80)); } } // ── Helpers ── private (bool Ok, string? Warning) EvaluateVersion(string? detectedVersion) { if (_requiredVersion is null) return (true, null); // Unpinned — always ok if (detectedVersion is null) return (_options.FailFastOnMissingVersion ? false : true, "Gateway version not detected while version pin is set."); if (!string.Equals(detectedVersion, _requiredVersion, StringComparison.OrdinalIgnoreCase)) return (false, $"Version drift: detected '{detectedVersion}', required '{_requiredVersion}'."); return (true, null); } private async Task FetchGatewayVersionAsync(CancellationToken ct) { var client = _httpClientFactory.CreateClient("gateway"); try { using var request = new HttpRequestMessage(HttpMethod.Get, "/health"); ApplyAuth(request); using var response = await client.SendAsync(request, ct); if (!response.IsSuccessStatusCode) return null; // Check X-OpenClaw-Version header first if (response.Headers.TryGetValues("X-OpenClaw-Version", out var headerValues)) { var version = headerValues.FirstOrDefault(); if (!string.IsNullOrWhiteSpace(version)) return version.Trim(); } // Try JSON body var body = await response.Content.ReadAsStringAsync(ct); if (!string.IsNullOrWhiteSpace(body)) { try { using var doc = JsonDocument.Parse(body); var root = doc.RootElement; var v = TryGetString(root, "version") ?? TryGetString(root, "gatewayVersion") ?? TryGetString(root, "openclawVersion"); if (v is not null) return v; } catch { /* Not JSON */ } } return null; } catch { return null; } } private TimeSpan ComputeBackoff() { var attempts = GetReconnectAttempts(); var backoffMs = (int)Math.Min( _options.ReconnectInitialDelayMs * Math.Pow(2, attempts), _options.ReconnectMaxDelayMs); // Add jitter (±25%) var jitter = (int)(backoffMs * 0.25 * (Random.Shared.NextDouble() * 2 - 1)); return TimeSpan.FromMilliseconds(Math.Max(100, backoffMs + jitter)); } private void IncrementReconnectAttempts() { lock (_lock) { _reconnectAttempts++; } } private int GetReconnectAttempts() { lock (_lock) { return _reconnectAttempts; } } private void SetState(GatewayConnectionState state, string? message = null) { lock (_lock) { _state = state; if (message is not null) _statusMessage = message; } } private string? ResolveToken() { // Token NEVER logged — only read from config here var token = _configuration["Integrations:OpenClaw:Password"] ?? _configuration["Integrations:OpenClaw:Token"]; return string.IsNullOrWhiteSpace(token) ? null : token; } private void ApplyAuth(HttpRequestMessage request) { var token = ResolveToken(); if (token is not null) request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", token); } private static void ConfigureWebSocket(ClientWebSocket ws) { // Add auth header to the initial HTTP upgrade request ws.Options.KeepAliveInterval = TimeSpan.FromSeconds(30); ws.Options.UseDefaultCredentials = false; } // Overload with auth header via cookie (WebSocket can't do Authorization header natively in all runtimes) private void ConfigureWebSocketWithAuth(ClientWebSocket ws) { var token = ResolveToken(); if (token is not null) { ws.Options.SetRequestHeader("Authorization", "Bearer " + token); } } private static string ConvertToWebSocketUrl(string baseUrl) { var trimmed = baseUrl.TrimEnd('/'); if (trimmed.StartsWith("https://", StringComparison.OrdinalIgnoreCase)) return "wss://" + trimmed["https://".Length..]; if (trimmed.StartsWith("http://", StringComparison.OrdinalIgnoreCase)) return "ws://" + trimmed["http://".Length..]; return "ws://" + trimmed; } private static string? TryGetString(JsonElement root, string property) => root.ValueKind == JsonValueKind.Object && root.TryGetProperty(property, out var value) && value.ValueKind == JsonValueKind.String ? value.GetString() : null; private static string Truncate(string value, int maxLength) => value.Length <= maxLength ? value : value[..maxLength] + "\u2026"; } /// /// Configuration options for the GatewayConnector background service. /// public sealed class GatewayConnectorOptions { public const string SectionName = "GatewayConnector"; /// /// WebSocket path relative to the gateway base URL. /// Default: "/ws" /// public string WebSocketPath { get; set; } = "/ws"; /// /// Initial reconnect delay in milliseconds. /// Default: 1000 (1 second) /// public int ReconnectInitialDelayMs { get; set; } = 1000; /// /// Maximum reconnect delay in milliseconds. /// Default: 300_000 (5 minutes) /// public int ReconnectMaxDelayMs { get; set; } = 300_000; /// /// Maximum number of consecutive reconnect attempts before entering Failed state. /// Default: 0 (no limit — keep retrying forever) /// public int ReconnectMaxAttempts { get; set; } /// /// When true, the connector enters Failed state immediately if the gateway version /// doesn't match the pinned required version (instead of connecting with a warning). /// Default: true /// public bool FailFastOnVersionMismatch { get; set; } = true; /// /// When true, missing version (gateway doesn't report one) with a version pin set /// also triggers fail-fast. Default: false (allows pinned-version gateways that /// don't expose a version endpoint). /// public bool FailFastOnMissingVersion { get; set; } }