Alpha/Webflux-Payments
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Updates and payments additions

Browse Source
This commit is contained in:
spiro-alvin-nyasimi
2026-07-02 16:45:48 +03:00
parent 36c25a8fe6
commit e9d216be78
98 changed files with 2623 additions and 1445 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
src/main/java/com/test/payment/client/AirtelClient.java Normal file
View File

@@ -0,0 +1,69 @@
package com.test.payment.client;
import com.test.payment.dto.AirtelPaymentRequestDto;
import com.test.payment.dto.AirtelResponseDto;
import com.test.payment.exceptions.ProviderBusyException;
import com.test.payment.service.AirtelTokenService;
import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import io.github.resilience4j.ratelimiter.annotation.RateLimiter;
import io.github.resilience4j.retry.annotation.Retry;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.env.Environment;
import org.springframework.http.HttpStatusCode;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import java.time.Duration;
@Component
@RequiredArgsConstructor
@Slf4j
public class AirtelClient {
private final WebClient airtelWebClient;
private final AirtelTokenService tokenService;
private final Environment environment;
@CircuitBreaker(name = "airtelCircuitBreaker")
@RateLimiter(name = "airtelLimiter")
@Retry(name = "airtelRetry")
public Mono<AirtelResponseDto> pay(AirtelPaymentRequestDto request) {
return tokenService.getToken()
.flatMap(token -> airtelWebClient.post()
.uri("/merchant/v1/payments/")
.headers(h -> {
h.setBearerAuth(token);
h.set("X-Country", environment.getProperty("airtel.country", "KE"));
h.set("X-Currency", environment.getProperty("airtel.currency", "KES"));
})
.bodyValue(request)
.retrieve()
.onStatus(HttpStatusCode::isError,
resp -> ProviderHttpErrors.map(resp, "AIRTEL", tokenService.evictToken()))
.bodyToMono(AirtelResponseDto.class))
.retryWhen(reactor.util.retry.Retry.backoff(3, Duration.ofSeconds(3))
.maxBackoff(Duration.ofSeconds(30))
.jitter(0.5)
.filter(ProviderBusyException.class::isInstance)
.onRetryExhaustedThrow((spec, signal) -> signal.failure()));
}
@CircuitBreaker(name = "airtelCircuitBreaker")
@RateLimiter(name = "airtelLimiter")
public Mono<AirtelResponseDto> status(String transactionId) {
return tokenService.getToken()
.flatMap(token -> airtelWebClient.get()
.uri("/standard/v1/payments/{id}", transactionId)
.headers(h -> {
h.setBearerAuth(token);
h.set("X-Country", environment.getProperty("airtel.country", "KE"));
h.set("X-Currency", environment.getProperty("airtel.currency", "KES"));
})
.retrieve()
.onStatus(HttpStatusCode::isError,
resp -> ProviderHttpErrors.map(resp, "AIRTEL", tokenService.evictToken()))
.bodyToMono(AirtelResponseDto.class));
}
}

82
src/main/java/com/test/payment/client/MpesaClient.java Normal file
View File

@@ -0,0 +1,82 @@
package com.test.payment.client;
import com.test.payment.dto.MpesaRequestDto;
import com.test.payment.dto.MpesaResponse;
import com.test.payment.dto.StkQueryRequestDto;
import com.test.payment.dto.StkQueryResponseDto;
import com.test.payment.exceptions.ProviderBusyException;
import com.test.payment.exceptions.ProviderProcessingException;
import com.test.payment.service.MpesaTokenService;
import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import io.github.resilience4j.ratelimiter.annotation.RateLimiter;
import io.github.resilience4j.retry.annotation.Retry;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatusCode;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.ClientResponse;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import java.time.Duration;
/**
* All outbound Safaricom calls live here so the Resilience4j annotations are applied
* through Spring AOP (they would be silently skipped on self-invocation inside a service).
*
* The Resilience4j @Retry handles network-level failures; the reactive backoff retry
* below handles ProviderBusyException (rate limiting / "system busy"). Both coexist
* intentionally.
*/
@Component
@RequiredArgsConstructor
@Slf4j
public class MpesaClient {
private final WebClient mpesaWebClient;
private final MpesaTokenService tokenService;
@CircuitBreaker(name = "mpesaCircuitBreaker")
@RateLimiter(name = "mpesaLimiter")
@Retry(name = "mpesaRetry")
public Mono<MpesaResponse> stkPush(MpesaRequestDto request) {
return tokenService.getToken()
.flatMap(token -> mpesaWebClient.post()
.uri("/mpesa/stkpush/v1/processrequest")
.headers(h -> h.setBearerAuth(token))
.bodyValue(request)
.retrieve()
.onStatus(HttpStatusCode::isError, resp -> mapError(resp, false))
.bodyToMono(MpesaResponse.class))
.retryWhen(reactor.util.retry.Retry.backoff(3, Duration.ofSeconds(3))
.maxBackoff(Duration.ofSeconds(30))
.jitter(0.5)
.filter(ProviderBusyException.class::isInstance)
.onRetryExhaustedThrow((spec, signal) -> signal.failure()));
}
@CircuitBreaker(name = "mpesaCircuitBreaker")
@RateLimiter(name = "mpesaLimiter")
public Mono<StkQueryResponseDto> stkQuery(StkQueryRequestDto request) {
return tokenService.getToken()
.flatMap(token -> mpesaWebClient.post()
.uri("/mpesa/stkpushquery/v1/query")
.headers(h -> h.setBearerAuth(token))
.bodyValue(request)
.retrieve()
.onStatus(HttpStatusCode::isError, resp -> mapError(resp, true))
.bodyToMono(StkQueryResponseDto.class));
}
private Mono<? extends Throwable> mapError(ClientResponse response, boolean isQuery) {
if (isQuery && !response.statusCode().is2xxSuccessful()) {
// "The transaction is being processed" arrives as an error body (500.001.1001)
return response.bodyToMono(String.class)
.defaultIfEmpty("")
.flatMap(body -> body.contains("500.001.1001")
? Mono.error(new ProviderProcessingException("Transaction is still being processed"))
: ProviderHttpErrors.mapBody(response.statusCode(), body, "MPESA", tokenService.evictToken()));
}
return ProviderHttpErrors.map(response, "MPESA", tokenService.evictToken());
}
}

80
src/main/java/com/test/payment/client/MtnClient.java Normal file
View File

@@ -0,0 +1,80 @@
package com.test.payment.client;
import com.test.payment.dto.MtnPayRequestDto;
import com.test.payment.dto.MtnStatusResponseDto;
import com.test.payment.exceptions.ProviderBusyException;
import com.test.payment.service.MtnTokenService;
import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import io.github.resilience4j.ratelimiter.annotation.RateLimiter;
import io.github.resilience4j.retry.annotation.Retry;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.env.Environment;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatusCode;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import java.time.Duration;
@Component
@RequiredArgsConstructor
@Slf4j
public class MtnClient {
private final WebClient mtnWebClient;
private final MtnTokenService tokenService;
private final Environment environment;
/**
* MoMo request-to-pay returns 202 Accepted with an empty body; the referenceId
* we generate is the id used for the status check and callback correlation.
*/
@CircuitBreaker(name = "mtnCircuitBreaker")
@RateLimiter(name = "mtnLimiter")
@Retry(name = "mtnRetry")
public Mono<Void> requestToPay(String referenceId, MtnPayRequestDto request) {
return tokenService.getToken()
.flatMap(token -> mtnWebClient.post()
.uri("/collection/v1_0/requesttopay")
.headers(h -> {
commonHeaders(h, token);
h.set("X-Reference-Id", referenceId);
String callbackUrl = environment.getProperty("mtn.callback-url", "");
if (!callbackUrl.isBlank()) {
h.set("X-Callback-Url", callbackUrl);
}
})
.bodyValue(request)
.retrieve()
.onStatus(HttpStatusCode::isError,
resp -> ProviderHttpErrors.map(resp, "MTN", tokenService.evictToken()))
.toBodilessEntity()
.then())
.retryWhen(reactor.util.retry.Retry.backoff(3, Duration.ofSeconds(3))
.maxBackoff(Duration.ofSeconds(30))
.jitter(0.5)
.filter(ProviderBusyException.class::isInstance)
.onRetryExhaustedThrow((spec, signal) -> signal.failure()));
}
@CircuitBreaker(name = "mtnCircuitBreaker")
@RateLimiter(name = "mtnLimiter")
public Mono<MtnStatusResponseDto> status(String referenceId) {
return tokenService.getToken()
.flatMap(token -> mtnWebClient.get()
.uri("/collection/v1_0/requesttopay/{referenceId}", referenceId)
.headers(h -> commonHeaders(h, token))
.retrieve()
.onStatus(HttpStatusCode::isError,
resp -> ProviderHttpErrors.map(resp, "MTN", tokenService.evictToken()))
.bodyToMono(MtnStatusResponseDto.class));
}
private void commonHeaders(HttpHeaders headers, String token) {
headers.setBearerAuth(token);
headers.set("X-Target-Environment", environment.getProperty("mtn.target-environment", "sandbox"));
headers.set("Ocp-Apim-Subscription-Key", environment.getProperty("mtn.subscription-key", ""));
}
}

55
src/main/java/com/test/payment/client/ProviderHttpErrors.java Normal file
View File

@@ -0,0 +1,55 @@
package com.test.payment.client;
import com.test.payment.exceptions.ProviderBusyException;
import com.test.payment.exceptions.ProviderPermanentException;
import com.test.payment.exceptions.ProviderTransientException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatusCode;
import org.springframework.web.reactive.function.client.ClientResponse;
import reactor.core.publisher.Mono;
/**
* Shared HTTP error mapping for provider calls:
* 429/busy body → ProviderBusyException (reactive backoff retries),
* 401 → evict token + ProviderTransientException (Resilience4j retry refetches),
* other 4xx → ProviderPermanentException (never retried),
* 5xx → ProviderTransientException (retried).
*/
@Slf4j
public final class ProviderHttpErrors {
private ProviderHttpErrors() {
}
public static Mono<? extends Throwable> map(ClientResponse response, String provider, Mono<Void> evictToken) {
HttpStatusCode status = response.statusCode();
if (status.value() == 429) {
log.warn("[{}] rate limit (429) — will retry with backoff", provider);
return Mono.error(new ProviderBusyException(provider + " rate limit exceeded"));
}
return response.bodyToMono(String.class)
.defaultIfEmpty("")
.flatMap(body -> mapBody(status, body, provider, evictToken));
}
public static Mono<? extends Throwable> mapBody(HttpStatusCode status, String body, String provider, Mono<Void> evictToken) {
log.error("[{}] returned {} with body: {}", provider, status, body);
String lower = body.toLowerCase();
if (status.value() == 429
|| lower.contains("system is busy")
|| lower.contains("too many requests")
|| lower.contains("rate limit")
|| lower.contains("429.001")) {
return Mono.error(new ProviderBusyException(provider + " busy: " + body));
}
if (status.value() == 401) {
// stored token is stale/invalid — evict so the retry fetches a fresh one
return evictToken.then(Mono.error(new ProviderTransientException(
provider + " rejected the access token (401); token evicted for refresh", null)));
}
if (status.is4xxClientError()) {
return Mono.error(new ProviderPermanentException(provider + " rejected the request: " + body));
}
return Mono.error(new ProviderTransientException(provider + " error " + status + ": " + body, null));
}
}

