completed group batch feature

2026-06-08 09:18:06 +02:00
parent c87f1e5e23
commit 24f435a09f
60 changed files with 2006 additions and 323 deletions
--- a/src/main/java/zw/qantra/tm/domain/services/JobService.java
+++ b/src/main/java/zw/qantra/tm/domain/services/JobService.java
@@ -0,0 +1,113 @@
+package zw.qantra.tm.domain.services;
+
+import io.nflow.engine.service.WorkflowInstanceService;
+import io.nflow.engine.workflow.instance.WorkflowInstance;
+import io.nflow.engine.workflow.instance.WorkflowInstanceAction;
+import lombok.RequiredArgsConstructor;
+import org.jobrunr.jobs.context.JobRunrDashboardLogger;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+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.processors.workflows.WorkflowUtils;
+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;
+
+import static org.joda.time.DateTime.now;
+
+@Component
+@RequiredArgsConstructor
+public class JobService {
+    Logger log = new JobRunrDashboardLogger(LoggerFactory.getLogger(JobService.class));
+
+    private final WorkflowInstanceService workflowInstanceService;
+    private final WorkflowUtils workflowUtils;
+    private final TransactionService transactionService;
+
+    public void checkOrphanedTransactions() {
+        List<Transaction> transactions = transactionService.getTransactionRepository()
+                .findAllByPaymentStatusAndPollingStatus(Status.SUCCESS, Status.PENDING);
+        if (!transactions.isEmpty())
+            log.info("Transaction count {}", transactions.size());
+
+        Integer attempts = 0;
+        Integer success = 0;
+        for (Transaction transaction: transactions) {
+            attempts++;
+            try {
+                // polling logic
+                WorkflowInstance update = new WorkflowInstance.Builder()
+                        .setId(transaction.getWorkflowId())
+                        .setState("pollStatus")  // The state you want to transition to
+                        .setNextActivation(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(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(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(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);
+
+                    success++;
+                }else {
+                    transaction.setPollingStatus(Status.FAILED);
+                    transaction.setIntegrationStatus(Status.FAILED);
+                    transactionService.save(transaction);
+                }
+
+            }catch (Exception e) {
+                log.debug(e.getMessage());
+            }
+        }
+
+        log.trace("Polling complete. {} transactions processed, {} successful", attempts, success);
+
+    }
+}