Bulk delete enhancements (#3505)

Add audit logging and improve throughput

docs/rest/BulkDelete.http

@@ -28,7 +28,7 @@ Authorization: Bearer {{bearer.response.body.access_token}}
 ### Record Bulk Delete content location
 @bulkDeleteLocation = {{bulkDelete.response.headers.Content-Location}}
 
-### Bulk Delete with search parameteres
+### Bulk Delete with search parameteres. This also sets data up for the soft deleted sample below.
 # @name bulkDelete
 DELETE https://{{hostname}}/$bulk-delete?_tag=oldData
 Prefer: respond-async
@@ -46,9 +46,16 @@ DELETE https://{{hostname}}/$bulk-delete?_tag=oldData&_purgeHistory=true
 Prefer: respond-async
 Authorization: Bearer {{bearer.response.body.access_token}}
 
-### Bulk Delete Encounters with hard delete
+### Bulk Delete Patient with hard delete
 # @name bulkDelete
-DELETE https://{{hostname}}/Encounter/$bulk-delete?_hardDelete=true
+DELETE https://{{hostname}}/Patient/$bulk-delete?_tag=oldData&_hardDelete=true
+Prefer: respond-async
+Authorization: Bearer {{bearer.response.body.access_token}}
+
+### Bulk Delete soft deleted resources. Run the $bulk-delete sample above to setup data for this test.
+# Since this test can't use _tag it will delete all soft deleted data in the database.
+# @name bulkDelete
+DELETE https://{{hostname}}/$bulk-delete-soft-deleted
 Prefer: respond-async
 Authorization: Bearer {{bearer.response.body.access_token}}
 

src/Microsoft.Health.Fhir.Api/Features/Routing/KnownRoutes.cs

@@ -87,8 +87,11 @@ internal class KnownRoutes
 
         public const string BulkDelete = "$bulk-delete";
         public const string BulkDeleteResourceType = ResourceType + "/" + BulkDelete;
+        public const string BulkDeleteSoftDeleted = "$bulk-delete-soft-deleted";
+        public const string BulkDeleteSoftDeletedResourceType = ResourceType + "/" + BulkDeleteSoftDeleted;
         public const string BulkDeleteJobLocation = OperationsConstants.Operations + "/" + OperationsConstants.BulkDelete + "/" + IdRouteSegment;
         public const string BulkDeleteOperationDefinition = OperationDefinition + "/" + OperationsConstants.BulkDelete;
         public const string ResourceTypeBulkDeleteOperationDefinition = OperationDefinition + "/" + OperationsConstants.ResourceTypeBulkDelete;
+        public const string BulkDeleteSoftDeletedOperationDefinition = OperationDefinition + "/" + OperationsConstants.BulkDeleteSoftDeleted;
     }
 }

src/Microsoft.Health.Fhir.Api/Features/Routing/RouteNames.cs

@@ -78,5 +78,7 @@ internal static class RouteNames
         internal const string CancelBulkDelete = nameof(CancelBulkDelete);
 
         internal const string BulkDeleteDefinition = nameof(BulkDeleteDefinition);
+
+        internal const string BulkDeleteSoftDeletedDefinition = nameof(BulkDeleteSoftDeletedDefinition);
     }
 }

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

