diff --git a/apps/backend-quarkus/src/main/java/app/playmancala/history/GameHistoryService.java b/apps/backend-quarkus/src/main/java/app/playmancala/history/GameHistoryService.java new file mode 100644 index 0000000..645d0e6 --- /dev/null +++ b/apps/backend-quarkus/src/main/java/app/playmancala/history/GameHistoryService.java @@ -0,0 +1,91 @@ +package app.playmancala.history; + +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.function.Supplier; +import java.util.stream.Collectors; + +import org.jboss.logging.Logger; + +import com.fasterxml.jackson.core.type.TypeReference; + +import app.playmancala.live.Cache; +import app.playmancala.model.FinishedGameSummary; +import app.playmancala.model.GameRecord; + +// Reads finished games for the API, including ones not saved yet. Lists are cached briefly and come back without the +// saved games while the database is unreachable, so the home page still loads. Blocks; call it off the game loop. +public class GameHistoryService { + public static final int MAX_LIST_LIMIT = 20; + private static final Logger LOG = Logger.getLogger(GameHistoryService.class); + private static final long CACHE_TTL_SECONDS = 10; + private static final String POPULAR_KEY = "popular-finished"; + private static final TypeReference> SUMMARIES = new TypeReference<>() { }; + + private static final Comparator BY_NEWEST = + Comparator.comparingLong(FinishedGameSummary::endedAt).reversed(); + private static final Comparator BY_POPULARITY = + Comparator.comparingInt(FinishedGameSummary::peakSpectators).reversed().thenComparing(BY_NEWEST); + + private final GameHistoryRepository repository; + private final Cache cache; + // Games that ended but are not in the database yet (being saved, or waiting for a retry). + private final Supplier> unsaved; + + public GameHistoryService(GameHistoryRepository repository, Cache cache, Supplier> unsaved) { + this.repository = repository; + this.cache = cache; + this.unsaved = unsaved; + } + + private static String recentKey(String userId) { + return "recent:" + userId; + } + + // Null if there is no such game. Throws if the database cannot be asked. + public GameRecord get(String id) { + return unsaved.get().stream().filter(record -> record.id().equals(id)).findFirst() + .orElseGet(() -> repository.get(id)); + } + + public List getRecentForUser(String userId, int limit) { + List saved = cached(recentKey(userId), () -> repository.getRecentForUser(userId, MAX_LIST_LIMIT)); + List unsavedGames = unsaved.get().stream().filter(record -> record.hasPlayer(userId)).toList(); + return merge(saved, unsavedGames, BY_NEWEST, limit); + } + + public List getPopular(int limit) { + List saved = cached(POPULAR_KEY, () -> repository.getPopular(MAX_LIST_LIMIT)); + return merge(saved, unsaved.get(), BY_POPULARITY, limit); + } + + // Call when a game has been saved, so lists show it right away. + public void onGameSaved(GameRecord record) { + cache.delete(POPULAR_KEY, recentKey(record.player1().userId()), recentKey(record.player2().userId())); + } + + private List cached(String key, Supplier> load) { + Optional> hit = cache.get(key, SUMMARIES); + if (hit.isPresent()) return hit.get(); + try { + List games = load.get(); + cache.set(key, games, CACHE_TTL_SECONDS); + return games; + } catch (RuntimeException e) { + LOG.errorf("(History) Could not load games: %s", e.getMessage()); + return List.of(); + } + } + + private static List merge(List saved, List unsaved, + Comparator order, int limit) { + Set savedIds = saved.stream().map(FinishedGameSummary::id).collect(Collectors.toSet()); + List all = new ArrayList<>(saved); + unsaved.stream().filter(record -> !savedIds.contains(record.id())).map(GameRecord::toSummary).forEach(all::add); + all.sort(order); + return all.stream().limit(limit).toList(); + } +} diff --git a/apps/backend-quarkus/src/main/java/app/playmancala/history/Migrations.java b/apps/backend-quarkus/src/main/java/app/playmancala/history/Migrations.java new file mode 100644 index 0000000..136b54c --- /dev/null +++ b/apps/backend-quarkus/src/main/java/app/playmancala/history/Migrations.java @@ -0,0 +1,100 @@ +package app.playmancala.history; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +import javax.sql.DataSource; + +import org.jboss.logging.Logger; + +// Applies the migrations in src/main/resources/db/migrations that have not run yet, each in its own transaction. +// Same table and lock as the Node backend's runner, so databases it migrated are recognised. +public final class Migrations { + private static final Logger LOG = Logger.getLogger(Migrations.class); + private static final long MIGRATION_LOCK_ID = 727274; + + // Applied in this order; never change a migration that has been released, add a new one instead. + public static final List VERSIONS = List.of("001_games", "002_clock"); + + private Migrations() { + } + + public static void run(DataSource dataSource) throws SQLException { + try (Connection connection = dataSource.getConnection()) { + connection.setAutoCommit(true); + advisoryLock(connection, "pg_advisory_lock"); + try { + try (Statement statement = connection.createStatement()) { + statement.execute(""" + CREATE TABLE IF NOT EXISTS schema_migrations ( + version text PRIMARY KEY, + applied_at timestamptz NOT NULL DEFAULT now() + )"""); + } + Set applied = appliedVersions(connection); + for (String version : VERSIONS) { + if (applied.contains(version)) continue; + apply(connection, version); + LOG.infof("(DB) Applied migration %s", version); + } + } finally { + try { + advisoryLock(connection, "pg_advisory_unlock"); + } catch (SQLException ignored) { + // Released with the connection. + } + } + } + } + + private static void advisoryLock(Connection connection, String function) throws SQLException { + try (PreparedStatement statement = connection.prepareStatement("SELECT " + function + "(?)")) { + statement.setLong(1, MIGRATION_LOCK_ID); + statement.execute(); + } + } + + private static Set appliedVersions(Connection connection) throws SQLException { + Set applied = new HashSet<>(); + try (Statement statement = connection.createStatement(); + ResultSet rows = statement.executeQuery("SELECT version FROM schema_migrations")) { + while (rows.next()) applied.add(rows.getString(1)); + } + return applied; + } + + private static void apply(Connection connection, String version) throws SQLException { + connection.setAutoCommit(false); + try (Statement statement = connection.createStatement(); + PreparedStatement insert = connection.prepareStatement("INSERT INTO schema_migrations (version) VALUES (?)")) { + statement.execute(sql(version)); + insert.setString(1, version); + insert.executeUpdate(); + connection.commit(); + } catch (SQLException e) { + connection.rollback(); + throw e; + } finally { + connection.setAutoCommit(true); + } + } + + private static String sql(String version) { + String path = "/db/migrations/" + version + ".sql"; + try (InputStream in = Migrations.class.getResourceAsStream(path)) { + if (in == null) throw new IllegalStateException("Missing migration " + path); + return new String(in.readAllBytes(), StandardCharsets.UTF_8); + } catch (IOException e) { + throw new IllegalStateException(e); + } + } +} diff --git a/apps/backend-quarkus/src/main/java/app/playmancala/history/PostgresGameHistoryRepository.java b/apps/backend-quarkus/src/main/java/app/playmancala/history/PostgresGameHistoryRepository.java new file mode 100644 index 0000000..b3ed987 --- /dev/null +++ b/apps/backend-quarkus/src/main/java/app/playmancala/history/PostgresGameHistoryRepository.java @@ -0,0 +1,200 @@ +package app.playmancala.history; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Types; +import java.util.ArrayList; +import java.util.List; +import java.util.regex.Pattern; + +import javax.sql.DataSource; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.core.type.TypeReference; + +import app.playmancala.model.EndReason; +import app.playmancala.model.FinishedGameSummary; +import app.playmancala.model.GameRecord; +import app.playmancala.model.GameResult; +import app.playmancala.model.Json; +import app.playmancala.model.MoveRecord; +import app.playmancala.model.PlayerSummary; + +// Throws IllegalStateException when the database cannot be reached or a query fails. +public class PostgresGameHistoryRepository implements GameHistoryRepository { + private static final Pattern UUID_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 static final String SUMMARY_COLUMNS = """ + g.id, g.result, g.end_reason, g.left_user_id, g.started_at, g.ended_at, g.move_count, + g.final_board, g.peak_spectators, g.clock_initial_ms, + p1.user_id AS p1_user_id, p1.name AS p1_name, p1.is_guest AS p1_is_guest, + p2.user_id AS p2_user_id, p2.name AS p2_name, p2.is_guest AS p2_is_guest"""; + + private static final String FROM_GAMES = """ + FROM games g + JOIN game_players p1 ON p1.game_id = g.id AND p1.seat = 1 + JOIN game_players p2 ON p2.game_id = g.id AND p2.seat = 2"""; + + private static final TypeReference> BOARD = new TypeReference<>() { }; + private static final TypeReference> MOVES = new TypeReference<>() { }; + + private final DataSource dataSource; + + public PostgresGameHistoryRepository(DataSource dataSource) { + this.dataSource = dataSource; + } + + @FunctionalInterface + private interface Work { + T run(Connection connection) throws SQLException, JsonProcessingException; + } + + @Override + public void save(GameRecord record) { + inTransaction(connection -> { + int inserted; + try (PreparedStatement statement = connection.prepareStatement(""" + INSERT INTO games (id, player1_id, player2_id, result, end_reason, left_user_id, started_at, ended_at, + move_count, final_board, moves, peak_spectators, total_spectators, clock_initial_ms) + VALUES (?::uuid, ?, ?, ?, ?, ?, to_timestamp(? / 1000.0), to_timestamp(? / 1000.0), + ?, ?::jsonb, ?::jsonb, ?, ?, ?) + ON CONFLICT (id) DO NOTHING""")) { + statement.setString(1, record.id()); + statement.setString(2, record.player1().userId()); + statement.setString(3, record.player2().userId()); + statement.setString(4, record.result() == null ? null : record.result().wireName()); + statement.setString(5, record.endReason().wireName()); + statement.setString(6, record.leftUserId()); + statement.setLong(7, record.startedAt()); + statement.setLong(8, record.endedAt()); + statement.setInt(9, record.moveCount()); + statement.setString(10, Json.write(record.board())); + statement.setString(11, Json.write(record.moves())); + statement.setInt(12, record.peakSpectators()); + statement.setInt(13, record.totalSpectators()); + if (record.clockInitialMs() == null) statement.setNull(14, Types.INTEGER); + else statement.setInt(14, Math.toIntExact(record.clockInitialMs())); + inserted = statement.executeUpdate(); + } + if (inserted == 0) return null; + try (PreparedStatement statement = connection.prepareStatement( + "INSERT INTO game_players (game_id, seat, user_id, name, is_guest) VALUES (?::uuid, ?, ?, ?, ?)")) { + PlayerSummary[] players = { record.player1(), record.player2() }; + for (int seat = 1; seat <= 2; seat++) { + PlayerSummary player = players[seat - 1]; + statement.setString(1, record.id()); + statement.setInt(2, seat); + statement.setString(3, player.userId()); + statement.setString(4, player.name()); + statement.setBoolean(5, player.isGuest()); + statement.addBatch(); + } + statement.executeBatch(); + } + return null; + }); + } + + @Override + public GameRecord get(String id) { + if (!UUID_PATTERN.matcher(id).matches()) return null; + return query(connection -> { + try (PreparedStatement statement = connection.prepareStatement( + "SELECT " + SUMMARY_COLUMNS + ", g.moves, g.total_spectators " + FROM_GAMES + " WHERE g.id = ?::uuid")) { + statement.setString(1, id); + try (ResultSet row = statement.executeQuery()) { + if (!row.next()) return null; + FinishedGameSummary summary = toSummary(row); + return new GameRecord(summary.id(), summary.player1(), summary.player2(), summary.board(), + summary.result(), summary.endReason(), summary.leftUserId(), summary.startedAt(), + summary.endedAt(), summary.moveCount(), summary.peakSpectators(), summary.clockInitialMs(), + Json.MAPPER.readValue(row.getString("moves"), MOVES), row.getInt("total_spectators")); + } + } + }); + } + + @Override + public List getRecentForUser(String userId, int limit) { + return query(connection -> { + try (PreparedStatement statement = connection.prepareStatement("SELECT " + SUMMARY_COLUMNS + " " + FROM_GAMES + + " WHERE g.id IN (SELECT game_id FROM game_players WHERE user_id = ?) ORDER BY g.ended_at DESC LIMIT ?")) { + statement.setString(1, userId); + statement.setInt(2, limit); + return summaries(statement); + } + }); + } + + @Override + public List getPopular(int limit) { + return query(connection -> { + try (PreparedStatement statement = connection.prepareStatement("SELECT " + SUMMARY_COLUMNS + " " + FROM_GAMES + + " ORDER BY g.peak_spectators DESC, g.ended_at DESC LIMIT ?")) { + statement.setInt(1, limit); + return summaries(statement); + } + }); + } + + private static List summaries(PreparedStatement statement) throws SQLException, JsonProcessingException { + List summaries = new ArrayList<>(); + try (ResultSet rows = statement.executeQuery()) { + while (rows.next()) summaries.add(toSummary(rows)); + } + return summaries; + } + + private static FinishedGameSummary toSummary(ResultSet row) throws SQLException, JsonProcessingException { + Object clockInitialMs = row.getObject("clock_initial_ms"); + return new FinishedGameSummary( + row.getString("id"), + player(row, "p1_"), + player(row, "p2_"), + Json.MAPPER.readValue(row.getString("final_board"), BOARD), + GameResult.fromWireName(row.getString("result")), + EndReason.fromWireName(row.getString("end_reason")), + row.getString("left_user_id"), + row.getTimestamp("started_at").getTime(), + row.getTimestamp("ended_at").getTime(), + row.getInt("move_count"), + row.getInt("peak_spectators"), + clockInitialMs == null ? null : ((Number) clockInitialMs).longValue()); + } + + private static PlayerSummary player(ResultSet row, String prefix) throws SQLException { + return new PlayerSummary(row.getString(prefix + "user_id"), row.getString(prefix + "name"), + row.getBoolean(prefix + "is_guest")); + } + + private T query(Work work) { + try (Connection connection = dataSource.getConnection()) { + return work.run(connection); + } catch (SQLException | JsonProcessingException e) { + throw new IllegalStateException(e.getMessage(), e); + } + } + + private void inTransaction(Work work) { + query(connection -> { + connection.setAutoCommit(false); + try { + work.run(connection); + connection.commit(); + } catch (SQLException | JsonProcessingException | RuntimeException e) { + try { + connection.rollback(); + } catch (SQLException ignored) { + // The connection is broken; the transaction ends with it. + } + throw e; + } finally { + connection.setAutoCommit(true); + } + return null; + }); + } +} diff --git a/apps/backend-quarkus/src/main/java/app/playmancala/live/Cache.java b/apps/backend-quarkus/src/main/java/app/playmancala/live/Cache.java new file mode 100644 index 0000000..d41dbf1 --- /dev/null +++ b/apps/backend-quarkus/src/main/java/app/playmancala/live/Cache.java @@ -0,0 +1,14 @@ +package app.playmancala.live; + +import java.util.Optional; + +import com.fasterxml.jackson.core.type.TypeReference; + +// Short-lived cache of JSON values. Never fails: errors count as a miss, so callers just compute the value. +public interface Cache { + Optional get(String key, TypeReference type); + + void set(String key, Object value, long ttlSeconds); + + void delete(String... keys); +} diff --git a/apps/backend-quarkus/src/main/java/app/playmancala/live/ValkeyCache.java b/apps/backend-quarkus/src/main/java/app/playmancala/live/ValkeyCache.java new file mode 100644 index 0000000..eced22f --- /dev/null +++ b/apps/backend-quarkus/src/main/java/app/playmancala/live/ValkeyCache.java @@ -0,0 +1,52 @@ +package app.playmancala.live; + +import java.util.Arrays; +import java.util.Optional; + +import com.fasterxml.jackson.core.type.TypeReference; + +import app.playmancala.model.Json; +import io.quarkus.redis.datasource.RedisDataSource; +import io.quarkus.redis.datasource.value.SetArgs; +import io.quarkus.redis.datasource.value.ValueCommands; + +public class ValkeyCache implements Cache { + private static final String PREFIX = "mancala:cache:"; + + private final RedisDataSource redis; + private final ValueCommands values; + + public ValkeyCache(RedisDataSource redis) { + this.redis = redis; + this.values = redis.value(String.class); + } + + @Override + public Optional get(String key, TypeReference type) { + try { + String value = values.get(PREFIX + key); + return value == null ? Optional.empty() : Optional.of(Json.MAPPER.readValue(value, type)); + } catch (Exception e) { + return Optional.empty(); + } + } + + @Override + public void set(String key, Object value, long ttlSeconds) { + try { + values.set(PREFIX + key, Json.write(value), new SetArgs().ex(ttlSeconds)); + } catch (RuntimeException e) { + // A missing cache entry only costs a database query. + } + } + + @Override + public void delete(String... keys) { + if (keys.length == 0) return; + try { + redis.key().del(Arrays.stream(keys).map(key -> PREFIX + key).toArray(String[]::new)); + } catch (RuntimeException e) { + // Entries expire on their own. + } + } +} diff --git a/apps/backend-quarkus/src/main/java/app/playmancala/live/ValkeyLiveGameStore.java b/apps/backend-quarkus/src/main/java/app/playmancala/live/ValkeyLiveGameStore.java new file mode 100644 index 0000000..34a4455 --- /dev/null +++ b/apps/backend-quarkus/src/main/java/app/playmancala/live/ValkeyLiveGameStore.java @@ -0,0 +1,87 @@ +package app.playmancala.live; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import org.jboss.logging.Logger; + +import com.fasterxml.jackson.databind.JsonNode; + +import app.playmancala.model.GameRecord; +import app.playmancala.model.Json; +import io.quarkus.redis.datasource.RedisDataSource; +import io.quarkus.redis.datasource.keys.KeyScanArgs; +import io.quarkus.redis.datasource.value.ValueCommands; + +// Keys and values are the Node backend's, so either backend reads what the other wrote. +public class ValkeyLiveGameStore implements LiveGameStore { + private static final Logger LOG = Logger.getLogger(ValkeyLiveGameStore.class); + private static final String LIVE_PREFIX = "mancala:live:"; + private static final String PENDING_SAVE_PREFIX = "mancala:pending-save:"; + + private final RedisDataSource redis; + private final ValueCommands values; + + public ValkeyLiveGameStore(RedisDataSource redis) { + this.redis = redis; + this.values = redis.value(String.class); + } + + @Override + public void saveGame(String gameId, Object snapshot) { + values.set(LIVE_PREFIX + gameId, Json.write(snapshot)); + } + + @Override + public void removeGame(String gameId) { + redis.key().del(LIVE_PREFIX + gameId); + } + + @Override + public List loadGames() { + return loadAll(LIVE_PREFIX); + } + + @Override + public void savePendingRecord(GameRecord record) { + values.set(PENDING_SAVE_PREFIX + record.id(), Json.write(record)); + } + + @Override + public void removePendingRecord(String gameId) { + redis.key().del(PENDING_SAVE_PREFIX + gameId); + } + + @Override + public List loadPendingRecords() { + List records = new ArrayList<>(); + for (JsonNode json : loadAll(PENDING_SAVE_PREFIX)) { + try { + records.add(Json.MAPPER.treeToValue(json, GameRecord.class)); + } catch (Exception e) { + LOG.errorf("(Valkey) Ignoring unreadable pending record %s", json.path("id").asText()); + } + } + return records; + } + + private List loadAll(String prefix) { + // SCAN rather than KEYS, which would block Valkey. + List keys = new ArrayList<>(); + redis.key().scan(new KeyScanArgs().match(prefix + "*").count(100)).toIterable().forEach(keys::add); + if (keys.isEmpty()) return List.of(); + Map stored = values.mget(keys.toArray(String[]::new)); + List parsed = new ArrayList<>(); + for (String key : keys) { + String value = stored.get(key); + if (value == null) continue; + try { + parsed.add(Json.parse(value)); + } catch (Exception e) { + LOG.errorf("(Valkey) Ignoring unreadable value in %s", key); + } + } + return parsed; + } +} diff --git a/apps/backend-quarkus/src/main/resources/application.properties b/apps/backend-quarkus/src/main/resources/application.properties index e69de29..4875640 100644 --- a/apps/backend-quarkus/src/main/resources/application.properties +++ b/apps/backend-quarkus/src/main/resources/application.properties @@ -0,0 +1,41 @@ +# Settings come from environment variables with the Node backend's names; see README.md. + +# HTTP and WebSocket +quarkus.http.port=${PORT:5005} +quarkus.http.test-port=0 +quarkus.http.cors.enabled=true +quarkus.http.cors.origins=* +quarkus.http.access-log.enabled=true +quarkus.http.access-log.pattern=common +quarkus.http.access-log.log-to-file=true +quarkus.http.access-log.base-file-name=access +quarkus.http.access-log.log-directory=. +quarkus.http.access-log.rotate=false +%test.quarkus.http.access-log.enabled=false + +# Keycloak +mancala.auth.issuer=${KEYCLOAK_ISSUER:http://localhost:8080/realms/mancala} +# Empty: the issuer's certs endpoint. +mancala.auth.jwks-url=${KEYCLOAK_JWKS_URL:} +mancala.auth.client-ids=${KEYCLOAK_CLIENT_IDS:mancala-web} +mancala.auth.legacy-userkey-auth=${LEGACY_USERKEY_AUTH:true} + +# Time each player has for a whole game (docs/specs/game-clock). +mancala.game-clock-ms=${GAME_CLOCK_MS:300000} + +# Postgres. Tests start one with Dev Services. The backend starts without it; migrations are retried. +quarkus.datasource.db-kind=postgresql +%dev,prod.quarkus.datasource.jdbc.url=${DATABASE_JDBC_URL:jdbc:postgresql://localhost:5433/mancala} +%dev,prod.quarkus.datasource.username=${DATABASE_USERNAME:mancala} +%dev,prod.quarkus.datasource.password=${DATABASE_PASSWORD:mancala} +quarkus.datasource.jdbc.max-size=5 +quarkus.datasource.jdbc.acquisition-timeout=5S +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.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/main/resources/db/migrations/001_games.sql b/apps/backend-quarkus/src/main/resources/db/migrations/001_games.sql new file mode 100644 index 0000000..f5eb0fc --- /dev/null +++ b/apps/backend-quarkus/src/main/resources/db/migrations/001_games.sql @@ -0,0 +1,27 @@ +CREATE TABLE games ( + id uuid PRIMARY KEY, + player1_id text NOT NULL, + player2_id text NOT NULL, + result text CHECK (result IN ('player1', 'player2', 'draw')), + end_reason text NOT NULL CHECK (end_reason IN ('completed', 'left', 'timeout', 'inactive')), + left_user_id text, + started_at timestamptz NOT NULL, + ended_at timestamptz NOT NULL, + move_count integer NOT NULL, + final_board jsonb NOT NULL, + moves jsonb NOT NULL, + peak_spectators integer NOT NULL DEFAULT 0, + total_spectators integer NOT NULL DEFAULT 0 +); + +CREATE TABLE game_players ( + game_id uuid NOT NULL REFERENCES games (id) ON DELETE CASCADE, + seat smallint NOT NULL CHECK (seat IN (1, 2)), + user_id text NOT NULL, + name text, + is_guest boolean NOT NULL, + PRIMARY KEY (game_id, seat) +); + +CREATE INDEX game_players_user_id ON game_players (user_id); +CREATE INDEX games_popular ON games (peak_spectators DESC, ended_at DESC); diff --git a/apps/backend-quarkus/src/main/resources/db/migrations/002_clock.sql b/apps/backend-quarkus/src/main/resources/db/migrations/002_clock.sql new file mode 100644 index 0000000..10c4a38 --- /dev/null +++ b/apps/backend-quarkus/src/main/resources/db/migrations/002_clock.sql @@ -0,0 +1,2 @@ +-- Time each player started with; NULL for games saved before the game clock. Each move's clock is in `moves`. +ALTER TABLE games ADD COLUMN clock_initial_ms integer; diff --git a/apps/backend-quarkus/src/test/java/app/playmancala/history/PostgresGameHistoryRepositoryTest.java b/apps/backend-quarkus/src/test/java/app/playmancala/history/PostgresGameHistoryRepositoryTest.java new file mode 100644 index 0000000..dcf8bb9 --- /dev/null +++ b/apps/backend-quarkus/src/test/java/app/playmancala/history/PostgresGameHistoryRepositoryTest.java @@ -0,0 +1,189 @@ +package app.playmancala.history; + +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 java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; + +import javax.sql.DataSource; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import com.fasterxml.jackson.databind.JsonNode; + +import app.playmancala.engine.GameStep; +import app.playmancala.model.EndReason; +import app.playmancala.model.FinishedGameSummary; +import app.playmancala.model.GameRecord; +import app.playmancala.model.GameResult; +import app.playmancala.model.Json; +import app.playmancala.model.MoveRecord; +import app.playmancala.model.PlayerSummary; +import io.quarkus.test.junit.QuarkusTest; +import jakarta.inject.Inject; + +// Ported from the Node backend's tests/history/PostgresGameHistoryRepository.test.ts. Postgres runs in Docker. +@QuarkusTest +class PostgresGameHistoryRepositoryTest { + static final PlayerSummary ALICE = new PlayerSummary("kc:alice", "alice", false); + static final PlayerSummary BOB = new PlayerSummary("kc:bob", "bob", false); + static final PlayerSummary GUEST = new PlayerSummary("kc:guest", "guest-1a2b3c4d", true); + static final PlayerSummary LEGACY = new PlayerSummary("legacy:0123456789abcdef0123", null, true); + + static long endedAt = Instant.parse("2026-10-01T00:00:00Z").toEpochMilli(); + + @Inject + DataSource dataSource; + + PostgresGameHistoryRepository repository; + + @BeforeEach + void setUp() throws Exception { + Migrations.run(dataSource); + execute("TRUNCATE games CASCADE"); + repository = new PostgresGameHistoryRepository(dataSource); + } + + void execute(String sql) throws Exception { + try (Connection connection = dataSource.getConnection(); Statement statement = connection.createStatement()) { + statement.execute(sql); + } + } + + static GameRecord record(PlayerSummary player1, PlayerSummary player2, GameResult result, EndReason endReason, + String leftUserId, int peakSpectators, int totalSpectators, Long clockInitialMs, Long remainingMs) { + endedAt += 60_000; + MoveRecord move = new MoveRecord(player1.userId(), 3, List.of(4, 4, 4, 0, 5, 5, 1, 4, 4, 4, 4, 4, 4, 0), + List.of(new GameStep(6, "GAME_STEP_LAST_STONE_IN_BANK")), endedAt - 599_000, remainingMs); + return new GameRecord(UUID.randomUUID().toString(), player1, player2, + List.of(0, 0, 0, 0, 0, 0, 30, 0, 0, 0, 0, 0, 0, 18), result, endReason, leftUserId, endedAt - 600_000, + endedAt, 1, peakSpectators, clockInitialMs, List.of(move), totalSpectators); + } + + static GameRecord record() { + return record(ALICE, BOB, GameResult.PLAYER1, EndReason.COMPLETED, null, 0, 0, null, null); + } + + static GameRecord record(PlayerSummary player1, PlayerSummary player2) { + return record(player1, player2, GameResult.PLAYER1, EndReason.COMPLETED, null, 0, 0, null, null); + } + + static GameRecord withPeak(int peakSpectators) { + return record(ALICE, BOB, GameResult.PLAYER1, EndReason.COMPLETED, null, peakSpectators, 0, null, null); + } + + @Test + void runsMigrationsOnlyOnce() throws Exception { + Migrations.run(dataSource); + List versions = new ArrayList<>(); + try (Connection connection = dataSource.getConnection(); Statement statement = connection.createStatement(); + ResultSet rows = statement.executeQuery("SELECT version FROM schema_migrations ORDER BY version")) { + while (rows.next()) versions.add(rows.getString(1)); + } + assertEquals(List.of("001_games", "002_clock"), versions); + } + + @Test + void savesAGameAndReadsItBackUnchanged() { + GameRecord game = record(ALICE, LEGACY, null, EndReason.INACTIVE, null, 3, 5, null, null); + repository.save(game); + assertEquals(Json.tree(game), Json.tree(repository.get(game.id()))); + } + + @Test + void keepsTheClockTheInitialTimeAndEachMovesTimeLeft() { + GameRecord game = record(ALICE, BOB, GameResult.PLAYER1, EndReason.COMPLETED, null, 0, 0, 300_000L, 291_500L); + repository.save(game); + assertEquals(Json.tree(game), Json.tree(repository.get(game.id()))); + assertEquals(300_000L, repository.getRecentForUser(ALICE.userId(), 1).getFirst().clockInitialMs()); + } + + @Test + void readsGamesSavedBeforeTheGameClockWithoutClockData() { + GameRecord game = record(); + repository.save(game); + JsonNode saved = Json.tree(repository.get(game.id())); + assertFalse(saved.has("clockInitialMs")); + assertFalse(saved.at("/moves/0").has("remainingMs")); + } + + @Test + void keepsLeftUserIdForGamesAPlayerLeft() { + GameRecord game = record(ALICE, BOB, GameResult.PLAYER2, EndReason.LEFT, ALICE.userId(), 0, 0, null, null); + repository.save(game); + assertEquals(ALICE.userId(), repository.get(game.id()).leftUserId()); + } + + @Test + void ignoresSavingTheSameGameAgain() { + GameRecord game = record(); + repository.save(game); + repository.save(new GameRecord(game.id(), game.player1(), game.player2(), game.board(), GameResult.DRAW, + game.endReason(), null, game.startedAt(), game.endedAt(), game.moveCount(), game.peakSpectators(), null, + game.moves(), game.totalSpectators())); + assertEquals(GameResult.PLAYER1, repository.get(game.id()).result()); + } + + @Test + void returnsNullForUnknownAndMalformedIds() { + assertNull(repository.get(UUID.randomUUID().toString())); + assertNull(repository.get("not-a-uuid")); + } + + @Test + void listsAUsersGamesNewestFirstWithoutTheMoves() { + GameRecord first = record(); + GameRecord second = record(GUEST, ALICE); + GameRecord otherPlayers = record(BOB, GUEST); + List.of(first, second, otherPlayers).forEach(repository::save); + + List recent = repository.getRecentForUser(ALICE.userId(), 3); + assertEquals(List.of(second.id(), first.id()), recent.stream().map(FinishedGameSummary::id).toList()); + assertFalse(Json.tree(recent.getFirst()).has("moves")); + assertEquals(1, repository.getRecentForUser(ALICE.userId(), 1).size()); + } + + @Test + void listsPopularGamesByPeakSpectatorsThenNewest() { + GameRecord quiet = withPeak(0); + GameRecord popular = withPeak(9); + GameRecord alsoPopularButNewer = withPeak(9); + List.of(quiet, popular, alsoPopularButNewer).forEach(repository::save); + + assertEquals(List.of(alsoPopularButNewer.id(), popular.id(), quiet.id()), + repository.getPopular(5).stream().map(FinishedGameSummary::id).toList()); + } + + @Test + void readsAGameSavedByTheNodeBackend() throws Exception { + // As the Node backend's repository inserted it. + String id = UUID.randomUUID().toString(); + execute(""" + INSERT INTO games (id, player1_id, player2_id, result, end_reason, left_user_id, started_at, ended_at, + move_count, final_board, moves, peak_spectators, total_spectators, clock_initial_ms) + VALUES ('%s', 'kc:alice', 'legacy:0123456789abcdef0123', 'player2', 'left', 'kc:alice', + to_timestamp(1792152000000 / 1000.0), to_timestamp(1792152061234 / 1000.0), 1, + '[4,4,4,0,5,5,1,4,4,4,4,4,4,0]', + '[{"playerId":"kc:alice","moveIndex":3,"boardSnapshot":[4,4,4,0,5,5,1,4,4,4,4,4,4,0],"gameSteps":[{"index":3,"type":"GAME_STEP_GAME_MOVE","data":null},{"index":6,"type":"GAME_STEP_BOARD_CLEARED","data":{"pitIndexesThatHasStone":[12]}}],"at":1792152001000,"remainingMs":299000}]', + 2, 4, 300000); + INSERT INTO game_players (game_id, seat, user_id, name, is_guest) + VALUES ('%s', 1, 'kc:alice', 'alice', false), ('%s', 2, 'legacy:0123456789abcdef0123', NULL, true); + """.formatted(id, id, id)); + GameRecord game = repository.get(id); + assertEquals(1792152061234L, game.endedAt()); + assertEquals(GameResult.PLAYER2, game.result()); + assertEquals(new PlayerSummary("legacy:0123456789abcdef0123", null, true), game.player2()); + assertEquals("{\"playerId\":\"kc:alice\",\"moveIndex\":3,\"boardSnapshot\":[4,4,4,0,5,5,1,4,4,4,4,4,4,0]," + + "\"gameSteps\":[{\"index\":3,\"type\":\"GAME_STEP_GAME_MOVE\",\"data\":null},{\"index\":6," + + "\"type\":\"GAME_STEP_BOARD_CLEARED\",\"data\":{\"pitIndexesThatHasStone\":[12]}}]," + + "\"at\":1792152001000,\"remainingMs\":299000}", Json.write(game.moves().getFirst())); + } +} diff --git a/apps/backend-quarkus/src/test/java/app/playmancala/live/ValkeyStoreTest.java b/apps/backend-quarkus/src/test/java/app/playmancala/live/ValkeyStoreTest.java new file mode 100644 index 0000000..f53217e --- /dev/null +++ b/apps/backend-quarkus/src/test/java/app/playmancala/live/ValkeyStoreTest.java @@ -0,0 +1,131 @@ +package app.playmancala.live; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.JsonNode; + +import app.playmancala.model.EndReason; +import app.playmancala.model.GameRecord; +import app.playmancala.model.GameResult; +import app.playmancala.model.Json; +import app.playmancala.model.PlayerSummary; +import io.quarkus.redis.client.RedisClientName; +import io.quarkus.redis.datasource.RedisDataSource; +import io.quarkus.test.junit.QuarkusTest; +import jakarta.inject.Inject; + +// Ported from the Node backend's tests/live/ValkeyStore.test.ts. Valkey runs in Docker. +@QuarkusTest +class ValkeyStoreTest { + static final TypeReference> NUMBERS = new TypeReference<>() { }; + + @Inject + RedisDataSource redis; + + // Points at a port nothing listens on. + @Inject + @RedisClientName("offline") + RedisDataSource offline; + + @BeforeEach + void flush() { + redis.flushall(); + } + + static GameRecord record(String id) { + return new GameRecord(id, new PlayerSummary("kc:alice", "alice", false), new PlayerSummary("kc:bob", "bob", false), + List.of(0, 0, 0, 0, 0, 0, 30, 0, 0, 0, 0, 0, 0, 18), GameResult.PLAYER1, EndReason.COMPLETED, null, 1, 2, 0, + 0, null, List.of(), 0); + } + + static List sortById(List games) { + return games.stream().sorted(Comparator.comparing(game -> game.get("id").asText())).toList(); + } + + @Test + void savesLoadsAndRemovesLiveGameSnapshots() { + ValkeyLiveGameStore store = new ValkeyLiveGameStore(redis); + store.saveGame("g1", Map.of("id", "g1", "moves", 1)); + store.saveGame("g2", Map.of("id", "g2", "moves", 2)); + store.saveGame("g1", Map.of("id", "g1", "moves", 3)); + assertEquals(List.of(Json.tree(Map.of("id", "g1", "moves", 3)), Json.tree(Map.of("id", "g2", "moves", 2))), + sortById(store.loadGames())); + + store.removeGame("g1"); + assertEquals(List.of(Json.tree(Map.of("id", "g2", "moves", 2))), store.loadGames()); + } + + @Test + void keepsPendingRecordsSeparateFromLiveGames() { + ValkeyLiveGameStore store = new ValkeyLiveGameStore(redis); + store.saveGame("g1", Map.of("id", "g1")); + store.savePendingRecord(record("r1")); + assertEquals(List.of(record("r1")), store.loadPendingRecords()); + assertEquals(List.of(Json.tree(Map.of("id", "g1"))), store.loadGames()); + + store.removePendingRecord("r1"); + assertEquals(List.of(), store.loadPendingRecords()); + } + + @Test + void skipsValuesItCannotRead() { + ValkeyLiveGameStore store = new ValkeyLiveGameStore(redis); + redis.value(String.class).set("mancala:live:broken", "{not json"); + store.saveGame("g1", Map.of("id", "g1")); + assertEquals(List.of(Json.tree(Map.of("id", "g1"))), store.loadGames()); + } + + @Test + void storesValuesAsTheNodeBackendDid() { + new ValkeyLiveGameStore(redis).saveGame("g1", Map.of("id", "g1")); + assertEquals("{\"id\":\"g1\"}", redis.value(String.class).get("mancala:live:g1")); + new ValkeyCache(redis).set("popular", List.of(1, 2), 60); + assertEquals("[1,2]", redis.value(String.class).get("mancala:cache:popular")); + } + + @Test + void cacheReturnsValuesUntilTheyExpire() throws InterruptedException { + ValkeyCache cache = new ValkeyCache(redis); + cache.set("popular", List.of(1, 2, 3), 1); + assertEquals(Optional.of(List.of(1, 2, 3)), cache.get("popular", NUMBERS)); + Thread.sleep(1500); + assertEquals(Optional.empty(), cache.get("popular", NUMBERS)); + } + + @Test + void cacheDeletesEntries() { + ValkeyCache cache = new ValkeyCache(redis); + cache.set("a", 1, 60); + cache.set("b", 2, 60); + cache.delete("a", "b"); + assertEquals(Optional.empty(), cache.get("a", NUMBERS)); + assertEquals(Optional.empty(), cache.get("b", NUMBERS)); + } + + @Test + void whenValkeyIsUnreachableTheCacheAnswersWithMisses() { + ValkeyCache cache = new ValkeyCache(offline); + cache.set("a", 1, 60); + assertEquals(Optional.empty(), cache.get("a", NUMBERS)); + cache.delete("a"); + } + + @Test + void whenValkeyIsUnreachableTheStoreFailsAtOnce() { + ValkeyLiveGameStore store = new ValkeyLiveGameStore(offline); + long started = System.currentTimeMillis(); + assertThrows(RuntimeException.class, () -> store.saveGame("g1", Map.of())); + assertTrue(System.currentTimeMillis() - started < 1000, "took " + (System.currentTimeMillis() - started) + " ms"); + } +} diff --git a/docs/specs/quarkus-backend/tasks.md b/docs/specs/quarkus-backend/tasks.md index 10bd8a0..653c7ec 100644 --- a/docs/specs/quarkus-backend/tasks.md +++ b/docs/specs/quarkus-backend/tasks.md @@ -10,7 +10,7 @@ Work through the tasks in order; each ends with its tests passing and a commit. - [x] 4. **Game logic**: `GameStore`, `MatchMaker`, `GameClocks`, `GameSerialization`, `GameRecorder`, `GameManager`, `GameClockScheduler`, `LiveGames`, `PublicIds`, `CrashLog`; the in-memory fakes and the ported unit tests. -- [ ] 5. **Storage**: migrations and their runner, `PostgresGameHistoryRepository`, `GameHistoryService`, +- [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` wiring and startup, configuration; `@QuarkusTest`s for the API and the WebSocket.