GameService (WebSockets w/ Cats Effect, FS2 and http4s)

GameService.scala

package com.pinkstack.tttx
package mk2.services

import mk2.services.Protocol.Connected

import cats.*
import cats.effect.*
import cats.effect.std.Queue
import cats.syntax.all.*
import fs2.Stream
import fs2.concurrent.Topic
import org.http4s.*
import org.http4s.dsl.io.*
import org.http4s.server.websocket.WebSocketBuilder
import org.http4s.websocket.WebSocketFrame
import org.http4s.websocket.WebSocketFrame.{Close, Text}

import java.util.UUID

type Connection = UUID

enum Protocol:
  case Connected(connection: Connection)
  case Disconnected(connection: Connection)
  case In(connection: Connection, message: String)
  case Out(connection: Connection)
  case Ping

final case class GameService(
  ws: WebSocketBuilder[IO],
  topic: Topic[IO, Protocol],
  queue: Queue[IO, Protocol]
) extends LoggingIO:
  import Protocol.*

  private def play(connection: Connection = UUID.randomUUID()): IO[Response[IO]] =
    val disconnected = queue.offer(Disconnected(connection))
    val connected    = Stream.emit(Connected(connection))

    val toClient: Stream[IO, WebSocketFrame] = topic
      .subscribe(500)
      .map {
        case Ping => WebSocketFrame.Ping()
        case m    => Text(m.toString)
      }
      .onFinalize(disconnected)

    def fromClient(wsStream: Stream[IO, WebSocketFrame]): Stream[IO, Unit] = {
      val parsedInput: Stream[IO, Protocol] = wsStream.collect {
        case Text(text, _) => In(connection, text.strip.trim)
        case Close(_)      => Disconnected(connection)
      }

      (connected ++ parsedInput)
        .covary[IO]
        .evalTap(msg => info(s"[$connection] ${msg}"))
        .evalTap(queue.offer)
        .void
    }.onFinalize(disconnected)

    ws.build(toClient, fromClient)

  def routes: HttpRoutes[IO] = HttpRoutes.of[IO] {
    case GET -> Root / "play"              => play()
    case GET -> Root / "play" / connection => IO(UUID.fromString(connection)).flatMap(play)
  }