Implement RemoveEndpoints in SQL stores

src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs

@@ -246,7 +246,21 @@ static async Task<bool> TryRecordEndpointThroughput(ServiceControlDbContext cont
         });
     }
 
-    public Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken = default) => throw new NotImplementedException();
+    public Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken = default) =>
+        ExecuteWithDbContext(async (context, token) =>
+        {
+            foreach (var endpointsBySource in endpointIds.GroupBy(endpoint => endpoint.ThroughputSource))
+            {
+                var normalizedNames = endpointsBySource
+                    .Select(endpoint => Normalize(endpoint.Name))
+                    .Distinct()
+                    .ToList();
+
+                await context.LicensingEndpoints
+                    .Where(endpoint => endpoint.ThroughputSource == endpointsBySource.Key && normalizedNames.Contains(endpoint.NormalizedName))
+                    .ExecuteDeleteAsync(token);
+            }
+        }, cancellationToken);
 
     public Task UpdateUserIndicatorOnEndpoints(List<UpdateUserIndicator> userIndicatorUpdates, CancellationToken cancellationToken = default) =>
         ExecuteWithDbContext(async (context, token) =>

src/ServiceControl.Persistence.Tests/Throughput/EndpointsTests.cs

@@ -47,6 +47,30 @@ public async Task Should_add_new_endpoint_when_name_is_the_same_but_source_diffe
         Assert.That(endpoints.Count(), Is.EqualTo(2));
     }
 
+    [Test]
+    public async Task Should_remove_only_endpoints_matching_name_and_source()
+    {
+        var endpoint1Audit = new Endpoint("Endpoint1", ThroughputSource.Audit);
+        var endpoint1Broker = new Endpoint("Endpoint1", ThroughputSource.Broker);
+        var endpoint2Audit = new Endpoint("Endpoint2", ThroughputSource.Audit);
+        var endpoint3Monitoring = new Endpoint("Endpoint3", ThroughputSource.Monitoring);
+
+        await LicensingDataStore.SaveEndpoint(endpoint1Audit);
+        await LicensingDataStore.SaveEndpoint(endpoint1Broker);
+        await LicensingDataStore.SaveEndpoint(endpoint2Audit);
+        await LicensingDataStore.SaveEndpoint(endpoint3Monitoring);
+
+        await LicensingDataStore.RemoveEndpoints([endpoint1Audit.Id, endpoint3Monitoring.Id]);
+
+        var remainingEndpoints = await LicensingDataStore.GetAllEndpoints(true);
+
+        Assert.That(remainingEndpoints.Select(endpoint => endpoint.Id), Is.EquivalentTo(new[]
+        {
+            endpoint1Broker.Id,
+            endpoint2Audit.Id
+        }));
+    }
+
     [Test]
     public async Task Should_update_endpoint_that_already_has_throughput_with_new_throughput()
     {