diff --git a/.gitignore b/.gitignore index 524f096..ba3cb82 100644 --- a/.gitignore +++ b/.gitignore @@ -1,16 +1,47 @@ -# Compiled class file -*.class +# --- Gradle --- +build/ +.gradle/ -# Log file +# --- IDE: IntelliJ --- +.idea/ +*.iml +*.ipr +*.iws +out/ + +# --- IDE: Eclipse / STS --- +.apt_generated +.classpath +.factorypath +.project +.settings/ +.springBeans +.sts4-cache +bin/ + +# --- IDE: VS Code --- +.vscode/ + +# --- IDE: NetBeans --- +/nbproject/private/ +nbbuild/ +dist/ +nbdist/ +.nb-gradle/ + +# --- OS --- +.DS_Store +Thumbs.db + +# --- Logs & temp --- *.log +logs/ +nohup.out +hs_err_pid* +replay_pid* -# BlueJ files -*.ctxt - -# Mobile Tools for Java (J2ME) -.mtj.tmp/ - -# Package Files # +# --- Compiled / packaged --- +*.class *.jar *.war *.nar @@ -18,7 +49,15 @@ *.zip *.tar.gz *.rar +*.ctxt +.mtj.tmp/ -# virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml -hs_err_pid* -replay_pid* +# Keep the Gradle wrapper despite the *.jar rule above (last match wins) +!gradle/wrapper/gradle-wrapper.jar +!gradle/wrapper/gradle-wrapper.properties + +# --- Local env / secrets (keep real credentials out of the repo) --- +.env +.env.* +application-local.yml +application-local.yaml diff --git a/.gradle/8.14.3/checksums/checksums.lock b/.gradle/8.14.3/checksums/checksums.lock deleted file mode 100644 index 9fd90df..0000000 Binary files a/.gradle/8.14.3/checksums/checksums.lock and /dev/null differ diff --git a/.gradle/8.14.3/checksums/md5-checksums.bin b/.gradle/8.14.3/checksums/md5-checksums.bin deleted file mode 100644 index 85c3bc5..0000000 Binary files a/.gradle/8.14.3/checksums/md5-checksums.bin and /dev/null differ diff --git a/.gradle/8.14.3/checksums/sha1-checksums.bin b/.gradle/8.14.3/checksums/sha1-checksums.bin deleted file mode 100644 index 028991a..0000000 Binary files a/.gradle/8.14.3/checksums/sha1-checksums.bin and /dev/null differ diff --git a/.gradle/8.14.3/executionHistory/executionHistory.bin b/.gradle/8.14.3/executionHistory/executionHistory.bin deleted file mode 100644 index 54e6c4a..0000000 Binary files a/.gradle/8.14.3/executionHistory/executionHistory.bin and /dev/null differ diff --git a/.gradle/8.14.3/executionHistory/executionHistory.lock b/.gradle/8.14.3/executionHistory/executionHistory.lock deleted file mode 100644 index 927f37c..0000000 Binary files a/.gradle/8.14.3/executionHistory/executionHistory.lock and /dev/null differ diff --git a/.gradle/8.14.3/fileChanges/last-build.bin b/.gradle/8.14.3/fileChanges/last-build.bin deleted file mode 100644 index f76dd23..0000000 Binary files a/.gradle/8.14.3/fileChanges/last-build.bin and /dev/null differ diff --git a/.gradle/8.14.3/fileHashes/fileHashes.bin b/.gradle/8.14.3/fileHashes/fileHashes.bin deleted file mode 100644 index d312e43..0000000 Binary files a/.gradle/8.14.3/fileHashes/fileHashes.bin and /dev/null differ diff --git a/.gradle/8.14.3/fileHashes/fileHashes.lock b/.gradle/8.14.3/fileHashes/fileHashes.lock deleted file mode 100644 index a6230ae..0000000 Binary files a/.gradle/8.14.3/fileHashes/fileHashes.lock and /dev/null differ diff --git a/.gradle/8.14.3/fileHashes/resourceHashesCache.bin b/.gradle/8.14.3/fileHashes/resourceHashesCache.bin deleted file mode 100644 index 31128db..0000000 Binary files a/.gradle/8.14.3/fileHashes/resourceHashesCache.bin and /dev/null differ diff --git a/.gradle/8.14.3/gc.properties b/.gradle/8.14.3/gc.properties deleted file mode 100644 index e69de29..0000000 diff --git a/.gradle/buildOutputCleanup/buildOutputCleanup.lock b/.gradle/buildOutputCleanup/buildOutputCleanup.lock deleted file mode 100644 index 8a3c9b3..0000000 Binary files a/.gradle/buildOutputCleanup/buildOutputCleanup.lock and /dev/null differ diff --git a/.gradle/buildOutputCleanup/cache.properties b/.gradle/buildOutputCleanup/cache.properties deleted file mode 100644 index c17daad..0000000 --- a/.gradle/buildOutputCleanup/cache.properties +++ /dev/null @@ -1,2 +0,0 @@ -#Fri Oct 10 12:56:24 EAT 2025 -gradle.version=8.14.3 diff --git a/.gradle/buildOutputCleanup/outputFiles.bin b/.gradle/buildOutputCleanup/outputFiles.bin deleted file mode 100644 index bb3069c..0000000 Binary files a/.gradle/buildOutputCleanup/outputFiles.bin and /dev/null differ diff --git a/.gradle/file-system.probe b/.gradle/file-system.probe deleted file mode 100644 index 0d90605..0000000 Binary files a/.gradle/file-system.probe and /dev/null differ diff --git a/.gradle/vcs-1/gc.properties b/.gradle/vcs-1/gc.properties deleted file mode 100644 index e69de29..0000000 diff --git a/.idea/.gitignore b/.idea/.gitignore deleted file mode 100644 index a0ccf77..0000000 --- a/.idea/.gitignore +++ /dev/null @@ -1,5 +0,0 @@ -# Default ignored files -/shelf/ -/workspace.xml -# Environment-dependent path to Maven home directory -/mavenHomeManager.xml diff --git a/.idea/.name b/.idea/.name deleted file mode 100644 index 4d4d2ce..0000000 --- a/.idea/.name +++ /dev/null @@ -1 +0,0 @@ -payment \ No newline at end of file diff --git a/.idea/compiler.xml b/.idea/compiler.xml deleted file mode 100644 index 625c380..0000000 --- a/.idea/compiler.xml +++ /dev/null @@ -1,18 +0,0 @@ - - - - - - - - - - - - - - - - - \ No newline at end of file diff --git a/.idea/gradle.xml b/.idea/gradle.xml deleted file mode 100644 index f9163b4..0000000 --- a/.idea/gradle.xml +++ /dev/null @@ -1,15 +0,0 @@ - - - - - - \ No newline at end of file diff --git a/.idea/misc.xml b/.idea/misc.xml deleted file mode 100644 index 4ba5914..0000000 --- a/.idea/misc.xml +++ /dev/null @@ -1,5 +0,0 @@ - - - - - \ No newline at end of file diff --git a/.idea/modules.xml b/.idea/modules.xml deleted file mode 100644 index d28af19..0000000 --- a/.idea/modules.xml +++ /dev/null @@ -1,8 +0,0 @@ - - - - - - - - \ No newline at end of file diff --git a/.idea/modules/payment.main.iml b/.idea/modules/payment.main.iml deleted file mode 100644 index afc1871..0000000 --- a/.idea/modules/payment.main.iml +++ /dev/null @@ -1,8 +0,0 @@ - - - - - - - - \ No newline at end of file diff --git a/.idea/vcs.xml b/.idea/vcs.xml deleted file mode 100644 index 35eb1dd..0000000 --- a/.idea/vcs.xml +++ /dev/null @@ -1,6 +0,0 @@ - - - - - - \ No newline at end of file diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..ee45d9c --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,75 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Build & Run Commands + +```bash +# Build +./gradlew build + +# Run the application +./gradlew bootRun + +# Run tests +./gradlew test + +# Run a single test class +./gradlew test --tests "com.test.payment.PaymentApplicationTests" + +# Clean build +./gradlew clean build +``` + +Requires JDK 21 to run Gradle (Gradle 8.14 cannot run on JDK 25; set `JAVA_HOME` accordingly). +The app starts on the default port 8080. Swagger UI is available at `http://localhost:8080/swagger-ui.html`. +Redis is optional at runtime (`docker run -d --name mpesa-redis -p 6379:6379 redis:7-alpine`) — the app degrades gracefully without it. + +## Architecture Overview + +Spring Boot 4 / Java 21 reactive (WebFlux) multi-provider mobile-money service (M-Pesa STK Push, Airtel Money, MTN MoMo), using R2DBC for async DB access, Redis (Lettuce) for token caching, and Resilience4j for fault tolerance. + +**Layering (same shape for every provider):** +`Controller` → `Service` (implements `PaymentProviderService`; parses provider payloads) → `PaymentLifecycleService` (ALL persistence: initiation/response/callback/transaction, dedup, status transitions) and `Client` (resilience-wrapped HTTP) → provider API. + +**Endpoints** (identical pattern per provider under `/api/mpesa`, `/api/airtel`, `/api/mtn`): +- `POST /pay` — validated request; persists an initiation, calls the provider, persists the linked response, returns the outcome with a `providerReference`. +- `POST /callback` — provider result callback (MTN also accepts PUT); stored 1:1 with the initiation, deduplicated, updates status. Always acks. +- `GET /status/{providerReference}` — DB state; if still PENDING, performs a live provider status query and updates the DB. Degrades to last-known state when the provider rate-limits. +- `GET /api/payments/transactions?provider=` — consolidated transactions across providers. + +**Provider specifics:** +- M-Pesa: `providerReference` = CheckoutRequestID, `secondaryReference` = MerchantRequestID; STK query "still processing" (errorCode 500.001.1001) maps to `ProviderProcessingException` → stays PENDING. Sandbox creds in yml are live. +- Airtel: reference is a generated `ATL` transaction id; status codes TIP (pending) / TS (success) / TF (failed); receipt = `airtel_money_id`. Credentials are placeholders (`airtel.*`). +- MTN MoMo: reference is the generated `X-Reference-Id` UUID (externalId is set to the same value for callback correlation); request-to-pay returns 202 with no body; status SUCCESSFUL/FAILED/PENDING; receipt = `financialTransactionId`. Sandbox needs `mtn.subscription-key`, `mtn.api-user`, `mtn.api-key`; sandbox currency is EUR. Placeholders in yml. + +**Key design points:** + +- All I/O is non-blocking (`Mono`/`Flux` throughout). Never block a reactive pipeline with `.block()` (startup schema init is the one deliberate exception). +- All outbound provider calls live in `client/*Client` classes so the Resilience4j annotations (`@CircuitBreaker`, `@RateLimiter`, `@Retry`) go through Spring AOP — they are silently skipped on self-invocation. Each provider has its own instances (`mpesa*`, `airtel*`, `mtn*`) configured **only** in `application.yml` (YAML anchors share the retry/circuit-breaker settings). +- Shared exception model in `exceptions/`: `ProviderBusyException` (429/busy → reactive backoff retry, 503), `ProviderTransientException` (Resilience4j retry, 502), `ProviderPermanentException` (never retried, CB-ignored, 400), `ProviderProcessingException` (still processing → stays PENDING). HTTP-to-exception mapping is shared in `client/ProviderHttpErrors` (401 evicts the token so the retry refetches). +- `TokenCacheService` is the tiered token cache for all providers: Redis (`:access_token`) → `PROVIDER_TOKENS` table → live OAuth fetch (persisted to both tiers). Redis being down never fails a request (800ms timeouts, falls through). `TokenService` classes only supply the fetch call. +- Boot 4 defaults to Jackson 3 (`tools.jackson.databind.ObjectMapper`) — inject that type, not `com.fasterxml`. The `com.fasterxml.jackson.annotation.*` annotations still work. +- Resilience4j 2.4.0 with `resilience4j-spring-boot4`; annotations require `aspectjweaver` (Boot 4 removed `spring-boot-starter-aop`). Boot 4 also split `WebClient` auto-config into `spring-boot-starter-webclient`. +- Package convention: `models` holds only database entities (`@Table`) plus `DatabaseSchema`; everything crossing an API boundary lives in `dto`. + +**Persistence (in-memory H2 via R2DBC):** +- The schema lives in code: `models/DatabaseSchema.STATEMENTS`, executed at startup by `configurations/DatabaseSchemaInitializer` (there is no schema.sql; `spring.sql.init` is not used). +- `PAYMENT_INITIATIONS` — one row per payment attempt, any provider (status: PENDING → SUCCESS/FAILED). +- `PAYMENT_RESPONSES` — the provider's answer, `initiation_id UNIQUE` (1:1) and `provider_reference UNIQUE` (lookup key for callbacks/status). +- `PAYMENT_CALLBACKS` — the provider result callback, `initiation_id UNIQUE`, duplicates ignored, raw payload stored as JSON. +- `TRANSACTIONS` — consolidated record upserted by `PaymentLifecycleService.recordTransaction` whenever an initiation reaches a terminal state, from whichever path resolved it (`resolvedBy`: CALLBACK, QUERY, REJECTION, ERROR, RECONCILIATION). `initiation_id UNIQUE`. +- `PROVIDER_TOKENS` — OAuth tokens per provider with expiry. +- `@Table` names must be UPPERCASE — H2 stores unquoted DDL identifiers uppercase and Spring Data quotes entity names verbatim. +- `PaymentReconciliationJob` reconciles PENDING initiations of **all** providers older than `payments.reconciliation.pending-age` (default 5m) by dispatching to the right `PaymentProviderService`; interval `payments.reconciliation.fixed-delay` (default 60s). + +**Configuration (`application.yml`):** +- Config is read via `Environment.getProperty` by project convention (no `@ConfigurationProperties`). +- `mpesa.*` (live sandbox keys), `airtel.*` and `mtn.*` (placeholders — fill in real credentials), `payments.*` (token buffer, reconciliation). +- Callback URLs (`mpesa.callback-url`, `mtn.callback-url`) must be publicly reachable for real callbacks. +- `spring.data.redis.*` — short 1s timeouts so a dead Redis degrades fast. + +**Sandbox gotchas (observed live):** +- Safaricom's STK query endpoint is aggressively rate-limited (~few calls/minute) — expect `ProviderBusyException` from status checks/reconciliation during testing; the status endpoint then serves last-known DB state. +- M-Pesa test phone 254708374149 typically ends as ResultCode 1037 ("DS timeout user cannot be reached") since no real handset confirms the push. +- Airtel/MTN flows were verified against a local mock of their APIs (`scratchpad/provider_mock.py` pattern) — point `airtel.base-url`/`mtn.base-url` at a stub to test without credentials. diff --git a/build.gradle b/build.gradle index 3dc7c69..b4d941f 100644 --- a/build.gradle +++ b/build.gradle @@ -1,6 +1,6 @@ plugins { id 'java' - id 'org.springframework.boot' version '3.5.6' + id 'org.springframework.boot' version '4.1.0' id 'io.spring.dependency-management' version '1.1.7' } @@ -10,7 +10,7 @@ description = 'Demo project for Spring Boot' java { toolchain { - languageVersion = JavaLanguageVersion.of(17) + languageVersion = JavaLanguageVersion.of(21) } } @@ -21,29 +21,27 @@ repositories { dependencies { implementation("org.springframework.boot:spring-boot-starter-webflux") + implementation("org.springframework.boot:spring-boot-starter-webclient") implementation("org.springframework.boot:spring-boot-starter-data-r2dbc")/* implementation("io.r2dbc:r2dbc-postgresql")*/ - /*implementation("org.springframework.boot:spring-boot-starter-data-redis-reactive") - */implementation("io.lettuce:lettuce-core:6.2.2.RELEASE") + implementation("org.springframework.boot:spring-boot-starter-data-redis-reactive") + implementation("org.springframework.boot:spring-boot-starter-validation") // Choose ONE driver depending on your DB: runtimeOnly("io.r2dbc:r2dbc-h2") // for in-memory - implementation("io.github.resilience4j:resilience4j-ratelimiter:2.0.2") - implementation("io.github.resilience4j:resilience4j-circuitbreaker:2.0.2") - implementation("io.github.resilience4j:resilience4j-retry:2.0.2") - implementation("org.springframework.boot:spring-boot-starter") - implementation("org.springdoc:springdoc-openapi-starter-webflux-ui:2.6.0") + implementation("org.aspectj:aspectjweaver") + implementation("org.springdoc:springdoc-openapi-starter-webflux-ui:3.0.3") - implementation 'io.github.resilience4j:resilience4j-spring-boot2:1.7.1' - implementation 'io.github.resilience4j:resilience4j-reactor:1.7.1' + implementation 'io.github.resilience4j:resilience4j-spring-boot4:2.4.0' + implementation 'io.github.resilience4j:resilience4j-reactor:2.4.0' - implementation 'org.projectlombok:lombok:1.18.32' // Use the latest stable version - annotationProcessor 'org.projectlombok:lombok:1.18.32' // For annotation processing + implementation 'org.projectlombok:lombok:1.18.46' // Use the latest stable version + annotationProcessor 'org.projectlombok:lombok:1.18.46' // For annotation processing testImplementation("org.springframework.boot:spring-boot-starter-test") } diff --git a/build/reports/problems/problems-report.html b/build/reports/problems/problems-report.html deleted file mode 100644 index 200197d..0000000 --- a/build/reports/problems/problems-report.html +++ /dev/null @@ -1,663 +0,0 @@ - - - - - - - - - - - - - Gradle Configuration Cache - - - -
- -
- Loading... -
- - - - - - diff --git a/build/resources/main/application.yml b/build/resources/main/application.yml deleted file mode 100644 index 8d45436..0000000 --- a/build/resources/main/application.yml +++ /dev/null @@ -1,70 +0,0 @@ -#spring: - # redis: - # host: localhost - # port: 6379 - -resilience4j: - ratelimiter: - instances: - mpesaLimiter: - limit-for-period: 10 - limit-refresh-period: 1s - timeout-duration: 0s - - retry: - instances: - mpesaRetry: - max-attempts: 3 - wait-duration: 2s - retry-exceptions: - - org.springframework.web.reactive.function.client.WebClientRequestException - - java.io.IOException - - com.example.mpesa.exceptions.MpesaBusyException - ignore-exceptions: - - com.test.payment.exceptions.MpesaPermanentException - - java.lang.IllegalArgumentException - - - circuitbreaker: - instances: - mpesaCircuitBreaker: - sliding-window-size: 10 - permitted-number-of-calls-in-half-open-state: 3 - failure-rate-threshold: 50 - wait-duration-in-open-state: 10s - - - - - -spring: - r2dbc: - url: r2dbc:h2:mem:///mpesa_db;DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE - username: sa - password: - sql: - init: - mode: always - schema-locations: classpath:schema.sql - main: - web-application-type: reactive - -logging: - level: - org.springframework.data.r2dbc: DEBUG - -springdoc: - swagger-ui: - path: /swagger-ui.html - operationsSorter: method - tagsSorter: alpha - api-docs: - path: /v3/api-docs - packages-to-scan: com.test.payment.controller - -mpesa: - base-url: https://sandbox.safaricom.co.ke - consumer-key: k6e7LtBNeVX7V8MPqB7P83FsZio8cRZD - consumer-secret: cGwiWzhDGopC3dho - business-short-code: 174379 - pass-key: bfb279f9aa9bdbcf158e97dd71a467cd2e0c893059b10f78e6b72ada1ed2c919 \ No newline at end of file diff --git a/build/resources/main/schema.sql b/build/resources/main/schema.sql deleted file mode 100644 index 9930216..0000000 --- a/build/resources/main/schema.sql +++ /dev/null @@ -1,8 +0,0 @@ -CREATE TABLE IF NOT EXISTS transactions ( - id SERIAL PRIMARY KEY, - mpesa_reference VARCHAR(255), - checkout_request_id VARCHAR(255), - status VARCHAR(50), - amount DECIMAL(10,2), - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP -); diff --git a/build/tmp/compileJava/compileTransaction/stash-dir/MpesaController.class.uniqueId1 b/build/tmp/compileJava/compileTransaction/stash-dir/MpesaController.class.uniqueId1 deleted file mode 100644 index 521c44f..0000000 Binary files a/build/tmp/compileJava/compileTransaction/stash-dir/MpesaController.class.uniqueId1 and /dev/null differ diff --git a/build/tmp/compileJava/compileTransaction/stash-dir/MpesaService.class.uniqueId0 b/build/tmp/compileJava/compileTransaction/stash-dir/MpesaService.class.uniqueId0 deleted file mode 100644 index a0c6481..0000000 Binary files a/build/tmp/compileJava/compileTransaction/stash-dir/MpesaService.class.uniqueId0 and /dev/null differ diff --git a/build/tmp/compileJava/previous-compilation-data.bin b/build/tmp/compileJava/previous-compilation-data.bin deleted file mode 100644 index 19a15ca..0000000 Binary files a/build/tmp/compileJava/previous-compilation-data.bin and /dev/null differ diff --git a/gradle/wrapper/gradle-wrapper.jar b/gradle/wrapper/gradle-wrapper.jar new file mode 100644 index 0000000..1b33c55 Binary files /dev/null and b/gradle/wrapper/gradle-wrapper.jar differ diff --git a/src/main/java/com/test/payment/client/AirtelClient.java b/src/main/java/com/test/payment/client/AirtelClient.java new file mode 100644 index 0000000..2ae9b2a --- /dev/null +++ b/src/main/java/com/test/payment/client/AirtelClient.java @@ -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 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 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)); + } +} diff --git a/src/main/java/com/test/payment/client/MpesaClient.java b/src/main/java/com/test/payment/client/MpesaClient.java new file mode 100644 index 0000000..933a145 --- /dev/null +++ b/src/main/java/com/test/payment/client/MpesaClient.java @@ -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 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 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 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()); + } +} diff --git a/src/main/java/com/test/payment/client/MtnClient.java b/src/main/java/com/test/payment/client/MtnClient.java new file mode 100644 index 0000000..117186f --- /dev/null +++ b/src/main/java/com/test/payment/client/MtnClient.java @@ -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 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 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", "")); + } +} diff --git a/src/main/java/com/test/payment/client/ProviderHttpErrors.java b/src/main/java/com/test/payment/client/ProviderHttpErrors.java new file mode 100644 index 0000000..b3840d0 --- /dev/null +++ b/src/main/java/com/test/payment/client/ProviderHttpErrors.java @@ -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 map(ClientResponse response, String provider, Mono 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 mapBody(HttpStatusCode status, String body, String provider, Mono 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)); + } +} diff --git a/src/main/java/com/test/payment/configurations/DatabaseSchemaInitializer.java b/src/main/java/com/test/payment/configurations/DatabaseSchemaInitializer.java new file mode 100644 index 0000000..2e124c6 --- /dev/null +++ b/src/main/java/com/test/payment/configurations/DatabaseSchemaInitializer.java @@ -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(); + } +} diff --git a/src/main/java/com/test/payment/configurations/RedisConfig.java b/src/main/java/com/test/payment/configurations/RedisConfig.java deleted file mode 100644 index 0bed43b..0000000 --- a/src/main/java/com/test/payment/configurations/RedisConfig.java +++ /dev/null @@ -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 redisTemplate(LettuceConnectionFactory factory) { - RedisTemplate template = new RedisTemplate<>(); - template.setConnectionFactory(factory); - return template; - }*/ -} diff --git a/src/main/java/com/test/payment/configurations/ResilienceConfig.java b/src/main/java/com/test/payment/configurations/ResilienceConfig.java deleted file mode 100644 index d4b9f43..0000000 --- a/src/main/java/com/test/payment/configurations/ResilienceConfig.java +++ /dev/null @@ -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()); - } -} diff --git a/src/main/java/com/test/payment/configurations/WebClientConfig.java b/src/main/java/com/test/payment/configurations/WebClientConfig.java index db8a9c8..0af94bf 100644 --- a/src/main/java/com/test/payment/configurations/WebClientConfig.java +++ b/src/main/java/com/test/payment/configurations/WebClientConfig.java @@ -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(); } } diff --git a/src/main/java/com/test/payment/controller/AirtelController.java b/src/main/java/com/test/payment/controller/AirtelController.java new file mode 100644 index 0000000..38eb885 --- /dev/null +++ b/src/main/java/com/test/payment/controller/AirtelController.java @@ -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 pay(@Valid @RequestBody PaymentRequest request) { + return airtelService.initiatePayment(request); + } + + @PostMapping("/callback") + public Mono callback(@RequestBody AirtelCallbackPayload payload) { + return airtelService.handleCallback(payload); + } + + @GetMapping("/status/{transactionId}") + public Mono status(@PathVariable String transactionId) { + return airtelService.checkStatus(transactionId); + } +} diff --git a/src/main/java/com/test/payment/controller/GlobalExceptionHandler.java b/src/main/java/com/test/payment/controller/GlobalExceptionHandler.java new file mode 100644 index 0000000..2e1d0cb --- /dev/null +++ b/src/main/java/com/test/payment/controller/GlobalExceptionHandler.java @@ -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 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 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"); + } +} diff --git a/src/main/java/com/test/payment/controller/MpesaController.java b/src/main/java/com/test/payment/controller/MpesaController.java index 4287e86..27f5ec3 100644 --- a/src/main/java/com/test/payment/controller/MpesaController.java +++ b/src/main/java/com/test/payment/controller/MpesaController.java @@ -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> 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 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 callback(@RequestBody StkCallbackPayload payload) { + return mpesaService.handleCallback(payload); + } + + @GetMapping("/status/{checkoutRequestId}") + public Mono status(@PathVariable String checkoutRequestId) { + return mpesaService.checkStatus(checkoutRequestId); } } - diff --git a/src/main/java/com/test/payment/controller/MtnController.java b/src/main/java/com/test/payment/controller/MtnController.java new file mode 100644 index 0000000..12e108e --- /dev/null +++ b/src/main/java/com/test/payment/controller/MtnController.java @@ -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 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 callbackPost(@RequestBody MtnStatusResponseDto payload) { + return mtnService.handleCallback(payload); + } + + @PutMapping("/callback") + public Mono callbackPut(@RequestBody MtnStatusResponseDto payload) { + return mtnService.handleCallback(payload); + } + + @GetMapping("/status/{referenceId}") + public Mono status(@PathVariable String referenceId) { + return mtnService.checkStatus(referenceId); + } +} diff --git a/src/main/java/com/test/payment/controller/PaymentsController.java b/src/main/java/com/test/payment/controller/PaymentsController.java new file mode 100644 index 0000000..b783bc5 --- /dev/null +++ b/src/main/java/com/test/payment/controller/PaymentsController.java @@ -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 transactions(@RequestParam(required = false) String provider) { + return lifecycle.listTransactions(provider == null ? null : provider.toUpperCase()); + } +} diff --git a/src/main/java/com/test/payment/dto/AirtelCallbackPayload.java b/src/main/java/com/test/payment/dto/AirtelCallbackPayload.java new file mode 100644 index 0000000..aaed1e2 --- /dev/null +++ b/src/main/java/com/test/payment/dto/AirtelCallbackPayload.java @@ -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":"","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; + } +} diff --git a/src/main/java/com/test/payment/dto/AirtelPaymentRequestDto.java b/src/main/java/com/test/payment/dto/AirtelPaymentRequestDto.java new file mode 100644 index 0000000..3ddcdeb --- /dev/null +++ b/src/main/java/com/test/payment/dto/AirtelPaymentRequestDto.java @@ -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; + } +} diff --git a/src/main/java/com/test/payment/dto/AirtelResponseDto.java b/src/main/java/com/test/payment/dto/AirtelResponseDto.java new file mode 100644 index 0000000..ec9185c --- /dev/null +++ b/src/main/java/com/test/payment/dto/AirtelResponseDto.java @@ -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; + } +} diff --git a/src/main/java/com/test/payment/dto/CallbackAckDto.java b/src/main/java/com/test/payment/dto/CallbackAckDto.java new file mode 100644 index 0000000..940de84 --- /dev/null +++ b/src/main/java/com/test/payment/dto/CallbackAckDto.java @@ -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); + } +} diff --git a/src/main/java/com/test/payment/dto/ErrorResponseDto.java b/src/main/java/com/test/payment/dto/ErrorResponseDto.java new file mode 100644 index 0000000..937d8e1 --- /dev/null +++ b/src/main/java/com/test/payment/dto/ErrorResponseDto.java @@ -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 details; + private LocalDateTime timestamp; + + public static ErrorResponseDto of(String error, String message) { + return new ErrorResponseDto("FAILED", error, message, null, LocalDateTime.now()); + } +} diff --git a/src/main/java/com/test/payment/models/MpesaResponse.java b/src/main/java/com/test/payment/dto/MpesaResponse.java similarity index 94% rename from src/main/java/com/test/payment/models/MpesaResponse.java rename to src/main/java/com/test/payment/dto/MpesaResponse.java index 4bda54f..859daa6 100644 --- a/src/main/java/com/test/payment/models/MpesaResponse.java +++ b/src/main/java/com/test/payment/dto/MpesaResponse.java @@ -1,4 +1,4 @@ -package com.test.payment.models; +package com.test.payment.dto; import com.fasterxml.jackson.annotation.JsonProperty; import lombok.AllArgsConstructor; diff --git a/src/main/java/com/test/payment/dto/MtnPayRequestDto.java b/src/main/java/com/test/payment/dto/MtnPayRequestDto.java new file mode 100644 index 0000000..f3d9411 --- /dev/null +++ b/src/main/java/com/test/payment/dto/MtnPayRequestDto.java @@ -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; + } +} diff --git a/src/main/java/com/test/payment/dto/MtnStatusResponseDto.java b/src/main/java/com/test/payment/dto/MtnStatusResponseDto.java new file mode 100644 index 0000000..e009e21 --- /dev/null +++ b/src/main/java/com/test/payment/dto/MtnStatusResponseDto.java @@ -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); + } +} diff --git a/src/main/java/com/test/payment/dto/OAuth2TokenResponse.java b/src/main/java/com/test/payment/dto/OAuth2TokenResponse.java new file mode 100644 index 0000000..b5e5cb9 --- /dev/null +++ b/src/main/java/com/test/payment/dto/OAuth2TokenResponse.java @@ -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; + } + } +} diff --git a/src/main/java/com/test/payment/dto/PaymentRequest.java b/src/main/java/com/test/payment/dto/PaymentRequest.java new file mode 100644 index 0000000..ad8a085 --- /dev/null +++ b/src/main/java/com/test/payment/dto/PaymentRequest.java @@ -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; +} diff --git a/src/main/java/com/test/payment/dto/PaymentResultDto.java b/src/main/java/com/test/payment/dto/PaymentResultDto.java new file mode 100644 index 0000000..c2f18de --- /dev/null +++ b/src/main/java/com/test/payment/dto/PaymentResultDto.java @@ -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; +} diff --git a/src/main/java/com/test/payment/dto/StkCallbackPayload.java b/src/main/java/com/test/payment/dto/StkCallbackPayload.java new file mode 100644 index 0000000..2ccf77a --- /dev/null +++ b/src/main/java/com/test/payment/dto/StkCallbackPayload.java @@ -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; + } + + @Data + @JsonIgnoreProperties(ignoreUnknown = true) + public static class Item { + @JsonProperty("Name") + private String name; + + @JsonProperty("Value") + private Object value; + } +} diff --git a/src/main/java/com/test/payment/dto/StkQueryRequestDto.java b/src/main/java/com/test/payment/dto/StkQueryRequestDto.java new file mode 100644 index 0000000..30041b2 --- /dev/null +++ b/src/main/java/com/test/payment/dto/StkQueryRequestDto.java @@ -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; +} diff --git a/src/main/java/com/test/payment/dto/StkQueryResponseDto.java b/src/main/java/com/test/payment/dto/StkQueryResponseDto.java new file mode 100644 index 0000000..0b95a06 --- /dev/null +++ b/src/main/java/com/test/payment/dto/StkQueryResponseDto.java @@ -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; +} diff --git a/src/main/java/com/test/payment/dto/TransactionStatusDto.java b/src/main/java/com/test/payment/dto/TransactionStatusDto.java new file mode 100644 index 0000000..987b556 --- /dev/null +++ b/src/main/java/com/test/payment/dto/TransactionStatusDto.java @@ -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; +} diff --git a/src/main/java/com/test/payment/exceptions/MpesaBusyException.java b/src/main/java/com/test/payment/exceptions/MpesaBusyException.java deleted file mode 100644 index 8f366ca..0000000 --- a/src/main/java/com/test/payment/exceptions/MpesaBusyException.java +++ /dev/null @@ -1,8 +0,0 @@ -package com.test.payment.exceptions; - - -public class MpesaBusyException extends RuntimeException { - public MpesaBusyException(String msg) { - super(msg); - } -} diff --git a/src/main/java/com/test/payment/exceptions/MpesaPermanentException.java b/src/main/java/com/test/payment/exceptions/MpesaPermanentException.java deleted file mode 100644 index 527c550..0000000 --- a/src/main/java/com/test/payment/exceptions/MpesaPermanentException.java +++ /dev/null @@ -1,8 +0,0 @@ -package com.test.payment.exceptions; - - -public class MpesaPermanentException extends RuntimeException { - public MpesaPermanentException(String msg) { - super(msg); - } -} diff --git a/src/main/java/com/test/payment/exceptions/MpesaTransientException.java b/src/main/java/com/test/payment/exceptions/MpesaTransientException.java deleted file mode 100644 index cc7d46a..0000000 --- a/src/main/java/com/test/payment/exceptions/MpesaTransientException.java +++ /dev/null @@ -1,7 +0,0 @@ -package com.test.payment.exceptions; - -public class MpesaTransientException extends RuntimeException { - public MpesaTransientException(String msg, Throwable cause) { - super(msg, cause); - } -} diff --git a/src/main/java/com/test/payment/exceptions/ProviderBusyException.java b/src/main/java/com/test/payment/exceptions/ProviderBusyException.java new file mode 100644 index 0000000..9b0034e --- /dev/null +++ b/src/main/java/com/test/payment/exceptions/ProviderBusyException.java @@ -0,0 +1,8 @@ +package com.test.payment.exceptions; + + +public class ProviderBusyException extends RuntimeException { + public ProviderBusyException(String msg) { + super(msg); + } +} diff --git a/src/main/java/com/test/payment/exceptions/ProviderPermanentException.java b/src/main/java/com/test/payment/exceptions/ProviderPermanentException.java new file mode 100644 index 0000000..9bbccfb --- /dev/null +++ b/src/main/java/com/test/payment/exceptions/ProviderPermanentException.java @@ -0,0 +1,8 @@ +package com.test.payment.exceptions; + + +public class ProviderPermanentException extends RuntimeException { + public ProviderPermanentException(String msg) { + super(msg); + } +} diff --git a/src/main/java/com/test/payment/exceptions/ProviderProcessingException.java b/src/main/java/com/test/payment/exceptions/ProviderProcessingException.java new file mode 100644 index 0000000..4666434 --- /dev/null +++ b/src/main/java/com/test/payment/exceptions/ProviderProcessingException.java @@ -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); + } +} diff --git a/src/main/java/com/test/payment/exceptions/ProviderTransientException.java b/src/main/java/com/test/payment/exceptions/ProviderTransientException.java new file mode 100644 index 0000000..a0da308 --- /dev/null +++ b/src/main/java/com/test/payment/exceptions/ProviderTransientException.java @@ -0,0 +1,7 @@ +package com.test.payment.exceptions; + +public class ProviderTransientException extends RuntimeException { + public ProviderTransientException(String msg, Throwable cause) { + super(msg, cause); + } +} diff --git a/src/main/java/com/test/payment/jobs/MpesaTransactionJob.java b/src/main/java/com/test/payment/jobs/MpesaTransactionJob.java deleted file mode 100644 index 6dfeda0..0000000 --- a/src/main/java/com/test/payment/jobs/MpesaTransactionJob.java +++ /dev/null @@ -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(); - } -} diff --git a/src/main/java/com/test/payment/jobs/PaymentReconciliationJob.java b/src/main/java/com/test/payment/jobs/PaymentReconciliationJob.java new file mode 100644 index 0000000..150b51c --- /dev/null +++ b/src/main/java/com/test/payment/jobs/PaymentReconciliationJob.java @@ -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 servicesByProvider; + private final AtomicBoolean running = new AtomicBoolean(false); + + public PaymentReconciliationJob(PaymentInitiationRepository initiationRepository, + PaymentLifecycleService lifecycle, + Environment environment, + List 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())); + } +} diff --git a/src/main/java/com/test/payment/models/DatabaseSchema.java b/src/main/java/com/test/payment/models/DatabaseSchema.java new file mode 100644 index 0000000..6971684 --- /dev/null +++ b/src/main/java/com/test/payment/models/DatabaseSchema.java @@ -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 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) + ) + """ + ); +} diff --git a/src/main/java/com/test/payment/models/PaymentCallback.java b/src/main/java/com/test/payment/models/PaymentCallback.java new file mode 100644 index 0000000..f371279 --- /dev/null +++ b/src/main/java/com/test/payment/models/PaymentCallback.java @@ -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; +} diff --git a/src/main/java/com/test/payment/models/PaymentInitiation.java b/src/main/java/com/test/payment/models/PaymentInitiation.java new file mode 100644 index 0000000..642b27e --- /dev/null +++ b/src/main/java/com/test/payment/models/PaymentInitiation.java @@ -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; +} diff --git a/src/main/java/com/test/payment/models/PaymentProviderType.java b/src/main/java/com/test/payment/models/PaymentProviderType.java new file mode 100644 index 0000000..05e619b --- /dev/null +++ b/src/main/java/com/test/payment/models/PaymentProviderType.java @@ -0,0 +1,7 @@ +package com.test.payment.models; + +public enum PaymentProviderType { + MPESA, + AIRTEL, + MTN +} diff --git a/src/main/java/com/test/payment/models/PaymentRequest.java b/src/main/java/com/test/payment/models/PaymentRequest.java deleted file mode 100644 index 7000d4a..0000000 --- a/src/main/java/com/test/payment/models/PaymentRequest.java +++ /dev/null @@ -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; -} diff --git a/src/main/java/com/test/payment/models/PaymentResponse.java b/src/main/java/com/test/payment/models/PaymentResponse.java new file mode 100644 index 0000000..71630f8 --- /dev/null +++ b/src/main/java/com/test/payment/models/PaymentResponse.java @@ -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; +} diff --git a/src/main/java/com/test/payment/models/ProviderToken.java b/src/main/java/com/test/payment/models/ProviderToken.java new file mode 100644 index 0000000..31012f5 --- /dev/null +++ b/src/main/java/com/test/payment/models/ProviderToken.java @@ -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; +} diff --git a/src/main/java/com/test/payment/models/Transaction.java b/src/main/java/com/test/payment/models/Transaction.java index 076a26b..815d51c 100644 --- a/src/main/java/com/test/payment/models/Transaction.java +++ b/src/main/java/com/test/payment/models/Transaction.java @@ -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; } - diff --git a/src/main/java/com/test/payment/models/TransactionStatus.java b/src/main/java/com/test/payment/models/TransactionStatus.java new file mode 100644 index 0000000..d5df8db --- /dev/null +++ b/src/main/java/com/test/payment/models/TransactionStatus.java @@ -0,0 +1,7 @@ +package com.test.payment.models; + +public enum TransactionStatus { + PENDING, + SUCCESS, + FAILED +} diff --git a/src/main/java/com/test/payment/repository/MpesaRepository.java b/src/main/java/com/test/payment/repository/MpesaRepository.java deleted file mode 100644 index 289fa77..0000000 --- a/src/main/java/com/test/payment/repository/MpesaRepository.java +++ /dev/null @@ -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 { - - /* public Mono saveTransaction(Transaction transaction) { - return Mono.fromRunnable(() -> - redisTemplate.opsForHash().put("mpesa:transactions", transaction.getId(), transaction.getStatus()) - ).then(); - } - - public Flux findAllTransactions() { - List values = redisTemplate.opsForHash().values("mpesa:transactions"); - return Flux.fromIterable(values).cast(Transaction.class); - }*/ - - Flux findByStatus(String status); - Mono findByCheckoutRequestId(String checkoutRequestId); - - /* private final RedisTemplate redisTemplate; - - public MpesaRepository(RedisTemplate redisTemplate) { - this.redisTemplate = redisTemplate; - } - - */ -} diff --git a/src/main/java/com/test/payment/repository/PaymentCallbackRepository.java b/src/main/java/com/test/payment/repository/PaymentCallbackRepository.java new file mode 100644 index 0000000..86a0bc7 --- /dev/null +++ b/src/main/java/com/test/payment/repository/PaymentCallbackRepository.java @@ -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 { + + Mono findByInitiationId(Long initiationId); +} diff --git a/src/main/java/com/test/payment/repository/PaymentInitiationRepository.java b/src/main/java/com/test/payment/repository/PaymentInitiationRepository.java new file mode 100644 index 0000000..b15206e --- /dev/null +++ b/src/main/java/com/test/payment/repository/PaymentInitiationRepository.java @@ -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 { + + Flux findByStatusAndCreatedAtBefore(String status, LocalDateTime cutoff); +} diff --git a/src/main/java/com/test/payment/repository/PaymentResponseRepository.java b/src/main/java/com/test/payment/repository/PaymentResponseRepository.java new file mode 100644 index 0000000..019d9cb --- /dev/null +++ b/src/main/java/com/test/payment/repository/PaymentResponseRepository.java @@ -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 { + + Mono findByInitiationId(Long initiationId); + + Mono findByProviderReference(String providerReference); +} diff --git a/src/main/java/com/test/payment/repository/ProviderTokenRepository.java b/src/main/java/com/test/payment/repository/ProviderTokenRepository.java new file mode 100644 index 0000000..d7d7a2a --- /dev/null +++ b/src/main/java/com/test/payment/repository/ProviderTokenRepository.java @@ -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 { + + Mono findFirstByProviderAndExpiresAtAfterOrderByIdDesc(String provider, LocalDateTime cutoff); + + Mono deleteByProvider(String provider); +} diff --git a/src/main/java/com/test/payment/repository/TransactionRepository.java b/src/main/java/com/test/payment/repository/TransactionRepository.java new file mode 100644 index 0000000..f941bce --- /dev/null +++ b/src/main/java/com/test/payment/repository/TransactionRepository.java @@ -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 { + + Mono findByInitiationId(Long initiationId); + + Flux findByProvider(String provider); + + Flux findByStatus(String status); +} diff --git a/src/main/java/com/test/payment/service/AirtelService.java b/src/main/java/com/test/payment/service/AirtelService.java new file mode 100644 index 0000000..5475671 --- /dev/null +++ b/src/main/java/com/test/payment/service/AirtelService.java @@ -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 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 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 checkStatus(String providerReference) { + return lifecycle.checkStatus(provider(), providerReference, this::queryProvider); + } + + private Mono 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; + } + } +} diff --git a/src/main/java/com/test/payment/service/AirtelTokenService.java b/src/main/java/com/test/payment/service/AirtelTokenService.java new file mode 100644 index 0000000..85b04a7 --- /dev/null +++ b/src/main/java/com/test/payment/service/AirtelTokenService.java @@ -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 getToken() { + return tokenCache.getToken(PaymentProviderType.AIRTEL.name(), this::fetchToken); + } + + public Mono evictToken() { + return tokenCache.evictToken(PaymentProviderType.AIRTEL.name()); + } + + private Mono 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())); + } +} diff --git a/src/main/java/com/test/payment/service/MpesaService.java b/src/main/java/com/test/payment/service/MpesaService.java index 9657783..f9497b6 100644 --- a/src/main/java/com/test/payment/service/MpesaService.java +++ b/src/main/java/com/test/payment/service/MpesaService.java @@ -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 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 initiatePayment(PaymentRequest request) { - String businessShortCode = environment.getProperty("mpesa.business-short-code"); + @Override + public Mono 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 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 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 checkStatus(String providerReference) { + return lifecycle.checkStatus(provider(), providerReference, this::queryProvider); + } + + private Mono 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 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 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 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 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 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; + } } } - diff --git a/src/main/java/com/test/payment/service/MpesaServiceaa.java b/src/main/java/com/test/payment/service/MpesaServiceaa.java deleted file mode 100644 index c8d9aa9..0000000 --- a/src/main/java/com/test/payment/service/MpesaServiceaa.java +++ /dev/null @@ -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 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 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 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> 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 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 getToken() { - return Mono.fromSupplier(() -> "Q3HCot6dNLTpUtpitkr5Tatsa4KB"); - } - - public Mono 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 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) - ); - }*/ -} diff --git a/src/main/java/com/test/payment/service/MpesaTokenService.java b/src/main/java/com/test/payment/service/MpesaTokenService.java index f39537d..219dd80 100644 --- a/src/main/java/com/test/payment/service/MpesaTokenService.java +++ b/src/main/java/com/test/payment/service/MpesaTokenService.java @@ -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 redisTemplate; + private final WebClient mpesaWebClient; + private final TokenCacheService tokenCache; + private final Environment environment; public Mono 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 evictToken() { + return tokenCache.evictToken(PaymentProviderType.MPESA.name()); + } + + private Mono 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()))); } } diff --git a/src/main/java/com/test/payment/service/MtnService.java b/src/main/java/com/test/payment/service/MtnService.java new file mode 100644 index 0000000..aea8025 --- /dev/null +++ b/src/main/java/com/test/payment/service/MtnService.java @@ -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 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 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 checkStatus(String providerReference) { + return lifecycle.checkStatus(provider(), providerReference, this::queryProvider); + } + + private Mono 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; + } + } +} diff --git a/src/main/java/com/test/payment/service/MtnTokenService.java b/src/main/java/com/test/payment/service/MtnTokenService.java new file mode 100644 index 0000000..515bace --- /dev/null +++ b/src/main/java/com/test/payment/service/MtnTokenService.java @@ -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 getToken() { + return tokenCache.getToken(PaymentProviderType.MTN.name(), this::fetchToken); + } + + public Mono evictToken() { + return tokenCache.evictToken(PaymentProviderType.MTN.name()); + } + + private Mono 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())); + } +} diff --git a/src/main/java/com/test/payment/service/PaymentLifecycleService.java b/src/main/java/com/test/payment/service/PaymentLifecycleService.java new file mode 100644 index 0000000..c380371 --- /dev/null +++ b/src/main/java/com/test/payment/service/PaymentLifecycleService.java @@ -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 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 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 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 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 checkStatus(String provider, String providerReference, + Function> 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 reconcile(PaymentInitiation initiation, + Function> 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 listTransactions(String provider) { + return provider == null ? transactionRepository.findAll() : transactionRepository.findByProvider(provider); + } + + private Mono failTerminal(PaymentInitiation initiation, String reason) { + return updateStatus(initiation, TransactionStatus.FAILED.name()) + .flatMap(updated -> recordTransaction(updated, null, null, reason, null, null, "RECONCILIATION")); + } + + private Mono 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 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 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 cb = tuple.getT1(); + Optional 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 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 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); + } +} diff --git a/src/main/java/com/test/payment/service/PaymentProviderService.java b/src/main/java/com/test/payment/service/PaymentProviderService.java new file mode 100644 index 0000000..cf342d8 --- /dev/null +++ b/src/main/java/com/test/payment/service/PaymentProviderService.java @@ -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 initiatePayment(PaymentRequest request); + + Mono checkStatus(String providerReference); +} diff --git a/src/main/java/com/test/payment/service/TokenCacheService.java b/src/main/java/com/test/payment/service/TokenCacheService.java new file mode 100644 index 0000000..7690e3f --- /dev/null +++ b/src/main/java/com/test/payment/service/TokenCacheService.java @@ -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 getToken(String provider, Supplier> 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 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 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 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 fetchAndStore(String provider, Supplier> 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 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); + } +} diff --git a/src/main/java/com/test/payment/utils/MpesaUtils.java b/src/main/java/com/test/payment/utils/MpesaUtils.java index 6cedc3a..81060ee 100644 --- a/src/main/java/com/test/payment/utils/MpesaUtils.java +++ b/src/main/java/com/test/payment/utils/MpesaUtils.java @@ -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; diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 8d45436..4c95555 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -1,53 +1,61 @@ -#spring: - # redis: - # host: localhost - # port: 6379 - resilience4j: ratelimiter: instances: mpesaLimiter: - limit-for-period: 10 + limit-for-period: 5 limit-refresh-period: 1s - timeout-duration: 0s + timeout-duration: 5s + airtelLimiter: + limit-for-period: 5 + limit-refresh-period: 1s + timeout-duration: 5s + mtnLimiter: + limit-for-period: 5 + limit-refresh-period: 1s + timeout-duration: 5s retry: instances: - mpesaRetry: + mpesaRetry: &provider-retry max-attempts: 3 wait-duration: 2s retry-exceptions: - org.springframework.web.reactive.function.client.WebClientRequestException - java.io.IOException - - com.example.mpesa.exceptions.MpesaBusyException + - com.test.payment.exceptions.ProviderTransientException ignore-exceptions: - - com.test.payment.exceptions.MpesaPermanentException + - com.test.payment.exceptions.ProviderPermanentException - java.lang.IllegalArgumentException - + airtelRetry: *provider-retry + mtnRetry: *provider-retry circuitbreaker: instances: - mpesaCircuitBreaker: + mpesaCircuitBreaker: &provider-circuit-breaker sliding-window-size: 10 + minimum-number-of-calls: 5 permitted-number-of-calls-in-half-open-state: 3 failure-rate-threshold: 50 wait-duration-in-open-state: 10s - - - - + ignore-exceptions: + - com.test.payment.exceptions.ProviderPermanentException + - com.test.payment.exceptions.ProviderProcessingException + airtelCircuitBreaker: *provider-circuit-breaker + mtnCircuitBreaker: *provider-circuit-breaker spring: r2dbc: url: r2dbc:h2:mem:///mpesa_db;DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE username: sa password: - sql: - init: - mode: always - schema-locations: classpath:schema.sql main: web-application-type: reactive + data: + redis: + host: localhost + port: 6379 + connect-timeout: 1s + timeout: 1s logging: level: @@ -62,9 +70,35 @@ springdoc: path: /v3/api-docs packages-to-scan: com.test.payment.controller +payments: + token-expiry-buffer-seconds: 60 + reconciliation: + # transactions still PENDING after this age are re-checked against the provider + pending-age: 5m + # how often the reconciliation job runs + fixed-delay: 60s + mpesa: base-url: https://sandbox.safaricom.co.ke consumer-key: k6e7LtBNeVX7V8MPqB7P83FsZio8cRZD consumer-secret: cGwiWzhDGopC3dho business-short-code: 174379 - pass-key: bfb279f9aa9bdbcf158e97dd71a467cd2e0c893059b10f78e6b72ada1ed2c919 \ No newline at end of file + pass-key: bfb279f9aa9bdbcf158e97dd71a467cd2e0c893059b10f78e6b72ada1ed2c919 + callback-url: https://mydomain.com/api/mpesa/callback + +airtel: + base-url: https://openapiuat.airtel.africa + client-id: REPLACE_WITH_AIRTEL_CLIENT_ID + client-secret: REPLACE_WITH_AIRTEL_CLIENT_SECRET + country: KE + currency: KES + +mtn: + base-url: https://sandbox.momodeveloper.mtn.com + subscription-key: REPLACE_WITH_MTN_SUBSCRIPTION_KEY + api-user: REPLACE_WITH_MTN_API_USER + api-key: REPLACE_WITH_MTN_API_KEY + target-environment: sandbox + # MoMo sandbox only supports EUR + currency: EUR + callback-url: https://mydomain.com/api/mtn/callback diff --git a/src/main/resources/schema.sql b/src/main/resources/schema.sql deleted file mode 100644 index 9930216..0000000 --- a/src/main/resources/schema.sql +++ /dev/null @@ -1,8 +0,0 @@ -CREATE TABLE IF NOT EXISTS transactions ( - id SERIAL PRIMARY KEY, - mpesa_reference VARCHAR(255), - checkout_request_id VARCHAR(255), - status VARCHAR(50), - amount DECIMAL(10,2), - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP -);