[Feat] 영상 분석 파이프라인 견고화 및 미처리 영상 자동 분석 도입

.github/workflows/cicd-release.yml

@@ -78,13 +78,13 @@ jobs:
           DISCORD_MENTION_USER_ID: ${{ secrets.DISCORD_MENTION_USER_ID }}
           GCP_PROJECT_ID: ${{ secrets.GCP_PROJECT_ID }}
           YOUTUBE_API_KEY: ${{ secrets.YOUTUBE_API_KEY }}
-          YOUTUBE_PROXY_USERNAME: ${{ secrets.YOUTUBE_PROXY_USERNAME }}
+          YOUTUBE_PROXY_USERNAMES: ${{ secrets.YOUTUBE_PROXY_USERNAMES }}
           YOUTUBE_PROXY_PASSWORD: ${{ secrets.YOUTUBE_PROXY_PASSWORD }}
         with:
           host: ${{ secrets.SERVER_HOST }}
           username: ec2-user
           key: ${{ secrets.SERVER_KEY }}
-          envs: MYSQL_ROOT_PASSWORD,JWT_SECRET_KEY,DISCORD_WEBHOOK_ERROR_URL,DISCORD_MENTION_USER_ID,GCP_PROJECT_ID,YOUTUBE_API_KEY,YOUTUBE_PROXY_USERNAME,YOUTUBE_PROXY_PASSWORD
+          envs: MYSQL_ROOT_PASSWORD,JWT_SECRET_KEY,DISCORD_WEBHOOK_ERROR_URL,DISCORD_MENTION_USER_ID,GCP_PROJECT_ID,YOUTUBE_API_KEY,YOUTUBE_PROXY_USERNAMES,YOUTUBE_PROXY_PASSWORD
           script: |
             set -e
 
@@ -114,7 +114,7 @@ jobs:
             echo "DISCORD_MENTION_USER_ID=${DISCORD_MENTION_USER_ID}" >> $DEPLOY_DIR/.env
             echo "GCP_PROJECT_ID=${GCP_PROJECT_ID}" >> $DEPLOY_DIR/.env
             echo "YOUTUBE_API_KEY=${YOUTUBE_API_KEY}" >> $DEPLOY_DIR/.env
-            echo "YOUTUBE_PROXY_USERNAME=${YOUTUBE_PROXY_USERNAME}" >> $DEPLOY_DIR/.env
+            echo "YOUTUBE_PROXY_USERNAMES=${YOUTUBE_PROXY_USERNAMES}" >> $DEPLOY_DIR/.env
             echo "YOUTUBE_PROXY_PASSWORD=${YOUTUBE_PROXY_PASSWORD}" >> $DEPLOY_DIR/.env
             chmod 600 $DEPLOY_DIR/.env
 

docker/docker-compose.prod.yml

@@ -19,7 +19,7 @@ services:
       - GCP_CREDENTIALS_PATH=/app/config/gcp-credentials.json
       - GCP_VERTEX_AI_LOCATION=us-central1
       - YOUTUBE_API_KEY=${YOUTUBE_API_KEY}
-      - YOUTUBE_PROXY_USERNAME=${YOUTUBE_PROXY_USERNAME}
+      - YOUTUBE_PROXY_USERNAMES=${YOUTUBE_PROXY_USERNAMES}
       - YOUTUBE_PROXY_PASSWORD=${YOUTUBE_PROXY_PASSWORD}
     volumes:
       - ./gcp-credentials.json:/app/config/gcp-credentials.json:ro

linktrip-application/src/main/kotlin/com/linktrip/application/domain/video/KeywordAnalyzeService.kt

