Fix $bulk-delete bugs with custom search parameter

src/Microsoft.Health.Fhir.Core.UnitTests/Features/Operations/BulkDelete/BulkDeleteProcessingJobTests.cs

@@ -9,14 +9,15 @@
 using System.Threading;
 using System.Threading.Tasks;
 using MediatR;
+using Microsoft.Extensions.Logging;
 using Microsoft.Health.Core.Features.Context;
-using Microsoft.Health.Extensions.DependencyInjection;
 using Microsoft.Health.Fhir.Core.Features.Context;
 using Microsoft.Health.Fhir.Core.Features.Operations;
 using Microsoft.Health.Fhir.Core.Features.Operations.BulkDelete;
 using Microsoft.Health.Fhir.Core.Features.Persistence;
 using Microsoft.Health.Fhir.Core.Features.Search;
 using Microsoft.Health.Fhir.Core.Messages.Delete;
+using Microsoft.Health.Fhir.Core.Models;
 using Microsoft.Health.Fhir.Core.UnitTests.Extensions;
 using Microsoft.Health.Fhir.Tests.Common;
 using Microsoft.Health.JobManagement;
@@ -43,7 +44,14 @@ public BulkDeleteProcessingJobTests()
                 .Returns(Task.FromResult(new SearchResult(5, new List<Tuple<string, string>>())));
             _queueClient = Substitute.For<IQueueClient>();
             _deleter = Substitute.For<IDeletionService>();
-            _processingJob = new BulkDeleteProcessingJob(_deleter.CreateMockScopeFactory(), Substitute.For<RequestContextAccessor<IFhirRequestContext>>(), Substitute.For<IMediator>(), _searchService.CreateMockScopeFactory(), _queueClient);
+            _processingJob = new BulkDeleteProcessingJob(
+                _deleter.CreateMockScopeFactory(),
+                Substitute.For<RequestContextAccessor<IFhirRequestContext>>(),
+                Substitute.For<IMediator>(),
+                _searchService.CreateMockScopeFactory(),
+                _queueClient,
+                Substitute.For<IModelInfoProvider>(),
+                Substitute.For<ILogger<BulkDeleteProcessingJob>>());
         }
 
         [Fact]
@@ -58,11 +66,8 @@ public async Task GivenProcessingJob_WhenJobIsRun_ThenResourcesAreDeleted()
                 Definition = JsonConvert.SerializeObject(definition),
             };
 
-            var substituteResults = new Dictionary<string, long>();
-            substituteResults.Add("Patient", 3);
-
             _deleter.DeleteMultipleAsync(Arg.Any<ConditionalDeleteResourceRequest>(), Arg.Any<CancellationToken>())
-                .Returns(args => substituteResults);
+                .Returns(args => CreateResources(KnownResourceTypes.Patient, 3));
 
             var result = JsonConvert.DeserializeObject<BulkDeleteResult>(await _processingJob.ExecuteAsync(jobInfo, CancellationToken.None));
             Assert.Single(result.ResourcesDeleted);
@@ -83,11 +88,8 @@ public async Task GivenProcessingJob_WhenJobIsRunWithMultipleResourceTypes_ThenF
                 Definition = JsonConvert.SerializeObject(definition),
             };
 
-            var substituteResults = new Dictionary<string, long>();
-            substituteResults.Add("Patient", 3);
-
             _deleter.DeleteMultipleAsync(Arg.Any<ConditionalDeleteResourceRequest>(), Arg.Any<CancellationToken>())
-                .Returns(args => substituteResults);
+                .Returns(args => CreateResources(KnownResourceTypes.Patient, 3));
 
             var result = JsonConvert.DeserializeObject<BulkDeleteResult>(await _processingJob.ExecuteAsync(jobInfo, CancellationToken.None));
             Assert.Single(result.ResourcesDeleted);
@@ -104,5 +106,27 @@ public async Task GivenProcessingJob_WhenJobIsRunWithMultipleResourceTypes_ThenF
             var actualDefinition = JsonConvert.DeserializeObject<BulkDeleteDefinition>(definitions[0]);
             Assert.Equal(2, actualDefinition.Type.SplitByOrSeparator().Count());
         }
+
+        private List<ResourceWrapper> CreateResources(string resourceType, int count)
+        {
+            var resources = new List<ResourceWrapper>();
+            while (count-- > 0)
+            {
+                resources.Add(
+                    new ResourceWrapper(
+                        Guid.NewGuid().ToString(),
+                        Guid.NewGuid().ToString(),
+                        resourceType,
+                        null,
+                        null,
+                        DateTimeOffset.UtcNow,
+                        false,
+                        null,
+                        null,
+                        null));
+            }
+
+            return resources;
+        }
     }
 }

