feat(backend): keep live games and caches in Valkey

- Valkey client that fails commands at once while Valkey is down instead
  of queueing them, and logs connection errors once
- ValkeyLiveGameStore: live game snapshots and finished games waiting to
  be saved, loaded with SCAN on startup
- ValkeyCache: JSON values with a TTL; errors count as cache misses
- Integration tests against a real Valkey, including an unreachable one

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",

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<T>(key: string): Promise<T | undefined>;
+    setJson(key: string, value: unknown, ttlSeconds: number): Promise<void>;
+    delete(...keys: string[]): Promise<void>;
+}
+
+export class ValkeyCache implements Cache {
+    constructor(private readonly client: Valkey) { }
+
+    async getJson<T>(key: string): Promise<T | undefined> {
+        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<void> {
+        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<void> {
+        if (keys.length === 0) return;
+        try {
+            await this.client.del(...keys.map((key) => PREFIX + key));
+        } catch (err) {
+            // Entries expire on their own.
+        }
+    }
+}

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<void>;
+    removeGame(gameId: string): Promise<void>;
+    loadGames(): Promise<unknown[]>;
+    savePendingRecord(record: GameRecord): Promise<void>;
+    removePendingRecord(gameId: string): Promise<void>;
+    loadPendingRecords(): Promise<GameRecord[]>;
+}
+
+export class ValkeyLiveGameStore implements LiveGameStore {
+    constructor(private readonly client: Valkey) { }
+
+    async saveGame(gameId: string, snapshot: unknown): Promise<void> {
+        await this.client.set(LIVE_PREFIX + gameId, JSON.stringify(snapshot));
+    }
+
+    async removeGame(gameId: string): Promise<void> {
+        await this.client.del(LIVE_PREFIX + gameId);
+    }
+
+    loadGames(): Promise<unknown[]> {
+        return this.loadAll(LIVE_PREFIX);
+    }
+
+    async savePendingRecord(record: GameRecord): Promise<void> {
+        await this.client.set(PENDING_SAVE_PREFIX + record.id, JSON.stringify(record));
+    }
+
+    async removePendingRecord(gameId: string): Promise<void> {
+        await this.client.del(PENDING_SAVE_PREFIX + gameId);
+    }
+
+    loadPendingRecords(): Promise<GameRecord[]> {
+        return this.loadAll(PENDING_SAVE_PREFIX) as Promise<GameRecord[]>;
+    }
+
+    private async loadAll(prefix: string): Promise<unknown[]> {
+        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;
+    }
+}

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<string[]> {
+    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;
+}

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

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"

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.