Remove decommissioned endpoints from usage reports

src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs

@@ -179,6 +179,15 @@ public Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, Cancella
         return Task.CompletedTask;
     }
 
+    public Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken)
+    {
+        foreach (var id in endpointIds)
+        {
+            endpoints.Remove(id);
+        }
+        return Task.CompletedTask;
+    }
+
     class EndpointCollection : KeyedCollection<EndpointIdentifier, Endpoint>
     {
         protected override EndpointIdentifier GetKeyForItem(Endpoint item) => item.Id;

src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs

@@ -15,6 +15,8 @@ public interface ILicensingDataStore
 
     Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken);
 
+    Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken);
+
     Task<IDictionary<string, IEnumerable<ThroughputData>>> GetEndpointThroughputByQueueName(IList<string> queueNames, CancellationToken cancellationToken);
 
     Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, DateOnly date, long messageCount, CancellationToken cancellationToken) =>

src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs

@@ -203,6 +203,48 @@ await Task.Run(async () =>
         }
     }
 
+    [Test]
+    public async Task Should_only_create_new_endpoint_when_audit_counts_exist()
+    {
+        // Arrange
+        using var tokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(3));
+        var token = tokenSource.Token;
+        var fakeTimeProvider = new FakeTimeProvider();
+
+        var date = DateOnly.FromDateTime(DateTime.UtcNow.AddDays(-1));
+        var auditQuery = new AuditQuery_WithTwoEndpointsAndSelectiveCounts(
+            endpointWithoutCounts: "EndpointNoData",
+            endpointWithCounts: "EndpointWithData",
+            throughputDate: date,
+            throughputCount: 5);
+
+        using var auditThroughputCollectorHostedService = new AuditThroughputCollectorHostedService(
+            NullLogger<AuditThroughputCollectorHostedService>.Instance, configuration.ThroughputSettings, DataStore,
+            auditQuery, fakeTimeProvider)
+        { DelayStart = TimeSpan.Zero };
+
+        // Act
+        await auditThroughputCollectorHostedService.StartAsync(token);
+        await Task.Run(async () =>
+        {
+            do
+            {
+                await Task.Delay(TimeSpan.FromMilliseconds(50));
+            } while (!token.IsCancellationRequested);
+        });
+        await auditThroughputCollectorHostedService.StopAsync(token);
+
+        var endpointWithoutCounts = await DataStore.GetEndpoint("EndpointNoData", ThroughputSource.Audit, default);
+        var endpointWithCounts = await DataStore.GetEndpoint("EndpointWithData", ThroughputSource.Audit, default);
+
+        // Assert
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(endpointWithoutCounts, Is.Null, "Endpoint with empty auditCounts should not be created");
+            Assert.That(endpointWithCounts, Is.Not.Null, "Endpoint with auditCounts should be created");
+        }
+    }
+
     class AuditQuery_NoAuditRemotes : IAuditQuery
     {
         public SemanticVersion MinAuditCountsVersion => new(4, 29, 0);
@@ -335,4 +377,50 @@ public string SanitizeEndpointName(string endpointName)
 
         public string SanitizedEndpointNameCleanser(string endpointName) => endpointName;
     }
