Order Processing - Choreography - Payment Service

payment_service.kt

class PaymentService(
    private val customerAccounts: MutableMap<Int, Int> = mutableMapOf(),
    private val skuPrices: MutableMap<Int, Double> = mutableMapOf()
) {
    @KafkaListener(topics = [ORDERS])
    fun onOrder(order: Order) {
        if (order.status != SUBMITTED) return

        val total = order.items.sumOf { skuPrices[it] ?: throw RuntimeException("Customer unknown") }
        val status = paymentGateway.processPayment(
            total,
            customerAccounts[order.customerId] ?: throw RuntimeException("Customer unknown")
        )
        sendMessage(PAYMENTS, order.withStatus(if (status) PAID else PAYMENT_FAILED))
    }

    @KafkaListener(topics = [SKUS])
    fun onSku(sku: Sku) {
        if (sku.onStock > 0) skuPrices[sku.id] = sku.price else skuPrices.remove(sku.id)
    }

    @KafkaListener(topics = [CUSTOMERS])
    fun onCustomer(customer: Customer) {
        customerAccounts[customer.id] = customer.account
    }    
}