Updates

build.gradle

@@ -21,22 +21,26 @@ repositories {
 
 dependencies {
 	implementation("org.springframework.boot:spring-boot-starter-webflux")
-	implementation("org.springframework.boot:spring-boot-starter-data-jpa")
 	implementation("org.springframework.boot:spring-boot-starter-data-r2dbc")/*
 	implementation("io.r2dbc:r2dbc-postgresql")*/
 
-	implementation("org.springframework.boot:spring-boot-starter-data-redis-reactive")
+	/*implementation("org.springframework.boot:spring-boot-starter-data-redis-reactive")
-	implementation("io.lettuce:lettuce-core:6.2.2.RELEASE")
+	*/implementation("io.lettuce:lettuce-core:6.2.2.RELEASE")
 
-	implementation 'com.h2database:h2'
+// Choose ONE driver depending on your DB:
+	runtimeOnly("io.r2dbc:r2dbc-h2") // for in-memory
 
-	implementation("io.github.resilience4j:resilience4j-reactor:2.0.2")
 	implementation("io.github.resilience4j:resilience4j-ratelimiter:2.0.2")
 	implementation("io.github.resilience4j:resilience4j-circuitbreaker:2.0.2")
 	implementation("io.github.resilience4j:resilience4j-retry:2.0.2")
-	implementation("io.github.resilience4j:resilience4j-spring-boot3:2.0.2")
 
 	implementation("org.springframework.boot:spring-boot-starter")
+	implementation("org.springdoc:springdoc-openapi-starter-webflux-ui:2.6.0")
+
+	implementation 'io.github.resilience4j:resilience4j-spring-boot2:1.7.1'
+	implementation 'io.github.resilience4j:resilience4j-reactor:1.7.1'
+
+
 
 	implementation 'org.projectlombok:lombok:1.18.32' // Use the latest stable version
 	annotationProcessor 'org.projectlombok:lombok:1.18.32' // For annotation processing

build/resources/main/application.yml

@@ -19,8 +19,11 @@ resilience4j:
         retry-exceptions:
           - org.springframework.web.reactive.function.client.WebClientRequestException
           - java.io.IOException
+          - com.example.mpesa.exceptions.MpesaBusyException
         ignore-exceptions:
           - com.test.payment.exceptions.MpesaPermanentException
+          - java.lang.IllegalArgumentException
+
 
   circuitbreaker:
     instances:
@@ -30,11 +33,38 @@ resilience4j:
         failure-rate-threshold: 50
         wait-duration-in-open-state: 10s
 
-spring:
-  datasource:
-    url: jdbc:h2:mem:testdb;DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE
-    driverClassName: org.h2.Driver
-    username: sa
-    password: password
-    platform: h2
 
+
+
+
+spring:
+  r2dbc:
+    url: r2dbc:h2:mem:///mpesa_db;DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE
+    username: sa
+    password:
+  sql:
+    init:
+      mode: always
+      schema-locations: classpath:schema.sql
+  main:
+    web-application-type: reactive
+
+logging:
+  level:
+    org.springframework.data.r2dbc: DEBUG
+
+springdoc:
+  swagger-ui:
+    path: /swagger-ui.html
+    operationsSorter: method
+    tagsSorter: alpha
+  api-docs:
+    path: /v3/api-docs
+  packages-to-scan: com.test.payment.controller
+
+mpesa:
+  base-url: https://sandbox.safaricom.co.ke
+  consumer-key: k6e7LtBNeVX7V8MPqB7P83FsZio8cRZD
+  consumer-secret: cGwiWzhDGopC3dho
+  business-short-code: 174379
+  pass-key: bfb279f9aa9bdbcf158e97dd71a467cd2e0c893059b10f78e6b72ada1ed2c919

build/resources/main/schema.sql

@@ -0,0 +1,8 @@
+CREATE TABLE IF NOT EXISTS transactions (
+    id SERIAL PRIMARY KEY,
+    mpesa_reference VARCHAR(255),
+    checkout_request_id VARCHAR(255),
+    status VARCHAR(50),
+    amount DECIMAL(10,2),
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);

src/main/java/com/test/payment/PaymentApplication.java

@@ -2,10 +2,12 @@ package com.test.payment;
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.data.r2dbc.repository.config.EnableR2dbcRepositories;
 import org.springframework.scheduling.annotation.EnableScheduling;
 
 @SpringBootApplication
 @EnableScheduling
+@EnableR2dbcRepositories(basePackages = "com.test.payment.repository")
 public class PaymentApplication {
 
 	public static void main(String[] args) {

src/main/java/com/test/payment/configurations/RedisConfig.java

@@ -2,13 +2,13 @@ package com.test.payment.configurations;
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+//import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
-import org.springframework.data.redis.core.RedisTemplate;
+//import org.springframework.data.redis.core.RedisTemplate;
 
-@Configuration
+//@Configuration
 public class RedisConfig {
 
-    @Bean
+ /*   @Bean
     public LettuceConnectionFactory redisConnectionFactory() {
         return new LettuceConnectionFactory();
     }
@@ -18,5 +18,5 @@ public class RedisConfig {
         RedisTemplate<String, String> template = new RedisTemplate<>();
         template.setConnectionFactory(factory);
         return template;
-    }
+    }*/
 }

src/main/java/com/test/payment/controller/MpesaController.java

@@ -1,10 +1,10 @@
 package com.test.payment.controller;
 
 
-import com.test.payment.models.*;
 import com.test.payment.models.MpesaResponse;
 import com.test.payment.models.PaymentRequest;
 import com.test.payment.service.MpesaService;
+import com.test.payment.service.MpesaServiceaa;
 import lombok.RequiredArgsConstructor;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.*;
@@ -22,7 +22,7 @@ public class MpesaController {
         return mpesaService.initiatePayment(request)
                 .map(ResponseEntity::ok)
                 .onErrorResume(ex -> Mono.just(ResponseEntity.badRequest()
-                        .body(new MpesaResponse("FAILED", ex.getMessage()))));
+                        .body(new MpesaResponse("FAILED", ex.getMessage(),"","",""))));
     }
 }
 

src/main/java/com/test/payment/dto/MpesaRequestDto.java

@@ -0,0 +1,44 @@
+package com.test.payment.dto;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class MpesaRequestDto {
+    @JsonProperty("BusinessShortCode")
+    private Long businessShortCode;
+
+    @JsonProperty("Password")
+    private String password;
+
+    @JsonProperty("Timestamp")
+    private String timestamp;
+
+    @JsonProperty("TransactionType")
+    private String transactionType;
+
+    @JsonProperty("Amount")
+    private Integer amount;
+
+    @JsonProperty("PartyA")
+    private Long partyA;
+
+    @JsonProperty("PartyB")
+    private Long partyB;
+
+    @JsonProperty("PhoneNumber")
+    private Long phoneNumber;
+
+    @JsonProperty("CallBackURL")
+    private String callBackURL;
+
+    @JsonProperty("AccountReference")
+    private String accountReference;
+
+    @JsonProperty("TransactionDesc")
+    private String transactionDesc;
+}

src/main/java/com/test/payment/dto/MpesaTokenResponse.java

@@ -0,0 +1,14 @@
+package com.test.payment.dto;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.Data;
+
+@Data
+public class MpesaTokenResponse {
+
+    @JsonProperty("access_token")
+    private String accessToken;
+
+    @JsonProperty("expires_in")
+    private String expiresIn;
+}

src/main/java/com/test/payment/exceptions/MpesaBusyException.java

@@ -0,0 +1,8 @@
+package com.test.payment.exceptions;
+
+
+public class MpesaBusyException extends RuntimeException {
+    public MpesaBusyException(String msg) {
+        super(msg);
+    }
+}

src/main/java/com/test/payment/jobs/MpesaTransactionJob.java

@@ -16,7 +16,7 @@ public class MpesaTransactionJob {
 
     @Scheduled(fixedDelay = 60000)
     public void pullTransactions() {
-        mpesaRepository.findAllTransactions()
+        mpesaRepository.findAll()
                 .doOnNext(tx -> log.info("Checking transaction: {}", tx))
                 .subscribe();
     }

src/main/java/com/test/payment/models/MpesaResponse.java

@@ -1,5 +1,6 @@
 package com.test.payment.models;
 
+import com.fasterxml.jackson.annotation.JsonProperty;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
@@ -8,6 +9,17 @@ import lombok.NoArgsConstructor;
 @AllArgsConstructor
 @NoArgsConstructor
 public class MpesaResponse {
-    private String status;
+
-    private String message;
+    @JsonProperty("MerchantRequestID")
+    private String merchantRequestID;
+
+    @JsonProperty("ResponseCode")
+    private String responseCode;
+    @JsonProperty("CheckoutRequestID")
+    private String checkoutRequestId;
+    @JsonProperty("CustomerMessage")
+    private String customerMessage;
+    @JsonProperty("ResponseDescription")
+    private String responseDescription;
+
 }

src/main/java/com/test/payment/models/PaymentRequest.java

@@ -5,8 +5,8 @@ import lombok.Data;
 
 @Data
 public class PaymentRequest {
-    private String phoneNumber;
+    private long phoneNumber;
-    private double amount;
+    private int amount;
     private String accountReference;
     private String transactionDesc;
 }

src/main/java/com/test/payment/models/Transaction.java

@@ -1,21 +1,23 @@
 package com.test.payment.models;
 
-import jakarta.persistence.Entity;
+
-import jakarta.persistence.Id;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.relational.core.mapping.*;
 
 @Data
 @AllArgsConstructor
 @NoArgsConstructor
-@Entity
+@Table("transactions")
 public class Transaction {
 
     @Id
     private String id;
-    private String phoneNumber;
+    private long phoneNumber;
-    private double amount;
+    private long amount;
     private String status;
+    private String checkoutRequestId;
 }
 

src/main/java/com/test/payment/repository/MpesaRepository.java

@@ -2,23 +2,20 @@ package com.test.payment.repository;
 
 
 import com.test.payment.models.Transaction;
-import org.springframework.data.redis.core.RedisTemplate;
+//import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
 import org.springframework.stereotype.Repository;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.util.List;
+import java.util.Locale;
+
 
 @Repository
-public class MpesaRepository {
+public interface MpesaRepository extends ReactiveCrudRepository<Transaction, Long> {
 
-    private final RedisTemplate<String, String> redisTemplate;
+  /*  public Mono<Void> saveTransaction(Transaction transaction) {
-
-    public MpesaRepository(RedisTemplate<String, String> redisTemplate) {
-        this.redisTemplate = redisTemplate;
-    }
-
-    public Mono<Void> saveTransaction(Transaction transaction) {
         return Mono.fromRunnable(() ->
                 redisTemplate.opsForHash().put("mpesa:transactions", transaction.getId(), transaction.getStatus())
         ).then();
@@ -27,5 +24,16 @@ public class MpesaRepository {
     public Flux<Transaction> findAllTransactions() {
         List<Object> values = redisTemplate.opsForHash().values("mpesa:transactions");
         return Flux.fromIterable(values).cast(Transaction.class);
+    }*/
+
+    Flux<Transaction> findByStatus(String status);
+    Mono<Transaction> findByCheckoutRequestId(String checkoutRequestId);
+
+   /* private final RedisTemplate<String, String> redisTemplate;
+
+    public MpesaRepository(RedisTemplate<String, String> redisTemplate) {
+        this.redisTemplate = redisTemplate;
     }
+
+   */
 }

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

@@ -1,21 +1,27 @@
 package com.test.payment.service;
 
 
-import com.test.payment.exceptions.MpesaPermanentException;
+import com.test.payment.dto.MpesaRequestDto;
+import com.test.payment.dto.MpesaTokenResponse;
+import com.test.payment.exceptions.MpesaBusyException;
 import com.test.payment.exceptions.MpesaTransientException;
-import com.test.payment.models.*;
+import com.test.payment.models.MpesaResponse;
+import com.test.payment.models.PaymentRequest;
 import com.test.payment.repository.MpesaRepository;
-import io.github.resilience4j.circuitbreaker.*;
+import com.test.payment.utils.MpesaUtils;
-import io.github.resilience4j.ratelimiter.*;
-import io.github.resilience4j.retry.*;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.core.env.Environment;
+import org.springframework.http.HttpStatusCode;
 import org.springframework.stereotype.Service;
 import org.springframework.web.reactive.function.client.WebClient;
 import reactor.core.publisher.Mono;
+import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
+import io.github.resilience4j.ratelimiter.annotation.RateLimiter;
+import io.github.resilience4j.retry.annotation.Retry;
 
-import java.util.UUID;
+import java.time.Duration;
-import java.util.function.Supplier;
+import java.util.Base64;
 
 @Service
 @RequiredArgsConstructor
@@ -24,42 +30,175 @@ public class MpesaService {
 
     private final WebClient mpesaWebClient;
     private final MpesaRepository mpesaRepository;
-    private final MpesaTokenService tokenService;
+    private final Environment environment;
-    private final RateLimiter rateLimiter;
-    private final Retry retry;
-    private final CircuitBreaker circuitBreaker;
 
-    public Mono<MpesaResponse> initiatePayment(PaymentRequest request) {
+    public Mono<String> getToken() {
-        Supplier<Mono<MpesaResponse>> decoratedSupplier =
+        return Mono.just("hVG5ybD47dHUMQ6RfWby2ZpIs1Ul");
-                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) {
+    @CircuitBreaker(name = "mpesaCircuitBreaker", fallbackMethod = "mpesaFallback")
-        return tokenService.getToken()
+    @RateLimiter(name = "mpesaLimiter")
+    @Retry(name = "mpesaRetry")
+    public Mono<MpesaResponse> initiatePayment(PaymentRequest request) {
+        String businessShortCode = environment.getProperty("mpesa.business-short-code");
+        String passkey = environment.getProperty("mpesa.pass-key");
+        String callback = "https://mydomain.com/path";
+
+        MpesaUtils.MpesaAuthData authData = MpesaUtils.generateAuthData(businessShortCode, passkey);
+
+        MpesaRequestDto mpesaRequestDto = new MpesaRequestDto(
+                Long.valueOf(businessShortCode),
+                authData.getPassword(),
+                authData.getTimestamp(),
+                "CustomerPayBillOnline",
+                request.getAmount(),
+                request.getPhoneNumber(),
+                Long.valueOf(businessShortCode),
+                request.getPhoneNumber(),
+                callback,
+                request.getAccountReference(),
+                request.getTransactionDesc()
+        );
+
+        // Use Mono.defer so each subscription is independent (multi-user safe)
+        return Mono.defer(() ->
+                        getToken()
+                                .flatMap(token -> callMpesa(mpesaRequestDto))
+                )
+                // Reactive retry only on MpesaBusyException, with configurable backoff
+                .retryWhen(
+                        reactor.util.retry.Retry.backoff(3, Duration.ofSeconds(10))
+                                .filter(ex -> ex instanceof MpesaBusyException)
+                                .onRetryExhaustedThrow((spec, signal) -> signal.failure())
+                )
+                .doOnNext(resp -> log.info("M-Pesa STK Response: {}", resp))
+                .doOnError(e -> log.error("M-Pesa call failed: {}", e.getMessage()));
+    }
+
+/*    @CircuitBreaker(name = "mpesaCircuitBreaker", fallbackMethod = "mpesaFallback")
+    @RateLimiter(name = "mpesaLimiter")
+    @Retry(name = "mpesaRetry")
+    public Mono<MpesaResponse> initiatePayment(PaymentRequest request) {
+        String businessShortCode = environment.getProperty("mpesa.business-short-code");
+        String passkey = environment.getProperty("mpesa.pass-key");
+        String callback = "https://mydomain.com/path";
+
+        MpesaUtils.MpesaAuthData authData = MpesaUtils.generateAuthData(businessShortCode, passkey);
+
+
+        MpesaRequestDto mpesaRequestDto = new MpesaRequestDto(
+                Long.valueOf(businessShortCode),
+                authData.getPassword(),
+                authData.getTimestamp(),
+                "CustomerPayBillOnline",
+                request.getAmount(),
+                request.getPhoneNumber(),
+                Long.valueOf(businessShortCode),
+                request.getPhoneNumber(),
+                callback,
+                request.getAccountReference(),
+                request.getTransactionDesc()
+        );
+        *//*return Mono.delay(Duration.ofSeconds(1))
+                .then(callMpesa(mpesaRequestDto))
+                .doOnNext(resp -> log.info("M-Pesa STK Response: {}", resp))
+                .doOnError(e -> log.error("M-Pesa API failed: {}", e.getMessage()));*//*
+
+      *//*  return callMpesa(mpesaRequestDto)
+                .doOnNext(resp -> log.info("M-Pesa STK Response: {}", resp))
+                .doOnError(e -> {
+                    if (e.toString().contains("System is busy")) {
+                        log.warn("M-Pesa system is busy");
+                        throe Mono.error(new MpesaBusyException("System busy"));
+                    } else {
+                        log.error("M-Pesa API failed: {}", e.getMessage());
+                    }
+                });*//*
+
+        return getToken()
+                .flatMap(token -> callMpesa(mpesaRequestDto))
+                .retryWhen(
+                        reactor.util.retry.Retry.backoff(3, Duration.ofSeconds(10))
+                                .filter(ex -> ex instanceof MpesaBusyException)
+                                .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) ->
+                                        retrySignal.failure()
+                                )
+                )  .doOnError(e -> log.error("M-Pesa call failed: {}", e.getMessage()));
+    }*/
+
+    /*@Retry(name = "mpesaRetry", fallbackMethod = "mpesaFallback")
+    @RateLimiter(name = "mpesaLimiter")
+    @CircuitBreaker(name = "mpesaCB", fallbackMethod = "mpesaFallback")
+    public Mono<MpesaResponse> initiatePayment(PaymentRequest request) {
+
+        String businessShortCode = environment.getProperty("mpesa.business-short-code");
+        String passkey = environment.getProperty("mpesa.pass-key");
+        String callback = "https://mydomain.com/path";
+
+        MpesaUtils.MpesaAuthData authData = MpesaUtils.generateAuthData(businessShortCode, passkey);
+
+
+        MpesaRequestDto mpesaRequestDto = new MpesaRequestDto(
+                Long.valueOf(businessShortCode),
+                authData.getPassword(),
+                authData.getTimestamp(),
+                "CustomerPayBillOnline",
+                request.getAmount(),
+                request.getPhoneNumber(),
+                Long.valueOf(businessShortCode),
+                request.getPhoneNumber(),
+                callback,
+                request.getAccountReference(),
+                request.getTransactionDesc()
+        );
+
+        return mpesaWebClient.post()
+                .uri("/mpesa/stkpush/v1/processrequest")
+                .bodyValue(mpesaRequestDto)
+                .retrieve()
+                .bodyToMono(MpesaResponse.class)
+                .flatMap(resp -> {
+                    if (resp.toString().contains("System is busy")) {
+                        return Mono.error(new MpesaBusyException("System busy"));
+                    }
+                    return Mono.just(resp);
+                });
+    }*/
+
+
+
+    private Mono<MpesaResponse> callMpesa(MpesaRequestDto request) {
+        return getToken()
                 .flatMap(token ->
                         mpesaWebClient.post()
                                 .uri("/mpesa/stkpush/v1/processrequest")
                                 .header("Authorization", "Bearer " + token)
                                 .bodyValue(request)
                                 .retrieve()
+                                .onStatus(HttpStatusCode::isError, clientResponse ->
+                                        clientResponse.bodyToMono(String.class)
+                                                .flatMap(errorBody -> {
+                                                    log.error("M-Pesa returned {} with body: {}", clientResponse.statusCode(), errorBody);
+                                                    if (errorBody.contains("System is busy")) {
+                                                        return Mono.error(new MpesaBusyException("System busy"));
+                                                    }
+                                                    return Mono.error(new MpesaTransientException(errorBody, null));
+                                                })
+                                )
                                 .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));
-                                })
                 );
     }
+
+    // 🧯 Fallback if circuit is open or all retries fail
+    private Mono<MpesaResponse> mpesaFallback(PaymentRequest request, Throwable ex) {
+        log.error("⚠️ Mpesa fallback triggered: {}", ex.getMessage());
+        return Mono.just(new MpesaResponse(
+                "500",
+                "Fallback triggered due to service unavailability",
+                null,
+                "",
+                null
+        ));
+    }
 }
+

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

@@ -0,0 +1,190 @@
+package com.test.payment.service;
+
+
+import com.test.payment.dto.MpesaRequestDto;
+import com.test.payment.exceptions.MpesaTransientException;
+import com.test.payment.models.*;
+import com.test.payment.repository.MpesaRepository;
+import com.test.payment.utils.MpesaUtils;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.core.env.Environment;
+import org.springframework.http.HttpStatusCode;
+import org.springframework.stereotype.Service;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Mono;
+
+/*@Service
+@RequiredArgsConstructor
+@Slf4j*/
+public class MpesaServiceaa {
+/*
+    private final WebClient mpesaWebClient;
+    private final MpesaRepository mpesaRepository;
+    private final RateLimiter rateLimiter;
+    private final Retry retry;
+    private final CircuitBreaker circuitBreaker;
+    private final Environment environment;
+
+    public Mono<String> getToken() {
+*//*        // 1️⃣ Check Redis cache first
+        return redisTemplate.opsForValue().get("mpesa:token")
+                .flatMap(cachedToken -> {
+                    if (cachedToken != null) {
+                        return Mono.just(cachedToken);
+                    }
+                    return fetchNewToken();
+                })
+                // if no cached token, fetch and cache new one
+                .switchIfEmpty(fetchNewToken());*//*
+        return Mono.fromSupplier(() -> "Q3HCot6dNLTpUtpitkr5Tatsa4KB");
+            // return fetchNewToken();
+    }
+
+    private Mono<String> fetchNewToken() {
+
+        String baseUrl = environment.getProperty("mpesa.base-url");
+        String consumerKey = environment.getProperty("mpesa.consumer-key");
+        String consumerSecret = environment.getProperty("mpesa.consumer-secret");
+
+        String credentials = consumerKey + ":" + consumerSecret;
+        String encodedCredentials = Base64.getEncoder().encodeToString(credentials.getBytes());
+
+        return mpesaWebClient.get()
+                .uri(baseUrl + "/oauth/v1/generate?grant_type=client_credentials")
+                .header("Authorization", "Basic " + encodedCredentials)
+                .retrieve()
+                .bodyToMono(MpesaTokenResponse.class)
+                .map(MpesaTokenResponse::getAccessToken);
+                *//*.flatMap(token ->
+                        redisTemplate.opsForValue()
+                                .set("mpesa:token", token, Duration.ofMinutes(50))
+                                .thenReturn(token)
+                );*//*
+    }
+
+    public Mono<MpesaResponse> initiatePayment(PaymentRequest request) {
+
+        String businessShortCode = environment.getProperty("mpesa.business-short-code");
+
+        String passkey = environment.getProperty("mpesa.pass-key");
+
+        String callback = "https://mydomain.com/path";
+
+        MpesaUtils.MpesaAuthData authData = MpesaUtils.generateAuthData(businessShortCode, passkey);
+
+        MpesaRequestDto mpesaRequestDto = new MpesaRequestDto(Long.valueOf(businessShortCode),authData.getPassword(),authData.getTimestamp(),"CustomerPayBillOnline",request.getAmount(),request.getPhoneNumber(),
+                Long.valueOf(businessShortCode),request.getPhoneNumber(),callback,request.getAccountReference(), request.getTransactionDesc());
+
+        Supplier<Mono<MpesaResponse>> decoratedSupplier =
+                CircuitBreaker.decorateSupplier(circuitBreaker,
+                        RateLimiter.decorateSupplier(rateLimiter,
+                                Retry.decorateSupplier(retry, () -> callMpesa(mpesaRequestDto))
+                        )
+                );
+
+        return Mono.defer(decoratedSupplier)
+            *//*    .flatMap(response -> {
+                 *//**//*   Transaction tx = new Transaction(UUID.randomUUID().toString(),
+                            request.getPhoneNumber(), request.getAmount(), response.getResponseCode(), response.getCheckoutRequestId());
+                    return mpesaRepository.save(tx).thenReturn(response);*//**//*
+                    return  response;
+                })*//*
+                .doOnError(e -> log.error("M-Pesa API failed: {}", e.getMessage()));
+    }
+
+    private Mono<MpesaResponse> callMpesa(MpesaRequestDto request) {
+
+        getToken()
+                .map(token -> {
+                    System.out.println("Token: " + token);
+                    return token;
+                })
+                .subscribe();
+
+        return getToken()
+                .flatMap(token ->
+                        mpesaWebClient.post()
+                                .uri("/mpesa/stkpush/v1/processrequest")
+                                .header("Authorization", "Bearer " + token)
+                                .bodyValue(request)
+                                .retrieve()
+                                // Intercept 4xx/5xx and extract actual body
+                                .onStatus(HttpStatusCode::isError, clientResponse ->
+                                        clientResponse.bodyToMono(String.class)
+                                                .flatMap(errorBody -> {
+                                                    log.error("M-Pesa returned {} with body: {}", clientResponse.statusCode(), errorBody);
+                                                    // Return a Mono.error so onErrorResume below can handle it
+                                                    return Mono.error(new RuntimeException(errorBody));
+                                                })
+                                )
+                                .bodyToMono(MpesaResponse.class)
+                                // Catch the above RuntimeException and return the error body as normal data
+                                .onErrorResume(RuntimeException.class, ex -> {
+                                    log.error("Returning raw M-Pesa 500 error body: {}", ex.getMessage());
+                                    return Mono.error(new MpesaTransientException(ex.getMessage(), ex)); // This is the actual 500 error body from M-Pesa
+                                })
+                                .doOnNext(response -> log.info("M-Pesa STK Response: {}", response)));
+
+    }*/
+
+   /* private final WebClient mpesaWebClient;
+    private final MpesaRepository mpesaRepository;
+    private final io.github.resilience4j.ratelimiter.RateLimiter rateLimiter;
+    private final io.github.resilience4j.retry.Retry retry;
+    private final io.github.resilience4j.circuitbreaker.CircuitBreaker circuitBreaker;
+    private final Environment environment;
+
+    public Mono<String> getToken() {
+        return Mono.fromSupplier(() -> "Q3HCot6dNLTpUtpitkr5Tatsa4KB");
+    }
+
+    public Mono<MpesaResponse> initiatePayment(PaymentRequest request) {
+        String businessShortCode = environment.getProperty("mpesa.business-short-code");
+        String passkey = environment.getProperty("mpesa.pass-key");
+        String callback = "https://mydomain.com/path";
+
+        MpesaUtils.MpesaAuthData authData = MpesaUtils.generateAuthData(businessShortCode, passkey);
+
+        MpesaRequestDto mpesaRequestDto = new MpesaRequestDto(
+                Long.valueOf(businessShortCode),
+                authData.getPassword(),
+                authData.getTimestamp(),
+                "CustomerPayBillOnline",
+                request.getAmount(),
+                request.getPhoneNumber(),
+                Long.valueOf(businessShortCode),
+                request.getPhoneNumber(),
+                callback,
+                request.getAccountReference(),
+                request.getTransactionDesc()
+        );
+
+        // Reactive chaining with operators (not Supplier)
+        return callMpesa(mpesaRequestDto)
+                .transformDeferred(ReactorRateLimiterOperator.of(rateLimiter))
+                .transformDeferred(ReactorRetryOperator.of(retry))
+                .transformDeferred(ReactorCircuitBreakerOperator.of(circuitBreaker))
+                .doOnNext(resp -> log.info("M-Pesa STK Response: {}", resp))
+                .doOnError(e -> log.error("M-Pesa API failed: {}", e.getMessage()));
+    }
+
+    private Mono<MpesaResponse> callMpesa(MpesaRequestDto request) {
+        return getToken()
+                .flatMap(token ->
+                        mpesaWebClient.post()
+                                .uri("/mpesa/stkpush/v1/processrequest")
+                                .header("Authorization", "Bearer " + token)
+                                .bodyValue(request)
+                                .retrieve()
+                                .onStatus(HttpStatusCode::isError, clientResponse ->
+                                        clientResponse.bodyToMono(String.class)
+                                                .flatMap(errorBody -> {
+                                                    log.error("M-Pesa returned {} with body: {}", clientResponse.statusCode(), errorBody);
+                                                    return Mono.error(new MpesaTransientException(errorBody, null));
+                                                })
+                                )
+                                .bodyToMono(MpesaResponse.class)
+                );
+    }*/
+}

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

@@ -2,7 +2,7 @@ package com.test.payment.service;
 
 
 import lombok.RequiredArgsConstructor;
-import org.springframework.data.redis.core.RedisTemplate;
+//import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Mono;
 
@@ -12,16 +12,16 @@ import java.time.Duration;
 @RequiredArgsConstructor
 public class MpesaTokenService {
 
-    private final RedisTemplate<String, String> redisTemplate;
+   // private final RedisTemplate<String, String> redisTemplate;
 
     public Mono<String> getToken() {
-        String cachedToken = redisTemplate.opsForValue().get("mpesa:token");
+       /* String cachedToken = redisTemplate.opsForValue().get("mpesa:token");
         if (cachedToken != null) {
             return Mono.just(cachedToken);
         }
         // Simulate token fetch from M-Pesa auth endpoint
         String newToken = "access_token_" + System.currentTimeMillis();
-        redisTemplate.opsForValue().set("mpesa:token", newToken, Duration.ofMinutes(50));
+        redisTemplate.opsForValue().set("mpesa:token", newToken, Duration.ofMinutes(50));*/
-        return Mono.just(newToken);
+        return Mono.just("newToken");
     }
 }

src/main/java/com/test/payment/utils/MpesaUtils.java

@@ -0,0 +1,60 @@
+package com.test.payment.utils;
+
+
+
+import java.nio.charset.StandardCharsets;
+import java.text.SimpleDateFormat;
+import java.util.Base64;
+import java.util.Date;
+import java.util.TimeZone;
+
+public class MpesaUtils {
+
+    /**
+     * Generates a timestamp in the format yyyyMMddHHmmss
+     * Example: 20251010162455
+     */
+    public static String generateTimestamp() {
+        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
+        // Set timezone to Africa/Nairobi
+        sdf.setTimeZone(TimeZone.getTimeZone("Africa/Nairobi"));
+        return sdf.format(new Date());
+    }
+
+    /**
+     * Generates a Base64-encoded password for M-Pesa STK Push.
+     * Format: Base64(BusinessShortCode + Passkey + Timestamp)
+     */
+    public static String generatePassword(String businessShortCode, String passkey, String timestamp) {
+        String dataToEncode = businessShortCode + passkey + timestamp;
+        return Base64.getEncoder().encodeToString(dataToEncode.getBytes(StandardCharsets.UTF_8));
+    }
+
+    /**
+     * Helper method to generate both password and timestamp together.
+     */
+    public static MpesaAuthData generateAuthData(String businessShortCode, String passkey) {
+        String timestamp = generateTimestamp();
+        String password = generatePassword(businessShortCode, passkey, timestamp);
+        return new MpesaAuthData(password, timestamp);
+    }
+
+    // Inner class to hold both password and timestamp
+    public static class MpesaAuthData {
+        private final String password;
+        private final String timestamp;
+
+        public MpesaAuthData(String password, String timestamp) {
+            this.password = password;
+            this.timestamp = timestamp;
+        }
+
+        public String getPassword() {
+            return password;
+        }
+
+        public String getTimestamp() {
+            return timestamp;
+        }
+    }
+}

src/main/resources/application.yml

@@ -19,8 +19,11 @@ resilience4j:
         retry-exceptions:
           - org.springframework.web.reactive.function.client.WebClientRequestException
           - java.io.IOException
+          - com.example.mpesa.exceptions.MpesaBusyException
         ignore-exceptions:
           - com.test.payment.exceptions.MpesaPermanentException
+          - java.lang.IllegalArgumentException
+
 
   circuitbreaker:
     instances:
@@ -30,11 +33,38 @@ resilience4j:
         failure-rate-threshold: 50
         wait-duration-in-open-state: 10s
 
-spring:
-  datasource:
-    url: jdbc:h2:mem:testdb;DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE
-    driverClassName: org.h2.Driver
-    username: sa
-    password: password
-    platform: h2
 
+
+
+
+spring:
+  r2dbc:
+    url: r2dbc:h2:mem:///mpesa_db;DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE
+    username: sa
+    password:
+  sql:
+    init:
+      mode: always
+      schema-locations: classpath:schema.sql
+  main:
+    web-application-type: reactive
+
+logging:
+  level:
+    org.springframework.data.r2dbc: DEBUG
+
+springdoc:
+  swagger-ui:
+    path: /swagger-ui.html
+    operationsSorter: method
+    tagsSorter: alpha
+  api-docs:
+    path: /v3/api-docs
+  packages-to-scan: com.test.payment.controller
+
+mpesa:
+  base-url: https://sandbox.safaricom.co.ke
+  consumer-key: k6e7LtBNeVX7V8MPqB7P83FsZio8cRZD
+  consumer-secret: cGwiWzhDGopC3dho
+  business-short-code: 174379
+  pass-key: bfb279f9aa9bdbcf158e97dd71a467cd2e0c893059b10f78e6b72ada1ed2c919

src/main/resources/schema.sql

@@ -0,0 +1,8 @@
+CREATE TABLE IF NOT EXISTS transactions (
+    id SERIAL PRIMARY KEY,
+    mpesa_reference VARCHAR(255),
+    checkout_request_id VARCHAR(255),
+    status VARCHAR(50),
+    amount DECIMAL(10,2),
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);