@@ -34,6 +34,11 @@ public BulkDeleteOrchestratorJobTests()
         {
             _queueClient = Substitute.For<IQueueClient>();
             _searchService = Substitute.For<ISearchService>();
+            _searchService.SearchAsync(Arg.Any<string>(), Arg.Any<IReadOnlyList<Tuple<string, string>>>(), Arg.Any<CancellationToken>()).Returns((x) =>
+            {
+                var result = new SearchResult(2, new List<Tuple<string, string>>());
+                return Task.FromResult(result);
+            });
             _orchestratorJob = new BulkDeleteOrchestratorJob(_queueClient, _searchService);
 
             _progress = new Progress<string>((result) => { });
@@ -62,9 +67,14 @@ public async Task GivenBulkDeleteJob_WhenNoResourceTypeIsGiven_ThenProcessingJob
             await _queueClient.ReceivedWithAnyArgs(1).EnqueueAsync(Arg.Any<byte>(), Arg.Any<string[]>(), Arg.Any<long?>(), false, false, Arg.Any<CancellationToken>());
             await _searchService.ReceivedWithAnyArgs(1).GetUsedResourceTypes(Arg.Any<CancellationToken>());
 
-            // Checks that two processing jobs were queued
+            // Checks that one processing job was queued
             var calls = _queueClient.ReceivedCalls();
-            Assert.Equal(2, ((string[])calls.First().GetArguments()[1]).Length);
+            var definitions = (string[])calls.First().GetArguments()[1];
+            Assert.Single(definitions);
+
+            // Checks that the processing job lists both resource types
+            var actualDefinition = JsonConvert.DeserializeObject<BulkDeleteDefinition>(definitions[0]);
+            Assert.Equal(2, actualDefinition.Type.SplitByOrSeparator().Count());
         }
 
         [Fact]

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

@@ -5,6 +5,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using MediatR;
@@ -14,6 +15,7 @@
 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.Tests.Common;
 using Microsoft.Health.JobManagement;
@@ -31,13 +33,19 @@ public class BulkDeleteProcessingJobTests
         private IDeletionService _deleter;
         private IProgress<string> _progress;
         private BulkDeleteProcessingJob _processingJob;
+        private ISearchService _searchService;
+        private IQueueClient _queueClient;
 
         public BulkDeleteProcessingJobTests()
         {
+            _searchService = Substitute.For<ISearchService>();
+            _searchService.SearchAsync(Arg.Any<string>(), Arg.Any<IReadOnlyList<Tuple<string, string>>>(), Arg.Any<CancellationToken>(), resourceVersionTypes: Arg.Any<ResourceVersionType>())
+                .Returns(Task.FromResult(new SearchResult(5, new List<Tuple<string, string>>())));
+            _queueClient = Substitute.For<IQueueClient>();
             _deleter = Substitute.For<IDeletionService>();
             var deleter = Substitute.For<IScoped<IDeletionService>>();
             deleter.Value.Returns(_deleter);
-            _processingJob = new BulkDeleteProcessingJob(() => deleter, Substitute.For<RequestContextAccessor<IFhirRequestContext>>(), Substitute.For<IMediator>());
+            _processingJob = new BulkDeleteProcessingJob(() => deleter, Substitute.For<RequestContextAccessor<IFhirRequestContext>>(), Substitute.For<IMediator>(), _searchService, _queueClient);
 
             _progress = new Progress<string>((result) => { });
         }
@@ -55,13 +63,44 @@ public async Task GivenProcessingJob_WhenJobIsRun_ThenResourcesAreDeleted()
             };
 
             _deleter.DeleteMultipleAsync(Arg.Any<ConditionalDeleteResourceRequest>(), Arg.Any<CancellationToken>())
-                .Returns(args => new HashSet<string> { "1", "2", "3"});
+                .Returns(args => 3);
 
             var result = JsonConvert.DeserializeObject<BulkDeleteResult>(await _processingJob.ExecuteAsync(jobInfo, _progress, CancellationToken.None));
             Assert.Single(result.ResourcesDeleted);
             Assert.Equal(3, result.ResourcesDeleted["Patient"]);
 
             await _deleter.ReceivedWithAnyArgs(1).DeleteMultipleAsync(Arg.Any<ConditionalDeleteResourceRequest>(), Arg.Any<CancellationToken>());
         }
