<commit_before>package ca.uhn.fhir.jpa.dao.r4;

import ca.uhn.fhir.interceptor.executor.InterceptorService;
import ca.uhn.fhir.jpa.interceptor.UserRequestRetryVersionConflictsInterceptor;
import ca.uhn.fhir.jpa.partition.SystemRequestDetails;
import ca.uhn.fhir.jpa.searchparam.SearchParameterMap;
import ca.uhn.fhir.rest.api.PatchTypeEnum;
import ca.uhn.fhir.rest.api.server.IBundleProvider;
import ca.uhn.fhir.rest.param.TokenParam;
import ca.uhn.fhir.rest.server.RestfulServer;
import ca.uhn.fhir.rest.server.exceptions.PreconditionFailedException;
import ca.uhn.fhir.rest.server.exceptions.ResourceVersionConflictException;
import ca.uhn.fhir.rest.server.servlet.ServletRequestDetails;
import ca.uhn.fhir.util.BundleBuilder;
import ca.uhn.fhir.util.HapiExtensions;
import org.hl7.fhir.instance.model.api.IIdType;
import org.hl7.fhir.r4.model.BooleanType;
import org.hl7.fhir.r4.model.Bundle;
import org.hl7.fhir.r4.model.CodeType;
import org.hl7.fhir.r4.model.Coverage;
import org.hl7.fhir.r4.model.Enumerations;
import org.hl7.fhir.r4.model.ExplanationOfBenefit;
import org.hl7.fhir.r4.model.IdType;
import org.hl7.fhir.r4.model.Observation;
import org.hl7.fhir.r4.model.Parameters;
import org.hl7.fhir.r4.model.Patient;
import org.hl7.fhir.r4.model.Practitioner;
import org.hl7.fhir.r4.model.Reference;
import org.hl7.fhir.r4.model.SearchParameter;
import org.hl7.fhir.r4.model.StringType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.test.annotation.DirtiesContext;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

@SuppressWarnings({"unchecked", "deprecation", "Duplicates"})
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
public class FhirResourceDaoR4ConcurrentWriteTest extends BaseJpaR4Test {

	private static final org.slf4j.Logger ourLog = org.slf4j.LoggerFactory.getLogger(FhirResourceDaoR4ConcurrentWriteTest.class);
	private ExecutorService myExecutor;
	private UserRequestRetryVersionConflictsInterceptor myRetryInterceptor;


	@BeforeEach
	public void before() {
		myExecutor = Executors.newFixedThreadPool(10);
		myRetryInterceptor = new UserRequestRetryVersionConflictsInterceptor();

		RestfulServer server = new RestfulServer(myFhirCtx);
		when(mySrd.getServer()).thenReturn(server);

	}

	@AfterEach
	public void after() {
		myExecutor.shutdown();
		myInterceptorRegistry.unregisterInterceptor(myRetryInterceptor);
	}

