Feature/ip 98 infrastruktura sse powiadomien

backend/project-manager/src/main/java/pl/edu/agh/project_manager/ProjectManagerApplication.java

@@ -2,8 +2,10 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.scheduling.annotation.EnableScheduling;
 
 @SpringBootApplication
+@EnableScheduling
 public class ProjectManagerApplication {
 
 	public static void main(String[] args) {

backend/project-manager/src/main/java/pl/edu/agh/project_manager/controller/notification/NotificationController.java

@@ -3,12 +3,15 @@
 import lombok.RequiredArgsConstructor;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.web.PageableDefault;
+import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.security.access.prepost.PreAuthorize;
 import org.springframework.security.core.annotation.AuthenticationPrincipal;
 import org.springframework.web.bind.annotation.*;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 import pl.edu.agh.project_manager.controller.dto.PagedResponse;
 import pl.edu.agh.project_manager.controller.dto.notification.NotificationResponse;
+import pl.edu.agh.project_manager.infrastructure.notification.SseSubscriber;
 import pl.edu.agh.project_manager.security.UserPrincipal;
 import pl.edu.agh.project_manager.service.notification.NotificationService;
 
@@ -20,6 +23,13 @@
 public class NotificationController {
 
     private final NotificationService notificationService;
+    private final SseSubscriber sseSubscriber;
+
+    @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+    public ResponseEntity<SseEmitter> streamNotifications(@AuthenticationPrincipal UserPrincipal userPrincipal) {
+        SseEmitter emitter = sseSubscriber.subscribe(userPrincipal.userId());
+        return ResponseEntity.ok(emitter);
+    }
 
     @GetMapping
     public ResponseEntity<PagedResponse<NotificationResponse>> getUserNotifications(

backend/project-manager/src/main/java/pl/edu/agh/project_manager/infrastructure/notification/SseNotificationSender.java

@@ -0,0 +1,65 @@
+package pl.edu.agh.project_manager.infrastructure.notification;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+import pl.edu.agh.project_manager.controller.dto.notification.NotificationResponse;
+import pl.edu.agh.project_manager.service.notification.NotificationSender;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Slf4j
+@Component
+public class SseNotificationSender implements NotificationSender, SseSubscriber {
+
+    private final Map<UUID, SseEmitter> emitters = new ConcurrentHashMap<>();
+
+    @Override
+    public SseEmitter subscribe(UUID userId) {
+        SseEmitter emitter = new SseEmitter(3600000L);
+
+        emitters.put(userId, emitter);
+        log.info("User {} subscribed to SSE notifications", userId);
+
+        emitter.onCompletion(() -> {
+            emitters.remove(userId);
+            log.info("SSE connection completed for user {}", userId);
+        });
+        emitter.onTimeout(() -> {
+            emitters.remove(userId);
+            log.info("SSE connection timed out for user {}", userId);
+        });
+        emitter.onError((e) -> {
+            emitters.remove(userId);
+            log.error("SSE connection error for user {}: {}", userId, e.getMessage());
+        });
+
+        try {
+            emitter.send(SseEmitter.event().name("INIT").data("Connected"));
+        } catch (IOException e) {
+            emitters.remove(userId);
+        }
+
+        return emitter;
+    }
+
+    @Override
+    public void send(NotificationResponse notification, UUID recipientId) {
+        SseEmitter emitter = emitters.get(recipientId);
+
+        if (emitter != null) {
+            try {
+                emitter.send(SseEmitter.event()
+                        .name("NOTIFICATION")
+                        .data(notification));
+                log.info("Push notification sent to user {}", recipientId);
+            } catch (IOException e) {
+                log.warn("Failed to send notification to user {}. Removing dead emitter.", recipientId);
+                emitters.remove(recipientId);
+            }
+        }
+    }
+}

backend/project-manager/src/main/java/pl/edu/agh/project_manager/infrastructure/notification/SseSubscriber.java

@@ -0,0 +1,9 @@
+package pl.edu.agh.project_manager.infrastructure.notification;
+
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.util.UUID;
+
+public interface SseSubscriber {
+    SseEmitter subscribe(UUID userId);
+}

backend/project-manager/src/main/java/pl/edu/agh/project_manager/infrastructure/scheduler/NotificationCleanupScheduler.java

@@ -0,0 +1,21 @@
+package pl.edu.agh.project_manager.infrastructure.scheduler;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import pl.edu.agh.project_manager.service.notification.NotificationService;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class NotificationCleanupScheduler {
+
+    private final NotificationService notificationService;
+
+    @Scheduled(cron = "0 0 3 * * *")
+    public void scheduleNotificationCleanup() {
+        log.info("Czyszczenia starych powiadomień...");
+        notificationService.cleanupOldReadNotifications();
+    }
+}

backend/project-manager/src/main/java/pl/edu/agh/project_manager/repository/notification/NotificationRepository.java

@@ -8,6 +8,7 @@
 import org.springframework.data.repository.query.Param;
 import pl.edu.agh.project_manager.domain.entity.notification.Notification;
 
+import java.time.LocalDateTime;
 import java.util.List;
 import java.util.UUID;
 
@@ -19,7 +20,11 @@ public interface NotificationRepository extends JpaRepository<Notification, UUID
 
     Page<Notification> findAllByRecipientIdAndIsReadFalseOrderByCreatedAtDesc(UUID recipientId, Pageable pageable);
 
-    @Modifying
+    @Modifying(clearAutomatically = true, flushAutomatically = true)
     @Query("UPDATE Notification n SET n.isRead = true WHERE n.recipient.id = :userId AND n.isRead = false")
     void markAllAsReadByUserId(@Param("userId") UUID userId);
+
+    @Modifying(clearAutomatically = true, flushAutomatically = true)
+    @Query("DELETE FROM Notification n WHERE n.isRead = true AND n.createdAt < :thresholdDate")
+    int deleteReadAndOlderThan(@Param("thresholdDate") LocalDateTime thresholdDate);
 }

backend/project-manager/src/main/java/pl/edu/agh/project_manager/service/notification/NotificationSender.java

@@ -0,0 +1,9 @@
+package pl.edu.agh.project_manager.service.notification;
+
+import pl.edu.agh.project_manager.controller.dto.notification.NotificationResponse;
+
+import java.util.UUID;
+
+public interface NotificationSender {
+    void send(NotificationResponse notification, UUID recipientId);
+}

backend/project-manager/src/main/java/pl/edu/agh/project_manager/service/notification/NotificationService.java

@@ -1,6 +1,7 @@
 package pl.edu.agh.project_manager.service.notification;
 
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
@@ -18,14 +19,17 @@
 import pl.edu.agh.project_manager.domain.exception.ApplicationException;
 import pl.edu.agh.project_manager.repository.notification.NotificationRepository;
 
+import java.time.LocalDateTime;
 import java.util.List;
 import java.util.UUID;
 
 @Service
+@Slf4j
 @RequiredArgsConstructor
 public class NotificationService {
 
     private final NotificationRepository notificationRepository;
+    private final NotificationSender notificationSender;
 
     @TransactionalEventListener
     @Transactional(propagation = Propagation.REQUIRES_NEW)
@@ -40,7 +44,11 @@ private void create(User recipient, NotificationType type, String message, UUID
                 .message(message)
                 .referenceId(referenceId)
                 .build();
-        notificationRepository.save(notification);
+
+        Notification savedNotification = notificationRepository.save(notification);
+
+        NotificationResponse response = NotificationResponse.from(savedNotification);
+        notificationSender.send(response, recipient.getId());
     }
 
     @Transactional(readOnly = true)
@@ -73,4 +81,15 @@ public void markAsRead(UUID notificationId, UUID userId) {
     public void markAllAsRead(UUID userId) {
         notificationRepository.markAllAsReadByUserId(userId);
     }
+
+    @Transactional
+    public void cleanupOldReadNotifications() {
+        LocalDateTime thirtyDaysAgo = LocalDateTime.now().minusDays(30);
+
+        int deletedCount = notificationRepository.deleteReadAndOlderThan(thirtyDaysAgo);
+
+        if (deletedCount > 0) {
+            log.info("Zakończono czyszczenie powiadomień. Usunięto {} starych, przeczytanych rekordów.", deletedCount);
+        }
+    }
 }

backend/project-manager/src/test/java/pl/edu/agh/project_manager/repository/notification/NotificationRepositoryTest.java

@@ -0,0 +1,115 @@
+package pl.edu.agh.project_manager.repository.notification;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.data.jpa.test.autoconfigure.DataJpaTest;
+import org.springframework.boot.jdbc.test.autoconfigure.AutoConfigureTestDatabase;
+import org.springframework.jdbc.core.JdbcTemplate;
+import pl.edu.agh.project_manager.domain.entity.notification.Notification;
+import pl.edu.agh.project_manager.domain.entity.user.User;
+import pl.edu.agh.project_manager.domain.enums.NotificationType;
+import pl.edu.agh.project_manager.domain.enums.UserRole;
+import pl.edu.agh.project_manager.domain.enums.UserStatus;
+import pl.edu.agh.project_manager.repository.user.UserRepository;
+
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+@DataJpaTest
+@AutoConfigureTestDatabase(replace = AutoConfigureTestDatabase.Replace.NONE)
+class NotificationRepositoryTest {
+
+    @Autowired
+    private NotificationRepository notificationRepository;
+
+    @Autowired
+    private UserRepository userRepository;
+
+    @Autowired
+    private JdbcTemplate jdbcTemplate;
+
+    private User testUser1;
+    private User testUser2;
+
+    @BeforeEach
+    void setUp() {
+        testUser1 = userRepository.saveAndFlush(User.builder()
+                .email("user1@agh.edu.pl")
+                .name("Jan")
+                .surname("Testowy")
+                .password("hash")
+                .userRole(UserRole.COMMON)
+                .userStatus(UserStatus.ACTIVE)
+                .build());
+
+        testUser2 = userRepository.saveAndFlush(User.builder()
+                .email("user2@agh.edu.pl")
+                .name("Anna")
+                .surname("Inna")
+                .password("hash")
+                .userRole(UserRole.COMMON)
+                .userStatus(UserStatus.ACTIVE)
+                .build());
+    }
+
+    @Test
+    void shouldMarkAllUnreadNotificationsAsReadForSpecificUser() {
+        Notification n1 = saveNotification(testUser1, false);
+        Notification n2 = saveNotification(testUser1, false);
+        Notification n3 = saveNotification(testUser1, true);
+
+        Notification n4 = saveNotification(testUser2, false);
+
+        // When
+        notificationRepository.markAllAsReadByUserId(testUser1.getId());
+
+        // Then
+        assertTrue(notificationRepository.findById(n1.getId()).orElseThrow().isRead());
+        assertTrue(notificationRepository.findById(n2.getId()).orElseThrow().isRead());
+        assertTrue(notificationRepository.findById(n3.getId()).orElseThrow().isRead());
+
+        assertFalse(notificationRepository.findById(n4.getId()).orElseThrow().isRead());
+    }
+
+    @Test
+    void shouldDeleteOnlyReadAndOlderThanThreshold() {
+        // Given
+        LocalDateTime now = LocalDateTime.now();
+        LocalDateTime threshold = now.minusDays(30);
+
+        Notification readAndOld = saveNotification(testUser1, true);
+        ageNotificationInDatabase(readAndOld.getId(), now.minusDays(40));
+
+        Notification readAndNew = saveNotification(testUser1, true);
+        ageNotificationInDatabase(readAndNew.getId(), now.minusDays(10));
+
+        Notification unreadAndOld = saveNotification(testUser1, false);
+        ageNotificationInDatabase(unreadAndOld.getId(), now.minusDays(40));
+
+        // When
+        int deletedRecordsCount = notificationRepository.deleteReadAndOlderThan(threshold);
+
+        // Then
+        assertEquals(1, deletedRecordsCount);
+        assertFalse(notificationRepository.existsById(readAndOld.getId()));
+        assertTrue(notificationRepository.existsById(readAndNew.getId()));
+        assertTrue(notificationRepository.existsById(unreadAndOld.getId()));
+    }
+
+
+    private Notification saveNotification(User recipient, boolean isRead) {
+        return notificationRepository.saveAndFlush(Notification.builder()
+                .recipient(recipient)
+                .type(NotificationType.SYSTEM_NEW_EMPLOYEE)
+                .message("Test")
+                .isRead(isRead)
+                .build());
+    }
+
+    private void ageNotificationInDatabase(UUID notificationId, LocalDateTime newDate) {
+        jdbcTemplate.update("UPDATE notifications SET created_at = ? WHERE id = ?", newDate, notificationId);
+    }
+}