diff --git a/src/ServiceControl.Migration.AcceptanceTests/FakeBrokerThroughputQuery.cs b/src/ServiceControl.Migration.AcceptanceTests/FakeBrokerThroughputQuery.cs new file mode 100644 index 0000000000..5ceb938368 --- /dev/null +++ b/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; + +/// +/// 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. +/// +class FakeBrokerThroughputQuery : IBrokerThroughputQuery +{ + readonly ConcurrentDictionary> staged = new(); + readonly ConcurrentQueue<(string Queue, DateOnly StartDate)> requests = new(); + int queuesStillReading; + int completedPasses; + + /// + /// How many times the collector has finished reading every staged queue's days. + /// + public int CompletedPasses => Volatile.Read(ref completedPasses); + + /// + /// Every queue the collector asked for days of, with the first day it asked for, in the order it asked. + /// + public IReadOnlyList<(string Queue, DateOnly StartDate)> Requests => [.. requests]; + + /// + /// Makes the broker report count messages on the queue for that day, replacing any count staged for the same day. + /// + public void Stage(string queue, DateOnly day, long count) => staged.GetOrAdd(queue, _ => new())[day] = count; + + public Dictionary Data { get; } = []; + + public string MessageTransport => "FakeBroker"; + + public string ScopeType => null; + + public KeyDescriptionPair[] Settings => []; + + public void Initialize(ReadOnlyDictionary settings) { } + + public bool HasInitialisationErrors(out string errorMessage) + { + errorMessage = string.Empty; + return false; + } + + public Task<(bool Success, List Errors, string Diagnostics)> TestConnection(CancellationToken cancellationToken = default) => + Task.FromResult((true, new List(), "Fake broker reachable")); + + public async IAsyncEnumerable 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 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; +} diff --git a/src/ServiceControl.Migration.AcceptanceTests/LicensingEndpointsMigrationTests.cs b/src/ServiceControl.Migration.AcceptanceTests/LicensingEndpointsMigrationTests.cs new file mode 100644 index 0000000000..83e0f6a57a --- /dev/null +++ b/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()), + new MigrationRow("Sales.Orders /Broker", new Endpoint("Sales.Orders ", ThroughputSource.Broker) { SanitizedName = "sales.orders" }, new Dictionary()) + ], + "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()), + new MigrationRow("Sales.Orders/Broker", new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = null! }, new Dictionary()) + ], + "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"); + } + } +} diff --git a/src/ServiceControl.Migration.AcceptanceTests/LicensingSettingsMigrationTests.cs b/src/ServiceControl.Migration.AcceptanceTests/LicensingSettingsMigrationTests.cs new file mode 100644 index 0000000000..3655d52306 --- /dev/null +++ b/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); + } + } +} diff --git a/src/ServiceControl.Migration.AcceptanceTests/LicensingThroughputMigrationTests.cs b/src/ServiceControl.Migration.AcceptanceTests/LicensingThroughputMigrationTests.cs new file mode 100644 index 0000000000..7ce8608892 --- /dev/null +++ b/src/ServiceControl.Migration.AcceptanceTests/LicensingThroughputMigrationTests.cs @@ -0,0 +1,205 @@ +namespace ServiceControl.Migration.AcceptanceTests; + +using System; +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 LicensingThroughputMigrationTests : MigrationAcceptanceTest +{ + [Test] + public void Writing_counts_before_their_endpoint_is_refused_naming_the_endpoint() + { + var category = MigrationCategoryRegistry.All.Single(entry => entry.Id == "LicensingThroughput"); + var day = new ThroughputDayRow(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker), new DateOnly(2026, 1, 1), 100); + var batch = new MigrationBatch([new MigrationRow("Sales.Orders/Broker|2026-01-01", day, new Dictionary())], "Sales.Orders/Broker"); + var checkpointToExtend = new MigrationCheckpoint(category.Id, MigrationCategoryState.InProgress, batch.Cursor, 0, 0, null, null, null, null, null, null); + + var exception = Assert.ThrowsAsync(() => Target.Write(category, batch, checkpointToExtend)); + + Assert.That(exception!.Message, Does.Contain("Endpoint Sales.Orders from Broker does not exist"), "a missing endpoint is refused by name and source, as the product's own throughput write refuses it, rather than surfacing as a raw foreign key violation"); + } + + [Test] + public async Task An_endpoint_skipped_by_its_own_copy_blocks_the_counts_and_none_are_written() + { + await SeedThroughputSource(new Endpoint("Billing", ThroughputSource.Broker) { SanitizedName = new string('s', 451) }); + await SeedThroughputTimeSeries("Billing/Broker", new DateOnly(2026, 1, 1), 50); + + await CopyCategory("LicensingEndpoints"); + Assert.ThrowsAsync(() => CopyCategory("LicensingThroughput")); + + var endpoints = await ReadCheckpoint("LicensingEndpoints"); + var throughput = await ReadCheckpoint("LicensingThroughput"); + var written = await QueryTarget(dbContext => dbContext.LicensingEndpointThroughput.CountAsync()); + + using (Assert.EnterMultipleScope()) + { + Assert.That(endpoints.State, Is.EqualTo(MigrationCategoryState.CompleteWithErrors), "the endpoint whose sanitized name is too long is skipped as a fault"); + Assert.That(throughput.State, Is.EqualTo(MigrationCategoryState.Blocked), "a Failed predecessor blocks the counts, so the writer never meets a day whose endpoint was skipped"); + Assert.That(throughput.LastError, Does.Contain("must follow LicensingEndpoints, which is CompleteWithErrors")); + Assert.That(written, Is.Zero); + } + } + + [Test] + public async Task A_document_of_another_collection_is_not_read_so_the_endpoints_complete_and_the_counts_copy() + { + await SeedThroughputSource(new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" }); + await SeedThroughputTimeSeries("Sales.Orders/Broker", new DateOnly(2026, 1, 1), 100); + + using (var session = SourceStore.OpenAsyncSession(Source.ThroughputDatabase)) + { + var stray = new { Setting = "left by an older version" }; + await session.StoreAsync(stray, "LegacySettings/1"); + session.Advanced.GetMetadataFor(stray)["@collection"] = "LegacySettings"; + await session.SaveChangesAsync(); + } + + var endpoints = await CopyCategory("LicensingEndpoints"); + + using (Assert.EnterMultipleScope()) + { + Assert.That(endpoints.Saved.State, Is.EqualTo(MigrationCategoryState.Complete), "a document outside the EndpointDocuments collection is not an endpoint, so it is not skipped as a nameless one that blocks the counts"); + Assert.That(endpoints.Copied, Is.EqualTo(1)); + Assert.That(endpoints.Skipped, Is.Zero); + } + + var throughput = await CopyCategory("LicensingThroughput"); + + using (Assert.EnterMultipleScope()) + { + Assert.That(throughput.Saved.State, Is.EqualTo(MigrationCategoryState.Complete)); + Assert.That(throughput.Copied, Is.EqualTo(1)); + Assert.That(await QueryTarget(dbContext => dbContext.LicensingEndpointThroughput.CountAsync()), Is.EqualTo(1)); + } + } + + [Test] + public async Task The_registry_order_blocks_the_counts_until_their_endpoints_are_copied() + { + await SeedThroughputSource(new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" }); + await SeedThroughputTimeSeries("Sales.Orders/Broker", new DateOnly(2026, 1, 1), 100); + + Assert.ThrowsAsync(() => CopyCategory("LicensingThroughput")); + var blocked = await ReadCheckpoint("LicensingThroughput"); + + await CopyCategory("LicensingEndpoints"); + var result = await CopyCategory("LicensingThroughput"); + + using (Assert.EnterMultipleScope()) + { + Assert.That(blocked.State, Is.EqualTo(MigrationCategoryState.Blocked)); + Assert.That(blocked.LastError, Does.Contain("must follow LicensingEndpoints")); + Assert.That(result.Copied, Is.EqualTo(1)); + } + } + + [Test] + public async Task Running_the_category_twice_produces_the_same_count() + { + await SeedThroughputSource(new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" }); + await SeedThroughputTimeSeries("Sales.Orders/Broker", new DateOnly(2026, 1, 1), 100); + await CopyCategory("LicensingEndpoints"); + + await CopyCategory("LicensingThroughput"); + await DeleteCheckpoint("LicensingThroughput"); + var second = await CopyCategory("LicensingThroughput"); + + var total = await QueryTarget(dbContext => dbContext.LicensingEndpointThroughput + .Where(row => row.NormalizedName == "sales.orders" && row.ThroughputSource == ThroughputSource.Broker && row.DateUtc == new DateOnly(2026, 1, 1)) + .Select(row => row.MessageCount) + .SingleAsync()); + + Assert.That(second.Copied, Is.EqualTo(1), "the second run wrote nothing, so the count below cannot tell setting from adding"); + Assert.That(total, Is.EqualTo(100), "running the category twice must set the day's count, not add to it"); + } + + [Test] + public async Task Two_names_differing_only_in_a_trailing_space_set_one_day_without_failing_the_batch() + { + var endpoints = MigrationCategoryRegistry.All.Single(entry => entry.Id == "LicensingEndpoints"); + var endpointBatch = new MigrationBatch( + [ + new MigrationRow("Sales.Orders /Broker", new Endpoint("Sales.Orders ", ThroughputSource.Broker) { SanitizedName = "sales.orders" }, new Dictionary()), + new MigrationRow("Sales.Orders/Broker", new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" }, new Dictionary()) + ], + "Sales.Orders/Broker"); + await Target.Write(endpoints, endpointBatch, new MigrationCheckpoint(endpoints.Id, MigrationCategoryState.InProgress, endpointBatch.Cursor, 0, 0, null, null, null, null, null, null)); + + var category = MigrationCategoryRegistry.All.Single(entry => entry.Id == "LicensingThroughput"); + var day = new DateOnly(2026, 1, 1); + var batch = new MigrationBatch( + [ + new MigrationRow("Sales.Orders /Broker|2026-01-01", new ThroughputDayRow(new EndpointIdentifier("Sales.Orders ", ThroughputSource.Broker), day, 100), new Dictionary()), + new MigrationRow("Sales.Orders/Broker|2026-01-01", new ThroughputDayRow(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker), day, 250), new Dictionary()) + ], + "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 count = await QueryTarget(dbContext => dbContext.LicensingEndpointThroughput + .Where(row => row.NormalizedName == "sales.orders" && row.ThroughputSource == ThroughputSource.Broker && row.DateUtc == day) + .Select(row => row.MessageCount) + .SingleAsync()); + + using (Assert.EnterMultipleScope()) + { + Assert.That(result.Copied, Is.EqualTo(keyIgnoresTrailingSpace ? 1 : 2), "SQL Server's key ignores the trailing space, so the two days are one row; PostgreSQL's does not, so they are two"); + Assert.That(result.AlreadyPresent, Is.EqualTo(keyIgnoresTrailingSpace ? 1 : 0), "the day SQL Server's key folds into the other is already present, not lost or failed"); + Assert.That(result.Skipped, Is.Zero); + Assert.That(await Target.Count(category), Is.EqualTo(result.Copied), "in a single batch, every row reported as copied is a row in the table"); + Assert.That(count, Is.EqualTo(250), "the later row sets the day's count where the key makes the two one, and never adds to the earlier row's"); + } + } + + [Test] + public async Task The_copied_counts_are_what_the_collector_reads_as_already_collected() + { + await SeedThroughputSource(new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" }); + await SeedThroughputTimeSeries("Sales.Orders/Broker", new DateOnly(2026, 1, 1), 100); + await SeedThroughputTimeSeries("Sales.Orders/Broker", new DateOnly(2026, 1, 2), 200); + + await CopyCategory("LicensingEndpoints"); + await CopyCategory("LicensingThroughput"); + + var endpoint = await TargetLicensingDataStore.GetEndpoint(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker)); + + Assert.That(endpoint!.LastCollectedDate, Is.EqualTo(new DateOnly(2026, 1, 2)), + "LastCollectedDate is derived from MAX(DateUtc) over the copied rows, and it is the only thing stopping the collector re-collecting these two days and adding to them"); + } + + [Test] + public async Task A_day_whose_endpoint_has_no_name_is_skipped_as_a_required_value_missing() + { + await SeedThroughputSource(new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" }); + await CopyCategory("LicensingEndpoints"); + + var category = MigrationCategoryRegistry.All.Single(entry => entry.Id == "LicensingThroughput"); + var day = new DateOnly(2026, 1, 1); + var batch = new MigrationBatch( + [ + new MigrationRow("/Broker|2026-01-01", new ThroughputDayRow(new EndpointIdentifier(null!, ThroughputSource.Broker), day, 100), new Dictionary()), + new MigrationRow("Sales.Orders/Broker|2026-01-01", new ThroughputDayRow(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker), day, 250), new Dictionary()) + ], + "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); + + using (Assert.EnterMultipleScope()) + { + Assert.That(result.Copied, Is.EqualTo(1)); + Assert.That(result.SkippedIds, Is.EqualTo(new[] { "/Broker|2026-01-01" })); + Assert.That(result.SkipReasons![MigrationSkipReason.RequiredValueMissing], Is.EqualTo(1), "a nameless endpoint's day has no NormalizedName to key under, so it is skipped and counted rather than halting the category"); + } + } +} diff --git a/src/ServiceControl.Migration.AcceptanceTests/MigrationAcceptanceTest.cs b/src/ServiceControl.Migration.AcceptanceTests/MigrationAcceptanceTest.cs index 6574e6eced..637692496f 100644 --- a/src/ServiceControl.Migration.AcceptanceTests/MigrationAcceptanceTest.cs +++ b/src/ServiceControl.Migration.AcceptanceTests/MigrationAcceptanceTest.cs @@ -13,10 +13,14 @@ namespace ServiceControl.Migration.AcceptanceTests; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using NServiceBus.Extensibility; using NServiceBus.Transport; using NUnit.Framework; +using Particular.LicensingComponent.BrokerThroughput; +using Particular.LicensingComponent.Contracts; +using Particular.LicensingComponent.Persistence; using ServiceControl.Persistence.EFCore; using Particular.ServiceControl.Hosting; using ServiceBus.Management.Infrastructure.Settings; @@ -36,6 +40,7 @@ namespace ServiceControl.Migration.AcceptanceTests; using ServiceControl.Persistence.Infrastructure; using ServiceControl.Persistence.Tests; using ServiceControl.Persistence.UnitOfWork; +using ServiceControl.Transports.BrokerThroughput; using TestHelper; // Aliased rather than imported: Raven.Client.Documents carries LINQ extensions that collide with EF Core's. using IDocumentStore = Raven.Client.Documents.IDocumentStore; @@ -52,6 +57,11 @@ abstract class MigrationAcceptanceTest ServiceProvider targetServices; IMigrationCheckpointStore targetCheckpointStore; + readonly FakeBrokerThroughputQuery brokerQuery = new(); + + // RavenDB refuses an incremental time series whose name does not start with "INC:". + // It must equal LicensingDataStore.ThroughputTimeSeriesName, which is internal to the RavenDB persister. + const string ThroughputTimeSeriesName = "INC: throughput data"; protected Settings Settings { get; private set; } protected HttpClient HttpClient { get; private set; } @@ -156,6 +166,64 @@ protected async Task SeedSource(string database, params (string Id, object Docum protected Task SeedSource(params (string Id, object Document)[] documents) => SeedSource(Source.PrimaryDatabase, documents); + protected ILicensingDataStore TargetLicensingDataStore => targetServices.GetRequiredService(); + + protected async Task SeedThroughputSource(Endpoint endpoint) + { + var documentId = $"{endpoint.Id.Name}/{endpoint.Id.ThroughputSource}"; + + var document = new + { + Id = documentId, + EndpointId = new { endpoint.Id.Name, endpoint.Id.ThroughputSource }, + endpoint.SanitizedName, + EndpointIndicators = endpoint.EndpointIndicators ?? [], + UserIndicator = endpoint.UserIndicator ?? string.Empty, + Scope = endpoint.Scope ?? string.Empty + }; + + using var session = SourceStore.OpenAsyncSession(Source.ThroughputDatabase); + await session.StoreAsync(document, documentId); + session.Advanced.GetMetadataFor(document)["@collection"] = "EndpointDocuments"; + await session.SaveChangesAsync(); + } + + protected async Task SeedThroughputTimeSeries(string documentId, DateOnly day, long count) + { + using var session = SourceStore.OpenAsyncSession(Source.ThroughputDatabase); + session.IncrementalTimeSeriesFor(documentId, ThroughputTimeSeriesName) + .Increment(day.ToDateTime(TimeOnly.MinValue, DateTimeKind.Utc), count); + await session.SaveChangesAsync(); + } + + protected void BrokerReports(string queue, DateOnly day, long count) => brokerQuery.Stage(queue, day, count); + + protected IReadOnlyList<(string Queue, DateOnly StartDate)> BrokerRequests => brokerQuery.Requests; + + protected Task WaitForCollectorPass() => + WaitUntil(() => Task.FromResult(brokerQuery.CompletedPasses > 0), "the broker throughput collector completed a pass"); + + // The product adds the broker collector only when a query is registered before AddServiceControl, and customize runs after it. + protected void CollectFromTheBrokerImmediately(IServiceCollection services) + { + services.AddSingleton(brokerQuery); + services.AddHostedService(provider => new BrokerThroughputCollectorHostedService( + provider.GetRequiredService>(), + brokerQuery, + provider.GetRequiredService(), + provider.GetRequiredService(), + provider.GetRequiredService()) + { + DelayStart = TimeSpan.Zero + }); + } + + protected Task SeedThroughputReportMasks(IReadOnlyList maskedStrings) => + SeedSource(Source.ThroughputDatabase, ("ReportMasks", new { MaskedStrings = maskedStrings.ToList() })); + + protected Task SeedThroughputLicensedEndpointDetails(LicensedEndpointDetails details) => + SeedSource(Source.ThroughputDatabase, ("LicensedEndpointDetails", details)); + protected Task SeedSourceWithBody(FailedMessage document, byte[] body, string contentType) => SeedSourceWithBodies((document, body, contentType)); diff --git a/src/ServiceControl.Migration.AcceptanceTests/When_the_throughput_collector_runs_after_a_migration.cs b/src/ServiceControl.Migration.AcceptanceTests/When_the_throughput_collector_runs_after_a_migration.cs new file mode 100644 index 0000000000..0bf218a543 --- /dev/null +++ b/src/ServiceControl.Migration.AcceptanceTests/When_the_throughput_collector_runs_after_a_migration.cs @@ -0,0 +1,47 @@ +namespace ServiceControl.Migration.AcceptanceTests; + +using System; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Time.Testing; +using NUnit.Framework; +using Particular.LicensingComponent.Contracts; + +[TestFixture] +[NonParallelizable] +class When_the_throughput_collector_runs_after_a_migration : MigrationAcceptanceTest +{ + [Test] + public async Task Copied_days_are_not_collected_a_second_time_and_added_to() + { + var time = new FakeTimeProvider(); + var yesterday = DateOnly.FromDateTime(time.GetUtcNow().UtcDateTime).AddDays(-1); + + await SeedThroughputSource(new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" }); + await SeedThroughputTimeSeries("Sales.Orders/Broker", yesterday, 100); + BrokerReports(queue: "sales.orders", day: yesterday, count: 70); + + await RunRequiredCopy(customize: services => + { + services.AddSingleton(time); + CollectFromTheBrokerImmediately(services); + }); + + await WaitForCollectorPass(); + + var total = await QueryTarget(dbContext => dbContext.LicensingEndpointThroughput + .Where(row => row.NormalizedName == "sales.orders" && row.ThroughputSource == ThroughputSource.Broker && row.DateUtc == yesterday) + .Select(row => row.MessageCount) + .SingleAsync()); + + using (Assert.EnterMultipleScope()) + { + Assert.That(BrokerRequests, Is.EqualTo(new[] { ("sales.orders", yesterday.AddDays(1)) }), + "the collector asks only for days after the last copied one, so it never tried to collect the copied day again and a failed write cannot hide that it did"); + Assert.That(total, Is.EqualTo(100), + "100 is the copied count; 70 means the copy wrote nothing and the collector collected the day itself; 170 means the collector re-collected a copied day and the additive write path added the two, so the licensing report overstates usage"); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlMigrationSqlDialect.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlMigrationSqlDialect.cs index 20c1542231..6c4ec5155a 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlMigrationSqlDialect.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlMigrationSqlDialect.cs @@ -1,6 +1,7 @@ namespace ServiceControl.Persistence.EFCore.PostgreSql; using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Entities; using ServiceControl.Persistence.EFCore.Infrastructure; /// @@ -47,4 +48,30 @@ ON CONFLICT ({keyColumns}) DO NOTHING return inserted; } + + public async Task SetLicensingEndpointThroughput(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken = default) + { + var insert = MigrationInsert.For(dbContext); + var keyColumns = string.Join(", ", insert.KeyColumns); + var written = 0; + + foreach (var chunk in rows.Chunk(RowsPerStatement(insert.Columns.Count))) + { + written += (await insert.Execute( + dbContext, + $""" + INSERT INTO {insert.Table} ({string.Join(", ", insert.Columns)}) + VALUES + {ParameterRows(chunk.Length, insert.Columns.Count)} + ON CONFLICT ({keyColumns}) + DO UPDATE SET {string.Join(", ", insert.Columns.Except(insert.KeyColumns).Select(column => $"{column} = excluded.{column}"))} + RETURNING {keyColumns} + """, + chunk, + configureParameter: null, + cancellationToken)).Count; + } + + return written; + } } diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerMigrationSqlDialect.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerMigrationSqlDialect.cs index 5de111af7b..4d4d0044e8 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerMigrationSqlDialect.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerMigrationSqlDialect.cs @@ -6,6 +6,7 @@ namespace ServiceControl.Persistence.EFCore.SqlServer; using Microsoft.EntityFrameworkCore.Metadata; using Microsoft.EntityFrameworkCore.Storage; using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Entities; using ServiceControl.Persistence.EFCore.Infrastructure; /// @@ -113,6 +114,45 @@ WHEN NOT MATCHED THEN INSERT ({columns}) VALUES ({string.Join(", ", insert.Colum return inserted; } + public async Task SetLicensingEndpointThroughput(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken = default) + { + var insert = MigrationInsert.For(dbContext); + var columns = string.Join(", ", insert.Columns); + var partitionBy = string.Join(", ", insert.KeyColumns.Select(column => + Keys.Collations.TryGetValue((insert.Table, column), out var collation) ? $"{column} COLLATE {collation}" : column)); + var written = 0; + + foreach (var chunk in rows.Chunk(RowsPerStatement(insert.Columns.Count))) + { + // HOLDLOCK, or two writers can both find a day missing and both insert it. + // The output names the source's key values, because a stored name can differ from the chunk's by a trailing space and would match no row of it. + written += (await insert.Execute( + dbContext, + $""" + MERGE {insert.Table} WITH (HOLDLOCK) AS t + USING ( + SELECT {columns} + FROM ( + SELECT {columns}, ROW_NUMBER() OVER (PARTITION BY {partitionBy} ORDER BY [MigrationOrdinal] DESC) AS [MigrationDuplicate] + FROM (VALUES + {ParameterRowsWithOrdinal(chunk.Length, insert.Columns.Count)} + ) AS v ({columns}, [MigrationOrdinal]) + ) AS deduplicated + WHERE [MigrationDuplicate] = 1 + ) AS s ({columns}) + ON {string.Join(" AND ", insert.KeyColumns.Select(column => $"t.{column} = s.{column}"))} + WHEN MATCHED THEN UPDATE SET {string.Join(", ", insert.Columns.Except(insert.KeyColumns).Select(column => $"{column} = s.{column}"))} + WHEN NOT MATCHED THEN INSERT ({columns}) VALUES ({string.Join(", ", insert.Columns.Select(column => $"s.{column}"))}) + OUTPUT {string.Join(", ", insert.KeyColumns.Select(column => $"s.{column}"))}; + """, + chunk, + configureParameter: null, + cancellationToken)).Count; + } + + return written; + } + // EF declares unbounded text as nvarchar(4000) or nvarchar(max) by each value's length, and each new mix compiles its own plan. // Size -1 declares every such value as the nvarchar(max) its column already is. static void DeclareUnboundedTextAsMax(IProperty property, DbParameter parameter) diff --git a/src/ServiceControl.Persistence.EFCore/DataMigration/EFCoreMigrationTarget.cs b/src/ServiceControl.Persistence.EFCore/DataMigration/EFCoreMigrationTarget.cs index 85c4bea2be..e0b683b1c8 100644 --- a/src/ServiceControl.Persistence.EFCore/DataMigration/EFCoreMigrationTarget.cs +++ b/src/ServiceControl.Persistence.EFCore/DataMigration/EFCoreMigrationTarget.cs @@ -26,7 +26,11 @@ sealed class EFCoreMigrationTarget( { new KnownEndpointsWriter(migrationDialect), new EndpointSettingsWriter(migrationDialect), - new FailedMessagesWriter(migrationDialect, storagePersistence, settings) + new FailedMessagesWriter(migrationDialect, storagePersistence, settings), + new LicensingEndpointsWriter(migrationDialect), + new LicensingThroughputWriter(migrationDialect), + new LicensingReportMasksWriter(migrationDialect), + new LicensedEndpointDetailsWriter(migrationDialect) }.ToFrozenDictionary(writer => writer.CategoryId, StringComparer.Ordinal); public Task Open(CancellationToken cancellationToken = default) => diff --git a/src/ServiceControl.Persistence.EFCore/DataMigration/LicensingEndpointRowMapper.cs b/src/ServiceControl.Persistence.EFCore/DataMigration/LicensingEndpointRowMapper.cs new file mode 100644 index 0000000000..b5772d85fa --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/DataMigration/LicensingEndpointRowMapper.cs @@ -0,0 +1,29 @@ +namespace ServiceControl.Persistence.EFCore.DataMigration; + +using Particular.LicensingComponent.Contracts; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Implementation; + +/// +/// Turns a licensing endpoint read from the source into the row the target stores, with the same key and the same +/// stand-ins for missing values as the product's own save. +/// +static class LicensingEndpointRowMapper +{ + public static LicensingEndpointEntity Map(Endpoint endpoint) + { + var sanitizedName = endpoint.SanitizedName ?? string.Empty; + + return new LicensingEndpointEntity + { + NormalizedName = LicensingDataStore.Normalize(endpoint.Id.Name), + ThroughputSource = endpoint.Id.ThroughputSource, + Name = endpoint.Id.Name, + SanitizedName = sanitizedName, + NormalizedSanitizedName = LicensingDataStore.Normalize(sanitizedName), + UserIndicator = endpoint.UserIndicator ?? string.Empty, + Scope = endpoint.Scope, + EndpointIndicators = endpoint.EndpointIndicators?.ToList() ?? [] + }; + } +} diff --git a/src/ServiceControl.Persistence.EFCore/DataMigration/LicensingThroughputRowMapper.cs b/src/ServiceControl.Persistence.EFCore/DataMigration/LicensingThroughputRowMapper.cs new file mode 100644 index 0000000000..299fd8912d --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/DataMigration/LicensingThroughputRowMapper.cs @@ -0,0 +1,23 @@ +namespace ServiceControl.Persistence.EFCore.DataMigration; + +using ServiceControl.Persistence.DataMigration; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Implementation; + +/// +/// Turns one endpoint's daily count read from the source into the row the target stores, keyed by the same +/// normalised name the product's own licensing store computes. +/// +static class LicensingThroughputRowMapper +{ + public static LicensingEndpointThroughputEntity Map(ThroughputDayRow row) + { + return new LicensingEndpointThroughputEntity + { + NormalizedName = LicensingDataStore.Normalize(row.Endpoint.Name), + ThroughputSource = row.Endpoint.ThroughputSource, + DateUtc = row.DateUtc, + MessageCount = row.MessageCount + }; + } +} diff --git a/src/ServiceControl.Persistence.EFCore/DataMigration/SettingRowMapper.cs b/src/ServiceControl.Persistence.EFCore/DataMigration/SettingRowMapper.cs new file mode 100644 index 0000000000..5e43817ff8 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/DataMigration/SettingRowMapper.cs @@ -0,0 +1,13 @@ +namespace ServiceControl.Persistence.EFCore.DataMigration; + +using System.Text.Json; +using ServiceControl.Persistence.EFCore.Entities; + +/// +/// Builds the Settings row for a value, storing it under the given key as the JSON text that +/// SettingsStore.StoreSetting writes. is the type the product stores under that key. +/// +static class SettingRowMapper +{ + public static SettingEntity Map(string key, T value) => new() { Key = key, Value = JsonSerializer.Serialize(value) }; +} diff --git a/src/ServiceControl.Persistence.EFCore/DataMigration/Writers/LicensedEndpointDetailsWriter.cs b/src/ServiceControl.Persistence.EFCore/DataMigration/Writers/LicensedEndpointDetailsWriter.cs new file mode 100644 index 0000000000..9bcf8ec909 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/DataMigration/Writers/LicensedEndpointDetailsWriter.cs @@ -0,0 +1,31 @@ +namespace ServiceControl.Persistence.EFCore.DataMigration.Writers; + +using DbContexts; +using Entities; +using Infrastructure; +using Microsoft.EntityFrameworkCore; +using Particular.LicensingComponent.Contracts; +using ServiceControl.Persistence.DataMigration; + +/// +/// Writes the licensed endpoint details file as one row in Settings. A row the target already holds is left as it is +/// and counted as already present. +/// +sealed class LicensedEndpointDetailsWriter(IMigrationSqlDialect migrationDialect) : MigrationCategoryWriter +{ + public override string CategoryId => MigrationCategoryIds.LicensedEndpointDetails; + + const string SettingKey = SettingKeys.LicensedEndpointDetails; + + public override int BatchSize(ServiceControlDbContext dbContext) => migrationDialect.RowsPerStatementFor(dbContext); + + protected override Task PrepareDocuments(ServiceControlDbContext dbContext, IReadOnlyList<(MigrationRow Row, LicensedEndpointDetails Document)> documents, CancellationToken cancellationToken = default) + { + List rows = [.. documents.Select(document => SettingRowMapper.Map(SettingKey, document.Document))]; + + return Task.FromResult(new PreparedBatch(async (context, token) => (await migrationDialect.InsertMissing(context, rows, token)).Count, rows.Count, [], Merges: [])); + } + + public override Task Count(ServiceControlDbContext dbContext, CancellationToken cancellationToken = default) => + dbContext.Settings.LongCountAsync(setting => setting.Key == SettingKey, cancellationToken); +} diff --git a/src/ServiceControl.Persistence.EFCore/DataMigration/Writers/LicensingEndpointsWriter.cs b/src/ServiceControl.Persistence.EFCore/DataMigration/Writers/LicensingEndpointsWriter.cs new file mode 100644 index 0000000000..cac5d17b6b --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/DataMigration/Writers/LicensingEndpointsWriter.cs @@ -0,0 +1,54 @@ +namespace ServiceControl.Persistence.EFCore.DataMigration.Writers; + +using DbContexts; +using Entities; +using Infrastructure; +using Microsoft.EntityFrameworkCore; +using Particular.LicensingComponent.Contracts; +using ServiceControl.Persistence.DataMigration; + +/// +/// Writes the licensing endpoints. An endpoint the target already holds, or a second one in the same batch whose key +/// the column treats as the same, is left as it is and counted as already present. +/// +sealed class LicensingEndpointsWriter(IMigrationSqlDialect migrationDialect) : MigrationCategoryWriter +{ + public override string CategoryId => MigrationCategoryIds.LicensingEndpoints; + + public override int BatchSize(ServiceControlDbContext dbContext) => migrationDialect.RowsPerStatementFor(dbContext); + + public override Task Count(ServiceControlDbContext dbContext, CancellationToken cancellationToken = default) => + dbContext.LicensingEndpoints.LongCountAsync(cancellationToken); + + protected override Task PrepareDocuments(ServiceControlDbContext dbContext, IReadOnlyList<(MigrationRow Row, Endpoint Document)> documents, CancellationToken cancellationToken = default) + { + var insert = MigrationInsert.For(dbContext); + var rows = new List(documents.Count); + var skips = new List<(string SourceId, MigrationSkipReason Reason, string Detail)>(); + + foreach (var (row, endpoint) in documents.OrderBy(document => document.Row.SourceId, StringComparer.Ordinal)) + { + if (endpoint.Id.Name is null) + { + skips.Add((row.SourceId, MigrationSkipReason.RequiredValueMissing, "LicensingEndpoints.Name and NormalizedName are NOT NULL and the document has no endpoint name")); + continue; + } + + var entity = LicensingEndpointRowMapper.Map(endpoint); + + if (UnstorableValueSkip(insert, migrationDialect, row, entity) is { } unstorable) + { + skips.Add(unstorable); + continue; + } + + rows.Add(entity); + } + + return Task.FromResult(new PreparedBatch( + async (context, token) => (await migrationDialect.InsertMissing(context, rows, token)).Count, + rows.Count, + skips, + Merges: [])); + } +} diff --git a/src/ServiceControl.Persistence.EFCore/DataMigration/Writers/LicensingReportMasksWriter.cs b/src/ServiceControl.Persistence.EFCore/DataMigration/Writers/LicensingReportMasksWriter.cs new file mode 100644 index 0000000000..18823e068c --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/DataMigration/Writers/LicensingReportMasksWriter.cs @@ -0,0 +1,30 @@ +namespace ServiceControl.Persistence.EFCore.DataMigration.Writers; + +using DbContexts; +using Entities; +using Infrastructure; +using Microsoft.EntityFrameworkCore; +using ServiceControl.Persistence.DataMigration; + +/// +/// Writes the report masks as one row in Settings. A row the target already holds is left as it is and counted as +/// already present. +/// +sealed class LicensingReportMasksWriter(IMigrationSqlDialect migrationDialect) : MigrationCategoryWriter> +{ + public override string CategoryId => MigrationCategoryIds.LicensingReportMasks; + + const string SettingKey = SettingKeys.ReportMasks; + + public override int BatchSize(ServiceControlDbContext dbContext) => migrationDialect.RowsPerStatementFor(dbContext); + + protected override Task PrepareDocuments(ServiceControlDbContext dbContext, IReadOnlyList<(MigrationRow Row, List Document)> documents, CancellationToken cancellationToken = default) + { + List rows = [.. documents.Select(document => SettingRowMapper.Map(SettingKey, document.Document))]; + + return Task.FromResult(new PreparedBatch(async (context, token) => (await migrationDialect.InsertMissing(context, rows, token)).Count, rows.Count, [], Merges: [])); + } + + public override Task Count(ServiceControlDbContext dbContext, CancellationToken cancellationToken = default) => + dbContext.Settings.LongCountAsync(setting => setting.Key == SettingKey, cancellationToken); +} diff --git a/src/ServiceControl.Persistence.EFCore/DataMigration/Writers/LicensingThroughputWriter.cs b/src/ServiceControl.Persistence.EFCore/DataMigration/Writers/LicensingThroughputWriter.cs new file mode 100644 index 0000000000..e0af3e0161 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/DataMigration/Writers/LicensingThroughputWriter.cs @@ -0,0 +1,75 @@ +namespace ServiceControl.Persistence.EFCore.DataMigration.Writers; + +using DbContexts; +using Entities; +using Implementation; +using Infrastructure; +using Microsoft.EntityFrameworkCore; +using Particular.LicensingComponent.Contracts; +using ServiceControl.Persistence.DataMigration; + +/// +/// Writes the licensing endpoints' daily message counts, setting each day's count rather than adding to it, so a copy +/// run twice leaves the same numbers. A batch naming an endpoint the target does not hold is refused with an +/// naming that endpoint and its source. A day whose key repeats in the batch +/// keeps its last row, and the others are counted as already present. +/// +sealed class LicensingThroughputWriter(IMigrationSqlDialect migrationDialect) : MigrationCategoryWriter +{ + public override string CategoryId => MigrationCategoryIds.LicensingThroughput; + + public override int BatchSize(ServiceControlDbContext dbContext) => migrationDialect.RowsPerStatementFor(dbContext); + + public override Task Count(ServiceControlDbContext dbContext, CancellationToken cancellationToken = default) => + dbContext.LicensingEndpointThroughput.LongCountAsync(cancellationToken); + + protected override async Task PrepareDocuments(ServiceControlDbContext dbContext, IReadOnlyList<(MigrationRow Row, ThroughputDayRow Document)> documents, CancellationToken cancellationToken = default) + { + var insert = MigrationInsert.For(dbContext); + var nameComparer = migrationDialect.KeyComparer(typeof(LicensingEndpointThroughputEntity), nameof(LicensingEndpointThroughputEntity.NormalizedName)); + var mapped = new List(documents.Count); + var endpoints = new List(documents.Count); + var skips = new List<(string SourceId, MigrationSkipReason Reason, string Detail)>(); + + // Date order is the order the product's own throughput write takes row locks in. + foreach (var (row, day) in documents.OrderBy(document => document.Document.DateUtc).ThenBy(document => document.Row.SourceId, StringComparer.Ordinal)) + { + if (day.Endpoint.Name is null) + { + skips.Add((row.SourceId, MigrationSkipReason.RequiredValueMissing, "LicensingEndpointThroughput.NormalizedName is NOT NULL and the day's endpoint has no name")); + continue; + } + + var entity = LicensingThroughputRowMapper.Map(day); + + if (UnstorableValueSkip(insert, migrationDialect, row, entity) is { } unstorable) + { + skips.Add(unstorable); + continue; + } + + mapped.Add(entity); + endpoints.Add(day.Endpoint); + } + + foreach (var endpoint in endpoints.Distinct()) + { + var normalizedName = LicensingDataStore.Normalize(endpoint.Name); + + // Checked in SQL rather than in memory, so the name compares under the column's collation as the foreign key does. + if (!await dbContext.LicensingEndpoints.AnyAsync(row => row.NormalizedName == normalizedName && row.ThroughputSource == endpoint.ThroughputSource, cancellationToken)) + { + throw new InvalidOperationException($"Endpoint {endpoint.Name} from {endpoint.ThroughputSource} does not exist in the migration target, so its daily throughput cannot be copied. LicensingThroughput copies only after LicensingEndpoints has finished."); + } + } + + List rows = + [ + .. mapped + .GroupBy(day => (day.ThroughputSource, day.DateUtc)) + .SelectMany(sameDay => sameDay.GroupBy(day => day.NormalizedName, nameComparer).Select(sameKey => sameKey.Last())) + ]; + + return new PreparedBatch((context, token) => migrationDialect.SetLicensingEndpointThroughput(context, rows, token), mapped.Count, skips, Merges: []); + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs index 2683c5069c..3b13887cb5 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs @@ -328,5 +328,5 @@ static Endpoint ToEndpoint(LicensingEndpointEntity row, DateOnly? lastCollectedD LastCollectedDate = lastCollectedDate ?? default }; - static string Normalize(string name) => name.ToLowerInvariant(); + internal static string Normalize(string name) => name.ToLowerInvariant(); } diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/IMigrationSqlDialect.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/IMigrationSqlDialect.cs index 67f875e037..951088d8cb 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/IMigrationSqlDialect.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/IMigrationSqlDialect.cs @@ -1,6 +1,7 @@ namespace ServiceControl.Persistence.EFCore.Infrastructure; using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Entities; /// /// The migration's provider-specific SQL. Statements run inside the caller's transaction and insert only what is absent. @@ -39,4 +40,12 @@ public interface IMigrationSqlDialect /// Inserts the rows whose key the table does not hold, merges any whose keys the database treats as one, and returns the rows it inserted. /// Task> InsertMissing(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken = default) where TEntity : class; + + /// + /// Sets each day's message count: a day the table does not hold is inserted, and a day it holds gets its count replaced, never added to. + /// The rows go to the database in the order given. Every row's endpoint must already be in the table, or the foreign key fails the statement. + /// Two rows with the same key in one call are not allowed on PostgreSQL, which refuses the statement. SQL Server keeps the last of them, which also covers names that differ only by a trailing space. + /// Returns how many rows the statements inserted or updated. It runs on the caller's open transaction, and throws when there is none or when a returned key matches no row given. + /// + Task SetLicensingEndpointThroughput(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken = default); } diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/MigrationInsert.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/MigrationInsert.cs index 947e20afec..b53b885fa9 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/MigrationInsert.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/MigrationInsert.cs @@ -1,6 +1,7 @@ namespace ServiceControl.Persistence.EFCore.Infrastructure; using System.Data.Common; +using System.Reflection; using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore.Infrastructure; using Microsoft.EntityFrameworkCore.Metadata; @@ -16,6 +17,7 @@ public sealed class MigrationInsert where TEntity : class readonly IReadOnlyList properties; readonly IReadOnlyList keyProperties; + readonly IReadOnlyList keyReaders; readonly IReadOnlyList<(IProperty Property, string TableAndColumn, int MaxLength)> boundedText; readonly IReadOnlyList<(IProperty Property, string TableAndColumn)> text; @@ -24,6 +26,7 @@ public sealed class MigrationInsert where TEntity : class Table = quotedTable; this.properties = properties; this.keyProperties = keyProperties; + keyReaders = [.. keyProperties.Select(KeyReader)]; Columns = [.. properties.Select(property => quote(columnName(property)))]; KeyColumns = [.. keyProperties.Select(property => quote(columnName(property)))]; text = [.. properties.Where(property => property.ClrType == typeof(string)).Select(property => (property, $"{tableName}.{columnName(property)}"))]; @@ -157,10 +160,9 @@ public async Task> Execute(ServiceControlDbContext dbCont await using var reader = await command.ExecuteReaderAsync(cancellationToken); while (await reader.ReadAsync(cancellationToken)) { - object?[] key = [.. keyProperties.Select((property, column) => FromProvider(property, reader.GetValue(column)))]; + object?[] key = [.. keyProperties.Select((property, column) => FromProvider(property, keyReaders[column].Invoke(reader, BindingFlags.DoNotWrapExceptions, null, [column], null)!))]; - // A key can come back as a different CLR type than it went in as, a date column read back as - // DateTime say, and a plain lookup would report that as a bare KeyNotFoundException. + // A returned key that no row of the chunk holds exactly would be a bare KeyNotFoundException with a plain lookup, so it is refused by name. inserted.Add(rowsByKey.TryGetValue(key, out var row) ? row : throw new InvalidOperationException($"The {Table} insert returned the key [{string.Join(", ", key.Select(Describe))}], which matches no row in the batch that produced it. Compare it against the key types the batch carried: [{string.Join(", ", keyProperties.Select(property => property.ClrType.Name))}].")); @@ -173,6 +175,14 @@ public async Task> Execute(ServiceControlDbContext dbCont static string Describe(object? value) => value is null ? "null" : $"{value} ({value.GetType().Name})"; + // SQL Server's GetValue returns a date column as DateTime, which no DateOnly key equals, so each key is read as the type its mapping stores. + static MethodInfo KeyReader(IProperty property) + { + var mapping = property.GetRelationalTypeMapping(); + + return typeof(DbDataReader).GetMethod(nameof(DbDataReader.GetFieldValue))!.MakeGenericMethod(mapping.Converter?.ProviderClrType ?? mapping.ClrType); + } + static object? FromProvider(IProperty property, object value) => property.GetRelationalTypeMapping().Converter is { } converter ? converter.ConvertFromProvider(value) : value; diff --git a/src/ServiceControl.Persistence.RavenDB/DataMigration/RavenMigrationSource.cs b/src/ServiceControl.Persistence.RavenDB/DataMigration/RavenMigrationSource.cs index 22eefa3bbf..cfd20d1582 100644 --- a/src/ServiceControl.Persistence.RavenDB/DataMigration/RavenMigrationSource.cs +++ b/src/ServiceControl.Persistence.RavenDB/DataMigration/RavenMigrationSource.cs @@ -27,7 +27,11 @@ sealed class RavenMigrationSource(RavenReadOnlySourceLifecycle lifecycle) : IMig { new KnownEndpointsReader(lifecycle), new EndpointSettingsReader(lifecycle), - new FailedMessagesReader(lifecycle) + new FailedMessagesReader(lifecycle), + new LicensingEndpointsReader(lifecycle), + new LicensingThroughputReader(lifecycle), + new LicensingReportMasksReader(lifecycle), + new LicensedEndpointDetailsReader(lifecycle) }.ToFrozenDictionary(reader => reader.CategoryId, StringComparer.Ordinal); public Task Open(CancellationToken cancellationToken = default) => lifecycle.Open(cancellationToken); diff --git a/src/ServiceControl.Persistence.RavenDB/DataMigration/Readers/LicensedEndpointDetailsReader.cs b/src/ServiceControl.Persistence.RavenDB/DataMigration/Readers/LicensedEndpointDetailsReader.cs new file mode 100644 index 0000000000..477edb9b3b --- /dev/null +++ b/src/ServiceControl.Persistence.RavenDB/DataMigration/Readers/LicensedEndpointDetailsReader.cs @@ -0,0 +1,47 @@ +#nullable enable + +namespace ServiceControl.Persistence.RavenDB.DataMigration.Readers; + +using System; +using System.Collections.Generic; +using System.Collections.ObjectModel; +using System.Runtime.CompilerServices; +using System.Threading; +using Particular.LicensingComponent.Contracts; +using ServiceControl.Persistence.DataMigration; +using ServiceControl.Persistence.RavenDB.Throughput; + +/// +/// Reads the licensed endpoint details document, which becomes one Settings row. A read gives one batch holding the +/// whole document, or no rows when the document does not exist. The batch's cursor is the document's id, and a read +/// resumed after that id gives nothing. +/// +sealed class LicensedEndpointDetailsReader(RavenReadOnlySourceLifecycle lifecycle) : MigrationCategoryReader(lifecycle) +{ + const string DocumentId = LicensingDataStore.LicencedEndpointDetailsDocumentId; + + public override string CategoryId => MigrationCategoryIds.LicensedEndpointDetails; + + public override IAsyncEnumerable Read(string? resumeAfter, int batchSize, CancellationToken cancellationToken = default) => + ReadLicensedEndpointDetails(resumeAfter, cancellationToken); + + async IAsyncEnumerable ReadLicensedEndpointDetails(string? resumeAfter, [EnumeratorCancellation] CancellationToken cancellationToken) + { + if (resumeAfter == DocumentId) + { + yield break; + } + + if (resumeAfter is not null) + { + throw new InvalidOperationException($"The migration cannot resume the '{CategoryId}' category after '{resumeAfter}', because its only cursor is '{DocumentId}'. Run --migration-retry {CategoryId} to copy the category again from the start."); + } + + using var session = Lifecycle.OpenSession(Lifecycle.Settings.ThroughputDatabaseName); + var document = await session.LoadAsync(DocumentId, cancellationToken); + + yield return new MigrationBatch( + document is null ? [] : [new MigrationRow(DocumentId, document, ReadOnlyDictionary.Empty)], + DocumentId); + } +} diff --git a/src/ServiceControl.Persistence.RavenDB/DataMigration/Readers/LicensingEndpointsReader.cs b/src/ServiceControl.Persistence.RavenDB/DataMigration/Readers/LicensingEndpointsReader.cs new file mode 100644 index 0000000000..8ecc76e125 --- /dev/null +++ b/src/ServiceControl.Persistence.RavenDB/DataMigration/Readers/LicensingEndpointsReader.cs @@ -0,0 +1,20 @@ +#nullable enable + +namespace ServiceControl.Persistence.RavenDB.DataMigration.Readers; + +using System.Collections.Generic; +using System.Threading; +using Particular.LicensingComponent.Contracts; +using ServiceControl.Persistence.DataMigration; +using ServiceControl.Persistence.RavenDB.Throughput.Models; + +/// +/// Reads the licensing endpoints out of the throughput database, skipping every document outside the EndpointDocuments collection. +/// +sealed class LicensingEndpointsReader(RavenReadOnlySourceLifecycle lifecycle) : MigrationCategoryReader(lifecycle) +{ + public override string CategoryId => MigrationCategoryIds.LicensingEndpoints; + + public override IAsyncEnumerable Read(string? resumeAfter, int batchSize, CancellationToken cancellationToken = default) => + RavenDocumentStream.ByPrefix(Lifecycle, CategoryId, Lifecycle.Settings.ThroughputDatabaseName, string.Empty, resumeAfter, batchSize, RavenDocumentStream.MaxDocumentsWalkedPerBatch, ThroughputDocuments.Endpoint, cancellationToken); +} diff --git a/src/ServiceControl.Persistence.RavenDB/DataMigration/Readers/LicensingReportMasksReader.cs b/src/ServiceControl.Persistence.RavenDB/DataMigration/Readers/LicensingReportMasksReader.cs new file mode 100644 index 0000000000..6b74a4e20a --- /dev/null +++ b/src/ServiceControl.Persistence.RavenDB/DataMigration/Readers/LicensingReportMasksReader.cs @@ -0,0 +1,47 @@ +#nullable enable + +namespace ServiceControl.Persistence.RavenDB.DataMigration.Readers; + +using System; +using System.Collections.Generic; +using System.Collections.ObjectModel; +using System.Runtime.CompilerServices; +using System.Threading; +using ServiceControl.Persistence.DataMigration; +using ServiceControl.Persistence.RavenDB.Throughput; +using ServiceControl.Persistence.RavenDB.Throughput.Models; + +/// +/// Reads the report masks document, whose list of masked strings becomes one Settings row. A read gives one batch +/// holding the list, or no rows when the document does not exist, and a list stored as null reads as an empty one. +/// The batch's cursor is the document's id, and a read resumed after that id gives nothing. +/// +sealed class LicensingReportMasksReader(RavenReadOnlySourceLifecycle lifecycle) : MigrationCategoryReader>(lifecycle) +{ + const string DocumentId = LicensingDataStore.ReportMasksDocumentId; + + public override string CategoryId => MigrationCategoryIds.LicensingReportMasks; + + public override IAsyncEnumerable Read(string? resumeAfter, int batchSize, CancellationToken cancellationToken = default) => + ReadReportMasks(resumeAfter, cancellationToken); + + async IAsyncEnumerable ReadReportMasks(string? resumeAfter, [EnumeratorCancellation] CancellationToken cancellationToken) + { + if (resumeAfter == DocumentId) + { + yield break; + } + + if (resumeAfter is not null) + { + throw new InvalidOperationException($"The migration cannot resume the '{CategoryId}' category after '{resumeAfter}', because its only cursor is '{DocumentId}'. Run --migration-retry {CategoryId} to copy the category again from the start."); + } + + using var session = Lifecycle.OpenSession(Lifecycle.Settings.ThroughputDatabaseName); + var document = await session.LoadAsync(DocumentId, cancellationToken); + + yield return new MigrationBatch( + document is null ? [] : [new MigrationRow(DocumentId, document.MaskedStrings ?? [], ReadOnlyDictionary.Empty)], + DocumentId); + } +} diff --git a/src/ServiceControl.Persistence.RavenDB/DataMigration/Readers/LicensingThroughputReader.cs b/src/ServiceControl.Persistence.RavenDB/DataMigration/Readers/LicensingThroughputReader.cs new file mode 100644 index 0000000000..7a69b34118 --- /dev/null +++ b/src/ServiceControl.Persistence.RavenDB/DataMigration/Readers/LicensingThroughputReader.cs @@ -0,0 +1,56 @@ +#nullable enable + +namespace ServiceControl.Persistence.RavenDB.DataMigration.Readers; + +using System; +using System.Collections.Generic; +using System.Collections.ObjectModel; +using System.Globalization; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using Particular.LicensingComponent.Contracts; +using Raven.Client.Documents.Session.TimeSeries; +using ServiceControl.Persistence.DataMigration; +using ServiceControl.Persistence.RavenDB.Throughput; +using ServiceControl.Persistence.RavenDB.Throughput.Models; + +/// +/// Reads each licensing endpoint's daily message counts out of the throughput database, one endpoint per batch, +/// with the batch's cursor on that endpoint's document id. +/// +sealed class LicensingThroughputReader(RavenReadOnlySourceLifecycle lifecycle) : MigrationCategoryReader(lifecycle) +{ + public override string CategoryId => MigrationCategoryIds.LicensingThroughput; + + public override IAsyncEnumerable Read(string? resumeAfter, int batchSize, CancellationToken cancellationToken = default) => + ReadThroughputDays(resumeAfter, cancellationToken); + + async IAsyncEnumerable ReadThroughputDays(string? resumeAfter, [EnumeratorCancellation] CancellationToken cancellationToken) + { + // One endpoint per batch, because the batch size counts rows and one endpoint can bring a year or more of days. + await foreach (var endpoints in RavenDocumentStream.ByPrefix(Lifecycle, CategoryId, Lifecycle.Settings.ThroughputDatabaseName, string.Empty, resumeAfter, 1, RavenDocumentStream.MaxDocumentsWalkedPerBatch, ThroughputDocuments.Endpoint, cancellationToken)) + { + var days = new List(); + + foreach (var endpoint in endpoints.Rows) + { + var id = ((Endpoint)endpoint.Document).Id; + + foreach (var point in await ReadThroughputSeries(endpoint.SourceId, cancellationToken)) + { + days.Add(new MigrationRow($"{endpoint.SourceId}|{point.Timestamp.ToString("yyyy-MM-dd", CultureInfo.InvariantCulture)}", new ThroughputDayRow(id, DateOnly.FromDateTime(point.Timestamp), (long)point.Value), ReadOnlyDictionary.Empty)); + } + } + + yield return new MigrationBatch(days, endpoints.Cursor); + } + } + + // Each series read is a request, and one session refuses more than MaxNumberOfRequestsPerSession of them. + async Task ReadThroughputSeries(string documentId, CancellationToken cancellationToken) + { + using var session = Lifecycle.OpenSession(Lifecycle.Settings.ThroughputDatabaseName); + return await session.IncrementalTimeSeriesFor(documentId, LicensingDataStore.ThroughputTimeSeriesName).GetAsync(token: cancellationToken) ?? []; + } +} diff --git a/src/ServiceControl.Persistence.RavenDB/DataMigration/ThroughputDocuments.cs b/src/ServiceControl.Persistence.RavenDB/DataMigration/ThroughputDocuments.cs new file mode 100644 index 0000000000..b6a475ac3d --- /dev/null +++ b/src/ServiceControl.Persistence.RavenDB/DataMigration/ThroughputDocuments.cs @@ -0,0 +1,31 @@ +#nullable enable + +namespace ServiceControl.Persistence.RavenDB.DataMigration; + +using System; +using System.Collections.ObjectModel; +using Raven.Client; +using Raven.Client.Documents.Commands; +using Raven.Client.Documents.Conventions; +using ServiceControl.Persistence.DataMigration; +using ServiceControl.Persistence.RavenDB.Throughput; +using ServiceControl.Persistence.RavenDB.Throughput.Models; + +/// +/// What the throughput database's readers share: which documents are endpoints, and how an endpoint document becomes a row. +/// +static class ThroughputDocuments +{ + static readonly string EndpointCollection = DocumentConventions.DefaultGetCollectionName(typeof(EndpointDocument)); + + /// + /// Turns one streamed document into a row that carries an . Returns null when the document is not in the EndpointDocuments collection, which is the only place the product reads endpoints from. + /// The throughput database gives its documents no id prefix, so the stream also hands back its singleton documents and anything else stored there. + /// + public static MigrationRow? Endpoint(StreamResult entry) => + IsEndpoint(entry) ? new MigrationRow(entry.Id, entry.Document.ToEndpoint(), ReadOnlyDictionary.Empty) : null; + + // RavenDB matches collection names ignoring case. + static bool IsEndpoint(StreamResult entry) => + entry.Metadata.TryGetValue(Constants.Documents.Metadata.Collection, out var collection) && string.Equals(collection?.ToString(), EndpointCollection, StringComparison.OrdinalIgnoreCase); +} diff --git a/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs b/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs index 6e14a20845..f8c3a72ebc 100644 --- a/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs @@ -21,10 +21,10 @@ class LicensingDataStore( TimeProvider timeProvider) : ILicensingDataStore { internal const string ThroughputTimeSeriesName = "INC: throughput data"; - const string AuditServiceMetadataDocumentId = "AuditServiceMetadata"; - const string BrokerMetadataDocumentId = "BrokerMetadata"; - const string ReportMasksDocumentId = "ReportMasks"; - const string LicencedEndpointDetailsDocumentId = "LicensedEndpointDetails"; + internal const string AuditServiceMetadataDocumentId = "AuditServiceMetadata"; + internal const string BrokerMetadataDocumentId = "BrokerMetadata"; + internal const string ReportMasksDocumentId = "ReportMasks"; + internal const string LicencedEndpointDetailsDocumentId = "LicensedEndpointDetails"; static readonly AuditServiceMetadata DefaultAuditServiceMetadata = new([], []); static readonly BrokerMetadata DefaultBrokerMetadata = new(null, []); diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/LicensingEndpointsSourceTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/LicensingEndpointsSourceTests.cs new file mode 100644 index 0000000000..e07ff573fa --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/LicensingEndpointsSourceTests.cs @@ -0,0 +1,72 @@ +namespace ServiceControl.Persistence.Tests.RavenDB.DataMigration; + +using System.Linq; +using System.Threading.Tasks; +using NUnit.Framework; +using Particular.LicensingComponent.Contracts; +using ServiceControl.Persistence.DataMigration; +using RavenLicensingDataStore = ServiceControl.Persistence.RavenDB.Throughput.LicensingDataStore; +using ServiceControl.Persistence.RavenDB.Throughput.Models; + +class LicensingEndpointsSourceTests : RavenMigrationSourceTestBase +{ + [Test] + public async Task Reads_endpoints_and_skips_the_four_singleton_documents() + { + using (var session = DocumentStore.OpenAsyncSession(ThroughputDatabaseName)) + { + await session.StoreAsync(new EndpointDocument(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker)) + { + SanitizedName = "sales.orders" + }, "Sales.Orders/Broker"); + + await session.StoreAsync(new { }, RavenLicensingDataStore.AuditServiceMetadataDocumentId); + await session.StoreAsync(new { MaskedStrings = new[] { "secret" } }, RavenLicensingDataStore.ReportMasksDocumentId); + await session.StoreAsync(new { }, RavenLicensingDataStore.BrokerMetadataDocumentId); + await session.StoreAsync(new { LicenseId = "abc" }, RavenLicensingDataStore.LicencedEndpointDetailsDocumentId); + + await session.SaveChangesAsync(); + } + + await using var source = await OpenMigrationSource(); + var category = MigrationCategoryRegistry.All.Single(entry => entry.Id == "LicensingEndpoints"); + + var batches = await CollectBatches(source, category); + var rows = batches.SelectMany(batch => batch.Rows).ToList(); + + Assert.That(rows, Has.Count.EqualTo(1)); + var endpoint = (Endpoint)rows[0].Document; + Assert.That(endpoint.Id.Name, Is.EqualTo("Sales.Orders")); + } + + [Test] + public async Task A_document_of_another_collection_is_neither_read_nor_counted() + { + using (var session = DocumentStore.OpenAsyncSession(ThroughputDatabaseName)) + { + await session.StoreAsync(new EndpointDocument(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker)) + { + SanitizedName = "sales.orders" + }, "Sales.Orders/Broker"); + + var stray = new { Setting = "left by an older version" }; + await session.StoreAsync(stray, "LegacySettings/1"); + session.Advanced.GetMetadataFor(stray)["@collection"] = "LegacySettings"; + + await session.SaveChangesAsync(); + } + + await using var source = await OpenMigrationSource(); + var category = MigrationCategoryRegistry.All.Single(entry => entry.Id == MigrationCategoryIds.LicensingEndpoints); + + var rows = (await CollectBatches(source, category)).SelectMany(batch => batch.Rows).ToList(); + var count = await source.Count(category); + + using (Assert.EnterMultipleScope()) + { + Assert.That(rows.Select(row => row.SourceId), Has.None.StartsWith("LegacySettings/"), "the product reads endpoints only from the EndpointDocuments collection, so a document of any other kind is not an endpoint"); + Assert.That(rows, Has.Count.EqualTo(1)); + Assert.That(count, Is.EqualTo(rows.Count), "the count reads what the copy reads, so --migration-verify expects no row for the stray document"); + } + } +} diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/LicensingSettingsSourceTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/LicensingSettingsSourceTests.cs new file mode 100644 index 0000000000..6959b67b5a --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/LicensingSettingsSourceTests.cs @@ -0,0 +1,119 @@ +namespace ServiceControl.Persistence.Tests.RavenDB.DataMigration; + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text.Json; +using System.Threading.Tasks; +using NUnit.Framework; +using Particular.LicensingComponent.Contracts; +using ServiceControl.Persistence.DataMigration; +using ServiceControl.Persistence.RavenDB.Throughput.Models; +using RavenLicensingDataStore = ServiceControl.Persistence.RavenDB.Throughput.LicensingDataStore; + +class LicensingSettingsSourceTests : RavenMigrationSourceTestBase +{ + [Test] + public async Task Reads_the_report_masks_document_as_its_list_of_masked_strings() + { + await Seed(RavenLicensingDataStore.ReportMasksDocumentId, new ReportConfigurationDocument { MaskedStrings = ["contoso", "fabrikam"] }); + + var batches = await Read(MigrationCategoryIds.LicensingReportMasks); + + var row = batches.SelectMany(batch => batch.Rows).Single(); + using (Assert.EnterMultipleScope()) + { + Assert.That(row.SourceId, Is.EqualTo("ReportMasks")); + Assert.That((List)row.Document, Is.EqualTo(new[] { "contoso", "fabrikam" })); + Assert.That(batches.Last().Cursor, Is.EqualTo("ReportMasks")); + } + } + + [Test] + public async Task Reads_the_licensed_endpoint_details_document_whole() + { + var seeded = Details; + await Seed(RavenLicensingDataStore.LicencedEndpointDetailsDocumentId, seeded); + + var batches = await Read(MigrationCategoryIds.LicensedEndpointDetails); + + var details = (LicensedEndpointDetails)batches.SelectMany(batch => batch.Rows).Single().Document; + using (Assert.EnterMultipleScope()) + { + Assert.That(JsonSerializer.Serialize(details), Is.EqualTo(JsonSerializer.Serialize(seeded)), "every field, including the endpoint, queue and product arrays, must survive the read"); + Assert.That(batches.Last().Cursor, Is.EqualTo("LicensedEndpointDetails")); + } + } + + [TestCase(MigrationCategoryIds.LicensingReportMasks, "ReportMasks")] + [TestCase(MigrationCategoryIds.LicensedEndpointDetails, "LicensedEndpointDetails")] + public async Task An_absent_document_gives_one_empty_batch_that_still_records_the_cursor(string categoryId, string documentId) + { + var batches = await Read(categoryId); + + using (Assert.EnterMultipleScope()) + { + Assert.That(batches.SelectMany(batch => batch.Rows), Is.Empty); + Assert.That(batches.Select(batch => batch.Cursor), Is.EqualTo(new[] { documentId })); + } + } + + [TestCase(MigrationCategoryIds.LicensingReportMasks, "ReportMasks")] + [TestCase(MigrationCategoryIds.LicensedEndpointDetails, "LicensedEndpointDetails")] + public async Task A_read_resumed_after_the_documents_own_id_gives_nothing(string categoryId, string documentId) + { + await Seed("ReportMasks", new ReportConfigurationDocument { MaskedStrings = ["contoso"] }); + await Seed("LicensedEndpointDetails", new LicensedEndpointDetails { LicenseId = "abc-123" }); + + var batches = await Read(categoryId, documentId); + + Assert.That(batches, Is.Empty); + } + + [TestCase(MigrationCategoryIds.LicensingReportMasks)] + [TestCase(MigrationCategoryIds.LicensedEndpointDetails)] + public async Task A_read_resumed_after_a_cursor_the_reader_never_issued_is_refused(string categoryId) + { + await Seed("ReportMasks", new ReportConfigurationDocument { MaskedStrings = ["contoso"] }); + await Seed("LicensedEndpointDetails", new LicensedEndpointDetails { LicenseId = "abc-123" }); + + var exception = Assert.ThrowsAsync(() => Read(categoryId, "Sales.Orders/Broker")); + + Assert.That(exception!.Message, Does.Contain(categoryId).And.Contain("'Sales.Orders/Broker'")); + } + + [Test] + public async Task A_report_masks_document_whose_list_is_null_reads_as_an_empty_list() + { + await Seed(RavenLicensingDataStore.ReportMasksDocumentId, new ReportConfigurationDocument { MaskedStrings = null }); + + var batches = await Read(MigrationCategoryIds.LicensingReportMasks); + + var row = batches.SelectMany(batch => batch.Rows).Single(); + Assert.That(row.Document, Is.InstanceOf>().And.Empty, "the product shows a null list as no masks, and a null document halts the writer with a reader and writer mismatch"); + } + + 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 }] + }; + + async Task Seed(string id, object document) + { + using var session = DocumentStore.OpenAsyncSession(ThroughputDatabaseName); + await session.StoreAsync(document, id); + await session.SaveChangesAsync(); + } + + async Task> Read(string categoryId, string resumeAfter = null) + { + await using var source = await OpenMigrationSource(); + return await CollectBatches(source, MigrationCategoryRegistry.All.Single(entry => entry.Id == categoryId), resumeAfter); + } +} diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/LicensingThroughputSourceTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/LicensingThroughputSourceTests.cs new file mode 100644 index 0000000000..1ce913e7ca --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/LicensingThroughputSourceTests.cs @@ -0,0 +1,70 @@ +namespace ServiceControl.Persistence.Tests.RavenDB.DataMigration; + +using System; +using System.Linq; +using System.Threading.Tasks; +using NUnit.Framework; +using Particular.LicensingComponent.Contracts; +using Raven.Client.Documents.Conventions; +using ServiceControl.Persistence.DataMigration; +using RavenLicensingDataStore = ServiceControl.Persistence.RavenDB.Throughput.LicensingDataStore; +using ServiceControl.Persistence.RavenDB.Throughput.Models; + +class LicensingThroughputSourceTests : RavenMigrationSourceTestBase +{ + [Test] + public async Task Reads_one_endpoints_whole_time_series_as_one_batch_and_cursors_on_its_document_id() + { + var documentId = "Sales.Orders/Broker"; + + using (var session = DocumentStore.OpenAsyncSession(ThroughputDatabaseName)) + { + await session.StoreAsync(new EndpointDocument(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker)), documentId); + await session.SaveChangesAsync(); + + session.IncrementalTimeSeriesFor(documentId, RavenLicensingDataStore.ThroughputTimeSeriesName) + .Increment(new DateTime(2026, 1, 1), 100); + session.IncrementalTimeSeriesFor(documentId, RavenLicensingDataStore.ThroughputTimeSeriesName) + .Increment(new DateTime(2026, 1, 2), 250); + await session.SaveChangesAsync(); + } + + await using var source = await OpenMigrationSource(); + var category = MigrationCategoryRegistry.All.Single(entry => entry.Id == "LicensingThroughput"); + + var batches = await CollectBatches(source, category); + + Assert.That(batches, Has.Count.EqualTo(1)); + Assert.That(batches[0].Cursor, Is.EqualTo(documentId)); + Assert.That(batches[0].Rows.Select(row => row.Document), Is.EqualTo(new[] + { + new ThroughputDayRow(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker), new DateOnly(2026, 1, 1), 100), + new ThroughputDayRow(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker), new DateOnly(2026, 1, 2), 250) + }), "each row carries the endpoint its count belongs to, because EndpointDailyThroughput does not, and the day and total RavenDB holds"); + Assert.That(batches[0].Rows.Select(row => ((ThroughputDayRow)row.Document).Endpoint.Name), Is.EqualTo(new[] { "Sales.Orders", "Sales.Orders" }), "the endpoint's name keeps its case, which the record's equality ignores"); + } + + [Test] + public async Task Reads_more_endpoints_than_one_session_accepts_requests() + { + var endpointCount = new DocumentConventions().MaxNumberOfRequestsPerSession + 1; + + using (var session = DocumentStore.OpenAsyncSession(ThroughputDatabaseName)) + { + for (var index = 0; index < endpointCount; index++) + { + await session.StoreAsync(new EndpointDocument(new EndpointIdentifier($"Endpoint{index}", ThroughputSource.Broker)), $"Endpoint{index}/Broker"); + } + + await session.SaveChangesAsync(); + } + + await using var source = await OpenMigrationSource(); + var category = MigrationCategoryRegistry.All.Single(entry => entry.Id == "LicensingThroughput"); + + var batches = await CollectBatches(source, category); + + Assert.That(batches, Has.Count.EqualTo(endpointCount), "each endpoint's series is its own request, and one session refuses this many"); + Assert.That(await source.Count(category), Is.Zero); + } +} diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/RavenMigrationSourceTestBase.cs b/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/RavenMigrationSourceTestBase.cs index 358c3f9943..f940d6f8ba 100644 --- a/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/RavenMigrationSourceTestBase.cs +++ b/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/RavenMigrationSourceTestBase.cs @@ -13,6 +13,8 @@ abstract class RavenMigrationSourceTestBase : RavenPersistenceTestBase static readonly SettingsRootNamespace SettingsRoot = new("ServiceControl"); static readonly object SettingsGate = new(); + protected string ThroughputDatabaseName => ((RavenPersisterSettings)PersistenceSettings).ThroughputDatabaseName; + protected async Task OpenMigrationSource() { var settings = (RavenPersisterSettings)PersistenceSettings; diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/RavenMigrationSourceTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/RavenMigrationSourceTests.cs index e97f52a466..a1da5edf80 100644 --- a/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/RavenMigrationSourceTests.cs +++ b/src/ServiceControl.Persistence.Tests.RavenDB/DataMigration/RavenMigrationSourceTests.cs @@ -5,9 +5,12 @@ namespace ServiceControl.Persistence.Tests.RavenDB.DataMigration; using System.Linq; using System.Threading.Tasks; using NUnit.Framework; +using Particular.LicensingComponent.Contracts; using Raven.Client.Documents.Operations.Indexes; using ServiceControl.Operations; using ServiceControl.Persistence.DataMigration; +using ServiceControl.Persistence.RavenDB.Throughput.Models; +using RavenLicensingDataStore = ServiceControl.Persistence.RavenDB.Throughput.LicensingDataStore; class RavenMigrationSourceTests : RavenMigrationSourceTestBase { @@ -126,7 +129,11 @@ public async Task Reading_every_category_creates_no_index() { [MigrationCategoryIds.EndpointSettings] = tests => tests.SeedEndpointSettings(2), [MigrationCategoryIds.KnownEndpoints] = tests => tests.MonitoringDataStore.CreateIfNotExists(new EndpointDetails { Name = "Sales.Orders", HostId = Guid.NewGuid(), Host = "HOST01" }), - [MigrationCategoryIds.UnresolvedAndRetryIssuedFailedMessages] = tests => tests.SeedFailedMessage(new IngestedFailure().ToFailedMessage()) + [MigrationCategoryIds.UnresolvedAndRetryIssuedFailedMessages] = tests => tests.SeedFailedMessage(new IngestedFailure().ToFailedMessage()), + [MigrationCategoryIds.LicensingEndpoints] = tests => tests.SeedThroughputDocument("Sales.Orders/Broker", new EndpointDocument(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker))), + [MigrationCategoryIds.LicensingThroughput] = tests => tests.SeedThroughputDocument("Sales.Orders/Broker", new EndpointDocument(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker))), + [MigrationCategoryIds.LicensingReportMasks] = tests => tests.SeedThroughputDocument(RavenLicensingDataStore.ReportMasksDocumentId, new ReportConfigurationDocument { MaskedStrings = ["contoso"] }), + [MigrationCategoryIds.LicensedEndpointDetails] = tests => tests.SeedThroughputDocument(RavenLicensingDataStore.LicencedEndpointDetailsDocumentId, new LicensedEndpointDetails { LicenseId = "abc-123" }) }; string DatabaseName => ((RavenPersisterSettings)PersistenceSettings).DatabaseName; @@ -138,6 +145,13 @@ async Task IndexNames() => static readonly MigrationCategory EndpointSettingsCategory = MigrationCategoryRegistry.Find("EndpointSettings"); + async Task SeedThroughputDocument(string id, object document) + { + using var session = DocumentStore.OpenAsyncSession(ThroughputDatabaseName); + await session.StoreAsync(document, id); + await session.SaveChangesAsync(); + } + async Task SeedEndpointSettings(int count) { for (var index = 0; index < count; index++) diff --git a/src/ServiceControl.Persistence.Tests/EFCore/Migration/LicensedEndpointDetailsMigrationTargetTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/Migration/LicensedEndpointDetailsMigrationTargetTests.cs new file mode 100644 index 0000000000..1c6cba5d3d --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/Migration/LicensedEndpointDetailsMigrationTargetTests.cs @@ -0,0 +1,29 @@ +namespace ServiceControl.Persistence.Tests; + +using System.Linq; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using Particular.LicensingComponent.Contracts; +using ServiceControl.Persistence.DataMigration; + +class LicensedEndpointDetailsMigrationTargetTests : PersistenceTestBase +{ + [SetUp] + public Task OpenTarget() => Target.Open(); + + IMigrationTarget Target => ServiceProvider.GetRequiredService(); + + static readonly MigrationCategory LicensedEndpointDetailsCategory = MigrationCategoryRegistry.All.Single(category => category.Id == MigrationCategoryIds.LicensedEndpointDetails); + + [Test] + public async Task Count_is_the_licensed_endpoint_details_row_alone() + { + await LicensingDataStore.SaveReportMasks(["contoso"]); + Assert.That(await Target.Count(LicensedEndpointDetailsCategory), Is.Zero, "the report masks row is not a licensed endpoint details row"); + + await LicensingDataStore.SaveLicensedEndpointDetails(new LicensedEndpointDetails { LicenseId = "abc-123" }); + + Assert.That(await Target.Count(LicensedEndpointDetailsCategory), Is.EqualTo(1), "counted as the whole Settings table, this would also count the report masks row and every other row in the table"); + } +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/Migration/LicensingEndpointsMigrationTargetTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/Migration/LicensingEndpointsMigrationTargetTests.cs new file mode 100644 index 0000000000..2eb2c74e25 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/Migration/LicensingEndpointsMigrationTargetTests.cs @@ -0,0 +1,63 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using Particular.LicensingComponent.Contracts; +using ServiceControl.Persistence.DataMigration; +using ServiceControl.Persistence.EFCore.DataMigration.Writers; +using ServiceControl.Persistence.EFCore.Infrastructure; + +class LicensingEndpointsMigrationTargetTests : PersistenceTestBase +{ + [SetUp] + public Task OpenTarget() => Target.Open(); + + IMigrationTarget Target => ServiceProvider.GetRequiredService(); + + static readonly MigrationCategory LicensingEndpointsCategory = MigrationCategoryRegistry.All.Single(category => category.Id == MigrationCategoryIds.LicensingEndpoints); + + [Test] + public async Task Count_is_the_licensing_endpoint_rows_the_target_holds() + { + await LicensingDataStore.SaveEndpoint(new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" }); + await LicensingDataStore.SaveEndpoint(new Endpoint("Sales.Orders", ThroughputSource.Audit) { SanitizedName = "sales.orders" }); + await LicensingDataStore.RecordEndpointThroughput("Sales.Orders", ThroughputSource.Broker, new DateOnly(2026, 1, 1), 5); + await LicensingDataStore.RecordEndpointThroughput("Sales.Orders", ThroughputSource.Broker, new DateOnly(2026, 1, 2), 5); + await LicensingDataStore.RecordEndpointThroughput("Sales.Orders", ThroughputSource.Audit, new DateOnly(2026, 1, 1), 5); + + Assert.That(await Target.Count(LicensingEndpointsCategory), Is.EqualTo(2), "one row per name and source, and not the three throughput days that hang off them"); + } + + [Test] + public async Task A_licensing_endpoint_with_a_value_longer_than_its_column_is_skipped_as_value_too_long_and_the_rest_copy() + { + var fitsExactly = new Endpoint(new string('n', 300), ThroughputSource.Broker) { SanitizedName = new string('s', 450) }; + var longName = new Endpoint(new string('n', 301), ThroughputSource.Audit) { SanitizedName = "sales.orders" }; + var longSanitizedName = new Endpoint("Billing", ThroughputSource.Broker) { SanitizedName = new string('s', 451) }; + var batch = BatchOf(("Endpoint/1", fitsExactly), ("Endpoint/2", longName), ("Endpoint/3", longSanitizedName)); + + var details = await PreparedSkips.DetailsFrom(ServiceProvider, new LicensingEndpointsWriter(ServiceProvider.GetRequiredService()), batch); + var result = await Target.Write(LicensingEndpointsCategory, batch, CheckpointAfter("Endpoint/3")); + + using (Assert.EnterMultipleScope()) + { + Assert.That(result.Copied, Is.EqualTo(1), "a value exactly as long as its column fits"); + Assert.That(result.SkippedIds, Is.EqualTo(new[] { "Endpoint/2", "Endpoint/3" })); + Assert.That(result.SkipReasons, Is.EqualTo(new Dictionary { [MigrationSkipReason.ValueTooLong] = 2 }), "an over-long value would fail the whole insert and halt the category"); + Assert.That((await LicensingDataStore.GetAllEndpoints(includePlatformEndpoints: true)).Select(endpoint => endpoint.Id.Name), Is.EqualTo(new[] { new string('n', 300) })); + Assert.That(details, Has.Count.EqualTo(2)); + Assert.That(details[0], Does.EndWith("takes at most 300 characters and the value holds 301, on document 'Endpoint/2'")); + Assert.That(details[1], Does.EndWith("takes at most 450 characters and the value holds 451, on document 'Endpoint/3'")); + } + } + + static MigrationBatch BatchOf(params (string Id, object Document)[] rows) => + new([.. rows.Select(row => new MigrationRow(row.Id, row.Document, new Dictionary()))], rows[^1].Id); + + static MigrationCheckpoint CheckpointAfter(string cursor) => + new(MigrationCategoryIds.LicensingEndpoints, MigrationCategoryState.InProgress, cursor, 0, 0, null, null, null, null, null, null); +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/Migration/LicensingMigrationMapperCoverageTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/Migration/LicensingMigrationMapperCoverageTests.cs new file mode 100644 index 0000000000..f7c2d37322 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/Migration/LicensingMigrationMapperCoverageTests.cs @@ -0,0 +1,51 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using Particular.LicensingComponent.Contracts; +using ServiceControl.Persistence.DataMigration; +using ServiceControl.Persistence.EFCore.DataMigration; +using ServiceControl.Persistence.EFCore.DbContexts; + +class LicensingMigrationMapperCoverageTests : PersistenceTestBase +{ + [Test] + public async Task The_endpoint_mapper_sets_every_mapped_property() + { + var endpoint = new Endpoint("Sales.Orders", ThroughputSource.Audit) + { + SanitizedName = "Sales.Orders", + EndpointIndicators = [EndpointIndicator.PlatformEndpoint.ToString()], + UserIndicator = "Sales team", + Scope = "Sales" + }; + + MigrationEntityCoverage.AssertEveryMappedPropertyIsSet(await Model(), LicensingEndpointRowMapper.Map(endpoint)); + } + + [Test] + public async Task The_throughput_mapper_sets_every_mapped_property() + { + var day = new ThroughputDayRow(new EndpointIdentifier("Sales.Orders", ThroughputSource.Audit), new DateOnly(2026, 1, 1), 100); + + var row = LicensingThroughputRowMapper.Map(day); + + MigrationEntityCoverage.AssertEveryMappedPropertyIsSet(await Model(), row); + using (Assert.EnterMultipleScope()) + { + Assert.That(row.NormalizedName, Is.EqualTo("sales.orders"), "the key must be the name the endpoint row is stored under, or the foreign key refuses the day"); + Assert.That(row.ThroughputSource, Is.EqualTo(ThroughputSource.Audit)); + Assert.That(row.DateUtc, Is.EqualTo(new DateOnly(2026, 1, 1))); + Assert.That(row.MessageCount, Is.EqualTo(100)); + } + } + + async Task Model() + { + await using var scope = ServiceProvider.CreateAsyncScope(); + return scope.ServiceProvider.GetRequiredService().Model; + } +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/Migration/LicensingReportMasksMigrationTargetTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/Migration/LicensingReportMasksMigrationTargetTests.cs new file mode 100644 index 0000000000..45faa38e2e --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/Migration/LicensingReportMasksMigrationTargetTests.cs @@ -0,0 +1,29 @@ +namespace ServiceControl.Persistence.Tests; + +using System.Linq; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using Particular.LicensingComponent.Contracts; +using ServiceControl.Persistence.DataMigration; + +class LicensingReportMasksMigrationTargetTests : PersistenceTestBase +{ + [SetUp] + public Task OpenTarget() => Target.Open(); + + IMigrationTarget Target => ServiceProvider.GetRequiredService(); + + static readonly MigrationCategory LicensingReportMasksCategory = MigrationCategoryRegistry.All.Single(category => category.Id == MigrationCategoryIds.LicensingReportMasks); + + [Test] + public async Task Count_is_the_report_masks_row_alone() + { + await LicensingDataStore.SaveLicensedEndpointDetails(new LicensedEndpointDetails { LicenseId = "abc-123" }); + Assert.That(await Target.Count(LicensingReportMasksCategory), Is.Zero, "the licensed endpoint details row is not a report masks row"); + + await LicensingDataStore.SaveReportMasks(["contoso"]); + + Assert.That(await Target.Count(LicensingReportMasksCategory), Is.EqualTo(1), "counted as the whole Settings table, this would also count the licensed endpoint details row and every other row in the table"); + } +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/Migration/LicensingThroughputMigrationTargetTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/Migration/LicensingThroughputMigrationTargetTests.cs new file mode 100644 index 0000000000..5e4b3cea25 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/Migration/LicensingThroughputMigrationTargetTests.cs @@ -0,0 +1,165 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using Particular.LicensingComponent.Contracts; +using ServiceControl.Persistence.DataMigration; +using ServiceControl.Persistence.EFCore.DataMigration.Writers; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Infrastructure; + +class LicensingThroughputMigrationTargetTests : PersistenceTestBase +{ + [SetUp] + public Task OpenTarget() => Target.Open(); + + IMigrationTarget Target => ServiceProvider.GetRequiredService(); + + static readonly MigrationCategory LicensingThroughputCategory = MigrationCategoryRegistry.All.Single(category => category.Id == MigrationCategoryIds.LicensingThroughput); + + [Test] + public async Task Count_is_the_daily_throughput_rows_the_target_holds() + { + await LicensingDataStore.SaveEndpoint(new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" }); + await LicensingDataStore.SaveEndpoint(new Endpoint("Billing", ThroughputSource.Broker) { SanitizedName = "billing" }); + await LicensingDataStore.RecordEndpointThroughput("Sales.Orders", ThroughputSource.Broker, new DateOnly(2026, 1, 1), 5); + await LicensingDataStore.RecordEndpointThroughput("Sales.Orders", ThroughputSource.Broker, new DateOnly(2026, 1, 2), 5); + await LicensingDataStore.RecordEndpointThroughput("Billing", ThroughputSource.Broker, new DateOnly(2026, 1, 1), 5); + + Assert.That(await Target.Count(LicensingThroughputCategory), Is.EqualTo(3), "one row per endpoint and day, and not the two endpoint rows they hang off"); + } + + [Test] + public async Task A_day_whose_endpoint_name_is_longer_than_its_column_is_skipped_as_value_too_long_and_the_rest_copy() + { + await LicensingDataStore.SaveEndpoint(new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" }); + var day = new DateOnly(2026, 1, 1); + var batch = BatchOf( + ("Sales.Orders/Broker|2026-01-01", new ThroughputDayRow(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker), day, 100)), + ("long/Broker|2026-01-01", new ThroughputDayRow(new EndpointIdentifier(new string('n', 301), ThroughputSource.Broker), day, 250))); + + var details = await PreparedSkips.DetailsFrom(ServiceProvider, new LicensingThroughputWriter(ServiceProvider.GetRequiredService()), batch); + var result = await Target.Write(LicensingThroughputCategory, batch, CheckpointAfter("long/Broker|2026-01-01")); + + using (Assert.EnterMultipleScope()) + { + Assert.That(result.Copied, Is.EqualTo(1)); + Assert.That(result.SkippedIds, Is.EqualTo(new[] { "long/Broker|2026-01-01" })); + Assert.That(result.SkipReasons, Is.EqualTo(new Dictionary { [MigrationSkipReason.ValueTooLong] = 1 }), "the target cannot hold an endpoint with this name, so the day must be skipped here rather than fail the whole batch at the endpoint check"); + Assert.That(details, Is.EqualTo(new[] + { + $"{PreparedSkips.TableAndColumnOf(ServiceProvider, nameof(LicensingEndpointThroughputEntity.NormalizedName))} takes at most 300 characters and the value holds 301, on document 'long/Broker|2026-01-01'" + })); + } + } + + [Test] + public async Task A_day_repeated_in_one_batch_keeps_its_last_row_and_counts_the_other_as_already_present() + { + await LicensingDataStore.SaveEndpoint(new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" }); + var day = new DateOnly(2026, 1, 1); + var batch = BatchOf( + ("Sales.Orders/Broker|2026-01-01|b", new ThroughputDayRow(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker), day, 250)), + ("Sales.Orders/Broker|2026-01-01|a", new ThroughputDayRow(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker), day, 100))); + + var result = await Target.Write(LicensingThroughputCategory, batch, CheckpointAfter("Sales.Orders/Broker|2026-01-01|b")); + await using var scope = ServiceProvider.CreateAsyncScope(); + var stored = await scope.ServiceProvider.GetRequiredService().LicensingEndpointThroughput.Select(row => row.MessageCount).SingleAsync(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(result.Copied, Is.EqualTo(1)); + Assert.That(result.AlreadyPresent, Is.EqualTo(1), "PostgreSQL refuses a statement that updates one row twice, so the repeat is dropped before it and counted rather than failing the batch"); + Assert.That(stored, Is.EqualTo(250), "the row whose id sorts last by ordinal order sets the day, whatever order the batch arrives in"); + } + } + + [Test] + public async Task A_later_batch_replaces_the_count_an_earlier_batch_stored_for_the_same_day() + { + await LicensingDataStore.SaveEndpoint(new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" }); + var day = new DateOnly(2026, 1, 1); + + var first = await Target.Write(LicensingThroughputCategory, BatchOf(("Sales.Orders/Broker|2026-01-01", new ThroughputDayRow(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker), day, 100))), CheckpointAfter("Sales.Orders/Broker")); + var second = await Target.Write(LicensingThroughputCategory, BatchOf(("Sales.Orders/Broker|2026-01-01", new ThroughputDayRow(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker), day, 250))), first.Saved); + + using (Assert.EnterMultipleScope()) + { + Assert.That(second.Copied, Is.EqualTo(1), "an update is a row the statement wrote"); + Assert.That(await StoredCounts(), Is.EqualTo(new[] { 250L }), "the count RavenDB holds now replaces the stored one, and is never added to it or ignored"); + } + } + + [Test] + public async Task One_endpoint_with_more_days_than_one_statement_takes_stores_every_day_with_its_own_count() + { + await LicensingDataStore.SaveEndpoint(new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" }); + var dialect = ServiceProvider.GetRequiredService(); + int dayCount; + + await using (var scope = ServiceProvider.CreateAsyncScope()) + { + dayCount = dialect.RowsPerStatementFor(scope.ServiceProvider.GetRequiredService()) + 1; + } + + var firstDay = new DateOnly(2026, 1, 1); + var days = Enumerable.Range(0, dayCount).Select(offset => (Date: firstDay.AddDays(offset), Count: 1000L + offset)).ToArray(); + var batch = BatchOf([.. days.Select(day => ($"Sales.Orders/Broker|{day.Date:yyyy-MM-dd}", (object)new ThroughputDayRow(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker), day.Date, day.Count)))]); + + var result = await Target.Write(LicensingThroughputCategory, batch, CheckpointAfter("Sales.Orders/Broker")); + + await using var readScope = ServiceProvider.CreateAsyncScope(); + var stored = await readScope.ServiceProvider.GetRequiredService().LicensingEndpointThroughput + .OrderBy(row => row.DateUtc) + .Select(row => new { row.DateUtc, row.MessageCount }) + .ToListAsync(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(result.Copied, Is.EqualTo(dayCount), "the batch runs as two statements, and Copied adds up what both wrote"); + Assert.That(result.AlreadyPresent, Is.Zero, "a day the second statement wrote must not be counted as already present"); + Assert.That(stored.Select(row => (row.DateUtc, row.MessageCount)), Is.EqualTo(days), "every day is stored with its own count, including the day past the first statement"); + } + } + + [Test] + public async Task A_later_batch_sets_a_day_stored_under_a_name_that_differs_only_by_a_trailing_space() + { + var endpoints = MigrationCategoryRegistry.All.Single(category => category.Id == MigrationCategoryIds.LicensingEndpoints); + var endpointBatch = BatchOf( + ("Sales.Orders /Broker", new Endpoint("Sales.Orders ", ThroughputSource.Broker) { SanitizedName = "sales.orders" }), + ("Sales.Orders/Broker", new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" })); + await Target.Write(endpoints, endpointBatch, new MigrationCheckpoint(endpoints.Id, MigrationCategoryState.InProgress, endpointBatch.Cursor, 0, 0, null, null, null, null, null, null)); + var day = new DateOnly(2026, 1, 1); + + var first = await Target.Write(LicensingThroughputCategory, BatchOf(("Sales.Orders /Broker|2026-01-01", new ThroughputDayRow(new EndpointIdentifier("Sales.Orders ", ThroughputSource.Broker), day, 100))), CheckpointAfter("Sales.Orders /Broker")); + var second = await Target.Write(LicensingThroughputCategory, BatchOf(("Sales.Orders/Broker|2026-01-01", new ThroughputDayRow(new EndpointIdentifier("Sales.Orders", ThroughputSource.Broker), day, 250))), first.Saved); + + await using var scope = ServiceProvider.CreateAsyncScope(); + var keyIgnoresTrailingSpace = scope.ServiceProvider.GetRequiredService().Database.ProviderName!.Contains("SqlServer"); + + using (Assert.EnterMultipleScope()) + { + Assert.That(second.Copied, Is.EqualTo(1), "the second batch's day is matched back to the row that produced it, not refused because the stored name keeps its trailing space"); + Assert.That(await StoredCounts(), Is.EqualTo(keyIgnoresTrailingSpace ? new[] { 250L } : new[] { 100L, 250L }), "SQL Server's key makes the two names one day, which the later count sets; PostgreSQL's keeps two"); + } + } + + async Task StoredCounts() + { + await using var scope = ServiceProvider.CreateAsyncScope(); + return await scope.ServiceProvider.GetRequiredService().LicensingEndpointThroughput.Select(row => row.MessageCount).OrderBy(count => count).ToArrayAsync(); + } + + static MigrationBatch BatchOf(params (string Id, object Document)[] rows) => + new([.. rows.Select(row => new MigrationRow(row.Id, row.Document, new Dictionary()))], rows[^1].Id); + + static MigrationCheckpoint CheckpointAfter(string cursor) => + new(MigrationCategoryIds.LicensingThroughput, MigrationCategoryState.InProgress, cursor, 0, 0, null, null, null, null, null, null); +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/Migration/MigrationInsertDateKeyTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/Migration/MigrationInsertDateKeyTests.cs new file mode 100644 index 0000000000..629d6d1684 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/Migration/MigrationInsertDateKeyTests.cs @@ -0,0 +1,36 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using Particular.LicensingComponent.Contracts; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Infrastructure; + +class MigrationInsertDateKeyTests : PersistenceTestBase +{ + [Test] + public async Task A_row_whose_key_holds_a_date_is_matched_back_to_the_row_that_produced_it() + { + await LicensingDataStore.SaveEndpoint(new Endpoint("Sales.Orders", ThroughputSource.Broker) { SanitizedName = "sales.orders" }); + var dialect = ServiceProvider.GetRequiredService(); + var day = new LicensingEndpointThroughputEntity { NormalizedName = "sales.orders", ThroughputSource = ThroughputSource.Broker, DateUtc = new DateOnly(2026, 1, 1), MessageCount = 100 }; + + await using var scope = ServiceProvider.CreateAsyncScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + await dialect.Open(dbContext); + + var inserted = await dbContext.Database.CreateExecutionStrategy().ExecuteAsync(async () => + { + await using var transaction = await dbContext.Database.BeginTransactionAsync(); + var rows = await dialect.InsertMissing(dbContext, [day]); + await transaction.CommitAsync(); + return rows; + }); + + Assert.That(inserted, Is.EqualTo(new[] { day }), "a date key comes back from the database as DateTime unless it is read as the type its mapping stores, and the exact lookup then refuses the row"); + } +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/Migration/MigrationTargetCoverageTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/Migration/MigrationTargetCoverageTests.cs index eb7caebdcc..0cf8345618 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/Migration/MigrationTargetCoverageTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/Migration/MigrationTargetCoverageTests.cs @@ -1,11 +1,15 @@ namespace ServiceControl.Persistence.Tests; using System.Collections.Generic; +using System.Linq; using System.Threading; using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; using NUnit.Framework; using ServiceControl.Persistence.DataMigration; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Entities; class MigrationTargetCoverageTests : PersistenceTestBase { @@ -30,15 +34,29 @@ public void Every_category_the_target_supports_has_a_test_for_a_value_longer_tha { var target = ServiceProvider.GetRequiredService(); - Assert.That(target.SupportedCategoryIds, Is.EquivalentTo(ValueTooLongTests.Keys), + Assert.That(target.SupportedCategoryIds, Is.EquivalentTo(ValueTooLongTests.Keys.Concat(NoValueCanBeTooLong)), "a writer that does not run the length check sends an over-long value to the insert, which fails the whole statement and halts the category; give the new category a ValueTooLong test like those listed here, then add it to this list"); } + [Test] + public async Task The_settings_value_column_has_no_maximum_length_so_no_setting_can_be_too_long() + { + await using var scope = ServiceProvider.CreateAsyncScope(); + var model = scope.ServiceProvider.GetRequiredService().Model; + + Assert.That(model.FindEntityType(typeof(SettingEntity))!.FindProperty(nameof(SettingEntity.Value))!.GetMaxLength(), Is.Null, "NoValueCanBeTooLong exempts the settings categories because of this; give them a value-too-long test if the column gains a limit"); + } + + // Each is one Settings row under a constant key whose value column has no maximum length, so no source value can be too long for it. + static readonly string[] NoValueCanBeTooLong = [MigrationCategoryIds.LicensingReportMasks, MigrationCategoryIds.LicensedEndpointDetails]; + // Uses nameof, so renaming or deleting one of these tests fails the build instead of leaving this list out of date. static readonly Dictionary ValueTooLongTests = new() { [MigrationCategoryIds.KnownEndpoints] = nameof(KnownEndpointsMigrationTargetTests.A_known_endpoint_with_a_value_longer_than_its_column_is_skipped_as_value_too_long_and_the_rest_copy), [MigrationCategoryIds.EndpointSettings] = nameof(EndpointSettingsMigrationTargetTests.A_setting_whose_name_is_longer_than_its_column_is_skipped_as_value_too_long_and_the_rest_copy), - [MigrationCategoryIds.UnresolvedAndRetryIssuedFailedMessages] = nameof(FailedMessagesMigrationTargetTests.A_message_or_group_value_longer_than_its_column_skips_the_message_whole_and_the_rest_copy) + [MigrationCategoryIds.UnresolvedAndRetryIssuedFailedMessages] = nameof(FailedMessagesMigrationTargetTests.A_message_or_group_value_longer_than_its_column_skips_the_message_whole_and_the_rest_copy), + [MigrationCategoryIds.LicensingEndpoints] = nameof(LicensingEndpointsMigrationTargetTests.A_licensing_endpoint_with_a_value_longer_than_its_column_is_skipped_as_value_too_long_and_the_rest_copy), + [MigrationCategoryIds.LicensingThroughput] = nameof(LicensingThroughputMigrationTargetTests.A_day_whose_endpoint_name_is_longer_than_its_column_is_skipped_as_value_too_long_and_the_rest_copy) }; } diff --git a/src/ServiceControl.Persistence.Tests/EFCore/Migration/SettingRowMapperTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/Migration/SettingRowMapperTests.cs new file mode 100644 index 0000000000..84b27862f8 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/Migration/SettingRowMapperTests.cs @@ -0,0 +1,55 @@ +namespace ServiceControl.Persistence.Tests; + +using System.Collections.Generic; +using System.Text.Json; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using Particular.LicensingComponent.Contracts; +using ServiceControl.Persistence.EFCore.DataMigration; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Infrastructure; + +class SettingRowMapperTests : PersistenceTestBase +{ + static readonly List ReportMasks = ["contoso", "fabrikam"]; + + static readonly 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 Sets_every_mapped_property() + { + await using var scope = ServiceProvider.CreateAsyncScope(); + var model = scope.ServiceProvider.GetRequiredService().Model; + + MigrationEntityCoverage.AssertEveryMappedPropertyIsSet(model, SettingRowMapper.Map(SettingKeys.ReportMasks, ReportMasks)); + } + + [Test] + public async Task Licensing_settings_read_back_through_GetSetting_as_the_product_reads_them() + { + await using (var scope = ServiceProvider.CreateAsyncScope()) + { + var dbContext = scope.ServiceProvider.GetRequiredService(); + dbContext.Settings.AddRange(SettingRowMapper.Map(SettingKeys.ReportMasks, ReportMasks), SettingRowMapper.Map(SettingKeys.LicensedEndpointDetails, Details)); + await dbContext.SaveChangesAsync(); + } + + await using (var scope = ServiceProvider.CreateAsyncScope()) + { + var dbContext = scope.ServiceProvider.GetRequiredService(); + + Assert.That(await dbContext.GetSetting>(SettingKeys.ReportMasks), Is.EqualTo(ReportMasks)); + Assert.That(JsonSerializer.Serialize(await dbContext.GetSetting(SettingKeys.LicensedEndpointDetails)), Is.EqualTo(JsonSerializer.Serialize(Details))); + } + } +} diff --git a/src/ServiceControl.Persistence/DataMigration/ThroughputDayRow.cs b/src/ServiceControl.Persistence/DataMigration/ThroughputDayRow.cs new file mode 100644 index 0000000000..bcbfb5431c --- /dev/null +++ b/src/ServiceControl.Persistence/DataMigration/ThroughputDayRow.cs @@ -0,0 +1,10 @@ +namespace ServiceControl.Persistence.DataMigration; + +using System; +using Particular.LicensingComponent.Contracts; + +/// +/// One endpoint's message count for one UTC day, as a licensing throughput row crosses from source to target. +/// Endpoint is the endpoint the count belongs to, DateUtc is the day, and MessageCount is that day's whole total. +/// +public sealed record ThroughputDayRow(EndpointIdentifier Endpoint, DateOnly DateUtc, long MessageCount); diff --git a/src/ServiceControl.Persistence/ServiceControl.Persistence.csproj b/src/ServiceControl.Persistence/ServiceControl.Persistence.csproj index 131f2cfa0e..876ea8a78a 100644 --- a/src/ServiceControl.Persistence/ServiceControl.Persistence.csproj +++ b/src/ServiceControl.Persistence/ServiceControl.Persistence.csproj @@ -6,6 +6,7 @@ +