Reintroduce WebSockets for real-time run, judgement and viewer updates.

backend/src/main/kotlin/dev/dres/DRES.kt

@@ -117,7 +117,7 @@ object DRES {
         RunExecutor.init(store)
 
         /* Initialize EventStreamProcessor */
-        EventStreamProcessor.register( /* Add handlers here */)
+        EventStreamProcessor.register(RunExecutor)
         EventStreamProcessor.init()
 
         /* Initialize Rest API. */

backend/src/main/kotlin/dev/dres/api/rest/AccessManager.kt

@@ -165,4 +165,21 @@ object AccessManager {
     fun getRunManagerForUser(userId: UserId): Set<RunManager> = this.locks.read {
         return this.usersToRunMap[userId] ?: emptySet()
     }
+
+    /**
+     * Checks whether the user associated with the given [SessionToken] is allowed to observe
+     * (e.g., via WebSocket) the given [runManager], mirroring the access rules applied to the REST API.
+     *
+     * @param sessionId The [SessionToken] to check.
+     * @param runManager The [RunManager] the session wants to observe.
+     * @return True if access is permitted, false otherwise.
+     */
+    fun canViewEvaluation(sessionId: SessionToken?, runManager: RunManager): Boolean {
+        val roles = rolesOfSession(sessionId)
+        if (roles.contains(ApiRole.ADMIN)) return true
+        val userId = userIdForSession(sessionId) ?: return false
+        return (roles.contains(ApiRole.JUDGE) && runManager.template.judges.contains(userId)) ||
+            (roles.contains(ApiRole.VIEWER) && runManager.template.viewers.contains(userId)) ||
+            (roles.contains(ApiRole.PARTICIPANT) && runManager.template.hasParticipant(userId))
+    }
 }

backend/src/main/kotlin/dev/dres/api/rest/RestApi.kt

@@ -30,6 +30,7 @@ import dev.dres.api.rest.types.status.ErrorStatus
 import dev.dres.api.rest.types.users.ApiRole
 import dev.dres.data.model.config.Config
 import dev.dres.mgmt.cache.CacheManager
+import dev.dres.run.RunExecutor
 import dev.dres.utilities.NamedThreadFactory
 import io.javalin.Javalin
 import io.javalin.apibuilder.ApiBuilder.*
@@ -328,7 +329,7 @@ object RestApi {
                             }
                         }
                     }
-                    //ws("ws/run", runExecutor)
+                    ws("ws/run", RunExecutor::accept)
                 }
             }
 

backend/src/main/kotlin/dev/dres/api/rest/types/evaluation/websocket/ServerMessage.kt

@@ -1,13 +1,29 @@
 package dev.dres.api.rest.types.evaluation.websocket
 
+import dev.dres.api.rest.types.evaluation.ApiEvaluationOverview
+import dev.dres.api.rest.types.evaluation.ApiEvaluationState
+import dev.dres.api.rest.types.evaluation.ApiTeamTaskOverview
 import dev.dres.data.model.run.interfaces.EvaluationId
 import dev.dres.data.model.run.interfaces.TaskId
 
 /**
  * Message send by the DRES server via WebSocket to inform clients about the state of the run.
  *
+ * The optional [state] and [overview] fields carry a state diff so that receivers can update
+ * their local representation without issuing a separate HTTP request. [teamOverview] carries a
+ * scoped diff for a single team (e.g. in response to a submission) so that receivers don't need
+ * to be sent every team's overview just because one team's changed.
+ *
  * @author Ralph Gasser
- * @version 1.2.0
+ * @version 1.4.0
  */
-data class ServerMessage(val evaluationId: EvaluationId, val type: ServerMessageType, val taskId: TaskId? = null, val timestamp: Long = System.currentTimeMillis())
+data class ServerMessage(
+    val evaluationId: EvaluationId,
+    val type: ServerMessageType,
+    val taskId: TaskId? = null,
+    val timestamp: Long = System.currentTimeMillis(),
+    val state: ApiEvaluationState? = null,
+    val overview: ApiEvaluationOverview? = null,
+    val teamOverview: ApiTeamTaskOverview? = null
+)
 

backend/src/main/kotlin/dev/dres/api/rest/types/evaluation/websocket/ServerMessageType.kt

@@ -14,5 +14,6 @@ enum class ServerMessageType {
     TASK_START,             /** Task run started. */
     TASK_UPDATED,           /** State of task run has changed; mostly handles arrival of or changes to submissions. */
     TASK_END,               /** Tasks run ended. */
+    VIEWER_UPDATE,          /** A viewer connected or signalled ready — the viewer list has changed. */
     PING                    /** Keep alive */
 }

backend/src/main/kotlin/dev/dres/run/InteractiveAsynchronousRunManager.kt

@@ -21,6 +21,8 @@ import dev.dres.data.model.template.team.TeamId
 import dev.dres.run.RunManager.Companion.MAXIMUM_RUN_LOOP_ERROR_COUNT
 import dev.dres.run.audit.AuditLogSource
 import dev.dres.run.audit.AuditLogger
+import dev.dres.run.eventstream.EventStreamProcessor
+import dev.dres.run.eventstream.ViewerUpdateEvent
 import dev.dres.run.exceptions.IllegalRunStateException
 import dev.dres.run.exceptions.IllegalTeamIdException
 import dev.dres.run.score.scoreboard.Scoreboard
@@ -622,13 +624,15 @@ class InteractiveAsynchronousRunManager(
         val currentTaskId = this.currentTask(rac)?.taskId
         if (taskTemplateId == currentTaskId) {
             this.viewers[viewerInfo] = false
+            EventStreamProcessor.event(ViewerUpdateEvent(this.id))
         }
     }
 
     override fun viewerReady(taskTemplateId: TaskTemplateId, rac: RunActionContext, viewerInfo: ViewerInfo) {
         val currentTaskId = this.currentTask(rac)?.taskId
         if (taskTemplateId == currentTaskId) {
             this.viewers[viewerInfo] = true
+            EventStreamProcessor.event(ViewerUpdateEvent(this.id))
         }
     }
 

backend/src/main/kotlin/dev/dres/run/InteractiveSynchronousRunManager.kt

@@ -24,6 +24,7 @@ import dev.dres.run.audit.AuditLogSource
 import dev.dres.run.audit.AuditLogger
 import dev.dres.run.eventstream.EventStreamProcessor
 import dev.dres.run.eventstream.TaskEndEvent
+import dev.dres.run.eventstream.ViewerUpdateEvent
 import dev.dres.run.exceptions.IllegalRunStateException
 import dev.dres.run.score.scoreboard.Scoreboard
 import dev.dres.run.updatables.*
@@ -469,6 +470,7 @@ class InteractiveSynchronousRunManager(
 
         if (taskTemplateId == currentTemplateId) {
             this.readyLatch.register(viewerInfo)
+            EventStreamProcessor.event(ViewerUpdateEvent(this.id))
         }
 
     }
@@ -483,6 +485,7 @@ class InteractiveSynchronousRunManager(
             /* Since the viewer does send the ready message too early, we cannot care whether the task is (already) preparing or not */
                 this.readyLatch.register(viewerInfo) //avoid redying previously untracked viewers
                 this.readyLatch.setReady(viewerInfo)
+                EventStreamProcessor.event(ViewerUpdateEvent(this.id))
 //            }
         }