+
+    class AuditQuery_WithTwoEndpointsAndSelectiveCounts : IAuditQuery
+    {
+        public AuditQuery_WithTwoEndpointsAndSelectiveCounts(
+            string endpointWithoutCounts,
+            string endpointWithCounts,
+            DateOnly throughputDate,
+            long throughputCount)
+        {
+            this.endpointWithoutCounts = endpointWithoutCounts;
+            this.endpointWithCounts = endpointWithCounts;
+            this.throughputDate = throughputDate;
+            this.throughputCount = throughputCount;
+        }
+
+        public SemanticVersion MinAuditCountsVersion => new(4, 29, 0);
+        public Func<RemoteInstanceInformation, bool> ValidRemoteInstances => _ => true;
+
+        public Task<IEnumerable<ServiceControlEndpoint>> GetKnownEndpoints(CancellationToken cancellationToken) =>
+            Task.FromResult<IEnumerable<ServiceControlEndpoint>>(
+            [
+                new ServiceControlEndpoint { Name = endpointWithoutCounts, HeartbeatsEnabled = true },
+                new ServiceControlEndpoint { Name = endpointWithCounts, HeartbeatsEnabled = true }
+            ]);
+
+        public Task<IEnumerable<AuditCount>> GetAuditCountForEndpoint(string endpointUrlName, CancellationToken cancellationToken)
+        {
+            if (endpointUrlName == endpointWithCounts)
+            {
+                return Task.FromResult<IEnumerable<AuditCount>>([new AuditCount { UtcDate = throughputDate, Count = throughputCount }]);
+            }
+
+            return Task.FromResult<IEnumerable<AuditCount>>([]);
+        }
+
+        public Task<List<RemoteInstanceInformation>> GetAuditRemotes(CancellationToken cancellationToken) =>
+            Task.FromResult<List<RemoteInstanceInformation>>([]);
+
+        public Task<ConnectionSettingsTestResult> TestAuditConnection(CancellationToken cancellationToken) =>
+            Task.FromResult(new ConnectionSettingsTestResult { ConnectionSuccessful = true, ConnectionErrorMessages = [] });
+
+        readonly string endpointWithoutCounts;
+        readonly string endpointWithCounts;
+        readonly DateOnly throughputDate;
+        readonly long throughputCount;
+    }
 }

src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Throughput_Tests.cs

@@ -194,7 +194,7 @@ await DataStore.CreateBuilder()
     public async Task Should_return_correct_throughput_in_report_when_endpoint_has_no_throughput()
     {
         // Arrange
-        await DataStore.CreateBuilder().AddEndpoint().Build();
+        await DataStore.CreateBuilder().AddEndpoint().WithThroughput(ThroughputSource.Broker, data: [0]).Build();
 
         // Act
         var report = await ThroughputCollector.GenerateThroughputReport("", null, default);
@@ -211,6 +211,20 @@ public async Task Should_return_correct_throughput_in_report_when_endpoint_has_n
         }
     }
 
+    [Test]
+    public async Task Should_not_return_endpoint_in_report_when_endpoint_has_no_throughput()
+    {
+        // Arrange
+        await DataStore.CreateBuilder().AddEndpoint().Build();
+
+        // Act
+        var report = await ThroughputCollector.GenerateThroughputReport("", null, default);
+
+        // Assert
+        Assert.That(report, Is.Not.Null);
+        Assert.That(report.ReportData.Queues.Count, Is.Zero, "Invalid number of endpoints in throughput report");
+    }
+
     [Test]
     public async Task Should_return_correct_throughput_in_report_when_data_from_multiple_sources_and_name_is_different()
     {

src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_ThroughputSummary_Tests.cs

@@ -177,10 +177,11 @@ await DataStore.CreateBuilder()
     }
 
     [Test]
-    public async Task Should_return_correct_max_daily_throughput_in_summary_when_endpoint_has_no_throughput()
+    public async Task Should_return_correct_max_daily_throughput_in_summary_when_endpoint_has_zero_throughput()
     {
         // Arrange
-        await DataStore.CreateBuilder().AddEndpoint().Build();
+        await DataStore.CreateBuilder().AddEndpoint().WithThroughput(new ThroughputData([
+                new EndpointDailyThroughput(new DateOnly(2025, 1, 10), 0)])).Build();
 
         // Act
         var summary = await ThroughputCollector.GetThroughputSummary(default);
@@ -191,6 +192,20 @@ public async Task Should_return_correct_max_daily_throughput_in_summary_when_end
         Assert.That(summary[0].MaxDailyThroughput, Is.EqualTo(0), $"Incorrect MaxDailyThroughput recorded for {summary[0].Name}");
     }
 
