using System.Diagnostics; using System.Globalization; using System.Text.Json; using MeterVault.Core.Normalization; using MeterVault.Infrastructure.Normalization; using MeterVault.Infrastructure.Options; using MeterVault.Infrastructure.Persistence; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging; using Npgsql; using Testcontainers.PostgreSql; using Xunit.Abstractions; namespace MeterVault.Integration.Tests.Performance; /// How the measured database was prepared: loaded now or reused, and what each step took. internal sealed record PreparedDatabase( DatasetManifest Manifest, bool Loaded, RebuildStats? Rebuild, double? CompressSeconds, int CompressedChunks); /// The startup rebuild (NormalizationUpgrade) over the whole dataset. internal sealed record RebuildStats(double Seconds, int Meters, int Rebuilt, bool MeasuredInThisRun); /// /// The database of one performance run: a fresh TimescaleDB container (the image the fixture pins), or an existing /// database from METERVAULT_PERF_DB. Prepares it the way a real instance gets there — migrations, the default /// seed, the raw dataset, the startup rebuild (), and the compression policy's work /// on raw chunks older than 30 days — and hands out contexts for the readers. /// internal sealed class PerfDatabase : IDbContextFactory, IAsyncDisposable { /// The pinned image of . public const string Image = "timescale/timescaledb:2.17.2-pg16"; /// The frozen "now" the measured dataset ends at: 19 September 2026, 14:37 Berlin, as the cost tests use. public static readonly DateTimeOffset FrozenNow = new DateTimeOffset(2026, 9, 19, 14, 37, 0, TimeSpan.FromHours(2)).ToUniversalTime(); private const string RebuildKey = "perf_rebuild"; private readonly PostgreSqlContainer? _container; private PerfDatabase(PostgreSqlContainer? container, string connectionString) { _container = container; // Long statements are allowed: a 1,000-meter read on a busy machine must be measured, not time out. ConnectionString = new NpgsqlConnectionStringBuilder(connectionString) { CommandTimeout = 600, IncludeErrorDetail = true }.ConnectionString; } public string ConnectionString { get; } /// Where the database came from, for the report. public string Origin => _container is null ? "existing database (METERVAULT_PERF_DB)" : $"fresh container ({Image})"; public static async Task OpenAsync(PerfSettings settings, PerfLog log) { if (settings.Database is { } external) { log.Write("Using the database from METERVAULT_PERF_DB"); return new PerfDatabase(null, external); } log.Write($"Starting a {Image} container…"); var container = new PostgreSqlBuilder(Image) .WithDatabase("metervault") .WithUsername("metervault") .WithPassword("metervault") .Build(); await container.StartAsync(); return new PerfDatabase(container, container.GetConnectionString()); } public MeterVaultDbContext CreateDbContext() { var options = new DbContextOptionsBuilder() .UseNpgsql(ConnectionString, npgsql => npgsql.MigrationsAssembly(typeof(MeterVaultDbContext).Assembly.FullName)) .UseSnakeCaseNamingConvention() .Options; return new MeterVaultDbContext(options); } /// /// Migrates and seeds, loads the dataset unless the database already holds it, rebuilds what is not built yet, and /// compresses raw chunks older than 30 days (what the compression policy does on a running instance). /// public async Task PrepareAsync(PerfSettings settings, PerfLog log) { await using (var db = CreateDbContext()) { db.Database.SetCommandTimeout(TimeSpan.FromMinutes(10)); await db.Database.MigrateAsync(); await DatabaseSeeder.SeedAsync(db); // As in TimescaleFixture: the scheduled compression job must not race the load and rebuild. Compression // is applied explicitly below, once, the way the policy would have left the chunks. await db.Database.ExecuteSqlRawAsync( "SELECT alter_job(job_id, scheduled => false) FROM timescaledb_information.jobs " + "WHERE proc_name IN ('policy_compression', 'policy_columnstore');"); } await using var connection = new NpgsqlConnection(ConnectionString); await connection.OpenAsync(); var manifest = await SyntheticDataset.ReadMarkerAsync(connection); var loaded = false; if (manifest is null) { var zone = TimeZoneInfo.FindSystemTimeZoneById("Europe/Berlin"); log.Write(string.Create(CultureInfo.InvariantCulture, $"Loading the synthetic dataset (scale {settings.Scale}, now {FrozenNow:O})…")); manifest = await SyntheticDataset.LoadAsync(ConnectionString, FrozenNow, zone, settings.Scale, log.Write); loaded = true; } else { log.Write(string.Create(CultureInfo.InvariantCulture, $"Reusing the loaded dataset ({manifest.MeterCount} meters, now {manifest.Now:O})")); } var rebuild = await RebuildAsync(connection, manifest, log); var (compressSeconds, compressed) = await CompressAsync(connection, manifest.Now, log); if (loaded || rebuild?.MeasuredInThisRun == true || compressed > 0) { log.Write("VACUUM ANALYZE…"); await ExecuteAsync(connection, "VACUUM ANALYZE"); } return new PreparedDatabase(manifest, loaded, rebuild, compressSeconds, compressed); } /// /// Runs the startup rebuild when stored consumption is missing or outdated — timed, with a frozen clock at the /// dataset's "now" — and remembers its duration in the database, so a later reuse can still report it. /// private async Task RebuildAsync(NpgsqlConnection connection, DatasetManifest manifest, PerfLog log) { var meters = Convert.ToInt32(await ScalarAsync(connection, "SELECT count(*) FROM meter"), CultureInfo.InvariantCulture); var built = Convert.ToInt32( await ScalarAsync(connection, $"SELECT count(*) FROM meter_rollup_state WHERE revision >= {NormalizationUpgrade.CurrentRevision}"), CultureInfo.InvariantCulture); var revision = await ScalarAsync(connection, $"SELECT value::text FROM app_setting WHERE key = '{NormalizationUpgrade.SettingKey}'"); if (built >= meters && revision is not null) { return await ScalarAsync(connection, $"SELECT value::text FROM app_setting WHERE key = '{RebuildKey}'") is string json ? JsonSerializer.Deserialize(json)! with { MeasuredInThisRun = false } : null; } log.Write(string.Create(CultureInfo.InvariantCulture, $"Rebuilding {meters} meters through NormalizationUpgrade…")); var watch = Stopwatch.StartNew(); int rebuilt; await using (var db = CreateDbContext()) { var normalization = Normalization(db, manifest); var upgrade = new NormalizationUpgrade(db, normalization, new PerfLogger(log)); rebuilt = await upgrade.RunAsync(); } watch.Stop(); var stats = new RebuildStats(watch.Elapsed.TotalSeconds, meters, rebuilt, MeasuredInThisRun: true); log.Write(string.Create(CultureInfo.InvariantCulture, $"Rebuilt {rebuilt} of {meters} meters in {stats.Seconds:F1} s")); await using var command = new NpgsqlCommand( "INSERT INTO app_setting (key, value) VALUES (@key, @value) ON CONFLICT (key) DO UPDATE SET value = excluded.value", connection); command.Parameters.AddWithValue("key", RebuildKey); command.Parameters.Add(new NpgsqlParameter("value", NpgsqlTypes.NpgsqlDbType.Jsonb) { Value = JsonSerializer.Serialize(stats) }); await command.ExecuteNonQueryAsync(); return stats; } /// A normalization service as the app builds it, in the dataset's zone, on a clock frozen at its "now". public static NormalizationService Normalization(MeterVaultDbContext db, DatasetManifest manifest) => new(db, NormalizationEngine.CreateDefault(), Microsoft.Extensions.Options.Options.Create(new MeterVaultOptions { TimeZone = manifest.Zone }), new FixedTimeProvider(manifest.Now)); /// Compresses the raw chunks the 30-day policy would have compressed by the dataset's "now". private static async Task<(double? Seconds, int Chunks)> CompressAsync(NpgsqlConnection connection, DateTimeOffset now, PerfLog log) { await using var pending = new NpgsqlCommand( """ SELECT count(*) FROM timescaledb_information.chunks WHERE hypertable_name = 'reading' AND NOT is_compressed AND range_end <= @horizon """, connection); pending.Parameters.AddWithValue("horizon", now.AddDays(-30).UtcDateTime); var count = Convert.ToInt32(await pending.ExecuteScalarAsync(), CultureInfo.InvariantCulture); if (count == 0) { return (null, 0); } log.Write(string.Create(CultureInfo.InvariantCulture, $"Compressing {count} raw chunks older than 30 days…")); var watch = Stopwatch.StartNew(); await using var compress = new NpgsqlCommand( "SELECT count(compress_chunk(c, if_not_compressed => true)) FROM show_chunks('reading', older_than => @horizon) c", connection) { CommandTimeout = 0, }; compress.Parameters.AddWithValue("horizon", now.AddDays(-30).UtcDateTime); await compress.ExecuteScalarAsync(); return (watch.Elapsed.TotalSeconds, count); } public static async Task ScalarAsync(NpgsqlConnection connection, string sql) { await using var command = new NpgsqlCommand(sql, connection) { CommandTimeout = 0 }; var value = await command.ExecuteScalarAsync(); return value is DBNull ? null : value; } public static async Task ExecuteAsync(NpgsqlConnection connection, string sql) { await using var command = new NpgsqlCommand(sql, connection) { CommandTimeout = 0 }; await command.ExecuteNonQueryAsync(); } public async ValueTask DisposeAsync() { if (_container is not null) { await _container.DisposeAsync(); } } } /// Timestamped progress to perf-<label>.log in the output directory and to the test output. internal sealed class PerfLog : IDisposable { private readonly StreamWriter _file; private readonly ITestOutputHelper? _output; private readonly Stopwatch _clock = Stopwatch.StartNew(); public PerfLog(PerfSettings settings, string name, ITestOutputHelper? output) { Directory.CreateDirectory(settings.OutputDirectory); Path = System.IO.Path.Combine(settings.OutputDirectory, $"{name}-{settings.Label}.log"); _file = new StreamWriter(Path, append: false) { AutoFlush = true }; _output = output; } public string Path { get; } public void Write(string message) { var line = string.Create(CultureInfo.InvariantCulture, $"[{DateTimeOffset.Now:HH:mm:ss} +{_clock.Elapsed.TotalSeconds,7:F1}s] {message}"); lock (_file) { _file.WriteLine(line); } try { _output?.WriteLine(line); } catch (InvalidOperationException) { // The test has finished; the file still has it. } } public void Dispose() => _file.Dispose(); } /// Routes the upgrade's progress ("Rebuilt 100 of 1000 meter(s)") into the perf log. internal sealed class PerfLogger(PerfLog log) : ILogger { public IDisposable? BeginScope(TState state) where TState : notnull => null; public bool IsEnabled(LogLevel logLevel) => logLevel >= LogLevel.Information; public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) { if (IsEnabled(logLevel)) { log.Write($"{typeof(T).Name}: {formatter(state, exception)}{(exception is null ? string.Empty : " — " + exception.Message)}"); } } }