122 lines
5.4 KiB
Java
122 lines
5.4 KiB
Java
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;
|
|
private final SettingService settingService;
|
|
|
|
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 {
|
|
int maxRetries = settingService.getIntegerSetting("MAX_RETRIES");
|
|
int retries = transaction.getRetries() == null ? 0 : transaction.getRetries();
|
|
|
|
if (retries >= maxRetries)
|
|
break;
|
|
|
|
transaction.setRetries(retries + 1);
|
|
// 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);
|
|
|
|
}
|
|
}
|