using MeterVault.Core.Domain;
using MeterVault.Core.Normalization;
using MeterVault.Infrastructure.Ingestion;
using MeterVault.Infrastructure.Normalization;
using MeterVault.Infrastructure.Persistence;
using Microsoft.EntityFrameworkCore;
namespace MeterVault.Integration.Tests.Ingestion;
///
/// Recording meter events from the meter page: what a swap, reset, tank level, delivery or note
/// actually persists, that the derived consumption follows in the same step, and that a mistake can
/// be taken back without leaving the series worse than before.
///
[Collection("Timescale")]
public sealed class MeterEventServiceTests(TimescaleFixture fx)
{
private static readonly DateTimeOffset Yesterday = new(2026, 9, 16, 18, 0, 0, TimeSpan.Zero);
private static readonly DateTimeOffset SwapAt = new(2026, 9, 17, 9, 30, 0, TimeSpan.Zero);
[Fact]
public async Task A_swap_recorded_today_books_the_old_tail_and_lets_the_new_register_count_on()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
var ingestion = NewIngestion(db);
await ingestion.IngestByMeterAsync(meterId, Yesterday.AddDays(-30), 848, quality: ReadingQuality.Manual);
await ingestion.IngestByMeterAsync(meterId, Yesterday, 861, quality: ReadingQuality.Manual);
var result = await NewService(db).RecordAsync(meterId,
new MeterEventDraft(MeterEventType.MeterSwap, SwapAt) { PrevValue = 873, NewValue = 2 });
Assert.True(result.Succeeded, result.Problem.ToString());
// The event carries both registers; the new register's start is a real, flagged manual reading.
var swap = await db.MeterEvents.AsNoTracking().SingleAsync(e => e.MeterId == meterId);
Assert.Equal((873d, 2d, "m3"), (swap.PrevValue!.Value, swap.NewValue!.Value, swap.Unit));
var start = await db.Readings.AsNoTracking().SingleAsync(r => r.MeterId == meterId && r.Time == SwapAt);
Assert.Equal(2d, start.Value, 9);
Assert.Equal(ReadingQuality.Manual, start.Quality);
Assert.True(start.Flags.HasFlag(ReadingFlags.MeterSwap));
// The old meter's last 12 m³ land at the swap — not a −859 anomaly, not an 871 spike.
var atSwap = await db.Consumption.AsNoTracking().SingleAsync(c => c.MeterId == meterId && c.Time == SwapAt);
Assert.Equal(12d, atSwap.Amount, 9);
// A reading of the new register is accepted and counts from its start value.
var next = await ingestion.IngestByMeterAsync(meterId, SwapAt.AddHours(8), 2.4, quality: ReadingQuality.Manual);
Assert.Equal(IngestionOutcome.Written, next);
var afterSwap = await db.Consumption.AsNoTracking().SingleAsync(c => c.MeterId == meterId && c.Time == SwapAt.AddHours(8));
Assert.Equal(0.4, afterSwap.Amount, 9);
await CleanupAsync(db, meterId);
}
[Fact]
public async Task A_new_reading_typed_at_the_swap_instant_replaces_the_start_value_and_still_reconciles()
{
// The reading dialog's "meter swapped?" hand-off: the user records the swap at the time they
// were typing a reading, then saves that reading at the same instant.
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
var ingestion = NewIngestion(db);
await ingestion.IngestByMeterAsync(meterId, Yesterday, 861, quality: ReadingQuality.Manual);
await NewService(db).RecordAsync(meterId, new MeterEventDraft(MeterEventType.MeterSwap, SwapAt) { PrevValue = 873, NewValue = 0 });
var outcome = await ingestion.IngestByMeterAsync(meterId, SwapAt, 0.3, quality: ReadingQuality.Manual);
Assert.Equal(IngestionOutcome.Updated, outcome);
var atSwap = await db.Consumption.AsNoTracking().SingleAsync(c => c.MeterId == meterId && c.Time == SwapAt);
Assert.Equal(12.3, atSwap.Amount, 9); // 873 − 861, plus 0.3 on the new register
await CleanupAsync(db, meterId);
}
[Fact]
public async Task A_swap_is_refused_when_its_numbers_or_its_instant_cannot_be_right()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
var ingestion = NewIngestion(db);
await ingestion.IngestByMeterAsync(meterId, Yesterday, 861, quality: ReadingQuality.Manual);
var service = NewService(db);
// The old register cannot end below a reading already taken from it.
var below = await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.MeterSwap, SwapAt) { PrevValue = 850, NewValue = 0 });
Assert.Equal(MeterEventProblem.OldRegisterBelowPreviousReading, below.Problem);
// A reading already sits at that instant: the start reading would silently overwrite it.
var sameTime = await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.MeterSwap, Yesterday) { PrevValue = 861, NewValue = 0 });
Assert.Equal(MeterEventProblem.ReadingAtSameTime, sameTime.Problem);
// A tank event means nothing on a register.
var delivery = await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.Delivery, SwapAt) { Amount = 100 });
Assert.Equal(MeterEventProblem.NotRecordableForMode, delivery.Problem);
Assert.False(await db.MeterEvents.AnyAsync(e => e.MeterId == meterId));
// A second swap between the same two readings would never be applied.
Assert.True((await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.MeterSwap, SwapAt) { PrevValue = 873, NewValue = 0 })).Succeeded);
var context = await service.GetContextAsync(meterId, SwapAt.AddMinutes(-10));
Assert.True(context!.BoundaryInWindow);
var duplicate = await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.MeterSwap, SwapAt.AddMinutes(-10)) { PrevValue = 873, NewValue = 0 });
Assert.Equal(MeterEventProblem.BoundaryAlreadyRecorded, duplicate.Problem);
// ...but a genuine later swap, after the new register has been read, is fine.
await ingestion.IngestByMeterAsync(meterId, SwapAt.AddDays(10), 5, quality: ReadingQuality.Manual);
var later = await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.MeterSwap, SwapAt.AddDays(20)) { PrevValue = 9, NewValue = 0 });
Assert.True(later.Succeeded, later.Problem.ToString());
await CleanupAsync(db, meterId);
}
[Fact]
public async Task A_reset_with_the_last_register_value_keeps_the_stretch_before_it()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "kWh");
var ingestion = NewIngestion(db);
await ingestion.IngestByMeterAsync(meterId, Yesterday, 99_990, quality: ReadingQuality.Manual);
var result = await NewService(db).RecordAsync(meterId,
new MeterEventDraft(MeterEventType.CounterReset, SwapAt) { PrevValue = 99_999, NewValue = 0 });
Assert.True(result.Succeeded, result.Problem.ToString());
var start = await db.Readings.AsNoTracking().SingleAsync(r => r.MeterId == meterId && r.Time == SwapAt);
Assert.True(start.Flags.HasFlag(ReadingFlags.CounterReset));
var atReset = await db.Consumption.AsNoTracking().SingleAsync(c => c.MeterId == meterId && c.Time == SwapAt);
Assert.Equal(9d, atReset.Amount, 9);
await CleanupAsync(db, meterId);
}
[Fact]
public async Task Tank_levels_and_deliveries_drive_the_tank_and_centimetres_need_a_calibration()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.ConsumableBalance, "L");
var service = NewService(db);
// No tank yet: a dipstick reading in cm cannot be turned into litres.
var uncalibrated = await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.TankLevel, Yesterday) { Amount = 50, Unit = "cm" });
Assert.Equal(MeterEventProblem.LevelNeedsCalibration, uncalibrated.Problem);
db.Tanks.Add(new Tank
{
MeterId = meterId,
Capacity = 7000,
Unit = "L",
Calibration = MeterConfigFactory.SerializeCalibration(new CalibrationCurve(7000d / 150d)),
});
await db.SaveChangesAsync();
Assert.True((await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.TankLevel, Yesterday.AddDays(-60)) { Amount = 50, Unit = "cm" })).Succeeded);
Assert.Equal(MeterEventProblem.AmountOutOfRange,
(await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.Delivery, Yesterday.AddDays(-30)) { Amount = 0 })).Problem);
Assert.True((await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.Delivery, Yesterday.AddDays(-30)) { Amount = 2000 })).Succeeded);
var context = await service.GetContextAsync(meterId, Yesterday);
Assert.Equal(2333.33, context!.LastLevel!.Volume, 1);
Assert.Equal(2000d, context.DeliveredSinceLastLevel, 9);
Assert.Equal(600d, context.UsedSinceLastLevel(context.ToVolume(80, centimetres: true))!.Value, 1);
Assert.True((await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.TankLevel, Yesterday) { Amount = 80, Unit = "cm" })).Succeeded);
// 2333.3 L + 2000 L delivered − 3733.3 L now = 600 L drawn, booked at the new level.
var drawn = await db.Consumption.AsNoTracking().SingleAsync(c => c.MeterId == meterId);
Assert.Equal(600d, drawn.Amount, 1);
Assert.Equal(Yesterday, drawn.Time);
var delivery = await db.MeterEvents.AsNoTracking().SingleAsync(e => e.MeterId == meterId && e.EventType == MeterEventType.Delivery);
Assert.Equal("L", delivery.Unit);
await db.Tanks.Where(t => t.MeterId == meterId).ExecuteDeleteAsync();
await CleanupAsync(db, meterId);
}
[Fact]
public async Task A_note_needs_text_and_is_offered_on_every_mode()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.InstantRate, "W");
var service = NewService(db);
Assert.Equal(MeterEventProblem.NoteRequired,
(await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.Note, SwapAt) { Notes = " " })).Problem);
Assert.Equal(MeterEventProblem.NotRecordableForMode,
(await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.MeterSwap, SwapAt) { NewValue = 0 })).Problem);
var note = await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.Note, SwapAt) { Notes = " Sensor moved to the new fuse box " });
Assert.True(note.Succeeded);
Assert.Equal("Sensor moved to the new fuse box", (await db.MeterEvents.AsNoTracking().SingleAsync(e => e.Id == note.EventId)).Notes);
await CleanupAsync(db, meterId);
}
[Fact]
public async Task Deleting_a_swap_takes_its_start_reading_along_and_restores_the_series()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
var ingestion = NewIngestion(db);
await ingestion.IngestByMeterAsync(meterId, Yesterday, 861, quality: ReadingQuality.Manual);
var service = NewService(db);
var before = await ConsumptionAsync(db, meterId);
var swap = await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.MeterSwap, SwapAt) { PrevValue = 875, NewValue = 0 });
var deleted = await service.DeleteEventAsync(meterId, swap.EventId!.Value);
Assert.True(deleted.Succeeded);
Assert.False(await db.MeterEvents.AnyAsync(e => e.MeterId == meterId));
Assert.False(await db.Readings.AnyAsync(r => r.MeterId == meterId && r.Time == SwapAt));
Assert.Equal(before, await ConsumptionAsync(db, meterId));
// Re-recording it correctly works straight away — nothing was left behind to trip over.
Assert.True((await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.MeterSwap, SwapAt) { PrevValue = 873, NewValue = 0 })).Succeeded);
await CleanupAsync(db, meterId);
}
[Fact]
public async Task Deleting_a_swap_keeps_a_real_reading_later_typed_at_the_same_instant()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
var ingestion = NewIngestion(db);
await ingestion.IngestByMeterAsync(meterId, Yesterday, 861, quality: ReadingQuality.Manual);
var service = NewService(db);
var swap = await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.MeterSwap, SwapAt) { PrevValue = 873, NewValue = 0 });
await ingestion.IngestByMeterAsync(meterId, SwapAt, 0.3, quality: ReadingQuality.Manual);
await service.DeleteEventAsync(meterId, swap.EventId!.Value);
Assert.True(await db.Readings.AnyAsync(r => r.MeterId == meterId && r.Time == SwapAt));
await CleanupAsync(db, meterId);
}
[Fact]
public async Task Imported_events_and_non_manual_readings_are_not_deletable_here()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
var batch = new ImportBatch { SourceName = "test.csv", CreatedAt = DateTimeOffset.UtcNow };
db.ImportBatches.Add(batch);
await db.SaveChangesAsync();
var imported = new MeterEvent { MeterId = meterId, Time = Yesterday, EventType = MeterEventType.MeterSwap, PrevValue = 1, NewValue = 0, ImportBatchId = batch.Id };
db.MeterEvents.Add(imported);
db.Readings.Add(new Reading { MeterId = meterId, Time = SwapAt, Value = 5, Quality = ReadingQuality.Measured });
await db.SaveChangesAsync();
var service = NewService(db);
Assert.Equal(MeterEventProblem.Imported, (await service.DeleteEventAsync(meterId, imported.Id)).Problem);
Assert.Equal(MeterEventProblem.NotManual, (await service.DeleteManualReadingAsync(meterId, SwapAt)).Problem);
Assert.Equal(MeterEventProblem.NotFound, (await service.DeleteEventAsync(meterId + 100_000, imported.Id)).Problem);
await db.MeterEvents.Where(e => e.MeterId == meterId).ExecuteDeleteAsync();
await db.ImportBatches.Where(b => b.Id == batch.Id).ExecuteDeleteAsync();
await CleanupAsync(db, meterId);
}
[Fact]
public async Task A_mistyped_manual_reading_can_be_deleted_and_the_next_one_is_accepted_again()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
var ingestion = NewIngestion(db);
await ingestion.IngestByMeterAsync(meterId, Yesterday, 1873.4, quality: ReadingQuality.Manual);
await ingestion.IngestByMeterAsync(meterId, Yesterday.AddHours(1), 18734, quality: ReadingQuality.Manual); // typo
Assert.Equal(IngestionOutcome.RejectedDecrease,
await ingestion.IngestByMeterAsync(meterId, Yesterday.AddHours(2), 1874, quality: ReadingQuality.Manual));
var deleted = await NewService(db).DeleteManualReadingAsync(meterId, Yesterday.AddHours(1));
Assert.True(deleted.Succeeded);
Assert.DoesNotContain(await ConsumptionAsync(db, meterId), c => c > 10_000);
Assert.Equal(IngestionOutcome.Written,
await ingestion.IngestByMeterAsync(meterId, Yesterday.AddHours(2), 1874, quality: ReadingQuality.Manual));
await CleanupAsync(db, meterId);
}
[Fact]
public async Task The_context_warns_about_readings_after_a_backdated_event_and_live_sources()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
var ingestion = NewIngestion(db);
await ingestion.IngestByMeterAsync(meterId, Yesterday, 861, quality: ReadingQuality.Manual);
await ingestion.IngestByMeterAsync(meterId, SwapAt.AddHours(1), 862, quality: ReadingQuality.Manual);
await ingestion.IngestByMeterAsync(meterId, SwapAt.AddHours(2), 863, quality: ReadingQuality.Manual);
db.MeterSources.Add(new MeterSource { MeterId = meterId, SourceType = SourceType.HomeAssistant, IsEnabled = true });
db.MeterSources.Add(new MeterSource { MeterId = meterId, SourceType = SourceType.Manual, IsEnabled = true });
await db.SaveChangesAsync();
var context = await NewService(db).GetContextAsync(meterId, SwapAt);
Assert.NotNull(context);
Assert.Equal(new ReadingPoint(Yesterday, 861), context.Previous);
Assert.Equal(SwapAt.AddHours(1), context.Next!.Time);
Assert.Equal(2, context.ReadingsAfter);
Assert.False(context.ReadingAtTime);
Assert.Equal(1, context.LiveSources); // manual sources do not keep feeding the old register
Assert.Equal(12d, context.Tail(873)!.Value, 9);
await db.MeterSources.Where(s => s.MeterId == meterId).ExecuteDeleteAsync();
await CleanupAsync(db, meterId);
}
[Fact]
public async Task A_second_tank_level_at_the_same_instant_is_refused_instead_of_crashing_the_recompute()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.ConsumableBalance, "L");
var service = NewService(db);
Assert.True((await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.TankLevel, Yesterday.AddDays(-30)) { Amount = 3000 })).Succeeded);
Assert.True((await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.TankLevel, SwapAt) { Amount = 2500 })).Succeeded);
// Re-entered within the same minute to fix a typo: two levels at one instant cannot both book.
var again = await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.TankLevel, SwapAt) { Amount = 2400 });
Assert.Equal(MeterEventProblem.LevelAtSameTime, again.Problem);
Assert.Equal(500d, (await db.Consumption.AsNoTracking().SingleAsync(c => c.MeterId == meterId)).Amount, 9);
await CleanupAsync(db, meterId);
}
[Fact]
public async Task A_boundary_cannot_be_removed_while_a_later_swap_was_measured_against_it()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
var ingestion = NewIngestion(db);
await ingestion.IngestByMeterAsync(meterId, Yesterday, 861, quality: ReadingQuality.Manual);
var service = NewService(db);
// A reset, then — a minute later, with no reading in between — a swap whose old register is
// measured from the reset's start reading of 0.
var reset = await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.CounterReset, SwapAt) { PrevValue = 873, NewValue = 0 });
var swap = await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.MeterSwap, SwapAt.AddMinutes(1)) { PrevValue = 0, NewValue = 0 });
Assert.True(reset.Succeeded && swap.Succeeded);
var before = await ConsumptionAsync(db, meterId);
// Removing the reset (or just its start reading) would re-measure that swap from 861: −861 m³.
Assert.Equal(MeterEventProblem.LaterBoundaryDependsOnIt, (await service.DeleteEventAsync(meterId, reset.EventId!.Value)).Problem);
Assert.Equal(MeterEventProblem.LaterBoundaryDependsOnIt, (await service.DeleteManualReadingAsync(meterId, SwapAt)).Problem);
Assert.Equal(before, await ConsumptionAsync(db, meterId));
Assert.DoesNotContain(before, c => c < 0);
// Later first, then earlier: both go, and nothing negative is ever booked on the way.
Assert.True((await service.DeleteEventAsync(meterId, swap.EventId!.Value)).Succeeded);
Assert.True((await service.DeleteEventAsync(meterId, reset.EventId!.Value)).Succeeded);
Assert.DoesNotContain(await ConsumptionAsync(db, meterId), c => c < 0);
await CleanupAsync(db, meterId);
}
[Fact]
public async Task A_typo_cannot_be_deleted_out_from_under_a_swap_recorded_after_it()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3");
var ingestion = NewIngestion(db);
await ingestion.IngestByMeterAsync(meterId, Yesterday, 1873, quality: ReadingQuality.Manual);
await ingestion.IngestByMeterAsync(meterId, Yesterday.AddHours(1), 18734, quality: ReadingQuality.Manual); // typo
var service = NewService(db);
var swap = await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.MeterSwap, SwapAt) { PrevValue = 18734, NewValue = 0 });
Assert.True(swap.Succeeded);
Assert.Equal(MeterEventProblem.LaterBoundaryDependsOnIt,
(await service.DeleteManualReadingAsync(meterId, Yesterday.AddHours(1))).Problem);
await CleanupAsync(db, meterId);
}
[Fact]
public async Task Without_an_earlier_reading_the_old_register_is_measured_from_the_baseline()
{
await using var db = fx.CreateContext();
var meterId = await CreateMeterAsync(db, MeterMode.CumulativeCounter, "m3", initialBaseline: 900);
var service = NewService(db);
var context = await service.GetContextAsync(meterId, SwapAt);
Assert.Null(context!.Previous);
Assert.Equal(50d, context.Tail(950)!.Value, 9);
// Below the baseline would book a negative tail, exactly as below a reading would.
Assert.Equal(MeterEventProblem.OldRegisterBelowPreviousReading,
(await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.MeterSwap, SwapAt) { PrevValue = 873, NewValue = 0 })).Problem);
Assert.True((await service.RecordAsync(meterId, new MeterEventDraft(MeterEventType.MeterSwap, SwapAt) { PrevValue = 950, NewValue = 0 })).Succeeded);
Assert.Equal([50d], await ConsumptionAsync(db, meterId));
await CleanupAsync(db, meterId);
}
private static MeterEventService NewService(MeterVaultDbContext db)
{
var normalization = new NormalizationService(db, NormalizationEngine.CreateDefault());
return new MeterEventService(db, new IngestionService(db, normalization), normalization);
}
private static IngestionService NewIngestion(MeterVaultDbContext db) =>
new(db, new NormalizationService(db, NormalizationEngine.CreateDefault()));
private static async Task> ConsumptionAsync(MeterVaultDbContext db, int meterId) =>
await db.Consumption.AsNoTracking().Where(c => c.MeterId == meterId).OrderBy(c => c.Time).Select(c => c.Amount).ToListAsync();
private static async Task CreateMeterAsync(MeterVaultDbContext db, MeterMode mode, string unit, double initialBaseline = 0)
{
await DatabaseSeeder.SeedAsync(db);
var type = await db.EnergyTypes.FirstAsync(t => t.Key == "water");
var meter = new Meter { Name = $"events-{Guid.NewGuid():N}", EnergyTypeId = type.Id, Mode = mode, Unit = unit, InitialBaseline = initialBaseline };
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.MeterEvents.Where(e => e.MeterId == meterId).ExecuteDeleteAsync();
await db.Consumption.Where(c => c.MeterId == meterId).ExecuteDeleteAsync();
await db.Meters.Where(m => m.Id == meterId).ExecuteDeleteAsync();
}
}