feat(ranking): 스냅샷 job 실행 서비스를 추가한다

src/main/kotlin/kr/co/vividnext/sodalive/v2/ranking/application/CreatorRankingSnapshotJobService.kt

@@ -0,0 +1,45 @@
+package kr.co.vividnext.sodalive.v2.ranking.application
+
+import kr.co.vividnext.sodalive.v2.ranking.domain.CreatorRankingPeriodPolicy
+import kr.co.vividnext.sodalive.v2.ranking.port.out.CreatorRankingSnapshotJobPort
+import kr.co.vividnext.sodalive.v2.ranking.port.out.CreatorRankingSnapshotJobRecord
+import kr.co.vividnext.sodalive.v2.ranking.port.out.CreatorRankingSnapshotJobStatus
+import kr.co.vividnext.sodalive.v2.ranking.port.out.CreatorRankingSnapshotJobTrigger
+import org.springframework.stereotype.Service
+import java.time.LocalDateTime
+import java.time.ZonedDateTime
+
+@Service
+class CreatorRankingSnapshotJobService(
+    private val refreshService: CreatorRankingSnapshotRefreshService,
+    private val jobPort: CreatorRankingSnapshotJobPort,
+    private val nowProvider: () -> ZonedDateTime = { ZonedDateTime.now() }
+) {
+    private val periodPolicy = CreatorRankingPeriodPolicy()
+
+    fun refreshLastCompletedWeekByScheduledJob() {
+        val now = nowProvider()
+        val period = periodPolicy.resolveLastCompletedWeek(now)
+        val utcRange = periodPolicy.toUtcRange(period)
+        val job = jobPort.save(
+            CreatorRankingSnapshotJobRecord(
+                aggregationStartAtUtc = utcRange.startInclusiveUtc,
+                aggregationEndAtUtc = utcRange.endExclusiveUtc,
+                trigger = CreatorRankingSnapshotJobTrigger.SCHEDULED,
+                status = CreatorRankingSnapshotJobStatus.PENDING,
+                lastError = null,
+                processingStartedAt = null,
+                processedAt = null
+            )
+        )
+        val jobId = job.id ?: return
+        jobPort.markProcessing(jobId, LocalDateTime.now())
+        try {
+            refreshService.refreshLastCompletedWeek(now)
+            jobPort.markDone(jobId, LocalDateTime.now())
+        } catch (ex: Exception) {
+            jobPort.markFailed(jobId, LocalDateTime.now(), ex.message)
+            throw ex
+        }
+    }
+}

src/main/kotlin/kr/co/vividnext/sodalive/v2/ranking/application/CreatorRankingSnapshotRefreshService.kt

