diff --git a/.github/workflows/api-gateway.yml b/.github/workflows/api-gateway.yml index 154c486..a2ffddb 100644 --- a/.github/workflows/api-gateway.yml +++ b/.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 diff --git a/.github/workflows/auth-service.yml b/.github/workflows/auth-service.yml index 6402eaf..f895306 100644 --- a/.github/workflows/auth-service.yml +++ b/.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 diff --git a/.github/workflows/discovery-service.yml b/.github/workflows/discovery-service.yml index 6748a2f..708f9dd 100644 --- a/.github/workflows/discovery-service.yml +++ b/.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 diff --git a/.idea/compiler.xml b/.idea/compiler.xml index 111e76e..d4c6d35 100644 --- a/.idea/compiler.xml +++ b/.idea/compiler.xml @@ -9,7 +9,7 @@ - + diff --git a/auth-service/src/main/java/com/flowforge/auth_service/exception/GlobalExceptionHandler.java b/auth-service/src/main/java/com/flowforge/auth_service/exception/GlobalExceptionHandler.java index 565e7a4..1967f78 100644 --- a/auth-service/src/main/java/com/flowforge/auth_service/exception/GlobalExceptionHandler.java +++ b/auth-service/src/main/java/com/flowforge/auth_service/exception/GlobalExceptionHandler.java @@ -64,6 +64,10 @@ public ApiResponse handleInvitationExpired(Exception ex) { public ApiResponse handleOrganizationNotFound(Exception ex) { return ApiResponse.error(ex.getMessage()); } + @ExceptionHandler(InvitationEmailSentFailedException.class) + public ApiResponse handleInvitationEmailSentFailed(Exception ex) { + return ApiResponse.error(ex.getMessage()); + } @ExceptionHandler(MethodArgumentNotValidException.class) @ResponseStatus(HttpStatus.BAD_REQUEST) diff --git a/auth-service/src/main/java/com/flowforge/auth_service/exception/InvitationEmailSentFailedException.java b/auth-service/src/main/java/com/flowforge/auth_service/exception/InvitationEmailSentFailedException.java new file mode 100644 index 0000000..73b8959 --- /dev/null +++ b/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); + } +} diff --git a/auth-service/src/main/java/com/flowforge/auth_service/service/AuthenticationService.java b/auth-service/src/main/java/com/flowforge/auth_service/service/AuthenticationService.java index e2bd1a5..cb2f5d3 100644 --- a/auth-service/src/main/java/com/flowforge/auth_service/service/AuthenticationService.java +++ b/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; diff --git a/auth-service/src/main/java/com/flowforge/auth_service/service/MailService.java b/auth-service/src/main/java/com/flowforge/auth_service/service/MailService.java index 1d73a27..85047d4 100644 --- a/auth-service/src/main/java/com/flowforge/auth_service/service/MailService.java +++ b/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()); } } } \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml index 5b8602a..081dd37 100644 --- a/docker-compose.yml +++ b/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 diff --git a/worker-service/src/main/java/com/flowforge/worker_service/adapter/in/kafka/TaskCreatedConsumer.java b/worker-service/src/main/java/com/flowforge/worker_service/adapter/in/kafka/TaskCreatedConsumer.java index 035c64d..4523a37 100644 --- a/worker-service/src/main/java/com/flowforge/worker_service/adapter/in/kafka/TaskCreatedConsumer.java +++ b/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()); + } } } diff --git a/worker-service/src/main/java/com/flowforge/worker_service/adapter/in/kafka/TaskCreatedDltHandler.java b/worker-service/src/main/java/com/flowforge/worker_service/adapter/in/kafka/TaskCreatedDltHandler.java new file mode 100644 index 0000000..1eda7fe --- /dev/null +++ b/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); + } +} \ No newline at end of file diff --git a/worker-service/src/main/java/com/flowforge/worker_service/adapter/out/persistence/DeadLetterTaskEntity.java b/worker-service/src/main/java/com/flowforge/worker_service/adapter/out/persistence/DeadLetterTaskEntity.java new file mode 100644 index 0000000..69ef6b2 --- /dev/null +++ b/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; +} diff --git a/worker-service/src/main/java/com/flowforge/worker_service/adapter/out/persistence/DeadLetterTaskRepository.java b/worker-service/src/main/java/com/flowforge/worker_service/adapter/out/persistence/DeadLetterTaskRepository.java new file mode 100644 index 0000000..5f7f951 --- /dev/null +++ b/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 { +} \ No newline at end of file diff --git a/worker-service/src/main/java/com/flowforge/worker_service/application/service/TaskExecutorService.java b/worker-service/src/main/java/com/flowforge/worker_service/application/service/TaskExecutorService.java index 391f7f8..e00bbf7 100644 --- a/worker-service/src/main/java/com/flowforge/worker_service/application/service/TaskExecutorService.java +++ b/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(); diff --git a/worker-service/src/main/java/com/flowforge/worker_service/common/exception/WorkerExecutionException.java b/worker-service/src/main/java/com/flowforge/worker_service/common/exception/WorkerExecutionException.java new file mode 100644 index 0000000..a66d672 --- /dev/null +++ b/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); + } +} diff --git a/worker-service/src/main/java/com/flowforge/worker_service/config/KafkaConfig.java b/worker-service/src/main/java/com/flowforge/worker_service/config/KafkaConfig.java index c1f7979..3e68c92 100644 --- a/worker-service/src/main/java/com/flowforge/worker_service/config/KafkaConfig.java +++ b/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 { diff --git a/worker-service/src/main/java/com/flowforge/worker_service/domain/model/WorkerResult.java b/worker-service/src/main/java/com/flowforge/worker_service/domain/model/WorkerResult.java index 2b2f734..8a30923 100644 --- a/worker-service/src/main/java/com/flowforge/worker_service/domain/model/WorkerResult.java +++ b/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; diff --git a/worker-service/src/main/java/com/flowforge/worker_service/domain/model/WorkerTask.java b/worker-service/src/main/java/com/flowforge/worker_service/domain/model/WorkerTask.java index cda97c9..58d5a02 100644 --- a/worker-service/src/main/java/com/flowforge/worker_service/domain/model/WorkerTask.java +++ b/worker-service/src/main/java/com/flowforge/worker_service/domain/model/WorkerTask.java @@ -2,6 +2,7 @@ import com.flowforge.worker_service.domain.enums.TaskStatus; import lombok.AllArgsConstructor; +import lombok.Builder; import lombok.Data; import lombok.NoArgsConstructor; @@ -9,6 +10,7 @@ import java.util.UUID; @Data +@Builder @AllArgsConstructor @NoArgsConstructor public class WorkerTask { diff --git a/worker-service/src/main/resources/application.yaml b/worker-service/src/main/resources/application.yaml index e252758..119adef 100644 --- a/worker-service/src/main/resources/application.yaml +++ b/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: