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<List<FinishedGameSummary>> SUMMARIES = new TypeReference<>() { };
+
+    private static final Comparator<FinishedGameSummary> BY_NEWEST =
+            Comparator.comparingLong(FinishedGameSummary::endedAt).reversed();
+    private static final Comparator<FinishedGameSummary> 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<List<GameRecord>> unsaved;
+
+    public GameHistoryService(GameHistoryRepository repository, Cache cache, Supplier<List<GameRecord>> 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<FinishedGameSummary> getRecentForUser(String userId, int limit) {
+        List<FinishedGameSummary> saved = cached(recentKey(userId), () -> repository.getRecentForUser(userId, MAX_LIST_LIMIT));
+        List<GameRecord> unsavedGames = unsaved.get().stream().filter(record -> record.hasPlayer(userId)).toList();
+        return merge(saved, unsavedGames, BY_NEWEST, limit);
+    }
+
+    public List<FinishedGameSummary> getPopular(int limit) {
+        List<FinishedGameSummary> 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<FinishedGameSummary> cached(String key, Supplier<List<FinishedGameSummary>> load) {
+        Optional<List<FinishedGameSummary>> hit = cache.get(key, SUMMARIES);
+        if (hit.isPresent()) return hit.get();
+        try {
+            List<FinishedGameSummary> 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<FinishedGameSummary> merge(List<FinishedGameSummary> saved, List<GameRecord> unsaved,
+            Comparator<FinishedGameSummary> order, int limit) {
+        Set<String> savedIds = saved.stream().map(FinishedGameSummary::id).collect(Collectors.toSet());
+        List<FinishedGameSummary> 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();
+    }
+}

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<String> 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<String> 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<String> appliedVersions(Connection connection) throws SQLException {
+        Set<String> 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);
+        }
+    }
+}

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<List<Integer>> BOARD = new TypeReference<>() { };
+    private static final TypeReference<List<MoveRecord>> MOVES = new TypeReference<>() { };
+
+    private final DataSource dataSource;
+
+    public PostgresGameHistoryRepository(DataSource dataSource) {
+        this.dataSource = dataSource;
+    }
+
+    @FunctionalInterface
+    private interface Work<T> {
+        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<FinishedGameSummary> 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<FinishedGameSummary> 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<FinishedGameSummary> summaries(PreparedStatement statement) throws SQLException, JsonProcessingException {
+        List<FinishedGameSummary> 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> T query(Work<T> work) {
+        try (Connection connection = dataSource.getConnection()) {
+            return work.run(connection);
+        } catch (SQLException | JsonProcessingException e) {
+            throw new IllegalStateException(e.getMessage(), e);
+        }
+    }
+
+    private void inTransaction(Work<Void> 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;
+        });
+    }
+}

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 {
+    <T> Optional<T> get(String key, TypeReference<T> type);
+
+    void set(String key, Object value, long ttlSeconds);
+
+    void delete(String... keys);
+}

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<String, String> values;
+
+    public ValkeyCache(RedisDataSource redis) {
+        this.redis = redis;
+        this.values = redis.value(String.class);
+    }
+
+    @Override
+    public <T> Optional<T> get(String key, TypeReference<T> 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.
+        }
+    }
+}

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<String, String> 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<JsonNode> 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<GameRecord> loadPendingRecords() {
+        List<GameRecord> 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<JsonNode> loadAll(String prefix) {
+        // SCAN rather than KEYS, which would block Valkey.
+        List<String> keys = new ArrayList<>();
+        redis.key().scan(new KeyScanArgs().match(prefix + "*").count(100)).toIterable().forEach(keys::add);
+        if (keys.isEmpty()) return List.of();
+        Map<String, String> stored = values.mget(keys.toArray(String[]::new));
+        List<JsonNode> 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;
+    }
+}

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

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);

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;

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<String> 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<FinishedGameSummary> 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()));
+    }
+}

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<List<Integer>> 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<JsonNode> sortById(List<JsonNode> 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");
+    }
+}

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.