26
src/main/java/com/test/payment/configurations/DatabaseSchemaInitializer.java Normal file
View File

@@ -0,0 +1,26 @@
package com.test.payment.configurations;
import com.test.payment.models.DatabaseSchema;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.r2dbc.core.DatabaseClient;
import reactor.core.publisher.Flux;
/**
* Creates the schema from DatabaseSchema.STATEMENTS at startup
* (replaces spring.sql.init + schema.sql).
*/
@Configuration
@Slf4j
public class DatabaseSchemaInitializer {
@Bean
public org.springframework.beans.factory.InitializingBean schemaInitializer(DatabaseClient databaseClient) {
return () -> Flux.fromIterable(DatabaseSchema.STATEMENTS)
.concatMap(statement -> databaseClient.sql(statement).then())
.doOnComplete(() -> log.info("Database schema initialized ({} statements)", DatabaseSchema.STATEMENTS.size()))
.then()
.block();
}
}

22
src/main/java/com/test/payment/configurations/RedisConfig.java
View File

@@ -1,22 +0,0 @@
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.core.RedisTemplate;
//@Configuration
public class RedisConfig {
/* @Bean
public LettuceConnectionFactory redisConnectionFactory() {
return new LettuceConnectionFactory();
}
@Bean
public RedisTemplate<String, String> redisTemplate(LettuceConnectionFactory factory) {
RedisTemplate<String, String> template = new RedisTemplate<>();
template.setConnectionFactory(factory);
return template;
}*/
}

45
src/main/java/com/test/payment/configurations/ResilienceConfig.java
View File

@@ -1,45 +0,0 @@
package com.test.payment.configurations;
import com.test.payment.exceptions.MpesaPermanentException;
import io.github.resilience4j.circuitbreaker.*;
import io.github.resilience4j.ratelimiter.*;
import io.github.resilience4j.retry.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.reactive.function.client.WebClientRequestException;
import java.io.IOException;
import java.time.Duration;
@Configuration
public class ResilienceConfig {
@Bean
public RateLimiter rateLimiter() {
return RateLimiter.of("mpesaLimiter", RateLimiterConfig.custom()
.limitForPeriod(10)
.limitRefreshPeriod(Duration.ofSeconds(1))
.timeoutDuration(Duration.ZERO)
.build());
}
@Bean
public Retry retry() {
return Retry.of("mpesaRetry", RetryConfig.custom()
.maxAttempts(3)
.waitDuration(Duration.ofSeconds(2))
.retryExceptions(WebClientRequestException.class, IOException.class)
.ignoreExceptions(MpesaPermanentException.class)
.build());
}
@Bean
public CircuitBreaker circuitBreaker() {
return CircuitBreaker.of("mpesaCircuitBreaker", CircuitBreakerConfig.custom()
.failureRateThreshold(50)
.waitDurationInOpenState(Duration.ofSeconds(10))
.permittedNumberOfCallsInHalfOpenState(3)
.slidingWindowSize(10)
.build());
}
}

28
src/main/java/com/test/payment/configurations/WebClientConfig.java
View File

@@ -1,16 +1,40 @@
package com.test.payment.configurations;
import io.netty.channel.ChannelOption;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.netty.http.client.HttpClient;
import java.time.Duration;
@Configuration
public class WebClientConfig {
@Bean
public WebClient mpesaWebClient(WebClient.Builder builder) {
public WebClient mpesaWebClient(WebClient.Builder builder, Environment environment) {
return providerClient(builder, environment.getProperty("mpesa.base-url", "https://sandbox.safaricom.co.ke"));
}
@Bean
public WebClient airtelWebClient(WebClient.Builder builder, Environment environment) {
return providerClient(builder, environment.getProperty("airtel.base-url", "https://openapiuat.airtel.africa"));
}
@Bean
public WebClient mtnWebClient(WebClient.Builder builder, Environment environment) {
return providerClient(builder, environment.getProperty("mtn.base-url", "https://sandbox.momodeveloper.mtn.com"));
}
private WebClient providerClient(WebClient.Builder builder, String baseUrl) {
HttpClient httpClient = HttpClient.create()
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
.responseTimeout(Duration.ofSeconds(15));
return builder
.baseUrl("https://sandbox.safaricom.co.ke")
.baseUrl(baseUrl)
.clientConnector(new ReactorClientHttpConnector(httpClient))
.build();
}
}

40
src/main/java/com/test/payment/controller/AirtelController.java Normal file
View File

@@ -0,0 +1,40 @@
package com.test.payment.controller;
import com.test.payment.dto.AirtelCallbackPayload;
import com.test.payment.dto.CallbackAckDto;
import com.test.payment.dto.PaymentRequest;
import com.test.payment.dto.PaymentResultDto;
import com.test.payment.dto.TransactionStatusDto;
import com.test.payment.service.AirtelService;
import jakarta.validation.Valid;
import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;
@RestController
@RequestMapping("/api/airtel")
@RequiredArgsConstructor
public class AirtelController {
private final AirtelService airtelService;
@PostMapping("/pay")
public Mono<PaymentResultDto> pay(@Valid @RequestBody PaymentRequest request) {
return airtelService.initiatePayment(request);
}
@PostMapping("/callback")
public Mono<CallbackAckDto> callback(@RequestBody AirtelCallbackPayload payload) {
return airtelService.handleCallback(payload);
}
@GetMapping("/status/{transactionId}")
public Mono<TransactionStatusDto> status(@PathVariable String transactionId) {
return airtelService.checkStatus(transactionId);
}
}

90
src/main/java/com/test/payment/controller/GlobalExceptionHandler.java Normal file
View File

@@ -0,0 +1,90 @@
package com.test.payment.controller;
import com.test.payment.dto.ErrorResponseDto;
import com.test.payment.exceptions.ProviderBusyException;
import com.test.payment.exceptions.ProviderPermanentException;
import com.test.payment.exceptions.ProviderTransientException;
import io.github.resilience4j.circuitbreaker.CallNotPermittedException;
import io.github.resilience4j.ratelimiter.RequestNotPermitted;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestControllerAdvice;
import org.springframework.web.bind.support.WebExchangeBindException;
import org.springframework.web.reactive.function.client.WebClientRequestException;
import org.springframework.web.server.ResponseStatusException;
import java.util.List;
/**
* Every failure still returns a structured JSON value to the caller,
* with an HTTP status that reflects what actually happened upstream.
*/
@RestControllerAdvice
@Slf4j
public class GlobalExceptionHandler {
@ExceptionHandler(CallNotPermittedException.class)
@ResponseStatus(HttpStatus.SERVICE_UNAVAILABLE)
public ErrorResponseDto circuitOpen(CallNotPermittedException ex) {
log.warn("Circuit breaker open: {}", ex.getMessage());
return ErrorResponseDto.of("CIRCUIT_OPEN", "Payment service temporarily unavailable, please retry shortly");
}
@ExceptionHandler(RequestNotPermitted.class)
@ResponseStatus(HttpStatus.TOO_MANY_REQUESTS)
public ErrorResponseDto rateLimited(RequestNotPermitted ex) {
return ErrorResponseDto.of("RATE_LIMITED", "Too many payment requests, please retry shortly");
}
@ExceptionHandler(ProviderBusyException.class)
@ResponseStatus(HttpStatus.SERVICE_UNAVAILABLE)
public ErrorResponseDto providerBusy(ProviderBusyException ex) {
return ErrorResponseDto.of("PROVIDER_BUSY", "The payment provider is busy, please retry shortly");
}
@ExceptionHandler(ProviderPermanentException.class)
@ResponseStatus(HttpStatus.BAD_REQUEST)
public ErrorResponseDto providerRejected(ProviderPermanentException ex) {
return ErrorResponseDto.of("PROVIDER_REJECTED", ex.getMessage());
}
@ExceptionHandler(ProviderTransientException.class)
@ResponseStatus(HttpStatus.BAD_GATEWAY)
public ErrorResponseDto providerError(ProviderTransientException ex) {
return ErrorResponseDto.of("PROVIDER_ERROR", "Payment provider request failed, please retry");
}
@ExceptionHandler(WebClientRequestException.class)
@ResponseStatus(HttpStatus.BAD_GATEWAY)
public ErrorResponseDto providerUnreachable(WebClientRequestException ex) {
log.error("Could not reach payment provider: {}", ex.getMessage());
return ErrorResponseDto.of("PROVIDER_UNREACHABLE", "Could not reach the payment provider, please retry");
}
@ExceptionHandler(WebExchangeBindException.class)
@ResponseStatus(HttpStatus.BAD_REQUEST)
public ErrorResponseDto validationFailed(WebExchangeBindException ex) {
List<String> details = ex.getFieldErrors().stream()
.map(err -> err.getField() + ": " + err.getDefaultMessage())
.toList();
ErrorResponseDto body = ErrorResponseDto.of("VALIDATION_FAILED", "Request validation failed");
body.setDetails(details);
return body;
}
@ExceptionHandler(ResponseStatusException.class)
public ResponseEntity<ErrorResponseDto> responseStatus(ResponseStatusException ex) {
return ResponseEntity.status(ex.getStatusCode())
.body(ErrorResponseDto.of(String.valueOf(ex.getStatusCode().value()), ex.getReason()));
}
@ExceptionHandler(Exception.class)
@ResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR)
public ErrorResponseDto unexpected(Exception ex) {
log.error("Unexpected error", ex);
return ErrorResponseDto.of("INTERNAL_ERROR", "An unexpected error occurred");
}
}

40
src/main/java/com/test/payment/controller/MpesaController.java
View File