@@ -24,11 +24,6 @@ class CreatorRankingSnapshotRefreshService(
     private val periodPolicy = CreatorRankingPeriodPolicy()
     private val scorePolicy = CreatorRankingScorePolicy()
 
-    @Transactional
-    fun refreshLastCompletedWeek() {
-        refreshLastCompletedWeek(ZonedDateTime.now())
-    }
-
     @Transactional
     fun refreshLastCompletedWeek(now: ZonedDateTime) {
         val startedAt = System.currentTimeMillis()

src/test/kotlin/kr/co/vividnext/sodalive/v2/ranking/application/CreatorRankingSnapshotJobServiceTest.kt

@@ -0,0 +1,121 @@
+package kr.co.vividnext.sodalive.v2.ranking.application
+
+import kr.co.vividnext.sodalive.v2.ranking.port.out.CreatorRankingSnapshotJobPort
+import kr.co.vividnext.sodalive.v2.ranking.port.out.CreatorRankingSnapshotJobRecord
+import kr.co.vividnext.sodalive.v2.ranking.port.out.CreatorRankingSnapshotJobStatus
+import kr.co.vividnext.sodalive.v2.ranking.port.out.CreatorRankingSnapshotJobTrigger
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertThrows
+import org.junit.jupiter.api.DisplayName
+import org.junit.jupiter.api.Test
+import org.mockito.Mockito
+import java.time.LocalDateTime
+import java.time.ZoneId
+import java.time.ZonedDateTime
+
+class CreatorRankingSnapshotJobServiceTest {
+    @Test
+    @DisplayName("스케줄 실행은 집계 기간을 포함한 SCHEDULED job을 생성하고 성공 시 DONE으로 기록한다")
+    fun shouldCreateScheduledJobAndMarkDoneWhenRefreshSucceeds() {
+        val refreshService = Mockito.mock(CreatorRankingSnapshotRefreshService::class.java)
+        val jobPort = FakeCreatorRankingSnapshotJobPort()
+        val now = ZonedDateTime.of(2026, 6, 8, 7, 30, 0, 0, ZoneId.of("Asia/Seoul"))
+        val service = CreatorRankingSnapshotJobService(refreshService, jobPort) { now }
+
+        service.refreshLastCompletedWeekByScheduledJob()
+
+        val job = jobPort.jobs.single()
+        assertEquals(LocalDateTime.of(2026, 5, 31, 15, 0), job.aggregationStartAtUtc)
+        assertEquals(LocalDateTime.of(2026, 6, 7, 15, 0), job.aggregationEndAtUtc)
+        assertEquals(CreatorRankingSnapshotJobTrigger.SCHEDULED, job.trigger)
+        assertEquals(CreatorRankingSnapshotJobStatus.DONE, job.status)
+        assertEquals(null, job.lastError)
+        Mockito.verify(refreshService).refreshLastCompletedWeek(now)
+    }
+
+    @Test
+    @DisplayName("스케줄 실행 실패는 FAILED 상태와 실패 사유를 기록하고 예외를 전파한다")
+    fun shouldMarkScheduledJobFailedWhenRefreshFails() {
+        val refreshService = Mockito.mock(CreatorRankingSnapshotRefreshService::class.java)
+        val jobPort = FakeCreatorRankingSnapshotJobPort()
+        val now = ZonedDateTime.of(2026, 6, 8, 7, 30, 0, 0, ZoneId.of("Asia/Seoul"))
+        val service = CreatorRankingSnapshotJobService(refreshService, jobPort) { now }
+        Mockito.doThrow(IllegalStateException("aggregate failed"))
+            .`when`(refreshService).refreshLastCompletedWeek(now)
+
+        val exception = assertThrows(IllegalStateException::class.java) {
+            service.refreshLastCompletedWeekByScheduledJob()
+        }
+
+        assertEquals("aggregate failed", exception.message)
+        assertEquals(CreatorRankingSnapshotJobStatus.FAILED, jobPort.jobs.single().status)
+        assertEquals("aggregate failed", jobPort.jobs.single().lastError)
+    }
+}
+
+private class FakeCreatorRankingSnapshotJobPort : CreatorRankingSnapshotJobPort {
+    val jobs = mutableListOf<CreatorRankingSnapshotJobRecord>()
+    private var nextId = 1L
+
+    override fun save(job: CreatorRankingSnapshotJobRecord): CreatorRankingSnapshotJobRecord {
+        val saved = job.copy(id = job.id ?: nextId++)
+        jobs.add(saved)
+        return saved
+    }
+
+    override fun findById(jobId: Long): CreatorRankingSnapshotJobRecord? {
+        return jobs.firstOrNull { it.id == jobId }
+    }
+
+    override fun findByPeriodAndStatuses(
+        aggregationStartAtUtc: LocalDateTime,
+        aggregationEndAtUtc: LocalDateTime,
+        statuses: List<CreatorRankingSnapshotJobStatus>
+    ): List<CreatorRankingSnapshotJobRecord> {
+        return jobs.filter {
+            it.aggregationStartAtUtc == aggregationStartAtUtc &&
+                it.aggregationEndAtUtc == aggregationEndAtUtc &&
+                it.status in statuses
+        }
+    }
+
+    override fun markProcessing(jobId: Long, processingStartedAt: LocalDateTime): CreatorRankingSnapshotJobRecord? {
+        return update(jobId) {
+            it.copy(
+                status = CreatorRankingSnapshotJobStatus.PROCESSING,
+                processingStartedAt = processingStartedAt
+            )
+        }
+    }
+
+    override fun markDone(jobId: Long, processedAt: LocalDateTime): CreatorRankingSnapshotJobRecord? {
+        return update(jobId) {
+            it.copy(
+                status = CreatorRankingSnapshotJobStatus.DONE,
+                processedAt = processedAt,
+                lastError = null
+            )
+        }
+    }
+
+    override fun markFailed(jobId: Long, processedAt: LocalDateTime, lastError: String?): CreatorRankingSnapshotJobRecord? {
+        return update(jobId) {
+            it.copy(
+                status = CreatorRankingSnapshotJobStatus.FAILED,
+                processedAt = processedAt,
+                lastError = lastError
+            )
+        }
+    }
+
+    private fun update(
+        jobId: Long,
+        updater: (CreatorRankingSnapshotJobRecord) -> CreatorRankingSnapshotJobRecord
+    ): CreatorRankingSnapshotJobRecord? {
+        val index = jobs.indexOfFirst { it.id == jobId }
+        if (index < 0) return null
+        val updated = updater(jobs[index])
+        jobs[index] = updated
+        return updated
+    }
+}