MeterVault/tests/Integration.Tests/Performance/PerfDatabase.cs

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;

/// <summary>How the measured database was prepared: loaded now or reused, and what each step took.</summary>
internal sealed record PreparedDatabase(
    DatasetManifest Manifest,
    bool Loaded,
    RebuildStats? Rebuild,
    double? CompressSeconds,
    int CompressedChunks);

/// <summary>The startup rebuild (NormalizationUpgrade) over the whole dataset.</summary>
internal sealed record RebuildStats(double Seconds, int Meters, int Rebuilt, bool MeasuredInThisRun);

/// <summary>
/// The database of one performance run: a fresh TimescaleDB container (the image the fixture pins), or an existing
/// database from <c>METERVAULT_PERF_DB</c>. Prepares it the way a real instance gets there — migrations, the default
/// seed, the raw dataset, the startup rebuild (<see cref="NormalizationUpgrade"/>), and the compression policy's work
/// on raw chunks older than 30 days — and hands out contexts for the readers.
/// </summary>
internal sealed class PerfDatabase : IDbContextFactory<MeterVaultDbContext>, IAsyncDisposable
{
    /// <summary>The pinned image of <see cref="TimescaleFixture"/>.</summary>
    public const string Image = "timescale/timescaledb:2.17.2-pg16";

    /// <summary>The frozen "now" the measured dataset ends at: 19 September 2026, 14:37 Berlin, as the cost tests use.</summary>
    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; }

    /// <summary>Where the database came from, for the report.</summary>
    public string Origin => _container is null ? "existing database (METERVAULT_PERF_DB)" : $"fresh container ({Image})";

    public static async Task<PerfDatabase> 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<MeterVaultDbContext>()
            .UseNpgsql(ConnectionString, npgsql => npgsql.MigrationsAssembly(typeof(MeterVaultDbContext).Assembly.FullName))
            .UseSnakeCaseNamingConvention()
            .Options;
        return new MeterVaultDbContext(options);
    }

    /// <summary>
    /// 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).
    /// </summary>
    public async Task<PreparedDatabase> 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);
    }

    /// <summary>
    /// 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.
    /// </summary>
    private async Task<RebuildStats?> 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<RebuildStats>(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<NormalizationUpgrade>(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;
    }

    /// <summary>A normalization service as the app builds it, in the dataset's zone, on a clock frozen at its "now".</summary>
    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));

    /// <summary>Compresses the raw chunks the 30-day policy would have compressed by the dataset's "now".</summary>
    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<object?> 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();
        }
    }
}

/// <summary>Timestamped progress to <c>perf-&lt;label&gt;.log</c> in the output directory and to the test output.</summary>
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();
}

/// <summary>Routes the upgrade's progress ("Rebuilt 100 of 1000 meter(s)") into the perf log.</summary>
internal sealed class PerfLogger<T>(PerfLog log) : ILogger<T>
{
    public IDisposable? BeginScope<TState>(TState state)
        where TState : notnull => null;

    public bool IsEnabled(LogLevel logLevel) => logLevel >= LogLevel.Information;

    public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
    {
        if (IsEnabled(logLevel))
        {
            log.Write($"{typeof(T).Name}: {formatter(state, exception)}{(exception is null ? string.Empty : " — " + exception.Message)}");
        }
    }
}