MeterVault/src/Infrastructure/Ingestion/HomeAssistantWorker.cs

using System.Collections.Concurrent;
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>
/// Polls Home Assistant entity states on each source's configured interval and ingests them
/// (SDD §6.2 REST fallback). Home Assistant can alternatively push to MQTT (handled by
/// <see cref="MqttIngestionWorker"/>) or push to the app's REST API. Tokens are resolved from
/// environment variables named in the endpoint config — never stored in the database.
/// </summary>
public sealed class HomeAssistantWorker(
    IServiceScopeFactory scopeFactory,
    IHttpClientFactory httpClientFactory,
    ILogger<HomeAssistantWorker> logger) : BackgroundService
{
    private static readonly TimeSpan TickInterval = TimeSpan.FromSeconds(10);

    private readonly IServiceScopeFactory _scopeFactory = scopeFactory;
    private readonly IHttpClientFactory _httpClientFactory = httpClientFactory;
    private readonly ILogger<HomeAssistantWorker> _logger = logger;
    private readonly ConcurrentDictionary<int, DateTimeOffset> _nextPoll = new();

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        using var timer = new PeriodicTimer(TickInterval);
        do
        {
            try
            {
                await PollDueSourcesAsync(stoppingToken).ConfigureAwait(false);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "Home Assistant poll tick failed; will retry");
            }
        }
        while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false));
    }

    private async Task PollDueSourcesAsync(CancellationToken cancellationToken)
    {
        await using var scope = _scopeFactory.CreateAsyncScope();
        var db = scope.ServiceProvider.GetRequiredService<MeterVaultDbContext>();
        var ingestion = scope.ServiceProvider.GetRequiredService<IngestionService>();
        var client = new HaStateClient(_httpClientFactory.CreateClient());

        var endpoints = await db.IngestionEndpoints
            .Where(e => e.IsEnabled && e.Type == EndpointType.HomeAssistant)
            .ToDictionaryAsync(e => e.Id, cancellationToken).ConfigureAwait(false);
        if (endpoints.Count == 0)
        {
            return;
        }

        var sources = await db.MeterSources
            .Where(s => s.IsEnabled && s.SourceType == SourceType.HomeAssistant && s.EndpointId != null)
            .ToListAsync(cancellationToken).ConfigureAwait(false);

        // Drop schedule entries for sources that are gone/disabled so the dictionary doesn't grow.
        var liveIds = sources.Select(s => s.Id).ToHashSet();
        foreach (var staleId in _nextPoll.Keys.Where(id => !liveIds.Contains(id)).ToList())
        {
            _nextPoll.TryRemove(staleId, out _);
        }

        var now = DateTimeOffset.UtcNow;
        foreach (var source in sources)
        {
            if (_nextPoll.TryGetValue(source.Id, out var due) && due > now)
            {
                continue;
            }

            if (source.EndpointId is not { } endpointId || !endpoints.TryGetValue(endpointId, out var endpoint))
            {
                continue;
            }

            var config = SourceConfig.Parse(source.Config);
            var interval = TimeSpan.FromSeconds(Math.Max(5, config.PollSeconds ?? 60));
            _nextPoll[source.Id] = now + interval;

            await PollSourceAsync(client, ingestion, endpoint, source, config, cancellationToken).ConfigureAwait(false);
        }
    }

    private async Task PollSourceAsync(
        HaStateClient client, IngestionService ingestion, IngestionEndpoint endpoint,
        MeterSource source, SourceConfig config, CancellationToken cancellationToken)
    {
        var endpointConfig = ParseHaEndpoint(endpoint.Config);
        var token = string.IsNullOrWhiteSpace(endpointConfig.TokenEnv)
            ? null
            : Environment.GetEnvironmentVariable(endpointConfig.TokenEnv);

        if (string.IsNullOrWhiteSpace(endpointConfig.BaseUrl) || string.IsNullOrWhiteSpace(token)
            || string.IsNullOrWhiteSpace(config.EntityId))
        {
            return;
        }

        try
        {
            var state = await client.GetStateAsync(
                endpointConfig.BaseUrl, token, config.EntityId, config.Attribute, cancellationToken).ConfigureAwait(false);
            if (state is { } value)
            {
                await ingestion.IngestAsync(source.Id, value.Time, value.Value, cancellationToken).ConfigureAwait(false);
            }
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "Home Assistant poll failed for entity {EntityId}", config.EntityId);
        }
    }

    private static HaEndpoint ParseHaEndpoint(string? json)
    {
        if (string.IsNullOrWhiteSpace(json))
        {
            return new HaEndpoint();
        }

        try
        {
            return JsonSerializer.Deserialize<HaEndpoint>(json,
                new JsonSerializerOptions { PropertyNameCaseInsensitive = true }) ?? new HaEndpoint();
        }
        catch (JsonException)
        {
            return new HaEndpoint();
        }
    }

    private sealed record HaEndpoint
    {
        public string? BaseUrl { get; init; }

        public string? TokenEnv { get; init; }
    }
}