Import bundle

src/Microsoft.Health.Fhir.Api/Features/ActionResults/ImportBundleResult.cs

@@ -0,0 +1,25 @@
+// -------------------------------------------------------------------------------------------------
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License (MIT). See LICENSE in the repo root for license information.
+// -------------------------------------------------------------------------------------------------
+
+using System.Net;
+using EnsureThat;
+using Microsoft.Health.Fhir.Core.Features.Operations.Import;
+
+namespace Microsoft.Health.Fhir.Api.Features.ActionResults
+{
+    /// <summary>
+    /// Used to return the result of a import bundle operation.
+    /// </summary>
+    public class ImportBundleResult : ResourceActionResult<ImportJobResult>
+    {
+        public ImportBundleResult(int count, HttpStatusCode statusCode)
+            : base(null, statusCode)
+        {
+            LoadedResources = count;
+        }
+
+        public int LoadedResources { get; }
+    }
+}

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

@@ -66,6 +66,8 @@ internal class KnownRoutes
         public const string ImportDataOperationDefinition = OperationDefinition + "/" + OperationsConstants.Import;
         public const string ImportJobLocation = OperationsConstants.Operations + "/" + OperationsConstants.Import + "/" + IdRouteSegment;
 
+        public const string ImportBundle = "$import/Bundle";
+
         public const string CompartmentTypeByResourceType = CompartmentTypeRouteSegment + "/" + IdRouteSegment + "/" + CompartmentResourceTypeRouteSegment;
 
         public const string WellKnown = ".well-known";

src/Microsoft.Health.Fhir.Core/Extensions/ImportMediatorExtensions.cs