@@ -1,13 +1,19 @@
package com.test.payment.controller;
import com.test.payment.models.MpesaResponse;
import com.test.payment.models.PaymentRequest;
import com.test.payment.dto.CallbackAckDto;
import com.test.payment.dto.PaymentRequest;
import com.test.payment.dto.PaymentResultDto;
import com.test.payment.dto.StkCallbackPayload;
import com.test.payment.dto.TransactionStatusDto;
import com.test.payment.service.MpesaService;
import com.test.payment.service.MpesaServiceaa;
import jakarta.validation.Valid;
import lombok.RequiredArgsConstructor;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;
@RestController
@@ -18,11 +24,21 @@ public class MpesaController {
private final MpesaService mpesaService;
@PostMapping("/pay")
public Mono<ResponseEntity<MpesaResponse>> pay(@RequestBody PaymentRequest request) {
return mpesaService.initiatePayment(request)
.map(ResponseEntity::ok)
.onErrorResume(ex -> Mono.just(ResponseEntity.badRequest()
.body(new MpesaResponse("FAILED", ex.getMessage(),"","",""))));
public Mono<PaymentResultDto> pay(@Valid @RequestBody PaymentRequest request) {
return mpesaService.initiatePayment(request);
}
/**
* Safaricom posts the STK result here (mpesa.callback-url must point at this path).
* Always acknowledges with ResultCode 0 so Safaricom does not keep retrying.
*/
@PostMapping("/callback")
public Mono<CallbackAckDto> callback(@RequestBody StkCallbackPayload payload) {
return mpesaService.handleCallback(payload);
}
@GetMapping("/status/{checkoutRequestId}")
public Mono<TransactionStatusDto> status(@PathVariable String checkoutRequestId) {
return mpesaService.checkStatus(checkoutRequestId);
}
}

50
src/main/java/com/test/payment/controller/MtnController.java Normal file
View File

@@ -0,0 +1,50 @@
package com.test.payment.controller;
import com.test.payment.dto.CallbackAckDto;
import com.test.payment.dto.MtnStatusResponseDto;
import com.test.payment.dto.PaymentRequest;
import com.test.payment.dto.PaymentResultDto;
import com.test.payment.dto.TransactionStatusDto;
import com.test.payment.service.MtnService;
import jakarta.validation.Valid;
import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;
@RestController
@RequestMapping("/api/mtn")
@RequiredArgsConstructor
public class MtnController {
private final MtnService mtnService;
@PostMapping("/pay")
public Mono<PaymentResultDto> pay(@Valid @RequestBody PaymentRequest request) {
return mtnService.initiatePayment(request);
}
/**
* MoMo delivers the request-to-pay result here (mtn.callback-url); MTN may use
* PUT or POST depending on the API version, so both are accepted.
*/
@PostMapping("/callback")
public Mono<CallbackAckDto> callbackPost(@RequestBody MtnStatusResponseDto payload) {
return mtnService.handleCallback(payload);
}
@PutMapping("/callback")
public Mono<CallbackAckDto> callbackPut(@RequestBody MtnStatusResponseDto payload) {
return mtnService.handleCallback(payload);
}
@GetMapping("/status/{referenceId}")
public Mono<TransactionStatusDto> status(@PathVariable String referenceId) {
return mtnService.checkStatus(referenceId);
}
}

26
src/main/java/com/test/payment/controller/PaymentsController.java Normal file
View File

@@ -0,0 +1,26 @@
package com.test.payment.controller;
import com.test.payment.models.Transaction;
import com.test.payment.service.PaymentLifecycleService;
import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
@RestController
@RequestMapping("/api/payments")
@RequiredArgsConstructor
public class PaymentsController {
private final PaymentLifecycleService lifecycle;
/**
* All consolidated transactions across providers; optional ?provider=MPESA|AIRTEL|MTN filter.
*/
@GetMapping("/transactions")
public Flux<Transaction> transactions(@RequestParam(required = false) String provider) {
return lifecycle.listTransactions(provider == null ? null : provider.toUpperCase());
}
}

29
src/main/java/com/test/payment/dto/AirtelCallbackPayload.java Normal file
View File

@@ -0,0 +1,29 @@
package com.test.payment.dto;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
/**
* Airtel Money result callback:
* {"transaction":{"id":"<merchant tx id>","message":"...","status_code":"TS","airtel_money_id":"..."}}
*/
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class AirtelCallbackPayload {
private Transaction transaction;
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public static class Transaction {
private String id;
private String message;
@JsonProperty("status_code")
private String statusCode;
@JsonProperty("airtel_money_id")
private String airtelMoneyId;
}
}

37
src/main/java/com/test/payment/dto/AirtelPaymentRequestDto.java Normal file
View File

@@ -0,0 +1,37 @@
package com.test.payment.dto;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* Airtel Money USSD-push collection request (POST /merchant/v1/payments/).
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class AirtelPaymentRequestDto {
private String reference;
private Subscriber subscriber;
private Transaction transaction;
@Data
@AllArgsConstructor
@NoArgsConstructor
public static class Subscriber {
private String country;
private String currency;
private String msisdn;
}
@Data
@AllArgsConstructor
@NoArgsConstructor
public static class Transaction {
private String amount;
private String country;
private String currency;
private String id;
}
}

45
src/main/java/com/test/payment/dto/AirtelResponseDto.java Normal file
View File

@@ -0,0 +1,45 @@
package com.test.payment.dto;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
/**
* Airtel Money response envelope, shared by payment initiation and status enquiry.
* Transaction status codes: TIP = in progress, TS = success, TF = failed.
*/
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class AirtelResponseDto {
private DataBody data;
private Status status;
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public static class DataBody {
private Transaction transaction;
}
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public static class Transaction {
private String id;
private String status;
private String message;
@JsonProperty("airtel_money_id")
private String airtelMoneyId;
}
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public static class Status {
private String code;
private String message;
private Boolean success;
@JsonProperty("response_code")
private String responseCode;
}
}

25
src/main/java/com/test/payment/dto/CallbackAckDto.java Normal file
View File

@@ -0,0 +1,25 @@
package com.test.payment.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* Acknowledgement returned to Safaricom for a received STK callback.
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class CallbackAckDto {
@JsonProperty("ResultCode")
private int resultCode;
@JsonProperty("ResultDesc")
private String resultDesc;
public static CallbackAckDto accepted(String desc) {
return new CallbackAckDto(0, desc);
}
}

24
src/main/java/com/test/payment/dto/ErrorResponseDto.java Normal file
View File

@@ -0,0 +1,24 @@
package com.test.payment.dto;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.time.LocalDateTime;
import java.util.List;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class ErrorResponseDto {
private String status;
private String error;
private String message;
private List<String> details;
private LocalDateTime timestamp;
public static ErrorResponseDto of(String error, String message) {
return new ErrorResponseDto("FAILED", error, message, null, LocalDateTime.now());
}
}

2
src/main/java/com/test/payment/models/MpesaResponse.java → src/main/java/com/test/payment/dto/MpesaResponse.java
View File

@@ -1,4 +1,4 @@
package com.test.payment.models;
package com.test.payment.dto;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.AllArgsConstructor;

29
src/main/java/com/test/payment/dto/MtnPayRequestDto.java Normal file
View File

@@ -0,0 +1,29 @@
package com.test.payment.dto;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* MTN MoMo Collections request-to-pay body (POST /collection/v1_0/requesttopay).
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class MtnPayRequestDto {
private String amount;
private String currency;
private String externalId;
private Payer payer;
private String payerMessage;
private String payeeNote;
@Data
@AllArgsConstructor
@NoArgsConstructor
public static class Payer {
private String partyIdType;
private String partyId;
}
}

27
src/main/java/com/test/payment/dto/MtnStatusResponseDto.java Normal file
View File

@@ -0,0 +1,27 @@
package com.test.payment.dto;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import lombok.Data;
/**
* MTN MoMo request-to-pay resource (GET /collection/v1_0/requesttopay/{referenceId})
* and the shape MTN posts to the callback URL. status: PENDING | SUCCESSFUL | FAILED.
* The callback may carry referenceId; externalId is set by us to the same reference.
*/
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class MtnStatusResponseDto {
private String referenceId;
private String externalId;
private String amount;
private String currency;
private String status;
private Object reason;
private String financialTransactionId;
private MtnPayRequestDto.Payer payer;
public String reasonText() {
return reason == null ? null : String.valueOf(reason);
}
}

28
src/main/java/com/test/payment/dto/OAuth2TokenResponse.java Normal file
View File

@@ -0,0 +1,28 @@
package com.test.payment.dto;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
/**
* Generic OAuth2 client-credentials token response (Airtel, MTN).
* expires_in arrives as a number from MTN and sometimes a string from Airtel.
*/
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class OAuth2TokenResponse {
@JsonProperty("access_token")
private String accessToken;
@JsonProperty("expires_in")
private Object expiresIn;
public long expiresInSeconds() {
try {
return Long.parseLong(String.valueOf(expiresIn).trim());
} catch (NumberFormatException e) {
return 3600;
}
}
}

28
src/main/java/com/test/payment/dto/PaymentRequest.java Normal file
View File

@@ -0,0 +1,28 @@
package com.test.payment.dto;
import jakarta.validation.constraints.Min;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;
import jakarta.validation.constraints.Pattern;
import jakarta.validation.constraints.Size;
import lombok.Data;
@Data
public class PaymentRequest {
@NotBlank(message = "phoneNumber is required")
@Pattern(regexp = "^\\d{9,14}$", message = "phoneNumber must be an MSISDN of 9-14 digits, e.g. 254712345678")
private String phoneNumber;
@NotNull(message = "amount is required")
@Min(value = 1, message = "amount must be at least 1")
private Integer amount;
@NotBlank(message = "accountReference is required")
@Size(max = 12, message = "accountReference must be at most 12 characters")
private String accountReference;
@NotBlank(message = "transactionDesc is required")
@Size(max = 13, message = "transactionDesc must be at most 13 characters")
private String transactionDesc;
}

22
src/main/java/com/test/payment/dto/PaymentResultDto.java Normal file
View File

@@ -0,0 +1,22 @@
package com.test.payment.dto;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class PaymentResultDto {
private Long initiationId;
private String provider;
private String status;
private String providerReference;
private String secondaryReference;
private String responseCode;
private String responseDescription;
private String customerMessage;
}

63
src/main/java/com/test/payment/dto/StkCallbackPayload.java Normal file
View File

@@ -0,0 +1,63 @@
package com.test.payment.dto;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
import java.util.List;
/**
* Safaricom STK Push result callback payload:
* {"Body":{"stkCallback":{"MerchantRequestID":...,"CheckoutRequestID":...,"ResultCode":0,
* "ResultDesc":...,"CallbackMetadata":{"Item":[{"Name":"Amount","Value":1.0}, ...]}}}}
*/
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class StkCallbackPayload {
@JsonProperty("Body")
private Body body;
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public static class Body {
@JsonProperty("stkCallback")
private StkCallbackBody stkCallback;
}
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public static class StkCallbackBody {
@JsonProperty("MerchantRequestID")
private String merchantRequestId;
@JsonProperty("CheckoutRequestID")
private String checkoutRequestId;
@JsonProperty("ResultCode")
private Integer resultCode;
@JsonProperty("ResultDesc")
private String resultDesc;
@JsonProperty("CallbackMetadata")
private CallbackMetadata callbackMetadata;
}
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public static class CallbackMetadata {
@JsonProperty("Item")
private List<Item> item;
}
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public static class Item {
@JsonProperty("Name")
private String name;
@JsonProperty("Value")
private Object value;
}
}

24
src/main/java/com/test/payment/dto/StkQueryRequestDto.java Normal file
View File

@@ -0,0 +1,24 @@
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 StkQueryRequestDto {
@JsonProperty("BusinessShortCode")
private Long businessShortCode;
@JsonProperty("Password")
private String password;
@JsonProperty("Timestamp")
private String timestamp;
@JsonProperty("CheckoutRequestID")
private String checkoutRequestId;
}

28
src/main/java/com/test/payment/dto/StkQueryResponseDto.java Normal file
View File

@@ -0,0 +1,28 @@
package com.test.payment.dto;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class StkQueryResponseDto {
@JsonProperty("ResponseCode")
private String responseCode;
@JsonProperty("ResponseDescription")
private String responseDescription;
@JsonProperty("MerchantRequestID")
private String merchantRequestId;
@JsonProperty("CheckoutRequestID")
private String checkoutRequestId;
@JsonProperty("ResultCode")
private String resultCode;
@JsonProperty("ResultDesc")
private String resultDesc;
}

30
src/main/java/com/test/payment/dto/TransactionStatusDto.java Normal file
View File

@@ -0,0 +1,30 @@
package com.test.payment.dto;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.math.BigDecimal;
import java.time.LocalDateTime;
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class TransactionStatusDto {
private Long initiationId;
private String provider;
private String providerReference;
private String secondaryReference;
private String status;
private String phoneNumber;
private BigDecimal amount;
private String accountReference;
private String resultCode;
private String resultDesc;
private String receiptNumber;
private LocalDateTime createdAt;
private LocalDateTime updatedAt;
}

8
src/main/java/com/test/payment/exceptions/MpesaBusyException.java
View File

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

8
src/main/java/com/test/payment/exceptions/MpesaPermanentException.java
View File

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

7
src/main/java/com/test/payment/exceptions/MpesaTransientException.java
View File

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

8
src/main/java/com/test/payment/exceptions/ProviderBusyException.java Normal file
View File

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

8
src/main/java/com/test/payment/exceptions/ProviderPermanentException.java Normal file
View File

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

11
src/main/java/com/test/payment/exceptions/ProviderProcessingException.java Normal file
View File

@@ -0,0 +1,11 @@
package com.test.payment.exceptions;
/**
* The transaction is still being processed by M-Pesa (STK query errorCode 500.001.1001).
* Not a failure — the transaction stays PENDING until the callback or a later status check.
*/
public class ProviderProcessingException extends RuntimeException {
public ProviderProcessingException(String msg) {
super(msg);
}
}

7
src/main/java/com/test/payment/exceptions/ProviderTransientException.java Normal file
View File

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

23
src/main/java/com/test/payment/jobs/MpesaTransactionJob.java
View File

@@ -1,23 +0,0 @@
package com.test.payment.jobs;
import com.test.payment.repository.MpesaRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
@Component
@RequiredArgsConstructor
@Slf4j
public class MpesaTransactionJob {
private final MpesaRepository mpesaRepository;
@Scheduled(fixedDelay = 60000)
public void pullTransactions() {
mpesaRepository.findAll()
.doOnNext(tx -> log.info("Checking transaction: {}", tx))
.subscribe();
}
}

80
src/main/java/com/test/payment/jobs/PaymentReconciliationJob.java Normal file
View File

@@ -0,0 +1,80 @@
package com.test.payment.jobs;
import com.test.payment.models.TransactionStatus;
import com.test.payment.repository.PaymentInitiationRepository;
import com.test.payment.service.PaymentLifecycleService;
import com.test.payment.service.PaymentProviderService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.convert.DurationStyle;
import org.springframework.core.env.Environment;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import java.util.stream.Collectors;
/**
* Reconciles payments (any provider) stuck in PENDING longer than
* payments.reconciliation.pending-age by querying the provider for the real
* outcome and updating the database.
*/
@Component
@Slf4j
public class PaymentReconciliationJob {
private final PaymentInitiationRepository initiationRepository;
private final PaymentLifecycleService lifecycle;
private final Environment environment;
private final Map<String, PaymentProviderService> servicesByProvider;
private final AtomicBoolean running = new AtomicBoolean(false);
public PaymentReconciliationJob(PaymentInitiationRepository initiationRepository,
PaymentLifecycleService lifecycle,
Environment environment,
List<PaymentProviderService> providerServices) {
this.initiationRepository = initiationRepository;
this.lifecycle = lifecycle;
this.environment = environment;
this.servicesByProvider = providerServices.stream()
.collect(Collectors.toMap(PaymentProviderService::provider, Function.identity()));
}
@Scheduled(fixedDelayString = "${payments.reconciliation.fixed-delay:60s}")
public void reconcilePendingTransactions() {
if (!running.compareAndSet(false, true)) {
log.debug("Previous reconciliation run still in progress — skipping this tick");
return;
}
Duration pendingAge = DurationStyle.detectAndParse(
environment.getProperty("payments.reconciliation.pending-age", "5m"));
LocalDateTime cutoff = LocalDateTime.now().minus(pendingAge);
initiationRepository.findByStatusAndCreatedAtBefore(TransactionStatus.PENDING.name(), cutoff)
.concatMap(initiation -> {
PaymentProviderService service = servicesByProvider.get(initiation.getProvider());
if (service == null) {
log.error("No provider service registered for {} (initiation {})",
initiation.getProvider(), initiation.getId());
return Mono.empty();
}
return lifecycle.reconcile(initiation, service::checkStatus)
.doOnSubscribe(s -> log.info("[{}] reconciling PENDING initiation {} (created {})",
initiation.getProvider(), initiation.getId(), initiation.getCreatedAt()))
.onErrorResume(e -> {
log.error("[{}] reconciliation failed for initiation {}: {}",
initiation.getProvider(), initiation.getId(), e.toString());
return Mono.empty();
});
})
.doFinally(signal -> running.set(false))
.subscribe(
v -> { },
e -> log.error("Reconciliation run aborted: {}", e.toString()));
}
}

100
src/main/java/com/test/payment/models/DatabaseSchema.java Normal file
View File

@@ -0,0 +1,100 @@
package com.test.payment.models;
import java.util.List;
/**
* The database schema as code — executed statement-by-statement at startup by
* DatabaseSchemaInitializer (replaces the old classpath schema.sql).
*/
public final class DatabaseSchema {
private DatabaseSchema() {
}
public static final List<String> STATEMENTS = List.of(
"""
CREATE TABLE IF NOT EXISTS provider_tokens (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
provider VARCHAR(20) NOT NULL,
access_token VARCHAR(512) NOT NULL,
expires_at TIMESTAMP NOT NULL,
created_at TIMESTAMP NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS payment_initiations (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
provider VARCHAR(20) NOT NULL,
phone_number VARCHAR(15) NOT NULL,
amount DECIMAL(10,2) NOT NULL,
account_reference VARCHAR(50),
transaction_desc VARCHAR(100),
status VARCHAR(20) NOT NULL,
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP
)
""",
"""
CREATE INDEX IF NOT EXISTS idx_initiations_status_created
ON payment_initiations (status, created_at)
""",
// One response per initiation (UNIQUE on initiation_id enforces the 1:1 link).
// provider_reference is the ID used for callbacks/status checks
// (M-Pesa CheckoutRequestID, Airtel transaction id, MTN X-Reference-Id).
"""
CREATE TABLE IF NOT EXISTS payment_responses (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
initiation_id BIGINT NOT NULL UNIQUE,
provider VARCHAR(20) NOT NULL,
provider_reference VARCHAR(100) UNIQUE,
secondary_reference VARCHAR(100),
response_code VARCHAR(30),
response_description VARCHAR(255),
customer_message VARCHAR(255),
created_at TIMESTAMP NOT NULL,
CONSTRAINT fk_response_initiation FOREIGN KEY (initiation_id) REFERENCES payment_initiations (id)
)
""",
// One callback per initiation (UNIQUE on initiation_id enforces the 1:1 link)
"""
CREATE TABLE IF NOT EXISTS payment_callbacks (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
initiation_id BIGINT NOT NULL UNIQUE,
provider VARCHAR(20) NOT NULL,
provider_reference VARCHAR(100),
result_code VARCHAR(30),
result_desc VARCHAR(255),
receipt_number VARCHAR(50),
amount DECIMAL(10,2),
phone_number VARCHAR(15),
transaction_date VARCHAR(20),
raw_payload CLOB,
created_at TIMESTAMP NOT NULL,
CONSTRAINT fk_callback_initiation FOREIGN KEY (initiation_id) REFERENCES payment_initiations (id)
)
""",
// Consolidated transaction record, written when an initiation reaches a
// terminal state (SUCCESS/FAILED). UNIQUE initiation_id: one per initiation.
"""
CREATE TABLE IF NOT EXISTS transactions (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
initiation_id BIGINT NOT NULL UNIQUE,
provider VARCHAR(20) NOT NULL,
provider_reference VARCHAR(100),
secondary_reference VARCHAR(100),
phone_number VARCHAR(15),
amount DECIMAL(10,2),
account_reference VARCHAR(50),
status VARCHAR(20) NOT NULL,
result_code VARCHAR(30),
result_desc VARCHAR(255),
receipt_number VARCHAR(50),
transaction_date VARCHAR(20),
resolved_by VARCHAR(20),
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP,
CONSTRAINT fk_transaction_initiation FOREIGN KEY (initiation_id) REFERENCES payment_initiations (id)
)
"""
);
}

33
src/main/java/com/test/payment/models/PaymentCallback.java Normal file
View File

@@ -0,0 +1,33 @@
package com.test.payment.models;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.data.annotation.Id;
import org.springframework.data.relational.core.mapping.Table;
import java.math.BigDecimal;
import java.time.LocalDateTime;
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@Table("PAYMENT_CALLBACKS")
public class PaymentCallback {
@Id
private Long id;
private Long initiationId;
private String provider;
private String providerReference;
private String resultCode;
private String resultDesc;
private String receiptNumber;
private BigDecimal amount;
private String phoneNumber;
private String transactionDate;
private String rawPayload;
private LocalDateTime createdAt;
}

30
src/main/java/com/test/payment/models/PaymentInitiation.java Normal file
View File

@@ -0,0 +1,30 @@
package com.test.payment.models;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.data.annotation.Id;
import org.springframework.data.relational.core.mapping.Table;
import java.math.BigDecimal;
import java.time.LocalDateTime;
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@Table("PAYMENT_INITIATIONS")
public class PaymentInitiation {
@Id
private Long id;
private String provider;
private String phoneNumber;
private BigDecimal amount;
private String accountReference;
private String transactionDesc;
private String status;
private LocalDateTime createdAt;
private LocalDateTime updatedAt;
}

7
src/main/java/com/test/payment/models/PaymentProviderType.java Normal file
View File

@@ -0,0 +1,7 @@
package com.test.payment.models;
public enum PaymentProviderType {
MPESA,
AIRTEL,
MTN
}

12
src/main/java/com/test/payment/models/PaymentRequest.java
View File

@@ -1,12 +0,0 @@
package com.test.payment.models;
import lombok.Data;
@Data
public class PaymentRequest {
private long phoneNumber;
private int amount;
private String accountReference;
private String transactionDesc;
}

35
src/main/java/com/test/payment/models/PaymentResponse.java Normal file
View File

@@ -0,0 +1,35 @@
package com.test.payment.models;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.data.annotation.Id;
import org.springframework.data.relational.core.mapping.Table;
import java.time.LocalDateTime;
/**
* The provider's answer to an initiation. providerReference is the ID later used
* for callbacks and status checks (M-Pesa CheckoutRequestID, Airtel transaction id,
* MTN X-Reference-Id); secondaryReference is any additional provider ID
* (M-Pesa MerchantRequestID).
*/
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@Table("PAYMENT_RESPONSES")
public class PaymentResponse {
@Id
private Long id;
private Long initiationId;
private String provider;
private String providerReference;
private String secondaryReference;
private String responseCode;
private String responseDescription;
private String customerMessage;
private LocalDateTime createdAt;
}

25
src/main/java/com/test/payment/models/ProviderToken.java Normal file
View File

@@ -0,0 +1,25 @@
package com.test.payment.models;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.data.annotation.Id;
import org.springframework.data.relational.core.mapping.Table;
import java.time.LocalDateTime;
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@Table("PROVIDER_TOKENS")
public class ProviderToken {
@Id
private Long id;
private String provider;
private String accessToken;
private LocalDateTime expiresAt;
private LocalDateTime createdAt;
}

35
src/main/java/com/test/payment/models/Transaction.java
View File

@@ -1,23 +1,42 @@
package com.test.payment.models;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.data.annotation.Id;
import org.springframework.data.relational.core.mapping.*;
import org.springframework.data.relational.core.mapping.Table;
import java.math.BigDecimal;
import java.time.LocalDateTime;
/**
* Consolidated record written when a payment attempt reaches a terminal state
* (SUCCESS or FAILED) — one row per initiation, whatever path resolved it
* (callback, status query, immediate rejection, or reconciliation timeout).
*/
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@Table("transactions")
@Table("TRANSACTIONS")
public class Transaction {
@Id
private String id;
private long phoneNumber;
private long amount;
private Long id;
private Long initiationId;
private String provider;
private String providerReference;
private String secondaryReference;
private String phoneNumber;
private BigDecimal amount;
private String accountReference;
private String status;
private String checkoutRequestId;
private String resultCode;
private String resultDesc;
private String receiptNumber;
private String transactionDate;
private String resolvedBy;
private LocalDateTime createdAt;
private LocalDateTime updatedAt;
}

7
src/main/java/com/test/payment/models/TransactionStatus.java Normal file
View File

@@ -0,0 +1,7 @@
package com.test.payment.models;
public enum TransactionStatus {
PENDING,
SUCCESS,
FAILED
}

39
src/main/java/com/test/payment/repository/MpesaRepository.java
View File

@@ -1,39 +0,0 @@
package com.test.payment.repository;
import com.test.payment.models.Transaction;
//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 interface MpesaRepository extends ReactiveCrudRepository<Transaction, Long> {
/* public Mono<Void> saveTransaction(Transaction transaction) {
return Mono.fromRunnable(() ->
redisTemplate.opsForHash().put("mpesa:transactions", transaction.getId(), transaction.getStatus())
).then();
}
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;
}
*/
}

