diff --git a/apps/giga-wear/backend/src/main/kotlin/ru/shadowsparky/gigawear/backend/presentation/ChatRouting.kt b/apps/giga-wear/backend/src/main/kotlin/ru/shadowsparky/gigawear/backend/presentation/ChatRouting.kt index 7f4a908..ba65c8e 100644 --- a/apps/giga-wear/backend/src/main/kotlin/ru/shadowsparky/gigawear/backend/presentation/ChatRouting.kt +++ b/apps/giga-wear/backend/src/main/kotlin/ru/shadowsparky/gigawear/backend/presentation/ChatRouting.kt @@ -2,17 +2,16 @@ package ru.shadowsparky.gigawear.backend.presentation import io.ktor.server.request.receive import io.ktor.server.response.respond -import io.ktor.server.routing.Routing +import io.ktor.server.routing.Route import io.ktor.server.routing.post import ru.shadowsparky.chat.backend.domain.ProcessUserMessageUseCase import ru.shadowsparky.chat.domain.Message private const val MOCK_USER_ID = -1L -fun Routing.setupChat(useCase: ProcessUserMessageUseCase) { +fun Route.setupChat(useCase: ProcessUserMessageUseCase) { post("chat") { val message = call.receive().copy(timestamp = System.currentTimeMillis()) call.respond(useCase.execute(message, MOCK_USER_ID)) } - } diff --git a/apps/giga-wear/backend/src/main/kotlin/ru/shadowsparky/gigawear/backend/presentation/UpdateRouting.kt b/apps/giga-wear/backend/src/main/kotlin/ru/shadowsparky/gigawear/backend/presentation/UpdateRouting.kt new file mode 100644 index 0000000..43b0561 --- /dev/null +++ b/apps/giga-wear/backend/src/main/kotlin/ru/shadowsparky/gigawear/backend/presentation/UpdateRouting.kt @@ -0,0 +1,17 @@ +package ru.shadowsparky.gigawear.backend.presentation + +import io.ktor.server.request.receive +import io.ktor.server.response.respond +import io.ktor.server.routing.Route +import io.ktor.server.routing.Routing +import io.ktor.server.routing.post +import ru.shadowsparky.chat.backend.domain.ProcessUserMessageUseCase +import ru.shadowsparky.chat.domain.Message + +fun Route.setupUpdate(useCase: ProcessUserMessageUseCase) { + post("chat") { + val message = call.receive().copy(timestamp = System.currentTimeMillis()) + call.respond(useCase.execute(message, MOCK_USER_ID)) + } +} + 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 1a46b38..4d78bae 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 @@ -2,17 +2,19 @@ package ru.shadowsparky.vbox.backend import io.ktor.server.application.Application import io.ktor.server.application.ApplicationStopping +import io.ktor.server.routing.routing import kotlinx.coroutines.runBlocking import org.koin.core.annotation.KoinApplication import org.koin.core.component.KoinComponent import org.koin.core.component.get import org.koin.plugin.module.dsl.startKoin +import ru.shadowsparky.backend.presentation.installStatusPages import ru.shadowsparky.koin 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 import ru.shadowsparky.vbox.backend.presentation.configureWebSocket +import ru.shadowsparky.vbox.backend.presentation.routing.setupAuthMethods @KoinApplication(modules = [BackendModule::class]) object VBoxBackend @@ -30,5 +32,6 @@ fun Application.module() { configureSerialization(koin.get()) configureJwt(koin.get()) configureWebSocket(koin.get()) - configureRouting(koin.get(), koin.get()) + installStatusPages() + routing { setupAuthMethods(koin.get(), koin.get()) } } diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/BackendSearchRepository.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/BackendSearchRepository.kt index b1caec9..91a6cea 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/BackendSearchRepository.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/BackendSearchRepository.kt @@ -4,10 +4,11 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.flow import kotlinx.coroutines.withContext +import ru.shadowsparky.backend.domain.RemoteEventHandler +import ru.shadowsparky.backend.domain.notify import ru.shadowsparky.vbox.backend.AppDatabase import ru.shadowsparky.vbox.shared.di.factory.DispatcherProvider import ru.shadowsparky.vbox.shared.domain.RemoteEvent -import ru.shadowsparky.vbox.shared.domain.RemoteEventHandler import ru.shadowsparky.vbox.shared.domain.SearchRepository import java.sql.SQLException @@ -59,6 +60,9 @@ class BackendSearchRepository( } private suspend fun notifyChanged() { - eventHandler.notify(RemoteEvent.OnSearch(System.currentTimeMillis(), userId)) + eventHandler.notify( + RemoteEvent.OnSearch(System.currentTimeMillis(), userId), + RemoteEvent.serializer() + ) } } diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/recent/BackendRecentlyWatchedRepository.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/recent/BackendRecentlyWatchedRepository.kt index 1272091..b838e97 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/recent/BackendRecentlyWatchedRepository.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/recent/BackendRecentlyWatchedRepository.kt @@ -3,11 +3,12 @@ package ru.shadowsparky.vbox.backend.data.recent import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.flow import kotlinx.coroutines.withContext +import ru.shadowsparky.backend.domain.RemoteEventHandler +import ru.shadowsparky.backend.domain.notify import ru.shadowsparky.vbox.backend.AppDatabase import ru.shadowsparky.vbox.shared.di.factory.DispatcherProvider import ru.shadowsparky.vbox.shared.domain.RecentlyWatchedRepository import ru.shadowsparky.vbox.shared.domain.RemoteEvent -import ru.shadowsparky.vbox.shared.domain.RemoteEventHandler import ru.shadowsparky.vbox.shared.domain.model.RecentlyWatchedInfo import java.sql.SQLException @@ -85,7 +86,8 @@ class BackendRecentlyWatchedRepository( System.currentTimeMillis(), userId, seasonId - ) + ), + RemoteEvent.serializer() ) } } \ No newline at end of file diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/saved/BackendSavedMovieRepository.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/saved/BackendSavedMovieRepository.kt index f404a8f..4bf7334 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/saved/BackendSavedMovieRepository.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/saved/BackendSavedMovieRepository.kt @@ -7,10 +7,11 @@ import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.map import kotlinx.coroutines.withContext import ru.shadowsparky.backend.data.Logger +import ru.shadowsparky.backend.domain.RemoteEventHandler +import ru.shadowsparky.backend.domain.notify import ru.shadowsparky.vbox.backend.AppDatabase import ru.shadowsparky.vbox.shared.di.factory.DispatcherProvider import ru.shadowsparky.vbox.shared.domain.RemoteEvent -import ru.shadowsparky.vbox.shared.domain.RemoteEventHandler import ru.shadowsparky.vbox.shared.domain.SavedMovieRepository import ru.shadowsparky.vbox.shared.domain.model.SERIAL_FLAG import ru.shadowsparky.vbox.shared.domain.model.VideoDetails @@ -50,7 +51,10 @@ class BackendSavedMovieRepository( details.id, userId ) - eventHandler.notify(RemoteEvent.OnSaved(userId, details.id)) + eventHandler.notify( + RemoteEvent.OnSaved(userId, details.id), + RemoteEvent.serializer() + ) } catch (_: SQLException) { } } @@ -67,7 +71,10 @@ class BackendSavedMovieRepository( logger.debug(TAG, "remove(${id})") withContext(dispatcherProvider.io) { db.saved_movieQueries.removeSavedMovie(userId, id) - eventHandler.notify(RemoteEvent.OnSaved(userId, id)) + eventHandler.notify( + RemoteEvent.OnSaved(userId, id), + RemoteEvent.serializer() + ) } } diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/tags/BackendMovieTagRepository.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/tags/BackendMovieTagRepository.kt index dc5c2f1..a95896d 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/tags/BackendMovieTagRepository.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/tags/BackendMovieTagRepository.kt @@ -8,11 +8,12 @@ import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.map import kotlinx.coroutines.withContext import org.koin.core.annotation.Factory +import ru.shadowsparky.backend.domain.RemoteEventHandler +import ru.shadowsparky.backend.domain.notify import ru.shadowsparky.domain.DispatcherProvider import ru.shadowsparky.http.domain.BadRequestException import ru.shadowsparky.vbox.backend.AppDatabase import ru.shadowsparky.vbox.shared.domain.RemoteEvent -import ru.shadowsparky.vbox.shared.domain.RemoteEventHandler import ru.shadowsparky.vbox.shared.domain.tag.MovieTagRepository import ru.shadowsparky.vbox.shared.domain.tag.TagInfo @@ -62,12 +63,18 @@ class BackendMovieTagRepository( override suspend fun link(movieId: Long, tag: String) { dbQueries.linkMovieTag(movieId, findTagId(tag), userId).await() - movieEventHandler.notify(RemoteEvent.OnMovieTag(userId, movieId)) + movieEventHandler.notify( + RemoteEvent.OnMovieTag(userId, movieId), + RemoteEvent.serializer() + ) } override suspend fun unlink(movieId: Long, tag: String) { dbQueries.unlinkMovieTag(movieId, findTagId(tag), userId).await() - movieEventHandler.notify(RemoteEvent.OnMovieTag(userId, movieId)) + movieEventHandler.notify( + RemoteEvent.OnMovieTag(userId, movieId), + RemoteEvent.serializer() + ) } private suspend fun findTagId(tag: String): Long = withContext(dispatcherProvider.io) { diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/tags/BackendUserTagRepository.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/tags/BackendUserTagRepository.kt index 837b415..e1ffb23 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/tags/BackendUserTagRepository.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/tags/BackendUserTagRepository.kt @@ -5,10 +5,11 @@ import app.cash.sqldelight.coroutines.mapToList import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.map import org.koin.core.annotation.Factory +import ru.shadowsparky.backend.domain.RemoteEventHandler +import ru.shadowsparky.backend.domain.notify import ru.shadowsparky.vbox.backend.AppDatabase import ru.shadowsparky.vbox.shared.di.factory.DispatcherProvider import ru.shadowsparky.vbox.shared.domain.RemoteEvent -import ru.shadowsparky.vbox.shared.domain.RemoteEventHandler import ru.shadowsparky.vbox.shared.domain.tag.TagInfo import ru.shadowsparky.vbox.shared.domain.tag.UserTagRepository @@ -47,16 +48,25 @@ class BackendUserTagRepository( override suspend fun add(tag: String) { queries.insertUserTag(userId, tag).await() - userTagEventHandler.notify(RemoteEvent.OnUserTag(userId)) + userTagEventHandler.notify( + RemoteEvent.OnUserTag(userId), + RemoteEvent.serializer() + ) } override suspend fun edit(oldTag: String, newTag: String) { queries.updateTagText(newTag, oldTag, userId).await() - userTagEventHandler.notify(RemoteEvent.OnUserTag(userId)) + userTagEventHandler.notify( + RemoteEvent.OnUserTag(userId), + RemoteEvent.serializer() + ) } override suspend fun delete(tag: String) { queries.deleteTagById(tag, userId).await() - userTagEventHandler.notify(RemoteEvent.OnUserTag(userId)) + userTagEventHandler.notify( + RemoteEvent.OnUserTag(userId), + RemoteEvent.serializer() + ) } } diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/EntryPoint.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/EntryPoint.kt index 95a5d7d..eb3654e 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/EntryPoint.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/EntryPoint.kt @@ -3,11 +3,12 @@ package ru.shadowsparky.vbox.backend.di import kotlinx.serialization.json.Json import org.koin.core.annotation.Single import ru.shadowsparky.backend.data.TokenVerifier +import ru.shadowsparky.backend.data.event.SessionRegistry import ru.shadowsparky.backend.domain.JwtInfo import ru.shadowsparky.backend.domain.LoginVerifier +import ru.shadowsparky.backend.domain.RemoteEventHandler import ru.shadowsparky.chat.backend.domain.ProcessUserMessageUseCase import ru.shadowsparky.vbox.backend.data.BackendUpdateFetcherFactory -import ru.shadowsparky.vbox.backend.data.SessionRegistry import ru.shadowsparky.vbox.backend.data.auth.AuthTokenRepositoryFactory import ru.shadowsparky.vbox.backend.data.chat.BackendChatRepositoryFactory import ru.shadowsparky.vbox.backend.data.tags.MovieTagRepositoryFactory @@ -15,7 +16,6 @@ import ru.shadowsparky.vbox.backend.data.tags.UserTagRepositoryFactory import ru.shadowsparky.vbox.backend.di.factory.RecentlyWatchedRepositoryFactory import ru.shadowsparky.vbox.backend.di.factory.SavedMovieRepositoryFactory import ru.shadowsparky.vbox.backend.di.factory.SearchRepositoryFactory -import ru.shadowsparky.vbox.shared.domain.RemoteEventHandler import ru.shadowsparky.vbox.shared.domain.VideoApi @Single diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/factory/RecentlyWatchedRepositoryFactory.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/factory/RecentlyWatchedRepositoryFactory.kt index ead2198..f4f126f 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/factory/RecentlyWatchedRepositoryFactory.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/factory/RecentlyWatchedRepositoryFactory.kt @@ -1,13 +1,13 @@ package ru.shadowsparky.vbox.backend.di.factory import org.koin.core.annotation.Factory +import ru.shadowsparky.backend.domain.RemoteEventHandler import ru.shadowsparky.vbox.backend.AppDatabase 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 @Factory class RecentlyWatchedRepositoryFactory( diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/factory/SavedMovieRepositoryFactory.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/factory/SavedMovieRepositoryFactory.kt index f7403ae..a39a836 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/factory/SavedMovieRepositoryFactory.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/factory/SavedMovieRepositoryFactory.kt @@ -2,8 +2,8 @@ package ru.shadowsparky.vbox.backend.di.factory import org.koin.core.annotation.Factory import ru.shadowsparky.backend.data.Logger +import ru.shadowsparky.backend.data.event.BackendRemoteEventHandler import ru.shadowsparky.vbox.backend.AppDatabase -import ru.shadowsparky.vbox.backend.data.BackendRemoteEventHandler import ru.shadowsparky.vbox.backend.data.saved.BackendSavedMovieRepository import ru.shadowsparky.vbox.backend.data.saved.CacheSavedMovieRepository import ru.shadowsparky.vbox.backend.domain.CacheStorage diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/factory/SearchRepositoryFactory.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/factory/SearchRepositoryFactory.kt index 82d22da..366d528 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/factory/SearchRepositoryFactory.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/di/factory/SearchRepositoryFactory.kt @@ -1,10 +1,10 @@ package ru.shadowsparky.vbox.backend.di.factory import org.koin.core.annotation.Factory +import ru.shadowsparky.backend.domain.RemoteEventHandler import ru.shadowsparky.vbox.backend.AppDatabase import ru.shadowsparky.vbox.backend.data.BackendSearchRepository import ru.shadowsparky.vbox.shared.di.factory.DispatcherProvider -import ru.shadowsparky.vbox.shared.domain.RemoteEventHandler import ru.shadowsparky.vbox.shared.domain.SearchRepository @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 8af777c..76f8712 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 @@ -1,59 +1,17 @@ package ru.shadowsparky.vbox.backend.presentation -import io.ktor.http.HttpStatusCode import io.ktor.server.application.Application import io.ktor.server.application.ApplicationCall -import io.ktor.server.application.install -import io.ktor.server.auth.Authentication -import io.ktor.server.auth.authentication -import io.ktor.server.auth.jwt.JWTPrincipal -import io.ktor.server.auth.jwt.jwt -import io.ktor.server.response.respond -import ru.shadowsparky.backend.domain.INVALID_TOKEN_MSG -import ru.shadowsparky.backend.domain.LOGIN_NAME -import ru.shadowsparky.backend.domain.VerifyTokenException +import ru.shadowsparky.backend.presentation.configureJwt +import ru.shadowsparky.backend.presentation.obtainUserId import ru.shadowsparky.vbox.backend.di.AuthEntryPoint -import ru.shadowsparky.vbox.shared.domain.USER_ID_ARG -import ru.shadowsparky.vbox.shared.domain.model.ServerExceptionInfo const val AUTH_JWT_NAME = "auth-jwt" fun Application.configureJwt(authEntryPoint: AuthEntryPoint) = with(authEntryPoint) { - install(Authentication) { - jwt(AUTH_JWT_NAME) { - realm = jwtInfo.realm - verifier(tokenVerifier.verifier) - validate { credential -> - val login = credential.payload.getClaim(LOGIN_NAME).asString() - try { - if (login != null) { - loginVerifier.verify(login) - if (credential.payload.expiresAt == null) { - throw VerifyTokenException("Static tokens not supported!") - } - JWTPrincipal(credential.payload) - } else { - null - } - } catch (e: VerifyTokenException) { - routingLogger.error("verify token failed ${e.message}") - null - } - } - challenge { _, _ -> - call.respond( - HttpStatusCode.Unauthorized, - ServerExceptionInfo(INVALID_TOKEN_MSG) - ) - } - } - } + configureJwt(jwtInfo, tokenVerifier, loginVerifier) } fun ApplicationCall.obtainUserId(): Long { - val principal = authentication.principal() - ?: throw VerifyTokenException(INVALID_TOKEN_MSG) - return principal.payload - .getClaim(USER_ID_ARG) - .asLong() + return obtainUserId() } diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/Routing.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/Routing.kt deleted file mode 100644 index 497d337..0000000 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/Routing.kt +++ /dev/null @@ -1,42 +0,0 @@ -package ru.shadowsparky.vbox.backend.presentation - -import io.ktor.http.HttpStatusCode -import io.ktor.server.application.Application -import io.ktor.server.application.install -import io.ktor.server.plugins.statuspages.StatusPages -import io.ktor.server.response.respond -import io.ktor.server.routing.routing -import org.slf4j.Logger -import org.slf4j.LoggerFactory -import ru.shadowsparky.http.domain.HttpException -import ru.shadowsparky.vbox.backend.di.AuthEntryPoint -import ru.shadowsparky.vbox.backend.di.RoutingEntryPoint -import ru.shadowsparky.vbox.backend.presentation.routing.setupAuthMethods -import ru.shadowsparky.vbox.shared.domain.model.ServerExceptionInfo - -val routingLogger: Logger = LoggerFactory.getLogger("routing") - -fun Application.configureRouting( - routingEntryPoint: RoutingEntryPoint, - authEntryPoint: AuthEntryPoint -) { - install(StatusPages) { - exception { call, cause -> - routingLogger.error("http exception occurred. returns ${cause.httpCode}", cause) - call.respond( - status = HttpStatusCode.fromValue(cause.httpCode), - message = ServerExceptionInfo(cause.message ?: "Неизвестная ошибка") - ) - } - exception { call, cause -> - routingLogger.error("error occurred. returns 500...", cause) - call.respond( - status = HttpStatusCode.InternalServerError, - message = ServerExceptionInfo(cause.message ?: "Неизвестная ошибка") - ) - } - } - routing { - setupAuthMethods(routingEntryPoint, authEntryPoint) - } -} 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 f326110..0cf5067 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 @@ -1,100 +1,10 @@ package ru.shadowsparky.vbox.backend.presentation -import com.auth0.jwt.exceptions.TokenExpiredException -import com.auth0.jwt.interfaces.DecodedJWT -import io.ktor.http.HttpStatusCode import io.ktor.server.application.Application -import io.ktor.server.application.install -import io.ktor.server.routing.routing -import io.ktor.server.websocket.WebSockets -import io.ktor.server.websocket.pingPeriod -import io.ktor.server.websocket.timeout -import io.ktor.server.websocket.webSocket -import io.ktor.websocket.Frame -import io.ktor.websocket.readText -import kotlinx.coroutines.CompletableDeferred -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.backend.presentation.setupWebSocket 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.RemoteEventHandler -import ru.shadowsparky.vbox.shared.domain.USER_ID_ARG -import kotlin.time.Duration.Companion.seconds +import ru.shadowsparky.vbox.shared.domain.ON_EVENT fun Application.configureWebSocket(socketEntryPoint: WebSocketEntryPoint) = with(socketEntryPoint) { - install(WebSockets) { - pingPeriod = (30).seconds - timeout = (30).seconds - maxFrameSize = Long.MAX_VALUE - masking = false - } - routing { - webSocket(RemoteEventHandler.ON_EVENT) { - val userId = incoming.authFlow(json, tokenVerifier, outgoing) - val session = SessionRegistry.Writer { text -> outgoing.trySend(Frame.Text(text)) } - sessionRegistry.put(userId, session) - val deferred = CompletableDeferred() - try { - outgoing.invokeOnClose { deferred.complete(null) } - deferred.await() - } finally { - sessionRegistry.remove(userId, session) - } - } - } -} - -private suspend fun ReceiveChannel.receiveTextOrNull(): String? { - val frame = receive() - return if (frame is Frame.Text) frame.readText() else null -} - -private suspend fun ReceiveChannel.authFlow( - json: Json, - tokenVerifier: TokenVerifier, - sendChannel: SendChannel -): Long { - val rawRequest = receiveTextOrNull() ?: throw BadRequestException("Authentication request required") - val request = runCatching { json.decodeFromString(rawRequest) }.getOrNull() - val decodedJwt = if (request == null) { - tokenVerifier.verify(rawRequest) - } else { - authFlowV2(request, json, tokenVerifier, sendChannel) - } - return decodedJwt.getClaim(USER_ID_ARG).asLong() -} - -private const val MAX_ATTEMPTS = 3 - -private suspend fun ReceiveChannel.authFlowV2( - initialRequest: AuthRequest, - json: Json, - tokenVerifier: TokenVerifier, - sendChannel: SendChannel -): DecodedJWT { - var currentRequest = initialRequest - repeat(MAX_ATTEMPTS) { attempt -> - try { - val jwt = tokenVerifier.verify(currentRequest.token) - sendChannel.send(Frame.Text(json.encodeToString(AuthResponse(true)))) - return jwt - } catch (e: TokenExpiredException) { - eventLogger.error("token expired. attempt=$attempt", e) - if (attempt == MAX_ATTEMPTS - 1) return@repeat - sendChannel.send(Frame.Text(json.encodeToString(AuthResponse(false, e.message)))) - val nextRaw = receiveTextOrNull() ?: return@repeat - currentRequest = json.decodeFromString(nextRaw) - } catch (e: Exception) { - eventLogger.error("unable to verify token", e) - throw HttpException(HttpStatusCode.Unauthorized) - } - } - throw HttpException(HttpStatusCode.Unauthorized) + setupWebSocket(ON_EVENT, json, tokenVerifier, sessionRegistry) } diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/routing/ChatRouting.kt b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/routing/ChatRouting.kt index 5b2eea6..2690b3e 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/routing/ChatRouting.kt +++ b/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/presentation/routing/ChatRouting.kt @@ -5,6 +5,8 @@ import io.ktor.server.response.respond import io.ktor.server.routing.Route import io.ktor.server.routing.get import io.ktor.server.routing.post +import ru.shadowsparky.backend.domain.RemoteEventHandler +import ru.shadowsparky.backend.domain.notify import ru.shadowsparky.chat.backend.domain.ProcessUserMessageUseCase import ru.shadowsparky.chat.domain.ChatRepository import ru.shadowsparky.chat.domain.ChatRoles @@ -15,7 +17,6 @@ import ru.shadowsparky.vbox.backend.data.chat.BackendChatRepositoryFactory import ru.shadowsparky.vbox.backend.presentation.obtainUserId import ru.shadowsparky.vbox.shared.domain.DYNAMIC_PREFIX import ru.shadowsparky.vbox.shared.domain.RemoteEvent -import ru.shadowsparky.vbox.shared.domain.RemoteEventHandler fun Route.setupChat( chatRepositoryFactory: BackendChatRepositoryFactory, @@ -33,7 +34,10 @@ fun Route.setupChat( val info = call.receive() if (info.role != ChatRoles.USER) throw BadRequestException("invalid role") val response = processUserMessageUseCase.execute(info, userId) - remoteEventHandler.notify(RemoteEvent.OnChatUpdate(userId)) + remoteEventHandler.notify( + RemoteEvent.OnChatUpdate(userId), + RemoteEvent.serializer() + ) call.respond(response) } } diff --git a/apps/vbox/client/shared/src/commonMain/kotlin/ru/shadowsparky/vbox/shared/data/RemoteEventListenerImpl.kt b/apps/vbox/client/shared/src/commonMain/kotlin/ru/shadowsparky/vbox/shared/data/RemoteEventListenerImpl.kt index 84f5f7a..b1afab3 100644 --- a/apps/vbox/client/shared/src/commonMain/kotlin/ru/shadowsparky/vbox/shared/data/RemoteEventListenerImpl.kt +++ b/apps/vbox/client/shared/src/commonMain/kotlin/ru/shadowsparky/vbox/shared/data/RemoteEventListenerImpl.kt @@ -4,114 +4,50 @@ import io.ktor.client.HttpClient import io.ktor.client.plugins.websocket.webSocket import io.ktor.client.request.url import io.ktor.http.HttpMethod -import io.ktor.websocket.Frame -import io.ktor.websocket.readText -import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.SupervisorJob -import kotlinx.coroutines.channels.ReceiveChannel -import kotlinx.coroutines.channels.SendChannel -import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.SharingStarted +import kotlinx.coroutines.flow.emitAll import kotlinx.coroutines.flow.filterNotNull -import kotlinx.coroutines.flow.retryWhen +import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.shareIn import kotlinx.coroutines.flow.transformLatest import kotlinx.serialization.json.Json import org.koin.core.annotation.Single -import ru.shadowsparky.domain.Log +import ru.shadowsparky.http.data.RemoveEventProcessor +import ru.shadowsparky.http.data.retryExponential import ru.shadowsparky.http.domain.TokenStorage -import ru.shadowsparky.vbox.shared.domain.AuthRequest -import ru.shadowsparky.vbox.shared.domain.AuthResponse import ru.shadowsparky.vbox.shared.domain.HealthCheck +import ru.shadowsparky.vbox.shared.domain.ON_EVENT import ru.shadowsparky.vbox.shared.domain.RemoteEvent -import ru.shadowsparky.vbox.shared.domain.RemoteEventHandler import ru.shadowsparky.vbox.shared.domain.RemoteEventListener import ru.shadowsparky.vbox.shared.domain.ServerConfigurationRepository import ru.shadowsparky.vbox.shared.domain.getServerConfiguration -import kotlin.math.pow -import kotlin.time.Duration.Companion.milliseconds @Single class RemoteEventListenerImpl( private val httpClient: HttpClient, private val serverConfigurationRepository: ServerConfigurationRepository, - private val authTokenCache: TokenStorage, + authTokenCache: TokenStorage, private val healthCheck: HealthCheck, - private val json: Json, - private val log: Log + private val processor: RemoveEventProcessor, + private val json: Json ) : RemoteEventListener { private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) @OptIn(ExperimentalCoroutinesApi::class) - override val event = authTokenCache.token.filterNotNull().transformLatest { token -> + override val event: Flow = authTokenCache.token.filterNotNull().transformLatest { token -> val config = serverConfigurationRepository.getServerConfiguration() - log.d(TAG, "listener started") - try { - httpClient.webSocket( - method = HttpMethod.Get, - request = { url(config.asWebSocketStr() + "/${RemoteEventHandler.ON_EVENT}") } - ) { - authV2(token.token, incoming, outgoing) - for (frame in incoming) { - if (frame is Frame.Text) { - val text = frame.readText() - try { - val decoded = json.decodeFromString(text) - log.d(TAG, "receive $text $decoded") - emit(decoded) - } catch (e: Exception) { - log.e(TAG, e, "failed to decode message: $text") - } - } - } - } - } catch (e: Exception) { - log.e(TAG, e, "websocket closed with error") - throw e - } finally { - log.d(TAG, "listener finished") + httpClient.webSocket( + method = HttpMethod.Get, + request = { url(config.asWebSocketStr() + "/${ON_EVENT}") } + ) { + val flow = processor.process(token.token, this) { healthCheck.check() } + .map { json.decodeFromString(it) } + emitAll(flow) } }.retryExponential { true }.shareIn(scope, SharingStarted.WhileSubscribed(500), 0) - - private suspend fun authV2( - token: String, - input: ReceiveChannel, - output: SendChannel - ) { - output.send(Frame.Text(json.encodeToString(AuthRequest(token)))) - val rawFrame = (input.receive() as Frame.Text).readText() - val response = json.decodeFromString(rawFrame) - if (!response.ok) { - healthCheck.check() - error("Authentication failed") - } - } - - private fun Flow.retryExponential( - maxRetries: Int = Int.MAX_VALUE, - initialDelay: Long = 5000L, - maxDelay: Long = 300_000L, - factor: Double = 2.0, - shouldRetry: (Throwable) -> Boolean = { true } - ): Flow = retryWhen { cause, attempt -> - log.d(TAG, "error occurred, attempt=$attempt", cause) - if (!shouldRetry(cause) || attempt >= maxRetries || cause is CancellationException) { - false - } else { - val delayTime = (initialDelay * factor.pow(attempt.toDouble())) - .toLong() - .coerceAtMost(maxDelay) - log.d(TAG, "retry after delay $delayTime", cause) - delay(delayTime.milliseconds) - true - } - } - - private companion object { - const val TAG = "RemoteSearchEventListener" - } } diff --git a/apps/vbox/common/src/commonMain/kotlin/ru/shadowsparky/vbox/shared/domain/RemoteEventHandler.kt b/apps/vbox/common/src/commonMain/kotlin/ru/shadowsparky/vbox/shared/domain/RemoteEventHandler.kt index 2da5e93..d271448 100644 --- a/apps/vbox/common/src/commonMain/kotlin/ru/shadowsparky/vbox/shared/domain/RemoteEventHandler.kt +++ b/apps/vbox/common/src/commonMain/kotlin/ru/shadowsparky/vbox/shared/domain/RemoteEventHandler.kt @@ -4,13 +4,7 @@ import kotlinx.coroutines.flow.Flow import kotlinx.serialization.SerialName import kotlinx.serialization.Serializable -interface RemoteEventHandler { - suspend fun notify(eventInfo: RemoteEvent) - - companion object { - const val ON_EVENT = "$DYNAMIC_PREFIX/onEvent" - } -} +const val ON_EVENT = "$DYNAMIC_PREFIX/onEvent" interface RemoteEventListener { val event: Flow @@ -45,11 +39,3 @@ sealed interface RemoteEvent { @SerialName("OnChatUpdate") data class OnChatUpdate(override val userId: Long) : RemoteEvent } - -@Serializable -@SerialName("AuthRequest") -data class AuthRequest(val token: String) - -@Serializable -@SerialName("AuthResponse") -data class AuthResponse(val ok: Boolean, val error: String? = null) diff --git a/libs/backend-base/build.gradle.kts b/libs/backend-base/build.gradle.kts index 7400a92..e713e01 100644 --- a/libs/backend-base/build.gradle.kts +++ b/libs/backend-base/build.gradle.kts @@ -1,8 +1,16 @@ plugins { alias(libs.plugins.convention.jvm) alias(libs.plugins.convention.koin) + alias(libs.plugins.ktor) + alias(libs.plugins.convention.serialization) } dependencies { + implementation(project(":libs:http-client")) implementation(libs.java.jwt) implementation(libs.logback.classic) + implementation(libs.ktor.server.core.jvm) + implementation(libs.ktor.server.auth) + implementation(libs.ktor.server.websockets.jvm) + implementation(libs.ktor.server.auth.jwt) + implementation(libs.ktor.server.status.pages.jvm) } diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/BackendRemoteEventHandler.kt b/libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/data/event/BackendRemoteEventHandler.kt similarity index 54% rename from apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/BackendRemoteEventHandler.kt rename to libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/data/event/BackendRemoteEventHandler.kt index d1da013..14ef143 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/BackendRemoteEventHandler.kt +++ b/libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/data/event/BackendRemoteEventHandler.kt @@ -1,22 +1,24 @@ -package ru.shadowsparky.vbox.backend.data +package ru.shadowsparky.backend.data.event -import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonElement +import kotlinx.serialization.json.jsonObject +import kotlinx.serialization.json.jsonPrimitive +import kotlinx.serialization.json.longOrNull import org.koin.core.annotation.Single import org.slf4j.Logger import org.slf4j.LoggerFactory -import ru.shadowsparky.vbox.shared.domain.RemoteEvent -import ru.shadowsparky.vbox.shared.domain.RemoteEventHandler +import ru.shadowsparky.backend.domain.RemoteEventHandler val eventLogger: Logger = LoggerFactory.getLogger("events") @Single class BackendRemoteEventHandler( - private val json: Json, private val registry: SessionRegistry ) : RemoteEventHandler { - override suspend fun notify(eventInfo: RemoteEvent) { - val jsonText = json.encodeToString(eventInfo) - val writers = registry.get(eventInfo.userId) + override suspend fun notify(eventInfo: JsonElement) { + val userId = eventInfo.jsonObject["userId"]?.jsonPrimitive?.longOrNull + ?: error("User id field required") + val writers = registry.get(userId) if (writers.isNullOrEmpty()) { eventLogger.info("unable to notify {}. sessions not found, cache {}", eventInfo, registry) return @@ -24,10 +26,10 @@ class BackendRemoteEventHandler( writers.toList().forEach { writer -> eventLogger.info("notify[{}]. session {}", eventInfo, writer) try { - writer.writeText(jsonText) + writer.writeText(eventInfo.toString()) } catch (e: Exception) { eventLogger.error("unable to notify {}. delete session. Reason: {}", writer, e.message) - registry.remove(eventInfo.userId, writer) + registry.remove(userId, writer) } } } diff --git a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/SessionRegistry.kt b/libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/data/event/SessionRegistry.kt similarity index 95% rename from apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/SessionRegistry.kt rename to libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/data/event/SessionRegistry.kt index 88c2fcd..896fae6 100644 --- a/apps/vbox/backend/src/main/kotlin/ru/shadowsparky/vbox/backend/data/SessionRegistry.kt +++ b/libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/data/event/SessionRegistry.kt @@ -1,4 +1,4 @@ -package ru.shadowsparky.vbox.backend.data +package ru.shadowsparky.backend.data.event import org.koin.core.annotation.Single import java.util.concurrent.ConcurrentHashMap diff --git a/libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/domain/Events.kt b/libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/domain/Events.kt new file mode 100644 index 0000000..1c289ee --- /dev/null +++ b/libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/domain/Events.kt @@ -0,0 +1,18 @@ +package ru.shadowsparky.backend.domain + +import kotlinx.serialization.SerializationStrategy +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonElement +import org.koin.mp.KoinPlatform + +interface RemoteEventHandler { + suspend fun notify(eventInfo: JsonElement) +} + +suspend fun RemoteEventHandler.notify( + any: T, + serializer: SerializationStrategy, +) { + val json = KoinPlatform.getKoin().get() + notify(json.encodeToJsonElement(serializer, any)) +} diff --git a/libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/domain/ExceptionInfo.kt b/libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/domain/ExceptionInfo.kt new file mode 100644 index 0000000..07288c9 --- /dev/null +++ b/libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/domain/ExceptionInfo.kt @@ -0,0 +1,6 @@ +package ru.shadowsparky.backend.domain + +import kotlinx.serialization.Serializable + +@Serializable +data class ExceptionInfo(val msg: String) diff --git a/libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/presentation/KtorExtensions.kt b/libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/presentation/KtorExtensions.kt new file mode 100644 index 0000000..bd609cd --- /dev/null +++ b/libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/presentation/KtorExtensions.kt @@ -0,0 +1,90 @@ +package ru.shadowsparky.backend.presentation + +import io.ktor.http.HttpStatusCode +import io.ktor.server.application.Application +import io.ktor.server.application.ApplicationCall +import io.ktor.server.application.install +import io.ktor.server.auth.Authentication +import io.ktor.server.auth.authentication +import io.ktor.server.auth.jwt.JWTPrincipal +import io.ktor.server.auth.jwt.jwt +import io.ktor.server.plugins.statuspages.StatusPages +import io.ktor.server.response.respond +import org.slf4j.Logger +import org.slf4j.LoggerFactory +import ru.shadowsparky.backend.data.TokenVerifier +import ru.shadowsparky.backend.domain.ExceptionInfo +import ru.shadowsparky.backend.domain.INVALID_TOKEN_MSG +import ru.shadowsparky.backend.domain.JwtInfo +import ru.shadowsparky.backend.domain.LOGIN_NAME +import ru.shadowsparky.backend.domain.LoginVerifier +import ru.shadowsparky.backend.domain.USER_ID_ARG +import ru.shadowsparky.backend.domain.VerifyTokenException +import ru.shadowsparky.http.domain.HttpException + +const val AUTH_JWT_NAME = "auth-jwt" + +val routingLogger: Logger = LoggerFactory.getLogger("routing") + +fun ApplicationCall.obtainUserId(): Long { + val principal = authentication.principal() + ?: throw VerifyTokenException(INVALID_TOKEN_MSG) + return principal.payload + .getClaim(USER_ID_ARG) + .asLong() +} + +fun Application.configureJwt( + jwtInfo: JwtInfo, + tokenVerifier: TokenVerifier, + loginVerifier: LoginVerifier +) { + install(Authentication) { + jwt(AUTH_JWT_NAME) { + realm = jwtInfo.realm + verifier(tokenVerifier.verifier) + validate { credential -> + val login = credential.payload.getClaim(LOGIN_NAME).asString() + try { + if (login != null) { + loginVerifier.verify(login) + if (credential.payload.expiresAt == null) { + throw VerifyTokenException("Static tokens not supported!") + } + JWTPrincipal(credential.payload) + } else { + null + } + } catch (e: VerifyTokenException) { + routingLogger.error("verify token failed ${e.message}") + null + } + } + challenge { _, _ -> + call.respond( + HttpStatusCode.Unauthorized, + ExceptionInfo(INVALID_TOKEN_MSG) + ) + } + } + } +} + +fun Application.installStatusPages() { + install(StatusPages) { + exception { call, cause -> + routingLogger.error("http exception occurred. returns ${cause.httpCode}", cause) + call.respond( + status = HttpStatusCode.fromValue(cause.httpCode), + message = ExceptionInfo(cause.message ?: "Неизвестная ошибка") + ) + } + exception { call, cause -> + routingLogger.error("error occurred. returns 500...", cause) + call.respond( + status = HttpStatusCode.InternalServerError, + message = ExceptionInfo(cause.message ?: "Неизвестная ошибка") + ) + } + } +} diff --git a/libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/presentation/WebSocketExtensions.kt b/libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/presentation/WebSocketExtensions.kt new file mode 100644 index 0000000..f70c94d --- /dev/null +++ b/libs/backend-base/src/main/kotlin/ru/shadowsparky/backend/presentation/WebSocketExtensions.kt @@ -0,0 +1,105 @@ +package ru.shadowsparky.backend.presentation + +import com.auth0.jwt.exceptions.TokenExpiredException +import com.auth0.jwt.interfaces.DecodedJWT +import io.ktor.http.HttpStatusCode +import io.ktor.server.application.Application +import io.ktor.server.application.install +import io.ktor.server.routing.routing +import io.ktor.server.websocket.WebSockets +import io.ktor.server.websocket.pingPeriod +import io.ktor.server.websocket.timeout +import io.ktor.server.websocket.webSocket +import io.ktor.websocket.Frame +import io.ktor.websocket.readText +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.channels.ReceiveChannel +import kotlinx.coroutines.channels.SendChannel +import kotlinx.serialization.json.Json +import ru.shadowsparky.backend.data.TokenVerifier +import ru.shadowsparky.backend.data.event.SessionRegistry +import ru.shadowsparky.backend.data.event.eventLogger +import ru.shadowsparky.backend.domain.USER_ID_ARG +import ru.shadowsparky.http.domain.AuthRequest +import ru.shadowsparky.http.domain.AuthResponse +import ru.shadowsparky.http.domain.BadRequestException +import ru.shadowsparky.http.domain.HttpException +import kotlin.time.Duration.Companion.seconds + +fun Application.setupWebSocket( + path: String?, + json: Json, + tokenVerifier: TokenVerifier, + sessionRegistry: SessionRegistry +) { + install(WebSockets) { + pingPeriod = (30).seconds + timeout = (30).seconds + maxFrameSize = Long.MAX_VALUE + masking = false + } + path?.let { + routing { + webSocket(path) { + val userId = incoming.authFlow(json, tokenVerifier, outgoing) + val session = SessionRegistry.Writer { text -> outgoing.trySend(Frame.Text(text)) } + sessionRegistry.put(userId, session) + val deferred = CompletableDeferred() + try { + outgoing.invokeOnClose { deferred.complete(null) } + deferred.await() + } finally { + sessionRegistry.remove(userId, session) + } + } + } + } +} + +private suspend fun ReceiveChannel.receiveTextOrNull(): String? { + val frame = receive() + return if (frame is Frame.Text) frame.readText() else null +} + +private suspend fun ReceiveChannel.authFlow( + json: Json, + tokenVerifier: TokenVerifier, + sendChannel: SendChannel +): Long { + val rawRequest = receiveTextOrNull() ?: throw BadRequestException("Authentication request required") + val request = runCatching { json.decodeFromString(rawRequest) }.getOrNull() + val decodedJwt = if (request == null) { + tokenVerifier.verify(rawRequest) + } else { + authFlowV2(request, json, tokenVerifier, sendChannel) + } + return decodedJwt.getClaim(USER_ID_ARG).asLong() +} + +private const val MAX_ATTEMPTS = 3 + +private suspend fun ReceiveChannel.authFlowV2( + initialRequest: AuthRequest, + json: Json, + tokenVerifier: TokenVerifier, + sendChannel: SendChannel +): DecodedJWT { + var currentRequest = initialRequest + repeat(MAX_ATTEMPTS) { attempt -> + try { + val jwt = tokenVerifier.verify(currentRequest.token) + sendChannel.send(Frame.Text(json.encodeToString(AuthResponse(true)))) + return jwt + } catch (e: TokenExpiredException) { + eventLogger.error("token expired. attempt=$attempt", e) + if (attempt == MAX_ATTEMPTS - 1) return@repeat + sendChannel.send(Frame.Text(json.encodeToString(AuthResponse(false, e.message)))) + val nextRaw = receiveTextOrNull() ?: return@repeat + currentRequest = json.decodeFromString(nextRaw) + } catch (e: Exception) { + eventLogger.error("unable to verify token", e) + throw HttpException(HttpStatusCode.Unauthorized) + } + } + throw HttpException(HttpStatusCode.Unauthorized) +} diff --git a/libs/http-client/src/commonMain/kotlin/ru/shadowsparky/http/data/RemoveEventProcessor.kt b/libs/http-client/src/commonMain/kotlin/ru/shadowsparky/http/data/RemoveEventProcessor.kt new file mode 100644 index 0000000..0004bc6 --- /dev/null +++ b/libs/http-client/src/commonMain/kotlin/ru/shadowsparky/http/data/RemoveEventProcessor.kt @@ -0,0 +1,85 @@ +package ru.shadowsparky.http.data + +import io.ktor.client.plugins.websocket.DefaultClientWebSocketSession +import io.ktor.websocket.Frame +import io.ktor.websocket.readText +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.channels.ReceiveChannel +import kotlinx.coroutines.channels.SendChannel +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.retryWhen +import kotlinx.serialization.json.Json +import org.koin.core.annotation.Single +import ru.shadowsparky.domain.Log +import ru.shadowsparky.http.domain.AuthRequest +import ru.shadowsparky.http.domain.AuthResponse +import kotlin.math.pow +import kotlin.time.Duration.Companion.milliseconds + +@Single +class RemoveEventProcessor( + private val json: Json, + private val log: Log +) { + fun process( + token: String, + session: DefaultClientWebSocketSession, + onTokenInvalid: suspend () -> Unit + ): Flow { + val incoming = session.incoming + val outgoing = session.outgoing + return flow { + authV2(token, incoming, outgoing, onTokenInvalid) + for (frame in incoming) { + if (frame is Frame.Text) { + val text = frame.readText() + try { + log.d(TAG, "receive $text") + emit(text) + } catch (e: Exception) { + log.e(TAG, e, "failed to decode message: $text") + } + } + } + } + } + + private suspend fun authV2( + token: String, + input: ReceiveChannel, + output: SendChannel, + onTokenInvalid: suspend () -> Unit + ) { + output.send(Frame.Text(json.encodeToString(AuthRequest(token)))) + val rawFrame = (input.receive() as Frame.Text).readText() + val response = json.decodeFromString(rawFrame) + if (!response.ok) { + onTokenInvalid() + error("Authentication failed") + } + } + + private companion object { + const val TAG = "RemoveEventProcessor" + } +} + +fun Flow.retryExponential( + maxRetries: Int = Int.MAX_VALUE, + initialDelay: Long = 5000L, + maxDelay: Long = 300_000L, + factor: Double = 2.0, + shouldRetry: (Throwable) -> Boolean = { true } +): Flow = retryWhen { cause, attempt -> + if (!shouldRetry(cause) || attempt >= maxRetries || cause is CancellationException) { + false + } else { + val delayTime = (initialDelay * factor.pow(attempt.toDouble())) + .toLong() + .coerceAtMost(maxDelay) + delay(delayTime.milliseconds) + true + } +} diff --git a/libs/http-client/src/commonMain/kotlin/ru/shadowsparky/http/domain/Events.kt b/libs/http-client/src/commonMain/kotlin/ru/shadowsparky/http/domain/Events.kt new file mode 100644 index 0000000..670be34 --- /dev/null +++ b/libs/http-client/src/commonMain/kotlin/ru/shadowsparky/http/domain/Events.kt @@ -0,0 +1,12 @@ +package ru.shadowsparky.http.domain + +import kotlinx.serialization.SerialName +import kotlinx.serialization.Serializable + +@Serializable +@SerialName("AuthRequest") +data class AuthRequest(val token: String) + +@Serializable +@SerialName("AuthResponse") +data class AuthResponse(val ok: Boolean, val error: String? = null)