Migration engine: Copy licensing throughput history to SQL

src/ServiceControl.Migration.AcceptanceTests/FakeBrokerThroughputQuery.cs

@@ -0,0 +1,97 @@
+namespace ServiceControl.Migration.AcceptanceTests;
+
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Collections.ObjectModel;
+using System.Linq;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+using ServiceControl.Transports.BrokerThroughput;
+
+/// <summary>
+/// A broker that reports the daily counts a test stages, only for days on or after the start date the collector asks from,
+/// records the start date of every request, and counts how many whole collector passes have read it.
+/// </summary>
+class FakeBrokerThroughputQuery : IBrokerThroughputQuery
+{
+    readonly ConcurrentDictionary<string, ConcurrentDictionary<DateOnly, long>> staged = new();
+    readonly ConcurrentQueue<(string Queue, DateOnly StartDate)> requests = new();
+    int queuesStillReading;
+    int completedPasses;
+
+    /// <summary>
+    /// How many times the collector has finished reading every staged queue's days.
+    /// </summary>
+    public int CompletedPasses => Volatile.Read(ref completedPasses);
+
+    /// <summary>
+    /// Every queue the collector asked for days of, with the first day it asked for, in the order it asked.
+    /// </summary>
+    public IReadOnlyList<(string Queue, DateOnly StartDate)> Requests => [.. requests];
+
+    /// <summary>
+    /// Makes the broker report count messages on the queue for that day, replacing any count staged for the same day.
+    /// </summary>
+    public void Stage(string queue, DateOnly day, long count) => staged.GetOrAdd(queue, _ => new())[day] = count;
+
+    public Dictionary<string, string> Data { get; } = [];
+
+    public string MessageTransport => "FakeBroker";
+
+    public string ScopeType => null;
+
+    public KeyDescriptionPair[] Settings => [];
+
+    public void Initialize(ReadOnlyDictionary<string, string> settings) { }
+
+    public bool HasInitialisationErrors(out string errorMessage)
+    {
+        errorMessage = string.Empty;
+        return false;
+    }
+
+    public Task<(bool Success, List<string> Errors, string Diagnostics)> TestConnection(CancellationToken cancellationToken = default) =>
+        Task.FromResult((true, new List<string>(), "Fake broker reachable"));
+
+    public async IAsyncEnumerable<IBrokerQueue> GetQueueNames([EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        var queues = staged.Keys.ToArray();
+        Volatile.Write(ref queuesStillReading, queues.Length);
+
+        foreach (var queue in queues)
+        {
+            yield return new DefaultBrokerQueue(queue);
+        }
+
+        await Task.CompletedTask;
+    }
+
+    public async IAsyncEnumerable<QueueThroughput> GetThroughputPerDay(IBrokerQueue brokerQueue, DateOnly startDate, [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        requests.Enqueue((brokerQueue.QueueName, startDate));
+
+        try
+        {
+            foreach (var (day, count) in staged[brokerQueue.QueueName].Where(entry => entry.Key >= startDate).OrderBy(entry => entry.Key))
+            {
+                yield return new QueueThroughput { DateUTC = day, TotalThroughput = count };
+            }
+        }
+        finally
+        {
+            // The collector records each day before asking for the next, so the last queue's enumeration ending is the pass ending.
+            if (Interlocked.Decrement(ref queuesStillReading) == 0)
+            {
+                Interlocked.Increment(ref completedPasses);
+            }
+        }
+
+        await Task.CompletedTask;
+    }
+
+    public string SanitizeEndpointName(string endpointName) => endpointName;
+
+    public string SanitizedEndpointNameCleanser(string endpointName) => endpointName;
+}

src/ServiceControl.Migration.AcceptanceTests/LicensingEndpointsMigrationTests.cs

@@ -0,0 +1,124 @@
+namespace ServiceControl.Migration.AcceptanceTests;
+
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.EntityFrameworkCore;
+using NUnit.Framework;
+using Particular.LicensingComponent.Contracts;
+using ServiceControl.Persistence.DataMigration;
+
+[TestFixture]
+[NonParallelizable]
+class LicensingEndpointsMigrationTests : MigrationAcceptanceTest
+{
+    [Test]
+    public async Task Copies_an_endpoint_and_reads_it_back_through_the_products_own_API()
+    {
+        await SeedThroughputSource(new Endpoint("Sales.Orders", ThroughputSource.Broker)
+        {
+            SanitizedName = "Sales_Orders",
+            UserIndicator = "Sales team",
+            Scope = "Sales",
+            EndpointIndicators = [EndpointIndicator.PlatformEndpoint.ToString()]
+        });
+
+        await CopyCategory("LicensingEndpoints");
+
+        var fromTarget = await TargetLicensingDataStore.GetEndpoint("Sales.Orders", ThroughputSource.Broker);
+        var reportableEndpoints = await TargetLicensingDataStore.GetAllEndpoints(includePlatformEndpoints: false);
+        var row = await QueryTarget(dbContext => dbContext.LicensingEndpoints.AsNoTracking().SingleAsync());
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(fromTarget, Is.Not.Null);
+            Assert.That(fromTarget!.UserIndicator, Is.EqualTo("Sales team"));
+            Assert.That(fromTarget.Scope, Is.EqualTo("Sales"));
+            Assert.That(row.SanitizedName, Is.EqualTo("Sales_Orders"));
+            Assert.That(row.NormalizedSanitizedName, Is.EqualTo("sales_orders"), "GetEndpointThroughputByQueueName matches on this column, so an unnormalised value hides a migrated endpoint's throughput from the licensing report");
+            Assert.That(fromTarget.EndpointIndicators, Is.EqualTo(new[] { EndpointIndicator.PlatformEndpoint.ToString() }), "the indicators column holds what EF itself writes: JSON on SQL Server, a text[] on PostgreSQL");
+            Assert.That(reportableEndpoints, Is.Empty, "the product filters platform endpoints inside the database, so an indicator value EF cannot query puts a platform endpoint in the licensing report");
+        }
+    }
+
+    [Test]
+    public async Task The_same_name_under_two_sources_is_two_rows_and_not_a_collision()
+    {
+        await SeedThroughputSource(new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" });
+        await SeedThroughputSource(new Endpoint("SALES.ORDERS", ThroughputSource.Audit) { SanitizedName = "sales.orders" });
+
+        var result = await CopyCategory("LicensingEndpoints");
+
+        Assert.That(result.Copied, Is.EqualTo(2), "the key is (NormalizedName, ThroughputSource), so the throughput source must survive into the key or the second endpoint is treated as a duplicate of the first");
+        Assert.That(result.Skipped, Is.Zero);
+    }
+
+    [Test]
+    public async Task An_endpoint_the_target_already_holds_is_already_present_rather_than_skipped()
+    {
+        await SeedThroughputSource(new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" });
+
+        await CopyCategory("LicensingEndpoints");
+        await DeleteCheckpoint("LicensingEndpoints");
+        var second = await CopyCategory("LicensingEndpoints");
+
+        Assert.That(second.AlreadyPresent, Is.EqualTo(1));
+        Assert.That(second.Skipped, Is.Zero, "an endpoint the target already holds was not lost, and counting it as a skip makes --migration-verify subtract it twice");
+    }
+
+    [Test]
+    public async Task Two_names_differing_only_in_a_trailing_space_never_fail_the_batch()
+    {
+        var category = MigrationCategoryRegistry.All.Single(entry => entry.Id == "LicensingEndpoints");
+        var batch = new MigrationBatch(
+            [
+                new MigrationRow("Sales.Orders/Broker", new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" }, new Dictionary<string, object>()),
+                new MigrationRow("Sales.Orders /Broker", new Endpoint("Sales.Orders ", ThroughputSource.Broker) { SanitizedName = "sales.orders" }, new Dictionary<string, object>())
+            ],
+            "Sales.Orders/Broker");
+        var checkpointToExtend = new MigrationCheckpoint(category.Id, MigrationCategoryState.InProgress, batch.Cursor, 0, 0, null, null, null, null, null, null);
+
+        var result = await Target.Write(category, batch, checkpointToExtend);
+
+        var keyIgnoresTrailingSpace = (await QueryTarget(dbContext => Task.FromResult(dbContext.Database.ProviderName))).Contains("SqlServer");
+        var storedNames = await QueryTarget(dbContext => dbContext.LicensingEndpoints.AsNoTracking().Select(row => row.Name).ToListAsync());
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(result.Copied, Is.EqualTo(keyIgnoresTrailingSpace ? 1 : 2), "SQL Server's key ignores the trailing space and keeps the first row; PostgreSQL's does not and keeps both");
+            Assert.That(result.AlreadyPresent, Is.EqualTo(keyIgnoresTrailingSpace ? 1 : 0), "the row SQL Server's key treats as the first's is already present, not lost");
+            Assert.That(result.Skipped, Is.Zero);
+            Assert.That(await Target.Count(category), Is.EqualTo(result.Copied), "every row reported as copied is a row in the table");
+
+            if (keyIgnoresTrailingSpace)
+            {
+                Assert.That(storedNames, Is.EqualTo(new[] { "Sales.Orders " }), "the row whose id sorts first by ordinal order is the one a shared key keeps, whatever order the batch arrives in");
+            }
+        }
+    }
+
+    [Test]
+    public async Task An_endpoint_with_no_name_is_skipped_as_a_required_value_missing()
+    {
+        var category = MigrationCategoryRegistry.All.Single(entry => entry.Id == "LicensingEndpoints");
+        var batch = new MigrationBatch(
+            [
+                new MigrationRow("/Broker", new Endpoint(new EndpointIdentifier(null!, ThroughputSource.Broker)) { SanitizedName = "nameless" }, new Dictionary<string, object>()),
+                new MigrationRow("Sales.Orders/Broker", new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = null! }, new Dictionary<string, object>())
+            ],
+            "Sales.Orders/Broker");
+        var checkpointToExtend = new MigrationCheckpoint(category.Id, MigrationCategoryState.InProgress, batch.Cursor, 0, 0, null, null, null, null, null, null);
+
+        var result = await Target.Write(category, batch, checkpointToExtend);
+        var stored = await QueryTarget(dbContext => dbContext.LicensingEndpoints.AsNoTracking().SingleAsync());
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(stored.SanitizedName, Is.Empty, "a null sanitized name takes the empty string the product itself stores");
+            Assert.That(stored.NormalizedSanitizedName, Is.Empty);
+            Assert.That(result.Copied, Is.EqualTo(1));
+            Assert.That(result.SkippedIds, Is.EqualTo(new[] { "/Broker" }));
+            Assert.That(result.SkipReasons![MigrationSkipReason.RequiredValueMissing], Is.EqualTo(1), "Name and NormalizedName are NOT NULL and the product has no value of its own for a nameless endpoint, so the row is skipped and counted rather than halting the category");
+        }
+    }
+}

src/ServiceControl.Migration.AcceptanceTests/LicensingSettingsMigrationTests.cs

@@ -0,0 +1,98 @@
+namespace ServiceControl.Migration.AcceptanceTests;
+
+using System.Linq;
+using System.Text.Json;
+using System.Threading.Tasks;
+using NUnit.Framework;
+using Particular.LicensingComponent.Contracts;
+using ServiceControl.Persistence.DataMigration;
+
+[TestFixture]
+[NonParallelizable]
+class LicensingSettingsMigrationTests : MigrationAcceptanceTest
+{
+    static LicensedEndpointDetails Details => new LicensedEndpointDetails
+    {
+        LicenseId = "abc-123",
+        ValidId = true,
+        ServiceEndDate = "2027-01-01",
+        Endpoints = [new LicensedEndpoint { Name = "Sales.Orders", Classification = 2, EndpointSize = "Large", Queues = [new QueueIdentity { NameHash = "n1", ScopeHash = "s1" }] }],
+        InfrastructureQueues = [new QueueIdentity { NameHash = "n2", ScopeHash = "s2" }],
+        ExcludedQueues = [new QueueIdentity { NameHash = "n3", ScopeHash = "s3" }],
+        Products = [new Product { ProductCode = "P1", MonthlyThroughput = 500 }]
+    };
+
+    [Test]
+    public async Task Copies_report_masks_as_one_row_in_settings()
+    {
+        await SeedThroughputReportMasks(["contoso", "fabrikam"]);
+
+        var result = await CopyCategory("LicensingReportMasks");
+
+        Assert.That(result.Copied, Is.EqualTo(1));
+        var masks = await TargetLicensingDataStore.GetReportMasks();
+        Assert.That(masks, Is.EquivalentTo(new[] { "contoso", "fabrikam" }));
+    }
+
+    [Test]
+    public async Task Copies_the_licensed_endpoint_details_file_as_one_row_in_settings()
+    {
+        var seeded = Details;
+        await SeedThroughputLicensedEndpointDetails(seeded);
+
+        var result = await CopyCategory("LicensedEndpointDetails");
+
+        Assert.That(result.Copied, Is.EqualTo(1));
+        var details = await TargetLicensingDataStore.GetLicensedEndpointDetails();
+        Assert.That(JsonSerializer.Serialize(details), Is.EqualTo(JsonSerializer.Serialize(seeded)), "every field, including the endpoint, queue and product arrays, must reach the SQL row");
+    }
+
+    [Test]
+    public async Task Report_masks_an_earlier_pass_copied_are_kept_and_reported_as_already_present()
+    {
+        await TargetLicensingDataStore.SaveReportMasks(["copied earlier"]);
+        await SeedThroughputReportMasks(["contoso"]);
+
+        var result = await CopyCategory("LicensingReportMasks");
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(result.Copied, Is.Zero);
+            Assert.That(result.AlreadyPresent, Is.EqualTo(1));
+            Assert.That(await TargetLicensingDataStore.GetReportMasks(), Is.EquivalentTo(new[] { "copied earlier" }), "a copy inserts a settings row only where the target has none, so a value an earlier pass copied is never overwritten");
+        }
+    }
+
+    [Test]
+    public async Task Licensed_endpoint_details_an_earlier_pass_copied_are_kept_and_reported_as_already_present()
+    {
+        await TargetLicensingDataStore.SaveLicensedEndpointDetails(new LicensedEndpointDetails { LicenseId = "copied earlier" });
+        await SeedThroughputLicensedEndpointDetails(new LicensedEndpointDetails { LicenseId = "abc-123" });
+
+        var result = await CopyCategory("LicensedEndpointDetails");
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(result.Copied, Is.Zero);
+            Assert.That(result.AlreadyPresent, Is.EqualTo(1));
+            Assert.That((await TargetLicensingDataStore.GetLicensedEndpointDetails())!.LicenseId, Is.EqualTo("copied earlier"), "a copy inserts a settings row only where the target has none, so a value an earlier pass copied is never overwritten");
+        }
+    }
+
+    [Test]
+    public async Task An_absent_document_writes_no_row()
+    {
+        var masks = MigrationCategoryRegistry.All.Single(entry => entry.Id == "LicensingReportMasks");
+        var details = MigrationCategoryRegistry.All.Single(entry => entry.Id == "LicensedEndpointDetails");
+
+        var masksResult = await CopyCategory(masks.Id);
+        var detailsResult = await CopyCategory(details.Id);
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(masksResult.Copied + detailsResult.Copied, Is.Zero);
+            Assert.That(await Target.Count(masks), Is.Zero, "an absent RavenDB document hands over nothing, so no empty masking list is invented on the SQL side");
+            Assert.That(await Target.Count(details), Is.Zero);
+        }
+    }
+}