Notifications kafka

docker-compose.yml

@@ -8,6 +8,22 @@ services:
       - MYSQL_PASSWORD=${DB_PASSWORD}
     ports:
       - "3302:3306"
+
+  kafka:
+    image: wurstmeister/kafka
+    ports:
+      - "9092:9092"
+    environment:
+      - KAFKA_ADVERTISED_HOST_NAME=kafka
+      - KAFKA_ADVERTISED_PORT=9092
+      - KAFKA_CREATE_TOPICS=topic1:1:1
+      - KAFKA_ZOOKEEPER_CONNECT=${KAFKA_ZOOKEEPER_CONNECT}
+
+  zookeeper:
+    image: wurstmeister/zookeeper
+    ports:
+      - "2181:2181"
+
   app:
     build:
       context: .
@@ -30,4 +46,5 @@ services:
     ports:
       - "8080:8080"
     depends_on:
+      - kafka
       - mysql

pom.xml

@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
 	<modelVersion>4.0.0</modelVersion>
 	<parent>
 		<groupId>org.springframework.boot</groupId>
@@ -74,6 +74,15 @@
 			<artifactId>firebase-admin</artifactId>
 			<version>9.1.1</version>
 		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-websocket</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+		</dependency>
 	</dependencies>
 
 	<build>
@@ -113,4 +122,4 @@
 		</pluginRepository>
 	</pluginRepositories>
 
-</project>
+</project>

src/main/java/com/rm/mynotes/model/Note.java

@@ -12,6 +12,8 @@
 import javax.validation.constraints.NotEmpty;
 import javax.validation.constraints.NotNull;
 import java.time.OffsetDateTime;
+import java.util.ArrayList;
+import java.util.List;
 
 @Data
 @Builder
@@ -40,6 +42,8 @@ public class Note {
     @NotNull
     private OffsetDateTime lastUpdate;
 
+    private List<Reminder> reminders = new ArrayList<>();
+
     @NotNull
     private OffsetDateTime createdAt;
 
@@ -53,5 +57,6 @@ public Note(NoteDTO noteDTO) {
         this.description = noteDTO.getDescription();
         this.createdAt = CommonFunctions.getCurrentDatetime();
         this.lastUpdate = CommonFunctions.getCurrentDatetime();
+        this.reminders = new ArrayList<>();
     }
 }

src/main/java/com/rm/mynotes/model/Notification.java

@@ -0,0 +1,34 @@
+package com.rm.mynotes.model;
+
+import com.rm.mynotes.utils.constants.StatusTypes;
+import jakarta.persistence.*;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.NoArgsConstructor;
+
+import javax.validation.constraints.NotNull;
+import java.time.OffsetDateTime;
+import java.util.Set;
+
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@Entity
+public class Notification {
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    @NotNull
+    private OffsetDateTime createdAt;
+
+    private Boolean wasRead = false;
+
+    private Set<String> content;
+
+    @OneToMany(fetch = FetchType.LAZY, cascade = CascadeType.ALL, orphanRemoval = true)
+    @JoinTable(name = "notification_reminder", joinColumns = @JoinColumn(name = "notification_id", referencedColumnName = "id"),
+            inverseJoinColumns = @JoinColumn(name = "reminder_id", referencedColumnName = "id")
+    )
+    private Reminder reminder;
+}

src/main/java/com/rm/mynotes/model/Reminder.java

@@ -0,0 +1,35 @@
+package com.rm.mynotes.model;
+
+import com.rm.mynotes.utils.constants.StatusTypes;
+import jakarta.persistence.*;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.NoArgsConstructor;
+
+import javax.validation.constraints.NotNull;
+import java.time.OffsetDateTime;
+
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@Entity
+public class Reminder {
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    @NotNull
+    private String title;
+
+    @NotNull
+    private OffsetDateTime createdAt;
+
+    @NotNull
+    private OffsetDateTime lastUpdate;
+
+    @NotNull
+    private OffsetDateTime reminderDate;
+
+    @Enumerated(EnumType.STRING)
+    private StatusTypes status;
+}

src/main/java/com/rm/mynotes/repository/NotificationRepository.java

