diff --git a/apps/backend-quarkus/pom.xml b/apps/backend-quarkus/pom.xml
index 0569e61..52f430f 100644
--- a/apps/backend-quarkus/pom.xml
+++ b/apps/backend-quarkus/pom.xml
@@ -38,7 +38,11 @@
io.quarkus
- quarkus-rest-jackson
+ quarkus-rest
+
+
+ io.quarkus
+ quarkus-jackson
io.quarkus
diff --git a/apps/backend-quarkus/src/main/java/app/playmancala/Backend.java b/apps/backend-quarkus/src/main/java/app/playmancala/Backend.java
new file mode 100644
index 0000000..2386982
--- /dev/null
+++ b/apps/backend-quarkus/src/main/java/app/playmancala/Backend.java
@@ -0,0 +1,186 @@
+package app.playmancala;
+
+import java.sql.SQLException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import javax.sql.DataSource;
+
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.jboss.logging.Logger;
+
+import app.playmancala.auth.TokenVerifier;
+import app.playmancala.game.GameClockScheduler;
+import app.playmancala.game.GameManager;
+import app.playmancala.game.GameRecorder;
+import app.playmancala.game.GameStore;
+import app.playmancala.game.LiveGames;
+import app.playmancala.game.MatchMaker;
+import app.playmancala.history.GameHistoryService;
+import app.playmancala.history.Migrations;
+import app.playmancala.history.PostgresGameHistoryRepository;
+import app.playmancala.live.ValkeyCache;
+import app.playmancala.live.ValkeyLiveGameStore;
+import app.playmancala.loop.GameLoop;
+import app.playmancala.rtmt.RtmtServer;
+import io.quarkus.redis.datasource.RedisDataSource;
+import io.quarkus.runtime.Startup;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+
+// Builds the backend from the infrastructure Quarkus provides and starts it (docs/specs/quarkus-backend, Startup).
+@Startup
+@ApplicationScoped
+public class Backend {
+ private static final Logger LOG = Logger.getLogger(Backend.class);
+ private static final long MIGRATION_RETRY_MS = 30_000;
+ private static final long VALKEY_STARTUP_WAIT_MS = 5_000;
+
+ @Inject
+ DataSource dataSource;
+
+ @Inject
+ RedisDataSource redis;
+
+ @Inject
+ TokenVerifier tokenVerifier;
+
+ @ConfigProperty(name = "mancala.auth.legacy-userkey-auth")
+ String legacyUserKeyAuth;
+
+ @ConfigProperty(name = "mancala.game-clock-ms")
+ long gameClockMs;
+
+ private GameLoop loop;
+ // Valkey writes, in order.
+ private ExecutorService storeWriter;
+ // Migrations and saving finished games.
+ private ScheduledExecutorService recorderIo;
+ // Token checks.
+ private ExecutorService verifier;
+
+ private RtmtServer rtmt;
+ private GameStore gameStore;
+ private GameManager gameManager;
+ private LiveGames liveGames;
+ private GameHistoryService history;
+ private GameClockScheduler clockScheduler;
+
+ @PostConstruct
+ void start() {
+ loop = new GameLoop();
+ storeWriter = Executors.newSingleThreadExecutor(Thread.ofPlatform().name("valkey-writer").daemon().factory());
+ recorderIo = Executors.newSingleThreadScheduledExecutor(Thread.ofPlatform().name("game-recorder").daemon().factory());
+ verifier = Executors.newVirtualThreadPerTaskExecutor();
+
+ ValkeyLiveGameStore liveGameStore = new ValkeyLiveGameStore(redis);
+ PostgresGameHistoryRepository repository = new PostgresGameHistoryRepository(dataSource);
+ GameRecorder recorder = new GameRecorder(repository, liveGameStore, recorderIo);
+ history = new GameHistoryService(repository, new ValkeyCache(redis), recorder::getUnsaved);
+ recorder.onSaved(history::onGameSaved);
+
+ // Exactly "true", like the Node backend.
+ rtmt = new RtmtServer(tokenVerifier, "true".equals(legacyUserKeyAuth), loop, verifier,
+ RtmtServer.DEFAULT_AUTH_TIMEOUT_MS);
+ gameStore = new GameStore();
+ gameManager = new GameManager(gameStore, rtmt, new MatchMaker(rtmt), liveGameStore, recorder, storeWriter,
+ loop::now, gameClockMs);
+ liveGames = new LiveGames(rtmt, gameStore, gameManager, loop);
+ clockScheduler = new GameClockScheduler(gameManager::endGamesOutOfTime, loop,
+ GameClockScheduler.CLOCK_CHECK_INTERVAL_MS);
+
+ // The game runs without the database; saved games wait in the recorder until migrations have run.
+ recorderIo.execute(this::migrate);
+ recorderIo.scheduleWithFixedDelay(() -> guarded(recorder::retryPending), 0, GameRecorder.RETRY_INTERVAL_MS,
+ TimeUnit.MILLISECONDS);
+ // The first task on the loop, so socket events queue behind it and reconnecting players find their games.
+ loop.execute(() -> {
+ waitForValkey();
+ gameManager.restoreLiveGames();
+ clockScheduler.start();
+ rtmt.start();
+ });
+ }
+
+ @PreDestroy
+ void stop() {
+ loop.close();
+ storeWriter.shutdown();
+ recorderIo.shutdownNow();
+ verifier.shutdownNow();
+ }
+
+ private void migrate() {
+ try {
+ Migrations.run(dataSource);
+ } catch (SQLException | RuntimeException e) {
+ LOG.errorf("(DB) Migrations failed, retrying in %ds: %s", MIGRATION_RETRY_MS / 1000, e.getMessage());
+ recorderIo.schedule(this::migrate, MIGRATION_RETRY_MS, TimeUnit.MILLISECONDS);
+ }
+ }
+
+ private void waitForValkey() {
+ long deadline = System.currentTimeMillis() + VALKEY_STARTUP_WAIT_MS;
+ while (true) {
+ try {
+ redis.execute("PING");
+ return;
+ } catch (RuntimeException e) {
+ if (System.currentTimeMillis() >= deadline) {
+ LOG.error("(Valkey) Not reachable yet; starting without restoring live games");
+ return;
+ }
+ }
+ try {
+ Thread.sleep(250);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+ }
+
+ private static void guarded(Runnable task) {
+ try {
+ task.run();
+ } catch (RuntimeException e) {
+ LOG.error("(Backend) Task failed", e);
+ }
+ }
+
+ public GameLoop loop() {
+ return loop;
+ }
+
+ public RtmtServer rtmt() {
+ return rtmt;
+ }
+
+ public GameStore gameStore() {
+ return gameStore;
+ }
+
+ public GameManager gameManager() {
+ return gameManager;
+ }
+
+ public LiveGames liveGames() {
+ return liveGames;
+ }
+
+ public GameHistoryService history() {
+ return history;
+ }
+
+ public TokenVerifier tokenVerifier() {
+ return tokenVerifier;
+ }
+
+ public boolean legacyUserKeyAuth() {
+ return "true".equals(legacyUserKeyAuth);
+ }
+}
diff --git a/apps/backend-quarkus/src/main/java/app/playmancala/auth/TokenVerifierProducer.java b/apps/backend-quarkus/src/main/java/app/playmancala/auth/TokenVerifierProducer.java
new file mode 100644
index 0000000..da25e80
--- /dev/null
+++ b/apps/backend-quarkus/src/main/java/app/playmancala/auth/TokenVerifierProducer.java
@@ -0,0 +1,23 @@
+package app.playmancala.auth;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+
+import jakarta.enterprise.inject.Produces;
+import jakarta.inject.Singleton;
+
+public class TokenVerifierProducer {
+ @Produces
+ @Singleton
+ TokenVerifier tokenVerifier(
+ @ConfigProperty(name = "mancala.auth.issuer") String issuer,
+ @ConfigProperty(name = "mancala.auth.jwks-url") Optional jwksUrl,
+ @ConfigProperty(name = "mancala.auth.client-ids") String clientIds) {
+ // Comma-separated; spaces and empty entries are ignored, as in the Node backend.
+ List allowedClientIds = Arrays.stream(clientIds.split(",")).map(String::trim).filter(id -> !id.isEmpty()).toList();
+ return KeycloakTokenVerifier.create(issuer, allowedClientIds, jwksUrl.orElse(null));
+ }
+}
diff --git a/apps/backend-quarkus/src/main/java/app/playmancala/http/BearerAuth.java b/apps/backend-quarkus/src/main/java/app/playmancala/http/BearerAuth.java
new file mode 100644
index 0000000..cc69264
--- /dev/null
+++ b/apps/backend-quarkus/src/main/java/app/playmancala/http/BearerAuth.java
@@ -0,0 +1,35 @@
+package app.playmancala.http;
+
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.jboss.logging.Logger;
+
+import app.playmancala.auth.InvalidTokenException;
+import app.playmancala.auth.TokenVerifier;
+import app.playmancala.model.AuthUser;
+import jakarta.ws.rs.WebApplicationException;
+import jakarta.ws.rs.core.Response;
+
+// Requires an "Authorization: Bearer " header with the same Keycloak token the WebSocket uses.
+final class BearerAuth {
+ private static final Logger LOG = Logger.getLogger(BearerAuth.class);
+ private static final Pattern BEARER = Pattern.compile("^Bearer (.+)$");
+
+ private BearerAuth() {
+ }
+
+ // The user, or a 401 response.
+ static AuthUser requireUser(TokenVerifier tokenVerifier, String authorization) {
+ Matcher match = BEARER.matcher(authorization == null ? "" : authorization);
+ if (!match.matches()) throw new WebApplicationException(Response.Status.UNAUTHORIZED);
+ try {
+ return tokenVerifier.verify(match.group(1));
+ } catch (InvalidTokenException e) {
+ throw new WebApplicationException(Response.Status.UNAUTHORIZED);
+ } catch (RuntimeException e) {
+ LOG.error("(Auth) Token verification failed", e);
+ throw new WebApplicationException(Response.Status.UNAUTHORIZED);
+ }
+ }
+}
diff --git a/apps/backend-quarkus/src/main/java/app/playmancala/http/ErrorMapper.java b/apps/backend-quarkus/src/main/java/app/playmancala/http/ErrorMapper.java
new file mode 100644
index 0000000..7b01516
--- /dev/null
+++ b/apps/backend-quarkus/src/main/java/app/playmancala/http/ErrorMapper.java
@@ -0,0 +1,20 @@
+package app.playmancala.http;
+
+import org.jboss.logging.Logger;
+import org.jboss.resteasy.reactive.server.ServerExceptionMapper;
+
+import io.vertx.core.http.HttpServerRequest;
+import jakarta.ws.rs.WebApplicationException;
+import jakarta.ws.rs.core.Response;
+
+// A failing dependency (the database, mostly) answers 503, as the Express error handler did.
+public class ErrorMapper {
+ private static final Logger LOG = Logger.getLogger(ErrorMapper.class);
+
+ @ServerExceptionMapper
+ public Response map(RuntimeException error, HttpServerRequest request) {
+ if (error instanceof WebApplicationException webError) return webError.getResponse();
+ LOG.errorf("(HTTP) %s %s failed: %s", request.method(), request.path(), error.getMessage());
+ return Response.status(Response.Status.SERVICE_UNAVAILABLE).build();
+ }
+}
diff --git a/apps/backend-quarkus/src/main/java/app/playmancala/http/GameResource.java b/apps/backend-quarkus/src/main/java/app/playmancala/http/GameResource.java
new file mode 100644
index 0000000..9addfc3
--- /dev/null
+++ b/apps/backend-quarkus/src/main/java/app/playmancala/http/GameResource.java
@@ -0,0 +1,125 @@
+package app.playmancala.http;
+
+import java.math.BigInteger;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import app.playmancala.Backend;
+import app.playmancala.game.BackendGame;
+import app.playmancala.game.LiveGames;
+import app.playmancala.history.GameHistoryService;
+import app.playmancala.model.AuthUser;
+import app.playmancala.model.GameRecord;
+import app.playmancala.model.Json;
+import app.playmancala.model.LiveGameSummary;
+import jakarta.inject.Inject;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.HeaderParam;
+import jakarta.ws.rs.NotFoundException;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.PathParam;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.core.HttpHeaders;
+import jakarta.ws.rs.core.MediaType;
+import jakarta.ws.rs.core.Response;
+
+// The HTTP API. Blocking: runs on worker threads, reads live games on the game loop and saved games itself.
+// Bodies are written with the shared mapper, so they are the same JSON as the WebSocket messages.
+@Path("/")
+public class GameResource {
+ private static final Pattern LEADING_INTEGER = Pattern.compile("^\\s*([+-]?\\d+)");
+
+ @Inject
+ Backend backend;
+
+ @GET
+ @Produces(MediaType.TEXT_PLAIN)
+ public String root() {
+ return "Server up and running! ci test1";
+ }
+
+ // Issues userKeys for legacy clients; token clients get their identity from Keycloak.
+ @GET
+ @Path("register{slash: /?}")
+ @Produces(MediaType.TEXT_PLAIN)
+ public String register() {
+ if (!backend.legacyUserKeyAuth()) throw new NotFoundException();
+ return UUID.randomUUID().toString();
+ }
+
+ // A live game as players receive it, or the saved record of a finished game.
+ @GET
+ @Path("game/{id}")
+ public Response game(@PathParam("id") String id) {
+ // Serialised on the loop, because the client game refers to the live engine objects.
+ String live = backend.loop().call(() -> {
+ BackendGame game = backend.gameStore().get(id);
+ return game == null ? null : Json.write(backend.gameManager().toClientGame(game));
+ });
+ if (live != null) return json(live);
+ GameRecord record = backend.history().get(id);
+ if (record == null) throw new NotFoundException();
+ Map response = new LinkedHashMap<>();
+ response.put("status", "finished");
+ response.put("record", record);
+ return json(response);
+ }
+
+ // Live games if any are being played, otherwise the most spectated finished games. With kind=finished always the
+ // finished games, for clients that get live games over the WebSocket.
+ @GET
+ @Path("games/popular")
+ public Response popular(@QueryParam("limit") String limitParam, @QueryParam("kind") String kind) {
+ int limit = parseLimit(limitParam, 5);
+ List live = "finished".equals(kind) ? List.of()
+ : backend.loop().call(() -> backend.liveGames().getTop(limit));
+ Map response = new LinkedHashMap<>();
+ if (!live.isEmpty()) {
+ response.put("kind", "live");
+ response.put("games", live);
+ } else {
+ response.put("kind", "finished");
+ response.put("games", backend.history().getPopular(limit));
+ }
+ return json(response);
+ }
+
+ @GET
+ @Path("games/me/active")
+ public Response activeGame(@HeaderParam(HttpHeaders.AUTHORIZATION) String authorization) {
+ AuthUser user = BearerAuth.requireUser(backend.tokenVerifier(), authorization);
+ LiveGameSummary game = backend.loop().call(() -> {
+ BackendGame active = backend.gameStore().getGameByUser(user.id());
+ return active == null ? null : LiveGames.toLiveGameSummary(active);
+ });
+ Map response = new HashMap<>();
+ response.put("game", game);
+ return json(response);
+ }
+
+ @GET
+ @Path("games/me")
+ public Response myGames(@HeaderParam(HttpHeaders.AUTHORIZATION) String authorization,
+ @QueryParam("limit") String limit) {
+ AuthUser user = BearerAuth.requireUser(backend.tokenVerifier(), authorization);
+ return json(Map.of("games", backend.history().getRecentForUser(user.id(), parseLimit(limit, 3))));
+ }
+
+ private static Response json(Object body) {
+ return Response.ok(body instanceof String text ? text : Json.write(body), MediaType.APPLICATION_JSON).build();
+ }
+
+ // Like parseInt in the Node backend: the leading integer, clamped to 1..MAX_LIST_LIMIT.
+ static int parseLimit(String value, int fallback) {
+ Matcher match = LEADING_INTEGER.matcher(value == null ? "" : value);
+ if (!match.find()) return fallback;
+ BigInteger limit = new BigInteger(match.group(1));
+ return limit.max(BigInteger.ONE).min(BigInteger.valueOf(GameHistoryService.MAX_LIST_LIMIT)).intValue();
+ }
+}
diff --git a/apps/backend-quarkus/src/main/java/app/playmancala/rtmt/RtmtEndpoint.java b/apps/backend-quarkus/src/main/java/app/playmancala/rtmt/RtmtEndpoint.java
new file mode 100644
index 0000000..de16978
--- /dev/null
+++ b/apps/backend-quarkus/src/main/java/app/playmancala/rtmt/RtmtEndpoint.java
@@ -0,0 +1,93 @@
+package app.playmancala.rtmt;
+
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+
+import org.jboss.logging.Logger;
+
+import app.playmancala.Backend;
+import io.quarkus.websockets.next.CloseReason;
+import io.quarkus.websockets.next.OnClose;
+import io.quarkus.websockets.next.OnError;
+import io.quarkus.websockets.next.OnOpen;
+import io.quarkus.websockets.next.OnTextMessage;
+import io.quarkus.websockets.next.WebSocket;
+import io.quarkus.websockets.next.WebSocketConnection;
+import io.smallrye.common.annotation.NonBlocking;
+import jakarta.inject.Inject;
+
+// Clients connect to the root path, as they did to the Node backend. Events run on the connection's event loop, in
+// order, and are only handed to the game loop, where RtmtServer handles them.
+@WebSocket(path = "/")
+public class RtmtEndpoint {
+ private static final Logger LOG = Logger.getLogger(RtmtEndpoint.class);
+
+ @Inject
+ Backend backend;
+
+ @OnOpen
+ @NonBlocking
+ public void onOpen(WebSocketConnection connection) {
+ RtmtSocket socket = new ConnectionSocket(connection);
+ backend.loop().execute(() -> backend.rtmt().onOpen(socket));
+ }
+
+ @OnTextMessage
+ @NonBlocking
+ public void onTextMessage(WebSocketConnection connection, String text) {
+ RtmtSocket socket = new ConnectionSocket(connection);
+ backend.loop().execute(() -> backend.rtmt().onText(socket, text));
+ }
+
+ @OnClose
+ @NonBlocking
+ public void onClose(WebSocketConnection connection) {
+ RtmtSocket socket = new ConnectionSocket(connection);
+ backend.loop().execute(() -> backend.rtmt().onClose(socket));
+ }
+
+ @OnError
+ @NonBlocking
+ public void onError(WebSocketConnection connection, Throwable error) {
+ // A close event follows, which cleans up.
+ LOG.debugf("(RTMT) WebSocket error: %s", error.getMessage());
+ }
+
+ private record ConnectionSocket(WebSocketConnection connection) implements RtmtSocket {
+ @Override
+ public String id() {
+ return connection.id();
+ }
+
+ @Override
+ public String queryParam(String name) {
+ String query = connection.handshakeRequest().query();
+ if (query == null || query.isEmpty()) return null;
+ for (String pair : query.split("&")) {
+ int equals = pair.indexOf('=');
+ String key = equals < 0 ? pair : pair.substring(0, equals);
+ if (URLDecoder.decode(key, StandardCharsets.UTF_8).equals(name)) {
+ return equals < 0 ? "" : URLDecoder.decode(pair.substring(equals + 1), StandardCharsets.UTF_8);
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public boolean isOpen() {
+ return connection.isOpen();
+ }
+
+ @Override
+ public void send(String text) {
+ connection.sendText(text).subscribe().with(ok -> { },
+ error -> LOG.debugf("(RTMT) Could not send: %s", error.getMessage()));
+ }
+
+ @Override
+ public void close(int code, String reason) {
+ connection.close(new CloseReason(code, reason)).subscribe().with(ok -> { },
+ error -> LOG.debugf("(RTMT) Could not close: %s", error.getMessage()));
+ }
+ }
+}
diff --git a/apps/backend-quarkus/src/main/java/app/playmancala/rtmt/RtmtServer.java b/apps/backend-quarkus/src/main/java/app/playmancala/rtmt/RtmtServer.java
new file mode 100644
index 0000000..50eaa73
--- /dev/null
+++ b/apps/backend-quarkus/src/main/java/app/playmancala/rtmt/RtmtServer.java
@@ -0,0 +1,256 @@
+package app.playmancala.rtmt;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Executor;
+import java.util.function.BiConsumer;
+import java.util.regex.Pattern;
+
+import org.jboss.logging.Logger;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.NullNode;
+
+import app.playmancala.auth.InvalidTokenException;
+import app.playmancala.auth.TokenVerifier;
+import app.playmancala.loop.Cancellable;
+import app.playmancala.loop.Scheduler;
+import app.playmancala.model.AuthUser;
+import app.playmancala.model.Channels;
+import app.playmancala.model.Json;
+
+// The WebSocket protocol: authentication, routing messages by channel, users with several sockets and the ping
+// keep-alive. Messages are JSON objects { channel, message }. Everything here runs on the game loop; the endpoint
+// forwards socket events to it, and tokens are checked on `verifier` and the outcome handled back on the loop.
+public class RtmtServer implements Rtmt {
+ public static final long PING_INTERVAL_MS = 2_000;
+ public static final long DEFAULT_AUTH_TIMEOUT_MS = 5_000;
+ private static final Logger LOG = Logger.getLogger(RtmtServer.class);
+ private static final Pattern LEGACY_USER_KEY_PATTERN =
+ Pattern.compile("^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", Pattern.CASE_INSENSITIVE);
+
+ private record Envelope(String channel, Object message) {
+ }
+
+ private static final class Client {
+ final RtmtSocket socket;
+ boolean alive = true;
+ // Set once the socket is authenticated.
+ String userId;
+ Cancellable authTimeout;
+
+ Client(RtmtSocket socket) {
+ this.socket = socket;
+ }
+ }
+
+ private final TokenVerifier tokenVerifier;
+ private final boolean legacyUserKeyAuth;
+ private final Scheduler loop;
+ private final Executor verifier;
+ private final long authTimeoutMs;
+
+ private final Map> messageChannels = new HashMap<>();
+ // By socket id.
+ private final Map clients = new HashMap<>();
+ // A user can be connected from several devices or tabs at once.
+ private final Map> sockets = new HashMap<>();
+ private final Map users = new HashMap<>();
+ private final List> connectionChangeListeners = new ArrayList<>();
+ private Cancellable pingTimer;
+
+ public RtmtServer(TokenVerifier tokenVerifier, boolean legacyUserKeyAuth, Scheduler loop, Executor verifier,
+ long authTimeoutMs) {
+ this.tokenVerifier = tokenVerifier;
+ this.legacyUserKeyAuth = legacyUserKeyAuth;
+ this.loop = loop;
+ this.verifier = verifier;
+ this.authTimeoutMs = authTimeoutMs;
+ }
+
+ public void start() {
+ stop();
+ pingTimer = loop.scheduleAtFixedRate(this::ping, PING_INTERVAL_MS);
+ }
+
+ public void stop() {
+ if (pingTimer != null) pingTimer.cancel();
+ pingTimer = null;
+ }
+
+ public void onOpen(RtmtSocket socket) {
+ LOG.debug("(RTMT) Client connected");
+ Client client = new Client(socket);
+ clients.put(socket.id(), client);
+
+ String legacyUserKey = socket.queryParam("userKey");
+ if (legacyUserKey != null) {
+ if (legacyUserKeyAuth && LEGACY_USER_KEY_PATTERN.matcher(legacyUserKey).matches()) {
+ bindSocket(client, new AuthUser(legacyUserKey, null, true));
+ } else {
+ rejectSocket(client, "userKey login is not accepted");
+ }
+ return;
+ }
+
+ client.authTimeout = loop.schedule(() -> {
+ if (client.userId == null) rejectSocket(client, "authentication timeout");
+ }, authTimeoutMs);
+ }
+
+ public void onClose(RtmtSocket socket) {
+ Client client = clients.remove(socket.id());
+ if (client == null) return;
+ LOG.debug("(RTMT) Client disconnected");
+ if (client.authTimeout != null) client.authTimeout.cancel();
+ unbindSocket(client);
+ }
+
+ public void onText(RtmtSocket socket, String text) {
+ Client client = clients.get(socket.id());
+ if (client == null) return;
+ JsonNode envelope;
+ try {
+ envelope = Json.parse(text);
+ } catch (Exception e) {
+ LOG.debug("(RTMT) Ignoring malformed message");
+ return;
+ }
+ if (envelope == null || !envelope.path("channel").isTextual()) return;
+ String channel = envelope.get("channel").asText();
+ JsonNode message = envelope.has("message") ? envelope.get("message") : NullNode.getInstance();
+
+ if (channel.equals(Channels.PONG)) {
+ client.alive = true;
+ return;
+ }
+ if (channel.equals(Channels.AUTH)) {
+ authenticate(client, message);
+ return;
+ }
+ if (client.userId == null) {
+ LOG.debugf("(RTMT) Ignoring message from unauthenticated client on channel %s", channel);
+ return;
+ }
+ BiConsumer callback = messageChannels.get(channel);
+ if (callback != null) {
+ callback.accept(client.userId, message);
+ } else {
+ LOG.debugf("(RTMT) Channel callback not found! %s", channel);
+ }
+ }
+
+ private void authenticate(Client client, JsonNode message) {
+ JsonNode token = message.path("token");
+ if (!token.isTextual()) {
+ rejectSocket(client, "missing token");
+ return;
+ }
+ verifier.execute(() -> {
+ try {
+ AuthUser user = tokenVerifier.verify(token.asText());
+ loop.execute(() -> onAuthenticated(client, user));
+ } catch (InvalidTokenException e) {
+ loop.execute(() -> rejectSocket(client, "invalid token"));
+ } catch (RuntimeException e) {
+ LOG.error("(RTMT) Token verification failed", e);
+ loop.execute(() -> rejectSocket(client, "authentication failed"));
+ }
+ });
+ }
+
+ private void onAuthenticated(Client client, AuthUser user) {
+ if (!client.socket.isOpen() || clients.get(client.socket.id()) != client) return;
+ if (client.userId != null && !client.userId.equals(user.id())) {
+ rejectSocket(client, "token belongs to another user");
+ return;
+ }
+ if (client.authTimeout != null) client.authTimeout.cancel();
+ if (client.userId != null) {
+ // Token refresh on an already authenticated socket.
+ users.put(user.id(), user);
+ } else {
+ bindSocket(client, user);
+ }
+ send(client, Channels.ON_AUTH, Map.of("user", user));
+ }
+
+ private void rejectSocket(Client client, String reason) {
+ if (!client.socket.isOpen()) return;
+ send(client, Channels.ON_AUTH_ERROR, Map.of("reason", reason));
+ client.socket.close(Channels.CLOSE_AUTH_FAILED, reason);
+ }
+
+ private void bindSocket(Client client, AuthUser user) {
+ client.userId = user.id();
+ users.put(user.id(), user);
+ sockets.computeIfAbsent(user.id(), id -> new LinkedHashSet<>()).add(client);
+ // Fired for every new socket so each device gets the current game state.
+ List.copyOf(connectionChangeListeners).forEach(listener -> listener.accept(user.id(), true));
+ }
+
+ private void unbindSocket(Client client) {
+ String userId = client.userId;
+ if (userId == null) return;
+ Set userSockets = sockets.get(userId);
+ if (userSockets == null || !userSockets.remove(client)) return;
+ if (userSockets.isEmpty()) {
+ sockets.remove(userId);
+ users.remove(userId);
+ List.copyOf(connectionChangeListeners).forEach(listener -> listener.accept(userId, false));
+ }
+ }
+
+ private void ping() {
+ for (Client client : List.copyOf(clients.values())) {
+ if (!client.alive) {
+ LOG.debug("(RTMT) Closing a socket that did not answer the ping");
+ client.socket.close(1001, "ping timeout");
+ continue;
+ }
+ client.alive = false;
+ send(client, Channels.PING, Map.of());
+ }
+ }
+
+ @Override
+ public void sendMessage(String clientId, String channel, Object message) {
+ Set userSockets = sockets.get(clientId);
+ if (userSockets == null) {
+ LOG.debugf("(RTMT) Client not connected: %s", clientId);
+ return;
+ }
+ String text = Json.write(new Envelope(channel, message));
+ userSockets.forEach(client -> {
+ if (client.socket.isOpen()) client.socket.send(text);
+ });
+ }
+
+ private void send(Client client, String channel, Object message) {
+ if (client.socket.isOpen()) client.socket.send(Json.write(new Envelope(channel, message)));
+ }
+
+ @Override
+ public void listenMessage(String channel, BiConsumer callback) {
+ messageChannels.put(channel, callback);
+ }
+
+ @Override
+ public boolean isClientOnline(String clientId) {
+ return sockets.containsKey(clientId);
+ }
+
+ @Override
+ public AuthUser getUser(String clientId) {
+ return users.get(clientId);
+ }
+
+ @Override
+ public void listenOnClientConnectionChange(BiConsumer listener) {
+ connectionChangeListeners.add(listener);
+ }
+}
diff --git a/apps/backend-quarkus/src/main/java/app/playmancala/rtmt/RtmtSocket.java b/apps/backend-quarkus/src/main/java/app/playmancala/rtmt/RtmtSocket.java
new file mode 100644
index 0000000..f8af2a2
--- /dev/null
+++ b/apps/backend-quarkus/src/main/java/app/playmancala/rtmt/RtmtSocket.java
@@ -0,0 +1,15 @@
+package app.playmancala.rtmt;
+
+// A client's WebSocket as RtmtServer sees it. Sends and closes do not block.
+public interface RtmtSocket {
+ String id();
+
+ // A query parameter of the URL the client connected to, or null.
+ String queryParam(String name);
+
+ boolean isOpen();
+
+ void send(String text);
+
+ void close(int code, String reason);
+}
diff --git a/apps/backend-quarkus/src/main/resources/application.properties b/apps/backend-quarkus/src/main/resources/application.properties
index 4875640..b55fd58 100644
--- a/apps/backend-quarkus/src/main/resources/application.properties
+++ b/apps/backend-quarkus/src/main/resources/application.properties
@@ -35,7 +35,7 @@ quarkus.datasource.devservices.image-name=postgres:17.6
# Valkey. Tests start one with Dev Services. Short timeouts, so callers fall back quickly while it is down.
%dev,prod.quarkus.redis.hosts=${VALKEY_URL:redis://localhost:6379}
quarkus.redis.timeout=2S
-quarkus.redis.tcp.connect-timeout=1S
+quarkus.redis.tcp.connection-timeout=1S
quarkus.redis.devservices.image-name=valkey/valkey:9.1.2
# Nothing listens on port 1; for the tests of an unreachable Valkey.
%test.quarkus.redis.offline.hosts=redis://127.0.0.1:1
diff --git a/apps/backend-quarkus/src/test/java/app/playmancala/fakes/InMemoryCache.java b/apps/backend-quarkus/src/test/java/app/playmancala/fakes/InMemoryCache.java
new file mode 100644
index 0000000..f5a8148
--- /dev/null
+++ b/apps/backend-quarkus/src/test/java/app/playmancala/fakes/InMemoryCache.java
@@ -0,0 +1,35 @@
+package app.playmancala.fakes;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+
+import app.playmancala.live.Cache;
+import app.playmancala.model.Json;
+
+public class InMemoryCache implements Cache {
+ public final Map values = new HashMap<>();
+
+ @Override
+ public Optional get(String key, TypeReference type) {
+ String value = values.get(key);
+ if (value == null) return Optional.empty();
+ try {
+ return Optional.of(Json.MAPPER.readValue(value, type));
+ } catch (Exception e) {
+ return Optional.empty();
+ }
+ }
+
+ @Override
+ public void set(String key, Object value, long ttlSeconds) {
+ values.put(key, Json.write(value));
+ }
+
+ @Override
+ public void delete(String... keys) {
+ for (String key : keys) values.remove(key);
+ }
+}
diff --git a/apps/backend-quarkus/src/test/java/app/playmancala/history/GameHistoryServiceTest.java b/apps/backend-quarkus/src/test/java/app/playmancala/history/GameHistoryServiceTest.java
new file mode 100644
index 0000000..cbc75f7
--- /dev/null
+++ b/apps/backend-quarkus/src/test/java/app/playmancala/history/GameHistoryServiceTest.java
@@ -0,0 +1,102 @@
+package app.playmancala.history;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import app.playmancala.fakes.InMemoryCache;
+import app.playmancala.fakes.InMemoryGameHistoryRepository;
+import app.playmancala.model.EndReason;
+import app.playmancala.model.FinishedGameSummary;
+import app.playmancala.model.GameRecord;
+import app.playmancala.model.GameResult;
+import app.playmancala.model.PlayerSummary;
+
+// The list behaviour the Node backend's HTTP API tests checked: the cache, unsaved games and a database that is down.
+class GameHistoryServiceTest {
+ static final PlayerSummary ALICE = new PlayerSummary("kc:alice", "alice", false);
+ static final PlayerSummary BOB = new PlayerSummary("kc:bob", "bob", false);
+ static final PlayerSummary CAROL = new PlayerSummary("kc:carol", "carol", false);
+
+ InMemoryGameHistoryRepository repository;
+ InMemoryCache cache;
+ List unsaved;
+ GameHistoryService history;
+ long endedAt = 1_000_000;
+
+ @BeforeEach
+ void setUp() {
+ repository = new InMemoryGameHistoryRepository();
+ cache = new InMemoryCache();
+ unsaved = new ArrayList<>();
+ history = new GameHistoryService(repository, cache, () -> List.copyOf(unsaved));
+ }
+
+ GameRecord record(PlayerSummary player1, PlayerSummary player2, int peakSpectators) {
+ endedAt += 60_000;
+ return new GameRecord(UUID.randomUUID().toString(), player1, player2, List.of(), GameResult.PLAYER1,
+ EndReason.LEFT, null, endedAt - 1000, endedAt, 0, peakSpectators, 300_000L, List.of(), peakSpectators);
+ }
+
+ static List ids(List games) {
+ return games.stream().map(FinishedGameSummary::id).toList();
+ }
+
+ @Test
+ void returnsTheUsersLatestGamesAndShowsANewOneRightAwayDespiteTheCache() {
+ GameRecord first = record(ALICE, BOB, 0);
+ repository.save(first);
+ repository.save(record(CAROL, BOB, 0));
+ assertEquals(List.of(first.id()), ids(history.getRecentForUser(ALICE.userId(), 3)));
+
+ GameRecord second = record(BOB, ALICE, 0);
+ repository.save(second);
+ // Still cached.
+ assertEquals(List.of(first.id()), ids(history.getRecentForUser(ALICE.userId(), 3)));
+ history.onGameSaved(second);
+ assertEquals(List.of(second.id()), ids(history.getRecentForUser(ALICE.userId(), 1)));
+ }
+
+ @Test
+ void returnsAnEmptyListWhileTheDatabaseIsDown() {
+ repository.failing = true;
+ assertEquals(List.of(), history.getRecentForUser(ALICE.userId(), 3));
+ assertEquals(List.of(), history.getPopular(5));
+ }
+
+ @Test
+ void includesGamesThatEndedButAreNotSavedYet() {
+ repository.failing = true;
+ GameRecord game = record(ALICE, BOB, 0);
+ unsaved.add(game);
+ assertEquals(List.of(game.id()), ids(history.getRecentForUser(ALICE.userId(), 3)));
+ assertEquals(List.of(), ids(history.getRecentForUser(CAROL.userId(), 3)));
+ assertEquals(List.of(game.id()), ids(history.getPopular(5)));
+ assertSame(game, history.get(game.id()));
+ }
+
+ @Test
+ void listsPopularGamesBySpectatorsThenNewestWithoutDuplicates() {
+ GameRecord quiet = record(ALICE, BOB, 0);
+ GameRecord watched = record(ALICE, BOB, 3);
+ GameRecord newerQuiet = record(ALICE, BOB, 0);
+ List.of(quiet, watched).forEach(repository::save);
+ unsaved.add(newerQuiet);
+ unsaved.add(watched); // saved meanwhile
+ assertEquals(List.of(watched.id(), newerQuiet.id(), quiet.id()), ids(history.getPopular(5)));
+ assertEquals(List.of(watched.id()), ids(history.getPopular(1)));
+ }
+
+ @Test
+ void readingASavedGameFailsWhileTheDatabaseIsDown() {
+ repository.failing = true;
+ assertThrows(IllegalStateException.class, () -> history.get(UUID.randomUUID().toString()));
+ }
+}
diff --git a/apps/backend-quarkus/src/test/java/app/playmancala/http/ApiTest.java b/apps/backend-quarkus/src/test/java/app/playmancala/http/ApiTest.java
new file mode 100644
index 0000000..8bb82bf
--- /dev/null
+++ b/apps/backend-quarkus/src/test/java/app/playmancala/http/ApiTest.java
@@ -0,0 +1,221 @@
+package app.playmancala.http;
+
+import static io.restassured.RestAssured.given;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.Matchers.matchesPattern;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.net.URI;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import app.playmancala.game.PublicIds;
+import app.playmancala.model.Channels;
+import app.playmancala.model.Json;
+import app.playmancala.support.TestClient;
+import io.quarkus.test.common.http.TestHTTPResource;
+import io.quarkus.test.junit.QuarkusTest;
+
+// The WebSocket and the HTTP API end to end, with Postgres and Valkey in Docker and a fake token verifier.
+// Ported from the Node backend's tests/server/ExpressApp.test.ts and tests/rtmt/rtmt_websocket.test.ts.
+@QuarkusTest
+class ApiTest {
+ @TestHTTPResource("/")
+ URI base;
+
+ URI webSocketUri(String query) {
+ return URI.create(base.toString().replaceFirst("^http", "ws") + query);
+ }
+
+ static String unique(String name) {
+ return "kc:" + name + "-" + UUID.randomUUID().toString().substring(0, 8);
+ }
+
+ static JsonNode get(String path, String userId) {
+ var request = given();
+ if (userId != null) request = request.header("Authorization", "Bearer valid:" + userId);
+ String body = request.when().get(path).then().statusCode(200).extract().asString();
+ try {
+ return Json.parse(body);
+ } catch (Exception e) {
+ throw new IllegalStateException(body, e);
+ }
+ }
+
+ static List ids(JsonNode games) {
+ return games.valueStream().map(game -> game.get("id").asText()).toList();
+ }
+
+ // Both ask for a game and get paired; returns the on_game_start message.
+ static JsonNode startGame(TestClient first, TestClient second) {
+ first.send(Channels.NEW_GAME, Map.of());
+ second.send(Channels.NEW_GAME, Map.of());
+ JsonNode start = first.waitFor(Channels.ON_GAME_START);
+ assertEquals(start.get("id"), second.waitFor(Channels.ON_GAME_START).get("id"));
+ return start;
+ }
+
+ @Test
+ void answersOnTheRootPathAndIssuesLegacyUserKeys() {
+ given().when().get("/").then().statusCode(200);
+ String uuid = "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}";
+ given().when().get("/register/").then().statusCode(200).body(matchesPattern(uuid));
+ given().when().get("/register").then().statusCode(200).body(matchesPattern(uuid));
+ }
+
+ @Test
+ void allowsEveryOrigin() {
+ given().header("Origin", "https://playmancala.app").when().get("/games/popular").then().statusCode(200)
+ .header("Access-Control-Allow-Origin", matchesPattern("\\*|https://playmancala\\.app"));
+ }
+
+ @Test
+ void aGameFromStartToTheSavedRecord() {
+ String alice = unique("alice");
+ String bob = unique("bob");
+ String carol = unique("carol");
+ try (TestClient aliceClient = TestClient.connectAs(webSocketUri(""), alice);
+ TestClient bobClient = TestClient.connectAs(webSocketUri(""), bob);
+ TestClient carolClient = TestClient.connectAs(webSocketUri(""), carol)) {
+ JsonNode start = startGame(aliceClient, bobClient);
+ String gameId = start.get("id").asText();
+ assertEquals("initial", start.at("/mancalaGame/state").asText());
+ assertEquals(300_000, start.at("/clock/initialMs").asLong());
+
+ // A spectator, seen by the API but without spectator ids.
+ carolClient.send(Channels.LISTEN_GAME_EVENTS, gameId);
+ await().atMost(Duration.ofSeconds(5)).until(() -> get("/game/" + gameId, null).get("spectatorCount").asInt() == 1);
+ JsonNode live = get("/game/" + gameId, null);
+ assertEquals(gameId, live.get("id").asText());
+ assertFalse(live.has("spectatorIds"));
+ assertTrue(ids(get("/games/popular?limit=20", null).get("games")).contains(gameId));
+ assertEquals(gameId, get("/games/me/active", alice).at("/game/id").asText());
+ assertTrue(get("/games/me/active", carol).get("game").isNull());
+
+ // A move reaches the players and the spectator.
+ String turnPlayerId = start.at("/mancalaGame/turnPlayerId").asText();
+ (turnPlayerId.equals(alice) ? aliceClient : bobClient).send(Channels.GAME_MOVE, Map.of("index", 0));
+ for (TestClient client : List.of(aliceClient, bobClient, carolClient)) {
+ assertEquals(1, client.waitFor(Channels.ON_GAME_UPDATE).at("/mancalaGame/history").size());
+ }
+ assertEquals(1, get("/game/" + gameId, null).at("/mancalaGame/history").size());
+
+ // Alice leaves: Bob wins, and the game is saved.
+ aliceClient.send(Channels.LEAVE_GAME, Map.of());
+ JsonNode end = carolClient.waitFor(Channels.ON_GAME_END);
+ assertEquals("left", end.get("endReason").asText());
+ assertEquals(bob, end.get("winnerId").asText());
+ assertEquals(alice, bobClient.waitFor(Channels.ON_GAME_USER_LEAVE).asText());
+
+ JsonNode finished = get("/game/" + gameId, null);
+ assertEquals("finished", finished.get("status").asText());
+ assertEquals(1, finished.at("/record/moves").size());
+ assertEquals(alice, finished.at("/record/leftUserId").asText());
+ assertTrue(get("/games/me/active", alice).get("game").isNull());
+ await().atMost(Duration.ofSeconds(10)).until(() -> ids(get("/games/me", alice).get("games")).contains(gameId));
+ assertEquals(List.of(gameId), ids(get("/games/me?limit=1", bob).get("games")));
+ assertFalse(get("/games/me", bob).get("games").get(0).has("moves"));
+ assertTrue(ids(get("/games/popular?kind=finished&limit=20", null).get("games")).contains(gameId));
+ }
+ }
+
+ @Test
+ void rejectsANewGameWhileOneIsInProgress() {
+ try (TestClient alice = TestClient.connectAs(webSocketUri(""), unique("alice"));
+ TestClient bob = TestClient.connectAs(webSocketUri(""), unique("bob"))) {
+ String gameId = startGame(alice, bob).get("id").asText();
+ alice.send(Channels.NEW_GAME, Map.of());
+ assertEquals(gameId, alice.waitFor(Channels.ON_NEW_GAME_REJECTED).get("gameId").asText());
+ alice.send(Channels.LEAVE_GAME, Map.of());
+ bob.waitFor(Channels.ON_GAME_END);
+ }
+ }
+
+ @Test
+ void aReconnectingPlayerGetsTheGameAgain() {
+ String alice = unique("alice");
+ try (TestClient aliceClient = TestClient.connectAs(webSocketUri(""), alice);
+ TestClient bob = TestClient.connectAs(webSocketUri(""), unique("bob"))) {
+ String gameId = startGame(aliceClient, bob).get("id").asText();
+ aliceClient.close();
+ JsonNode offline = bob.waitFor(Channels.ON_USER_CONNECTION_CHANGE, 2).getLast();
+ assertTrue(offline.toString().contains("\"isOnline\":false"), offline.toString());
+
+ try (TestClient again = TestClient.connectAs(webSocketUri(""), alice)) {
+ assertEquals(gameId, again.waitFor(Channels.ON_GAME_UPDATE).get("id").asText());
+ again.send(Channels.LEAVE_GAME, Map.of());
+ bob.waitFor(Channels.ON_GAME_END);
+ }
+ }
+ }
+
+ @Test
+ void sendsLiveGamePreviewsToSubscribers() {
+ try (TestClient viewer = TestClient.connectAs(webSocketUri(""), unique("viewer"));
+ TestClient alice = TestClient.connectAs(webSocketUri(""), unique("alice"));
+ TestClient bob = TestClient.connectAs(webSocketUri(""), unique("bob"))) {
+ viewer.send(Channels.SUBSCRIBE_LIVE_GAMES, Map.of());
+ viewer.waitFor(Channels.ON_LIVE_GAMES);
+ String gameId = startGame(alice, bob).get("id").asText();
+ await().atMost(Duration.ofSeconds(5)).until(() -> viewer.messages(Channels.ON_LIVE_GAMES).stream()
+ .anyMatch(push -> ids(push.get("games")).contains(gameId)));
+ alice.send(Channels.LEAVE_GAME, Map.of());
+ bob.waitFor(Channels.ON_GAME_END);
+ }
+ }
+
+ @Test
+ void requiresAValidTokenForTheUsersGames() {
+ given().when().get("/games/me/active").then().statusCode(401);
+ given().header("Authorization", "Bearer forged").when().get("/games/me/active").then().statusCode(401);
+ given().when().get("/games/me").then().statusCode(401);
+ }
+
+ @Test
+ void answers404ForUnknownGames() {
+ given().when().get("/game/unknown").then().statusCode(404);
+ given().when().get("/game/" + UUID.randomUUID()).then().statusCode(404);
+ }
+
+ @Test
+ void closesASocketWithAnInvalidToken() throws Exception {
+ TestClient client = new TestClient(webSocketUri(""));
+ client.send(Channels.AUTH, Map.of("token", "forged"));
+ assertEquals(Json.tree(Map.of("reason", "invalid token")), client.waitFor(Channels.ON_AUTH_ERROR));
+ assertEquals(Channels.CLOSE_AUTH_FAILED, client.closed.get(5, TimeUnit.SECONDS));
+ }
+
+ @Test
+ void playsWithLegacyUserKeyClients() {
+ String key1 = UUID.randomUUID().toString();
+ String key2 = UUID.randomUUID().toString();
+ try (TestClient player1 = new TestClient(webSocketUri("?userKey=" + key1));
+ TestClient player2 = new TestClient(webSocketUri("?userKey=" + key2))) {
+ String gameId = startGame(player1, player2).get("id").asText();
+ JsonNode preview = get("/games/popular?limit=20", null).get("games").valueStream()
+ .filter(game -> game.get("id").asText().equals(gameId)).findFirst().orElseThrow();
+ assertFalse(preview.toString().contains(key1));
+ assertTrue(preview.toString().contains(PublicIds.publicUserId(key1)));
+ player1.send(Channels.LEAVE_GAME, Map.of());
+ player2.waitFor(Channels.ON_GAME_END);
+ }
+ }
+
+ @Test
+ void parsesLimitsLikeParseInt() {
+ assertEquals(3, GameResource.parseLimit(null, 3));
+ assertEquals(3, GameResource.parseLimit("abc", 3));
+ assertEquals(7, GameResource.parseLimit("7abc", 3));
+ assertEquals(1, GameResource.parseLimit("-4", 3));
+ assertEquals(20, GameResource.parseLimit("99999999999999999999", 3));
+ }
+}
diff --git a/apps/backend-quarkus/src/test/java/app/playmancala/rtmt/RtmtServerTest.java b/apps/backend-quarkus/src/test/java/app/playmancala/rtmt/RtmtServerTest.java
new file mode 100644
index 0000000..0d3d29f
--- /dev/null
+++ b/apps/backend-quarkus/src/test/java/app/playmancala/rtmt/RtmtServerTest.java
@@ -0,0 +1,324 @@
+package app.playmancala.rtmt;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import app.playmancala.auth.InvalidTokenException;
+import app.playmancala.auth.TokenVerifier;
+import app.playmancala.loop.ManualScheduler;
+import app.playmancala.model.AuthUser;
+import app.playmancala.model.Channels;
+import app.playmancala.model.Json;
+
+// Ported from the Node backend's tests/rtmt/rtmt_websocket.test.ts, with fake sockets instead of a server.
+class RtmtServerTest {
+ static final String LEGACY_KEY = "0b9f6d1e-7c1a-4f7e-9a51-3f0c2b6d8e11";
+ static final long AUTH_TIMEOUT_MS = 300;
+
+ // Accepts tokens of the form "valid:".
+ static final TokenVerifier FAKE_VERIFIER = token -> {
+ if (!token.startsWith("valid:")) throw new InvalidTokenException("bad token");
+ String id = token.substring("valid:".length());
+ return new AuthUser(id, "name-of-" + id, false);
+ };
+
+ final ManualScheduler loop = new ManualScheduler(0);
+ // Token checks wait here until the test runs them, as they would on a worker thread.
+ final List pendingChecks = new ArrayList<>();
+ final List connectionChanges = new ArrayList<>();
+ RtmtServer server;
+ int socketCount;
+
+ class FakeSocket implements RtmtSocket {
+ final String id = "socket-" + ++socketCount;
+ final String query;
+ final List received = new ArrayList<>();
+ boolean open = true;
+ Integer closeCode;
+
+ FakeSocket(String query) {
+ this.query = query;
+ }
+
+ @Override
+ public String id() {
+ return id;
+ }
+
+ @Override
+ public String queryParam(String name) {
+ return query != null && query.startsWith(name + "=") ? query.substring(name.length() + 1) : null;
+ }
+
+ @Override
+ public boolean isOpen() {
+ return open;
+ }
+
+ @Override
+ public void send(String text) {
+ try {
+ received.add(Json.parse(text));
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ // Closing also delivers the close event, as the endpoint would.
+ @Override
+ public void close(int code, String reason) {
+ if (!open) return;
+ open = false;
+ closeCode = code;
+ server.onClose(this);
+ }
+
+ void sendText(String text) {
+ server.onText(this, text);
+ runChecks();
+ }
+
+ void send(String channel, Object message) {
+ sendText(Json.write(Map.of("channel", channel, "message", message)));
+ }
+
+ List messages(String channel) {
+ return received.stream().filter(m -> m.get("channel").asText().equals(channel)).map(m -> m.get("message")).toList();
+ }
+ }
+
+ void start(boolean legacyUserKeyAuth) {
+ server = new RtmtServer(FAKE_VERIFIER, legacyUserKeyAuth, loop, pendingChecks::add, AUTH_TIMEOUT_MS);
+ server.listenOnClientConnectionChange((id, isOnline) -> connectionChanges.add(id + (isOnline ? " online" : " offline")));
+ server.start();
+ }
+
+ void runChecks() {
+ List checks = List.copyOf(pendingChecks);
+ pendingChecks.clear();
+ checks.forEach(Runnable::run);
+ }
+
+ FakeSocket connect(String query) {
+ FakeSocket socket = new FakeSocket(query);
+ server.onOpen(socket);
+ return socket;
+ }
+
+ FakeSocket connectAs(String id) {
+ FakeSocket socket = connect(null);
+ socket.send(Channels.AUTH, Map.of("token", "valid:" + id));
+ assertEquals(1, socket.messages(Channels.ON_AUTH).size());
+ return socket;
+ }
+
+ void disconnect(FakeSocket socket) {
+ socket.open = false;
+ server.onClose(socket);
+ }
+
+ @Test
+ void authenticatesASocketAndRoutesItsMessagesAsThatUser() {
+ start(false);
+ List received = new ArrayList<>();
+ server.listenMessage("hello", (id, message) -> received.add(id + " " + message));
+
+ FakeSocket socket = connect(null);
+ socket.send(Channels.AUTH, Map.of("token", "valid:alice"));
+ assertEquals(List.of(Json.tree(Map.of("user", Map.of("id", "alice", "name", "name-of-alice", "isGuest", false)))),
+ socket.messages(Channels.ON_AUTH));
+
+ socket.send("hello", Map.of("n", 1));
+ assertEquals(List.of("alice {\"n\":1}"), received);
+ assertTrue(server.isClientOnline("alice"));
+ assertEquals("name-of-alice", server.getUser("alice").name());
+ assertEquals(List.of("alice online"), connectionChanges);
+ }
+
+ @Test
+ void sendsMessagesAsChannelAndMessage() {
+ start(false);
+ FakeSocket socket = connectAs("alice");
+ server.sendMessage("alice", "news", "text");
+ assertEquals(Json.tree(Map.of("channel", "news", "message", "text")), socket.received.getLast());
+ assertEquals(List.of("channel", "message"), socket.received.getLast().propertyStream().map(Map.Entry::getKey).toList());
+ }
+
+ @Test
+ void ignoresMessagesBeforeAuthentication() {
+ start(false);
+ List received = new ArrayList<>();
+ server.listenMessage("hello", (id, message) -> received.add(id));
+ connect(null).send("hello", Map.of());
+ assertEquals(List.of(), received);
+ }
+
+ @Test
+ void closesTheSocketOnAnInvalidToken() {
+ start(false);
+ FakeSocket socket = connect(null);
+ socket.send(Channels.AUTH, Map.of("token", "forged"));
+ assertEquals(List.of(Json.tree(Map.of("reason", "invalid token"))), socket.messages(Channels.ON_AUTH_ERROR));
+ assertEquals(Channels.CLOSE_AUTH_FAILED, socket.closeCode);
+ }
+
+ @Test
+ void closesTheSocketWhenATokenIsMissing() {
+ start(false);
+ FakeSocket socket = connect(null);
+ socket.send(Channels.AUTH, Map.of());
+ assertEquals(List.of(Json.tree(Map.of("reason", "missing token"))), socket.messages(Channels.ON_AUTH_ERROR));
+ assertEquals(Channels.CLOSE_AUTH_FAILED, socket.closeCode);
+ }
+
+ @Test
+ void closesTheSocketWhenNoTokenArrivesInTime() {
+ start(false);
+ FakeSocket socket = connect(null);
+ loop.advance(AUTH_TIMEOUT_MS - 1);
+ assertTrue(socket.open);
+ loop.advance(1);
+ assertEquals(List.of(Json.tree(Map.of("reason", "authentication timeout"))), socket.messages(Channels.ON_AUTH_ERROR));
+ assertEquals(Channels.CLOSE_AUTH_FAILED, socket.closeCode);
+ }
+
+ @Test
+ void keepsAnAuthenticatedSocketPastTheTimeout() {
+ start(false);
+ FakeSocket socket = connectAs("alice");
+ loop.advance(AUTH_TIMEOUT_MS * 2);
+ assertNull(socket.closeCode);
+ }
+
+ @Test
+ void acceptsARefreshedTokenForTheSameUserButNotForAnotherUser() {
+ start(false);
+ FakeSocket socket = connectAs("alice");
+ socket.send(Channels.AUTH, Map.of("token", "valid:alice"));
+ assertEquals(2, socket.messages(Channels.ON_AUTH).size());
+ assertEquals(List.of("alice online"), connectionChanges);
+
+ socket.send(Channels.AUTH, Map.of("token", "valid:mallory"));
+ assertEquals(List.of(Json.tree(Map.of("reason", "token belongs to another user"))),
+ socket.messages(Channels.ON_AUTH_ERROR));
+ assertEquals(Channels.CLOSE_AUTH_FAILED, socket.closeCode);
+ assertFalse(server.isClientOnline("alice"));
+ }
+
+ @Test
+ void ignoresATokenCheckThatEndsAfterTheSocketClosed() {
+ start(false);
+ FakeSocket socket = connect(null);
+ server.onText(socket, Json.write(Map.of("channel", Channels.AUTH, "message", Map.of("token", "valid:alice"))));
+ disconnect(socket);
+ runChecks();
+ assertFalse(server.isClientOnline("alice"));
+ assertEquals(List.of(), connectionChanges);
+ }
+
+ @Test
+ void rejectsUserKeyConnectionsWhenLegacyAuthIsOff() {
+ start(false);
+ FakeSocket socket = connect("userKey=" + LEGACY_KEY);
+ assertEquals(Channels.CLOSE_AUTH_FAILED, socket.closeCode);
+ }
+
+ @Test
+ void survivesMalformedMessages() {
+ start(false);
+ FakeSocket socket = connectAs("alice");
+ socket.sendText("{not json");
+ socket.sendText("null");
+ socket.sendText("[1]");
+ socket.sendText("{\"channel\":5}");
+ socket.send(Channels.AUTH, Map.of("token", "valid:alice"));
+ assertEquals(2, socket.messages(Channels.ON_AUTH).size());
+ }
+
+ @Test
+ void deliversMessagesToEverySocketOfTheUser() {
+ start(false);
+ FakeSocket phone = connectAs("alice");
+ FakeSocket laptop = connectAs("alice");
+ server.sendMessage("alice", "news", Map.of("n", 1));
+ assertEquals(List.of(Json.tree(Map.of("n", 1))), phone.messages("news"));
+ assertEquals(List.of(Json.tree(Map.of("n", 1))), laptop.messages("news"));
+ assertEquals(List.of("alice online", "alice online"), connectionChanges);
+ }
+
+ @Test
+ void keepsTheUserOnlineUntilTheLastSocketCloses() {
+ start(false);
+ FakeSocket phone = connectAs("alice");
+ FakeSocket laptop = connectAs("alice");
+
+ disconnect(phone);
+ assertTrue(server.isClientOnline("alice"));
+ assertFalse(connectionChanges.contains("alice offline"));
+
+ disconnect(laptop);
+ assertFalse(server.isClientOnline("alice"));
+ assertNull(server.getUser("alice"));
+ assertTrue(connectionChanges.contains("alice offline"));
+ }
+
+ @Test
+ void doesNotDropAReconnectedSocketWhenTheOldOneClosesLate() {
+ start(false);
+ FakeSocket oldSocket = connectAs("alice");
+ FakeSocket newSocket = connectAs("alice");
+ disconnect(oldSocket);
+ server.sendMessage("alice", "news", Map.of());
+ assertEquals(List.of(Json.tree(Map.of())), newSocket.messages("news"));
+ }
+
+ @Test
+ void pingsEverySocketAndClosesThoseThatDoNotAnswer() {
+ start(false);
+ FakeSocket answering = connectAs("alice");
+ FakeSocket silent = connectAs("bob");
+ loop.advance(RtmtServer.PING_INTERVAL_MS);
+ assertEquals(1, answering.messages(Channels.PING).size());
+ assertEquals(Json.tree(Map.of()), silent.messages(Channels.PING).getFirst());
+
+ answering.send(Channels.PONG, Map.of());
+ loop.advance(RtmtServer.PING_INTERVAL_MS);
+ assertNull(answering.closeCode);
+ assertEquals(1001, silent.closeCode);
+ assertFalse(server.isClientOnline("bob"));
+ }
+
+ @Test
+ void acceptsAUuidUserKey() {
+ start(true);
+ connect("userKey=" + LEGACY_KEY);
+ assertTrue(server.isClientOnline(LEGACY_KEY));
+ assertEquals(new AuthUser(LEGACY_KEY, null, true), server.getUser(LEGACY_KEY));
+ }
+
+ @Test
+ void aUserKeyCannotClaimAKeycloakUsersId() {
+ start(true);
+ FakeSocket socket = connect("userKey=kc:" + LEGACY_KEY);
+ assertEquals(Channels.CLOSE_AUTH_FAILED, socket.closeCode);
+ }
+
+ @Test
+ void aLegacySocketCannotSwitchToATokenIdentity() {
+ start(true);
+ FakeSocket socket = connect("userKey=" + LEGACY_KEY);
+ socket.send(Channels.AUTH, Map.of("token", "valid:kc:alice"));
+ assertEquals(Channels.CLOSE_AUTH_FAILED, socket.closeCode);
+ assertFalse(server.isClientOnline(LEGACY_KEY));
+ }
+}
diff --git a/apps/backend-quarkus/src/test/java/app/playmancala/support/FakeTokenVerifier.java b/apps/backend-quarkus/src/test/java/app/playmancala/support/FakeTokenVerifier.java
new file mode 100644
index 0000000..797682c
--- /dev/null
+++ b/apps/backend-quarkus/src/test/java/app/playmancala/support/FakeTokenVerifier.java
@@ -0,0 +1,21 @@
+package app.playmancala.support;
+
+import app.playmancala.auth.InvalidTokenException;
+import app.playmancala.auth.TokenVerifier;
+import app.playmancala.model.AuthUser;
+import jakarta.annotation.Priority;
+import jakarta.enterprise.inject.Alternative;
+import jakarta.inject.Singleton;
+
+// Replaces Keycloak in @QuarkusTests: accepts tokens of the form "valid:".
+@Alternative
+@Priority(1)
+@Singleton
+public class FakeTokenVerifier implements TokenVerifier {
+ @Override
+ public AuthUser verify(String token) throws InvalidTokenException {
+ if (!token.startsWith("valid:")) throw new InvalidTokenException("bad token");
+ String id = token.substring("valid:".length());
+ return new AuthUser(id, "name-of-" + id, false);
+ }
+}
diff --git a/apps/backend-quarkus/src/test/java/app/playmancala/support/TestClient.java b/apps/backend-quarkus/src/test/java/app/playmancala/support/TestClient.java
new file mode 100644
index 0000000..89c362a
--- /dev/null
+++ b/apps/backend-quarkus/src/test/java/app/playmancala/support/TestClient.java
@@ -0,0 +1,96 @@
+package app.playmancala.support;
+
+import static org.awaitility.Awaitility.await;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.WebSocket;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import app.playmancala.model.Channels;
+import app.playmancala.model.Json;
+
+// A WebSocket client speaking the game protocol; answers pings like the real clients.
+public class TestClient implements WebSocket.Listener, AutoCloseable {
+ public final List messages = new CopyOnWriteArrayList<>();
+ public final CompletableFuture closed = new CompletableFuture<>();
+ private final StringBuilder partial = new StringBuilder();
+ private final WebSocket webSocket;
+
+ public TestClient(URI uri) {
+ webSocket = HttpClient.newHttpClient().newWebSocketBuilder().buildAsync(uri, this).join();
+ }
+
+ // Connects and logs in as the user, with a token the FakeTokenVerifier accepts.
+ public static TestClient connectAs(URI uri, String userId) {
+ TestClient client = new TestClient(uri);
+ client.send(Channels.AUTH, Map.of("token", "valid:" + userId));
+ client.waitFor(Channels.ON_AUTH);
+ return client;
+ }
+
+ public void send(String channel, Object message) {
+ webSocket.sendText(Json.write(Map.of("channel", channel, "message", message)), true).join();
+ }
+
+ public List messages(String channel) {
+ return messages.stream().filter(m -> m.get("channel").asText().equals(channel)).map(m -> m.get("message")).toList();
+ }
+
+ // The first message on the channel, waiting up to 10 seconds for it.
+ public JsonNode waitFor(String channel) {
+ return waitFor(channel, 1).getFirst();
+ }
+
+ public List waitFor(String channel, int count) {
+ await().atMost(Duration.ofSeconds(10)).until(() -> messages(channel).size() >= count);
+ return messages(channel);
+ }
+
+ @Override
+ public void onOpen(WebSocket webSocket) {
+ webSocket.request(1);
+ }
+
+ @Override
+ public CompletionStage> onText(WebSocket webSocket, CharSequence data, boolean last) {
+ partial.append(data);
+ if (last) {
+ try {
+ JsonNode message = Json.parse(partial.toString());
+ if (message.get("channel").asText().equals(Channels.PING)) {
+ webSocket.sendText(Json.write(Map.of("channel", Channels.PONG, "message", Map.of())), true);
+ }
+ messages.add(message);
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ partial.setLength(0);
+ }
+ webSocket.request(1);
+ return null;
+ }
+
+ @Override
+ public CompletionStage> onClose(WebSocket webSocket, int statusCode, String reason) {
+ closed.complete(statusCode);
+ return null;
+ }
+
+ @Override
+ public void onError(WebSocket webSocket, Throwable error) {
+ closed.completeExceptionally(error);
+ }
+
+ @Override
+ public void close() {
+ if (!webSocket.isOutputClosed()) webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "").join();
+ }
+}
diff --git a/docs/specs/quarkus-backend/tasks.md b/docs/specs/quarkus-backend/tasks.md
index 653c7ec..d30c6a5 100644
--- a/docs/specs/quarkus-backend/tasks.md
+++ b/docs/specs/quarkus-backend/tasks.md
@@ -12,7 +12,7 @@ Work through the tasks in order; each ends with its tests passing and a commit.
unit tests.
- [x] 5. **Storage**: migrations and their runner, `PostgresGameHistoryRepository`, `GameHistoryService`,
`ValkeyLiveGameStore`, `ValkeyCache`; integration tests with Dev Services.
-- [ ] 6. **WebSocket and HTTP**: `RtmtServer` with unit tests, `RtmtEndpoint`, the HTTP resources, `Backend`
+- [x] 6. **WebSocket and HTTP**: `RtmtServer` with unit tests, `RtmtEndpoint`, the HTTP resources, `Backend`
wiring and startup, configuration; `@QuarkusTest`s for the API and the WebSocket.
- [ ] 7. **Replace the Node backend**: move the project to `apps/backend`, Dockerfile, `.dockerignore`, compose,
remove the Node sources; build the image.