diff --git a/src/main/kotlin/kr/co/vividnext/sodalive/v2/content/recommendation/application/AudioRecommendationSnapshotFallbackService.kt b/src/main/kotlin/kr/co/vividnext/sodalive/v2/content/recommendation/application/AudioRecommendationSnapshotFallbackService.kt new file mode 100644 index 00000000..8580d1c1 --- /dev/null +++ b/src/main/kotlin/kr/co/vividnext/sodalive/v2/content/recommendation/application/AudioRecommendationSnapshotFallbackService.kt @@ -0,0 +1,155 @@ +package kr.co.vividnext.sodalive.v2.content.recommendation.application + +import kr.co.vividnext.sodalive.v2.recommendation.domain.RecommendedSectionType +import kr.co.vividnext.sodalive.v2.recommendation.port.out.RecommendationSnapshotPort +import kr.co.vividnext.sodalive.v2.recommendation.port.out.RecommendationSnapshotRecord +import org.redisson.api.RedissonClient +import org.slf4j.LoggerFactory +import org.springframework.stereotype.Service +import java.time.LocalDateTime +import java.time.ZoneId +import java.util.concurrent.CompletableFuture +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.Executor +import java.util.concurrent.ExecutorService +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit +import java.util.concurrent.TimeoutException +import java.util.concurrent.atomic.AtomicReference +import javax.annotation.PreDestroy + +@Service +class AudioRecommendationSnapshotFallbackService( + private val snapshotPort: RecommendationSnapshotPort, + private val refreshService: AudioRecommendationSnapshotRefreshService, + private val redissonClient: RedissonClient, + executor: Executor? = null, + private val homeWaitMillis: Long = HOME_WAIT_MILLIS +) { + private val log = LoggerFactory.getLogger(javaClass) + private val ownedExecutor: ExecutorService? = if (executor == null) { + Executors.newFixedThreadPool(DEFAULT_WORKER_THREADS) + } else { + null + } + private val workerExecutor: Executor = executor ?: ownedExecutor!! + private val refreshFutures = ConcurrentHashMap?>>() + + fun refreshIfMissing( + sectionType: RecommendedSectionType, + offset: Long, + limit: Int, + now: LocalDateTime = LocalDateTime.now(KST_ZONE) + ): List { + val existing = snapshotPort.findLatestSnapshots(sectionType, offset, limit) + if (existing.isNotEmpty()) return existing + val snapshotAt = snapshotAt(now) + if (snapshotPort.existsSnapshot(sectionType, snapshotAt)) { + return snapshotPort.findLatestSnapshots(sectionType, offset, limit) + } + + val future = getOrStartRefresh(sectionType, now) + return try { + future.get(homeWaitMillis, TimeUnit.MILLISECONDS) + snapshotPort.findLatestSnapshots(sectionType, offset, limit) + } catch (ex: TimeoutException) { + log.warn( + "event=audio_recommendation_snapshot_fallback_timeout sectionType={} homeWaitMs={}", + sectionType, + homeWaitMillis + ) + emptyList() + } catch (ex: InterruptedException) { + Thread.currentThread().interrupt() + log.warn( + "event=audio_recommendation_snapshot_fallback_failure sectionType={} error={}", + sectionType, + ex.message, + ex + ) + emptyList() + } catch (ex: Exception) { + log.warn( + "event=audio_recommendation_snapshot_fallback_failure sectionType={} error={}", + sectionType, + ex.message, + ex + ) + emptyList() + } + } + + @Synchronized + private fun getOrStartRefresh(sectionType: RecommendedSectionType, now: LocalDateTime): CompletableFuture { + val reference = refreshFutures.computeIfAbsent(sectionType) { AtomicReference() } + reference.get()?.let { return it } + + val newFuture = CompletableFuture.runAsync({ refreshInWorker(sectionType, now) }, workerExecutor) + reference.set(newFuture) + newFuture.whenComplete { _, _ -> reference.compareAndSet(newFuture, null) } + return newFuture + } + + private fun refreshInWorker(sectionType: RecommendedSectionType, now: LocalDateTime) { + val lock = redissonClient.getLock(lockKey(sectionType)) + try { + if (!lock.tryLock(LOCK_WAIT_MILLIS, -1, TimeUnit.MILLISECONDS)) { + log.info( + "event=audio_recommendation_snapshot_fallback_lock_missed sectionType={} lockKey={} lockWaitMs={}", + sectionType, + lockKey(sectionType), + LOCK_WAIT_MILLIS + ) + return + } + + log.info( + "event=audio_recommendation_snapshot_fallback_lock_acquired sectionType={} lockKey={}", + sectionType, + lockKey(sectionType) + ) + if (snapshotPort.existsSnapshot(sectionType, snapshotAt(now))) return + + log.info("event=audio_recommendation_snapshot_fallback_refresh_start sectionType={}", sectionType) + val refreshedCount = refreshService.refreshSection(sectionType, now) + log.info( + "event=audio_recommendation_snapshot_fallback_refresh_success sectionType={} refreshedCount={}", + sectionType, + refreshedCount + ) + } catch (ex: Exception) { + log.warn( + "event=audio_recommendation_snapshot_fallback_refresh_failure sectionType={} error={}", + sectionType, + ex.message, + ex + ) + } finally { + if (lock.isHeldByCurrentThread) { + lock.unlock() + } + } + } + + @PreDestroy + fun shutdown() { + ownedExecutor?.shutdown() + } + + companion object { + private val KST_ZONE: ZoneId = ZoneId.of("Asia/Seoul") + private const val LOCK_WAIT_MILLIS = 300L + private const val HOME_WAIT_MILLIS = 1_500L + private const val DEFAULT_WORKER_THREADS = 2 + + fun lockKey(sectionType: RecommendedSectionType): String { + return "lock:audio-recommendation-snapshot-refresh:$sectionType" + } + + private fun snapshotAt(now: LocalDateTime): LocalDateTime { + return now.toLocalDate() + .minusDays(1) + .atTime(23, 59, 59) + } + } +} diff --git a/src/main/kotlin/kr/co/vividnext/sodalive/v2/content/recommendation/application/AudioRecommendationSnapshotRefreshService.kt b/src/main/kotlin/kr/co/vividnext/sodalive/v2/content/recommendation/application/AudioRecommendationSnapshotRefreshService.kt index f2e76301..e56fb14e 100644 --- a/src/main/kotlin/kr/co/vividnext/sodalive/v2/content/recommendation/application/AudioRecommendationSnapshotRefreshService.kt +++ b/src/main/kotlin/kr/co/vividnext/sodalive/v2/content/recommendation/application/AudioRecommendationSnapshotRefreshService.kt @@ -62,6 +62,48 @@ class AudioRecommendationSnapshotRefreshService( } } + @Transactional(propagation = Propagation.REQUIRES_NEW) + fun refreshSection(sectionType: RecommendedSectionType, now: LocalDateTime = LocalDateTime.now(KST_ZONE)): Int { + val snapshotAt = snapshotAt(now.atZone(KST_ZONE)) + val newAndHotWindowStart = windowStart(snapshotAt, days = 3) + val mostCommentedWindowStart = windowStart(snapshotAt, days = 7) + + when (sectionType) { + RecommendedSectionType.NEW_AND_HOT_AUDIO_SAFE -> replaceNewAndHotSnapshots( + newAndHotWindowStart, + snapshotAt, + AudioRecommendationVisibility.SAFE + ) + RecommendedSectionType.NEW_AND_HOT_AUDIO_ALL -> replaceNewAndHotSnapshots( + newAndHotWindowStart, + snapshotAt, + AudioRecommendationVisibility.ALL + ) + RecommendedSectionType.MOST_COMMENTED_AUDIO_SAFE -> replaceMostCommentedSnapshots( + mostCommentedWindowStart, + snapshotAt, + AudioRecommendationVisibility.SAFE + ) + RecommendedSectionType.MOST_COMMENTED_AUDIO_ALL -> replaceMostCommentedSnapshots( + mostCommentedWindowStart, + snapshotAt, + AudioRecommendationVisibility.ALL + ) + RecommendedSectionType.RECOMMENDED_AUDIO_SAFE -> replaceRecommendedAudioSnapshots( + mostCommentedWindowStart, + snapshotAt, + AudioRecommendationVisibility.SAFE + ) + RecommendedSectionType.RECOMMENDED_AUDIO_ALL -> replaceRecommendedAudioSnapshots( + mostCommentedWindowStart, + snapshotAt, + AudioRecommendationVisibility.ALL + ) + else -> error("Unsupported audio recommendation sectionType: $sectionType") + } + return 1 + } + private fun replaceNewAndHotSnapshots( windowStart: LocalDateTime, snapshotAt: LocalDateTime, diff --git a/src/test/kotlin/kr/co/vividnext/sodalive/v2/content/recommendation/application/AudioRecommendationSnapshotFallbackServiceTest.kt b/src/test/kotlin/kr/co/vividnext/sodalive/v2/content/recommendation/application/AudioRecommendationSnapshotFallbackServiceTest.kt new file mode 100644 index 00000000..7ec06cb1 --- /dev/null +++ b/src/test/kotlin/kr/co/vividnext/sodalive/v2/content/recommendation/application/AudioRecommendationSnapshotFallbackServiceTest.kt @@ -0,0 +1,295 @@ +package kr.co.vividnext.sodalive.v2.content.recommendation.application + +import kr.co.vividnext.sodalive.v2.recommendation.domain.RecommendedSectionType +import kr.co.vividnext.sodalive.v2.recommendation.port.out.RecommendationSnapshotPort +import kr.co.vividnext.sodalive.v2.recommendation.port.out.RecommendationSnapshotRecord +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.DisplayName +import org.junit.jupiter.api.Test +import org.mockito.Mockito +import org.redisson.api.RLock +import org.redisson.api.RedissonClient +import java.time.LocalDateTime +import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executor +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit + +class AudioRecommendationSnapshotFallbackServiceTest { + @Test + @DisplayName("오디오 fallback은 최신 스냅샷이 이미 있으면 lock과 refresh를 실행하지 않는다") + fun shouldReturnExistingAudioSnapshotsWithoutLockOrRefresh() { + val snapshotPort = FakeAudioFallbackSnapshotPort() + val snapshotAt = LocalDateTime.of(2026, 7, 9, 23, 59, 59) + snapshotPort.replaceSnapshots( + RecommendedSectionType.MOST_COMMENTED_AUDIO_SAFE, + snapshotAt, + listOf(snapshot(RecommendedSectionType.MOST_COMMENTED_AUDIO_SAFE, 10L, snapshotAt)) + ) + val refreshService = Mockito.mock(AudioRecommendationSnapshotRefreshService::class.java) + val redissonClient = Mockito.mock(RedissonClient::class.java) + val service = AudioRecommendationSnapshotFallbackService(snapshotPort, refreshService, redissonClient, directExecutor()) + + val snapshots = service.refreshIfMissing( + RecommendedSectionType.MOST_COMMENTED_AUDIO_SAFE, + offset = 0, + limit = 5, + now = LocalDateTime.of(2026, 7, 10, 0, 0) + ) + + assertEquals(listOf(10L), snapshots.map { it.targetId }) + Mockito.verifyNoInteractions(redissonClient) + Mockito.verifyNoInteractions(refreshService) + } + + @Test + @DisplayName("오디오 fallback은 section lock을 잡고 section refresh 후 최신 스냅샷을 다시 조회한다") + fun shouldRefreshMissingAudioSnapshotsWithSectionLock() { + val snapshotPort = FakeAudioFallbackSnapshotPort() + val refreshService = Mockito.mock(AudioRecommendationSnapshotRefreshService::class.java) + val redissonClient = Mockito.mock(RedissonClient::class.java) + val lock = Mockito.mock(RLock::class.java) + val now = LocalDateTime.of(2026, 7, 10, 0, 0) + val sectionType = RecommendedSectionType.RECOMMENDED_AUDIO_ALL + Mockito.`when`( + redissonClient.getLock(AudioRecommendationSnapshotFallbackService.lockKey(sectionType)) + ).thenReturn(lock) + Mockito.`when`(lock.tryLock(300, -1, TimeUnit.MILLISECONDS)).thenReturn(true) + Mockito.`when`(lock.isHeldByCurrentThread).thenReturn(true) + Mockito.doAnswer { + snapshotPort.replaceSnapshots( + sectionType, + LocalDateTime.of(2026, 7, 9, 23, 59, 59), + listOf(snapshot(sectionType, 20L)) + ) + 1 + }.`when`(refreshService).refreshSection(sectionType, now) + val service = AudioRecommendationSnapshotFallbackService(snapshotPort, refreshService, redissonClient, directExecutor()) + + val snapshots = service.refreshIfMissing( + sectionType, + offset = 0, + limit = 20, + now = now + ) + + assertEquals(listOf(20L), snapshots.map { it.targetId }) + Mockito.verify(redissonClient) + .getLock(AudioRecommendationSnapshotFallbackService.lockKey(sectionType)) + Mockito.verify(lock).tryLock(300, -1, TimeUnit.MILLISECONDS) + Mockito.verify(refreshService).refreshSection(sectionType, now) + Mockito.verify(lock).unlock() + } + + @Test + @DisplayName("오디오 fallback은 빈 스냅샷 marker가 있으면 refresh를 반복하지 않는다") + fun shouldSkipAudioRefreshWhenEmptySnapshotMarkerExists() { + val snapshotPort = FakeAudioFallbackSnapshotPort() + val now = LocalDateTime.of(2026, 7, 10, 0, 0) + snapshotPort.replaceSnapshots( + RecommendedSectionType.NEW_AND_HOT_AUDIO_SAFE, + LocalDateTime.of(2026, 7, 9, 23, 59, 59), + emptyList() + ) + val refreshService = Mockito.mock(AudioRecommendationSnapshotRefreshService::class.java) + val redissonClient = Mockito.mock(RedissonClient::class.java) + val service = AudioRecommendationSnapshotFallbackService(snapshotPort, refreshService, redissonClient, directExecutor()) + + val snapshots = service.refreshIfMissing( + RecommendedSectionType.NEW_AND_HOT_AUDIO_SAFE, + offset = 0, + limit = 12, + now = now + ) + + assertEquals(emptyList(), snapshots) + Mockito.verifyNoInteractions(redissonClient) + Mockito.verifyNoInteractions(refreshService) + } + + @Test + @DisplayName("오디오 fallback은 lock 획득 실패 시 refresh를 실행하지 않고 빈 배열을 반환한다") + fun shouldReturnEmptyWhenAudioSectionLockIsNotAcquired() { + val snapshotPort = FakeAudioFallbackSnapshotPort() + val refreshService = Mockito.mock(AudioRecommendationSnapshotRefreshService::class.java) + val redissonClient = Mockito.mock(RedissonClient::class.java) + val lock = Mockito.mock(RLock::class.java) + val sectionType = RecommendedSectionType.MOST_COMMENTED_AUDIO_ALL + Mockito.`when`( + redissonClient.getLock(AudioRecommendationSnapshotFallbackService.lockKey(sectionType)) + ).thenReturn(lock) + Mockito.`when`(lock.tryLock(300, -1, TimeUnit.MILLISECONDS)).thenReturn(false) + Mockito.`when`(lock.isHeldByCurrentThread).thenReturn(false) + val service = AudioRecommendationSnapshotFallbackService(snapshotPort, refreshService, redissonClient, directExecutor()) + + val snapshots = service.refreshIfMissing( + sectionType, + offset = 0, + limit = 5, + now = LocalDateTime.of(2026, 7, 10, 0, 0) + ) + + assertEquals(emptyList(), snapshots) + Mockito.verifyNoInteractions(refreshService) + Mockito.verify(lock, Mockito.never()).unlock() + } + + @Test + @DisplayName("오디오 fallback은 refresh 실패 시 예외를 던지지 않고 빈 배열을 반환한다") + fun shouldReturnEmptyWhenAudioRefreshFails() { + val snapshotPort = FakeAudioFallbackSnapshotPort() + val refreshService = Mockito.mock(AudioRecommendationSnapshotRefreshService::class.java) + val redissonClient = Mockito.mock(RedissonClient::class.java) + val lock = Mockito.mock(RLock::class.java) + val now = LocalDateTime.of(2026, 7, 10, 0, 0) + val sectionType = RecommendedSectionType.RECOMMENDED_AUDIO_SAFE + Mockito.`when`( + redissonClient.getLock(AudioRecommendationSnapshotFallbackService.lockKey(sectionType)) + ).thenReturn(lock) + Mockito.`when`(lock.tryLock(300, -1, TimeUnit.MILLISECONDS)).thenReturn(true) + Mockito.`when`(lock.isHeldByCurrentThread).thenReturn(true) + Mockito.doThrow(IllegalStateException("boom")).`when`(refreshService) + .refreshSection(sectionType, now) + val service = AudioRecommendationSnapshotFallbackService(snapshotPort, refreshService, redissonClient, directExecutor()) + + val snapshots = service.refreshIfMissing( + sectionType, + offset = 0, + limit = 20, + now = now + ) + + assertEquals(emptyList(), snapshots) + Mockito.verify(lock).unlock() + } + + @Test + @DisplayName("오디오 fallback은 timeout 시 빈 배열을 반환하고 worker를 취소하지 않는다") + fun shouldReturnEmptyOnAudioTimeoutAndKeepWorkerRunning() { + val snapshotPort = FakeAudioFallbackSnapshotPort() + val refreshStarted = CountDownLatch(1) + val allowRefreshComplete = CountDownLatch(1) + val refreshService = BlockingAudioRefreshService(snapshotPort, refreshStarted, allowRefreshComplete) + val redissonClient = Mockito.mock(RedissonClient::class.java) + val lock = Mockito.mock(RLock::class.java) + val sectionType = RecommendedSectionType.MOST_COMMENTED_AUDIO_SAFE + Mockito.`when`( + redissonClient.getLock(AudioRecommendationSnapshotFallbackService.lockKey(sectionType)) + ).thenReturn(lock) + Mockito.`when`(lock.tryLock(300, -1, TimeUnit.MILLISECONDS)).thenReturn(true) + Mockito.`when`(lock.isHeldByCurrentThread).thenReturn(true) + val executor = Executors.newFixedThreadPool(2) + val service = AudioRecommendationSnapshotFallbackService( + snapshotPort, + refreshService, + redissonClient, + executor, + homeWaitMillis = 50 + ) + + val first = service.refreshIfMissing( + sectionType, + offset = 0, + limit = 5, + now = LocalDateTime.of(2026, 7, 10, 0, 0) + ) + assertEquals(true, refreshStarted.await(1, TimeUnit.SECONDS)) + allowRefreshComplete.countDown() + executor.shutdown() + assertEquals(true, executor.awaitTermination(1, TimeUnit.SECONDS)) + val second = service.refreshIfMissing( + sectionType, + offset = 0, + limit = 5, + now = LocalDateTime.of(2026, 7, 10, 0, 0) + ) + + assertEquals(emptyList(), first) + assertEquals(listOf(99L), second.map { it.targetId }) + } + + private fun directExecutor(): Executor = Executor { command -> command.run() } +} + +private class BlockingAudioRefreshService( + private val snapshotPort: RecommendationSnapshotPort, + private val refreshStarted: CountDownLatch, + private val allowRefreshComplete: CountDownLatch +) : AudioRecommendationSnapshotRefreshService( + snapshotPort, + Mockito.mock(kr.co.vividnext.sodalive.v2.content.recommendation.port.out.AudioRecommendationQueryPort::class.java) +) { + override fun refreshSection(sectionType: RecommendedSectionType, now: LocalDateTime): Int { + refreshStarted.countDown() + allowRefreshComplete.await(1, TimeUnit.SECONDS) + snapshotPort.replaceSnapshots( + sectionType, + LocalDateTime.of(2026, 7, 9, 23, 59, 59), + listOf(snapshot(sectionType, 99L)) + ) + return 1 + } +} + +private class FakeAudioFallbackSnapshotPort : RecommendationSnapshotPort { + private val snapshots = mutableListOf() + + override fun findLatestSnapshots( + sectionType: RecommendedSectionType, + offset: Long, + limit: Int + ): List { + val latestSnapshotAt = snapshots.filter { it.sectionType == sectionType }.maxOfOrNull { it.snapshotAt } + return snapshots + .filter { it.sectionType == sectionType && it.snapshotAt == latestSnapshotAt && it.targetId != 0L } + .drop(offset.toInt()) + .take(limit) + } + + override fun findSnapshots( + sectionType: RecommendedSectionType, + snapshotAt: LocalDateTime, + offset: Long, + limit: Int + ): List { + return snapshots + .filter { it.sectionType == sectionType && it.snapshotAt == snapshotAt && it.targetId != 0L } + .drop(offset.toInt()) + .take(limit) + } + + override fun existsLatestSnapshot(sectionType: RecommendedSectionType): Boolean { + return snapshots.any { it.sectionType == sectionType } + } + + override fun existsSnapshot(sectionType: RecommendedSectionType, snapshotAt: LocalDateTime): Boolean { + return snapshots.any { it.sectionType == sectionType && it.snapshotAt == snapshotAt } + } + + override fun replaceSnapshots( + sectionType: RecommendedSectionType, + snapshotAt: LocalDateTime, + newSnapshots: List + ) { + snapshots.removeIf { it.sectionType == sectionType && it.snapshotAt == snapshotAt } + if (newSnapshots.isEmpty()) { + snapshots.add(snapshot(sectionType, 0L, snapshotAt)) + return + } + snapshots.addAll(newSnapshots) + } +} + +private fun snapshot( + sectionType: RecommendedSectionType, + targetId: Long, + snapshotAt: LocalDateTime = LocalDateTime.of(2026, 7, 9, 23, 59, 59) +): RecommendationSnapshotRecord { + return RecommendationSnapshotRecord( + sectionType = sectionType, + targetId = targetId, + score = 1.0, + snapshotAt = snapshotAt, + randomTieBreaker = 1.0 + ) +} diff --git a/src/test/kotlin/kr/co/vividnext/sodalive/v2/content/recommendation/application/AudioRecommendationSnapshotRefreshServiceTest.kt b/src/test/kotlin/kr/co/vividnext/sodalive/v2/content/recommendation/application/AudioRecommendationSnapshotRefreshServiceTest.kt index ddee8e35..168bacc3 100644 --- a/src/test/kotlin/kr/co/vividnext/sodalive/v2/content/recommendation/application/AudioRecommendationSnapshotRefreshServiceTest.kt +++ b/src/test/kotlin/kr/co/vividnext/sodalive/v2/content/recommendation/application/AudioRecommendationSnapshotRefreshServiceTest.kt @@ -104,4 +104,23 @@ class AudioRecommendationSnapshotRefreshServiceTest { 100 ) } + + @Test + @DisplayName("section refresh는 요청한 오디오 스냅샷 section만 교체한다") + fun shouldRefreshRequestedAudioSnapshotSectionOnly() { + val now = LocalDateTime.of(2026, 6, 24, 0, 0) + val snapshotAt = LocalDateTime.of(2026, 6, 23, 23, 59, 59) + val windowStart = LocalDateTime.of(2026, 6, 17, 0, 0) + + service.refreshSection(RecommendedSectionType.MOST_COMMENTED_AUDIO_ALL, now) + + Mockito.verify(queryPort).findMostCommentedSnapshots( + windowStart, + snapshotAt, + AudioRecommendationVisibility.ALL, + AudioRecommendationSnapshotRefreshService.MOST_COMMENTED_LIMIT + ) + Mockito.verify(snapshotPort).replaceSnapshots(RecommendedSectionType.MOST_COMMENTED_AUDIO_ALL, snapshotAt, emptyList()) + Mockito.verifyNoMoreInteractions(queryPort) + } }