+    [Test]
+    public async Task Should_not_return_endpoint_in_summary_when_endpoint_has_no_throughput()
+    {
+        // Arrange
+        await DataStore.CreateBuilder().AddEndpoint().Build();
+
+        // Act
+        var summary = await ThroughputCollector.GetThroughputSummary(default);
+
+        // Assert
+        Assert.That(summary, Is.Not.Null);
+        Assert.That(summary, Is.Empty, "Invalid number of endpoints in throughput summary");
+    }
+
     [Test]
     public async Task Should_return_correct_max_daily_throughput_in_summary_when_data_from_multiple_sources_and_name_is_different()
     {

src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs

@@ -74,11 +74,15 @@ async Task GatherThroughput(CancellationToken cancellationToken)
 
             var auditCounts = (await auditQuery.GetAuditCountForEndpoint(knownEndpointsLookup[endpointId].UrlName, cancellationToken)).ToList();
 
-            if (endpoint == null)
+            if (endpoint == null && auditCounts.Count > 0)
             {
                 endpoint = ConvertToEndpoint(knownEndpointsLookup[endpointId]);
                 await dataStore.SaveEndpoint(endpoint, cancellationToken);
             }
+            else if (endpoint is null)
+            {
+                continue;
+            }
 
             var missingAuditThroughput = auditCounts
                 .Where(auditCount => auditCount.UtcDate > endpoint.LastCollectedDate &&

src/Particular.LicensingComponent/ThroughputCollector.cs

@@ -224,7 +224,14 @@ async IAsyncEnumerable<EndpointData> GetDistinctEndpointData([EnumeratorCancella
 
             var userIndicator = UserIndicator(endpointGroupPerQueue) ?? null;
 
-            yield return new EndpointData(endpointName, throughputData, userIndicator, EndpointScope(endpointGroupPerQueue), EndpointIndicators(endpointGroupPerQueue), IsKnownEndpoint(endpointGroupPerQueue));
+            if (throughputData.Any(x => x.Any()))
+            {
+                yield return new EndpointData(endpointName, throughputData, userIndicator, EndpointScope(endpointGroupPerQueue), EndpointIndicators(endpointGroupPerQueue), IsKnownEndpoint(endpointGroupPerQueue));
+            }
+            else
+            {
+                await dataStore.RemoveEndpoints([.. endpointGroupPerQueue.Select(endpoint => endpoint.Id)], cancellationToken);
+            }
         }
     }
 

src/Particular.LicensingComponent/ThroughputDataExtensions.cs

@@ -51,5 +51,5 @@ public static long AverageMonthlyThroughput(this List<ThroughputData> throughput
     }
 
     public static bool HasDataFromSource(this IDictionary<string, IEnumerable<ThroughputData>> throughputPerQueue, ThroughputSource source) =>
-        throughputPerQueue.Any(queueName => queueName.Value.Any(data => data.ThroughputSource == source && data.Count > 0));
+        throughputPerQueue.Any(queueThroughput => queueThroughput.Value.Any(data => data.ThroughputSource == source && data.Count > 0));
 }

src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndex.cs

@@ -21,13 +21,14 @@ from message in messages
                     TimeSent = (DateTime)message.MessageMetadata["TimeSent"],
                     ProcessedAt = message.ProcessedAt,
                     ReceivingEndpointName = ((EndpointDetails)message.MessageMetadata["ReceivingEndpoint"]).Name,
+                    SendingEndpointName = ((EndpointDetails)message.MessageMetadata["SendingEndpoint"]).Name,
                     CriticalTime = (TimeSpan?)message.MessageMetadata["CriticalTime"],
                     ProcessingTime = (TimeSpan?)message.MessageMetadata["ProcessingTime"],
                     DeliveryTime = (TimeSpan?)message.MessageMetadata["DeliveryTime"],
                     Query = message.MessageMetadata.Select(_ => _.Value.ToString()).Union(new[]
                     {
-                                      string.Join(" ", message.Headers.Select(x => x.Value))
-                                  }).ToArray(),
+                        string.Join(" ", message.Headers.Select(x => x.Value))
+                    }).ToArray(),
                     ConversationId = (string)message.MessageMetadata["ConversationId"]
                 };
 
