Refactor Kafka message handling and introduce SummaryBook entity

Author: maxiplux

src/main/java/app/quantun/summary/message/consumer/MyKafkaConsumer.java

@@ -1,11 +1,9 @@
 package app.quantun.summary.message.consumer;
 
-import app.quantun.summary.model.contract.MessagePayload;
-import org.springframework.kafka.annotation.KafkaListener;
+import app.quantun.summary.model.contract.message.BookFilePayload;
 
 import java.util.List;
-import java.util.Map;
 
 public interface MyKafkaConsumer {
-    void processMessages(List<MessagePayload> messages);
+    void processMessages(List<BookFilePayload> messages);
 }

src/main/java/app/quantun/summary/message/consumer/MyKafkaConsumerImpl.java

@@ -1,26 +1,28 @@
 package app.quantun.summary.message.consumer;
 
-import app.quantun.summary.model.contract.MessagePayload;
+import app.quantun.summary.model.contract.message.BookFilePayload;
+import app.quantun.summary.repository.SummaryBookRepository;
+import lombok.NoArgsConstructor;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.support.Acknowledgment;
-import org.springframework.messaging.handler.annotation.Payload;
 import org.springframework.retry.annotation.Backoff;
 import org.springframework.retry.annotation.Retryable;
 import org.springframework.stereotype.Service;
 
 import java.util.List;
-import java.util.Map;
 
 /**
  * Kafka consumer service for processing messages from the "bulk-data" topic.
  * This class listens to Kafka messages and logs the received messages.
  */
 @Service
 @Slf4j
