diff --git a/README.md b/README.md index 5ceda89..ba063d9 100644 --- a/README.md +++ b/README.md @@ -41,9 +41,13 @@ Configuration is via environment variables (`Section__Key` double-underscore map | `ConnectionStrings__Default` | PostgreSQL/Timescale connection string | | `MeterVault__TimeZone` | Local timezone for buckets/display (default `Europe/Berlin`) | | `MeterVault__ApiKeys__0` | An API key accepted on the `X-Api-Key` header | +| `MeterVault__AllowAnonymousApi` | `true` to open the REST API without a key (trusted LAN only) | | `MeterVault__ReverseProxyTrust` | `true` to honour `X-Forwarded-User` behind an auth proxy | | `MeterVault__EnableLiveIngestion` | `false` to disable the MQTT/HA workers | +The REST API is **closed by default**: with no `ApiKeys` configured and `AllowAnonymousApi` off, it +returns 401. Set at least one API key (or open it explicitly for a trusted network). + Secrets (broker/HA tokens) are **never** stored in the database — endpoint configs hold the *name* of an environment variable, resolved at runtime. diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index 21d8c96..64bc03a 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -33,6 +33,9 @@ services: MeterVault__TimeZone: ${METERVAULT_TIMEZONE:-Europe/Berlin} MeterVault__Currency: ${METERVAULT_CURRENCY:-EUR} MeterVault__Locale: ${METERVAULT_LOCALE:-en} + # REST API is closed by default. Set a key to enable it (or AllowAnonymousApi on a trusted LAN): + # MeterVault__ApiKeys__0: your-secret-key + # MeterVault__AllowAnonymousApi: "true" ports: - "${METERVAULT_PORT:-8080}:8080" healthcheck: diff --git a/src/App/Api/ApiEndpoints.cs b/src/App/Api/ApiEndpoints.cs index 137b2fa..cccd478 100644 --- a/src/App/Api/ApiEndpoints.cs +++ b/src/App/Api/ApiEndpoints.cs @@ -17,18 +17,25 @@ public sealed record EventPush(int MeterId, DateTimeOffset Time, MeterEventType public sealed record TariffPush(TariffScope ScopeType, int? ScopeId, TariffComponent Component, double Value, string Unit, DateOnly ValidFrom, DateOnly? ValidTo); -public sealed record IngestResult(int Written, int Updated, int Rejected); +public sealed record IngestResult(int Written, int Updated, int Rejected, int Ignored); /// Maps the versioned REST API. All endpoints require a valid API key (SDD §9). public static class ApiEndpoints { + private const int MaxReadingsPerRequest = 5000; + public static IEndpointRouteBuilder MapMeterVaultApi(this IEndpointRouteBuilder app) { var api = app.MapGroup("/api/v1").AddEndpointFilter().WithTags("MeterVault"); api.MapPost("/readings", async (ReadingPush[] readings, IngestionService ingestion, CancellationToken ct) => { - int written = 0, updated = 0, rejected = 0; + if (readings.Length > MaxReadingsPerRequest) + { + return Results.BadRequest(new { error = $"At most {MaxReadingsPerRequest} readings per request." }); + } + + int written = 0, updated = 0, rejected = 0, ignored = 0; foreach (var r in readings) { switch (await ingestion.IngestByMeterAsync(r.MeterId, r.Time, r.Value, ct)) @@ -36,11 +43,11 @@ public static class ApiEndpoints case IngestionOutcome.Written: written++; break; case IngestionOutcome.Updated: updated++; break; case IngestionOutcome.RejectedDecrease: rejected++; break; - default: break; + default: ignored++; break; // unknown meter } } - return Results.Ok(new IngestResult(written, updated, rejected)); + return Results.Ok(new IngestResult(written, updated, rejected, ignored)); }).WithSummary("Ingest one or more readings (idempotent). Lets Home Assistant push."); api.MapGet("/meters", async (MeterVaultDbContext db, CancellationToken ct) => @@ -70,6 +77,14 @@ public static class ApiEndpoints api.MapPost("/events", async (EventPush push, MeterVaultDbContext db, NormalizationService normalization, CancellationToken ct) => { + if (!await db.Meters.AnyAsync(m => m.Id == push.MeterId, ct)) + { + return Results.NotFound(new { error = $"Meter {push.MeterId} does not exist." }); + } + + // One transaction so the event insert + consumption recompute (which deletes then + // re-inserts) are atomic — a failure must not leave the meter with no consumption. + await using var tx = await db.Database.BeginTransactionAsync(ct); db.MeterEvents.Add(new MeterEvent { MeterId = push.MeterId, @@ -84,6 +99,7 @@ public static class ApiEndpoints await db.SaveChangesAsync(ct); await normalization.RecomputeMeterAsync(push.MeterId, null, ct); await db.SaveChangesAsync(ct); + await tx.CommitAsync(ct); return Results.Ok(); }).WithSummary("Record a delivery / swap / tank level / correction and recompute the meter."); diff --git a/src/App/Api/ApiKeyFilter.cs b/src/App/Api/ApiKeyFilter.cs index e91688f..eca0595 100644 --- a/src/App/Api/ApiKeyFilter.cs +++ b/src/App/Api/ApiKeyFilter.cs @@ -17,7 +17,11 @@ public sealed class ApiKeyFilter(IOptions options) : IEndpoin { if (_options.ApiKeys.Count == 0) { - return await next(context).ConfigureAwait(false); + // Fail closed: an unconfigured deployment must not expose export/import. Opt in explicitly. + return _options.AllowAnonymousApi + ? await next(context).ConfigureAwait(false) + : Results.Problem(statusCode: StatusCodes.Status401Unauthorized, + title: "API is closed: configure MeterVault:ApiKeys or set AllowAnonymousApi=true."); } var provided = context.HttpContext.Request.Headers[HeaderName].ToString(); diff --git a/src/App/Components/Pages/Admin/EnergyTypes.razor b/src/App/Components/Pages/Admin/EnergyTypes.razor index 594400a..a2995e2 100644 --- a/src/App/Components/Pages/Admin/EnergyTypes.razor +++ b/src/App/Components/Pages/Admin/EnergyTypes.razor @@ -1,6 +1,6 @@ @page "/admin/energy-types" @rendermode InteractiveServer -@inject MeterVault.Infrastructure.Persistence.MeterVaultDbContext Db +@inject Microsoft.EntityFrameworkCore.IDbContextFactory DbFactory @using Microsoft.EntityFrameworkCore MeterVault — Energy types @@ -32,6 +32,9 @@ else @code { private List? _types; - protected override async Task OnInitializedAsync() => - _types = await Db.EnergyTypes.AsNoTracking().OrderBy(t => t.Id).ToListAsync(); + protected override async Task OnInitializedAsync() + { + await using var db = await DbFactory.CreateDbContextAsync(); + _types = await db.EnergyTypes.AsNoTracking().OrderBy(t => t.Id).ToListAsync(); + } } diff --git a/src/App/Components/Pages/Admin/Tariffs.razor b/src/App/Components/Pages/Admin/Tariffs.razor index 0a5012b..bd40bd8 100644 --- a/src/App/Components/Pages/Admin/Tariffs.razor +++ b/src/App/Components/Pages/Admin/Tariffs.razor @@ -1,6 +1,6 @@ @page "/admin/tariffs" @rendermode InteractiveServer -@inject MeterVault.Infrastructure.Persistence.MeterVaultDbContext Db +@inject Microsoft.EntityFrameworkCore.IDbContextFactory DbFactory @using Microsoft.EntityFrameworkCore MeterVault — Tariffs @@ -40,8 +40,11 @@ else @code { private List? _tariffs; - protected override async Task OnInitializedAsync() => - _tariffs = await Db.Tariffs.AsNoTracking() + protected override async Task OnInitializedAsync() + { + await using var db = await DbFactory.CreateDbContextAsync(); + _tariffs = await db.Tariffs.AsNoTracking() .OrderBy(t => t.Component).ThenBy(t => t.ValidFrom) .ToListAsync(); + } } diff --git a/src/App/Components/Pages/Meters.razor b/src/App/Components/Pages/Meters.razor index f4a6932..d8bc99c 100644 --- a/src/App/Components/Pages/Meters.razor +++ b/src/App/Components/Pages/Meters.razor @@ -1,6 +1,6 @@ @page "/meters" @rendermode InteractiveServer -@inject MeterVault.Infrastructure.Persistence.MeterVaultDbContext Db +@inject Microsoft.EntityFrameworkCore.IDbContextFactory DbFactory @using Microsoft.EntityFrameworkCore MeterVault — Meters @@ -54,7 +54,8 @@ else protected override async Task OnInitializedAsync() { - _meters = await Db.Meters + await using var db = await DbFactory.CreateDbContextAsync(); + _meters = await db.Meters .AsNoTracking() .Include(m => m.EnergyType) .Include(m => m.Sources) diff --git a/src/App/Components/Pages/Trends.razor b/src/App/Components/Pages/Trends.razor index d617345..6c2d042 100644 --- a/src/App/Components/Pages/Trends.razor +++ b/src/App/Components/Pages/Trends.razor @@ -13,7 +13,7 @@ Last 24 months Last 48 months - Apply + Apply @if (_loading) @@ -38,10 +38,21 @@ private async Task LoadAsync() { + if (_loading) + { + return; // guard against overlapping loads + } + _loading = true; - var asOf = DateOnly.FromDateTime(DateTime.UtcNow); - var from = asOf.AddMonths(-_months); - _points = await Dash.GetMonthlyTrendAsync(new DateOnly(from.Year, from.Month, 1), asOf.AddMonths(1)); - _loading = false; + try + { + var asOf = DateOnly.FromDateTime(DateTime.UtcNow); + var from = asOf.AddMonths(-_months); + _points = await Dash.GetMonthlyTrendAsync(new DateOnly(from.Year, from.Month, 1), asOf.AddMonths(1)); + } + finally + { + _loading = false; + } } } diff --git a/src/App/Program.cs b/src/App/Program.cs index c2c3ec1..9b54fa1 100644 --- a/src/App/Program.cs +++ b/src/App/Program.cs @@ -39,6 +39,10 @@ try builder.Services.AddRazorComponents() .AddInteractiveServerComponents(); + // Serialize/accept enums as strings on the REST API (e.g. event Type "Delivery"). + builder.Services.ConfigureHttpJsonOptions(o => + o.SerializerOptions.Converters.Add(new System.Text.Json.Serialization.JsonStringEnumConverter())); + builder.Services.AddEndpointsApiExplorer(); builder.Services.AddSwaggerGen(c => c.SwaggerDoc("v1", new() { Title = "MeterVault API", Version = "v1" })); diff --git a/src/Core/Normalization/Normalizers/VirtualNormalizer.cs b/src/Core/Normalization/Normalizers/VirtualNormalizer.cs index 8659dd3..03fb37b 100644 --- a/src/Core/Normalization/Normalizers/VirtualNormalizer.cs +++ b/src/Core/Normalization/Normalizers/VirtualNormalizer.cs @@ -47,11 +47,17 @@ public sealed class VirtualNormalizer : IMeterNormalizer variables[$"m{meterId}"] = byMeter[meterId].GetValueOrDefault(time); } + var amount = evaluator.Evaluate(variables); + if (double.IsNaN(amount) || double.IsInfinity(amount)) + { + amount = 0; // e.g. division by zero in a bucket — don't poison downstream sums + } + yield return new Consumption { MeterId = context.Meter.MeterId, Time = time, - Amount = evaluator.Evaluate(variables), + Amount = amount, Kind = ConsumptionKind.Consumption, Quality = ReadingQuality.Estimated, }; diff --git a/src/Infrastructure/Backup/ExportService.cs b/src/Infrastructure/Backup/ExportService.cs index 0c89655..7d6a0bf 100644 --- a/src/Infrastructure/Backup/ExportService.cs +++ b/src/Infrastructure/Backup/ExportService.cs @@ -118,6 +118,7 @@ public sealed class ExportService(MeterVaultDbContext db) cost.Id = 0; cost.CategoryId = cost.CategoryId is { } cid ? categoryMap.GetValueOrDefault(cid, cid) : null; cost.MeterId = cost.MeterId is { } mid ? meterMap.GetValueOrDefault(mid, mid) : null; + cost.ImportBatchId = null; // the original import batch isn't part of the export _db.ManualCosts.Add(cost); } @@ -125,6 +126,7 @@ public sealed class ExportService(MeterVaultDbContext db) { meterEvent.Id = 0; meterEvent.MeterId = meterMap.GetValueOrDefault(meterEvent.MeterId, meterEvent.MeterId); + meterEvent.ImportBatchId = null; _db.MeterEvents.Add(meterEvent); } diff --git a/src/Infrastructure/Costing/CostService.cs b/src/Infrastructure/Costing/CostService.cs index 80f29cc..97d28b9 100644 --- a/src/Infrastructure/Costing/CostService.cs +++ b/src/Infrastructure/Costing/CostService.cs @@ -10,24 +10,28 @@ namespace MeterVault.Infrastructure.Costing; /// Computes cost by joining bucketed consumption with time-ranged tariffs (SDD §7.5). Consumption /// is aggregated in SQL (Dapper, local-timezone buckets); the active price for each bucket is /// resolved in C# via using the month's dominant price (SDD §14.3). -/// Categories roll up their member meters' costs plus meterless manual costs. +/// Categories roll up their member meters' costs plus meterless manual costs. Uses a DbContext +/// factory (short-lived context per operation) so it is safe from a Blazor circuit. /// -public sealed class CostService(MeterVaultDbContext db) +public sealed class CostService(IDbContextFactory contextFactory) { - private readonly MeterVaultDbContext _db = db; + private readonly IDbContextFactory _contextFactory = contextFactory; public async Task> GetMeterCostsAsync( int meterId, DateTimeOffset from, DateTimeOffset to, CostBucket bucket = CostBucket.Month, CancellationToken cancellationToken = default) { - var meter = await _db.Meters.FirstOrDefaultAsync(m => m.Id == meterId, cancellationToken).ConfigureAwait(false); + await using var db = await _contextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false); + + var meter = await db.Meters.AsNoTracking() + .FirstOrDefaultAsync(m => m.Id == meterId, cancellationToken).ConfigureAwait(false); if (meter is null) { return []; } - var tariffs = await _db.Tariffs.AsNoTracking().ToListAsync(cancellationToken).ConfigureAwait(false); - var series = await QueryConsumptionAsync(meterId, from, to, bucket, cancellationToken).ConfigureAwait(false); + var tariffs = await db.Tariffs.AsNoTracking().ToListAsync(cancellationToken).ConfigureAwait(false); + var series = await QueryConsumptionAsync(db, meterId, from, to, bucket, cancellationToken).ConfigureAwait(false); var results = new List(); foreach (var period in series.Keys.OrderBy(k => k)) @@ -39,7 +43,8 @@ public sealed class CostService(MeterVaultDbContext db) var basePrice = TariffResolver.ResolveValue(tariffs, TariffComponent.BasePrice, meterId, meter.EnergyTypeId, mid); var feedIn = TariffResolver.ResolveValue(tariffs, TariffComponent.FeedIn, meterId, meter.EnergyTypeId, mid); - var cost = (consumption * unitPrice) + basePrice - (generation * feedIn); + // Base price is a monthly standing charge: prorate it to the bucket length. + var cost = (consumption * unitPrice) + (basePrice * MonthsInBucket(period, bucket)) - (generation * feedIn); results.Add(new MeterCostBucket(period, consumption, generation, cost)); } @@ -49,7 +54,9 @@ public sealed class CostService(MeterVaultDbContext db) public async Task> GetCategoryCostsAsync( int categoryId, DateTimeOffset from, DateTimeOffset to, CancellationToken cancellationToken = default) { - var members = await _db.CostCategoryMembers + await using var db = await _contextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false); + + var members = await db.CostCategoryMembers.AsNoTracking() .Where(m => m.CategoryId == categoryId) .ToListAsync(cancellationToken).ConfigureAwait(false); @@ -63,7 +70,7 @@ public sealed class CostService(MeterVaultDbContext db) if (member.EnergyTypeId is { } energyTypeId) { - var byType = await _db.Meters.Where(m => m.EnergyTypeId == energyTypeId).Select(m => m.Id) + var byType = await db.Meters.AsNoTracking().Where(m => m.EnergyTypeId == energyTypeId).Select(m => m.Id) .ToListAsync(cancellationToken).ConfigureAwait(false); meterIds.UnionWith(byType); } @@ -78,9 +85,10 @@ public sealed class CostService(MeterVaultDbContext db) } } - var manualCosts = await _db.ManualCosts - .Where(c => c.CategoryId == categoryId && c.PeriodStart >= DateOnly.FromDateTime(from.UtcDateTime) - && c.PeriodStart < DateOnly.FromDateTime(to.UtcDateTime)) + var fromDate = DateOnly.FromDateTime(from.Date); + var toDate = DateOnly.FromDateTime(to.Date); + var manualCosts = await db.ManualCosts.AsNoTracking() + .Where(c => c.CategoryId == categoryId && c.PeriodStart >= fromDate && c.PeriodStart < toDate) .ToListAsync(cancellationToken).ConfigureAwait(false); foreach (var cost in manualCosts) { @@ -91,8 +99,9 @@ public sealed class CostService(MeterVaultDbContext db) return [.. totals.OrderBy(kv => kv.Key).Select(kv => new CategoryCostBucket(kv.Key, kv.Value))]; } - private async Task> QueryConsumptionAsync( - int meterId, DateTimeOffset from, DateTimeOffset to, CostBucket bucket, CancellationToken cancellationToken) + private static async Task> QueryConsumptionAsync( + MeterVaultDbContext db, int meterId, DateTimeOffset from, DateTimeOffset to, CostBucket bucket, + CancellationToken cancellationToken) { var interval = bucket switch { @@ -107,7 +116,7 @@ public sealed class CostService(MeterVaultDbContext db) "FROM consumption WHERE meter_id = @meterId AND \"time\" >= @from AND \"time\" < @to " + "GROUP BY period, kind"; - var connection = _db.Database.GetDbConnection(); + var connection = db.Database.GetDbConnection(); var command = new CommandDefinition(sql, new { meterId, from, to }, cancellationToken: cancellationToken); var rows = await connection.QueryAsync(command).ConfigureAwait(false); @@ -123,6 +132,13 @@ public sealed class CostService(MeterVaultDbContext db) return result; } + private static double MonthsInBucket(DateOnly period, CostBucket bucket) => bucket switch + { + CostBucket.Day => 1.0 / DateTime.DaysInMonth(period.Year, period.Month), + CostBucket.Year => 12.0, + _ => 1.0, + }; + private static DateOnly RepresentativeDate(DateOnly period, CostBucket bucket) => bucket switch { CostBucket.Day => period, diff --git a/src/Infrastructure/Dashboard/DashboardService.cs b/src/Infrastructure/Dashboard/DashboardService.cs index 9a0142f..4603956 100644 --- a/src/Infrastructure/Dashboard/DashboardService.cs +++ b/src/Infrastructure/Dashboard/DashboardService.cs @@ -7,36 +7,41 @@ namespace MeterVault.Infrastructure.Dashboard; /// /// Read model for the dashboard (SDD §8): overview KPIs with period-over-period deltas, the cost /// breakdown by category, the "what cost more / less" difference view, and monthly trends. Reads -/// only aggregated cost — never the raw hypertable. +/// only aggregated cost — never the raw hypertable. Uses a DbContext factory (short-lived context +/// per operation) so it is safe from a Blazor circuit. /// -public sealed class DashboardService(MeterVaultDbContext db, CostService costService) +public sealed class DashboardService(IDbContextFactory contextFactory, CostService costService) { - private readonly MeterVaultDbContext _db = db; + private readonly IDbContextFactory _contextFactory = contextFactory; private readonly CostService _costService = costService; public async Task GetSummaryAsync(DateOnly asOf, CancellationToken cancellationToken = default) { + await using var db = await _contextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false); + var monthStart = new DateOnly(asOf.Year, asOf.Month, 1); var prevMonthStart = monthStart.AddMonths(-1); var yearStart = new DateOnly(asOf.Year, 1, 1); var prevYearStart = yearStart.AddYears(-1); var month = new CostKpi( - await TotalCostAsync(monthStart, monthStart.AddMonths(1), cancellationToken).ConfigureAwait(false), - await TotalCostAsync(prevMonthStart, monthStart, cancellationToken).ConfigureAwait(false)); + await TotalCostAsync(db, monthStart, monthStart.AddMonths(1), cancellationToken).ConfigureAwait(false), + await TotalCostAsync(db, prevMonthStart, monthStart, cancellationToken).ConfigureAwait(false)); var year = new CostKpi( - await TotalCostAsync(yearStart, yearStart.AddYears(1), cancellationToken).ConfigureAwait(false), - await TotalCostAsync(prevYearStart, yearStart, cancellationToken).ConfigureAwait(false)); + await TotalCostAsync(db, yearStart, yearStart.AddYears(1), cancellationToken).ConfigureAwait(false), + await TotalCostAsync(db, prevYearStart, yearStart, cancellationToken).ConfigureAwait(false)); - var latest = await LatestMonthCostAsync(cancellationToken).ConfigureAwait(false); + var latest = await LatestMonthCostAsync(db, cancellationToken).ConfigureAwait(false); return new DashboardSummary(asOf, month, year, latest); } public async Task> GetCategoryBreakdownAsync( DateOnly from, DateOnly to, CancellationToken cancellationToken = default) { - var categories = await _db.CostCategories.OrderBy(c => c.Sort).ToListAsync(cancellationToken).ConfigureAwait(false); + await using var db = await _contextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false); + var categories = await db.CostCategories.AsNoTracking().OrderBy(c => c.Sort).ToListAsync(cancellationToken).ConfigureAwait(false); + var slices = new List(); foreach (var category in categories) { @@ -55,12 +60,13 @@ public sealed class DashboardService(MeterVaultDbContext db, CostService costSer public async Task> GetCategoryDifferenceAsync( DateOnly currentStart, DateOnly previousStart, DateOnly span, CancellationToken cancellationToken = default) { - // span length in months from currentStart. var months = ((span.Year - currentStart.Year) * 12) + span.Month - currentStart.Month; var currentEnd = currentStart.AddMonths(Math.Max(1, months)); var previousEnd = previousStart.AddMonths(Math.Max(1, months)); - var categories = await _db.CostCategories.OrderBy(c => c.Sort).ToListAsync(cancellationToken).ConfigureAwait(false); + await using var db = await _contextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false); + var categories = await db.CostCategories.AsNoTracking().OrderBy(c => c.Sort).ToListAsync(cancellationToken).ConfigureAwait(false); + var rows = new List(); foreach (var category in categories) { @@ -78,8 +84,9 @@ public sealed class DashboardService(MeterVaultDbContext db, CostService costSer public async Task> GetMonthlyTrendAsync( DateOnly from, DateOnly to, CancellationToken cancellationToken = default) { + await using var db = await _contextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false); var totals = new Dictionary(); - foreach (var meterId in await ActiveMeterIdsAsync(cancellationToken).ConfigureAwait(false)) + foreach (var meterId in await ActiveMeterIdsAsync(db, cancellationToken).ConfigureAwait(false)) { foreach (var bucket in await _costService.GetMeterCostsAsync(meterId, ToUtc(from), ToUtc(to), CostBucket.Month, cancellationToken).ConfigureAwait(false)) { @@ -90,25 +97,25 @@ public sealed class DashboardService(MeterVaultDbContext db, CostService costSer return [.. totals.OrderBy(kv => kv.Key).Select(kv => new TrendPoint(kv.Key, kv.Value))]; } - private async Task TotalCostAsync(DateOnly from, DateOnly to, CancellationToken cancellationToken) + private async Task TotalCostAsync(MeterVaultDbContext db, DateOnly from, DateOnly to, CancellationToken cancellationToken) { double total = 0; - foreach (var meterId in await ActiveMeterIdsAsync(cancellationToken).ConfigureAwait(false)) + foreach (var meterId in await ActiveMeterIdsAsync(db, cancellationToken).ConfigureAwait(false)) { var costs = await _costService.GetMeterCostsAsync(meterId, ToUtc(from), ToUtc(to), CostBucket.Month, cancellationToken).ConfigureAwait(false); total += costs.Sum(c => c.Cost); } - var manual = await _db.ManualCosts + var manual = await db.ManualCosts.AsNoTracking() .Where(c => c.PeriodStart >= from && c.PeriodStart < to) .SumAsync(c => (double?)c.Amount, cancellationToken).ConfigureAwait(false); return total + (manual ?? 0); } - private async Task LatestMonthCostAsync(CancellationToken cancellationToken) + private async Task LatestMonthCostAsync(MeterVaultDbContext db, CancellationToken cancellationToken) { - var latest = await _db.Consumption + var latest = await db.Consumption.AsNoTracking() .OrderByDescending(c => c.Time) .Select(c => (DateTimeOffset?)c.Time) .FirstOrDefaultAsync(cancellationToken).ConfigureAwait(false); @@ -118,11 +125,11 @@ public sealed class DashboardService(MeterVaultDbContext db, CostService costSer } var monthStart = new DateOnly(latest.Value.Year, latest.Value.Month, 1); - return await TotalCostAsync(monthStart, monthStart.AddMonths(1), cancellationToken).ConfigureAwait(false); + return await TotalCostAsync(db, monthStart, monthStart.AddMonths(1), cancellationToken).ConfigureAwait(false); } - private async Task> ActiveMeterIdsAsync(CancellationToken cancellationToken) => - await _db.Meters.Select(m => m.Id).ToListAsync(cancellationToken).ConfigureAwait(false); + private static async Task> ActiveMeterIdsAsync(MeterVaultDbContext db, CancellationToken cancellationToken) => + await db.Meters.AsNoTracking().Select(m => m.Id).ToListAsync(cancellationToken).ConfigureAwait(false); private static DateTimeOffset ToUtc(DateOnly date) => new(date.Year, date.Month, date.Day, 0, 0, 0, TimeSpan.Zero); } diff --git a/src/Infrastructure/DependencyInjection.cs b/src/Infrastructure/DependencyInjection.cs index 76363cb..7d97f7c 100644 --- a/src/Infrastructure/DependencyInjection.cs +++ b/src/Infrastructure/DependencyInjection.cs @@ -15,11 +15,16 @@ public static class DependencyInjection this IServiceCollection services, string connectionString) { - services.AddDbContext(options => + // A factory (for per-operation contexts in Blazor components/read services — a shared + // circuit-scoped DbContext is not thread-safe) plus a scoped context (from the factory) + // for request/tick-scoped services (API, workers, importers) that inject it directly. + services.AddDbContextFactory(options => options .UseNpgsql(connectionString, npgsql => npgsql.MigrationsAssembly(typeof(MeterVaultDbContext).Assembly.FullName)) .UseSnakeCaseNamingConvention()); + services.AddScoped(sp => + sp.GetRequiredService>().CreateDbContext()); services.AddSingleton(_ => NormalizationEngine.CreateDefault()); services.AddScoped(); diff --git a/src/Infrastructure/Import/CsvImporter.cs b/src/Infrastructure/Import/CsvImporter.cs index df45991..9be28f0 100644 --- a/src/Infrastructure/Import/CsvImporter.cs +++ b/src/Infrastructure/Import/CsvImporter.cs @@ -46,7 +46,7 @@ public sealed class CsvImporter foreach (var column in profile.Columns) { - StageColumn(column, row, time, r, staged, previousByMeter); + StageColumn(column, row, time, r, staged, previousByMeter, profile.DetectCumulativeSwaps); } } @@ -54,7 +54,7 @@ public sealed class CsvImporter } private static void StageColumn(ColumnMapping column, IReadOnlyList row, DateTimeOffset time, - int rowIndex, StagedImport staged, Dictionary previousByMeter) + int rowIndex, StagedImport staged, Dictionary previousByMeter, bool detectSwaps) { if (column.Role == MappingRole.Ignore || column.Index >= row.Count) { @@ -70,7 +70,7 @@ public sealed class CsvImporter switch (column.Role) { case MappingRole.Reading: - StageReading(column, row, cell, time, rowIndex, staged, previousByMeter); + StageReading(column, row, cell, time, rowIndex, staged, previousByMeter, detectSwaps); break; case MappingRole.Delivery: @@ -126,7 +126,7 @@ public sealed class CsvImporter } private static void StageReading(ColumnMapping column, IReadOnlyList row, string cell, - DateTimeOffset time, int rowIndex, StagedImport staged, Dictionary previousByMeter) + DateTimeOffset time, int rowIndex, StagedImport staged, Dictionary previousByMeter, bool detectSwaps) { var split = ValueCell.Split(cell); if (split is null) @@ -144,7 +144,7 @@ public sealed class CsvImporter $"Row {rowIndex + 1}: expected unit '{column.Unit}' but found '{split.Value.Unit}'."); } - if (previousByMeter.TryGetValue(meterId, out var previous) && value < previous) + if (detectSwaps && previousByMeter.TryGetValue(meterId, out var previous) && value < previous) { var override_ = ReadSwapOverride(column, row); staged.Events.Add(new MeterEvent diff --git a/src/Infrastructure/Import/ReferenceDataImporter.cs b/src/Infrastructure/Import/ReferenceDataImporter.cs index ba89f20..988f109 100644 --- a/src/Infrastructure/Import/ReferenceDataImporter.cs +++ b/src/Infrastructure/Import/ReferenceDataImporter.cs @@ -61,7 +61,10 @@ public sealed class ReferenceDataImporter(MeterVaultDbContext db, ImportService AddElectricityTariffs(electricity); AddTariff(TariffScope.EnergyType, water, 5.00, "EUR/m3", new DateOnly(2022, 11, 1)); - await LinkCategoriesAsync(haus.Id, netz.Id, auto.Id, solar1.Id, solar2.Id, wasser.Id, oilTank.Id, cancellationToken).ConfigureAwait(false); + // Strom and Wasser costs are computed from meters + tariffs; only Heizung comes from the + // Kosten sheet (oil has no tariff). Linking a meter AND importing its Kosten column into the + // same category would double-count, so we keep exactly one cost source per category. + await LinkStromAndWasserAsync(haus.Id, netz.Id, auto.Id, solar1.Id, solar2.Id, wasser.Id, cancellationToken).ConfigureAwait(false); await _db.SaveChangesAsync(cancellationToken).ConfigureAwait(false); var meterIds = new ReferenceMeterIds(haus.Id, netz.Id, auto.Id, solar1.Id, solar2.Id, wasser.Id, oilTank.Id, burner.Id); @@ -70,9 +73,19 @@ public sealed class ReferenceDataImporter(MeterVaultDbContext db, ImportService await ImportSheetAsync(sampleDataDirectory, ElectricityFile, ReferenceProfiles.Electricity(meterIds), cancellationToken).ConfigureAwait(false); await ImportSheetAsync(sampleDataDirectory, WaterFile, ReferenceProfiles.Water(meterIds), cancellationToken).ConfigureAwait(false); await ImportSheetAsync(sampleDataDirectory, OilFile, ReferenceProfiles.HeatingOil(meterIds), cancellationToken).ConfigureAwait(false); - await ImportSheetAsync(sampleDataDirectory, CostsFile, ReferenceProfiles.Costs(categoryIds), cancellationToken).ConfigureAwait(false); + await ImportSheetAsync(sampleDataDirectory, CostsFile, HeizungCostsProfile(categoryIds.Heizung), cancellationToken).ConfigureAwait(false); } + /// Kosten profile that imports only the Heizung column — Strom/Wasser are metered. + private static MappingProfile HeizungCostsProfile(int heizungCategoryId) => new() + { + Name = "Energiebilanz — Kosten (Heizung)", + DateColumn = 0, + DateKind = DateKind.MonthName, + FirstDataRowIndex = 1, + Columns = [new ColumnMapping { Index = 3, Role = MappingRole.ManualCost, CategoryId = heizungCategoryId }], + }; + private async Task ImportSheetAsync(string dir, string file, MappingProfile profile, CancellationToken cancellationToken) { var path = Path.Combine(dir, file); @@ -121,8 +134,8 @@ public sealed class ReferenceDataImporter(MeterVaultDbContext db, ImportService ValidFrom = validFrom, }); - private async Task LinkCategoriesAsync( - int haus, int netz, int auto, int solar1, int solar2, int wasser, int oilTank, CancellationToken cancellationToken) + private async Task LinkStromAndWasserAsync( + int haus, int netz, int auto, int solar1, int solar2, int wasser, CancellationToken cancellationToken) { var categories = await CategoryIdsAsync(cancellationToken).ConfigureAwait(false); foreach (var meterId in new[] { haus, netz, auto, solar1, solar2 }) @@ -131,7 +144,7 @@ public sealed class ReferenceDataImporter(MeterVaultDbContext db, ImportService } _db.CostCategoryMembers.Add(new CostCategoryMember { CategoryId = categories.Wasser, MeterId = wasser }); - _db.CostCategoryMembers.Add(new CostCategoryMember { CategoryId = categories.Heizung, MeterId = oilTank }); + // Heizung (oil) cost comes from the imported Kosten column, not a meter — no link (avoids double-count). } private async Task CategoryIdsAsync(CancellationToken cancellationToken) diff --git a/src/Infrastructure/Ingestion/HomeAssistantWorker.cs b/src/Infrastructure/Ingestion/HomeAssistantWorker.cs index 724a415..02b1395 100644 --- a/src/Infrastructure/Ingestion/HomeAssistantWorker.cs +++ b/src/Infrastructure/Ingestion/HomeAssistantWorker.cs @@ -67,6 +67,13 @@ public sealed class HomeAssistantWorker( .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) { diff --git a/src/Infrastructure/Ingestion/IngestionService.cs b/src/Infrastructure/Ingestion/IngestionService.cs index b371fa2..f86b10d 100644 --- a/src/Infrastructure/Ingestion/IngestionService.cs +++ b/src/Infrastructure/Ingestion/IngestionService.cs @@ -112,7 +112,7 @@ public sealed class IngestionService(MeterVaultDbContext db) var previous = await _db.Readings .Where(r => r.MeterId == meterId && r.Time < time) .OrderByDescending(r => r.Time) - .Select(r => (double?)r.Value) + .Select(r => new { r.Value, r.Time }) .FirstOrDefaultAsync(cancellationToken).ConfigureAwait(false); if (previous is null || value >= previous.Value) @@ -120,11 +120,12 @@ public sealed class IngestionService(MeterVaultDbContext db) return false; } - // A reset/swap event between the previous reading and this one explains the decrease. + // Only a reset/swap in the window (previousReading, thisReading] explains the decrease — + // an old historical reset must not permanently disable the guard. var explained = await _db.MeterEvents.AnyAsync( e => e.MeterId == meterId && (e.EventType == MeterEventType.CounterReset || e.EventType == MeterEventType.MeterSwap) - && e.Time <= time, + && e.Time > previous.Time && e.Time <= time, cancellationToken).ConfigureAwait(false); return !explained; diff --git a/src/Infrastructure/Ingestion/MqttIngestionWorker.cs b/src/Infrastructure/Ingestion/MqttIngestionWorker.cs index 8c674ed..f9702e6 100644 --- a/src/Infrastructure/Ingestion/MqttIngestionWorker.cs +++ b/src/Infrastructure/Ingestion/MqttIngestionWorker.cs @@ -24,42 +24,52 @@ public sealed class MqttIngestionWorker( private readonly ILogger _logger = logger; private readonly MqttClientFactory _factory = new(); private readonly ConcurrentDictionary _clients = new(); + private readonly ConcurrentDictionary> _subscribed = new(); protected override async Task ExecuteAsync(CancellationToken stoppingToken) { using var timer = new PeriodicTimer(ReconnectInterval); - do + try { - try + do { - await EnsureConnectionsAsync(stoppingToken).ConfigureAwait(false); - } - catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) - { - break; - } - catch (Exception ex) - { - _logger.LogWarning(ex, "MQTT ingestion tick failed; will retry"); - } - } - while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false)); - - foreach (var client in _clients.Values) - { - try - { - if (client.IsConnected) + try { - await client.DisconnectAsync(cancellationToken: CancellationToken.None).ConfigureAwait(false); + await EnsureConnectionsAsync(stoppingToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + catch (Exception ex) + { + _logger.LogWarning(ex, "MQTT ingestion tick failed; will retry"); } } - catch (Exception ex) + while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false)); + } + catch (OperationCanceledException) + { + // Normal shutdown while idle waiting for the next tick. + } + finally + { + foreach (var client in _clients.Values) { - _logger.LogDebug(ex, "Error disconnecting MQTT client during shutdown"); - } + try + { + if (client.IsConnected) + { + await client.DisconnectAsync(cancellationToken: CancellationToken.None).ConfigureAwait(false); + } + } + catch (Exception ex) + { + _logger.LogDebug(ex, "Error disconnecting MQTT client during shutdown"); + } - client.Dispose(); + client.Dispose(); + } } } @@ -75,13 +85,32 @@ public sealed class MqttIngestionWorker( foreach (var endpoint in endpoints) { var client = _clients.GetOrAdd(endpoint.Id, _ => CreateClient()); - if (client.IsConnected) - { - continue; - } - var topics = await ResolveTopicsAsync(db, endpoint, cancellationToken).ConfigureAwait(false); - await ConnectAndSubscribeAsync(endpoint, client, topics, cancellationToken).ConfigureAwait(false); + + if (!client.IsConnected) + { + _subscribed[endpoint.Id] = []; + await ConnectAndSubscribeAsync(endpoint, client, topics, cancellationToken).ConfigureAwait(false); + } + else + { + // Already connected — subscribe to any topics added since the last tick. + await SubscribeNewTopicsAsync(endpoint.Id, client, topics, cancellationToken).ConfigureAwait(false); + } + } + } + + private async Task SubscribeNewTopicsAsync( + int endpointId, IMqttClient client, IReadOnlyList topics, CancellationToken cancellationToken) + { + var known = _subscribed.GetOrAdd(endpointId, _ => []); + foreach (var topic in topics) + { + if (known.Add(topic)) + { + await client.SubscribeAsync(topic, cancellationToken: cancellationToken).ConfigureAwait(false); + _logger.LogInformation("MQTT subscribed to new topic {Topic} on endpoint {Endpoint}", topic, endpointId); + } } } @@ -114,9 +143,11 @@ public sealed class MqttIngestionWorker( try { await client.ConnectAsync(builder.Build(), cancellationToken).ConfigureAwait(false); + var known = _subscribed.GetOrAdd(endpoint.Id, _ => []); foreach (var topic in topics) { await client.SubscribeAsync(topic, cancellationToken: cancellationToken).ConfigureAwait(false); + known.Add(topic); } _logger.LogInformation("MQTT connected to {Host}:{Port} ({TopicCount} topics)", diff --git a/src/Infrastructure/Options/MeterVaultOptions.cs b/src/Infrastructure/Options/MeterVaultOptions.cs index bc69c1d..a828095 100644 --- a/src/Infrastructure/Options/MeterVaultOptions.cs +++ b/src/Infrastructure/Options/MeterVaultOptions.cs @@ -33,4 +33,10 @@ public sealed class MeterVaultOptions /// Honour X-Forwarded-User/Remote-User from a trusted reverse proxy (SDD §10). public bool ReverseProxyTrust { get; set; } + + /// + /// Allow the REST API without any API key. Off by default: with no keys configured the API is + /// closed (401) rather than open, so an unconfigured deployment doesn't expose export/import. + /// + public bool AllowAnonymousApi { get; set; } } diff --git a/src/Infrastructure/Persistence/Migrations/20260713085642_TimescaleHypertables.cs b/src/Infrastructure/Persistence/Migrations/20260713085642_TimescaleHypertables.cs index 167297f..18b8c3d 100644 --- a/src/Infrastructure/Persistence/Migrations/20260713085642_TimescaleHypertables.cs +++ b/src/Infrastructure/Persistence/Migrations/20260713085642_TimescaleHypertables.cs @@ -18,8 +18,9 @@ namespace MeterVault.Infrastructure.Persistence.Migrations protected override void Up(MigrationBuilder migrationBuilder) { // Raw readings: 30-day chunks; the (meter_id, time) PK contains the partition column. + // if_not_exists keeps a down-then-up cycle safe (Down leaves the hypertable in place). migrationBuilder.Sql( - "SELECT create_hypertable('reading', 'time', chunk_time_interval => INTERVAL '30 days');"); + "SELECT create_hypertable('reading', 'time', chunk_time_interval => INTERVAL '30 days', if_not_exists => TRUE);"); // Columnar compression, segmented by meter (monotonic sensor data compresses ~10-20x). migrationBuilder.Sql( @@ -29,11 +30,11 @@ namespace MeterVault.Infrastructure.Persistence.Migrations "timescaledb.compress_orderby = 'time DESC');"); migrationBuilder.Sql( - "SELECT add_compression_policy('reading', INTERVAL '30 days');"); + "SELECT add_compression_policy('reading', INTERVAL '30 days', if_not_exists => true);"); // Normalized consumption: 90-day chunks. Kept effectively forever (small), so no compression policy. migrationBuilder.Sql( - "SELECT create_hypertable('consumption', 'time', chunk_time_interval => INTERVAL '90 days');"); + "SELECT create_hypertable('consumption', 'time', chunk_time_interval => INTERVAL '90 days', if_not_exists => TRUE);"); } /// diff --git a/src/Infrastructure/Persistence/Migrations/20260713094634_ContinuousAggregates.cs b/src/Infrastructure/Persistence/Migrations/20260713094634_ContinuousAggregates.cs index 99df849..efc5ee8 100644 --- a/src/Infrastructure/Persistence/Migrations/20260713094634_ContinuousAggregates.cs +++ b/src/Infrastructure/Persistence/Migrations/20260713094634_ContinuousAggregates.cs @@ -40,8 +40,11 @@ namespace MeterVault.Infrastructure.Persistence.Migrations private static void CreateAggregate(MigrationBuilder builder, string name, string bucket) { + // IF NOT EXISTS + if_not_exists keep this re-run-safe: because the statements suppress + // the transaction, a crash before __EFMigrationsHistory is written would otherwise brick + // the next migrate by re-creating an object that already exists. builder.Sql( - $"CREATE MATERIALIZED VIEW {name} WITH (timescaledb.continuous) AS " + + $"CREATE MATERIALIZED VIEW IF NOT EXISTS {name} WITH (timescaledb.continuous) AS " + $"SELECT time_bucket(INTERVAL '{bucket}', time, '{Timezone}') AS bucket, " + "meter_id, kind, sum(amount) AS amount " + "FROM consumption GROUP BY bucket, meter_id, kind WITH NO DATA;", @@ -54,7 +57,8 @@ namespace MeterVault.Infrastructure.Persistence.Migrations $"SELECT add_continuous_aggregate_policy('{name}', " + $"start_offset => INTERVAL '{startOffset}', " + $"end_offset => INTERVAL '{endOffset}', " + - "schedule_interval => INTERVAL '1 hour');", + "schedule_interval => INTERVAL '1 hour', " + + "if_not_exists => true);", suppressTransaction: true); } } diff --git a/tests/Integration.Tests/ApiTests.cs b/tests/Integration.Tests/ApiTests.cs index 43c4f8b..c713598 100644 --- a/tests/Integration.Tests/ApiTests.cs +++ b/tests/Integration.Tests/ApiTests.cs @@ -52,6 +52,37 @@ public sealed class ApiTests(TimescaleFixture fx) } } + private sealed record EventPush(int MeterId, DateTimeOffset Time, string Type, + double? Amount, double? PrevValue, double? NewValue, string? Unit, string? Notes); + + [Fact] + public async Task Api_is_closed_when_no_keys_are_configured() + { + using var factory = new MeterVaultAppFactory(fx.ConnectionString, configureApiKey: false); + using var client = factory.CreateClient(); + + var response = await client.PostAsJsonAsync("/api/v1/readings", + new[] { new ReadingPush(1, DateTimeOffset.UtcNow, 1) }); + + Assert.Equal(HttpStatusCode.Unauthorized, response.StatusCode); + } + + [Fact] + public async Task Events_for_a_missing_meter_return_404_not_500() + { + using var factory = new MeterVaultAppFactory(fx.ConnectionString); + using var client = factory.CreateClient(); + + using var request = new HttpRequestMessage(HttpMethod.Post, "/api/v1/events") + { + Content = JsonContent.Create(new EventPush(999999, DateTimeOffset.UtcNow, "Delivery", 100, null, null, "L", null)), + }; + request.Headers.Add("X-Api-Key", MeterVaultAppFactory.ApiKey); + var response = await client.SendAsync(request); + + Assert.Equal(HttpStatusCode.NotFound, response.StatusCode); + } + [Fact] public async Task Meters_endpoint_and_swagger_are_available() { diff --git a/tests/Integration.Tests/Costing/CostReconciliationTests.cs b/tests/Integration.Tests/Costing/CostReconciliationTests.cs index 70c20e1..2572e1b 100644 --- a/tests/Integration.Tests/Costing/CostReconciliationTests.cs +++ b/tests/Integration.Tests/Costing/CostReconciliationTests.cs @@ -36,7 +36,7 @@ public sealed class CostReconciliationTests(TimescaleFixture fx) }); await db.SaveChangesAsync(); - var costs = await new CostService(db).GetMeterCostsAsync(meterId, From, To); + var costs = await new CostService(fx).GetMeterCostsAsync(meterId, From, To); var computed = costs.ToDictionary(c => c.Period, c => c.Cost); var oracle = OracleByMonth(ReadRows(Water), dateColumn: 0, valueColumn: 4, firstDataRow: 1); // Kosten @@ -64,7 +64,7 @@ public sealed class CostReconciliationTests(TimescaleFixture fx) db.CostCategoryMembers.Add(new CostCategoryMember { CategoryId = wasserCategory.Id, MeterId = meterId }); await db.SaveChangesAsync(); - var rollup = await new CostService(db).GetCategoryCostsAsync(wasserCategory.Id, From, To); + var rollup = await new CostService(fx).GetCategoryCostsAsync(wasserCategory.Id, From, To); var dec2022 = rollup.Single(r => r.Period == new DateOnly(2022, 12, 1)); Assert.Equal(70d, dec2022.Cost, 2); // Dez 2022: 14 m³ × 5,00 € diff --git a/tests/Integration.Tests/DashboardRenderTests.cs b/tests/Integration.Tests/DashboardRenderTests.cs index b69b120..bc93f3f 100644 --- a/tests/Integration.Tests/DashboardRenderTests.cs +++ b/tests/Integration.Tests/DashboardRenderTests.cs @@ -1,3 +1,4 @@ +using MeterVault.Infrastructure.Costing; using MeterVault.Infrastructure.Import; using MeterVault.Infrastructure.Persistence; using Microsoft.EntityFrameworkCore; @@ -31,6 +32,15 @@ public sealed class DashboardRenderTests(TimescaleFixture fx) Assert.Contains(await db.Meters.Select(m => m.Name).ToListAsync(), n => n == "Zähler Haus"); Assert.True(await db.Meters.CountAsync() >= 8); Assert.True(await db.Consumption.AnyAsync()); + + // Regression (audit): Wasser is metered (water tariff), the Kosten Wasser column is + // NOT imported, so the category is not double-counted — Dez 2022 = 14 m³ × 5 € = 70 €. + var wasser = await db.CostCategories.FirstAsync(c => c.Name == "Wasser"); + var rollup = await new CostService(fx).GetCategoryCostsAsync( + wasser.Id, + new DateTimeOffset(2022, 12, 1, 0, 0, 0, TimeSpan.Zero), + new DateTimeOffset(2023, 1, 1, 0, 0, 0, TimeSpan.Zero)); + Assert.Equal(70d, rollup.Sum(r => r.Cost), 1); } using var client = factory.CreateClient(); diff --git a/tests/Integration.Tests/Ingestion/IngestionServiceTests.cs b/tests/Integration.Tests/Ingestion/IngestionServiceTests.cs index 17c0dfb..6714051 100644 --- a/tests/Integration.Tests/Ingestion/IngestionServiceTests.cs +++ b/tests/Integration.Tests/Ingestion/IngestionServiceTests.cs @@ -49,6 +49,28 @@ public sealed class IngestionServiceTests(TimescaleFixture fx) await CleanupAsync(db, meterId); } + [Fact] + public async Task Old_reset_does_not_permanently_disable_the_decrease_guard() + { + await using var db = fx.CreateContext(); + var (meterId, sourceId) = await SetupAsync(db, MeterMode.CumulativeCounter); + var service = new IngestionService(db); + + // A reset early on explains an early decrease... + await service.IngestAsync(sourceId, T0, 100); + db.MeterEvents.Add(new MeterEvent { MeterId = meterId, Time = T0.AddMinutes(10), EventType = MeterEventType.CounterReset, NewValue = 0 }); + await db.SaveChangesAsync(); + Assert.Equal(IngestionOutcome.Written, await service.IngestAsync(sourceId, T0.AddMinutes(20), 30)); + await service.IngestAsync(sourceId, T0.AddHours(1), 200); + + // ...but a later spurious decrease with NO event in its window must still be rejected. + var outcome = await service.IngestAsync(sourceId, T0.AddHours(2), 150); + + Assert.Equal(IngestionOutcome.RejectedDecrease, outcome); + + await CleanupAsync(db, meterId); + } + [Fact] public async Task Allows_decrease_when_a_swap_event_explains_it() { diff --git a/tests/Integration.Tests/MeterVaultAppFactory.cs b/tests/Integration.Tests/MeterVaultAppFactory.cs index 1fc695c..cb928cd 100644 --- a/tests/Integration.Tests/MeterVaultAppFactory.cs +++ b/tests/Integration.Tests/MeterVaultAppFactory.cs @@ -7,7 +7,7 @@ namespace MeterVault.Integration.Tests; /// Boots the real ASP.NET Core app in-memory against the shared Timescale container. /// Migrations are already applied by , so startup migration is off. /// -public sealed class MeterVaultAppFactory(string connectionString) : WebApplicationFactory +public sealed class MeterVaultAppFactory(string connectionString, bool configureApiKey = true) : WebApplicationFactory { public const string ApiKey = "test-api-key"; @@ -17,6 +17,9 @@ public sealed class MeterVaultAppFactory(string connectionString) : WebApplicati builder.UseSetting("ConnectionStrings:Default", connectionString); builder.UseSetting("MeterVault:RunMigrationsAtStartup", "false"); builder.UseSetting("MeterVault:EnableLiveIngestion", "false"); - builder.UseSetting("MeterVault:ApiKeys:0", ApiKey); + if (configureApiKey) + { + builder.UseSetting("MeterVault:ApiKeys:0", ApiKey); + } } } diff --git a/tests/Integration.Tests/TimescaleFixture.cs b/tests/Integration.Tests/TimescaleFixture.cs index 53da599..cef1668 100644 --- a/tests/Integration.Tests/TimescaleFixture.cs +++ b/tests/Integration.Tests/TimescaleFixture.cs @@ -10,7 +10,7 @@ namespace MeterVault.Integration.Tests; /// The image tag is pinned: the compression DDL (add_compression_policy) was renamed toward /// add_columnstore_policy in newer Timescale, so a floating tag would risk breaking migrations. /// -public sealed class TimescaleFixture : IAsyncLifetime +public sealed class TimescaleFixture : IAsyncLifetime, IDbContextFactory { private readonly PostgreSqlContainer _db = new PostgreSqlBuilder("timescale/timescaledb:2.17.2-pg16") .WithDatabase("metervault") @@ -30,6 +30,9 @@ public sealed class TimescaleFixture : IAsyncLifetime return new MeterVaultDbContext(options); } + /// Lets tests construct services that take an . + public MeterVaultDbContext CreateDbContext() => CreateContext(); + public async Task InitializeAsync() { await _db.StartAsync();