12
src/main/java/com/test/payment/repository/PaymentCallbackRepository.java Normal file
View File

@@ -0,0 +1,12 @@
package com.test.payment.repository;
import com.test.payment.models.PaymentCallback;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Mono;
@Repository
public interface PaymentCallbackRepository extends ReactiveCrudRepository<PaymentCallback, Long> {
Mono<PaymentCallback> findByInitiationId(Long initiationId);
}

14
src/main/java/com/test/payment/repository/PaymentInitiationRepository.java Normal file
View File

@@ -0,0 +1,14 @@
package com.test.payment.repository;
import com.test.payment.models.PaymentInitiation;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Flux;
import java.time.LocalDateTime;
@Repository
public interface PaymentInitiationRepository extends ReactiveCrudRepository<PaymentInitiation, Long> {
Flux<PaymentInitiation> findByStatusAndCreatedAtBefore(String status, LocalDateTime cutoff);
}

14
src/main/java/com/test/payment/repository/PaymentResponseRepository.java Normal file
View File

@@ -0,0 +1,14 @@
package com.test.payment.repository;
import com.test.payment.models.PaymentResponse;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Mono;
@Repository
public interface PaymentResponseRepository extends ReactiveCrudRepository<PaymentResponse, Long> {
Mono<PaymentResponse> findByInitiationId(Long initiationId);
Mono<PaymentResponse> findByProviderReference(String providerReference);
}

