feat(content-ranking): 랭킹 스냅샷 job 저장소를 추가한다

2026-06-24 16:21:00 +09:00
parent f1e03706c7
commit 453d914f44
5 changed files with 359 additions and 0 deletions
--- a/src/main/kotlin/kr/co/vividnext/sodalive/v2/content/ranking/adapter/out/persistence/AudioRankingSnapshotJob.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/v2/content/ranking/adapter/out/persistence/AudioRankingSnapshotJob.kt
@@ -0,0 +1,46 @@
+package kr.co.vividnext.sodalive.v2.content.ranking.adapter.out.persistence
+
+import kr.co.vividnext.sodalive.common.BaseEntity
+import kr.co.vividnext.sodalive.v2.content.ranking.domain.AudioRankingType
+import kr.co.vividnext.sodalive.v2.content.ranking.port.out.AudioRankingSnapshotJobStatus
+import kr.co.vividnext.sodalive.v2.content.ranking.port.out.AudioRankingSnapshotJobTrigger
+import java.time.LocalDateTime
+import javax.persistence.Column
+import javax.persistence.Entity
+import javax.persistence.EnumType
+import javax.persistence.Enumerated
+import javax.persistence.Table
+
+@Entity
+@Table(name = "content_ranking_snapshot_job")
+class AudioRankingSnapshotJob(
+    @Enumerated(EnumType.STRING)
+    @Column(name = "ranking_type", nullable = false, length = 30)
+    val rankingType: AudioRankingType,
+
+    @Column(name = "aggregation_start_at_utc", nullable = false)
+    val aggregationStartAtUtc: LocalDateTime,
+
+    @Column(name = "aggregation_end_at_utc", nullable = false)
+    val aggregationEndAtUtc: LocalDateTime,
+
+    @Column(name = "visible_from_at", nullable = false)
+    val visibleFromAtUtc: LocalDateTime,
+
+    @Enumerated(EnumType.STRING)
+    @Column(name = "trigger_type", nullable = false, length = 20)
+    val trigger: AudioRankingSnapshotJobTrigger,
+
+    @Enumerated(EnumType.STRING)
+    @Column(name = "status", nullable = false, length = 20)
+    var status: AudioRankingSnapshotJobStatus = AudioRankingSnapshotJobStatus.PENDING,
+
+    @Column(name = "last_error", columnDefinition = "text")
+    var lastError: String? = null,
+
+    @Column(name = "processing_started_at")
+    var processingStartedAt: LocalDateTime? = null,
+
+    @Column(name = "processed_at")
+    var processedAt: LocalDateTime? = null
+) : BaseEntity()
--- a/src/main/kotlin/kr/co/vividnext/sodalive/v2/content/ranking/adapter/out/persistence/AudioRankingSnapshotJobRepository.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/v2/content/ranking/adapter/out/persistence/AudioRankingSnapshotJobRepository.kt
@@ -0,0 +1,31 @@
+package kr.co.vividnext.sodalive.v2.content.ranking.adapter.out.persistence
+
+import kr.co.vividnext.sodalive.v2.content.ranking.domain.AudioRankingType
+import kr.co.vividnext.sodalive.v2.content.ranking.port.out.AudioRankingSnapshotJobStatus
+import kr.co.vividnext.sodalive.v2.content.ranking.port.out.AudioRankingSnapshotJobTrigger
+import org.springframework.data.jpa.repository.JpaRepository
+import org.springframework.data.jpa.repository.Lock
+import org.springframework.data.jpa.repository.Query
+import org.springframework.data.repository.query.Param
+import java.time.LocalDateTime
+import javax.persistence.LockModeType
+
+interface AudioRankingSnapshotJobRepository : JpaRepository<AudioRankingSnapshotJob, Long> {
+    @Lock(LockModeType.PESSIMISTIC_WRITE)
+    @Query("select j from AudioRankingSnapshotJob j where j.id = :jobId")
+    fun findByIdForUpdate(@Param("jobId") jobId: Long): AudioRankingSnapshotJob?
+
+    fun findAllByRankingTypeAndAggregationStartAtUtcAndAggregationEndAtUtcAndStatusInOrderByCreatedAtDesc(
+        rankingType: AudioRankingType,
+        aggregationStartAtUtc: LocalDateTime,
+        aggregationEndAtUtc: LocalDateTime,
+        statuses: List<AudioRankingSnapshotJobStatus>
+    ): List<AudioRankingSnapshotJob>
+
+    fun countByRankingTypeAndAggregationStartAtUtcAndAggregationEndAtUtcAndTrigger(
+        rankingType: AudioRankingType,
+        aggregationStartAtUtc: LocalDateTime,
+        aggregationEndAtUtc: LocalDateTime,
+        trigger: AudioRankingSnapshotJobTrigger
+    ): Long
+}
--- a/src/main/kotlin/kr/co/vividnext/sodalive/v2/content/ranking/adapter/out/persistence/DefaultAudioRankingSnapshotJobRepository.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/v2/content/ranking/adapter/out/persistence/DefaultAudioRankingSnapshotJobRepository.kt
@@ -0,0 +1,112 @@
+package kr.co.vividnext.sodalive.v2.content.ranking.adapter.out.persistence
+
+import kr.co.vividnext.sodalive.v2.content.ranking.domain.AudioRankingType
+import kr.co.vividnext.sodalive.v2.content.ranking.port.out.AudioRankingSnapshotJobPort
+import kr.co.vividnext.sodalive.v2.content.ranking.port.out.AudioRankingSnapshotJobRecord
+import kr.co.vividnext.sodalive.v2.content.ranking.port.out.AudioRankingSnapshotJobStatus
+import kr.co.vividnext.sodalive.v2.content.ranking.port.out.AudioRankingSnapshotJobTrigger
+import org.springframework.stereotype.Repository
+import org.springframework.transaction.annotation.Transactional
+import java.time.LocalDateTime
+
+@Repository
+class DefaultAudioRankingSnapshotJobRepository(
+    private val repository: AudioRankingSnapshotJobRepository
+) : AudioRankingSnapshotJobPort {
+    @Transactional
+    override fun save(job: AudioRankingSnapshotJobRecord): AudioRankingSnapshotJobRecord {
+        return repository.save(job.toEntity()).toRecord()
+    }
+
+    override fun findById(jobId: Long): AudioRankingSnapshotJobRecord? {
+        return repository.findById(jobId).orElse(null)?.toRecord()
+    }
+
+    override fun findByRankingTypeAndPeriodAndStatuses(
+        rankingType: AudioRankingType,
+        aggregationStartAtUtc: LocalDateTime,
+        aggregationEndAtUtc: LocalDateTime,
+        statuses: List<AudioRankingSnapshotJobStatus>
+    ): List<AudioRankingSnapshotJobRecord> {
+        return repository.findAllByRankingTypeAndAggregationStartAtUtcAndAggregationEndAtUtcAndStatusInOrderByCreatedAtDesc(
+            rankingType = rankingType,
+            aggregationStartAtUtc = aggregationStartAtUtc,
+            aggregationEndAtUtc = aggregationEndAtUtc,
+            statuses = statuses
+        ).map { it.toRecord() }
+    }
+
+    override fun countByRankingTypeAndPeriodAndTrigger(
+        rankingType: AudioRankingType,
+        aggregationStartAtUtc: LocalDateTime,
+        aggregationEndAtUtc: LocalDateTime,
+        trigger: AudioRankingSnapshotJobTrigger
+    ): Long {
+        return repository.countByRankingTypeAndAggregationStartAtUtcAndAggregationEndAtUtcAndTrigger(
+            rankingType = rankingType,
+            aggregationStartAtUtc = aggregationStartAtUtc,
+            aggregationEndAtUtc = aggregationEndAtUtc,
+            trigger = trigger
+        )
+    }
+
+    @Transactional
+    override fun markProcessing(jobId: Long, processingStartedAt: LocalDateTime): AudioRankingSnapshotJobRecord? {
+        val job = repository.findByIdForUpdate(jobId) ?: return null
+        job.status = AudioRankingSnapshotJobStatus.PROCESSING
+        job.processingStartedAt = processingStartedAt
+        job.lastError = null
+        return job.toRecord()
+    }
+
+    @Transactional
+    override fun markDone(jobId: Long, processedAt: LocalDateTime): AudioRankingSnapshotJobRecord? {
+        val job = repository.findByIdForUpdate(jobId) ?: return null
+        job.status = AudioRankingSnapshotJobStatus.DONE
+        job.processedAt = processedAt
+        job.lastError = null
+        return job.toRecord()
+    }
+
+    @Transactional
+    override fun markFailed(jobId: Long, processedAt: LocalDateTime, lastError: String?): AudioRankingSnapshotJobRecord? {
+        val job = repository.findByIdForUpdate(jobId) ?: return null
+        job.status = AudioRankingSnapshotJobStatus.FAILED
+        job.processedAt = processedAt
+        job.lastError = lastError?.take(MAX_ERROR_LENGTH)
+        return job.toRecord()
+    }
+
+    private fun AudioRankingSnapshotJobRecord.toEntity(): AudioRankingSnapshotJob {
+        return AudioRankingSnapshotJob(
+            rankingType = rankingType,
+            aggregationStartAtUtc = aggregationStartAtUtc,
+            aggregationEndAtUtc = aggregationEndAtUtc,
+            visibleFromAtUtc = visibleFromAtUtc,
+            trigger = trigger,
+            status = status,
+            lastError = lastError,
+            processingStartedAt = processingStartedAt,
+            processedAt = processedAt
+        )
+    }
+
+    private fun AudioRankingSnapshotJob.toRecord(): AudioRankingSnapshotJobRecord {
+        return AudioRankingSnapshotJobRecord(
+            id = id,
+            rankingType = rankingType,
+            aggregationStartAtUtc = aggregationStartAtUtc,
+            aggregationEndAtUtc = aggregationEndAtUtc,
+            visibleFromAtUtc = visibleFromAtUtc,
+            trigger = trigger,
+            status = status,
+            lastError = lastError,
+            processingStartedAt = processingStartedAt,
+            processedAt = processedAt
+        )
+    }
+
+    companion object {
+        private const val MAX_ERROR_LENGTH = 1000
+    }
+}
--- a/src/main/kotlin/kr/co/vividnext/sodalive/v2/content/ranking/port/out/AudioRankingSnapshotJobPort.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/v2/content/ranking/port/out/AudioRankingSnapshotJobPort.kt
@@ -0,0 +1,56 @@
+package kr.co.vividnext.sodalive.v2.content.ranking.port.out
+
+import kr.co.vividnext.sodalive.v2.content.ranking.domain.AudioRankingType
+import java.time.LocalDateTime
+
+interface AudioRankingSnapshotJobPort {
+    fun save(job: AudioRankingSnapshotJobRecord): AudioRankingSnapshotJobRecord
+
+    fun findById(jobId: Long): AudioRankingSnapshotJobRecord?
+
+    fun findByRankingTypeAndPeriodAndStatuses(
+        rankingType: AudioRankingType,
+        aggregationStartAtUtc: LocalDateTime,
+        aggregationEndAtUtc: LocalDateTime,
+        statuses: List<AudioRankingSnapshotJobStatus>
+    ): List<AudioRankingSnapshotJobRecord>
+
+    fun countByRankingTypeAndPeriodAndTrigger(
+        rankingType: AudioRankingType,
+        aggregationStartAtUtc: LocalDateTime,
+        aggregationEndAtUtc: LocalDateTime,
+        trigger: AudioRankingSnapshotJobTrigger
+    ): Long
+
+    fun markProcessing(jobId: Long, processingStartedAt: LocalDateTime): AudioRankingSnapshotJobRecord?
+
+    fun markDone(jobId: Long, processedAt: LocalDateTime): AudioRankingSnapshotJobRecord?
+
+    fun markFailed(jobId: Long, processedAt: LocalDateTime, lastError: String?): AudioRankingSnapshotJobRecord?
+}
+
+enum class AudioRankingSnapshotJobStatus {
+    PENDING,
+    PROCESSING,
+    DONE,
+    FAILED
+}
+
+enum class AudioRankingSnapshotJobTrigger {
+    SCHEDULED,
+    MANUAL,
+    FALLBACK
+}
+
+data class AudioRankingSnapshotJobRecord(
+    val id: Long? = null,
+    val rankingType: AudioRankingType,
+    val aggregationStartAtUtc: LocalDateTime,
+    val aggregationEndAtUtc: LocalDateTime,
+    val visibleFromAtUtc: LocalDateTime,
+    val trigger: AudioRankingSnapshotJobTrigger,
+    val status: AudioRankingSnapshotJobStatus,
+    val lastError: String?,
+    val processingStartedAt: LocalDateTime?,
+    val processedAt: LocalDateTime?
+)