Add async processing for patient bulk uploads

.github/workflows/test.yml

@@ -34,6 +34,7 @@ jobs:
           --health-interval 10s
           --health-timeout 5s
           --health-retries 5
+          --name sr_test_db
         ports:
           - 5432:5432
     steps:
@@ -60,6 +61,7 @@ jobs:
         run: |
           chmod 0600 $PGPASSFILE
           db-setup/create-db.sh
+          docker restart --time 0 sr_test_db
       - name: Run tests
         env:
           OKTA_TESTING_DISABLEHTTPS: true

backend/db-setup/reset-db.sql

@@ -1,3 +1,5 @@
+ALTER SYSTEM SET max_connections = 500;
+
 DROP SCHEMA IF EXISTS simple_report CASCADE;
 
 CREATE SCHEMA IF NOT EXISTS simple_report;

backend/src/main/java/gov/cdc/usds/simplereport/SimpleReportApplication.java

@@ -14,6 +14,7 @@
 import gov.cdc.usds.simplereport.service.OrganizationInitializingService;
 import gov.cdc.usds.simplereport.service.ScheduledTasksService;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -24,7 +25,9 @@
 import org.springframework.cloud.openfeign.EnableFeignClients;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Profile;
+import org.springframework.scheduling.annotation.EnableAsync;
 import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.security.core.context.SecurityContextHolder;
 
 @Slf4j
 @SpringBootApplication
@@ -40,13 +43,20 @@
   CorsProperties.class,
   AzureStorageQueueReportingProperties.class
 })
