schmidt.florian
c0bbaba99f
ci / build-test (push) Successful in 1m8s
MqttMessageRouter matched purely on topic with no endpoint predicate, and RouteAsync was not even passed an endpoint id. Topic filters routinely overlap between brokers — every Tasmota install publishes tele/+/SENSOR — so with two brokers a message on A was ingested by a source bound to B. HA enforced the binding on both workers; MQTT enforced it only at subscribe time. Pass the endpoint id through: MQTTnet's event args carry the topic but not the delivering connection, so CreateClient captures the id in the handler closure. ResolveTopicsAsync drops its `|| EndpointId == null` clause to match, since an unbound source is no longer routed and subscribing its topic everywhere would only invite traffic nothing consumes. That last part would silently kill unbound sources that work today, so a data migration binds them to the single broker when exactly one exists — the case where old and new behaviour coincide. Two or more brokers is left alone: the old behaviour was already ambiguous and a guess could route a meter's data to the wrong broker. HA sources are excluded; they have always required an endpoint, so binding them would activate ingestion never previously running. Claude-Session: https://claude.ai/code/session_01V6joyergfvVLFEizH1hJLd
64 lines
2.7 KiB
C#
64 lines
2.7 KiB
C#
using MeterVault.Core.Domain;
|
|
using MeterVault.Infrastructure.Persistence;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.Logging;
|
|
|
|
namespace MeterVault.Infrastructure.Ingestion;
|
|
|
|
/// <summary>
|
|
/// Routes an incoming MQTT message to every enabled MQTT/Tasmota source that is bound to the
|
|
/// delivering broker <em>and</em> whose topic filter covers it, extracts the value (and payload
|
|
/// timestamp), and ingests it (SDD §6.1). Decoupled from the broker client so it can be exercised
|
|
/// directly against the database in tests.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// The endpoint predicate is load-bearing, not defensive: topic filters routinely overlap between
|
|
/// brokers (every Tasmota install publishes <c>tele/+/SENSOR</c>), so matching on topic alone would
|
|
/// let a message from one broker be ingested by a source bound to another.
|
|
/// </remarks>
|
|
public sealed class MqttMessageRouter(
|
|
MeterVaultDbContext db, IngestionService ingestion, ILogger<MqttMessageRouter> logger)
|
|
{
|
|
private readonly MeterVaultDbContext _db = db;
|
|
private readonly IngestionService _ingestion = ingestion;
|
|
private readonly ILogger<MqttMessageRouter> _logger = logger;
|
|
|
|
public async Task<int> RouteAsync(
|
|
int endpointId, string topic, string payload, CancellationToken cancellationToken = default)
|
|
{
|
|
var sources = await _db.MeterSources
|
|
.Where(s => s.IsEnabled
|
|
&& (s.SourceType == SourceType.Mqtt || s.SourceType == SourceType.Tasmota)
|
|
&& s.EndpointId == endpointId)
|
|
.ToListAsync(cancellationToken).ConfigureAwait(false);
|
|
|
|
var routed = 0;
|
|
foreach (var source in sources)
|
|
{
|
|
var config = SourceConfig.Parse(source.Config);
|
|
if (!MqttTopicMatcher.Matches(config.Topic, topic))
|
|
{
|
|
continue;
|
|
}
|
|
|
|
if (!PayloadExtractor.TryExtractValue(payload, config.Path, out var value))
|
|
{
|
|
_logger.LogDebug("Source {SourceId}: no value at path '{Path}' in topic {Topic}",
|
|
source.Id, config.Path, topic);
|
|
continue;
|
|
}
|
|
|
|
// Prefer the payload's own time; Tasmota SENSOR messages carry "Time"; else receive time.
|
|
var timePath = config.TimePath ?? (source.SourceType == SourceType.Tasmota ? "Time" : null);
|
|
var time = PayloadExtractor.TryExtractTime(payload, timePath, out var payloadTime)
|
|
? payloadTime
|
|
: DateTimeOffset.UtcNow;
|
|
|
|
await _ingestion.IngestAsync(source.Id, time, value, cancellationToken).ConfigureAwait(false);
|
|
routed++;
|
|
}
|
|
|
|
return routed;
|
|
}
|
|
}
|