Merge branch 'feature/add-polling-cron' into 'release/0.0.1-RELEASE'

Feature/add polling cron

See merge request qantra/qpay-transaction-manager-spring-boot!7

src/main/java/zw/qantra/tm/domain/repositories/TransactionRepository.java

@@ -5,6 +5,7 @@ import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
 import org.springframework.stereotype.Repository;
 
 import zw.qantra.tm.domain.enums.RequestType;
+import zw.qantra.tm.domain.enums.Status;
 import zw.qantra.tm.domain.models.Transaction;
 
 import java.util.List;
@@ -16,4 +17,7 @@ public interface TransactionRepository extends JpaRepository<Transaction, UUID>,
     Transaction findByTraceAndType(String trace, RequestType type);
     Optional<Transaction> findByReference(String reference);
     List<Transaction> findAllByUserId(String uid);
-} 
+    List<Transaction> findAllByConfirmationStatusAndPaymentStatusAndIntegrationStatus(
+            Status s1, Status s2, Status s3);
+    List<Transaction> findAllByConfirmationStatusAndIntegrationStatus(Status s1, Status s2);
+}

src/main/java/zw/qantra/tm/domain/workflows/PollCronWorkflow.java

@@ -0,0 +1,123 @@
+package zw.qantra.tm.domain.workflows;
+
+import io.nflow.engine.service.WorkflowInstanceService;
+import io.nflow.engine.workflow.curated.State;
+import io.nflow.engine.workflow.definition.NextAction;
+import io.nflow.engine.workflow.definition.StateExecution;
+import io.nflow.engine.workflow.definition.WorkflowDefinition;
+import io.nflow.engine.workflow.definition.WorkflowStateType;
+import io.nflow.engine.workflow.instance.WorkflowInstance;
+import io.nflow.engine.workflow.instance.WorkflowInstanceAction;
+import org.joda.time.DateTime;
+import org.slf4j.Logger;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import zw.qantra.tm.domain.dtos.StateResponse;
+import zw.qantra.tm.domain.enums.Status;
+import zw.qantra.tm.domain.models.Transaction;
+import zw.qantra.tm.domain.services.TransactionService;
+import zw.qantra.tm.utils.LogUtils;
+import zw.qantra.tm.utils.Utils;
+
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+@Component
+public class PollCronWorkflow extends WorkflowDefinition {
+    Logger log = org.slf4j.LoggerFactory.getLogger(PollCronWorkflow.class);
+
+    @Autowired
+    private TransactionService transactionService;
+    @Autowired
+    private WorkflowInstanceService workflowInstanceService;
+    @Autowired
+    private WorkflowUtils workflowUtils;
+
+    public static final String TYPE = "pollCronWorkflow";
+
+    public static final State PROCESS = new State("process", WorkflowStateType.start);
+    private static final State ERROR = new State("error", WorkflowStateType.manual, "Error state");
+
+    public PollCronWorkflow() {
+        super(TYPE, PROCESS, ERROR);
+        permit(PROCESS, PROCESS);
+    }
+
+    public NextAction process(StateExecution execution) {
+        List<Transaction> transactions = transactionService.getTransactionRepository()
+                .findAllByConfirmationStatusAndIntegrationStatus(Status.SUCCESS, Status.PENDING);
+        log.info("Transaction count {}", transactions.size());
+
+        for (Transaction transaction: transactions) {
+            try {
+                // polling logic
+                WorkflowInstance update = new WorkflowInstance.Builder()
+                        .setId(transaction.getWorkflowId())
+                        .setState("pollStatus")  // The state you want to transition to
+                        .setNextActivation(DateTime.now())  // Schedule immediate execution
+                        .setStateVariables(new HashMap<>(){{ put("body", Utils.toJson(transaction)); }})
+                        .setStateText("Resumed from manual state")
+                        .build();
+
+                WorkflowInstanceAction action = new WorkflowInstanceAction.Builder(update)
+                        .setType(WorkflowInstanceAction.WorkflowActionType.externalChange)
+                        .setExecutionEnd(DateTime.now())
+                        .build();
+
+                // Apply the update
+                workflowInstanceService.updateWorkflowInstance(update, action);
+
+                CompletableFuture<Object> future = workflowUtils
+                        .waitForState(transaction.getWorkflowId(), new String[]{"done", "failed"},
+                                15000, 50);
+                Object response = future.get();
+                LogUtils.logPrettyJson(log, "polling result", response);
+
+                StateResponse wfResp = Utils.deepCopy(response, StateResponse.class);
+                LinkedHashMap tran = (LinkedHashMap) wfResp.getBody();
+                if(tran == null)
+                    break;
+
+                // only attempt integration logic if polling succeeded
+                if(tran.get("status").equals(Status.SUCCESS.name())) {
+                    // integration logic
+                    WorkflowInstance update1 = new WorkflowInstance.Builder()
+                            .setId(transaction.getWorkflowId())
+                            .setState("integration")  // The state you want to transition to
+                            .setNextActivation(DateTime.now())  // Schedule immediate execution
+                            .setStateVariables(new HashMap<>(){{ put("body", Utils.toJson(tran)); }})
+                            .setStateText("Resumed from manual state")
+                            .build();
+
+                    WorkflowInstanceAction action1 = new WorkflowInstanceAction.Builder(update1)
+                            .setType(WorkflowInstanceAction.WorkflowActionType.externalChange)
+                            .setExecutionEnd(DateTime.now())
+                            .build();
+
+                    // Apply the update
+                    workflowInstanceService.updateWorkflowInstance(update1, action1);
+
+                    CompletableFuture<Object> future1 = workflowUtils
+                            .waitForState(transaction.getWorkflowId(), new String[]{"purchaseInvoicePaymentSuccess", "failed"},
+                                    15000, 50);
+                    Object response1 = future1.get();
+                    LogUtils.logPrettyJson(log, "integration result", response1);
+                }else {
+                    transaction.setPollingStatus(Status.FAILED);
+                    transaction.setIntegrationStatus(Status.FAILED);
+                    transactionService.save(transaction);
+                }
+
+            }catch (Exception e) {
+                log.error(e.getMessage());
+            }
+        }
+
+        return NextAction.moveToStateAfter(
+                PROCESS, DateTime.now().plusMinutes(2),
+                String.format("Cron complete on %s transactions", transactions.size()));
+    }
+
+}