+@EnableAsync
 @EnableScheduling
 @EnableFeignClients
 public class SimpleReportApplication {
   public static void main(String[] args) {
     SpringApplication.run(SimpleReportApplication.class, args);
   }
 
+  @Bean
+  public InitializingBean initializingBean() {
+    return () ->
+        SecurityContextHolder.setStrategyName(SecurityContextHolder.MODE_INHERITABLETHREADLOCAL);
+  }
+
   @Bean
   public CommandLineRunner initDiseasesOnStartup(DiseaseService initService) {
     return args -> initService.initDiseases();

backend/src/main/java/gov/cdc/usds/simplereport/idp/repository/DemoOktaRepository.java

@@ -26,8 +26,11 @@
 import java.util.UUID;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.support.ScopeNotActiveException;
 import org.springframework.context.annotation.Profile;
 import org.springframework.http.HttpStatus;
+import org.springframework.security.core.GrantedAuthority;
+import org.springframework.security.core.context.SecurityContextHolder;
 import org.springframework.stereotype.Service;
 
 /** Handles all user/organization management in Okta */
@@ -309,13 +312,20 @@ private Optional<OrganizationRoleClaims> getOrganizationRoleClaimsFromTenantData
 
   public Optional<OrganizationRoleClaims> getOrganizationRoleClaimsForUser(String username) {
     // when accessing tenant data, bypass okta and get org from the altered authorities
-    if (tenantDataContextHolder.hasBeenPopulated()
-        && username.equals(tenantDataContextHolder.getUsername())) {
-      return getOrganizationRoleClaimsFromTenantDataAccess(
-          tenantDataContextHolder.getAuthorities());
+    try {
+      if (tenantDataContextHolder.hasBeenPopulated()
+          && username.equals(tenantDataContextHolder.getUsername())) {
+        return getOrganizationRoleClaimsFromTenantDataAccess(
+            tenantDataContextHolder.getAuthorities());
+      }
+      return Optional.ofNullable(usernameOrgRolesMap.get(username));
+    } catch (ScopeNotActiveException e) {
+      Set<String> authorities =
+          SecurityContextHolder.getContext().getAuthentication().getAuthorities().stream()
+              .map(GrantedAuthority::getAuthority)
+              .collect(Collectors.toSet());
+      return getOrganizationRoleClaimsFromTenantDataAccess(authorities);
     }
-
-    return Optional.ofNullable(usernameOrgRolesMap.get(username));
   }
 
   public void reset() {

backend/src/main/java/gov/cdc/usds/simplereport/service/ApiUserService.java

@@ -37,6 +37,7 @@
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.support.ScopeNotActiveException;
 import org.springframework.security.access.AccessDeniedException;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
@@ -505,13 +506,17 @@ private ApiUser getCurrentApiUser() {
       return getCurrentApiUserNoCache();
     }
 
-    if (_apiUserContextHolder.hasBeenPopulated()) {
-      log.debug("Retrieving user from request context");
-      return _apiUserContextHolder.getCurrentApiUser();
+    try {
+      if (_apiUserContextHolder.hasBeenPopulated()) {
+        log.debug("Retrieving user from request context");
+        return _apiUserContextHolder.getCurrentApiUser();
+      }
+      ApiUser user = getCurrentApiUserNoCache();
+      _apiUserContextHolder.setCurrentApiUser(user);
+      return user;
+    } catch (ScopeNotActiveException e) {
+      return getCurrentApiUserNoCache();
     }
-    ApiUser user = getCurrentApiUserNoCache();
-    _apiUserContextHolder.setCurrentApiUser(user);
-    return user;
   }
 
   private ApiUser getCurrentApiUserNoCache() {

backend/src/main/java/gov/cdc/usds/simplereport/service/OrganizationService.java

@@ -27,6 +27,7 @@
 import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.support.ScopeNotActiveException;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -51,12 +52,16 @@ public void resetOrganizationRolesContext() {
   }
 
   public Optional<OrganizationRoles> getCurrentOrganizationRoles() {
-    if (organizationRolesContext.hasBeenPopulated()) {
-      return organizationRolesContext.getOrganizationRoles();
+    try {
+      if (organizationRolesContext.hasBeenPopulated()) {
+        return organizationRolesContext.getOrganizationRoles();
+      }
+      var result = fetchCurrentOrganizationRoles();
+      organizationRolesContext.setOrganizationRoles(result);
+      return result;
+    } catch (ScopeNotActiveException e) {
+      return fetchCurrentOrganizationRoles();
     }
-    var result = fetchCurrentOrganizationRoles();
-    organizationRolesContext.setOrganizationRoles(result);
-    return result;
   }
 
   private Optional<OrganizationRoles> fetchCurrentOrganizationRoles() {

backend/src/main/java/gov/cdc/usds/simplereport/service/PatientBulkUploadService.java

@@ -1,75 +1,39 @@
 package gov.cdc.usds.simplereport.service;
 
-import static gov.cdc.usds.simplereport.api.Translators.parsePersonRole;
-import static gov.cdc.usds.simplereport.api.Translators.parsePhoneType;
-import static gov.cdc.usds.simplereport.api.Translators.parseUserShortDate;
-import static gov.cdc.usds.simplereport.api.Translators.parseYesNo;
-import static gov.cdc.usds.simplereport.validators.CsvValidatorUtils.convertEthnicityToDatabaseValue;
-import static gov.cdc.usds.simplereport.validators.CsvValidatorUtils.convertRaceToDatabaseValue;
-import static gov.cdc.usds.simplereport.validators.CsvValidatorUtils.convertSexToDatabaseValue;
-
-import com.fasterxml.jackson.databind.MappingIterator;
 import gov.cdc.usds.simplereport.api.model.errors.CsvProcessingException;
 import gov.cdc.usds.simplereport.api.model.filerow.PatientUploadRow;
 import gov.cdc.usds.simplereport.api.uploads.PatientBulkUploadResponse;
 import gov.cdc.usds.simplereport.config.AuthorizationConfiguration;
-import gov.cdc.usds.simplereport.db.model.Facility;
-import gov.cdc.usds.simplereport.db.model.Organization;
-import gov.cdc.usds.simplereport.db.model.Person;
-import gov.cdc.usds.simplereport.db.model.PhoneNumber;
-import gov.cdc.usds.simplereport.db.model.auxiliary.StreetAddress;
 import gov.cdc.usds.simplereport.db.model.auxiliary.UploadStatus;
 import gov.cdc.usds.simplereport.service.model.reportstream.FeedbackMessage;
-import gov.cdc.usds.simplereport.validators.CsvValidatorUtils;
 import gov.cdc.usds.simplereport.validators.FileValidator;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
-import java.util.ArrayList;
-import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
 import java.util.UUID;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Transactional;
 
 /**
  * Service to upload a roster of patient data given a CSV input. Formerly restricted to superusers
- * but now (almost) available to end users.
- *
- * <p>Updated by emmastephenson on 10/24/2022
+ * but now available to end users.
  */
 @Service
-@Transactional
 @RequiredArgsConstructor
 @Slf4j
 public class PatientBulkUploadService {
 
-  private final PersonService _personService;
-  private final AddressValidationService _addressValidationService;
-  private final OrganizationService _organizationService;
-  private final FileValidator<PatientUploadRow> _patientBulkUploadFileValidator;
+  private final FileValidator<PatientUploadRow> patientUploadRowFileValidator;
+  private final PatientBulkUploadServiceAsync patientBulkUploadServiceAsync;
 
-  // This authorization will change once we open the feature to end users
-  @AuthorizationConfiguration.RequireGlobalAdminUser
+  @AuthorizationConfiguration.RequirePermissionCreatePatientAtFacility
   public PatientBulkUploadResponse processPersonCSV(InputStream csvStream, UUID facilityId)
       throws IllegalArgumentException {
 
     PatientBulkUploadResponse result = new PatientBulkUploadResponse();
 
-    Organization currentOrganization = _organizationService.getCurrentOrganization();
-
-    // Patients do not need to be assigned to a facility, but if an id is given it must be valid
-    Optional<Facility> assignedFacility =
-        Optional.ofNullable(facilityId).map(_organizationService::getFacilityInCurrentOrg);
-
-    Set<Person> patientsList = new HashSet<>();
-    List<PhoneNumber> phoneNumbersList = new ArrayList<>();
-
     byte[] content;
 
     try {
@@ -80,110 +44,16 @@ public PatientBulkUploadResponse processPersonCSV(InputStream csvStream, UUID fa
     }
 
     List<FeedbackMessage> errors =
-        _patientBulkUploadFileValidator.validate(new ByteArrayInputStream(content));
+        patientUploadRowFileValidator.validate(new ByteArrayInputStream(content));
 
     if (!errors.isEmpty()) {
       result.setStatus(UploadStatus.FAILURE);
       result.setErrors(errors.toArray(FeedbackMessage[]::new));
       return result;
     }
 
-    // This is the point where we need to figure out multithreading
-    // because what needs to happen is that we return a success message to the end user
-    // but continue to process the csv (create person records) in the background.
-    // Putting a pin in it for now.
-
-    final MappingIterator<Map<String, String>> valueIterator =
-        CsvValidatorUtils.getIteratorForCsv(new ByteArrayInputStream(content));
-
-    while (valueIterator.hasNext()) {
-      final Map<String, String> row = CsvValidatorUtils.getNextRow(valueIterator);
-
-      try {
-
-        PatientUploadRow extractedData = new PatientUploadRow(row);
-
-        // Fetch address information
-        StreetAddress address =
-            _addressValidationService.getValidatedAddress(
-                extractedData.getStreet().getValue(),
-                extractedData.getStreet2().getValue(),
-                extractedData.getCity().getValue(),
-                extractedData.getState().getValue(),
-                extractedData.getZipCode().getValue(),
-                null);
-
-        String country =
-            extractedData.getCountry().getValue() == null
-                ? "USA"
-                : extractedData.getCountry().getValue();
-
-        if (_personService.isDuplicatePatient(
-            extractedData.getFirstName().getValue(),
-            extractedData.getLastName().getValue(),
-            parseUserShortDate(extractedData.getDateOfBirth().getValue()),
-            currentOrganization,
-            assignedFacility)) {
-          continue;
-        }
-
-        // create new person with current organization, then add to new patients list
-        Person newPatient =
-            new Person(
-                currentOrganization,
-                assignedFacility.orElse(null),
-                null, // lookupid
-                extractedData.getFirstName().getValue(),
-                extractedData.getMiddleName().getValue(),
-                extractedData.getLastName().getValue(),
-                extractedData.getSuffix().getValue(),
-                parseUserShortDate(extractedData.getDateOfBirth().getValue()),
-                address,
-                country,
-                parsePersonRole(extractedData.getRole().getValue(), false),
-                List.of(extractedData.getEmail().getValue()),
-                convertRaceToDatabaseValue(extractedData.getRace().getValue()),
-                convertEthnicityToDatabaseValue(extractedData.getEthnicity().getValue()),
-                null, // tribalAffiliation
-                convertSexToDatabaseValue(extractedData.getBiologicalSex().getValue()),
-                parseYesNo(extractedData.getResidentCongregateSetting().getValue()),
-                parseYesNo(extractedData.getEmployedInHealthcare().getValue()),
-                null, // preferredLanguage
-                null // testResultDeliveryPreference
-                );
-
-        if (!patientsList.contains(newPatient)) {
-          // collect phone numbers and associate them with the patient
-          // then add to phone numbers list and set primary phone, if exists
-          List<PhoneNumber> newPhoneNumbers =
-              _personService.assignPhoneNumbersToPatient(
-                  newPatient,
-                  List.of(
-                      new PhoneNumber(
-                          parsePhoneType(extractedData.getPhoneNumberType().getValue()),
-                          extractedData.getPhoneNumber().getValue())));
-          phoneNumbersList.addAll(newPhoneNumbers);
-          newPhoneNumbers.stream().findFirst().ifPresent(newPatient::setPrimaryPhone);
-
-          patientsList.add(newPatient);
-        }
-      } catch (IllegalArgumentException e) {
-        String errorMessage = "Error uploading patient roster";
-        log.error(
-            errorMessage
-                + " for organization "
-                + currentOrganization.getExternalId()
-                + " and facility "
-                + facilityId);
-        throw new IllegalArgumentException(errorMessage);
-      }
-    }
-
-    _personService.addPatientsAndPhoneNumbers(patientsList, phoneNumbersList);
-
-    log.info("CSV patient upload completed for {}", currentOrganization.getOrganizationName());
+    patientBulkUploadServiceAsync.savePatients(content, facilityId);
     result.setStatus(UploadStatus.SUCCESS);
-    // eventually want to send an email here instead of return success
     return result;
   }
 }