@@ -64,7 +64,10 @@ class KeywordAnalyzeService(
 
     private fun analyzeVideo(videoId: String): VideoAnalysisTask? =
         try {
-            videoAnalyzeUseCase.analyzeVideo(VideoAnalysisTask.buildUrl(videoId))
+            videoAnalyzeUseCase.analyzeVideo(
+                youtubeUrl = VideoAnalysisTask.buildUrl(videoId),
+                source = Source.BATCH,
+            )
         } catch (e: Exception) {
             logger.warn { "영상 분석 요청 스킵: videoId=$videoId, reason=${e.message}" }
             null

linktrip-application/src/main/kotlin/com/linktrip/application/domain/video/Source.kt

@@ -0,0 +1,14 @@
+package com.linktrip.application.domain.video
+
+/**
+ * 영상 분석 요청의 출처. 큐 우선순위 결정에 사용된다.
+ *
+ * priority 가 작을수록 먼저 dequeue 된다 (Java [java.util.PriorityQueue] 컨벤션).
+ */
+enum class Source(val priority: Int) {
+    /** 사용자 직접 요청 (POST /video/analyze 등). 항상 먼저 처리. */
+    USER(0),
+
+    /** 시스템 배치/수집 (YouTube 정기 수집, stranded backfill  등). USER 가 비었을 때만 처리. */
+    BATCH(10),
+}

linktrip-application/src/main/kotlin/com/linktrip/application/domain/video/VideoAnalysisTask.kt

@@ -10,6 +10,8 @@ data class VideoAnalysisTask(
     val youtubeUrl: String,
     val valid: Boolean,
     val status: VideoAnalysisTaskStatus,
+    /** 이 task 를 생성한 출처 (audit + 큐 재시도 priority 결정용). 한 번 정해지면 변경하지 않는다. */
+    val source: Source,
     val summary: String? = null,
     val estimatedMinCost: Long? = null,
     val estimatedMaxCost: Long? = null,
@@ -19,7 +21,7 @@ data class VideoAnalysisTask(
     val updatedAt: LocalDateTime = LocalDateTime.now(),
 ) {
     companion object {
-        private const val YOUTUBE_VIDEO_BASE_URL = "https://www.youtube.com/watch?v="
+        const val YOUTUBE_VIDEO_BASE_URL = "https://www.youtube.com/watch?v="
 
         /**
          * 지원 URL 형식:
@@ -45,13 +47,17 @@ data class VideoAnalysisTask(
         private val VIDEO_ID_REGEX =
             Regex("(?:youtube\\.com/(?:watch\\?v=|embed/|shorts/)|youtu\\.be/)([\\w-]+)")
 
-        fun create(youtubeUrl: String): VideoAnalysisTask {
+        fun create(
+            youtubeUrl: String,
+            source: Source,
+        ): VideoAnalysisTask {
             val normalizedUrl = normalizeUrl(youtubeUrl)
             return VideoAnalysisTask(
                 id = IdGenerator.generate(),
                 youtubeUrl = normalizedUrl,
                 valid = false,
                 status = VideoAnalysisTaskStatus.PENDING,
+                source = source,
             )
         }
 

linktrip-application/src/main/kotlin/com/linktrip/application/domain/video/VideoAnalyzeEvent.kt

@@ -3,4 +3,5 @@ package com.linktrip.application.domain.video
 data class VideoAnalyzeEvent(
     val videoAnalysisTaskId: String,
     val youtubeUrl: String,
+    val source: Source,
 )

linktrip-application/src/main/kotlin/com/linktrip/application/domain/video/VideoAnalyzeEventListener.kt

@@ -16,7 +16,9 @@ class VideoAnalyzeEventListener(
     @Async("VideoAnalyzeExecutor")
     @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
     fun handle(event: VideoAnalyzeEvent) {
-        logger.info { "영상 분석 큐 등록: id=${event.videoAnalysisTaskId}, url=${event.youtubeUrl}" }
-        videoAnalysisQueuePort.enqueue(event.videoAnalysisTaskId, event.youtubeUrl)
+        logger.info {
+            "영상 분석 큐 등록: id=${event.videoAnalysisTaskId}, source=${event.source}, url=${event.youtubeUrl}"
+        }
+        videoAnalysisQueuePort.enqueue(event.videoAnalysisTaskId, event.youtubeUrl, event.source)
     }
 }

linktrip-application/src/main/kotlin/com/linktrip/application/domain/video/VideoAnalyzeService.kt

@@ -11,20 +11,24 @@ class VideoAnalyzeService(
     private val videoAnalysisTaskPersistencePort: VideoAnalysisTaskPersistencePort,
 ) : VideoAnalyzeUseCase {
     @Transactional
-    override fun analyzeVideo(youtubeUrl: String): VideoAnalysisTask {
+    override fun analyzeVideo(
+        youtubeUrl: String,
+        source: Source,
+    ): VideoAnalysisTask {
         val normalizedUrl = VideoAnalysisTask.normalizeUrl(youtubeUrl)
 
         videoAnalysisTaskPersistencePort.findByYoutubeUrl(normalizedUrl)?.let { existing ->
             // FAILED 가 아니면 현재 상태를 그대로 반환 (이미 처리 중이거나 완료된 건 재분석하지 않음).
             if (existing.status != VideoAnalysisTaskStatus.FAILED) return existing
 
             videoAnalysisTaskPersistencePort.updateStatus(existing.id, VideoAnalysisTaskStatus.PENDING)
-            Events.raise(VideoAnalyzeEvent(existing.id, normalizedUrl))
+            Events.raise(VideoAnalyzeEvent(existing.id, normalizedUrl, source))
             return existing.copy(status = VideoAnalysisTaskStatus.PENDING)
         }
 
-        val videoAnalysisTask = videoAnalysisTaskPersistencePort.save(VideoAnalysisTask.create(normalizedUrl))
-        Events.raise(VideoAnalyzeEvent(videoAnalysisTask.id, normalizedUrl))
+        val videoAnalysisTask =
+            videoAnalysisTaskPersistencePort.save(VideoAnalysisTask.create(normalizedUrl, source))
+        Events.raise(VideoAnalyzeEvent(videoAnalysisTask.id, normalizedUrl, source))
 
         return videoAnalysisTask
     }

linktrip-application/src/main/kotlin/com/linktrip/application/domain/youtube/YouTubeCollectService.kt

@@ -1,5 +1,6 @@
 package com.linktrip.application.domain.youtube
 
+import com.linktrip.application.domain.video.Source
 import com.linktrip.application.domain.video.VideoAnalysisTask
 import com.linktrip.application.port.input.VideoAnalyzeUseCase
 import com.linktrip.application.port.output.external.YouTubePort
@@ -95,7 +96,10 @@ class YouTubeCollectService(
     private fun requestVideoAnalysis(videos: List<YouTubeVideoMeta>) {
         videos.forEach { video ->
             try {
-                videoAnalyzeUseCase.analyzeVideo(VideoAnalysisTask.buildUrl(video.videoId))
+                videoAnalyzeUseCase.analyzeVideo(
+                    youtubeUrl = VideoAnalysisTask.buildUrl(video.videoId),
+                    source = Source.BATCH,
+                )
             } catch (e: Exception) {
                 logger.warn(e) { "영상 분석 요청 실패: videoId=${video.videoId}" }
             }

linktrip-application/src/main/kotlin/com/linktrip/application/port/input/VideoAnalyzeUseCase.kt

@@ -1,7 +1,11 @@
 package com.linktrip.application.port.input
 
+import com.linktrip.application.domain.video.Source
 import com.linktrip.application.domain.video.VideoAnalysisTask
 
 interface VideoAnalyzeUseCase {
-    fun analyzeVideo(youtubeUrl: String): VideoAnalysisTask
+    fun analyzeVideo(
+        youtubeUrl: String,
+        source: Source,
+    ): VideoAnalysisTask
 }

linktrip-application/src/main/kotlin/com/linktrip/application/port/output/persistence/YouTubeVideoPersistencePort.kt

@@ -20,4 +20,10 @@ interface YouTubeVideoPersistencePort {
         cursor: LocalDateTime?,
         size: Int,
     ): CursorPage<YouTubeVideoMeta>
+
+    /**
+     * youtube_video 에 있지만 아직 video_analysis_task 가 생성되지 않은 (= 한 번도 분석 요청된 적 없는) videoId 목록.
+     * 오래된 것부터 [limit] 건. backfill  스케줄러에서 stranded 영상 소진용.
+     */
+    fun findUnanalyzedVideoIds(limit: Int): List<String>
 }

linktrip-application/src/main/kotlin/com/linktrip/application/port/output/queue/VideoAnalysisQueuePort.kt

@@ -1,11 +1,13 @@
 package com.linktrip.application.port.output.queue
 
+import com.linktrip.application.domain.video.Source
 import com.linktrip.application.domain.video.VideoAnalyzeEvent
 
 interface VideoAnalysisQueuePort {
     fun enqueue(
         videoAnalysisTaskId: String,
         youtubeUrl: String,
+        source: Source,
     )
 
     fun dequeue(): VideoAnalyzeEvent?

linktrip-application/src/test/kotlin/com/linktrip/application/domain/video/VideoAnalysisQueueConsumerTest.kt

@@ -79,7 +79,7 @@ class VideoAnalysisQueueConsumerTest {
         // when - processAnalysis를 직접 호출 (리플렉션 대신 consumer 내부 메서드 테스트)
         val method = consumer.javaClass.getDeclaredMethod("processAnalysis", VideoAnalyzeEvent::class.java)
         method.isAccessible = true
-        method.invoke(consumer, VideoAnalyzeEvent("s1", "https://youtube.com/1"))
+        method.invoke(consumer, VideoAnalyzeEvent("s1", "https://youtube.com/1", Source.USER))
 
         // then
         verify(videoAnalysisResultSaver).save(
@@ -113,7 +113,7 @@ class VideoAnalysisQueueConsumerTest {
         // when
         val method = consumer.javaClass.getDeclaredMethod("processAnalysis", VideoAnalyzeEvent::class.java)
         method.isAccessible = true
-        method.invoke(consumer, VideoAnalyzeEvent("s1", "https://youtube.com/1"))
+        method.invoke(consumer, VideoAnalyzeEvent("s1", "https://youtube.com/1", Source.USER))
 
         // then
         verify(videoAnalysisTaskPersistencePort).updateValidAndStatus(
@@ -142,7 +142,7 @@ class VideoAnalysisQueueConsumerTest {
         // when
         val method = consumer.javaClass.getDeclaredMethod("processAnalysis", VideoAnalyzeEvent::class.java)
         method.isAccessible = true
-        method.invoke(consumer, VideoAnalyzeEvent("s1", "https://youtube.com/1"))
+        method.invoke(consumer, VideoAnalyzeEvent("s1", "https://youtube.com/1", Source.USER))
 
         // then
         verify(videoAnalysisTaskPersistencePort).updateStatus("s1", VideoAnalysisTaskStatus.FAILED)
@@ -166,7 +166,7 @@ class VideoAnalysisQueueConsumerTest {
         // when
         val method = consumer.javaClass.getDeclaredMethod("processAnalysis", VideoAnalyzeEvent::class.java)
         method.isAccessible = true
-        method.invoke(consumer, VideoAnalyzeEvent("s1", "https://youtube.com/1"))
+        method.invoke(consumer, VideoAnalyzeEvent("s1", "https://youtube.com/1", Source.USER))
 
         // then - 분석 결과는 저장되고, 알림도 전송된다
         verify(videoAnalysisResultSaver).save(
@@ -205,7 +205,7 @@ class VideoAnalysisQueueConsumerTest {
         // when
         val method = consumer.javaClass.getDeclaredMethod("processAnalysis", VideoAnalyzeEvent::class.java)
         method.isAccessible = true
-        method.invoke(consumer, VideoAnalyzeEvent("s1", "https://youtube.com/1"))
+        method.invoke(consumer, VideoAnalyzeEvent("s1", "https://youtube.com/1", Source.USER))
 
         // then
         assertEquals(true, request1.processed)
@@ -259,7 +259,7 @@ class VideoAnalysisQueueConsumerTest {
         // when
         val method = consumer.javaClass.getDeclaredMethod("processAnalysis", VideoAnalyzeEvent::class.java)
         method.isAccessible = true
-        method.invoke(consumer, VideoAnalyzeEvent("s1", "https://youtube.com/1"))
+        method.invoke(consumer, VideoAnalyzeEvent("s1", "https://youtube.com/1", Source.USER))
 
         // then
         val itemsCaptor = argumentCaptor<List<TravelItineraryItem>>()

linktrip-application/src/test/kotlin/com/linktrip/application/domain/video/VideoAnalysisTaskTest.kt

@@ -59,24 +59,34 @@ class VideoAnalysisTaskTest {
     }
 
     @Test
-    fun `create는 정규화된 URL로 PENDING 상태의 VideoAnalysisTask를 생성한다`() {
-        // given - youtu.be 단축 URL
+    fun `create는 정규화된 URL로 PENDING 상태와 지정된 source로 VideoAnalysisTask를 생성한다`() {
+        // given - youtu.be 단축 URL + USER source
         val shortUrl = "https://youtu.be/abc123"
 
         // when
-        val task = VideoAnalysisTask.create(shortUrl)
+        val task = VideoAnalysisTask.create(shortUrl, Source.USER)
 
-        // then - 정규화된 URL, PENDING 상태, valid=false
+        // then - 정규화된 URL, PENDING 상태, valid=false, source=USER
         assertEquals("https://www.youtube.com/watch?v=abc123", task.youtubeUrl)
         assertEquals(VideoAnalysisTaskStatus.PENDING, task.status)
+        assertEquals(Source.USER, task.source)
         assertFalse(task.valid)
         assertTrue(task.id.isNotBlank())
     }
 
+    @Test
+    fun `create에 BATCH source를 전달하면_source=BATCH로 task가 생성되어 통계 분류가 가능하다`() {
+        // when - BATCH source 로 생성
+        val task = VideoAnalysisTask.create("https://youtu.be/abc123", Source.BATCH)
+
+        // then - source=BATCH 로 audit 정보가 박혀 통계 집계 가능
+        assertEquals(Source.BATCH, task.source)
+    }
+
     @Test
     fun `create에 유효하지 않은 URL을 전달하면 예외가 발생한다`() {
         assertThrows<LinktripException> {
-            VideoAnalysisTask.create("https://naver.com/video")
+            VideoAnalysisTask.create("https://naver.com/video", Source.USER)
         }
     }
 }

linktrip-application/src/test/kotlin/com/linktrip/application/domain/video/VideoAnalyzeEventListenerTest.kt

@@ -17,14 +17,26 @@ class VideoAnalyzeEventListenerTest {
     lateinit var listener: VideoAnalyzeEventListener
 
     @Test
-    fun `이벤트를 수신하면_큐에 enqueue만 수행한다`() {
-        // given
-        val event = VideoAnalyzeEvent("s1", "https://youtube.com/1")
+    fun `USER 이벤트를 수신하면_큐에 USER source 로 enqueue 한다`() {
+        // given - 사용자 직접 요청에서 발생한 이벤트
+        val event = VideoAnalyzeEvent("s1", "https://youtube.com/1", Source.USER)
 
         // when
         listener.handle(event)
 
-        // then - 큐에 넣기만 하고 직접 분석하지 않는다
-        verify(videoAnalysisQueuePort).enqueue("s1", "https://youtube.com/1")
+        // then - event 의 source 가 그대로 큐로 전달되어야 USER 우선순위가 보장된다
+        verify(videoAnalysisQueuePort).enqueue("s1", "https://youtube.com/1", Source.USER)
+    }
+
+    @Test
+    fun `BATCH 이벤트를 수신하면_큐에 BATCH source 로 enqueue 한다`() {
+        // given - 배치/수집에서 발생한 이벤트
+        val event = VideoAnalyzeEvent("s2", "https://youtube.com/2", Source.BATCH)
+
+        // when
+        listener.handle(event)
+
+        // then - BATCH source 가 보존되어 사용자 요청보다 후순위로 처리됨
+        verify(videoAnalysisQueuePort).enqueue("s2", "https://youtube.com/2", Source.BATCH)
     }
 }