Request for exposing client id and username in batch job status api (#5526)

* initial tests and implementation of the feature persistence layer.

* initial tests and implementation pointcut to modify JobInstance before persistence.

* spotless happy.

* Addressing comments from first code review.

* happy spotless

---------

Co-authored-by: peartree <etienne.poirier@smilecdr.com>

hapi-fhir-base/src/main/java/ca/uhn/fhir/interceptor/api/Pointcut.java

@@ -2935,6 +2935,31 @@ public enum Pointcut implements IPointcut {
 			"ca.uhn.fhir.rest.api.server.RequestDetails",
 			"org.hl7.fhir.instance.model.api.IBaseResource"),
 
+	/**
+	 * <b>Storage Hook:</b>
+	 * Invoked before a batch job is persisted to the database.
+	 * <p>
+	 * Hooks will have access to the content of the job being created
+	 * and may choose to make modifications to it. These changes will be
+	 * reflected in permanent storage.
+	 * </p>
+	 * Hooks may accept the following parameters:
+	 * <ul>
+	 * <li>
+	 * ca.uhn.fhir.batch2.model.JobInstance
+	 * </li>
+	 * <li>
+	 * ca.uhn.fhir.rest.api.server.RequestDetails - A bean containing details about the request that lead to the creation
+	 * of the jobInstance.
+	 * </li>
+	 * </ul>
+	 * <p>
+	 * Hooks should return <code>void</code>.
+	 * </p>
+	 */
+	STORAGE_PRESTORAGE_BATCH_JOB_CREATE(
+			void.class, "ca.uhn.fhir.batch2.model.JobInstance", "ca.uhn.fhir.rest.api.server.RequestDetails"),
+
 	/**
 	 * This pointcut is used only for unit tests. Do not use in production code as it may be changed or
 	 * removed at any time.

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/batch2/JobInstanceUtil.java

@@ -61,6 +61,8 @@ class JobInstanceUtil {
 		retVal.setReport(theEntity.getReport());
 		retVal.setEstimatedTimeRemaining(theEntity.getEstimatedTimeRemaining());
 		retVal.setWarningMessages(theEntity.getWarningMessages());
+		retVal.setTriggeringUsername(theEntity.getTriggeringUsername());
+		retVal.setTriggeringClientId(theEntity.getTriggeringClientId());
 		return retVal;
 	}
 
@@ -95,6 +97,8 @@ class JobInstanceUtil {
 		theJobInstanceEntity.setReport(theJobInstance.getReport());
 		theJobInstanceEntity.setEstimatedTimeRemaining(theJobInstance.getEstimatedTimeRemaining());
 		theJobInstanceEntity.setWarningMessages(theJobInstance.getWarningMessages());
+		theJobInstanceEntity.setTriggeringUsername(theJobInstance.getTriggeringUsername());
+		theJobInstanceEntity.setTriggeringClientId(theJobInstance.getTriggeringClientId());
 	}
 
 	/**

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/batch2/JpaBatch2Config.java

@@ -21,6 +21,7 @@ package ca.uhn.fhir.jpa.batch2;
 
 import ca.uhn.fhir.batch2.api.IJobPersistence;
 import ca.uhn.fhir.batch2.config.BaseBatch2Config;
+import ca.uhn.fhir.interceptor.api.IInterceptorBroadcaster;
 import ca.uhn.fhir.jpa.bulk.export.job.BulkExportJobConfig;
 import ca.uhn.fhir.jpa.dao.data.IBatch2JobInstanceRepository;
 import ca.uhn.fhir.jpa.dao.data.IBatch2WorkChunkRepository;
@@ -29,7 +30,6 @@ import jakarta.persistence.EntityManager;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
-import org.springframework.context.annotation.Primary;
 
 @Configuration
 @Import({BulkExportJobConfig.class})
@@ -40,28 +40,13 @@ public class JpaBatch2Config extends BaseBatch2Config {
 			IBatch2JobInstanceRepository theJobInstanceRepository,
 			IBatch2WorkChunkRepository theWorkChunkRepository,
 			IHapiTransactionService theTransactionService,
-			EntityManager theEntityManager) {
+			EntityManager theEntityManager,
+			IInterceptorBroadcaster theInterceptorBroadcaster) {
 		return new JpaJobPersistenceImpl(
-				theJobInstanceRepository, theWorkChunkRepository, theTransactionService, theEntityManager);
+				theJobInstanceRepository,
-	}
+				theWorkChunkRepository,
-
+				theTransactionService,
-	@Primary
+				theEntityManager,
-	@Bean
+				theInterceptorBroadcaster);
-	public IJobPersistence batch2JobInstancePersisterWrapper(
-			IBatch2JobInstanceRepository theJobInstanceRepository,
-			IBatch2WorkChunkRepository theWorkChunkRepository,
-			IHapiTransactionService theTransactionService,
-			EntityManager theEntityManager) {
-		IJobPersistence retVal = batch2JobInstancePersister(
-				theJobInstanceRepository, theWorkChunkRepository, theTransactionService, theEntityManager);
-		// Avoid H2 synchronization issues caused by
-		// https://github.com/h2database/h2database/issues/1808
-		// TODO: Update 2023-03-14 - The bug above appears to be fixed. I'm going to try
-		// disabing this and see if we can get away without it. If so, we can delete
-		// this entirely
-		//		if (HapiSystemProperties.isUnitTestModeEnabled()) {
-		//			retVal = ProxyUtil.synchronizedProxy(IJobPersistence.class, retVal);
-		//		}
-		return retVal;
 	}
 }

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/batch2/JpaJobPersistenceImpl.java

@@ -30,12 +30,17 @@ import ca.uhn.fhir.batch2.model.WorkChunkCreateEvent;
 import ca.uhn.fhir.batch2.model.WorkChunkErrorEvent;
 import ca.uhn.fhir.batch2.model.WorkChunkStatusEnum;
 import ca.uhn.fhir.batch2.models.JobInstanceFetchRequest;
+import ca.uhn.fhir.interceptor.api.HookParams;
+import ca.uhn.fhir.interceptor.api.IInterceptorBroadcaster;
+import ca.uhn.fhir.interceptor.api.Pointcut;
 import ca.uhn.fhir.jpa.dao.data.IBatch2JobInstanceRepository;
 import ca.uhn.fhir.jpa.dao.data.IBatch2WorkChunkRepository;
 import ca.uhn.fhir.jpa.dao.tx.IHapiTransactionService;
 import ca.uhn.fhir.jpa.entity.Batch2JobInstanceEntity;
 import ca.uhn.fhir.jpa.entity.Batch2WorkChunkEntity;
 import ca.uhn.fhir.model.api.PagingIterator;
+import ca.uhn.fhir.rest.api.server.RequestDetails;
+import ca.uhn.fhir.rest.api.server.SystemRequestDetails;
 import ca.uhn.fhir.util.Batch2JobDefinitionConstants;
 import ca.uhn.fhir.util.Logs;
 import com.fasterxml.jackson.core.JsonParser;
@@ -82,6 +87,7 @@ public class JpaJobPersistenceImpl implements IJobPersistence {
 	private final IBatch2WorkChunkRepository myWorkChunkRepository;
 	private final EntityManager myEntityManager;
 	private final IHapiTransactionService myTransactionService;
+	private final IInterceptorBroadcaster myInterceptorBroadcaster;
 
 	/**
 	 * Constructor
@@ -90,13 +96,15 @@ public class JpaJobPersistenceImpl implements IJobPersistence {
 			IBatch2JobInstanceRepository theJobInstanceRepository,
 			IBatch2WorkChunkRepository theWorkChunkRepository,
 			IHapiTransactionService theTransactionService,
-			EntityManager theEntityManager) {
+			EntityManager theEntityManager,
+			IInterceptorBroadcaster theInterceptorBroadcaster) {
 		Validate.notNull(theJobInstanceRepository);
 		Validate.notNull(theWorkChunkRepository);
 		myJobInstanceRepository = theJobInstanceRepository;
 		myWorkChunkRepository = theWorkChunkRepository;
 		myTransactionService = theTransactionService;
 		myEntityManager = theEntityManager;
+		myInterceptorBroadcaster = theInterceptorBroadcaster;
 	}
 
 	@Override
@@ -143,6 +151,8 @@ public class JpaJobPersistenceImpl implements IJobPersistence {
 	public String storeNewInstance(JobInstance theInstance) {
 		Validate.isTrue(isBlank(theInstance.getInstanceId()));
 
+		invokePreStorageBatchHooks(theInstance);
+
 		Batch2JobInstanceEntity entity = new Batch2JobInstanceEntity();
 		entity.setId(UUID.randomUUID().toString());
 		entity.setDefinitionId(theInstance.getJobDefinitionId());
@@ -154,6 +164,8 @@ public class JpaJobPersistenceImpl implements IJobPersistence {
 		entity.setCreateTime(new Date());
 		entity.setStartTime(new Date());
 		entity.setReport(theInstance.getReport());
+		entity.setTriggeringUsername(theInstance.getTriggeringUsername());
+		entity.setTriggeringClientId(theInstance.getTriggeringClientId());
 
 		entity = myJobInstanceRepository.save(entity);
 		return entity.getId();
@@ -520,4 +532,14 @@ public class JpaJobPersistenceImpl implements IJobPersistence {
 			}
 		}
 	}
+
+	private void invokePreStorageBatchHooks(JobInstance theJobInstance) {
+		if (myInterceptorBroadcaster.hasHooks(Pointcut.STORAGE_PRESTORAGE_BATCH_JOB_CREATE)) {
+			HookParams params = new HookParams()
+					.add(JobInstance.class, theJobInstance)
+					.add(RequestDetails.class, new SystemRequestDetails());
+
+			myInterceptorBroadcaster.callHooks(Pointcut.STORAGE_PRESTORAGE_BATCH_JOB_CREATE, params);
+		}
+	}
 }

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/entity/Batch2JobInstanceEntity.java

@@ -55,6 +55,8 @@ public class Batch2JobInstanceEntity implements Serializable {
 	public static final int TIME_REMAINING_LENGTH = 100;
 	public static final int PARAMS_JSON_MAX_LENGTH = 2000;
 	private static final long serialVersionUID = 8187134261799095422L;
+	public static final int INITIATING_USER_NAME_MAX_LENGTH = 200;
+	public static final int INITIATING_CLIENT_ID_MAX_LENGTH = 200;
 
 	@Id
 	@Column(name = "ID", length = JobDefinition.ID_MAX_LENGTH, nullable = false)
@@ -130,6 +132,12 @@ public class Batch2JobInstanceEntity implements Serializable {
 	@Column(name = "WARNING_MSG", length = WARNING_MSG_MAX_LENGTH, nullable = true)
 	private String myWarningMessages;
 
+	@Column(name = "USER_NAME", length = INITIATING_USER_NAME_MAX_LENGTH, nullable = true)
+	private String myTriggeringUsername;
+
+	@Column(name = "CLIENT_ID", length = INITIATING_CLIENT_ID_MAX_LENGTH, nullable = true)
+	private String myTriggeringClientId;
+
 	/**
 	 * Any output from the job can be held in this column
 	 * Even serialized json
@@ -316,6 +324,24 @@ public class Batch2JobInstanceEntity implements Serializable {
 		myWarningMessages = theWarningMessages;
 	}
 
+	public String getTriggeringUsername() {
+		return myTriggeringUsername;
+	}
+
+	public Batch2JobInstanceEntity setTriggeringUsername(String theTriggeringUsername) {
+		myTriggeringUsername = theTriggeringUsername;
+		return this;
+	}
+
+	public String getTriggeringClientId() {
+		return myTriggeringClientId;
+	}
+
+	public Batch2JobInstanceEntity setTriggeringClientId(String theTriggeringClientId) {
+		myTriggeringClientId = theTriggeringClientId;
+		return this;
+	}
+
 	@Override
 	public String toString() {
 		return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)
@@ -338,6 +364,8 @@ public class Batch2JobInstanceEntity implements Serializable {
 				.append("estimatedTimeRemaining", myEstimatedTimeRemaining)
 				.append("report", myReport)
 				.append("warningMessages", myWarningMessages)
+				.append("initiatingUsername", myTriggeringUsername)
+				.append("initiatingclientId", myTriggeringClientId)
 				.toString();
 	}
 

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/migrate/tasks/HapiFhirJpaMigrationTasks.java

@@ -135,6 +135,12 @@ public class HapiFhirJpaMigrationTasks extends BaseMigrationTasks<VersionEnum> {
 
 		// For resolving references that don't supply the type.
 		hfjResource.addIndex("20231027.3", "IDX_RES_FHIR_ID").unique(false).withColumns("FHIR_ID");
+
+		Builder.BuilderWithTableName batch2JobInstanceTable = version.onTable("BT2_JOB_INSTANCE");
+
+		batch2JobInstanceTable.addColumn("20231128.1", "USER_NAME").nullable().type(ColumnTypeEnum.STRING, 200);
+
+		batch2JobInstanceTable.addColumn("20231128.2", "CLIENT_ID").nullable().type(ColumnTypeEnum.STRING, 200);
 	}
 
 	protected void init680() {

hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/batch2/JobInstanceUtilTest.java

@@ -24,7 +24,6 @@ class JobInstanceUtilTest {
 
 		assertTrue(EqualsBuilder.reflectionEquals(instance, instanceCopyBack));
 
-
 	}
 
 }

hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/batch2/JobInstanceRepositoryTest.java

@@ -18,6 +18,7 @@ import java.util.List;
 import java.util.Set;
 
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 
 public class JobInstanceRepositoryTest extends BaseJpaR4Test {
@@ -30,6 +31,9 @@ public class JobInstanceRepositoryTest extends BaseJpaR4Test {
 	private static final String JOB_DEFINITION_ID = "my-job-def-id";
 	private static final String INSTANCE_ID = "abc-123";
 
+	private static final String TRIGGERING_USER_NAME = "triggeringUser";
+	private static final String TRIGGERING_CLIENT_ID = "clientId";
+
 	@Test
 	public void testSearchByJobParamsAndStatuses_SingleStatus() {
 		Set<StatusEnum> statuses = Set.of(StatusEnum.IN_PROGRESS);
@@ -69,6 +73,16 @@ public class JobInstanceRepositoryTest extends BaseJpaR4Test {
 		assertThat(jobInstances, hasSize(2));
 	}
 
+	@Test
+	public void testPersistInitiatingUsernameAndClientId() {
+		Set<StatusEnum> statuses = Set.of(StatusEnum.IN_PROGRESS);
+		List<Batch2JobInstanceEntity> instancesByJobIdParamsAndStatus = runInTransaction(()->myJobInstanceRepository.findInstancesByJobIdParamsAndStatus(JOB_DEFINITION_ID, PARAMS, statuses, PageRequest.of(0, 10)));
+		assertThat(instancesByJobIdParamsAndStatus, hasSize(1));
+		Batch2JobInstanceEntity batch2JobInstanceEntity = instancesByJobIdParamsAndStatus.get(0);
+		assertThat(TRIGGERING_USER_NAME, equalTo(batch2JobInstanceEntity.getTriggeringUsername()));
+		assertThat(TRIGGERING_CLIENT_ID, equalTo(batch2JobInstanceEntity.getTriggeringClientId()));
+	}
+
 	@BeforeEach
 	public void beforeEach() {
 		//Create in-progress job.
@@ -78,6 +92,8 @@ public class JobInstanceRepositoryTest extends BaseJpaR4Test {
 		instance.setCreateTime(new Date());
 		instance.setDefinitionId(JOB_DEFINITION_ID);
 		instance.setParams(PARAMS);
+		instance.setTriggeringUsername(TRIGGERING_USER_NAME);
+		instance.setTriggeringClientId(TRIGGERING_CLIENT_ID);
 		myJobInstanceRepository.save(instance);
 
 		Batch2JobInstanceEntity completedInstance = new Batch2JobInstanceEntity();

hapi-fhir-jpaserver-test-r4/src/test/java/ca/uhn/fhir/jpa/batch2/JpaJobPersistenceImplTest.java

@@ -11,6 +11,8 @@ import ca.uhn.fhir.batch2.model.WorkChunkCreateEvent;
 import ca.uhn.fhir.batch2.model.WorkChunkErrorEvent;
 import ca.uhn.fhir.batch2.model.WorkChunkStatusEnum;
 import ca.uhn.fhir.batch2.models.JobInstanceFetchRequest;
+import ca.uhn.fhir.interceptor.api.IAnonymousInterceptor;
+import ca.uhn.fhir.interceptor.api.Pointcut;
 import ca.uhn.fhir.jpa.dao.data.IBatch2JobInstanceRepository;
 import ca.uhn.fhir.jpa.dao.data.IBatch2WorkChunkRepository;
 import ca.uhn.fhir.jpa.entity.Batch2JobInstanceEntity;
@@ -633,6 +635,30 @@ public class JpaJobPersistenceImplTest extends BaseJpaR4Test {
 		}
 	}
 
+	@Test
+	public void testPrestorageInterceptor_whenModifyingJobInstance_modifiedJobInstanceIsPersisted(){
+		String expectedTriggeringUserName = "bobTheUncle";
+
+		IAnonymousInterceptor prestorageBatchJobCreateInterceptor = (pointcut, params) -> {
+			JobInstance jobInstance = params.get(JobInstance.class);
+			jobInstance.setTriggeringUsername(expectedTriggeringUserName);
+		};
+
+		try{
+			myInterceptorRegistry.registerAnonymousInterceptor(Pointcut.STORAGE_PRESTORAGE_BATCH_JOB_CREATE, prestorageBatchJobCreateInterceptor);
+			JobInstance instance = createInstance();
+			String instanceId = mySvc.storeNewInstance(instance);
+
+			JobInstance foundInstance = mySvc.fetchInstance(instanceId).orElseThrow(IllegalStateException::new);
+
+			assertEquals(expectedTriggeringUserName, foundInstance.getTriggeringUsername());
+
+		} finally {
+			myInterceptorRegistry.unregisterInterceptor(prestorageBatchJobCreateInterceptor);
+		}
+
+	}
+
 	private WorkChunk freshFetchWorkChunk(String chunkId) {
 		return runInTransaction(() ->
 			myWorkChunkRepository.findById(chunkId)

hapi-fhir-storage-batch2/src/main/java/ca/uhn/fhir/batch2/model/JobInstance.java

@@ -116,6 +116,12 @@ public class JobInstance implements IModelJson, IJobInstance {
 	@JsonProperty(value = "warningMessages", access = JsonProperty.Access.READ_ONLY)
 	private String myWarningMessages;
 
+	@JsonProperty(value = "triggeringUsername", access = JsonProperty.Access.READ_ONLY)
+	private String myTriggeringUsername;
+
+	@JsonProperty(value = "triggeringClientId", access = JsonProperty.Access.READ_ONLY)
+	private String myTriggeringClientId;
+
 	/**
 	 * Constructor
 	 */
@@ -149,6 +155,8 @@ public class JobInstance implements IModelJson, IJobInstance {
 		setCurrentGatedStepId(theJobInstance.getCurrentGatedStepId());
 		setReport(theJobInstance.getReport());
 		setWarningMessages(theJobInstance.getWarningMessages());
+		setTriggeringUsername(theJobInstance.getTriggeringUsername());
+		setTriggeringClientId(theJobInstance.getTriggeringClientId());
 	}
 
 	public String getJobDefinitionId() {
@@ -375,6 +383,24 @@ public class JobInstance implements IModelJson, IJobInstance {
 		myReport = theReport;
 	}
 
+	public String getTriggeringUsername() {
+		return myTriggeringUsername;
+	}
+
+	public JobInstance setTriggeringUsername(String theTriggeringUsername) {
+		myTriggeringUsername = theTriggeringUsername;
+		return this;
+	}
+
+	public String getTriggeringClientId() {
+		return myTriggeringClientId;
+	}
+
+	public JobInstance setTriggeringClientId(String theTriggeringClientId) {
+		myTriggeringClientId = theTriggeringClientId;
+		return this;
+	}
+
 	@Override
 	public String toString() {
 		return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)
@@ -396,6 +422,8 @@ public class JobInstance implements IModelJson, IJobInstance {
 				.append("estimatedTimeRemaining", myEstimatedTimeRemaining)
 				.append("report", myReport)
 				.append("warningMessages", myWarningMessages)
+				.append("triggeringUsername", myTriggeringUsername)
+				.append("triggeringClientId", myTriggeringClientId)
 				.toString();
 	}