	@Test
	public void testConcurrentTransactionCreates() {
		myDaoConfig.setMatchUrlCache(true);

		AtomicInteger counter = new AtomicInteger(0);
		Runnable creator = () -> {
			BundleBuilder bb = new BundleBuilder(myFhirCtx);
			String patientId = "Patient/PT" + counter.get();
			IdType practitionerId = IdType.newRandomUuid();
			IdType practitionerId2 = IdType.newRandomUuid();

			ExplanationOfBenefit eob = new ExplanationOfBenefit();
			eob.addIdentifier().setSystem("foo").setValue("" + counter.get());
			eob.getPatient().setReference(patientId);
			eob.addCareTeam().getProvider().setReference(practitionerId.getValue());
			eob.addCareTeam().getProvider().setReference(practitionerId2.getValue());
			bb.addTransactionUpdateEntry(eob).conditional("ExplanationOfBenefit?identifier=foo|" + counter.get());

			Patient pt = new Patient();
			pt.setId(patientId);
			pt.setActive(true);
			bb.addTransactionUpdateEntry(pt);

			Coverage coverage = new Coverage();
			coverage.addIdentifier().setSystem("foo").setValue("" + counter.get());
			coverage.getBeneficiary().setReference(patientId);
			bb.addTransactionUpdateEntry(coverage).conditional("Coverage?identifier=foo|" + counter.get());

			Practitioner practitioner = new Practitioner();
			practitioner.setId(practitionerId);
			practitioner.addIdentifier().setSystem("foo").setValue("" + counter.get());
			bb.addTransactionCreateEntry(practitioner).conditional("Practitioner?identifier=foo|" + counter.get());

			Practitioner practitioner2 = new Practitioner();
			practitioner2.setId(practitionerId2);
			practitioner2.addIdentifier().setSystem("foo2").setValue("" + counter.get());
			bb.addTransactionCreateEntry(practitioner2).conditional("Practitioner?identifier=foo2|" + counter.get());

			Observation obs = new Observation();
			obs.setId("Observation/OBS" + counter);
			obs.getSubject().setReference(pt.getId());
			bb.addTransactionUpdateEntry(obs);

			Bundle input = (Bundle) bb.getBundle();
			mySystemDao.transaction(new SystemRequestDetails(), input);
		};

		for (int i = 0; i < 10; i++) {
			counter.set(i);
			ourLog.info("*********************************************************************************");
			ourLog.info("Starting pass {}", i);
			ourLog.info("*********************************************************************************");

			List<Future<?>> futures = new ArrayList<>();
			for (int j = 0; j < 10; j++) {
				futures.add(myExecutor.submit(creator));
			}

			for (Future<?> next : futures) {
				try {
					next.get();
				} catch (Exception e) {
					// ignore
				}
			}

			creator.run();
		}

		runInTransaction(() -> {
			Map<String, Integer> counts = new TreeMap<>();
			myResourceTableDao
				.findAll()
				.stream()
				.forEach(t -> {
					counts.putIfAbsent(t.getResourceType(), 0);
					int value = counts.get(t.getResourceType());
					value++;
					counts.put(t.getResourceType(), value);
				});
			ourLog.info("Counts: {}", counts);

			assertEquals(10, counts.get("Patient"), counts.toString());
		});

	}


