CLI client, SSE

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<T> {
-                abstract fun toList(): List<T>
+                fun toList(): List<T>
             }
             data class DirectData<T>(val data: List<T>) : IData<T> {
                 override fun toList(): List<T> = data
@@ -110,6 +110,7 @@ object CLI {
             data class FileData<T>(val file: InputStream, val parse: (String) -> T?) : IData<T> {
                 override fun toList(): List<T> {
                     val raw = file.bufferedReader().use { it.readText() }.split(',')
+                    
                     val parsed = ArrayList<T>(raw.size)
                     val errors = mutableListOf<String>()
                     raw.forEach {

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

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

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
+                        _auth.onRefresh(it)
-                        val ref = _auth.refresh.value ?: return@refreshTokens null
+                        val res = _auth.asBearer()
-                        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
                         res
                     }
+                    res
                 }
             }
         }
@@ -94,14 +87,19 @@ class Client private constructor(private val _auth: AuthProvider) {
     ): Either<ErrorResponse, TRes> =
         callRoute(_client, route, body)
 
-    suspend fun <T> connectSSE(route: String, serializer: KSerializer<T>, onEvent: suspend (T) -> Unit) {
+    suspend fun <TParams : Any, TEvent : Any> callSSE(
-        println("Client tries to set up SSE to $route")
+        route: SSERoute<TParams, TEvent>,
-        val server = _auth.server.value ?: throw IllegalStateException("No server URL set.")
+        params: TParams,
-        _client.sse(urlString = "$server$route", showCommentEvents = true, showRetryEvents = true) {
+        handler: suspend (Either<ErrorResponse, TEvent>) -> Unit
-            incoming.collect {
+    ): Either<ErrorResponse, Unit> {
-                println("RECEIVE: $it")
+        try {
-                if(it.data != null) onEvent(Json.decodeFromString(serializer, it.data!!))
+            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
         }
     }
 

clients/compose/src/commonMain/kotlin/com/jaytux/phoebench/clients/Util.kt

@@ -29,7 +29,7 @@ fun <T> MutableState<T>.immutable(): State<T> = this
 fun <T> T.ignore() {}
 
 inline fun <reified R> 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 <T, X : Comparable<X>> List<T>.insort(elem: T, crossinline sortBy: (T) -> X): List<T> {
+    val insertionPoint = binarySearchBy(sortBy(elem), selector = sortBy)
+    if(insertionPoint >= 0) return this
+    val index = -insertionPoint - 1
+
+    return toMutableList().apply { add(index, elem) }
+}

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<String?>(null)
     private val _isAdmin = mutableStateOf(false)
@@ -33,8 +40,6 @@ class HomeVM(
     private val _users = mutableStateOf(listOf<UserListResponse.UserData>())
     private val _invites = mutableStateOf(listOf<InviteListResponse.Invite>())
 
-    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<HomeResponse.ProjectSummary>()) {
-                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
+                _ownProjects.value = it.ownProjects.sortedBy { p -> p.name }
-                _publicProjects.value = it.publicProjects
+                _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 {
+                    _invites.value = r.uuids
-                        it.copy(expires = it.expires)
-                    }
                 }
 
                 _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()
             }
         }
     }

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<Float>, measurements: List<Float>,
                          unit: TimeUnit): Either<ErrorResponse, EntryResponse>
