feat(backend): save finished games in Postgres

- Connection pool and a small migration runner (schema_migrations, one
  transaction per migration, advisory lock)
- games and game_players tables with indexes for a user's games and for
  popular games
- PostgresGameHistoryRepository: idempotent save, get, recent games of a
  user, popular games
- Integration tests against a real Postgres via testcontainers; a custom
  Jest environment gives tests Node's web globals, which Jest 27 leaves out
- GameRecord carries the number of distinct spectators
This commit is contained in:
Halit Aksoy 2026-10-16 00:11:12 +03:00
parent ed2a909f16
commit f404b92aaf
13 changed files with 1386 additions and 12 deletions

View File

@ -1,6 +1,6 @@
module.exports = { module.exports = {
transform: {'^.+\\.ts?$': 'ts-jest'}, transform: {'^.+\\.ts?$': 'ts-jest'},
testEnvironment: 'node', testEnvironment: '<rootDir>/tests/support/NodeEnvironment.js',
testRegex: '/tests/.*\\.(test|spec)?\\.(ts|tsx)$', testRegex: '/tests/.*\\.(test|spec)?\\.(ts|tsx)$',
moduleFileExtensions: ['ts', 'tsx', 'js', 'jsx', 'json', 'node'] moduleFileExtensions: ['ts', 'tsx', 'js', 'jsx', 'json', 'node'],
}; };

View File

@ -24,6 +24,7 @@
"jose": "5.10.0", "jose": "5.10.0",
"mancala.js": "link:../../packages/mancala.js", "mancala.js": "link:../../packages/mancala.js",
"morgan": "^1.10.0", "morgan": "^1.10.0",
"pg": "8.23.1",
"uuid": "^8.3.2", "uuid": "^8.3.2",
"ws": "^7.5.0" "ws": "^7.5.0"
}, },
@ -31,8 +32,10 @@
"@types/express": "^4.17.12", "@types/express": "^4.17.12",
"@types/jest": "^26.0.23", "@types/jest": "^26.0.23",
"@types/node": "^15.12.4", "@types/node": "^15.12.4",
"@types/pg": "8.23.1",
"jest": "^27.0.5", "jest": "^27.0.5",
"nodemon": "^2.0.7", "nodemon": "^2.0.7",
"testcontainers": "12.2.0",
"ts-jest": "^27.0.3", "ts-jest": "^27.0.3",
"ts-node": "^10.0.0", "ts-node": "^10.0.0",
"typescript": "^4.7.4" "typescript": "^4.7.4"

View File

@ -0,0 +1,29 @@
export const sql = `
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);
`;

View File

@ -0,0 +1,11 @@
import { sql as games001 } from "./001_games";
export interface Migration {
version: string;
sql: string;
}
// Applied in this order; never change a migration that has been released, add a new one instead.
export const migrations: Migration[] = [
{ version: "001_games", sql: games001 },
];

View File

@ -0,0 +1,41 @@
import { Pool } from "pg";
import { Migration, migrations as allMigrations } from "./migrations";
const MIGRATION_LOCK_ID = 727274;
export function createPool(databaseUrl: string): Pool {
const pool = new Pool({ connectionString: databaseUrl, max: 5, connectionTimeoutMillis: 5000 });
// Without a listener, an error on an idle connection (e.g. Postgres restarting) would crash the process.
pool.on("error", (err) => console.error("(DB) Idle connection error", err.message));
return pool;
}
// Applies the migrations that have not run yet, each in its own transaction.
export async function runMigrations(pool: Pool, migrations: Migration[] = allMigrations): Promise<void> {
const client = await pool.connect();
try {
await client.query("SELECT pg_advisory_lock($1)", [MIGRATION_LOCK_ID]);
await client.query(`CREATE TABLE IF NOT EXISTS schema_migrations (
version text PRIMARY KEY,
applied_at timestamptz NOT NULL DEFAULT now()
)`);
const { rows } = await client.query<{ version: string }>("SELECT version FROM schema_migrations");
const applied = new Set(rows.map((row) => row.version));
for (const migration of migrations) {
if (applied.has(migration.version)) continue;
try {
await client.query("BEGIN");
await client.query(migration.sql);
await client.query("INSERT INTO schema_migrations (version) VALUES ($1)", [migration.version]);
await client.query("COMMIT");
} catch (err) {
await client.query("ROLLBACK");
throw err;
}
console.info(`(DB) Applied migration ${migration.version}`);
}
} finally {
await client.query("SELECT pg_advisory_unlock($1)", [MIGRATION_LOCK_ID]).catch(() => undefined);
client.release();
}
}

