diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/Main.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/Main.kt index e3baeb2..1a46b38 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/Main.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/Main.kt @@ -27,7 +27,7 @@ fun Application.module() { runBlocking { koin.getOrNull()?.close() } } val koin = object : KoinComponent {} - configureSerialization() + configureSerialization(koin.get()) configureJwt(koin.get()) configureWebSocket(koin.get()) configureRouting(koin.get(), koin.get()) diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/BackendUpdateFetcher.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/BackendUpdateFetcher.kt index 2198ece..05735a9 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/BackendUpdateFetcher.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/BackendUpdateFetcher.kt @@ -1,6 +1,9 @@ package ru.shadowsparky.vbox.backend.data import io.ktor.http.HttpStatusCode +import io.ktor.util.cio.readChannel +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withContext import kotlinx.serialization.json.Json import org.koin.core.annotation.Factory @@ -28,8 +31,18 @@ class BackendUpdateFetcher( private val dispatcherProvider: DispatcherProvider, private val userId: Long ) : UpdateFetcher { + private val fetchMutex = Mutex() + private val updateFileMutex = Mutex() - override suspend fun fetch(versionCode: Long): UpdateInfo = withContext(dispatcherProvider.io) { + override suspend fun fetch(versionCode: Long): UpdateInfo { + updateInfo?.let { return it } + fetchMutex.withLock { + updateInfo?.let { return it } + return fetchInternal().apply { updateInfo = this } + } + } + + private suspend fun fetchInternal(): UpdateInfo = withContext(dispatcherProvider.io) { val configPath = envFetcher.get("UPDATE_CONFIG") if (configPath.isBlank()) { UpdateInfo() @@ -38,11 +51,25 @@ class BackendUpdateFetcher( } } - suspend fun updateFile(): File = withContext(dispatcherProvider.io) { + suspend fun updateFile(): File { + updateFile?.let { return it } + updateFileMutex.withLock { + updateFile?.let { return it } + return updateFileInternal().apply { updateFile = this } + } + } + + private suspend fun updateFileInternal(): File = withContext(dispatcherProvider.io) { val updateFilePath = envFetcher.get("UPDATE_FILE") if (updateFilePath.isBlank()) throw HttpException(HttpStatusCode.NotFound.value, "Update not found") val file = File(updateFilePath) + file.readChannel() if (!file.exists()) throw HttpException(HttpStatusCode.NotFound.value, "Update not exists") file } + + private companion object { + var updateInfo: UpdateInfo? = null + var updateFile: File? = null + } } diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/FuzzySearchFilter.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/FuzzySearchFilter.kt index c1410c9..1e7d293 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/FuzzySearchFilter.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/FuzzySearchFilter.kt @@ -49,6 +49,6 @@ class FuzzyMovieSearchFilter { } private companion object { - const val THRESHOLD = 55 + const val THRESHOLD = 70 } } diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/RedisCacheStorage.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/RedisCacheStorage.kt index 8c1a704..3c05afa 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/RedisCacheStorage.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/RedisCacheStorage.kt @@ -3,7 +3,7 @@ package ru.shadowsparky.vbox.backend.data import io.lettuce.core.ExperimentalLettuceCoroutinesApi import io.lettuce.core.RedisClient import io.lettuce.core.SetArgs -import io.lettuce.core.api.StatefulRedisConnection +import io.lettuce.core.api.StatefulConnection import io.lettuce.core.api.coroutines import io.lettuce.core.api.coroutines.RedisCoroutinesCommands import kotlinx.coroutines.sync.Mutex @@ -18,6 +18,7 @@ import ru.shadowsparky.domain.DispatcherProvider import ru.shadowsparky.vbox.backend.domain.CacheStorage import java.util.concurrent.ConcurrentHashMap import kotlin.time.Duration +import kotlin.time.toJavaDuration @OptIn(ExperimentalLettuceCoroutinesApi::class) @Single @@ -27,25 +28,23 @@ class RedisCacheStorage( private val json: Json ) : CacheStorage { private val logger = LoggerFactory.getLogger("RedisCacheStorage") - - private var client: RedisClient? = null - private var connection: StatefulRedisConnection? = null - private var commands: RedisCoroutinesCommands? = null + private var session: Session? = null private val connectionMutex = Mutex() + private val locks = ConcurrentHashMap() override suspend fun getOrCreate( key: String, serializer: KSerializer, - ttl: Duration?, - createNew: suspend () -> T, + ttl: Duration, + factory: 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 { + factory().also { write(key, it, serializer, ttl) } } finally { @@ -58,16 +57,11 @@ class RedisCacheStorage( key: String, value: T, serializer: KSerializer, - duration: Duration? = null + duration: Duration ) { - 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) - } + val args = SetArgs().ex(duration.toJavaDuration()) + getCommands().set(key, rawValue, args) } private suspend fun read(key: String, serializer: KSerializer): T? { @@ -76,50 +70,40 @@ class RedisCacheStorage( } override suspend fun delete(key: String) { - (getCommands().del(key) ?: 0L) > 0L + getCommands().del(key) } override suspend fun close() { connectionMutex.withLock { - if (connection != null || client != null) { + session?.let { 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 + runCatching { + it.connection.close() + it.client.shutdown() } + session = null } } } private suspend fun getCommands(): RedisCoroutinesCommands { - commands?.let { return it } + session?.commands?.let { return it } return connectionMutex.withLock { - commands?.let { return@withLock it } + session?.commands?.let { return@withLock it } withContext(dispatcherProvider.io) { - try { - val redisUri = envFetcher.get("REDIS_URI", "redis://redis:6379") - logger.info("Динамическое подключение к Redis: $redisUri") - val redisClient = RedisClient.create(redisUri) - val conn = redisClient.connect() - val cmds = conn.coroutines() - - client = redisClient - connection = conn - commands = cmds - - logger.info("Успешно подключено к Redis") - cmds - } catch (e: Exception) { - logger.error("Ошибка динамического подключения к Redis: ${e.message}", e) - throw e + val redisUri = envFetcher.get("REDIS_URI", "redis://redis:6379") + val redisClient = RedisClient.create(redisUri) + val conn = redisClient.connect() + conn.coroutines().apply { + session = Session(redisClient, conn, this) } } } } + + private class Session( + val client: RedisClient, + val connection: StatefulConnection, + val commands: RedisCoroutinesCommands + ) } diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/domain/CacheStorage.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/domain/CacheStorage.kt index 35a0125..78c42ce 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/domain/CacheStorage.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/domain/CacheStorage.kt @@ -8,8 +8,8 @@ interface CacheStorage { suspend fun getOrCreate( key: String, serializer: KSerializer, - ttl: Duration? = null, - createNew: suspend () -> T, + ttl: Duration, + factory: suspend () -> T ): T suspend fun delete(key: String) suspend fun close() @@ -18,12 +18,12 @@ interface CacheStorage { suspend inline fun CacheStorage.getOrCreate( key: String, ttl: Duration, - noinline createNew: suspend () -> T, + noinline factory: suspend () -> T, ): T { return getOrCreate( key = key, serializer = serializer(), ttl = ttl, - createNew = createNew + factory = factory ) } diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/Jwt.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/Jwt.kt index f6f5e85..8af777c 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/Jwt.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/Jwt.kt @@ -18,18 +18,16 @@ import ru.shadowsparky.vbox.shared.domain.model.ServerExceptionInfo const val AUTH_JWT_NAME = "auth-jwt" -fun Application.configureJwt(authEntryPoint: AuthEntryPoint) { +fun Application.configureJwt(authEntryPoint: AuthEntryPoint) = with(authEntryPoint) { install(Authentication) { jwt(AUTH_JWT_NAME) { - val jwtInfo = authEntryPoint.jwtInfo realm = jwtInfo.realm - verifier(authEntryPoint.tokenVerifier.verifier) + verifier(tokenVerifier.verifier) validate { credential -> - val tokenRepo = authEntryPoint.loginVerifier val login = credential.payload.getClaim(LOGIN_NAME).asString() try { if (login != null) { - tokenRepo.verify(login) + loginVerifier.verify(login) if (credential.payload.expiresAt == null) { throw VerifyTokenException("Static tokens not supported!") } diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/Serialization.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/Serialization.kt index bcd316d..89533f1 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/Serialization.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/Serialization.kt @@ -4,9 +4,8 @@ import io.ktor.serialization.kotlinx.json.json import io.ktor.server.application.Application import io.ktor.server.application.install import io.ktor.server.plugins.contentnegotiation.ContentNegotiation +import kotlinx.serialization.json.Json -fun Application.configureSerialization() { - install(ContentNegotiation) { - json() - } +fun Application.configureSerialization(json: Json) { + install(ContentNegotiation) { json(json) } } diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/WebSocket.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/WebSocket.kt index 54e3c13..f326110 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/WebSocket.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/WebSocket.kt @@ -13,24 +13,19 @@ import io.ktor.server.websocket.webSocket import io.ktor.websocket.Frame import io.ktor.websocket.readText import kotlinx.coroutines.CompletableDeferred -import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.channels.ReceiveChannel import kotlinx.coroutines.channels.SendChannel import kotlinx.serialization.json.Json import ru.shadowsparky.backend.data.TokenVerifier +import ru.shadowsparky.http.domain.BadRequestException import ru.shadowsparky.http.domain.HttpException import ru.shadowsparky.vbox.backend.data.SessionRegistry import ru.shadowsparky.vbox.backend.data.eventLogger import ru.shadowsparky.vbox.backend.di.WebSocketEntryPoint import ru.shadowsparky.vbox.shared.domain.AuthRequest import ru.shadowsparky.vbox.shared.domain.AuthResponse -import ru.shadowsparky.vbox.shared.domain.RecentlyWatchedRepository import ru.shadowsparky.vbox.shared.domain.RemoteEventHandler -import ru.shadowsparky.vbox.shared.domain.SavedMovieRepository -import ru.shadowsparky.vbox.shared.domain.SearchRepository import ru.shadowsparky.vbox.shared.domain.USER_ID_ARG -import ru.shadowsparky.vbox.shared.domain.tag.MovieTagRepository -import ru.shadowsparky.vbox.shared.domain.tag.UserTagRepository import kotlin.time.Duration.Companion.seconds fun Application.configureWebSocket(socketEntryPoint: WebSocketEntryPoint) = with(socketEntryPoint) { @@ -41,15 +36,6 @@ fun Application.configureWebSocket(socketEntryPoint: WebSocketEntryPoint) = with masking = false } routing { - listOf( - "${SearchRepository.PREFIX}/onChange", - "${RecentlyWatchedRepository.PREFIX}/onChange", - "${SavedMovieRepository.PREFIX}/onChange", - "${UserTagRepository.PREFIX}/onChange", - "${MovieTagRepository.PREFIX}/onChange" - ).forEach { - webSocket(it) { awaitCancellation() } - } webSocket(RemoteEventHandler.ON_EVENT) { val userId = incoming.authFlow(json, tokenVerifier, outgoing) val session = SessionRegistry.Writer { text -> outgoing.trySend(Frame.Text(text)) } @@ -75,7 +61,7 @@ private suspend fun ReceiveChannel.authFlow( tokenVerifier: TokenVerifier, sendChannel: SendChannel ): Long { - val rawRequest = receiveTextOrNull() ?: throw HttpException(HttpStatusCode.BadRequest) + val rawRequest = receiveTextOrNull() ?: throw BadRequestException("Authentication request required") val request = runCatching { json.decodeFromString(rawRequest) }.getOrNull() val decodedJwt = if (request == null) { tokenVerifier.verify(rawRequest)