@@ -0,0 +1,9 @@
+package com.rm.mynotes.repository;
+
+import com.rm.mynotes.model.Notification;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface NotificationRepository extends JpaRepository<Notification, Long> {
+}

src/main/java/com/rm/mynotes/repository/ReminderRepository.java

@@ -0,0 +1,9 @@
+package com.rm.mynotes.repository;
+
+import com.rm.mynotes.model.Reminder;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface ReminderRepository extends JpaRepository<Reminder, Long> {
+}

src/main/java/com/rm/mynotes/resource/ReminderResource.java

@@ -0,0 +1,28 @@
+package com.rm.mynotes.resource;
+
+import com.rm.mynotes.model.Reminder;
+import com.rm.mynotes.service.mold.ReminderService;
+import com.rm.mynotes.utils.dto.requests.ReminderDTO;
+import lombok.RequiredArgsConstructor;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.ResponseEntity;
+import org.springframework.security.core.Authentication;
+import org.springframework.web.bind.annotation.*;
+
+@RestController
+@RequiredArgsConstructor
+@RequestMapping
+public class ReminderResource {
+    @Autowired
+    private ReminderService reminderService;
+
+    @GetMapping("/app/reminder/1")
+    public String test() {
+        return "você fez uma requisição!";
+    }
+
+    @PostMapping("/app/reminder/{noteId}")
+    public ResponseEntity<Reminder> createReminder(Authentication authentication, @RequestBody ReminderDTO reminderDTO, @RequestParam(required = true, name = "noteId") Long noteId) {
+        return reminderService.createReminder(authentication, reminderDTO, noteId);
+    }
+}

src/main/java/com/rm/mynotes/resource/WebSocketResource.java

@@ -0,0 +1,21 @@
+package com.rm.mynotes.resource;
+
+import com.rm.mynotes.model.Reminder;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.messaging.handler.annotation.MessageMapping;
+import org.springframework.messaging.handler.annotation.SendTo;
+import org.springframework.stereotype.Controller;
+
+@Controller
+public class WebSocketResource {
+    @Autowired
+    private KafkaTemplate<String, String> kafkaTemplate;
+
+    @MessageMapping("/createReminder")
+    @SendTo("/topic/reminders")
+    public Reminder createEvent(Reminder reminder) {
+        kafkaTemplate.send("remindersTopic", reminder.toString());
+        return reminder;
+    }
+}

src/main/java/com/rm/mynotes/service/impl/ReminderServiceImplementation.java

@@ -0,0 +1,71 @@
+package com.rm.mynotes.service.impl;
+
+import com.rm.mynotes.model.Note;
+import com.rm.mynotes.model.Reminder;
+import com.rm.mynotes.model.UserEntity;
+import com.rm.mynotes.repository.NoteRepository;
+import com.rm.mynotes.repository.ReminderRepository;
+import com.rm.mynotes.repository.UserRepository;
+import com.rm.mynotes.service.mold.ReminderService;
+import com.rm.mynotes.utils.constants.RoutePaths;
+import com.rm.mynotes.utils.constants.StatusTypes;
+import com.rm.mynotes.utils.dto.payloads.ResponseDTO;
+import com.rm.mynotes.utils.dto.requests.ReminderDTO;
+import com.rm.mynotes.utils.functions.CommonFunctions;
+import lombok.RequiredArgsConstructor;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.ResponseEntity;
+import org.springframework.security.authentication.BadCredentialsException;
+import org.springframework.security.core.Authentication;
+import org.springframework.stereotype.Service;
+import org.springframework.web.bind.annotation.ExceptionHandler;
+import org.springframework.web.servlet.support.ServletUriComponentsBuilder;
+
+import java.net.URI;
+import java.util.List;
+import java.util.Objects;
+
+@Service
+@RequiredArgsConstructor
+public class ReminderServiceImplementation implements ReminderService {
+    @Autowired
+    private UserRepository userRepository;
+
+    @Autowired
+    private NoteRepository noteRepository;
+
+    @Autowired
+    private ReminderRepository reminderRepository;
+
+    @Autowired
+    private CommonFunctions commonFunctions;
+
+    @Override
+    public ResponseEntity<Reminder> createReminder(Authentication authentication, ReminderDTO reqReminder, Long noteId) {
+        UserEntity user = commonFunctions.getCurrentUser(authentication);
+        Note note = user.getNotes().stream().filter(userNote -> Objects.equals(userNote.getId(), noteId))
+                .findFirst().orElseThrow(() -> new BadCredentialsException("A anotação informada não existe."));
+
+        Reminder reminder = Reminder.builder()
+                .reminderDate(reqReminder.getReminderDate())
+                .createdAt(CommonFunctions.getCurrentDatetime())
+                .lastUpdate(CommonFunctions.getCurrentDatetime())
+                .status(StatusTypes.PENDING)
+                .build();
+
+        List<Reminder> reminders = note.getReminders();
+        reminders.add(reminderRepository.save(reminder));
+
+        note.setReminders(reminders);
+        noteRepository.save(note);
+
+        URI uri = URI.create(ServletUriComponentsBuilder.fromCurrentContextPath().path(RoutePaths.REMINDER).toUriString());
+
+        return ResponseEntity.created(uri).body(reminder);
+    }
+
+    @ExceptionHandler(Exception.class)
+    public ResponseEntity<ResponseDTO> handleException(Exception exception) {
+        return CommonFunctions.errorHandling(exception);
+    }
+}

