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

src/Infrastructure/DependencyInjection.cs

@@ -57,6 +57,7 @@ public static class DependencyInjection
         services.AddHttpClient();
         services.AddHostedService<MqttIngestionWorker>();
         services.AddHostedService<HomeAssistantWorker>();
+        services.AddHostedService<HomeAssistantWebSocketWorker>();
         return services;
     }
 }

src/Infrastructure/Ingestion/HaEndpointConfig.cs

@@ -22,6 +22,13 @@ public sealed record HaEndpointConfig
     /// <summary>Name of the environment variable holding the long-lived access token.</summary>
     public string? TokenEnv { get; init; }
 
+    /// <summary>
+    /// When true, a persistent WebSocket subscription pushes state changes in real time
+    /// (<see cref="HomeAssistantWebSocketWorker"/>); when false (default) the REST poll worker
+    /// samples on each source's interval. An endpoint is handled by exactly one of the two.
+    /// </summary>
+    public bool UseWebSocket { get; init; }
+
     public static HaEndpointConfig Parse(string? json)
     {
         if (string.IsNullOrWhiteSpace(json))

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);
 
+    /// <summary>
+    /// Extracts a numeric value + last-updated time from a Home Assistant state object (the REST
+    /// entity payload, or a <c>new_state</c> from a WebSocket <c>state_changed</c> event — they share
+    /// the same shape). Returns null for <c>unavailable</c>/<c>unknown</c>/non-numeric states.
+    /// </summary>
+    internal static HaState? ParseStateElement(JsonElement root, string? attribute)
+    {
         double value;
         if (string.IsNullOrWhiteSpace(attribute))
         {

src/Infrastructure/Ingestion/HaWebSocketProtocol.cs

@@ -0,0 +1,70 @@
+using System.Text.Json;
+
+namespace MeterVault.Infrastructure.Ingestion;
+
+/// <summary>
+/// Pure helpers for the Home Assistant WebSocket API (<c>/api/websocket</c>): the auth handshake
+/// (<c>auth_required</c> → <c>auth</c> → <c>auth_ok</c>), subscribing to <c>state_changed</c> events,
+/// and reading the entity id + <c>new_state</c> out of an event frame. No I/O — the transport lives
+/// in <see cref="HomeAssistantWebSocketWorker"/>, which keeps this unit-testable.
+/// </summary>
+internal static class HaWebSocketProtocol
+{
+    /// <summary>Derives the WebSocket endpoint from an HTTP base URL (http→ws, https→wss, path <c>/api/websocket</c>).</summary>
+    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";
+
+    /// <summary>
+    /// If <paramref name="root"/> is a <c>state_changed</c> event with a numeric-capable
+    /// <c>new_state</c> object, yields the entity id and that state object. The caller extracts the
+    /// numeric value (state or a named attribute) per source via
+    /// <see cref="HaStateClient.ParseStateElement"/>.
+    /// </summary>
+    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;
+    }
+}

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;
+
+/// <summary>
+/// Real-time Home Assistant ingestion via the WebSocket API (SDD §6.2 push path). For each enabled
+/// HA endpoint whose config sets <c>UseWebSocket</c>, holds a persistent connection that authenticates,
+/// subscribes to <c>state_changed</c> events, and ingests changes for the endpoint's configured
+/// entities as they happen. Reconnects with capped backoff. Endpoints without <c>UseWebSocket</c> stay
+/// on the REST poll worker (<see cref="HomeAssistantWorker"/>) — each endpoint is served by exactly one.
+/// Tokens are resolved from environment variables named in the endpoint config, never stored plaintext.
+/// </summary>
+public sealed class HomeAssistantWebSocketWorker(
+    IServiceScopeFactory scopeFactory,
+    ILogger<HomeAssistantWebSocketWorker> 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<HomeAssistantWebSocketWorker> _logger = logger;
+    private readonly ConcurrentDictionary<int, Task> _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));
+    }
+
+    /// <summary>Starts a connection loop for each WebSocket-enabled endpoint that isn't already running.</summary>
+    private async Task SuperviseAsync(CancellationToken stoppingToken)
+    {
+        List<int> wsEndpointIds;
+        await using (var scope = _scopeFactory.CreateAsyncScope())
+        {
+            var db = scope.ServiceProvider.GetRequiredService<MeterVaultDbContext>();
+            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));
+        }
+    }
+
+    /// <summary>Connect → listen → reconnect loop for one endpoint, until it is disabled or the app stops.</summary>
+    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<MeterVaultDbContext>();
+                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);
+                }
+            }
+        }
+    }
+
+    /// <summary>Maps each configured entity id to the sources (and optional attribute) that read it.</summary>
+    private async Task<Dictionary<string, List<(int SourceId, string? Attribute)>>> LoadEntityMapAsync(
+        int endpointId, CancellationToken cancellationToken)
+    {
+        await using var scope = _scopeFactory.CreateAsyncScope();
+        var db = scope.ServiceProvider.GetRequiredService<MeterVaultDbContext>();
+        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<string, List<(int, string?)>>(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<IngestionService>();
+        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<MeterVaultDbContext>();
+            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);
+
+    /// <summary>Reads one (possibly fragmented) text message and parses it as JSON.</summary>
+    private static async Task<JsonDocument> 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);
+    }
+}

src/Infrastructure/Ingestion/HomeAssistantWorker.cs

@@ -54,9 +54,14 @@ public sealed class HomeAssistantWorker(
         var ingestion = scope.ServiceProvider.GetRequiredService<IngestionService>();
         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;