perf: improve extraction cache throughput

src/main/kotlin/dev/typetype/server/services/CachedChannelService.kt

@@ -1,6 +1,5 @@
 package dev.typetype.server.services
 
-import dev.typetype.server.cache.CacheJson
 import dev.typetype.server.cache.CacheService
 import dev.typetype.server.models.ChannelResponse
 import dev.typetype.server.models.ExtractionResult
@@ -10,21 +9,12 @@ class CachedChannelService(
     private val cache: CacheService,
 ) : ChannelService {
 
-    companion object {
-        private const val CHANNEL_CACHE_TTL_SECONDS = 1800L
-    }
-
-    override suspend fun getChannel(url: String, nextpage: String?, sort: String?): ExtractionResult<ChannelResponse> {
-        val key = "channel:$url:${nextpage ?: "null"}:${sort ?: "default"}"
-        runCatching { cache.get(key) }.getOrNull()?.let { cached ->
-            return runCatching { ExtractionResult.Success(CacheJson.decodeFromString<ChannelResponse>(cached)) }.getOrElse {
-                delegate.getChannel(url, nextpage, sort)
-            }
-        }
-        val result = delegate.getChannel(url, nextpage, sort)
-        if (result is ExtractionResult.Success) {
-            runCatching { cache.set(key, CacheJson.encodeToString(ChannelResponse.serializer(), result.data), CHANNEL_CACHE_TTL_SECONDS) }
-        }
-        return result
-    }
+    override suspend fun getChannel(url: String, nextpage: String?, sort: String?): ExtractionResult<ChannelResponse> =
+        PublicExtractionCache.getOrLoad(
+            cache = cache,
+            area = "channel",
+            key = PublicCacheKey.of("channel", url, nextpage, sort),
+            serializer = ChannelResponse.serializer(),
+            ttlSeconds = { PublicCachePolicy.channelTtl(url, nextpage, sort) },
+        ) { delegate.getChannel(url, nextpage, sort) }
 }

src/main/kotlin/dev/typetype/server/services/CachedCommentService.kt

@@ -1,6 +1,5 @@
 package dev.typetype.server.services
 
-import dev.typetype.server.cache.CacheJson
 import dev.typetype.server.cache.CacheService
 import dev.typetype.server.models.CommentsPageResponse
 import dev.typetype.server.models.ExtractionResult
@@ -10,18 +9,12 @@ class CachedCommentService(
     private val cache: CacheService,
 ) : CommentService {
 
-    override suspend fun getComments(url: String, nextpage: String?): ExtractionResult<CommentsPageResponse> {
-        val key = "comments:$url:${nextpage ?: "null"}"
-        runCatching { cache.get(key) }.getOrNull()?.let { cached ->
-            return runCatching { ExtractionResult.Success(CacheJson.decodeFromString<CommentsPageResponse>(cached)) }.getOrElse {
-                delegate.getComments(url, nextpage)
-            }
-        }
-        val result = delegate.getComments(url, nextpage)
-        if (result is ExtractionResult.Success) {
-            runCatching { cache.set(key, CacheJson.encodeToString(CommentsPageResponse.serializer(), result.data), 300L) }
-        }
-        return result
-    }
+    override suspend fun getComments(url: String, nextpage: String?): ExtractionResult<CommentsPageResponse> =
+        PublicExtractionCache.getOrLoad(
+            cache = cache,
+            area = "comments",
+            key = PublicCacheKey.of("comments", url, nextpage),
+            serializer = CommentsPageResponse.serializer(),
+            ttlSeconds = { PublicCachePolicy.commentsTtl(url, nextpage) },
+        ) { delegate.getComments(url, nextpage) }
 }
-

src/main/kotlin/dev/typetype/server/services/CachedSearchService.kt

@@ -1,6 +1,5 @@
 package dev.typetype.server.services
 
-import dev.typetype.server.cache.CacheJson
 import dev.typetype.server.cache.CacheService
 import dev.typetype.server.models.ExtractionResult
 import dev.typetype.server.models.SearchPageResponse
