Initial commit

src/main/java/com/test/payment/service/MpesaService.java

@@ -0,0 +1,65 @@
+package com.test.payment.service;
+
+
+import com.test.payment.exceptions.MpesaPermanentException;
+import com.test.payment.exceptions.MpesaTransientException;
+import com.test.payment.models.*;
+import com.test.payment.repository.MpesaRepository;
+import io.github.resilience4j.circuitbreaker.*;
+import io.github.resilience4j.ratelimiter.*;
+import io.github.resilience4j.retry.*;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Mono;
+
+import java.util.UUID;
+import java.util.function.Supplier;
+
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class MpesaService {
+
+    private final WebClient mpesaWebClient;
+    private final MpesaRepository mpesaRepository;
+    private final MpesaTokenService tokenService;
+    private final RateLimiter rateLimiter;
+    private final Retry retry;
+    private final CircuitBreaker circuitBreaker;
+
+    public Mono<MpesaResponse> initiatePayment(PaymentRequest request) {
+        Supplier<Mono<MpesaResponse>> decoratedSupplier =
+                CircuitBreaker.decorateSupplier(circuitBreaker,
+                        RateLimiter.decorateSupplier(rateLimiter,
+                                Retry.decorateSupplier(retry, () -> callMpesa(request))
+                        )
+                );
+
+        return Mono.defer(decoratedSupplier)
+                .flatMap(response -> {
+                    Transaction tx = new Transaction(UUID.randomUUID().toString(),
+                            request.getPhoneNumber(), request.getAmount(), response.getStatus());
+                    return mpesaRepository.saveTransaction(tx).thenReturn(response);
+                })
+                .doOnError(e -> log.error("M-Pesa API failed: {}", e.getMessage()));
+    }
+
+    private Mono<MpesaResponse> callMpesa(PaymentRequest request) {
+        return tokenService.getToken()
+                .flatMap(token ->
+                        mpesaWebClient.post()
+                                .uri("/mpesa/stkpush/v1/processrequest")
+                                .header("Authorization", "Bearer " + token)
+                                .bodyValue(request)
+                                .retrieve()
+                                .bodyToMono(MpesaResponse.class)
+                                .timeout(java.time.Duration.ofSeconds(20))
+                                .onErrorResume(ex -> {
+                                    log.error("Error calling M-Pesa: {}", ex.getMessage());
+                                    return Mono.error(new MpesaTransientException("Temporary M-Pesa issue", ex));
+                                })
+                );
+    }
+}