using Dapper;
using MeterVault.Core.Analysis;
using MeterVault.Core.Analysis.Coverage;
using MeterVault.Core.Analysis.Rollups;
using MeterVault.Core.Domain;
using MeterVault.Core.Normalization;
using MeterVault.Infrastructure.Import;
using MeterVault.Infrastructure.Ingestion;
using MeterVault.Infrastructure.Normalization;
using MeterVault.Infrastructure.Options;
using MeterVault.Infrastructure.Persistence;
using MeterVault.Infrastructure.Persistence.Analysis;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging.Abstractions;
using static MeterVault.Integration.Tests.Reconciliation.ReconciliationSupport;
namespace MeterVault.Integration.Tests.Analysis;
///
/// The analysis tables (D-12): day and month rollups, coverage runs and rollup state, written by the recompute from
/// the same rows it stores as consumption, by diff, in the caller's transaction — and rebuilt by the startup upgrade
/// (D-16). Checked against the database's own sums of consumption, the way a reader would query them.
///
[Collection("Timescale")]
public sealed class AnalysisDataTests(TimescaleFixture fx)
{
private const string BerlinId = "Europe/Berlin";
private static readonly TimeZoneInfo Berlin = TimeZoneInfo.FindSystemTimeZoneById(BerlinId);
/// The frozen "now" the recomputes stamp with.
private static readonly DateTimeOffset BuildTime = new(2026, 9, 19, 8, 0, 0, TimeSpan.Zero);
[Fact]
public async Task Rollups_equal_consumption_summed_by_local_day_and_month_for_every_reference_meter()
{
await using var db = fx.CreateContext();
var meters = await ImportReferenceSheetsAsync(db);
try
{
foreach (var meterId in meters.All)
{
var days = await db.ConsumptionRollups.AsNoTracking().Where(r => r.MeterId == meterId)
.OrderBy(r => r.Day).ThenBy(r => r.Kind).ToListAsync();
var months = await db.ConsumptionRollupMonths.AsNoTracking().Where(r => r.MeterId == meterId)
.OrderBy(r => r.Month).ThenBy(r => r.Kind).ToListAsync();
Assert.NotEmpty(days);
AssertSameBuckets(await ConsumptionByAsync(db, meterId, "day"), days.Select(d => d.ToBucket()).ToList(), meterId);
AssertSameBuckets(await ConsumptionByAsync(db, meterId, "month"), months.Select(m => m.ToBucket()).ToList(), meterId);
}
// Generation stays generation; the electricity sheet's Solar 1 has only generation rows.
Assert.All(
await db.ConsumptionRollupMonths.AsNoTracking().Where(r => r.MeterId == meters.Solar1).ToListAsync(),
r => Assert.Equal(ConsumptionKind.Generation, r.Kind));
// Water, December 2022: the sheet's 14 m³, one imported month row, recorded until the month ends.
var december = await db.ConsumptionRollupMonths.AsNoTracking()
.SingleAsync(r => r.MeterId == meters.Wasser && r.Month == new DateOnly(2022, 12, 1));
Assert.Equal(14, december.Amount, 9);
Assert.Equal(14, december.Imported, 9);
Assert.Equal(1, december.Rows);
Assert.Equal(Provenance.Imported, december.ToBucket().Provenance);
}
finally
{
await DeleteAsync(db, meters.All);
}
}
[Fact]
public async Task A_label_rows_interval_ends_where_its_month_ends()
{
// A-05: "Dezember 2022" is the register at the end of December, stamped on the 1st. Its day and month
// rollups are recorded until the local midnight that ends December, so a reader whose now is inside
// December does not count it as an actual.
await using var db = fx.CreateContext();
var meters = await ImportReferenceSheetsAsync(db);
try
{
var endOfDecember = GapAttribution.LocalMidnight(new DateOnly(2023, 1, 1), Berlin);
var month = await db.ConsumptionRollupMonths.AsNoTracking()
.SingleAsync(r => r.MeterId == meters.Wasser && r.Month == new DateOnly(2022, 12, 1));
var day = await db.ConsumptionRollups.AsNoTracking()
.SingleAsync(r => r.MeterId == meters.Wasser && r.Day == new DateOnly(2022, 12, 1));
Assert.Equal(endOfDecember, month.MaxIntervalEnd);
Assert.Equal(endOfDecember, day.MaxIntervalEnd);
Assert.Equal(TimeSpan.Zero, month.MaxIntervalEnd.Offset);
// Every imported month of the sheet ends at its own month's end.
var all = await db.ConsumptionRollupMonths.AsNoTracking().Where(r => r.MeterId == meters.Wasser).ToListAsync();
Assert.All(all, r => Assert.Equal(GapAttribution.LocalMidnight(r.Month.AddMonths(1), Berlin), r.MaxIntervalEnd));
}
finally
{
await DeleteAsync(db, meters.All);
}
}
[Fact]
public async Task Coverage_runs_are_stored_for_the_seeded_water_meter()
{
await using var db = fx.CreateContext();
var meters = await ImportReferenceSheetsAsync(db);
try
{
var stored = await db.MeterCoverage.AsNoTracking().Where(r => r.MeterId == meters.Wasser)
.OrderBy(r => r.SpanFrom).ToListAsync();
// One month run from November 2022 to the end of May 2026, divided at months, across the swap.
var run = Assert.Single(stored);
Assert.Equal(GapAttribution.LocalMidnight(new DateOnly(2022, 11, 1), Berlin), run.SpanFrom);
Assert.Equal(GapAttribution.LocalMidnight(new DateOnly(2026, 6, 1), Berlin), run.SpanTo);
Assert.Equal(GapAttribution.LocalMidnight(new DateOnly(2026, 5, 1), Berlin), run.LastIntervalStart);
Assert.Equal(ResolutionClass.Month, run.ResolutionClass);
Assert.True(run.DividedAtMonths);
Assert.Equal(CoverageGapReason.None, run.GapReason);
// What is stored is exactly what the builder makes of the engine's rows.
Assert.Equal(await ExpectedCoverageAsync(db, meters.Wasser), stored.Select(r => r.ToRun()).ToList());
// The burner's twelve-year first interval is its own coarse run, and the tank's runs start at its
// first dipstick, not its first delivery (the rows stored match the builder there too).
Assert.Equal(await ExpectedCoverageAsync(db, meters.Burner), await RunsAsync(db, meters.Burner));
Assert.Equal(ResolutionClass.Coarse, (await RunsAsync(db, meters.Burner))[0].Resolution);
Assert.Equal(await ExpectedCoverageAsync(db, meters.OilTank), await RunsAsync(db, meters.OilTank));
}
finally
{
await DeleteAsync(db, meters.All);
}
}
[Fact]
public async Task Recomputing_unchanged_data_rewrites_nothing_and_a_new_reading_touches_only_what_it_moved()
{
await using var db = fx.CreateContext();
var meters = await ImportReferenceSheetsAsync(db);
var water = meters.Wasser;
try
{
var before = await VersionsAsync(db, water);
var state = await StateAsync(db, water);
Assert.Equal(BuildTime, state.BuiltAt);
// Same inputs, a later clock: not one row version changes, and the state keeps its build time.
await using (var again = fx.CreateContext())
{
await using var tx = await again.Database.BeginTransactionAsync();
await Normalization(again, time: new FixedTimeProvider(BuildTime.AddHours(1))).RecomputeMeterAsync(water, null);
await again.SaveChangesAsync();
await tx.CommitAsync();
}
Assert.Equal(before, await VersionsAsync(db, water));
Assert.Equal(BuildTime, (await StateAsync(db, water)).BuiltAt);
// A live reading ten days after the sheet's last month: one new day, one new month, the one run
// extended — every other row keeps its version.
var later = BuildTime.AddHours(2);
await using (var live = fx.CreateContext())
{
var ingestion = new IngestionService(live, Normalization(live, time: new FixedTimeProvider(later)));
var outcome = await ingestion.IngestByMeterAsync(water, Local(2026, 6, 11, 12), 700);
Assert.Equal(IngestionOutcome.Written, outcome);
}
var after = await VersionsAsync(db, water);
Assert.Equal(before.Days.Count + 1, after.Days.Count);
Assert.Empty(before.Days.Except(after.Days));
Assert.Equal("2026-06-11", Assert.Single(after.Days.Except(before.Days)).Split('|')[0]);
Assert.Equal(before.Months.Count + 1, after.Months.Count);
Assert.Empty(before.Months.Except(after.Months));
Assert.Single(after.Coverage);
Assert.NotEqual(before.Coverage, after.Coverage);
var run = await db.MeterCoverage.AsNoTracking().SingleAsync(r => r.MeterId == water);
Assert.Equal(Local(2026, 6, 11, 12), run.SpanTo);
Assert.Equal(GapAttribution.LocalMidnight(new DateOnly(2026, 6, 1), Berlin), run.LastIntervalStart);
var june = await db.ConsumptionRollupMonths.AsNoTracking().SingleAsync(r => r.MeterId == water && r.Month == new DateOnly(2026, 6, 1));
Assert.Equal(13, june.Amount, 9);
Assert.Equal(13, june.Measured, 9);
Assert.Equal(later, (await StateAsync(db, water)).BuiltAt);
}
finally
{
await DeleteAsync(db, meters.All);
}
}
[Fact]
public async Task Several_recomputes_before_one_save_leave_exactly_the_last_result()
{
// One context may recompute a meter repeatedly before it saves. Rows a pass removed come back when the next
// pass wants them again, and rows a pass added go when the next one does not.
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
try
{
await using (var setup = fx.CreateContext())
{
var ingestion = new IngestionService(setup, Normalization(setup));
await ingestion.IngestByMeterAsync(meterId, Local(2026, 8, 3, 9), 700, quality: ReadingQuality.Manual);
await ingestion.IngestByMeterAsync(meterId, Local(2026, 8, 4, 9), 710, quality: ReadingQuality.Manual);
}
var original = await DaysAsync(db, meterId);
Assert.Equal([new DateOnly(2026, 8, 3), new DateOnly(2026, 8, 4)], original.Select(d => d.Start));
await using (var ctx = fx.CreateContext())
{
await using var tx = await ctx.Database.BeginTransactionAsync();
var normalization = Normalization(ctx);
// Pass 1: 4 August gone, 5 August new.
await ctx.Readings.Where(r => r.MeterId == meterId && r.Time == Local(2026, 8, 4, 9)).ExecuteDeleteAsync();
await InsertReadingAsync(ctx, meterId, Local(2026, 8, 5, 9), 715);
await normalization.RecomputeMeterAsync(meterId, null);
// Pass 2: back to 3 and 4 August.
await InsertReadingAsync(ctx, meterId, Local(2026, 8, 4, 9), 710);
await ctx.Readings.Where(r => r.MeterId == meterId && r.Time == Local(2026, 8, 5, 9)).ExecuteDeleteAsync();
await normalization.RecomputeMeterAsync(meterId, null);
await ctx.SaveChangesAsync();
await tx.CommitAsync();
}
Assert.Equal(original, await DaysAsync(db, meterId));
Assert.Equal(710, (await db.ConsumptionRollupMonths.AsNoTracking().SingleAsync(r => r.MeterId == meterId)).Amount, 9);
}
finally
{
await DeleteAsync(db, meterId);
}
}
[Fact]
public async Task Rows_removed_behind_the_trackers_back_are_written_again()
{
// A long-lived context (a worker's scope) still tracks the rows it saved. When they are deleted outside it,
// the next recompute must add them again rather than trust its stale copies.
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
try
{
await using var worker = fx.CreateContext();
var ingestion = new IngestionService(worker, Normalization(worker));
await ingestion.IngestByMeterAsync(meterId, Local(2026, 8, 3, 9), 700);
await ingestion.IngestByMeterAsync(meterId, Local(2026, 8, 4, 9), 710);
await db.ConsumptionRollups.Where(r => r.MeterId == meterId).ExecuteDeleteAsync();
await db.MeterCoverage.Where(r => r.MeterId == meterId).ExecuteDeleteAsync();
Assert.Equal(IngestionOutcome.Written, await ingestion.IngestByMeterAsync(meterId, Local(2026, 8, 5, 9), 725));
Assert.Equal(
[(new DateOnly(2026, 8, 3), 700d), (new DateOnly(2026, 8, 4), 10d), (new DateOnly(2026, 8, 5), 15d)],
(await DaysAsync(db, meterId)).Select(d => (d.Start, d.Amount)));
Assert.Equal(Local(2026, 8, 5, 9), Assert.Single(await RunsAsync(db, meterId)).To);
}
finally
{
await DeleteAsync(db, meterId);
}
}
[Fact]
public async Task An_instant_first_reading_flags_its_day_as_an_opening_balance_until_an_install_date_says_since_when()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
try
{
await using (var live = fx.CreateContext())
{
var ingestion = new IngestionService(live, Normalization(live));
await ingestion.IngestByMeterAsync(meterId, Local(2026, 8, 3, 9), 700, quality: ReadingQuality.Manual);
await ingestion.IngestByMeterAsync(meterId, Local(2026, 8, 4, 9), 710, quality: ReadingQuality.Manual);
}
var days = await DaysAsync(db, meterId);
Assert.Equal([RollupFlags.OpeningBalance, RollupFlags.None], days.Select(d => d.Flags));
Assert.Equal(RollupFlags.OpeningBalance, (await db.ConsumptionRollupMonths.AsNoTracking().SingleAsync(r => r.MeterId == meterId)).Flags);
// The opening balance is no coverage (A-01): only the day between the two readings is covered.
var run = Assert.Single(await RunsAsync(db, meterId));
Assert.Equal(Local(2026, 8, 3, 9), run.From);
Assert.Equal(Local(2026, 8, 4, 9), run.To);
// An install date is an engine input (D-10): the first reading now counts from it.
await db.Meters.Where(m => m.Id == meterId)
.ExecuteUpdateAsync(s => s.SetProperty(m => m.InstalledAt, (DateOnly?)new DateOnly(2026, 7, 1)));
await RecomputeAsync(meterId);
Assert.All(await DaysAsync(db, meterId), d => Assert.Equal(RollupFlags.None, d.Flags));
Assert.Equal(GapAttribution.LocalMidnight(new DateOnly(2026, 7, 1), Berlin), (await RunsAsync(db, meterId))[0].From);
}
finally
{
await DeleteAsync(db, meterId);
}
}
[Fact]
public async Task A_virtual_meter_stores_no_series_only_its_state()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "kWh");
var legacyId = await CreateMeterAsync(db, MeterMode.Virtual, "kWh");
try
{
await using (var live = fx.CreateContext())
{
var ingestion = new IngestionService(live, Normalization(live));
await ingestion.IngestByMeterAsync(meterId, Local(2026, 8, 3, 9), 700);
await ingestion.IngestByMeterAsync(meterId, Local(2026, 8, 4, 9), 710);
}
Assert.NotEmpty(await DaysAsync(db, meterId));
// The meter becomes a virtual generation sum: its physical series, rollups and coverage go.
await db.Meters.Where(m => m.Id == meterId).ExecuteUpdateAsync(s => s
.SetProperty(m => m.Mode, MeterMode.Virtual)
.SetProperty(m => m.Meta, """{"expression":"m1 + m2","referencedMeterIds":[1,2],"resultKind":"generation","resultUnit":"kWh","costRule":"sourceCosts"}"""));
await RecomputeAsync(meterId);
await RecomputeAsync(legacyId);
Assert.False(await db.Consumption.AnyAsync(c => c.MeterId == meterId));
Assert.Empty(await DaysAsync(db, meterId));
Assert.False(await db.ConsumptionRollupMonths.AnyAsync(r => r.MeterId == meterId));
Assert.Empty(await RunsAsync(db, meterId));
var state = await StateAsync(db, meterId);
Assert.Equal(QuantityKind.Generation, state.Kind);
Assert.Equal("kWh", state.NormalizedUnit);
Assert.Equal(NormalizationUpgrade.CurrentRevision, state.Revision);
Assert.Equal(BerlinId, state.Zone);
// A legacy virtual meter without a definition is recorded as undeclared consumption in its own unit.
var legacy = await StateAsync(db, legacyId);
Assert.Equal(QuantityKind.Consumption, legacy.Kind);
Assert.Equal("kWh", legacy.NormalizedUnit);
}
finally
{
await DeleteAsync(db, meterId, legacyId);
}
}
[Fact]
public async Task The_state_records_the_normalized_quantity_not_the_raw_unit()
{
await using var db = fx.CreateContext();
var water = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
var export = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "kWh");
var sensor = await CreateMeterAsync(db, MeterMode.InstantRate, "kW");
await db.Meters.Where(m => m.Id == export).ExecuteUpdateAsync(s => s.SetProperty(m => m.Meta, """{"role":"grid_export"}"""));
try
{
await RecomputeAsync(water, export, sensor);
Assert.Equal((QuantityKind.Consumption, "m³"), Quantity(await StateAsync(db, water)));
Assert.Equal((QuantityKind.Export, "kWh"), Quantity(await StateAsync(db, export)));
Assert.Equal((QuantityKind.Consumption, "kWh"), Quantity(await StateAsync(db, sensor)));
}
finally
{
await DeleteAsync(db, water, export, sensor);
}
static (QuantityKind, string) Quantity(MeterRollupState state) => (state.Kind, state.NormalizedUnit);
}
[Fact]
public async Task Deleting_a_meter_takes_its_analysis_rows_with_it()
{
await using var db = fx.CreateContext();
var meters = await ImportReferenceSheetsAsync(db);
var ids = meters.All;
Assert.True(await db.ConsumptionRollups.AnyAsync(r => ids.Contains(r.MeterId)));
Assert.True(await db.ConsumptionRollupMonths.AnyAsync(r => ids.Contains(r.MeterId)));
Assert.True(await db.MeterCoverage.AnyAsync(r => ids.Contains(r.MeterId)));
Assert.Equal(ids.Length, await db.MeterRollupStates.CountAsync(r => ids.Contains(r.MeterId)));
// As the meter list deletes: consumption and readings first (restrict), then the meter itself.
await DeleteAsync(db, ids);
Assert.False(await db.ConsumptionRollups.AnyAsync(r => ids.Contains(r.MeterId)));
Assert.False(await db.ConsumptionRollupMonths.AnyAsync(r => ids.Contains(r.MeterId)));
Assert.False(await db.MeterCoverage.AnyAsync(r => ids.Contains(r.MeterId)));
Assert.False(await db.MeterRollupStates.AnyAsync(r => ids.Contains(r.MeterId)));
}
[Fact]
public async Task The_upgrade_builds_rollups_and_state_for_revision_2_data_including_virtual_meters()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
var virtualId = await CreateMeterAsync(db, MeterMode.Virtual, "m3");
db.Readings.AddRange(
new Reading { MeterId = meterId, Time = Local(2026, 8, 1, 9), Value = 700, Quality = ReadingQuality.Manual },
new Reading { MeterId = meterId, Time = Local(2026, 9, 16, 18), Value = 746, Quality = ReadingQuality.Manual });
// What revision 2 stored: consumption only, and a series for the virtual meter nothing ever read.
db.Consumption.AddRange(
new Consumption { MeterId = meterId, Time = Local(2026, 8, 1, 9), Amount = 700, Quality = ReadingQuality.Manual },
new Consumption { MeterId = meterId, Time = Local(2026, 9, 16, 18), Amount = 46, Quality = ReadingQuality.Manual },
new Consumption { MeterId = virtualId, Time = Local(2026, 9, 1, 0), Amount = 5, Quality = ReadingQuality.Estimated });
await db.SaveChangesAsync();
db.ChangeTracker.Clear();
await SetSettingsAsync(db, revision: 2, zone: BerlinId);
try
{
Assert.True(await Upgrade(db).RunAsync() >= 2);
var state = await StateAsync(db, meterId);
Assert.Equal(NormalizationUpgrade.CurrentRevision, state.Revision);
Assert.Equal(BerlinId, state.Zone);
Assert.Equal(BuildTime, state.BuiltAt);
Assert.Equal((QuantityKind.Consumption, "m³"), (state.Kind, state.NormalizedUnit));
// Rollups exist and agree with the rebuilt consumption, August holding its share of the six weeks.
AssertSameBuckets(await ConsumptionByAsync(db, meterId, "month"),
(await db.ConsumptionRollupMonths.AsNoTracking().Where(r => r.MeterId == meterId).OrderBy(r => r.Month).ToListAsync())
.Select(m => m.ToBucket()).ToList(), meterId);
Assert.True((await db.ConsumptionRollupMonths.AsNoTracking()
.SingleAsync(r => r.MeterId == meterId && r.Month == new DateOnly(2026, 8, 1))).Amount > 730);
Assert.NotEmpty(await RunsAsync(db, meterId));
// The virtual meter is part of the rebuild: its stray series is purged and its state recorded.
Assert.False(await db.Consumption.AnyAsync(c => c.MeterId == virtualId));
Assert.Equal(NormalizationUpgrade.CurrentRevision, (await StateAsync(db, virtualId)).Revision);
var revision = await db.AppSettings.AsNoTracking().SingleAsync(s => s.Key == NormalizationUpgrade.SettingKey);
Assert.Equal(NormalizationUpgrade.CurrentRevision.ToString(System.Globalization.CultureInfo.InvariantCulture), revision.Value);
Assert.Equal(0, await Upgrade(db).RunAsync());
}
finally
{
await DeleteAsync(db, meterId, virtualId);
}
}
[Fact]
public async Task The_upgrade_rebuilds_a_meter_whose_state_is_missing_although_the_revision_is_current()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
try
{
// The instance is current; only the new meter, never recomputed, lacks a state.
await SetSettingsAsync(db, revision: NormalizationUpgrade.CurrentRevision, zone: BerlinId);
Assert.False(await db.MeterRollupStates.AnyAsync(s => s.MeterId == meterId));
Assert.True(await Upgrade(db).RunAsync() >= 1);
Assert.Equal(BerlinId, (await StateAsync(db, meterId)).Zone);
Assert.Equal(0, await Upgrade(db).RunAsync());
}
finally
{
await DeleteAsync(db, meterId);
}
}
[Fact]
public async Task A_zone_change_rebuilds_rollups_in_the_new_zone()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
var first = new DateTimeOffset(2026, 1, 10, 12, 0, 0, TimeSpan.Zero);
var late = new DateTimeOffset(2026, 1, 20, 23, 30, 0, TimeSpan.Zero); // 20 January in London, 21st in Berlin
try
{
await using (var london = fx.CreateContext())
{
var ingestion = new IngestionService(london, Normalization(london, "Europe/London"));
await ingestion.IngestByMeterAsync(meterId, first, 100);
await ingestion.IngestByMeterAsync(meterId, late, 130);
}
Assert.Equal("Europe/London", (await StateAsync(db, meterId)).Zone);
Assert.Contains(new DateOnly(2026, 1, 20), (await DaysAsync(db, meterId)).Select(d => d.Start));
await SetSettingsAsync(db, revision: NormalizationUpgrade.CurrentRevision, zone: "Europe/London");
Assert.True(await Upgrade(db).RunAsync() >= 1);
Assert.Equal(BerlinId, (await StateAsync(db, meterId)).Zone);
var days = (await DaysAsync(db, meterId)).Select(d => d.Start).ToList();
Assert.Contains(new DateOnly(2026, 1, 21), days);
Assert.DoesNotContain(new DateOnly(2026, 1, 20), days);
var zone = await db.AppSettings.AsNoTracking().SingleAsync(s => s.Key == NormalizationUpgrade.ZoneSettingKey);
Assert.Equal($"\"{BerlinId}\"", zone.Value);
}
finally
{
await DeleteAsync(db, meterId);
}
}
[Fact]
public async Task The_upgrade_skips_a_meter_whose_history_is_older_than_its_readings_instead_of_cutting_it_off()
{
await using var db = fx.CreateContext();
var truncated = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
var midnight = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3", installedAt: new DateOnly(2026, 5, 1));
// Consumption from 2020 whose readings are gone — what raw retention would leave behind.
db.Readings.AddRange(
new Reading { MeterId = truncated, Time = Local(2026, 6, 1, 9), Value = 900, Quality = ReadingQuality.Manual },
new Reading { MeterId = truncated, Time = Local(2026, 6, 2, 9), Value = 910, Quality = ReadingQuality.Manual });
db.Consumption.AddRange(
new Consumption { MeterId = truncated, Time = Local(2020, 3, 1, 9), Amount = 400, Quality = ReadingQuality.Manual },
new Consumption { MeterId = truncated, Time = Local(2026, 6, 1, 9), Amount = 500, Quality = ReadingQuality.Manual });
// A first reading at exactly local midnight after an install date is booked one second before it (D-11):
// that is not lost history.
db.Readings.Add(new Reading { MeterId = midnight, Time = Local(2026, 6, 1, 0), Value = 50, Quality = ReadingQuality.Manual });
await db.SaveChangesAsync();
db.ChangeTracker.Clear();
try
{
await RecomputeAsync(midnight);
Assert.Equal(Local(2026, 6, 1, 0).AddSeconds(-1), await db.Consumption.Where(c => c.MeterId == midnight).MinAsync(c => c.Time));
// As revision 2 left it: consumption, but no analysis data yet.
await db.MeterRollupStates.Where(s => s.MeterId == midnight).ExecuteDeleteAsync();
await db.ConsumptionRollups.Where(r => r.MeterId == midnight).ExecuteDeleteAsync();
await SetSettingsAsync(db, revision: 2, zone: BerlinId);
await Upgrade(db).RunAsync();
Assert.Equal(900, await db.Consumption.Where(c => c.MeterId == truncated).SumAsync(c => c.Amount), 9);
Assert.Equal(Local(2020, 3, 1, 9), await db.Consumption.Where(c => c.MeterId == truncated).MinAsync(c => c.Time));
Assert.False(await db.MeterRollupStates.AnyAsync(s => s.MeterId == truncated));
Assert.Equal(NormalizationUpgrade.CurrentRevision, (await StateAsync(db, midnight)).Revision);
Assert.True(await db.ConsumptionRollups.AnyAsync(r => r.MeterId == midnight));
// Checked again at the next start, and skipped again — never counted as rebuilt, never pending.
Assert.Equal(0, await Upgrade(db).RunAsync());
Assert.False(await db.MeterRollupStates.AnyAsync(s => s.MeterId == truncated));
var pending = await db.AppSettings.AsNoTracking().SingleAsync(s => s.Key == NormalizationUpgrade.PendingSettingKey);
Assert.Equal("[]", pending.Value);
}
finally
{
await DeleteAsync(db, truncated, midnight);
}
}
// ---- helpers ----
private sealed record ReferenceMeters(int Haus, int Netz, int Auto, int Solar1, int Solar2, int Wasser, int OilTank, int Burner)
{
public int[] All => [Haus, Netz, Auto, Solar1, Solar2, Wasser, OilTank, Burner];
}
/// A Berlin wall-clock time as the UTC instant the database stores.
private static DateTimeOffset Local(int year, int month, int day, int hour, int minute = 0)
{
var wall = new DateTime(year, month, day, hour, minute, 0);
return new DateTimeOffset(wall, Berlin.GetUtcOffset(wall)).ToUniversalTime();
}
private static NormalizationService Normalization(MeterVaultDbContext db, string zone = BerlinId, TimeProvider? time = null) =>
new(db, NormalizationEngine.CreateDefault(),
Microsoft.Extensions.Options.Options.Create(new MeterVaultOptions { TimeZone = zone }),
time ?? new FixedTimeProvider(BuildTime));
private static NormalizationUpgrade Upgrade(MeterVaultDbContext db) =>
new(db, Normalization(db), NullLogger.Instance);
private async Task RecomputeAsync(params int[] meterIds)
{
await using var db = fx.CreateContext();
await using var tx = await db.Database.BeginTransactionAsync();
foreach (var meterId in meterIds)
{
await Normalization(db).RecomputeMeterAsync(meterId, null);
}
await db.SaveChangesAsync();
await tx.CommitAsync();
}
private static async Task CreateMeterAsync(MeterVaultDbContext db, MeterMode mode, string unit, DateOnly? installedAt = null)
{
await DatabaseSeeder.SeedAsync(db);
var type = await db.EnergyTypes.FirstAsync(t => t.Key == "water");
var meter = new Meter
{
Name = $"analysis-{Guid.NewGuid():N}",
EnergyTypeId = type.Id,
Mode = mode,
Unit = unit,
InstalledAt = installedAt,
};
db.Meters.Add(meter);
await db.SaveChangesAsync();
db.ChangeTracker.Clear();
return meter.Id;
}
private static async Task InsertReadingAsync(MeterVaultDbContext db, int meterId, DateTimeOffset time, double value) =>
await db.Database.ExecuteSqlInterpolatedAsync(
$"INSERT INTO reading (meter_id, \"time\", value, quality, flags) VALUES ({meterId}, {time.ToUniversalTime()}, {value}, {(short)ReadingQuality.Manual}, 0)");
/// Creates the eight reference meters under fresh names and imports three sheets into them, in Berlin.
private static async Task ImportReferenceSheetsAsync(MeterVaultDbContext db)
{
await DatabaseSeeder.SeedAsync(db);
var electricity = (await db.EnergyTypes.FirstAsync(t => t.Key == "electricity")).Id;
var water = (await db.EnergyTypes.FirstAsync(t => t.Key == "water")).Id;
var oil = (await db.EnergyTypes.FirstAsync(t => t.Key == "heating_oil")).Id;
var suffix = Guid.NewGuid().ToString("N");
Meter Create(string name, short type, MeterMode mode, string unit, double baseline = 0) =>
new() { Name = $"{name} {suffix}", EnergyTypeId = type, Mode = mode, Unit = unit, InitialBaseline = baseline };
var haus = Create("Haus", electricity, MeterMode.CumulativeCounter, "kWh");
var netz = Create("Netz", electricity, MeterMode.CumulativeCounter, "kWh");
var auto = Create("Auto", electricity, MeterMode.CumulativeCounter, "kWh");
var solar1 = Create("Solar 1", electricity, MeterMode.GenerationCounter, "kWh");
var solar2 = Create("Solar 2", electricity, MeterMode.GenerationCounter, "kWh");
var wasser = Create("Wasser", water, MeterMode.CumulativeCounter, "m3", baseline: 820);
var tank = Create("Öltank", oil, MeterMode.ConsumableBalance, "L");
var burner = Create("Brenner", oil, MeterMode.RuntimeCounter, "h");
db.Meters.AddRange(haus, netz, auto, solar1, solar2, wasser, tank, burner);
await db.SaveChangesAsync();
db.Tanks.Add(new Tank
{
MeterId = tank.Id,
Capacity = 7000,
Unit = "L",
Calibration = MeterConfigFactory.SerializeCalibration(new CalibrationCurve(ReferenceProfiles.OilLitresPerCm)),
});
await db.SaveChangesAsync();
var ids = new ReferenceMeterIds(haus.Id, netz.Id, auto.Id, solar1.Id, solar2.Id, wasser.Id, tank.Id, burner.Id);
var import = new ImportService(db, Normalization(db));
foreach (var (profile, file) in new[]
{
(ReferenceProfiles.Electricity(ids), Electricity),
(ReferenceProfiles.Water(ids), Water),
(ReferenceProfiles.HeatingOil(ids), Oil),
})
{
await import.CommitAsync(Stage(profile, file), file, null);
}
db.ChangeTracker.Clear();
return new ReferenceMeters(haus.Id, netz.Id, auto.Id, solar1.Id, solar2.Id, wasser.Id, tank.Id, burner.Id);
}
/// Removes meters the way the meter list does; the analysis tables go with them (cascade).
private static async Task DeleteAsync(MeterVaultDbContext db, params int[] meterIds)
{
var batches = await db.Readings.Where(r => meterIds.Contains(r.MeterId) && r.ImportBatchId != null)
.Select(r => r.ImportBatchId!.Value).Distinct().ToListAsync();
await db.Consumption.Where(c => meterIds.Contains(c.MeterId)).ExecuteDeleteAsync();
await db.Readings.Where(r => meterIds.Contains(r.MeterId)).ExecuteDeleteAsync();
await db.MeterEvents.Where(e => meterIds.Contains(e.MeterId)).ExecuteDeleteAsync();
await db.Meters.Where(m => meterIds.Contains(m.Id)).ExecuteDeleteAsync();
await db.ImportBatches.Where(b => batches.Contains(b.Id)).ExecuteDeleteAsync();
}
private static async Task SetSettingsAsync(MeterVaultDbContext db, int revision, string zone)
{
foreach (var (key, value) in new[]
{
(NormalizationUpgrade.SettingKey, revision.ToString(System.Globalization.CultureInfo.InvariantCulture)),
(NormalizationUpgrade.ZoneSettingKey, System.Text.Json.JsonSerializer.Serialize(zone)),
(NormalizationUpgrade.PendingSettingKey, "[]"),
})
{
var setting = await db.AppSettings.FirstOrDefaultAsync(s => s.Key == key);
if (setting is null)
{
db.AppSettings.Add(new AppSetting { Key = key, Value = value });
}
else
{
setting.Value = value;
}
}
await db.SaveChangesAsync();
db.ChangeTracker.Clear();
}
private static async Task StateAsync(MeterVaultDbContext db, int meterId) =>
await db.MeterRollupStates.AsNoTracking().SingleAsync(s => s.MeterId == meterId);
private static async Task> DaysAsync(MeterVaultDbContext db, int meterId) =>
(await db.ConsumptionRollups.AsNoTracking().Where(r => r.MeterId == meterId)
.OrderBy(r => r.Day).ThenBy(r => r.Kind).ToListAsync())
.Select(r => r.ToBucket()).ToList();
private static async Task> RunsAsync(MeterVaultDbContext db, int meterId) =>
(await db.MeterCoverage.AsNoTracking().Where(r => r.MeterId == meterId).OrderBy(r => r.SpanFrom).ToListAsync())
.Select(r => r.ToRun()).ToList();
/// The coverage the builder makes of a fresh normalization of the meter's stored inputs, in Berlin.
private static async Task> ExpectedCoverageAsync(MeterVaultDbContext db, int meterId)
{
var meter = await db.Meters.AsNoTracking().SingleAsync(m => m.Id == meterId);
var tank = await db.Tanks.AsNoTracking().FirstOrDefaultAsync(t => t.MeterId == meterId);
var rows = NormalizationEngine.CreateDefault().Normalize(new NormalizationContext
{
Meter = MeterConfigFactory.FromMeter(meter, tank),
Readings = await db.Readings.AsNoTracking().Where(r => r.MeterId == meterId).OrderBy(r => r.Time).ToListAsync(),
Events = await db.MeterEvents.AsNoTracking().Where(e => e.MeterId == meterId).OrderBy(e => e.Time).ToListAsync(),
TimeZone = Berlin,
});
return [.. CoverageBuilder.Build(rows, Berlin)];
}
/// What consumption itself sums to per Berlin day or month, split by quality like the rollups.
private static async Task> ConsumptionByAsync(MeterVaultDbContext db, int meterId, string unit)
{
var bucket = unit == "day"
? "(\"time\" AT TIME ZONE 'Europe/Berlin')::date"
: "date_trunc('month', \"time\" AT TIME ZONE 'Europe/Berlin')::date";
var sql = $"""
SELECT {bucket} AS start, kind,
sum(amount) AS amount,
coalesce(sum(amount) FILTER (WHERE quality = 0), 0) AS measured,
coalesce(sum(amount) FILTER (WHERE quality = 2), 0) AS manual,
coalesce(sum(amount) FILTER (WHERE quality = 3), 0) AS imported,
coalesce(sum(amount) FILTER (WHERE quality IN (1, 4)), 0) AS estimated,
count(*)::int AS rows
FROM consumption WHERE meter_id = @meterId
GROUP BY 1, 2 ORDER BY 1, 2
""";
var rows = await db.Database.GetDbConnection().QueryAsync(sql, new { meterId });
return rows.Select(r => new RollupBucket(
r.Start, (ConsumptionKind)r.Kind, r.Amount, r.Measured, r.Manual, r.Imported, r.Estimated,
r.Rows, RollupFlags.None, DateTimeOffset.MinValue)).ToList();
}
private static void AssertSameBuckets(List expected, List actual, int meterId)
{
Assert.Equal(expected.Select(b => (b.Start, b.Kind)), actual.Select(b => (b.Start, b.Kind)));
foreach (var (e, a) in expected.Zip(actual))
{
var label = $"meter {meterId}, {e.Start:yyyy-MM-dd} {e.Kind}";
AssertClose(e.Amount, a.Amount, label + " amount");
AssertClose(e.Measured, a.Measured, label + " measured");
AssertClose(e.Manual, a.Manual, label + " manual");
AssertClose(e.Imported, a.Imported, label + " imported");
AssertClose(e.Estimated, a.Estimated, label + " estimated");
Assert.True(e.Rows == a.Rows, $"{label}: {a.Rows} rows, consumption has {e.Rows}");
}
}
private static void AssertClose(double expected, double actual, string label) =>
Assert.True(Math.Abs(expected - actual) <= 1e-9 * Math.Max(1, Math.Abs(expected)), $"{label}: {actual} vs {expected}");
/// Every analysis row of the meter with its tuple version: an UPDATE or a DELETE + INSERT changes it.
private static async Task VersionsAsync(MeterVaultDbContext db, int meterId)
{
var connection = db.Database.GetDbConnection();
async Task> Query(string sql) => [.. await connection.QueryAsync(sql, new { meterId })];
return new RowVersions(
await Query("SELECT concat_ws('|', day, kind, xmin, ctid) FROM consumption_rollup WHERE meter_id = @meterId ORDER BY 1"),
await Query("SELECT concat_ws('|', month, kind, xmin, ctid) FROM consumption_rollup_month WHERE meter_id = @meterId ORDER BY 1"),
await Query("SELECT concat_ws('|', span_from, xmin, ctid) FROM meter_coverage WHERE meter_id = @meterId ORDER BY 1"),
await Query("SELECT concat_ws('|', meter_id, xmin, ctid) FROM meter_rollup_state WHERE meter_id = @meterId ORDER BY 1"));
}
private sealed record RowVersions(List Days, List Months, List Coverage, List State)
{
public bool Equals(RowVersions? other) =>
other is not null && Days.SequenceEqual(other.Days) && Months.SequenceEqual(other.Months)
&& Coverage.SequenceEqual(other.Coverage) && State.SequenceEqual(other.State);
public override int GetHashCode() => HashCode.Combine(Days.Count, Months.Count, Coverage.Count, State.Count);
}
private sealed class SqlBucket
{
public DateOnly Start { get; set; }
public short Kind { get; set; }
public double Amount { get; set; }
public double Measured { get; set; }
public double Manual { get; set; }
public double Imported { get; set; }
public double Estimated { get; set; }
public int Rows { get; set; }
}
}