src/Microsoft.Health.Fhir.Core/Features/Definition/SearchParameterDefinitionManager.cs

@@ -269,13 +269,15 @@
 
             // find all derived resources from the list of base resources
             var allResourceTypes = GetDerivedResourceTypes(searchParameterInfo.BaseResourceTypes);
             foreach (var resourceType in allResourceTypes)
             {
-                TypeLookup[resourceType].TryRemove(searchParameterInfo.Code, out var removedParam);
-                if (removedParam.Url != searchParameterInfo.Url)
+                if (TypeLookup[resourceType].TryRemove(searchParameterInfo.Code, out var removedParam))
                 {
-                    _logger.LogError("Error, Search Param {RemovedParam} removed from Search Param Definition manager.  It does not match deleted Search Param {Url}", removedParam.Url, url);
+                    if (removedParam?.Url != searchParameterInfo.Url)
+                    {
+                        _logger.LogError("Error, Search Param {RemovedParam} removed from Search Param Definition manager.  It does not match deleted Search Param {Url}", removedParam?.Url, url);
+                    }
                 }
             }
 
             if (calculateHash)

src/Microsoft.Health.Fhir.Core/Features/Operations/BulkDelete/BulkDeleteProcessingJob.cs

@@ -11,15 +11,18 @@
 using System.Threading.Tasks;
 using EnsureThat;
 using MediatR;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Primitives;
 using Microsoft.Health.Core.Features.Context;
 using Microsoft.Health.Extensions.DependencyInjection;
 using Microsoft.Health.Fhir.Core.Exceptions;
+using Microsoft.Health.Fhir.Core.Extensions;
 using Microsoft.Health.Fhir.Core.Features.Context;
 using Microsoft.Health.Fhir.Core.Features.Operations.BulkDelete.Messages;
 using Microsoft.Health.Fhir.Core.Features.Persistence;
 using Microsoft.Health.Fhir.Core.Features.Search;
 using Microsoft.Health.Fhir.Core.Messages.Delete;
+using Microsoft.Health.Fhir.Core.Models;
 using Microsoft.Health.JobManagement;
 using Newtonsoft.Json;
 
@@ -33,19 +36,25 @@
         private readonly IMediator _mediator;
         private readonly Func<IScoped<ISearchService>> _searchService;
         private readonly IQueueClient _queueClient;
+        private readonly IModelInfoProvider _modelInfoProvider;
+        private readonly ILogger<BulkDeleteProcessingJob> _logger;
 
         public BulkDeleteProcessingJob(
             Func<IScoped<IDeletionService>> deleterFactory,
             RequestContextAccessor<IFhirRequestContext> contextAccessor,
             IMediator mediator,
             Func<IScoped<ISearchService>> searchService,
-            IQueueClient queueClient)
+            IQueueClient queueClient,
+            IModelInfoProvider modelInfoProvider,
+            ILogger<BulkDeleteProcessingJob> logger)
         {
             _deleterFactory = EnsureArg.IsNotNull(deleterFactory, nameof(deleterFactory));
             _contextAccessor = EnsureArg.IsNotNull(contextAccessor, nameof(contextAccessor));
             _mediator = EnsureArg.IsNotNull(mediator, nameof(mediator));
             _searchService = EnsureArg.IsNotNull(searchService, nameof(searchService));
             _queueClient = EnsureArg.IsNotNull(queueClient, nameof(queueClient));
+            _modelInfoProvider = EnsureArg.IsNotNull(modelInfoProvider, nameof(modelInfoProvider));
+            _logger = EnsureArg.IsNotNull(logger, nameof(logger));
         }
 
         public async Task<string> ExecuteAsync(JobInfo jobInfo, CancellationToken cancellationToken)
@@ -73,14 +82,14 @@
 
                 _contextAccessor.RequestContext = fhirRequestContext;
                 var result = new BulkDeleteResult();
