From 8550ed8d9e9d35d0f6eccaabdc0ced498dd585cb Mon Sep 17 00:00:00 2001 From: Florian Schmidt Date: Fri, 17 Jul 2026 11:05:01 +0200 Subject: [PATCH] Ingestion: Home Assistant WebSocket push path HomeAssistantWebSocketWorker holds a persistent state_changed subscription per HA endpoint that opts in via the connector's WebSocket toggle (HaEndpointConfig.UseWebSocket): auth handshake, subscribe, ingest in real time, capped-backoff reconnect. The REST poll worker skips WS endpoints so each is served once. HaWebSocketProtocol holds the pure handshake/parse logic. Verified by 11 protocol unit tests + a live integration test against an in-process fake HA server. CLAUDE.md updated. Claude-Session: https://claude.ai/code/session_01Kib2MniVFbD95fkgLgBBnB --- CLAUDE.md | 2 +- .../Components/Pages/Admin/Connectors.razor | 9 +- src/Infrastructure/DependencyInjection.cs | 1 + .../Ingestion/HaEndpointConfig.cs | 7 + src/Infrastructure/Ingestion/HaStateClient.cs | 12 +- .../Ingestion/HaWebSocketProtocol.cs | 70 +++++ .../Ingestion/HomeAssistantWebSocketWorker.cs | 292 ++++++++++++++++++ .../Ingestion/HomeAssistantWorker.cs | 9 +- .../Ingestion/HaWebSocketProtocolTests.cs | 119 +++++++ .../HomeAssistantWebSocketWorkerTests.cs | 182 +++++++++++ 10 files changed, 695 insertions(+), 8 deletions(-) create mode 100644 src/Infrastructure/Ingestion/HaWebSocketProtocol.cs create mode 100644 src/Infrastructure/Ingestion/HomeAssistantWebSocketWorker.cs create mode 100644 tests/Integration.Tests/Ingestion/HaWebSocketProtocolTests.cs create mode 100644 tests/Integration.Tests/Ingestion/HomeAssistantWebSocketWorkerTests.cs diff --git a/CLAUDE.md b/CLAUDE.md index 53ae339..62cbb09 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -6,7 +6,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co MeterVault is a self-hosted, local-first energy & utility metering platform: it ingests meter data from Home Assistant, Tasmota and MQTT on a schedule, stores every reading timestamped and immutable, normalizes it into consumption, and turns it into cost dashboards. Energy types (electricity, water, heating oil, gas, …) and meters are **user-defined, never hardcoded**. -**Status: implemented (M0–M7) + SDD §8 panels.** The full solution is built and green — five projects, ~108 tests, working Docker deploy. `docs/SDD.md` remains the design reference; the milestone map (§12) matches the git history (M0…M7 commits). The dedicated **PV/Solar** (`/solar`), **Oil/consumable** (`/consumables`) and **meter-detail** (`/meters/{id}`) views (SDD §8.4–§8.6) are implemented as read models in `Infrastructure/Dashboard` (`SolarService`, `ConsumableService`, `MeterDetailService`) — PV meters are found by `Mode == GenerationCounter` and grid/load meters by a `role` tag in `Meter.Meta` (`MeterRoles`/`MeterMeta`), so nothing is hardcoded by name. **Admin write-CRUD** (SDD §8.7) is implemented as MudBlazor inline-dialog pages: energy types, meters (+ recompute on mode/baseline change), a meter's ingest sources (meter-detail Sources tab), tariffs, cost categories + members, and connectors (`ingestion_endpoint`, secrets by env-var reference only). `/admin/settings` is a read-only effective-config view (settings are env-driven and reproducible, not DB-stored). **Home Assistant reading** is configured here: an HA connector (`BaseUrl` + `TokenEnv`) + an HA source (entity id) drives `HomeAssistantWorker`'s REST poll; `HaConnectionTester` powers the connector "Test connection" button. **Meter topology & flow**: `MeterLink` (a directed `from→to` edge; a downstream meter is a *subsection* of an upstream one, multi-parent allowed) drives a per-energy-type page `/energy/{id}` with a hand-rolled SVG **Sankey** (`SankeyChart.razor`, since ApexCharts has no Sankey type) computed by `FlowService` (link value = downstream consumption, split proportionally across multiple parents; unaccounted remainder → an "Other" node). Upstream meters are wired cycle-safely in the meter editor; the nav lists a link per energy type. Remaining refinements (HA WebSocket *push* — REST poll works today; commit-arbitrary-CSV-from-UI needs a meter-mapping wizard; full de-DE UI-string localization) are noted at their commits. Set `MeterVault__SeedReferenceData=true` (compose: `METERVAULT_SEED=true`) for a one-command populated demo. +**Status: implemented (M0–M7) + SDD §8 panels.** The full solution is built and green — five projects, ~108 tests, working Docker deploy. `docs/SDD.md` remains the design reference; the milestone map (§12) matches the git history (M0…M7 commits). The dedicated **PV/Solar** (`/solar`), **Oil/consumable** (`/consumables`) and **meter-detail** (`/meters/{id}`) views (SDD §8.4–§8.6) are implemented as read models in `Infrastructure/Dashboard` (`SolarService`, `ConsumableService`, `MeterDetailService`) — PV meters are found by `Mode == GenerationCounter` and grid/load meters by a `role` tag in `Meter.Meta` (`MeterRoles`/`MeterMeta`), so nothing is hardcoded by name. **Admin write-CRUD** (SDD §8.7) is implemented as MudBlazor inline-dialog pages: energy types, meters (+ recompute on mode/baseline change), a meter's ingest sources (meter-detail Sources tab), tariffs, cost categories + members, and connectors (`ingestion_endpoint`, secrets by env-var reference only). `/admin/settings` is a read-only effective-config view (settings are env-driven and reproducible, not DB-stored). **Home Assistant reading** is configured here: an HA connector (`BaseUrl` + `TokenEnv`) + an HA source (entity id) drives `HomeAssistantWorker`'s REST poll, or — with the connector's **WebSocket push** toggle (`HaEndpointConfig.UseWebSocket`) — `HomeAssistantWebSocketWorker` holds a persistent `state_changed` subscription and ingests in real time (the poll worker skips WS endpoints, so each is served once; `HaWebSocketProtocol` is the pure, unit-tested handshake/parse logic). `HaConnectionTester` powers the connector "Test connection" button. **Meter topology & flow**: `MeterLink` (a directed `from→to` edge; a downstream meter is a *subsection* of an upstream one, multi-parent allowed) drives a per-energy-type page `/energy/{id}` with a hand-rolled SVG **Sankey** (`SankeyChart.razor`, since ApexCharts has no Sankey type) computed by `FlowService` (link value = downstream consumption, split proportionally across multiple parents; unaccounted remainder → an "Other" node). Upstream meters are wired cycle-safely in the meter editor; the nav lists a link per energy type. **CSV mapping wizard** (`/import/wizard`): upload an arbitrary CSV, map columns → meters/roles, dry-run preview, then commit as a revertible `import_batch` (the `/import` page lists batches with one-click revert). The `instant_rate` mode is normalized (`InstantRateNormalizer`: rate integrated over time, trapezoidal). Remaining refinement: full de-DE UI-string localization (data parsing is already de-DE) — noted at its commit. Set `MeterVault__SeedReferenceData=true` (compose: `METERVAULT_SEED=true`) for a one-command populated demo. ## Source of truth diff --git a/src/App/Components/Pages/Admin/Connectors.razor b/src/App/Components/Pages/Admin/Connectors.razor index b76ca20..f4856b5 100644 --- a/src/App/Components/Pages/Admin/Connectors.razor +++ b/src/App/Components/Pages/Admin/Connectors.razor @@ -71,6 +71,10 @@ else { + + + On: subscribe to state changes and ingest in real time. Off: REST poll on each source's interval. + Test connection @@ -130,7 +134,7 @@ else _working = new EditModel { Id = endpoint.Id, Type = endpoint.Type, Name = endpoint.Name, IsEnabled = endpoint.IsEnabled, - BaseUrl = ha.BaseUrl, TokenEnv = ha.TokenEnv, + BaseUrl = ha.BaseUrl, TokenEnv = ha.TokenEnv, UseWebSocket = ha.UseWebSocket, }; } else @@ -170,7 +174,7 @@ else } var config = _working.Type == EndpointType.HomeAssistant - ? new HaEndpointConfig { BaseUrl = Trim(_working.BaseUrl), TokenEnv = Trim(_working.TokenEnv) }.ToJson() + ? new HaEndpointConfig { BaseUrl = Trim(_working.BaseUrl), TokenEnv = Trim(_working.TokenEnv), UseWebSocket = _working.UseWebSocket }.ToJson() : new EndpointConfig { Host = string.IsNullOrWhiteSpace(_working.Host) ? "localhost" : _working.Host.Trim(), @@ -234,6 +238,7 @@ else // Home Assistant public string? BaseUrl { get; set; } public string? TokenEnv { get; set; } + public bool UseWebSocket { get; set; } public string? TestEntityId { get; set; } // MQTT broker diff --git a/src/Infrastructure/DependencyInjection.cs b/src/Infrastructure/DependencyInjection.cs index 41bf6ad..bf7c11a 100644 --- a/src/Infrastructure/DependencyInjection.cs +++ b/src/Infrastructure/DependencyInjection.cs @@ -57,6 +57,7 @@ public static class DependencyInjection services.AddHttpClient(); services.AddHostedService(); services.AddHostedService(); + services.AddHostedService(); return services; } } diff --git a/src/Infrastructure/Ingestion/HaEndpointConfig.cs b/src/Infrastructure/Ingestion/HaEndpointConfig.cs index 8ef9a6b..ad78948 100644 --- a/src/Infrastructure/Ingestion/HaEndpointConfig.cs +++ b/src/Infrastructure/Ingestion/HaEndpointConfig.cs @@ -22,6 +22,13 @@ public sealed record HaEndpointConfig /// Name of the environment variable holding the long-lived access token. public string? TokenEnv { get; init; } + /// + /// When true, a persistent WebSocket subscription pushes state changes in real time + /// (); when false (default) the REST poll worker + /// samples on each source's interval. An endpoint is handled by exactly one of the two. + /// + public bool UseWebSocket { get; init; } + public static HaEndpointConfig Parse(string? json) { if (string.IsNullOrWhiteSpace(json)) diff --git a/src/Infrastructure/Ingestion/HaStateClient.cs b/src/Infrastructure/Ingestion/HaStateClient.cs index 21821ad..403654a 100644 --- a/src/Infrastructure/Ingestion/HaStateClient.cs +++ b/src/Infrastructure/Ingestion/HaStateClient.cs @@ -31,10 +31,16 @@ public sealed class HaStateClient(HttpClient httpClient) return ParseState(await JsonDocument.ParseAsync(stream, cancellationToken: cancellationToken).ConfigureAwait(false), attribute); } - internal static HaState? ParseState(JsonDocument document, string? attribute) - { - var root = document.RootElement; + internal static HaState? ParseState(JsonDocument document, string? attribute) => + ParseStateElement(document.RootElement, attribute); + /// + /// Extracts a numeric value + last-updated time from a Home Assistant state object (the REST + /// entity payload, or a new_state from a WebSocket state_changed event — they share + /// the same shape). Returns null for unavailable/unknown/non-numeric states. + /// + internal static HaState? ParseStateElement(JsonElement root, string? attribute) + { double value; if (string.IsNullOrWhiteSpace(attribute)) { diff --git a/src/Infrastructure/Ingestion/HaWebSocketProtocol.cs b/src/Infrastructure/Ingestion/HaWebSocketProtocol.cs new file mode 100644 index 0000000..c5af248 --- /dev/null +++ b/src/Infrastructure/Ingestion/HaWebSocketProtocol.cs @@ -0,0 +1,70 @@ +using System.Text.Json; + +namespace MeterVault.Infrastructure.Ingestion; + +/// +/// Pure helpers for the Home Assistant WebSocket API (/api/websocket): the auth handshake +/// (auth_required → auth → auth_ok), subscribing to state_changed events, +/// and reading the entity id + new_state out of an event frame. No I/O — the transport lives +/// in , which keeps this unit-testable. +/// +internal static class HaWebSocketProtocol +{ + /// Derives the WebSocket endpoint from an HTTP base URL (http→ws, https→wss, path /api/websocket). + public static Uri WebSocketUri(string baseUrl) + { + var http = new Uri(baseUrl.TrimEnd('/') + "/api/websocket", UriKind.Absolute); + var scheme = http.Scheme.Equals("https", StringComparison.OrdinalIgnoreCase) ? "wss" : "ws"; + return new UriBuilder(http) { Scheme = scheme }.Uri; + } + + public static string AuthMessage(string token) => + JsonSerializer.Serialize(new { type = "auth", access_token = token }); + + public static string SubscribeStateChanged(int id) => + JsonSerializer.Serialize(new { id, type = "subscribe_events", event_type = "state_changed" }); + + public static string? MessageType(JsonElement root) => + root.TryGetProperty("type", out var type) && type.ValueKind == JsonValueKind.String ? type.GetString() : null; + + public static bool IsAuthRequired(JsonElement root) => MessageType(root) == "auth_required"; + + public static bool IsAuthOk(JsonElement root) => MessageType(root) == "auth_ok"; + + public static bool IsAuthInvalid(JsonElement root) => MessageType(root) == "auth_invalid"; + + /// + /// If is a state_changed event with a numeric-capable + /// new_state object, yields the entity id and that state object. The caller extracts the + /// numeric value (state or a named attribute) per source via + /// . + /// + public static bool TryReadStateChanged(JsonElement root, out string entityId, out JsonElement newState) + { + entityId = string.Empty; + newState = default; + + if (MessageType(root) != "event" + || !root.TryGetProperty("event", out var evt) + || !evt.TryGetProperty("event_type", out var evtType) + || evtType.GetString() != "state_changed" + || !evt.TryGetProperty("data", out var data)) + { + return false; + } + + if (!data.TryGetProperty("entity_id", out var id) || id.ValueKind != JsonValueKind.String) + { + return false; + } + + if (!data.TryGetProperty("new_state", out var state) || state.ValueKind != JsonValueKind.Object) + { + return false; // entity removed (new_state null) — nothing to ingest. + } + + entityId = id.GetString()!; + newState = state; + return true; + } +} diff --git a/src/Infrastructure/Ingestion/HomeAssistantWebSocketWorker.cs b/src/Infrastructure/Ingestion/HomeAssistantWebSocketWorker.cs new file mode 100644 index 0000000..3c37d27 --- /dev/null +++ b/src/Infrastructure/Ingestion/HomeAssistantWebSocketWorker.cs @@ -0,0 +1,292 @@ +using System.Collections.Concurrent; +using System.Net.WebSockets; +using System.Text; +using System.Text.Json; +using MeterVault.Core.Domain; +using MeterVault.Infrastructure.Persistence; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +namespace MeterVault.Infrastructure.Ingestion; + +/// +/// Real-time Home Assistant ingestion via the WebSocket API (SDD §6.2 push path). For each enabled +/// HA endpoint whose config sets UseWebSocket, holds a persistent connection that authenticates, +/// subscribes to state_changed events, and ingests changes for the endpoint's configured +/// entities as they happen. Reconnects with capped backoff. Endpoints without UseWebSocket stay +/// on the REST poll worker () — each endpoint is served by exactly one. +/// Tokens are resolved from environment variables named in the endpoint config, never stored plaintext. +/// +public sealed class HomeAssistantWebSocketWorker( + IServiceScopeFactory scopeFactory, + ILogger logger) : BackgroundService +{ + private static readonly TimeSpan SuperviseInterval = TimeSpan.FromSeconds(15); + private static readonly TimeSpan InitialBackoff = TimeSpan.FromSeconds(2); + private static readonly TimeSpan MaxBackoff = TimeSpan.FromSeconds(60); + private static readonly TimeSpan EntityMapTtl = TimeSpan.FromSeconds(30); + + private readonly IServiceScopeFactory _scopeFactory = scopeFactory; + private readonly ILogger _logger = logger; + private readonly ConcurrentDictionary _connections = new(); + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + using var timer = new PeriodicTimer(SuperviseInterval); + do + { + try + { + await SuperviseAsync(stoppingToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Home Assistant WebSocket supervisor tick failed; will retry"); + } + } + while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false)); + } + + /// Starts a connection loop for each WebSocket-enabled endpoint that isn't already running. + private async Task SuperviseAsync(CancellationToken stoppingToken) + { + List wsEndpointIds; + await using (var scope = _scopeFactory.CreateAsyncScope()) + { + var db = scope.ServiceProvider.GetRequiredService(); + var endpoints = await db.IngestionEndpoints + .Where(e => e.IsEnabled && e.Type == EndpointType.HomeAssistant) + .ToListAsync(stoppingToken).ConfigureAwait(false); + wsEndpointIds = endpoints + .Where(e => HaEndpointConfig.Parse(e.Config).UseWebSocket) + .Select(e => e.Id) + .ToList(); + } + + // Reap finished connection loops (endpoint disabled/removed, or terminal failure). + foreach (var id in _connections.Where(kv => kv.Value.IsCompleted).Select(kv => kv.Key).ToList()) + { + _connections.TryRemove(id, out _); + } + + foreach (var endpointId in wsEndpointIds) + { + // Fire-and-forget: the loop stores its own Task in _connections and self-terminates when + // the endpoint is disabled/removed; the supervisor reaps completed entries above. + _ = _connections.GetOrAdd(endpointId, id => Task.Run(() => RunConnectionAsync(id, stoppingToken), stoppingToken)); + } + } + + /// Connect → listen → reconnect loop for one endpoint, until it is disabled or the app stops. + private async Task RunConnectionAsync(int endpointId, CancellationToken stoppingToken) + { + var backoff = InitialBackoff; + while (!stoppingToken.IsCancellationRequested) + { + HaEndpointConfig config; + string? token; + await using (var scope = _scopeFactory.CreateAsyncScope()) + { + var db = scope.ServiceProvider.GetRequiredService(); + var endpoint = await db.IngestionEndpoints + .FirstOrDefaultAsync(e => e.Id == endpointId, stoppingToken).ConfigureAwait(false); + if (endpoint is null || !endpoint.IsEnabled) + { + return; // gone/disabled — stop; the supervisor will restart it if it comes back. + } + + config = HaEndpointConfig.Parse(endpoint.Config); + if (!config.UseWebSocket) + { + return; // switched to poll mode. + } + + token = config.ResolveToken(); + } + + if (string.IsNullOrWhiteSpace(config.BaseUrl) || string.IsNullOrWhiteSpace(token)) + { + _logger.LogWarning("HA WebSocket endpoint {EndpointId} missing base URL or token; retrying", endpointId); + } + else + { + try + { + await ListenAsync(endpointId, config.BaseUrl!, token!, stoppingToken).ConfigureAwait(false); + backoff = InitialBackoff; // clean close → reset backoff. + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + return; + } + catch (Exception ex) + { + _logger.LogWarning(ex, "HA WebSocket connection to endpoint {EndpointId} dropped; reconnecting", endpointId); + await UpdateEndpointStatusAsync(endpointId, "disconnected", CancellationToken.None).ConfigureAwait(false); + } + } + + try + { + await Task.Delay(backoff, stoppingToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + return; + } + + backoff = TimeSpan.FromSeconds(Math.Min(MaxBackoff.TotalSeconds, backoff.TotalSeconds * 2)); + } + } + + private async Task ListenAsync(int endpointId, string baseUrl, string token, CancellationToken stoppingToken) + { + using var ws = new ClientWebSocket(); + await ws.ConnectAsync(HaWebSocketProtocol.WebSocketUri(baseUrl), stoppingToken).ConfigureAwait(false); + + // Handshake: auth_required → auth → auth_ok. + using (var required = await ReceiveJsonAsync(ws, stoppingToken).ConfigureAwait(false)) + { + if (!HaWebSocketProtocol.IsAuthRequired(required.RootElement)) + { + // Some setups may not send auth_required first; proceed to auth regardless. + _logger.LogDebug("HA WebSocket did not send auth_required first (type={Type})", HaWebSocketProtocol.MessageType(required.RootElement)); + } + } + + await SendAsync(ws, HaWebSocketProtocol.AuthMessage(token), stoppingToken).ConfigureAwait(false); + using (var authResult = await ReceiveJsonAsync(ws, stoppingToken).ConfigureAwait(false)) + { + if (!HaWebSocketProtocol.IsAuthOk(authResult.RootElement)) + { + throw new InvalidOperationException( + $"Home Assistant WebSocket auth failed (type={HaWebSocketProtocol.MessageType(authResult.RootElement)})."); + } + } + + await SendAsync(ws, HaWebSocketProtocol.SubscribeStateChanged(1), stoppingToken).ConfigureAwait(false); + await UpdateEndpointStatusAsync(endpointId, "connected (ws)", stoppingToken).ConfigureAwait(false); + _logger.LogInformation("HA WebSocket connected for endpoint {EndpointId}", endpointId); + + var entityMap = await LoadEntityMapAsync(endpointId, stoppingToken).ConfigureAwait(false); + var mapLoadedAt = DateTimeOffset.UtcNow; + + while (!stoppingToken.IsCancellationRequested && ws.State == WebSocketState.Open) + { + using var doc = await ReceiveJsonAsync(ws, stoppingToken).ConfigureAwait(false); + if (!HaWebSocketProtocol.TryReadStateChanged(doc.RootElement, out var entityId, out var newState)) + { + continue; + } + + if (DateTimeOffset.UtcNow - mapLoadedAt > EntityMapTtl) + { + entityMap = await LoadEntityMapAsync(endpointId, stoppingToken).ConfigureAwait(false); + mapLoadedAt = DateTimeOffset.UtcNow; + } + + if (!entityMap.TryGetValue(entityId, out var sources)) + { + continue; + } + + foreach (var (sourceId, attribute) in sources) + { + if (HaStateClient.ParseStateElement(newState, attribute) is { } state) + { + await IngestAsync(sourceId, state, stoppingToken).ConfigureAwait(false); + } + } + } + } + + /// Maps each configured entity id to the sources (and optional attribute) that read it. + private async Task>> LoadEntityMapAsync( + int endpointId, CancellationToken cancellationToken) + { + await using var scope = _scopeFactory.CreateAsyncScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var sources = await db.MeterSources.AsNoTracking() + .Where(s => s.IsEnabled && s.SourceType == SourceType.HomeAssistant && s.EndpointId == endpointId) + .ToListAsync(cancellationToken).ConfigureAwait(false); + + var map = new Dictionary>(StringComparer.OrdinalIgnoreCase); + foreach (var source in sources) + { + var config = SourceConfig.Parse(source.Config); + if (string.IsNullOrWhiteSpace(config.EntityId)) + { + continue; + } + + if (!map.TryGetValue(config.EntityId, out var list)) + { + map[config.EntityId] = list = []; + } + + list.Add((source.Id, config.Attribute)); + } + + return map; + } + + private async Task IngestAsync(int sourceId, HaState state, CancellationToken cancellationToken) + { + await using var scope = _scopeFactory.CreateAsyncScope(); + var ingestion = scope.ServiceProvider.GetRequiredService(); + await ingestion.IngestAsync(sourceId, state.Time, state.Value, cancellationToken).ConfigureAwait(false); + } + + private async Task UpdateEndpointStatusAsync(int endpointId, string status, CancellationToken cancellationToken) + { + try + { + await using var scope = _scopeFactory.CreateAsyncScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var endpoint = await db.IngestionEndpoints + .FirstOrDefaultAsync(e => e.Id == endpointId, cancellationToken).ConfigureAwait(false); + if (endpoint is not null) + { + endpoint.LastStatus = status; + endpoint.LastSeenAt = DateTimeOffset.UtcNow; + await db.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + } + } + catch (Exception ex) + { + _logger.LogDebug(ex, "Failed to update HA endpoint {EndpointId} status", endpointId); + } + } + + private static Task SendAsync(ClientWebSocket ws, string json, CancellationToken cancellationToken) => + ws.SendAsync(Encoding.UTF8.GetBytes(json), WebSocketMessageType.Text, endOfMessage: true, cancellationToken); + + /// Reads one (possibly fragmented) text message and parses it as JSON. + private static async Task ReceiveJsonAsync(ClientWebSocket ws, CancellationToken cancellationToken) + { + using var buffer = new MemoryStream(); + var chunk = new byte[8192]; + WebSocketReceiveResult result; + do + { + result = await ws.ReceiveAsync(chunk, cancellationToken).ConfigureAwait(false); + if (result.MessageType == WebSocketMessageType.Close) + { + throw new InvalidOperationException( + $"Home Assistant closed the WebSocket ({result.CloseStatus}: {result.CloseStatusDescription})."); + } + + buffer.Write(chunk, 0, result.Count); + } + while (!result.EndOfMessage); + + buffer.Position = 0; + return await JsonDocument.ParseAsync(buffer, cancellationToken: cancellationToken).ConfigureAwait(false); + } +} diff --git a/src/Infrastructure/Ingestion/HomeAssistantWorker.cs b/src/Infrastructure/Ingestion/HomeAssistantWorker.cs index 4dffd34..5a8f7d8 100644 --- a/src/Infrastructure/Ingestion/HomeAssistantWorker.cs +++ b/src/Infrastructure/Ingestion/HomeAssistantWorker.cs @@ -54,9 +54,14 @@ public sealed class HomeAssistantWorker( var ingestion = scope.ServiceProvider.GetRequiredService(); var client = new HaStateClient(_httpClientFactory.CreateClient()); - var endpoints = await db.IngestionEndpoints + // Endpoints using the WebSocket push path are served by HomeAssistantWebSocketWorker; skip + // them here so a source is never both polled and pushed. + var enabled = await db.IngestionEndpoints .Where(e => e.IsEnabled && e.Type == EndpointType.HomeAssistant) - .ToDictionaryAsync(e => e.Id, cancellationToken).ConfigureAwait(false); + .ToListAsync(cancellationToken).ConfigureAwait(false); + var endpoints = enabled + .Where(e => !HaEndpointConfig.Parse(e.Config).UseWebSocket) + .ToDictionary(e => e.Id); if (endpoints.Count == 0) { return; diff --git a/tests/Integration.Tests/Ingestion/HaWebSocketProtocolTests.cs b/tests/Integration.Tests/Ingestion/HaWebSocketProtocolTests.cs new file mode 100644 index 0000000..f8675f5 --- /dev/null +++ b/tests/Integration.Tests/Ingestion/HaWebSocketProtocolTests.cs @@ -0,0 +1,119 @@ +using System.Text.Json; +using MeterVault.Infrastructure.Ingestion; + +namespace MeterVault.Integration.Tests.Ingestion; + +/// +/// The pure Home Assistant WebSocket protocol helpers: URL derivation, the auth/subscribe frames, +/// and reading the entity id + numeric value out of a state_changed event (the transport is +/// exercised separately). No network — these are the parts that must be provably correct. +/// +public sealed class HaWebSocketProtocolTests +{ + [Theory] + [InlineData("http://ha.local:8123", "ws", "ha.local", 8123)] + [InlineData("http://ha.local:8123/", "ws", "ha.local", 8123)] + [InlineData("https://ha.example.com", "wss", "ha.example.com", 443)] + public void WebSocketUri_maps_scheme_and_appends_api_path(string baseUrl, string scheme, string host, int port) + { + var uri = HaWebSocketProtocol.WebSocketUri(baseUrl); + + Assert.Equal(scheme, uri.Scheme); + Assert.Equal(host, uri.Host); + Assert.Equal(port, uri.Port); + Assert.Equal("/api/websocket", uri.AbsolutePath); + } + + [Fact] + public void AuthMessage_carries_type_and_token() + { + using var doc = JsonDocument.Parse(HaWebSocketProtocol.AuthMessage("secret-token")); + + Assert.Equal("auth", doc.RootElement.GetProperty("type").GetString()); + Assert.Equal("secret-token", doc.RootElement.GetProperty("access_token").GetString()); + } + + [Fact] + public void SubscribeStateChanged_requests_state_changed_events() + { + using var doc = JsonDocument.Parse(HaWebSocketProtocol.SubscribeStateChanged(7)); + + Assert.Equal(7, doc.RootElement.GetProperty("id").GetInt32()); + Assert.Equal("subscribe_events", doc.RootElement.GetProperty("type").GetString()); + Assert.Equal("state_changed", doc.RootElement.GetProperty("event_type").GetString()); + } + + [Theory] + [InlineData("{\"type\":\"auth_required\",\"ha_version\":\"2024.6\"}", true, false, false)] + [InlineData("{\"type\":\"auth_ok\"}", false, true, false)] + [InlineData("{\"type\":\"auth_invalid\",\"message\":\"bad token\"}", false, false, true)] + public void Auth_message_types_are_recognized(string json, bool required, bool ok, bool invalid) + { + using var doc = JsonDocument.Parse(json); + var root = doc.RootElement; + + Assert.Equal(required, HaWebSocketProtocol.IsAuthRequired(root)); + Assert.Equal(ok, HaWebSocketProtocol.IsAuthOk(root)); + Assert.Equal(invalid, HaWebSocketProtocol.IsAuthInvalid(root)); + } + + [Fact] + public void TryReadStateChanged_extracts_entity_and_numeric_state() + { + using var doc = JsonDocument.Parse(StateChangedEvent(state: "1234.5")); + + Assert.True(HaWebSocketProtocol.TryReadStateChanged(doc.RootElement, out var entityId, out var newState)); + Assert.Equal("sensor.house_power", entityId); + + var parsed = HaStateClient.ParseStateElement(newState, attribute: null); + Assert.NotNull(parsed); + Assert.Equal(1234.5, parsed!.Value.Value, 3); + Assert.Equal(new DateTimeOffset(2024, 6, 15, 10, 0, 0, TimeSpan.Zero), parsed.Value.Time); + } + + [Fact] + public void TryReadStateChanged_reads_a_named_attribute() + { + using var doc = JsonDocument.Parse(StateChangedEvent(state: "on")); + + Assert.True(HaWebSocketProtocol.TryReadStateChanged(doc.RootElement, out _, out var newState)); + // state "on" is non-numeric, but the 'current' attribute is a number. + Assert.Null(HaStateClient.ParseStateElement(newState, attribute: null)); + var byAttribute = HaStateClient.ParseStateElement(newState, attribute: "current"); + Assert.Equal(42, byAttribute!.Value.Value, 3); + } + + [Fact] + public void TryReadStateChanged_ignores_removed_entities_and_other_events() + { + using var removed = JsonDocument.Parse( + "{\"type\":\"event\",\"event\":{\"event_type\":\"state_changed\",\"data\":{\"entity_id\":\"sensor.x\",\"new_state\":null}}}"); + Assert.False(HaWebSocketProtocol.TryReadStateChanged(removed.RootElement, out _, out _)); + + using var other = JsonDocument.Parse( + "{\"type\":\"event\",\"event\":{\"event_type\":\"call_service\",\"data\":{}}}"); + Assert.False(HaWebSocketProtocol.TryReadStateChanged(other.RootElement, out _, out _)); + + using var result = JsonDocument.Parse("{\"id\":1,\"type\":\"result\",\"success\":true}"); + Assert.False(HaWebSocketProtocol.TryReadStateChanged(result.RootElement, out _, out _)); + } + + private static string StateChangedEvent(string state) => $$""" + { + "id": 1, + "type": "event", + "event": { + "event_type": "state_changed", + "data": { + "entity_id": "sensor.house_power", + "new_state": { + "entity_id": "sensor.house_power", + "state": "{{state}}", + "attributes": { "unit_of_measurement": "W", "current": 42 }, + "last_updated": "2024-06-15T10:00:00+00:00" + } + } + } + } + """; +} diff --git a/tests/Integration.Tests/Ingestion/HomeAssistantWebSocketWorkerTests.cs b/tests/Integration.Tests/Ingestion/HomeAssistantWebSocketWorkerTests.cs new file mode 100644 index 0000000..4635aa2 --- /dev/null +++ b/tests/Integration.Tests/Ingestion/HomeAssistantWebSocketWorkerTests.cs @@ -0,0 +1,182 @@ +using System.Net.WebSockets; +using System.Text; +using System.Text.Json; +using MeterVault.Core.Domain; +using MeterVault.Infrastructure.Ingestion; +using MeterVault.Infrastructure.Persistence; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Hosting.Server; +using Microsoft.AspNetCore.Hosting.Server.Features; +using Microsoft.AspNetCore.Http; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; + +namespace MeterVault.Integration.Tests.Ingestion; + +/// +/// End-to-end proof of the HA WebSocket push path against an in-process fake Home Assistant server: +/// the worker performs the auth handshake, subscribes to state_changed, and a pushed change +/// for a configured entity lands as a reading. This exercises the real ClientWebSocket +/// transport and handshake sequencing that the pure-protocol unit tests can't. +/// +[Collection("Timescale")] +public sealed class HomeAssistantWebSocketWorkerTests(TimescaleFixture fx) +{ + private const string TokenEnvVar = "MV_TEST_HA_WS_TOKEN"; + + [Fact] + public async Task Pushed_state_change_becomes_a_reading() + { + await using var db = fx.CreateContext(); + Environment.SetEnvironmentVariable(TokenEnvVar, "test-token"); + await using var fake = await FakeHaServer.StartAsync(entityId: "sensor.house_power", state: "4711"); + + try + { + var type = new EnergyType { Key = "ha_ws_test", DisplayName = "HA WS", BaseUnit = "kWh", DefaultMode = MeterMode.CumulativeCounter }; + db.EnergyTypes.Add(type); + await db.SaveChangesAsync(); + + var meter = new Meter { Name = "HA WS Meter", EnergyTypeId = type.Id, Mode = MeterMode.CumulativeCounter, Unit = "kWh" }; + db.Meters.Add(meter); + await db.SaveChangesAsync(); + + var endpoint = new IngestionEndpoint + { + Type = EndpointType.HomeAssistant, + Name = "Fake HA", + IsEnabled = true, + Config = new HaEndpointConfig { BaseUrl = fake.BaseUrl, TokenEnv = TokenEnvVar, UseWebSocket = true }.ToJson(), + }; + db.IngestionEndpoints.Add(endpoint); + await db.SaveChangesAsync(); + + db.MeterSources.Add(new MeterSource + { + MeterId = meter.Id, + SourceType = SourceType.HomeAssistant, + EndpointId = endpoint.Id, + IsEnabled = true, + Config = JsonSerializer.Serialize(new { entityId = "sensor.house_power" }), + }); + await db.SaveChangesAsync(); + + await using var provider = BuildProvider(fx.ConnectionString); + var worker = new HomeAssistantWebSocketWorker( + provider.GetRequiredService(), NullLogger.Instance); + + await worker.StartAsync(CancellationToken.None); + try + { + Reading? reading = null; + for (var i = 0; i < 60 && reading is null; i++) + { + await Task.Delay(200); + reading = await db.Readings.AsNoTracking().FirstOrDefaultAsync(r => r.MeterId == meter.Id); + } + + Assert.NotNull(reading); + Assert.Equal(4711, reading!.Value, 3); + Assert.Equal(ReadingQuality.Measured, reading.Quality); + } + finally + { + await worker.StopAsync(CancellationToken.None); + } + } + finally + { + Environment.SetEnvironmentVariable(TokenEnvVar, null); + await db.Readings.ExecuteDeleteAsync(); + await db.MeterSources.ExecuteDeleteAsync(); + await db.IngestionEndpoints.ExecuteDeleteAsync(); + await db.Meters.ExecuteDeleteAsync(); + await db.EnergyTypes.Where(t => t.Key == "ha_ws_test").ExecuteDeleteAsync(); + } + } + + private static ServiceProvider BuildProvider(string connectionString) + { + var services = new ServiceCollection(); + services.AddLogging(); + services.AddDbContextFactory(o => o + .UseNpgsql(connectionString, n => n.MigrationsAssembly(typeof(MeterVaultDbContext).Assembly.FullName)) + .UseSnakeCaseNamingConvention()); + services.AddScoped(sp => sp.GetRequiredService>().CreateDbContext()); + services.AddScoped(); + return services.BuildServiceProvider(); + } + + /// A minimal Home Assistant WebSocket server: handshake, then push one state_changed event. + private sealed class FakeHaServer : IAsyncDisposable + { + private readonly WebApplication _app; + + private FakeHaServer(WebApplication app, string baseUrl) + { + _app = app; + BaseUrl = baseUrl; + } + + public string BaseUrl { get; } + + public static async Task StartAsync(string entityId, string state) + { + var builder = WebApplication.CreateBuilder(); + builder.Logging.ClearProviders(); + builder.WebHost.UseUrls("http://127.0.0.1:0"); + var app = builder.Build(); + app.UseWebSockets(); + + var eventJson = + "{\"id\":1,\"type\":\"event\",\"event\":{\"event_type\":\"state_changed\",\"data\":{\"entity_id\":\"" + + entityId + "\",\"new_state\":{\"entity_id\":\"" + entityId + "\",\"state\":\"" + state + + "\",\"attributes\":{},\"last_updated\":\"2024-06-15T10:00:00+00:00\"}}}}"; + + app.Map("/api/websocket", async context => + { + if (!context.WebSockets.IsWebSocketRequest) + { + context.Response.StatusCode = 400; + return; + } + + using var ws = await context.WebSockets.AcceptWebSocketAsync(); + await SendAsync(ws, """{"type":"auth_required","ha_version":"2024.6"}"""); + await ReceiveAsync(ws); // client "auth" + await SendAsync(ws, """{"type":"auth_ok"}"""); + await ReceiveAsync(ws); // client "subscribe_events" + await SendAsync(ws, """{"id":1,"type":"result","success":true}"""); + await SendAsync(ws, eventJson); + + try + { + await Task.Delay(Timeout.Infinite, context.RequestAborted); + } + catch (OperationCanceledException) + { + // client/test closed — expected. + } + }); + + await app.StartAsync(); + var address = app.Services.GetRequiredService().Features + .Get()!.Addresses.First(); + return new FakeHaServer(app, address); + } + + public async ValueTask DisposeAsync() => await _app.DisposeAsync(); + + private static Task SendAsync(WebSocket ws, string json) => + ws.SendAsync(Encoding.UTF8.GetBytes(json), WebSocketMessageType.Text, endOfMessage: true, CancellationToken.None); + + private static async Task ReceiveAsync(WebSocket ws) + { + var buffer = new byte[8192]; + await ws.ReceiveAsync(buffer, CancellationToken.None); + } + } +}