Allow candidate exchange via REST+SSE

Closes #19

src/main/kotlin/com/faforever/icebreaker/security/CurrentUserService.kt

@@ -0,0 +1,17 @@
+package com.faforever.icebreaker.security
+
+import io.quarkus.oidc.runtime.OidcJwtCallerPrincipal
+import io.quarkus.security.identity.SecurityIdentity
+import jakarta.enterprise.context.ApplicationScoped
+
+@ApplicationScoped
+class CurrentUserService(
+    private val securityIdentity: SecurityIdentity,
+) {
+    fun getCurrentUserId(): Long? {
+        val principal = (securityIdentity.principal as? OidcJwtCallerPrincipal)
+        val subject = principal?.claims?.claimsMap?.get("sub") as? String
+
+        return subject?.toLong()
+    }
+}

src/main/kotlin/com/faforever/icebreaker/service/CandidatesMessage.kt

@@ -0,0 +1,59 @@
+package com.faforever.icebreaker.service
+
+import com.fasterxml.jackson.annotation.JsonInclude
+import com.fasterxml.jackson.annotation.JsonSubTypes
+import com.fasterxml.jackson.annotation.JsonTypeInfo
+import com.fasterxml.jackson.annotation.JsonValue
+
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "eventType")
+@JsonSubTypes(
+    JsonSubTypes.Type(value = CandidatesMessage::class, name = "candidates"),
+    JsonSubTypes.Type(value = ConnectedMessage::class, name = "connected"),
+)
+@JsonInclude(JsonInclude.Include.ALWAYS)
+interface EventMessage {
+    val gameId: Long
+    val senderId: Long
+    val recipientId: Long?
+}
+
+@JvmRecord
+data class ConnectedMessage(
+    override val gameId: Long,
+    override val senderId: Long,
+    override val recipientId: Long? = null,
+) : EventMessage
+
+@JvmRecord
+data class CandidatesMessage(
+    override val gameId: Long,
+    override val senderId: Long,
+    override val recipientId: Long,
+    val candidates: List<CandidateDescriptor>,
+) : EventMessage {
+
+    enum class CandidateType(@JsonValue val jsonValue: String) {
+        PEER_REFLEXIVE_CANDIDATE("prflx"),
+        SERVER_REFLEXIVE_CANDIDATE("srflx"),
+        RELAYED_CANDIDATE("relay"),
+        HOST_CANDIDATE("host"),
+        LOCAL_CANDIDATE("local"),
+        STUN_CANDIDATE("stun"),
+    }
+
+    @JvmRecord
+    data class CandidateDescriptor(
+        val foundation: String,
+        val protocol: String,
+        val priority: Long,
+        val ip: String?,
+        val port: Int,
+        val type: CandidateType,
+        val generation: Int,
+        val id: String,
+        val relAddr: String?,
+        val relPort: Int,
+    ) : Comparable<CandidateDescriptor> {
+        override operator fun compareTo(other: CandidateDescriptor) = (other.priority - priority).toInt()
+    }
+}

src/main/kotlin/com/faforever/icebreaker/service/SessionService.kt

@@ -3,12 +3,15 @@ package com.faforever.icebreaker.service
 import com.faforever.icebreaker.config.FafProperties
 import com.faforever.icebreaker.persistence.IceSessionEntity
 import com.faforever.icebreaker.persistence.IceSessionRepository
+import com.faforever.icebreaker.security.CurrentUserService
 import com.faforever.icebreaker.util.AsyncRunner
 import io.quarkus.scheduler.Scheduled
 import io.quarkus.security.ForbiddenException
 import io.quarkus.security.UnauthorizedException
 import io.quarkus.security.identity.SecurityIdentity
 import io.smallrye.jwt.build.Jwt
+import io.smallrye.mutiny.Multi
+import io.smallrye.mutiny.helpers.MultiEmitterProcessor
 import jakarta.enterprise.inject.Instance
 import jakarta.inject.Singleton
 import jakarta.transaction.Transactional
