From 0c112eb71606d0743a031a39e6b1b87a7c71e31a Mon Sep 17 00:00:00 2001 From: Halit Aksoy Date: Fri, 16 Oct 2026 11:33:07 +0300 Subject: [PATCH] feat(backend): serve the WebSocket protocol and the HTTP API RtmtServer ports the WebSocket protocol (auth handshake, sockets per user, ping) and runs on the game loop; the websockets-next endpoint at / only forwards socket events. The HTTP routes, CORS and the access log match the Express app. Backend wires everything and restores live games before handling socket events. End-to-end tests play games over real sockets against Postgres and Valkey. --- apps/backend-quarkus/pom.xml | 6 +- .../main/java/app/playmancala/Backend.java | 186 ++++++++++ .../auth/TokenVerifierProducer.java | 23 ++ .../java/app/playmancala/http/BearerAuth.java | 35 ++ .../app/playmancala/http/ErrorMapper.java | 20 ++ .../app/playmancala/http/GameResource.java | 125 +++++++ .../app/playmancala/rtmt/RtmtEndpoint.java | 93 +++++ .../java/app/playmancala/rtmt/RtmtServer.java | 256 ++++++++++++++ .../java/app/playmancala/rtmt/RtmtSocket.java | 15 + .../src/main/resources/application.properties | 2 +- .../app/playmancala/fakes/InMemoryCache.java | 35 ++ .../history/GameHistoryServiceTest.java | 102 ++++++ .../java/app/playmancala/http/ApiTest.java | 221 ++++++++++++ .../app/playmancala/rtmt/RtmtServerTest.java | 324 ++++++++++++++++++ .../support/FakeTokenVerifier.java | 21 ++ .../app/playmancala/support/TestClient.java | 96 ++++++ docs/specs/quarkus-backend/tasks.md | 2 +- 17 files changed, 1559 insertions(+), 3 deletions(-) create mode 100644 apps/backend-quarkus/src/main/java/app/playmancala/Backend.java create mode 100644 apps/backend-quarkus/src/main/java/app/playmancala/auth/TokenVerifierProducer.java create mode 100644 apps/backend-quarkus/src/main/java/app/playmancala/http/BearerAuth.java create mode 100644 apps/backend-quarkus/src/main/java/app/playmancala/http/ErrorMapper.java create mode 100644 apps/backend-quarkus/src/main/java/app/playmancala/http/GameResource.java create mode 100644 apps/backend-quarkus/src/main/java/app/playmancala/rtmt/RtmtEndpoint.java create mode 100644 apps/backend-quarkus/src/main/java/app/playmancala/rtmt/RtmtServer.java create mode 100644 apps/backend-quarkus/src/main/java/app/playmancala/rtmt/RtmtSocket.java create mode 100644 apps/backend-quarkus/src/test/java/app/playmancala/fakes/InMemoryCache.java create mode 100644 apps/backend-quarkus/src/test/java/app/playmancala/history/GameHistoryServiceTest.java create mode 100644 apps/backend-quarkus/src/test/java/app/playmancala/http/ApiTest.java create mode 100644 apps/backend-quarkus/src/test/java/app/playmancala/rtmt/RtmtServerTest.java create mode 100644 apps/backend-quarkus/src/test/java/app/playmancala/support/FakeTokenVerifier.java create mode 100644 apps/backend-quarkus/src/test/java/app/playmancala/support/TestClient.java 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.