Refactor user insert logic into UserInsertProcessor class

Move two-phase user insert (bulk + enrich) from UserMetadataUpdater
to new UserInsertProcessor class for better separation of concerns.
Update UserMetadataUpdater to delegate to the new class. Add
UserInsertProcessor.cs to project. Improves maintainability,
testability, and code organization.

Author: sambetts

src/AnalyticsEngine/WebJob.Office365ActivityImporter.Engine/Graph/User/UserInsertProcessor.cs

@@ -0,0 +1,272 @@
+using Common.Entities;
+using DataUtils;
+using System;
+using System.Collections.Generic;
+using System.Data;
+using System.Data.Entity;
+using System.Data.SqlClient;
+using System.Diagnostics;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace WebJob.Office365ActivityImporter.Engine.Graph
+{
+    /// <summary>
+    /// Handles inserting new users into the database via two-phase approach:
+    /// fast bulk insert (SqlBulkCopy), then metadata enrichment in batches.
+    /// </summary>
+    internal class UserInsertProcessor
+    {
+        private readonly AnalyticsLogger _telemetry;
+        private readonly UserBatchProcessor _batchProcessor;
+        private const int BULK_INSERT_BATCH_SIZE = 10000;
+        private const int METADATA_BATCH_SIZE = 500;
+
+        public UserInsertProcessor(AnalyticsLogger telemetry, UserBatchProcessor batchProcessor)
+        {
+            _telemetry = telemetry ?? throw new ArgumentNullException(nameof(telemetry));
+            _batchProcessor = batchProcessor ?? throw new ArgumentNullException(nameof(batchProcessor));
+        }
+
+        /// <summary>
+        /// Inserts missing users into DB using two-phase approach: fast bulk insert, then metadata enrichment
+        /// </summary>
+        public async Task<List<Common.Entities.User>> InsertMissingUsers(
+            AnalyticsEntitiesContext db,
+            List<GraphUser> allGraphUsers,
+            List<Common.Entities.User> graphMentionedDbUsers,
+            bool readUserSkus,
+            UserMetadataCache userMetaCache,
+            UserDataMapper dataMapper,
+            UserLicenseProcessor licenseProcessor,
+            Func<AnalyticsEntitiesContext, GraphUser, List<GraphUser>, List<Common.Entities.User>, Common.Entities.User, bool, Dictionary<string, Common.Entities.User>, Task> updateAction)
+        {
+            _telemetry.LogInformation($"User import - Inserting missing users (two-phase: bulk insert + metadata enrichment)...");
+
+            // Create HashSet for O(1) lookup of existing DB users
+            var existingUpns = new HashSet<string>(
+                graphMentionedDbUsers.Select(u => u.UserPrincipalName?.ToLower()).Where(upn => !string.IsNullOrEmpty(upn)),
+                StringComparer.OrdinalIgnoreCase);
+
+            // Build list of users to insert - optimized with HashSet lookup
+            var usersToInsert = new List<GraphUser>();
+            foreach (var graphUser in allGraphUsers)
+            {
+                var upn = graphUser.UserPrincipalName?.ToLower();
+                if (!string.IsNullOrEmpty(upn) && !existingUpns.Contains(upn))
+                {
+                    usersToInsert.Add(graphUser);
+                    existingUpns.Add(upn); // Prevent duplicate UPNs from Graph
+                }
+            }
+
+            _telemetry.LogInformation($"User import - Found {usersToInsert.Count.ToString("N0")} new users to insert");
+
+            if (usersToInsert.Count == 0)
+            {
+                return new List<Common.Entities.User>();
+            }
+
+            // PHASE 1: Fast bulk insert with minimal data
+            _telemetry.LogInformation($"User import - Phase 1: Starting bulk insert of {usersToInsert.Count.ToString("N0")} users...");
+            await BulkInsertUsers(db, usersToInsert, BULK_INSERT_BATCH_SIZE);
+            _telemetry.LogInformation($"User import - Phase 1: Bulk insert completed");
+
+            // PHASE 2: Load inserted users and enrich with metadata
+            _telemetry.LogInformation($"User import - Phase 2: Starting metadata enrichment...");
+            var insertedUserUpns = usersToInsert.Select(u => u.UserPrincipalName.ToLower()).ToList();
+            var insertedDbUsers = await EnrichInsertedUsersWithMetadata(
+                db,
+                allGraphUsers,
+                graphMentionedDbUsers,
+                insertedUserUpns,
+                readUserSkus,
+                METADATA_BATCH_SIZE,
+                userMetaCache,
+                updateAction);
+
+            _telemetry.LogInformation($"User import - Phase 2: Metadata enrichment completed for {insertedDbUsers.Count.ToString("N0")} users");
+
+            // Cleanup
+            existingUpns.Clear();
+            usersToInsert.Clear();
+            insertedUserUpns.Clear();
+
+            _telemetry.LogInformation($"User import - Completed inserting and enriching {insertedDbUsers.Count.ToString("N0")} new users");
+
+            return insertedDbUsers;
+        }
+
+        /// <summary>
+        /// Phase 1: Uses SqlBulkCopy for fast bulk insert of minimal user data
+        /// </summary>
+        private async Task BulkInsertUsers(AnalyticsEntitiesContext db, List<GraphUser> graphUsers, int batchSize)
+        {
+            var connectionString = db.Database.Connection.ConnectionString;
+            var totalInserted = 0;
+
+            // Process in batches to manage memory
+            for (int batchStart = 0; batchStart < graphUsers.Count; batchStart += batchSize)
+            {
+                var batch = graphUsers.Skip(batchStart).Take(batchSize).ToList();
+                var dataTable = CreateUserDataTable(batch);
+
+                using (var bulkCopy = new SqlBulkCopy(connectionString))
+                {
+                    bulkCopy.DestinationTableName = "dbo.users";
+                    bulkCopy.BatchSize = batchSize;
+                    bulkCopy.BulkCopyTimeout = 600; // 10 minutes
+
+                    // Map only columns that exist in both GraphUser and the User table
+                    bulkCopy.ColumnMappings.Add("UserPrincipalName", "user_name");
+                    bulkCopy.ColumnMappings.Add("AzureAdId", "azure_ad_id");
+                    bulkCopy.ColumnMappings.Add("AccountEnabled", "account_enabled");
+                    bulkCopy.ColumnMappings.Add("Mail", "mail");
+                    bulkCopy.ColumnMappings.Add("PostalCode", "postalcode");
+
+                    await bulkCopy.WriteToServerAsync(dataTable);
+                }
+
+                totalInserted += batch.Count;
+                _telemetry.LogInformation($"User import - Bulk inserted {totalInserted.ToString("N0")}/{graphUsers.Count.ToString("N0")} users to SQL");
+
+                dataTable.Clear();
+                dataTable.Dispose();
+            }
+        }
+
+        /// <summary>
+        /// Creates a DataTable from GraphUser list with minimal essential columns for bulk insert
+        /// </summary>
+        private DataTable CreateUserDataTable(List<GraphUser> graphUsers)
+        {
+            var dataTable = new DataTable();
+
+            // Add only columns that exist in both GraphUser and the User database table
+            dataTable.Columns.Add("UserPrincipalName", typeof(string));
+            dataTable.Columns.Add("AzureAdId", typeof(string));
+            dataTable.Columns.Add("AccountEnabled", typeof(bool));
+            dataTable.Columns.Add("Mail", typeof(string));
+            dataTable.Columns.Add("PostalCode", typeof(string));
+
+            // Populate rows
+            foreach (var graphUser in graphUsers)
+            {
+                var row = dataTable.NewRow();
+                row["UserPrincipalName"] = graphUser.UserPrincipalName ?? (object)DBNull.Value;
+                row["AzureAdId"] = graphUser.Id ?? (object)DBNull.Value;
+                row["AccountEnabled"] = graphUser.AccountEnabled ?? false;
+                row["Mail"] = graphUser.Mail ?? (object)DBNull.Value;
+                row["PostalCode"] = graphUser.PostalCode ?? (object)DBNull.Value;
+
+                dataTable.Rows.Add(row);
+            }
+
+            return dataTable;
+        }
+
+        /// <summary>
+        /// Phase 2: Loads newly inserted users and enriches them with metadata (managers, licenses, etc.)
+        /// </summary>
+        private async Task<List<Common.Entities.User>> EnrichInsertedUsersWithMetadata(
+            AnalyticsEntitiesContext db,
+            List<GraphUser> allGraphUsers,
+            List<Common.Entities.User> graphMentionedDbUsers,
+            List<string> insertedUserUpns,
+            bool readUserSkus,
+            int batchSize,
+            UserMetadataCache userMetaCache,
+            Func<AnalyticsEntitiesContext, GraphUser, List<GraphUser>, List<Common.Entities.User>, Common.Entities.User, bool, Dictionary<string, Common.Entities.User>, Task> updateAction)
+        {
+            var enrichedUsers = new List<Common.Entities.User>(insertedUserUpns.Count);
+
+            // Create dictionary for fast Graph user lookup - pre-allocate capacity
+            var graphUsersByUpn = new Dictionary<string, GraphUser>(allGraphUsers.Count, StringComparer.OrdinalIgnoreCase);
+            foreach (var graphUser in allGraphUsers)
+            {
+                if (!string.IsNullOrEmpty(graphUser.UserPrincipalName))
+                {
+                    graphUsersByUpn[graphUser.UserPrincipalName] = graphUser;
+                }
+            }
+
+            // Build DB user dictionary for manager resolution from existing users
+            // (newly inserted users added incrementally as each batch is loaded with tracking;
+            // cross-batch managers resolved via DB fallback in UpdateUserManager)
+            var dbUsersByAadId = new Dictionary<string, Common.Entities.User>(
+                graphMentionedDbUsers.Count, StringComparer.OrdinalIgnoreCase);
+
+            foreach (var user in graphMentionedDbUsers)
+            {
+                if (!string.IsNullOrEmpty(user.AzureAdId) && !dbUsersByAadId.ContainsKey(user.AzureAdId))
+                {
+                    dbUsersByAadId[user.AzureAdId] = user;
+                }
+            }
+
+            // Process in batches - use GetRange for O(1) extraction
+            var enrichSw = Stopwatch.StartNew();
+            for (int batchStart = 0; batchStart < insertedUserUpns.Count; batchStart += batchSize)
+            {
+                var batchCount = Math.Min(batchSize, insertedUserUpns.Count - batchStart);
+                var batchUpns = insertedUserUpns.GetRange(batchStart, batchCount);
+
+                // Load batch of newly inserted users from database WITH TRACKING for updates
+                var batchUsers = await db.users
+                    .Where(u => batchUpns.Contains(u.UserPrincipalName.ToLower()))
+                    .Include(u => u.LicenseLookups)
+                    .ToListAsync();
+
+                // Update dbUsersByAadId with TRACKED entities from this batch
+                foreach (var trackedUser in batchUsers)
+                {
+                    if (!string.IsNullOrEmpty(trackedUser.AzureAdId))
+                    {
+                        dbUsersByAadId[trackedUser.AzureAdId] = trackedUser;
+                    }
+                }
+
+                // Pre-populate cache with tracked entities from this batch to prevent duplicate inserts
+                foreach (var trackedUser in batchUsers)
+                {
+                    var upnLower = trackedUser.UserPrincipalName?.ToLower();
+                    if (!string.IsNullOrEmpty(upnLower))
+                    {
+                        await userMetaCache.UserCache.GetOrCreateNewResource(upnLower, trackedUser);
+                    }
+                }
+
+                // Update metadata for each user
+                foreach (var dbUser in batchUsers)
+                {
+                    var upnLower = dbUser.UserPrincipalName?.ToLower();
+                    if (!string.IsNullOrEmpty(upnLower) && graphUsersByUpn.TryGetValue(upnLower, out var graphUser))
+                    {
+                        await updateAction(db, graphUser, allGraphUsers, new List<Common.Entities.User>(), dbUser, readUserSkus, dbUsersByAadId);
+                    }
+                }
+
+                // Save batch
+                db.ChangeTracker.DetectChanges();
+                await db.SaveChangesAsync();
+
+                enrichedUsers.AddRange(batchUsers);
+                var percentDone = (double)enrichedUsers.Count / insertedUserUpns.Count * 100;
+                var elapsedMs = enrichSw.ElapsedMilliseconds;
+                var estimatedTotalMs = elapsedMs / percentDone * 100;
+                var remainingMs = estimatedTotalMs - elapsedMs;
+                var remaining = TimeSpan.FromMilliseconds(remainingMs);
+                _telemetry.LogInformation($"User import - Enriched metadata for {enrichedUsers.Count.ToString("N0")}/{insertedUserUpns.Count.ToString("N0")} users ({percentDone:F1}% done, estimated {remaining.Hours}h {remaining.Minutes}m {remaining.Seconds}s remaining)");
+
+                // Clear change tracker to free memory after each batch
+                _batchProcessor.DetachAllEntitiesExceptLookups(db);
+            }
+
+            // Cleanup
+            graphUsersByUpn.Clear();
+            dbUsersByAadId.Clear();
+
+            return enrichedUsers;
+        }
+    }
+}