Ingestion: route MQTT messages only to sources bound to the delivering broker

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

tests/Integration.Tests/Ingestion/IngestionServiceTests.cs

@@ -100,10 +100,13 @@ public sealed class IngestionServiceTests(TimescaleFixture fx)
     public async Task Mqtt_router_ingests_a_tasmota_payload()
     {
         await using var db = fx.CreateContext();
-        var (meterId, _) = await SetupAsync(db, MeterMode.CumulativeCounter, topic: "tele/plug7/SENSOR");
+        var brokerId = await CreateBrokerAsync(db);
+        var (meterId, _) = await SetupAsync(
+            db, MeterMode.CumulativeCounter, topic: "tele/plug7/SENSOR", endpointId: brokerId);
         var router = new MqttMessageRouter(db, new IngestionService(db), NullLogger<MqttMessageRouter>.Instance);
 
         var routed = await router.RouteAsync(
+            brokerId,
             "tele/plug7/SENSOR",
             """{"Time":"2024-03-01T10:00:00","ENERGY":{"Total":8421.0}}""");
 
@@ -115,9 +118,39 @@ public sealed class IngestionServiceTests(TimescaleFixture fx)
         await CleanupAsync(db, meterId);
     }
 
+    [Fact]
+    public async Task Mqtt_router_ignores_a_source_bound_to_another_broker()
+    {
+        await using var db = fx.CreateContext();
+        var brokerA = await CreateBrokerAsync(db);
+        var brokerB = await CreateBrokerAsync(db);
+
+        // Topic filter that both brokers' traffic would match — the binding is the only thing
+        // separating them.
+        var (meterId, _) = await SetupAsync(
+            db, MeterMode.CumulativeCounter, topic: "tele/+/SENSOR", endpointId: brokerB);
+        var router = new MqttMessageRouter(db, new IngestionService(db), NullLogger<MqttMessageRouter>.Instance);
+
+        var routed = await router.RouteAsync(
+            brokerA,
+            "tele/plug7/SENSOR",
+            """{"Time":"2024-03-01T10:00:00","ENERGY":{"Total":8421.0}}""");
+
+        Assert.Equal(0, routed);
+        Assert.False(await db.Readings.AnyAsync(r => r.MeterId == meterId));
+
+        // Same message on the broker it is actually bound to does land.
+        Assert.Equal(1, await router.RouteAsync(
+            brokerB,
+            "tele/plug7/SENSOR",
+            """{"Time":"2024-03-01T10:00:00","ENERGY":{"Total":8421.0}}"""));
+
+        await CleanupAsync(db, meterId);
+    }
+
     private static async Task<(int MeterId, int SourceId)> SetupAsync(
         MeterVaultDbContext db, MeterMode mode, double scale = 1, double offset = 0,
-        string topic = "tele/x/SENSOR", string? path = "ENERGY.Total")
+        string topic = "tele/x/SENSOR", string? path = "ENERGY.Total", int? endpointId = null)
     {
         await DatabaseSeeder.SeedAsync(db);
         var type = await db.EnergyTypes.FirstAsync(t => t.Key == "electricity");
@@ -136,6 +169,7 @@ public sealed class IngestionServiceTests(TimescaleFixture fx)
         {
             MeterId = meter.Id,
             SourceType = SourceType.Tasmota,
+            EndpointId = endpointId ?? await CreateBrokerAsync(db),
             ValueKind = SourceValueKind.Register,
             Scale = scale,
             Offset = offset,
@@ -147,10 +181,24 @@ public sealed class IngestionServiceTests(TimescaleFixture fx)
         return (meter.Id, source.Id);
     }
 
+    private static async Task<int> CreateBrokerAsync(MeterVaultDbContext db)
+    {
+        var endpoint = new IngestionEndpoint
+        {
+            Type = EndpointType.MqttBroker,
+            Name = $"broker-{Guid.NewGuid():N}",
+            Config = """{"host":"localhost","port":1883}""",
+        };
+        db.IngestionEndpoints.Add(endpoint);
+        await db.SaveChangesAsync();
+        return endpoint.Id;
+    }
+
     private static async Task CleanupAsync(MeterVaultDbContext db, int meterId)
     {
         await db.Readings.Where(r => r.MeterId == meterId).ExecuteDeleteAsync();
         await db.MeterEvents.Where(e => e.MeterId == meterId).ExecuteDeleteAsync();
         await db.Meters.Where(m => m.Id == meterId).ExecuteDeleteAsync();
+        await db.IngestionEndpoints.Where(e => e.Name.StartsWith("broker-")).ExecuteDeleteAsync();
     }
 }