@@ -14,18 +13,11 @@ class CachedSearchService(
         query: String,
         serviceId: Int,
         nextpage: String?,
-    ): ExtractionResult<SearchPageResponse> {
-        val key = "search:$serviceId:$query:${nextpage ?: "null"}"
-        runCatching { cache.get(key) }.getOrNull()?.let { cached ->
-            return runCatching { ExtractionResult.Success(CacheJson.decodeFromString<SearchPageResponse>(cached)) }.getOrElse {
-                delegate.search(query, serviceId, nextpage)
-            }
-        }
-        val result = delegate.search(query, serviceId, nextpage)
-        if (result is ExtractionResult.Success) {
-            runCatching { cache.set(key, CacheJson.encodeToString(SearchPageResponse.serializer(), result.data), 300L) }
-        }
-        return result
-    }
+    ): ExtractionResult<SearchPageResponse> = PublicExtractionCache.getOrLoad(
+        cache = cache,
+        area = "search",
+        key = PublicCacheKey.of("search", serviceId.toString(), query, nextpage),
+        serializer = SearchPageResponse.serializer(),
+        ttlSeconds = { PublicCachePolicy.searchTtl(serviceId, nextpage) },
+    ) { delegate.search(query, serviceId, nextpage) }
 }
-

src/main/kotlin/dev/typetype/server/services/CachedStreamService.kt

@@ -4,29 +4,57 @@ import dev.typetype.server.cache.CacheJson
 import dev.typetype.server.cache.CacheService
 import dev.typetype.server.models.ExtractionResult
 import dev.typetype.server.models.StreamResponse