16
src/main/java/com/test/payment/repository/ProviderTokenRepository.java Normal file
View File

@@ -0,0 +1,16 @@
package com.test.payment.repository;
import com.test.payment.models.ProviderToken;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Mono;
import java.time.LocalDateTime;
@Repository
public interface ProviderTokenRepository extends ReactiveCrudRepository<ProviderToken, Long> {
Mono<ProviderToken> findFirstByProviderAndExpiresAtAfterOrderByIdDesc(String provider, LocalDateTime cutoff);
Mono<Void> deleteByProvider(String provider);
}

17
src/main/java/com/test/payment/repository/TransactionRepository.java Normal file
View File

@@ -0,0 +1,17 @@
package com.test.payment.repository;
import com.test.payment.models.Transaction;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@Repository
public interface TransactionRepository extends ReactiveCrudRepository<Transaction, Long> {
Mono<Transaction> findByInitiationId(Long initiationId);
Flux<Transaction> findByProvider(String provider);
Flux<Transaction> findByStatus(String status);
}

138
src/main/java/com/test/payment/service/AirtelService.java Normal file
View File

@@ -0,0 +1,138 @@
package com.test.payment.service;
import com.test.payment.client.AirtelClient;
import com.test.payment.dto.AirtelCallbackPayload;
import com.test.payment.dto.AirtelPaymentRequestDto;
import com.test.payment.dto.AirtelResponseDto;
import com.test.payment.dto.CallbackAckDto;
import com.test.payment.dto.PaymentRequest;
import com.test.payment.dto.PaymentResultDto;
import com.test.payment.dto.TransactionStatusDto;
import com.test.payment.exceptions.ProviderBusyException;
import com.test.payment.models.PaymentProviderType;
import com.test.payment.models.PaymentResponse;
import com.test.payment.models.TransactionStatus;
import com.test.payment.service.PaymentLifecycleService.CallbackData;
import com.test.payment.service.PaymentLifecycleService.ProviderResponseData;
import com.test.payment.service.PaymentLifecycleService.QueryOutcome;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import tools.jackson.databind.ObjectMapper;
import java.util.UUID;
/**
* Airtel Money USSD-push collections. Transaction status codes:
* TIP = in progress, TS = success, TF = failed.
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class AirtelService implements PaymentProviderService {
private final AirtelClient airtelClient;
private final PaymentLifecycleService lifecycle;
private final Environment environment;
private final ObjectMapper objectMapper;
@Override
public String provider() {
return PaymentProviderType.AIRTEL.name();
}
@Override
public Mono<PaymentResultDto> initiatePayment(PaymentRequest request) {
return lifecycle.saveInitiation(provider(), request)
.flatMap(initiation -> Mono.defer(() -> {
String reference = "ATL" + UUID.randomUUID().toString().replace("-", "");
return airtelClient.pay(buildRequest(request, reference))
.map(response -> toResponseData(response, reference));
})
.flatMap(data -> lifecycle.persistResponse(initiation, data))
.onErrorResume(ex -> lifecycle.markFailed(initiation, ex)));
}
public Mono<CallbackAckDto> handleCallback(AirtelCallbackPayload payload) {
if (payload == null || payload.getTransaction() == null) {
log.warn("Received malformed Airtel callback payload");
return Mono.just(CallbackAckDto.accepted("Ignored: empty callback"));
}
AirtelCallbackPayload.Transaction transaction = payload.getTransaction();
boolean success = "TS".equalsIgnoreCase(transaction.getStatusCode());
CallbackData data = new CallbackData(
transaction.getId(),
transaction.getStatusCode(),
transaction.getMessage(),
transaction.getAirtelMoneyId(),
null, null, null,
success);
return lifecycle.applyCallback(provider(), data, toJson(payload));
}
@Override
public Mono<TransactionStatusDto> checkStatus(String providerReference) {
return lifecycle.checkStatus(provider(), providerReference, this::queryProvider);
}
private Mono<QueryOutcome> queryProvider(PaymentResponse response) {
return airtelClient.status(response.getProviderReference())
.map(result -> {
AirtelResponseDto.Transaction tx = result.getData() != null ? result.getData().getTransaction() : null;
String status = tx != null ? tx.getStatus() : null;
String message = tx != null && tx.getMessage() != null
? tx.getMessage()
: (result.getStatus() != null ? result.getStatus().getMessage() : null);
return new QueryOutcome(mapStatus(status), status, message,
tx != null ? tx.getAirtelMoneyId() : null);
})
.onErrorResume(ProviderBusyException.class,
e -> Mono.just(QueryOutcome.pending("Airtel status query rate-limited — showing last known state")));
}
private AirtelPaymentRequestDto buildRequest(PaymentRequest request, String reference) {
String country = environment.getProperty("airtel.country", "KE");
String currency = environment.getProperty("airtel.currency", "KES");
return new AirtelPaymentRequestDto(
request.getAccountReference(),
new AirtelPaymentRequestDto.Subscriber(country, currency, request.getPhoneNumber()),
new AirtelPaymentRequestDto.Transaction(
String.valueOf(request.getAmount()), country, currency, reference));
}
private ProviderResponseData toResponseData(AirtelResponseDto response, String reference) {
AirtelResponseDto.Status status = response.getStatus();
boolean accepted = status != null && Boolean.TRUE.equals(status.getSuccess());
String transactionStatus = response.getData() != null && response.getData().getTransaction() != null
? response.getData().getTransaction().getStatus()
: null;
return new ProviderResponseData(
reference,
null,
status != null ? status.getCode() : null,
status != null ? status.getMessage() : null,
transactionStatus,
accepted);
}
private TransactionStatus mapStatus(String airtelStatus) {
if ("TS".equalsIgnoreCase(airtelStatus)) {
return TransactionStatus.SUCCESS;
}
if ("TF".equalsIgnoreCase(airtelStatus)) {
return TransactionStatus.FAILED;
}
return TransactionStatus.PENDING; // TIP or unknown — keep waiting
}
private String toJson(Object value) {
try {
return objectMapper.writeValueAsString(value);
} catch (Exception e) {
log.warn("Could not serialize callback payload: {}", e.toString());
return null;
}
}
}

42
src/main/java/com/test/payment/service/AirtelTokenService.java Normal file
View File

@@ -0,0 +1,42 @@
package com.test.payment.service;
import com.test.payment.dto.OAuth2TokenResponse;
import com.test.payment.models.PaymentProviderType;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import java.util.Map;
@Service
@RequiredArgsConstructor
@Slf4j
public class AirtelTokenService {
private final WebClient airtelWebClient;
private final TokenCacheService tokenCache;
private final Environment environment;
public Mono<String> getToken() {
return tokenCache.getToken(PaymentProviderType.AIRTEL.name(), this::fetchToken);
}
public Mono<Void> evictToken() {
return tokenCache.evictToken(PaymentProviderType.AIRTEL.name());
}
private Mono<TokenCacheService.FetchedToken> fetchToken() {
return airtelWebClient.post()
.uri("/auth/oauth2/token")
.bodyValue(Map.of(
"client_id", environment.getProperty("airtel.client-id", ""),
"client_secret", environment.getProperty("airtel.client-secret", ""),
"grant_type", "client_credentials"))
.retrieve()
.bodyToMono(OAuth2TokenResponse.class)
.map(resp -> new TokenCacheService.FetchedToken(resp.getAccessToken(), resp.expiresInSeconds()));
}
}

294
src/main/java/com/test/payment/service/MpesaService.java
View File

@@ -1,204 +1,156 @@
package com.test.payment.service;
import com.test.payment.client.MpesaClient;
import com.test.payment.dto.CallbackAckDto;
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.MpesaResponse;
import com.test.payment.models.PaymentRequest;
import com.test.payment.repository.MpesaRepository;
import com.test.payment.dto.MpesaResponse;
import com.test.payment.dto.PaymentRequest;
import com.test.payment.dto.PaymentResultDto;
import com.test.payment.dto.StkCallbackPayload;
import com.test.payment.dto.StkQueryRequestDto;
import com.test.payment.dto.TransactionStatusDto;
import com.test.payment.exceptions.ProviderBusyException;
import com.test.payment.exceptions.ProviderProcessingException;
import com.test.payment.models.PaymentProviderType;
import com.test.payment.models.PaymentResponse;
import com.test.payment.models.TransactionStatus;
import com.test.payment.service.PaymentLifecycleService.CallbackData;
import com.test.payment.service.PaymentLifecycleService.ProviderResponseData;
import com.test.payment.service.PaymentLifecycleService.QueryOutcome;
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;
import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker;
import io.github.resilience4j.ratelimiter.annotation.RateLimiter;
import io.github.resilience4j.retry.annotation.Retry;
import tools.jackson.databind.ObjectMapper;
import java.time.Duration;
import java.util.Base64;
import java.math.BigDecimal;
import java.util.Map;
import java.util.stream.Collectors;
@Service
@RequiredArgsConstructor
@Slf4j
public class MpesaService {
public class MpesaService implements PaymentProviderService {
private final WebClient mpesaWebClient;
private final MpesaRepository mpesaRepository;
private final MpesaClient mpesaClient;
private final PaymentLifecycleService lifecycle;
private final Environment environment;
private final ObjectMapper objectMapper;
public Mono<String> getToken() {
return Mono.just("hVG5ybD47dHUMQ6RfWby2ZpIs1Ul");
@Override
public String provider() {
return PaymentProviderType.MPESA.name();
}
@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");
@Override
public Mono<PaymentResultDto> initiatePayment(PaymentRequest request) {
return lifecycle.saveInitiation(provider(), request)
.flatMap(initiation -> Mono.defer(() -> mpesaClient.stkPush(buildStkRequest(request)))
.flatMap(response -> lifecycle.persistResponse(initiation, toResponseData(response)))
.onErrorResume(ex -> lifecycle.markFailed(initiation, ex)));
}
public Mono<CallbackAckDto> handleCallback(StkCallbackPayload payload) {
if (payload == null || payload.getBody() == null || payload.getBody().getStkCallback() == null) {
log.warn("Received malformed STK callback payload");
return Mono.just(CallbackAckDto.accepted("Ignored: empty callback"));
}
StkCallbackPayload.StkCallbackBody callback = payload.getBody().getStkCallback();
Map<String, Object> metadata = extractMetadata(callback);
boolean success = callback.getResultCode() != null && callback.getResultCode() == 0;
CallbackData data = new CallbackData(
callback.getCheckoutRequestId(),
callback.getResultCode() == null ? null : String.valueOf(callback.getResultCode()),
callback.getResultDesc(),
asString(metadata.get("MpesaReceiptNumber")),
metadata.get("Amount") != null ? new BigDecimal(metadata.get("Amount").toString()) : null,
asString(metadata.get("PhoneNumber")),
asString(metadata.get("TransactionDate")),
success);
return lifecycle.applyCallback(provider(), data, toJson(payload));
}
@Override
public Mono<TransactionStatusDto> checkStatus(String providerReference) {
return lifecycle.checkStatus(provider(), providerReference, this::queryProvider);
}
private Mono<QueryOutcome> queryProvider(PaymentResponse response) {
String shortCode = environment.getProperty("mpesa.business-short-code");
String passkey = environment.getProperty("mpesa.pass-key");
String callback = "https://mydomain.com/path";
MpesaUtils.MpesaAuthData auth = MpesaUtils.generateAuthData(shortCode, passkey);
StkQueryRequestDto query = new StkQueryRequestDto(
Long.valueOf(shortCode), auth.getPassword(), auth.getTimestamp(), response.getProviderReference());
MpesaUtils.MpesaAuthData authData = MpesaUtils.generateAuthData(businessShortCode, passkey);
return mpesaClient.stkQuery(query)
.map(result -> new QueryOutcome(
mapQueryResult(result.getResultCode()), result.getResultCode(), result.getResultDesc(), null))
.onErrorResume(ProviderProcessingException.class,
e -> Mono.just(QueryOutcome.pending("Transaction is still being processed by M-Pesa")))
.onErrorResume(ProviderBusyException.class,
e -> Mono.just(QueryOutcome.pending("M-Pesa status query rate-limited — showing last known state")));
}
MpesaRequestDto mpesaRequestDto = new MpesaRequestDto(
Long.valueOf(businessShortCode),
authData.getPassword(),
authData.getTimestamp(),
private MpesaRequestDto buildStkRequest(PaymentRequest request) {
String shortCode = environment.getProperty("mpesa.business-short-code");
String passkey = environment.getProperty("mpesa.pass-key");
String callbackUrl = environment.getProperty("mpesa.callback-url");
MpesaUtils.MpesaAuthData auth = MpesaUtils.generateAuthData(shortCode, passkey);
Long phone = Long.valueOf(request.getPhoneNumber());
return new MpesaRequestDto(
Long.valueOf(shortCode),
auth.getPassword(),
auth.getTimestamp(),
"CustomerPayBillOnline",
request.getAmount(),
request.getPhoneNumber(),
Long.valueOf(businessShortCode),
request.getPhoneNumber(),
callback,
phone,
Long.valueOf(shortCode),
phone,
callbackUrl,
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)
);
private ProviderResponseData toResponseData(MpesaResponse response) {
boolean accepted = "0".equals(response.getResponseCode());
return new ProviderResponseData(
response.getCheckoutRequestId(),
response.getMerchantRequestID(),
response.getResponseCode(),
response.getResponseDescription(),
response.getCustomerMessage(),
accepted);
}
// 🧯 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
));
private TransactionStatus mapQueryResult(String resultCode) {
if (resultCode == null) {
return TransactionStatus.PENDING;
}
return "0".equals(resultCode) ? TransactionStatus.SUCCESS : TransactionStatus.FAILED;
}
private Map<String, Object> extractMetadata(StkCallbackPayload.StkCallbackBody callback) {
if (callback.getCallbackMetadata() == null || callback.getCallbackMetadata().getItem() == null) {
return Map.of();
}
return callback.getCallbackMetadata().getItem().stream()
.filter(item -> item.getName() != null && item.getValue() != null)
.collect(Collectors.toMap(StkCallbackPayload.Item::getName, StkCallbackPayload.Item::getValue, (a, b) -> a));
}
private String asString(Object value) {
return value == null ? null : value.toString();
}
private String toJson(Object value) {
try {
return objectMapper.writeValueAsString(value);
} catch (Exception e) {
log.warn("Could not serialize callback payload: {}", e.toString());
return null;
}
}
}

190
src/main/java/com/test/payment/service/MpesaServiceaa.java
View File

@@ -1,190 +0,0 @@
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)
);
}*/
}