@@ -26,8 +29,11 @@ class SessionService(
     private val fafProperties: FafProperties,
     private val iceSessionRepository: IceSessionRepository,
     private val securityIdentity: SecurityIdentity,
+    private val currentUserService: CurrentUserService,
 ) {
     private val activeSessionHandlers = sessionHandlers.filter { it.active }
+    private val eventEmitter = MultiEmitterProcessor.create<EventMessage>()
+    private val eventBroadcast: Multi<EventMessage> = eventEmitter.toMulti().broadcast().toAllSubscribers()
 
     fun buildToken(gameId: Long): String {
         val userId =
@@ -117,4 +123,27 @@ class SessionService(
                 iceSessionRepository.delete(iceSession)
             }
     }
+
+    fun listenForEventMessages(gameId: Long): Multi<EventMessage> {
+        val userId = currentUserService.getCurrentUserId()
+        eventEmitter.emit(ConnectedMessage(gameId = gameId, senderId = currentUserService.getCurrentUserId()!!))
+
+        return eventBroadcast.filter {
+            it.gameId == gameId && (it.recipientId == userId || (it.recipientId == null && it.senderId != userId))
+        }
+    }
+
+    fun onCandidatesReceived(gameId: Long, candidatesMessage: CandidatesMessage) {
+        // Check messages for manipulation. We need to prevent cross-channel vulnerabilities.
+        check(candidatesMessage.gameId == gameId) {
+            "gameId $gameId from endpoint does not match gameId ${candidatesMessage.gameId} in candidateMessage"
+        }
+
+        val currentUserId = currentUserService.getCurrentUserId()
+        check(candidatesMessage.senderId == currentUserService.getCurrentUserId()) {
+            "current user id $currentUserId from endpoint does not match sourceId ${candidatesMessage.senderId} in candidateMessage"
+        }
+
+        eventEmitter.emit(candidatesMessage)
+    }
 }

src/main/kotlin/com/faforever/icebreaker/web/SessionController.kt

@@ -1,9 +1,12 @@
 package com.faforever.icebreaker.web
 
+import com.faforever.icebreaker.service.CandidatesMessage
+import com.faforever.icebreaker.service.EventMessage
 import com.faforever.icebreaker.service.Session
 import com.faforever.icebreaker.service.SessionService
 import io.quarkus.runtime.annotations.RegisterForReflection
 import io.quarkus.security.PermissionsAllowed
+import io.smallrye.mutiny.Multi
 import jakarta.inject.Singleton
 import jakarta.ws.rs.Consumes
 import jakarta.ws.rs.GET
@@ -12,6 +15,7 @@ import jakarta.ws.rs.Path
 import jakarta.ws.rs.Produces
 import jakarta.ws.rs.core.MediaType
 import org.jboss.resteasy.reactive.RestPath
+import org.jboss.resteasy.reactive.RestStreamElementType
 
 @Path("/session")
 @Singleton
@@ -61,4 +65,18 @@ class SessionController(
                 ),
             )
         }
+
+    @POST
+    @Path("/game/{gameId}/events")
+    @PermissionsAllowed("USER:lobby")
+    @Consumes(MediaType.APPLICATION_JSON)
+    fun postEvent(@RestPath gameId: Long, candidatesMessage: CandidatesMessage) {
+        sessionService.onCandidatesReceived(gameId, candidatesMessage)
+    }
+
+    @GET
+    @Path("/game/{gameId}/events")
+    @PermissionsAllowed("USER:lobby")
+    @RestStreamElementType(MediaType.APPLICATION_JSON)
+    fun getSessionEvents(@RestPath gameId: Long): Multi<EventMessage> = sessionService.listenForEventMessages(gameId)
 }

src/main/resources/application.yaml

@@ -101,11 +101,11 @@ smallrye:
          tQIDAQAB
          -----END PUBLIC KEY-----
     log:
-      level: DEBUG
+      level: INFO
       category:
         "org.hibernate.SQL":
           level: DEBUG
         "com.faforever":
            level: DEBUG
         "io.quarkus":
-           level: DEBUG
+           level: INFO