+import kotlinx.coroutines.CompletableDeferred
+import java.util.concurrent.ConcurrentHashMap
 
 class CachedStreamService(
     private val delegate: StreamService,
     private val cache: CacheService,
 ) : StreamService {
 
+    private val inFlight = ConcurrentHashMap<String, CompletableDeferred<ExtractionResult<StreamResponse>>>()
+
     companion object {
         fun cacheKey(url: String): String = "stream:$url"
     }
 
     override suspend fun getStreamInfo(url: String): ExtractionResult<StreamResponse> {
         val key = cacheKey(url)
-        runCatching { cache.get(key) }.getOrNull()?.let { cached ->
-            return runCatching { ExtractionResult.Success(CacheJson.decodeFromString<StreamResponse>(cached)) }.getOrElse {
-                delegate.getStreamInfo(url)
-            }
+        cachedStream(key)?.let { return it }
+        val pending = CompletableDeferred<ExtractionResult<StreamResponse>>()
+        val existing = inFlight.putIfAbsent(key, pending)
+        if (existing != null) return existing.await()
+        return try {
+            val result = getCachedOrLoad(url, key)
+            pending.complete(result)
+            result
+        } catch (error: Throwable) {
+            pending.completeExceptionally(error)
+            throw error
+        } finally {
+            inFlight.remove(key, pending)
         }
+    }
+
+    private suspend fun getCachedOrLoad(url: String, key: String): ExtractionResult<StreamResponse> {
+        cachedStream(key)?.let { return it }
         val result = delegate.getStreamInfo(url)
         if (result is ExtractionResult.Success) {
             val ttl = result.data.streamCacheTtlSeconds()
-            if (ttl > 0) runCatching { cache.set(key, CacheJson.encodeToString(StreamResponse.serializer(), result.data), ttl) }
+            if (ttl > 0) {
+                runCatching {
+                    cache.set(key, CacheJson.encodeToString(StreamResponse.serializer(), result.data), ttl)
+                }
+            }
         }
         return result
     }
 
+    private suspend fun cachedStream(key: String): ExtractionResult<StreamResponse>? = runCatching { cache.get(key) }
+        .getOrNull()
+        ?.let { cached ->
+            runCatching {
+                ExtractionResult.Success(CacheJson.decodeFromString<StreamResponse>(cached))
+            }.getOrNull()
+        }
 }

src/main/kotlin/dev/typetype/server/services/CachedSuggestionService.kt

@@ -1,6 +1,5 @@
 package dev.typetype.server.services
 
-import dev.typetype.server.cache.CacheJson
 import dev.typetype.server.cache.CacheService
 import dev.typetype.server.models.ExtractionResult
 import kotlinx.serialization.builtins.ListSerializer
@@ -11,24 +10,14 @@ class CachedSuggestionService(
     private val cache: CacheService,
 ) : SuggestionService {
 
-    override suspend fun getSuggestions(query: String, serviceId: Int): ExtractionResult<List<String>> {
-        val key = "suggestions:$serviceId:$query"
-        runCatching { cache.get(key) }.getOrNull()?.let { cached ->
-            return runCatching {
-                ExtractionResult.Success(CacheJson.decodeFromString(ListSerializer(String.serializer()), cached))
-            }.getOrElse { delegate.getSuggestions(query, serviceId) }
-        }
-        val result = delegate.getSuggestions(query, serviceId)
-        if (result is ExtractionResult.Success) {
-            runCatching {
-                cache.set(key, CacheJson.encodeToString(ListSerializer(String.serializer()), result.data), SUGGESTIONS_TTL_SECONDS)
-            }
-        }
-        return result
-    }
+    private val listSerializer = ListSerializer(String.serializer())
 
-    private companion object {
-        const val SUGGESTIONS_TTL_SECONDS = 300L
-    }
+    override suspend fun getSuggestions(query: String, serviceId: Int): ExtractionResult<List<String>> =
+        PublicExtractionCache.getOrLoad(
+            cache = cache,
+            area = "suggestions",
+            key = PublicCacheKey.of("suggestions", serviceId.toString(), query),
+            serializer = listSerializer,
+            ttlSeconds = { PublicCachePolicy.suggestionTtl(serviceId) },
+        ) { delegate.getSuggestions(query, serviceId) }
 }
-

src/main/kotlin/dev/typetype/server/services/CachedTrendingService.kt

@@ -1,6 +1,5 @@
 package dev.typetype.server.services
 
-import dev.typetype.server.cache.CacheJson
 import dev.typetype.server.cache.CacheService
 import dev.typetype.server.models.ExtractionResult
 import dev.typetype.server.models.VideoItem
@@ -13,18 +12,11 @@ class CachedTrendingService(
 
     private val listSerializer = ListSerializer(VideoItem.serializer())
 
-    override suspend fun getTrending(serviceId: Int): ExtractionResult<List<VideoItem>> {
-        val key = "trending:$serviceId"
-        runCatching { cache.get(key) }.getOrNull()?.let { cached ->
-            return runCatching { ExtractionResult.Success(CacheJson.decodeFromString(listSerializer, cached)) }.getOrElse {
-                delegate.getTrending(serviceId)
-            }
-        }
-        val result = delegate.getTrending(serviceId)
-        if (result is ExtractionResult.Success) {
-            runCatching { cache.set(key, CacheJson.encodeToString(listSerializer, result.data), 900L) }
-        }
-        return result
-    }
+    override suspend fun getTrending(serviceId: Int): ExtractionResult<List<VideoItem>> = PublicExtractionCache.getOrLoad(
+        cache = cache,
+        area = "trending",
+        key = PublicCacheKey.of("trending", serviceId.toString()),
+        serializer = listSerializer,
+        ttlSeconds = { PublicCachePolicy.trendingTtl(serviceId) },
+    ) { delegate.getTrending(serviceId) }
 }
-

src/main/kotlin/dev/typetype/server/services/PublicCacheKey.kt

@@ -0,0 +1,22 @@
+package dev.typetype.server.services
+
+import java.security.MessageDigest
+
+internal object PublicCacheKey {
+    private val hex = "0123456789abcdef".toCharArray()
+
+    fun of(area: String, vararg parts: String?): String = "$area:v2:${digest(parts)}"
+
+    private fun digest(parts: Array<out String?>): String {
+        val raw = parts.joinToString(separator = "\u001f") { it.orEmpty() }
+        return MessageDigest.getInstance("SHA-256").digest(raw.toByteArray()).toHex().take(32)
+    }
+
+    private fun ByteArray.toHex(): String = buildString(size * 2) {
+        this@toHex.forEach { byte ->
+            val value = byte.toInt() and 0xff
+            append(hex[value ushr 4])
+            append(hex[value and 0x0f])
+        }
+    }
+}

src/main/kotlin/dev/typetype/server/services/PublicCachePolicy.kt

@@ -0,0 +1,46 @@
+package dev.typetype.server.services
+
+internal object PublicCachePolicy {
+    fun trendingTtl(serviceId: Int): Long = when (serviceId) {
+        BILIBILI_SERVICE_ID, NICONICO_SERVICE_ID -> 600L
+        YOUTUBE_SERVICE_ID -> 1_800L
+        else -> 3_600L
+    }
+
+    fun suggestionTtl(serviceId: Int): Long = when (serviceId) {
+        BILIBILI_SERVICE_ID, NICONICO_SERVICE_ID -> 900L
+        else -> 1_800L
+    }
+
+    fun searchTtl(serviceId: Int, nextpage: String?): Long {
+        if (nextpage != null) return 300L
+        return when (serviceId) {
+            BILIBILI_SERVICE_ID, NICONICO_SERVICE_ID -> 300L
+            YOUTUBE_SERVICE_ID -> 600L
+            else -> 900L
+        }
+    }
+
+    fun channelTtl(url: String, nextpage: String?, sort: String?): Long = when {
+        url.contains("/search", ignoreCase = true) -> 600L
+        nextpage != null -> 1_800L
+        url.contains("/shorts", ignoreCase = true) -> 900L
+        sort.equals("latest", ignoreCase = true) -> 900L
+        else -> 3_600L
+    }
+
+    fun commentsTtl(url: String, nextpage: String?): Long {
+        if (nextpage != null) return 600L
+        return when (url.serviceHint()) {
+            BILIBILI_SERVICE_ID, NICONICO_SERVICE_ID -> 300L
+            else -> 180L
+        }
+    }
+}
+
+private fun String.serviceHint(): Int? = when {
+    contains("bilibili.com", ignoreCase = true) -> BILIBILI_SERVICE_ID
+    contains("nicovideo.jp", ignoreCase = true) -> NICONICO_SERVICE_ID
+    contains("youtube.com", ignoreCase = true) || contains("youtu.be", ignoreCase = true) -> YOUTUBE_SERVICE_ID
+    else -> null
+}

src/main/kotlin/dev/typetype/server/services/PublicExtractionCache.kt

@@ -0,0 +1,59 @@
+package dev.typetype.server.services
+
+import dev.typetype.server.cache.CacheJson
+import dev.typetype.server.cache.CacheService
+import dev.typetype.server.models.ExtractionResult
+import kotlinx.serialization.KSerializer
+import org.slf4j.LoggerFactory
+
+internal object PublicExtractionCache {
+    private val logger = LoggerFactory.getLogger(PublicExtractionCache::class.java)
+
+    suspend fun <T> getOrLoad(
+        cache: CacheService,
+        area: String,
+        key: String,
+        serializer: KSerializer<T>,
+        ttlSeconds: (T) -> Long,
+        load: suspend () -> ExtractionResult<T>,
+    ): ExtractionResult<T> {
+        val cached = runCatching { cache.get(key) }
+            .onFailure { logger.warn("cache event=get_failed area={} key={} error={}", area, key, it.message) }
+            .getOrNull()
+        if (cached != null) {
+            val decoded = runCatching { CacheJson.decodeFromString(serializer, cached) }
+            decoded.getOrNull()?.let {
+                logger.info("cache event=hit area={} key={}", area, key)
+                return ExtractionResult.Success(it)
+            }
+            logger.warn(
+                "cache event=decode_failed area={} key={} error={}",
+                area,
+                key,
+                decoded.exceptionOrNull()?.message,
+            )
+        }
+        logger.info("cache event=miss area={} key={}", area, key)
+        val result = load()
+        if (result is ExtractionResult.Success) {
+            store(cache, area, key, serializer, result.data, ttlSeconds(result.data))
+        }
+        return result
+    }
+
+    private suspend fun <T> store(
+        cache: CacheService,
+        area: String,
+        key: String,
+        serializer: KSerializer<T>,
+        data: T,
+        ttlSeconds: Long,
+    ) {
+        if (ttlSeconds <= 0L) return
+        runCatching { cache.set(key, CacheJson.encodeToString(serializer, data), ttlSeconds) }
+            .onSuccess { logger.info("cache event=store area={} key={} ttlSeconds={}", area, key, ttlSeconds) }
+            .onFailure {
+                logger.warn("cache event=set_failed area={} key={} error={}", area, key, it.message)
+            }
+    }
+}