Task 1 completed

application.yml

@@ -0,0 +1,35 @@
+spring:
+  kafka:
+    bootstrap-servers: localhost:9092
+    producer:
+      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
+    consumer:
+      group-id: midas-core-group
+      auto-offset-reset: earliest
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
+      properties:
+        spring.json.trusted.packages: "*"
+
+  datasource:
+    url: jdbc:h2:mem:midasdb  # ✅ In-memory H2 database
+    driver-class-name: org.h2.Driver
+    username: sa
+    password:
+
+  jpa:
+    database-platform: org.hibernate.dialect.H2Dialect
+    hibernate:
+      ddl-auto: update  # ✅ Automatically updates schema
+
+  h2:
+    console:
+      enabled: true  # ✅ Enables H2 web console
+
+server:
+  port: 33400  # ✅ Set application to run on port 33400
+
+
+general:
+  kafka-topic: transactions

pom.xml

@@ -17,6 +17,67 @@
 		<java.version>17</java.version>
 	</properties>
 	<dependencies>
+		<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-data-jpa -->
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-jpa</artifactId>
+			<version>3.2.5</version>
+		</dependency>
+
+		<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-web -->
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+			<version>3.2.5</version>
+		</dependency>
+
+		<!-- https://mvnrepository.com/artifact/com.h2database/h2 -->
+		<dependency>
+			<groupId>com.h2database</groupId>
+			<artifactId>h2</artifactId>
+			<version>2.2.224</version>
+			<scope>test</scope>
+		</dependency>
+
+		<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-test -->
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<version>3.2.5</version>
+			<scope>test</scope>
+		</dependency>
+
+		<!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka-test -->
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<version>3.1.4</version>
+			<scope>test</scope>
+		</dependency>
+
+		<!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<version>3.1.4</version>
+		</dependency>
+
+
+
+
+		<!-- https://mvnrepository.com/artifact/org.testcontainers/kafka -->
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>kafka</artifactId>
+			<version>1.19.1</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.maven.plugins</groupId>
+			<artifactId>maven-surefire-plugin</artifactId>
+			<version>3.1.2</version> <!-- Ensure this is correct -->
+		</dependency>
+
 	</dependencies>
 
 	<build>
@@ -25,6 +86,11 @@
 				<groupId>org.springframework.boot</groupId>
 				<artifactId>spring-boot-maven-plugin</artifactId>
 			</plugin>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-surefire-plugin</artifactId>
+				<version>3.1.2</version> <!-- You can also try the latest version -->
+			</plugin>
 		</plugins>
 	</build>
 

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

