using Dapper; using MeterVault.Core.Domain; using MeterVault.Core.Normalization; using MeterVault.Infrastructure.Costing; using MeterVault.Infrastructure.Ingestion; using MeterVault.Infrastructure.Normalization; using MeterVault.Infrastructure.Options; using MeterVault.Infrastructure.Persistence; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging.Abstractions; namespace MeterVault.Integration.Tests.Ingestion; /// /// Consumption between two readings shows up in the months it accrued in — checked the way the charts /// see it, bucketed by the database in the instance timezone — and stored data built under the old rule /// is rebuilt once instead of waiting for each meter's next reading. /// [Collection("Timescale")] public sealed class MonthAttributionTests(TimescaleFixture fx) { private static readonly TimeZoneInfo Berlin = TimeZoneInfo.FindSystemTimeZoneById("Europe/Berlin"); /// A Berlin wall-clock time as the UTC instant the database stores. private static DateTimeOffset BerlinTime(int year, int month, int day, int hour) => new DateTimeOffset(new DateTime(year, month, day, hour, 0, 0), Berlin.GetUtcOffset(new DateTime(year, month, day, hour, 0, 0))) .ToUniversalTime(); [Fact] public async Task A_reading_six_weeks_after_the_last_fills_both_months_it_covers() { await using var db = fx.CreateContext(); var meterId = await CreateMeterAsync(db); var ingestion = new IngestionService(db, BerlinNormalization(db)); var august1 = BerlinTime(2026, 8, 1, 9); var september16 = BerlinTime(2026, 9, 16, 18); await ingestion.IngestByMeterAsync(meterId, august1, 700, quality: ReadingQuality.Manual); await ingestion.IngestByMeterAsync(meterId, september16, 746, quality: ReadingQuality.Manual); var months = await MonthlyAsync(db, meterId); var expectedAugust = 46 * ((BerlinTime(2026, 9, 1, 0) - august1) / (september16 - august1)); // August also holds the first reading's 700, counted from the meter's zero baseline. Assert.Equal(700 + expectedAugust, months[new DateOnly(2026, 8, 1)], 6); Assert.Equal(46 - expectedAugust, months[new DateOnly(2026, 9, 1)], 6); await CleanupAsync(db, meterId); } [Fact] public async Task Stored_consumption_from_an_older_revision_is_rebuilt_once() { await using var db = fx.CreateContext(); var meterId = await CreateMeterAsync(db); db.Readings.AddRange( new Reading { MeterId = meterId, Time = BerlinTime(2026, 8, 1, 9), Value = 700, Quality = ReadingQuality.Manual }, new Reading { MeterId = meterId, Time = BerlinTime(2026, 9, 16, 18), Value = 746, Quality = ReadingQuality.Manual }); // What the previous rule stored: the whole six weeks on the September reading. db.Consumption.AddRange( new Consumption { MeterId = meterId, Time = BerlinTime(2026, 8, 1, 9), Amount = 700, Quality = ReadingQuality.Manual }, new Consumption { MeterId = meterId, Time = BerlinTime(2026, 9, 16, 18), Amount = 46, Quality = ReadingQuality.Manual }); await SetRevisionAsync(db, "1"); var upgrade = new NormalizationUpgrade(db, BerlinNormalization(db), NullLogger.Instance); var rebuilt = await upgrade.RunAsync(); Assert.True(rebuilt >= 1); var months = await MonthlyAsync(db, meterId); Assert.True(months[new DateOnly(2026, 8, 1)] > 700 + 30, "August did not get its share back."); var stored = await db.AppSettings.AsNoTracking().SingleAsync(s => s.Key == NormalizationUpgrade.SettingKey); Assert.Equal(NormalizationUpgrade.CurrentRevision.ToString(System.Globalization.CultureInfo.InvariantCulture), stored.Value); // Up to date now: the next start does nothing. Assert.Equal(0, await new NormalizationUpgrade(db, BerlinNormalization(db), NullLogger.Instance).RunAsync()); await CleanupAsync(db, meterId); } [Fact] public async Task A_meter_that_cannot_be_rebuilt_neither_stops_startup_nor_the_other_meters() { await using var db = fx.CreateContext(); var good = await CreateMeterAsync(db); var broken = await CreateMeterAsync(db); db.Readings.AddRange( new Reading { MeterId = good, Time = BerlinTime(2026, 8, 1, 9), Value = 700, Quality = ReadingQuality.Manual }, new Reading { MeterId = good, Time = BerlinTime(2026, 9, 16, 18), Value = 746, Quality = ReadingQuality.Manual }); await db.SaveChangesAsync(); // A mode this build cannot read: loading the meter throws, as any unexpected data would. await db.Database.ExecuteSqlInterpolatedAsync($"UPDATE meter SET mode = 'FromTheFuture' WHERE id = {broken}"); await SetRevisionAsync(db, "1"); try { var rebuilt = await Upgrade(db).RunAsync(); Assert.True(rebuilt >= 1); Assert.True((await MonthlyAsync(db, good)).ContainsKey(new DateOnly(2026, 8, 1)), "The healthy meter was not rebuilt."); Assert.Contains(broken, await PendingAsync(db)); // The next start retries only what failed — and once it can be read, it is rebuilt. Assert.Equal(0, await Upgrade(db).RunAsync()); await db.Database.ExecuteSqlInterpolatedAsync($"UPDATE meter SET mode = 'CumulativeCounter' WHERE id = {broken}"); Assert.Equal(1, await Upgrade(db).RunAsync()); Assert.Empty(await PendingAsync(db)); Assert.Equal(0, await Upgrade(db).RunAsync()); } finally { await db.Database.ExecuteSqlInterpolatedAsync($"UPDATE meter SET mode = 'CumulativeCounter' WHERE id = {broken}"); await CleanupAsync(db, good); await CleanupAsync(db, broken); } } [Fact] public async Task Rows_of_earlier_monthly_imports_are_marked_as_months_before_the_rebuild() { await using var db = fx.CreateContext(); var monthly = await CreateMeterAsync(db); var autoMonthly = await CreateMeterAsync(db); var daily = await CreateMeterAsync(db); var monthBatch = new ImportBatch { SourceName = "months.csv", Mapping = """{"dateKind":"MonthName"}""" }; // The wizard's default: auto-detected dates, which do not say whether a row named a month or a day. var autoMonthBatch = new ImportBatch { SourceName = "auto-months.csv", Mapping = """{"dateKind":"Auto"}""" }; var dayBatch = new ImportBatch { SourceName = "days.csv", Mapping = """{"dateKind":"Auto"}""" }; db.ImportBatches.AddRange(monthBatch, autoMonthBatch, dayBatch); await db.SaveChangesAsync(); // As an import before revision 2 stored them: nothing but the midnight stamp on the 1st. foreach (var (meter, batch) in new[] { (monthly, monthBatch.Id), (autoMonthly, autoMonthBatch.Id), (daily, dayBatch.Id) }) { db.Readings.AddRange( new Reading { MeterId = meter, Time = Utc(2026, 6, 1), Value = 100, Quality = ReadingQuality.Imported, ImportBatchId = batch }, new Reading { MeterId = meter, Time = Utc(2026, 7, 1), Value = 130, Quality = ReadingQuality.Imported, ImportBatchId = batch }); } // A reading off the 1st gives the auto-detected batch away as day-dated. db.Readings.Add(new Reading { MeterId = daily, Time = Utc(2026, 7, 15), Value = 140, Quality = ReadingQuality.Imported, ImportBatchId = dayBatch.Id }); await db.SaveChangesAsync(); db.ChangeTracker.Clear(); await SetRevisionAsync(db, "1"); try { await Upgrade(db).RunAsync(); var flags = await db.Readings.AsNoTracking() .Where(r => r.MeterId == monthly || r.MeterId == autoMonthly || r.MeterId == daily) .Select(r => new { r.MeterId, r.Flags }) .ToListAsync(); Assert.All(flags.Where(f => f.MeterId == monthly), f => Assert.True(f.Flags.HasFlag(ReadingFlags.MonthLabel))); // Every row on the 1st across months: a monthly table, as those rows were always attributed. Assert.All(flags.Where(f => f.MeterId == autoMonthly), f => Assert.True(f.Flags.HasFlag(ReadingFlags.MonthLabel))); Assert.All(flags.Where(f => f.MeterId == daily), f => Assert.False(f.Flags.HasFlag(ReadingFlags.MonthLabel))); // "Juli 2026" now books July's 30 under July, not June. Assert.Equal(30, (await MonthlyAsync(db, monthly))[new DateOnly(2026, 7, 1)], 6); Assert.Equal(30, (await MonthlyAsync(db, autoMonthly))[new DateOnly(2026, 7, 1)], 6); } finally { await CleanupAsync(db, monthly); await CleanupAsync(db, autoMonthly); await CleanupAsync(db, daily); await db.ImportBatches.Where(b => b.Id == monthBatch.Id || b.Id == autoMonthBatch.Id || b.Id == dayBatch.Id).ExecuteDeleteAsync(); } } [Fact] public async Task Costs_are_bucketed_in_the_zone_months_are_divided_in() { // London is an hour behind Berlin: August's share is stamped at 23:59:59 London time, which is // already September in Berlin. Bucketed in a hard-coded Berlin, the fix would not show. await using var db = fx.CreateContext(); var meterId = await CreateMeterAsync(db); var london = Microsoft.Extensions.Options.Options.Create(new MeterVaultOptions { TimeZone = "Europe/London" }); var ingestion = new IngestionService(db, new NormalizationService(db, NormalizationEngine.CreateDefault(), london)); await ingestion.IngestByMeterAsync(meterId, Utc(2026, 8, 1).AddHours(8), 700, quality: ReadingQuality.Manual); await ingestion.IngestByMeterAsync(meterId, Utc(2026, 9, 16).AddHours(17), 746, quality: ReadingQuality.Manual); try { // A clock after the range (D-01): the service stops actuals at now, and the September reading must be past. var costs = await new CostService(fx, london, new FixedTimeProvider(Utc(2026, 10, 1))).GetMeterCostsAsync(meterId, Utc(2026, 7, 1), Utc(2026, 10, 1)); var august = Assert.Single(costs, c => c.Period == new DateOnly(2026, 8, 1)); var september = Assert.Single(costs, c => c.Period == new DateOnly(2026, 9, 1)); Assert.True(august.Consumption > 730, $"August holds {august.Consumption}"); Assert.True(september.Consumption < 16, $"September holds {september.Consumption}"); } finally { await CleanupAsync(db, meterId); } } [Fact] public async Task A_reading_inside_an_imported_month_is_judged_against_the_month_before_it() { // "August 2026" = 731 is stamped on 1 August but is the register on 31 August. A photo of the meter // from 20 August showing 720 is not a drop from 731. await using var db = fx.CreateContext(); var meterId = await CreateMeterAsync(db); db.Readings.AddRange( new Reading { MeterId = meterId, Time = Utc(2026, 7, 1), Value = 699, Quality = ReadingQuality.Imported, Flags = ReadingFlags.MonthLabel }, new Reading { MeterId = meterId, Time = Utc(2026, 8, 1), Value = 731, Quality = ReadingQuality.Imported, Flags = ReadingFlags.MonthLabel }); await db.SaveChangesAsync(); db.ChangeTracker.Clear(); var ingestion = new IngestionService(db, BerlinNormalization(db)); try { Assert.Equal(IngestionOutcome.Written, await ingestion.IngestByMeterAsync(meterId, BerlinTime(2026, 8, 20, 9), 720, quality: ReadingQuality.Manual)); // ...while a real drop below July's register still is one. Assert.Equal(IngestionOutcome.RejectedDecrease, await ingestion.IngestByMeterAsync(meterId, BerlinTime(2026, 8, 21, 9), 650, quality: ReadingQuality.Manual)); var months = await MonthlyAsync(db, meterId); Assert.Equal(32, months[new DateOnly(2026, 8, 1)], 6); } finally { await CleanupAsync(db, meterId); } } [Fact] public async Task A_live_value_written_onto_a_month_row_turns_it_into_a_reading_at_that_instant() { await using var db = fx.CreateContext(); var meterId = await CreateMeterAsync(db); db.Readings.AddRange( new Reading { MeterId = meterId, Time = Utc(2026, 7, 1), Value = 699, Quality = ReadingQuality.Imported, Flags = ReadingFlags.MonthLabel }, new Reading { MeterId = meterId, Time = Utc(2026, 8, 1), Value = 731, Quality = ReadingQuality.Imported, Flags = ReadingFlags.MonthLabel }); await db.SaveChangesAsync(); db.ChangeTracker.Clear(); var ingestion = new IngestionService(db, BerlinNormalization(db)); try { // A hand correction keeps the month row a month row... await ingestion.IngestByMeterAsync(meterId, Utc(2026, 7, 1), 700, quality: ReadingQuality.Manual); // ...an API/HA value at that instant is what the register showed then. await ingestion.IngestByMeterAsync(meterId, Utc(2026, 8, 1), 702); var rows = await db.Readings.AsNoTracking().Where(r => r.MeterId == meterId).OrderBy(r => r.Time).ToListAsync(); Assert.True(rows[0].Flags.HasFlag(ReadingFlags.MonthLabel)); Assert.False(rows[1].Flags.HasFlag(ReadingFlags.MonthLabel)); Assert.Equal(ReadingQuality.Measured, rows[1].Quality); } finally { await CleanupAsync(db, meterId); } } [Fact] public async Task Behind_utc_a_period_starts_at_local_midnight_so_month_end_shares_stay_in_it() { // New York: December's share of a 10 December to 20 January interval is stamped at 23:59:59 on // 31 December local, which is already 1 January in UTC. A year requested as UTC midnights lost it. await using var db = fx.CreateContext(); var newYork = Microsoft.Extensions.Options.Options.Create(new MeterVaultOptions { TimeZone = "America/New_York" }); var type = new EnergyType { Key = $"flow-{Guid.NewGuid():N}", DisplayName = "Flow test", BaseUnit = "m3", DefaultMode = MeterMode.CumulativeCounter }; db.EnergyTypes.Add(type); await db.SaveChangesAsync(); var meter = new Meter { Name = $"ny-{Guid.NewGuid():N}", EnergyTypeId = type.Id, Mode = MeterMode.CumulativeCounter, Unit = "m3" }; db.Meters.Add(meter); await db.SaveChangesAsync(); var ingestion = new IngestionService(db, new NormalizationService(db, NormalizationEngine.CreateDefault(), newYork)); await ingestion.IngestByMeterAsync(meter.Id, new DateTimeOffset(2026, 12, 10, 17, 0, 0, TimeSpan.Zero), 100, quality: ReadingQuality.Manual); await ingestion.IngestByMeterAsync(meter.Id, new DateTimeOffset(2027, 1, 20, 17, 0, 0, TimeSpan.Zero), 141, quality: ReadingQuality.Manual); try { // Read once both readings lie in the past: actuals stop at now (D-04), and both are after today. var later = new FixedTimeProvider(new DateTimeOffset(2027, 6, 1, 0, 0, 0, TimeSpan.Zero)); var december = await new MeterVault.Infrastructure.Dashboard.FlowService(fx, newYork, later) .GetFlowAsync(type.Id, new DateOnly(2026, 12, 1), new DateOnly(2027, 1, 1)); var node = Assert.Single(december.Nodes, n => n.MeterId == meter.Id); // The first reading's 100 plus December's 21.5 of the 41 days' 41 m3 (noon on the 10th to midnight). Assert.Equal(121.5, node.Value, 3); } finally { await CleanupAsync(db, meter.Id); await db.EnergyTypes.Where(t => t.Id == type.Id).ExecuteDeleteAsync(); } } private static DateTimeOffset Utc(int year, int month, int day) => new(year, month, day, 0, 0, 0, TimeSpan.Zero); private static NormalizationUpgrade Upgrade(MeterVaultDbContext db) => new(db, BerlinNormalization(db), NullLogger.Instance); private static async Task PendingAsync(MeterVaultDbContext db) { var setting = await db.AppSettings.AsNoTracking().FirstOrDefaultAsync(s => s.Key == NormalizationUpgrade.PendingSettingKey); return setting is null ? [] : System.Text.Json.JsonSerializer.Deserialize(setting.Value) ?? []; } private static NormalizationService BerlinNormalization(MeterVaultDbContext db) => new(db, NormalizationEngine.CreateDefault(), Microsoft.Extensions.Options.Options.Create(new MeterVaultOptions { TimeZone = "Europe/Berlin" })); private static async Task> MonthlyAsync(MeterVaultDbContext db, int meterId) { const string sql = "SELECT (time_bucket(INTERVAL '1 month', \"time\", 'Europe/Berlin') AT TIME ZONE 'Europe/Berlin')::date AS period, " + "sum(amount) AS amount FROM consumption WHERE meter_id = @meterId GROUP BY period ORDER BY period"; var rows = await db.Database.GetDbConnection().QueryAsync<(DateOnly Period, double Amount)>(sql, new { meterId }); return rows.ToDictionary(r => r.Period, r => r.Amount); } private static async Task SetRevisionAsync(MeterVaultDbContext db, string revision) { var setting = await db.AppSettings.FirstOrDefaultAsync(s => s.Key == NormalizationUpgrade.SettingKey); if (setting is null) { db.AppSettings.Add(new AppSetting { Key = NormalizationUpgrade.SettingKey, Value = revision }); } else { setting.Value = revision; } await db.SaveChangesAsync(); db.ChangeTracker.Clear(); } private static async Task CreateMeterAsync(MeterVaultDbContext db) { await DatabaseSeeder.SeedAsync(db); var type = await db.EnergyTypes.FirstAsync(t => t.Key == "water"); var meter = new Meter { Name = $"months-{Guid.NewGuid():N}", EnergyTypeId = type.Id, Mode = MeterMode.CumulativeCounter, Unit = "m3" }; db.Meters.Add(meter); await db.SaveChangesAsync(); return meter.Id; } private static async Task CleanupAsync(MeterVaultDbContext db, int meterId) { await db.Readings.Where(r => r.MeterId == meterId).ExecuteDeleteAsync(); await db.Consumption.Where(c => c.MeterId == meterId).ExecuteDeleteAsync(); await db.Meters.Where(m => m.Id == meterId).ExecuteDeleteAsync(); } }