diff --git a/apps/backend/package.json b/apps/backend/package.json index 3a9d604..2b08a10 100644 --- a/apps/backend/package.json +++ b/apps/backend/package.json @@ -21,6 +21,7 @@ "@types/ws": "^7.4.5", "cors": "^2.8.5", "express": "^4.17.1", + "iovalkey": "0.4.0", "jose": "5.10.0", "mancala.js": "link:../../packages/mancala.js", "morgan": "^1.10.0", diff --git a/apps/backend/src/cache/Cache.ts b/apps/backend/src/cache/Cache.ts new file mode 100644 index 0000000..eb2e3d5 --- /dev/null +++ b/apps/backend/src/cache/Cache.ts @@ -0,0 +1,40 @@ +import { Valkey } from "iovalkey"; + +const PREFIX = "mancala:cache:"; + +// Short-lived cache of JSON values. Never fails: errors count as a miss, so callers just compute the value. +export interface Cache { + getJson(key: string): Promise; + setJson(key: string, value: unknown, ttlSeconds: number): Promise; + delete(...keys: string[]): Promise; +} + +export class ValkeyCache implements Cache { + constructor(private readonly client: Valkey) { } + + async getJson(key: string): Promise { + try { + const value = await this.client.get(PREFIX + key); + return value === null ? undefined : JSON.parse(value) as T; + } catch (err) { + return undefined; + } + } + + async setJson(key: string, value: unknown, ttlSeconds: number): Promise { + try { + await this.client.set(PREFIX + key, JSON.stringify(value), "EX", ttlSeconds); + } catch (err) { + // A missing cache entry only costs a database query. + } + } + + async delete(...keys: string[]): Promise { + if (keys.length === 0) return; + try { + await this.client.del(...keys.map((key) => PREFIX + key)); + } catch (err) { + // Entries expire on their own. + } + } +} diff --git a/apps/backend/src/live/LiveGameStore.ts b/apps/backend/src/live/LiveGameStore.ts new file mode 100644 index 0000000..a8abcdf --- /dev/null +++ b/apps/backend/src/live/LiveGameStore.ts @@ -0,0 +1,61 @@ +import { GameRecord } from "@mancala/core"; +import { Valkey } from "iovalkey"; +import { scanKeys } from "./valkey"; + +const LIVE_PREFIX = "mancala:live:"; +const PENDING_SAVE_PREFIX = "mancala:pending-save:"; + +// Durable copy of state that otherwise only lives in memory: live games, so they survive a restart, and finished +// games that could not be saved to the database yet. Methods reject when Valkey is unreachable. +export interface LiveGameStore { + saveGame(gameId: string, snapshot: unknown): Promise; + removeGame(gameId: string): Promise; + loadGames(): Promise; + savePendingRecord(record: GameRecord): Promise; + removePendingRecord(gameId: string): Promise; + loadPendingRecords(): Promise; +} + +export class ValkeyLiveGameStore implements LiveGameStore { + constructor(private readonly client: Valkey) { } + + async saveGame(gameId: string, snapshot: unknown): Promise { + await this.client.set(LIVE_PREFIX + gameId, JSON.stringify(snapshot)); + } + + async removeGame(gameId: string): Promise { + await this.client.del(LIVE_PREFIX + gameId); + } + + loadGames(): Promise { + return this.loadAll(LIVE_PREFIX); + } + + async savePendingRecord(record: GameRecord): Promise { + await this.client.set(PENDING_SAVE_PREFIX + record.id, JSON.stringify(record)); + } + + async removePendingRecord(gameId: string): Promise { + await this.client.del(PENDING_SAVE_PREFIX + gameId); + } + + loadPendingRecords(): Promise { + return this.loadAll(PENDING_SAVE_PREFIX) as Promise; + } + + private async loadAll(prefix: string): Promise { + const keys = await scanKeys(this.client, prefix + "*"); + if (keys.length === 0) return []; + const values = await this.client.mget(keys); + const parsed: unknown[] = []; + values.forEach((value, i) => { + if (value === null) return; + try { + parsed.push(JSON.parse(value)); + } catch { + console.error(`(Valkey) Ignoring unreadable value in ${keys[i]}`); + } + }); + return parsed; + } +} diff --git a/apps/backend/src/live/valkey.ts b/apps/backend/src/live/valkey.ts new file mode 100644 index 0000000..82aee4a --- /dev/null +++ b/apps/backend/src/live/valkey.ts @@ -0,0 +1,33 @@ +import { RedisOptions, Valkey } from "iovalkey"; + +export function createValkeyClient(url: string, options: RedisOptions = {}): Valkey { + const client = new Valkey(url, { + // Fail commands at once while Valkey is down instead of queueing them; callers fall back. + enableOfflineQueue: false, + maxRetriesPerRequest: 1, + ...options, + }); + // Without a listener, connection errors would be unhandled 'error' events and crash the process. + let lastError = ""; + client.on("error", (err: Error) => { + if (err.message !== lastError) console.error("(Valkey) " + err.message); + lastError = err.message; + }); + client.on("ready", () => { + lastError = ""; + console.info("(Valkey) Connected"); + }); + return client; +} + +// Keys matching the pattern, without blocking Valkey the way KEYS would. +export async function scanKeys(client: Valkey, pattern: string): Promise { + const keys: string[] = []; + let cursor = "0"; + do { + const [next, batch] = await client.scan(cursor, "MATCH", pattern, "COUNT", 100); + keys.push(...batch); + cursor = next; + } while (cursor !== "0"); + return keys; +} diff --git a/apps/backend/tests/live/ValkeyStore.test.ts b/apps/backend/tests/live/ValkeyStore.test.ts new file mode 100644 index 0000000..6446f26 --- /dev/null +++ b/apps/backend/tests/live/ValkeyStore.test.ts @@ -0,0 +1,113 @@ +import { GameRecord } from "@mancala/core"; +import { Valkey } from "iovalkey"; +import { ValkeyCache } from "../../src/cache/Cache"; +import { ValkeyLiveGameStore } from "../../src/live/LiveGameStore"; +import { createValkeyClient } from "../../src/live/valkey"; +import { RunningService, startValkey } from "../support/containers"; + +jest.setTimeout(120_000); + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +const record = (id: string): GameRecord => ({ + id, + player1: { userId: "kc:alice", name: "alice", isGuest: false }, + player2: { userId: "kc:bob", name: "bob", isGuest: false }, + board: [0, 0, 0, 0, 0, 0, 30, 0, 0, 0, 0, 0, 0, 18], + result: "player1", endReason: "completed", + startedAt: 1, endedAt: 2, moveCount: 0, peakSpectators: 0, totalSpectators: 0, moves: [], +}); + +describe("Valkey store and cache", () => { + let valkey: RunningService; + let client: Valkey; + + beforeAll(async () => { + valkey = await startValkey(); + client = createValkeyClient(valkey.url); + await new Promise((resolve) => client.once("ready", resolve)); + }); + + afterAll(async () => { + client?.disconnect(); + await valkey?.container.stop(); + }); + + beforeEach(() => client.flushall()); + + describe("ValkeyLiveGameStore", () => { + it("saves, loads and removes live game snapshots", async () => { + const store = new ValkeyLiveGameStore(client); + await store.saveGame("g1", { id: "g1", moves: 1 }); + await store.saveGame("g2", { id: "g2", moves: 2 }); + await store.saveGame("g1", { id: "g1", moves: 3 }); + const sortById = (games: unknown[]) => (games as { id: string }[]).sort((a, b) => a.id.localeCompare(b.id)); + expect(sortById(await store.loadGames())).toEqual([{ id: "g1", moves: 3 }, { id: "g2", moves: 2 }]); + + await store.removeGame("g1"); + expect(await store.loadGames()).toEqual([{ id: "g2", moves: 2 }]); + }); + + it("keeps pending records separate from live games", async () => { + const store = new ValkeyLiveGameStore(client); + await store.saveGame("g1", { id: "g1" }); + await store.savePendingRecord(record("r1")); + expect(await store.loadPendingRecords()).toEqual([record("r1")]); + expect(await store.loadGames()).toEqual([{ id: "g1" }]); + + await store.removePendingRecord("r1"); + expect(await store.loadPendingRecords()).toEqual([]); + }); + + it("skips values it cannot read", async () => { + const store = new ValkeyLiveGameStore(client); + await client.set("mancala:live:broken", "{not json"); + await store.saveGame("g1", { id: "g1" }); + expect(await store.loadGames()).toEqual([{ id: "g1" }]); + }); + }); + + describe("ValkeyCache", () => { + it("returns cached values until they expire", async () => { + const cache = new ValkeyCache(client); + await cache.setJson("popular", [1, 2, 3], 1); + expect(await cache.getJson("popular")).toEqual([1, 2, 3]); + await sleep(1500); + expect(await cache.getJson("popular")).toBeUndefined(); + }); + + it("deletes entries", async () => { + const cache = new ValkeyCache(client); + await cache.setJson("a", 1, 60); + await cache.setJson("b", 2, 60); + await cache.delete("a", "b"); + expect(await cache.getJson("a")).toBeUndefined(); + expect(await cache.getJson("b")).toBeUndefined(); + }); + }); + + describe("when Valkey is unreachable", () => { + let offline: Valkey; + + beforeAll(() => { + // Nothing listens on port 1. No reconnect attempts, so nothing is left running after the tests. + offline = createValkeyClient("redis://127.0.0.1:1", { retryStrategy: () => null, disconnectTimeout: 50 }); + }); + + afterAll(() => offline.disconnect()); + + it("the cache answers with misses instead of failing", async () => { + const cache = new ValkeyCache(offline); + await expect(cache.setJson("a", 1, 60)).resolves.toBeUndefined(); + await expect(cache.getJson("a")).resolves.toBeUndefined(); + await expect(cache.delete("a")).resolves.toBeUndefined(); + }); + + it("the store rejects at once, so callers can fall back", async () => { + const store = new ValkeyLiveGameStore(offline); + const started = Date.now(); + await expect(store.saveGame("g1", {})).rejects.toThrow(); + expect(Date.now() - started).toBeLessThan(1000); + }); + }); +}); diff --git a/apps/backend/yarn.lock b/apps/backend/yarn.lock index da0e486..f8ab311 100644 --- a/apps/backend/yarn.lock +++ b/apps/backend/yarn.lock @@ -331,6 +331,11 @@ protobufjs "^7.5.5" yargs "^17.7.2" +"@iovalkey/commands@^0.1.0": + version "0.1.0" + resolved "https://registry.yarnpkg.com/@iovalkey/commands/-/commands-0.1.0.tgz#62648b19106690ee8ce7d11aaf6249a242f927b0" + integrity sha512-/B9W4qKSSITDii5nkBCHyPkIkAi+ealUtr1oqBJsLxjSRLka4pxun2VvMNSmcwgAMxgXtQfl0qRv7TE+udPJzg== + "@isaacs/cliui@^8.0.2": version "8.0.2" resolved "https://registry.yarnpkg.com/@isaacs/cliui/-/cliui-8.0.2.tgz#b37667b7bc181c168782259bab42474fbf52b550" @@ -1518,6 +1523,11 @@ clone-response@^1.0.2: dependencies: mimic-response "^1.0.0" +cluster-key-slot@^1.1.0: + version "1.1.2" + resolved "https://registry.yarnpkg.com/cluster-key-slot/-/cluster-key-slot-1.1.2.tgz#88ddaa46906e303b5de30d3153b7d9fe0a0c19ac" + integrity sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA== + co@^4.6.0: version "4.6.0" resolved "https://registry.yarnpkg.com/co/-/co-4.6.0.tgz#6ea6bdf3d853ae54ccb8e47bfa0bf3f9031fb184" @@ -1725,7 +1735,7 @@ debug@^3.2.7: dependencies: ms "^2.1.1" -debug@^4.4.3: +debug@^4.3.4, debug@^4.4.3: version "4.4.3" resolved "https://registry.yarnpkg.com/debug/-/debug-4.4.3.tgz#c6ae432d9bd9662582fce08709b038c58e9e3d6a" integrity sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA== @@ -1774,6 +1784,11 @@ delayed-stream@~1.0.0: resolved "https://registry.yarnpkg.com/delayed-stream/-/delayed-stream-1.0.0.tgz#df3ae199acadfb7d440aaae0b29e2272b24ec619" integrity sha1-3zrhmayt+31ECqrgsp4icrJOxhk= +denque@^2.1.0: + version "2.1.0" + resolved "https://registry.yarnpkg.com/denque/-/denque-2.1.0.tgz#e93e1a6569fb5e66f16a3c2a2964617d349d6ab1" + integrity sha512-HVQE3AAb/pxF8fQAoiqpvg9i3evqug3hoiwakOyZAwJm+6vZehbkYXZ0l4JxS+I3QxM97v5aaRNhj8v5oBhekw== + depd@2.0.0, depd@~2.0.0: version "2.0.0" resolved "https://registry.yarnpkg.com/depd/-/depd-2.0.0.tgz#b696163cc757560d09cf22cc8fad1571b79e76df" @@ -2399,6 +2414,21 @@ ini@~1.3.0: resolved "https://registry.yarnpkg.com/ini/-/ini-1.3.8.tgz#a29da425b48806f34767a4efce397269af28432c" integrity sha512-JV/yugV2uzW5iMRSiZAyDtQd+nxtUnjeLt0acNdw98kKLrvuRVyB80tsREOE7yvGVgalhZ6RNXCmEHkUKBKxew== +iovalkey@0.4.0: + version "0.4.0" + resolved "https://registry.yarnpkg.com/iovalkey/-/iovalkey-0.4.0.tgz#65316e7aabc980191f2122709569ec878b880d3b" + integrity sha512-OSUKxJ+s44CLdUTaicX4+pVrBN/zHKIYwr2oKhRDuczr19FwcSCXEZHzcMUYWZNh1mNSCV9bNJHW/T6cHVm9Aw== + dependencies: + "@iovalkey/commands" "^0.1.0" + cluster-key-slot "^1.1.0" + debug "^4.3.4" + denque "^2.1.0" + lodash.defaults "^4.2.0" + lodash.isarguments "^3.1.0" + redis-errors "^1.2.0" + redis-parser "^3.0.0" + standard-as-callback "^2.1.0" + ipaddr.js@1.9.1: version "1.9.1" resolved "https://registry.yarnpkg.com/ipaddr.js/-/ipaddr.js-1.9.1.tgz#bff38543eeb8984825079ff3a2a8e6cbd46781b3" @@ -3108,6 +3138,16 @@ lodash.camelcase@^4.3.0: resolved "https://registry.yarnpkg.com/lodash.camelcase/-/lodash.camelcase-4.3.0.tgz#b28aa6288a2b9fc651035c7711f65ab6190331a6" integrity sha512-TwuEnCnxbc3rAvhf/LbG7tJUDzhqXyFnv3dtzLOPgCG/hODL7WFnsbwktkD7yUV0RrreP/l1PALq/YSg6VvjlA== +lodash.defaults@^4.2.0: + version "4.2.0" + resolved "https://registry.yarnpkg.com/lodash.defaults/-/lodash.defaults-4.2.0.tgz#d09178716ffea4dde9e5fb7b37f6f0802274580c" + integrity sha512-qjxPLHd3r5DnsdGacqOMU6pb/avJzdh9tFX2ymgoZE27BmjXrNy/y4LoaiTeAb+O3gL8AfpJGtqfX/ae2leYYQ== + +lodash.isarguments@^3.1.0: + version "3.1.0" + resolved "https://registry.yarnpkg.com/lodash.isarguments/-/lodash.isarguments-3.1.0.tgz#2f573d85c6a24289ff00663b491c1d338ff3458a" + integrity sha512-chi4NHZlZqZD18a0imDHnZPrDeBbTtVN7GXMwuGdRH9qotxAjYs3aVLKc7zNOG9eddR5Ksd8rvFEBc9SsggPpg== + lodash.memoize@4.x: version "4.1.2" resolved "https://registry.yarnpkg.com/lodash.memoize/-/lodash.memoize-4.1.2.tgz#bcc6c49a42a2840ed997f323eada5ecd182e0bfe" @@ -3815,6 +3855,18 @@ readdirp@~3.6.0: dependencies: picomatch "^2.2.1" +redis-errors@^1.0.0, redis-errors@^1.2.0: + version "1.2.0" + resolved "https://registry.yarnpkg.com/redis-errors/-/redis-errors-1.2.0.tgz#eb62d2adb15e4eaf4610c04afe1529384250abad" + integrity sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w== + +redis-parser@^3.0.0: + version "3.0.0" + resolved "https://registry.yarnpkg.com/redis-parser/-/redis-parser-3.0.0.tgz#b66d828cdcafe6b4b8a428a7def4c6bcac31c8b4" + integrity sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A== + dependencies: + redis-errors "^1.0.0" + registry-auth-token@^4.0.0: version "4.2.1" resolved "https://registry.yarnpkg.com/registry-auth-token/-/registry-auth-token-4.2.1.tgz#6d7b4006441918972ccd5fedcd41dc322c79b250" @@ -4059,6 +4111,11 @@ stack-utils@^2.0.3: dependencies: escape-string-regexp "^2.0.0" +standard-as-callback@^2.1.0: + version "2.1.0" + resolved "https://registry.yarnpkg.com/standard-as-callback/-/standard-as-callback-2.1.0.tgz#8953fc05359868a77b5b9739a665c5977bb7df45" + integrity sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A== + statuses@2.0.1: version "2.0.1" resolved "https://registry.yarnpkg.com/statuses/-/statuses-2.0.1.tgz#55cb000ccf1d48728bd23c685a063998cf1a1b63" diff --git a/docs/specs/game-history/tasks.md b/docs/specs/game-history/tasks.md index 253b17c..77812ef 100644 --- a/docs/specs/game-history/tasks.md +++ b/docs/specs/game-history/tasks.md @@ -11,7 +11,7 @@ Work through the tasks in order; each ends with its tests passing and a commit. info once at game start. - [x] 4. **Postgres repository**: pool, migration runner, `games` and `game_players`, `PostgresGameHistoryRepository` (save, get, recent for user, popular), integration tests. -- [ ] 5. **Valkey store**: live game snapshots and restore, pending saves, cache; integration tests. +- [x] 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), timeout watcher, `offlineSince`, rejecting a new game while playing, saving with retry, restore on startup; unit tests.