@@ -2,6 +2,8 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+import org.springframework.web.client.RestTemplate;
 
 @SpringBootApplication
 public class MidasCoreApplication {
@@ -10,4 +12,9 @@ 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,24 @@
+package com.jpmc.midascore.controller;
+
+import com.jpmc.midascore.model.Balance;
+import com.jpmc.midascore.entity.UserRecord;
+import com.jpmc.midascore.repository.UserRepository;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.Optional;
+
+@RestController
+@RequestMapping
+public class BalanceController {
+
+    @Autowired
+    private UserRepository userRepository;
+
+    @GetMapping("/balance")
+    public Balance getBalance(@RequestParam Long userId) {
+        Optional<UserRecord> userOpt = userRepository.findById(userId);
+        float balance = userOpt.map(UserRecord::getBalance).orElse(0.0f);
+        return new Balance(balance);
+    }
+}

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

@@ -0,0 +1,55 @@
+package com.jpmc.midascore.kafka;
+
+import com.jpmc.midascore.entity.UserRecord;
+import com.jpmc.midascore.foundation.Transaction;
+import com.jpmc.midascore.model.TransactionRecord;
+import com.jpmc.midascore.repository.TransactionRepository;
+import com.jpmc.midascore.repository.UserRepository;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.Optional;
+
+@Service
+public class TransactionListener {
+
+    @Autowired
+    private UserRepository userRepository;
+
+    @Autowired
+    private TransactionRepository transactionRepository;
+
+    @KafkaListener(topics = "${general.kafka-topic}", groupId = "midas-core-group")
+    @Transactional
+    public void listen(Transaction transaction) {
+        Optional<UserRecord> senderOpt = Optional.ofNullable(userRepository.findById(transaction.getSenderId()));
+        Optional<UserRecord> recipientOpt = Optional.ofNullable(userRepository.findById(transaction.getRecipientId()));
+
+        if (senderOpt.isPresent() && recipientOpt.isPresent()) {
+            UserRecord sender = senderOpt.get();
+            UserRecord recipient = recipientOpt.get();
+
+            if (sender.getBalance() >= transaction.getAmount()) {
+                // Deduct from sender and add to recipient
+                sender.setBalance(sender.getBalance() - transaction.getAmount());
+                recipient.setBalance(recipient.getBalance() + transaction.getAmount());
+
+                // Save updated users
+                userRepository.save(sender);
+                userRepository.save(recipient);
+
+                // Save transaction record
+                TransactionRecord record = new TransactionRecord(sender, recipient, transaction.getAmount(), 0);
+                transactionRepository.save(record);
+
+                System.out.println("✅ Transaction recorded: " + record.getSender() + " Balance: " + transaction.getAmount());
+            } else {
+                System.out.println("❌ Transaction failed: Insufficient balance for sender ID " + sender.getId());
+            }
+        } else {
+            System.out.println("❌ Transaction failed: Invalid sender or recipient ID.");
+        }
+    }
+}

src/main/java/com/jpmc/midascore/kafka/TransactionProducer.java

@@ -0,0 +1,29 @@
+package com.jpmc.midascore.kafka;
+
+import com.jpmc.midascore.foundation.Transaction;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+
+import java.util.Properties;
+
+public class TransactionProducer {
+    public static void main(String[] args) {
+        String topic = "transactions";
+
+        Properties props = new Properties();
+        props.put("bootstrap.servers", "localhost:9092");
+        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+
+        Producer<String, Transaction> producer = new KafkaProducer<>(props);
+
+        Transaction transaction = new Transaction(101, 202, 500.75f);
+        ProducerRecord<String, Transaction> record = new ProducerRecord<String, Transaction>(topic, "txn-key",  transaction);
+
+        producer.send(record);
+        producer.close();
+
+        System.out.println("Transaction sent successfully");
+    }
+}

src/main/java/com/jpmc/midascore/model/Balance.java

@@ -0,0 +1,17 @@
+package com.jpmc.midascore.model;
+
+public class Balance {
+    private float balance;
+
+    public Balance(float balance) {
+        this.balance = balance;
+    }
+
+    public float getBalance() {
+        return balance;
+    }
+
+    public void setBalance(float balance) {
+        this.balance = balance;
+    }
+}

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

@@ -0,0 +1,13 @@
+package com.jpmc.midascore.model;
+
+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/model/TransactionRecord.java

@@ -0,0 +1,43 @@
+package com.jpmc.midascore.model;
+
+
+import com.jpmc.midascore.entity.UserRecord;
+import jakarta.persistence.*;
+
+@Entity
+public class TransactionRecord {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private long id;
+
+    @ManyToOne
+    @JoinColumn(name = "sender_id", nullable = false)
+    private UserRecord sender;
+
+    @ManyToOne
+    @JoinColumn(name = "recipient_id", nullable = false)
+    private UserRecord recipient;
+
+    @Column(nullable = false)
+    private float amount;
+
+    @Column(nullable = false)
+    private float incentive;
+
+    public TransactionRecord() {}
+
+    public TransactionRecord(UserRecord sender, UserRecord recipient, float amount, float incentive) {
+        this.sender = sender;
+        this.recipient = recipient;
+        this.amount = amount;
+        this.incentive = incentive;
+    }
+
+    public float getIncentive() {return incentive;}
+    public void setIncentive(float incentive){this.incentive = incentive;}
+    public long getId() {return id;}
+    public UserRecord getSender() {return sender;}
+    public UserRecord getRecipient() {return recipient;}
+    public float getAmount() {return amount;}
+}

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

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

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

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

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

@@ -0,0 +1,21 @@
+package com.jpmc.midascore.service;
+
+import com.jpmc.midascore.foundation.Transaction;
+import com.jpmc.midascore.model.Incentive;
+import org.springframework.stereotype.Service;
+import org.springframework.web.client.RestTemplate;
+
+@Service
+public class IncentiveService {
+
+    private final RestTemplate restTemplate;
+
+    public IncentiveService(RestTemplate restTemplate) {
+        this.restTemplate = restTemplate;
+    }
+
+    public Incentive getIncentive(Transaction transaction) {
+        String url = "http://localhost:8080/incentive";
+        return restTemplate.postForObject(url, transaction, Incentive.class);
+    }
+}