Completed JPMC Forage Program Tasks 2-5: Kafka, H2, Incentive API, Ba…

pom.xml

@@ -17,6 +17,52 @@
 		<java.version>17</java.version>
 	</properties>
 	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-jpa</artifactId>
+			<version>3.2.5</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+			<version>3.2.5</version>
+		</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>
+			<version>2.2.224</version>
+			<scope>runtime</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<version>3.2.5</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<version>3.1.4</version>
+			<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,12 +2,23 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.web.client.RestTemplate;
 
 @SpringBootApplication
+@EnableKafka
 public class MidasCoreApplication {
 
     public static void main(String[] args) {
         SpringApplication.run(MidasCoreApplication.class, args);
     }
 
+    @Bean
+    public RestTemplate restTemplate() {
+        return new RestTemplate();
+    }
+
 }
+
+

src/main/java/com/jpmc/midascore/controller/BalanceController.java

@@ -0,0 +1,29 @@
+package com.jpmc.midascore.controller;
+
+import com.jpmc.midascore.entity.UserRecord;
+import com.jpmc.midascore.repository.UserRepository;
+import com.jpmc.midascore.foundation.Balance;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.Optional;
+
+@RestController
+public class BalanceController {
+
+    private final UserRepository userRepo;
+
+    public BalanceController(UserRepository userRepo) {
+        this.userRepo = userRepo;
+    }
+
+    @GetMapping("/balance")
+    public Balance getBalance(@RequestParam long userId) {
+        Optional<UserRecord> userOpt = userRepo.findById(userId);
+
+        if (userOpt.isPresent()) {
+            return new Balance(userOpt.get().getBalance());
+        } else {
+            return new Balance(0f);
+        }
+    }
+}

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

@@ -0,0 +1,36 @@
+package com.jpmc.midascore.entity;
+
+import jakarta.persistence.*;
+import java.time.LocalDateTime;
+
+@Entity
+public class TransactionRecord {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    @ManyToOne
+    private UserRecord sender;
+
+    @ManyToOne
+    private UserRecord recipient;
+
+    private float amount;
+
+    private float incentive;   // ✅ new field
+
+    private LocalDateTime timestamp = LocalDateTime.now();
+
+    // getters + setters
+    public float getIncentive() { return incentive; }
+    public void setIncentive(float incentive) { this.incentive = incentive; }
+    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; }
+    public LocalDateTime getTimestamp() { return timestamp; }
+}

src/main/java/com/jpmc/midascore/foundation/Incentive.java

@@ -0,0 +1,13 @@
+package com.jpmc.midascore.foundation;
+
+public class Incentive {
+    private float amount;
+
+    public float getAmount() {
+        return amount;
+    }
+
+    public void setAmount(float amount) {
+        this.amount = amount;
+    }
+}

src/main/java/com/jpmc/midascore/listener/TransactionListener.java

@@ -0,0 +1,74 @@
+package com.jpmc.midascore.listener;
+
+import com.jpmc.midascore.foundation.Incentive;
+import com.jpmc.midascore.foundation.Transaction;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+import com.jpmc.midascore.entity.UserRecord;
+import com.jpmc.midascore.entity.TransactionRecord;
+import com.jpmc.midascore.repository.UserRepository;
+import com.jpmc.midascore.repository.TransactionRecordRepository;
+
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.web.client.RestTemplate;
+
+import java.util.Optional;
+
+
+@Component
+public class TransactionListener {
+
+    private final UserRepository userRepo;
+    private final TransactionRecordRepository txRepo;
+
+
+    public TransactionListener(UserRepository userRepo, TransactionRecordRepository txRepo) {
+        this.userRepo = userRepo;
+        this.txRepo = txRepo;
+    }
+    @Autowired
+    private RestTemplate restTemplate;
+
+    @Transactional
+    @KafkaListener(topics = "${general.kafka-topic}", groupId = "midas-group")
+    public void listen(Transaction transaction) {
+        Optional<UserRecord> senderOpt = userRepo.findById(transaction.getSenderId());
+        Optional<UserRecord> recipientOpt = userRepo.findById(transaction.getRecipientId());
+
+        if (senderOpt.isPresent() && recipientOpt.isPresent()) {
+            UserRecord sender = senderOpt.get();
+            UserRecord recipient = recipientOpt.get();
+
+            if (sender.getBalance() >= transaction.getAmount()) {
+                // Deduct from sender
+                sender.setBalance(sender.getBalance() - transaction.getAmount());
+
+                // Call Incentive API
+                Incentive incentive = restTemplate.postForObject(
+                        "http://localhost:8080/incentive", transaction, Incentive.class);
+
+                float incentiveAmount = (incentive != null) ? incentive.getAmount() : 0f;
+
+                // Add to recipient
+                recipient.setBalance(recipient.getBalance() + transaction.getAmount() + incentiveAmount);
+
+                // Save users
+                userRepo.save(sender);
+                userRepo.save(recipient);
+
+                // Record transaction with incentive
+                TransactionRecord record = new TransactionRecord();
+                record.setSender(sender);
+                record.setRecipient(recipient);
+                record.setAmount(transaction.getAmount());
+                record.setIncentive(incentiveAmount);
+                txRepo.save(record);
+
+                System.out.println("✅ Transaction + incentive recorded: " + record);
+            }
+        }
+    }
+
+}

src/main/java/com/jpmc/midascore/repository/TransactionRecordRepository.java

@@ -0,0 +1,7 @@
+package com.jpmc.midascore.repository;
+
+import com.jpmc.midascore.entity.TransactionRecord;
+import org.springframework.data.jpa.repository.JpaRepository;
+
+public interface TransactionRecordRepository extends JpaRepository<TransactionRecord, Long> {
+}

src/main/java/com/jpmc/midascore/repository/UserRepository.java

@@ -1,8 +1,10 @@
 package com.jpmc.midascore.repository;
 
+import com.jpmc.midascore.entity.UserRecord;
+import org.springframework.data.jpa.repository.JpaRepository;
 import com.jpmc.midascore.entity.UserRecord;
 import org.springframework.data.repository.CrudRepository;
 
 public interface UserRepository extends CrudRepository<UserRecord, Long> {
-    UserRecord findById(long id);
+    UserRecord findByName(String name);
 }

src/test/resources/application.properties

@@ -0,0 +1,5 @@
+general.kafka-topic=test-topic
+spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
+spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
+spring.kafka.consumer.properties.spring.json.trusted.packages=*
+server.port=33400