feat(home): 추천 스냅샷 fallback을 공통화한다
This commit is contained in:
@@ -0,0 +1,13 @@
|
||||
package kr.co.vividnext.sodalive.v2.recommendation.application
|
||||
|
||||
import kr.co.vividnext.sodalive.v2.recommendation.port.out.RecommendationSnapshotRecord
|
||||
import java.time.LocalDateTime
|
||||
import java.time.ZoneOffset
|
||||
|
||||
interface AiCharacterSnapshotFallbackPort {
|
||||
fun refreshIfMissing(
|
||||
offset: Long,
|
||||
limit: Int,
|
||||
nowUtc: LocalDateTime = LocalDateTime.now(ZoneOffset.UTC)
|
||||
): List<RecommendationSnapshotRecord>
|
||||
}
|
||||
@@ -1,137 +0,0 @@
|
||||
package kr.co.vividnext.sodalive.v2.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 org.springframework.transaction.PlatformTransactionManager
|
||||
import org.springframework.transaction.TransactionDefinition
|
||||
import org.springframework.transaction.support.TransactionTemplate
|
||||
import java.time.LocalDateTime
|
||||
import java.time.ZoneOffset
|
||||
import java.util.concurrent.CompletableFuture
|
||||
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 AiCharacterSnapshotFallbackService(
|
||||
private val snapshotPort: RecommendationSnapshotPort,
|
||||
private val refreshService: RecommendationSnapshotRefreshService,
|
||||
private val redissonClient: RedissonClient,
|
||||
executor: Executor? = null,
|
||||
private val homeWaitMillis: Long = HOME_WAIT_MILLIS,
|
||||
transactionManager: PlatformTransactionManager? = null
|
||||
) : AiCharacterSnapshotFallbackPort {
|
||||
private val log = LoggerFactory.getLogger(javaClass)
|
||||
private val ownedExecutor: ExecutorService? = if (executor == null) Executors.newSingleThreadExecutor() else null
|
||||
private val workerExecutor: Executor = executor ?: ownedExecutor!!
|
||||
private val refreshFuture = AtomicReference<CompletableFuture<Void>?>()
|
||||
private val readTransactionTemplate = transactionManager?.let {
|
||||
TransactionTemplate(it).also { template ->
|
||||
template.propagationBehavior = TransactionDefinition.PROPAGATION_REQUIRES_NEW
|
||||
template.isReadOnly = true
|
||||
}
|
||||
}
|
||||
|
||||
override fun refreshIfMissing(
|
||||
offset: Long,
|
||||
limit: Int,
|
||||
nowUtc: LocalDateTime
|
||||
): List<RecommendationSnapshotRecord> {
|
||||
val existing = findLatestSnapshots(offset, limit)
|
||||
if (existing.isNotEmpty()) return existing
|
||||
if (hasLatestSnapshot()) return findLatestSnapshots(offset, limit)
|
||||
|
||||
val future = getOrStartRefresh(nowUtc)
|
||||
return try {
|
||||
future.get(homeWaitMillis, TimeUnit.MILLISECONDS)
|
||||
findLatestSnapshots(offset, limit)
|
||||
} catch (ex: TimeoutException) {
|
||||
log.warn("event=ai_character_snapshot_fallback_timeout homeWaitMs={}", homeWaitMillis)
|
||||
emptyList()
|
||||
} catch (ex: InterruptedException) {
|
||||
Thread.currentThread().interrupt()
|
||||
log.warn("event=ai_character_snapshot_fallback_failure error={}", ex.message, ex)
|
||||
emptyList()
|
||||
} catch (ex: Exception) {
|
||||
log.warn("event=ai_character_snapshot_fallback_failure error={}", ex.message, ex)
|
||||
emptyList()
|
||||
}
|
||||
}
|
||||
|
||||
private fun findLatestSnapshots(offset: Long, limit: Int): List<RecommendationSnapshotRecord> {
|
||||
return readSnapshotPort {
|
||||
snapshotPort.findLatestSnapshots(RecommendedSectionType.AI_CHARACTER, offset, limit)
|
||||
}
|
||||
}
|
||||
|
||||
private fun hasLatestSnapshot(): Boolean {
|
||||
return readSnapshotPort {
|
||||
snapshotPort.existsLatestSnapshot(RecommendedSectionType.AI_CHARACTER)
|
||||
}
|
||||
}
|
||||
|
||||
private fun <T> readSnapshotPort(action: () -> T): T {
|
||||
val template = readTransactionTemplate ?: return action()
|
||||
return template.execute { action() } ?: action()
|
||||
}
|
||||
|
||||
@Synchronized
|
||||
private fun getOrStartRefresh(nowUtc: LocalDateTime): CompletableFuture<Void> {
|
||||
refreshFuture.get()?.let { return it }
|
||||
|
||||
val newFuture = CompletableFuture.runAsync({ refreshInWorker(nowUtc) }, workerExecutor)
|
||||
refreshFuture.set(newFuture)
|
||||
newFuture.whenComplete { _, _ -> refreshFuture.compareAndSet(newFuture, null) }
|
||||
return newFuture
|
||||
}
|
||||
|
||||
private fun refreshInWorker(nowUtc: LocalDateTime) {
|
||||
val lock = redissonClient.getLock(LOCK_KEY)
|
||||
try {
|
||||
if (!lock.tryLock(LOCK_WAIT_MILLIS, -1, TimeUnit.MILLISECONDS)) {
|
||||
log.info("event=ai_character_snapshot_fallback_lock_missed lockKey={} lockWaitMs={}", LOCK_KEY, LOCK_WAIT_MILLIS)
|
||||
return
|
||||
}
|
||||
|
||||
log.info("event=ai_character_snapshot_fallback_lock_acquired lockKey={}", LOCK_KEY)
|
||||
if (hasLatestSnapshot()) return
|
||||
|
||||
log.info("event=ai_character_snapshot_fallback_refresh_start lockKey={}", LOCK_KEY)
|
||||
val refreshedCount = refreshService.refreshAiCharacterSnapshots(nowUtc)
|
||||
log.info("event=ai_character_snapshot_fallback_refresh_success refreshedCount={}", refreshedCount)
|
||||
} catch (ex: Exception) {
|
||||
log.warn("event=ai_character_snapshot_fallback_refresh_failure error={}", ex.message, ex)
|
||||
} finally {
|
||||
if (lock.isHeldByCurrentThread) {
|
||||
lock.unlock()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
fun shutdown() {
|
||||
ownedExecutor?.shutdown()
|
||||
}
|
||||
|
||||
companion object {
|
||||
const val LOCK_KEY = "lock:recommendation-snapshot-refresh:AI_CHARACTER"
|
||||
private const val LOCK_WAIT_MILLIS = 300L
|
||||
private const val HOME_WAIT_MILLIS = 1_500L
|
||||
}
|
||||
}
|
||||
|
||||
interface AiCharacterSnapshotFallbackPort {
|
||||
fun refreshIfMissing(
|
||||
offset: Long,
|
||||
limit: Int,
|
||||
nowUtc: LocalDateTime = LocalDateTime.now(ZoneOffset.UTC)
|
||||
): List<RecommendationSnapshotRecord>
|
||||
}
|
||||
@@ -0,0 +1,222 @@
|
||||
package kr.co.vividnext.sodalive.v2.recommendation.application
|
||||
|
||||
import kr.co.vividnext.sodalive.v2.recommendation.domain.RecommendationSnapshotWindowPolicy
|
||||
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.context.annotation.Primary
|
||||
import org.springframework.stereotype.Service
|
||||
import org.springframework.transaction.PlatformTransactionManager
|
||||
import org.springframework.transaction.TransactionDefinition
|
||||
import org.springframework.transaction.support.TransactionTemplate
|
||||
import java.time.LocalDateTime
|
||||
import java.time.ZoneOffset
|
||||
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
|
||||
|
||||
@Primary
|
||||
@Service
|
||||
class RecommendationSnapshotFallbackService(
|
||||
private val snapshotPort: RecommendationSnapshotPort,
|
||||
private val refreshService: RecommendationSnapshotRefreshService,
|
||||
private val redissonClient: RedissonClient,
|
||||
executor: Executor? = null,
|
||||
private val homeWaitMillis: Long = HOME_WAIT_MILLIS,
|
||||
transactionManager: PlatformTransactionManager? = null
|
||||
) : AiCharacterSnapshotFallbackPort, CheerCreatorSnapshotFallbackPort {
|
||||
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 snapshotWindowPolicy = RecommendationSnapshotWindowPolicy()
|
||||
private val refreshFutures = ConcurrentHashMap<RecommendedSectionType, AtomicReference<CompletableFuture<Void>?>>()
|
||||
private val readTransactionTemplate = transactionManager?.let {
|
||||
TransactionTemplate(it).also { template ->
|
||||
template.propagationBehavior = TransactionDefinition.PROPAGATION_REQUIRES_NEW
|
||||
template.isReadOnly = true
|
||||
}
|
||||
}
|
||||
|
||||
override fun refreshIfMissing(
|
||||
offset: Long,
|
||||
limit: Int,
|
||||
nowUtc: LocalDateTime
|
||||
): List<RecommendationSnapshotRecord> {
|
||||
return refreshIfMissing(target(RecommendedSectionType.AI_CHARACTER, AI_CHARACTER_LOCK_KEY), offset, limit, nowUtc)
|
||||
}
|
||||
|
||||
override fun refreshCheerCreatorIfMissing(
|
||||
offset: Long,
|
||||
limit: Int,
|
||||
nowUtc: LocalDateTime
|
||||
): List<RecommendationSnapshotRecord> {
|
||||
return refreshIfMissing(target(RecommendedSectionType.CHEER_CREATOR, CHEER_CREATOR_LOCK_KEY), offset, limit, nowUtc)
|
||||
}
|
||||
|
||||
private fun refreshIfMissing(
|
||||
target: FallbackTarget,
|
||||
offset: Long,
|
||||
limit: Int,
|
||||
nowUtc: LocalDateTime
|
||||
): List<RecommendationSnapshotRecord> {
|
||||
val snapshotAt = snapshotWindowPolicy.previousKstDayUtcWindow(nowUtc).snapshotAt
|
||||
val existing = findSnapshots(target.sectionType, snapshotAt, offset, limit)
|
||||
if (existing.isNotEmpty()) return existing
|
||||
if (hasSnapshot(target.sectionType, snapshotAt)) return findSnapshots(target.sectionType, snapshotAt, offset, limit)
|
||||
|
||||
val future = getOrStartRefresh(target, nowUtc)
|
||||
return try {
|
||||
future.get(homeWaitMillis, TimeUnit.MILLISECONDS)
|
||||
findSnapshots(target.sectionType, snapshotAt, offset, limit)
|
||||
} catch (ex: TimeoutException) {
|
||||
log.warn(
|
||||
"event=recommendation_snapshot_fallback_timeout sectionType={} homeWaitMs={}",
|
||||
target.sectionType,
|
||||
homeWaitMillis
|
||||
)
|
||||
emptyList()
|
||||
} catch (ex: InterruptedException) {
|
||||
Thread.currentThread().interrupt()
|
||||
log.warn(
|
||||
"event=recommendation_snapshot_fallback_failure sectionType={} error={}",
|
||||
target.sectionType,
|
||||
ex.message,
|
||||
ex
|
||||
)
|
||||
emptyList()
|
||||
} catch (ex: Exception) {
|
||||
log.warn(
|
||||
"event=recommendation_snapshot_fallback_failure sectionType={} error={}",
|
||||
target.sectionType,
|
||||
ex.message,
|
||||
ex
|
||||
)
|
||||
emptyList()
|
||||
}
|
||||
}
|
||||
|
||||
private fun findSnapshots(
|
||||
sectionType: RecommendedSectionType,
|
||||
snapshotAt: LocalDateTime,
|
||||
offset: Long,
|
||||
limit: Int
|
||||
): List<RecommendationSnapshotRecord> {
|
||||
return readSnapshotPort { snapshotPort.findSnapshots(sectionType, snapshotAt, offset, limit) }
|
||||
}
|
||||
|
||||
private fun hasSnapshot(sectionType: RecommendedSectionType, snapshotAt: LocalDateTime): Boolean {
|
||||
return readSnapshotPort { snapshotPort.existsSnapshot(sectionType, snapshotAt) }
|
||||
}
|
||||
|
||||
private fun <T> readSnapshotPort(action: () -> T): T {
|
||||
val template = readTransactionTemplate ?: return action()
|
||||
return template.execute { action() } ?: action()
|
||||
}
|
||||
|
||||
@Synchronized
|
||||
private fun getOrStartRefresh(target: FallbackTarget, nowUtc: LocalDateTime): CompletableFuture<Void> {
|
||||
val reference = refreshFutures.computeIfAbsent(target.sectionType) { AtomicReference() }
|
||||
reference.get()?.let { return it }
|
||||
|
||||
val newFuture = CompletableFuture.runAsync({ refreshInWorker(target, nowUtc) }, workerExecutor)
|
||||
reference.set(newFuture)
|
||||
newFuture.whenComplete { _, _ -> reference.compareAndSet(newFuture, null) }
|
||||
return newFuture
|
||||
}
|
||||
|
||||
private fun refreshInWorker(target: FallbackTarget, nowUtc: LocalDateTime) {
|
||||
val lock = redissonClient.getLock(target.lockKey)
|
||||
try {
|
||||
if (!lock.tryLock(LOCK_WAIT_MILLIS, -1, TimeUnit.MILLISECONDS)) {
|
||||
log.info(
|
||||
"event=recommendation_snapshot_fallback_lock_missed sectionType={} lockKey={} lockWaitMs={}",
|
||||
target.sectionType,
|
||||
target.lockKey,
|
||||
LOCK_WAIT_MILLIS
|
||||
)
|
||||
return
|
||||
}
|
||||
|
||||
log.info(
|
||||
"event=recommendation_snapshot_fallback_lock_acquired sectionType={} lockKey={}",
|
||||
target.sectionType,
|
||||
target.lockKey
|
||||
)
|
||||
val snapshotAt = snapshotWindowPolicy.previousKstDayUtcWindow(nowUtc).snapshotAt
|
||||
if (hasSnapshot(target.sectionType, snapshotAt)) return
|
||||
|
||||
log.info(
|
||||
"event=recommendation_snapshot_fallback_refresh_start sectionType={} lockKey={}",
|
||||
target.sectionType,
|
||||
target.lockKey
|
||||
)
|
||||
val refreshedCount = refresh(target.sectionType, nowUtc)
|
||||
log.info(
|
||||
"event=recommendation_snapshot_fallback_refresh_success sectionType={} refreshedCount={}",
|
||||
target.sectionType,
|
||||
refreshedCount
|
||||
)
|
||||
} catch (ex: Exception) {
|
||||
log.warn(
|
||||
"event=recommendation_snapshot_fallback_refresh_failure sectionType={} error={}",
|
||||
target.sectionType,
|
||||
ex.message,
|
||||
ex
|
||||
)
|
||||
} finally {
|
||||
if (lock.isHeldByCurrentThread) {
|
||||
lock.unlock()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun refresh(sectionType: RecommendedSectionType, nowUtc: LocalDateTime): Int {
|
||||
return when (sectionType) {
|
||||
RecommendedSectionType.AI_CHARACTER -> refreshService.refreshAiCharacterSnapshots(nowUtc)
|
||||
RecommendedSectionType.CHEER_CREATOR -> refreshService.refreshCheerCreatorSnapshots(nowUtc)
|
||||
else -> error("Unsupported fallback sectionType: $sectionType")
|
||||
}
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
fun shutdown() {
|
||||
ownedExecutor?.shutdown()
|
||||
}
|
||||
|
||||
private fun target(sectionType: RecommendedSectionType, lockKey: String): FallbackTarget {
|
||||
return FallbackTarget(sectionType, lockKey)
|
||||
}
|
||||
|
||||
private data class FallbackTarget(
|
||||
val sectionType: RecommendedSectionType,
|
||||
val lockKey: String
|
||||
)
|
||||
|
||||
companion object {
|
||||
const val AI_CHARACTER_LOCK_KEY = "lock:recommendation-snapshot-refresh:AI_CHARACTER"
|
||||
const val CHEER_CREATOR_LOCK_KEY = "lock:recommendation-snapshot-refresh:CHEER_CREATOR"
|
||||
private const val LOCK_WAIT_MILLIS = 300L
|
||||
private const val HOME_WAIT_MILLIS = 1_500L
|
||||
private const val DEFAULT_WORKER_THREADS = 2
|
||||
}
|
||||
}
|
||||
|
||||
interface CheerCreatorSnapshotFallbackPort {
|
||||
fun refreshCheerCreatorIfMissing(
|
||||
offset: Long,
|
||||
limit: Int,
|
||||
nowUtc: LocalDateTime = LocalDateTime.now(ZoneOffset.UTC)
|
||||
): List<RecommendationSnapshotRecord>
|
||||
}
|
||||
Reference in New Issue
Block a user