create cache storage

apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/Main.kt

@@ -8,7 +8,7 @@ import org.koin.core.component.KoinComponent
 import org.koin.core.component.get
 import org.koin.plugin.module.dsl.startKoin
 import ru.shadowsparky.koin
-import ru.shadowsparky.vbox.backend.data.RedisCache
+import ru.shadowsparky.vbox.backend.domain.CacheStorage
 import ru.shadowsparky.vbox.backend.presentation.configureJwt
 import ru.shadowsparky.vbox.backend.presentation.configureRouting
 import ru.shadowsparky.vbox.backend.presentation.configureSerialization
@@ -24,7 +24,7 @@ fun main(args: Array<String>) {
 
 fun Application.module() {
     monitor.subscribe(ApplicationStopping) {
-        runBlocking { koin.getOrNull<RedisCache>()?.disconnect() }
+        runBlocking { koin.getOrNull<CacheStorage>()?.close() }
     }
     val koin = object : KoinComponent {}
     configureSerialization()

apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/RedisCacheStorage.kt

@@ -9,27 +9,93 @@ import io.lettuce.core.api.coroutines.RedisCoroutinesCommands
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withContext
+import kotlinx.serialization.KSerializer
 import kotlinx.serialization.json.Json
-import org.koin.core.annotation.Factory
+import org.koin.core.annotation.Single
 import org.slf4j.LoggerFactory
 import ru.shadowsparky.backend.data.EnvFetcher
 import ru.shadowsparky.domain.DispatcherProvider
+import ru.shadowsparky.vbox.backend.domain.CacheStorage
+import java.util.concurrent.ConcurrentHashMap
 import kotlin.time.Duration
 
 @OptIn(ExperimentalLettuceCoroutinesApi::class)
-@Factory
+@Single
-class RedisCache(
+class RedisCacheStorage(
     private val envFetcher: EnvFetcher,
     private val dispatcherProvider: DispatcherProvider,
-    val json: Json
+    private val json: Json
-) {
+) : CacheStorage {
-    private val logger = LoggerFactory.getLogger(RedisCache::class.java)
+    private val logger = LoggerFactory.getLogger("RedisCacheStorage")
 
     private var client: RedisClient? = null
     private var connection: StatefulRedisConnection<String, String>? = null
     private var commands: RedisCoroutinesCommands<String, String>? = null
-
     private val connectionMutex = Mutex()
+    private val locks = ConcurrentHashMap<String, Mutex>()
+
+    override suspend fun <T> getOrCreate(
+        key: String,
+        serializer: KSerializer<T>,
+        ttl: Duration?,
+        createNew: suspend () -> T,
+    ): T {
+        read(key, serializer)?.let { return it }
+        val mutex = locks.computeIfAbsent(key) { Mutex() }
+        return mutex.withLock {
+            try {
+                read(key, serializer)?.let { return@withLock it }
+                createNew().also {
+                    write(key, it, serializer, ttl)
+                }
+            } finally {
+                locks.remove(key, mutex)
+            }
+        }
+    }
+
+    private suspend fun <T> write(
+        key: String,
+        value: T,
+        serializer: KSerializer<T>,
+        duration: Duration? = null
+    ) {
+        val ttlInSeconds = duration?.inWholeSeconds
+        val rawValue = json.encodeToString(serializer, value)
+        if (ttlInSeconds != null && ttlInSeconds > 0) {
+            val args = SetArgs().ex(ttlInSeconds)
+            getCommands().set(key, rawValue, args)
+        } else {
+            getCommands().set(key, rawValue)
+        }
+    }
+
+    private suspend fun <T> read(key: String, serializer: KSerializer<T>): T? {
+        val raw = getCommands().get(key) ?: return null
+        return json.decodeFromString(serializer, raw)
+    }
+
+    override suspend fun delete(key: String) {
+        (getCommands().del(key) ?: 0L) > 0L
+    }
+
+    override suspend fun close() {
+        connectionMutex.withLock {
+            if (connection != null || client != null) {
+                logger.info("closing redis...")
+                try {
+                    connection?.close()
+                    client?.shutdown()
+                } catch (e: Exception) {
+                    logger.error("unable to close redis: ${e.message}", e)
+                } finally {
+                    connection = null
+                    client = null
+                    commands = null
+                }
+            }
+        }
+    }
 
     private suspend fun getCommands(): RedisCoroutinesCommands<String, String> {
         commands?.let { return it }
@@ -56,61 +122,4 @@ class RedisCache(
             }
         }
     }
-
-    suspend inline fun <reified T> write(key: String, value: T, duration: Duration? = null) {
-        write(key, json.encodeToString(value), duration)
-    }
-
-    suspend fun write(key: String, value: String, duration: Duration? = null) {
-        val ttlInSeconds = duration?.inWholeSeconds
-        if (ttlInSeconds != null && ttlInSeconds > 0) {
-            val args = SetArgs().ex(ttlInSeconds)
-            getCommands().set(key, value, args)
-        } else {
-            getCommands().set(key, value)
-        }
-    }
-
-    suspend inline fun <reified T> transact(
-        key: String,
-        ttl: Duration? = null,
-        createNew: suspend () -> T
-    ): T {
-        read<T>(key)?.let { return it }
-        val result = createNew()
-        write(key, result, ttl)
-        return result
-    }
-
-    suspend inline fun <reified T> read(key: String): T? {
-        return json.decodeFromString(readInternal(key) ?: return null)
-    }
-
-    suspend fun readInternal(key: String): String? {
-        return getCommands().get(key)
-    }
-
-    suspend fun delete(key: String): Boolean {
-        return (getCommands().del(key) ?: 0L) > 0L
-    }
-
-    suspend fun disconnect() {
-        connectionMutex.withLock {
-            if (connection != null || client != null) {
-                logger.info("Закрытие соединений с Redis...")
-                try {
-                    connection?.close()
-                    client?.shutdown()
-                } catch (e: Exception) {
-                    logger.error("Ошибка при закрытии Redis: ${e.message}", e)
-                } finally {
-                    connection = null
-                    client = null
-                    commands = null
-                }
-                logger.info("Redis успешно отключен")
-            }
-        }
-    }
 }
-

apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/http/CacheVideoApi.kt

@@ -1,7 +1,8 @@
 package ru.shadowsparky.vbox.backend.data.http
 
 import ru.shadowsparky.vbox.backend.data.FuzzyMovieSearchFilter
-import ru.shadowsparky.vbox.backend.data.RedisCache
+import ru.shadowsparky.vbox.backend.domain.CacheStorage
+import ru.shadowsparky.vbox.backend.domain.getOrCreate
 import ru.shadowsparky.vbox.shared.domain.VideoApi
 import ru.shadowsparky.vbox.shared.domain.model.VideoDetails
 import ru.shadowsparky.vbox.shared.domain.model.VideoLinksResponse
@@ -11,12 +12,12 @@ import kotlin.time.Duration.Companion.days
 class CacheVideoApi(
     private val wrapper: VideoApi,
     private val filter: FuzzyMovieSearchFilter,
-    private val redis: RedisCache
+    private val redis: CacheStorage
 ) : VideoApi {
     private val ttl = 3.days
 
     override suspend fun fetchNewVideos(query: String?): VideosResponse {
-        return redis.transact("$PREFIX:videos:$query", ttl) {
+        return redis.getOrCreate("$PREFIX:videos:$query", ttl) {
             wrapper.fetchNewVideos(query).let {
                 if (query != null) {
                     it.copy(items = filter.filter(it.items, query))
@@ -28,11 +29,11 @@ class CacheVideoApi(
     }
 
     override suspend fun fetchDetails(id: Long): VideoDetails {
-        return redis.transact("$PREFIX:details:$id", ttl) { wrapper.fetchDetails(id) }
+        return redis.getOrCreate("$PREFIX:details:$id", ttl) { wrapper.fetchDetails(id) }
     }
 
     override suspend fun fetchVideoLinks(id: Long, seasonId: Long?): VideoLinksResponse {
-        return redis.transact("$PREFIX:links:$id:$seasonId", ttl) {
+        return redis.getOrCreate("$PREFIX:links:$id:$seasonId", ttl) {
             wrapper.fetchVideoLinks(id, seasonId)
         }
     }

apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/recent/CacheRecentlyWatchedRepository.kt

@@ -3,14 +3,15 @@ package ru.shadowsparky.vbox.backend.data.recent
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flow
-import ru.shadowsparky.vbox.backend.data.RedisCache
+import ru.shadowsparky.vbox.backend.domain.CacheStorage
+import ru.shadowsparky.vbox.backend.domain.getOrCreate
 import ru.shadowsparky.vbox.shared.domain.RecentlyWatchedRepository
 import ru.shadowsparky.vbox.shared.domain.model.RecentlyWatchedInfo
 import kotlin.time.Duration.Companion.days
 
 class CacheRecentlyWatchedRepository(
     private val wrapper: RecentlyWatchedRepository,
-    private val redis: RedisCache,
+    private val redis: CacheStorage,
     userId: Long
 ) : RecentlyWatchedRepository {
     private val prefix = "recently_watched:$userId"
@@ -38,7 +39,7 @@ class CacheRecentlyWatchedRepository(
     ): Flow<List<RecentlyWatchedInfo>> {
         return flow {
             emit(
-                redis.transact(
+                redis.getOrCreate(
                     "$prefix:$movieId:$seasonId",
                     7.days
                 ) { wrapper.queryRecentlyWatched(movieId, seasonId).first() }
@@ -47,7 +48,7 @@ class CacheRecentlyWatchedRepository(
     }
 
     override suspend fun getAllRecentlyWatched(): List<RecentlyWatchedInfo> {
-        return redis.transact(
+        return redis.getOrCreate(
             prefix,
             7.days
         ) { wrapper.getAllRecentlyWatched() }

apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/saved/CacheSavedMovieRepository.kt

@@ -3,14 +3,15 @@ package ru.shadowsparky.vbox.backend.data.saved
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flow
-import ru.shadowsparky.vbox.backend.data.RedisCache
+import ru.shadowsparky.vbox.backend.domain.CacheStorage
+import ru.shadowsparky.vbox.backend.domain.getOrCreate
 import ru.shadowsparky.vbox.shared.domain.SavedMovieRepository
 import ru.shadowsparky.vbox.shared.domain.model.VideoDetails
 import kotlin.time.Duration.Companion.days
 
 class CacheSavedMovieRepository(
     private val wrapper: SavedMovieRepository,
-    private val redis: RedisCache,
+    private val redis: CacheStorage,
     userId: Long
 ) : SavedMovieRepository {
     private val ttl = 7.days
@@ -27,13 +28,13 @@ class CacheSavedMovieRepository(
 
     override fun getAll(): Flow<List<VideoDetails>> {
         return flow {
-            emit(redis.transact(prefix, ttl) { wrapper.getAll().first() })
+            emit(redis.getOrCreate(prefix, ttl) { wrapper.getAll().first() })
         }
     }
 
     override fun isSaved(id: Long): Flow<Boolean> {
         return flow {
-            emit(redis.transact("$prefix:$id") { wrapper.isSaved(id).first() })
+            emit(redis.getOrCreate("$prefix:$id", ttl) { wrapper.isSaved(id).first() })
         }
     }
 

apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/HttpModule.kt

@@ -3,9 +3,9 @@ package ru.shadowsparky.vbox.backend.di
 import org.koin.core.annotation.Factory
 import org.koin.core.annotation.Module
 import ru.shadowsparky.vbox.backend.data.FuzzyMovieSearchFilter
-import ru.shadowsparky.vbox.backend.data.RedisCache
-import ru.shadowsparky.vbox.backend.data.http.ExternalBackendApi
 import ru.shadowsparky.vbox.backend.data.http.CacheVideoApi
+import ru.shadowsparky.vbox.backend.data.http.ExternalBackendApi
+import ru.shadowsparky.vbox.backend.domain.CacheStorage
 import ru.shadowsparky.vbox.shared.domain.VideoApi
 
 @Module
@@ -15,8 +15,8 @@ class HttpModule {
     fun provideVideoApi(
         impl: ExternalBackendApi,
         filter: FuzzyMovieSearchFilter,
-        redisCache: RedisCache
+        cacheStorage: CacheStorage
     ): VideoApi {
-        return CacheVideoApi(impl, filter, redisCache)
+        return CacheVideoApi(impl, filter, cacheStorage)
     }
 }

apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/factory/RecentlyWatchedRepositoryFactory.kt

@@ -2,9 +2,9 @@ package ru.shadowsparky.vbox.backend.di.factory
 
 import org.koin.core.annotation.Factory
 import ru.shadowsparky.vbox.backend.AppDatabase
-import ru.shadowsparky.vbox.backend.data.RedisCache
 import ru.shadowsparky.vbox.backend.data.recent.BackendRecentlyWatchedRepository
 import ru.shadowsparky.vbox.backend.data.recent.CacheRecentlyWatchedRepository
+import ru.shadowsparky.vbox.backend.domain.CacheStorage
 import ru.shadowsparky.vbox.shared.di.factory.DispatcherProvider
 import ru.shadowsparky.vbox.shared.domain.RecentlyWatchedRepository
 import ru.shadowsparky.vbox.shared.domain.RemoteEventHandler
@@ -14,13 +14,13 @@ class RecentlyWatchedRepositoryFactory(
     private val db: AppDatabase,
     private val dispatcherProvider: DispatcherProvider,
     private val remoteEventHandler: RemoteEventHandler,
-    private val redisCache: RedisCache
+    private val cacheStorage: CacheStorage
 ) {
     fun create(userId: Long): RecentlyWatchedRepository {
         val impl = BackendRecentlyWatchedRepository(db, userId, dispatcherProvider, remoteEventHandler)
         return CacheRecentlyWatchedRepository(
             impl,
-            redisCache,
+            cacheStorage,
             userId
         )
     }

apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/factory/SavedMovieRepositoryFactory.kt

@@ -4,9 +4,9 @@ import org.koin.core.annotation.Factory
 import ru.shadowsparky.backend.data.Logger
 import ru.shadowsparky.vbox.backend.AppDatabase
 import ru.shadowsparky.vbox.backend.data.BackendRemoteEventHandler
-import ru.shadowsparky.vbox.backend.data.RedisCache
 import ru.shadowsparky.vbox.backend.data.saved.BackendSavedMovieRepository
 import ru.shadowsparky.vbox.backend.data.saved.CacheSavedMovieRepository
+import ru.shadowsparky.vbox.backend.domain.CacheStorage
 import ru.shadowsparky.vbox.shared.di.factory.DispatcherProvider
 import ru.shadowsparky.vbox.shared.domain.SavedMovieRepository
 
@@ -16,10 +16,10 @@ class SavedMovieRepositoryFactory(
     private val dispatcherProvider: DispatcherProvider,
     private val logger: Logger,
     private val eventHandler: BackendRemoteEventHandler,
-    private val redisCache: RedisCache
+    private val cacheStorage: CacheStorage
 ) {
     fun create(userId: Long): SavedMovieRepository {
         val impl = BackendSavedMovieRepository(db, userId, dispatcherProvider, logger, eventHandler)
-        return CacheSavedMovieRepository(impl, redisCache, userId)
+        return CacheSavedMovieRepository(impl, cacheStorage, userId)
     }
 }

apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/domain/CacheStorage.kt

@@ -0,0 +1,29 @@
+package ru.shadowsparky.vbox.backend.domain
+
+import kotlinx.serialization.KSerializer
+import kotlinx.serialization.serializer
+import kotlin.time.Duration
+
+interface CacheStorage {
+    suspend fun <T> getOrCreate(
+        key: String,
+        serializer: KSerializer<T>,
+        ttl: Duration? = null,
+        createNew: suspend () -> T,
+    ): T
+    suspend fun delete(key: String)
+    suspend fun close()
+}
+
+suspend inline fun <reified T> CacheStorage.getOrCreate(
+    key: String,
+    ttl: Duration,
+    noinline createNew: suspend () -> T,
+): T {
+    return getOrCreate(
+        key = key,
+        serializer = serializer<T>(),
+        ttl = ttl,
+        createNew = createNew
+    )
+}