@@ -17,6 +17,17 @@ namespace Microsoft.Health.Fhir.Core.Extensions
 {
     public static class ImportMediatorExtensions
     {
+        public static async Task<ImportBundleResponse> ImportBundleAsync(
+            this IMediator mediator,
+            IList<ImportResource> resources,
+            CancellationToken cancellationToken)
+        {
+            EnsureArg.IsNotNull(mediator, nameof(mediator));
+            var request = new ImportBundleRequest(resources);
+            var response = await mediator.Send(request, cancellationToken);
+            return response;
+        }
+
         public static async Task<CreateImportResponse> ImportAsync(
             this IMediator mediator,
             Uri requestUri,

src/Microsoft.Health.Fhir.Core/Features/Operations/Import/ImportBundleMetricsNotification.cs

@@ -0,0 +1,37 @@
+// -------------------------------------------------------------------------------------------------
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License (MIT). See LICENSE in the repo root for license information.
+// -------------------------------------------------------------------------------------------------
+
+using System;
+using Microsoft.Health.Fhir.Core.Features.Metrics;
+using Microsoft.Health.Fhir.ValueSets;
+
+namespace Microsoft.Health.Fhir.Core.Features.Operations.Import
+{
+    public class ImportBundleMetricsNotification : IMetricsNotification
+    {
+        public ImportBundleMetricsNotification(
+            DateTimeOffset startTime,
+            DateTimeOffset endTime,
+            long succeededCount)
+        {
+            FhirOperation = AuditEventSubType.ImportBundle;
+            ResourceType = null;
+
+            StartTime = startTime;
+            EndTime = endTime;
+            SucceededCount = succeededCount;
+        }
+
+        public string FhirOperation { get; }
+
+        public string ResourceType { get; }
+
+        public DateTimeOffset StartTime { get; }
+
+        public DateTimeOffset EndTime { get; }
+
+        public long SucceededCount { get; }
+    }
+}

src/Microsoft.Health.Fhir.Core/Features/Operations/Import/ImportBundleRequestHandler.cs

@@ -0,0 +1,56 @@
+// -------------------------------------------------------------------------------------------------
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License (MIT). See LICENSE in the repo root for license information.
+// -------------------------------------------------------------------------------------------------
+
+using System;
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using EnsureThat;
+using Hl7.Fhir.Serialization;
+using MediatR;
+using Microsoft.Extensions.Logging;
+using Microsoft.Health.Core.Features.Security.Authorization;
+using Microsoft.Health.Fhir.Core.Exceptions;
+using Microsoft.Health.Fhir.Core.Features.Persistence;
+using Microsoft.Health.Fhir.Core.Features.Security;
+using Microsoft.Health.Fhir.Core.Messages.Import;
+using Microsoft.Health.JobManagement;
+
+namespace Microsoft.Health.Fhir.Core.Features.Operations.Import
+{
+    /// <summary>
+    /// MediatR request handler. Called when the ImportController processes ImportBundleRequest.
+    /// </summary>
+    public class ImportBundleRequestHandler : IRequestHandler<ImportBundleRequest, ImportBundleResponse>
+    {
+        private readonly IFhirDataStore _store;
+        private readonly ILogger<ImportBundleRequestHandler> _logger;
+        private readonly IAuthorizationService<DataActions> _authorizationService;
+
+        public ImportBundleRequestHandler(
+            IFhirDataStore store,
+            ILogger<ImportBundleRequestHandler> logger,
+            IAuthorizationService<DataActions> authorizationService)
+        {
+            _store = EnsureArg.IsNotNull(store, nameof(store));
+            _logger = EnsureArg.IsNotNull(logger, nameof(logger));
+            _authorizationService = EnsureArg.IsNotNull(authorizationService, nameof(authorizationService));
+        }
+
+        public async Task<ImportBundleResponse> Handle(ImportBundleRequest request, CancellationToken cancellationToken)
+        {
+            EnsureArg.IsNotNull(request, nameof(request));
+            if (await _authorizationService.CheckAccess(DataActions.Write, cancellationToken) != DataActions.Write)
+            {
+                throw new UnauthorizedFhirActionException();
+            }
+
+            var input = request.Resources.Select(_ => new ResourceWrapperOperation(_.ResourceWrapper, true, false, null, false, false, null)).ToList();
+            await _store.MergeAsync(input, new MergeOptions(false), cancellationToken);
+            return await Task.FromResult(new ImportBundleResponse(request.Resources.Count));
+        }
+    }
+}

src/Microsoft.Health.Fhir.Core/Messages/BulkImport/ImportBundleRequest.cs

@@ -0,0 +1,28 @@
+// -------------------------------------------------------------------------------------------------
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License (MIT). See LICENSE in the repo root for license information.
+// -------------------------------------------------------------------------------------------------
+
+using System;
+using System.Collections.Generic;
+using EnsureThat;
+using MediatR;
+using Microsoft.Health.Fhir.Core.Features.Operations.Import;
+using Microsoft.Health.Fhir.Core.Features.Operations.Import.Models;
+using Microsoft.Health.Fhir.Core.Features.Persistence;
+
+namespace Microsoft.Health.Fhir.Core.Messages.Import
+{
+    public class ImportBundleRequest : IRequest<ImportBundleResponse>
+    {
+        public ImportBundleRequest(IList<ImportResource> resources)
+        {
+            Resources = EnsureArg.IsNotNull(resources, nameof(resources));
+        }
+
+        /// <summary>
+        /// Resources to import
+        /// </summary>
+        public IList<ImportResource> Resources { get; }
+    }
+}

src/Microsoft.Health.Fhir.Core/Messages/BulkImport/ImportBundleResponse.cs

@@ -0,0 +1,20 @@
+// -------------------------------------------------------------------------------------------------
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License (MIT). See LICENSE in the repo root for license information.
+// -------------------------------------------------------------------------------------------------
+
+namespace Microsoft.Health.Fhir.Core.Messages.Import
+{
+    public class ImportBundleResponse
+    {
+        public ImportBundleResponse(int count)
+        {
+            LoadedResources = count;
+        }
+
+        /// <summary>
+        /// Number of loaded resources
+        /// </summary>
+        public int LoadedResources { get; }
+    }
+}

src/Microsoft.Health.Fhir.Shared.Api.UnitTests/Controllers/ImportControllerTests.cs

@@ -18,6 +18,7 @@
 using Microsoft.Health.Fhir.Core.Features.Context;
 using Microsoft.Health.Fhir.Core.Features.Operations.Import;
 using Microsoft.Health.Fhir.Core.Features.Operations.Import.Models;
+using Microsoft.Health.Fhir.Core.Features.Persistence;
 using Microsoft.Health.Fhir.Core.Features.Routing;
 using Microsoft.Health.Fhir.Tests.Common;
 using Microsoft.Health.Test.Utilities;
@@ -100,6 +101,7 @@ private ImportController GetController(ImportTaskConfiguration bulkImportConfig)
                 _urlResolver,
                 optionsOperationConfiguration,
                 optionsFeatures,
+                Substitute.For<IResourceWrapperFactory>(),
                 NullLogger<ImportController>.Instance);
         }
 

src/Microsoft.Health.Fhir.Shared.Api/Controllers/ImportController.cs

@@ -5,21 +5,31 @@
 
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
+using System.IO;
 using System.Linq;
 using System.Net;
+using System.Net.Http;
+using System.Text;
+using System.Threading;
 using System.Threading.Tasks;
 using EnsureThat;
 using Hl7.Fhir.Model;
+using Hl7.Fhir.Serialization;
 using MediatR;
+using Microsoft.AspNetCore.DataProtection.KeyManagement;
+using Microsoft.AspNetCore.Http;
 using Microsoft.AspNetCore.Mvc;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
+using Microsoft.Extensions.Primitives;
 using Microsoft.Health.Api.Features.Audit;
 using Microsoft.Health.Core.Features.Context;
 using Microsoft.Health.Fhir.Api.Configs;
 using Microsoft.Health.Fhir.Api.Features.ActionResults;
 using Microsoft.Health.Fhir.Api.Features.Filters;
 using Microsoft.Health.Fhir.Api.Features.Headers;
+using Microsoft.Health.Fhir.Api.Features.Operations;
 using Microsoft.Health.Fhir.Api.Features.Operations.Import;
 using Microsoft.Health.Fhir.Api.Features.Routing;
 using Microsoft.Health.Fhir.Core.Configs;
@@ -29,9 +39,11 @@
 using Microsoft.Health.Fhir.Core.Features.Operations;
 using Microsoft.Health.Fhir.Core.Features.Operations.Import;
 using Microsoft.Health.Fhir.Core.Features.Operations.Import.Models;
+using Microsoft.Health.Fhir.Core.Features.Persistence;
 using Microsoft.Health.Fhir.Core.Features.Routing;
 using Microsoft.Health.Fhir.Core.Messages.Import;
 using Microsoft.Health.Fhir.ValueSets;
+using Newtonsoft.Json;
 
 namespace Microsoft.Health.Fhir.Api.Controllers
 {
@@ -58,13 +70,15 @@ public class ImportController : Controller
         private readonly FeatureConfiguration _features;
         private readonly ILogger<ImportController> _logger;
         private readonly ImportTaskConfiguration _importConfig;
+        private readonly IResourceWrapperFactory _resourceWrapperFactory;
 
         public ImportController(
             IMediator mediator,
             RequestContextAccessor<IFhirRequestContext> fhirRequestContextAccessor,
             IUrlResolver urlResolver,
             IOptions<OperationsConfiguration> operationsConfig,
             IOptions<FeatureConfiguration> features,
+            IResourceWrapperFactory resourceWrapperFactory,
             ILogger<ImportController> logger)
         {
             EnsureArg.IsNotNull(fhirRequestContextAccessor, nameof(fhirRequestContextAccessor));
@@ -79,9 +93,80 @@ public ImportController(
             _urlResolver = urlResolver;
             _features = features.Value;
             _mediator = mediator;
+            _resourceWrapperFactory = EnsureArg.IsNotNull(resourceWrapperFactory, nameof(resourceWrapperFactory));
             _logger = logger;
         }
 
+        [HttpPost]
+        [Route(KnownRoutes.ImportBundle)]
+        [AuditEventType(AuditEventSubType.ImportBundle)]
+        public async Task<IActionResult> ImportBundle()
+        {
+            var sw = Stopwatch.StartNew();
+            var startDate = DateTime.UtcNow;
+            var resources = new List<ImportResource>();
+            var keys = new HashSet<ResourceKey>();
+            var fhirParser = new FhirJsonParser();
+            var importParser = new ImportResourceParser(fhirParser, _resourceWrapperFactory);
+            var index = 0L;
+            using var reader = new StreamReader(Request.Body, Encoding.UTF8);
+            try
+            {
+                if (Request.ContentType != null && Request.ContentType.Contains("application/fhir+ndjson", StringComparison.OrdinalIgnoreCase))
+                {
+                    var line = await reader.ReadLineAsync();
+                    while (line != null)
+                    {
+                        var importResource = importParser.Parse(index, 0, 0, line, ImportMode.IncrementalLoad);
+                        var key = importResource.ResourceWrapper.ToResourceKey(true);
+                        if (!keys.Add(key))
+                        {
+                            throw new RequestNotValidException(string.Format(Resources.ResourcesMustBeUnique, key));
+                        }
+
+                        resources.Add(importResource);
+                        line = await reader.ReadLineAsync();
+                        index++;
+                    }
+                }
+                else
+                {
+                    var str = await reader.ReadToEndAsync();
+                    var resource = await fhirParser.ParseAsync(str);
+                    var bundle = resource.ToResourceElement().ToPoco<Bundle>();
+                    foreach (var entry in bundle.Entry)
+                    {
+                        if (entry.Request?.Method != Bundle.HTTPVerb.PUT)
+                        {
+                            throw new RequestNotValidException(string.Format(Resources.InvalidBundleEntry, entry.Request?.Method, KnownRoutes.ImportBundle));
+                        }
+
+                        var importResource = importParser.Parse(index, 0, 0, entry.Resource, ImportMode.IncrementalLoad);
+                        var key = importResource.ResourceWrapper.ToResourceKey(true);
+                        if (!keys.Add(key))
+                        {
+                            throw new RequestNotValidException(string.Format(Resources.ResourcesMustBeUnique, key));
+                        }
+
+                        resources.Add(importResource);
+                        index++;
+                    }
+                }
+            }
+            catch (Exception)
+            {
+                throw new RequestNotValidException(string.Format(Resources.ParsingError, $"Unable to parse resource at index={index}"));
+            }
+
+            var request = new ImportBundleRequest(resources);
+            var response = await _mediator.ImportBundleAsync(request.Resources, HttpContext.RequestAborted);
+            var result = new ImportBundleResult(response.LoadedResources, HttpStatusCode.OK);
+            result.Headers["LoadedResources"] = response.LoadedResources.ToString();
+            await _mediator.Publish(new ImportBundleMetricsNotification(startDate, DateTime.UtcNow, response.LoadedResources), CancellationToken.None);
+            _logger.LogInformation("Loaded {LoadedResources} resources, elapsed {Milliseconds} milliseconds.", response.LoadedResources, (int)sw.Elapsed.TotalMilliseconds);
+            return result;
+        }
+
         [HttpPost]
         [Route(KnownRoutes.Import)]
         [ServiceFilter(typeof(ValidateImportRequestFilterAttribute))]
@@ -106,14 +191,14 @@ public async Task<IActionResult> Import([FromBody] Parameters importTaskParamete
                 throw new RequestNotValidException(Resources.InitialImportModeNotEnabled);
             }
 
-            CreateImportResponse response = await _mediator.ImportAsync(
-                 _fhirRequestContextAccessor.RequestContext.Uri,
-                 importRequest.InputFormat,
-                 importRequest.InputSource,
-                 importRequest.Input,
-                 importRequest.StorageDetail,
-                 initialLoad ? ImportMode.InitialLoad : ImportMode.IncrementalLoad, // default to incremental mode
-                 HttpContext.RequestAborted);
+            var response = await _mediator.ImportAsync(
+                    _fhirRequestContextAccessor.RequestContext.Uri,
+                    importRequest.InputFormat,
+                    importRequest.InputSource,
+                    importRequest.Input,
+                    importRequest.StorageDetail,
+                    initialLoad ? ImportMode.InitialLoad : ImportMode.IncrementalLoad, // default to incremental mode
+                    HttpContext.RequestAborted);
 
             var bulkImportResult = ImportResult.Accepted();
             bulkImportResult.SetContentLocationHeader(_urlResolver, OperationsConstants.Import, response.TaskId);

src/Microsoft.Health.Fhir.Shared.Client/FhirClient.cs

@@ -445,6 +445,19 @@ public async Task<FhirResponse> ConvertDataAsync(Parameters parameters, Cancella
             return new FhirResponse(response);
         }
 
+        public async Task<HttpResponseMessage> ImportBundleAsync(string bundle, bool isNdJson = false, CancellationToken cancellationToken = default)
+        {
+            string requestPath = "$import/Bundle";
+            using var message = new HttpRequestMessage(HttpMethod.Post, requestPath)
+            {
+                Content = new StringContent(bundle, Encoding.UTF8, isNdJson ? "application/fhir+ndjson" : _contentType),
+            };
+
+            using var response = await HttpClient.SendAsync(message, cancellationToken);
+            await EnsureSuccessStatusCodeAsync(response);
+            return response;
+        }
+
         public async Task<Uri> ImportAsync(Parameters parameters, CancellationToken cancellationToken = default)
         {
             string requestPath = "$import";