-                IDictionary<string, long> resourcesDeleted = new Dictionary<string, long>();
+                var resourcesDeleted = new List<ResourceWrapper>();
                 using IScoped<IDeletionService> deleter = _deleterFactory.Invoke();
                 Exception exception = null;
                 List<string> types = definition.Type.SplitByOrSeparator().ToList();
 
                 try
                 {
-                    resourcesDeleted = await deleter.Value.DeleteMultipleAsync(
+                    var deleteResult = await deleter.Value.DeleteMultipleAsync(
                         new ConditionalDeleteResourceRequest(
                             types[0],
                             (IReadOnlyList<Tuple<string, string>>)definition.SearchParameters,
@@ -90,23 +99,30 @@
                             versionType: definition.VersionType,
                             allowPartialSuccess: false), // Explicitly setting to call out that this can be changed in the future if we want to. Bulk delete offers the possibility of automatically rerunning the operation until it succeeds, fully automating the process.
                         cancellationToken);
+                    resourcesDeleted.AddRange(deleteResult);
                 }
-                catch (IncompleteOperationException<IDictionary<string, long>> ex)
+                catch (IncompleteOperationException<IList<ResourceWrapper>> ex)
                 {
-                    resourcesDeleted = ex.PartialResults;
+                    resourcesDeleted.AddRange(ex.PartialResults);
                     result.Issues.Add(ex.Message);
                     exception = ex;
                 }
 
-                foreach (var (key, value) in resourcesDeleted)
+                var deletedResourceCountMap = resourcesDeleted.GroupBy(x => x.ResourceTypeName).ToDictionary(x => x.Key, x => (long)x.Count());
+                foreach (var item in deletedResourceCountMap)
                 {
-                    if (!result.ResourcesDeleted.TryAdd(key, value))
+                    if (!result.ResourcesDeleted.TryAdd(item.Key, item.Value))
                     {
-                        result.ResourcesDeleted[key] += value;
+                        result.ResourcesDeleted[item.Key] += item.Value;
                     }
                 }
 
-                await _mediator.Publish(new BulkDeleteMetricsNotification(jobInfo.Id, resourcesDeleted.Sum(resource => resource.Value)), cancellationToken);
+                var notification = new BulkDeleteMetricsNotification(jobInfo.Id, resourcesDeleted.Count)
+                {
+                    Content = CreateNotificationContent(resourcesDeleted),
+                };
+
+                await _mediator.Publish(notification, cancellationToken);
 
                 if (exception != null)
                 {
@@ -132,5 +148,27 @@
                 _contextAccessor.RequestContext = existingFhirRequestContext;
             }
         }
+
+        private string CreateNotificationContent(List<ResourceWrapper> resources)
+        {
+            try
+            {
+                var searchParameterUrls = resources
+                    .Where(x => x.ResourceTypeName == KnownResourceTypes.SearchParameter && x.RawResource != null)
+                    .Select(x => _modelInfoProvider.ToTypedElement(x.RawResource).GetStringScalar("url"))
+                    .ToList();
+                if (searchParameterUrls.Any())
+                {
+                    return JsonConvert.SerializeObject(searchParameterUrls);
+                }
+
+                return null;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, $"Failed to create the notification content for {resources.Count} search parameters.");
+                throw;
+            }
+        }
     }
 }

src/Microsoft.Health.Fhir.Core/Features/Operations/BulkDelete/Messages/BulkDeleteMetricsNotification.cs

@@ -29,5 +29,11 @@ public BulkDeleteMetricsNotification(long id, long resourcesDeleted)
         public long JobId { get; }
 
         public OperationStatus Status { get; }
+
+        /// <summary>
+        /// Gets or sets a content associated with this notification. (e.g. A list of deleted search parameter urls.)
+        /// TODO: this should probably be Stream but MediatR doesn't look like cloning a notification before it sends to handlers.
+        /// </summary>
+        public object Content { get; set; }
     }
 }

src/Microsoft.Health.Fhir.Core/Features/Persistence/IDeletionService.cs

@@ -14,6 +14,6 @@ public interface IDeletionService
     {
         public Task<ResourceKey> DeleteAsync(DeleteResourceRequest request, CancellationToken cancellationToken);
 
-        public Task<IDictionary<string, long>> DeleteMultipleAsync(ConditionalDeleteResourceRequest request, CancellationToken cancellationToken);
+        public Task<List<ResourceWrapper>> DeleteMultipleAsync(ConditionalDeleteResourceRequest request, CancellationToken cancellationToken);
     }
 }

src/Microsoft.Health.Fhir.Core/Features/Search/Parameters/SearchParameterOperations.cs

@@ -6,24 +6,29 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Text.Json;
 using System.Threading;
 using System.Threading.Tasks;
 using EnsureThat;
 using Hl7.Fhir.ElementModel;
+using MediatR;
 using Microsoft.Extensions.Logging;
 using Microsoft.Health.Extensions.DependencyInjection;
 using Microsoft.Health.Fhir.Core.Exceptions;
 using Microsoft.Health.Fhir.Core.Extensions;
 using Microsoft.Health.Fhir.Core.Features.Definition;
 using Microsoft.Health.Fhir.Core.Features.Definition.BundleWrappers;
