Kafka listener in notification service

lombok.config

@@ -0,0 +1 @@
+lombok.addLombokGeneratedAnnotation = true

pom.xml

@@ -15,6 +15,7 @@
 	<description>Notification service for rbc library</description>
 	<properties>
 		<java.version>17</java.version>
+		<mapstruct.version>1.4.2.Final</mapstruct.version>
 		<sonar.organization>productdock</sonar.organization>
 		<sonar.host.url>https://sonarcloud.io</sonar.host.url>
 		<sonar.projectKey>ProductDock_rbc-library-notification</sonar.projectKey>
@@ -24,6 +25,19 @@
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-web</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-oauth2-resource-server</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-security</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.security</groupId>
+			<artifactId>spring-security-test</artifactId>
+			<scope>test</scope>
+		</dependency>
 
 		<dependency>
 			<groupId>org.projectlombok</groupId>
@@ -39,6 +53,48 @@
 			<artifactId>spring-boot-starter-test</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<version>2.6.3</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>junit-jupiter</artifactId>
+			<version>1.17.0</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>kafka</artifactId>
+			<version>1.15.3</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-mongodb</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>de.flapdoodle.embed</groupId>
+			<artifactId>de.flapdoodle.embed.mongo</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.mapstruct</groupId>
+			<artifactId>mapstruct</artifactId>
+			<version>${mapstruct.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.mapstruct</groupId>
+			<artifactId>mapstruct-processor</artifactId>
+			<version>${mapstruct.version}</version>
+		</dependency>
 	</dependencies>
 
 	<profiles>

src/main/java/com/productdock/library/notification/NotificationApplication.java

@@ -1,9 +1,15 @@
 package com.productdock.library.notification;
 
 import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.data.mongodb.repository.config.EnableMongoRepositories;
+import org.springframework.kafka.annotation.EnableKafka;
 
+@EnableKafka
 @SpringBootApplication
+@EnableAutoConfiguration
+@EnableMongoRepositories
 public class NotificationApplication {
 
 	public static void main(String[] args) {

src/main/java/com/productdock/library/notification/adapter/in/kafka/KafkaConsumer.java

@@ -0,0 +1,34 @@
+package com.productdock.library.notification.adapter.in.kafka;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.productdock.library.notification.adapter.in.kafka.messages.NotificationMapper;
+import com.productdock.library.notification.adapter.in.kafka.messages.NotificationMessage;
+import com.productdock.library.notification.application.port.in.AddNotificationUseCase;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+import java.time.OffsetDateTime;
+
+@Component
+@Slf4j
+@AllArgsConstructor
+public class KafkaConsumer {
+
+    private NotificationMapper notificationMapper;
+    private ObjectMapper objectMapper;
+    private AddNotificationUseCase addNotificationUseCase;
+
+    @KafkaListener(topics = "${spring.kafka.topic.notifications}")
+    public synchronized void listenNotifications(ConsumerRecord<String, String> message) throws JsonProcessingException {
+        log.info("Received notification kafka message: {}", message);
+
+        var notificationMessage = objectMapper.readValue(message.value(), NotificationMessage.class);
+        var notification = notificationMapper.toDomain(notificationMessage);
+        notification.setCreatedDate(OffsetDateTime.now());
+        addNotificationUseCase.saveNotification(notification);
+    }
+}

src/main/java/com/productdock/library/notification/adapter/in/kafka/messages/ActionMessage.java

@@ -0,0 +1,13 @@
+package com.productdock.library.notification.adapter.in.kafka.messages;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+
+@AllArgsConstructor
+@Getter
+@Builder
+public class ActionMessage {
+    private String type;
+    private String target;
+}

src/main/java/com/productdock/library/notification/adapter/in/kafka/messages/NotificationMapper.java

@@ -0,0 +1,15 @@
+package com.productdock.library.notification.adapter.in.kafka.messages;
+
+import com.productdock.library.notification.domain.Notification;
+import org.mapstruct.Mapper;
+import org.mapstruct.Mapping;
+import org.mapstruct.ReportingPolicy;
+
+@Mapper(unmappedTargetPolicy = ReportingPolicy.IGNORE, componentModel = "spring")
+public interface NotificationMapper {
+
+    @Mapping(target = "read", ignore = true)
+    @Mapping(target = "id", ignore = true)
+    @Mapping(target = "createdDate", ignore = true)
+    public Notification toDomain(NotificationMessage source);
+}

src/main/java/com/productdock/library/notification/adapter/in/kafka/messages/NotificationMessage.java

@@ -0,0 +1,13 @@
+package com.productdock.library.notification.adapter.in.kafka.messages;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+
+@Builder
+@AllArgsConstructor
+public class NotificationMessage {
+    public final String title;
+    public final String description;
+    public final String userId;
+    public final ActionMessage action;
+}

src/main/java/com/productdock/library/notification/adapter/in/web/NotificationsApi.java

@@ -0,0 +1,46 @@
+package com.productdock.library.notification.adapter.in.web;
+
+import com.productdock.library.notification.application.port.in.GetNotificationsQuery;
+import com.productdock.library.notification.application.port.in.ReadNotificationsUseCase;
+import com.productdock.library.notification.domain.Notification;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.security.core.Authentication;
+import org.springframework.security.oauth2.jwt.Jwt;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.List;
+
+@RestController
+@RequestMapping("api/notifications")
+@Slf4j
+@AllArgsConstructor
+public class NotificationsApi {
+
+    private GetNotificationsQuery getNotificationsQuery;
+    private ReadNotificationsUseCase readNotificationsUseCase;
+
+    public static final String CLAIM_EMAIL = "email";
+
+    @GetMapping
+    public List<Notification> getNotifications(Authentication authentication) {
+        var userId = getUserId(authentication);
+        return getNotificationsQuery.getNotifications(userId);
+    }
+
+    @GetMapping("/unread")
+    public int getUnreadCount(Authentication authentication){
+        var userId = getUserId(authentication);
+        return getNotificationsQuery.getUnreadNotificationsCount(userId);
+    }
+
+    @PostMapping("/read")
+    public void readNotifications(Authentication authentication) {
+        var userId = getUserId(authentication);
+        readNotificationsUseCase.markAsReadNotifications(userId);
+    }
+
+    private String getUserId(Authentication authentication) {
+        return ((Jwt) authentication.getCredentials()).getClaim(CLAIM_EMAIL).toString();
+    }
+}

src/main/java/com/productdock/library/notification/adapter/out/mongo/NotificationPersistenceAdapter.java

@@ -0,0 +1,47 @@
+package com.productdock.library.notification.adapter.out.mongo;
+
+import com.productdock.library.notification.adapter.out.mongo.enitity.NotificationEntity;
+import com.productdock.library.notification.adapter.out.mongo.mapper.NotificationEntityMapper;
+import com.productdock.library.notification.application.port.out.persistence.NotificationPersistenceOutPort;
+import com.productdock.library.notification.domain.Notification;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.stereotype.Repository;
+
+import java.util.List;
+
+@Repository
+@Slf4j
+@AllArgsConstructor
+public class NotificationPersistenceAdapter implements NotificationPersistenceOutPort {
+
+    private NotificationRepository notificationRepository;
+    private NotificationEntityMapper mapper;
+    private MongoTemplate mongoTemplate;
+
+    @Override
+    public List<Notification> getAllByUserId(String userId) {
+        return notificationRepository.findAllByUserId(userId)
+                .stream().map(notificationEntity -> mapper.toDomain(notificationEntity)).toList();
+    }
+
+    @Override
+    public List<Notification> getUnreadByUserId(String userId) {
+        return notificationRepository.findAllByReadAndUserId(false, userId)
+                .stream().map(notificationEntity -> mapper.toDomain(notificationEntity)).toList();
+    }
+
+    @Override
+    public void save(Notification notification) {
+        log.info("Notification for saving: {}", notification);
+        var saved = notificationRepository.save(mapper.toEntity(notification));
+        List<NotificationEntity> savedNotification = notificationRepository.findAllByUserId(notification.getUserId());
+        log.info("Saved notification: {}", savedNotification);
+    }
+
+    @Override
+    public void saveAll(List<Notification> notifications) {
+        notificationRepository.saveAll(notifications.stream().map(notification -> mapper.toEntity(notification)).toList());
+    }
+}

src/main/java/com/productdock/library/notification/adapter/out/mongo/NotificationRepository.java

@@ -0,0 +1,15 @@
+package com.productdock.library.notification.adapter.out.mongo;
+
+import com.productdock.library.notification.adapter.out.mongo.enitity.NotificationEntity;
+import org.springframework.data.mongodb.repository.MongoRepository;
+import org.springframework.stereotype.Repository;
+
+import java.util.List;
+
+@Repository
+public interface NotificationRepository extends MongoRepository<NotificationEntity, String> {
+
+    List<NotificationEntity> findAllByUserId(String userId);
+
+    List<NotificationEntity> findAllByReadAndUserId(boolean read, String userId);
+}

src/main/java/com/productdock/library/notification/adapter/out/mongo/enitity/ActionEntity.java

@@ -0,0 +1,17 @@
+package com.productdock.library.notification.adapter.out.mongo.enitity;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.io.Serializable;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@Builder
+public class ActionEntity implements Serializable {
+    private String type;
+    private String target;
+}

src/main/java/com/productdock/library/notification/adapter/out/mongo/enitity/NotificationEntity.java

@@ -0,0 +1,26 @@
+package com.productdock.library.notification.adapter.out.mongo.enitity;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+import java.io.Serializable;
+
+@Document("notifications")
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@Builder
+public class NotificationEntity implements Serializable {
+    @Id
+    private String id;
+    private String title;
+    private String description;
+    private String userId;
+    private boolean read;
+    private String createdDate;
+    private ActionEntity action;
+}

src/main/java/com/productdock/library/notification/adapter/out/mongo/mapper/NotificationEntityMapper.java

@@ -0,0 +1,22 @@
+package com.productdock.library.notification.adapter.out.mongo.mapper;
+
+import com.productdock.library.notification.adapter.out.mongo.enitity.NotificationEntity;
+import com.productdock.library.notification.domain.Notification;
+import org.mapstruct.Mapper;
+import org.mapstruct.ReportingPolicy;
+
+import java.time.OffsetDateTime;
+
+@Mapper(unmappedTargetPolicy = ReportingPolicy.IGNORE, componentModel = "spring")
+public interface NotificationEntityMapper {
+    NotificationEntity toEntity(Notification source);
+    Notification toDomain(NotificationEntity source);
+
+    default String map(OffsetDateTime value){
+        return value.toString();
+    }
+
+    default OffsetDateTime map(String value) {
+        return OffsetDateTime.parse(value);
+    }
+}

src/main/java/com/productdock/library/notification/application/port/in/AddNotificationUseCase.java

@@ -0,0 +1,8 @@
+package com.productdock.library.notification.application.port.in;
+
+import com.productdock.library.notification.domain.Notification;
+
+public interface AddNotificationUseCase {
+
+    void saveNotification(Notification notification);
+}

src/main/java/com/productdock/library/notification/application/port/in/GetNotificationsQuery.java

@@ -0,0 +1,12 @@
+package com.productdock.library.notification.application.port.in;
+
+import com.productdock.library.notification.domain.Notification;
+
+import java.util.List;
+
+public interface GetNotificationsQuery {
+
+    List<Notification> getNotifications(String userId);
+
+    int getUnreadNotificationsCount(String userId);
+}

src/main/java/com/productdock/library/notification/application/port/in/ReadNotificationsUseCase.java

@@ -0,0 +1,6 @@
+package com.productdock.library.notification.application.port.in;
+
+public interface ReadNotificationsUseCase {
+
+    void markAsReadNotifications(String userId);
+}

src/main/java/com/productdock/library/notification/application/port/out/persistence/NotificationPersistenceOutPort.java

@@ -0,0 +1,16 @@
+package com.productdock.library.notification.application.port.out.persistence;
+
+import com.productdock.library.notification.domain.Notification;
+
+import java.util.List;
+
+public interface NotificationPersistenceOutPort {
+
+    List<Notification> getAllByUserId(String userId);
+
+    List<Notification> getUnreadByUserId(String userId);
+
+    void save(Notification notification);
+
+    void saveAll(List<Notification> notifications);
+}