View File

@ -0,0 +1,114 @@
import { FinishedGameSummary, GameRecord, MoveRecord, PlayerSummary } from "@mancala/core";
import { Pool, PoolClient } from "pg";
// Saved, finished games. Ids in records are public ids (see publicIds.ts).
export interface GameHistoryRepository {
// Saving the same game twice is a no-op, so failed saves can be retried.
save(record: GameRecord): Promise<void>;
get(id: string): Promise<GameRecord | undefined>;
// The user's finished games, most recent first.
getRecentForUser(userId: string, limit: number): Promise<FinishedGameSummary[]>;
// Finished games with the most spectators, most recent first among equals.
getPopular(limit: number): Promise<FinishedGameSummary[]>;
}
const UUID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
const 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,
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`;
const 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`;
type SummaryRow = {
id: string; result: FinishedGameSummary["result"]; end_reason: FinishedGameSummary["endReason"];
left_user_id: string | null; started_at: Date; ended_at: Date; move_count: number; final_board: number[];
peak_spectators: number;
p1_user_id: string; p1_name: string | null; p1_is_guest: boolean;
p2_user_id: string; p2_name: string | null; p2_is_guest: boolean;
};
const toPlayer = (userId: string, name: string | null, isGuest: boolean): PlayerSummary =>
name === null ? { userId, isGuest } : { userId, name, isGuest };
function toSummary(row: SummaryRow): FinishedGameSummary {
const summary: FinishedGameSummary = {
id: row.id,
player1: toPlayer(row.p1_user_id, row.p1_name, row.p1_is_guest),
player2: toPlayer(row.p2_user_id, row.p2_name, row.p2_is_guest),
board: row.final_board,
result: row.result,
endReason: row.end_reason,
startedAt: row.started_at.getTime(),
endedAt: row.ended_at.getTime(),
moveCount: row.move_count,
peakSpectators: row.peak_spectators,
};
if (row.left_user_id !== null) summary.leftUserId = row.left_user_id;
return summary;
}
export class PostgresGameHistoryRepository implements GameHistoryRepository {
constructor(private readonly pool: Pool) { }
async save(record: GameRecord): Promise<void> {
await this.inTransaction(async (client) => {
const inserted = await client.query(
`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)
VALUES ($1, $2, $3, $4, $5, $6, to_timestamp($7 / 1000.0), to_timestamp($8 / 1000.0),
$9, $10, $11, $12, $13)
ON CONFLICT (id) DO NOTHING`,
[record.id, record.player1.userId, record.player2.userId, record.result, record.endReason,
record.leftUserId ?? null, record.startedAt, record.endedAt, record.moveCount,
JSON.stringify(record.board), JSON.stringify(record.moves), record.peakSpectators,
record.totalSpectators]);
if (inserted.rowCount === 0) return;
for (const [seat, player] of [[1, record.player1], [2, record.player2]] as const) {
await client.query(
"INSERT INTO game_players (game_id, seat, user_id, name, is_guest) VALUES ($1, $2, $3, $4, $5)",
[record.id, seat, player.userId, player.name ?? null, player.isGuest]);
}
});
}
async get(id: string): Promise<GameRecord | undefined> {
if (!UUID_PATTERN.test(id)) return undefined;
const { rows } = await this.pool.query<SummaryRow & { moves: MoveRecord[], total_spectators: number }>(
`SELECT ${SUMMARY_COLUMNS}, g.moves, g.total_spectators ${FROM_GAMES} WHERE g.id = $1`, [id]);
if (rows.length === 0) return undefined;
return { ...toSummary(rows[0]), moves: rows[0].moves, totalSpectators: rows[0].total_spectators };
}
async getRecentForUser(userId: string, limit: number): Promise<FinishedGameSummary[]> {
const { rows } = await this.pool.query<SummaryRow>(
`SELECT ${SUMMARY_COLUMNS} ${FROM_GAMES}
WHERE g.id IN (SELECT game_id FROM game_players WHERE user_id = $1)
ORDER BY g.ended_at DESC LIMIT $2`, [userId, limit]);
return rows.map(toSummary);
}
async getPopular(limit: number): Promise<FinishedGameSummary[]> {
const { rows } = await this.pool.query<SummaryRow>(
`SELECT ${SUMMARY_COLUMNS} ${FROM_GAMES} ORDER BY g.peak_spectators DESC, g.ended_at DESC LIMIT $1`,
[limit]);
return rows.map(toSummary);
}
private async inTransaction(work: (client: PoolClient) => Promise<void>): Promise<void> {
const client = await this.pool.connect();
try {
await client.query("BEGIN");
await work(client);
await client.query("COMMIT");
} catch (err) {
await client.query("ROLLBACK").catch(() => undefined);
throw err;
} finally {
client.release();
}
}
}

View File

@ -0,0 +1,111 @@
import { GameRecord } from "@mancala/core";
import { Pool } from "pg";
import { v4 as uuid } from "uuid";
import { createPool, runMigrations } from "../../src/db/postgres";
import { PostgresGameHistoryRepository } from "../../src/history/GameHistoryRepository";
import { RunningService, startPostgres } from "../support/containers";
jest.setTimeout(120_000);
const ALICE = { userId: "kc:alice", name: "alice", isGuest: false };
const BOB = { userId: "kc:bob", name: "bob", isGuest: false };
const GUEST = { userId: "kc:guest", name: "guest-1a2b3c4d", isGuest: true };
const LEGACY = { userId: "legacy:0123456789abcdef0123", isGuest: true };
let endedAt = Date.UTC(2026, 9, 1);
const record = (overrides: Partial<GameRecord> = {}): GameRecord => {
endedAt += 60_000;
return {
id: uuid(),
player1: ALICE,
player2: BOB,
board: [0, 0, 0, 0, 0, 0, 30, 0, 0, 0, 0, 0, 0, 18],
result: "player1",
endReason: "completed",
startedAt: endedAt - 600_000,
endedAt,
moveCount: 1,
peakSpectators: 0,
totalSpectators: 0,
moves: [{
playerId: ALICE.userId, moveIndex: 3,
boardSnapshot: [4, 4, 4, 0, 5, 5, 1, 4, 4, 4, 4, 4, 4, 0],
gameSteps: [{ index: 6, type: "GAME_STEP_LAST_STONE_IN_BANK", data: null }],
at: endedAt - 599_000,
}],
...overrides,
};
};
describe("PostgresGameHistoryRepository", () => {
let postgres: RunningService;
let pool: Pool;
let repository: PostgresGameHistoryRepository;
beforeAll(async () => {
postgres = await startPostgres();
pool = createPool(postgres.url);
await runMigrations(pool);
repository = new PostgresGameHistoryRepository(pool);
});
afterAll(async () => {
await pool?.end();
await postgres?.container.stop();
});
beforeEach(() => pool.query("TRUNCATE games CASCADE"));
it("runs migrations only once", async () => {
await runMigrations(pool);
const { rows } = await pool.query("SELECT version FROM schema_migrations");
expect(rows).toEqual([{ version: "001_games" }]);
});
it("saves a game and reads it back unchanged", async () => {
const game = record({ player2: LEGACY, result: null, endReason: "inactive", peakSpectators: 3, totalSpectators: 5 });
await repository.save(game);
expect(await repository.get(game.id)).toEqual(game);
});
it("keeps leftUserId for games a player left", async () => {
const game = record({ result: "player2", endReason: "left", leftUserId: ALICE.userId });
await repository.save(game);
expect((await repository.get(game.id))?.leftUserId).toBe(ALICE.userId);
});
it("ignores saving the same game again", async () => {
const game = record();
await repository.save(game);
await repository.save({ ...game, result: "draw" });
expect((await repository.get(game.id))?.result).toBe("player1");
});
it("returns undefined for unknown and malformed ids", async () => {
expect(await repository.get(uuid())).toBeUndefined();
expect(await repository.get("not-a-uuid")).toBeUndefined();
});
it("lists a user's games newest first, without the moves", async () => {
const first = record();
const second = record({ player1: GUEST, player2: ALICE });
const otherPlayers = record({ player1: BOB, player2: GUEST });
for (const game of [first, second, otherPlayers]) await repository.save(game);
const recent = await repository.getRecentForUser(ALICE.userId, 3);
expect(recent.map((game) => game.id)).toEqual([second.id, first.id]);
expect(recent[0]).not.toHaveProperty("moves");
expect(await repository.getRecentForUser(ALICE.userId, 1)).toHaveLength(1);
});
it("lists popular games by peak spectators, then newest", async () => {
const quiet = record({ peakSpectators: 0 });
const popular = record({ peakSpectators: 9 });
const alsoPopularButNewer = record({ peakSpectators: 9 });
for (const game of [quiet, popular, alsoPopularButNewer]) await repository.save(game);
const ids = (await repository.getPopular(5)).map((game) => game.id);
expect(ids).toEqual([alsoPopularButNewer.id, popular.id, quiet.id]);
});
});

View File