@@ -48,6 +49,7 @@ public class SortAndFilterOptions
             public MessageStatus Status { get; set; }
             public DateTime ProcessedAt { get; set; }
             public string ReceivingEndpointName { get; set; }
+            public string SendingEndpointName { get; set; }
             public TimeSpan? CriticalTime { get; set; }
             public TimeSpan? ProcessingTime { get; set; }
             public TimeSpan? DeliveryTime { get; set; }

src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndexWithFullTextSearch.cs

@@ -21,14 +21,15 @@ from message in messages
                     TimeSent = (DateTime)message.MessageMetadata["TimeSent"],
                     ProcessedAt = message.ProcessedAt,
                     ReceivingEndpointName = ((EndpointDetails)message.MessageMetadata["ReceivingEndpoint"]).Name,
+                    SendingEndpointName = ((EndpointDetails)message.MessageMetadata["SendingEndpoint"]).Name,
                     CriticalTime = (TimeSpan?)message.MessageMetadata["CriticalTime"],
                     ProcessingTime = (TimeSpan?)message.MessageMetadata["ProcessingTime"],
                     DeliveryTime = (TimeSpan?)message.MessageMetadata["DeliveryTime"],
                     Query = message.MessageMetadata.Select(_ => _.Value.ToString()).Union(new[]
                     {
-                                    string.Join(" ", message.Headers.Select(x => x.Value)),
-                                    LoadAttachment(message, "body").GetContentAsString()
-                                }).ToArray(),
+                        string.Join(" ", message.Headers.Select(x => x.Value)),
+                        LoadAttachment(message, "body").GetContentAsString()
+                    }).ToArray(),
                     ConversationId = (string)message.MessageMetadata["ConversationId"]
                 };
 

src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs

@@ -138,6 +138,9 @@ public async Task<QueryResult<IList<AuditCount>>> QueryAuditCounts(string endpoi
                 .OrderBy(m => m.ProcessedAt)
                 .FirstOrDefaultAsync(token: cancellationToken);
 
+            var hasSent = await session.Query<MessagesViewIndex.SortAndFilterOptions>(indexName)
+                .AnyAsync(m => m.SendingEndpointName == endpointName, token: cancellationToken);
+
             if (oldestMsg != null)
             {
                 var endDate = DateTime.UtcNow.Date.AddDays(1);
@@ -166,6 +169,14 @@ public async Task<QueryResult<IList<AuditCount>>> QueryAuditCounts(string endpoi
                     }
                 }
             }
+            else if (hasSent)
+            {
+                results.Add(new AuditCount
+                {
+                    UtcDate = DateTime.UtcNow.Date,
+                    Count = 0
+                });
+            }
 
             return new QueryResult<IList<AuditCount>>(results, QueryStatsInfo.Zero);
         }

src/ServiceControl.Audit.Persistence.Tests/AuditCountingTests.cs

@@ -67,7 +67,36 @@ string ScrubDates(string input)
             }, ScrubDates);
         }
 