+using Microsoft.Health.Fhir.Core.Features.Operations.BulkDelete.Messages;
 using Microsoft.Health.Fhir.Core.Features.Persistence;
 using Microsoft.Health.Fhir.Core.Features.Search.Registry;
 using Microsoft.Health.Fhir.Core.Models;
 
 namespace Microsoft.Health.Fhir.Core.Features.Search.Parameters
 {
-    public class SearchParameterOperations : ISearchParameterOperations
+    public class SearchParameterOperations : ISearchParameterOperations, INotificationHandler<BulkDeleteMetricsNotification>
     {
+        private const int DefaultDeleteTasksPerPage = 5;
+
         private readonly ISearchParameterDefinitionManager _searchParameterDefinitionManager;
         private readonly SearchParameterStatusManager _searchParameterStatusManager;
         private readonly IModelInfoProvider _modelInfoProvider;
@@ -278,6 +283,27 @@ await _searchParameterStatusManager.ApplySearchParameterStatus(
                 cancellationToken);
         }
 
+        public async Task Handle(BulkDeleteMetricsNotification notification, CancellationToken cancellationToken)
+        {
+            var content = notification?.Content?.ToString();
+            if (!string.IsNullOrWhiteSpace(content))
+            {
+                try
+                {
+                    var urls = JsonSerializer.Deserialize<List<string>>(content);
+                    if (urls.Any())
+                    {
+                        await DeleteSearchParametersAsync(urls, DefaultDeleteTasksPerPage, cancellationToken);
+                    }
+                }
+                catch (JsonException ex)
+                {
+                    _logger.LogError(ex, "Failed to deserialize the notification content.");
+                    throw;
+                }
+            }
+        }
+
         private void DeleteSearchParameter(string url)
         {
             try
@@ -311,5 +337,69 @@ private async Task<ITypedElement> GetSearchParameterByUrl(string url, Cancellati
 
             return null;
         }
+
+        private async Task DeleteSearchParametersAsync(List<string> urls, int pageSize, CancellationToken cancellationToken)
+        {
+            if (urls?.Any() ?? false)
+            {
+                // We need to make sure we have the latest search parameters before trying to delete
+                // existing search parameter. This is to avoid trying to update a search parameter that
+                // was recently added and that hasn't propogated to all fhir-server instances.
+                await GetAndApplySearchParameterUpdates(cancellationToken);
+
+                var count = 0;
+                while (count < urls.Count)
+                {
+                    var urlsToDelete = urls.Skip(count).Take(pageSize).ToList();
+                    count += urlsToDelete.Count;
+
+                    var tasks = new List<Task>();
+                    foreach (var url in urlsToDelete)
+                    {
+                        if (string.IsNullOrWhiteSpace(url))
+                        {
+                            _logger.LogWarning("Skipping the null or empty url.");
+                            continue;
+                        }
+
+                        tasks.Add(
+                            Task.Run(
+                                async () =>
+                                {
+                                    try
+                                    {
+                                        // First we delete the status metadata from the data store as this function depends on
+                                        // the in memory definition manager.  Once complete we remove the SearchParameter from
+                                        // the definition manager.
+                                        _logger.LogInformation("Deleting the search parameter '{Url}'", url);
+                                        await _searchParameterStatusManager.UpdateSearchParameterStatusAsync(new List<string>() { url }, SearchParameterStatus.PendingDelete, cancellationToken);
+
+                                        // Update the status of the search parameter in the definition manager once the status is updated in the store.
+                                        _searchParameterDefinitionManager.UpdateSearchParameterStatus(url, SearchParameterStatus.PendingDelete);
+                                    }
+                                    catch (Exception ex)
+                                    {
+                                        _logger.LogError(ex, $"Failed to delete a search parameter with url: {url}");
+                                        throw;
+                                    }
+                                },
+                                cancellationToken));
+                    }
+
+                    try
+                    {
+                        if (tasks.Any())
+                        {
+                            await Task.WhenAll(tasks);
+                        }
+                    }
+                    catch (Exception ex)
+                    {
+                        _logger.LogError(ex, $"Failed to delete {tasks.Where(x => !x.IsCompleted).Count()} search parameters.");
+                        throw;
+                    }
+                }
+            }
+        }
     }
 }

src/Microsoft.Health.Fhir.Shared.Core/Features/Resources/Delete/ConditionalDeleteResourceHandler.cs

@@ -146,8 +146,9 @@ private async Task<DeleteResourceResponse> DeleteSingleAsync(ConditionalDeleteRe
 
         private async Task<DeleteResourceResponse> DeleteMultipleAsync(ConditionalDeleteResourceRequest request, CancellationToken cancellationToken)
         {
-            long numDeleted = (await _deleter.DeleteMultipleAsync(request, cancellationToken)).Sum(result => result.Value);
-            return new DeleteResourceResponse((int)numDeleted);
+            var result = await _deleter.DeleteMultipleAsync(request, cancellationToken);
+            var stats = result.GroupBy(x => x.ResourceTypeName).ToDictionary(x => x.Key, x => (long)x.Count());
+            return new DeleteResourceResponse((int)stats.Sum(x => x.Value));
         }
     }
 }