feat(content): 오디오 스냅샷 섹션 fallback을 추가한다
This commit is contained in:
@@ -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<RecommendedSectionType, AtomicReference<CompletableFuture<Void>?>>()
|
||||
|
||||
fun refreshIfMissing(
|
||||
sectionType: RecommendedSectionType,
|
||||
offset: Long,
|
||||
limit: Int,
|
||||
now: LocalDateTime = LocalDateTime.now(KST_ZONE)
|
||||
): List<RecommendationSnapshotRecord> {
|
||||
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<Void> {
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user