diff --git a/src/Infrastructure/Ingestion/MqttIngestionWorker.cs b/src/Infrastructure/Ingestion/MqttIngestionWorker.cs index f9702e6..8ac3243 100644 --- a/src/Infrastructure/Ingestion/MqttIngestionWorker.cs +++ b/src/Infrastructure/Ingestion/MqttIngestionWorker.cs @@ -84,7 +84,7 @@ public sealed class MqttIngestionWorker( foreach (var endpoint in endpoints) { - var client = _clients.GetOrAdd(endpoint.Id, _ => CreateClient()); + var client = _clients.GetOrAdd(endpoint.Id, id => CreateClient(id)); var topics = await ResolveTopicsAsync(db, endpoint, cancellationToken).ConfigureAwait(false); if (!client.IsConnected) @@ -114,10 +114,13 @@ public sealed class MqttIngestionWorker( } } - private IMqttClient CreateClient() + // One client per endpoint, with the endpoint id captured in the handler: MQTTnet's event args + // carry the topic but not which connection delivered it, and the router needs that to keep + // sources bound to one broker from ingesting another's traffic. + private IMqttClient CreateClient(int endpointId) { var client = _factory.CreateMqttClient(); - client.ApplicationMessageReceivedAsync += OnMessageAsync; + client.ApplicationMessageReceivedAsync += args => OnMessageAsync(endpointId, args); return client; } @@ -163,10 +166,12 @@ public sealed class MqttIngestionWorker( private static async Task> ResolveTopicsAsync( MeterVaultDbContext db, IngestionEndpoint endpoint, CancellationToken cancellationToken) { + // Bound sources only, matching MqttMessageRouter: an unbound source is not routed, so + // subscribing its topic on every broker would only invite traffic nothing consumes. var sourceConfigs = await db.MeterSources .Where(s => s.IsEnabled && (s.SourceType == SourceType.Mqtt || s.SourceType == SourceType.Tasmota) - && (s.EndpointId == endpoint.Id || s.EndpointId == null)) + && s.EndpointId == endpoint.Id) .Select(s => s.Config) .ToListAsync(cancellationToken).ConfigureAwait(false); @@ -188,7 +193,7 @@ public sealed class MqttIngestionWorker( return [.. topics]; } - private async Task OnMessageAsync(MqttApplicationMessageReceivedEventArgs args) + private async Task OnMessageAsync(int endpointId, MqttApplicationMessageReceivedEventArgs args) { var topic = args.ApplicationMessage.Topic; var payload = args.ApplicationMessage.ConvertPayloadToString() ?? string.Empty; @@ -197,7 +202,7 @@ public sealed class MqttIngestionWorker( { await using var scope = _scopeFactory.CreateAsyncScope(); var router = scope.ServiceProvider.GetRequiredService(); - await router.RouteAsync(topic, payload).ConfigureAwait(false); + await router.RouteAsync(endpointId, topic, payload).ConfigureAwait(false); } catch (Exception ex) { diff --git a/src/Infrastructure/Ingestion/MqttMessageRouter.cs b/src/Infrastructure/Ingestion/MqttMessageRouter.cs index 5eb6d62..1969f6d 100644 --- a/src/Infrastructure/Ingestion/MqttMessageRouter.cs +++ b/src/Infrastructure/Ingestion/MqttMessageRouter.cs @@ -6,10 +6,16 @@ using Microsoft.Extensions.Logging; namespace MeterVault.Infrastructure.Ingestion; /// -/// Routes an incoming MQTT message to every enabled MQTT/Tasmota source 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. +/// Routes an incoming MQTT message to every enabled MQTT/Tasmota source that is bound to the +/// delivering broker and 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. /// +/// +/// The endpoint predicate is load-bearing, not defensive: topic filters routinely overlap between +/// brokers (every Tasmota install publishes tele/+/SENSOR), so matching on topic alone would +/// let a message from one broker be ingested by a source bound to another. +/// public sealed class MqttMessageRouter( MeterVaultDbContext db, IngestionService ingestion, ILogger logger) { @@ -17,10 +23,13 @@ public sealed class MqttMessageRouter( private readonly IngestionService _ingestion = ingestion; private readonly ILogger _logger = logger; - public async Task RouteAsync(string topic, string payload, CancellationToken cancellationToken = default) + public async Task 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)) + .Where(s => s.IsEnabled + && (s.SourceType == SourceType.Mqtt || s.SourceType == SourceType.Tasmota) + && s.EndpointId == endpointId) .ToListAsync(cancellationToken).ConfigureAwait(false); var routed = 0; diff --git a/src/Infrastructure/Persistence/Migrations/20260718091623_BindUnboundMqttSourcesToSoleBroker.Designer.cs b/src/Infrastructure/Persistence/Migrations/20260718091623_BindUnboundMqttSourcesToSoleBroker.Designer.cs new file mode 100644 index 0000000..8861af9 --- /dev/null +++ b/src/Infrastructure/Persistence/Migrations/20260718091623_BindUnboundMqttSourcesToSoleBroker.Designer.cs @@ -0,0 +1,931 @@ +// +using System; +using MeterVault.Infrastructure.Persistence; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata; + +#nullable disable + +namespace MeterVault.Infrastructure.Persistence.Migrations +{ + [DbContext(typeof(MeterVaultDbContext))] + [Migration("20260718091623_BindUnboundMqttSourcesToSoleBroker")] + partial class BindUnboundMqttSourcesToSoleBroker + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.9") + .HasAnnotation("Relational:MaxIdentifierLength", 63); + + NpgsqlModelBuilderExtensions.HasPostgresExtension(modelBuilder, "timescaledb"); + NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder); + + modelBuilder.Entity("MeterVault.Core.Domain.AppSetting", b => + { + b.Property("Key") + .HasMaxLength(128) + .HasColumnType("character varying(128)") + .HasColumnName("key"); + + b.Property("Value") + .IsRequired() + .HasColumnType("jsonb") + .HasColumnName("value"); + + b.HasKey("Key") + .HasName("pk_app_setting"); + + b.ToTable("app_setting", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Consumption", b => + { + b.Property("MeterId") + .HasColumnType("integer") + .HasColumnName("meter_id"); + + b.Property("Time") + .HasColumnType("timestamp with time zone") + .HasColumnName("time"); + + b.Property("Kind") + .HasColumnType("smallint") + .HasColumnName("kind"); + + b.Property("Amount") + .HasColumnType("double precision") + .HasColumnName("amount"); + + b.Property("ImportBatchId") + .HasColumnType("integer") + .HasColumnName("import_batch_id"); + + b.Property("Quality") + .HasColumnType("smallint") + .HasColumnName("quality"); + + b.HasKey("MeterId", "Time", "Kind") + .HasName("pk_consumption"); + + b.HasIndex("ImportBatchId") + .HasDatabaseName("ix_consumption_import_batch_id"); + + b.ToTable("consumption", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.CostCategory", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("ColorHex") + .HasColumnType("text") + .HasColumnName("color_hex"); + + b.Property("Name") + .IsRequired() + .HasMaxLength(128) + .HasColumnType("character varying(128)") + .HasColumnName("name"); + + b.Property("Sort") + .HasColumnType("integer") + .HasColumnName("sort"); + + b.HasKey("Id") + .HasName("pk_cost_category"); + + b.ToTable("cost_category", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.CostCategoryMember", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("CategoryId") + .HasColumnType("integer") + .HasColumnName("category_id"); + + b.Property("EnergyTypeId") + .HasColumnType("smallint") + .HasColumnName("energy_type_id"); + + b.Property("MeterId") + .HasColumnType("integer") + .HasColumnName("meter_id"); + + b.HasKey("Id") + .HasName("pk_cost_category_member"); + + b.HasIndex("CategoryId") + .HasDatabaseName("ix_cost_category_member_category_id"); + + b.HasIndex("EnergyTypeId") + .HasDatabaseName("ix_cost_category_member_energy_type_id"); + + b.HasIndex("MeterId") + .HasDatabaseName("ix_cost_category_member_meter_id"); + + b.ToTable("cost_category_member", null, t => + { + t.HasCheckConstraint("ck_cost_category_member_target", "meter_id IS NOT NULL OR energy_type_id IS NOT NULL"); + }); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.EnergyType", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("smallint") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("BaseUnit") + .IsRequired() + .HasMaxLength(16) + .HasColumnType("character varying(16)") + .HasColumnName("base_unit"); + + b.Property("ColorHex") + .HasColumnType("text") + .HasColumnName("color_hex"); + + b.Property("CreatedAt") + .ValueGeneratedOnAdd() + .HasColumnType("timestamp with time zone") + .HasColumnName("created_at") + .HasDefaultValueSql("now()"); + + b.Property("DefaultMode") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("character varying(32)") + .HasColumnName("default_mode"); + + b.Property("DisplayName") + .IsRequired() + .HasMaxLength(128) + .HasColumnType("character varying(128)") + .HasColumnName("display_name"); + + b.Property("Icon") + .HasColumnType("text") + .HasColumnName("icon"); + + b.Property("Key") + .IsRequired() + .HasMaxLength(64) + .HasColumnType("character varying(64)") + .HasColumnName("key"); + + b.HasKey("Id") + .HasName("pk_energy_type"); + + b.HasIndex("Key") + .IsUnique() + .HasDatabaseName("ix_energy_type_key"); + + b.ToTable("energy_type", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.ImportBatch", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("CreatedAt") + .ValueGeneratedOnAdd() + .HasColumnType("timestamp with time zone") + .HasColumnName("created_at") + .HasDefaultValueSql("now()"); + + b.Property("Mapping") + .HasColumnType("jsonb") + .HasColumnName("mapping"); + + b.Property("RevertedAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("reverted_at"); + + b.Property("RowCount") + .HasColumnType("integer") + .HasColumnName("row_count"); + + b.Property("SourceName") + .HasColumnType("text") + .HasColumnName("source_name"); + + b.HasKey("Id") + .HasName("pk_import_batch"); + + b.ToTable("import_batch", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.IngestionEndpoint", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("Config") + .IsRequired() + .ValueGeneratedOnAdd() + .HasColumnType("jsonb") + .HasColumnName("config") + .HasDefaultValueSql("'{}'::jsonb"); + + b.Property("IsEnabled") + .HasColumnType("boolean") + .HasColumnName("is_enabled"); + + b.Property("LastSeenAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("last_seen_at"); + + b.Property("LastStatus") + .HasColumnType("text") + .HasColumnName("last_status"); + + b.Property("Name") + .IsRequired() + .HasMaxLength(128) + .HasColumnType("character varying(128)") + .HasColumnName("name"); + + b.Property("Type") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("character varying(32)") + .HasColumnName("type"); + + b.HasKey("Id") + .HasName("pk_ingestion_endpoint"); + + b.ToTable("ingestion_endpoint", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.ManualCost", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("Amount") + .HasColumnType("double precision") + .HasColumnName("amount"); + + b.Property("CategoryId") + .HasColumnType("integer") + .HasColumnName("category_id"); + + b.Property("Currency") + .IsRequired() + .HasMaxLength(8) + .HasColumnType("character varying(8)") + .HasColumnName("currency"); + + b.Property("ImportBatchId") + .HasColumnType("integer") + .HasColumnName("import_batch_id"); + + b.Property("MeterId") + .HasColumnType("integer") + .HasColumnName("meter_id"); + + b.Property("Notes") + .HasColumnType("text") + .HasColumnName("notes"); + + b.Property("PeriodEnd") + .HasColumnType("date") + .HasColumnName("period_end"); + + b.Property("PeriodStart") + .HasColumnType("date") + .HasColumnName("period_start"); + + b.HasKey("Id") + .HasName("pk_manual_cost"); + + b.HasIndex("CategoryId") + .HasDatabaseName("ix_manual_cost_category_id"); + + b.HasIndex("ImportBatchId") + .HasDatabaseName("ix_manual_cost_import_batch_id"); + + b.HasIndex("MeterId") + .HasDatabaseName("ix_manual_cost_meter_id"); + + b.ToTable("manual_cost", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Meter", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("CreatedAt") + .ValueGeneratedOnAdd() + .HasColumnType("timestamp with time zone") + .HasColumnName("created_at") + .HasDefaultValueSql("now()"); + + b.Property("EnergyTypeId") + .HasColumnType("smallint") + .HasColumnName("energy_type_id"); + + b.Property("InitialBaseline") + .HasColumnType("double precision") + .HasColumnName("initial_baseline"); + + b.Property("InstalledAt") + .HasColumnType("date") + .HasColumnName("installed_at"); + + b.Property("IsActive") + .ValueGeneratedOnAdd() + .HasColumnType("boolean") + .HasDefaultValue(true) + .HasColumnName("is_active"); + + b.Property("Location") + .HasColumnType("text") + .HasColumnName("location"); + + b.Property("Manufacturer") + .HasColumnType("text") + .HasColumnName("manufacturer"); + + b.Property("Meta") + .IsRequired() + .ValueGeneratedOnAdd() + .HasColumnType("jsonb") + .HasColumnName("meta") + .HasDefaultValueSql("'{}'::jsonb"); + + b.Property("Mode") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("character varying(32)") + .HasColumnName("mode"); + + b.Property("Model") + .HasColumnType("text") + .HasColumnName("model"); + + b.Property("Name") + .IsRequired() + .HasColumnType("text") + .HasColumnName("name"); + + b.Property("RetiredAt") + .HasColumnType("date") + .HasColumnName("retired_at"); + + b.Property("SerialNumber") + .HasColumnType("text") + .HasColumnName("serial_number"); + + b.Property("Unit") + .IsRequired() + .HasColumnType("text") + .HasColumnName("unit"); + + b.Property("UpdatedAt") + .ValueGeneratedOnAdd() + .HasColumnType("timestamp with time zone") + .HasColumnName("updated_at") + .HasDefaultValueSql("now()"); + + b.HasKey("Id") + .HasName("pk_meter"); + + b.HasIndex("EnergyTypeId", "IsActive") + .HasDatabaseName("ix_meter_energy_type_id_is_active"); + + b.ToTable("meter", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.MeterEvent", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("Amount") + .HasColumnType("double precision") + .HasColumnName("amount"); + + b.Property("EventType") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("character varying(32)") + .HasColumnName("event_type"); + + b.Property("ImportBatchId") + .HasColumnType("integer") + .HasColumnName("import_batch_id"); + + b.Property("Meta") + .IsRequired() + .ValueGeneratedOnAdd() + .HasColumnType("jsonb") + .HasColumnName("meta") + .HasDefaultValueSql("'{}'::jsonb"); + + b.Property("MeterId") + .HasColumnType("integer") + .HasColumnName("meter_id"); + + b.Property("NewValue") + .HasColumnType("double precision") + .HasColumnName("new_value"); + + b.Property("Notes") + .HasColumnType("text") + .HasColumnName("notes"); + + b.Property("PrevValue") + .HasColumnType("double precision") + .HasColumnName("prev_value"); + + b.Property("Time") + .HasColumnType("timestamp with time zone") + .HasColumnName("time"); + + b.Property("Unit") + .HasColumnType("text") + .HasColumnName("unit"); + + b.HasKey("Id") + .HasName("pk_meter_event"); + + b.HasIndex("ImportBatchId") + .HasDatabaseName("ix_meter_event_import_batch_id"); + + b.HasIndex("MeterId", "Time") + .HasDatabaseName("ix_meter_event_meter_id_time"); + + b.ToTable("meter_event", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.MeterLink", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("FromMeterId") + .HasColumnType("integer") + .HasColumnName("from_meter_id"); + + b.Property("ToMeterId") + .HasColumnType("integer") + .HasColumnName("to_meter_id"); + + b.HasKey("Id") + .HasName("pk_meter_link"); + + b.HasIndex("ToMeterId") + .HasDatabaseName("ix_meter_link_to_meter_id"); + + b.HasIndex("FromMeterId", "ToMeterId") + .IsUnique() + .HasDatabaseName("ix_meter_link_from_meter_id_to_meter_id"); + + b.ToTable("meter_link", null, t => + { + t.HasCheckConstraint("ck_meter_link_distinct", "from_meter_id <> to_meter_id"); + }); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.MeterSource", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("Config") + .IsRequired() + .ValueGeneratedOnAdd() + .HasColumnType("jsonb") + .HasColumnName("config") + .HasDefaultValueSql("'{}'::jsonb"); + + b.Property("EndpointId") + .HasColumnType("integer") + .HasColumnName("endpoint_id"); + + b.Property("IsEnabled") + .HasColumnType("boolean") + .HasColumnName("is_enabled"); + + b.Property("LastSeenAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("last_seen_at"); + + b.Property("LastStatus") + .HasColumnType("text") + .HasColumnName("last_status"); + + b.Property("LastValue") + .HasColumnType("double precision") + .HasColumnName("last_value"); + + b.Property("MeterId") + .HasColumnType("integer") + .HasColumnName("meter_id"); + + b.Property("Offset") + .HasColumnType("double precision") + .HasColumnName("offset"); + + b.Property("Priority") + .HasColumnType("integer") + .HasColumnName("priority"); + + b.Property("Scale") + .ValueGeneratedOnAdd() + .HasColumnType("double precision") + .HasDefaultValue(1.0) + .HasColumnName("scale"); + + b.Property("SourceType") + .IsRequired() + .HasMaxLength(32) + .HasColumnType("character varying(32)") + .HasColumnName("source_type"); + + b.Property("ValueKind") + .IsRequired() + .HasMaxLength(16) + .HasColumnType("character varying(16)") + .HasColumnName("value_kind"); + + b.HasKey("Id") + .HasName("pk_meter_source"); + + b.HasIndex("EndpointId") + .HasDatabaseName("ix_meter_source_endpoint_id"); + + b.HasIndex("MeterId") + .HasDatabaseName("ix_meter_source_meter_id"); + + b.ToTable("meter_source", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Reading", b => + { + b.Property("MeterId") + .HasColumnType("integer") + .HasColumnName("meter_id"); + + b.Property("Time") + .HasColumnType("timestamp with time zone") + .HasColumnName("time"); + + b.Property("Flags") + .HasColumnType("integer") + .HasColumnName("flags"); + + b.Property("ImportBatchId") + .HasColumnType("integer") + .HasColumnName("import_batch_id"); + + b.Property("Quality") + .HasColumnType("smallint") + .HasColumnName("quality"); + + b.Property("SourceId") + .HasColumnType("integer") + .HasColumnName("source_id"); + + b.Property("Value") + .HasColumnType("double precision") + .HasColumnName("value"); + + b.HasKey("MeterId", "Time") + .HasName("pk_reading"); + + b.HasIndex("ImportBatchId") + .HasDatabaseName("ix_reading_import_batch_id"); + + b.ToTable("reading", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Tank", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("CachedAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("cached_at"); + + b.Property("CachedBalance") + .HasColumnType("double precision") + .HasColumnName("cached_balance"); + + b.Property("Calibration") + .HasColumnType("jsonb") + .HasColumnName("calibration"); + + b.Property("Capacity") + .HasColumnType("double precision") + .HasColumnName("capacity"); + + b.Property("FixedRate") + .HasColumnType("double precision") + .HasColumnName("fixed_rate"); + + b.Property("LowThreshold") + .HasColumnType("double precision") + .HasColumnName("low_threshold"); + + b.Property("MeterId") + .HasColumnType("integer") + .HasColumnName("meter_id"); + + b.Property("RateMode") + .IsRequired() + .HasMaxLength(16) + .HasColumnType("character varying(16)") + .HasColumnName("rate_mode"); + + b.Property("ReorderThreshold") + .HasColumnType("double precision") + .HasColumnName("reorder_threshold"); + + b.Property("Unit") + .IsRequired() + .HasMaxLength(16) + .HasColumnType("character varying(16)") + .HasColumnName("unit"); + + b.HasKey("Id") + .HasName("pk_tank"); + + b.HasIndex("MeterId") + .IsUnique() + .HasDatabaseName("ix_tank_meter_id"); + + b.ToTable("tank", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Tariff", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("Component") + .IsRequired() + .HasMaxLength(16) + .HasColumnType("character varying(16)") + .HasColumnName("component"); + + b.Property("Currency") + .IsRequired() + .HasMaxLength(8) + .HasColumnType("character varying(8)") + .HasColumnName("currency"); + + b.Property("Notes") + .HasColumnType("text") + .HasColumnName("notes"); + + b.Property("ScopeId") + .HasColumnType("integer") + .HasColumnName("scope_id"); + + b.Property("ScopeType") + .IsRequired() + .HasMaxLength(16) + .HasColumnType("character varying(16)") + .HasColumnName("scope_type"); + + b.Property("Unit") + .IsRequired() + .HasMaxLength(16) + .HasColumnType("character varying(16)") + .HasColumnName("unit"); + + b.Property("ValidFrom") + .HasColumnType("date") + .HasColumnName("valid_from"); + + b.Property("ValidTo") + .HasColumnType("date") + .HasColumnName("valid_to"); + + b.Property("Value") + .HasColumnType("double precision") + .HasColumnName("value"); + + b.HasKey("Id") + .HasName("pk_tariff"); + + b.HasIndex("ScopeType", "ScopeId", "Component", "ValidFrom") + .HasDatabaseName("ix_tariff_scope_type_scope_id_component_valid_from"); + + b.ToTable("tariff", (string)null); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Consumption", b => + { + b.HasOne("MeterVault.Core.Domain.Meter", null) + .WithMany() + .HasForeignKey("MeterId") + .OnDelete(DeleteBehavior.Restrict) + .IsRequired() + .HasConstraintName("fk_consumption_meter_meter_id"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.CostCategoryMember", b => + { + b.HasOne("MeterVault.Core.Domain.CostCategory", "Category") + .WithMany("Members") + .HasForeignKey("CategoryId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired() + .HasConstraintName("fk_cost_category_member_cost_category_category_id"); + + b.HasOne("MeterVault.Core.Domain.EnergyType", null) + .WithMany() + .HasForeignKey("EnergyTypeId") + .OnDelete(DeleteBehavior.Cascade) + .HasConstraintName("fk_cost_category_member_energy_type_energy_type_id"); + + b.HasOne("MeterVault.Core.Domain.Meter", null) + .WithMany() + .HasForeignKey("MeterId") + .OnDelete(DeleteBehavior.Cascade) + .HasConstraintName("fk_cost_category_member_meter_meter_id"); + + b.Navigation("Category"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.ManualCost", b => + { + b.HasOne("MeterVault.Core.Domain.CostCategory", null) + .WithMany() + .HasForeignKey("CategoryId") + .OnDelete(DeleteBehavior.SetNull) + .HasConstraintName("fk_manual_cost_cost_category_category_id"); + + b.HasOne("MeterVault.Core.Domain.Meter", null) + .WithMany() + .HasForeignKey("MeterId") + .OnDelete(DeleteBehavior.SetNull) + .HasConstraintName("fk_manual_cost_meter_meter_id"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Meter", b => + { + b.HasOne("MeterVault.Core.Domain.EnergyType", "EnergyType") + .WithMany("Meters") + .HasForeignKey("EnergyTypeId") + .OnDelete(DeleteBehavior.Restrict) + .IsRequired() + .HasConstraintName("fk_meter_energy_type_energy_type_id"); + + b.Navigation("EnergyType"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.MeterEvent", b => + { + b.HasOne("MeterVault.Core.Domain.Meter", null) + .WithMany() + .HasForeignKey("MeterId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired() + .HasConstraintName("fk_meter_event_meter_meter_id"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.MeterLink", b => + { + b.HasOne("MeterVault.Core.Domain.Meter", "FromMeter") + .WithMany() + .HasForeignKey("FromMeterId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired() + .HasConstraintName("fk_meter_link_meter_from_meter_id"); + + b.HasOne("MeterVault.Core.Domain.Meter", "ToMeter") + .WithMany() + .HasForeignKey("ToMeterId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired() + .HasConstraintName("fk_meter_link_meter_to_meter_id"); + + b.Navigation("FromMeter"); + + b.Navigation("ToMeter"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.MeterSource", b => + { + b.HasOne("MeterVault.Core.Domain.IngestionEndpoint", "Endpoint") + .WithMany() + .HasForeignKey("EndpointId") + .OnDelete(DeleteBehavior.SetNull) + .HasConstraintName("fk_meter_source_ingestion_endpoints_endpoint_id"); + + b.HasOne("MeterVault.Core.Domain.Meter", "Meter") + .WithMany("Sources") + .HasForeignKey("MeterId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired() + .HasConstraintName("fk_meter_source_meter_meter_id"); + + b.Navigation("Endpoint"); + + b.Navigation("Meter"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Reading", b => + { + b.HasOne("MeterVault.Core.Domain.Meter", null) + .WithMany() + .HasForeignKey("MeterId") + .OnDelete(DeleteBehavior.Restrict) + .IsRequired() + .HasConstraintName("fk_reading_meter_meter_id"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Tank", b => + { + b.HasOne("MeterVault.Core.Domain.Meter", "Meter") + .WithMany() + .HasForeignKey("MeterId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired() + .HasConstraintName("fk_tank_meter_meter_id"); + + b.Navigation("Meter"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.CostCategory", b => + { + b.Navigation("Members"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.EnergyType", b => + { + b.Navigation("Meters"); + }); + + modelBuilder.Entity("MeterVault.Core.Domain.Meter", b => + { + b.Navigation("Sources"); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/Infrastructure/Persistence/Migrations/20260718091623_BindUnboundMqttSourcesToSoleBroker.cs b/src/Infrastructure/Persistence/Migrations/20260718091623_BindUnboundMqttSourcesToSoleBroker.cs new file mode 100644 index 0000000..dd73e62 --- /dev/null +++ b/src/Infrastructure/Persistence/Migrations/20260718091623_BindUnboundMqttSourcesToSoleBroker.cs @@ -0,0 +1,47 @@ +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace MeterVault.Infrastructure.Persistence.Migrations +{ + /// + public partial class BindUnboundMqttSourcesToSoleBroker : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + // MQTT routing now honours meter_source.endpoint_id (SDD §6.1): a source is served only + // by the broker it is bound to. Previously an unbound source was subscribed on every + // broker and matched on topic alone, so unbound sources that work today would silently + // go quiet after this deploy. + // + // Backfill them onto the single broker only when exactly one exists — then the old + // "any broker" behaviour and the new "its broker" behaviour are the same thing, so the + // rewrite is provably lossless. With zero brokers there is nothing to bind to; with two + // or more the old behaviour was already ambiguous and a guess could route a meter's + // data to the wrong broker, so those are left for the operator to resolve in the UI. + // + // Enums persist as their C# names (HasConversion), hence 'Mqtt'/'MqttBroker'. + // HomeAssistant sources are deliberately excluded: the HA workers have always required + // endpoint_id, so an unbound HA source is already inert and binding it here would + // activate ingestion the operator never had running. + migrationBuilder.Sql(""" + UPDATE meter_source AS s + SET endpoint_id = sole.id + FROM (SELECT id FROM ingestion_endpoint WHERE type = 'MqttBroker') AS sole + WHERE s.endpoint_id IS NULL + AND s.source_type IN ('Mqtt', 'Tasmota') + AND (SELECT count(*) FROM ingestion_endpoint WHERE type = 'MqttBroker') = 1; + """); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + // Intentionally empty. The rows this bound are indistinguishable from ones the operator + // bound by hand, so clearing endpoint_id on the way down would discard real + // configuration. Leaving the binding in place is harmless under the old routing, which + // ignored endpoint_id entirely. + } + } +} diff --git a/tests/Integration.Tests/Ingestion/IngestionServiceTests.cs b/tests/Integration.Tests/Ingestion/IngestionServiceTests.cs index 6714051..f6a8df5 100644 --- a/tests/Integration.Tests/Ingestion/IngestionServiceTests.cs +++ b/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.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.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 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(); } }