Implement $mdm-link-history on JPA (#4648)

* First commit:  prototyping code.

* Push down hard-coding of MdmLink revision results into MdmLinkQuerySvcImplSvc.

* More fixes and push hard-coded MdmLink revisions to DAO.  Add unit test for converter.  Add largely disabled new Spring config to load the AuditReader.  Add new interface method to Dao interface.

* Add config key for enabling envers.  Add generated javadoc to EnversAuditConfig.  Add properties for both the new config and the new mdm history REST API.

* First commit post-merge with new logic to retrieve audited MdmLinks and convert them to JSON.

* Change revision timestamp long to a Date.

* Add a separate inheritance hierarchy for BasePartitionable classes that use Hibernate Envers.  Ensure MdmLink is part of this new hierarchy.  Fix HapiFhirJpaMigrationTasks to properly migrate the modification of HFJ_REVINFO from long to timestamp and to add partition_id and partition_date to the MdmLink audit table.

* Deprecate IMdmLinkDao.findHistory() and mark for removal as well as related methods.

* Add changelog.   Handle empty query results.  Remove hard-coded disabling of envers.  Clean up JPA code.  Clean up unit tests and make them pass.

* Add new hapi-fhir system property to disable envers but leave it enabled by default.  Fix nasty validation bug in IdHelperService.  Tweak MDM dao to throw another Exception and code if envers is disabled.

* Fix error code messages.  More cleanup.

* Another error code fix.

* Add documentation for new feature.

* Fix unit test.  Fix migration tasks to drop and add column instead of modifying it because of a postgres error.

* Cleanup TODOs, delete dead code, tweak unit tests, move/rename classes.

* Default implementation of new history DAO method to avoid need for bump.

* Non-dupe Msg code.

* Since 6.5.7.

* Remove misleading comment.

* Set disabled to TRUE, not FALSE.

* First round of code review fixes.

* Apply documentation suggestions from code review

Co-authored-by: michaelabuckley <michaelabuckley@gmail.com>

* Code review fix:  Split out $mdm-link-history into a separate provider that's enabled by configuration.

* Change configuration strategy from system properties to JpaStorageSettings and JpaStorageSettingsConfigurer.   Hook into these new settings from HibernatePropertiesProvider.  Update more documentation.

* Fix unit test failure in SearchQueryBuilder.

* Apply suggested Javadoc and documentation changes from code review

Co-authored-by: michaelabuckley <michaelabuckley@gmail.com>

* Change semantics for this feature from disabled to enabled.

* Fix failing unit test.

* Fix conditional logic for reversal of envers disabled/enabled.

---------

Co-authored-by: michaelabuckley <michaelabuckley@gmail.com>

hapi-fhir-cli/hapi-fhir-cli-jpaserver/src/main/java/ca/uhn/fhir/jpa/demo/CommonConfig.java

@@ -22,6 +22,7 @@ package ca.uhn.fhir.jpa.demo;
 import ca.uhn.fhir.jpa.api.config.JpaStorageSettings;
 import ca.uhn.fhir.jpa.model.config.PartitionSettings;
 import ca.uhn.fhir.jpa.model.dialect.HapiFhirH2Dialect;
+import ca.uhn.fhir.rest.api.Constants;
 import org.apache.commons.dbcp2.BasicDataSource;
 import org.hibernate.search.mapper.orm.cfg.HibernateOrmMapperSettings;
 import org.springframework.context.annotation.Bean;
@@ -83,6 +84,7 @@ public class CommonConfig {
 		extraProperties.put("hibernate.cache.use_minimal_puts", "false");
 		extraProperties.put("hibernate.search.backend.type", "lucene");
 		extraProperties.put(HibernateOrmMapperSettings.ENABLED, "false");
+		extraProperties.put(Constants.HIBERNATE_INTEGRATION_ENVERS_ENABLED, storageSettings().isNonResourceDbHistoryEnabled());
 
 		return extraProperties;
 	}

hapi-fhir-docs/src/main/resources/ca/uhn/hapi/fhir/changelog/6_6_0/4641-introduce-operation-mdm-link-history.yaml

@@ -0,0 +1,4 @@
+---
+type: add
+issue: 4651
+title: "Introduce configuration to enable the feature that stores the history of MdmLinks.  Introduce the $mdm-link-history operation."

hapi-fhir-docs/src/main/resources/ca/uhn/hapi/fhir/docs/server_jpa/configuration.md

@@ -170,3 +170,12 @@ Delete with expunge submits a job to delete and expunge the requested resources.
 ?_expunge=true syntax is used to trigger the delete expunge, then the batch size will be determined by the value
 of [Expunge Batch Size](/apidocs/hapi-fhir-storage/ca/uhn/fhir/jpa/api/config/JpaStorageSettings.html#getExpungeBatchSize())
 property.
+
+# Disabling Non Resource DB History
+
+This setting controls whether MdmLink and any other non-resource (ex: Patient is a FHIR resource, MdmLink is not) DB history is enabled.  Presently, this only affects the history for MDM links, but the functionality may be extended to other domains.
+
+Clients may want to disable this setting for performance reasons as it populates a new set of database tables when enabled.
+
+Setting this property explicitly to false disables the feature:  [Non Resource DB History](/apidocs/hapi-fhir-storage/ca/uhn/fhir/jpa/api/config/JpaStorageSettings.html#isNonResourceDbHistoryEnabled())
+

hapi-fhir-docs/src/main/resources/ca/uhn/hapi/fhir/docs/server_jpa/performance.md

@@ -60,3 +60,11 @@ X-Upsert-Extistence-Check: disabled
 This should improve write performance, so this header can be useful when large amounts of data will be created using client assigned IDs in a controlled fashion.
 
 If this setting is used and a resource already exists with a given client-assigned ID, a database constraint error will prevent any duplicate records from being created, and the operation will fail.
+
+# Disabling Non Resource DB History
+
+This setting controls whether non-resource (ex: Patient is a resource, MdmLink is not) DB history is enabled.  Presently, this only affects the history for MDM links, but the functionality may be extended to other domains.
+
+Clients may want to disable this setting for performance reasons as it populates a new set of database tables when enabled.
+
+Setting this property explicitly to false disables the feature:  [Non Resource DB History](/apidocs/hapi-fhir-storage/ca/uhn/fhir/jpa/api/config/JpaStorageSettings.html#isNonResourceDbHistoryEnabled())

hapi-fhir-docs/src/main/resources/ca/uhn/hapi/fhir/docs/server_jpa_mdm/mdm_operations.md

@@ -210,6 +210,162 @@ This operation returns a `Parameters` resource that looks like the following:
 }
 ```
 
+## Link History
+
+Use the `$mdm-link-history` operation to request a list of historical entries for a given set of `goldenResourceId`s or `sourceResourceId`s.  Either parameter is optional but **at least one** must be provided.
+
+MDM link history is made possible by a back-end configuration that enables saving the historical entries to a new audit table in the database.  This feature is enabled by default.  Some clients may wish to leave this feature disabled in order to save disk space.
+
+Setting this property explicitly to false disables the feature:  [Non Resource DB History](/apidocs/hapi-fhir-storage/ca/uhn/fhir/jpa/api/config/JpaStorageSettings.html#isNonResourceDbHistoryEnabled())
+
+This operation takes the following parameters:
+
+<table class="table table-striped table-condensed">
+    <thead>
+        <tr>
+            <th>Name</th>
+            <th>Type</th>
+            <th>Cardinality</th>
+            <th>Description</th>
+        </tr>
+    </thead>
+    <tbody>
+        <tr>
+            <td>goldenResourceId</td>
+            <td>String</td>
+            <td>0..*</td>
+            <td>
+                The id of the Golden Resource (e.g. Golden Patient Resource).
+            </td>
+        </tr>
+        <tr>
+            <td>resourceId</td>
+            <td>String</td>
+            <td>0..*</td>
+            <td>
+                The id of the source resource (e.g. Patient resource).
+            </td>
+        </tr>
+    </tbody>
+</table>
+
+
+This operation returns a `Parameters` resource that looks like the following, in the example case where an MdmLink was updated from MATCH to NO_MATCH, with the MDM revisions sorted in *descending* order:
+
+If there are any duplication between results returned by a combination of golden resource IDs and source IDs, they will be included only once.  So, for example, if there is one historical MDM link for golden resource 123 and source resource 456, and both of these identifiers are in the query, only a single historical entry will be returned.
+
+### Example Use
+
+```url
+http://example.com/$mdm-link-history?goldenResourceId=1553&sourceResourceId=1552
+```
+
+```json
+{
+    "resourceType": "Parameters",
+    "parameter": [
+        {
+            "name": "historical link",
+            "part": [
+                {
+                    "name": "goldenResourceId",
+                    "valueString": "Patient/1553"
+                },
+                {
+                    "name": "revisionTimestamp",
+                    "valueString": "2023-03-16 15:14:39.17"
+                },
+                {
+                    "name": "sourceResourceId",
+                    "valueString": "Patient/1552"
+                },
+                {
+                    "name": "matchResult",
+                    "valueString": "NO_MATCH"
+                },
+                {
+                    "name": "score",
+                    "valueDecimal": 1
+                },
+                {
+                    "name": "linkSource",
+                    "valueString": "MANUAL"
+                },
+                {
+                    "name": "eidMatch",
+                    "valueBoolean": false
+                },
+                {
+                    "name": "hadToCreateNewResource",
+                    "valueBoolean": true
+                },
+                {
+                    "name": "score",
+                    "valueDecimal": 1
+                },
+                {
+                    "name": "linkCreated",
+                    "valueDecimal": 1678994017461
+                },
+                {
+                    "name": "linkUpdated",
+                    "valueDecimal": 1678994079155
+                }
+            ]
+        },
+        {
+            "name": "historical link",
+            "part": [
+                {
+                    "name": "goldenResourceId",
+                    "valueString": "Patient/1553"
+                },
+                {
+                    "name": "revisionTimestamp",
+                    "valueString": "2023-03-16 15:13:37.469"
+                },
+                {
+                    "name": "sourceResourceId",
+                    "valueString": "Patient/1552"
+                },
+                {
+                    "name": "matchResult",
+                    "valueString": "MATCH"
+                },
+                {
+                    "name": "score",
+                    "valueDecimal": 1
+                },
+                {
+                    "name": "linkSource",
+                    "valueString": "AUTO"
+                },
+                {
+                    "name": "eidMatch",
+                    "valueBoolean": false
+                },
+                {
+                    "name": "hadToCreateNewResource",
+                    "valueBoolean": true
+                },
+                {
+                    "name": "score",
+                    "valueDecimal": 1
+                },
+                {
+                    "name": "linkCreated",
+                    "valueDecimal": 1678994017461
+                },
+                {
+                    "name": "linkUpdated",
+                    "valueDecimal": 1678994017461
+                }
+            ]
+        }
+    ]
+}
+```
+
 ## Query Duplicate Golden Resources
 
 Use the `$mdm-duplicate-golden-resources` operation to request a list of duplicate Golden Resources. 

hapi-fhir-jpa/src/main/java/ca/uhn/fhir/jpa/config/HapiFhirLocalContainerEntityManagerFactoryBean.java

@@ -20,6 +20,7 @@
 package ca.uhn.fhir.jpa.config;
 
 import ca.uhn.fhir.rest.api.Constants;
+import ca.uhn.fhir.system.HapiSystemProperties;
 import com.google.common.base.Strings;
 import org.hibernate.cfg.AvailableSettings;
 import org.hibernate.query.criteria.LiteralHandlingMode;
@@ -50,11 +51,6 @@ public class HapiFhirLocalContainerEntityManagerFactoryBean extends LocalContain
 	public Map<String, Object> getJpaPropertyMap() {
 		Map<String, Object> retVal = super.getJpaPropertyMap();
 
-		// TODO: LD:  expose configuration for this in a future MR
-		if (!retVal.containsKey(Constants.HIBERNATE_INTEGRATION_ENVERS_ENABLED)) {
-			retVal.put(Constants.HIBERNATE_INTEGRATION_ENVERS_ENABLED, false);
-		}
-
 		// SOMEDAY these defaults can be set in the constructor.  setJpaProperties does a merge.
 		if (!retVal.containsKey(AvailableSettings.CRITERIA_LITERAL_HANDLING_MODE)) {
 			retVal.put(AvailableSettings.CRITERIA_LITERAL_HANDLING_MODE, LiteralHandlingMode.BIND);

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/config/EnversAuditConfig.java

@@ -0,0 +1,43 @@
+package ca.uhn.fhir.jpa.config;
+
+/*-
+ * #%L
+ * HAPI FHIR JPA Server
+ * %%
+ * Copyright (C) 2014 - 2023 Smile CDR, Inc.
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import org.hibernate.envers.AuditReader;
+import org.hibernate.envers.AuditReaderFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import javax.persistence.EntityManagerFactory;
+
+@Configuration
+public class EnversAuditConfig {
+	private final EntityManagerFactory myEntityManagerFactory;
+
+	public EnversAuditConfig(EntityManagerFactory entityManagerFactory) {
+		this.myEntityManagerFactory = entityManagerFactory;
+	}
+
+	@Bean
+	AuditReader auditReader() {
+		return AuditReaderFactory.get(myEntityManagerFactory.createEntityManager());
+	}
+
+}

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/config/HibernatePropertiesProvider.java

@@ -19,6 +19,8 @@
  */
 package ca.uhn.fhir.jpa.config;
 
+import ca.uhn.fhir.jpa.api.config.JpaStorageSettings;
+import ca.uhn.fhir.rest.api.Constants;
 import ca.uhn.fhir.util.ReflectionUtil;
 import com.google.common.annotations.VisibleForTesting;
 import org.apache.commons.lang3.StringUtils;
@@ -29,6 +31,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean;
 
 import javax.sql.DataSource;
+import java.util.Map;
 
 public class HibernatePropertiesProvider {
 
@@ -37,6 +40,9 @@ public class HibernatePropertiesProvider {
 	private Dialect myDialect;
 	private String myHibernateSearchBackend;
 
+	@Autowired
+	private JpaStorageSettings myStorageSettings;
+
 	@VisibleForTesting
 	public void setDialectForUnitTest(Dialect theDialect) {
 		myDialect = theDialect;
@@ -50,6 +56,14 @@ public class HibernatePropertiesProvider {
 			Validate.notNull(dialect, "Unable to create instance of class: %s", dialectClass);
 			myDialect = dialect;
 		}
+
+		if (myEntityManagerFactory != null) {
+			final Map<String, Object> jpaPropertyMap = myEntityManagerFactory.getJpaPropertyMap();
+			if (! jpaPropertyMap.containsKey(Constants.HIBERNATE_INTEGRATION_ENVERS_ENABLED)) {
+				jpaPropertyMap.put(Constants.HIBERNATE_INTEGRATION_ENVERS_ENABLED, myStorageSettings.isNonResourceDbHistoryEnabled());
+			}
+		}
+
 		return dialect;
 	}
 

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/config/JpaConfig.java

@@ -209,7 +209,8 @@ import java.util.Date;
 	Batch2SupportConfig.class,
 	JpaBulkExportConfig.class,
 	SearchConfig.class,
-	PackageLoaderConfig.class
+	PackageLoaderConfig.class,
+	EnversAuditConfig.class
 })
 public class JpaConfig {
 	public static final String JPA_VALIDATION_SUPPORT_CHAIN = "myJpaValidationSupportChain";

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/index/IdHelperService.java

@@ -711,6 +711,9 @@ public class IdHelperService implements IIdHelperService<JpaPid> {
 	public JpaPid getPidOrThrowException(@Nonnull RequestPartitionId theRequestPartitionId, IIdType theId) {
 		List<IIdType> ids = Collections.singletonList(theId);
 		List<JpaPid> resourcePersistentIds = resolveResourcePersistentIdsWithCache(theRequestPartitionId, ids);
+		if (resourcePersistentIds.isEmpty()) {
+			throw new InvalidRequestException(Msg.code(2295) + "Invalid ID was provided: [" + theId.getIdPart() + "]");
+		}
 		return resourcePersistentIds.get(0);
 	}
 

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/mdm/MdmLinkDaoJpaImpl.java

@@ -23,20 +23,28 @@ import ca.uhn.fhir.i18n.Msg;
 import ca.uhn.fhir.interceptor.model.RequestPartitionId;
 import ca.uhn.fhir.jpa.api.svc.IIdHelperService;
 import ca.uhn.fhir.jpa.dao.data.IMdmLinkJpaRepository;
+import ca.uhn.fhir.jpa.entity.HapiFhirEnversRevision;
 import ca.uhn.fhir.jpa.entity.MdmLink;
 import ca.uhn.fhir.jpa.model.dao.JpaPid;
+import ca.uhn.fhir.jpa.model.entity.EnversRevision;
 import ca.uhn.fhir.mdm.api.IMdmLink;
+import ca.uhn.fhir.mdm.api.MdmHistorySearchParameters;
 import ca.uhn.fhir.mdm.api.MdmLinkSourceEnum;
+import ca.uhn.fhir.mdm.api.MdmLinkWithRevision;
 import ca.uhn.fhir.mdm.api.MdmMatchResultEnum;
 import ca.uhn.fhir.mdm.api.MdmQuerySearchParameters;
 import ca.uhn.fhir.mdm.api.paging.MdmPageRequest;
 import ca.uhn.fhir.mdm.dao.IMdmLinkDao;
 import ca.uhn.fhir.mdm.model.MdmPidTuple;
-import ca.uhn.fhir.rest.api.Constants;
 import ca.uhn.fhir.rest.api.SortOrderEnum;
 import ca.uhn.fhir.rest.server.exceptions.UnprocessableEntityException;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.collections4.ListUtils;
+import org.apache.commons.lang3.Validate;
+import org.hibernate.envers.AuditReader;
+import org.hibernate.envers.RevisionType;
+import org.hibernate.envers.query.AuditEntity;
+import org.hibernate.envers.query.AuditQueryCreator;
 import org.hl7.fhir.instance.model.api.IIdType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -48,6 +56,7 @@ import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.history.Revisions;
 
+import javax.annotation.Nonnull;
 import javax.persistence.EntityManager;
 import javax.persistence.TypedQuery;
 import javax.persistence.criteria.CriteriaBuilder;
@@ -82,6 +91,8 @@ public class MdmLinkDaoJpaImpl implements IMdmLinkDao<JpaPid, MdmLink> {
 	protected EntityManager myEntityManager;
 	@Autowired
 	private IIdHelperService<JpaPid> myIdHelperService;
+	@Autowired
+	private AuditReader myAuditReader;
 
 	@Override
 	public int deleteWithAnyReferenceToPid(JpaPid thePid) {
@@ -277,7 +288,6 @@ public class MdmLinkDaoJpaImpl implements IMdmLinkDao<JpaPid, MdmLink> {
 		return andPredicates;
 	}
 
-
 	private List<Order> getOrderList(MdmQuerySearchParameters theParams, CriteriaBuilder criteriaBuilder, Root<MdmLink> from) {
 		if (CollectionUtils.isEmpty(theParams.getSort())) {
 			return Collections.emptyList();
@@ -306,13 +316,61 @@ public class MdmLinkDaoJpaImpl implements IMdmLinkDao<JpaPid, MdmLink> {
 		}
 	}
 
+	// TODO: LD:  delete for good on the next bump
 	@Override
+	@Deprecated(since = "6.5.6", forRemoval = true)
 	public Revisions<Long, MdmLink> findHistory(JpaPid theMdmLinkPid) {
-		// TODO:  LD:  future MR for MdmdLink History return some other object than Revisions, like a Map of List, Pageable, etc?
 		final Revisions<Long, MdmLink> revisions = myMdmLinkDao.findRevisions(theMdmLinkPid.getId());
 
 		revisions.forEach(revision -> ourLog.debug("MdmLink revision: {}", revision));
 
 		return revisions;
 	}
+
+	@Override
+	public List<MdmLinkWithRevision<MdmLink>> getHistoryForIds(MdmHistorySearchParameters theMdmHistorySearchParameters) {
+		final AuditQueryCreator auditQueryCreator = myAuditReader.createQuery();
+
+		try {
+			@SuppressWarnings("unchecked")
+			final List<Object[]> mdmLinksWithRevisions = auditQueryCreator.forRevisionsOfEntity(MdmLink.class, false, false)
+				.add(AuditEntity.or(AuditEntity.property(GOLDEN_RESOURCE_PID_NAME).in(convertToLongIds(theMdmHistorySearchParameters.getGoldenResourceIds())),
+					AuditEntity.property(SOURCE_PID_NAME).in(convertToLongIds(theMdmHistorySearchParameters.getSourceIds()))))
+				.addOrder(AuditEntity.property(GOLDEN_RESOURCE_PID_NAME).asc())
+				.addOrder(AuditEntity.property(SOURCE_PID_NAME).asc())
+				.addOrder(AuditEntity.revisionNumber().desc())
+				.getResultList();
+
+			return mdmLinksWithRevisions.stream()
+				.map(this::buildRevisionFromObjectArray)
+				.collect(Collectors.toUnmodifiableList());
+		} catch (IllegalStateException exception) {
+			ourLog.error("got an Exception when trying to invoke Envers:", exception);
+			throw new IllegalStateException(Msg.code(2291) + "Hibernate envers AuditReader is returning Service is not yet initialized but front-end validation has not caught the error that envers is disabled");
+		}
+	}
+
+	@Nonnull
+	private List<Long> convertToLongIds(List<IIdType> theMdmHistorySearchParameters) {
+		return theMdmHistorySearchParameters.stream()
+			.map(id -> myIdHelperService.getPidOrThrowException(RequestPartitionId.allPartitions(), id))
+			.map(JpaPid::getId)
+			.collect(Collectors.toUnmodifiableList());
+	}
+
+	@SuppressWarnings("unchecked")
+	private MdmLinkWithRevision<MdmLink> buildRevisionFromObjectArray(Object[] theArray) {
+		final Object mdmLinkUncast = theArray[0];
+		final Object revisionUncast = theArray[1];
+		final Object revisionTypeUncast = theArray[2];
+
+		Validate.isInstanceOf(MdmLink.class, mdmLinkUncast);
+		Validate.isInstanceOf(HapiFhirEnversRevision.class, revisionUncast);
+		Validate.isInstanceOf(RevisionType.class, revisionTypeUncast);
+
+		final HapiFhirEnversRevision revision = (HapiFhirEnversRevision) revisionUncast;
+
+		return new MdmLinkWithRevision<>((MdmLink) mdmLinkUncast,
+			new EnversRevision((RevisionType)revisionTypeUncast, revision.getRev(), revision.getRevtstmp()));
+	}
 }

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

@@ -32,6 +32,7 @@ import javax.persistence.Id;
 import javax.persistence.SequenceGenerator;
 import javax.persistence.Table;
 import java.io.Serializable;
+import java.util.Date;
 
 /**
  * This class exists strictly to override the default names used to generate Hibernate Envers revision table.
@@ -62,7 +63,7 @@ public class HapiFhirEnversRevision implements Serializable {
 
 	@RevisionTimestamp
 	@Column(name = "REVTSTMP")
-	private long myRevtstmp;
+	private Date myRevtstmp;
 
 	public long getRev() {
 		return myRev;
@@ -72,11 +73,11 @@ public class HapiFhirEnversRevision implements Serializable {
 		myRev = theRev;
 	}
 
-	public long getRevtstmp() {
+	public Date getRevtstmp() {
 		return myRevtstmp;
 	}
 
-	public void setRevtstmp(long theRevtstmp) {
+	public void setRevtstmp(Date theRevtstmp) {
 		myRevtstmp = theRevtstmp;
 	}
 

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

@@ -20,7 +20,7 @@
 package ca.uhn.fhir.jpa.entity;
 
 import ca.uhn.fhir.jpa.model.dao.JpaPid;
-import ca.uhn.fhir.jpa.model.entity.BasePartitionable;
+import ca.uhn.fhir.jpa.model.entity.AuditableBasePartitionable;
 import ca.uhn.fhir.jpa.model.entity.ResourceTable;
 import ca.uhn.fhir.mdm.api.IMdmLink;
 import ca.uhn.fhir.mdm.api.MdmLinkSourceEnum;
@@ -62,7 +62,7 @@ import java.util.Date;
 @Audited
 // This is the table name generated by default by envers, but we set it explicitly for clarity
 @AuditTable("MPI_LINK_AUD")
-public class MdmLink extends BasePartitionable implements IMdmLink<JpaPid> {
+public class MdmLink extends AuditableBasePartitionable implements IMdmLink<JpaPid> {
 	public static final int VERSION_LENGTH = 16;
 	private static final int MATCH_RESULT_LENGTH = 16;
 	private static final int LINK_SOURCE_LENGTH = 16;

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

@@ -137,13 +137,15 @@ public class HapiFhirJpaMigrationTasks extends BaseMigrationTasks<VersionEnum> {
 		final String revColumnName = "REV";
 		final String enversRevisionTable = "HFJ_REVINFO";
 		final String enversMpiLinkAuditTable = "MPI_LINK_AUD";
+		final String revTstmpColumnName = "REVTSTMP";
 
+		{
 			version.addIdGenerator("20230306.1", "SEQ_HFJ_REVINFO");
 
 			final Builder.BuilderAddTableByColumns enversRevInfo = version.addTableByColumns("20230306.2", enversRevisionTable, revColumnName);
 
 			enversRevInfo.addColumn(revColumnName).nonNullable().type(ColumnTypeEnum.LONG);
-		enversRevInfo.addColumn("REVTSTMP").nullable().type(ColumnTypeEnum.LONG);
+			enversRevInfo.addColumn(revTstmpColumnName).nullable().type(ColumnTypeEnum.LONG);
 
 			final Builder.BuilderAddTableByColumns empiLink = version.addTableByColumns("20230306.6", enversMpiLinkAuditTable, "PID", revColumnName);
 
@@ -151,20 +153,17 @@ public class HapiFhirJpaMigrationTasks extends BaseMigrationTasks<VersionEnum> {
 			empiLink.addColumn("REV").nonNullable().type(ColumnTypeEnum.LONG);
 			empiLink.addColumn("REVTYPE").nullable().type(ColumnTypeEnum.TINYINT);
 			empiLink.addColumn("PERSON_PID").nullable().type(ColumnTypeEnum.LONG);
-		// TODO:  LD: if we want to fully audit partition_id we need to make BasePartitionable  @Auditable, which means adding a bunch of different _AUD migrations here, even if those tables will never be used
-//		empiLink.addColumn("PARTITION_ID").nullable().type(ColumnTypeEnum.INT);
 			empiLink.addColumn("GOLDEN_RESOURCE_PID").nullable().type(ColumnTypeEnum.LONG);
-		// TODO:  LD: figure out a way to set this to 100:  perhaps a migration of MdmLink proper (not _AUD) to alter table to 100?
+			empiLink.addColumn("TARGET_TYPE").nullable().type(ColumnTypeEnum.STRING, 40);
-		empiLink.addColumn( "TARGET_TYPE").nullable().type(ColumnTypeEnum.STRING, 40);
+			empiLink.addColumn("RULE_COUNT").nullable().type(ColumnTypeEnum.LONG);
-		empiLink.addColumn( "RULE_COUNT").nullable().type(ColumnTypeEnum.LONG);
 			empiLink.addColumn("TARGET_PID").nullable().type(ColumnTypeEnum.LONG);
 			empiLink.addColumn("MATCH_RESULT").nullable().type(ColumnTypeEnum.INT);
 			empiLink.addColumn("LINK_SOURCE").nullable().type(ColumnTypeEnum.INT);
 			empiLink.addColumn("CREATED").nullable().type(ColumnTypeEnum.DATE_TIMESTAMP);
 			empiLink.addColumn("UPDATED").nullable().type(ColumnTypeEnum.DATE_TIMESTAMP);
 			empiLink.addColumn("VERSION").nullable().type(ColumnTypeEnum.STRING, 16);
-		empiLink.addColumn("EID_MATCH") .nullable().type(ColumnTypeEnum.BOOLEAN);
+			empiLink.addColumn("EID_MATCH").nullable().type(ColumnTypeEnum.BOOLEAN);
-		empiLink.addColumn("NEW_PERSON") .nullable().type(ColumnTypeEnum.BOOLEAN);
+			empiLink.addColumn("NEW_PERSON").nullable().type(ColumnTypeEnum.BOOLEAN);
 			empiLink.addColumn("VECTOR").nullable().type(ColumnTypeEnum.LONG);
 			empiLink.addColumn("SCORE").nullable().type(ColumnTypeEnum.FLOAT);
 
@@ -174,6 +173,35 @@ public class HapiFhirJpaMigrationTasks extends BaseMigrationTasks<VersionEnum> {
 				.references(enversRevisionTable, revColumnName);
 		}
 
+		{
+			// The pre-release already contains the long version of this column
+			// We do this becausea doing a modifyColumn on Postgres (and possibly other RDBMS's) will fail with a nasty error:
+			// column "revtstmp" cannot be cast automatically to type timestamp without time zone Hint: You might need to specify "USING revtstmp::timestamp without time zone".
+			version
+				.onTable(enversRevisionTable)
+				.dropColumn("20230316.1", revTstmpColumnName);
+
+			version
+				.onTable(enversRevisionTable)
+				.addColumn("20230316.2", revTstmpColumnName)
+				.nullable()
+				.type(ColumnTypeEnum.DATE_TIMESTAMP);
+
+			// New columns from AuditableBasePartitionable
+			version
+				.onTable(enversMpiLinkAuditTable)
+				.addColumn("20230316.3", "PARTITION_ID")
+				.nullable()
+				.type(ColumnTypeEnum.INT);
+
+			version
+				.onTable(enversMpiLinkAuditTable)
+			   .addColumn("20230316.4", "PARTITION_DATE")
+				.nullable()
+				.type(ColumnTypeEnum.DATE_ONLY);
+		}
+	}
+
 	protected void init640() {
 		Builder version = forVersion(VersionEnum.V6_3_0);
 

hapi-fhir-jpaserver-mdm/src/main/java/ca/uhn/fhir/jpa/mdm/dao/MdmLinkDaoSvc.java

@@ -24,7 +24,9 @@ import ca.uhn.fhir.interceptor.model.RequestPartitionId;
 import ca.uhn.fhir.jpa.api.svc.IIdHelperService;
 import ca.uhn.fhir.jpa.model.entity.PartitionablePartitionId;
 import ca.uhn.fhir.mdm.api.IMdmLink;
+import ca.uhn.fhir.mdm.api.MdmHistorySearchParameters;
 import ca.uhn.fhir.mdm.api.MdmLinkSourceEnum;
+import ca.uhn.fhir.mdm.api.MdmLinkWithRevision;
 import ca.uhn.fhir.mdm.api.MdmMatchOutcome;
 import ca.uhn.fhir.mdm.api.MdmMatchResultEnum;
 import ca.uhn.fhir.mdm.api.MdmQuerySearchParameters;
@@ -387,7 +389,14 @@ public class MdmLinkDaoSvc<P extends IResourcePersistentId, M extends IMdmLink<P
 		myMdmLinkDao.deleteLinksWithAnyReferenceToPids(theGoldenResourcePids);
 	}
 
+	// TODO: LD:  delete for good on the next bump
+	@Deprecated(since = "6.5.7", forRemoval = true)
 	public Revisions<Long, M> findMdmLinkHistory(M mdmLink) {
 		return myMdmLinkDao.findHistory(mdmLink.getId());
 	}
+
+	@Transactional
+	public List<MdmLinkWithRevision<M>> findMdmLinkHistory(MdmHistorySearchParameters theMdmHistorySearchParameters) {
+		return myMdmLinkDao.getHistoryForIds(theMdmHistorySearchParameters);
+	}
 }

hapi-fhir-jpaserver-mdm/src/main/java/ca/uhn/fhir/jpa/mdm/svc/IMdmModelConverterSvc.java

@@ -21,6 +21,8 @@ package ca.uhn.fhir.jpa.mdm.svc;
 
 import ca.uhn.fhir.mdm.api.IMdmLink;
 import ca.uhn.fhir.mdm.api.MdmLinkJson;
+import ca.uhn.fhir.mdm.api.MdmLinkWithRevisionJson;
+import ca.uhn.fhir.mdm.api.MdmLinkWithRevision;
 
 /**
  * Contract for decoupling API dependency from the base / JPA modules.
@@ -33,6 +35,13 @@ public interface IMdmModelConverterSvc {
 	 * @param theLink Link to convert
 	 * @return Returns the converted link
 	 */
-	public MdmLinkJson toJson(IMdmLink theLink);
+	MdmLinkJson toJson(IMdmLink theLink);
 
+	/**
+	 * Creates JSON representation of the provided MDM link with revision data
+	 *
+	 * @param theMdmLinkRevision Link with revision data to convert
+	 * @return Returns the converted link
+	 */
+	MdmLinkWithRevisionJson toJson(MdmLinkWithRevision<? extends IMdmLink<?>> theMdmLinkRevision);
 }

hapi-fhir-jpaserver-mdm/src/main/java/ca/uhn/fhir/jpa/mdm/svc/MdmControllerSvcImpl.java

@@ -31,7 +31,9 @@ import ca.uhn.fhir.mdm.api.IMdmControllerSvc;
 import ca.uhn.fhir.mdm.api.IMdmLinkCreateSvc;
 import ca.uhn.fhir.mdm.api.IMdmLinkQuerySvc;
 import ca.uhn.fhir.mdm.api.IMdmLinkUpdaterSvc;
+import ca.uhn.fhir.mdm.api.MdmHistorySearchParameters;
 import ca.uhn.fhir.mdm.api.MdmLinkJson;
+import ca.uhn.fhir.mdm.api.MdmLinkWithRevisionJson;
 import ca.uhn.fhir.mdm.api.MdmMatchResultEnum;
 import ca.uhn.fhir.mdm.api.MdmQuerySearchParameters;
 import ca.uhn.fhir.mdm.api.paging.MdmPageRequest;
@@ -135,6 +137,11 @@ public class MdmControllerSvcImpl implements IMdmControllerSvc {
 		return resultPage;
 	}
 
+	@Override
+	public List<MdmLinkWithRevisionJson> queryLinkHistory(MdmHistorySearchParameters theMdmHistorySearchParameters, RequestDetails theRequestDetails) {
+		return myMdmLinkQuerySvc.queryLinkHistory(theMdmHistorySearchParameters);
+	}
+
 	@Override
 	@Deprecated
 	public Page<MdmLinkJson> queryLinksFromPartitionList(@Nullable String theGoldenResourceId, @Nullable String theSourceResourceId,

hapi-fhir-jpaserver-mdm/src/main/java/ca/uhn/fhir/jpa/mdm/svc/MdmLinkQuerySvcImplSvc.java

@@ -22,8 +22,11 @@ package ca.uhn.fhir.jpa.mdm.svc;
 import ca.uhn.fhir.jpa.mdm.dao.MdmLinkDaoSvc;
 import ca.uhn.fhir.mdm.api.IMdmLink;
 import ca.uhn.fhir.mdm.api.IMdmLinkQuerySvc;
+import ca.uhn.fhir.mdm.api.MdmHistorySearchParameters;
 import ca.uhn.fhir.mdm.api.MdmLinkJson;
+import ca.uhn.fhir.mdm.api.MdmLinkWithRevisionJson;
 import ca.uhn.fhir.mdm.api.MdmLinkSourceEnum;
+import ca.uhn.fhir.mdm.api.MdmLinkWithRevision;
 import ca.uhn.fhir.mdm.api.MdmMatchResultEnum;
 import ca.uhn.fhir.mdm.api.MdmQuerySearchParameters;
 import ca.uhn.fhir.mdm.api.paging.MdmPageRequest;
@@ -36,6 +39,7 @@ import org.springframework.data.domain.Page;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.util.List;
+import java.util.stream.Collectors;
 
 public class MdmLinkQuerySvcImplSvc implements IMdmLinkQuerySvc {
 
@@ -102,4 +106,13 @@ public class MdmLinkQuerySvcImplSvc implements IMdmLinkQuerySvc {
 		Page<? extends IMdmLink> mdmLinkPage = myMdmLinkDaoSvc.executeTypedQuery(mdmQuerySearchParameters);
 		return mdmLinkPage.map(myMdmModelConverterSvc::toJson);
 	}
+
+	@Override
+	public List<MdmLinkWithRevisionJson> queryLinkHistory(MdmHistorySearchParameters theMdmHistorySearchParameters) {
+		final List<MdmLinkWithRevision<? extends IMdmLink<?>>> mdmLinkHistoryFromDao = myMdmLinkDaoSvc.findMdmLinkHistory(theMdmHistorySearchParameters);
+
+		return mdmLinkHistoryFromDao.stream()
+			.map(myMdmModelConverterSvc::toJson)
+			.collect(Collectors.toUnmodifiableList());
+	}
 }

hapi-fhir-jpaserver-mdm/src/main/java/ca/uhn/fhir/jpa/mdm/svc/MdmModelConverterSvcImpl.java

@@ -22,6 +22,8 @@ package ca.uhn.fhir.jpa.mdm.svc;
 import ca.uhn.fhir.jpa.api.svc.IIdHelperService;
 import ca.uhn.fhir.mdm.api.IMdmLink;
 import ca.uhn.fhir.mdm.api.MdmLinkJson;
+import ca.uhn.fhir.mdm.api.MdmLinkWithRevisionJson;
+import ca.uhn.fhir.mdm.api.MdmLinkWithRevision;
 import org.springframework.beans.factory.annotation.Autowired;
 
 public class MdmModelConverterSvcImpl implements IMdmModelConverterSvc {
@@ -49,4 +51,10 @@ public class MdmModelConverterSvcImpl implements IMdmModelConverterSvc {
 		return retVal;
 	}
 
+	@Override
+	public MdmLinkWithRevisionJson toJson(MdmLinkWithRevision<? extends IMdmLink<?>> theMdmLinkRevision) {
+		final MdmLinkJson mdmLinkJson = toJson(theMdmLinkRevision.getMdmLink());
+
+		return new MdmLinkWithRevisionJson(mdmLinkJson, theMdmLinkRevision.getEnversRevision().getRevisionNumber(), theMdmLinkRevision.getEnversRevision().getRevisionTimestamp());
+	}
 }

hapi-fhir-jpaserver-mdm/src/test/java/ca/uhn/fhir/jpa/mdm/BaseMdmR4Test.java

@@ -623,4 +623,34 @@ abstract public class BaseMdmR4Test extends BaseJpaR4Test {
 		retval.setRestOperation(MdmTransactionContext.OperationType.UPDATE_LINK);
 		return retval;
 	}
+
+	protected MdmLink createGoldenPatientAndLinkToSourcePatient(Long thePatientPid, MdmMatchResultEnum theMdmMatchResultEnum) {
+		Patient patient = createPatient();
+
+		MdmLink mdmLink = (MdmLink) myMdmLinkDaoSvc.newMdmLink();
+		mdmLink.setLinkSource(MdmLinkSourceEnum.MANUAL);
+		mdmLink.setMatchResult(theMdmMatchResultEnum);
+		mdmLink.setCreated(new Date());
+		mdmLink.setUpdated(new Date());
+		mdmLink.setGoldenResourcePersistenceId(JpaPid.fromId(thePatientPid));
+		mdmLink.setSourcePersistenceId(runInTransaction(()->myIdHelperService.getPidOrNull(RequestPartitionId.allPartitions(), patient)));
+		return myMdmLinkDao.save(mdmLink);
+	}
+
+	protected MdmLink createGoldenPatientAndLinkToSourcePatient(MdmMatchResultEnum theMdmMatchResultEnum, MdmLinkSourceEnum theMdmLinkSourceEnum, String theVersion, Date theCreateTime, Date theUpdateTime, boolean theLinkCreatedNewResource) {
+		final Patient goldenPatient = createPatient();
+		final Patient sourcePatient = createPatient();
+
+		final MdmLink mdmLink = (MdmLink) myMdmLinkDaoSvc.newMdmLink();
+		mdmLink.setLinkSource(theMdmLinkSourceEnum);
+		mdmLink.setMatchResult(theMdmMatchResultEnum);
+		mdmLink.setCreated(theCreateTime);
+		mdmLink.setUpdated(theUpdateTime);
+		mdmLink.setVersion(theVersion);
+		mdmLink.setGoldenResourcePersistenceId(runInTransaction(()->myIdHelperService.getPidOrNull(RequestPartitionId.allPartitions(), goldenPatient)));
+		mdmLink.setSourcePersistenceId(runInTransaction(()->myIdHelperService.getPidOrNull(RequestPartitionId.allPartitions(), sourcePatient)));
+		mdmLink.setHadToCreateNewGoldenResource(theLinkCreatedNewResource);
+
+		return myMdmLinkDao.save(mdmLink);
+	}
 }

hapi-fhir-jpaserver-mdm/src/test/java/ca/uhn/fhir/jpa/mdm/dao/MdmLinkDaoSvcTest.java

@@ -5,23 +5,30 @@ import ca.uhn.fhir.jpa.entity.MdmLink;
 import ca.uhn.fhir.jpa.mdm.BaseMdmR4Test;
 import ca.uhn.fhir.jpa.model.dao.JpaPid;
 import ca.uhn.fhir.jpa.util.TestUtil;
+import ca.uhn.fhir.jpa.model.entity.EnversRevision;
 import ca.uhn.fhir.mdm.api.IMdmLink;
+import ca.uhn.fhir.mdm.api.MdmHistorySearchParameters;
 import ca.uhn.fhir.mdm.api.MdmLinkSourceEnum;
+import ca.uhn.fhir.mdm.api.MdmLinkWithRevision;
 import ca.uhn.fhir.mdm.api.MdmMatchResultEnum;
 import ca.uhn.fhir.mdm.model.MdmPidTuple;
 import ca.uhn.fhir.mdm.rules.json.MdmRulesJson;
+import org.hibernate.envers.RevisionType;
 import org.hl7.fhir.r4.model.Patient;
 import org.junit.jupiter.api.Test;
-import org.springframework.data.history.Revision;
+import org.slf4j.Logger;
-import org.springframework.data.history.Revisions;
+import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nonnull;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Date;
 import java.util.List;
+import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.in;
@@ -30,10 +37,13 @@ import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class MdmLinkDaoSvcTest extends BaseMdmR4Test {
 
+	private static final Logger ourLog = LoggerFactory.getLogger(MdmLinkDaoSvcTest.class);
+
 	@Test
 	public void testCreate() {
 		MdmLink mdmLink = createResourcesAndBuildTestMDMLink();
@@ -71,12 +81,12 @@ public class MdmLinkDaoSvcTest extends BaseMdmR4Test {
 		//Create 10 linked patients.
 		List<MdmLink> mdmLinks = new ArrayList<>();
 		for (int i = 0; i < 10; i++) {
-			mdmLinks.add(createPatientAndLinkTo(golden.getIdElement().getIdPartAsLong(), MdmMatchResultEnum.MATCH));
+			mdmLinks.add(createGoldenPatientAndLinkToSourcePatient(golden.getIdElement().getIdPartAsLong(), MdmMatchResultEnum.MATCH));
 		}
 
 		//Now lets connect a few as just POSSIBLE_MATCHes and ensure they aren't returned.
 		for (int i = 0 ; i < 5; i++) {
-			createPatientAndLinkTo(golden.getIdElement().getIdPartAsLong(), MdmMatchResultEnum.POSSIBLE_MATCH);
+			createGoldenPatientAndLinkToSourcePatient(golden.getIdElement().getIdPartAsLong(), MdmMatchResultEnum.POSSIBLE_MATCH);
 		}
 
 		List<Long> expectedExpandedPids = mdmLinks.stream().map(MdmLink::getSourcePid).collect(Collectors.toList());
@@ -94,47 +104,113 @@ public class MdmLinkDaoSvcTest extends BaseMdmR4Test {
 	}
 
 	@Test
-	public void testMdmLinkHistoryCreateUpdateDelete() {
+	public void testHistoryForMultipleIdsCrud() {
-		final MdmLink mdmLink = createResourcesAndBuildTestMDMLink();
+		final List<MdmLink> mdmLinksWithLinkedPatients1 = createMdmLinksWithLinkedPatients(MdmMatchResultEnum.MATCH, 3);
-		assertThat(mdmLink.getCreated(), is(nullValue()));
+		final List<MdmLink> mdmLinksWithLinkedPatients2 = createMdmLinksWithLinkedPatients(MdmMatchResultEnum.MATCH, 4);
-		assertThat(mdmLink.getUpdated(), is(nullValue()));
+		final List<MdmLink> mdmLinksWithLinkedPatients3 = createMdmLinksWithLinkedPatients(MdmMatchResultEnum.MATCH, 2);
-		myMdmLinkDaoSvc.save(mdmLink);
+		flipLinksTo(mdmLinksWithLinkedPatients3, MdmMatchResultEnum.NO_MATCH);
-		assertThat(mdmLink.getCreated(), is(notNullValue()));
-		assertThat(mdmLink.getUpdated(), is(notNullValue()));
-		assertTrue(mdmLink.getUpdated().getTime() - mdmLink.getCreated().getTime() < 1000);
 
-		final Revisions<Long, MdmLink> mdmLinkHistoryCreate = myMdmLinkDaoSvc.findMdmLinkHistory(mdmLink);
+		final MdmHistorySearchParameters mdmHistorySearchParameters =
+			new MdmHistorySearchParameters()
+				.setGoldenResourceIds(getIdsFromMdmLinks(MdmLink::getGoldenResourcePersistenceId, mdmLinksWithLinkedPatients1.get(0), mdmLinksWithLinkedPatients3.get(0)))
+				.setSourceIds(getIdsFromMdmLinks(MdmLink::getSourcePersistenceId, mdmLinksWithLinkedPatients1.get(0), mdmLinksWithLinkedPatients2.get(0)));
 
-		assertThat(mdmLinkHistoryCreate.stream().map(revision -> revision.getEntity().getMatchResult()).toList(),
+		final List<MdmLinkWithRevision<MdmLink>> actualMdmLinkRevisions = myMdmLinkDaoSvc.findMdmLinkHistory(mdmHistorySearchParameters);
-			contains(MdmMatchResultEnum.MATCH));
 
-		mdmLink.setMatchResult(MdmMatchResultEnum.NO_MATCH);
+		final JpaPid goldenResourceId1 = mdmLinksWithLinkedPatients1.get(0).getGoldenResourcePersistenceId();
+		final JpaPid goldenResourceId2 = mdmLinksWithLinkedPatients2.get(0).getGoldenResourcePersistenceId();
+		final JpaPid goldenResourceId3 = mdmLinksWithLinkedPatients3.get(0).getGoldenResourcePersistenceId();
 
-		myMdmLinkDaoSvc.save(mdmLink);
+		final JpaPid sourceId1_1 = mdmLinksWithLinkedPatients1.get(0).getSourcePersistenceId();
+		final JpaPid sourceId1_2 = mdmLinksWithLinkedPatients1.get(1).getSourcePersistenceId();
+		final JpaPid sourceId1_3 = mdmLinksWithLinkedPatients1.get(2).getSourcePersistenceId();
 
-		final Revisions<Long, MdmLink> mdmLinkHistoryUpdate = myMdmLinkDaoSvc.findMdmLinkHistory(mdmLink);
+		final JpaPid sourceId2_1 = mdmLinksWithLinkedPatients2.get(0).getSourcePersistenceId();
 
-		assertThat(mdmLinkHistoryUpdate.stream().map(revision -> revision.getEntity().getMatchResult()).toList(),
+		final JpaPid sourceId3_1 = mdmLinksWithLinkedPatients3.get(0).getSourcePersistenceId();
-			contains(MdmMatchResultEnum.MATCH, MdmMatchResultEnum.NO_MATCH));
+		final JpaPid sourceId3_2 = mdmLinksWithLinkedPatients3.get(1).getSourcePersistenceId();
 
-		myMdmLinkDaoSvc.deleteLink(mdmLink);
+		final List<MdmLinkWithRevision<MdmLink>> expectedMdLinkRevisions = List.of(
+			buildMdmLinkWithRevision(1, RevisionType.ADD, MdmMatchResultEnum.MATCH, goldenResourceId1, sourceId1_1),
+			buildMdmLinkWithRevision(2, RevisionType.ADD, MdmMatchResultEnum.MATCH, goldenResourceId1, sourceId1_2),
+			buildMdmLinkWithRevision(3, RevisionType.ADD, MdmMatchResultEnum.MATCH, goldenResourceId1, sourceId1_3),
+			buildMdmLinkWithRevision(4, RevisionType.ADD, MdmMatchResultEnum.MATCH, goldenResourceId2, sourceId2_1),
+			buildMdmLinkWithRevision(10, RevisionType.MOD, MdmMatchResultEnum.NO_MATCH, goldenResourceId3, sourceId3_1),
+			buildMdmLinkWithRevision(8, RevisionType.ADD, MdmMatchResultEnum.MATCH, goldenResourceId3, sourceId3_1),
+			buildMdmLinkWithRevision(11, RevisionType.MOD, MdmMatchResultEnum.NO_MATCH, goldenResourceId3, sourceId3_2),
+			buildMdmLinkWithRevision(9, RevisionType.ADD, MdmMatchResultEnum.MATCH, goldenResourceId3, sourceId3_2)
+		);
 
-		final Revisions<Long, MdmLink> mdmLinkHistoryDelete = myMdmLinkDaoSvc.findMdmLinkHistory(mdmLink);
+		assertMdmRevisionsEqual(expectedMdLinkRevisions, actualMdmLinkRevisions);
-
-		assertThat(mdmLinkHistoryDelete.stream().map(Revision::getEntity).map(MdmLink::getMatchResult).toList(),
-			contains(MdmMatchResultEnum.MATCH, MdmMatchResultEnum.NO_MATCH, null));
 	}
 
-	private MdmLink createPatientAndLinkTo(Long thePatientPid, MdmMatchResultEnum theMdmMatchResultEnum) {
+	@Nonnull
-		Patient patient = createPatient();
+	private static List<String> getIdsFromMdmLinks(Function<MdmLink, JpaPid> getIdFunction, MdmLink... mdmLinks) {
+		return Arrays.stream(mdmLinks)
+			.map(getIdFunction)
+			.map(JpaPid::getId).map(along -> Long.toString(along))
+			.collect(Collectors.toUnmodifiableList());
+	}
+
+	private void assertMdmRevisionsEqual(List<MdmLinkWithRevision<MdmLink>> expectedMdmLinkRevisions, List<MdmLinkWithRevision<MdmLink>> actualMdmLinkRevisions) {
+		assertNotNull(actualMdmLinkRevisions);
+
+		assertEquals(expectedMdmLinkRevisions.size(), actualMdmLinkRevisions.size());
+
+		for (int index = 0; index < expectedMdmLinkRevisions.size(); index++) {
+			final MdmLinkWithRevision<MdmLink> expectedMdmLinkRevision = expectedMdmLinkRevisions.get(index);
+			final MdmLinkWithRevision<MdmLink> actualMdmLinkRevision = actualMdmLinkRevisions.get(index);
+
+			final EnversRevision expectedEnversRevision = expectedMdmLinkRevision.getEnversRevision();
+			final EnversRevision actualEnversRevision = actualMdmLinkRevision.getEnversRevision();
+			final MdmLink expectedMdmLink = expectedMdmLinkRevision.getMdmLink();
+			final MdmLink actualMdmLink = actualMdmLinkRevision.getMdmLink();
+
+			assertEquals(expectedMdmLink.getMatchResult(), actualMdmLink.getMatchResult());
+			assertEquals(expectedMdmLink.getGoldenResourcePersistenceId(), actualMdmLinkRevision.getMdmLink().getGoldenResourcePersistenceId());
+			assertEquals(expectedMdmLink.getSourcePersistenceId(), actualMdmLinkRevision.getMdmLink().getSourcePersistenceId());
+
+			assertEquals(expectedEnversRevision.getRevisionType(), actualEnversRevision.getRevisionType());
+			// TODO:  LD:  when running this unit test on a pipeline, it's impossible to assert a revision number because of all the other MdmLinks
+			// created by other tests.  So for now, simply assert the revision is greater than 0
+			assertTrue(actualEnversRevision.getRevisionNumber() > 0);
+		}
+	}
+
+	private MdmLinkWithRevision<MdmLink> buildMdmLinkWithRevision(long theRevisionNumber, RevisionType theRevisionType, MdmMatchResultEnum theMdmMatchResultEnum, JpaPid theGolderResourceId, JpaPid theSourceId) {
+		final MdmLink mdmLink = new MdmLink();
+
+		mdmLink.setMatchResult(theMdmMatchResultEnum);
+		mdmLink.setGoldenResourcePersistenceId(theGolderResourceId);
+		mdmLink.setSourcePersistenceId(theSourceId);
+
+		final MdmLinkWithRevision<MdmLink> mdmLinkWithRevision = new MdmLinkWithRevision<>(mdmLink, new EnversRevision(theRevisionType, theRevisionNumber, new Date()));
+
+		return mdmLinkWithRevision;
+	}
+
+	private void flipLinksTo(List<MdmLink> theMdmLinksWithLinkedPatients, MdmMatchResultEnum theMdmMatchResultEnum) {
+		theMdmLinksWithLinkedPatients.forEach(mdmLink -> {
+			mdmLink.setMatchResult(theMdmMatchResultEnum);
+			myMdmLinkDaoSvc.save(mdmLink);
+		});
+	}
+
+	private List<MdmLink> createMdmLinksWithLinkedPatients(MdmMatchResultEnum theFirstMdmMatchResultEnum, int numTargetPatients) {
+		final Patient goldenPatient = createPatient();
+
+		return IntStream.range(0, numTargetPatients).mapToObj(myInt -> {
+			final Patient targetPatient = createPatient();
 
 			MdmLink mdmLink = (MdmLink) myMdmLinkDaoSvc.newMdmLink();
 			mdmLink.setLinkSource(MdmLinkSourceEnum.MANUAL);
-		mdmLink.setMatchResult(theMdmMatchResultEnum);
+			mdmLink.setMatchResult(theFirstMdmMatchResultEnum);
 			mdmLink.setCreated(new Date());
 			mdmLink.setUpdated(new Date());
-		mdmLink.setGoldenResourcePersistenceId(JpaPid.fromId(thePatientPid));
+			mdmLink.setGoldenResourcePersistenceId(runInTransaction(() -> myIdHelperService.getPidOrNull(RequestPartitionId.allPartitions(), goldenPatient)));
-		mdmLink.setSourcePersistenceId(runInTransaction(()->myIdHelperService.getPidOrNull(RequestPartitionId.allPartitions(), patient)));
+			mdmLink.setSourcePersistenceId(runInTransaction(() -> myIdHelperService.getPidOrNull(RequestPartitionId.allPartitions(), targetPatient)));
 			return myMdmLinkDao.save(mdmLink);
+
+		}).toList();
 	}
 }

hapi-fhir-jpaserver-mdm/src/test/java/ca/uhn/fhir/jpa/mdm/svc/MdmModelConverterSvcImplTest.java

@@ -0,0 +1,99 @@
+package ca.uhn.fhir.jpa.mdm.svc;
+
+import ca.uhn.fhir.jpa.entity.MdmLink;
+import ca.uhn.fhir.jpa.mdm.BaseMdmR4Test;
+import ca.uhn.fhir.jpa.model.entity.EnversRevision;
+import ca.uhn.fhir.mdm.api.IMdmLink;
+import ca.uhn.fhir.mdm.api.MdmLinkJson;
+import ca.uhn.fhir.mdm.api.MdmLinkWithRevisionJson;
+import ca.uhn.fhir.mdm.api.MdmLinkSourceEnum;
+import ca.uhn.fhir.mdm.api.MdmLinkWithRevision;
+import ca.uhn.fhir.mdm.api.MdmMatchResultEnum;
+import ca.uhn.fhir.rest.api.server.storage.IResourcePersistentId;
+import org.hibernate.envers.RevisionType;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.time.LocalDateTime;
+import java.time.Month;
+import java.time.ZoneId;
+import java.util.Date;
+
+public class MdmModelConverterSvcImplTest extends BaseMdmR4Test {
+	private static final Logger ourLog = LoggerFactory.getLogger(MdmModelConverterSvcImplTest.class);
+
+	@Autowired
+	IMdmModelConverterSvc myMdmModelConverterSvc;
+
+	@Test
+	public void testBasicMdmLinkConversion() {
+		final Date createTime = new Date();
+		final Date updateTime = new Date();
+		final String version = "1";
+		final boolean isLinkCreatedResource = false;
+
+		final MdmLink mdmLink = createGoldenPatientAndLinkToSourcePatient(MdmMatchResultEnum.MATCH, MdmLinkSourceEnum.MANUAL, version, createTime, updateTime, isLinkCreatedResource);
+		myMdmLinkDao.save(mdmLink);
+
+		final MdmLinkJson actualMdmLinkJson = myMdmModelConverterSvc.toJson(mdmLink);
+
+		ourLog.info("actualMdmLinkJson: {}", actualMdmLinkJson);
+
+		assertEquals(getExepctedMdmLinkJson(mdmLink.getGoldenResourcePersistenceId().getId(), mdmLink.getSourcePersistenceId().getId(), MdmMatchResultEnum.MATCH, MdmLinkSourceEnum.MANUAL, version, createTime, updateTime, isLinkCreatedResource), actualMdmLinkJson);
+	}
+
+	@Test
+	public void testBasicMdmLinkRevisionConversion() {
+		final Date createTime = new Date();
+		final Date updateTime = new Date();
+		final Date revisionTimestamp = Date.from(LocalDateTime
+			.of(2023, Month.MARCH, 16, 15, 23, 0)
+				.atZone(ZoneId.systemDefault())
+			.toInstant());
+		final String version = "1";
+		final boolean isLinkCreatedResource = false;
+		final long revisionNumber = 2L;
+
+		final MdmLink mdmLink = createGoldenPatientAndLinkToSourcePatient(MdmMatchResultEnum.MATCH, MdmLinkSourceEnum.MANUAL, version, createTime, updateTime, isLinkCreatedResource);
+
+		final MdmLinkWithRevision<IMdmLink<? extends IResourcePersistentId<?>>> revision = new MdmLinkWithRevision<>(mdmLink, new EnversRevision(RevisionType.ADD, revisionNumber, revisionTimestamp));
+
+		final MdmLinkWithRevisionJson actualMdmLinkWithRevisionJson = myMdmModelConverterSvc.toJson(revision);
+
+		final MdmLinkWithRevisionJson expectedMdmLinkWithRevisionJson =
+			new MdmLinkWithRevisionJson(getExepctedMdmLinkJson(mdmLink.getGoldenResourcePersistenceId().getId(), mdmLink.getSourcePersistenceId().getId(), MdmMatchResultEnum.MATCH, MdmLinkSourceEnum.MANUAL, version, createTime, updateTime, isLinkCreatedResource), revisionNumber, revisionTimestamp);
+
+		assertMdmLinkRevisionsEqual(expectedMdmLinkWithRevisionJson, actualMdmLinkWithRevisionJson);
+	}
+
+	private void assertMdmLinkRevisionsEqual(MdmLinkWithRevisionJson theExpectedMdmLinkWithRevisionJson, MdmLinkWithRevisionJson theActualMdmLinkWithRevisionJson) {
+		final MdmLinkJson expectedMdmLink = theExpectedMdmLinkWithRevisionJson.getMdmLink();
+		final MdmLinkJson actualMdmLink = theActualMdmLinkWithRevisionJson.getMdmLink();
+		assertEquals(expectedMdmLink.getGoldenResourceId(), actualMdmLink.getGoldenResourceId());
+		assertEquals(expectedMdmLink.getSourceId(), actualMdmLink.getSourceId());
+		assertEquals(expectedMdmLink.getMatchResult(), actualMdmLink.getMatchResult());
+		assertEquals(expectedMdmLink.getLinkSource(), actualMdmLink.getLinkSource());
+
+		assertEquals(theExpectedMdmLinkWithRevisionJson.getRevisionNumber(), theActualMdmLinkWithRevisionJson.getRevisionNumber());
+		assertEquals(theExpectedMdmLinkWithRevisionJson.getRevisionTimestamp(), theActualMdmLinkWithRevisionJson.getRevisionTimestamp());
+	}
+
+	private MdmLinkJson getExepctedMdmLinkJson(Long theGoldenPatientId, Long theSourceId, MdmMatchResultEnum theMdmMatchResultEnum, MdmLinkSourceEnum theMdmLinkSourceEnum, String version, Date theCreateTime, Date theUpdateTime, boolean theLinkCreatedNewResource) {
+		final MdmLinkJson mdmLinkJson = new MdmLinkJson();
+
+		mdmLinkJson.setGoldenResourceId("Patient/" + theGoldenPatientId);
+		mdmLinkJson.setSourceId("Patient/" + theSourceId);
+		mdmLinkJson.setMatchResult(theMdmMatchResultEnum);
+		mdmLinkJson.setLinkSource(theMdmLinkSourceEnum);
+		mdmLinkJson.setVersion(version);
+		mdmLinkJson.setCreated(theCreateTime);
+		mdmLinkJson.setUpdated(theUpdateTime);
+		mdmLinkJson.setLinkCreatedNewResource(theLinkCreatedNewResource);
+
+		return mdmLinkJson;
+	}
+}

hapi-fhir-jpaserver-model/src/main/java/ca/uhn/fhir/jpa/model/entity/AuditableBasePartitionable.java

@@ -0,0 +1,61 @@
+package ca.uhn.fhir.jpa.model.entity;
+
+/*-
+ * #%L
+ * HAPI FHIR JPA Model
+ * %%
+ * Copyright (C) 2014 - 2023 Smile CDR, Inc.
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import org.hibernate.envers.Audited;
+import org.hibernate.envers.NotAudited;
+
+import javax.annotation.Nullable;
+import javax.persistence.Column;
+import javax.persistence.Embedded;
+import javax.persistence.MappedSuperclass;
+import java.io.Serializable;
+
+/**
+ * This is a copy of (@link {@link BasePartitionable} used ONLY for entities that are audited by Hibernate Envers.
+ * <p>
+ * The reason for this is Envers will generate _AUD table schema for all auditable tables, even those marked as {@link NotAudited}.
+ * <p>
+ * Should we make more entities envers auditable in the future, they would need to extend this class and not {@link BasePartitionable}.
+ */
+@Audited
+@MappedSuperclass
+public class AuditableBasePartitionable implements Serializable {
+	@Embedded
+	private PartitionablePartitionId myPartitionId;
+
+	/**
+	 * This is here to support queries only, do not set this field directly
+	 */
+	@SuppressWarnings("unused")
+	@Column(name = PartitionablePartitionId.PARTITION_ID, insertable = false, updatable = false, nullable = true)
+	private Integer myPartitionIdValue;
+
+	@Nullable
+	public PartitionablePartitionId getPartitionId() {
+		return myPartitionId;
+	}
+
+	public void setPartitionId(PartitionablePartitionId thePartitionId) {
+		myPartitionId = thePartitionId;
+	}
+
+}

hapi-fhir-jpaserver-model/src/main/java/ca/uhn/fhir/jpa/model/entity/BasePartitionable.java

@@ -25,6 +25,11 @@ import javax.persistence.Embedded;
 import javax.persistence.MappedSuperclass;
 import java.io.Serializable;
 
+/**
+ * This is the base class for entities with partitioning that does NOT include Hibernate Envers logging.
+ * <p>
+ * If your entity needs Envers auditing, please have it extend {@link AuditableBasePartitionable} instead.
+ */
 @MappedSuperclass
 public abstract class BasePartitionable implements Serializable {
 

hapi-fhir-jpaserver-model/src/main/java/ca/uhn/fhir/jpa/model/entity/EnversRevision.java

@@ -0,0 +1,74 @@
+package ca.uhn.fhir.jpa.model.entity;
+
+/*-
+ * #%L
+ * HAPI FHIR JPA Model
+ * %%
+ * Copyright (C) 2014 - 2023 Smile CDR, Inc.
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.hibernate.envers.RevisionType;
+
+import java.util.Date;
+import java.util.Objects;
+
+public class EnversRevision {
+	private final RevisionType myRevisionType;
+	private final long myRevisionNumber;
+	private final Date myRevisionTimestamp;
+
+	public EnversRevision(RevisionType theRevisionType, long theRevisionNumber, Date theRevisionTimestamp) {
+		myRevisionType = theRevisionType;
+		myRevisionNumber = theRevisionNumber;
+		myRevisionTimestamp = theRevisionTimestamp;
+	}
+
+	public RevisionType getRevisionType() {
+		return myRevisionType;
+	}
+
+	public long getRevisionNumber() {
+		return myRevisionNumber;
+	}
+
+	public Date getRevisionTimestamp() {
+		return myRevisionTimestamp;
+	}
+
+	@Override
+	public boolean equals(Object theO) {
+
+		if (this == theO) return true;
+		if (theO == null || getClass() != theO.getClass()) return false;
+		final EnversRevision that = (EnversRevision) theO;
+		return myRevisionNumber == that.myRevisionNumber && myRevisionTimestamp == that.myRevisionTimestamp && myRevisionType == that.myRevisionType;
+	}
+
+	@Override
+	public int hashCode() {
+		return Objects.hash(myRevisionType, myRevisionNumber, myRevisionTimestamp);
+	}
+
+	@Override
+	public String toString() {
+		return new ToStringBuilder(this)
+			.append("myRevisionType", myRevisionType)
+			.append("myRevisionNumber", myRevisionNumber)
+			.append("myRevisionTimestamp", myRevisionTimestamp)
+			.toString();
+	}
+}

hapi-fhir-jpaserver-test-utilities/src/main/java/ca/uhn/fhir/jpa/test/config/TestHSearchAddInConfig.java

@@ -136,7 +136,6 @@ public class TestHSearchAddInConfig {
 			luceneHeapProperties.put(BackendSettings.backendKey(LuceneBackendSettings.LUCENE_VERSION), "LUCENE_CURRENT");
 			luceneHeapProperties.put(HibernateOrmMapperSettings.ENABLED, "true");
 			luceneHeapProperties.put(BackendSettings.backendKey(LuceneIndexSettings.IO_WRITER_INFOSTREAM), "true");
-			// TODO: LD: rely on config properties to set this in a future MR
 			luceneHeapProperties.put(Constants.HIBERNATE_INTEGRATION_ENVERS_ENABLED, "true");
 
 			return (theProperties) -> {

hapi-fhir-server-mdm/src/main/java/ca/uhn/fhir/mdm/api/IMdmControllerSvc.java

@@ -46,6 +46,8 @@ public interface IMdmControllerSvc {
 
 	Page<MdmLinkJson> queryLinksFromPartitionList(MdmQuerySearchParameters theMdmQuerySearchParameters, MdmTransactionContext theMdmTransactionContext);
 
+	List<MdmLinkWithRevisionJson> queryLinkHistory(MdmHistorySearchParameters theMdmHistorySearchParameters, RequestDetails theRequestDetails);
+
 	Page<MdmLinkJson> getDuplicateGoldenResources(MdmTransactionContext theMdmTransactionContext, MdmPageRequest thePageRequest);
 
 	Page<MdmLinkJson> getDuplicateGoldenResources(MdmTransactionContext theMdmTransactionContext, MdmPageRequest thePageRequest, RequestDetails theRequestDetails, String theRequestResourceType);

hapi-fhir-server-mdm/src/main/java/ca/uhn/fhir/mdm/api/IMdmLinkQuerySvc.java

@@ -37,4 +37,6 @@ public interface IMdmLinkQuerySvc {
 	Page<MdmLinkJson> queryLinks(MdmQuerySearchParameters theMdmQuerySearchParameters, MdmTransactionContext theMdmContext);
 	Page<MdmLinkJson> getDuplicateGoldenResources(MdmTransactionContext theMdmContext, MdmPageRequest thePageRequest);
 	Page<MdmLinkJson> getDuplicateGoldenResources(MdmTransactionContext theMdmContext, MdmPageRequest thePageRequest, List<Integer> thePartitionId, String theRequestResourceType);
+
+    List<MdmLinkWithRevisionJson> queryLinkHistory(MdmHistorySearchParameters theMdmHistorySearchParameters);
 }

hapi-fhir-server-mdm/src/main/java/ca/uhn/fhir/mdm/api/MdmHistorySearchParameters.java

@@ -0,0 +1,90 @@
+package ca.uhn.fhir.mdm.api;
+
+/*-
+ * #%L
+ * HAPI FHIR - Master Data Management
+ * %%
+ * Copyright (C) 2014 - 2023 Smile CDR, Inc.
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import ca.uhn.fhir.mdm.provider.MdmControllerUtil;
+import ca.uhn.fhir.rest.server.provider.ProviderConstants;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.hl7.fhir.instance.model.api.IIdType;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+public class MdmHistorySearchParameters {
+	private List<IIdType> myGoldenResourceIds;
+	private List<IIdType> mySourceIds;
+
+	public MdmHistorySearchParameters() {}
+
+	public List<IIdType> getGoldenResourceIds() {
+		return myGoldenResourceIds;
+	}
+
+	public List<IIdType> getSourceIds() {
+		return mySourceIds;
+	}
+
+	public MdmHistorySearchParameters setGoldenResourceIds(List<String> theGoldenResourceIds) {
+		myGoldenResourceIds = extractId(theGoldenResourceIds);
+		return this;
+	}
+
+	public MdmHistorySearchParameters setSourceIds(List<String> theSourceIds) {
+		mySourceIds = extractId(theSourceIds);
+		return this;
+	}
+
+	@Override
+	public boolean equals(Object theO) {
+		if (this == theO) return true;
+		if (theO == null || getClass() != theO.getClass()) return false;
+		final MdmHistorySearchParameters that = (MdmHistorySearchParameters) theO;
+		return Objects.equals(myGoldenResourceIds, that.myGoldenResourceIds) && Objects.equals(mySourceIds, that.mySourceIds);
+	}
+
+	@Override
+	public int hashCode() {
+		return Objects.hash(myGoldenResourceIds, mySourceIds);
+	}
+
+	@Override
+	public String toString() {
+		return new ToStringBuilder(this)
+			.append("myMdmGoldenResourceIds", myGoldenResourceIds)
+			.append("myMdmTargetResourceIds", mySourceIds)
+			.toString();
+	}
+
+	@Nonnull
+	private static List<IIdType> extractId(List<String> theTheGoldenResourceIds) {
+		return theTheGoldenResourceIds.stream()
+			.map(MdmHistorySearchParameters::extractId)
+			.collect(Collectors.toUnmodifiableList());
+	}
+
+	@Nullable
+	private static IIdType extractId(String theTheGoldenResourceId) {
+		return MdmControllerUtil.extractGoldenResourceIdDtOrNull(ProviderConstants.MDM_QUERY_LINKS_GOLDEN_RESOURCE_ID, theTheGoldenResourceId);
+	}
+}

hapi-fhir-server-mdm/src/main/java/ca/uhn/fhir/mdm/api/MdmLinkJson.java

@@ -23,6 +23,7 @@ import ca.uhn.fhir.model.api.IModelJson;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
 import java.util.Date;
+import java.util.Objects;
 
 public class MdmLinkJson implements IModelJson {
 
@@ -175,6 +176,41 @@ public class MdmLinkJson implements IModelJson {
 		myRuleCount = theRuleCount;
 	}
 
+	@Override
+	public boolean equals(Object theO) {
+		if (this == theO) return true;
+		if (theO == null || getClass() != theO.getClass()) return false;
+		final MdmLinkJson that = (MdmLinkJson) theO;
+		return Objects.equals(myGoldenResourceId, that.myGoldenResourceId) &&
+			Objects.equals(mySourceId, that.mySourceId) &&
+			myMatchResult == that.myMatchResult &&
+			myLinkSource == that.myLinkSource &&
+			Objects.equals(myCreated, that.myCreated) &&
+			Objects.equals(myUpdated, that.myUpdated) &&
+			Objects.equals(myVersion, that.myVersion) &&
+			Objects.equals(myEidMatch, that.myEidMatch) &&
+			Objects.equals(myLinkCreatedNewResource, that.myLinkCreatedNewResource) &&
+			Objects.equals(myVector, that.myVector) &&
+			Objects.equals(myScore, that.myScore) &&
+			Objects.equals(myRuleCount, that.myRuleCount);
+	}
+
+	@Override
+	public int hashCode() {
+		return Objects.hash(myGoldenResourceId,
+			mySourceId,
+			myMatchResult,
+			myLinkSource,
+			myCreated,
+			myUpdated,
+			myVersion,
+			myEidMatch,
+			myLinkCreatedNewResource,
+			myVector,
+			myScore,
+			myRuleCount);
+	}
+
 	@Override
 	public String toString() {
 		return "MdmLinkJson{" +

hapi-fhir-server-mdm/src/main/java/ca/uhn/fhir/mdm/api/MdmLinkWithRevision.java

@@ -0,0 +1,66 @@
+package ca.uhn.fhir.mdm.api;
+
+/*-
+ * #%L
+ * HAPI FHIR - Master Data Management
+ * %%
+ * Copyright (C) 2014 - 2023 Smile CDR, Inc.
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import ca.uhn.fhir.jpa.model.entity.EnversRevision;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+
+import java.util.Objects;
+
+public class MdmLinkWithRevision<T extends IMdmLink<?>> {
+	private final T myMdmLink;
+	private final EnversRevision myEnversRevision;
+
+	public MdmLinkWithRevision(T theMdmLink, EnversRevision theEnversRevision) {
+		myMdmLink = theMdmLink;
+		myEnversRevision = theEnversRevision;
+	}
+
+	public T getMdmLink() {
+		return myMdmLink;
+	}
+
+	public EnversRevision getEnversRevision() {
+		return myEnversRevision;
+	}
+
+	@Override
+	public boolean equals(Object theO) {
+		if (this == theO) return true;
+		if (theO == null || getClass() != theO.getClass()) return false;
+		final MdmLinkWithRevision<?> that = (MdmLinkWithRevision<?>) theO;
+		return Objects.equals(myMdmLink, that.myMdmLink) && Objects.equals(myEnversRevision, that.myEnversRevision);
+	}
+
+	@Override
+	public int hashCode() {
+		return Objects.hash(myMdmLink, myEnversRevision);
+	}
+
+	@Override
+	public String toString() {
+		return new ToStringBuilder(this)
+			.append("myMdmLink", myMdmLink)
+			.append("myEnversRevision", myEnversRevision)
+			.toString();
+	}
+
+}

hapi-fhir-server-mdm/src/main/java/ca/uhn/fhir/mdm/api/MdmLinkWithRevisionJson.java

@@ -0,0 +1,79 @@
+package ca.uhn.fhir.mdm.api;
+
+/*-
+ * #%L
+ * HAPI FHIR - Master Data Management
+ * %%
+ * Copyright (C) 2014 - 2023 Smile CDR, Inc.
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import ca.uhn.fhir.model.api.IModelJson;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+
+import java.util.Date;
+import java.util.Objects;
+
+public class MdmLinkWithRevisionJson implements IModelJson {
+	@JsonProperty(value = "mdmLink", required = true)
+	MdmLinkJson myMdmLink;
+
+	@JsonProperty(value = "revisionNumber", required = true)
+	Long myRevisionNumber;
+
+	@JsonProperty(value = "revisionTimestamp", required = true)
+	Date myRevisionTimestamp;
+
+	public MdmLinkWithRevisionJson(MdmLinkJson theMdmLink, Long theRevisionNumber, Date theRevisionTimestamp) {
+		myMdmLink = theMdmLink;
+		myRevisionNumber = theRevisionNumber;
+		myRevisionTimestamp = theRevisionTimestamp;
+	}
+
+	public MdmLinkJson getMdmLink() {
+		return myMdmLink;
+	}
+
+	public Long getRevisionNumber() {
+		return myRevisionNumber;
+	}
+
+	public Date getRevisionTimestamp() {
+		return myRevisionTimestamp;
+	}
+
+	@Override
+	public boolean equals(Object theO) {
+		if (this == theO) return true;
+		if (theO == null || getClass() != theO.getClass()) return false;
+		final MdmLinkWithRevisionJson that = (MdmLinkWithRevisionJson) theO;
+		return myMdmLink.equals(that.myMdmLink) && myRevisionNumber.equals(that.myRevisionNumber) && myRevisionTimestamp.equals(that.myRevisionTimestamp);
+	}
+
+	@Override
+	public int hashCode() {
+		return Objects.hash(myMdmLink, myRevisionNumber, myRevisionTimestamp);
+	}
+
+	@Override
+	public String toString() {
+		return new ToStringBuilder(this)
+			.append("myMdmLink", myMdmLink)
+			.append("myRevisionNumber", myRevisionNumber)
+			.append("myRevisionTimestamp", myRevisionTimestamp)
+			.toString();
+	}
+}

hapi-fhir-server-mdm/src/main/java/ca/uhn/fhir/mdm/dao/IMdmLinkDao.java

@@ -19,8 +19,11 @@
  */
 package ca.uhn.fhir.mdm.dao;
 
+import ca.uhn.fhir.i18n.Msg;
 import ca.uhn.fhir.mdm.api.IMdmLink;
+import ca.uhn.fhir.mdm.api.MdmHistorySearchParameters;
 import ca.uhn.fhir.mdm.api.MdmLinkSourceEnum;
+import ca.uhn.fhir.mdm.api.MdmLinkWithRevision;
 import ca.uhn.fhir.mdm.api.MdmMatchResultEnum;
 import ca.uhn.fhir.mdm.api.MdmQuerySearchParameters;
 import ca.uhn.fhir.mdm.api.paging.MdmPageRequest;
@@ -32,6 +35,7 @@ import org.springframework.data.domain.Example;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.history.Revisions;
+import org.springframework.data.history.Revision;
 
 import java.util.Date;
 import java.util.List;
@@ -83,5 +87,13 @@ public interface IMdmLinkDao<P extends IResourcePersistentId, M extends IMdmLink
 
 	void deleteLinksWithAnyReferenceToPids(List<P> theResourcePersistentIds);
 
-	Revisions<Long, M> findHistory(P thePid);
+	// TODO: LD:  delete for good on the next bump
+	@Deprecated(since = "6.5.6", forRemoval = true)
+	default Revisions<Long, M> findHistory(P thePid) {
+		throw new UnsupportedOperationException(Msg.code(2296) + "Deprecated and not supported in non-JPA");
+	}
+
+	default List<MdmLinkWithRevision<M>> getHistoryForIds(MdmHistorySearchParameters theMdmHistorySearchParameters) {
+		throw new UnsupportedOperationException(Msg.code(2299) + "not yet implemented");
+	}
 }

hapi-fhir-server-mdm/src/main/java/ca/uhn/fhir/mdm/provider/BaseMdmProvider.java

@@ -22,6 +22,7 @@ package ca.uhn.fhir.mdm.provider;
 import ca.uhn.fhir.context.FhirContext;
 import ca.uhn.fhir.i18n.Msg;
 import ca.uhn.fhir.mdm.api.MdmLinkJson;
+import ca.uhn.fhir.mdm.api.MdmLinkWithRevisionJson;
 import ca.uhn.fhir.mdm.api.MdmMatchResultEnum;
 import ca.uhn.fhir.mdm.api.paging.MdmPageLinkBuilder;
 import ca.uhn.fhir.mdm.api.paging.MdmPageLinkTuple;
@@ -38,7 +39,11 @@ import org.hl7.fhir.instance.model.api.IBaseParameters;
 import org.hl7.fhir.instance.model.api.IPrimitiveType;
 import org.springframework.data.domain.Page;
 
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
 
 
 public abstract class BaseMdmProvider {
@@ -63,6 +68,16 @@ public abstract class BaseMdmProvider {
 		}
 	}
 
+	protected void validateMdmLinkHistoryParameters(List<IPrimitiveType<String>> theGoldenResourceIds, List<IPrimitiveType<String>> theSourceIds) {
+		validateBothCannotBeNullOrEmpty(ProviderConstants.MDM_QUERY_LINKS_GOLDEN_RESOURCE_ID, theGoldenResourceIds, ProviderConstants.MDM_QUERY_LINKS_RESOURCE_ID, theSourceIds);
+	}
+
+	private void validateBothCannotBeNullOrEmpty(String theFirstName, List<IPrimitiveType<String>> theFirstList, String theSecondName, List<IPrimitiveType<String>> theSecondList) {
+		if ((theFirstList == null || theFirstList.isEmpty()) || (theSecondList == null || theSecondList.isEmpty())) {
+			throw new InvalidRequestException(Msg.code(2292) + "both ["+theFirstName+"] and ["+theSecondName+"] cannot be null or empty");
+		}
+	}
+
 	protected void validateUpdateLinkParameters(IPrimitiveType<String> theGoldenResourceId, IPrimitiveType<String> theResourceId, IPrimitiveType<String> theMatchResult) {
 		validateNotNull(ProviderConstants.MDM_UPDATE_LINK_GOLDEN_RESOURCE_ID, theGoldenResourceId);
 		validateNotNull(ProviderConstants.MDM_UPDATE_LINK_RESOURCE_ID, theResourceId);
@@ -107,6 +122,13 @@ public abstract class BaseMdmProvider {
 		return mdmTransactionContext;
 	}
 
+	@Nonnull
+	protected List<String> convertToStringsIfNotNull(List<IPrimitiveType<String>> thePrimitiveTypeStrings) {
+		return thePrimitiveTypeStrings == null
+			? Collections.emptyList()
+			: thePrimitiveTypeStrings.stream().map(this::extractStringOrNull).collect(Collectors.toUnmodifiableList());
+	}
+
 	protected String extractStringOrNull(IPrimitiveType<String> theString) {
 		if (theString == null) {
 			return null;
@@ -135,6 +157,32 @@ public abstract class BaseMdmProvider {
 		});
 		return retval;
 	}
+	protected void parametersFromMdmLinkRevisions(IBaseParameters theRetVal, List<MdmLinkWithRevisionJson> theMdmLinkRevisions) {
+		if (theMdmLinkRevisions.isEmpty()) {
+			final IBase resultPart = ParametersUtil.addParameterToParameters(myFhirContext, theRetVal, "historical links not found for query parameters");
+
+			ParametersUtil.addPartString(myFhirContext, resultPart, "theResults", "historical links not found for query parameters");
+		}
+
+		theMdmLinkRevisions.forEach(mdmLinkRevision -> parametersFromMdmLinkRevision(theRetVal, mdmLinkRevision));
+	}
+
+	private void parametersFromMdmLinkRevision(IBaseParameters retVal, MdmLinkWithRevisionJson mdmLinkRevision) {
+		final IBase resultPart = ParametersUtil.addParameterToParameters(myFhirContext, retVal, "historical link");
+		final MdmLinkJson mdmLink = mdmLinkRevision.getMdmLink();
+
+		ParametersUtil.addPartString(myFhirContext, resultPart, "goldenResourceId", mdmLink.getGoldenResourceId());
+		ParametersUtil.addPartString(myFhirContext, resultPart, "revisionTimestamp", mdmLinkRevision.getRevisionTimestamp().toString());
+		ParametersUtil.addPartString(myFhirContext, resultPart, "sourceResourceId", mdmLink.getSourceId());
+		ParametersUtil.addPartString(myFhirContext, resultPart, "matchResult", mdmLink.getMatchResult().name());
+		ParametersUtil.addPartDecimal(myFhirContext, resultPart, "score", mdmLink.getScore());
+		ParametersUtil.addPartString(myFhirContext, resultPart, "linkSource", mdmLink.getLinkSource().name());
+		ParametersUtil.addPartBoolean(myFhirContext, resultPart, "eidMatch", mdmLink.getEidMatch());
+		ParametersUtil.addPartBoolean(myFhirContext, resultPart, "hadToCreateNewResource", mdmLink.getLinkCreatedNewResource());
+		ParametersUtil.addPartDecimal(myFhirContext, resultPart, "score", mdmLink.getScore());
+		ParametersUtil.addPartDecimal(myFhirContext, resultPart, "linkCreated", (double) mdmLink.getCreated().getTime());
+		ParametersUtil.addPartDecimal(myFhirContext, resultPart, "linkUpdated", (double) mdmLink.getUpdated().getTime());
+	}
 
 	protected void addPagingParameters(IBaseParameters theParameters, Page<MdmLinkJson> theCurrentPage, ServletRequestDetails theServletRequestDetails, MdmPageRequest thePageRequest) {
 		MdmPageLinkTuple mdmPageLinkTuple = MdmPageLinkBuilder.buildMdmPageLinks(theServletRequestDetails, theCurrentPage, thePageRequest);

hapi-fhir-server-mdm/src/main/java/ca/uhn/fhir/mdm/provider/MdmLinkHistoryProviderDstu3Plus.java

@@ -0,0 +1,71 @@
+package ca.uhn.fhir.mdm.provider;
+
+/*-
+ * #%L
+ * HAPI FHIR - Master Data Management
+ * %%
+ * Copyright (C) 2014 - 2023 Smile CDR, Inc.
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import ca.uhn.fhir.context.FhirContext;
+import ca.uhn.fhir.mdm.api.IMdmControllerSvc;
+import ca.uhn.fhir.mdm.api.MdmHistorySearchParameters;
+import ca.uhn.fhir.mdm.api.MdmLinkWithRevisionJson;
+import ca.uhn.fhir.rest.annotation.Operation;
+import ca.uhn.fhir.rest.annotation.OperationParam;
+import ca.uhn.fhir.rest.server.provider.ProviderConstants;
+import ca.uhn.fhir.rest.server.servlet.ServletRequestDetails;
+import ca.uhn.fhir.util.ParametersUtil;
+import org.hl7.fhir.instance.model.api.IBaseParameters;
+import org.hl7.fhir.instance.model.api.IPrimitiveType;
+import org.slf4j.Logger;
+
+import java.util.List;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+public class MdmLinkHistoryProviderDstu3Plus extends BaseMdmProvider {
+	private static final Logger ourLog = getLogger(MdmLinkHistoryProviderDstu3Plus.class);
+
+	private final IMdmControllerSvc myMdmControllerSvc;
+
+	public MdmLinkHistoryProviderDstu3Plus(FhirContext theFhirContext, IMdmControllerSvc theMdmControllerSvc) {
+		super(theFhirContext);
+		myMdmControllerSvc = theMdmControllerSvc;
+	}
+
+	@Operation(name = ProviderConstants.MDM_LINK_HISTORY, idempotent = true)
+	public IBaseParameters historyLinks(@OperationParam(name = ProviderConstants.MDM_QUERY_LINKS_GOLDEN_RESOURCE_ID, min = 0, max = OperationParam.MAX_UNLIMITED, typeName = "string") List<IPrimitiveType<String>> theMdmGoldenResourceIds,
+													@OperationParam(name = ProviderConstants.MDM_QUERY_LINKS_RESOURCE_ID, min = 0, max = OperationParam.MAX_UNLIMITED, typeName = "string") List<IPrimitiveType<String>> theResourceIds,
+													ServletRequestDetails theRequestDetails) {
+		validateMdmLinkHistoryParameters(theMdmGoldenResourceIds, theResourceIds);
+
+		final List<String> goldenResourceIdsToUse = convertToStringsIfNotNull(theMdmGoldenResourceIds);
+		final List<String> resourceIdsToUse = convertToStringsIfNotNull(theResourceIds);
+
+		final IBaseParameters retVal = ParametersUtil.newInstance(myFhirContext);
+
+		final MdmHistorySearchParameters mdmHistorySearchParameters = new MdmHistorySearchParameters()
+			.setGoldenResourceIds(goldenResourceIdsToUse)
+			.setSourceIds(resourceIdsToUse);
+
+		final List<MdmLinkWithRevisionJson> mdmLinkRevisionsFromSvc = myMdmControllerSvc.queryLinkHistory(mdmHistorySearchParameters, theRequestDetails);
+
+		parametersFromMdmLinkRevisions(retVal, mdmLinkRevisionsFromSvc);
+
+		return retVal;
+	}
+}

hapi-fhir-server-mdm/src/main/java/ca/uhn/fhir/mdm/provider/MdmProviderDstu3Plus.java

@@ -25,7 +25,9 @@ import ca.uhn.fhir.mdm.api.IMdmControllerSvc;
 import ca.uhn.fhir.mdm.api.IMdmSettings;
 import ca.uhn.fhir.mdm.api.IMdmSubmitSvc;
 import ca.uhn.fhir.mdm.api.MdmConstants;
+import ca.uhn.fhir.mdm.api.MdmHistorySearchParameters;
 import ca.uhn.fhir.mdm.api.MdmLinkJson;
+import ca.uhn.fhir.mdm.api.MdmLinkWithRevisionJson;
 import ca.uhn.fhir.mdm.api.MdmQuerySearchParameters;
 import ca.uhn.fhir.mdm.api.paging.MdmPageRequest;
 import ca.uhn.fhir.mdm.model.MdmTransactionContext;
@@ -36,8 +38,10 @@ import ca.uhn.fhir.rest.annotation.OperationParam;
 import ca.uhn.fhir.rest.api.Constants;
 import ca.uhn.fhir.rest.api.server.RequestDetails;
 import ca.uhn.fhir.rest.server.exceptions.InvalidRequestException;
+import ca.uhn.fhir.rest.server.exceptions.MethodNotAllowedException;
 import ca.uhn.fhir.rest.server.provider.ProviderConstants;
 import ca.uhn.fhir.rest.server.servlet.ServletRequestDetails;
+import ca.uhn.fhir.system.HapiSystemProperties;
 import ca.uhn.fhir.util.ParametersUtil;
 import org.apache.commons.lang3.StringUtils;
 import org.hl7.fhir.instance.model.api.IAnyResource;
@@ -174,6 +178,7 @@ public class MdmProviderDstu3Plus extends BaseMdmProvider {
 		}
 	}
 
+	// Is a set of the OR sufficient ot the contenxt she's investigating?
 	@Operation(name = ProviderConstants.MDM_QUERY_LINKS, idempotent = true)
 	public IBaseParameters queryLinks(@OperationParam(name = ProviderConstants.MDM_QUERY_LINKS_GOLDEN_RESOURCE_ID, min = 0, max = 1, typeName = "string") IPrimitiveType<String> theGoldenResourceId,
 												 @OperationParam(name = ProviderConstants.MDM_QUERY_LINKS_RESOURCE_ID, min = 0, max = 1, typeName = "string") IPrimitiveType<String> theResourceId,
@@ -210,6 +215,7 @@ public class MdmProviderDstu3Plus extends BaseMdmProvider {
 		return parametersFromMdmLinks(mdmLinkJson, true, theRequestDetails, mdmPageRequest);
 	}
 
+
 	@Operation(name = ProviderConstants.MDM_DUPLICATE_GOLDEN_RESOURCES, idempotent = true)
 	public IBaseParameters getDuplicateGoldenResources(
 		@Description(formalDefinition="Results from this method are returned across multiple pages. This parameter controls the offset when fetching a page.")

hapi-fhir-server-mdm/src/main/java/ca/uhn/fhir/mdm/provider/MdmProviderLoader.java

@@ -22,6 +22,7 @@ package ca.uhn.fhir.mdm.provider;
 import ca.uhn.fhir.context.ConfigurationException;
 import ca.uhn.fhir.context.FhirContext;
 import ca.uhn.fhir.i18n.Msg;
+import ca.uhn.fhir.jpa.api.config.JpaStorageSettings;
 import ca.uhn.fhir.mdm.api.IMdmControllerSvc;
 import ca.uhn.fhir.mdm.api.IMdmSettings;
 import ca.uhn.fhir.mdm.api.IMdmSubmitSvc;
@@ -45,6 +46,8 @@ public class MdmProviderLoader {
 	private IMdmSubmitSvc myMdmSubmitSvc;
 	@Autowired
 	private IMdmSettings myMdmSettings;
+	@Autowired
+	private JpaStorageSettings myStorageSettings;
 
 	private BaseMdmProvider myMdmProvider;
 
@@ -58,6 +61,9 @@ public class MdmProviderLoader {
 						myMdmSubmitSvc,
 						myMdmSettings
 						));
+				if (myStorageSettings.isNonResourceDbHistoryEnabled()) {
+					myResourceProviderFactory.addSupplier(() -> new MdmLinkHistoryProviderDstu3Plus(myFhirContext, myMdmControllerSvc));
+				}
 				break;
 			default:
 				throw new ConfigurationException(Msg.code(1497) + "MDM not supported for FHIR version " + myFhirContext.getVersion().getVersion());

hapi-fhir-server/src/main/java/ca/uhn/fhir/rest/server/provider/ProviderConstants.java

@@ -85,6 +85,7 @@ public class ProviderConstants {
 	public static final String MDM_CREATE_LINK_MATCH_RESULT = "matchResult";
 
 	public static final String MDM_QUERY_LINKS = "$mdm-query-links";
+	public static final String MDM_LINK_HISTORY = "$mdm-link-history";
 	public static final String MDM_QUERY_LINKS_GOLDEN_RESOURCE_ID = "goldenResourceId";
 	public static final String MDM_QUERY_LINKS_RESOURCE_ID = "resourceId";
 	public static final String MDM_QUERY_PARTITION_IDS = "partitionIds";

hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/api/config/JpaStorageSettings.java

@@ -311,6 +311,12 @@ public class JpaStorageSettings extends StorageSettings {
 	 */
 	private boolean myJobFastTrackingEnabled = false;
 
+	/**
+	 * Since 6.6.0
+	 * Applies to MDM links.
+	 */
+	private boolean myNonResourceDbHistoryEnabled = true;
+
 	/**
 	 * Constructor
 	 */
@@ -2331,6 +2337,24 @@ public class JpaStorageSettings extends StorageSettings {
 		myJobFastTrackingEnabled = theJobFastTrackingEnabled;
 	}
 
+	/**
+	 * This setting controls whether MdmLink and other non-resource DB history is enabled.
+	 * This setting controls whether non-resource DB history is enabled
+	 * <p/>
+	 * By default, this is enabled unless explicitly disabled.
+	 *
+	 * @return Whether non-resource DB history is enabled (default is true);
+	 * @since 6.6.0
+	 */
+	public boolean isNonResourceDbHistoryEnabled() {
+		return myNonResourceDbHistoryEnabled;
+	}
+
+
+	public void setNonResourceDbHistoryEnabled(boolean theNonResourceDbHistoryEnabled) {
+		myNonResourceDbHistoryEnabled = theNonResourceDbHistoryEnabled;
+	}
+
 	public enum StoreMetaSourceInformationEnum {
 		NONE(false, false),
 		SOURCE_URI(true, false),