src/main/java/zw/qantra/tm/domain/workflows/StartupApplicationListener.java

@@ -0,0 +1,35 @@
+package zw.qantra.tm.domain.workflows;
+
+import io.nflow.engine.service.WorkflowInstanceService;
+import io.nflow.engine.workflow.instance.WorkflowInstanceFactory;
+import lombok.RequiredArgsConstructor;
+import org.joda.time.DateTime;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.event.ContextRefreshedEvent;
+import org.springframework.stereotype.Component;
+import zw.qantra.tm.utils.Utils;
+
+@Component
+@RequiredArgsConstructor
+public class StartupApplicationListener implements ApplicationListener<ContextRefreshedEvent> {
+
+    private final Logger log = LoggerFactory.getLogger(this.getClass());
+
+    private final WorkflowInstanceService workflowInstanceService;
+    private final WorkflowInstanceFactory workflowInstanceFactory;
+
+    @Override public void onApplicationEvent(ContextRefreshedEvent event) {
+        var instance = workflowInstanceFactory.newWorkflowInstanceBuilder()
+                .setType(PollCronWorkflow.TYPE)
+                .setExternalId("PollCronWorkflow")
+                .setNextActivation(DateTime.now())
+                .build();
+
+        long workflowId = workflowInstanceService.insertWorkflowInstance(instance);
+
+        log.info("Cron workflow ID {} started", workflowId);
+    }
+}

src/main/java/zw/qantra/tm/domain/workflows/TransactionWorkflow.java

@@ -24,7 +24,7 @@ import java.util.LinkedHashMap;
 
 @Component
 public class TransactionWorkflow extends WorkflowDefinition {
-    Logger log = org.slf4j.LoggerFactory.getLogger(RegistrationWorkflow.class);
+    Logger log = org.slf4j.LoggerFactory.getLogger(TransactionWorkflow.class);
 
     @Autowired
     private WorkflowHandlerFactory workflowHandlerFactory;
@@ -256,7 +256,7 @@ public class TransactionWorkflow extends WorkflowDefinition {
         } catch (Exception e) {
             log.info(e.getMessage());
             execution.setVariable("body", Utils.toJson(transaction));
-            return NextAction.retryAfter(DateTime.now().plusSeconds(30), e.getMessage());
+            return NextAction.moveToState(TRAN_FAILED, "Polling failed");
         }
     }