FINERACT-1972 Custom Snapshot Event Triggered by COB

2024-01-22 15:29:21 +01:00
parent 4cb8946bc1
commit 8b379695d7
24 changed files with 1109 additions and 16 deletions
@@ -23,5 +23,7 @@ public final class FineractProfiles {
    public static final String LIQUIBASE_ONLY = "liquibase-only";
    public static final String DIAGNOSTICS = "diagnostics";

+    public static final String TEST = "test";
+
    private FineractProfiles() {}
 }
@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.fineract.infrastructure.event.external.api;
+
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.DELETE;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.core.MediaType;
+import java.util.List;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.infrastructure.core.boot.FineractProfiles;
+import org.apache.fineract.infrastructure.core.serialization.DefaultToApiJsonSerializer;
+import org.apache.fineract.infrastructure.event.external.service.InternalExternalEventService;
+import org.apache.fineract.infrastructure.event.external.service.validation.ExternalEventDTO;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.annotation.Profile;
+import org.springframework.stereotype.Component;
+
+@Profile(FineractProfiles.TEST)
+@Component
+@Path("/v1/internal/externalevents")
+@RequiredArgsConstructor
+@Slf4j
+public class InternalExternalEventsApiResource implements InitializingBean {
+
+    private final InternalExternalEventService internalExternalEventService;
+    private final DefaultToApiJsonSerializer<List<ExternalEventDTO>> jsonSerializer;
+
+    @Override
+    @SuppressFBWarnings("SLF4J_SIGN_ONLY_FORMAT")
+    public void afterPropertiesSet() throws Exception {
+        log.warn("------------------------------------------------------------");
+        log.warn("                                                            ");
+        log.warn("DO NOT USE THIS IN PRODUCTION!");
+        log.warn("Internal client services mode is enabled");
+        log.warn("DO NOT USE THIS IN PRODUCTION!");
+        log.warn("                                                            ");
+        log.warn("------------------------------------------------------------");
+    }
+
+    @GET
+    @Consumes({ MediaType.APPLICATION_JSON })
+    @Produces({ MediaType.APPLICATION_JSON })
+    public String getAllExternalEvents(@QueryParam("idempotencyKey") final String idempotencyKey, @QueryParam("type") final String type,
+            @QueryParam("category") final String category, @QueryParam("aggregateRootId") final Long aggregateRootId) {
+        log.debug("getAllExternalEvents called with params idempotencyKey:{}, type:{}, category:{}, aggregateRootId:{}  ", idempotencyKey,
+                type, category, aggregateRootId);
+        List<ExternalEventDTO> allExternalEvents = internalExternalEventService.getAllExternalEvents(idempotencyKey, type, category,
+                aggregateRootId);
+        return jsonSerializer.serialize(allExternalEvents);
+    }
+
+    @DELETE
+    public void deleteAllExternalEvents() {
+        log.debug("deleteAllExternalEvents called");
+        internalExternalEventService.deleteAllExternalEvents();
+    }
+
+}
@@ -26,11 +26,12 @@ import org.apache.fineract.infrastructure.event.external.repository.domain.Exter
 import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEventView;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
 import org.springframework.data.jpa.repository.Modifying;
 import org.springframework.data.jpa.repository.Query;
 import org.springframework.data.repository.query.Param;

