diff --git a/src/main/java/com/iemr/flw/repo/iemr/DiagnosticOrderRepo.java b/src/main/java/com/iemr/flw/repo/iemr/DiagnosticOrderRepo.java index bad44ff2..86e811c8 100644 --- a/src/main/java/com/iemr/flw/repo/iemr/DiagnosticOrderRepo.java +++ b/src/main/java/com/iemr/flw/repo/iemr/DiagnosticOrderRepo.java @@ -26,21 +26,19 @@ Optional findFirstByBeneficiaryIdAndOrderTypeAndDeletedFalseOrd @Query("SELECT o FROM DiagnosticOrder o WHERE o.beneficiaryId = :beneficiaryId AND o.deleted = false ORDER BY o.createdDate DESC") List findByBeneficiaryId(@Param("beneficiaryId") Long beneficiaryId); - // Polling and the daily close only touch orders created within [dayStart, dayEnd) — i.e. today's. @Query("SELECT o FROM DiagnosticOrder o WHERE o.orderType = 'XRAY_CHEST' " + "AND o.status IN ('PENDING', 'IN_PROGRESS') " + "AND o.deleted = false " + - "AND o.createdDate >= :dayStart AND o.createdDate < :dayEnd " + "ORDER BY o.lastPolledAt ASC NULLS FIRST") - List findXrayDueForPoll(@Param("dayStart") Timestamp dayStart, @Param("dayEnd") Timestamp dayEnd); + List findXrayDueForPoll(); @Query("SELECT o FROM DiagnosticOrder o WHERE o.orderType IN ('MTB', 'MTB_PLUS', 'MDR_RIF') " + "AND o.status IN ('PENDING', 'IN_PROGRESS') " + "AND o.deleted = false " + - "AND o.createdDate >= :dayStart AND o.createdDate < :dayEnd " + "ORDER BY o.lastPolledAt ASC NULLS FIRST") - List findTrueNatDueForPoll(@Param("dayStart") Timestamp dayStart, @Param("dayEnd") Timestamp dayEnd); + List findTrueNatDueForPoll(); + // The (disabled) daily close only touches orders created within [dayStart, dayEnd) — i.e. today's. @Query("SELECT o FROM DiagnosticOrder o WHERE o.status IN ('PENDING', 'IN_PROGRESS') AND o.deleted = false " + "AND o.createdDate >= :dayStart AND o.createdDate < :dayEnd") List findAllOutstandingOrders(@Param("dayStart") Timestamp dayStart, @Param("dayEnd") Timestamp dayEnd); diff --git a/src/main/java/com/iemr/flw/service/DiagnosticOrderService.java b/src/main/java/com/iemr/flw/service/DiagnosticOrderService.java index 250aa143..f2f39dba 100644 --- a/src/main/java/com/iemr/flw/service/DiagnosticOrderService.java +++ b/src/main/java/com/iemr/flw/service/DiagnosticOrderService.java @@ -23,8 +23,18 @@ public interface DiagnosticOrderService { DiagnosticOrderResultDto processResult(DiagnosticOrder order, DiagnosticPollResult result) throws Exception; + /** + * Polls the vendor once for this order. The order is re-read first and skipped (returns null) if it + * is no longer PENDING/IN_PROGRESS, e.g. a manual result landed after the scheduler loaded it. + */ DiagnosticPollResult pollOnce(DiagnosticOrder order) throws Exception; + /** + * Daily cutoff: marks this order CLOSED with the given reason and notifies the vendor, unless it was + * resolved (e.g. by a manual result) after the scheduler loaded it. + */ + void expireOrder(DiagnosticOrder order, String reason) throws Exception; + DiagnosticOrderResultDto triggerManualPoll(Long beneficiaryId, String orderType, Long visitCode) throws Exception; DiagnosticOrder retryPoll(Long beneficiaryId, String orderType, Long visitCode) throws Exception; diff --git a/src/main/java/com/iemr/flw/service/impl/DiagnosticOrderServiceImpl.java b/src/main/java/com/iemr/flw/service/impl/DiagnosticOrderServiceImpl.java index 25747002..54121070 100644 --- a/src/main/java/com/iemr/flw/service/impl/DiagnosticOrderServiceImpl.java +++ b/src/main/java/com/iemr/flw/service/impl/DiagnosticOrderServiceImpl.java @@ -62,6 +62,12 @@ public class DiagnosticOrderServiceImpl implements DiagnosticOrderService { DiagnosticOrderStatus.FAILED.name(), DiagnosticOrderStatus.CLOSED.name()); + // Statuses the scheduler polls and the daily cutoff expires — matches findXrayDueForPoll / + // findTrueNatDueForPoll / findAllOutstandingOrders. + private static final Set OUTSTANDING_STATUSES = Set.of( + DiagnosticOrderStatus.PENDING.name(), + DiagnosticOrderStatus.IN_PROGRESS.name()); + // Invalid test outcomes: the result is still recorded (result row, tb_suspected write-back, documents) // but the order is CLOSED instead of COMPLETED, so a retest can be pushed. The vendor reports an // invalid MTB/MDR_RIF run as "Error-2"; a manual entry uses "Invalid/Error". @@ -163,6 +169,20 @@ private T withBeneficiaryLockBlocking(Long beneficiaryId, Callable action } } + // For the poll scheduler: never waits. A busy beneficiary (user push/manual entry in progress) is + // simply left for the next tick, rather than stalling the single scheduler thread behind it. + private Optional withBeneficiaryLockIfFree(Long beneficiaryId, Callable action) throws Exception { + Lock lock = beneficiaryLock(beneficiaryId); + if (!lock.tryLock()) { + return Optional.empty(); + } + try { + return Optional.ofNullable(action.call()); + } finally { + lock.unlock(); + } + } + private Lock beneficiaryLock(Long beneficiaryId) { if (beneficiaryId == null) { throw new IllegalArgumentException("beneficiaryId is required"); @@ -541,8 +561,67 @@ private void ingestAssets(DiagnosticOrder order, DiagnosticPollResult pollResult } } + // The scheduler loads its due-for-poll batch at the start of a tick and polls each order in turn, so + // by the time an order's turn comes it may have been completed/closed by a manual entry. Polling and + // saving that stale copy would overwrite the manual result (whole-row save, no @Version), so the + // order is re-read under the beneficiary lock and skipped unless it is still awaiting a result. + // Failures are recorded on that same fresh row, inside the lock, for the same reason. @Override public DiagnosticPollResult pollOnce(DiagnosticOrder order) throws Exception { + Optional polled = withBeneficiaryLockIfFree(order.getBeneficiaryId(), () -> { + Optional current = findOutstandingOrder(order.getId()); + if (current.isEmpty()) { + logger.info("Skipping poll for orderId={}: no longer PENDING/IN_PROGRESS", order.getId()); + return null; + } + DiagnosticOrder fresh = current.get(); + try { + return pollOnceLocked(fresh); + } catch (Exception e) { + fresh.setRetryCount(fresh.getRetryCount() + 1); + fresh.setLastPolledAt(new Timestamp(System.currentTimeMillis())); + fresh.setErrorMessage(e.getMessage()); + fresh.setModifiedBy("SYSTEM"); + fresh.setProcessed("N"); + diagnosticOrderRepo.save(fresh); + throw e; + } + }); + if (polled.isEmpty()) { + logger.debug("Poll for orderId={} skipped or deferred (beneficiary busy / order resolved)", order.getId()); + } + return polled.orElse(null); + } + + // Same stale-copy problem as pollOnce: the daily expiry batch is loaded up front, so each order is + // re-read under the lock and only closed if it is still outstanding. + @Override + public void expireOrder(DiagnosticOrder order, String reason) throws Exception { + withBeneficiaryLock(order.getBeneficiaryId(), () -> { + Optional current = findOutstandingOrder(order.getId()); + if (current.isEmpty()) { + logger.info("Skipping expiry for orderId={}: no longer PENDING/IN_PROGRESS", order.getId()); + return null; + } + DiagnosticOrder fresh = current.get(); + fresh.setStatus(DiagnosticOrderStatus.CLOSED.name()); + fresh.setReasonToClose(reason); + fresh.setLastPolledAt(new Timestamp(System.currentTimeMillis())); + fresh.setModifiedBy("SYSTEM"); + fresh.setProcessed("N"); + diagnosticOrderRepo.save(fresh); + notifyProviderOrderClosed(fresh, reason); + return null; + }); + } + + private Optional findOutstandingOrder(Long orderId) { + return diagnosticOrderRepo.findById(orderId) + .filter(o -> !Boolean.TRUE.equals(o.getDeleted())) + .filter(o -> OUTSTANDING_STATUSES.contains(o.getStatus())); + } + + private DiagnosticPollResult pollOnceLocked(DiagnosticOrder order) throws Exception { // A null pushResponseJson means the order was never actually confirmed as pushed to the // vendor (e.g. a stale/legacy row) — polling the vendor for it would be meaningless, so // fail it outright instead of burning a vendor call every tick. @@ -601,15 +680,33 @@ private DiagnosticOrder resolveOrder(Long beneficiaryId, String orderType, Long @Override public DiagnosticOrderResultDto triggerManualPoll(Long beneficiaryId, String orderType, Long visitCode) throws Exception { - DiagnosticOrder order = resolveOrder(beneficiaryId, orderType, visitCode); - DiagnosticProvider provider = providerFactory.getProvider(order.getProviderCode()); - DiagnosticPollResult pollResult = provider.pollResult(order, true); - boolean invalid = closeIfInvalidPolledResult(order, pollResult); - return processResult(order, pollResult, invalid, "SYSTEM"); + // Locked, and the order read inside the lock, so this can't interleave with a manual entry or a + // scheduler poll for the same beneficiary and save over what they just wrote. + return withBeneficiaryLock(beneficiaryId, () -> { + DiagnosticOrder order = resolveOrder(beneficiaryId, orderType, visitCode); + if (DiagnosticOrderStatus.MANUAL_ENTRY.name().equals(order.getStatus())) { + throw new IllegalStateException("Cannot poll a MANUAL_ENTRY order — no vendor is involved, " + + "submit the result via manualResult instead"); + } + if (order.getManuallyEnteredBy() != null) { + throw new IllegalStateException("Cannot poll this order: its result was entered manually by " + + order.getManuallyEnteredBy() + " and a vendor poll would overwrite it"); + } + DiagnosticProvider provider = providerFactory.getProvider(order.getProviderCode()); + DiagnosticPollResult pollResult = provider.pollResult(order, true); + boolean invalid = closeIfInvalidPolledResult(order, pollResult); + return processResult(order, pollResult, invalid, "SYSTEM"); + }); } @Override public DiagnosticOrder retryPoll(Long beneficiaryId, String orderType, Long visitCode) throws Exception { + // Locked so the terminal-status check and the reset to PENDING can't straddle a manual entry + // that completes the order in between (which would reopen it for polling). + return withBeneficiaryLock(beneficiaryId, () -> retryPollLocked(beneficiaryId, orderType, visitCode)); + } + + private DiagnosticOrder retryPollLocked(Long beneficiaryId, String orderType, Long visitCode) throws Exception { DiagnosticOrder order = resolveOrder(beneficiaryId, orderType, visitCode); String status = order.getStatus(); @@ -730,6 +827,12 @@ public DiagnosticOrderResultDto submitManualResult(ManualDiagnosticResultRequest return toResultDto(closed); } + return withBeneficiaryLock(request.getBeneficiaryId(), + () -> submitManualResultLocked(request, actingUserName)); + } + + private DiagnosticOrderResultDto submitManualResultLocked(ManualDiagnosticResultRequestDto request, + String actingUserName) throws Exception { DiagnosticOrder order = findLatestOrder(request.getBeneficiaryId(), request.getOrderType()); if (DiagnosticOrderStatus.COMPLETED.name().equals(order.getStatus())) { throw new IllegalStateException("Cannot submit manual result: order is already COMPLETED (beneficiaryId=" diff --git a/src/main/java/com/iemr/flw/service/impl/DiagnosticPollSchedulerService.java b/src/main/java/com/iemr/flw/service/impl/DiagnosticPollSchedulerService.java index 07f04110..1a36e8cd 100644 --- a/src/main/java/com/iemr/flw/service/impl/DiagnosticPollSchedulerService.java +++ b/src/main/java/com/iemr/flw/service/impl/DiagnosticPollSchedulerService.java @@ -10,6 +10,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; +import org.springframework.web.client.ResourceAccessException; import java.sql.Timestamp; import java.time.LocalDate; @@ -33,13 +34,13 @@ public class DiagnosticPollSchedulerService { @Scheduled(fixedDelayString = "${diagnostic.poll.xray.tick-ms:15000}") public void pollXrayOrders() { pollDueOrders("XRAY", DiagnosticOrderType.XRAY_CHEST, - () -> diagnosticOrderRepo.findXrayDueForPoll(startOfToday(), startOfTomorrow())); + diagnosticOrderRepo::findXrayDueForPoll); } @Scheduled(fixedDelayString = "${diagnostic.poll.truenat.tick-ms:60000}") public void pollTrueNatOrders() { pollDueOrders("TrueNat", DiagnosticOrderType.MTB, - () -> diagnosticOrderRepo.findTrueNatDueForPoll(startOfToday(), startOfTomorrow())); + diagnosticOrderRepo::findTrueNatDueForPoll); } private void pollDueOrders(String label, DiagnosticOrderType orderType, Supplier> dueOrders) { @@ -88,7 +89,7 @@ private void pollDueOrders(String label, DiagnosticOrderType orderType, Supplier // } // } - // Polling and the daily close only handle orders created today. JVM default zone on purpose: it's the + // The (disabled) daily close only handles orders created today. JVM default zone on purpose: it's the // same clock created_date is stamped with (@CreationTimestamp), so the day boundaries line up. private static Timestamp startOfToday() { return Timestamp.valueOf(LocalDate.now().atStartOfDay()); @@ -103,16 +104,15 @@ private boolean noActiveVendor(DiagnosticOrderType orderType) { return providerCode == null || providerCode.isBlank(); } - // Used only by the (disabled) auto-closure job above. + // Used only by the (disabled) auto-closure job above. expireOrder re-reads the order under the + // beneficiary lock, so an order completed manually after this batch was loaded is left alone. // private void expire(DiagnosticOrder order) { // String reason = "Order validity expired for the day " + LocalDate.now(ZoneId.of("Asia/Kolkata")); - // order.setStatus(DiagnosticOrderStatus.CLOSED.name()); - // order.setReasonToClose(reason); - // order.setLastPolledAt(new Timestamp(System.currentTimeMillis())); - // order.setModifiedBy("SYSTEM"); - // order.setProcessed("N"); - // diagnosticOrderRepo.save(order); - // diagnosticOrderService.notifyProviderOrderClosed(order, reason); + // try { + // diagnosticOrderService.expireOrder(order, reason); + // } catch (Exception e) { + // logger.error("Expiry failed for orderId={}: {}", order.getId(), e.getMessage()); + // } // } // Returns false only when the vendor couldn't be reached at all (connect/read timeout, refused @@ -120,19 +120,14 @@ private boolean noActiveVendor(DiagnosticOrderType orderType) { // (bad data, vendor-side rejection), so it's recorded and the tick carries on. private boolean pollSingle(DiagnosticOrder order) { try { - // pollOnce now saves the result/order fields and ingests any documents itself — - // nothing left to do here on success. diagnosticOrderService.pollOnce(order); return true; + } catch (ResourceAccessException e) { + logger.error("Vendor unreachable polling orderId={}: {}", order.getId(), e.getMessage()); + return false; } catch (Exception e) { logger.error("Poll failed for orderId={}: {}", order.getId(), e.getMessage()); - order.setRetryCount(order.getRetryCount() + 1); - order.setLastPolledAt(new Timestamp(System.currentTimeMillis())); - order.setErrorMessage(e.getMessage()); - order.setModifiedBy("SYSTEM"); - order.setProcessed("N"); - diagnosticOrderRepo.save(order); - return false; + return true; } }