M3: live ingestion (MQTT/Tasmota + Home Assistant)

- PayloadExtractor: dot-path value/time extraction (Tasmota ENERGY.Total, bare scalars). - MqttTopicMatcher: standard +/# wildcard matching. - IngestionService: scale/offset, idempotent upsert on (meter_id, time), and a spurious- decrease guard for monotonic registers (allowed only with a reset/swap event) + source last-seen status. - MqttMessageRouter + MqttIngestionWorker (MQTTnet 5): per-endpoint persistent connections, topic subscription, graceful degradation; secrets resolved by env-var reference. - Home Assistant: HaStateClient (REST /api/states parse) + HomeAssistantWorker polling on each source's interval. HA-via-MQTT also works through the MQTT path. - Ingestion workers gated by MeterVault:EnableLiveIngestion (off in tests). 85 tests green (53 Core + 32 integration): Tasmota payload → reading verified end to end. Follow-up (polish): HA WebSocket push (state_changed) as an alternative to REST poll; source-topic index caching in the router. Claude-Session: https://claude.ai/code/session_01WujdMtMJPbxDpDnMeK22rr
2026-07-13 11:45:20 +02:00
parent 5977c81002
commit 4b0cad67df
17 changed files with 1060 additions and 0 deletions
@@ -0,0 +1,115 @@
+using MeterVault.Core.Domain;
+using MeterVault.Infrastructure.Persistence;
+using Microsoft.EntityFrameworkCore;
+
+namespace MeterVault.Infrastructure.Ingestion;
+
+/// <summary>The outcome of ingesting one reading.</summary>
+public enum IngestionOutcome
+{
+    Written,
+    Updated,
+    RejectedDecrease,
+    UnknownSource,
+}
+
+/// <summary>
+/// Persists a single incoming reading (SDD §6.1, FR-4): applies the source's scale/offset, is
+/// idempotent on (meter_id, time), and guards monotonic registers against spurious decreases
+/// unless an active reset/swap event explains them. Updates the source's last-seen status.
+/// Consumption normalization is recomputed separately (batch/scheduled), not per message.
+/// </summary>
+public sealed class IngestionService(MeterVaultDbContext db)
+{
+    private static readonly HashSet<MeterMode> MonotonicModes =
+        [MeterMode.CumulativeCounter, MeterMode.GenerationCounter, MeterMode.RuntimeCounter];
+
+    private readonly MeterVaultDbContext _db = db;
+
+    public async Task<IngestionOutcome> IngestAsync(
+        int sourceId, DateTimeOffset time, double rawValue, CancellationToken cancellationToken = default)
+    {
+        var source = await _db.MeterSources
+            .FirstOrDefaultAsync(s => s.Id == sourceId, cancellationToken).ConfigureAwait(false);
+        if (source is null)
+        {
+            return IngestionOutcome.UnknownSource;
+        }
+
+        var meter = await _db.Meters
+            .FirstOrDefaultAsync(m => m.Id == source.MeterId, cancellationToken).ConfigureAwait(false);
+        if (meter is null)
+        {
+            return IngestionOutcome.UnknownSource;
+        }
+
+        var value = (rawValue * source.Scale) + source.Offset;
+        var utc = time.ToUniversalTime();
+
+        if (MonotonicModes.Contains(meter.Mode)
+            && await IsSpuriousDecreaseAsync(meter.Id, utc, value, cancellationToken).ConfigureAwait(false))
+        {
+            await UpdateSourceStatusAsync(source, utc, value, "rejected: decrease", cancellationToken).ConfigureAwait(false);
+            return IngestionOutcome.RejectedDecrease;
+        }
+
+        var existing = await _db.Readings
+            .FirstOrDefaultAsync(r => r.MeterId == meter.Id && r.Time == utc, cancellationToken).ConfigureAwait(false);
+
+        IngestionOutcome outcome;
+        if (existing is null)
+        {
+            _db.Readings.Add(new Reading
+            {
+                MeterId = meter.Id,
+                Time = utc,
+                Value = value,
+                SourceId = source.Id,
+                Quality = ReadingQuality.Measured,
+            });
+            outcome = IngestionOutcome.Written;
+        }
+        else
+        {
+            existing.Value = value;
+            existing.SourceId = source.Id;
+            outcome = IngestionOutcome.Updated;
+        }
+
+        await UpdateSourceStatusAsync(source, utc, value, "ok", cancellationToken).ConfigureAwait(false);
+        return outcome;
+    }
+
+    private async Task<bool> IsSpuriousDecreaseAsync(
+        int meterId, DateTimeOffset time, double value, CancellationToken cancellationToken)
+    {
+        var previous = await _db.Readings
+            .Where(r => r.MeterId == meterId && r.Time < time)
+            .OrderByDescending(r => r.Time)
+            .Select(r => (double?)r.Value)
+            .FirstOrDefaultAsync(cancellationToken).ConfigureAwait(false);
+
+        if (previous is null || value >= previous.Value)
+        {
+            return false;
+        }
+
+        // A reset/swap event between the previous reading and this one explains the decrease.
+        var explained = await _db.MeterEvents.AnyAsync(
+            e => e.MeterId == meterId
+                 && (e.EventType == MeterEventType.CounterReset || e.EventType == MeterEventType.MeterSwap)
+                 && e.Time <= time,
+            cancellationToken).ConfigureAwait(false);
+
+        return !explained;
+    }
+
+    private async Task UpdateSourceStatusAsync(
+        MeterSource source, DateTimeOffset time, double value, string status, CancellationToken cancellationToken)
+    {
+        source.LastSeenAt = time;
+        source.LastValue = value;
+        source.LastStatus = status;
+        await _db.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
+    }
+}