Remove duplicate transactions from input CSV (#63)

* remove duplicate transactions

* enable ordering

* camel route works when ordering is enabled

* GOV-325: set note for duplicate transactions

* GOV-325: perform deduplication when enabled

* add null pointer checks

* address review changes

* address review comments

Author: arkadasfynarfin

src/main/java/org/mifos/processor/bulk/camel/routes/OrderingRoute.java

@@ -60,6 +60,9 @@ public void configure() {
                             case "account_number":
                                 key = transaction.getAccountNumber();
                                 break;
+                            case "payee_identifier":
+                                key = transaction.getPayeeIdentifier();
+                                break;
                             case "amount":
                                 key = transaction.getAmount();
                                 break;

src/main/java/org/mifos/processor/bulk/zeebe/worker/OrderingWorker.java

@@ -3,11 +3,17 @@
 import org.apache.camel.Exchange;
 import org.apache.camel.support.DefaultExchange;
 import org.mifos.processor.bulk.camel.routes.RouteId;
+import org.mifos.processor.bulk.schema.Transaction;
 import org.springframework.stereotype.Component;
 
+import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.HashSet;
+import java.util.Objects;
 
 import static org.mifos.processor.bulk.camel.config.CamelProperties.SERVER_FILE_NAME;
+import static org.mifos.processor.bulk.camel.config.CamelProperties.TRANSACTION_LIST;
 import static org.mifos.processor.bulk.zeebe.ZeebeVariables.*;
 
 @Component
@@ -26,26 +32,83 @@ public void setup() {
         newWorker(Worker.ORDERING, (client, job) -> {
             logger.debug("Job '{}' started from process '{}' with key {}", job.getType(), job.getBpmnProcessId(), job.getKey());
             Map<String, Object> variables = job.getVariablesAsMap();
+            Exchange exchange = new DefaultExchange(camelContext);
+
             if (workerConfig.isOrderingWorkerEnabled) {
                 variables.put(ORDERING_FAILED, false);
+                String filename = (String) variables.get(FILE_NAME);
+                exchange.setProperty(SERVER_FILE_NAME, filename);
+
+                try {
+                    sendToCamelRoute(RouteId.ORDERING, exchange);
+                    assert !exchange.getProperty(ORDERING_FAILED, Boolean.class);
+                } catch (Exception e) {
+                    variables.put(ORDERING_FAILED, true);
+                }
+                variables.put(ORDERING_FAILED, false);
+                variables.put(ORDERED_BY, exchange.getProperty(ORDERED_BY));
             }
 
-            String filename = (String) variables.get(FILE_NAME);
-            Exchange exchange = new DefaultExchange(camelContext);
-            exchange.setProperty(SERVER_FILE_NAME, filename);
+            if(workerConfig.isTransactionDeduplicationEnabled){
+                List<Transaction> transactionList = exchange.getProperty(TRANSACTION_LIST, List.class);
+                removeDuplicates(transactionList, workerConfig.isOrderingWorkerEnabled);
+                variables.put(TRANSACTION_LIST, transactionList);
+            }
+            client.newCompleteCommand(job.getKey()).variables(variables).send();
+        });
+    }
+
+    private void removeDuplicates(List<Transaction> transactionList, boolean orderingEnabled){
+        if(orderingEnabled){
+            removeDuplicatesIfOrderingEnabled(transactionList);
+            return;
+        }
+        removeDuplicatesIfOrderingDisabled(transactionList);
+    }
+
+    private void removeDuplicatesIfOrderingEnabled(List<Transaction> transactionList){
 
-            try {
-                sendToCamelRoute(RouteId.ORDERING, exchange);
-                assert !exchange.getProperty(ORDERING_FAILED, Boolean.class);
-            } catch (Exception e) {
-                variables.put(ORDERING_FAILED, true);
+        for (int i = 0; i <transactionList.size()-1; i++) {
+            Transaction currentTransaction = transactionList.get(i);
+            Transaction nextTransaction = transactionList.get(i + 1);
+
+            if (currentTransaction == null || nextTransaction == null) {
+                continue;
+            }
+            String currentPayeeDetail = fetchPayeeDetail(currentTransaction);
+            String nextPayeeDetail = fetchPayeeDetail(nextTransaction);
+
+            if (currentPayeeDetail.equals(nextPayeeDetail)) {
+                currentTransaction.setNote("Duplicate transaction.");
             }
+        }
+    }
 
-            variables.put(ORDERING_FAILED, false);
-            variables.put(ORDERED_BY, exchange.getProperty(ORDERED_BY));
+    private void removeDuplicatesIfOrderingDisabled(List<Transaction> transactionList){
+        Set<String> set = new HashSet<>();
 
-            client.newCompleteCommand(job.getKey()).variables(variables).send();
-        });
+        if(Objects.isNull(transactionList)){
+            return;
+        }
+
+        for(Transaction transaction : transactionList){
+            String payeeDetail = fetchPayeeDetail(transaction);
+            if(set.contains(payeeDetail)){
+                transaction.setNote("Duplicate transaction.");
+            }
+            else{
+                set.add(payeeDetail);
+            }
+        }
+    }
+
+    private String fetchPayeeDetail(Transaction transaction){
+        String payeeIdentifier = transaction.getPayeeIdentifier();
+        String payeeIdentifierType = transaction.getPayeeIdentifierType();
+        String amount = transaction.getAmount();
+        String currency = transaction.getCurrency();
+
+        return String.format("%s%s%s%s", payeeIdentifier, payeeIdentifierType, amount, currency);
     }
 
 }

src/main/java/org/mifos/processor/bulk/zeebe/worker/WorkerConfig.java

@@ -27,4 +27,7 @@ public class WorkerConfig {
     @Value("${config.completion-threshold-check.enable}")
     public boolean isCompletionThresholdCheckEnabled;
 
+    @Value("config.deduplication.enabled")
+    public boolean isTransactionDeduplicationEnabled;
+
 }

src/main/resources/application.yaml

@@ -84,8 +84,8 @@ config:
   approval:
     enable: true
   ordering:
-    enable: false
-    field: "payment_mode"
+    enable: true
+    field: "payee_identifier"
   splitting:
     enable: false
     sub-batch-size: 5
@@ -101,6 +101,8 @@ config:
     completion-threshold: 95 # in percentage
     max-retry: 4 #can be as high as 30
     delay: 60 # in seconds
+  deduplication:
+    enabled: true
 
 callback:
   max-retry: 3