-public interface ExternalEventRepository extends JpaRepository<ExternalEvent, Long> {
+public interface ExternalEventRepository extends JpaRepository<ExternalEvent, Long>, JpaSpecificationExecutor<ExternalEvent> {

    List<ExternalEventView> findByStatusOrderById(ExternalEventStatus status, Pageable batchSize);

@@ -0,0 +1,147 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.fineract.infrastructure.event.external.service;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.fineract.avro.BulkMessageItemV1;
+import org.apache.fineract.infrastructure.core.boot.FineractProfiles;
+import org.apache.fineract.infrastructure.event.external.repository.ExternalEventRepository;
+import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEvent;
+import org.apache.fineract.infrastructure.event.external.service.validation.ExternalEventDTO;
+import org.springframework.context.annotation.Profile;
+import org.springframework.data.jpa.domain.Specification;
+import org.springframework.stereotype.Service;
+
+@Service
+@Profile(FineractProfiles.TEST)
+@Slf4j
+@AllArgsConstructor
+public class InternalExternalEventService {
+
+    private final ExternalEventRepository externalEventRepository;
+
+    public void deleteAllExternalEvents() {
+        externalEventRepository.deleteAll();
+    }
+
+    public List<ExternalEventDTO> getAllExternalEvents(String idempotencyKey, String type, String category, Long aggregateRootId) {
+        List<Specification<ExternalEvent>> specifications = new ArrayList<>();
+
+        if (StringUtils.isNotEmpty(idempotencyKey)) {
+            specifications.add(hasIdempotencyKey(idempotencyKey));
+        }
+
+        if (StringUtils.isNotEmpty(type)) {
+            specifications.add(hasType(type));
+        }
+
+        if (StringUtils.isNotEmpty(category)) {
+            specifications.add(hasCategory(category));
+        }
+
+        if (aggregateRootId != null) {
+            specifications.add(hasAggregateRootId(aggregateRootId));
+        }
+
+        Specification<ExternalEvent> reducedSpecification = specifications.stream().reduce(Specification::and)
+                .orElse((Specification<ExternalEvent>) (root, query, criteriaBuilder) -> null);
+        List<ExternalEvent> externalEvents = externalEventRepository.findAll(reducedSpecification);
+
+        try {
+            return convertToReadableFormat(externalEvents);
+        } catch (ClassNotFoundException | NoSuchMethodException | InvocationTargetException | IllegalAccessException
+                | JsonProcessingException e) {
+            throw new RuntimeException("Error while converting external events to readable format", e);
+        }
+    }
+
+    private Specification<ExternalEvent> hasIdempotencyKey(String idempotencyKey) {
+        return (root, query, cb) -> cb.equal(root.get("idempotencyKey"), idempotencyKey);
+    }
+
+    private Specification<ExternalEvent> hasType(String type) {
+        return (root, query, cb) -> cb.equal(root.get("type"), type);
+    }
+
+    private Specification<ExternalEvent> hasCategory(String category) {
+        return (root, query, cb) -> cb.equal(root.get("category"), category);
+    }
+
+    private Specification<ExternalEvent> hasAggregateRootId(Long aggregateRootId) {
+        return (root, query, cb) -> cb.equal(root.get("aggregateRootId"), aggregateRootId);
+    }
+
+    private List<ExternalEventDTO> convertToReadableFormat(List<ExternalEvent> externalEvents) throws ClassNotFoundException,
+            NoSuchMethodException, InvocationTargetException, IllegalAccessException, JsonProcessingException {
+        List<ExternalEventDTO> eventMessages = new ArrayList<>();
+        for (ExternalEvent externalEvent : externalEvents) {
+            Class<?> payLoadClass = Class.forName(externalEvent.getSchema());
+            ByteBuffer byteBuffer = ByteBuffer.wrap(externalEvent.getData());
+            Method method = payLoadClass.getMethod("fromByteBuffer", ByteBuffer.class);
+            Object payLoad = method.invoke(null, byteBuffer);
+            if (externalEvent.getType().equalsIgnoreCase("BulkBusinessEvent")) {
+                Method methodToGetDatas = payLoad.getClass().getMethod("getDatas", (Class<?>) null);
+                List<BulkMessageItemV1> bulkMessages = (List<BulkMessageItemV1>) methodToGetDatas.invoke(payLoad);
+                StringBuilder bulkMessagePayload = new StringBuilder();
+                for (BulkMessageItemV1 bulkMessage : bulkMessages) {
+                    ExternalEventDTO bulkMessageData = retrieveBulkMessage(bulkMessage, externalEvent);
+                    bulkMessagePayload.append(bulkMessageData);
+                    bulkMessagePayload.append(System.lineSeparator());
+                }
+                eventMessages.add(new ExternalEventDTO(externalEvent.getId(), externalEvent.getType(), externalEvent.getCategory(),
+                        externalEvent.getCreatedAt(), toJsonMap(bulkMessagePayload.toString()), externalEvent.getBusinessDate(),
+                        externalEvent.getSchema(), externalEvent.getAggregateRootId()));
+
+            } else {
+                eventMessages.add(new ExternalEventDTO(externalEvent.getId(), externalEvent.getType(), externalEvent.getCategory(),
+                        externalEvent.getCreatedAt(), toJsonMap(payLoad.toString()), externalEvent.getBusinessDate(),
+                        externalEvent.getSchema(), externalEvent.getAggregateRootId()));
+            }
+        }
+
+        return eventMessages;
+    }
+
+    private ExternalEventDTO retrieveBulkMessage(BulkMessageItemV1 messageItem, ExternalEvent externalEvent) throws ClassNotFoundException,
+            InvocationTargetException, IllegalAccessException, NoSuchMethodException, JsonProcessingException {
+        Class<?> messageBulkMessagePayLoad = Class.forName(messageItem.getDataschema());
+        Method methodForPayLoad = messageBulkMessagePayLoad.getMethod("fromByteBuffer", ByteBuffer.class);
+        Object payLoadBulkItem = methodForPayLoad.invoke(null, messageItem.getData());
+        return new ExternalEventDTO((long) messageItem.getId(), messageItem.getType(), messageItem.getCategory(),
+                externalEvent.getCreatedAt(), toJsonMap(payLoadBulkItem.toString()), externalEvent.getBusinessDate(),
+                externalEvent.getSchema(), externalEvent.getAggregateRootId());
+    }
+
+    private Map<String, Object> toJsonMap(String json) throws JsonProcessingException {
+        ObjectMapper objectMapper = new ObjectMapper();
+        return objectMapper.readValue(json, new TypeReference<>() {});
+    }
+
+}
@@ -0,0 +1,41 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.fineract.infrastructure.event.external.service.validation;
+
+import java.time.LocalDate;
+import java.time.OffsetDateTime;
+import java.util.Map;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.ToString;
+
+@Getter
+@AllArgsConstructor
+@ToString
+public class ExternalEventDTO {
+
+    private final Long eventId;
+    private final String type;
+    private final String category;
+    private final OffsetDateTime createdAt;
+    private final Map<String, Object> payLoad;
+    private final LocalDate businessDate;
+    private final String schema;
+    private final Long aggregateRootId;
+}