From bbc545f2907004c5231f0f639b5b07658441252a Mon Sep 17 00:00:00 2001 From: Vusumuzi Khoza Date: Thu, 4 Dec 2025 09:29:22 +0200 Subject: [PATCH 1/2] initial progress --- .../repositories/TransactionRepository.java | 5 +- .../tm/domain/workflows/PollCronWorkflow.java | 122 ++++++++++++++++++ .../workflows/StartupApplicationListener.java | 35 +++++ .../domain/workflows/TransactionWorkflow.java | 2 +- 4 files changed, 162 insertions(+), 2 deletions(-) create mode 100644 src/main/java/zw/qantra/tm/domain/workflows/PollCronWorkflow.java create mode 100644 src/main/java/zw/qantra/tm/domain/workflows/StartupApplicationListener.java diff --git a/src/main/java/zw/qantra/tm/domain/repositories/TransactionRepository.java b/src/main/java/zw/qantra/tm/domain/repositories/TransactionRepository.java index c81e709..1351a57 100644 --- a/src/main/java/zw/qantra/tm/domain/repositories/TransactionRepository.java +++ b/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,6 @@ public interface TransactionRepository extends JpaRepository, Transaction findByTraceAndType(String trace, RequestType type); Optional findByReference(String reference); List findAllByUserId(String uid); -} \ No newline at end of file + List findAllByConfirmationStatusAndPaymentStatusAndIntegrationStatus( + Status s1, Status s2, Status s3); +} \ No newline at end of file diff --git a/src/main/java/zw/qantra/tm/domain/workflows/PollCronWorkflow.java b/src/main/java/zw/qantra/tm/domain/workflows/PollCronWorkflow.java new file mode 100644 index 0000000..c6e2121 --- /dev/null +++ b/src/main/java/zw/qantra/tm/domain/workflows/PollCronWorkflow.java @@ -0,0 +1,122 @@ +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.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.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 transactions = transactionService.getTransactionRepository() + .findAllByConfirmationStatusAndPaymentStatusAndIntegrationStatus( + Status.SUCCESS, Status.PENDING, 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 future = workflowUtils + .waitForState(transaction.getWorkflowId(), new String[]{"done", "failed"}, + 15000, 50); + Object response = future.get(); + LogUtils.logPrettyJson(log, "polling result", response); + + Transaction tran = Utils.deepCopy(response, Transaction.class); + // only attempt integration logic if polling succeeded + if(tran.getStatus().equals(Status.SUCCESS)) { + // 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(transaction)); }}) + .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 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()); + + transaction.setPollingStatus(Status.FAILED); + transaction.setIntegrationStatus(Status.FAILED); + transactionService.save(transaction); + } + } + + return NextAction.moveToStateAfter( + PROCESS, DateTime.now().plusMinutes(2), + String.format("Cron complete on %s transactions", transactions.size())); + } + +} diff --git a/src/main/java/zw/qantra/tm/domain/workflows/StartupApplicationListener.java b/src/main/java/zw/qantra/tm/domain/workflows/StartupApplicationListener.java new file mode 100644 index 0000000..1981e11 --- /dev/null +++ b/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 { + + 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); + } +} diff --git a/src/main/java/zw/qantra/tm/domain/workflows/TransactionWorkflow.java b/src/main/java/zw/qantra/tm/domain/workflows/TransactionWorkflow.java index c57ffc1..879b339 100644 --- a/src/main/java/zw/qantra/tm/domain/workflows/TransactionWorkflow.java +++ b/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; From 97dae2e9ee8c98c3efd33e884609104d503a59c4 Mon Sep 17 00:00:00 2001 From: Vusumuzi Khoza Date: Mon, 8 Dec 2025 14:01:51 +0200 Subject: [PATCH 2/2] completed poll logic --- .../repositories/TransactionRepository.java | 1 + .../tm/domain/workflows/PollCronWorkflow.java | 19 ++++++++++--------- .../workflows/StartupApplicationListener.java | 4 ++-- .../domain/workflows/TransactionWorkflow.java | 2 +- 4 files changed, 14 insertions(+), 12 deletions(-) diff --git a/src/main/java/zw/qantra/tm/domain/repositories/TransactionRepository.java b/src/main/java/zw/qantra/tm/domain/repositories/TransactionRepository.java index 1351a57..281d2fb 100644 --- a/src/main/java/zw/qantra/tm/domain/repositories/TransactionRepository.java +++ b/src/main/java/zw/qantra/tm/domain/repositories/TransactionRepository.java @@ -19,4 +19,5 @@ public interface TransactionRepository extends JpaRepository, List findAllByUserId(String uid); List findAllByConfirmationStatusAndPaymentStatusAndIntegrationStatus( Status s1, Status s2, Status s3); + List findAllByConfirmationStatusAndIntegrationStatus(Status s1, Status s2); } \ No newline at end of file diff --git a/src/main/java/zw/qantra/tm/domain/workflows/PollCronWorkflow.java b/src/main/java/zw/qantra/tm/domain/workflows/PollCronWorkflow.java index c6e2121..58e4efa 100644 --- a/src/main/java/zw/qantra/tm/domain/workflows/PollCronWorkflow.java +++ b/src/main/java/zw/qantra/tm/domain/workflows/PollCronWorkflow.java @@ -12,6 +12,7 @@ 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; @@ -19,6 +20,7 @@ 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; @@ -45,8 +47,7 @@ public class PollCronWorkflow extends WorkflowDefinition { public NextAction process(StateExecution execution) { List transactions = transactionService.getTransactionRepository() - .findAllByConfirmationStatusAndPaymentStatusAndIntegrationStatus( - Status.SUCCESS, Status.PENDING, Status.PENDING); + .findAllByConfirmationStatusAndIntegrationStatus(Status.SUCCESS, Status.PENDING); log.info("Transaction count {}", transactions.size()); for (Transaction transaction: transactions) { @@ -74,15 +75,19 @@ public class PollCronWorkflow extends WorkflowDefinition { Object response = future.get(); LogUtils.logPrettyJson(log, "polling result", response); - Transaction tran = Utils.deepCopy(response, Transaction.class); + 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.getStatus().equals(Status.SUCCESS)) { + 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(transaction)); }}) + .setStateVariables(new HashMap<>(){{ put("body", Utils.toJson(tran)); }}) .setStateText("Resumed from manual state") .build(); @@ -107,10 +112,6 @@ public class PollCronWorkflow extends WorkflowDefinition { }catch (Exception e) { log.error(e.getMessage()); - - transaction.setPollingStatus(Status.FAILED); - transaction.setIntegrationStatus(Status.FAILED); - transactionService.save(transaction); } } diff --git a/src/main/java/zw/qantra/tm/domain/workflows/StartupApplicationListener.java b/src/main/java/zw/qantra/tm/domain/workflows/StartupApplicationListener.java index 1981e11..473c091 100644 --- a/src/main/java/zw/qantra/tm/domain/workflows/StartupApplicationListener.java +++ b/src/main/java/zw/qantra/tm/domain/workflows/StartupApplicationListener.java @@ -28,8 +28,8 @@ public class StartupApplicationListener implements ApplicationListener