-        ProcessedMessage MakeMessage(string processingEndpoint, DateTime processedAt, bool systemMessage)
+        [Test]
+        public async Task Should_return_zero_throughput_entry_when_SendOnly()
+        {
+            // Arrange
+            var today = DateTime.UtcNow.Date;
+            const string sendOnlyEndpoint = "SendOnlyEndpoint";
+
+            var messages = new[]
+            {
+                // Endpoint sent a message, but did not receive any
+                MakeMessage("SomeOtherEndpoint", sendOnlyEndpoint, today, false)
+            };
+
+            await IngestProcessedMessagesAudits(messages);
+
+            // Act
+            var result = (await DataStore.QueryAuditCounts(sendOnlyEndpoint, TestContext.CurrentContext.CancellationToken)).Results;
+
+            // Assert
+            Assert.That(result, Is.Not.Empty, "Expected non-empty result for endpoint that only sent messages");
+            Assert.That(result, Has.Count.EqualTo(1), "Expected single audit count for send-only endpoint");
+            using (Assert.EnterMultipleScope())
+            {
+                Assert.That(result[0].UtcDate, Is.EqualTo(today), "Expected today's date placeholder");
+                Assert.That(result[0].Count, Is.Zero, "Expected zero throughput count for send-only endpoint");
+            }
+        }
+
+        static ProcessedMessage MakeMessage(string processingEndpoint, DateTime processedAt, bool systemMessage) => MakeMessage(processingEndpoint, null, processedAt, systemMessage);
+        static ProcessedMessage MakeMessage(string processingEndpoint, string sendingEndpoint, DateTime processedAt, bool systemMessage)
         {
             var messageId = Guid.NewGuid().ToString();
             var messageType = "MyMessageType";
@@ -85,8 +114,12 @@ ProcessedMessage MakeMessage(string processingEndpoint, DateTime processedAt, bo
                 { "MessageType", messageType },
                 { "IsRetried", false },
                 { "ConversationId", messageId },
-                { "ReceivingEndpoint", new EndpointDetails { Name = processingEndpoint } }
+                { "ReceivingEndpoint", new EndpointDetails { Name = processingEndpoint } },
             };
+            if (!string.IsNullOrEmpty(sendingEndpoint))
+            {
+                metadata.Add("SendingEndpoint", new EndpointDetails { Name = sendingEndpoint });
+            }
 
             var headers = new Dictionary<string, string>
             {

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

@@ -14,6 +14,7 @@ class LicensingDataStore : ILicensingDataStore
     public Task<bool> IsThereThroughputForLastXDays(int days, CancellationToken cancellationToken) => throw new NotImplementedException();
     public Task<bool> IsThereThroughputForLastXDaysForSource(int days, ThroughputSource throughputSource, bool includeToday, CancellationToken cancellationToken) => throw new NotImplementedException();
     public Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, IList<EndpointDailyThroughput> throughput, CancellationToken cancellationToken) => throw new NotImplementedException();
+    public Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken) => throw new NotImplementedException();
     public Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMetadata, CancellationToken cancellationToken) => throw new NotImplementedException();
     public Task SaveBrokerMetadata(BrokerMetadata brokerMetadata, CancellationToken cancellationToken) => throw new NotImplementedException();
     public Task SaveEndpoint(Particular.LicensingComponent.Contracts.Endpoint endpoint, CancellationToken cancellationToken) => throw new NotImplementedException();

src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs

@@ -25,6 +25,8 @@ class LicensingDataStore(
     const string ReportMasksDocumentId = "ReportMasks";
     const string LicencedEndpointDetailsDocumentId = "LicensedEndpointDetails";
 
+    const int ThroughputPeriodMonths = 14;
+
     static readonly AuditServiceMetadata DefaultAuditServiceMetadata = new([], []);
     static readonly BrokerMetadata DefaultBrokerMetadata = new(null, []);
     static readonly ReportConfigurationDocument DefaultReportConfiguration = new();
@@ -121,14 +123,28 @@ public async Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellation
         await session.SaveChangesAsync(cancellationToken);
     }
 
+    public async Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken)
+    {
+        var documentIds = endpointIds.Select(id => id.GenerateDocumentId());
+
+        var store = await storeProvider.GetDocumentStore(cancellationToken);
+        using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name);
+
+        foreach (var documentId in documentIds)
+        {
+            session.Delete(documentId);
+        }
+        await session.SaveChangesAsync(cancellationToken);
+    }
+
     public async Task<IDictionary<string, IEnumerable<ThroughputData>>> GetEndpointThroughputByQueueName(IList<string> queueNames, CancellationToken cancellationToken)
     {
         var results = queueNames.ToDictionary(queueName => queueName, _ => new List<ThroughputData>() as IEnumerable<ThroughputData>);
 
         var store = await storeProvider.GetDocumentStore(cancellationToken);
         using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name);
 
-        var from = DateTime.UtcNow.AddMonths(-14);
+        var from = DateTime.UtcNow.AddMonths(-ThroughputPeriodMonths);
         var query = session.Query<EndpointDocument>()
             .Where(document => document.SanitizedName.In(queueNames))
             .Include(builder => builder.IncludeTimeSeries(ThroughputTimeSeriesName, from));