+
+        [Fact]
+        public async Task GivenProcessingJob_WhenJobIsRunWithMultipleResourceTypes_ThenFollowupJobIsCreated()
+        {
+            _deleter.ClearReceivedCalls();
+
+            var definition = new BulkDeleteDefinition(JobType.BulkDeleteProcessing, DeleteOperation.HardDelete, "Patient,Observation,Device", new List<Tuple<string, string>>(), "https:\\\\test.com", "https:\\\\test.com", "test");
+            var jobInfo = new JobInfo()
+            {
+                Id = 1,
+                Definition = JsonConvert.SerializeObject(definition),
+            };
+
+            _deleter.DeleteMultipleAsync(Arg.Any<ConditionalDeleteResourceRequest>(), Arg.Any<CancellationToken>())
+                .Returns(args => 3);
+
+            var result = JsonConvert.DeserializeObject<BulkDeleteResult>(await _processingJob.ExecuteAsync(jobInfo, _progress, CancellationToken.None));
+            Assert.Single(result.ResourcesDeleted);
+            Assert.Equal(3, result.ResourcesDeleted["Patient"]);
+
+            await _deleter.ReceivedWithAnyArgs(1).DeleteMultipleAsync(Arg.Any<ConditionalDeleteResourceRequest>(), Arg.Any<CancellationToken>());
+
+            // Checks that one processing job was queued
+            var calls = _queueClient.ReceivedCalls();
+            var definitions = (string[])calls.First().GetArguments()[1];
+            Assert.Single(definitions);
+
+            // Checks that the processing job removed the resource type that was processed and lists the remaining two resource types
+            var actualDefinition = JsonConvert.DeserializeObject<BulkDeleteDefinition>(definitions[0]);
+            Assert.Equal(2, actualDefinition.Type.SplitByOrSeparator().Count());
+        }
     }
 }

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

@@ -86,7 +86,7 @@ public async Task GivenBulkDeleteRequest_WhenJobCreationRequested_ThenJobIsCreat
                     };
             });
 
-            var request = new CreateBulkDeleteRequest(DeleteOperation.HardDelete, KnownResourceTypes.Patient, searchParams);
+            var request = new CreateBulkDeleteRequest(DeleteOperation.HardDelete, KnownResourceTypes.Patient, searchParams, false);
 
             var response = await _handler.Handle(request, CancellationToken.None);
             Assert.NotNull(response);
@@ -105,7 +105,7 @@ public async Task GivenBulkDeleteRequestWithInvalidSearchParameter_WhenJobCreati
             _authorizationService.CheckAccess(Arg.Any<DataActions>(), Arg.Any<CancellationToken>()).Returns(DataActions.HardDelete | DataActions.Delete);
             _contextAccessor.RequestContext.BundleIssues.Add(new OperationOutcomeIssue(OperationOutcomeConstants.IssueSeverity.Warning, OperationOutcomeConstants.IssueType.Conflict));
 
-            var request = new CreateBulkDeleteRequest(DeleteOperation.HardDelete, KnownResourceTypes.Patient, searchParams);
+            var request = new CreateBulkDeleteRequest(DeleteOperation.HardDelete, KnownResourceTypes.Patient, searchParams, false);
 
             await Assert.ThrowsAsync<BadRequestException>(async () => await _handler.Handle(request, CancellationToken.None));
         }
@@ -118,7 +118,7 @@ public async Task GivenUnauthorizedUser_WhenJobCreationRequested_ThenUnauthorize
         {
             _authorizationService.CheckAccess(Arg.Any<DataActions>(), Arg.Any<CancellationToken>()).Returns(userRole);
 
-            var request = new CreateBulkDeleteRequest(deleteOperation, null, null);
+            var request = new CreateBulkDeleteRequest(deleteOperation, null, null, false);
             await Assert.ThrowsAsync<UnauthorizedFhirActionException>(async () => await _handler.Handle(request, CancellationToken.None));
         }
 
@@ -129,7 +129,7 @@ public async Task GivenBulkDeleteRequest_WhenJobCreationFails_ThenExceptionIsThr
             _contextAccessor.RequestContext.BundleIssues.Clear();
             _queueClient.EnqueueAsync((byte)QueueType.BulkDelete, Arg.Any<string[]>(), Arg.Any<long?>(), false, false, Arg.Any<CancellationToken>()).Returns(new List<JobInfo>());
 
-            var request = new CreateBulkDeleteRequest(DeleteOperation.HardDelete, null, new List<Tuple<string, string>>());
+            var request = new CreateBulkDeleteRequest(DeleteOperation.HardDelete, null, new List<Tuple<string, string>>(), false);
             await Assert.ThrowsAsync<JobNotExistException>(async () => await _handler.Handle(request, CancellationToken.None));
         }
     }