-    suspend fun updateEntry(id: Uuid, label: Uuid? = null, timestamp: Instant? = null, warmups: List<Float>? = null,
-                            measurements: List<Float>? = null, unit: TimeUnit? = null): Either<ErrorResponse, Unit>
     suspend fun deleteEntry(id: Uuid): Either<ErrorResponse, Unit>
 
     companion object {
@@ -60,14 +58,6 @@ interface IProjectRepo {
             ): Either<ErrorResponse, EntryResponse> =
                 _client.callRoute(Routes.Entry.new, EntryRequest(label, timestamp, _projectId, warmups, measurements, unit))
 
-            override suspend fun updateEntry(id: Uuid, label: Uuid?, timestamp: Instant?,
-                                             warmups: List<Float>?, measurements: List<Float>?,
-                                             unit: TimeUnit?
-            ): Either<ErrorResponse, Unit> =
-                _client.callRoute(Routes.Entry.update,
-                    id to PartialEntryRequest(label, timestamp, null, warmups, measurements, unit)
-                ).ignoreValue()
-
             override suspend fun deleteEntry(id: Uuid): Either<ErrorResponse, Unit> =
                 _client.callRoute(Routes.Entry.delete, id).ignoreValue()
         }

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<ErrorResponse, HomeEvent>) -> Unit): Either<ErrorResponse, Unit>
+    suspend fun connectAdmin(onEvent: suspend (Either<ErrorResponse, AdminEvent>) -> Unit): Either<ErrorResponse, Unit>
+    suspend fun connectProject(id: Uuid, onEvent: suspend (Either<ErrorResponse, ProjectEvent>) -> Unit): Either<ErrorResponse, Unit>
+
+    companion object {
+        class Default(private val _client: Client) : ISSERepo {
+            override suspend fun connectHome(onEvent: suspend (Either<ErrorResponse, HomeEvent>) -> Unit): Either<ErrorResponse, Unit> =
+                _client.callSSE(Routes.SSE.home, Unit, onEvent)
+
+            override suspend fun connectAdmin(onEvent: suspend (Either<ErrorResponse, AdminEvent>) -> Unit): Either<ErrorResponse, Unit> =
+                _client.callSSE(Routes.SSE.admin, Unit, onEvent)
+
+            override suspend fun connectProject(id: Uuid, onEvent: suspend (Either<ErrorResponse, ProjectEvent>) -> Unit): Either<ErrorResponse, Unit> =
+                _client.callSSE(Routes.SSE.projectSpecific, id, onEvent)
+        }
+
+        fun default(client: Client) = Default(client)
+    }
+}

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<Float>?, measurements: List<Float>?, 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 {

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 }

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)

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()
+}

common/src/commonMain/kotlin/com/jaytux/phoebench/common/Routes.kt

@@ -38,7 +38,12 @@ object Routes {
 
     object Entry {
         val new = ApiRoute.post<EntryRequest, EntryResponse>("/entry", Elevation.AUTH)
-        val update = ApiRoute.patchUuidNoRes<PartialEntryRequest>("/entry", Elevation.AUTH)
         val delete = ApiRoute.deleteUuidNoRes("/entry", Elevation.AUTH)
     }
+
+    object SSE {
+        val home = SSERoute.noArgs<HomeEvent>("/rt/home", Elevation.AUTH)
+        val admin = SSERoute.noArgs<AdminEvent>("/rt/admin", Elevation.ADMIN)
+        val projectSpecific = SSERoute.uuid<ProjectEvent>("/rt/project", Elevation.AUTH)
+    }
 }

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<TParams, TEvent: Any>(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<ErrorResponse, TEvent>) -> Unit): Either<ErrorResponse, Unit> {
+        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<ErrorResponse>()
+            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<TEvent: Any>(path: String, elevation: Elevation, resType: TypeInfo) : SSERoute<Unit, TEvent>(path, elevation, resType) {
+        override fun extractParams(reqParams: Parameters) {}
+    }
+
+    class SSERoute1<T1, TEvent: Any>(path: String, elevation: Elevation, resType: TypeInfo, val urlEncode: (T1) -> String, val urlDecode: (String?) -> T1?)
+        : SSERoute<T1, TEvent>(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 <reified TEvent: Any> noArgs(path: String, elevation: Elevation) =
+            SSERoute0<TEvent>(path, elevation, typeInfo<TEvent>())
+
+        inline fun <T, reified TEvent: Any> single(path: String, elevation: Elevation,
+            noinline urlEncode: (T) -> String = { it.toString() }, noinline urlDecode: (String?) -> T?
+        ) = SSERoute1<T, TEvent>(path, elevation, typeInfo<TEvent>(), urlEncode, urlDecode)
+
+        inline fun <reified TEvent: Any> uuid(path: String, elevation: Elevation) = single<Uuid, TEvent>(path, elevation) {
+            it?.let { p -> Uuid.parseOrNull(p) }
+        }
+    }
+}

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<Uuid, SSEBus<ProjectEvent>>()
+
+    val homeBus = SSEBus<HomeEvent>(typeOf<HomeEvent>(), serializer<HomeEvent>())
+    val adminBus = SSEBus<AdminEvent>(typeOf<AdminEvent>(), serializer<AdminEvent>())
+
+    fun projectBus(id: Uuid) = _projectBuses.computeIfAbsent(id) {
+        SSEBus<ProjectEvent>(typeOf<ProjectEvent>(), serializer<ProjectEvent>())
+    }
+
+    fun allBuses(): List<SSEBus<*>> {
+        val res = ArrayList<SSEBus<*>>(_projectBuses.size + 2)
+        res.addAll(_projectBuses.values)
+        res.add(homeBus)
+        res.add(adminBus)
+        return res
+    }
+}

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.*
-import com.jaytux.phoebench.common.HomeResponse
-import com.jaytux.phoebench.common.Routes
-import com.jaytux.phoebench.common.fold
 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.*