@ -0,0 +1,19 @@
// Jest 27's node environment runs tests in a context that lacks Node's newer web globals (ReadableStream, Blob,
// ...), which the HTTP client used by testcontainers needs. Copy whatever the context is missing from Node.
const NodeEnvironment = require("jest-environment-node");
// Reading these makes Node warn about its experimental web storage.
const SKIPPED = new Set(["localStorage", "sessionStorage"]);
class NodeEnvironmentWithWebGlobals extends NodeEnvironment {
constructor(config, context) {
super(config, context);
for (const name of Object.getOwnPropertyNames(globalThis)) {
if (!SKIPPED.has(name) && !(name in this.global)) {
this.global[name] = globalThis[name];
}
}
}
}
module.exports = NodeEnvironmentWithWebGlobals;

View File

@ -0,0 +1,31 @@
import { GenericContainer, StartedTestContainer, Wait } from "testcontainers";
// Same images as apps/compose.yaml.
const POSTGRES_IMAGE = "postgres:17.6";
const VALKEY_IMAGE = "valkey/valkey:9.1.2";
export interface RunningService {
url: string;
container: StartedTestContainer;
}
export async function startPostgres(): Promise<RunningService> {
const container = await new GenericContainer(POSTGRES_IMAGE)
.withEnvironment({ POSTGRES_USER: "mancala", POSTGRES_PASSWORD: "mancala", POSTGRES_DB: "mancala" })
.withExposedPorts(5432)
// Postgres logs this twice: once for the init run, once when it is really up.
.withWaitStrategy(Wait.forLogMessage(/database system is ready to accept connections/, 2))
.start();
return {
url: `postgres://mancala:mancala@${container.getHost()}:${container.getMappedPort(5432)}/mancala`,
container,
};
}
export async function startValkey(): Promise<RunningService> {
const container = await new GenericContainer(VALKEY_IMAGE)
.withExposedPorts(6379)
.withWaitStrategy(Wait.forLogMessage(/Ready to accept connections/))
.start();
return { url: `redis://${container.getHost()}:${container.getMappedPort(6379)}`, container };
}

File diff suppressed because it is too large Load Diff

View File

@ -87,7 +87,7 @@ interface FinishedGameSummary {
result: GameResult | null; endReason: EndReason; leftUserId?: string; result: GameResult | null; endReason: EndReason; leftUserId?: string;
startedAt: number; endedAt: number; moveCount: number; peakSpectators: number; startedAt: number; endedAt: number; moveCount: number; peakSpectators: number;
} }
interface GameRecord extends FinishedGameSummary { moves: MoveRecord[] } // replay payload interface GameRecord extends FinishedGameSummary { moves: MoveRecord[]; totalSpectators: number } // replay payload
``` ```
Live `Game` messages gain `startedAt` and `moveTimes: number[]` (one per history entry), and `UserConnectionInfo` Live `Game` messages gain `startedAt` and `moveTimes: number[]` (one per history entry), and `UserConnectionInfo`

View File

@ -9,7 +9,7 @@ Work through the tasks in order; each ends with its tests passing and a commit.
`createGameAtMove` with unit tests; localized strings. `createGameAtMove` with unit tests; localized strings.
- [x] 3. **Backend fixes**: remove spectators on disconnect, remove finished games from memory, send the connection - [x] 3. **Backend fixes**: remove spectators on disconnect, remove finished games from memory, send the connection
info once at game start. info once at game start.
- [ ] 4. **Postgres repository**: pool, migration runner, `games` and `game_players`, `PostgresGameHistoryRepository` - [x] 4. **Postgres repository**: pool, migration runner, `games` and `game_players`, `PostgresGameHistoryRepository`
(save, get, recent for user, popular), integration tests. (save, get, recent for user, popular), integration tests.
- [ ] 5. **Valkey store**: live game snapshots and restore, pending saves, cache; integration tests. - [ ] 5. **Valkey store**: live game snapshots and restore, pending saves, cache; integration tests.
- [ ] 6. **Game lifecycle**: start and move times, spectator counters, outcomes (completed, left, timeout, inactive), - [ ] 6. **Game lifecycle**: start and move times, spectator counters, outcomes (completed, left, timeout, inactive),

View File

@ -55,6 +55,8 @@ export interface FinishedGameSummary {
export interface GameRecord extends FinishedGameSummary { export interface GameRecord extends FinishedGameSummary {
moves: MoveRecord[]; moves: MoveRecord[];
// Distinct users who watched the game.
totalSpectators: number;
} }
// GET /game/:id for a game that has ended. // GET /game/:id for a game that has ended.