[Feat] SSE 알림 서비스 구현

src/main/java/com/kuit/agarang/domain/ai/service/MusicGenService.java

@@ -5,6 +5,7 @@
 import com.kuit.agarang.domain.ai.utils.MusicGenClientUtil;
 import com.kuit.agarang.domain.memory.model.entity.Memory;
 import com.kuit.agarang.domain.memory.repository.MemoryRepository;
+import com.kuit.agarang.domain.notification.model.SseEmitters;
 import com.kuit.agarang.global.common.exception.exception.OpenAPIException;
 import com.kuit.agarang.global.common.model.dto.BaseResponseStatus;
 import com.kuit.agarang.global.s3.model.dto.S3File;
@@ -28,14 +29,15 @@ public class MusicGenService {
   private final MusicGenClientUtil musicGenClientUtil;
   private final S3Util s3Util;
   private final MemoryRepository memoryRepository;
+  private final SseEmitters sseEmitters;
 
   private static final String WEBHOOK_URI = "/api/ai/music-gen/webhook";
   private static final Integer MUSIC_DURATION = 40;
 
   public String getMusic(String prompt) {
     MusicGenResponse response
-      = musicGenClientUtil.post(
-      new MusicGenRequest(version, baseUrl + WEBHOOK_URI, prompt, MUSIC_DURATION), MusicGenResponse.class);
+        = musicGenClientUtil.post(
+        new MusicGenRequest(version, baseUrl + WEBHOOK_URI, prompt, MUSIC_DURATION), MusicGenResponse.class);
     log.info("created musicgen id : {}", response.getId());
     return response.getId();
   }
@@ -53,6 +55,9 @@ public void saveMusic(MusicGenResponse response) {
     Memory memory = optionalMemory.get();
     memory.setMusicUrl(s3File.getObjectUrl());
     memoryRepository.save(memory);
+
+    String message = "음악 생성이 완료되었습니다";
+    sseEmitters.sendNotification(message);
   }
 
   private static void checkStatus(MusicGenResponse response) {

src/main/java/com/kuit/agarang/domain/notification/controller/SseController.java

@@ -0,0 +1,36 @@
+package com.kuit.agarang.domain.notification.controller;
+
+import com.kuit.agarang.domain.notification.model.SseEmitters;
+import com.kuit.agarang.global.common.exception.exception.BusinessException;
+import com.kuit.agarang.global.common.model.dto.BaseResponseStatus;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.MediaType;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.io.IOException;
+
+@Slf4j
+@RestController
+@RequiredArgsConstructor
+public class SseController {
+
+  private final SseEmitters sseEmitters;
+
+  @GetMapping(value = "/connect", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+  public SseEmitter connect() {
+    SseEmitter emitter = new SseEmitter(10 * 60 * 1000L);
+    sseEmitters.add(emitter);
+    try {
+      emitter.send(SseEmitter.event()
+          .name("connect")
+          .data("connected!")); // 503에러 방지를 위한 더미데이터
+    } catch (IOException e) {
+      emitter.completeWithError(e);
+      throw new BusinessException(BaseResponseStatus.FAIL_CREATE_EMITTER);
+    }
+    return emitter;
+  }
+}

src/main/java/com/kuit/agarang/domain/notification/model/SseEmitters.java

@@ -0,0 +1,47 @@
+package com.kuit.agarang.domain.notification.model;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+@Slf4j
+@Component
+public class SseEmitters {
+  private final List<SseEmitter> emitters = new CopyOnWriteArrayList<>();
+
+  public SseEmitter add(SseEmitter emitter) {
+    this.emitters.add(emitter);
+    log.info("new emitter added: {}", emitter);
+    log.info("emitter list size: {}", emitters.size());
+
+    emitter.onCompletion(() -> {
+      log.info("onCompletion callback");
+      this.emitters.remove(emitter);
+    });
+
+    emitter.onTimeout(() -> {
+      log.info("onTimeout callback");
+      emitter.complete();
+    });
+
+    return emitter;
+  }
+
+  public void sendNotification(String message) {
+    emitters.forEach(emitter -> {
+      try {
+        emitter.send(SseEmitter.event()
+            .name("notification")
+            .data(message));
+        log.info("Notification sent: {}", message);
+      } catch (IOException e) {
+        log.error("Error sending notification: {}", e.getMessage());
+        emitter.complete();
+      }
+    });
+  }
+}

src/main/java/com/kuit/agarang/global/common/model/dto/BaseResponseStatus.java

@@ -39,8 +39,8 @@ public enum BaseResponseStatus {
   FAIL_REDIS_CONNECTION(false, HttpStatus.SERVICE_UNAVAILABLE, 5002, "레디스 서버에 연결 실패했습니다."),
   FAIL_S3_UPLOAD(false, HttpStatus.SERVICE_UNAVAILABLE, 5003, "S3 파일 서버 업로드에 실패했습니다."),
   INVALID_GPT_RESPONSE(false, HttpStatus.INTERNAL_SERVER_ERROR, 5004, "GPT 응답이 유효하지 않습니다."),
-  FAIL_CREATE_MUSIC(false, HttpStatus.INTERNAL_SERVER_ERROR, 5005, "music gen 음악 생성에 실패했습니다.");
-
+  FAIL_CREATE_MUSIC(false, HttpStatus.INTERNAL_SERVER_ERROR, 5005, "music gen 음악 생성에 실패했습니다."),
+  FAIL_CREATE_EMITTER(false, HttpStatus.INTERNAL_SERVER_ERROR, 5006, "SseEmitter 생성에 실패했습니다.");
 
   private final boolean isSuccess;
   @JsonIgnore