From 8f4c6cc630cf358ea781134a09603aef6ee77fbc Mon Sep 17 00:00:00 2001 From: jay-tux Date: Sat, 8 Aug 2026 20:56:32 +0200 Subject: [PATCH] CLI client, SSE --- .../com/jaytux/phoebench/clients/cli/CLI.kt | 3 +- .../phoebench/clients/cli/ProjectHandlers.kt | 1 + .../jaytux/phoebench/clients/AuthProvider.kt | 7 - .../com/jaytux/phoebench/clients/Client.kt | 56 ++++---- .../com/jaytux/phoebench/clients/Util.kt | 10 +- .../jaytux/phoebench/clients/data/HomeVM.kt | 91 +++++++++---- .../phoebench/clients/data/IProjectRepo.kt | 10 -- .../jaytux/phoebench/clients/data/ISSERepo.kt | 31 +++++ .../phoebench/clients/data/ProjectVM.kt | 46 +++++-- .../jaytux/phoebench/clients/ui/HomeView.kt | 1 - .../phoebench/clients/ui/ProjectView.kt | 6 +- .../com/jaytux/phoebench/common/Events.kt | 59 +++++++++ .../com/jaytux/phoebench/common/Routes.kt | 7 +- .../com/jaytux/phoebench/common/SSERoute.kt | 94 ++++++++++++++ .../com/jaytux/phoebench/server/Buses.kt | 29 +++++ .../com/jaytux/phoebench/server/Main.kt | 90 ++----------- .../com/jaytux/phoebench/server/SSEBus.kt | 41 +++--- .../phoebench/server/handlers/AuthHandler.kt | 30 +++++ .../phoebench/server/handlers/Bridge.kt | 120 +++++++++++++++++- .../server/handlers/ProjectHandler.kt | 87 +++++++++---- .../phoebench/server/handlers/ServerScope.kt | 11 ++ .../main/resources/simplelogger.properties | 1 - 22 files changed, 611 insertions(+), 220 deletions(-) create mode 100644 clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/data/ISSERepo.kt create mode 100644 common/src/commonMain/kotlin/com/jaytux/phoebench/common/Events.kt create mode 100644 common/src/commonMain/kotlin/com/jaytux/phoebench/common/SSERoute.kt create mode 100644 server/src/main/kotlin/com/jaytux/phoebench/server/Buses.kt create mode 100644 server/src/main/kotlin/com/jaytux/phoebench/server/handlers/ServerScope.kt diff --git a/clients/cli/src/main/kotlin/com/jaytux/phoebench/clients/cli/CLI.kt b/clients/cli/src/main/kotlin/com/jaytux/phoebench/clients/cli/CLI.kt index 3bc4d19..83fa5ad 100644 --- a/clients/cli/src/main/kotlin/com/jaytux/phoebench/clients/cli/CLI.kt +++ b/clients/cli/src/main/kotlin/com/jaytux/phoebench/clients/cli/CLI.kt @@ -102,7 +102,7 @@ object CLI { data class ID(val id: Uuid) : IProjectIdentification, ILabelIdentification sealed interface IData { - abstract fun toList(): List + fun toList(): List } data class DirectData(val data: List) : IData { override fun toList(): List = data @@ -110,6 +110,7 @@ object CLI { data class FileData(val file: InputStream, val parse: (String) -> T?) : IData { override fun toList(): List { val raw = file.bufferedReader().use { it.readText() }.split(',') + val parsed = ArrayList(raw.size) val errors = mutableListOf() raw.forEach { diff --git a/clients/cli/src/main/kotlin/com/jaytux/phoebench/clients/cli/ProjectHandlers.kt b/clients/cli/src/main/kotlin/com/jaytux/phoebench/clients/cli/ProjectHandlers.kt index 84b42ba..1cb91e5 100644 --- a/clients/cli/src/main/kotlin/com/jaytux/phoebench/clients/cli/ProjectHandlers.kt +++ b/clients/cli/src/main/kotlin/com/jaytux/phoebench/clients/cli/ProjectHandlers.kt @@ -147,6 +147,7 @@ object ProjectHandlers { } val warmupData = warmup.ensure("warmup").toList() val measureData = measurement.ensure("measurements").toList() + println("Loaded ${warmupData.size} warmup elements, ${measureData.size} measurements") val timeUnit = unit.maybePrompt("time unit for data") { when(it) { in setOf("ns", "nano", "nanosec") -> TimeUnit.NANOS diff --git a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/AuthProvider.kt b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/AuthProvider.kt index 13a5c36..a0c9ea9 100644 --- a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/AuthProvider.kt +++ b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/AuthProvider.kt @@ -28,14 +28,11 @@ class AuthProvider private constructor() { init { _refresh.value = _refreshAccessor.load() - println("Refresh token: ${_refresh.value}") _server.value = _serverAccessor.load() - println("Server: ${_server.value}") if(_server.value == null) onLogout() } fun setServer(server: String, protocol: String) { - println("Setting server to $server, protocol version $protocol") _server.value = server _protocolVersion.value = protocol _serverAccessor.save(server) @@ -45,7 +42,6 @@ class AuthProvider private constructor() { fun onLogin(tokens: TokenResponse) { _lock.withLock { if (server.value == null) throw IllegalStateException("Server is null, cannot log in.") - println("Saving new refresh token (${tokens.refresh})") _refresh.value = tokens.refresh _access.value = tokens.access _refreshAccessor.save(tokens.refresh) @@ -55,7 +51,6 @@ class AuthProvider private constructor() { fun onRefresh(tokens: TokenResponse) { _lock.withLock { - println("Saving new refresh token (${tokens.refresh})") _refresh.value = tokens.refresh _access.value = tokens.access _refreshAccessor.save(tokens.refresh) @@ -64,7 +59,6 @@ class AuthProvider private constructor() { fun onLogout() { _lock.withLock { - println("Erasing refresh token ${_refresh.value}") _refresh.value = null _access.value = null _refreshAccessor.erase() @@ -73,7 +67,6 @@ class AuthProvider private constructor() { } fun asBearer(): BearerTokens? = _lock.withLock { - println("Bearer tokens requested (access: ${access.value}, refresh: ${_refresh.value})") BearerTokens( accessToken = _access.value ?: return null, refreshToken = _refresh.value?.toString() ?: return null diff --git a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/Client.kt b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/Client.kt index 20077b4..efabfd7 100644 --- a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/Client.kt +++ b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/Client.kt @@ -2,6 +2,7 @@ package com.jaytux.phoebench.clients import com.jaytux.phoebench.common.* import io.ktor.client.* +import io.ktor.client.plugins.* import io.ktor.client.plugins.auth.* import io.ktor.client.plugins.auth.providers.* import io.ktor.client.plugins.contentnegotiation.* @@ -10,8 +11,6 @@ import io.ktor.serialization.kotlinx.json.* import io.ktor.utils.io.* import kotlinx.atomicfu.locks.ReentrantLock import kotlinx.atomicfu.locks.withLock -import kotlinx.serialization.KSerializer -import kotlinx.serialization.json.Json import kotlin.time.Duration.Companion.seconds class Client private constructor(private val _auth: AuthProvider) { @@ -19,7 +18,6 @@ class Client private constructor(private val _auth: AuthProvider) { install(ContentNegotiation) { json() } } private val _client = platformClient { - var tryingRefresh = false install(ContentNegotiation) { json() } install(SSE) { @@ -28,38 +26,33 @@ class Client private constructor(private val _auth: AuthProvider) { bufferPolicy = SSEBufferPolicy.LastEvents(5) } + install(HttpTimeout) { + requestTimeoutMillis = 10000 + socketTimeoutMillis = 10000 + } + install(Auth) { bearer { cacheTokens = false loadTokens { val res = _auth.asBearer() - println("Client requested bearer tokens and got $res") res } refreshTokens { - if (tryingRefresh) { + val ref = _auth.refresh.value ?: return@refreshTokens null + + val res = callRoute(_authClient, Routes.Auth.refresh, RefreshRequest(ref), true).foldSuspend({ + if (it.msg.startsWith(COROUTINE_CANCELLED)) println("Coro calling refresh was cancelled") _auth.onLogout() null - } else { - tryingRefresh = true - val ref = _auth.refresh.value ?: return@refreshTokens null - println("Trying to re-authenticate using $ref") - - val res = callRoute(_authClient, Routes.Auth.refresh, RefreshRequest(ref), true).foldSuspend({ - if (it.msg.startsWith(COROUTINE_CANCELLED)) println("Coro calling refresh was cancelled") - _auth.onLogout() - null - }) { - _auth.onRefresh(it) - val res = _auth.asBearer() - println("Client requested bearer tokens (from refresh) and got $res") - res - } - tryingRefresh = false + }) { + _auth.onRefresh(it) + val res = _auth.asBearer() res } + res } } } @@ -94,14 +87,19 @@ class Client private constructor(private val _auth: AuthProvider) { ): Either = callRoute(_client, route, body) - suspend fun connectSSE(route: String, serializer: KSerializer, onEvent: suspend (T) -> Unit) { - println("Client tries to set up SSE to $route") - val server = _auth.server.value ?: throw IllegalStateException("No server URL set.") - _client.sse(urlString = "$server$route", showCommentEvents = true, showRetryEvents = true) { - incoming.collect { - println("RECEIVE: $it") - if(it.data != null) onEvent(Json.decodeFromString(serializer, it.data!!)) - } + suspend fun callSSE( + route: SSERoute, + params: TParams, + handler: suspend (Either) -> Unit + ): Either { + try { + println("SSE connection to ${route.path} using server ${_auth.server.value}") + val client = + IClient.Default(_client, _auth.server.value ?: throw IllegalStateException("No server URL set.")) + return route.call(client, params, handler) + } catch (e: CancellationException) { + handler(ErrorResponse("Event stream connected to ${route.path} was cancelled.").error()) + throw e } } diff --git a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/Util.kt b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/Util.kt index c0d4261..6a2e058 100644 --- a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/Util.kt +++ b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/Util.kt @@ -29,7 +29,7 @@ fun MutableState.immutable(): State = this fun T.ignore() {} inline fun ViewModel.withScope(crossinline block: suspend () -> R) = viewModelScope.launch { - withContext(Dispatchers.Unconfined) { block() } + withContext(Dispatchers.Default) { block() } } val formatter = LocalDateTime.Format { @@ -83,3 +83,11 @@ fun Float.fmt(): String { val decInt = (decimals * 1000).roundToInt().toFloat() / 1000f return (integer + decInt).toString() } + +inline fun > List.insort(elem: T, crossinline sortBy: (T) -> X): List { + val insertionPoint = binarySearchBy(sortBy(elem), selector = sortBy) + if(insertionPoint >= 0) return this + val index = -insertionPoint - 1 + + return toMutableList().apply { add(index, elem) } +} \ No newline at end of file diff --git a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/data/HomeVM.kt b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/data/HomeVM.kt index 059eab4..64ef0c9 100644 --- a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/data/HomeVM.kt +++ b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/data/HomeVM.kt @@ -9,11 +9,17 @@ import com.jaytux.phoebench.clients.SnackProvider import com.jaytux.phoebench.clients.SnackProvider.Companion.snackOnError import com.jaytux.phoebench.clients.SnackProvider.Companion.snackOr import com.jaytux.phoebench.clients.immutable +import com.jaytux.phoebench.clients.insort import com.jaytux.phoebench.clients.toClipEntry import com.jaytux.phoebench.clients.withScope +import com.jaytux.phoebench.common.AdminEvent +import com.jaytux.phoebench.common.Either +import com.jaytux.phoebench.common.HomeEvent import com.jaytux.phoebench.common.HomeResponse import com.jaytux.phoebench.common.InviteListResponse +import com.jaytux.phoebench.common.Routes import com.jaytux.phoebench.common.UserListResponse +import com.jaytux.phoebench.common.value import kotlinx.coroutines.Job import kotlinx.serialization.serializer import kotlin.uuid.Uuid @@ -22,7 +28,8 @@ class HomeVM( private val _client: Client = Client.get(), private val _auth: AuthProvider = AuthProvider.get(), private val _snack: SnackProvider = SnackProvider.get(), - private val _repo: IHomeRepo = IHomeRepo.default(_client) + private val _repo: IHomeRepo = IHomeRepo.default(_client), + private val _sseRepo: ISSERepo = ISSERepo.default(_client) ) : ViewModel() { private val _username = mutableStateOf(null) private val _isAdmin = mutableStateOf(false) @@ -33,8 +40,6 @@ class HomeVM( private val _users = mutableStateOf(listOf()) private val _invites = mutableStateOf(listOf()) - private var _tokenWasNull = _auth.refresh.value == null - val username = _username.immutable() val isAdmin = _isAdmin.immutable() val projectLimit = _projectLimit.immutable() @@ -45,8 +50,15 @@ class HomeVM( val users = _users.immutable() private var _listenJob: Job? = null + private var _adminJob: Job? = null init { + _listenJob = withScope { + _sseRepo.connectHome { + it.snackOr(this::homeEventHandler) + }.snackOnError() + } + refresh() } @@ -65,24 +77,7 @@ class HomeVM( _publicProjects.value = listOf() } - fun refreshIfNeeded() { - println("Check for refresh: $_tokenWasNull && ${_auth.refresh.value != null}") - if(_tokenWasNull && _auth.refresh.value != null) refresh() - _tokenWasNull = _auth.refresh.value == null - } - fun refresh() { - _listenJob = withScope { - _client.connectSSE("/rt/project", serializer()) { - if(it.owner.name == _username.value) { - if(_ownProjects.value.none { p -> p.id == it.id }) _ownProjects.value += it - } - if(it.isPublic) { - if(_publicProjects.value.none { p -> p.id == it.id }) _publicProjects.value += it - } - } - } - withScope { resetAdmin() _repo.getHome().snackOr { @@ -90,24 +85,30 @@ class HomeVM( _username.value = it.username _isAdmin.value = it.isAdmin _projectLimit.value = it.projectLimit - _ownProjects.value = it.ownProjects - _publicProjects.value = it.publicProjects + _ownProjects.value = it.ownProjects.sortedBy { p -> p.name } + _publicProjects.value = it.publicProjects.sortedBy { p -> p.name } refreshAdmin() } } } fun refreshAdmin() { + if(_isAdmin.value) { + _adminJob = withScope { + _sseRepo.connectAdmin { + it.snackOr(this::adminEventHandler) + }.snackOnError() + } + } + withScope { if(_isAdmin.value) { _repo.inviteList().snackOr { r -> - _invites.value = r.uuids.map { - it.copy(expires = it.expires) - } + _invites.value = r.uuids } _repo.userList().snackOr { r -> - _users.value = r.users + _users.value = r.users.sortedBy { it.name } } } else { @@ -117,6 +118,42 @@ class HomeVM( } } + private fun homeEventHandler(event: HomeEvent) { + when (event) { + is HomeEvent.Changed -> { + arrayOf(_ownProjects, _publicProjects).forEach { state -> + state.value = state.value.filter { p -> p.id != event.summary.id } + .insort(event.summary, HomeResponse.ProjectSummary::name) + } + } + + is HomeEvent.Created -> { + if (event.summary.owner.name == username.value) { + _ownProjects.value = _ownProjects.value.insort(event.summary, HomeResponse.ProjectSummary::name) + } + if (event.summary.isPublic) { + _publicProjects.value = _publicProjects.value.insort(event.summary, HomeResponse.ProjectSummary::name) + } + } + + is HomeEvent.Deleted -> { + arrayOf(_ownProjects, _publicProjects).forEach { state -> + state.value = state.value.filter { p -> p.id != event.id } + } + } + } + } + + private fun adminEventHandler(event: AdminEvent) { + when(event) { + is AdminEvent.InviteDeleted -> _invites.value = _invites.value.filter { it.code != event.id } + is AdminEvent.NewInvite -> _invites.value += event.invite + is AdminEvent.NewUser -> _users.value = _users.value.insort(event.user, UserListResponse.UserData::name) + is AdminEvent.UserChanged -> _users.value = _users.value.filter { it.id != event.user.id }.insort(event.user, UserListResponse.UserData::name) + is AdminEvent.UserDeleted -> _users.value = _users.value.filter { it.id != event.id } + } + } + fun mkInvite(clipboard: Clipboard, asAdmin: Boolean) { withScope { _repo.newInvite(asAdmin).snackOr { @@ -171,7 +208,7 @@ class HomeVM( fun mkProject(name: String, isPublic: Boolean) { withScope { _repo.newProject(name, isPublic).snackOr { - refresh() + if(_listenJob == null) refresh() } } } diff --git a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/data/IProjectRepo.kt b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/data/IProjectRepo.kt index a71997f..0c8b006 100644 --- a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/data/IProjectRepo.kt +++ b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/data/IProjectRepo.kt @@ -31,8 +31,6 @@ interface IProjectRepo { suspend fun newEntry(label: Uuid, timestamp: Instant, warmups: List, measurements: List, unit: TimeUnit): Either - suspend fun updateEntry(id: Uuid, label: Uuid? = null, timestamp: Instant? = null, warmups: List? = null, - measurements: List? = null, unit: TimeUnit? = null): Either suspend fun deleteEntry(id: Uuid): Either companion object { @@ -60,14 +58,6 @@ interface IProjectRepo { ): Either = _client.callRoute(Routes.Entry.new, EntryRequest(label, timestamp, _projectId, warmups, measurements, unit)) - override suspend fun updateEntry(id: Uuid, label: Uuid?, timestamp: Instant?, - warmups: List?, measurements: List?, - unit: TimeUnit? - ): Either = - _client.callRoute(Routes.Entry.update, - id to PartialEntryRequest(label, timestamp, null, warmups, measurements, unit) - ).ignoreValue() - override suspend fun deleteEntry(id: Uuid): Either = _client.callRoute(Routes.Entry.delete, id).ignoreValue() } diff --git a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/data/ISSERepo.kt b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/data/ISSERepo.kt new file mode 100644 index 0000000..bde2e87 --- /dev/null +++ b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/data/ISSERepo.kt @@ -0,0 +1,31 @@ +package com.jaytux.phoebench.clients.data + +import com.jaytux.phoebench.clients.Client +import com.jaytux.phoebench.common.AdminEvent +import com.jaytux.phoebench.common.Either +import com.jaytux.phoebench.common.ErrorResponse +import com.jaytux.phoebench.common.HomeEvent +import com.jaytux.phoebench.common.ProjectEvent +import com.jaytux.phoebench.common.Routes +import kotlin.uuid.Uuid + +interface ISSERepo { + suspend fun connectHome(onEvent: suspend (Either) -> Unit): Either + suspend fun connectAdmin(onEvent: suspend (Either) -> Unit): Either + suspend fun connectProject(id: Uuid, onEvent: suspend (Either) -> Unit): Either + + companion object { + class Default(private val _client: Client) : ISSERepo { + override suspend fun connectHome(onEvent: suspend (Either) -> Unit): Either = + _client.callSSE(Routes.SSE.home, Unit, onEvent) + + override suspend fun connectAdmin(onEvent: suspend (Either) -> Unit): Either = + _client.callSSE(Routes.SSE.admin, Unit, onEvent) + + override suspend fun connectProject(id: Uuid, onEvent: suspend (Either) -> Unit): Either = + _client.callSSE(Routes.SSE.projectSpecific, id, onEvent) + } + + fun default(client: Client) = Default(client) + } +} \ No newline at end of file diff --git a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/data/ProjectVM.kt b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/data/ProjectVM.kt index 64f0c98..2cf48a4 100644 --- a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/data/ProjectVM.kt +++ b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/data/ProjectVM.kt @@ -6,14 +6,19 @@ import androidx.lifecycle.ViewModel import com.jaytux.phoebench.clients.AuthProvider import com.jaytux.phoebench.clients.Client import com.jaytux.phoebench.clients.SnackProvider +import com.jaytux.phoebench.clients.SnackProvider.Companion.snackOnError import com.jaytux.phoebench.clients.SnackProvider.Companion.snackOr import com.jaytux.phoebench.clients.hexString import com.jaytux.phoebench.clients.immutable +import com.jaytux.phoebench.clients.insort import com.jaytux.phoebench.clients.systemTz import com.jaytux.phoebench.clients.withScope import com.jaytux.phoebench.common.EntryResponse import com.jaytux.phoebench.common.LabelResponse +import com.jaytux.phoebench.common.ProjectEvent import com.jaytux.phoebench.common.TimeUnit +import com.jaytux.phoebench.common.foldSuspend +import kotlinx.coroutines.Job import kotlin.time.Clock import kotlin.time.Instant import kotlin.uuid.Uuid @@ -24,6 +29,7 @@ class ProjectVM( private val _snack: SnackProvider = SnackProvider.get(), private val _client: Client = Client.get(), private val _repo: IProjectRepo = IProjectRepo.default(_client, _id), + private val _sseRepo: ISSERepo = ISSERepo.default(_client), private val _forceBack: () -> Unit ) : ViewModel() { data class Label(val id: Uuid, val name: String, val colorStr: String, val uiColor: Color = parseColor(colorStr)) { @@ -74,7 +80,16 @@ class ProjectVM( val labels = _labels.immutable() val entries = _entries.immutable() + private var _job: Job? = null + init { + _job = withScope { + _sseRepo.connectProject(_id) { + it.snackOr(this::handleProjectEvent) + }.snackOnError() + + } + refresh() } @@ -91,6 +106,22 @@ class ProjectVM( } } + private fun handleProjectEvent(event: ProjectEvent) { + when(event) { + ProjectEvent.Deleted -> back() + is ProjectEvent.EntryDeleted -> _entries.value = _entries.value.filter { it.id != event.id } + is ProjectEvent.LabelChanged -> _labels.value += (event.label.id to Label.fromResponse(event.label)) + is ProjectEvent.LabelDeleted -> _labels.value = _labels.value.filter { it.value.id != event.id } + is ProjectEvent.NewEntry -> _entries.value = _entries.value.insort(Entry.fromResponse(event.entry, _labels.value), Entry::timeStamp) + is ProjectEvent.NewLabel -> _labels.value += (event.label.id to Label.fromResponse(event.label)) + is ProjectEvent.Updated -> { + _name.value = event.changes.name + _owner.value = event.changes.owner.name + _public.value = event.changes.isPublic + } + } + } + fun update(name: String?, isPublic: Boolean?) { withScope { _repo.update(name, isPublic).snackOr { refresh() } @@ -99,10 +130,15 @@ class ProjectVM( fun delete() { withScope { - _repo.delete().snackOr { _forceBack() } + _repo.delete().snackOr { back() } } } + fun back() { + _job?.cancel() + _forceBack() + } + fun mkLabel(name: String, color: Color) { withScope { _repo.newLabel(name, color).snackOr { @@ -136,14 +172,6 @@ class ProjectVM( } } - fun updateEntry(id: Uuid, label: Label?, warmups: List?, measurements: List?, unit: TimeUnit?) { - withScope { - _repo.updateEntry(id, label?.id, Clock.System.now(), warmups, measurements, unit).snackOr { - refresh() - } - } - } - fun deleteEntry(id: Uuid) { withScope { _repo.deleteEntry(id).snackOr { diff --git a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/ui/HomeView.kt b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/ui/HomeView.kt index db759ae..c7fe521 100644 --- a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/ui/HomeView.kt +++ b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/ui/HomeView.kt @@ -275,7 +275,6 @@ fun AuthenticatedRoot() { ) { insets -> Surface(Modifier.padding(insets), color = MaterialTheme.colorScheme.surface) { currentProject?.let { - BackHandler { leaveProject() } ProjectView(it, ::leaveProject) } ?: run { HomeView { currentProject = it } diff --git a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/ui/ProjectView.kt b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/ui/ProjectView.kt index dd65330..d9db40d 100644 --- a/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/ui/ProjectView.kt +++ b/clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/ui/ProjectView.kt @@ -9,7 +9,9 @@ import androidx.compose.foundation.lazy.items import androidx.compose.material3.* import androidx.compose.runtime.* import androidx.compose.ui.Alignment +import androidx.compose.ui.ExperimentalComposeUiApi import androidx.compose.ui.Modifier +import androidx.compose.ui.backhandler.BackHandler import androidx.compose.ui.draw.scale import androidx.compose.ui.graphics.Color import androidx.compose.ui.graphics.SolidColor @@ -54,9 +56,11 @@ import io.github.koalaplot.core.xygraph.rememberGridStyle import kotlin.time.Instant import kotlin.uuid.Uuid +@OptIn(ExperimentalComposeUiApi::class) @Composable fun ProjectView(id: Uuid, forceBack: () -> Unit) { val vm = viewModel(key = id.toString()) { ProjectVM(id, _forceBack = forceBack) } + BackHandler { vm.back() } val name by vm.name val owner by vm.owner @@ -489,7 +493,7 @@ fun ProjectPlotArea(vm: ProjectVM){ } else { LazyColumn(Modifier.padding(start = 5.dp)) { - items(labels.toList()) { (_, lbl) -> + items(labels.toList().sortedBy { it.second.name }) { (_, lbl) -> Box(Modifier.fillMaxWidth().clickable { labelFilter.toggle(lbl) }) { Box { QuickLabel(lbl) diff --git a/common/src/commonMain/kotlin/com/jaytux/phoebench/common/Events.kt b/common/src/commonMain/kotlin/com/jaytux/phoebench/common/Events.kt new file mode 100644 index 0000000..fa65b98 --- /dev/null +++ b/common/src/commonMain/kotlin/com/jaytux/phoebench/common/Events.kt @@ -0,0 +1,59 @@ +package com.jaytux.phoebench.common + +import kotlinx.serialization.Serializable +import kotlin.uuid.Uuid + +@Serializable +sealed class HomeEvent { + @Serializable + data class Created(val summary: HomeResponse.ProjectSummary) : HomeEvent() + + @Serializable + data class Changed(val summary: HomeResponse.ProjectSummary) : HomeEvent() + + @Serializable + data class Deleted(val id: Uuid) : HomeEvent() + +} + +@Serializable +sealed class AdminEvent { + @Serializable + data class NewUser(val user: UserListResponse.UserData) : AdminEvent() + + @Serializable + data class UserChanged(val user: UserListResponse.UserData) : AdminEvent() + + @Serializable + data class UserDeleted(val id: Uuid) : AdminEvent() + + @Serializable + data class NewInvite(val invite: InviteListResponse.Invite) : AdminEvent() + + @Serializable + data class InviteDeleted(val id: Uuid) : AdminEvent() +} + +@Serializable +sealed class ProjectEvent { + @Serializable + object Deleted : ProjectEvent() + + @Serializable + data class Updated(val changes: HomeResponse.ProjectSummary) : ProjectEvent() + + @Serializable + data class NewLabel(val label: LabelResponse) : ProjectEvent() + + @Serializable + data class LabelChanged(val label: LabelResponse) : ProjectEvent() + + @Serializable + data class LabelDeleted(val id: Uuid) : ProjectEvent() + + @Serializable + data class NewEntry(val entry: EntryResponse) : ProjectEvent() + + @Serializable + data class EntryDeleted(val id: Uuid) : ProjectEvent() +} \ No newline at end of file diff --git a/common/src/commonMain/kotlin/com/jaytux/phoebench/common/Routes.kt b/common/src/commonMain/kotlin/com/jaytux/phoebench/common/Routes.kt index af5d04e..41ebaee 100644 --- a/common/src/commonMain/kotlin/com/jaytux/phoebench/common/Routes.kt +++ b/common/src/commonMain/kotlin/com/jaytux/phoebench/common/Routes.kt @@ -38,7 +38,12 @@ object Routes { object Entry { val new = ApiRoute.post("/entry", Elevation.AUTH) - val update = ApiRoute.patchUuidNoRes("/entry", Elevation.AUTH) val delete = ApiRoute.deleteUuidNoRes("/entry", Elevation.AUTH) } + + object SSE { + val home = SSERoute.noArgs("/rt/home", Elevation.AUTH) + val admin = SSERoute.noArgs("/rt/admin", Elevation.ADMIN) + val projectSpecific = SSERoute.uuid("/rt/project", Elevation.AUTH) + } } \ No newline at end of file diff --git a/common/src/commonMain/kotlin/com/jaytux/phoebench/common/SSERoute.kt b/common/src/commonMain/kotlin/com/jaytux/phoebench/common/SSERoute.kt new file mode 100644 index 0000000..69e7e9b --- /dev/null +++ b/common/src/commonMain/kotlin/com/jaytux/phoebench/common/SSERoute.kt @@ -0,0 +1,94 @@ +package com.jaytux.phoebench.common + +import io.ktor.client.call.body +import io.ktor.client.plugins.ResponseException +import io.ktor.client.plugins.sse.SSEClientException +import io.ktor.client.plugins.sse.sse +import io.ktor.client.plugins.sse.sseSession +import io.ktor.http.Parameters +import io.ktor.http.buildUrl +import io.ktor.util.reflect.TypeInfo +import io.ktor.util.reflect.typeInfo +import io.ktor.utils.io.CancellationException +import kotlinx.coroutines.NonCancellable +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.withContext +import kotlinx.serialization.decodeFromString +import kotlinx.serialization.json.Json +import kotlinx.serialization.serializer +import kotlin.uuid.Uuid + +sealed class SSERoute(val path: String, val elevation: Elevation, private val _resType: TypeInfo) { + open val pattern = path + private val deserializer = Json.serializersModule.serializer(_resType.kotlinType!!) + + protected open fun buildUrl(params: TParams): String = path + + abstract fun extractParams(reqParams: Parameters): TParams? + + suspend fun call(client: IClient, params: TParams, handler: suspend (Either) -> Unit): Either { + val fullUrl = "${client.serverUrl}${buildUrl(params)}" + try { + client.client.sse(urlString = fullUrl, showCommentEvents = true, showRetryEvents = true) { + incoming.collect { + try { + val data = it.data + @Suppress("UNCHECKED_CAST") + if (data != null) { + println("SSE [$path] with data $data") + handler((Json.decodeFromString(deserializer, data) as TEvent).value()) + } + } catch (e: CancellationException) { + withContext(NonCancellable) { + handler(ErrorResponse("The stream to $fullUrl was disconnected.").error()) + } + throw e + } catch (e: Exception) { + handler(ErrorResponse(e.message ?: "Unknown error while streaming $fullUrl").error()) + } + } + } + return Unit.value() + } + catch(e: ResponseException) { + val error = e.response.body() + return error.error() + } + catch(e: SSEClientException) { + return ErrorResponse(e.message ?: "Could not set up event stream for $fullUrl.").error() + } + catch(e: CancellationException) { + throw e + } + catch(e: Exception) { + return ErrorResponse(e.message ?: "SSE connection failed.").error() + } + } + + class SSERoute0(path: String, elevation: Elevation, resType: TypeInfo) : SSERoute(path, elevation, resType) { + override fun extractParams(reqParams: Parameters) {} + } + + class SSERoute1(path: String, elevation: Elevation, resType: TypeInfo, val urlEncode: (T1) -> String, val urlDecode: (String?) -> T1?) + : SSERoute(path, elevation, resType) + { + override val pattern: String = "$path/{param}" + + override fun buildUrl(params: T1): String = "$path/${urlEncode(params)}" + + override fun extractParams(reqParams: Parameters): T1? = urlDecode(reqParams["param"]) + } + + companion object { + inline fun noArgs(path: String, elevation: Elevation) = + SSERoute0(path, elevation, typeInfo()) + + inline fun single(path: String, elevation: Elevation, + noinline urlEncode: (T) -> String = { it.toString() }, noinline urlDecode: (String?) -> T? + ) = SSERoute1(path, elevation, typeInfo(), urlEncode, urlDecode) + + inline fun uuid(path: String, elevation: Elevation) = single(path, elevation) { + it?.let { p -> Uuid.parseOrNull(p) } + } + } +} \ No newline at end of file diff --git a/server/src/main/kotlin/com/jaytux/phoebench/server/Buses.kt b/server/src/main/kotlin/com/jaytux/phoebench/server/Buses.kt new file mode 100644 index 0000000..054b776 --- /dev/null +++ b/server/src/main/kotlin/com/jaytux/phoebench/server/Buses.kt @@ -0,0 +1,29 @@ +package com.jaytux.phoebench.server + +import com.jaytux.phoebench.common.AdminEvent +import com.jaytux.phoebench.common.HomeEvent +import com.jaytux.phoebench.common.ProjectEvent +import io.ktor.util.reflect.typeInfo +import kotlinx.serialization.serializer +import java.util.concurrent.ConcurrentHashMap +import kotlin.reflect.typeOf +import kotlin.uuid.Uuid + +object Buses { + private val _projectBuses = ConcurrentHashMap>() + + val homeBus = SSEBus(typeOf(), serializer()) + val adminBus = SSEBus(typeOf(), serializer()) + + fun projectBus(id: Uuid) = _projectBuses.computeIfAbsent(id) { + SSEBus(typeOf(), serializer()) + } + + fun allBuses(): List> { + val res = ArrayList>(_projectBuses.size + 2) + res.addAll(_projectBuses.values) + res.add(homeBus) + res.add(adminBus) + return res + } +} \ No newline at end of file diff --git a/server/src/main/kotlin/com/jaytux/phoebench/server/Main.kt b/server/src/main/kotlin/com/jaytux/phoebench/server/Main.kt index 962a36d..b901dc6 100644 --- a/server/src/main/kotlin/com/jaytux/phoebench/server/Main.kt +++ b/server/src/main/kotlin/com/jaytux/phoebench/server/Main.kt @@ -1,24 +1,10 @@ package com.jaytux.phoebench.server -import com.jaytux.phoebench.common.ErrorResponse -import com.jaytux.phoebench.common.HomeResponse -import com.jaytux.phoebench.common.Routes -import com.jaytux.phoebench.common.fold +import com.jaytux.phoebench.common.* import com.jaytux.phoebench.server.db.DB import com.jaytux.phoebench.server.db.User -import com.jaytux.phoebench.server.handlers.AuthHandler -import com.jaytux.phoebench.server.handlers.ProjectHandler -import com.jaytux.phoebench.server.handlers.RouteError -import com.jaytux.phoebench.server.handlers.deleteAdmin -import com.jaytux.phoebench.server.handlers.deleteAuth -import com.jaytux.phoebench.server.handlers.get -import com.jaytux.phoebench.server.handlers.getAdmin -import com.jaytux.phoebench.server.handlers.getAuth -import com.jaytux.phoebench.server.handlers.patchAdmin -import com.jaytux.phoebench.server.handlers.patchAuth -import com.jaytux.phoebench.server.handlers.post -import com.jaytux.phoebench.server.handlers.postAdmin -import com.jaytux.phoebench.server.handlers.postAuth +import com.jaytux.phoebench.server.handlers.* +import com.jaytux.phoebench.server.handlers.ProjectHandler.accessibleProject import io.ktor.http.* import io.ktor.serialization.kotlinx.json.* import io.ktor.server.application.* @@ -28,23 +14,15 @@ import io.ktor.server.netty.* import io.ktor.server.plugins.autohead.* import io.ktor.server.plugins.calllogging.* import io.ktor.server.plugins.contentnegotiation.* -import io.ktor.server.plugins.cors.routing.CORS -import io.ktor.server.plugins.statuspages.StatusPages +import io.ktor.server.plugins.cors.routing.* +import io.ktor.server.plugins.statuspages.* import io.ktor.server.request.* import io.ktor.server.response.* import io.ktor.server.routing.* -import io.ktor.server.sse.SSE -import io.ktor.server.sse.sse -import kotlinx.serialization.encodeToString +import io.ktor.server.sse.* import kotlinx.serialization.json.Json -import kotlinx.serialization.serializer import org.jetbrains.exposed.v1.jdbc.transactions.transaction -import java.net.URI import kotlin.uuid.Uuid -import com.jaytux.phoebench.server.handlers.RouteError.Companion.respondJson -import io.ktor.server.sse.heartbeat -import io.ktor.sse.ServerSentEvent -import io.ktor.utils.io.CancellationException fun main(args: Array) { DB.db @@ -156,59 +134,15 @@ fun Application.module() { deleteAuth(Routes.Label.delete, ProjectHandler::deleteLabel) postAuth(Routes.Entry.new, ProjectHandler::createEntry) - patchAuth(Routes.Entry.update, ProjectHandler::updateEntry) deleteAuth(Routes.Entry.delete, ProjectHandler::deleteEntry) - val projectBus = SSEBus.register("/rt/project") - sse("/rt/project") { - try { - println("Attempt to set up SSE") - val principal = call.principal() - val userId = principal?.payload?.getClaim(com.jaytux.phoebench.common.Auth.JWT_CLAIM)?.asString() - ?: throw RouteError("Missing user claim", HttpStatusCode.Unauthorized) - println(" -- SSE: userId = $userId") - val user = transaction { - User.findById(Uuid.parse(userId)) ?: throw RouteError( - "User not found", - HttpStatusCode.Unauthorized - ) - } - println(" -- SSE: user = ${user.username}") - - heartbeat {} - - val flow = projectBus.register(user.id.value) - - try { - flow.collect { - it.fold({ - throw Exception() // force end of collecting - }) { event -> - send(ServerSentEvent(data = Json.encodeToString(projectBus.serializer, event))) - } - } - } - catch(e: CancellationException) { - projectBus.disconnect(user.id.value) - throw e - } + sseAuth(Routes.SSE.home, { _, _ -> }) { _, _ -> Buses.homeBus } + sseAdmin(Routes.SSE.admin) { _, _ -> Buses.adminBus } + sseAuth(Routes.SSE.projectSpecific, + { user, uuid -> + transaction { accessibleProject(user, uuid, false) } } - catch(e: RouteError) { - call.respondText( - status = e.status, - text = Json.encodeToString(ErrorResponse(e.message ?: "Unknown error")), - contentType = ContentType.Application.Json - ) - } - - println("--- SSE Session ended ---") - } - - sse("/rt/users") {} - - sse("/rt/invites") {} - - sse("/rt/project/{project-id}") {} + ) { _, id -> Buses.projectBus(id) } } get("{...}") { diff --git a/server/src/main/kotlin/com/jaytux/phoebench/server/SSEBus.kt b/server/src/main/kotlin/com/jaytux/phoebench/server/SSEBus.kt index 0cb0e01..d0819ec 100644 --- a/server/src/main/kotlin/com/jaytux/phoebench/server/SSEBus.kt +++ b/server/src/main/kotlin/com/jaytux/phoebench/server/SSEBus.kt @@ -4,6 +4,7 @@ import com.jaytux.phoebench.common.Either import com.jaytux.phoebench.common.error import com.jaytux.phoebench.common.value import io.ktor.util.reflect.instanceOf +import io.ktor.websocket.Serializer import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.SharedFlow import kotlinx.serialization.KSerializer @@ -13,14 +14,17 @@ import kotlin.reflect.KType import kotlin.reflect.typeOf import kotlin.uuid.Uuid -class SSEBus private constructor (private val _containedType: KType, val serializer: KSerializer) { +class SSEBus(private val _containedType: KType, val serializer: KSerializer) { object Cancellation + private val _unAuthFlow = MutableSharedFlow() private val _flows = ConcurrentHashMap>>>() fun register(user: Uuid): SharedFlow> = _flows.computeIfAbsent(user) { 1 to MutableSharedFlow(extraBufferCapacity = 64) }.second + fun unRegistered(): SharedFlow = _unAuthFlow + suspend fun send(user: Uuid, data: T) { _flows[user]?.second?.emit(data.value()) } @@ -29,6 +33,10 @@ class SSEBus private constructor (private val _containedType: KType, val seri _flows.forEach { it.value.second.emit(data.value()) } } + suspend fun sendUnAuth(data: T) { + _unAuthFlow.emit(data) + } + fun disconnect(user: Uuid) { _flows.compute(user) { _, data -> if(data == null) null @@ -46,30 +54,11 @@ class SSEBus private constructor (private val _containedType: KType, val seri _flows.remove(user) } - companion object { - private val _busCache = mutableMapOf>() - - fun register(topic: String, contained: KType, serializer: KSerializer): SSEBus { - val bus = SSEBus(contained, serializer) - _busCache.compute(topic) { k, existing -> - if(existing != null) throw IllegalArgumentException("Bus for $topic exists already") - bus - } - return bus - } - - inline fun register(topic: String) = - register(topic, typeOf(), serializer()) - - fun getBus(topic: String, contained: KType): SSEBus? { - val bus = _busCache[topic] ?: return null - if(contained != bus._containedType) throw IllegalArgumentException("Type mismatch for bus for $topic") - - @Suppress("UNCHECKED_CAST") - return bus as SSEBus - } - - inline fun getBus(topic: String) = - getBus(topic, typeOf()) + suspend fun forceDisconnectExcept(user: Uuid) { + synchronized(_flows) { + val map = _flows.values.filter { it != _flows[user] } + _flows.keys.retainAll(setOf(user)) + map + }.forEach { it.second.emit(Cancellation.error()) } } } \ No newline at end of file diff --git a/server/src/main/kotlin/com/jaytux/phoebench/server/handlers/AuthHandler.kt b/server/src/main/kotlin/com/jaytux/phoebench/server/handlers/AuthHandler.kt index 48b5343..9d67d76 100644 --- a/server/src/main/kotlin/com/jaytux/phoebench/server/handlers/AuthHandler.kt +++ b/server/src/main/kotlin/com/jaytux/phoebench/server/handlers/AuthHandler.kt @@ -2,6 +2,8 @@ package com.jaytux.phoebench.server.handlers import com.jaytux.phoebench.common.* import com.jaytux.phoebench.server.Auth +import com.jaytux.phoebench.server.Buses +import com.jaytux.phoebench.server.SSEBus import com.jaytux.phoebench.server.db.Invite import com.jaytux.phoebench.server.db.RefreshToken import com.jaytux.phoebench.server.db.RefreshTokens @@ -14,6 +16,7 @@ import com.jaytux.phoebench.server.nowPlusMinutes import com.jaytux.phoebench.server.systemTZ import io.ktor.http.HttpStatusCode import io.ktor.util.logging.KtorSimpleLogger +import kotlinx.coroutines.launch import kotlinx.datetime.DateTimeUnit import org.jetbrains.exposed.v1.core.Transaction import org.jetbrains.exposed.v1.core.eq @@ -71,6 +74,13 @@ object AuthHandler { _hasOwner = true } logger.info("New user: ${user.username} (${user.id.value}; is admin? ${user.isAdmin})") + ServerScope.launch { + Buses.adminBus.sendAll( + AdminEvent.NewUser( + UserListResponse.UserData(user.id.value, user.username, user.isAdmin, user.projectLimit, 0) + ) + ) + } val access = Auth.generate(user.id.value) val refresh = newRefreshToken(user) @@ -150,6 +160,9 @@ object AuthHandler { inviteAsAdmin = if(user.isOwner) req.asAdmin else false expires = nowPlus(48, DateTimeUnit.HOUR) } + ServerScope.launch { + Buses.adminBus.sendAll(AdminEvent.NewInvite(InviteListResponse.Invite(invite.id.value, invite.expires, invite.inviteAsAdmin))) + } success(UuidResponse(invite.id.value)) } @@ -162,6 +175,9 @@ object AuthHandler { suspend fun deleteInvite(user: User, req: Uuid) = transaction { val inv = Invite.findById(req) ?: throw RouteError("Invalid invite code", HttpStatusCode.NotFound) inv.delete() + ServerScope.launch { + Buses.adminBus.sendAll(AdminEvent.InviteDeleted(req)) + } success(EmptyResponse()) } @@ -184,6 +200,16 @@ object AuthHandler { if(user.isOwner) target.isAdmin = it else throw RouteError("Only the server owner can change admin status.", HttpStatusCode.Forbidden) } + + ServerScope.launch { + Buses.adminBus.sendAll( + AdminEvent.UserChanged( + UserListResponse.UserData(user.id.value, user.username, user.isAdmin, user.projectLimit, 0) + ) + ) + if(changes.isAdmin == false) Buses.adminBus.forceDisconnect(req.first) + } + success(EmptyResponse()) } @@ -195,6 +221,10 @@ object AuthHandler { target.delete() } else throw RouteError("Only the owner can delete admin accounts.", HttpStatusCode.Forbidden) + ServerScope.launch { + Buses.adminBus.sendAll(AdminEvent.UserDeleted(req)) + Buses.allBuses().forEach { it.forceDisconnect(req) } + } success(EmptyResponse()) } diff --git a/server/src/main/kotlin/com/jaytux/phoebench/server/handlers/Bridge.kt b/server/src/main/kotlin/com/jaytux/phoebench/server/handlers/Bridge.kt index 36bcc4f..668b67e 100644 --- a/server/src/main/kotlin/com/jaytux/phoebench/server/handlers/Bridge.kt +++ b/server/src/main/kotlin/com/jaytux/phoebench/server/handlers/Bridge.kt @@ -4,20 +4,41 @@ import com.jaytux.phoebench.server.handlers.RouteError.Companion.wrapped import com.jaytux.phoebench.server.handlers.RouteError.Companion.wrappedAdmin import com.jaytux.phoebench.server.handlers.RouteError.Companion.wrappedAuth import com.jaytux.phoebench.common.ApiRoute +import com.jaytux.phoebench.common.Either import com.jaytux.phoebench.common.Elevation import com.jaytux.phoebench.common.EmptyRequest +import com.jaytux.phoebench.common.ErrorResponse +import com.jaytux.phoebench.common.SSERoute +import com.jaytux.phoebench.common.foldSuspend +import com.jaytux.phoebench.server.Auth.setup +import com.jaytux.phoebench.server.SSEBus import com.jaytux.phoebench.server.db.User +import com.jaytux.phoebench.server.handlers.RouteError +import io.ktor.http.ContentType import io.ktor.http.HttpStatusCode import io.ktor.server.application.ApplicationCall +import io.ktor.server.auth.jwt.JWTPrincipal +import io.ktor.server.auth.principal +import io.ktor.server.plugins.BadRequestException import io.ktor.server.plugins.ContentTransformationException import io.ktor.server.request.receive +import io.ktor.server.response.respondText import io.ktor.server.routing.Route import io.ktor.server.routing.RoutingContext import io.ktor.server.routing.get import io.ktor.server.routing.post import io.ktor.server.routing.delete import io.ktor.server.routing.patch +import io.ktor.server.sse.heartbeat +import io.ktor.server.sse.sse +import io.ktor.sse.ServerSentEvent import io.ktor.util.reflect.typeInfo +import io.ktor.utils.io.CancellationException +import kotlinx.coroutines.flow.SharedFlow +import kotlinx.serialization.json.Json +import org.jetbrains.exposed.v1.jdbc.transactions.transaction +import kotlin.time.Duration.Companion.seconds +import kotlin.uuid.Uuid suspend inline fun ApiRoute.paramArgs(call: ApplicationCall): TReq = parseParams(call.parameters) ?: throw RouteError( @@ -140,4 +161,101 @@ inline fun Route.patchAuth(api: ApiRoute< wrapperAuth(api, Route::patch, handler) inline fun Route.patchAdmin(api: ApiRoute, noinline handler: suspend (User, TReq) -> Pair): Route = - wrapperAdmin(api, Route::patch, handler) \ No newline at end of file + wrapperAdmin(api, Route::patch, handler) + +inline fun Route.wrapSSE( + api: SSERoute, noinline extra: suspend (ApplicationCall, TParams) -> TInter, + noinline prepare: suspend (TInter, TParams) -> SSEBus, + noinline extract: suspend (SSEBus, TInter, TParams) -> SharedFlow, + noinline handler: suspend (TFlow, sender: suspend (TEvent) -> Unit) -> Unit, + noinline onCancel: suspend (SSEBus, TInter, TParams, CancellationException) -> Unit +) { + sse(api.pattern) { + try { + heartbeat { + period = 1.seconds + } + + val params = api.extractParams(call.parameters) ?: throw RouteError( + "Missing or malformed parameters for SSE ${api.pattern}", + HttpStatusCode.BadRequest + ) + + val inter = extra(call, params) + val bus = prepare(inter, params) + val stream = extract(bus, inter, params) + try { + stream.collect { + handler(it) { ev -> send(ServerSentEvent(data = Json.encodeToString(bus.serializer, ev))) } + } + } + catch(e: CancellationException) { + onCancel(bus, inter, params, e) + throw e + } + } + catch(e: BadRequestException) { + call.respondText( + status = HttpStatusCode.BadRequest, + text = Json.encodeToString(ErrorResponse(e.message ?: "Unknown error")), + contentType = ContentType.Application.Json + ) + } + catch(e: RouteError) { + call.respondText( + status = e.status, + text = Json.encodeToString(ErrorResponse(e.message ?: "Unknown error")), + contentType = ContentType.Application.Json + ) + } + } +} + +inline fun Route.wrapAuthSSE( + api: SSERoute, + noinline verifyUser: suspend (User, TParams) -> Unit, + noinline prepare: suspend (User, TParams) -> SSEBus +) = wrapSSE(api, + extra = { call, params -> + val principal = call.principal() + val userId = principal?.payload?.getClaim(com.jaytux.phoebench.common.Auth.JWT_CLAIM)?.asString() + ?: throw RouteError("Missing user claim", HttpStatusCode.Unauthorized) + val user = transaction { + User.findById(Uuid.parse(userId)) ?: throw RouteError( + "User not found", + HttpStatusCode.Unauthorized + ) + } + verifyUser(user, params) + user + }, + prepare = prepare, + extract = { bus, user, _ -> bus.register(user.id.value) }, + handler = { event, sender -> + event.foldSuspend({ + throw RouteError("This event stream has been discontinued.", HttpStatusCode.Locked) + }) { sender(it) } + }, + onCancel = { bus, user, _, _ -> bus.disconnect(user.id.value) } + ) + +inline fun Route.sse(api: SSERoute, noinline setup: suspend (TParams) -> SSEBus) = + wrapSSE(api, + extra = { _, _ -> }, + prepare = { _, params -> setup(params) }, + extract = { bus, _, _ -> bus.unRegistered() }, + handler = { it, sender -> sender(it) }, + onCancel = { _, _, _, _ -> } + ) + +inline fun Route.sseAuth(api: SSERoute, + noinline verifyUser: suspend (User, TParams) -> Unit, noinline setup: suspend (User, TParams) -> SSEBus +) = wrapAuthSSE(api, verifyUser, setup) + +inline fun Route.sseAdmin(api: SSERoute, + noinline setup: suspend (User, TParams) -> SSEBus +) = wrapAuthSSE(api, { user, _ -> + if(!user.isAdmin) { + throw RouteError("Admin access required", HttpStatusCode.Forbidden) + } +}, setup) \ No newline at end of file diff --git a/server/src/main/kotlin/com/jaytux/phoebench/server/handlers/ProjectHandler.kt b/server/src/main/kotlin/com/jaytux/phoebench/server/handlers/ProjectHandler.kt index 34d024d..45eea7a 100644 --- a/server/src/main/kotlin/com/jaytux/phoebench/server/handlers/ProjectHandler.kt +++ b/server/src/main/kotlin/com/jaytux/phoebench/server/handlers/ProjectHandler.kt @@ -4,6 +4,7 @@ import com.jaytux.phoebench.common.EmptyRequest import com.jaytux.phoebench.common.EmptyResponse import com.jaytux.phoebench.common.EntryRequest import com.jaytux.phoebench.common.EntryResponse +import com.jaytux.phoebench.common.HomeEvent import com.jaytux.phoebench.common.HomeResponse import com.jaytux.phoebench.common.LabelRequest import com.jaytux.phoebench.common.LabelResponse @@ -11,8 +12,10 @@ import com.jaytux.phoebench.common.NamedID import com.jaytux.phoebench.common.PartialEntryRequest import com.jaytux.phoebench.common.PartialLabelRequest import com.jaytux.phoebench.common.PartialProjectRequest +import com.jaytux.phoebench.common.ProjectEvent import com.jaytux.phoebench.common.ProjectRequest import com.jaytux.phoebench.common.ProjectResponse +import com.jaytux.phoebench.server.Buses import com.jaytux.phoebench.server.SSEBus import com.jaytux.phoebench.server.db.Entries import com.jaytux.phoebench.server.db.Entry @@ -23,6 +26,8 @@ import com.jaytux.phoebench.server.db.Projects import com.jaytux.phoebench.server.db.User import com.jaytux.phoebench.server.handlers.RouteError.Companion.success import io.ktor.http.HttpStatusCode +import kotlinx.coroutines.GlobalScope +import kotlinx.coroutines.launch import org.jetbrains.exposed.v1.core.SortOrder import org.jetbrains.exposed.v1.core.Transaction import org.jetbrains.exposed.v1.core.eq @@ -31,10 +36,10 @@ import kotlin.uuid.Uuid object ProjectHandler { context(trns: Transaction) - private fun Project.isEditableBy(user: User): Boolean = ownerId.value == user.id.value + fun Project.isEditableBy(user: User): Boolean = ownerId.value == user.id.value context(trns: Transaction) - private fun Project.isAccessible(user: User, forEditing: Boolean): Project { + fun Project.isAccessible(user: User, forEditing: Boolean): Project { return when { isEditableBy(user) -> this isPublic && !forEditing -> this @@ -42,7 +47,7 @@ object ProjectHandler { } } - private fun Transaction.accessibleProject(user: User, id: Uuid, forEditing: Boolean): Project { + fun Transaction.accessibleProject(user: User, id: Uuid, forEditing: Boolean): Project { val project = Project.findById(id) ?: throw RouteError("Invalid project ID.", HttpStatusCode.NotFound) return project.isAccessible(user, forEditing) } @@ -77,10 +82,12 @@ object ProjectHandler { success(proj.toResponse(user)) }.also { (_, proj) -> - val bus = SSEBus.getBus("/rt/project") ?: return@also - val summary = HomeResponse.ProjectSummary(proj.id, proj.name, proj.isPublic, NamedID(user.username, user.id.value)) - if(proj.isPublic) bus.sendAll(summary) - else bus.send(user.id.value, summary) + ServerScope.launch { + val bus = Buses.homeBus + val summary = HomeResponse.ProjectSummary(proj.id, proj.name, proj.isPublic, NamedID(user.username, user.id.value)) + if(proj.isPublic) bus.sendAll(HomeEvent.Created(summary)) + else bus.send(user.id.value, HomeEvent.Created(summary)) + } } fun getProject(user: User, req: Uuid) = transaction { @@ -93,11 +100,30 @@ object ProjectHandler { val changes = req.second changes.name?.let { proj.name = it } changes.isPublic?.let { proj.isPublic = it } + + ServerScope.launch { + val bus = Buses.homeBus + val summary = HomeResponse.ProjectSummary(proj.id.value, proj.name, proj.isPublic, NamedID(user.username, user.id.value)) + if(proj.isPublic) bus.sendAll(HomeEvent.Changed(summary)) + else bus.send(user.id.value, HomeEvent.Changed(summary)) + + val projectBus = Buses.projectBus(req.first) + projectBus.sendAll(ProjectEvent.Updated(summary)) + if(changes.isPublic == false) projectBus.forceDisconnectExcept(proj.owner.id.value) + } + success(EmptyResponse()) } fun deleteProject(user: User, req: Uuid) = transaction { - accessibleProject(user, req, true).delete() + val proj = accessibleProject(user, req, true) + proj.delete() + ServerScope.launch { + val bus = Buses.homeBus + if(proj.isPublic) bus.sendAll(HomeEvent.Deleted(req)) + else bus.send(user.id.value, HomeEvent.Deleted(req)) + Buses.projectBus(req).sendAll(ProjectEvent.Deleted) + } success(EmptyResponse()) } @@ -109,7 +135,13 @@ object ProjectHandler { color = req.color project = proj } - success(LabelResponse(lbl.id.value, lbl.label, lbl.color)) + + val res = LabelResponse(lbl.id.value, lbl.label, lbl.color) + ServerScope.launch { + Buses.projectBus(req.projectId).sendAll(ProjectEvent.NewLabel(res)) + } + + success(res) } fun updateLabel(user: User, req: Pair) = transaction { @@ -121,6 +153,13 @@ object ProjectHandler { if(it.length != 7) throw RouteError("Expected Hex-RGB color (7 characters).", HttpStatusCode.BadRequest) lbl.color = it } + + ServerScope.launch { + Buses.projectBus(lbl.projectId.value).sendAll(ProjectEvent.LabelChanged( + LabelResponse(lbl.id.value, lbl.label, lbl.color) + )) + } + success(EmptyResponse()) } @@ -128,6 +167,11 @@ object ProjectHandler { val lbl = Label.findById(req) ?: throw RouteError("Invalid label ID.", HttpStatusCode.NotFound) lbl.project.isAccessible(user, true) lbl.delete() + + ServerScope.launch { + Buses.projectBus(lbl.id.value).sendAll(ProjectEvent.LabelDeleted(req)) + } + success(EmptyResponse()) } @@ -145,32 +189,21 @@ object ProjectHandler { warmups = req.warmups unit = req.unit } - success(EntryResponse(entry.id.value, entry.label.id.value, entry.timestamp, entry.warmups, entry.measurements, entry.unit)) - } - fun updateEntry(user: User, req: Pair) = transaction { - val entry = Entry.findById(req.first) ?: throw RouteError("Invalid entry ID.", HttpStatusCode.NotFound) - entry.project.isAccessible(user, true) - val changes = req.second - changes.label?.let { - val lbl = Label.findById(it) - when { - lbl == null -> throw RouteError("Invalid label ID.", HttpStatusCode.NotFound) - lbl.projectId.value != entry.projectId.value -> throw RouteError("Label is attached to a different project.", HttpStatusCode.Conflict) - else -> entry.label = lbl - } + val response = EntryResponse(entry.id.value, entry.label.id.value, entry.timestamp, entry.warmups, entry.measurements, entry.unit) + ServerScope.launch { + Buses.projectBus(proj.id.value).sendAll(ProjectEvent.NewEntry(response)) } - changes.timestamp?.let { entry.timestamp = it } - changes.warmups?.let { entry.warmups = it } - changes.measurements?.let { entry.measurements = it } - changes.unit?.let { entry.unit = it } - success(EmptyResponse()) + success(response) } fun deleteEntry(user: User, req: Uuid) = transaction { val entry = Entry.findById(req) ?: throw RouteError("Invalid entry ID.", HttpStatusCode.NotFound) entry.project.isAccessible(user, true) entry.delete() + ServerScope.launch { + Buses.projectBus(entry.project.id.value).sendAll(ProjectEvent.EntryDeleted(req)) + } success(EmptyResponse()) } } \ No newline at end of file diff --git a/server/src/main/kotlin/com/jaytux/phoebench/server/handlers/ServerScope.kt b/server/src/main/kotlin/com/jaytux/phoebench/server/handlers/ServerScope.kt new file mode 100644 index 0000000..0b83aef --- /dev/null +++ b/server/src/main/kotlin/com/jaytux/phoebench/server/handlers/ServerScope.kt @@ -0,0 +1,11 @@ +package com.jaytux.phoebench.server.handlers + +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlin.coroutines.CoroutineContext + +object ServerScope : CoroutineScope { + private val _actualScope = CoroutineScope(SupervisorJob() + Dispatchers.Default) + override val coroutineContext: CoroutineContext = _actualScope.coroutineContext +} \ No newline at end of file diff --git a/server/src/main/resources/simplelogger.properties b/server/src/main/resources/simplelogger.properties index e67eb96..e69de29 100644 --- a/server/src/main/resources/simplelogger.properties +++ b/server/src/main/resources/simplelogger.properties @@ -1 +0,0 @@ -org.slf4j.simpleLogger.log.io.ktor.server.plugins.cors.CORS=trace \ No newline at end of file