	@Test
	public void testCreateWithClientAssignedId() {
		myInterceptorRegistry.registerInterceptor(myRetryInterceptor);
		String value = UserRequestRetryVersionConflictsInterceptor.RETRY + "; " + UserRequestRetryVersionConflictsInterceptor.MAX_RETRIES + "=10";
		when(mySrd.getHeaders(eq(UserRequestRetryVersionConflictsInterceptor.HEADER_NAME))).thenReturn(Collections.singletonList(value));

		List<Future<?>> futures = new ArrayList<>();
		for (int i = 0; i < 5; i++) {
			Patient p = new Patient();
			p.setId("ABC");
			p.setActive(true);
			p.addIdentifier().setValue("VAL" + i);
			Runnable task = () -> myPatientDao.update(p, mySrd);
			Future<?> future = myExecutor.submit(task);<commit_msg>Add test<commit_after>package ca.uhn.fhir.jpa.dao.r4;

import ca.uhn.fhir.interceptor.executor.InterceptorService;
import ca.uhn.fhir.jpa.interceptor.UserRequestRetryVersionConflictsInterceptor;
import ca.uhn.fhir.jpa.partition.SystemRequestDetails;
import ca.uhn.fhir.jpa.searchparam.SearchParameterMap;
import ca.uhn.fhir.rest.api.PatchTypeEnum;
import ca.uhn.fhir.rest.api.server.IBundleProvider;
import ca.uhn.fhir.rest.param.TokenParam;
import ca.uhn.fhir.rest.server.RestfulServer;
import ca.uhn.fhir.rest.server.exceptions.PreconditionFailedException;
import ca.uhn.fhir.rest.server.exceptions.ResourceVersionConflictException;
import ca.uhn.fhir.rest.server.servlet.ServletRequestDetails;
import ca.uhn.fhir.util.BundleBuilder;
import ca.uhn.fhir.util.HapiExtensions;
import org.hl7.fhir.instance.model.api.IIdType;
import org.hl7.fhir.r4.model.BooleanType;
import org.hl7.fhir.r4.model.Bundle;
import org.hl7.fhir.r4.model.CodeType;
import org.hl7.fhir.r4.model.Coverage;
import org.hl7.fhir.r4.model.Enumerations;
import org.hl7.fhir.r4.model.ExplanationOfBenefit;
import org.hl7.fhir.r4.model.IdType;
import org.hl7.fhir.r4.model.Observation;
import org.hl7.fhir.r4.model.Parameters;
import org.hl7.fhir.r4.model.Patient;
import org.hl7.fhir.r4.model.Practitioner;
import org.hl7.fhir.r4.model.Reference;
import org.hl7.fhir.r4.model.SearchParameter;
import org.hl7.fhir.r4.model.StringType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.test.annotation.DirtiesContext;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

@SuppressWarnings({"unchecked", "deprecation", "Duplicates"})
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS)
public class FhirResourceDaoR4ConcurrentWriteTest extends BaseJpaR4Test {

	private static final org.slf4j.Logger ourLog = org.slf4j.LoggerFactory.getLogger(FhirResourceDaoR4ConcurrentWriteTest.class);
	private ExecutorService myExecutor;
	private UserRequestRetryVersionConflictsInterceptor myRetryInterceptor;


	@BeforeEach
	public void before() {
		myExecutor = Executors.newFixedThreadPool(10);
		myRetryInterceptor = new UserRequestRetryVersionConflictsInterceptor();

		RestfulServer server = new RestfulServer(myFhirCtx);
		when(mySrd.getServer()).thenReturn(server);

	}

	@AfterEach
	public void after() {
		myExecutor.shutdown();
		myInterceptorRegistry.unregisterInterceptor(myRetryInterceptor);
	}

	@Test
	public void testConcurrentTransactionCreates() {
		myDaoConfig.setMatchUrlCache(true);

		AtomicInteger counter = new AtomicInteger(0);
		Runnable creator = () -> {
			BundleBuilder bb = new BundleBuilder(myFhirCtx);
			String patientId = "Patient/PT" + counter.get();
			IdType practitionerId = IdType.newRandomUuid();
			IdType practitionerId2 = IdType.newRandomUuid();

			ExplanationOfBenefit eob = new ExplanationOfBenefit();
			eob.addIdentifier().setSystem("foo").setValue("" + counter.get());
			eob.getPatient().setReference(patientId);
			eob.addCareTeam().getProvider().setReference(practitionerId.getValue());
			eob.addCareTeam().getProvider().setReference(practitionerId2.getValue());
			bb.addTransactionUpdateEntry(eob).conditional("ExplanationOfBenefit?identifier=foo|" + counter.get());

			Patient pt = new Patient();
			pt.setId(patientId);
			pt.setActive(true);
			bb.addTransactionUpdateEntry(pt);

			Coverage coverage = new Coverage();
			coverage.addIdentifier().setSystem("foo").setValue("" + counter.get());
			coverage.getBeneficiary().setReference(patientId);
			bb.addTransactionUpdateEntry(coverage).conditional("Coverage?identifier=foo|" + counter.get());

			Practitioner practitioner = new Practitioner();
			practitioner.setId(practitionerId);
			practitioner.addIdentifier().setSystem("foo").setValue("" + counter.get());
			bb.addTransactionCreateEntry(practitioner).conditional("Practitioner?identifier=foo|" + counter.get());

			Practitioner practitioner2 = new Practitioner();
			practitioner2.setId(practitionerId2);
			practitioner2.addIdentifier().setSystem("foo2").setValue("" + counter.get());
			bb.addTransactionCreateEntry(practitioner2).conditional("Practitioner?identifier=foo2|" + counter.get());

			Observation obs = new Observation();
			obs.setId("Observation/OBS" + counter);
			obs.getSubject().setReference(pt.getId());
			bb.addTransactionUpdateEntry(obs);

			Bundle input = (Bundle) bb.getBundle();
			mySystemDao.transaction(new SystemRequestDetails(), input);
		};

		for (int i = 0; i < 10; i++) {
			counter.set(i);
			ourLog.info("*********************************************************************************");
			ourLog.info("Starting pass {}", i);
			ourLog.info("*********************************************************************************");

			List<Future<?>> futures = new ArrayList<>();
			for (int j = 0; j < 10; j++) {
				futures.add(myExecutor.submit(creator));
			}

			for (Future<?> next : futures) {
				try {
					next.get();
				} catch (Exception e) {
					// ignore
				}
			}

			creator.run();
		}

		runInTransaction(() -> {
			Map<String, Integer> counts = new TreeMap<>();
			myResourceTableDao
				.findAll()
				.stream()
				.forEach(t -> {
					counts.putIfAbsent(t.getResourceType(), 0);
					int value = counts.get(t.getResourceType());
					value++;
					counts.put(t.getResourceType(), value);
				});
			ourLog.info("Counts: {}", counts);

			assertEquals(10, counts.get("Patient"), counts.toString());
		});

	}


	/**
	 * Make a transaction with conditional updates that will fail due to
	 * constraint errors and be retried automatically. Make sure that the
	 * retry succeeds and that the data ultimately gets written.
	 */
	@Test
	public void testConcurrentTransactionConditionalUpdates() throws ExecutionException, InterruptedException {
		myInterceptorRegistry.registerInterceptor(myRetryInterceptor);

		Patient pt = new Patient();
		pt.setId("Patient/A");
		pt.addIdentifier().setSystem("http://foo").setValue("pt1");
		myPatientDao.update(pt);

		Observation obs = new Observation();
		obs.setId("Observation/O");
		obs.addIdentifier().setSystem("http://foo").setValue("obs1");
		myObservationDao.update(obs);

		AtomicInteger counter = new AtomicInteger(0);
		Runnable creator = () -> {
			BundleBuilder bb = new BundleBuilder(myFhirCtx);

			Patient patient = new Patient();
			patient.setId(IdType.newRandomUuid());
			patient.addIdentifier().setSystem("http://foo").setValue("pt1");
			patient.addName().setFamily("fam-" + counter.incrementAndGet());
			bb.addTransactionUpdateEntry(patient).conditional("Patient?identifier=http://foo|pt1");

			Observation observation = new Observation();
			observation.setId(IdType.newRandomUuid());
			observation.addIdentifier().setSystem("http://foo").setValue("obs1");
			observation.getCode().setText("obs-" + counter.incrementAndGet());
			observation.getSubject().setReference(patient.getId());
			bb.addTransactionUpdateEntry(observation).conditional("Observation?identifier=http://foo|obs1");

			Bundle input = (Bundle) bb.getBundle();
			input.getEntry().get(0).getResource().setId("Patient/A");
			input.getEntry().get(1).getResource().setId("Observation/O");
			SystemRequestDetails requestDetails = new SystemRequestDetails();
			UserRequestRetryVersionConflictsInterceptor.addRetryHeader(requestDetails, 20);
			mySystemDao.transaction(requestDetails, input);
		};

		List<Future<?>> futures = new ArrayList<>();
		for (int j = 0; j < 10; j++) {
			futures.add(myExecutor.submit(creator));
		}

		for (Future<?> next : futures) {
			next.get();
		}

		runInTransaction(() -> {
			Map<String, Integer> counts = new TreeMap<>();
			myResourceTableDao
				.findAll()
				.stream()
				.forEach(t -> {
					counts.putIfAbsent(t.getResourceType(), 0);
					int value = counts.get(t.getResourceType());
					value++;
					counts.put(t.getResourceType(), value);
				});
			ourLog.info("Counts: {}", counts);

			assertEquals(1, counts.get("Patient"), counts.toString());
			assertEquals(1, counts.get("Observation"), counts.toString());
			logAllResourceLinks();
			assertEquals(2, myResourceLinkDao.count());
			assertEquals(22, myResourceHistoryTableDao.count());
		});

	}


	@Test
	public void testCreateWithClientAssignedId() {
		myInterceptorRegistry.registerInterceptor(myRetryInterceptor);
		String value = UserRequestRetryVersionConflictsInterceptor.RETRY + "; " + UserRequestRetryVersionConflictsInterceptor.MAX_RETRIES + "=10";
		when(mySrd.getHeaders(eq(UserRequestRetryVersionConflictsInterceptor.HEADER_NAME))).thenReturn(Collections.singletonList(value));

		List<Future<?>> futures = new ArrayList<>();
		for (int i = 0; i < 5; i++) {
			Patient p = new Patient();
			p.setId("ABC");
			p.setActive(true);
			p.addIdentifier().setValue("VAL" + i);
			Runnable task = () -> myPatientDao.update(p, mySrd);
			Future<?> future = myExecutor.submit(task);
{'fraction_non_alphanumeric': 0.15584179970972423, 'fraction_numerical': 0.008466376390904693, 'mean_word_length': 5.505507474429583, 'pattern_counts': {'":': 0, '<': 36, '<?xml version=': 0, '>': 40, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4447120', 'n_tokens_mistral': 5843, 'n_tokens_neox': 5249, 'n_words': 811}