diff --git a/README.md b/README.md index b067a71026..d21a04c127 100644 --- a/README.md +++ b/README.md @@ -69,7 +69,7 @@ You must have two resources: } ``` -## Optional +## Optional[app-transaction-report-service](app-transaction-report-service) You can use any approach to store transaction data but you should consider that we may deal with high volume scenarios where we have a huge amount of writes and reads for the same data at the same time. How would you tackle this requirement? @@ -80,3 +80,47 @@ You can use Graphql; When you finish your challenge, after forking a repository, you **must** open a pull request to our repository. There are no limitations to the implementation, you can follow the programming paradigm, modularization, and style that you feel is the most appropriate solution. If you have any questions, please let us know. + + + + +# Resumen de Implementación + +## Endpoints Implementados + +### 1. **Gateway de Orquestación** + +* `POST /api/transactions` → Crea transacción (proxy hacia Transaction Service) +* `GET /api/transactions/{id}` → Consulta transacción + + +### 2. **Anti-Fraud Listener (Kafka)** + +* Consumidor del tópico: `transactions.validated` +* Procesa validación antifraude y actualiza estado de transacción + +### 3. **Producer de Eventos** + +* Tópico: `transactions.created` +* Publicación cuando se crea una transacción + +--- + +## Tecnologías Implementadas + +| Tecnología | Uso | +| -------------------------------- | ----------------------------------------------------- | +| **Spring Boot 3.3.5** | Core del proyecto | +| **Spring WebFlux** | API reactiva y backpressure | +| **Spring Cloud Gateway 2023.x** | API Gateway reactivo | +| **Resilience4J Circuit Breaker** | Protección ante fallos en downstream | +| **Redis Rate Limiting** | Límite de requests por IP | +| **Apache Kafka** | Broker de eventos | +| **Reactor Kafka** | Publisher/Subscriber reactivo | +| **Schema Registry (Confluent)** | Gestión y versionamiento de Avro schemas | +| **Avro** | Serialización eficiente binaria para eventos | +| **Docker Compose** | Orquestación de Kafka + Registry + Redis | +| **RecordNameStrategy** | Estrategia recomendada para versionamiento de schemas | + +--- + diff --git a/api-transaction-gateway/.gitattributes b/api-transaction-gateway/.gitattributes new file mode 100644 index 0000000000..3b41682ac5 --- /dev/null +++ b/api-transaction-gateway/.gitattributes @@ -0,0 +1,2 @@ +/mvnw text eol=lf +*.cmd text eol=crlf diff --git a/api-transaction-gateway/.gitignore b/api-transaction-gateway/.gitignore new file mode 100644 index 0000000000..667aaef0c8 --- /dev/null +++ b/api-transaction-gateway/.gitignore @@ -0,0 +1,33 @@ +HELP.md +target/ +.mvn/wrapper/maven-wrapper.jar +!**/src/main/**/target/ +!**/src/test/**/target/ + +### STS ### +.apt_generated +.classpath +.factorypath +.project +.settings +.springBeans +.sts4-cache + +### IntelliJ IDEA ### +.idea +*.iws +*.iml +*.ipr + +### NetBeans ### +/nbproject/private/ +/nbbuild/ +/dist/ +/nbdist/ +/.nb-gradle/ +build/ +!**/src/main/**/build/ +!**/src/test/**/build/ + +### VS Code ### +.vscode/ diff --git a/api-transaction-gateway/pom.xml b/api-transaction-gateway/pom.xml new file mode 100644 index 0000000000..c1fd3311b8 --- /dev/null +++ b/api-transaction-gateway/pom.xml @@ -0,0 +1,82 @@ + + + 4.0.0 + + org.springframework.boot + spring-boot-starter-parent + 3.3.5 + + + + com.pe.yape.service.challenge + api-transaction-gateway + 0.0.1-SNAPSHOT + API Transaction Gateway Service Application + + + + 17 + + + + + + org.springframework.cloud + spring-cloud-dependencies + 2023.0.3 + pom + import + + + + + + + + org.springframework.cloud + spring-cloud-starter-gateway + + + + org.springframework.cloud + spring-cloud-starter-circuitbreaker-reactor-resilience4j + + + + org.springframework.boot + spring-boot-starter-data-redis-reactive + + + + org.springframework.boot + spring-boot-starter-webflux + + + + io.github.resilience4j + resilience4j-spring-boot3 + + + + org.springframework.boot + spring-boot-starter-actuator + + + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + + org.springframework.boot + spring-boot-maven-plugin + + + + + diff --git a/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/ApiTransactionGateway.java b/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/ApiTransactionGateway.java new file mode 100644 index 0000000000..51724ec2f9 --- /dev/null +++ b/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/ApiTransactionGateway.java @@ -0,0 +1,13 @@ +package com.pe.yape.service.challenge.transactions; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class ApiTransactionGateway { + + public static void main(String[] args) { + SpringApplication.run(ApiTransactionGateway.class, args); + } + +} diff --git a/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/config/WebClientConfig.java b/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/config/WebClientConfig.java new file mode 100644 index 0000000000..12fe30b640 --- /dev/null +++ b/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/config/WebClientConfig.java @@ -0,0 +1,23 @@ +package com.pe.yape.service.challenge.transactions.config; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +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.Builder webClientBuilder() { + return WebClient.builder() + .clientConnector( + new ReactorClientHttpConnector( + HttpClient.create() + .responseTimeout(Duration.ofSeconds(5)) + )); + } +} diff --git a/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/controller/TransactionGatewayController.java b/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/controller/TransactionGatewayController.java new file mode 100644 index 0000000000..243cfb5448 --- /dev/null +++ b/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/controller/TransactionGatewayController.java @@ -0,0 +1,63 @@ +package com.pe.yape.service.challenge.transactions.controller; + +import com.pe.yape.service.challenge.transactions.httpclient.dto.*; +import io.github.resilience4j.circuitbreaker.annotation.CircuitBreaker; +import io.github.resilience4j.retry.annotation.Retry; +import io.github.resilience4j.timelimiter.annotation.TimeLimiter; +import org.springframework.web.bind.annotation.*; +import org.springframework.web.reactive.function.client.WebClient; +import reactor.core.publisher.Mono; + +import java.math.BigDecimal; +import java.time.LocalDateTime; + +@RestController +@RequestMapping("/api/transactions") +public class TransactionGatewayController { + + private final WebClient client; + + public TransactionGatewayController(WebClient.Builder builder) { + this.client = builder.build(); + } + + @GetMapping("/{transactionExternalId}") + @CircuitBreaker(name = "transactionServices", fallbackMethod = "fallbackTransaction") + @Retry(name = "transactionRetry") + @TimeLimiter(name = "transactionTimeout") + public Mono getById(@PathVariable String transactionExternalId) { + return client.get() + .uri("http://localhost:8085/api/transactions-report/transactions/{id}", transactionExternalId) + .retrieve() + .bodyToMono(SearchTransactionResponse.class); + } + + @PostMapping("/") + @CircuitBreaker(name = "transactionServices", fallbackMethod = "fallbackCreateTransaction") + @Retry(name = "transactionRetry") + @TimeLimiter(name = "transactionTimeout") + public Mono create(@RequestBody CreateTransactionRequest request) { + + return client.post() + .uri("http://localhost:8080/api/transactions/") + .bodyValue(request) + .retrieve() + .bodyToMono(CreateTransactionResponse.class); + } + + + public Mono fallbackTransaction(String id, Throwable ex) { + return Mono.just(new SearchTransactionResponse( + id, + new TransactionType("UNKNOWN"), + new TransactionStatus("FAILED"), + new BigDecimal("0"), + LocalDateTime.now() + )); + } + + public Mono fallbackCreateTransaction(CreateTransactionRequest request, + Throwable ex) { + return Mono.just(new CreateTransactionResponse("none")); + } +} diff --git a/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/httpclient/dto/CreateTransactionRequest.java b/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/httpclient/dto/CreateTransactionRequest.java new file mode 100644 index 0000000000..ef66534703 --- /dev/null +++ b/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/httpclient/dto/CreateTransactionRequest.java @@ -0,0 +1,7 @@ +package com.pe.yape.service.challenge.transactions.httpclient.dto; + +public record CreateTransactionRequest(String accountExternalIdDebit, + String accountExternalIdCredit, + Integer transferTypeId, + Double value) { +} diff --git a/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/httpclient/dto/CreateTransactionResponse.java b/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/httpclient/dto/CreateTransactionResponse.java new file mode 100644 index 0000000000..131c171a63 --- /dev/null +++ b/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/httpclient/dto/CreateTransactionResponse.java @@ -0,0 +1,4 @@ +package com.pe.yape.service.challenge.transactions.httpclient.dto; + +public record CreateTransactionResponse(String transactionExternalId) { +} diff --git a/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/httpclient/dto/SearchTransactionResponse.java b/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/httpclient/dto/SearchTransactionResponse.java new file mode 100644 index 0000000000..5f175204b4 --- /dev/null +++ b/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/httpclient/dto/SearchTransactionResponse.java @@ -0,0 +1,9 @@ +package com.pe.yape.service.challenge.transactions.httpclient.dto; + +import java.math.BigDecimal; +import java.time.LocalDateTime; + +public record SearchTransactionResponse(String transactionExternalId, TransactionType transactionType, + TransactionStatus transactionStatus, BigDecimal value, + LocalDateTime createdAt) { +} diff --git a/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/httpclient/dto/TransactionStatus.java b/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/httpclient/dto/TransactionStatus.java new file mode 100644 index 0000000000..804ed8dfcf --- /dev/null +++ b/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/httpclient/dto/TransactionStatus.java @@ -0,0 +1,4 @@ +package com.pe.yape.service.challenge.transactions.httpclient.dto; + +public record TransactionStatus(String name) { +} \ No newline at end of file diff --git a/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/httpclient/dto/TransactionType.java b/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/httpclient/dto/TransactionType.java new file mode 100644 index 0000000000..598fb5135f --- /dev/null +++ b/api-transaction-gateway/src/main/java/com/pe/yape/service/challenge/transactions/httpclient/dto/TransactionType.java @@ -0,0 +1,5 @@ +package com.pe.yape.service.challenge.transactions.httpclient.dto; + +public record TransactionType(String name) { +} + diff --git a/api-transaction-gateway/src/main/resources/application.yml b/api-transaction-gateway/src/main/resources/application.yml new file mode 100644 index 0000000000..30e7aa463a --- /dev/null +++ b/api-transaction-gateway/src/main/resources/application.yml @@ -0,0 +1,32 @@ +server: + port: 9000 +spring: + redis: + host: localhost + port: 6379 + +resilience4j: + circuitbreaker: + instances: + transactionServices: + sliding-window-size: 6 + permitted-number-of-calls-in-half-open-state: 2 + failure-rate-threshold: 50 + wait-duration-in-open-state: 10s + + retry: + instances: + transactionRetry: + max-attempts: 3 + wait-duration: 2s + + timelimiter: + instances: + transactionTimeout: + timeout-duration: 4s + + +logging: + level: + root: INFO + org.springframework.web: DEBUG diff --git a/app-anti-fraud-service/.gitattributes b/app-anti-fraud-service/.gitattributes new file mode 100644 index 0000000000..3b41682ac5 --- /dev/null +++ b/app-anti-fraud-service/.gitattributes @@ -0,0 +1,2 @@ +/mvnw text eol=lf +*.cmd text eol=crlf diff --git a/app-anti-fraud-service/.gitignore b/app-anti-fraud-service/.gitignore new file mode 100644 index 0000000000..667aaef0c8 --- /dev/null +++ b/app-anti-fraud-service/.gitignore @@ -0,0 +1,33 @@ +HELP.md +target/ +.mvn/wrapper/maven-wrapper.jar +!**/src/main/**/target/ +!**/src/test/**/target/ + +### STS ### +.apt_generated +.classpath +.factorypath +.project +.settings +.springBeans +.sts4-cache + +### IntelliJ IDEA ### +.idea +*.iws +*.iml +*.ipr + +### NetBeans ### +/nbproject/private/ +/nbbuild/ +/dist/ +/nbdist/ +/.nb-gradle/ +build/ +!**/src/main/**/build/ +!**/src/test/**/build/ + +### VS Code ### +.vscode/ diff --git a/app-anti-fraud-service/pom.xml b/app-anti-fraud-service/pom.xml new file mode 100644 index 0000000000..ad681a48e2 --- /dev/null +++ b/app-anti-fraud-service/pom.xml @@ -0,0 +1,134 @@ + + 4.0.0 + + + org.springframework.boot + spring-boot-starter-parent + 3.3.5 + + + + com.pe.yape.service.challenge + app-anti-fraud-service + 0.0.1-SNAPSHOT + jar + + + 17 + 3.3.5 + 1.11.4 + 1.3.22 + + + + + confluent + https://packages.confluent.io/maven/ + + + + + + + org.springframework.boot + spring-boot-starter-webflux + + + + org.springframework.boot + spring-boot-starter-logging + + + + + + + io.projectreactor.kafka + reactor-kafka + 1.3.22 + + + + + org.apache.avro + avro + 1.11.4 + + + + io.confluent + kafka-avro-serializer + 7.5.0 + + + + + + org.springframework.boot + spring-boot-starter-log4j2 + + + + + org.springframework.boot + spring-boot-starter-test + test + + + org.junit.vintage + junit-vintage-engine + + + + + + org.projectlombok + lombok + 1.18.38 + provided + + + + + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + + org.apache.avro + avro-maven-plugin + ${avro.version} + + + generate-sources + + schema + + + ${project.basedir}/src/main/avro + ${project.build.directory}/generated-sources/avro + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + 9 + 9 + + + + + diff --git a/app-anti-fraud-service/src/main/avro/TransactionCreatedEvent.avsc b/app-anti-fraud-service/src/main/avro/TransactionCreatedEvent.avsc new file mode 100644 index 0000000000..76615a0f7c --- /dev/null +++ b/app-anti-fraud-service/src/main/avro/TransactionCreatedEvent.avsc @@ -0,0 +1,12 @@ +{ + "type": "record", + "name": "TransactionCreatedEvent", + "namespace": "com.pe.yape.service.challenge.transactions.avro", + "fields": [ + {"name": "transactionExternalId", "type": "string"}, + {"name": "accountExternalIdDebit", "type": "string"}, + {"name": "accountExternalIdCredit", "type": "string"}, + {"name": "transferTypeId", "type": "int"}, + {"name": "value", "type": "double"} + ] +} diff --git a/app-anti-fraud-service/src/main/avro/TransactionValidatedEvent.avsc b/app-anti-fraud-service/src/main/avro/TransactionValidatedEvent.avsc new file mode 100644 index 0000000000..36cde2e1e6 --- /dev/null +++ b/app-anti-fraud-service/src/main/avro/TransactionValidatedEvent.avsc @@ -0,0 +1,11 @@ +{ + "type": "record", + "name": "TransactionValidatedEvent", + "namespace": "com.pe.yape.service.challenge.transactions.avro", + "fields": [ + {"name": "transactionExternalId", "type": "string"}, + {"name": "status", "type": "string"}, + {"name": "reasonCode", "type": ["null","string"], "default": null}, + {"name": "validatedAt", "type": "string"} + ] +} diff --git a/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/AntiFraudServiceApplication.java b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/AntiFraudServiceApplication.java new file mode 100644 index 0000000000..6b9f39184f --- /dev/null +++ b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/AntiFraudServiceApplication.java @@ -0,0 +1,13 @@ +package com.pe.yape.service.challenge.antifraud; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class AntiFraudServiceApplication { + + public static void main(String[] args) { + SpringApplication.run(AntiFraudServiceApplication.class, args); + } + +} diff --git a/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/config/KafkaConfig.java b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/config/KafkaConfig.java new file mode 100644 index 0000000000..c3f6fecbe2 --- /dev/null +++ b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/config/KafkaConfig.java @@ -0,0 +1,64 @@ +package com.pe.yape.service.challenge.antifraud.config; + +import com.pe.yape.service.challenge.transactions.avro.TransactionCreatedEvent; +import com.pe.yape.service.challenge.transactions.avro.TransactionValidatedEvent; +import io.confluent.kafka.serializers.KafkaAvroDeserializer; +import io.confluent.kafka.serializers.KafkaAvroSerializer; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import reactor.kafka.receiver.ReceiverOptions; +import reactor.kafka.sender.SenderOptions; + +import java.util.Map; + +@Configuration +public class KafkaConfig { + + public static final String TOPIC_TRANSACTIONS_CREATED = "transactions.created"; + public static final String TOPIC_TRANSACTIONS_VALIDATED = "transactions.validated"; + + @Bean + public NewTopic transactionsCreatedTopic() { + return new NewTopic(TOPIC_TRANSACTIONS_CREATED, 3, (short) 1); + } + + @Bean + public NewTopic transactionsValidatedTopic() { + return new NewTopic(TOPIC_TRANSACTIONS_VALIDATED, 3, (short) 1); + } + + @Bean + public ReceiverOptions receiverOptions() { + Map props = Map.of( + ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092", + ConsumerConfig.GROUP_ID_CONFIG, "anti-fraud-service-group", + ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest", + "schema.registry.url", "http://localhost:8081", + ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class, + ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class, + "specific.avro.reader",true + ); + + return ReceiverOptions.create(props) + .subscription(java.util.List.of(TOPIC_TRANSACTIONS_CREATED)); + } + + @Bean + public SenderOptions senderOptions(){ + Map props = Map.of( + ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092", + ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class, + ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class, + "schema.registry.url", "http://localhost:8081" + ); + + return SenderOptions.create(props); + } + + +} diff --git a/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/config/properties/AntifraudProperties.java b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/config/properties/AntifraudProperties.java new file mode 100644 index 0000000000..64ba6e17e5 --- /dev/null +++ b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/config/properties/AntifraudProperties.java @@ -0,0 +1,18 @@ +package com.pe.yape.service.challenge.antifraud.config.properties; + +import lombok.Getter; +import lombok.Setter; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.stereotype.Component; + +@Component +@ConfigurationProperties(prefix = "application.rules") + +@Getter +@Setter +public class AntifraudProperties { + + private String amountLimit; + +} diff --git a/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/domain/TransactionStatus.java b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/domain/TransactionStatus.java new file mode 100644 index 0000000000..d196f62ac9 --- /dev/null +++ b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/domain/TransactionStatus.java @@ -0,0 +1,8 @@ +package com.pe.yape.service.challenge.antifraud.domain; + +public enum TransactionStatus { + + APPROVED, + + REJECTED +} diff --git a/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/domain/ValidationResult.java b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/domain/ValidationResult.java new file mode 100644 index 0000000000..f4123ee958 --- /dev/null +++ b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/domain/ValidationResult.java @@ -0,0 +1,16 @@ +package com.pe.yape.service.challenge.antifraud.domain; + +import lombok.*; + +@Getter +@Setter +@Builder +@AllArgsConstructor +@NoArgsConstructor +public class ValidationResult { + + private TransactionStatus transactionStatus; + + private String reasonCode; + +} diff --git a/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/messaging/TransactionCreatedConsumer.java b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/messaging/TransactionCreatedConsumer.java new file mode 100644 index 0000000000..baeef2f6d9 --- /dev/null +++ b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/messaging/TransactionCreatedConsumer.java @@ -0,0 +1,68 @@ +package com.pe.yape.service.challenge.antifraud.messaging; + + +import com.pe.yape.service.challenge.transactions.avro.TransactionCreatedEvent; +import com.pe.yape.service.challenge.transactions.avro.TransactionValidatedEvent; +import com.pe.yape.service.challenge.antifraud.service.AntifraudService; + +import jakarta.annotation.PostConstruct; +import lombok.extern.slf4j.Slf4j; +import reactor.kafka.receiver.KafkaReceiver; +import reactor.kafka.receiver.ReceiverOptions; +import reactor.core.publisher.Flux; +import org.springframework.stereotype.Component; +import reactor.kafka.receiver.ReceiverRecord; + +import java.time.OffsetDateTime; +import java.time.format.DateTimeFormatter; + +@Component +@Slf4j +public class TransactionCreatedConsumer { + + private final TransactionValidatedProducer producer; + private final AntifraudService antifraudService; + private final ReceiverOptions receiverOptions; + + public TransactionCreatedConsumer(TransactionValidatedProducer producer, + AntifraudService antifraudService, + ReceiverOptions receiverOptions) { + this.producer = producer; + this.antifraudService = antifraudService; + this.receiverOptions = receiverOptions; + } + + @PostConstruct + public void startConsumer() { + log.info("Starting Kafka TransactionCreatedEvent consumer..."); + + Flux> kafkaFlux = + KafkaReceiver.create(receiverOptions).receive(); + + kafkaFlux.flatMap(record -> { + TransactionCreatedEvent transactionCreatedEvent = record.value(); + log.debug("Received message for transaction: {}", transactionCreatedEvent.getTransactionExternalId()); + + return antifraudService.validate(transactionCreatedEvent) + .map(validationResult -> + TransactionValidatedEvent.newBuilder() + .setTransactionExternalId(transactionCreatedEvent.getTransactionExternalId()) + .setStatus(validationResult.getTransactionStatus().toString()) + .setReasonCode(validationResult.getReasonCode()) + .setValidatedAt(OffsetDateTime.now().format(DateTimeFormatter.ISO_OFFSET_DATE_TIME)) // Use ISO_OFFSET_DATE_TIME + .build() + ).flatMap(transactionValidatedEvent -> + producer.publish(transactionValidatedEvent, transactionCreatedEvent.getTransactionExternalId().toString()) + ) + .doOnNext(s -> { + record.receiverOffset().acknowledge(); + log.info("Acknowledged transaction ID: {}", transactionCreatedEvent.getTransactionExternalId()); + }); + }) + .subscribe( + result -> log.debug("Processing loop successful for a record."), + error -> log.error("Error during Kafka consumption process: {}", error.getMessage(), error), + () -> log.warn("Kafka consumer flux completed unexpectedly.") + ); + } +} diff --git a/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/messaging/TransactionValidatedProducer.java b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/messaging/TransactionValidatedProducer.java new file mode 100644 index 0000000000..1f8056a230 --- /dev/null +++ b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/messaging/TransactionValidatedProducer.java @@ -0,0 +1,34 @@ +package com.pe.yape.service.challenge.antifraud.messaging; + + +import com.pe.yape.service.challenge.transactions.avro.TransactionValidatedEvent; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import reactor.core.publisher.Mono; +import reactor.kafka.sender.KafkaSender; +import reactor.kafka.sender.SenderOptions; +import reactor.kafka.sender.SenderRecord; +import reactor.kafka.sender.SenderResult; + +import static com.pe.yape.service.challenge.antifraud.config.KafkaConfig.TOPIC_TRANSACTIONS_VALIDATED; + +@Component +@Slf4j +public class TransactionValidatedProducer { + + private final KafkaSender sender; + + public TransactionValidatedProducer(SenderOptions senderOptions) { + this.sender = KafkaSender.create(senderOptions); + } + + public Mono publish(TransactionValidatedEvent event, String key) { + SenderRecord record = + SenderRecord.create(TOPIC_TRANSACTIONS_VALIDATED, null, null, key, event, key); + + return sender.send(Mono.just(record)) + .doOnNext(result -> log.info("Send message {} to {}" , key , event) ) + .map(SenderResult::correlationMetadata) + .ignoreElements(); + } +} diff --git a/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/service/AntifraudService.java b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/service/AntifraudService.java new file mode 100644 index 0000000000..335ce70edc --- /dev/null +++ b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/service/AntifraudService.java @@ -0,0 +1,40 @@ +package com.pe.yape.service.challenge.antifraud.service; + +import com.pe.yape.service.challenge.transactions.avro.TransactionCreatedEvent; +import com.pe.yape.service.challenge.antifraud.config.properties.AntifraudProperties; +import com.pe.yape.service.challenge.antifraud.domain.TransactionStatus; +import com.pe.yape.service.challenge.antifraud.domain.ValidationResult; +import com.pe.yape.service.challenge.antifraud.util.ReasonCodes; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import reactor.core.publisher.Mono; + +@Service +@RequiredArgsConstructor +public class AntifraudService { + + private final AntifraudProperties antifraudProperties; + + public Mono validate(TransactionCreatedEvent event) { + return Mono.fromCallable(() -> { + + ValidationResult.ValidationResultBuilder validationResultBuilder = ValidationResult.builder(); + + if (Double.compare(event.getValue(), 0.0) == 0) { + return validationResultBuilder.transactionStatus(TransactionStatus.REJECTED) + .reasonCode(ReasonCodes.MISSING_VALUE) + .build(); + } + if (event.getValue() > Integer.parseInt(antifraudProperties.getAmountLimit())) { + return validationResultBuilder.transactionStatus(TransactionStatus.REJECTED) + .reasonCode(ReasonCodes.VALUE_EXCEEDS_LIMIT) + .build(); + } + return validationResultBuilder.transactionStatus(TransactionStatus.APPROVED) + .build(); + }); + + } + + +} diff --git a/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/util/ReasonCodes.java b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/util/ReasonCodes.java new file mode 100644 index 0000000000..7bd5062207 --- /dev/null +++ b/app-anti-fraud-service/src/main/java/com/pe/yape/service/challenge/antifraud/util/ReasonCodes.java @@ -0,0 +1,8 @@ +package com.pe.yape.service.challenge.antifraud.util; + +public final class ReasonCodes { + public static final String VALUE_EXCEEDS_LIMIT = "VALUE_EXCEEDS_LIMIT"; + public static final String MISSING_VALUE = "MISSING_VALUE"; + + private ReasonCodes() {} +} diff --git a/app-anti-fraud-service/src/main/resources/application.yml b/app-anti-fraud-service/src/main/resources/application.yml new file mode 100644 index 0000000000..92a47e2095 --- /dev/null +++ b/app-anti-fraud-service/src/main/resources/application.yml @@ -0,0 +1,29 @@ +--- +server: + port: 8082 +spring: + kafka: + bootstrap-servers: localhost:9092 + topics: + transactions-created: transactions.created + transactions-validated: transactions.validated + schema-registry-url: http://localhost:8081 +application: + rules: + amount-limit: "1000" +logging: + level: + root: INFO + org.springframework.web: DEBUG + org.springframework.r2dbc: DEBUG + org.springframework.kafka: INFO +management: + endpoints: + web: + exposure: + include: health,info,metrics + health: + db: + enabled: true + kafka: + enabled: true diff --git a/app-transaction-report-service/.gitattributes b/app-transaction-report-service/.gitattributes new file mode 100644 index 0000000000..3b41682ac5 --- /dev/null +++ b/app-transaction-report-service/.gitattributes @@ -0,0 +1,2 @@ +/mvnw text eol=lf +*.cmd text eol=crlf diff --git a/app-transaction-report-service/.gitignore b/app-transaction-report-service/.gitignore new file mode 100644 index 0000000000..667aaef0c8 --- /dev/null +++ b/app-transaction-report-service/.gitignore @@ -0,0 +1,33 @@ +HELP.md +target/ +.mvn/wrapper/maven-wrapper.jar +!**/src/main/**/target/ +!**/src/test/**/target/ + +### STS ### +.apt_generated +.classpath +.factorypath +.project +.settings +.springBeans +.sts4-cache + +### IntelliJ IDEA ### +.idea +*.iws +*.iml +*.ipr + +### NetBeans ### +/nbproject/private/ +/nbbuild/ +/dist/ +/nbdist/ +/.nb-gradle/ +build/ +!**/src/main/**/build/ +!**/src/test/**/build/ + +### VS Code ### +.vscode/ diff --git a/app-transaction-report-service/pom.xml b/app-transaction-report-service/pom.xml new file mode 100644 index 0000000000..d1dfe2dcf8 --- /dev/null +++ b/app-transaction-report-service/pom.xml @@ -0,0 +1,122 @@ + + + 4.0.0 + + org.springframework.boot + spring-boot-starter-parent + 3.3.5 + + + + com.pe.yape.service.challenge + app-transaction-report-service + 0.0.1-SNAPSHOT + Transaction Report service application + + + + 17 + + + + + org.springframework.boot + spring-boot-starter-data-r2dbc + + + + org.springframework.boot + spring-boot-starter-webflux + + + + org.postgresql + postgresql + runtime + 42.7.8 + + + + org.postgresql + r2dbc-postgresql + runtime + 1.0.9.RELEASE + + + + org.projectlombok + lombok + + + + org.springframework.boot + spring-boot-starter-test + test + + + + io.projectreactor + reactor-test + test + + + + jakarta.validation + jakarta.validation-api + 3.0.2 + + + + org.hibernate.validator + hibernate-validator + 8.0.1.Final + + + + org.mapstruct + mapstruct + 1.5.5.Final + + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + + + org.projectlombok + lombok + + + org.springframework.boot + spring-boot-configuration-processor + + + org.mapstruct + mapstruct-processor + 1.5.5.Final + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + org.projectlombok + lombok + + + + + + + + diff --git a/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/AppTransactionReportServiceApplication.java b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/AppTransactionReportServiceApplication.java new file mode 100644 index 0000000000..717feceaa3 --- /dev/null +++ b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/AppTransactionReportServiceApplication.java @@ -0,0 +1,13 @@ +package com.pe.yape.service.challenge.transactions; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class AppTransactionReportServiceApplication { + + public static void main(String[] args) { + SpringApplication.run(AppTransactionReportServiceApplication.class, args); + } + +} diff --git a/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/config/AppConfig.java b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/config/AppConfig.java new file mode 100644 index 0000000000..b5d118d912 --- /dev/null +++ b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/config/AppConfig.java @@ -0,0 +1,20 @@ +package com.pe.yape.service.challenge.transactions.config; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter; + +import java.nio.charset.StandardCharsets; + +@Configuration +public class AppConfig { + + @Bean + public MappingJackson2HttpMessageConverter mappingJackson2HttpMessageConverter() { + MappingJackson2HttpMessageConverter jsonConverter = new MappingJackson2HttpMessageConverter(); + jsonConverter.setDefaultCharset(StandardCharsets.UTF_8); + return jsonConverter; + } + + +} diff --git a/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/controller/TransactionReportController.java b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/controller/TransactionReportController.java new file mode 100644 index 0000000000..d910e7b53c --- /dev/null +++ b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/controller/TransactionReportController.java @@ -0,0 +1,28 @@ +package com.pe.yape.service.challenge.transactions.controller; + +import com.pe.yape.service.challenge.transactions.dto.TransactionResponse; +import com.pe.yape.service.challenge.transactions.service.SearchTransactionService; +import lombok.extern.slf4j.Slf4j; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.*; +import reactor.core.publisher.Mono; + +@CrossOrigin(allowedHeaders = "*") +@RestController +@RequestMapping("/transactions") +@Slf4j +public class TransactionReportController { + + private final SearchTransactionService searchTransactionService; + + public TransactionReportController(SearchTransactionService searchTransactionService) { + this.searchTransactionService = searchTransactionService; + } + + @GetMapping(value = "/{transactionId}", produces = MediaType.APPLICATION_JSON_VALUE) + public Mono> retrieve(@PathVariable String transactionId) { + return searchTransactionService.search(transactionId) + .map(ResponseEntity::ok); + } +} diff --git a/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/domain/model/Transaction.java b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/domain/model/Transaction.java new file mode 100644 index 0000000000..466a186b08 --- /dev/null +++ b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/domain/model/Transaction.java @@ -0,0 +1,29 @@ +package com.pe.yape.service.challenge.transactions.domain.model; + +import lombok.*; +import org.springframework.data.annotation.Id; +import org.springframework.data.relational.core.mapping.Table; +import java.time.OffsetDateTime; +import java.util.UUID; + + +@Table(value = "transaction" , schema = "anti_fraud") +@Getter +@Setter +@Builder +public class Transaction { + + @Id + private UUID transactionExternalId; + private UUID accountExternalIdDebit; + private UUID accountExternalIdCredit; + private Integer transferTypeId; + private Double value; + private String status; + private OffsetDateTime createdAt; + private OffsetDateTime updatedAt; + + + + +} diff --git a/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionResponse.java b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionResponse.java new file mode 100644 index 0000000000..e6b2587b00 --- /dev/null +++ b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionResponse.java @@ -0,0 +1,15 @@ +package com.pe.yape.service.challenge.transactions.dto; + +import java.math.BigDecimal; +import java.time.LocalDateTime; + +public record TransactionResponse( + String transactionExternalId, + TransactionType transactionType, + TransactionStatus transactionStatus, + BigDecimal value, + LocalDateTime createdAt + ) {} + + + diff --git a/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionStatus.java b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionStatus.java new file mode 100644 index 0000000000..d163ba8adf --- /dev/null +++ b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionStatus.java @@ -0,0 +1,4 @@ +package com.pe.yape.service.challenge.transactions.dto; + +public record TransactionStatus(String name) { +} \ No newline at end of file diff --git a/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionType.java b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionType.java new file mode 100644 index 0000000000..56bab80c36 --- /dev/null +++ b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionType.java @@ -0,0 +1,5 @@ +package com.pe.yape.service.challenge.transactions.dto; + +public record TransactionType(String name) { +} + diff --git a/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/exception/handler/ApiException.java b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/exception/handler/ApiException.java new file mode 100644 index 0000000000..cdb61f8200 --- /dev/null +++ b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/exception/handler/ApiException.java @@ -0,0 +1,14 @@ +package com.pe.yape.service.challenge.transactions.exception.handler; + +import lombok.Getter; + +@Getter +public class ApiException extends RuntimeException { + + private final Integer code; + + public ApiException(Integer code, String message){ + super(message); + this.code = code; + } +} diff --git a/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/exception/handler/CustomErrorAttributes.java b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/exception/handler/CustomErrorAttributes.java new file mode 100644 index 0000000000..0df866a63b --- /dev/null +++ b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/exception/handler/CustomErrorAttributes.java @@ -0,0 +1,60 @@ +package com.pe.yape.service.challenge.transactions.exception.handler; + +import org.springframework.boot.web.error.ErrorAttributeOptions; +import org.springframework.boot.web.reactive.error.DefaultErrorAttributes; +import org.springframework.stereotype.Component; +import org.springframework.web.bind.support.WebExchangeBindException; +import org.springframework.web.reactive.function.server.ServerRequest; +import org.springframework.web.server.ServerWebInputException; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +@Component +public class CustomErrorAttributes extends DefaultErrorAttributes { + + @Override + public Map getErrorAttributes(ServerRequest request, ErrorAttributeOptions options) { + Throwable error = getError(request); + + Map body = new LinkedHashMap<>(); + body.put("path", request.path()); + body.put("status", getStatus(error)); + body.put("message", buildMessage(error)); + body.put("errors", extractValidationErrors(error)); + + return body; + } + + private int getStatus(Throwable error) { + if (error instanceof ServerWebInputException ex) { + return ex.getStatusCode().value(); + } + else if (error instanceof ApiException) { + return ((ApiException) error).getCode(); + } + return 500; + } + + private String buildMessage(Throwable error) { + if (error instanceof ServerWebInputException) { + return "Existen errores de validación"; + } + else if (error instanceof ApiException) { + return error.getMessage(); + } + return error.getMessage(); + } + + private List> extractValidationErrors(Throwable error) { + if (error instanceof WebExchangeBindException ex) { + return ex.getFieldErrors().stream() + .map(er -> Map.of( + "field", er.getField(), + "message", er.getDefaultMessage() + )).toList(); + } + return null; + } +} diff --git a/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/exception/handler/GlobalErrorHandler.java b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/exception/handler/GlobalErrorHandler.java new file mode 100644 index 0000000000..56b294dd81 --- /dev/null +++ b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/exception/handler/GlobalErrorHandler.java @@ -0,0 +1,44 @@ +package com.pe.yape.service.challenge.transactions.exception.handler; + +import org.springframework.boot.autoconfigure.web.WebProperties; +import org.springframework.boot.autoconfigure.web.reactive.error.AbstractErrorWebExceptionHandler; +import org.springframework.boot.web.error.ErrorAttributeOptions; +import org.springframework.boot.web.reactive.error.ErrorAttributes; +import org.springframework.context.ApplicationContext; +import org.springframework.core.annotation.Order; +import org.springframework.http.MediaType; +import org.springframework.http.codec.ServerCodecConfigurer; +import org.springframework.stereotype.Component; +import org.springframework.web.reactive.function.server.*; +import reactor.core.publisher.Mono; + +import java.util.Map; + +@Component +@Order(-2) +public class GlobalErrorHandler extends AbstractErrorWebExceptionHandler { + + public GlobalErrorHandler(CustomErrorAttributes attributes, + ApplicationContext context, + ServerCodecConfigurer codecConfigurer) { + super(attributes, new WebProperties.Resources(), context); + super.setMessageWriters(codecConfigurer.getWriters()); + super.setMessageReaders(codecConfigurer.getReaders()); + } + + @Override + protected RouterFunction getRoutingFunction(ErrorAttributes errorAttributes) { + return RouterFunctions.route(RequestPredicates.all(), this::renderErrorResponse); + } + + private Mono renderErrorResponse(ServerRequest request) { + Map error = getErrorAttributes(request, ErrorAttributeOptions.defaults()); + + int status = (int) error.get("status"); + + return ServerResponse + .status(status) + .contentType(MediaType.APPLICATION_JSON) + .bodyValue(error); + } +} diff --git a/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/mapper/TransactionMapper.java b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/mapper/TransactionMapper.java new file mode 100644 index 0000000000..ed17a5e220 --- /dev/null +++ b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/mapper/TransactionMapper.java @@ -0,0 +1,42 @@ +package com.pe.yape.service.challenge.transactions.mapper; + +import com.pe.yape.service.challenge.transactions.domain.model.Transaction; +import com.pe.yape.service.challenge.transactions.dto.TransactionResponse; +import com.pe.yape.service.challenge.transactions.dto.TransactionStatus; +import com.pe.yape.service.challenge.transactions.dto.TransactionType; +import org.mapstruct.Mapper; +import org.mapstruct.Mapping; +import java.time.LocalDateTime; +import java.time.OffsetDateTime; + +@Mapper(componentModel = "spring") +public interface TransactionMapper { + + @Mapping(target = "transactionExternalId", expression = "java(transaction.getTransactionExternalId().toString())") + @Mapping(target = "transactionType", expression = "java(mapType(transaction.getTransferTypeId()))") + @Mapping(target = "transactionStatus", expression = "java(mapStatus(transaction.getStatus()))") + @Mapping(target = "value", expression = "java(BigDecimal.valueOf(transaction.getValue()))") + @Mapping(target = "createdAt", expression = "java(toLocalDateTime(transaction.getCreatedAt()))") + TransactionResponse toResponse(Transaction transaction); + + default LocalDateTime toLocalDateTime(OffsetDateTime odt) { + return odt == null ? null : odt.toLocalDateTime(); + } + + default TransactionType mapType(Integer id) { + return new TransactionType(resolveTypeName(id)); + } + + default TransactionStatus mapStatus(String status) { + return new TransactionStatus(status); + } + + private static String resolveTypeName(Integer id) { + return switch (id) { + case 1 -> "DEPOSIT"; + case 2 -> "TRANSFER"; + case 3 -> "PAYMENT"; + default -> "UNKNOWN"; + }; + } +} diff --git a/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/repository/TransactionRepository.java b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/repository/TransactionRepository.java new file mode 100644 index 0000000000..ef8525bd71 --- /dev/null +++ b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/repository/TransactionRepository.java @@ -0,0 +1,11 @@ +package com.pe.yape.service.challenge.transactions.repository; + +import com.pe.yape.service.challenge.transactions.domain.model.Transaction; +import org.springframework.data.repository.reactive.ReactiveCrudRepository; +import org.springframework.stereotype.Repository; + +import java.util.UUID; + +@Repository +public interface TransactionRepository extends ReactiveCrudRepository { +} \ No newline at end of file diff --git a/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/service/SearchTransactionService.java b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/service/SearchTransactionService.java new file mode 100644 index 0000000000..eb74046035 --- /dev/null +++ b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/service/SearchTransactionService.java @@ -0,0 +1,15 @@ +package com.pe.yape.service.challenge.transactions.service; + +import com.pe.yape.service.challenge.transactions.dto.TransactionResponse; +import reactor.core.publisher.Mono; + +public interface SearchTransactionService { + + /** + * + * @param transactionId + * @return + */ + Mono search(String transactionId); + +} diff --git a/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/service/impl/SearchTransactionServiceImpl.java b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/service/impl/SearchTransactionServiceImpl.java new file mode 100644 index 0000000000..bbbad8bbc2 --- /dev/null +++ b/app-transaction-report-service/src/main/java/com/pe/yape/service/challenge/transactions/service/impl/SearchTransactionServiceImpl.java @@ -0,0 +1,34 @@ +package com.pe.yape.service.challenge.transactions.service.impl; + +import com.pe.yape.service.challenge.transactions.dto.TransactionResponse; +import com.pe.yape.service.challenge.transactions.exception.handler.ApiException; +import com.pe.yape.service.challenge.transactions.mapper.TransactionMapper; +import com.pe.yape.service.challenge.transactions.repository.TransactionRepository; +import com.pe.yape.service.challenge.transactions.service.SearchTransactionService; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import reactor.core.publisher.Mono; + +import java.util.UUID; + +@Slf4j +@Service +public class SearchTransactionServiceImpl implements SearchTransactionService { + + private final TransactionRepository transactionRepository; + + private final TransactionMapper transactionMapper; + + public SearchTransactionServiceImpl(TransactionRepository transactionRepository, + TransactionMapper transactionMapper) { + this.transactionRepository = transactionRepository; + this.transactionMapper = transactionMapper; + } + + @Override + public Mono search(String transactionId) { + return transactionRepository.findById(UUID.fromString(transactionId)) + .switchIfEmpty(Mono.error(new ApiException(404,String.format("Recurso %s no existe", transactionId)))) + .map(transactionMapper::toResponse); + } +} diff --git a/app-transaction-report-service/src/main/resources/application.yml b/app-transaction-report-service/src/main/resources/application.yml new file mode 100644 index 0000000000..94f5711758 --- /dev/null +++ b/app-transaction-report-service/src/main/resources/application.yml @@ -0,0 +1,28 @@ +--- +server: + port: 8085 + servlet: + context-path: /api/transactions-report +spring: + application: + name: transaction-report-service + webflux: + base-path: /api/transactions-report + r2dbc: + url: r2dbc:postgresql://localhost:5432/postgres + username: postgres + password: postgres +logging: + level: + root: INFO + org.springframework.web: DEBUG + org.springframework.r2dbc: DEBUG + org.springframework.kafka: INFO +management: + endpoints: + web: + exposure: + include: health,info,metrics + health: + db: + enabled: true diff --git a/app-transaction-service/.gitattributes b/app-transaction-service/.gitattributes new file mode 100644 index 0000000000..3b41682ac5 --- /dev/null +++ b/app-transaction-service/.gitattributes @@ -0,0 +1,2 @@ +/mvnw text eol=lf +*.cmd text eol=crlf diff --git a/app-transaction-service/.gitignore b/app-transaction-service/.gitignore new file mode 100644 index 0000000000..667aaef0c8 --- /dev/null +++ b/app-transaction-service/.gitignore @@ -0,0 +1,33 @@ +HELP.md +target/ +.mvn/wrapper/maven-wrapper.jar +!**/src/main/**/target/ +!**/src/test/**/target/ + +### STS ### +.apt_generated +.classpath +.factorypath +.project +.settings +.springBeans +.sts4-cache + +### IntelliJ IDEA ### +.idea +*.iws +*.iml +*.ipr + +### NetBeans ### +/nbproject/private/ +/nbbuild/ +/dist/ +/nbdist/ +/.nb-gradle/ +build/ +!**/src/main/**/build/ +!**/src/test/**/build/ + +### VS Code ### +.vscode/ diff --git a/app-transaction-service/pom.xml b/app-transaction-service/pom.xml new file mode 100644 index 0000000000..8bf16fbf34 --- /dev/null +++ b/app-transaction-service/pom.xml @@ -0,0 +1,179 @@ + + + 4.0.0 + + org.springframework.boot + spring-boot-starter-parent + 3.4.12 + + + + com.pe.yape.service.challenge + app-transaction-service + 0.0.1-SNAPSHOT + app-transaction-service + Transaction service application + + + + 17 + + + + + confluent + https://packages.confluent.io/maven/ + + + + + + org.springframework.boot + spring-boot-starter-data-r2dbc + + + + org.springframework.boot + spring-boot-starter-webflux + + + + org.postgresql + postgresql + runtime + 42.7.8 + + + + org.postgresql + r2dbc-postgresql + runtime + 1.0.9.RELEASE + + + + org.projectlombok + lombok + + + + org.springframework.boot + spring-boot-starter-test + test + + + + io.projectreactor + reactor-test + test + + + + jakarta.validation + jakarta.validation-api + 3.0.2 + + + + org.hibernate.validator + hibernate-validator + 8.0.1.Final + + + + org.flywaydb + flyway-core + 11.18.0 + + + + org.flywaydb + flyway-database-postgresql + + + + + io.projectreactor.kafka + reactor-kafka + 1.3.22 + + + + + org.apache.avro + avro + 1.11.4 + + + + io.confluent + kafka-avro-serializer + 7.5.0 + + + + io.confluent + kafka-schema-registry-client + 7.5.0 + + + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + + + org.projectlombok + lombok + + + org.springframework.boot + spring-boot-configuration-processor + + + org.mapstruct + mapstruct-processor + 1.5.5.Final + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + org.projectlombok + lombok + + + + + + org.apache.avro + avro-maven-plugin + 1.11.4 + + + generate-sources + + schema + + + ${project.basedir}/src/main/avro + ${project.build.directory}/generated-sources/avro + + + + + + + + diff --git a/app-transaction-service/src/main/avro/TransactionCreatedEvent.avsc b/app-transaction-service/src/main/avro/TransactionCreatedEvent.avsc new file mode 100644 index 0000000000..76615a0f7c --- /dev/null +++ b/app-transaction-service/src/main/avro/TransactionCreatedEvent.avsc @@ -0,0 +1,12 @@ +{ + "type": "record", + "name": "TransactionCreatedEvent", + "namespace": "com.pe.yape.service.challenge.transactions.avro", + "fields": [ + {"name": "transactionExternalId", "type": "string"}, + {"name": "accountExternalIdDebit", "type": "string"}, + {"name": "accountExternalIdCredit", "type": "string"}, + {"name": "transferTypeId", "type": "int"}, + {"name": "value", "type": "double"} + ] +} diff --git a/app-transaction-service/src/main/avro/TransactionValidatedEvent.avsc b/app-transaction-service/src/main/avro/TransactionValidatedEvent.avsc new file mode 100644 index 0000000000..36cde2e1e6 --- /dev/null +++ b/app-transaction-service/src/main/avro/TransactionValidatedEvent.avsc @@ -0,0 +1,11 @@ +{ + "type": "record", + "name": "TransactionValidatedEvent", + "namespace": "com.pe.yape.service.challenge.transactions.avro", + "fields": [ + {"name": "transactionExternalId", "type": "string"}, + {"name": "status", "type": "string"}, + {"name": "reasonCode", "type": ["null","string"], "default": null}, + {"name": "validatedAt", "type": "string"} + ] +} diff --git a/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/AppTransactionServiceApplication.java b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/AppTransactionServiceApplication.java new file mode 100644 index 0000000000..7095cfc532 --- /dev/null +++ b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/AppTransactionServiceApplication.java @@ -0,0 +1,16 @@ +package com.pe.yape.service.challenge.transactions; + +import com.pe.yape.service.challenge.transactions.config.KafkaReactiveConfig; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.Import; + +@SpringBootApplication +@Import(KafkaReactiveConfig.class) +public class AppTransactionServiceApplication { + + public static void main(String[] args) { + SpringApplication.run(AppTransactionServiceApplication.class, args); + } + +} diff --git a/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/config/AppConfig.java b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/config/AppConfig.java new file mode 100644 index 0000000000..b5d118d912 --- /dev/null +++ b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/config/AppConfig.java @@ -0,0 +1,20 @@ +package com.pe.yape.service.challenge.transactions.config; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter; + +import java.nio.charset.StandardCharsets; + +@Configuration +public class AppConfig { + + @Bean + public MappingJackson2HttpMessageConverter mappingJackson2HttpMessageConverter() { + MappingJackson2HttpMessageConverter jsonConverter = new MappingJackson2HttpMessageConverter(); + jsonConverter.setDefaultCharset(StandardCharsets.UTF_8); + return jsonConverter; + } + + +} diff --git a/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/config/KafkaReactiveConfig.java b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/config/KafkaReactiveConfig.java new file mode 100644 index 0000000000..fc22215f73 --- /dev/null +++ b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/config/KafkaReactiveConfig.java @@ -0,0 +1,64 @@ +package com.pe.yape.service.challenge.transactions.config; + +import com.pe.yape.service.challenge.transactions.avro.TransactionCreatedEvent; +import com.pe.yape.service.challenge.transactions.avro.TransactionValidatedEvent; +import io.confluent.kafka.serializers.KafkaAvroDeserializer; +import io.confluent.kafka.serializers.KafkaAvroSerializer; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import reactor.kafka.receiver.ReceiverOptions; +import reactor.kafka.sender.SenderOptions; + +import java.util.Map; + +@Configuration +public class KafkaReactiveConfig { + + public static final String TOPIC_TRANSACTIONS_CREATED = "transactions.created"; + public static final String TOPIC_TRANSACTIONS_VALIDATED = "transactions.validated"; + + @Bean + public NewTopic transactionsCreatedTopic() { + return new NewTopic(TOPIC_TRANSACTIONS_CREATED, 3, (short) 1); + } + + @Bean + public NewTopic transactionsValidatedTopic() { + return new NewTopic(TOPIC_TRANSACTIONS_VALIDATED, 3, (short) 1); + } + + @Bean + public ReceiverOptions receiverOptions() { + Map consumerProperties = Map.of( + ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092", + ConsumerConfig.GROUP_ID_CONFIG, "anti-fraud-service-group", + ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest", + "schema.registry.url", "http://localhost:8081", + ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class, + ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class + ); + + return ReceiverOptions.create(consumerProperties) + .subscription(java.util.List.of(TOPIC_TRANSACTIONS_VALIDATED)); + } + + @Bean + public SenderOptions senderOptions(){ + Map props = Map.of( + ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092", + ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class, + ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class, + "schema.registry.url", "http://localhost:8081" + ); + + return SenderOptions.create(props); + } + + + +} diff --git a/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/controller/TransactionController.java b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/controller/TransactionController.java new file mode 100644 index 0000000000..be69175fae --- /dev/null +++ b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/controller/TransactionController.java @@ -0,0 +1,31 @@ +package com.pe.yape.service.challenge.transactions.controller; + +import com.pe.yape.service.challenge.transactions.dto.TransactionRequest; +import com.pe.yape.service.challenge.transactions.dto.TransactionResponse; +import com.pe.yape.service.challenge.transactions.service.TransactionService; +import jakarta.validation.Valid; +import lombok.extern.slf4j.Slf4j; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.*; +import reactor.core.publisher.Mono; + +@CrossOrigin(allowedHeaders = "*") +@RestController +@RequestMapping("/transactions") +@Slf4j +public class TransactionController { + + private final TransactionService transactionService; + + public TransactionController(TransactionService transactionService) { + this.transactionService = transactionService; + } + + @PostMapping(value = "/", produces = MediaType.APPLICATION_JSON_VALUE) + public Mono> create(@Valid @RequestBody TransactionRequest request) { + return transactionService.create(request). + map(ResponseEntity::ok) + .doOnError(throwable -> log.info("Error : {}" , throwable.getMessage())); + } +} diff --git a/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/domain/model/Transaction.java b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/domain/model/Transaction.java new file mode 100644 index 0000000000..466a186b08 --- /dev/null +++ b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/domain/model/Transaction.java @@ -0,0 +1,29 @@ +package com.pe.yape.service.challenge.transactions.domain.model; + +import lombok.*; +import org.springframework.data.annotation.Id; +import org.springframework.data.relational.core.mapping.Table; +import java.time.OffsetDateTime; +import java.util.UUID; + + +@Table(value = "transaction" , schema = "anti_fraud") +@Getter +@Setter +@Builder +public class Transaction { + + @Id + private UUID transactionExternalId; + private UUID accountExternalIdDebit; + private UUID accountExternalIdCredit; + private Integer transferTypeId; + private Double value; + private String status; + private OffsetDateTime createdAt; + private OffsetDateTime updatedAt; + + + + +} diff --git a/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/domain/model/TransactionStatus.java b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/domain/model/TransactionStatus.java new file mode 100644 index 0000000000..f334e12db1 --- /dev/null +++ b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/domain/model/TransactionStatus.java @@ -0,0 +1,10 @@ +package com.pe.yape.service.challenge.transactions.domain.model; + +public enum TransactionStatus { + + PENDING, + + APPROVED, + + REJECTED +} diff --git a/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionRequest.java b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionRequest.java new file mode 100644 index 0000000000..30901ef148 --- /dev/null +++ b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionRequest.java @@ -0,0 +1,31 @@ +package com.pe.yape.service.challenge.transactions.dto; + +import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.NotNull; +import jakarta.validation.constraints.Pattern; +import jakarta.validation.constraints.Positive; +import lombok.Getter; +import lombok.Setter; + +@Getter +@Setter +public class TransactionRequest { + + @NotBlank(message = "Attribute accountExternalIdDebit is required") + @Pattern(regexp = "^[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[1-5][a-fA-F0-9]{3}-[89abAB][a-fA-F0-9]{3}-[a-fA-F0-9]{12}$" , + message = "Attribute accountExternalIdDebit is invalid") + private String accountExternalIdDebit; + + @NotBlank(message = "Attribute accountExternalIdCredit is required") + @Pattern(regexp = "^[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[1-5][a-fA-F0-9]{3}-[89abAB][a-fA-F0-9]{3}-[a-fA-F0-9]{12}$" , + message = "Attribute accountExternalIdDebit is invalid") + private String accountExternalIdCredit; + + @NotNull(message = "Attribute transferTypeId is required") + private Integer transferTypeId; + + @NotNull(message = "Attribute value is required") + @Positive(message = "Attribute value is invalid") + private Double value; + +} diff --git a/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionResponse.java b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionResponse.java new file mode 100644 index 0000000000..e47487230a --- /dev/null +++ b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionResponse.java @@ -0,0 +1,14 @@ +package com.pe.yape.service.challenge.transactions.dto; + +import lombok.*; + +@Getter +@Setter +@AllArgsConstructor +@NoArgsConstructor +@Builder +public class TransactionResponse { + + private String transactionExternalId; + +} diff --git a/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionUpdateRequest.java b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionUpdateRequest.java new file mode 100644 index 0000000000..393d22f625 --- /dev/null +++ b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/dto/TransactionUpdateRequest.java @@ -0,0 +1,27 @@ +package com.pe.yape.service.challenge.transactions.dto; + +import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.NotNull; +import jakarta.validation.constraints.Pattern; +import jakarta.validation.constraints.Positive; + +public class TransactionUpdateRequest { + + @NotBlank(message = "Attribute accountExternalIdDebit is required") + @Pattern(regexp = "^[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[1-5][a-fA-F0-9]{3}-[89abAB][a-fA-F0-9]{3}-[a-fA-F0-9]{12}$" , + message = "Attribute accountExternalIdDebit is invalid") + private String accountExternalIdDebit; + + @NotBlank(message = "Attribute accountExternalIdCredit is required") + @Pattern(regexp = "^[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[1-5][a-fA-F0-9]{3}-[89abAB][a-fA-F0-9]{3}-[a-fA-F0-9]{12}$" , + message = "Attribute accountExternalIdDebit is invalid") + private String accountExternalIdCredit; + + @NotNull(message = "Attribute transferTypeId is required") + private Integer transferTypeId; + + @NotNull(message = "Attribute value is required") + @Positive(message = "Attribute value is invalid") + private Double value; + +} diff --git a/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/exception/handler/CustomErrorAttributes.java b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/exception/handler/CustomErrorAttributes.java new file mode 100644 index 0000000000..9b5566b2ac --- /dev/null +++ b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/exception/handler/CustomErrorAttributes.java @@ -0,0 +1,60 @@ +package com.pe.yape.service.challenge.transactions.exception.handler; + +import org.springframework.boot.web.error.ErrorAttributeOptions; +import org.springframework.boot.web.reactive.error.DefaultErrorAttributes; +import org.springframework.stereotype.Component; +import org.springframework.web.bind.support.WebExchangeBindException; +import org.springframework.web.reactive.function.server.ServerRequest; +import org.springframework.web.server.ResponseStatusException; +import org.springframework.web.server.ServerWebInputException; + +import java.time.Instant; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +@Component +public class CustomErrorAttributes extends DefaultErrorAttributes { + + @Override + public Map getErrorAttributes(ServerRequest request, ErrorAttributeOptions options) { + Throwable error = getError(request); + + Map body = new LinkedHashMap<>(); + body.put("timestamp", Instant.now()); + body.put("path", request.path()); + body.put("status", getStatus(error)); + body.put("message", buildMessage(error)); + body.put("errors", extractValidationErrors(error)); + + return body; + } + + private int getStatus(Throwable error) { + if (error instanceof ServerWebInputException ex) { + return ex.getStatusCode().value(); + } + if (error instanceof WebExchangeBindException) { + return 400; + } + return 500; + } + + private String buildMessage(Throwable error) { + if (error instanceof ServerWebInputException) { + return "Existen errores de validación"; + } + return error.getMessage(); + } + + private List> extractValidationErrors(Throwable error) { + if (error instanceof WebExchangeBindException ex) { + return ex.getFieldErrors().stream() + .map(er -> Map.of( + "field", er.getField(), + "message", er.getDefaultMessage() + )).toList(); + } + return null; + } +} diff --git a/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/exception/handler/GlobalErrorHandler.java b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/exception/handler/GlobalErrorHandler.java new file mode 100644 index 0000000000..56b294dd81 --- /dev/null +++ b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/exception/handler/GlobalErrorHandler.java @@ -0,0 +1,44 @@ +package com.pe.yape.service.challenge.transactions.exception.handler; + +import org.springframework.boot.autoconfigure.web.WebProperties; +import org.springframework.boot.autoconfigure.web.reactive.error.AbstractErrorWebExceptionHandler; +import org.springframework.boot.web.error.ErrorAttributeOptions; +import org.springframework.boot.web.reactive.error.ErrorAttributes; +import org.springframework.context.ApplicationContext; +import org.springframework.core.annotation.Order; +import org.springframework.http.MediaType; +import org.springframework.http.codec.ServerCodecConfigurer; +import org.springframework.stereotype.Component; +import org.springframework.web.reactive.function.server.*; +import reactor.core.publisher.Mono; + +import java.util.Map; + +@Component +@Order(-2) +public class GlobalErrorHandler extends AbstractErrorWebExceptionHandler { + + public GlobalErrorHandler(CustomErrorAttributes attributes, + ApplicationContext context, + ServerCodecConfigurer codecConfigurer) { + super(attributes, new WebProperties.Resources(), context); + super.setMessageWriters(codecConfigurer.getWriters()); + super.setMessageReaders(codecConfigurer.getReaders()); + } + + @Override + protected RouterFunction getRoutingFunction(ErrorAttributes errorAttributes) { + return RouterFunctions.route(RequestPredicates.all(), this::renderErrorResponse); + } + + private Mono renderErrorResponse(ServerRequest request) { + Map error = getErrorAttributes(request, ErrorAttributeOptions.defaults()); + + int status = (int) error.get("status"); + + return ServerResponse + .status(status) + .contentType(MediaType.APPLICATION_JSON) + .bodyValue(error); + } +} diff --git a/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/messaging/TransactionCreatedProducer.java b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/messaging/TransactionCreatedProducer.java new file mode 100644 index 0000000000..a5424dd776 --- /dev/null +++ b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/messaging/TransactionCreatedProducer.java @@ -0,0 +1,48 @@ +package com.pe.yape.service.challenge.transactions.messaging; + + +import com.pe.yape.service.challenge.transactions.avro.TransactionCreatedEvent; +import com.pe.yape.service.challenge.transactions.domain.model.Transaction; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import reactor.core.publisher.Mono; +import reactor.kafka.sender.KafkaSender; +import reactor.kafka.sender.SenderOptions; +import reactor.kafka.sender.SenderRecord; +import reactor.kafka.sender.SenderResult; + +import static com.pe.yape.service.challenge.transactions.config.KafkaReactiveConfig.TOPIC_TRANSACTIONS_CREATED; + +@Component +@Slf4j +public class TransactionCreatedProducer { + + private final KafkaSender sender; + + public TransactionCreatedProducer(SenderOptions senderOptions) { + this.sender = KafkaSender.create(senderOptions); + } + + public Mono publish(Transaction transaction) { + + TransactionCreatedEvent transactionCreatedEvent = TransactionCreatedEvent.newBuilder() + .setTransactionExternalId(transaction.getTransactionExternalId().toString()) + .setAccountExternalIdCredit(transaction.getAccountExternalIdCredit().toString()) + .setAccountExternalIdDebit(transaction.getAccountExternalIdDebit().toString()) + .setValue(transaction.getValue()) + .setTransferTypeId(transaction.getTransferTypeId()) + .build(); + + SenderRecord record = + SenderRecord.create(TOPIC_TRANSACTIONS_CREATED, null, null, + transactionCreatedEvent.getTransactionExternalId().toString(), + transactionCreatedEvent, + transactionCreatedEvent.getTransactionExternalId().toString()); + + + return sender.send(Mono.just(record)) + .doOnNext(result -> log.info("Send message {} to {}" , transactionCreatedEvent.getTransactionExternalId() , transactionCreatedEvent) ) + .map(SenderResult::correlationMetadata) + .single(); + } +} diff --git a/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/messaging/TransactionValidatedConsumer.java b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/messaging/TransactionValidatedConsumer.java new file mode 100644 index 0000000000..f7bdf4e5be --- /dev/null +++ b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/messaging/TransactionValidatedConsumer.java @@ -0,0 +1,52 @@ +package com.pe.yape.service.challenge.transactions.messaging; + + +import com.pe.yape.service.challenge.transactions.avro.TransactionValidatedEvent; +import com.pe.yape.service.challenge.transactions.service.TransactionService; +import jakarta.annotation.PostConstruct; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import reactor.core.publisher.Flux; +import reactor.kafka.receiver.KafkaReceiver; +import reactor.kafka.receiver.ReceiverOptions; +import reactor.kafka.receiver.ReceiverRecord; + +@Component +@Slf4j +public class TransactionValidatedConsumer { + + private final TransactionService transactionService; + + private final ReceiverOptions receiverOptions; + + public TransactionValidatedConsumer(TransactionService transactionService, + ReceiverOptions receiverOptions) { + this.transactionService = transactionService; + this.receiverOptions = receiverOptions; + } + + @PostConstruct + private void startConsumer() { + + Flux> kafkaFlux = + KafkaReceiver.create(receiverOptions) + .receive(); + + kafkaFlux.flatMap(record -> { + + TransactionValidatedEvent transactionValidatedEvent = record.value(); + + return transactionService.update(transactionValidatedEvent) + .doOnNext(s -> { + log.info("Update transaction {} successfully to {}" , transactionValidatedEvent.getTransactionExternalId() , + transactionValidatedEvent.getStatus()); + record.receiverOffset() + .acknowledge(); + }); + }) .subscribe( + result -> log.info("Processing loop successful for a record."), + error -> log.error("Error during Kafka consumption process: {}", error.getMessage(), error), + () -> log.warn("Kafka consumer flux completed unexpectedly.") + );; + } +} diff --git a/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/repository/TransactionRepository.java b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/repository/TransactionRepository.java new file mode 100644 index 0000000000..ef8525bd71 --- /dev/null +++ b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/repository/TransactionRepository.java @@ -0,0 +1,11 @@ +package com.pe.yape.service.challenge.transactions.repository; + +import com.pe.yape.service.challenge.transactions.domain.model.Transaction; +import org.springframework.data.repository.reactive.ReactiveCrudRepository; +import org.springframework.stereotype.Repository; + +import java.util.UUID; + +@Repository +public interface TransactionRepository extends ReactiveCrudRepository { +} \ No newline at end of file diff --git a/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/service/TransactionService.java b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/service/TransactionService.java new file mode 100644 index 0000000000..54aeef2cdb --- /dev/null +++ b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/service/TransactionService.java @@ -0,0 +1,24 @@ +package com.pe.yape.service.challenge.transactions.service; + +import com.pe.yape.service.challenge.transactions.avro.TransactionValidatedEvent; +import com.pe.yape.service.challenge.transactions.dto.TransactionRequest; +import com.pe.yape.service.challenge.transactions.dto.TransactionResponse; +import reactor.core.publisher.Mono; + +public interface TransactionService { + + /** + * + * @param transactionRequest + * @return + */ + Mono create(TransactionRequest transactionRequest); + + /** + * + * @param transactionValidatedEvent + * @return + */ + Mono update(TransactionValidatedEvent transactionValidatedEvent); + +} diff --git a/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/service/impl/TransactionServiceImpl.java b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/service/impl/TransactionServiceImpl.java new file mode 100644 index 0000000000..fe53a1514f --- /dev/null +++ b/app-transaction-service/src/main/java/com/pe/yape/service/challenge/transactions/service/impl/TransactionServiceImpl.java @@ -0,0 +1,71 @@ +package com.pe.yape.service.challenge.transactions.service.impl; + +import com.pe.yape.service.challenge.transactions.avro.TransactionValidatedEvent; +import com.pe.yape.service.challenge.transactions.domain.model.Transaction; +import com.pe.yape.service.challenge.transactions.domain.model.TransactionStatus; +import com.pe.yape.service.challenge.transactions.dto.TransactionRequest; +import com.pe.yape.service.challenge.transactions.dto.TransactionResponse; +import com.pe.yape.service.challenge.transactions.messaging.TransactionCreatedProducer; +import com.pe.yape.service.challenge.transactions.repository.TransactionRepository; +import com.pe.yape.service.challenge.transactions.service.TransactionService; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import reactor.core.publisher.Mono; + +import java.time.OffsetDateTime; +import java.util.UUID; + +@Slf4j +@Service +public class TransactionServiceImpl implements TransactionService { + + private final TransactionRepository transactionRepository; + private final TransactionCreatedProducer transactionCreatedProducer; + + public TransactionServiceImpl(TransactionRepository transactionRepository, + TransactionCreatedProducer transactionCreatedProducer) { + this.transactionRepository = transactionRepository; + this.transactionCreatedProducer = transactionCreatedProducer; + } + + public Mono create(TransactionRequest request) { + Transaction transaction = Transaction.builder() + .accountExternalIdCredit(UUID.fromString(request.getAccountExternalIdCredit())) + .accountExternalIdDebit(UUID.fromString(request.getAccountExternalIdDebit())) + .transferTypeId(request.getTransferTypeId()) + .value(request.getValue()) + .status(TransactionStatus.PENDING.toString()) + .createdAt(OffsetDateTime.now()) + .build(); + + return transactionRepository.save(transaction) + .flatMap(transactionCreated -> transactionCreatedProducer.publish(transactionCreated) + .thenReturn(transactionCreated)) + .map(saved -> TransactionResponse.builder() + .transactionExternalId(saved.getTransactionExternalId().toString()) + .build() + ) + .doOnSuccess(transactionResponse -> log.info("Transaction created and published: {}", transactionResponse.getTransactionExternalId())) + .doOnError(error -> log.error("Error creating transaction or publishing event", error)); + } + + + @Override + public Mono update(TransactionValidatedEvent event) { + + return transactionRepository.findById(UUID.fromString(event.getTransactionExternalId().toString())) + .switchIfEmpty(Mono.error(new IllegalArgumentException("Transaction not found"))) + .flatMap(existingTransaction -> { + existingTransaction.setStatus(event.getStatus().toString()); + existingTransaction.setUpdatedAt(OffsetDateTime.now()); + return transactionRepository.save(existingTransaction); + }) + .map(updated -> TransactionResponse.builder() + .transactionExternalId(updated.getTransactionExternalId().toString()) + .build() + ) + .doOnSuccess(transactionResponse -> log.info("Transaction {} updated ", transactionResponse.getTransactionExternalId())) + .doOnError(error -> log.error("Error updating transaction or publishing event", error)); + } + +} diff --git a/app-transaction-service/src/main/resources/application.yml b/app-transaction-service/src/main/resources/application.yml new file mode 100644 index 0000000000..e11f3da1d2 --- /dev/null +++ b/app-transaction-service/src/main/resources/application.yml @@ -0,0 +1,48 @@ +--- +server: + port: 8080 + servlet: + context-path: /api +spring: + application: + name: transaction-service + webflux: + base-path: /api + r2dbc: + url: r2dbc:postgresql://localhost:5432/postgres + username: postgres + password: postgres + flyway: + enabled: true + url: jdbc:postgresql://localhost:5432/postgres + user: postgres + password: postgres + ignoreUnsupportedDatabase: true + locations: classpath:db/migration + kafka: + bootstrap-servers: "localhost:9092" + properties: + schema.registry.url: "https://localhost:8081" + consumer: + group-id: anti-fraud-service-group + specific-avro-reader: true + enable-auto-commit: false + auto-offset-reset: latest + producer: + acks: all +logging: + level: + root: INFO + org.springframework.web: DEBUG + org.springframework.r2dbc: DEBUG + org.springframework.kafka: INFO +management: + endpoints: + web: + exposure: + include: health,info,metrics + health: + db: + enabled: true + kafka: + enabled: true diff --git a/app-transaction-service/src/main/resources/db/migration/V1__create_transaction_table.sql b/app-transaction-service/src/main/resources/db/migration/V1__create_transaction_table.sql new file mode 100644 index 0000000000..fb1e23f4ae --- /dev/null +++ b/app-transaction-service/src/main/resources/db/migration/V1__create_transaction_table.sql @@ -0,0 +1,26 @@ +CREATE EXTENSION IF NOT EXISTS "pgcrypto"; + +CREATE SCHEMA "anti_fraud"; + + +CREATE TABLE "anti_fraud".transaction ( + transaction_external_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + account_external_id_debit UUID NOT NULL, + account_external_id_credit UUID NOT NULL, + transfer_type_id INT NOT NULL, + value NUMERIC(18,2) NOT NULL, + status VARCHAR(20) NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NULL +); + +-- Restricción para estados válidos +ALTER TABLE "anti_fraud".transaction +ADD CONSTRAINT chk_transaction_status +CHECK (status IN ('PENDING', 'APPROVED', 'REJECTED')); + +-- Índices recomendados +CREATE INDEX idx_transaction_status ON "anti_fraud".transaction(status); +CREATE INDEX idx_transaction_debit ON "anti_fraud".transaction(account_external_id_debit); +CREATE INDEX idx_transaction_credit ON "anti_fraud".transaction(account_external_id_credit); +CREATE INDEX idx_transaction_created_at ON "anti_fraud".transaction(created_at); diff --git a/docker-compose.yml b/docker-compose.yml index 0e8807f21c..3f937743a6 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,5 +1,10 @@ version: "3.7" services: + redis: + image: redis:7 + container_name: redis + ports: + - "6379:6379" postgres: image: postgres:14 ports: @@ -23,3 +28,15 @@ services: KAFKA_JMX_PORT: 9991 ports: - 9092:9092 + schema-registry: + image: confluentinc/cp-schema-registry:7.6.1 + container_name: schema-registry + depends_on: + - kafka + ports: + - "8081:8081" + environment: + SCHEMA_REGISTRY_HOST_NAME: schema-registry + SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka:29092 + +