39
src/main/java/com/test/payment/service/MpesaTokenService.java
View File

@@ -1,27 +1,40 @@
package com.test.payment.service;
import com.test.payment.dto.MpesaTokenResponse;
import com.test.payment.models.PaymentProviderType;
import lombok.RequiredArgsConstructor;
//import org.springframework.data.redis.core.RedisTemplate;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import java.time.Duration;
@Service
@RequiredArgsConstructor
@Slf4j
public class MpesaTokenService {
// private final RedisTemplate<String, String> redisTemplate;
private final WebClient mpesaWebClient;
private final TokenCacheService tokenCache;
private final Environment environment;
public Mono<String> getToken() {
/* 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));*/
return Mono.just("newToken");
return tokenCache.getToken(PaymentProviderType.MPESA.name(), this::fetchToken);
}
public Mono<Void> evictToken() {
return tokenCache.evictToken(PaymentProviderType.MPESA.name());
}
private Mono<TokenCacheService.FetchedToken> fetchToken() {
return mpesaWebClient.get()
.uri("/oauth/v1/generate?grant_type=client_credentials")
.headers(h -> h.setBasicAuth(
environment.getProperty("mpesa.consumer-key", ""),
environment.getProperty("mpesa.consumer-secret", "")))
.retrieve()
.bodyToMono(MpesaTokenResponse.class)
.map(resp -> new TokenCacheService.FetchedToken(
resp.getAccessToken(), Long.parseLong(resp.getExpiresIn().trim())));
}
}

124
src/main/java/com/test/payment/service/MtnService.java Normal file
View File

@@ -0,0 +1,124 @@
package com.test.payment.service;
import com.test.payment.client.MtnClient;
import com.test.payment.dto.CallbackAckDto;
import com.test.payment.dto.MtnPayRequestDto;
import com.test.payment.dto.MtnStatusResponseDto;
import com.test.payment.dto.PaymentRequest;
import com.test.payment.dto.PaymentResultDto;
import com.test.payment.dto.TransactionStatusDto;
import com.test.payment.exceptions.ProviderBusyException;
import com.test.payment.models.PaymentProviderType;
import com.test.payment.models.PaymentResponse;
import com.test.payment.models.TransactionStatus;
import com.test.payment.service.PaymentLifecycleService.CallbackData;
import com.test.payment.service.PaymentLifecycleService.ProviderResponseData;
import com.test.payment.service.PaymentLifecycleService.QueryOutcome;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import tools.jackson.databind.ObjectMapper;
import java.math.BigDecimal;
import java.util.UUID;
/**
* MTN MoMo Collections (request-to-pay). We generate the X-Reference-Id UUID and use
* it as the provider reference; externalId is set to the same value so callbacks can
* be correlated. Status values: PENDING | SUCCESSFUL | FAILED.
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class MtnService implements PaymentProviderService {
private final MtnClient mtnClient;
private final PaymentLifecycleService lifecycle;
private final Environment environment;
private final ObjectMapper objectMapper;
@Override
public String provider() {
return PaymentProviderType.MTN.name();
}
@Override
public Mono<PaymentResultDto> initiatePayment(PaymentRequest request) {
return lifecycle.saveInitiation(provider(), request)
.flatMap(initiation -> Mono.defer(() -> {
String reference = UUID.randomUUID().toString();
// 202 Accepted, empty body — the reference is all we get back
return mtnClient.requestToPay(reference, buildRequest(request, reference))
.thenReturn(new ProviderResponseData(
reference, null, "202", "Accepted", "Request to pay accepted", true));
})
.flatMap(data -> lifecycle.persistResponse(initiation, data))
.onErrorResume(ex -> lifecycle.markFailed(initiation, ex)));
}
public Mono<CallbackAckDto> handleCallback(MtnStatusResponseDto payload) {
if (payload == null || (payload.getReferenceId() == null && payload.getExternalId() == null)) {
log.warn("Received malformed MTN callback payload");
return Mono.just(CallbackAckDto.accepted("Ignored: empty callback"));
}
String reference = payload.getReferenceId() != null ? payload.getReferenceId() : payload.getExternalId();
boolean success = "SUCCESSFUL".equalsIgnoreCase(payload.getStatus());
CallbackData data = new CallbackData(
reference,
payload.getStatus(),
success ? "Payment successful" : payload.reasonText(),
payload.getFinancialTransactionId(),
payload.getAmount() != null ? new BigDecimal(payload.getAmount()) : null,
payload.getPayer() != null ? payload.getPayer().getPartyId() : null,
null,
success);
return lifecycle.applyCallback(provider(), data, toJson(payload));
}
@Override
public Mono<TransactionStatusDto> checkStatus(String providerReference) {
return lifecycle.checkStatus(provider(), providerReference, this::queryProvider);
}
private Mono<QueryOutcome> queryProvider(PaymentResponse response) {
return mtnClient.status(response.getProviderReference())
.map(result -> new QueryOutcome(
mapStatus(result.getStatus()),
result.getStatus(),
"SUCCESSFUL".equalsIgnoreCase(result.getStatus()) ? "Payment successful" : result.reasonText(),
result.getFinancialTransactionId()))
.onErrorResume(ProviderBusyException.class,
e -> Mono.just(QueryOutcome.pending("MTN status query rate-limited — showing last known state")));
}
private MtnPayRequestDto buildRequest(PaymentRequest request, String reference) {
return new MtnPayRequestDto(
String.valueOf(request.getAmount()),
environment.getProperty("mtn.currency", "EUR"),
reference,
new MtnPayRequestDto.Payer("MSISDN", request.getPhoneNumber()),
request.getTransactionDesc(),
request.getAccountReference());
}
private TransactionStatus mapStatus(String mtnStatus) {
if ("SUCCESSFUL".equalsIgnoreCase(mtnStatus)) {
return TransactionStatus.SUCCESS;
}
if ("FAILED".equalsIgnoreCase(mtnStatus)) {
return TransactionStatus.FAILED;
}
return TransactionStatus.PENDING;
}
private String toJson(Object value) {
try {
return objectMapper.writeValueAsString(value);
} catch (Exception e) {
log.warn("Could not serialize callback payload: {}", e.toString());
return null;
}
}
}

42
src/main/java/com/test/payment/service/MtnTokenService.java Normal file
View File

@@ -0,0 +1,42 @@
package com.test.payment.service;
import com.test.payment.dto.OAuth2TokenResponse;
import com.test.payment.models.PaymentProviderType;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
@Service
@RequiredArgsConstructor
@Slf4j
public class MtnTokenService {
private final WebClient mtnWebClient;
private final TokenCacheService tokenCache;
private final Environment environment;
public Mono<String> getToken() {
return tokenCache.getToken(PaymentProviderType.MTN.name(), this::fetchToken);
}
public Mono<Void> evictToken() {
return tokenCache.evictToken(PaymentProviderType.MTN.name());
}
private Mono<TokenCacheService.FetchedToken> fetchToken() {
return mtnWebClient.post()
.uri("/collection/token/")
.headers(h -> {
h.setBasicAuth(
environment.getProperty("mtn.api-user", ""),
environment.getProperty("mtn.api-key", ""));
h.set("Ocp-Apim-Subscription-Key", environment.getProperty("mtn.subscription-key", ""));
})
.retrieve()
.bodyToMono(OAuth2TokenResponse.class)
.map(resp -> new TokenCacheService.FetchedToken(resp.getAccessToken(), resp.expiresInSeconds()));
}
}

327
src/main/java/com/test/payment/service/PaymentLifecycleService.java Normal file
View File

@@ -0,0 +1,327 @@
package com.test.payment.service;
import com.test.payment.dto.CallbackAckDto;
import com.test.payment.dto.PaymentRequest;
import com.test.payment.dto.PaymentResultDto;
import com.test.payment.dto.TransactionStatusDto;
import com.test.payment.models.PaymentCallback;
import com.test.payment.models.PaymentInitiation;
import com.test.payment.models.PaymentResponse;
import com.test.payment.models.Transaction;
import com.test.payment.models.TransactionStatus;
import com.test.payment.repository.PaymentCallbackRepository;
import com.test.payment.repository.PaymentInitiationRepository;
import com.test.payment.repository.PaymentResponseRepository;
import com.test.payment.repository.TransactionRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
import org.springframework.web.server.ResponseStatusException;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.Optional;
import java.util.function.Function;
/**
* Provider-agnostic persistence and lifecycle: initiation → response → callback /
* status query → consolidated transaction. Provider services delegate here and only
* contribute the provider-specific HTTP calls and payload parsing.
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class PaymentLifecycleService {
private final PaymentInitiationRepository initiationRepository;
private final PaymentResponseRepository responseRepository;
private final PaymentCallbackRepository callbackRepository;
private final TransactionRepository transactionRepository;
/** The provider's answer to an initiation. */
public record ProviderResponseData(String providerReference, String secondaryReference, String responseCode,
String responseDescription, String customerMessage, boolean accepted) {
}
/** A parsed provider callback. */
public record CallbackData(String providerReference, String resultCode, String resultDesc, String receiptNumber,
BigDecimal amount, String phoneNumber, String transactionDate, boolean success) {
}
/** The outcome of a live status query. */
public record QueryOutcome(TransactionStatus newStatus, String resultCode, String resultDesc, String receiptNumber) {
public static QueryOutcome pending(String note) {
return new QueryOutcome(TransactionStatus.PENDING, null, note, null);
}
}
public Mono<PaymentInitiation> saveInitiation(String provider, PaymentRequest request) {
return initiationRepository.save(PaymentInitiation.builder()
.provider(provider)
.phoneNumber(request.getPhoneNumber())
.amount(BigDecimal.valueOf(request.getAmount()))
.accountReference(request.getAccountReference())
.transactionDesc(request.getTransactionDesc())
.status(TransactionStatus.PENDING.name())
.createdAt(LocalDateTime.now())
.build());
}
public Mono<PaymentResultDto> persistResponse(PaymentInitiation initiation, ProviderResponseData data) {
String newStatus = data.accepted() ? TransactionStatus.PENDING.name() : TransactionStatus.FAILED.name();
PaymentResponse entity = PaymentResponse.builder()
.initiationId(initiation.getId())
.provider(initiation.getProvider())
.providerReference(data.providerReference())
.secondaryReference(data.secondaryReference())
.responseCode(data.responseCode())
.responseDescription(data.responseDescription())
.customerMessage(data.customerMessage())
.createdAt(LocalDateTime.now())
.build();
// one response per initiation — an existing row wins, a concurrent insert falls back to it
return responseRepository.findByInitiationId(initiation.getId())
.switchIfEmpty(Mono.defer(() -> responseRepository.save(entity)
.onErrorResume(DuplicateKeyException.class,
e -> responseRepository.findByInitiationId(initiation.getId()))))
.flatMap(saved -> updateStatus(initiation, newStatus)
.flatMap(updated -> data.accepted()
? Mono.just(updated)
: recordTransaction(updated, saved, null,
data.responseDescription(), null, null, "REJECTION")
.thenReturn(updated))
.map(updated -> PaymentResultDto.builder()
.initiationId(updated.getId())
.provider(updated.getProvider())
.status(updated.getStatus())
.providerReference(saved.getProviderReference())
.secondaryReference(saved.getSecondaryReference())
.responseCode(saved.getResponseCode())
.responseDescription(saved.getResponseDescription())
.customerMessage(saved.getCustomerMessage())
.build()));
}
public Mono<PaymentResultDto> markFailed(PaymentInitiation initiation, Throwable ex) {
log.error("[{}] payment failed for initiation {}: {}", initiation.getProvider(), initiation.getId(), ex.toString());
String reason = ex.getMessage() == null ? ex.getClass().getSimpleName() : ex.getMessage();
return updateStatus(initiation, TransactionStatus.FAILED.name())
.flatMap(updated -> recordTransaction(updated, null, null, truncate(reason), null, null, "ERROR"))
.then(Mono.error(ex));
}
public Mono<CallbackAckDto> applyCallback(String provider, CallbackData data, String rawPayload) {
if (data.providerReference() == null) {
log.warn("[{}] callback without a provider reference ignored", provider);
return Mono.just(CallbackAckDto.accepted("Ignored: no reference"));
}
return responseRepository.findByProviderReference(data.providerReference())
.filter(response -> provider.equals(response.getProvider()))
.flatMap(response -> callbackRepository.findByInitiationId(response.getInitiationId())
.map(existing -> {
log.info("[{}] duplicate callback for {} ignored", provider, data.providerReference());
return CallbackAckDto.accepted("Duplicate callback ignored");
})
.switchIfEmpty(Mono.defer(() -> saveCallback(response, data, rawPayload))))
.switchIfEmpty(Mono.fromSupplier(() -> {
log.warn("[{}] callback for unknown reference {}", provider, data.providerReference());
return CallbackAckDto.accepted("Unknown reference");
}));
}
/**
* Returns the transaction state from the database; if it is still PENDING, runs the
* provider's live status query (querier) and applies the outcome.
*/
public Mono<TransactionStatusDto> checkStatus(String provider, String providerReference,
Function<PaymentResponse, Mono<QueryOutcome>> querier) {
return responseRepository.findByProviderReference(providerReference)
.filter(response -> provider.equals(response.getProvider()))
.switchIfEmpty(Mono.error(new ResponseStatusException(HttpStatus.NOT_FOUND,
"No " + provider + " transaction found for reference " + providerReference)))
.flatMap(response -> initiationRepository.findById(response.getInitiationId())
.flatMap(initiation -> TransactionStatus.PENDING.name().equals(initiation.getStatus())
? querier.apply(response).flatMap(outcome -> applyQueryOutcome(initiation, response, outcome))
: buildStatusDto(initiation, response)));
}
/**
* Used by the reconciliation job for PENDING initiations older than the configured age.
*/
public Mono<Void> reconcile(PaymentInitiation initiation,
Function<String, Mono<TransactionStatusDto>> statusChecker) {
return responseRepository.findByInitiationId(initiation.getId())
.flatMap(response -> {
if (response.getProviderReference() == null) {
return failTerminal(initiation, "No provider reference on response").thenReturn(true);
}
return statusChecker.apply(response.getProviderReference()).thenReturn(true);
})
.switchIfEmpty(Mono.defer(() -> {
log.warn("[{}] initiation {} never received a provider response — marking FAILED",
initiation.getProvider(), initiation.getId());
return failTerminal(initiation, "No provider response received").thenReturn(false);
}))
.then();
}
public Flux<Transaction> listTransactions(String provider) {
return provider == null ? transactionRepository.findAll() : transactionRepository.findByProvider(provider);
}
private Mono<Transaction> failTerminal(PaymentInitiation initiation, String reason) {
return updateStatus(initiation, TransactionStatus.FAILED.name())
.flatMap(updated -> recordTransaction(updated, null, null, reason, null, null, "RECONCILIATION"));
}
private Mono<CallbackAckDto> saveCallback(PaymentResponse response, CallbackData data, String rawPayload) {
PaymentCallback entity = PaymentCallback.builder()
.initiationId(response.getInitiationId())
.provider(response.getProvider())
.providerReference(data.providerReference())
.resultCode(data.resultCode())
.resultDesc(data.resultDesc())
.receiptNumber(data.receiptNumber())
.amount(data.amount())
.phoneNumber(data.phoneNumber())
.transactionDate(data.transactionDate())
.rawPayload(rawPayload)
.createdAt(LocalDateTime.now())
.build();
String newStatus = data.success() ? TransactionStatus.SUCCESS.name() : TransactionStatus.FAILED.name();
return callbackRepository.save(entity)
.onErrorResume(DuplicateKeyException.class,
e -> callbackRepository.findByInitiationId(response.getInitiationId()))
.flatMap(saved -> initiationRepository.findById(response.getInitiationId())
.flatMap(initiation -> updateStatus(initiation, newStatus))
.flatMap(updated -> recordTransaction(updated, response, data.resultCode(),
data.resultDesc(), data.receiptNumber(), data.transactionDate(), "CALLBACK")))
.doOnNext(tx -> log.info("[{}] callback processed for initiation {} — status {}",
tx.getProvider(), tx.getInitiationId(), tx.getStatus()))
.thenReturn(CallbackAckDto.accepted("Callback processed"));
}
private Mono<TransactionStatusDto> applyQueryOutcome(PaymentInitiation initiation, PaymentResponse response,
QueryOutcome outcome) {
if (outcome.newStatus() == TransactionStatus.PENDING) {
return buildStatusDto(initiation, response)
.map(dto -> {
if (outcome.resultDesc() != null) {
dto.setResultDesc(outcome.resultDesc());
}
return dto;
});
}
return updateStatus(initiation, outcome.newStatus().name())
.flatMap(updated -> recordTransaction(updated, response, outcome.resultCode(),
outcome.resultDesc(), outcome.receiptNumber(), null, "QUERY")
.then(buildStatusDto(updated, response)))
.map(dto -> {
dto.setResultCode(outcome.resultCode());
dto.setResultDesc(outcome.resultDesc());
if (outcome.receiptNumber() != null) {
dto.setReceiptNumber(outcome.receiptNumber());
}
return dto;
});
}
private Mono<TransactionStatusDto> buildStatusDto(PaymentInitiation initiation, PaymentResponse response) {
// result details come from the callback when we have one, otherwise from the
// consolidated transaction row (e.g. when a status query resolved the payment)
return callbackRepository.findByInitiationId(initiation.getId())
.map(Optional::of)
.defaultIfEmpty(Optional.empty())
.zipWith(transactionRepository.findByInitiationId(initiation.getId())
.map(Optional::of)
.defaultIfEmpty(Optional.empty()))
.map(tuple -> {
Optional<PaymentCallback> cb = tuple.getT1();
Optional<Transaction> tx = tuple.getT2();
return TransactionStatusDto.builder()
.initiationId(initiation.getId())
.provider(initiation.getProvider())
.providerReference(response.getProviderReference())
.secondaryReference(response.getSecondaryReference())
.status(initiation.getStatus())
.phoneNumber(initiation.getPhoneNumber())
.amount(initiation.getAmount())
.accountReference(initiation.getAccountReference())
.resultCode(cb.map(PaymentCallback::getResultCode)
.or(() -> tx.map(Transaction::getResultCode)).orElse(null))
.resultDesc(cb.map(PaymentCallback::getResultDesc)
.or(() -> tx.map(Transaction::getResultDesc)).orElse(null))
.receiptNumber(cb.map(PaymentCallback::getReceiptNumber)
.or(() -> tx.map(Transaction::getReceiptNumber)).orElse(null))
.createdAt(initiation.getCreatedAt())
.updatedAt(initiation.getUpdatedAt())
.build();
});
}
private Mono<PaymentInitiation> updateStatus(PaymentInitiation initiation, String status) {
initiation.setStatus(status);
initiation.setUpdatedAt(LocalDateTime.now());
return initiationRepository.save(initiation);
}
/**
* Upserts the consolidated TRANSACTIONS row for an initiation that reached a
* terminal state. Keyed by initiation_id (UNIQUE) so it can never duplicate;
* a later, richer resolution (e.g. a callback after a query) updates the row.
*/
private Mono<Transaction> recordTransaction(PaymentInitiation initiation, PaymentResponse response,
String resultCode, String resultDesc, String receiptNumber,
String transactionDate, String resolvedBy) {
return transactionRepository.findByInitiationId(initiation.getId())
.flatMap(existing -> {
existing.setStatus(initiation.getStatus());
if (resultCode != null) {
existing.setResultCode(resultCode);
}
if (resultDesc != null) {
existing.setResultDesc(resultDesc);
}
if (receiptNumber != null) {
existing.setReceiptNumber(receiptNumber);
}
if (transactionDate != null) {
existing.setTransactionDate(transactionDate);
}
existing.setResolvedBy(resolvedBy);
existing.setUpdatedAt(LocalDateTime.now());
return transactionRepository.save(existing);
})
.switchIfEmpty(Mono.defer(() -> transactionRepository.save(Transaction.builder()
.initiationId(initiation.getId())
.provider(initiation.getProvider())
.providerReference(response != null ? response.getProviderReference() : null)
.secondaryReference(response != null ? response.getSecondaryReference() : null)
.phoneNumber(initiation.getPhoneNumber())
.amount(initiation.getAmount())
.accountReference(initiation.getAccountReference())
.status(initiation.getStatus())
.resultCode(resultCode)
.resultDesc(resultDesc)
.receiptNumber(receiptNumber)
.transactionDate(transactionDate)
.resolvedBy(resolvedBy)
.createdAt(LocalDateTime.now())
.build())
.onErrorResume(DuplicateKeyException.class,
e -> transactionRepository.findByInitiationId(initiation.getId()))))
.doOnNext(tx -> log.info("[{}] transaction {} recorded for initiation {} — status {} (via {})",
initiation.getProvider(), tx.getId(), initiation.getId(), tx.getStatus(), resolvedBy));
}
private String truncate(String value) {
return value == null || value.length() <= 255 ? value : value.substring(0, 255);
}
}

