Merge branch 'issue46' of https://github.com/matheusandre1/timeless into issue46

Author: matheusandre1

timeless-api/src/main/java/dev/matheuscruz/infra/queue/SQS.java

@@ -15,13 +15,19 @@
 import dev.matheuscruz.infra.ai.data.SimpleMessage;
 import io.quarkus.narayana.jta.QuarkusTransaction;
 import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.Multi;
 import jakarta.enterprise.context.ApplicationScoped;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Optional;
 import org.eclipse.microprofile.reactive.messaging.Incoming;
 import org.eclipse.microprofile.reactive.messaging.Message;
 import org.eclipse.microprofile.reactive.messaging.Outgoing;
+import org.eclipse.microprofile.reactive.messaging.Incoming;
+import org.eclipse.microprofile.reactive.messaging.Message;
+import org.eclipse.microprofile.reactive.messaging.Outgoing;
 import org.jboss.logging.Logger;
 import software.amazon.awssdk.services.sqs.SqsClient;
 
@@ -49,6 +55,10 @@ public SQS(SqsClient sqs, ObjectMapper objectMapper, TextAiService aiService, Re
         this.userRepository = userRepository;
     }
 
+    @Incoming("whatsapp-incoming")
+    @Outgoing("whatsapp-recognized")
+    public Multi<Message<String>> receiveMessages(Message<String> message) {
+        String body = message.getPayload();
     @Incoming("whatsapp-incoming")
     @Outgoing("whatsapp-recognized")
     public Multi<Message<String>> receiveMessages(Message<String> message) {
@@ -59,13 +69,31 @@ public Multi<Message<String>> receiveMessages(Message<String> message) {
             return Multi.createFrom().item(message);
         }
 
+        if (!MessageKind.TEXT.equals(incomingMessage.kind())) {
+            return Multi.createFrom().item(message);
+        }
+
         Optional<User> user = this.userRepository.findByPhoneNumber(incomingMessage.sender());
 
         if (user.isEmpty()) {
             logger.error("User not found.");
             return Multi.createFrom().item(message);
         }
 
+        return Multi.createFrom().iterable(handleUserMessage(user.get(), incomingMessage)).map(processedMessage -> {
+            try {
+                String processedBody = objectMapper.writeValueAsString(processedMessage);
+                return Message.of(processedBody).withAck(() -> message.ack())
+                        .withNack(throwable -> message.nack(throwable));
+            } catch (JsonProcessingException e) {
+                logger.error("Failed to serialize message", e);
+                throw new RuntimeException(e);
+            }
+        });
+            logger.error("User not found.");
+            return Multi.createFrom().item(message);
+        }
+
         return Multi.createFrom().iterable(handleUserMessage(user.get(), incomingMessage)).map(processedMessage -> {
             try {
                 String processedBody = objectMapper.writeValueAsString(processedMessage);
@@ -87,7 +115,7 @@ private List<Object> handleUserMessage(User user, IncomingMessage message) {
             for (RecognizedOperation recognizedOperation : allRecognizedOperations.all()) {
                 switch (recognizedOperation.operation()) {
                     case AiOperations.ADD_TRANSACTION ->
-                            results.add(processAddTransactionMessage(user, message, recognizedOperation));
+                        results.add(processAddTransactionMessage(user, message, recognizedOperation));
                     case AiOperations.GET_BALANCE -> {
                         logger.info("Processing GET_BALANCE operation" + recognizedOperation.recognizedTransaction());
                         results.add(processSimpleMessage(user, message, recognizedOperation));