diff --git a/payment-orchestrator/payment-orchestrator/src/main/java/com/stablecoin/payments/orchestrator/domain/workflow/PaymentWorkflowImpl.java b/payment-orchestrator/payment-orchestrator/src/main/java/com/stablecoin/payments/orchestrator/domain/workflow/PaymentWorkflowImpl.java index 60a90625..b16c0dcc 100644 --- a/payment-orchestrator/payment-orchestrator/src/main/java/com/stablecoin/payments/orchestrator/domain/workflow/PaymentWorkflowImpl.java +++ b/payment-orchestrator/payment-orchestrator/src/main/java/com/stablecoin/payments/orchestrator/domain/workflow/PaymentWorkflowImpl.java @@ -6,6 +6,7 @@ import com.stablecoin.payments.orchestrator.domain.workflow.activity.FxLockActivity; import com.stablecoin.payments.orchestrator.domain.workflow.activity.FxLockRequest; import com.stablecoin.payments.orchestrator.domain.workflow.activity.FxLockResult; +import com.stablecoin.payments.orchestrator.domain.workflow.activity.FxReleaseRequest; import com.stablecoin.payments.orchestrator.domain.workflow.dto.CancelRequest; import com.stablecoin.payments.orchestrator.domain.workflow.dto.ChainConfirmedSignal; import com.stablecoin.payments.orchestrator.domain.workflow.dto.FiatCollectedSignal; @@ -19,6 +20,7 @@ import java.time.Duration; import java.util.ArrayDeque; import java.util.Deque; +import java.util.UUID; /** * Deterministic Temporal workflow implementation for the cross-border payment saga. @@ -202,11 +204,14 @@ public String getPaymentState() { return currentState; } + private static final String RELEASE_FX_LOCK_PREFIX = "RELEASE_FX_LOCK:"; + /** * Runs compensation stack in LIFO order and returns a FAILED result. *

- * Currently logs compensation steps. Actual compensation activities - * (e.g., release FX lock, refund fiat) will be added in Phase 3. + * Each compensation step is parsed and the corresponding activity is invoked. + * Compensation failures are logged but do not prevent remaining compensations + * from executing — the saga always reaches FAILED state. */ private PaymentResult handleCancellation(PaymentRequest request) { currentState = "COMPENSATING"; @@ -218,11 +223,25 @@ private PaymentResult handleCancellation(PaymentRequest request) { while (!compensationStack.isEmpty()) { var step = compensationStack.pop(); log.info("Compensating step: {} for paymentId={}", step, request.paymentId()); - // Phase 3: execute compensation activities here - // e.g., if step starts with "RELEASE_FX_LOCK:" → call fxLockActivity.releaseLock(quoteId) + executeCompensationStep(step, request.paymentId(), reason); } currentState = "FAILED"; return PaymentResult.failed(request.paymentId(), "Cancelled: " + reason); } + + private void executeCompensationStep(String step, UUID paymentId, String reason) { + try { + if (step.startsWith(RELEASE_FX_LOCK_PREFIX)) { + var lockId = UUID.fromString(step.substring(RELEASE_FX_LOCK_PREFIX.length())); + fxLockActivity.releaseLock(new FxReleaseRequest(lockId, paymentId, reason)); + log.info("Successfully released FX lock {} for paymentId={}", lockId, paymentId); + } else { + log.warn("Unknown compensation step: {} for paymentId={}", step, paymentId); + } + } catch (Exception e) { + log.error("Compensation step failed: {} for paymentId={}, error={}", + step, paymentId, e.getMessage(), e); + } + } } diff --git a/payment-orchestrator/payment-orchestrator/src/test/java/com/stablecoin/payments/orchestrator/domain/workflow/PaymentWorkflowTest.java b/payment-orchestrator/payment-orchestrator/src/test/java/com/stablecoin/payments/orchestrator/domain/workflow/PaymentWorkflowTest.java index c5f8abad..71d22301 100644 --- a/payment-orchestrator/payment-orchestrator/src/test/java/com/stablecoin/payments/orchestrator/domain/workflow/PaymentWorkflowTest.java +++ b/payment-orchestrator/payment-orchestrator/src/test/java/com/stablecoin/payments/orchestrator/domain/workflow/PaymentWorkflowTest.java @@ -6,6 +6,7 @@ import com.stablecoin.payments.orchestrator.domain.workflow.activity.FxLockActivity; import com.stablecoin.payments.orchestrator.domain.workflow.activity.FxLockRequest; import com.stablecoin.payments.orchestrator.domain.workflow.activity.FxLockResult; +import com.stablecoin.payments.orchestrator.domain.workflow.activity.FxReleaseRequest; import com.stablecoin.payments.orchestrator.domain.workflow.dto.CancelRequest; import com.stablecoin.payments.orchestrator.domain.workflow.dto.PaymentResult; import io.temporal.client.WorkflowClient; @@ -33,6 +34,8 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.then; +import static org.mockito.BDDMockito.willThrow; +import static org.mockito.Mockito.atLeast; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -154,17 +157,14 @@ void shouldFailWhenFxLockFails(WorkflowClient workflowClient, class CancelSignal { @Test - @DisplayName("should run compensation when cancel signal received after FX lock") - void shouldCompensateOnCancelAfterFxLock(WorkflowClient workflowClient, - Worker worker) { - // Set up: compliance passes, FX locks, but then cancel + @DisplayName("should release FX lock when cancel signal received after FX lock succeeds") + void shouldReleaseFxLockOnCancelAfterFxLock(WorkflowClient workflowClient, + Worker worker) { given(complianceActivity.checkCompliance(any())) .willReturn(new ComplianceResult(CHECK_ID, PASSED, null)); - // FX lock will succeed, but we send cancel before it returns given(fxLockActivity.lockFxRate(any())) .willAnswer(invocation -> { - // Send cancel signal during FX lock activity execution var stub = workflowClient.newWorkflowStub( PaymentWorkflow.class, "payment-" + PAYMENT_ID); @@ -184,6 +184,71 @@ LOCK_ID, QUOTE_ID, new BigDecimal("0.92"), assertThat(result) .usingRecursiveComparison() .isEqualTo(expected); + + then(fxLockActivity).should().releaseLock(new FxReleaseRequest( + LOCK_ID, PAYMENT_ID, "Customer requested cancellation")); + } + + @Test + @DisplayName("should not call releaseLock when cancel before FX lock") + void shouldNotReleaseLockWhenCancelBeforeFxLock(WorkflowClient workflowClient, + Worker worker) { + given(complianceActivity.checkCompliance(any())) + .willAnswer(invocation -> { + var stub = workflowClient.newWorkflowStub( + PaymentWorkflow.class, + "payment-" + PAYMENT_ID); + stub.cancelPayment(new CancelRequest( + PAYMENT_ID, "Changed mind", "customer")); + return new ComplianceResult(CHECK_ID, PASSED, null); + }); + + var workflow = startWorkflow(workflowClient, worker); + var result = getResult(workflowClient); + + var expected = PaymentResult.failed(PAYMENT_ID, "Cancelled: Changed mind"); + assertThat(result) + .usingRecursiveComparison() + .isEqualTo(expected); + + then(fxLockActivity).should(never()).lockFxRate(any()); + then(fxLockActivity).should(never()).releaseLock(any()); + } + + @Test + @DisplayName("should return FAILED even when compensation activity throws") + void shouldReturnFailedWhenCompensationThrows(WorkflowClient workflowClient, + Worker worker) { + given(complianceActivity.checkCompliance(any())) + .willReturn(new ComplianceResult(CHECK_ID, PASSED, null)); + + given(fxLockActivity.lockFxRate(any())) + .willAnswer(invocation -> { + var stub = workflowClient.newWorkflowStub( + PaymentWorkflow.class, + "payment-" + PAYMENT_ID); + stub.cancelPayment(new CancelRequest( + PAYMENT_ID, "Timeout", "system")); + return new FxLockResult( + LOCK_ID, QUOTE_ID, new BigDecimal("0.92"), + new BigDecimal("920.00"), "EUR", + LOCKED, null); + }); + + willThrow(new RuntimeException("FX engine unavailable")) + .given(fxLockActivity).releaseLock(any()); + + var workflow = startWorkflow(workflowClient, worker); + var result = getResult(workflowClient); + + var expected = PaymentResult.failed(PAYMENT_ID, "Cancelled: Timeout"); + assertThat(result) + .usingRecursiveComparison() + .isEqualTo(expected); + + // Temporal retries the activity (maxAttempts=3) before propagating the failure + then(fxLockActivity).should(atLeast(1)).releaseLock(new FxReleaseRequest( + LOCK_ID, PAYMENT_ID, "Timeout")); } }