20
src/main/java/com/test/payment/service/PaymentProviderService.java Normal file
View File

@@ -0,0 +1,20 @@
package com.test.payment.service;
import com.test.payment.dto.PaymentRequest;
import com.test.payment.dto.PaymentResultDto;
import com.test.payment.dto.TransactionStatusDto;
import reactor.core.publisher.Mono;
/**
* One implementation per provider (MpesaService, AirtelService, MtnService).
* The reconciliation job dispatches PENDING initiations to the right provider
* through this interface.
*/
public interface PaymentProviderService {
String provider();
Mono<PaymentResultDto> initiatePayment(PaymentRequest request);
Mono<TransactionStatusDto> checkStatus(String providerReference);
}

111
src/main/java/com/test/payment/service/TokenCacheService.java Normal file
View File

@@ -0,0 +1,111 @@
package com.test.payment.service;
import com.test.payment.models.ProviderToken;
import com.test.payment.repository.ProviderTokenRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.env.Environment;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.function.Supplier;
/**
* Tiered OAuth token cache shared by all providers: Redis first, then the database,
* then a live fetch (persisted to both tiers on success). Redis being down never
* fails a request — reads/writes time out and degrade to the next tier.
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class TokenCacheService {
private static final Duration REDIS_TIMEOUT = Duration.ofMillis(800);
private final ProviderTokenRepository tokenRepository;
private final ReactiveStringRedisTemplate redisTemplate;
private final Environment environment;
public record FetchedToken(String accessToken, long expiresInSeconds) {
}
public Mono<String> getToken(String provider, Supplier<Mono<FetchedToken>> fetcher) {
return fromRedis(provider)
.switchIfEmpty(Mono.defer(() -> fromDatabase(provider)))
.switchIfEmpty(Mono.defer(() -> fetchAndStore(provider, fetcher)));
}
/**
* Drops the cached token from Redis and the database (used on 401 from the provider).
*/
public Mono<Void> evictToken(String provider) {
return redisTemplate.opsForValue().delete(redisKey(provider))
.timeout(REDIS_TIMEOUT)
.onErrorResume(e -> {
log.warn("[{}] Redis unavailable while evicting token: {}", provider, e.toString());
return Mono.just(false);
})
.then(tokenRepository.deleteByProvider(provider));
}
private Mono<String> fromRedis(String provider) {
return redisTemplate.opsForValue().get(redisKey(provider))
.timeout(REDIS_TIMEOUT)
.doOnNext(t -> log.debug("[{}] token served from Redis", provider))
.onErrorResume(e -> {
log.warn("[{}] Redis unavailable for token read ({}) — falling back to database", provider, e.toString());
return Mono.empty();
});
}
private Mono<String> fromDatabase(String provider) {
int buffer = expiryBufferSeconds();
return tokenRepository.findFirstByProviderAndExpiresAtAfterOrderByIdDesc(provider, LocalDateTime.now().plusSeconds(buffer))
.flatMap(token -> {
long ttl = Duration.between(LocalDateTime.now(), token.getExpiresAt()).getSeconds() - buffer;
log.debug("[{}] token served from database", provider);
return cacheInRedis(provider, token.getAccessToken(), ttl).thenReturn(token.getAccessToken());
});
}
private Mono<String> fetchAndStore(String provider, Supplier<Mono<FetchedToken>> fetcher) {
return fetcher.get()
.flatMap(fetched -> {
ProviderToken token = ProviderToken.builder()
.provider(provider)
.accessToken(fetched.accessToken())
.expiresAt(LocalDateTime.now().plusSeconds(fetched.expiresInSeconds()))
.createdAt(LocalDateTime.now())
.build();
log.info("[{}] fetched new access token (expires in {}s)", provider, fetched.expiresInSeconds());
return tokenRepository.save(token)
.then(cacheInRedis(provider, fetched.accessToken(), fetched.expiresInSeconds() - expiryBufferSeconds()))
.thenReturn(fetched.accessToken());
});
}
private Mono<Void> cacheInRedis(String provider, String token, long ttlSeconds) {
if (ttlSeconds <= 0) {
return Mono.empty();
}
return redisTemplate.opsForValue().set(redisKey(provider), token, Duration.ofSeconds(ttlSeconds))
.timeout(REDIS_TIMEOUT)
.doOnNext(ok -> log.debug("[{}] token cached in Redis for {}s", provider, ttlSeconds))
.onErrorResume(e -> {
log.warn("[{}] Redis unavailable for token write ({}) — token persisted in database only", provider, e.toString());
return Mono.just(false);
})
.then();
}
private String redisKey(String provider) {
return provider.toLowerCase() + ":access_token";
}
private int expiryBufferSeconds() {
return environment.getProperty("payments.token-expiry-buffer-seconds", Integer.class, 60);
}
}

34
src/main/java/com/test/payment/utils/MpesaUtils.java
View File

@@ -1,24 +1,25 @@
package com.test.payment.utils;
import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Base64;
import java.util.Date;
import java.util.TimeZone;
public class MpesaUtils {
private static final ZoneId NAIROBI = ZoneId.of("Africa/Nairobi");
private static final DateTimeFormatter TIMESTAMP_FORMAT = DateTimeFormatter.ofPattern("yyyyMMddHHmmss");
private MpesaUtils() {
}
/**
* Generates a timestamp in the format yyyyMMddHHmmss
* Generates a timestamp in the format yyyyMMddHHmmss (Africa/Nairobi time).
* 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());
return ZonedDateTime.now(NAIROBI).format(TIMESTAMP_FORMAT);
}
/**
@@ -31,7 +32,8 @@ public class MpesaUtils {
}
/**
* Helper method to generate both password and timestamp together.
* Generates both password and timestamp together — always call this per attempt
* so the password matches a fresh timestamp.
*/
public static MpesaAuthData generateAuthData(String businessShortCode, String passkey) {
String timestamp = generateTimestamp();
@@ -39,15 +41,7 @@ public class MpesaUtils {
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 record MpesaAuthData(String password, String timestamp) {
public String getPassword() {
return password;