Merge pull request #110 from kjiyun/chat

kjiyun · web-flow · commit 29e598d16e6e · 2025-02-14T17:12:54.000+09:00
diff --git a/src/main/java/com/memesphere/domain/chartdata/repository/ChartDataRepository.java b/src/main/java/com/memesphere/domain/chartdata/repository/ChartDataRepository.java
@@ -30,8 +30,7 @@ public interface ChartDataRepository extends JpaRepository<ChartData, Long> {
 
     List<ChartData> findByMemeCoinOrderByRecordedTimeDesc(MemeCoin memeCoin);
 
-    //TODO: 위아래 코드 합치는 방법 찾기
-    List<ChartData> findByMemeCoinOrderByRecordedTimeDesc(MemeCoin memeCoin, Pageable pageable);
+    List<ChartData> findByMemeCoinAndRecordedTimeAfterOrderByRecordedTimeDesc(MemeCoin memeCoin, LocalDateTime recordedTime, Pageable pageable);
 
     @Query("SELECT c FROM ChartData c " +
             "WHERE c.memeCoin.id = :coinId " +
diff --git a/src/main/java/com/memesphere/domain/chartdata/scheduler/ChartDataScheduler.java b/src/main/java/com/memesphere/domain/chartdata/scheduler/ChartDataScheduler.java
@@ -8,8 +8,11 @@
 import com.memesphere.domain.memecoin.entity.MemeCoin;
 import com.memesphere.domain.memecoin.repository.MemeCoinRepository;
 import com.memesphere.domain.memecoin.service.MemeCoinQueryService;
+import com.memesphere.global.jwt.CustomUserDetails;
 import lombok.RequiredArgsConstructor;
 import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.security.core.Authentication;
+import org.springframework.security.core.context.SecurityContextHolder;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -24,9 +27,18 @@ public class ChartDataScheduler {
     private final BinanceQueryService binanceQueryService;
     private final MemeCoinQueryService memeCoinQueryService;
 
-    @Scheduled(cron = "0 0/10 * * * ?") // 0, 10, 20, 30, 40, 50분에 실행
+    @Scheduled(cron = "0 0/1 * * * ?") // 0, 10, 20, 30, 40, 50분에 실행
     @Transactional
     public void updateChartData() {
+
+        Long userId = null;
+
+        Authentication authentication = SecurityContextHolder.getContext().getAuthentication();
+        // `authentication.getPrincipal()`을 `CustomUserDetails`로 변환
+        if (authentication != null && authentication.getPrincipal() instanceof CustomUserDetails) {
+            CustomUserDetails userDetails = (CustomUserDetails) authentication.getPrincipal();
+            userId = userDetails.getUser().getId();
+        }
         List<MemeCoin> memeCoins = memeCoinRepository.findAll();
 
         for (MemeCoin memeCoin : memeCoins) {
@@ -36,7 +48,7 @@ public void updateChartData() {
 
                 ChartData chartData = toChartData(memeCoin,response);
 
-                memeCoinQueryService.updateChartData(memeCoin.getId(), chartData);
+                memeCoinQueryService.updateChartData(memeCoin.getId(), chartData, userId);
 
             } catch (Exception e) {
                 throw new GeneralException(ErrorStatus.CANNOT_LOAD_CHARTDATA);
diff --git a/src/main/java/com/memesphere/domain/chartdata/service/ChartDataQueryService.java b/src/main/java/com/memesphere/domain/chartdata/service/ChartDataQueryService.java
@@ -4,5 +4,5 @@
 import com.memesphere.domain.memecoin.entity.MemeCoin;
 
 public interface ChartDataQueryService {
-    void saveChartData(MemeCoin memeCoin, ChartData chartData);
+    void saveChartData(MemeCoin memeCoin, ChartData chartData, Long userId);
 }
diff --git a/src/main/java/com/memesphere/domain/chartdata/service/ChartDataQueryServiceImpl.java b/src/main/java/com/memesphere/domain/chartdata/service/ChartDataQueryServiceImpl.java
@@ -3,6 +3,7 @@
 import com.memesphere.domain.chartdata.entity.ChartData;
 import com.memesphere.domain.chartdata.repository.ChartDataRepository;
 import com.memesphere.domain.memecoin.entity.MemeCoin;
+import com.memesphere.domain.notification.service.PushNotificationService;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -12,12 +13,13 @@
 @Service
 @RequiredArgsConstructor
 public class ChartDataQueryServiceImpl implements ChartDataQueryService {
+    private final PushNotificationService pushNotificationService;
     private final ChartDataRepository chartDataRepository;
     private static final int MAX_CHART_DATA_CNT = 6;
 
     @Override
     @Transactional
-    public void saveChartData(MemeCoin memeCoin, ChartData newChartData) {
+    public void saveChartData(MemeCoin memeCoin, ChartData newChartData, Long userId) {
         List<ChartData> chartDataList = chartDataRepository.findByMemeCoinOrderByRecordedTimeDesc(memeCoin);
 
         if (chartDataList.size() >= MAX_CHART_DATA_CNT) {
@@ -28,5 +30,7 @@ public void saveChartData(MemeCoin memeCoin, ChartData newChartData) {
 
         newChartData.setMemeCoin(memeCoin);
         chartDataRepository.save(newChartData);
+
+        pushNotificationService.send(userId);
     }
 }
diff --git a/src/main/java/com/memesphere/domain/memecoin/service/MemeCoinQueryService.java b/src/main/java/com/memesphere/domain/memecoin/service/MemeCoinQueryService.java
@@ -3,5 +3,5 @@
 import com.memesphere.domain.chartdata.entity.ChartData;
 
 public interface MemeCoinQueryService {
-    void updateChartData(Long memeCoinId, ChartData newChartData);
+    void updateChartData(Long memeCoinId, ChartData newChartData, Long userId);
 }
diff --git a/src/main/java/com/memesphere/domain/memecoin/service/MemeCoinQueryServiceImpl.java b/src/main/java/com/memesphere/domain/memecoin/service/MemeCoinQueryServiceImpl.java
@@ -18,10 +18,10 @@ public class MemeCoinQueryServiceImpl implements MemeCoinQueryService {
 
     @Transactional
     @Override
-    public void updateChartData(Long memeCoinId, ChartData newChartData) {
+    public void updateChartData(Long memeCoinId, ChartData newChartData, Long userId) {
         MemeCoin memeCoin = memeCoinRepository.findById(memeCoinId)
                 .orElseThrow(() -> new GeneralException(ErrorStatus.MEMECOIN_NOT_FOUND));
 
-        chartDataService.saveChartData(memeCoin, newChartData);
+        chartDataService.saveChartData(memeCoin, newChartData, userId);
     }
 }
diff --git a/src/main/java/com/memesphere/domain/notification/service/PushNotificationService.java b/src/main/java/com/memesphere/domain/notification/service/PushNotificationService.java
@@ -1,9 +1,8 @@
 package com.memesphere.domain.notification.service;
 
-import com.memesphere.domain.notification.entity.Notification;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
 public interface PushNotificationService {
     SseEmitter subscribe(Long userId, String lastEventId);
-    void send(Notification notification, Long userId);
+    void send(Long userId);
 }
diff --git a/src/main/java/com/memesphere/domain/notification/service/PushNotificationServiceImpl.java b/src/main/java/com/memesphere/domain/notification/service/PushNotificationServiceImpl.java
@@ -11,6 +11,7 @@
 import com.memesphere.global.apipayload.ApiResponse;
 import com.memesphere.global.apipayload.code.status.ErrorStatus;
 import com.memesphere.global.apipayload.exception.GeneralException;
+import com.memesphere.global.jwt.CustomUserDetails;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
 import org.springframework.data.domain.PageRequest;
@@ -24,6 +25,8 @@
 import java.math.RoundingMode;
 import java.time.LocalDateTime;
 import java.util.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.stream.Collectors;
 
 @Log4j2
@@ -34,6 +37,7 @@ public class PushNotificationServiceImpl implements PushNotificationService {
     private final EmitterRepository emitterRepository;
     private final NotificationRepository notificationRepository;
     private final ChartDataRepository chartDataRepository;
+    private final ExecutorService executorService = Executors.newSingleThreadExecutor();
 
     // 연결 지속 시간 설정 : 한시간
     private static final Long DEFAULT_TIMEOUT = 60L * 1000 * 60;
@@ -61,18 +65,6 @@ public SseEmitter subscribe(Long userId, String lastEventId) {
                     .forEach(entry -> sendToClient(emitter, entry.getKey(), entry.getValue()));
         }
 
-        List<Notification> notifications = notificationRepository.findAllByUserId(userId); // 사용자가 등록한 알림 전부 가져오기
-
-        // 변동성을 초과하는 알림 필터링
-        List<Notification> filteredNotifications = notifications.stream()
-                .filter(notification -> isVolatilityExceeded(notification))
-                .collect(Collectors.toList());
-
-        if (!filteredNotifications.isEmpty()) {
-            notifications.forEach(notification -> {
-                send(notification, userId);
-            });
-        }
         return emitter;
     }
 
@@ -91,15 +83,35 @@ private void sendToClient(SseEmitter emitter, String emitterId, Object data) {
     }
 
     @Override
-    public void send(Notification notification, Long userId) {
+    public void send(Long userId) {
 
-        // 실시간 알림 전송 - 로그인 한 유저의 SseEmitter 모두 가져오기
-        Map<String, SseEmitter> sseEmitters = emitterRepository.findAllEmitterStartWithByUserId(String.valueOf(userId));
+        List<Notification> notifications = notificationRepository.findAllByUserId(userId); // 사용자가 등록한 알림 전부 가져오기
 
-        sseEmitters.forEach((key, emitter) -> {
-                emitterRepository.saveEventCache(key, notification);
-                sendToClient(emitter, key, NotificationConverter.toNotificationCreateResponse(notification, notification.getMemeCoin()));
-        });
+        // 변동성을 초과하는 알림 필터링
+        List<Notification> filteredNotifications = notifications.stream()
+                .filter(notification -> isVolatilityExceeded(notification))
+                .collect(Collectors.toList());
+
+        if (!filteredNotifications.isEmpty()) {
+            // 실시간 알림 전송 - 로그인 한 유저의 SseEmitter 모두 가져오기
+            Map<String, SseEmitter> sseEmitters = emitterRepository.findAllEmitterStartWithByUserId(String.valueOf(userId));
+
+            sseEmitters.forEach((key, emitter) -> {
+                executorService.submit(() -> {
+                    filteredNotifications.forEach(notification -> {
+
+                        emitterRepository.saveEventCache(key, notification);
+                        sendToClient(emitter, key, NotificationConverter.toNotificationCreateResponse(notification, notification.getMemeCoin()));
+
+                        try {
+                            Thread.sleep(500); // 0.5초 간격으로 전송
+                        } catch (InterruptedException e) {
+                            Thread.currentThread().interrupt();
+                        }
+                    });
+                });
+            });
+        }
     }
 
     private boolean isVolatilityExceeded(Notification notification) {
@@ -113,18 +125,20 @@ private boolean isVolatilityExceeded(Notification notification) {
             throw new GeneralException(ErrorStatus.CANNOT_LOAD_CHARTDATA);
         }
 
+        LocalDateTime notificationTime = notification.getCreatedAt();
+
         Integer count = notification.getStTime() / 10; //몇 번 가져올 것인지 결정
         Pageable pageable = (Pageable) PageRequest.of(0, count, Sort.by(Sort.Direction.DESC, "createdAt"));
-        List<ChartData> lastNData = chartDataRepository.findByMemeCoinOrderByRecordedTimeDesc(memeCoin, pageable);
+        List<ChartData> lastNData = chartDataRepository.findByMemeCoinAndRecordedTimeAfterOrderByRecordedTimeDesc(memeCoin, notificationTime, pageable);
 
         if (lastNData.size() < count) {
             return false; // 비교할 데이터가 부족하면 알림을 보내지 않음
         }
 
         BigDecimal sum = lastNData.stream()
-                .map(ChartData::getPrice)
+                .map(ChartData::getPriceChangeRate)
                 .reduce(BigDecimal.ZERO, BigDecimal::add);
-        BigDecimal average = sum.divide(BigDecimal.valueOf(lastNData.size()), 4, RoundingMode.HALF_UP);
+        BigDecimal average = sum.divide(BigDecimal.valueOf(count), 4, RoundingMode.HALF_UP);
         BigDecimal definedVolatility = new BigDecimal(notification.getVolatility());
 
         if (notification.getIsRising()) { // 상승인 경우
@@ -133,5 +147,4 @@ private boolean isVolatilityExceeded(Notification notification) {
             return average.compareTo(definedVolatility) < 0;
         }
     }
-
 }

Original file line number	Diff line number	Diff line change
`@@ -4,5 +4,5 @@`
`4`	`4`	`import com.memesphere.domain.memecoin.entity.MemeCoin;`
`5`	`5`
`6`	`6`	`public interface ChartDataQueryService {`
`7`		`- void saveChartData(MemeCoin memeCoin, ChartData chartData);`
	`7`	`+ void saveChartData(MemeCoin memeCoin, ChartData chartData, Long userId);`
`8`	`8`	`}`
Original file line number	Diff line number	Diff line change
`@@ -3,5 +3,5 @@`
`3`	`3`	`import com.memesphere.domain.chartdata.entity.ChartData;`
`4`	`4`
`5`	`5`	`public interface MemeCoinQueryService {`
`6`		`- void updateChartData(Long memeCoinId, ChartData newChartData);`
	`6`	`+ void updateChartData(Long memeCoinId, ChartData newChartData, Long userId);`
`7`	`7`	`}`
Original file line number	Diff line number	Diff line change
`@@ -18,10 +18,10 @@ public class MemeCoinQueryServiceImpl implements MemeCoinQueryService {`
`18`	`18`
`19`	`19`	`@Transactional`
`20`	`20`	`@Override`
`21`		`- public void updateChartData(Long memeCoinId, ChartData newChartData) {`
	`21`	`+ public void updateChartData(Long memeCoinId, ChartData newChartData, Long userId) {`
`22`	`22`	`MemeCoin memeCoin = memeCoinRepository.findById(memeCoinId)`
`23`	`23`	`.orElseThrow(() -> new GeneralException(ErrorStatus.MEMECOIN_NOT_FOUND));`
`24`	`24`
`25`		`- chartDataService.saveChartData(memeCoin, newChartData);`
	`25`	`+ chartDataService.saveChartData(memeCoin, newChartData, userId);`
`26`	`26`	`}`
`27`	`27`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,9 +1,8 @@`
`1`	`1`	`package com.memesphere.domain.notification.service;`
`2`	`2`
`3`		`-import com.memesphere.domain.notification.entity.Notification;`
`4`	`3`	`import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;`
`5`	`4`
`6`	`5`	`public interface PushNotificationService {`
`7`	`6`	`SseEmitter subscribe(Long userId, String lastEventId);`
`8`		`- void send(Notification notification, Long userId);`
	`7`	`+ void send(Long userId);`
`9`	`8`	`}`