MODLD-131: transform DI event message to LD resource (#92)

* MODLD-131: transform DI event message to LD resource

* MODLD-131: fix DataImportEventListenerIT

* MODLD-131: fix DataImportEventListenerIT

pom.xml

@@ -43,6 +43,7 @@
     <lombok.mapstruct-binding.version>0.2.0</lombok.mapstruct-binding.version>
     <hibernate-types.version>2.21.1</hibernate-types.version>
     <lib-linked-data-dictionary.version>1.0.0-SNAPSHOT</lib-linked-data-dictionary.version>
+    <lib-linked-data-marc2ld.version>1.0.0-SNAPSHOT</lib-linked-data-marc2ld.version>
 
     <!-- Test dependencies versions -->
     <testcontainers.version>1.19.1</testcontainers.version>
@@ -80,6 +81,11 @@
       <artifactId>lib-linked-data-dictionary</artifactId>
       <version>${lib-linked-data-dictionary.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.folio</groupId>
+      <artifactId>lib-linked-data-marc2ld</artifactId>
+      <version>${lib-linked-data-marc2ld.version}</version>
+    </dependency>
 
     <dependency>
       <groupId>org.springframework.boot</groupId>

src/main/java/org/folio/linked/data/LinkedDataApplication.java

@@ -4,12 +4,14 @@
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cache.annotation.EnableCaching;
 import org.springframework.cloud.openfeign.EnableFeignClients;
+import org.springframework.context.annotation.ComponentScan;
 import org.springframework.scheduling.annotation.EnableAsync;
 
 @EnableCaching
 @EnableAsync
 @EnableFeignClients
 @SpringBootApplication
+@ComponentScan({"org.folio.linked.data", "org.folio.marc2ld"})
 public class LinkedDataApplication {
 
 

src/main/java/org/folio/linked/data/configuration/KafkaListenerConfiguration.java

@@ -7,12 +7,10 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.function.Consumer;
 import lombok.RequiredArgsConstructor;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.folio.search.domain.dto.DataImportEvent;
 import org.folio.spring.tools.kafka.FolioKafkaProperties;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.Bean;
@@ -55,10 +53,4 @@ public ConsumerFactory<String, DataImportEvent> dataImportEventConsumerFactory()
     return new DefaultKafkaConsumerFactory<>(config, new StringDeserializer(), deserializer);
   }
 
-  @Bean
-  @Qualifier("dataImportEventProcessor")
-  // TODO (MODLD-131) - Delete this bean & replace with the actual component that process data import events
-  public Consumer<DataImportEvent> dataImportEventProcessor() {
-    return event -> {};
-  }
 }

src/main/java/org/folio/linked/data/configuration/json/ObjectMapperConfig.java

@@ -15,7 +15,9 @@
 import org.folio.search.domain.dto.DataImportEvent;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
 
+@Primary
 @Configuration
 public class ObjectMapperConfig {
 

src/main/java/org/folio/linked/data/integration/KafkaMessageListener.java

@@ -2,11 +2,10 @@
 
 import static org.folio.linked.data.util.Constants.FOLIO_PROFILE;
 
-import java.util.function.Consumer;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
+import org.folio.linked.data.integration.consumer.DataImportEventHandler;
 import org.folio.search.domain.dto.DataImportEvent;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.Profile;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
@@ -18,9 +17,7 @@
 public class KafkaMessageListener {
 
   private static final String DI_INSTANCE_CREATED_LISTENER = "mod-linked-data-data-import-instance-created-listener";
-
-  @Qualifier("dataImportEventProcessor")
-  private final Consumer<DataImportEvent> eventProcessor;
+  private final DataImportEventHandler dataImportEventHandler;
 
   @KafkaListener(
     id = DI_INSTANCE_CREATED_LISTENER,
@@ -30,6 +27,6 @@ public class KafkaMessageListener {
     topicPattern = "#{folioKafkaProperties.listener['data-import-instance-create'].topicPattern}")
   public void handleDataImportInstanceCreatedEvent(DataImportEvent event) {
     log.info("Received event: {}", event);
-    eventProcessor.accept(event);
+    dataImportEventHandler.handle(event);
   }
 }

src/main/java/org/folio/linked/data/integration/consumer/DataImportEventHandler.java

@@ -0,0 +1,33 @@
+package org.folio.linked.data.integration.consumer;
+
+import static org.apache.commons.lang3.ObjectUtils.isNotEmpty;
+import static org.folio.linked.data.util.Constants.FOLIO_PROFILE;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import org.folio.linked.data.service.ResourceService;
+import org.folio.marc2ld.mapper.Marc2BibframeMapper;
+import org.folio.search.domain.dto.DataImportEvent;
+import org.springframework.context.annotation.Profile;
+import org.springframework.stereotype.Component;
+
+@Log4j2
+@Component
+@Profile(FOLIO_PROFILE)
+@RequiredArgsConstructor
+public class DataImportEventHandler {
+
+  private final Marc2BibframeMapper marc2BibframeMapper;
+  private final ResourceService resourceService;
+
+  public void handle(DataImportEvent event) {
+    if (isNotEmpty(event.getMarc())) {
+      var marc2ldResource = marc2BibframeMapper.map(event.getMarc());
+      Long id = resourceService.createResource(marc2ldResource);
+      log.info("DataImportEvent with id [{}] was saved as LD resource with id [{}]", event.getId(), id);
+    } else {
+      log.error("DataImportEvent with id [{}], tenant [{}], eventType [{}] has no Marc record inside",
+        event.getId(), event.getTenant(), event.getEventType());
+    }
+  }
+}

src/main/java/org/folio/linked/data/mapper/ResourceMapper.java

@@ -1,5 +1,6 @@
 package org.folio.linked.data.mapper;
 
+import static java.util.stream.Collectors.toSet;
 import static org.folio.ld.dictionary.ResourceTypeDictionary.INSTANCE;
 import static org.mapstruct.MappingConstants.ComponentModel.SPRING;
 
@@ -8,6 +9,8 @@
 import lombok.NonNull;
 import lombok.SneakyThrows;
 import lombok.extern.log4j.Log4j2;
+import org.folio.ld.dictionary.PredicateDictionary;
+import org.folio.ld.dictionary.ResourceTypeDictionary;
 import org.folio.linked.data.domain.dto.InstanceField;
 import org.folio.linked.data.domain.dto.ResourceDto;
 import org.folio.linked.data.domain.dto.ResourceField;
@@ -18,7 +21,11 @@
 import org.folio.linked.data.mapper.resource.common.inner.InnerResourceMapper;
 import org.folio.linked.data.mapper.resource.kafka.KafkaMessageMapper;
 import org.folio.linked.data.model.ResourceShortInfo;
+import org.folio.linked.data.model.entity.PredicateEntity;
 import org.folio.linked.data.model.entity.Resource;
+import org.folio.linked.data.model.entity.ResourceEdge;
+import org.folio.linked.data.model.entity.ResourceTypeEntity;
+import org.folio.linked.data.model.entity.pk.ResourceEdgePk;
 import org.folio.search.domain.dto.BibframeIndex;
 import org.mapstruct.Mapper;
 import org.mapstruct.Mapping;
@@ -46,6 +53,31 @@ public abstract class ResourceMapper {
 
   public abstract ResourceShortInfoPage map(Page<ResourceShort> page);
 
+  public Resource toEntity(org.folio.marc2ld.model.Resource marc2ldResource) {
+    Resource resource = new Resource();
+    resource.setLabel(marc2ldResource.getLabel());
+    resource.setDoc(marc2ldResource.getDoc());
+    resource.setResourceHash(marc2ldResource.getResourceHash());
+    resource.setTypes(marc2ldResource.getTypes().stream().map(this::toEntity).collect(toSet()));
+    resource.setOutgoingEdges(marc2ldResource.getOutgoingEdges().stream().map(marc2ldEdge -> toEntity(marc2ldEdge,
+      resource)).collect(toSet()));
+    return resource;
+  }
+
+  public abstract ResourceTypeEntity toEntity(ResourceTypeDictionary dictionary);
+
+  public ResourceEdge toEntity(org.folio.marc2ld.model.ResourceEdge marc2ldEdge, Resource source) {
+    var edge = new ResourceEdge();
+    edge.setId(new ResourceEdgePk(marc2ldEdge.getSource().getResourceHash(),
+      marc2ldEdge.getTarget().getResourceHash(), marc2ldEdge.getPredicate().getHash()));
+    edge.setSource(source);
+    edge.setTarget(toEntity(marc2ldEdge.getTarget()));
+    edge.setPredicate(toEntity(marc2ldEdge.getPredicate()));
+    return edge;
+  }
+
+  public abstract PredicateEntity toEntity(PredicateDictionary dictionary);
+
   @SneakyThrows
   public Resource toEntity(ResourceDto dto) {
     try {

src/main/java/org/folio/linked/data/model/entity/pk/ResourceEdgePk.java

@@ -2,13 +2,15 @@
 
 import jakarta.persistence.Embeddable;
 import java.io.Serializable;
+import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 import lombok.experimental.Accessors;
 
 @Data
 @Embeddable
 @NoArgsConstructor
+@AllArgsConstructor
 @Accessors(chain = true)
 public class ResourceEdgePk implements Serializable {
 

src/main/java/org/folio/linked/data/service/ResourceService.java

@@ -7,6 +7,8 @@ public interface ResourceService {
 
   ResourceDto createResource(ResourceDto resourceRequest);
 
+  Long createResource(org.folio.marc2ld.model.Resource resource);
+
   ResourceDto getResourceById(Long id);
 
   ResourceDto updateResource(Long id, ResourceDto bibframeRequest);

src/main/java/org/folio/linked/data/service/ResourceServiceImpl.java

@@ -41,6 +41,13 @@ public ResourceDto createResource(ResourceDto resourceDto) {
     return resourceMapper.toDto(persisted);
   }
 
+  public Long createResource(org.folio.marc2ld.model.Resource marc2ldResource) {
+    var mapped = resourceMapper.toEntity(marc2ldResource);
+    var persisted = resourceRepo.save(mapped);
+    kafkaSender.sendResourceCreated(resourceMapper.mapToIndex(persisted));
+    return persisted.getResourceHash();
+  }
+
   @Override
   @Transactional(readOnly = true)
   public ResourceDto getResourceById(Long id) {

src/main/resources/changelog/scripts/v-1.0.0/update-mod-linked-data-types-and-predicate-hashes.xml

@@ -87,5 +87,15 @@
     </update>
   </changeSet>
 
-
+  <changeSet id="12_add_Identifier_type" author="[email protected]">
+    <preConditions onFail="MARK_RAN">
+      <sqlCheck expectedResult="0">
+        select count(*) from type_lookup tl where tl.type_uri = 'http://bibfra.me/vocab/lite/Identifier'
+      </sqlCheck>
+    </preConditions>
+    <insert tableName="type_lookup">
+      <column name="type_uri" value="http://bibfra.me/vocab/lite/Identifier"/>
+      <column name="type_hash" valueComputed="hashing.hash_text('http://bibfra.me/vocab/lite/Identifier')"/>
+    </insert>
+  </changeSet>
 </databaseChangeLog>

src/test/java/org/folio/linked/data/e2e/DatabaseFolioIT.java

@@ -1,7 +1,10 @@
 package org.folio.linked.data.e2e;
 
+import static org.folio.linked.data.test.TestUtil.FOLIO_TEST_PROFILE;
+import static org.folio.linked.data.util.Constants.FOLIO_PROFILE;
+
 import org.springframework.test.context.ActiveProfiles;
 
-@ActiveProfiles({"folio", "test-folio"})
+@ActiveProfiles({FOLIO_PROFILE, FOLIO_TEST_PROFILE})
 class DatabaseFolioIT extends DatabaseIT {
 }

src/test/java/org/folio/linked/data/e2e/ReIndexControllerITFolio.java → src/test/java/org/folio/linked/data/e2e/ReIndexControllerFolioIT.java

@@ -1,6 +1,10 @@
 package org.folio.linked.data.e2e;
 
 import static java.util.Objects.nonNull;
+import static org.folio.linked.data.test.TestUtil.FOLIO_TEST_PROFILE;
+import static org.folio.linked.data.test.TestUtil.TENANT_ID;
+import static org.folio.linked.data.util.Constants.FOLIO_PROFILE;
+import static org.folio.linked.data.util.Constants.SEARCH_PROFILE;
 import static org.hamcrest.Matchers.containsString;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -9,20 +13,29 @@
 import org.folio.linked.data.model.entity.Resource;
 import org.folio.linked.data.test.kafka.KafkaSearchIndexTopicListener;
 import org.folio.search.domain.dto.ResourceEventType;
+import org.folio.spring.tools.kafka.KafkaAdminService;
 import org.hamcrest.MatcherAssert;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Disabled;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.test.context.ActiveProfiles;
 
-@ActiveProfiles({"folio", "test-folio", "search"})
-public class ReIndexControllerITFolio extends ReIndexControllerIT {
+@ActiveProfiles({FOLIO_PROFILE, FOLIO_TEST_PROFILE, SEARCH_PROFILE})
+@Disabled("To be enabled in part 3 task")
+public class ReIndexControllerFolioIT extends ReIndexControllerIT {
 
   @Autowired
   private KafkaSearchIndexTopicListener consumer;
 
+  @BeforeAll
+  static void beforeAll(@Autowired KafkaAdminService kafkaAdminService) {
+    kafkaAdminService.createTopics(TENANT_ID);
+  }
+
   @Override
   @SneakyThrows
   protected void checkKafkaMessageSent(Resource persisted) {
-    boolean messageConsumed = consumer.getLatch().await(10, TimeUnit.SECONDS);
+    boolean messageConsumed = consumer.getLatch().await(30, TimeUnit.SECONDS);
     assertTrue(messageConsumed);
     if (nonNull(persisted)) {
       MatcherAssert.assertThat(consumer.getPayload(), containsString(persisted.getResourceHash().toString()));

src/test/java/org/folio/linked/data/e2e/ReIndexControllerIT.java

@@ -1,6 +1,5 @@
 package org.folio.linked.data.e2e;
 
-import static org.folio.linked.data.test.TestUtil.TENANT_ID;
 import static org.folio.linked.data.test.TestUtil.defaultHeaders;
 import static org.folio.linked.data.test.TestUtil.getBibframeSample;
 import static org.folio.linked.data.test.TestUtil.getBibframeSampleTest;
@@ -19,9 +18,7 @@
 import org.folio.linked.data.repo.ResourceRepository;
 import org.folio.linked.data.test.ResourceEdgeRepository;
 import org.folio.spring.test.extension.impl.OkapiConfiguration;
-import org.folio.spring.tools.kafka.KafkaAdminService;
 import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.env.Environment;
@@ -49,11 +46,6 @@ class ReIndexControllerIT {
   @Autowired
   private Environment env;
 
-  @BeforeAll
-  static void beforeAll(@Autowired KafkaAdminService kafkaAdminService) {
-    kafkaAdminService.createTopics(TENANT_ID);
-  }
-
   @AfterEach
   public void clean() {
     resourceEdgeRepository.deleteAll();