-import com.jaytux.phoebench.server.handlers.ProjectHandler
+import com.jaytux.phoebench.server.handlers.ProjectHandler.accessibleProject
-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 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.cors.routing.*
-import io.ktor.server.plugins.statuspages.StatusPages
+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.*
-import io.ktor.server.sse.sse
-import kotlinx.serialization.encodeToString
 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<String>) {
     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<HomeResponse.ProjectSummary>("/rt/project")
+            sseAuth(Routes.SSE.home, { _, _ -> }) { _, _ -> Buses.homeBus }
-            sse("/rt/project") {
+            sseAdmin(Routes.SSE.admin) { _, _ -> Buses.adminBus }
-                try {
+            sseAuth(Routes.SSE.projectSpecific,
-                    println("Attempt to set up SSE")
+                { user, uuid ->
-                    val principal = call.principal<JWTPrincipal>()
+                    transaction { accessibleProject(user, uuid, false) }
-                    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
-                    }
                 }
-                catch(e: RouteError) {
+            ) { _, id -> Buses.projectBus(id) }
-                    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}") {}
         }
 
         get("{...}") {

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<T> private constructor (private val _containedType: KType, val serializer: KSerializer<T>) {
+class SSEBus<T>(private val _containedType: KType, val serializer: KSerializer<T>) {
     object Cancellation
 
+    private val _unAuthFlow = MutableSharedFlow<T>()
     private val _flows = ConcurrentHashMap<Uuid, Pair<Int, MutableSharedFlow<Either<Cancellation, T>>>>()
 
     fun register(user: Uuid): SharedFlow<Either<Cancellation, T>> =
         _flows.computeIfAbsent(user) { 1 to MutableSharedFlow(extraBufferCapacity = 64) }.second
 
+    fun unRegistered(): SharedFlow<T> = _unAuthFlow
+
     suspend fun send(user: Uuid, data: T) {
         _flows[user]?.second?.emit(data.value())
     }
@@ -29,6 +33,10 @@ class SSEBus<T> 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<T> private constructor (private val _containedType: KType, val seri
         _flows.remove(user)
     }
 
-    companion object {
+    suspend fun forceDisconnectExcept(user: Uuid) {
-        private val _busCache = mutableMapOf<String, SSEBus<*>>()
+        synchronized(_flows) {
-
+            val map = _flows.values.filter { it != _flows[user] }
-        fun <T> register(topic: String, contained: KType, serializer: KSerializer<T>): SSEBus<T> {
+            _flows.keys.retainAll(setOf(user))
-            val bus = SSEBus<T>(contained, serializer)
+            map
-            _busCache.compute(topic) { k, existing ->
+        }.forEach { it.second.emit(Cancellation.error()) }
-                if(existing != null) throw IllegalArgumentException("Bus for $topic exists already")
-                bus
-            }
-            return bus
-        }
-
-        inline fun <reified T> register(topic: String) =
-            register<T>(topic, typeOf<T>(), serializer<T>())
-
-        fun <T> getBus(topic: String, contained: KType): SSEBus<T>? {
-            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<T>
-        }
-
-        inline fun <reified T> getBus(topic: String) =
-            getBus<T>(topic, typeOf<T>())
     }
 }

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())
     }
 

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 <reified TReq: Any> ApiRoute<TReq, *>.paramArgs(call: ApplicationCall): TReq =
     parseParams(call.parameters) ?: throw RouteError(
@@ -140,4 +161,101 @@ inline fun <reified TReq: Any, reified TRes: Any> Route.patchAuth(api: ApiRoute<
     wrapperAuth(api, Route::patch, handler)
 
 inline fun <reified TReq: Any, reified TRes: Any> Route.patchAdmin(api: ApiRoute<TReq, TRes>, noinline handler: suspend (User, TReq) -> Pair<HttpStatusCode, TRes>): Route =
-    wrapperAdmin(api, Route::patch, handler)
+    wrapperAdmin(api, Route::patch, handler)
+
+inline fun <reified TParams: Any, reified TEvent: Any, TInter, TFlow> Route.wrapSSE(
+    api: SSERoute<TParams, TEvent>, noinline extra: suspend (ApplicationCall, TParams) -> TInter,
+    noinline prepare: suspend (TInter, TParams) -> SSEBus<TEvent>,
+    noinline extract: suspend (SSEBus<TEvent>, TInter, TParams) -> SharedFlow<TFlow>,
+    noinline handler: suspend (TFlow, sender: suspend (TEvent) -> Unit) -> Unit,
+    noinline onCancel: suspend (SSEBus<TEvent>, 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 <reified TParams: Any, reified TEvent: Any> Route.wrapAuthSSE(
+    api: SSERoute<TParams, TEvent>,
+    noinline verifyUser: suspend (User, TParams) -> Unit,
+    noinline prepare: suspend (User, TParams) -> SSEBus<TEvent>
+) = wrapSSE(api,
+        extra = { call, params ->
+            val principal = call.principal<JWTPrincipal>()
+            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 <reified TParams: Any, reified TEvent: Any> Route.sse(api: SSERoute<TParams, TEvent>, noinline setup: suspend (TParams) -> SSEBus<TEvent>) =
+    wrapSSE(api,
+        extra = { _, _ -> },
+        prepare = { _, params -> setup(params)  },
+        extract = { bus, _, _ -> bus.unRegistered() },
+        handler = { it, sender -> sender(it) },
+        onCancel = { _, _, _, _ -> }
+    )
+
+inline fun <reified TParams: Any, reified TEvent: Any> Route.sseAuth(api: SSERoute<TParams, TEvent>,
+    noinline verifyUser: suspend (User, TParams) -> Unit, noinline setup: suspend (User, TParams) -> SSEBus<TEvent>
+) = wrapAuthSSE(api, verifyUser, setup)
+
+inline fun <reified TParams: Any, reified TEvent: Any> Route.sseAdmin(api: SSERoute<TParams, TEvent>,
+    noinline setup: suspend (User, TParams) -> SSEBus<TEvent>
+) = wrapAuthSSE(api, { user, _ ->
+    if(!user.isAdmin) {
+        throw RouteError("Admin access required", HttpStatusCode.Forbidden)
+    }
+}, setup)

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<HomeResponse.ProjectSummary>("/rt/project") ?: return@also
+        ServerScope.launch {
-        val summary = HomeResponse.ProjectSummary(proj.id, proj.name, proj.isPublic, NamedID(user.username, user.id.value))
+            val bus = Buses.homeBus
-        if(proj.isPublic) bus.sendAll(summary)
+            val summary = HomeResponse.ProjectSummary(proj.id, proj.name, proj.isPublic, NamedID(user.username, user.id.value))
-        else bus.send(user.id.value, summary)
+            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<Uuid, PartialLabelRequest>) = 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<Uuid, PartialEntryRequest>) = transaction {
+        val response = EntryResponse(entry.id.value, entry.label.id.value, entry.timestamp, entry.warmups, entry.measurements, entry.unit)
-        val entry = Entry.findById(req.first) ?: throw RouteError("Invalid entry ID.", HttpStatusCode.NotFound)
+        ServerScope.launch {
-        entry.project.isAccessible(user, true)
+            Buses.projectBus(proj.id.value).sendAll(ProjectEvent.NewEntry(response))
-        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
-            }
         }
-        changes.timestamp?.let { entry.timestamp = it }
+        success(response)
-        changes.warmups?.let { entry.warmups = it }
-        changes.measurements?.let { entry.measurements = it }
-        changes.unit?.let { entry.unit = it }
-        success(EmptyResponse())
     }
 
     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())
     }
 }

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
+}

server/src/main/resources/simplelogger.properties

@@ -1 +0,0 @@
-org.slf4j.simpleLogger.log.io.ktor.server.plugins.cors.CORS=trace