Fetch popularity transfers from database (#767)

The `db2azuresearch` and `auxiliary2azuresearch` jobs needs to get the latest popularity transfers from the database.

⚠ This change is not unit testable as it interacts with a database. This will be covered by end-to-end tests and monitoring.

Previous changes: NuGet/NuGet.Services.Metadata#765 and NuGet/NuGet.Services.Metadata#766.
Part of NuGet/NuGetGallery#7898

src/NuGet.Services.AzureSearch/AzureSearchTelemetryService.cs

@@ -189,6 +189,17 @@ public void TrackReadLatestIndexedPopularityTransfers(int outgoingTransfers, Tim
                 });
         }
 
+        public void TrackReadLatestPopularityTransfersFromDatabase(int outgoingTransfers, TimeSpan elapsed)
+        {
+            _telemetryClient.TrackMetric(
+                Prefix + "ReadLatestPopularityTransfersFromDatabase",
+                elapsed.TotalSeconds,
+                new Dictionary<string, string>
+                {
+                    { "OutgoingTransfers", outgoingTransfers.ToString() }
+                });
+        }
+
         public void TrackPopularityTransfersSetComparison(int oldCount, int newCount, int changeCount, TimeSpan elapsed)
         {
             _telemetryClient.TrackMetric(

src/NuGet.Services.AzureSearch/DatabaseAuxiliaryDataFetcher.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Data;
 using System.Data.Entity;
 using System.Diagnostics;
 using System.Linq;
@@ -35,6 +36,20 @@ FROM PackageRegistrations pr (NOLOCK)
 FROM PackageRegistrations pr (NOLOCK)
 INNER JOIN PackageRegistrationOwners pro (NOLOCK) ON pro.PackageRegistrationKey = pr.[Key]
 INNER JOIN Users u (NOLOCK) ON pro.UserKey = u.[Key]
+";
+
+        private const int GetPopularityTransfersPageSize = 1000;
+        private const string GetPopularityTransfersSkipParameter = "@skip";
+        private const string GetPopularityTransfersTakeParameter = "@take";
+        private const string GetPopularityTransfersSql = @"
+SELECT TOP (@take)
+    fpr.Id AS FromPackageId,
+    tpr.Id AS ToPackageId
+FROM PackageRenames r (NOLOCK)
+INNER JOIN PackageRegistrations fpr (NOLOCK) ON fpr.[Key] = r.[FromPackageRegistrationKey]
+INNER JOIN PackageRegistrations tpr (NOLOCK) ON tpr.[Key] = r.[ToPackageRegistrationKey]
+WHERE r.TransferPopularity != 0 AND r.[Key] >= @skip
+ORDER BY r.[Key] ASC
 ";
 
         public DatabaseAuxiliaryDataFetcher(
@@ -137,6 +152,52 @@ public async Task<SortedDictionary<string, SortedSet<string>>> GetPackageIdToOwn
                 }
             }
         }
+
+        public async Task<SortedDictionary<string, SortedSet<string>>> GetPackageIdToPopularityTransfersAsync()
+        {
+            var stopwatch = Stopwatch.StartNew();
+            var builder = new PackageIdToPopularityTransfersBuilder(_logger);
+            using (var connection = await _connectionFactory.OpenAsync())
+            using (var command = connection.CreateCommand())
+            {
+                command.CommandText = GetPopularityTransfersSql;
+                command.Parameters.Add(GetPopularityTransfersSkipParameter, SqlDbType.Int);
+                command.Parameters.AddWithValue(GetPopularityTransfersTakeParameter, GetPopularityTransfersPageSize);
+
+                // Load popularity transfers by paging through the database.
+                // We continue paging until we receive fewer results than the page size.
+                int currentPageResults;
+                int totalResults = 0;
+                do
+                {
+                    command.Parameters[GetPopularityTransfersSkipParameter].Value = totalResults;
+
+                    using (var reader = await command.ExecuteReaderAsync())
+                    {
+                        currentPageResults = 0;
+
+                        while (await reader.ReadAsync())
+                        {
+                            currentPageResults++;
+
+                            var fromId = reader.GetString(0);
+                            var toId = reader.GetString(1);
+
+                            builder.Add(fromId, toId);
+                        }
+                    }
+
+                    totalResults += currentPageResults;
+                }
+                while (currentPageResults == GetPopularityTransfersPageSize);
+
+                var output = builder.GetResult();
+                stopwatch.Stop();
+                _telemetryService.TrackReadLatestPopularityTransfersFromDatabase(output.Count, stopwatch.Elapsed);
+
+                return output;
+            }
+        }
     }
 }
 

src/NuGet.Services.AzureSearch/IAzureSearchTelemetryService.cs

@@ -22,6 +22,7 @@ public interface IAzureSearchTelemetryService
         void TrackReadLatestOwnersFromDatabase(int packageIdCount, TimeSpan elapsed);
         void TrackPopularityTransfersSetComparison(int oldCount, int newCount, int changeCount, TimeSpan elapsed);
         void TrackReadLatestIndexedPopularityTransfers(int outgoingTransfers, TimeSpan elapsed);
+        void TrackReadLatestPopularityTransfersFromDatabase(int outgoingTransfers, TimeSpan elapsed);
         void TrackReadLatestVerifiedPackagesFromDatabase(int packageIdCount, TimeSpan elapsed);
         IDisposable TrackReplaceLatestIndexedOwners(int packageIdCount);
         IDisposable TrackUploadOwnerChangeHistory(int packageIdCount);

src/NuGet.Services.AzureSearch/IDatabaseAuxiliaryDataFetcher.cs

@@ -24,6 +24,13 @@ public interface IDatabaseAuxiliaryDataFetcher
         /// </summary>
         Task<SortedDictionary<string, SortedSet<string>>> GetPackageIdToOwnersAsync();
 
+        /// <summary>
+        /// Fetch a mapping of package IDs to set of replacement package IDs for each renamed packages that transfer
+        /// popularity in the gallery database.
+        /// </summary>
+        /// <returns></returns>
+        Task<SortedDictionary<string, SortedSet<string>>> GetPackageIdToPopularityTransfersAsync();
+
         /// <summary>
         /// Fetch the set of all verified package IDs.
         /// </summary>