From 4197a1781c14a349c525512f5816a2dc9b5218d9 Mon Sep 17 00:00:00 2001 From: "@tanya_r" Date: Fri, 19 Jun 2026 19:35:14 -0300 Subject: [PATCH] fix(payments): orchestrate a payment automatically after it is initiated A payment created through the API stayed in INITIATED forever: the orchestrator that screens and submits it had no production caller (only the retry path touched it, and only for payments already in SCREENING). Publish an in-process event when a payment is persisted and drive orchestration from an after-commit, asynchronous listener on a bounded executor, so screening and bank submission run off the request thread, after the row is durably committed, and never inside a transaction. A bank failure still leaves the payment in SCREENING for the scheduled retry. Idempotent replays do not re-trigger. Closes #317 --- .../PaymentOrchestrationTrigger.kt | 39 +++ .../application/PaymentServiceImpl.kt | 237 +++++++++--------- .../event/PaymentInitiatedEvent.kt | 11 + .../payments/config/PaymentAsyncConfig.kt | 33 +++ .../PaymentOrchestrationTriggerTest.kt | 36 +++ .../application/PaymentServiceImplTest.kt | 5 + 6 files changed, 246 insertions(+), 115 deletions(-) create mode 100644 services/payments/src/main/kotlin/com/fincore/payments/application/PaymentOrchestrationTrigger.kt create mode 100644 services/payments/src/main/kotlin/com/fincore/payments/application/event/PaymentInitiatedEvent.kt create mode 100644 services/payments/src/main/kotlin/com/fincore/payments/config/PaymentAsyncConfig.kt create mode 100644 services/payments/src/test/kotlin/com/fincore/payments/application/PaymentOrchestrationTriggerTest.kt diff --git a/services/payments/src/main/kotlin/com/fincore/payments/application/PaymentOrchestrationTrigger.kt b/services/payments/src/main/kotlin/com/fincore/payments/application/PaymentOrchestrationTrigger.kt new file mode 100644 index 0000000..d6ca409 --- /dev/null +++ b/services/payments/src/main/kotlin/com/fincore/payments/application/PaymentOrchestrationTrigger.kt @@ -0,0 +1,39 @@ +// SPDX-License-Identifier: BUSL-1.1 +// SPDX-FileCopyrightText: 2026 FinCore Engine Authors + +package com.fincore.payments.application + +import com.fincore.payments.application.event.PaymentInitiatedEvent +import org.slf4j.LoggerFactory +import org.springframework.scheduling.annotation.Async +import org.springframework.stereotype.Component +import org.springframework.transaction.event.TransactionPhase +import org.springframework.transaction.event.TransactionalEventListener + +/** + * Drives orchestration once an initiated payment's transaction has committed, off the request thread. AFTER_COMMIT + * guarantees the payment row is durably visible before [PaymentOrchestrator.process] screens and submits it (the + * external bank call must never run inside a transaction). A [com.fincore.payments.application.bank.BankProviderException] + * leaves the payment in SCREENING for the scheduled retry, so it is caught here rather than failing the async worker. + */ +@Component +class PaymentOrchestrationTrigger( + private val orchestrator: PaymentOrchestrator, +) { + private val log = LoggerFactory.getLogger(javaClass) + + @Async("paymentOrchestrationExecutor") + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) + @Suppress("TooGenericExceptionCaught") // a failed orchestration must not crash the async worker; it falls to retry + fun onPaymentInitiated(event: PaymentInitiatedEvent) { + try { + orchestrator.process(event.paymentId) + } catch (ex: Exception) { + log + .atWarn() + .addKeyValue("paymentId", event.paymentId.toString()) + .setCause(ex) + .log("orchestration after initiate failed; left for the scheduled retry") + } + } +} diff --git a/services/payments/src/main/kotlin/com/fincore/payments/application/PaymentServiceImpl.kt b/services/payments/src/main/kotlin/com/fincore/payments/application/PaymentServiceImpl.kt index 797ead9..a13c6bb 100644 --- a/services/payments/src/main/kotlin/com/fincore/payments/application/PaymentServiceImpl.kt +++ b/services/payments/src/main/kotlin/com/fincore/payments/application/PaymentServiceImpl.kt @@ -8,6 +8,7 @@ import com.fincore.core.PaymentId import com.fincore.events.EventEnvelope import com.fincore.events.EventType import com.fincore.events.PaymentEvents +import com.fincore.payments.application.event.PaymentInitiatedEvent import com.fincore.payments.domain.Payment import com.fincore.payments.domain.enum.PaymentStatus import com.fincore.payments.domain.exception.PaymentNotFoundException @@ -16,6 +17,7 @@ import com.fincore.payments.infrastructure.persistence.PaymentEventEntity import com.fincore.payments.infrastructure.persistence.PaymentEventRepository import com.fincore.payments.infrastructure.persistence.PaymentPersistenceAdapter import com.fincore.payments.infrastructure.persistence.PaymentRepository +import org.springframework.context.ApplicationEventPublisher import org.springframework.data.domain.PageRequest import org.springframework.data.domain.Sort import org.springframework.stereotype.Service @@ -24,133 +26,138 @@ import java.time.Instant import java.util.UUID @Service -class PaymentServiceImpl( - private val idempotencyStore: PaymentIdempotencyStore, - private val paymentRepository: PaymentRepository, - private val paymentEventRepository: PaymentEventRepository, - private val adapter: PaymentPersistenceAdapter, - private val outboxPublisher: PaymentOutboxEventPublisher, - private val objectMapper: ObjectMapper, - private val metrics: PaymentMetrics, -) : PaymentService { - override fun initiate(command: InitiatePaymentCommand): Payment { - val keyHash = Sha256.hex(command.idempotencyKey) - var attempt = 0 - while (true) { - try { - return idempotencyStore.reserveOrRun(keyHash) { createPayment(command) } - } catch (race: PaymentIdempotencyRaceException) { - attempt++ - if (attempt >= MAX_ATTEMPTS) throw PaymentConcurrencyException(race) +class PaymentServiceImpl + @Suppress("LongParameterList") // cohesive collaborators of the payment write side; splitting them would not aid clarity + constructor( + private val idempotencyStore: PaymentIdempotencyStore, + private val paymentRepository: PaymentRepository, + private val paymentEventRepository: PaymentEventRepository, + private val adapter: PaymentPersistenceAdapter, + private val outboxPublisher: PaymentOutboxEventPublisher, + private val objectMapper: ObjectMapper, + private val metrics: PaymentMetrics, + private val eventPublisher: ApplicationEventPublisher, + ) : PaymentService { + override fun initiate(command: InitiatePaymentCommand): Payment { + val keyHash = Sha256.hex(command.idempotencyKey) + var attempt = 0 + while (true) { + try { + return idempotencyStore.reserveOrRun(keyHash) { createPayment(command) } + } catch (race: PaymentIdempotencyRaceException) { + attempt++ + if (attempt >= MAX_ATTEMPTS) throw PaymentConcurrencyException(race) + } } } - } - @Transactional(readOnly = true) - override fun list( - page: Int, - size: Int, - ): PaymentPage { - val pageable = PageRequest.of(page, size, Sort.by(Sort.Order.desc("createdAt"), Sort.Order.desc("id"))) - val result = paymentRepository.findAll(pageable) - return PaymentPage( - items = result.content.map(adapter::toDomain), - page = page, - size = size, - totalElements = result.totalElements, - totalPages = result.totalPages, - ) - } + @Transactional(readOnly = true) + override fun list( + page: Int, + size: Int, + ): PaymentPage { + val pageable = PageRequest.of(page, size, Sort.by(Sort.Order.desc("createdAt"), Sort.Order.desc("id"))) + val result = paymentRepository.findAll(pageable) + return PaymentPage( + items = result.content.map(adapter::toDomain), + page = page, + size = size, + totalElements = result.totalElements, + totalPages = result.totalPages, + ) + } - @Transactional(readOnly = true) - override fun get(id: PaymentId): Payment = - adapter.toDomain(paymentRepository.findById(id.value).orElseThrow { PaymentNotFoundException(id) }) + @Transactional(readOnly = true) + override fun get(id: PaymentId): Payment = + adapter.toDomain(paymentRepository.findById(id.value).orElseThrow { PaymentNotFoundException(id) }) - @Transactional - override fun cancel(id: PaymentId): Payment = transition(id, PaymentStatus.CANCELLED, PaymentEvents.PaymentCancelled) + @Transactional + override fun cancel(id: PaymentId): Payment = transition(id, PaymentStatus.CANCELLED, PaymentEvents.PaymentCancelled) - @Transactional - override fun screen(id: PaymentId): Payment { - val entity = paymentRepository.findById(id.value).orElseThrow { PaymentNotFoundException(id) } - val payment = adapter.toDomain(entity) - payment.transitionTo(PaymentStatus.SCREENING) - entity.status = payment.status - paymentRepository.saveAndFlush(entity) - metrics.record(PaymentStatus.SCREENING) - return payment - } + @Transactional + override fun screen(id: PaymentId): Payment { + val entity = paymentRepository.findById(id.value).orElseThrow { PaymentNotFoundException(id) } + val payment = adapter.toDomain(entity) + payment.transitionTo(PaymentStatus.SCREENING) + entity.status = payment.status + paymentRepository.saveAndFlush(entity) + metrics.record(PaymentStatus.SCREENING) + return payment + } - @Transactional - override fun markSubmitted( - id: PaymentId, - providerReference: String, - ): Payment { - val entity = paymentRepository.findById(id.value).orElseThrow { PaymentNotFoundException(id) } - val payment = adapter.toDomain(entity) - payment.transitionTo(PaymentStatus.SUBMITTED) - entity.status = payment.status - entity.providerReference = providerReference - paymentRepository.saveAndFlush(entity) - recordEvent(payment, PaymentEvents.PaymentScreened) - metrics.record(PaymentStatus.SUBMITTED) - return payment - } + @Transactional + override fun markSubmitted( + id: PaymentId, + providerReference: String, + ): Payment { + val entity = paymentRepository.findById(id.value).orElseThrow { PaymentNotFoundException(id) } + val payment = adapter.toDomain(entity) + payment.transitionTo(PaymentStatus.SUBMITTED) + entity.status = payment.status + entity.providerReference = providerReference + paymentRepository.saveAndFlush(entity) + recordEvent(payment, PaymentEvents.PaymentScreened) + metrics.record(PaymentStatus.SUBMITTED) + return payment + } - @Transactional - override fun markFailed( - id: PaymentId, - reason: String, - ): Payment = transition(id, PaymentStatus.FAILED, PaymentEvents.PaymentFailed, reason) + @Transactional + override fun markFailed( + id: PaymentId, + reason: String, + ): Payment = transition(id, PaymentStatus.FAILED, PaymentEvents.PaymentFailed, reason) - @Transactional - override fun markSettled(id: PaymentId): Payment = transition(id, PaymentStatus.SETTLED, PaymentEvents.PaymentSettled) + @Transactional + override fun markSettled(id: PaymentId): Payment = transition(id, PaymentStatus.SETTLED, PaymentEvents.PaymentSettled) - private fun transition( - id: PaymentId, - target: PaymentStatus, - type: EventType, - detail: String? = null, - ): Payment { - val entity = paymentRepository.findById(id.value).orElseThrow { PaymentNotFoundException(id) } - val payment = adapter.toDomain(entity) - payment.transitionTo(target) - entity.status = payment.status - paymentRepository.saveAndFlush(entity) - recordEvent(payment, type, detail) - metrics.record(target) - return payment - } + private fun transition( + id: PaymentId, + target: PaymentStatus, + type: EventType, + detail: String? = null, + ): Payment { + val entity = paymentRepository.findById(id.value).orElseThrow { PaymentNotFoundException(id) } + val payment = adapter.toDomain(entity) + payment.transitionTo(target) + entity.status = payment.status + paymentRepository.saveAndFlush(entity) + recordEvent(payment, type, detail) + metrics.record(target) + return payment + } - private fun createPayment(command: InitiatePaymentCommand): Payment { - val payment = Payment(PaymentId.generate(), command.amount, command.reference) - paymentRepository.saveAndFlush(adapter.toNewEntity(payment, Instant.now())) - recordEvent(payment, PaymentEvents.PaymentInitiated) - metrics.record(PaymentStatus.INITIATED) - return payment - } + private fun createPayment(command: InitiatePaymentCommand): Payment { + val payment = Payment(PaymentId.generate(), command.amount, command.reference) + paymentRepository.saveAndFlush(adapter.toNewEntity(payment, Instant.now())) + recordEvent(payment, PaymentEvents.PaymentInitiated) + metrics.record(PaymentStatus.INITIATED) + // Must stay inside this transaction: an AFTER_COMMIT listener drives orchestration only if the tx commits. + eventPublisher.publishEvent(PaymentInitiatedEvent(payment.id)) + return payment + } - private fun recordEvent( - payment: Payment, - type: EventType, - detail: String? = null, - ) { - val now = Instant.now() - val envelope = - EventEnvelope.of( - source = SOURCE, - type = type, - data = PaymentEventData.from(payment, detail), - subject = payment.id.toString(), + private fun recordEvent( + payment: Payment, + type: EventType, + detail: String? = null, + ) { + val now = Instant.now() + val envelope = + EventEnvelope.of( + source = SOURCE, + type = type, + data = PaymentEventData.from(payment, detail), + subject = payment.id.toString(), + ) + paymentEventRepository.saveAndFlush( + PaymentEventEntity(UUID.randomUUID(), payment.id.value, type.fullType, objectMapper.writeValueAsString(envelope), now), ) - paymentEventRepository.saveAndFlush( - PaymentEventEntity(UUID.randomUUID(), payment.id.value, type.fullType, objectMapper.writeValueAsString(envelope), now), - ) - outboxPublisher.publish(envelope, AGGREGATE_TYPE, payment.id.toString(), type.fullType, now) - } + outboxPublisher.publish(envelope, AGGREGATE_TYPE, payment.id.toString(), type.fullType, now) + } - private companion object { - const val MAX_ATTEMPTS = 3 - const val SOURCE = "payments" - const val AGGREGATE_TYPE = "Payment" + private companion object { + const val MAX_ATTEMPTS = 3 + const val SOURCE = "payments" + const val AGGREGATE_TYPE = "Payment" + } } -} diff --git a/services/payments/src/main/kotlin/com/fincore/payments/application/event/PaymentInitiatedEvent.kt b/services/payments/src/main/kotlin/com/fincore/payments/application/event/PaymentInitiatedEvent.kt new file mode 100644 index 0000000..ac13951 --- /dev/null +++ b/services/payments/src/main/kotlin/com/fincore/payments/application/event/PaymentInitiatedEvent.kt @@ -0,0 +1,11 @@ +// SPDX-License-Identifier: BUSL-1.1 +// SPDX-FileCopyrightText: 2026 FinCore Engine Authors + +package com.fincore.payments.application.event + +import com.fincore.core.PaymentId + +/** Published in-process once a payment is persisted as INITIATED, to drive orchestration after the transaction commits. */ +data class PaymentInitiatedEvent( + val paymentId: PaymentId, +) diff --git a/services/payments/src/main/kotlin/com/fincore/payments/config/PaymentAsyncConfig.kt b/services/payments/src/main/kotlin/com/fincore/payments/config/PaymentAsyncConfig.kt new file mode 100644 index 0000000..bcabaef --- /dev/null +++ b/services/payments/src/main/kotlin/com/fincore/payments/config/PaymentAsyncConfig.kt @@ -0,0 +1,33 @@ +// SPDX-License-Identifier: BUSL-1.1 +// SPDX-FileCopyrightText: 2026 FinCore Engine Authors + +package com.fincore.payments.config + +import org.springframework.context.annotation.Bean +import org.springframework.context.annotation.Configuration +import org.springframework.scheduling.annotation.EnableAsync +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor + +/** Enables @Async and provides a bounded executor for post-commit payment orchestration. */ +@Configuration +@EnableAsync +class PaymentAsyncConfig { + @Bean + fun paymentOrchestrationExecutor(): ThreadPoolTaskExecutor = + ThreadPoolTaskExecutor().apply { + corePoolSize = CORE_POOL_SIZE + maxPoolSize = MAX_POOL_SIZE + setQueueCapacity(QUEUE_CAPACITY) + setThreadNamePrefix("payment-orch-") + setWaitForTasksToCompleteOnShutdown(true) + setAwaitTerminationSeconds(AWAIT_TERMINATION_SECONDS) + initialize() + } + + private companion object { + const val CORE_POOL_SIZE = 2 + const val MAX_POOL_SIZE = 8 + const val QUEUE_CAPACITY = 100 + const val AWAIT_TERMINATION_SECONDS = 20 + } +} diff --git a/services/payments/src/test/kotlin/com/fincore/payments/application/PaymentOrchestrationTriggerTest.kt b/services/payments/src/test/kotlin/com/fincore/payments/application/PaymentOrchestrationTriggerTest.kt new file mode 100644 index 0000000..6cc6957 --- /dev/null +++ b/services/payments/src/test/kotlin/com/fincore/payments/application/PaymentOrchestrationTriggerTest.kt @@ -0,0 +1,36 @@ +// SPDX-License-Identifier: BUSL-1.1 +// SPDX-FileCopyrightText: 2026 FinCore Engine Authors + +package com.fincore.payments.application + +import com.fincore.core.PaymentId +import com.fincore.payments.application.event.PaymentInitiatedEvent +import com.fincore.payments.domain.Payment +import io.mockk.every +import io.mockk.mockk +import io.mockk.verify +import org.junit.jupiter.api.Test + +class PaymentOrchestrationTriggerTest { + private val orchestrator = mockk() + private val trigger = PaymentOrchestrationTrigger(orchestrator) + private val id = PaymentId.generate() + + @Test + fun `processes the payment when it is initiated`() { + every { orchestrator.process(id) } returns mockk() + + trigger.onPaymentInitiated(PaymentInitiatedEvent(id)) + + verify { orchestrator.process(id) } + } + + @Test + fun `swallows an orchestration failure so the async worker survives`() { + every { orchestrator.process(id) } throws RuntimeException("bank down") + + trigger.onPaymentInitiated(PaymentInitiatedEvent(id)) + + verify { orchestrator.process(id) } + } +} diff --git a/services/payments/src/test/kotlin/com/fincore/payments/application/PaymentServiceImplTest.kt b/services/payments/src/test/kotlin/com/fincore/payments/application/PaymentServiceImplTest.kt index 76eb5a8..27b026d 100644 --- a/services/payments/src/test/kotlin/com/fincore/payments/application/PaymentServiceImplTest.kt +++ b/services/payments/src/test/kotlin/com/fincore/payments/application/PaymentServiceImplTest.kt @@ -8,6 +8,7 @@ import com.fincore.core.Currency import com.fincore.core.Money import com.fincore.core.PaymentId import com.fincore.events.PaymentEvents +import com.fincore.payments.application.event.PaymentInitiatedEvent import com.fincore.payments.domain.Payment import com.fincore.payments.domain.enum.PaymentStatus import com.fincore.payments.domain.exception.PaymentDomainException @@ -24,6 +25,7 @@ import io.mockk.every import io.mockk.mockk import io.mockk.verify import org.junit.jupiter.api.Test +import org.springframework.context.ApplicationEventPublisher import java.math.BigDecimal import java.time.Instant import java.util.Optional @@ -36,6 +38,7 @@ class PaymentServiceImplTest { private val outboxPublisher = mockk(relaxed = true) private val objectMapper = mockk { every { writeValueAsString(any()) } returns "{}" } private val metrics = mockk(relaxed = true) + private val eventPublisher = mockk(relaxed = true) private val service = PaymentServiceImpl( idempotencyStore, @@ -45,6 +48,7 @@ class PaymentServiceImplTest { outboxPublisher, objectMapper, metrics, + eventPublisher, ) init { @@ -60,6 +64,7 @@ class PaymentServiceImplTest { payment.status shouldBe PaymentStatus.INITIATED verify { outboxPublisher.publish(any(), "Payment", any(), PaymentEvents.PaymentInitiated.fullType, any()) } + verify { eventPublisher.publishEvent(PaymentInitiatedEvent(payment.id)) } } @Test