Completed Backend Core: Kafka → DB Flow + Incentive API

README.md

@@ -1,2 +1,58 @@
-# Midas
+# MidasFinCore
 Project repo for the JPMC Advanced Software Engineering Forage program
+
+# Midas Core System
+
+## Overview
+
+**Midas Core** is a key component of the larger **Midas System**, a high-profile fintech platform designed to handle financial transactions with speed, reliability, and integrity. This project is part of a distributed system that integrates real-time transaction processing, data validation, and external API incentivization.
+
+The goal of Midas Core is to **receive**, **validate**, and **record** incoming financial transactions. It leverages modern backend technologies such as **Spring Boot**, **Apache Kafka**, **SQL databases**, and **REST APIs** to ensure secure, scalable, and efficient processing.
+
+---
+
+## Tech Stack
+
+- **Java 17**
+- **Spring Boot**
+- **Apache Kafka** – for receiving streaming financial transaction data
+- **PostgreSQL / MySQL** – for transaction validation and persistence
+- **External REST APIs** – for handling transaction incentivization
+- **Maven / Gradle** – for build and dependency management
+
+---
+
+## Architecture
+
+Midas Core operates as a service within a microservices architecture:
+
+1. **Receives transactions** via Kafka topics.
+2. **Validates** each transaction against business rules and database records.
+3. **Persists** the transaction data to a SQL database.
+4. **Communicates** with an external REST API to trigger incentive mechanisms.
+
+All components are wired using **Spring Boot’s dependency injection**, allowing clean modular development and easier testing.
+
+---
+
+## Getting Started
+
+### Prerequisites
+
+- Java 17+
+- Maven or Gradle
+- Docker (optional, for local Kafka or DB setup)
+- PostgreSQL or MySQL running locally or via Docker
+
+### Setup
+
+```bash
+# Clone the repository
+git clone https://github.com/YOUR-USERNAME/midas-core.git
+cd midas-core
+
+# Build the project
+./mvnw clean install
+
+# Run the application
+./mvnw spring-boot:run

application.yml

@@ -0,0 +1,51 @@
+general:
+  kafka-topic: midas-transactions
+
+incentive:
+  api:
+    url: http://localhost:8080/transaction
+
+server:
+  port: 33400
+
+spring:
+  datasource:
+    url: jdbc:h2:mem:midasdb;DB_CLOSE_DELAY=-1
+    driver-class-name: org.h2.Driver
+    username: sa
+    password:
+
+  kafka:
+    consumer:
+      bootstrap-servers: ${spring.embedded.kafka.brokers:localhost:9092}
+      group-id: midas-consumer-group
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
+      properties:
+        spring.json.trusted.packages: "com.jpmc.midascore.entity,com.jpmc.midascore.foundation"
+        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
+      auto-offset-reset: earliest
+    producer:
+      bootstrap-servers: ${spring.embedded.kafka.brokers:localhost:9092}
+      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
+
+  jpa:
+    hibernate:
+      ddl-auto: update
+    show-sql: true
+    properties:
+      hibernate:
+        format_sql: true
+
+  h2:
+    console:
+      enabled: true
+      path: /h2-console
+
+logging:
+  level:
+    org.springframework: INFO
+    com.jpmc.midascore: DEBUG
+    org.hibernate.SQL: DEBUG
+    org.hibernate.type: TRACE

pom.xml

@@ -16,7 +16,52 @@
 	<properties>
 		<java.version>17</java.version>
 	</properties>
+
 	<dependencies>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-actuator</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-jpa</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<version>3.1.4</version>
+		</dependency>
+		<dependency>
+			<groupId>com.h2database</groupId>
+			<artifactId>h2</artifactId>
+			<scope>runtime</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>kafka</artifactId>
+			<version>1.19.1</version>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 
 	<build>

src/main/java/com/jpmc/midascore/MidasCoreApplication.java

@@ -2,8 +2,12 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.domain.EntityScan;
+import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
 
 @SpringBootApplication
+@EntityScan("com.jpmc.midascore.entity")  // Explicitly scan entity package
+@EnableJpaRepositories("com.jpmc.midascore.repository")  // Explicitly scan repository package
 public class MidasCoreApplication {
 
     public static void main(String[] args) {

src/main/java/com/jpmc/midascore/component/IncentiveService.java

@@ -0,0 +1,28 @@
+package com.jpmc.midascore.component;
+
+import org.springframework.stereotype.Component;
+import org.springframework.web.client.RestTemplate;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.jpmc.midascore.exception.DependencyException;
+import com.jpmc.midascore.foundation.Transaction;
+
+@Component
+public class IncentiveService {
+
+    private final RestTemplate restTemplate = new RestTemplate();
+
+    private final static String INCENTIVE_URL = "http://localhost:8080/incentive";
+
+    private final ObjectMapper objectMapper = new ObjectMapper();
+
+    public float getIncentive(Transaction transaction) throws DependencyException {
+        try {
+            String responseEntityStr = restTemplate.postForObject(INCENTIVE_URL, transaction, String.class);
+            return Float.parseFloat(objectMapper.readTree(responseEntityStr).get("amount").asText());
+        } catch (Exception e) {
+            throw new DependencyException(e.getMessage());
+        }
+    }
+
+}

src/main/java/com/jpmc/midascore/component/TransactionConsumer.java

@@ -0,0 +1,90 @@
+package com.jpmc.midascore.component;
+
+import com.jpmc.midascore.entity.TransactionRecord;
+import com.jpmc.midascore.entity.UserRecord;
+import com.jpmc.midascore.exception.DependencyException;
+import com.jpmc.midascore.exception.ValidationException;
+import com.jpmc.midascore.foundation.Transaction;
+import com.jpmc.midascore.repository.TransactionRecordRepository;
+import com.jpmc.midascore.repository.UserRepository;
+
+import java.util.Optional;
+
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+@Component
+public class TransactionConsumer {
+
+    private final UserRepository userRepository;
+
+    private final TransactionRecordRepository transactionRecordRepository;
+
+    private final IncentiveService incentiveService;
+
+    public TransactionConsumer(UserRepository userRepository,
+        TransactionRecordRepository transactionRecordRepository,
+        IncentiveService incentiveService) {
+        this.userRepository = userRepository;
+        this.transactionRecordRepository = transactionRecordRepository;
+        this.incentiveService = incentiveService;
+    }
+
+    @KafkaListener(topics = "${general.kafka-topic}", groupId = "${spring.kafka.consumer.group-id}")
+    public void listen(Transaction transaction) {
+        try {
+            final UserRecord receiver = validateAndGetReceiver(transaction);
+            final UserRecord sender = validateAndGetSender(transaction);
+            final TransactionRecord transactionRecord = new TransactionRecord(sender, receiver,
+                    transaction.getAmount());
+            final float incentive = getIncentive(transaction);
+            receiver.setBalance(receiver.getBalance() + transaction.getAmount() + incentive);
+            sender.setBalance(sender.getBalance() - transaction.getAmount());
+            updateDatabaseInOneTransaction(transactionRecord, sender, receiver);
+            System.out.println(sender.getName() + " - balance: " + sender.getBalance());
+            System.out.println(receiver.getName() + " - balance: " + receiver.getBalance());
+        } catch (ValidationException ve) {
+            System.out.println("no modification to db because of error: " + ve.getMessage());
+        } catch (DependencyException de) {
+            System.out.println("no modification to db because of error: " + de.getMessage());
+        }
+    }
+
+    private float getIncentive(Transaction transaction) throws DependencyException {
+        return incentiveService.getIncentive(transaction);
+    }
+
+    @Transactional
+    private void updateDatabaseInOneTransaction(final TransactionRecord transactionRecord,
+        final UserRecord sender, final UserRecord receiver) {
+        transactionRecordRepository.save(transactionRecord);
+        userRepository.save(sender);
+        userRepository.save(receiver);
+    }
+
+    private UserRecord validateAndGetReceiver(final Transaction transaction) throws ValidationException {
+        final Long recipientId = transaction.getRecipientId();
+        Optional<UserRecord> userRecord = userRepository.findById(recipientId);
+        if (!userRecord.isPresent()) {
+            final String errorMsg = "Recipient not found with id " + recipientId;
+            throw new ValidationException(errorMsg);
+        }
+        return userRecord.get();
+    }
+
+    private UserRecord validateAndGetSender(final Transaction transaction) throws ValidationException {
+        final Long senderId = transaction.getSenderId();
+        Optional<UserRecord> userRecord = userRepository.findById(senderId);
+        if (!userRecord.isPresent()) {
+            final String errorMsg = "Sender not found with id " + senderId;
+            throw new ValidationException(errorMsg);
+        }
+        final UserRecord sender = userRecord.get();
+        if (sender.getBalance() < transaction.getAmount()) {
+            final String errorMsg = "Sender does not have enough balance to complete the transaction";
+            throw new ValidationException(errorMsg);
+        }
+        return sender;
+    }
+}

src/main/java/com/jpmc/midascore/entity/TransactionRecord.java

@@ -0,0 +1,75 @@
+package com.jpmc.midascore.entity;
+
+import jakarta.persistence.*;
+import com.fasterxml.jackson.annotation.JsonIdentityInfo;
+import com.fasterxml.jackson.annotation.ObjectIdGenerators;
+
+@Entity
+@JsonIdentityInfo(
+  generator = ObjectIdGenerators.PropertyGenerator.class, 
+  property = "id")
+public class TransactionRecord {
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    @ManyToOne(fetch = FetchType.LAZY)
+    @JoinColumn(name = "senderId", nullable = false)
+    private UserRecord sender;
+
+    @ManyToOne(fetch = FetchType.LAZY)
+    @JoinColumn(name = "recipientId", nullable = false)
+    private UserRecord recipient;
+
+    @Column(nullable = false)
+    private float amount;
+
+    // Default constructor required by JPA
+    public TransactionRecord() {}
+
+    // All-args constructor for convenience
+    public TransactionRecord(UserRecord sender, UserRecord recipient, float amount) {
+        this.sender = sender;
+        this.recipient = recipient;
+        this.amount = amount;
+    }
+
+    // Getters and Setters
+    public Long getId() {
+        return id;
+    }
+
+    public UserRecord getSender() {
+        return sender;
+    }
+
+    public void setSender(UserRecord sender) {
+        this.sender = sender;
+    }
+
+    public UserRecord getRecipient() {
+        return recipient;
+    }
+
+    public void setRecipient(UserRecord recipient) {
+        this.recipient = recipient;
+    }
+
+    public float getAmount() {
+        return amount;
+    }
+
+    public void setAmount(float amount) {
+        this.amount = amount;
+    }
+
+    @Override
+    public String toString() {
+        return "Transaction{" +
+                "id=" + id +
+                ", senderId=" + (sender != null ? sender.getId() : "null") +
+                ", recipientId=" + (recipient != null ? recipient.getId() : "null") +
+                ", amount=" + amount +
+                '}';
+    }
+}