+@RequiredArgsConstructor
 public class MyKafkaConsumerImpl implements MyKafkaConsumer {
 
+  private final SummaryBookRepository summaryBookRepository;
 
   /**
    * Process incoming messages from the "bulk-data" topic.
@@ -40,7 +42,7 @@ public class MyKafkaConsumerImpl implements MyKafkaConsumer {
           maxAttempts = 3,
           backoff = @Backoff(delay = 2000)
   )
-  public void listen(List<MessagePayload> messages, Acknowledgment ack) {
+  public void listen(List<BookFilePayload> messages, Acknowledgment ack) {
     try {
       log.info("Received batch of {} messages", messages.size());
 
@@ -62,9 +64,10 @@ public void listen(List<MessagePayload> messages, Acknowledgment ack) {
    *
    * @param messages List of MessagePayload to process.
    */
-  public void processMessages(List<MessagePayload> messages) {
+  public void processMessages(List<BookFilePayload> messages) {
     // Your processing logic here
-    for (MessagePayload message : messages) {
+    for (BookFilePayload message : messages) {
+      this.summaryBookRepository.findById(message.id());
       log.info("Processing message: {}", message);
     }
 

src/main/java/app/quantun/summary/message/producer/KafkaProducerService.java

@@ -1,5 +1,9 @@
 package app.quantun.summary.message.producer;
 
+import app.quantun.summary.model.contract.message.BookFilePayload;
+
 public interface KafkaProducerService {
-  void sendHashMapMessage();
+
+
+  void sendBookToBeProcessed(BookFilePayload message);
 }

src/main/java/app/quantun/summary/message/producer/KafkaProducerServiceImpl.java

@@ -1,14 +1,12 @@
 package app.quantun.summary.message.producer;
 
-import app.quantun.summary.model.contract.MessagePayload;
+import app.quantun.summary.model.contract.message.BookFilePayload;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Service;
-import org.springframework.util.concurrent.ListenableFuture;
-import org.springframework.util.concurrent.ListenableFutureCallback;
 
 import java.util.concurrent.CompletableFuture;
 import java.util.stream.IntStream;
@@ -22,38 +20,27 @@
 @Slf4j
 public class KafkaProducerServiceImpl implements KafkaProducerService {
 
-  private final KafkaTemplate<Object, MessagePayload> kafkaTemplate;
+  private final KafkaTemplate<Object, BookFilePayload> kafkaTemplate;
 
     // Externalize the topic name for better configurability
     @Value("${kafka.topic.bulk-data}")
     private String bulkDataTopic;
 
 
-    /**
-     * Sends a batch of HashMap messages to the "bulk-data" topic.
-     * Each message contains an ID, name, and value.
-     */
-    @Override
-    public void sendHashMapMessage() {
-        IntStream.rangeClosed(1, 10)
-                .forEach(i -> {
-                    MessagePayload messagePayload = new MessagePayload(
-                            String.valueOf(i),
-                            "User" + i,
-                            "Some data for user " + i
-                    );
 
-                    CompletableFuture<SendResult<Object, MessagePayload>> future = kafkaTemplate.send(bulkDataTopic, messagePayload).toCompletableFuture();
 
-                    future.thenAccept(result ->
-                                    log.info("Successfully sent message {} with payload: {}", i, messagePayload))
+    @Override
+    public void sendBookToBeProcessed(BookFilePayload message) {
+        CompletableFuture<SendResult<Object, BookFilePayload>> future = kafkaTemplate.send(bulkDataTopic, message).toCompletableFuture();
 
+        future.thenAccept(result ->
+                        log.info("Successfully sent message with payload: {}", message))
 
-                            .exceptionally(ex -> {
-                                log.error("Failed to send message {} with payload: {}. Exception: {}", i, messagePayload, ex.getMessage());
-                                return null;
-                            });
+                .exceptionally(ex -> {
+                    log.error("Failed to send message with payload: {}. Exception: {}", message, ex.getMessage());
+                    return null;
                 });
+
     }
 
 }

src/main/java/app/quantun/summary/model/contract/MessagePayload.java

@@ -0,0 +1,3 @@
+package app.quantun.summary.model.contract.message;
+
+public record BookFilePayload(Long id, String name) {}

src/main/java/app/quantun/summary/model/contract/message/BookFilePayload.java

@@ -0,0 +1,29 @@
+package app.quantun.summary.model.entity;
+
+import jakarta.persistence.*;
+import lombok.*;
+
+@Data
+@EqualsAndHashCode(of = "id")
+@Entity
+@Table(indexes = {
+        @Index(name = "idx_summarybook_uuid", columnList = "uuid")
+})
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+
+public class SummaryBook {
+    @Id
+    @GeneratedValue(strategy = GenerationType.SEQUENCE)
+    @Column(nullable = false)
+    private Long id;
+    private String name;
+    private String path;
+    @Column(columnDefinition = "TEXT")
+    // add index
+
+    private String uuid;
+
+
+}

src/main/java/app/quantun/summary/model/entity/SummaryBook.java

@@ -0,0 +1,7 @@
+package app.quantun.summary.repository;
+
+import app.quantun.summary.model.entity.SummaryBook;
+import org.springframework.data.jpa.repository.JpaRepository;
+
+public interface SummaryBookRepository extends JpaRepository<SummaryBook, Long> {
+}

src/main/java/app/quantun/summary/repository/SummaryBookRepository.java

@@ -92,6 +92,6 @@ private String saveFileToDisk(MultipartFile file, Path uploadPath) throws IOExce
     Path filePath = uploadPath.resolve(fileName);
     log.info("Saving file to disk at: {}", filePath);
     file.transferTo(filePath);
-    return fileName;
+    return filePath.toString();
   }
 }

src/main/java/app/quantun/summary/service/impl/FileStorageServiceImpl.java

@@ -2,6 +2,9 @@
 
 import app.quantun.summary.message.producer.KafkaProducerService;
 import app.quantun.summary.model.contract.dto.TableIndexContent;
+import app.quantun.summary.model.contract.message.BookFilePayload;
+import app.quantun.summary.model.entity.SummaryBook;
+import app.quantun.summary.repository.SummaryBookRepository;
 import app.quantun.summary.service.FileStorageService;
 import app.quantun.summary.service.PdfServices;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -22,6 +25,7 @@
 import org.springframework.web.multipart.MultipartFile;
 
 import java.util.Map;
+import java.util.UUID;
 
 /**
  * Service implementation for PDF-related operations.
@@ -39,18 +43,11 @@ public class PdfServicesImpl implements PdfServices {
   private final ChatClient openAiChatClient;
   @Qualifier("geminiChatClient")
   private final ChatClient geminiChatClient;
-  private final ObjectMapper objectMapper;
+
   private final ResourceLoader resourceLoader;
-  @Value("classpath:/templates/rag-prompt-template.st")
-  private Resource ressourceRagPromptTemplate;
-  @Value("classpath:/templates/system-message.st")
-  private Resource resourceSystemMessageTemplate;
-  @Value("classpath:/templates/rag-prompt-without-metadata-template.st")
-  private Resource resourceRagPromptWithoutMedataTemplate;
-  @Value("classpath:templates/get-capital-prompt.st")
-  private Resource resourceCapitalPromptTemplate;
-  @Value("classpath:templates/get-capital-with-info.st")
-  private Resource resourceCapitalWithInfoPromptTemplate;
+
+  private final SummaryBookRepository summaryBookRepository;
+
   @Value("classpath:templates/system/table.content.st")
   private Resource tableOfContentPromptTemplate;
 
@@ -84,8 +81,16 @@ public TableIndexContent getBookTableOfContentPages(String message) {
    */
   @Override
   public String storePdfFile(MultipartFile file) {
-    val message = this.fileStorageService.storePdfFile(file);
-    this.kafkaProducerService.sendHashMapMessage();
-    return message;
+    val bookPath = this.fileStorageService.storePdfFile(file);
+    val summaryBook = SummaryBook.builder()
+        .uuid(UUID.randomUUID().toString())
+        .name(file.getOriginalFilename())
+        .path(bookPath)
+        .build();
+    val savedSummaryBook = this.summaryBookRepository.save(summaryBook);
+
+    val message=new BookFilePayload( savedSummaryBook.getId(),savedSummaryBook.getName());
+    this.kafkaProducerService.sendBookToBeProcessed(message);
+    return message.toString();
   }
 }