src/main/java/com/rm/mynotes/service/mold/ReminderService.java

@@ -0,0 +1,10 @@
+package com.rm.mynotes.service.mold;
+
+import com.rm.mynotes.model.Reminder;
+import com.rm.mynotes.utils.dto.requests.ReminderDTO;
+import org.springframework.http.ResponseEntity;
+import org.springframework.security.core.Authentication;
+
+public interface ReminderService {
+    ResponseEntity<Reminder> createReminder(Authentication authentication, ReminderDTO reqReminder, Long noteId);
+}

src/main/java/com/rm/mynotes/utils/config/KafkaConsumerConfig.java

@@ -0,0 +1,27 @@
+package com.rm.mynotes.utils.config;
+
+import com.rm.mynotes.model.Notification;
+import com.rm.mynotes.model.Reminder;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.messaging.simp.SimpMessagingTemplate;
+
+@Configuration
+@EnableKafka
+public class KafkaConsumerConfig {
+    @Autowired
+    private SimpMessagingTemplate messagingTemplate;
+
+    @KafkaListener(topics = "remindersTopic", groupId = "group1")
+    public void listen(String message) {
+        Reminder reminder = parseReminder(message);
+        Notification notification = new Notification();
+        messagingTemplate.convertAndSend("/app/topic/notifications", notification);
+    }
+
+    private Reminder parseReminder(String message) {
+        return new Reminder();
+    }
+}

src/main/java/com/rm/mynotes/utils/config/KafkaProducerConfig.java

@@ -0,0 +1,32 @@
+package com.rm.mynotes.utils.config;
+
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaProducerConfig {
+    @Value("${spring.kafka.bootstrap-servers}")
+    private String bootstrapServers;
+
+    @Bean
+    public ProducerFactory<String, String> producerFactory() {
+        Map<String, Object> configProps = new HashMap<>();
+        configProps.put(org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+        configProps.put(org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        configProps.put(org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        return new DefaultKafkaProducerFactory<>(configProps);
+    }
+
+    @Bean
+    public KafkaTemplate<String, String> kafkaTemplate() {
+        return new KafkaTemplate<>(producerFactory());
+    }
+}

src/main/java/com/rm/mynotes/utils/config/WebSocketConfig.java

@@ -0,0 +1,22 @@
+package com.rm.mynotes.utils.config;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.simp.config.MessageBrokerRegistry;
+import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
+import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
+import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
+
+@Configuration
+@EnableWebSocketMessageBroker
+public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
+    @Override
+    public void configureMessageBroker(MessageBrokerRegistry config) {
+        config.enableSimpleBroker("/topic");
+        config.setApplicationDestinationPrefixes("/app");
+    }
+
+    @Override
+    public void registerStompEndpoints(StompEndpointRegistry registry) {
+        registry.addEndpoint("/websocket").withSockJS();
+    }
+}