Feature/worker service

.github/workflows/api-gateway.yml

@@ -32,7 +32,7 @@ jobs:
           cache: maven
 
       - name: Build Project
-        run: mvn clean package
+        run: mvn clean package -DskipTests
 
       - name: Login to Docker Hub
         uses: docker/login-action@v3

.github/workflows/auth-service.yml

@@ -32,7 +32,7 @@ jobs:
           cache: maven
 
       - name: Build Project
-        run: mvn clean package
+        run: mvn clean package -DskipTests
 
       - name: Login to Docker Hub
         uses: docker/login-action@v3

.github/workflows/discovery-service.yml

@@ -32,7 +32,7 @@ jobs:
           cache: maven
 
       - name: Build Project
-        run: mvn clean package
+        run: mvn clean package -DskipTests
 
       - name: Login to Docker Hub
         uses: docker/login-action@v3

auth-service/src/main/java/com/flowforge/auth_service/exception/GlobalExceptionHandler.java

@@ -64,6 +64,10 @@ public ApiResponse<Void> handleInvitationExpired(Exception ex) {
     public ApiResponse<Void> handleOrganizationNotFound(Exception ex) {
         return ApiResponse.error(ex.getMessage());
     }
+    @ExceptionHandler(InvitationEmailSentFailedException.class)
+    public ApiResponse<Void> handleInvitationEmailSentFailed(Exception ex) {
+        return ApiResponse.error(ex.getMessage());
+    }
 
     @ExceptionHandler(MethodArgumentNotValidException.class)
     @ResponseStatus(HttpStatus.BAD_REQUEST)

auth-service/src/main/java/com/flowforge/auth_service/exception/InvitationEmailSentFailedException.java

@@ -0,0 +1,7 @@
+package com.flowforge.auth_service.exception;
+
+public class InvitationEmailSentFailedException extends RuntimeException {
+    public InvitationEmailSentFailedException(String message) {
+        super(message);
+    }
+}

auth-service/src/main/java/com/flowforge/auth_service/service/AuthenticationService.java

@@ -7,6 +7,9 @@
 import com.flowforge.auth_service.model.Invitation;
 import com.flowforge.auth_service.model.Organization;
 import com.flowforge.auth_service.model.User;
+import com.flowforge.auth_service.repository.InvitationRepository;
+import com.flowforge.auth_service.repository.OrganizationRepository;
+import com.flowforge.auth_service.repository.UserRepository;
 import jakarta.transaction.Transactional;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Service;
@@ -15,25 +18,26 @@
 @RequiredArgsConstructor
 public class AuthenticationService {
     private final UserService userService;
+    private final UserRepository userRepository;
+    private final OrganizationRepository organizationRepository;
     private final OrganizationMapper organizationMapper;
-    private final OrganizationService organizationService;
+
     private final UserMapper userMapper;
     private final InvitationService invitationService;
 
-//repo
     @Transactional
     public User registerOrganization(RegisterOrgRequest request) {
         Organization org = organizationMapper.toEntity(request);
-        Organization savedOrg=organizationService.save(org);
+        Organization savedOrg=organizationRepository.save(org);
 
         User user = userMapper.toManagerEntity(request, savedOrg, userService.encodePassword(request.getPassword()));
-        return userService.save(user);
+        return userRepository.save(user);
     }
 
     @Transactional
     public User registerWithInvite(RegisterWithInviteRequest request, Invitation invitation) {
         User user = userMapper.toEntity(request, invitation, userService.encodePassword(request.getPassword()));
-        userService.save(user);
+        userRepository.save(user);
         invitationService.markAccepted(invitation);
         return user;
 

auth-service/src/main/java/com/flowforge/auth_service/service/MailService.java

@@ -1,5 +1,6 @@
 package com.flowforge.auth_service.service;
 
+import com.flowforge.auth_service.exception.InvitationEmailSentFailedException;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
@@ -35,8 +36,8 @@ public void sendInvitationEmail(String toEmail, String token, String orgName) {
             mailSender.send(message);
             log.info("Invitation email sent successfully to {}", toEmail);
         } catch (Exception e) {
-            //throw
             log.error("Failed to send invitation email to {}: {}. Dev invitation token: {}", toEmail, e.getMessage(), token);
+            throw new InvitationEmailSentFailedException(e.getMessage());
         }
     }
 }

docker-compose.yml

@@ -49,20 +49,20 @@ services:
       - "1025:1025"
       - "8025:8025"
 
-#  workflow-service:
-#    image: anzyl/workflow-service:v1
-#    container_name: workflow-service
-#    ports:
-#      - "8082:8082"
-#    depends_on:
-#      - db
-#      - discovery-service
-#    networks:
-#      - default
-#    environment:
-#      SPRING_PROFILES_ACTIVE: docker
-#    volumes:
-#      - ./database:/app/database
+  workflow-service:
+    image: anzyl/workflow-service:v1
+    container_name: workflow-service
+    ports:
+      - "8082:8082"
+    depends_on:
+      - db
+      - discovery-service
+    networks:
+      - default
+    environment:
+      SPRING_PROFILES_ACTIVE: docker
+    volumes:
+      - ./database:/app/database
 
   worker-service:
     image: aswinrj/worker-service

worker-service/src/main/java/com/flowforge/worker_service/adapter/in/kafka/TaskCreatedConsumer.java

@@ -3,10 +3,15 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.flowforge.worker_service.application.port.in.ExecuteTaskUseCase;
+import com.flowforge.worker_service.common.exception.WorkerExecutionException;
+import com.flowforge.worker_service.domain.model.WorkerResult;
 import com.flowforge.worker_service.domain.model.WorkerTask;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.annotation.RetryableTopic;
+import org.springframework.kafka.retrytopic.TopicSuffixingStrategy;
+import org.springframework.retry.annotation.Backoff;
 import org.springframework.stereotype.Component;
 
 @Component
@@ -17,6 +22,14 @@ public class TaskCreatedConsumer {
     private final ExecuteTaskUseCase executeTaskUseCase;
     private final ObjectMapper objectMapper;
 
+    @RetryableTopic(
+            attempts = "3",
+            backoff = @Backoff(delay = 1000, multiplier = 2),
+            autoCreateTopics = "true",
+            topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE,
+            include = { WorkerExecutionException.class },
+            exclude = { JsonProcessingException.class }
+    )
     @KafkaListener(
             topics = "task-created",
             groupId = "worker-group",
@@ -37,6 +50,11 @@ public void consume(String message) throws JsonProcessingException {
         );
 
         log.info("Received task-created event for task {} of type {}", task.getId(), task.getType());
-        executeTaskUseCase.execute(task);
+        WorkerResult result = executeTaskUseCase.execute(task);
+
+        if (!result.isSuccess()) {
+            log.warn("Task {} failed on this attempt: {}", task.getId(), result.getMessage());
+            throw new WorkerExecutionException("Task " + task.getId() + " failed: " + result.getMessage());
+        }
     }
 }

worker-service/src/main/java/com/flowforge/worker_service/adapter/in/kafka/TaskCreatedDltHandler.java

@@ -0,0 +1,38 @@
+package com.flowforge.worker_service.adapter.in.kafka;
+
+import com.flowforge.worker_service.adapter.out.persistence.DeadLetterTaskEntity;
+import com.flowforge.worker_service.adapter.out.persistence.DeadLetterTaskRepository;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.annotation.DltHandler;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class TaskCreatedDltHandler {
+    private final DeadLetterTaskRepository deadLetterTaskRepository;
+    @DltHandler
+    @KafkaListener(
+            topics = "task-created-dlt",
+            groupId = "worker-group-dlt"
+    )
+    public void handleDlt(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, @Header(KafkaHeaders.EXCEPTION_MESSAGE) String exceptionMessage) {
+
+        log.error("Task-created message sent to DLT from topic {}: {} | reason: {}",
+                topic, message, exceptionMessage);
+
+        DeadLetterTaskEntity entity = DeadLetterTaskEntity.builder()
+                .originalTopic(topic)
+                .taskPayload(message)
+                .exceptionMessage(exceptionMessage)
+                .failedAt(LocalDateTime.now())
+                .build();
+        deadLetterTaskRepository.save(entity);
+    }
+}

worker-service/src/main/java/com/flowforge/worker_service/adapter/out/persistence/DeadLetterTaskEntity.java

@@ -0,0 +1,36 @@
+package com.flowforge.worker_service.adapter.out.persistence;
+
+
+import jakarta.persistence.*;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+@Entity
+@Table(name = "dead_letter_tasks")
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class DeadLetterTaskEntity {
+
+    @Id
+    @GeneratedValue
+    private UUID id;
+
+    @Column(name = "original_topic", nullable = false)
+    private String originalTopic;
+
+    @Column(name = "task_payload", columnDefinition = "TEXT", nullable = false)
+    private String taskPayload;
+
+    @Column(name = "exception_message", columnDefinition = "TEXT")
+    private String exceptionMessage;
+
+    @Column(name = "failed_at", nullable = false)
+    private LocalDateTime failedAt;
+}

worker-service/src/main/java/com/flowforge/worker_service/adapter/out/persistence/DeadLetterTaskRepository.java

@@ -0,0 +1,8 @@
+package com.flowforge.worker_service.adapter.out.persistence;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+
+import java.util.UUID;
+
+public interface DeadLetterTaskRepository extends JpaRepository<DeadLetterTaskEntity, UUID> {
+}

worker-service/src/main/java/com/flowforge/worker_service/application/service/TaskExecutorService.java

@@ -20,7 +20,6 @@ public class TaskExecutorService implements ExecuteTaskUseCase {
     private final EventPublisherPort eventPublisherPort;
 
     @Override
-    @Transactional
     public WorkerResult execute(WorkerTask task) {
         try {
             task.markRunning();

worker-service/src/main/java/com/flowforge/worker_service/common/exception/WorkerExecutionException.java

@@ -0,0 +1,7 @@
+package com.flowforge.worker_service.common.exception;
+
+public class WorkerExecutionException extends RuntimeException {
+    public WorkerExecutionException(String message) {
+        super(message);
+    }
+}

worker-service/src/main/java/com/flowforge/worker_service/config/KafkaConfig.java

@@ -5,13 +5,16 @@
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.annotation.EnableKafkaRetryTopic;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 
 import java.util.HashMap;
 import java.util.Map;
-
+@EnableKafka
+@EnableKafkaRetryTopic
 @Configuration
 public class KafkaConfig {
 

worker-service/src/main/java/com/flowforge/worker_service/domain/model/WorkerResult.java

@@ -1,6 +1,7 @@
 package com.flowforge.worker_service.domain.model;
 
 import lombok.AllArgsConstructor;
+import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
@@ -9,6 +10,7 @@
 @Data
 @NoArgsConstructor
 @AllArgsConstructor
+@Builder
 public class WorkerResult {
     private UUID taskId;
     private boolean success;

worker-service/src/main/java/com/flowforge/worker_service/domain/model/WorkerTask.java

@@ -2,13 +2,15 @@
 
 import com.flowforge.worker_service.domain.enums.TaskStatus;
 import lombok.AllArgsConstructor;
+import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
 import java.time.LocalDateTime;
 import java.util.UUID;
 
 @Data
+@Builder
 @AllArgsConstructor
 @NoArgsConstructor
 public class WorkerTask {

worker-service/src/main/resources/application.yaml

@@ -34,7 +34,7 @@ spring:
         spring.json.trusted.packages: "*"
 
   mail:
-    host: localhost
+    host: mailpit
     port: 1025
 eureka:
   client: