From 6ba788bb6e72b97e956ccfc61a27612df767e567 Mon Sep 17 00:00:00 2001 From: Developer Date: Mon, 13 Jul 2026 14:13:39 +0200 Subject: [PATCH] P2.1: Add GatewayConnector BackgroundService with WebSocket client - GatewayConnector: BackgroundService maintaining persistent WebSocket connection to OpenClaw Gateway with Bearer token auth from backend env only - IGatewayConnector: Interface exposing connection state, version info, health - Exponential backoff reconnect: 1s initial, max 5min, with jitter - Version pinning: RequiredVersion from config; fail-fast or warn on mismatch - Pre-flight HTTP version check before WebSocket connect - GatewayHealthController: GET /api/health/gateway with connection status - GatewayConnectorOptions: Configurable via GatewayConnector config section - Token never logged; read from Integrations:OpenClaw:Password/Token only - 16 tests: options defaults, config binding, health endpoint responses, connection states, DI registration, routing attributes --- backend-tests/GatewayConnectorTests.cs | 311 +++++++++++ .../Controllers/GatewayHealthController.cs | 33 ++ .../Extensions/ServiceCollectionExtensions.cs | 6 + backend/Services/GatewayConnector.cs | 511 ++++++++++++++++++ backend/Services/IGatewayConnector.cs | 57 ++ backend/appsettings.json | 7 + 6 files changed, 925 insertions(+) create mode 100644 backend-tests/GatewayConnectorTests.cs create mode 100644 backend/Controllers/GatewayHealthController.cs create mode 100644 backend/Services/GatewayConnector.cs create mode 100644 backend/Services/IGatewayConnector.cs diff --git a/backend-tests/GatewayConnectorTests.cs b/backend-tests/GatewayConnectorTests.cs new file mode 100644 index 0000000..5884139 --- /dev/null +++ b/backend-tests/GatewayConnectorTests.cs @@ -0,0 +1,311 @@ +using System.Text.Json; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Http.Json; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Nexus.Api.Controllers; +using Nexus.Api.Services; +using Xunit; + +namespace Nexus.Api.Tests; + +public class GatewayConnectorTests +{ + // ── Options / Configuration tests ── + + [Fact] + public void Options_DefaultWebSocketPath_IsWs() + { + var options = new GatewayConnectorOptions(); + Assert.Equal("/ws", options.WebSocketPath); + } + + [Fact] + public void Options_DefaultReconnectInitialDelay_Is1000ms() + { + var options = new GatewayConnectorOptions(); + Assert.Equal(1000, options.ReconnectInitialDelayMs); + } + + [Fact] + public void Options_DefaultReconnectMaxDelay_Is5Minutes() + { + var options = new GatewayConnectorOptions(); + Assert.Equal(300_000, options.ReconnectMaxDelayMs); + } + + [Fact] + public void Options_FailFastOnVersionMismatch_IsTrueByDefault() + { + var options = new GatewayConnectorOptions(); + Assert.True(options.FailFastOnVersionMismatch); + } + + [Fact] + public void Options_FailFastOnMissingVersion_IsFalseByDefault() + { + var options = new GatewayConnectorOptions(); + Assert.False(options.FailFastOnMissingVersion); + } + + // ── Configuration binding tests ── + + [Fact] + public void Options_BindFromConfiguration() + { + var config = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary + { + ["GatewayConnector:WebSocketPath"] = "/events", + ["GatewayConnector:ReconnectInitialDelayMs"] = "2000", + ["GatewayConnector:ReconnectMaxDelayMs"] = "60000", + ["GatewayConnector:FailFastOnVersionMismatch"] = "false", + ["GatewayConnector:FailFastOnMissingVersion"] = "true" + }) + .Build(); + + var services = new ServiceCollection(); + services.AddSingleton(config); + services.AddOptions() + .BindConfiguration(GatewayConnectorOptions.SectionName); + var sp = services.BuildServiceProvider(); + var options = sp.GetRequiredService>().Value; + + Assert.Equal("/events", options.WebSocketPath); + Assert.Equal(2000, options.ReconnectInitialDelayMs); + Assert.Equal(60000, options.ReconnectMaxDelayMs); + Assert.False(options.FailFastOnVersionMismatch); + Assert.True(options.FailFastOnMissingVersion); + } + + // ── Health endpoint tests ── + + [Fact] + public async Task HealthEndpoint_ReturnsCorrectStructure() + { + var controller = new GatewayHealthController(new FakeGatewayConnector( + GatewayConnectionState.Connected, "2.103.0", "2.103.0", + DateTimeOffset.UtcNow, 0, "Gateway verbunden")); + + var body = await ExecuteAndReadJsonAsync(controller.GetGatewayHealth()); + + using var doc = JsonDocument.Parse(body); + var root = doc.RootElement; + + Assert.Equal("connected", root.GetProperty("status").GetString()); + Assert.True(root.GetProperty("connected").GetBoolean()); + Assert.Equal("2.103.0", root.GetProperty("gatewayVersion").GetString()); + Assert.Equal("2.103.0", root.GetProperty("requiredVersion").GetString()); + Assert.True(root.GetProperty("versionPinned").GetBoolean()); + Assert.Equal(0, root.GetProperty("reconnectAttempts").GetInt32()); + } + + [Fact] + public async Task HealthEndpoint_Disconnected_ShowsDisconnected() + { + var controller = new GatewayHealthController(new FakeGatewayConnector( + GatewayConnectionState.Disconnected, null, "2.103.0", + null, 5, "Connection refused")); + + var body = await ExecuteAndReadJsonAsync(controller.GetGatewayHealth()); + + using var doc = JsonDocument.Parse(body); + var root = doc.RootElement; + + Assert.Equal("disconnected", root.GetProperty("status").GetString()); + Assert.False(root.GetProperty("connected").GetBoolean()); + Assert.Equal("unknown", root.GetProperty("gatewayVersion").GetString()); + Assert.Equal(5, root.GetProperty("reconnectAttempts").GetInt32()); + Assert.Equal("Connection refused", root.GetProperty("message").GetString()); + } + + [Fact] + public async Task HealthEndpoint_Reconnecting_ShowsReconnecting() + { + var controller = new GatewayHealthController(new FakeGatewayConnector( + GatewayConnectionState.Reconnecting, null, null, + null, 3, "Reconnecting...")); + + var body = await ExecuteAndReadJsonAsync(controller.GetGatewayHealth()); + + using var doc = JsonDocument.Parse(body); + var root = doc.RootElement; + + Assert.Equal("reconnecting", root.GetProperty("status").GetString()); + Assert.False(root.GetProperty("connected").GetBoolean()); + Assert.False(root.GetProperty("versionPinned").GetBoolean()); + } + + [Fact] + public async Task HealthEndpoint_Failed_ShowsFailed() + { + var controller = new GatewayHealthController(new FakeGatewayConnector( + GatewayConnectionState.Failed, "2.100.0", "2.103.0", + null, 10, "Version mismatch — fail-fast")); + + var body = await ExecuteAndReadJsonAsync(controller.GetGatewayHealth()); + + using var doc = JsonDocument.Parse(body); + var root = doc.RootElement; + + Assert.Equal("failed", root.GetProperty("status").GetString()); + Assert.False(root.GetProperty("connected").GetBoolean()); + Assert.Equal("2.100.0", root.GetProperty("gatewayVersion").GetString()); + Assert.Equal("2.103.0", root.GetProperty("requiredVersion").GetString()); + Assert.Equal(10, root.GetProperty("reconnectAttempts").GetInt32()); + Assert.True(root.TryGetProperty("timestamp", out _)); + } + + [Fact] + public async Task HealthEndpoint_Unpinned_ShowsVersionPinnedFalse() + { + var controller = new GatewayHealthController(new FakeGatewayConnector( + GatewayConnectionState.Connected, "2.103.0", null, + DateTimeOffset.UtcNow, 0, null)); + + var body = await ExecuteAndReadJsonAsync(controller.GetGatewayHealth()); + + using var doc = JsonDocument.Parse(body); + var root = doc.RootElement; + + Assert.False(root.GetProperty("versionPinned").GetBoolean()); + Assert.Equal(JsonValueKind.Null, root.GetProperty("requiredVersion").ValueKind); + } + + // ── Connection state transition tests ── + + [Fact] + public void Connector_InitialState_IsInitializing() + { + var connector = new FakeGatewayConnector( + GatewayConnectionState.Initializing, null, null, null, 0, null); + + Assert.Equal(GatewayConnectionState.Initializing, connector.ConnectionState); + Assert.Null(connector.GatewayVersion); + Assert.Null(connector.LastConnectedAt); + Assert.Equal(0, connector.ReconnectAttempts); + } + + [Fact] + public void Connector_AfterConnection_ReportsConnected() + { + var now = DateTimeOffset.UtcNow; + var connector = new FakeGatewayConnector( + GatewayConnectionState.Connected, "2.103.0", "2.103.0", + now, 0, null); + + Assert.Equal(GatewayConnectionState.Connected, connector.ConnectionState); + Assert.Equal("2.103.0", connector.GatewayVersion); + Assert.Equal("2.103.0", connector.RequiredVersion); + Assert.NotNull(connector.LastConnectedAt); + Assert.Equal(0, connector.ReconnectAttempts); + } + + // ── DI registration tests ── + + [Fact] + public void Connector_CanBeRegisteredInDi() + { + var config = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary + { + ["Integrations:OpenClaw:BaseUrl"] = "http://localhost:18789", + ["Integrations:OpenClaw:RequiredVersion"] = "2.103.0" + }) + .Build(); + + var services = new ServiceCollection(); + services.AddSingleton(config); + services.AddLogging(); + services.AddHttpClient("gateway"); + services.AddOptions() + .BindConfiguration(GatewayConnectorOptions.SectionName); + services.AddSingleton(); + + var sp = services.BuildServiceProvider(); + var connector = sp.GetRequiredService(); + + Assert.NotNull(connector); + Assert.Equal(GatewayConnectionState.Initializing, connector.ConnectionState); + } + + // ── HTTP endpoint routing test ── + + [Fact] + public void HealthEndpoint_HasAllowAnonymous() + { + var type = typeof(GatewayHealthController); + var attr = type.GetCustomAttributes(true) + .OfType() + .FirstOrDefault(); + Assert.NotNull(attr); + } + + [Fact] + public void HealthEndpoint_HasCorrectRoute() + { + var method = typeof(GatewayHealthController).GetMethod("GetGatewayHealth"); + Assert.NotNull(method); + var routeAttr = method!.GetCustomAttributes(true) + .OfType() + .FirstOrDefault(); + Assert.NotNull(routeAttr); + Assert.Contains("/api/health/gateway", routeAttr!.Template!); + } + + // ── Helpers ── + + private static async Task ExecuteAndReadJsonAsync(IResult result) + { + var services = new ServiceCollection(); + services.AddLogging(); + services.ConfigureHttpJsonOptions(options => + options.SerializerOptions.PropertyNamingPolicy = System.Text.Json.JsonNamingPolicy.CamelCase); + var sp = services.BuildServiceProvider(); + + var httpContext = new DefaultHttpContext { RequestServices = sp }; + // HttpResults.Ok needs a response body to write to + httpContext.Response.Body = new MemoryStream(); + await result.ExecuteAsync(httpContext); + httpContext.Response.Body.Position = 0; + using var reader = new StreamReader(httpContext.Response.Body); + return reader.ReadToEnd(); + } +} + +/// +/// Test double implementing all members of IGatewayConnector. +/// +public sealed class FakeGatewayConnector : IGatewayConnector +{ + private readonly GatewayConnectionState _state; + private readonly string? _gatewayVersion; + private readonly string? _requiredVersion; + private readonly DateTimeOffset? _lastConnectedAt; + private readonly int _reconnectAttempts; + private readonly string? _statusMessage; + + public FakeGatewayConnector( + GatewayConnectionState state, + string? gatewayVersion, + string? requiredVersion, + DateTimeOffset? lastConnectedAt, + int reconnectAttempts, + string? statusMessage) + { + _state = state; + _gatewayVersion = gatewayVersion; + _requiredVersion = requiredVersion; + _lastConnectedAt = lastConnectedAt; + _reconnectAttempts = reconnectAttempts; + _statusMessage = statusMessage; + } + + public GatewayConnectionState ConnectionState => _state; + public string? GatewayVersion => _gatewayVersion; + public string? RequiredVersion => _requiredVersion; + public DateTimeOffset? LastConnectedAt => _lastConnectedAt; + public int ReconnectAttempts => _reconnectAttempts; + public string? StatusMessage => _statusMessage; +} diff --git a/backend/Controllers/GatewayHealthController.cs b/backend/Controllers/GatewayHealthController.cs new file mode 100644 index 0000000..58cd169 --- /dev/null +++ b/backend/Controllers/GatewayHealthController.cs @@ -0,0 +1,33 @@ +using Microsoft.AspNetCore.Authorization; +using Microsoft.AspNetCore.Mvc; +using Nexus.Api.Services; + +namespace Nexus.Api.Controllers; + +/// +/// Health-check endpoint for the Gateway WebSocket connection. +/// +[ApiController] +[AllowAnonymous] +public class GatewayHealthController(IGatewayConnector connector) : ControllerBase +{ + /// + /// Returns the current Gateway WebSocket connection health. + /// + [HttpGet("/api/health/gateway")] + public IResult GetGatewayHealth() + { + return Results.Ok(new + { + status = connector.ConnectionState.ToString().ToLowerInvariant(), + connected = connector.ConnectionState == GatewayConnectionState.Connected, + gatewayVersion = connector.GatewayVersion ?? "unknown", + requiredVersion = connector.RequiredVersion, + versionPinned = connector.RequiredVersion is not null, + lastConnectedAt = connector.LastConnectedAt?.ToString("o"), + reconnectAttempts = connector.ReconnectAttempts, + message = connector.StatusMessage, + timestamp = DateTimeOffset.UtcNow.ToString("o") + }); + } +} diff --git a/backend/Extensions/ServiceCollectionExtensions.cs b/backend/Extensions/ServiceCollectionExtensions.cs index f2b0962..f76181e 100644 --- a/backend/Extensions/ServiceCollectionExtensions.cs +++ b/backend/Extensions/ServiceCollectionExtensions.cs @@ -229,6 +229,12 @@ public static class ServiceCollectionExtensions services.AddScoped(); services.AddHostedService(); + // ── Gateway WebSocket Connector ── + services.AddOptions() + .BindConfiguration(GatewayConnectorOptions.SectionName); + services.AddSingleton(); + services.AddHostedService(sp => (GatewayConnector)sp.GetRequiredService()); + // ── Backend Bridge (Agent-Command-Service) ── services.AddScoped(); diff --git a/backend/Services/GatewayConnector.cs b/backend/Services/GatewayConnector.cs new file mode 100644 index 0000000..a658a2d --- /dev/null +++ b/backend/Services/GatewayConnector.cs @@ -0,0 +1,511 @@ +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; } +} diff --git a/backend/Services/IGatewayConnector.cs b/backend/Services/IGatewayConnector.cs new file mode 100644 index 0000000..943b9c5 --- /dev/null +++ b/backend/Services/IGatewayConnector.cs @@ -0,0 +1,57 @@ +using Nexus.Api.Models; + +namespace Nexus.Api.Services; + +/// +/// Maintains a persistent WebSocket connection to the OpenClaw Gateway +/// for real-time event streaming and health monitoring. +/// +public interface IGatewayConnector +{ + /// + /// Current WebSocket connection state. + /// + GatewayConnectionState ConnectionState { get; } + + /// + /// Gateway version reported at last connection. + /// + string? GatewayVersion { get; } + + /// + /// Required version from configuration, or null when unpinned. + /// + string? RequiredVersion { get; } + + /// + /// Timestamp of the last successful connection attempt. + /// + DateTimeOffset? LastConnectedAt { get; } + + /// + /// Number of consecutive failed connection attempts. + /// + int ReconnectAttempts { get; } + + /// + /// Detailed status message (e.g. error or version info). + /// + string? StatusMessage { get; } +} + +/// +/// Represents the current connection state of the GatewayConnector. +/// +public enum GatewayConnectionState +{ + /// Not yet attempted or initializing. + Initializing, + /// Connected and healthy. + Connected, + /// Disconnected, waiting for reconnect. + Disconnected, + /// Recoverable error, retrying. + Reconnecting, + /// Failed after maximum retries. + Failed +} diff --git a/backend/appsettings.json b/backend/appsettings.json index 60e94ab..70b04f3 100644 --- a/backend/appsettings.json +++ b/backend/appsettings.json @@ -27,5 +27,12 @@ "StaleHours": 2, "IntervalMinutes": 30 }, + "GatewayConnector": { + "WebSocketPath": "/ws", + "ReconnectInitialDelayMs": 1000, + "ReconnectMaxDelayMs": 300000, + "FailFastOnVersionMismatch": true, + "FailFastOnMissingVersion": false + }, "AllowedHosts": "*" }