[DDING-51] VodProcessingJob 완료 시 notification 전송 구현

src/main/java/ddingdong/ddingdongBE/common/exception/SseException.java

@@ -0,0 +1,19 @@
+package ddingdong.ddingdongBE.common.exception;
+
+import static org.springframework.http.HttpStatus.INTERNAL_SERVER_ERROR;
+
+sealed public class SseException extends CustomException {
+
+    public static final String SSE_NOT_FOUND_ERROR_MESSAGE = "SSE Not Found = ";
+
+    public SseException(String message, int errorCode) {
+        super(message, errorCode);
+    }
+
+    public static final class SseEmitterNotFound extends SseException {
+
+        public SseEmitterNotFound(String sseId) {
+            super(SSE_NOT_FOUND_ERROR_MESSAGE + sseId, INTERNAL_SERVER_ERROR.value());
+        }
+    }
+}

src/main/java/ddingdong/ddingdongBE/domain/feed/entity/Feed.java

@@ -44,7 +44,8 @@ public class Feed extends BaseEntity {
     private LocalDateTime deletedAt;
 
     @Builder
-    private Feed(String activityContent, Club club, FeedType feedType) {
+    private Feed(Long id, String activityContent, Club club, FeedType feedType) {
+        this.id = id;
         this.activityContent = activityContent;
         this.club = club;
         this.feedType = feedType;

src/main/java/ddingdong/ddingdongBE/domain/feed/service/FacadeClubFeedServiceImpl.java

@@ -7,18 +7,27 @@
 import ddingdong.ddingdongBE.domain.feed.service.dto.command.UpdateFeedCommand;
 import ddingdong.ddingdongBE.domain.filemetadata.entity.DomainType;
 import ddingdong.ddingdongBE.domain.filemetadata.service.FileMetaDataService;
+import ddingdong.ddingdongBE.domain.vodprocessing.entity.ConvertJobStatus;
+import ddingdong.ddingdongBE.domain.vodprocessing.entity.VodProcessingJob;
+import ddingdong.ddingdongBE.domain.vodprocessing.entity.VodProcessingNotification;
+import ddingdong.ddingdongBE.domain.vodprocessing.service.VodProcessingJobService;
+import ddingdong.ddingdongBE.sse.service.SseConnectionService;
+import ddingdong.ddingdongBE.sse.service.dto.SseEvent;
+import java.time.LocalDateTime;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 @Service
 @RequiredArgsConstructor
 @Transactional(readOnly = true)
-public class FacadeClubFeedServiceImpl implements FacadeClubFeedService{
+public class FacadeClubFeedServiceImpl implements FacadeClubFeedService {
 
     private final ClubService clubService;
     private final FileMetaDataService fileMetaDataService;
     private final FeedService feedService;
+    private final VodProcessingJobService vodProcessingJobService;
+    private final SseConnectionService sseConnectionService;
 
     @Override
     @Transactional
@@ -34,6 +43,7 @@ public void create(CreateFeedCommand command) {
 
         if (feed.isVideo()) {
             fileMetaDataService.updateStatusToCoupled(command.mediaId(), DomainType.FEED_VIDEO, createdId);
+            checkVodProcessingJobAndNotify(feed);
         }
     }
 
@@ -52,4 +62,16 @@ public void delete(Long feedId) {
         feedService.delete(feed);
         fileMetaDataService.updateStatusToDelete(feed.getFeedType().getDomainType(), feed.getId());
     }
+
+    private void checkVodProcessingJobAndNotify(Feed feed) {
+        VodProcessingJob vodProcessingJob = vodProcessingJobService.getByVideoFeedId(feed.getId());
+        if (vodProcessingJob.isPossibleNotify()) {
+            SseEvent<ConvertJobStatus> sseEvent = SseEvent.of(
+                    "vod-processing",
+                    vodProcessingJob.getConvertJobStatus(),
+                    LocalDateTime.now()
+            );
+            sseConnectionService.sendVodProcessingNotification(vodProcessingJob, sseEvent);
+        }
+    }
 }

src/main/java/ddingdong/ddingdongBE/domain/feed/service/FeedService.java

@@ -2,6 +2,7 @@
 
 import ddingdong.ddingdongBE.domain.feed.entity.Feed;
 import java.util.List;
+import java.util.Optional;
 
 public interface FeedService {
 
@@ -11,6 +12,8 @@ public interface FeedService {
 
     Feed getById(Long feedId);
 
+    Optional<Feed> findById(Long feedId);
+
     Long create(Feed feed);
 
     void delete(Feed feed);

src/main/java/ddingdong/ddingdongBE/domain/feed/service/GeneralFeedService.java

@@ -4,6 +4,7 @@
 import ddingdong.ddingdongBE.domain.feed.entity.Feed;
 import ddingdong.ddingdongBE.domain.feed.repository.FeedRepository;
 import java.util.List;
+import java.util.Optional;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -31,6 +32,11 @@ public Feed getById(Long feedId) {
             .orElseThrow(() -> new ResourceNotFound("Feed(id: " + feedId + ")를 찾을 수 없습니다."));
     }
 
+    @Override
+    public Optional<Feed> findById(Long feedId) {
+        return feedRepository.findById(feedId);
+    }
+
     @Override
     @Transactional
     public Long create(Feed feed) {

src/main/java/ddingdong/ddingdongBE/domain/vodprocessing/controller/dto/request/UpdateVodProcessingJobStatusRequest.java

@@ -3,11 +3,9 @@
 import ddingdong.ddingdongBE.domain.vodprocessing.entity.ConvertJobStatus;
 import ddingdong.ddingdongBE.domain.vodprocessing.service.dto.command.UpdateVodProcessingJobStatusCommand;
 import jakarta.validation.constraints.NotNull;
-import org.hibernate.validator.constraints.UUID;
 
 public record UpdateVodProcessingJobStatusRequest(
         @NotNull(message = "변환 작업 ID는 필수입니다")
-        @UUID
         String convertJobId,
         @NotNull(message = "상태는 필수입니다")
         ConvertJobStatus status

src/main/java/ddingdong/ddingdongBE/domain/vodprocessing/entity/VodNotificationStatus.java

@@ -1,4 +1,9 @@
 package ddingdong.ddingdongBE.domain.vodprocessing.entity;
 
 public enum VodNotificationStatus {
+    PENDING,
+    SENT,
+    COMPLETED,
+    FAILED,
+    EXPIRED
 }

src/main/java/ddingdong/ddingdongBE/domain/vodprocessing/entity/VodProcessingJob.java

@@ -57,4 +57,8 @@ private VodProcessingJob(Long id, VodProcessingNotification vodProcessingNotific
     public void updateConvertJobStatus(ConvertJobStatus convertJobStatus) {
         this.convertJobStatus = convertJobStatus;
     }
+
+    public boolean isPossibleNotify() {
+        return this.convertJobStatus == ConvertJobStatus.COMPLETE || this.convertJobStatus == ConvertJobStatus.ERROR;
+    }
 }

src/main/java/ddingdong/ddingdongBE/domain/vodprocessing/entity/VodProcessingNotification.java

@@ -10,6 +10,7 @@
 import jakarta.persistence.Id;
 import java.time.LocalDateTime;
 import lombok.AccessLevel;
+import lombok.Builder;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 
@@ -33,4 +34,26 @@ public class VodProcessingNotification extends BaseEntity {
     @Column(name = "notification_status", nullable = false)
     private VodNotificationStatus vodNotificationStatus;
 
+    @Builder
+    private VodProcessingNotification(Long id, LocalDateTime expiredAt, LocalDateTime sentAt, int retryCount,
+                                     VodNotificationStatus vodNotificationStatus) {
+        this.id = id;
+        this.expiredAt = expiredAt;
+        this.sentAt = sentAt;
+        this.retryCount = retryCount;
+        this.vodNotificationStatus = vodNotificationStatus;
+    }
+
+    public static VodProcessingNotification creatPending() {
+        return VodProcessingNotification.builder()
+                .retryCount(0)
+                .vodNotificationStatus(VodNotificationStatus.PENDING)
+                .build();
+    }
+
+    public void updateVodNotificationStatusToSent(LocalDateTime sentAt) {
+        this.sentAt = sentAt;
+        this.vodNotificationStatus = VodNotificationStatus.SENT;
+    }
+
 }

src/main/java/ddingdong/ddingdongBE/domain/vodprocessing/repository/VodProcessingJobRepository.java

@@ -1,13 +1,27 @@
 package ddingdong.ddingdongBE.domain.vodprocessing.repository;
 
+import ddingdong.ddingdongBE.domain.filemetadata.entity.DomainType;
 import ddingdong.ddingdongBE.domain.vodprocessing.entity.VodProcessingJob;
 import java.util.Optional;
 import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
 import org.springframework.stereotype.Repository;
 
 @Repository
 public interface VodProcessingJobRepository extends JpaRepository<VodProcessingJob, Long> {
 
     Optional<VodProcessingJob> findByConvertJobId(String convertJobId);
 
+    @Query("""
+            SELECT vj FROM VodProcessingJob vj
+            JOIN FETCH vj.fileMetaData fm
+            WHERE fm.entityId = :entityId
+            AND fm.domainType = :domainType
+            """)
+    Optional<VodProcessingJob> findFirstByFileMetaDataEntityIdAndDomainType(
+            @Param("entityId") Long entityId,
+            @Param("domainType") DomainType domainType
+    );
+
 }

src/main/java/ddingdong/ddingdongBE/domain/vodprocessing/repository/VodProcessingNotificationRepository.java

@@ -0,0 +1,10 @@
+package ddingdong.ddingdongBE.domain.vodprocessing.repository;
+
+import ddingdong.ddingdongBE.domain.vodprocessing.entity.VodProcessingNotification;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface VodProcessingNotificationRepository extends JpaRepository<VodProcessingNotification, Long> {
+
+}

src/main/java/ddingdong/ddingdongBE/domain/vodprocessing/service/FacadeVodProcessingJobServiceImpl.java

@@ -1,10 +1,18 @@
 package ddingdong.ddingdongBE.domain.vodprocessing.service;
 
+import ddingdong.ddingdongBE.domain.feed.entity.Feed;
+import ddingdong.ddingdongBE.domain.feed.service.FeedService;
 import ddingdong.ddingdongBE.domain.filemetadata.entity.FileMetaData;
 import ddingdong.ddingdongBE.domain.filemetadata.service.FileMetaDataService;
+import ddingdong.ddingdongBE.domain.vodprocessing.entity.ConvertJobStatus;
 import ddingdong.ddingdongBE.domain.vodprocessing.entity.VodProcessingJob;
+import ddingdong.ddingdongBE.domain.vodprocessing.entity.VodProcessingNotification;
 import ddingdong.ddingdongBE.domain.vodprocessing.service.dto.command.CreatePendingVodProcessingJobCommand;
 import ddingdong.ddingdongBE.domain.vodprocessing.service.dto.command.UpdateVodProcessingJobStatusCommand;
+import ddingdong.ddingdongBE.sse.service.SseConnectionService;
+import ddingdong.ddingdongBE.sse.service.dto.SseEvent;
+import java.time.LocalDateTime;
+import java.util.Optional;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -16,19 +24,43 @@ public class FacadeVodProcessingJobServiceImpl implements FacadeVodProcessingJob
 
     private final VodProcessingJobService vodProcessingJobService;
     private final FileMetaDataService fileMetaDataService;
+    private final VodProcessingNotificationService vodProcessingNotificationService;
+    private final SseConnectionService sseConnectionService;
+    private final FeedService feedService;
 
     @Override
     @Transactional
     public Long create(CreatePendingVodProcessingJobCommand command) {
         FileMetaData fileMetaData = fileMetaDataService.getById(command.fileId());
-        return vodProcessingJobService.save(command.toPendingVodProcessingJob(fileMetaData));
+        VodProcessingNotification pendingNotification =
+                vodProcessingNotificationService.save(VodProcessingNotification.creatPending());
+        return vodProcessingJobService.save(command.toPendingVodProcessingJob(pendingNotification, fileMetaData));
     }
 
     @Override
     @Transactional
     public void updateVodProcessingJobStatus(UpdateVodProcessingJobStatusCommand command) {
         VodProcessingJob vodProcessingJob = vodProcessingJobService.getByConvertJobId(command.convertJobId());
         vodProcessingJob.updateConvertJobStatus(command.status());
+        checkVodProcessingJobStatus(vodProcessingJob);
+    }
+
+    private void checkVodProcessingJobStatus(VodProcessingJob vodProcessingJob) {
+        if (vodProcessingJob.isPossibleNotify()) {
+            checkExistingFeedAndNotify(vodProcessingJob);
+        }
+    }
+
+    private void checkExistingFeedAndNotify(VodProcessingJob vodProcessingJob) {
+        Optional<Feed> optionalFeed = feedService.findById(vodProcessingJob.getFileMetaData().getEntityId());
+        if (optionalFeed.isPresent()) {
+            SseEvent<ConvertJobStatus> sseEvent = SseEvent.of(
+                    "vod-processing",
+                    vodProcessingJob.getConvertJobStatus(),
+                    LocalDateTime.now()
+            );
+            sseConnectionService.sendVodProcessingNotification(vodProcessingJob, sseEvent);
+        }
     }
 
 }

src/main/java/ddingdong/ddingdongBE/domain/vodprocessing/service/GeneralVodProcessingJobService.java

@@ -1,6 +1,7 @@
 package ddingdong.ddingdongBE.domain.vodprocessing.service;
 
 import ddingdong.ddingdongBE.common.exception.PersistenceException.ResourceNotFound;
+import ddingdong.ddingdongBE.domain.filemetadata.entity.DomainType;
 import ddingdong.ddingdongBE.domain.vodprocessing.entity.VodProcessingJob;
 import ddingdong.ddingdongBE.domain.vodprocessing.repository.VodProcessingJobRepository;
 import lombok.RequiredArgsConstructor;
@@ -34,4 +35,13 @@ public VodProcessingJob getByConvertJobId(String convertJobId) {
                 .orElseThrow(() -> new ResourceNotFound(
                         "VodProcessingJob(convertJobId=" + convertJobId + ")를 찾을 수 없습니다."));
     }
+
+    @Override
+    public VodProcessingJob getByVideoFeedId(Long videoFeedId) {
+        return vodProcessingJobRepository.findFirstByFileMetaDataEntityIdAndDomainType(
+                        videoFeedId,
+                        DomainType.FEED_VIDEO)
+                .orElseThrow(() -> new ResourceNotFound(
+                        "VodProcessingJob(videoFeedId=" + videoFeedId + ")를 찾을 수 없습니다."));
+    }
 }

src/main/java/ddingdong/ddingdongBE/domain/vodprocessing/service/GeneralVodProcessingNotificationService.java

@@ -0,0 +1,22 @@
+package ddingdong.ddingdongBE.domain.vodprocessing.service;
+
+import ddingdong.ddingdongBE.domain.vodprocessing.entity.VodProcessingNotification;
+import ddingdong.ddingdongBE.domain.vodprocessing.repository.VodProcessingNotificationRepository;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+@Service
+@Transactional(readOnly = true)
+@RequiredArgsConstructor
+public class GeneralVodProcessingNotificationService implements VodProcessingNotificationService {
+
+    private final VodProcessingNotificationRepository notificationRepository;
+
+    @Override
+    @Transactional
+    public VodProcessingNotification save(VodProcessingNotification vodProcessingNotification) {
+        return notificationRepository.save(vodProcessingNotification);
+    }
+
+}

src/main/java/ddingdong/ddingdongBE/domain/vodprocessing/service/VodProcessingJobService.java

@@ -10,4 +10,6 @@ public interface VodProcessingJobService {
 
     VodProcessingJob getByConvertJobId(String convertJobId);
 
+    VodProcessingJob getByVideoFeedId(Long videoFeedId);
+
 }

src/main/java/ddingdong/ddingdongBE/domain/vodprocessing/service/VodProcessingNotificationService.java

@@ -0,0 +1,9 @@
+package ddingdong.ddingdongBE.domain.vodprocessing.service;
+
+import ddingdong.ddingdongBE.domain.vodprocessing.entity.VodProcessingNotification;
+
+public interface VodProcessingNotificationService {
+
+    VodProcessingNotification save(VodProcessingNotification vodProcessingNotification);
+
+}

src/main/java/ddingdong/ddingdongBE/domain/vodprocessing/service/dto/command/CreatePendingVodProcessingJobCommand.java

@@ -3,19 +3,21 @@
 import ddingdong.ddingdongBE.domain.filemetadata.entity.FileMetaData;
 import ddingdong.ddingdongBE.domain.vodprocessing.entity.ConvertJobStatus;
 import ddingdong.ddingdongBE.domain.vodprocessing.entity.VodProcessingJob;
+import ddingdong.ddingdongBE.domain.vodprocessing.entity.VodProcessingNotification;
 
 public record CreatePendingVodProcessingJobCommand(
         String convertJobId,
         String userId,
         String fileId
 ) {
 
-    public VodProcessingJob toPendingVodProcessingJob(FileMetaData fileMetaData) {
+    public VodProcessingJob toPendingVodProcessingJob(VodProcessingNotification notification, FileMetaData fileMetaData) {
         return VodProcessingJob.builder()
                 .convertJobId(convertJobId)
                 .fileMetaData(fileMetaData)
                 .userId(userId)
                 .convertJobStatus(ConvertJobStatus.PENDING)
+                .vodProcessingNotification(notification)
                 .build();
     }
 

src/main/java/ddingdong/ddingdongBE/sse/repository/SseConnectionMapRepository.java

@@ -1,6 +1,7 @@
 package ddingdong.ddingdongBE.sse.repository;
 
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import org.springframework.stereotype.Repository;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
@@ -16,8 +17,8 @@ public void save(String id, SseEmitter sseEmitter) {
     }
 
     @Override
-    public SseEmitter findById(String id) {
-        return emitters.get(id);
+    public Optional<SseEmitter> findById(String id) {
+        return Optional.ofNullable(emitters.get(id));
     }
 
     @Override