From fb553c2fd083cbb19128af363467c3b75c2472d6 Mon Sep 17 00:00:00 2001 From: Devin Zuczek Date: Sat, 11 Jul 2026 12:37:22 -0400 Subject: [PATCH] move presence to d1 --- apps/auth/src/auth.app.ts | 13 +- apps/auth/src/context.ts | 8 +- apps/auth/src/test/integration/api.test.ts | 16 ++- apps/auth/wrangler.jsonc | 8 -- apps/match/src/context.ts | 11 +- apps/match/src/match.app.ts | 133 ++++++++++-------- apps/match/src/test/integration/api.test.ts | 128 ++++++++++++++++- apps/match/wrangler.jsonc | 17 +-- apps/rooms/migrations/0006_presence.sql | 19 +++ apps/rooms/src/context.ts | 8 +- apps/rooms/src/rooms.app.ts | 16 +-- apps/rooms/src/test/integration/api.test.ts | 16 ++- apps/rooms/wrangler.jsonc | 8 -- packages/domain/src/index.ts | 1 + packages/domain/src/presence-db.ts | 148 ++++++++++++++++++++ packages/domain/src/room-instance-db.ts | 33 +++++ 16 files changed, 452 insertions(+), 131 deletions(-) create mode 100644 apps/rooms/migrations/0006_presence.sql create mode 100644 packages/domain/src/presence-db.ts diff --git a/apps/auth/src/auth.app.ts b/apps/auth/src/auth.app.ts index 0e4cf8d..24cce15 100644 --- a/apps/auth/src/auth.app.ts +++ b/apps/auth/src/auth.app.ts @@ -10,6 +10,7 @@ import { RoomInstanceType, setLastLoginTime, setPasswordHash, + setPresence, } from '@repo/domain' import { logger, withNotFound, withOnError } from '@repo/hono-helpers' import { generateToken, TOKEN_TTL_SECONDS, validateAndGetAccountId } from '@repo/jwt' @@ -48,16 +49,14 @@ const ORIENTATION_ROOM_ID = 13 * client treats presence as out-of-sync and bounces the player to the dorm. */ const ORIENTATION_INSTANCE_ID = -2 -/** Presence TTL (s) — matches the match worker; refreshed by each heartbeat. */ -const PRESENCE_TTL = 900 /** * Seed a freshly created account's match presence to the Orientation room. The * client is placed into Orientation by its new-user flow without a matchmake * call, so the match heartbeat would otherwise report no/stale (dorm) presence * and bounce the player out. We write the Orientation instance (built from the - * shared rooms D1, matching the match worker's `roomInstanceFromRoom` shape) so - * the heartbeat keeps them there. + * shared rooms D1, matching the match worker's `roomInstanceFromRoom` shape) into + * the shared `presence` table (see @repo/domain) so the heartbeat keeps them there. */ async function placeNewPlayerInOrientation(env: App['Bindings'], accountId: number): Promise { const row = await env.DB.prepare('SELECT data FROM room WHERE room_id = ?1') @@ -92,16 +91,14 @@ async function placeNewPlayerInOrientation(env: App['Bindings'], accountId: numb isInProgress: false, EncryptVoiceChat: false, } - const presence = { + await setPresence(env.DB, { + accountId, roomInstance, statusVisibility: 0, deviceClass: 0, vrMovementMode: 1, platform: 0, appVersion: '20230302', - } - await env.RECFLARE_MATCH_PRESENCE.put(`presence:${accountId}`, JSON.stringify(presence), { - expirationTtl: PRESENCE_TTL, }) } diff --git a/apps/auth/src/context.ts b/apps/auth/src/context.ts index 2844359..abc1fb9 100644 --- a/apps/auth/src/context.ts +++ b/apps/auth/src/context.ts @@ -3,12 +3,10 @@ import type { SharedHonoEnv, SharedHonoVariables } from '@repo/hono-helpers/src/ export type Env = SharedHonoEnv & { // Shared rooms/accounts D1 database. The `auth` worker owns the `accounts` - // table (creates accounts on signup, seeds the system + Coach accounts). + // table (creates accounts on signup, seeds the system + Coach accounts). Also + // seeds new players' presence to the Orientation room (the shared `presence` + // table, see @repo/domain) so the match heartbeat keeps them there. DB: D1Database - // Shared match-presence KV (owned by the `match` worker). On account creation - // the new player's presence is seeded to the Orientation room so the match - // heartbeat keeps them there instead of bouncing them to the dorm. - RECFLARE_MATCH_PRESENCE: KVNamespace // Shared Secrets Store binding for the HS256 signing key. Resolve the value with // `await env.JWT_SECRET.get()`. Every worker binds the same store, so tokens // signed here verify in all of them. Provisioned via `wrangler secrets-store`; diff --git a/apps/auth/src/test/integration/api.test.ts b/apps/auth/src/test/integration/api.test.ts index f11d443..ed34c0d 100644 --- a/apps/auth/src/test/integration/api.test.ts +++ b/apps/auth/src/test/integration/api.test.ts @@ -4,7 +4,7 @@ import { beforeAll, describe, expect, test } from 'vitest' import '../../auth.app' -import { SCHEMA_DDL } from '@repo/domain' +import { PRESENCE_SCHEMA_DDL, SCHEMA_DDL } from '@repo/domain' import { hashPassword } from '../../password' import { REFRESH_SCHEMA_DDL } from '../../refresh-db' @@ -32,6 +32,8 @@ beforeAll(async () => { await adminSecretsStore(env.JWT_SECRET).create('test-signing-key') for (const stmt of SCHEMA_DDL) await env.DB.prepare(stmt).run() for (const stmt of REFRESH_SCHEMA_DDL) await env.DB.prepare(stmt).run() + // Presence table (owned by the rooms worker) — signup seeds the Orientation row. + for (const stmt of PRESENCE_SCHEMA_DDL) await env.DB.prepare(stmt).run() // Seed the accounts the credential-login tests use, each with LOGIN_PASSWORD set. const hash = await hashPassword(LOGIN_PASSWORD) @@ -312,11 +314,15 @@ describe('auth worker routes', () => { test('POST /connect/token create_account seeds the new player into Orientation', async () => { const payload = await tokenFor('grant_type=create_account&platform_id=steam-456') const sub = payload.sub as string - const presence = await env.RECFLARE_MATCH_PRESENCE.get<{ + // Presence is written to the shared `presence` D1 table (account_id keyed). + const row = await env.DB.prepare('SELECT data FROM presence WHERE account_id = ?1') + .bind(Number(sub)) + .first<{ data: string }>() + expect(row).not.toBeNull() + const presence = JSON.parse(row!.data) as { roomInstance: { roomInstanceId: number; roomId: number; location: string; name: string } - }>(`presence:${sub}`, 'json') - expect(presence).not.toBeNull() - expect(presence!.roomInstance).toMatchObject({ + } + expect(presence.roomInstance).toMatchObject({ roomInstanceId: -2, roomId: 13, location: ORIENTATION_SCENE, diff --git a/apps/auth/wrangler.jsonc b/apps/auth/wrangler.jsonc index 2294e0e..6e0e8bb 100644 --- a/apps/auth/wrangler.jsonc +++ b/apps/auth/wrangler.jsonc @@ -18,14 +18,6 @@ "migrations_table": "d1_migrations_auth" } ], - // Shared match-presence KV (owned by the `match` worker) — used to place new - // accounts into the Orientation room on signup. - "kv_namespaces": [ - { - "binding": "RECFLARE_MATCH_PRESENCE", - "id": "local" - } - ], "logpush": false, // Shared Secrets Store holding the HS256 JWT signing key. Every worker binds the // same store as JWT_SECRET so tokens signed by `auth` verify here. The "local" diff --git a/apps/match/src/context.ts b/apps/match/src/context.ts index 8b34674..8b09bbf 100644 --- a/apps/match/src/context.ts +++ b/apps/match/src/context.ts @@ -6,12 +6,11 @@ export type Env = SharedHonoEnv & { // with `await env.JWT_SECRET.get()`; all workers bind the same store so tokens // signed by `auth` verify here. JWT_SECRET: SecretsStoreSecret - // Per-player presence (the room instance they're currently in). Written by - // matchmake/goto, read by the heartbeat, cleared on login — mirrors the - // reference server's HeartbeatDB. - RECFLARE_MATCH_PRESENCE: KVNamespace - // Shared rooms DB (owned by the `rooms` worker). Read-only here to resolve a - // room's real scene/subroom when matchmaking into it. + // Shared `recflare` DB. Resolves room scenes for matchmaking (read), writes a + // player's personal dorm room on first entry, and holds player presence — the + // room instance each player is currently in (written by matchmake/heartbeat, + // read by the heartbeat and the batch `/player` lookup). See @repo/domain's + // presence-db (table owned/migrated by the `rooms` worker). DB: D1Database } diff --git a/apps/match/src/match.app.ts b/apps/match/src/match.app.ts index 00dee88..8699a3d 100644 --- a/apps/match/src/match.app.ts +++ b/apps/match/src/match.app.ts @@ -5,23 +5,28 @@ import { createRoomInstance, getJoinableInstance, getOrCreateDormRoom, + getPresence, + getPresences, getRoomById, getRoomByName, getRoomInstancesByRoom, + refreshInstanceFullness, RoomInstanceType, + setPresence, setRoomInstanceInProgress, } from '@repo/domain' import { withNotFound, withOnError } from '@repo/hono-helpers' import { validateAndGetAccountId } from '@repo/jwt' -import type { Room } from '@repo/domain' +import type { Room, StoredPresence } from '@repo/domain' import type { Context } from 'hono' import type { App } from './context' /** * The matchmaking surface. Rooms and room instances are D1-backed (matchmaking * finds/creates a `room_instance` row per session); player lookups still fall back - * to default values when nothing is found. Presence lives in the match KV. + * to default values when nothing is found. Presence is D1-backed too (the + * `presence` table; see @repo/domain's presence-db). * * Auth-gated routes still validate the Bearer JWT issued by the `auth` worker. */ @@ -72,20 +77,19 @@ type RoomInstance = ReturnType /** * Stored presence for a player — the room instance they matchmade into plus the - * status fields the heartbeat echoes back. Mirrors the reference server's - * HeartbeatDB row. + * status fields the heartbeat echoes back. The generic StoredPresence lives in + * @repo/domain; here it's specialized to the match worker's RoomInstance shape. */ -interface Presence { - roomInstance: RoomInstance | null - statusVisibility: number - deviceClass: number - vrMovementMode: number - platform: number - appVersion: string -} +type Presence = StoredPresence -/** Presence is kept this long (s) after the last matchmake/heartbeat refresh. */ -const PRESENCE_TTL = 900 +/** + * A heartbeat that changes nothing re-writes presence only once its TTL drops + * within this window of expiring (s), instead of on every beat. So a player who's + * sitting still is refreshed at most once per (PRESENCE_TTL_SECONDS − this) rather + * than on every heartbeat — far fewer D1 writes, while still staying comfortably + * ahead of expiry (the client heartbeats many times inside this window). + */ +const PRESENCE_REFRESH_THRESHOLD = 300 /** * Game build version reported in presence. This is a server-side constant — the @@ -94,24 +98,11 @@ const PRESENCE_TTL = 900 */ const GAME_VERSION = '20230302' -const presenceKey = (id: number) => `presence:${id}` - -/** Persist the player's presence (room instance + status), refreshing the TTL. */ -async function setPresence(c: Context, id: number, presence: Presence): Promise { - await c.env.RECFLARE_MATCH_PRESENCE.put(presenceKey(id), JSON.stringify(presence), { - expirationTtl: PRESENCE_TTL, - }) -} - -/** Read the player's stored presence, or null when they aren't in a room. */ -async function getPresence(c: Context, id: number): Promise { - return c.env.RECFLARE_MATCH_PRESENCE.get(presenceKey(id), 'json') -} - /** Store the room instance the player just matchmade into, preserving status. */ async function enterRoom(c: Context, id: number, roomInstance: RoomInstance): Promise { - const prev = await getPresence(c, id) - await setPresence(c, id, { + const prev = await getPresence(c.env.DB, id) + await setPresence(c.env.DB, { + accountId: id, roomInstance, statusVisibility: prev?.statusVisibility ?? 0, deviceClass: prev?.deviceClass ?? 0, @@ -119,6 +110,15 @@ async function enterRoom(c: Context, id: number, roomInstance: RoomInstance platform: prev?.platform ?? 0, appVersion: prev?.appVersion || GAME_VERSION, }) + // Keep the destination instance's is_full flag in sync with live presence (the + // player's own presence, just written, is counted). Then re-evaluate the + // instance they left — its head-count dropped — so a full room frees up when + // players move on. Both no-op for the synthetic dorm/orientation instances. + await refreshInstanceFullness(c.env.DB, roomInstance.roomInstanceId) + const leftId = prev?.roomInstance?.roomInstanceId + if (leftId != null && leftId !== roomInstance.roomInstanceId) { + await refreshInstanceFullness(c.env.DB, leftId) + } } /** @@ -330,21 +330,22 @@ const app = new Hono() .filter((n) => !Number.isNaN(n)) if (!ids || ids.length === 0) return c.json(DEFAULT_GET_PLAYER) - const players = await Promise.all( - ids.map(async (playerId) => { - const p = await getPresence(c, playerId) - return { - playerId, - statusVisibility: p?.statusVisibility ?? 0, - deviceClass: p?.deviceClass ?? 0, - vrMovementMode: p?.vrMovementMode ?? 1, - roomInstance: p?.roomInstance ?? null, - isOnline: p?.roomInstance != null, - appVersion: p?.appVersion || GAME_VERSION, - platform: p?.platform ?? 0, - } - }) - ) + // One query for the whole batch (D1 `WHERE account_id IN (…)`), rather than a + // point read per id as the KV store required. + const presences = await getPresences(c.env.DB, ids) + const players = ids.map((playerId) => { + const p = presences.get(playerId) + return { + playerId, + statusVisibility: p?.statusVisibility ?? 0, + deviceClass: p?.deviceClass ?? 0, + vrMovementMode: p?.vrMovementMode ?? 1, + roomInstance: p?.roomInstance ?? null, + isOnline: p?.roomInstance != null, + appVersion: p?.appVersion || GAME_VERSION, + platform: p?.platform ?? 0, + } + }) return c.json(players) }) @@ -367,16 +368,34 @@ const app = new Hono() // Return the player's stored presence (set by matchmake/goto), mirroring the // reference server's HeartbeatDB.GetPlayerHeartbeat. No presence → the player // isn't in a room yet, so roomInstance=null / isOnline=false. Posted status - // fields are merged back and the TTL refreshed so presence stays alive. - const presence = await getPresence(c, id) + // fields are merged back; the row is re-written (refreshing the TTL) only when + // something changed or its TTL is close to lapsing — see below. + const presence = await getPresence(c.env.DB, id) if (presence) { - if (hb.statusVisibility !== undefined) presence.statusVisibility = hb.statusVisibility - if (hb.deviceClass !== undefined) presence.deviceClass = hb.deviceClass - if (hb.vrMovementMode !== undefined) presence.vrMovementMode = hb.vrMovementMode - if (hb.platform !== undefined) presence.platform = hb.platform - if (hb.appVersion) presence.appVersion = hb.appVersion - if (!presence.appVersion) presence.appVersion = GAME_VERSION - await setPresence(c, id, presence) + // Merge the posted status fields, tracking whether any actually changed. + let changed = false + const apply = (key: K, value: Presence[K]) => { + if (presence[key] !== value) { + presence[key] = value + changed = true + } + } + if (hb.statusVisibility !== undefined) apply('statusVisibility', hb.statusVisibility) + if (hb.deviceClass !== undefined) apply('deviceClass', hb.deviceClass) + if (hb.vrMovementMode !== undefined) apply('vrMovementMode', hb.vrMovementMode) + if (hb.platform !== undefined) apply('platform', hb.platform) + if (hb.appVersion) apply('appVersion', hb.appVersion) + if (!presence.appVersion) apply('appVersion', GAME_VERSION) + + // Extending the TTL means re-writing the row, so skip the write on an + // unchanged heartbeat until the TTL is within PRESENCE_REFRESH_THRESHOLD + // (s) of lapsing — a still player is refreshed periodically rather than on + // every beat. `expiresAt` is epoch seconds (set by setPresence). + const nowSeconds = Math.floor(Date.now() / 1000) + const dueForRefresh = presence.expiresAt - nowSeconds <= PRESENCE_REFRESH_THRESHOLD + if (changed || dueForRefresh) { + await setPresence(c.env.DB, presence) + } } return c.json({ @@ -397,10 +416,10 @@ const app = new Hono() const body = await c.req.parseBody().catch(() => ({}) as Record) const sv = typeof body.statusVisibility === 'string' ? Number.parseInt(body.statusVisibility, 10) : NaN - const presence = await getPresence(c, id) + const presence = await getPresence(c.env.DB, id) if (presence && !Number.isNaN(sv)) { presence.statusVisibility = sv - await setPresence(c, id, presence) + await setPresence(c.env.DB, presence) } } return c.body(null, 200) @@ -434,7 +453,7 @@ const app = new Hono() // So: preserve existing presence; only fall back to the offline dorm when the // player has none (e.g. the title screen before they've entered any room). if (id !== null) { - const presence = await getPresence(c, id) + const presence = await getPresence(c.env.DB, id) if (presence?.roomInstance) { return c.json({ errorCode: 0, roomInstance: presence.roomInstance }) } diff --git a/apps/match/src/test/integration/api.test.ts b/apps/match/src/test/integration/api.test.ts index cddab8e..67e3d51 100644 --- a/apps/match/src/test/integration/api.test.ts +++ b/apps/match/src/test/integration/api.test.ts @@ -2,7 +2,12 @@ import { adminSecretsStore, env } from 'cloudflare:test' import { exports } from 'cloudflare:workers' import { beforeAll, describe, expect, test } from 'vitest' -import { ROOM_INSTANCE_SCHEMA_DDL } from '@repo/domain' +import { + countPlayersInInstance, + getRoomInstance, + PRESENCE_SCHEMA_DDL, + ROOM_INSTANCE_SCHEMA_DDL, +} from '@repo/domain' import '../../match.app' @@ -40,6 +45,14 @@ const TEST_ROOMS = [ CreatorAccountId: 42, SubRooms: [{ SubRoomId: 3, UnitySceneId: RECCENTER_SCENE, MaxPlayers: 8 }], }, + { + // A single-seat room so one player fills its instance (fullness tests). + RoomId: 5, + Name: 'SoloRoom', + IsDorm: false, + Accessibility: 1, + SubRooms: [{ SubRoomId: 5, UnitySceneId: RECCENTER_SCENE, MaxPlayers: 1 }], + }, ] beforeAll(async () => { @@ -58,6 +71,8 @@ beforeAll(async () => { await env.DB.batch(TEST_ROOMS.map((r) => insert.bind(JSON.stringify(r)))) // Room instances (owned by the rooms worker) — matchmaking finds/creates here. for (const stmt of ROOM_INSTANCE_SCHEMA_DDL) await env.DB.prepare(stmt).run() + // Presence table (owned by the rooms worker) — written/read by matchmake + heartbeat. + for (const stmt of PRESENCE_SCHEMA_DDL) await env.DB.prepare(stmt).run() // Accounts table (owned by the auth worker) — dorm creation reads the username // to name the room. Seed the players the dorm tests authenticate as. @@ -458,6 +473,117 @@ describe('auth-gated endpoints', () => { }) }) + // Seed presence directly into D1 with a chosen `expiresAt` (epoch seconds) so the + // TTL-refresh branch can be exercised deterministically (independent of timing). + const seedPresence = (id: number, expiresAt: number) => + env.DB.prepare('INSERT OR REPLACE INTO presence (data) VALUES (?1)') + .bind( + JSON.stringify({ + accountId: id, + roomInstance: { roomInstanceId: 1000042, roomId: 1 }, + statusVisibility: 0, + deviceClass: 0, + vrMovementMode: 1, + platform: 0, + appVersion: '20230302', + expiresAt, + }) + ) + .run() + + const storedExpiresAt = async (id: number): Promise => { + const row = await env.DB.prepare('SELECT data FROM presence WHERE account_id = ?1') + .bind(id) + .first<{ data: string }>() + return (JSON.parse(row!.data) as { expiresAt: number }).expiresAt + } + + const nowSeconds = () => Math.floor(Date.now() / 1000) + + test('heartbeat refreshes presence when its TTL is close to lapsing', async () => { + // TTL about to lapse (well inside the refresh window). + const nearExpiry = nowSeconds() + 10 + await seedPresence(700, nearExpiry) + await exports.default.fetch(`${ORIGIN}/player/heartbeat`, { + method: 'POST', + headers: await bearer('700'), + }) + // The heartbeat re-wrote the row, pushing expiry ~PRESENCE_TTL_SECONDS ahead. + expect(await storedExpiresAt(700)).toBeGreaterThan(nearExpiry + 60) + }) + + test('heartbeat skips the write when nothing changed and the TTL is healthy', async () => { + // A distinctive, far-future expiry (outside the refresh window) survives + // untouched — proving the unchanged heartbeat did not re-write the row. + const healthyExpiry = nowSeconds() + 800 + await seedPresence(701, healthyExpiry) + await exports.default.fetch(`${ORIGIN}/player/heartbeat`, { + method: 'POST', + headers: await bearer('701'), + }) + expect(await storedExpiresAt(701)).toBe(healthyExpiry) + }) + + test('countPlayersInInstance counts live players in a room instance (excludes expired)', async () => { + // Three players in instance 1000099 — two live, one expired. + const seedInInstance = (id: number, expiresAt: number) => + env.DB.prepare('INSERT OR REPLACE INTO presence (data) VALUES (?1)') + .bind( + JSON.stringify({ + accountId: id, + roomInstance: { roomInstanceId: 1000099, roomId: 2 }, + statusVisibility: 0, + deviceClass: 0, + vrMovementMode: 1, + platform: 0, + appVersion: '20230302', + expiresAt, + }) + ) + .run() + await seedInInstance(710, nowSeconds() + 800) + await seedInInstance(711, nowSeconds() + 800) + await seedInInstance(712, nowSeconds() - 10) // already expired → not counted + expect(await countPlayersInInstance(env.DB, 1000099)).toBe(2) + expect(await countPlayersInInstance(env.DB, 999999)).toBe(0) + }) + + // Matchmake into a room, returning the resulting instance id. + const matchmakeInto = async (room: string, sub: string): Promise => { + const res = (await ( + await exports.default.fetch(`${ORIGIN}/matchmake/${room}`, { + method: 'POST', + headers: await bearer(sub), + }) + ).json()) as { roomInstance: { roomInstanceId: number } } + return res.roomInstance.roomInstanceId + } + + test('matchmaking flags an instance full once it reaches capacity, and routes the next player elsewhere', async () => { + // SoloRoom (RoomId 5, MaxPlayers 1): one player fills its instance. + const first = await matchmakeInto('5', '820') + expect((await getRoomInstance(env.DB, first))?.isFull).toBe(true) + // A second player can't join the full instance — matchmaking makes a fresh one. + const second = await matchmakeInto('5', '821') + expect(second).not.toBe(first) + expect((await getRoomInstance(env.DB, second))?.isFull).toBe(true) + }) + + test('matchmaking leaves an instance not full below capacity', async () => { + // RecCenter (RoomId 2, MaxPlayers 12): one player does not fill it. + const instanceId = await matchmakeInto('2', '822') + expect((await getRoomInstance(env.DB, instanceId))?.isFull).toBe(false) + }) + + test('leaving a full instance clears its full flag', async () => { + // Fill SoloRoom, then the same player matchmakes into RecCenter — the SoloRoom + // instance they left should no longer be full. + const solo = await matchmakeInto('5', '823') + expect((await getRoomInstance(env.DB, solo))?.isFull).toBe(true) + await matchmakeInto('2', '823') + expect((await getRoomInstance(env.DB, solo))?.isFull).toBe(false) + }) + test('player/login, exclusivelogin and logout all preserve presence', async () => { const headers = await bearer('9') await exports.default.fetch(`${ORIGIN}/matchmake/dorm`, { method: 'POST', headers }) diff --git a/apps/match/wrangler.jsonc b/apps/match/wrangler.jsonc index 125c112..784ce8e 100644 --- a/apps/match/wrangler.jsonc +++ b/apps/match/wrangler.jsonc @@ -4,19 +4,10 @@ "main": "src/match.app.ts", "compatibility_date": "2025-09-20", "compatibility_flags": ["nodejs_compat"], - // Per-player presence store (room instance the player is currently in). Create - // with `wrangler kv namespace create RECFLARE_MATCH_PRESENCE`, then put the id in - // the root .env under RECFLARE_KV (see .env.example) — it is spliced into the - // "local" placeholder below at deploy time. - "kv_namespaces": [ - { - "binding": "RECFLARE_MATCH_PRESENCE", - "id": "local" - } - ], - // Shared `recflare` DB. Mostly read (room scenes for matchmaking); also writes a - // player's personal dorm room on first dorm entry (see getOrCreateDormRoom). The - // "local" placeholder is replaced with the real id from RECFLARE_D1 at deploy time. + // Shared `recflare` DB. Reads room scenes for matchmaking, writes a player's + // personal dorm room on first dorm entry (see getOrCreateDormRoom), and holds + // player presence (the `presence` table — owned/migrated by the `rooms` worker). + // The "local" placeholder is replaced with the real id from RECFLARE_D1 at deploy time. "d1_databases": [ { "binding": "DB", diff --git a/apps/rooms/migrations/0006_presence.sql b/apps/rooms/migrations/0006_presence.sql new file mode 100644 index 0000000..97eb3be --- /dev/null +++ b/apps/rooms/migrations/0006_presence.sql @@ -0,0 +1,19 @@ +-- Player presence — the room instance a player is currently in, plus the status +-- fields the match heartbeat echoes back. Stored as a JSON blob in `data` with +-- generated (virtual) columns for the fields we query on, the same pattern as the +-- rooms/room_instance tables. One row per account (unique `account_id`); writes +-- upsert via INSERT OR REPLACE. Rows carry an absolute `expires_at` (epoch +-- seconds) — reads filter expired rows out and a cleanup pass purges them. +-- Generated from packages/domain/src/presence-db.ts (PRESENCE_SCHEMA_DDL) — keep +-- in sync. Written by the match/auth workers, read by match/rooms. + +CREATE TABLE IF NOT EXISTS presence ( + data TEXT NOT NULL, + account_id INTEGER GENERATED ALWAYS AS (json_extract(data, '$.accountId')) VIRTUAL, + room_instance_id INTEGER GENERATED ALWAYS AS (json_extract(data, '$.roomInstance.roomInstanceId')) VIRTUAL, + room_id INTEGER GENERATED ALWAYS AS (json_extract(data, '$.roomInstance.roomId')) VIRTUAL, + expires_at INTEGER GENERATED ALWAYS AS (json_extract(data, '$.expiresAt')) VIRTUAL + ); +CREATE UNIQUE INDEX IF NOT EXISTS idx_presence_account ON presence (account_id); +CREATE INDEX IF NOT EXISTS idx_presence_room_instance ON presence (room_instance_id); +CREATE INDEX IF NOT EXISTS idx_presence_expires ON presence (expires_at); diff --git a/apps/rooms/src/context.ts b/apps/rooms/src/context.ts index 0a6da1e..87e8e14 100644 --- a/apps/rooms/src/context.ts +++ b/apps/rooms/src/context.ts @@ -9,12 +9,10 @@ export type Env = SharedHonoEnv & { // with `await env.JWT_SECRET.get()`; all workers bind the same store so tokens // signed by `auth` verify here. JWT_SECRET: SecretsStoreSecret - // D1 database holding rooms (JSON blob + generated columns). See rooms-db.ts. + // Shared `recflare` D1. Owns the rooms tables (JSON blob + generated columns, + // see rooms-db.ts) and migrates the shared `presence` table, which is read here + // to resolve the caller's current room instance for the photon access token. DB: D1Database - // Shared player-presence KV (owned by the `match` worker). Read here to resolve - // the caller's current room instance for the photon access token — the - // equivalent of the reference server's HeartbeatDB.GetPlayerHeartbeat. - RECFLARE_MATCH_PRESENCE: KVNamespace // SignalR notifications hub (DO owned by the `notify` worker). Bound here to // push RoomUpdate notifications when a room is mutated. RECFLARE_NOTIFICATIONS_HUB: DurableObjectNamespace diff --git a/apps/rooms/src/rooms.app.ts b/apps/rooms/src/rooms.app.ts index e0b1207..dfd7fb0 100644 --- a/apps/rooms/src/rooms.app.ts +++ b/apps/rooms/src/rooms.app.ts @@ -10,6 +10,7 @@ import { getFeaturedRooms, getHotRooms, getInteraction, + getPresence, getPublicRoomsByCreator, getRecommendedRooms, getRoomById, @@ -66,13 +67,9 @@ function allIds(idParam: string): number[] { * hardcoded moderator/dev accounts. */ const MAKER_PEN_ACCOUNT_IDS = new Set([1, 2, 3]) -/** Presence KV key (owned by the `match` worker); keep in sync with match's `presenceKey`. */ -const presenceKey = (id: number) => `presence:${id}` - -/** The slice of the match worker's presence record we read — the caller's current - * room instance. Mirrors the reference server's HeartbeatDB row. */ +/** The slice of the shared presence row we read — the caller's current room instance. */ interface PresenceView { - roomInstance?: { roomInstanceId?: number } | null + roomInstanceId?: number } /** @@ -117,15 +114,12 @@ function photonAccessToken(accountId: number, roomInstanceId: number | null) { /** * Photon access-token handler (served bare and under `/roomserver`). Auth-gated: * resolves the caller, reads their current room instance from the shared - * presence KV, and returns the permissions + token. + * `presence` table (see @repo/domain), and returns the permissions + token. */ async function handlePhotonAccessToken(c: Context) { const accountId = await authedAccountId(c) if (accountId === null) return unauthorized(c) - const presence = await c.env.RECFLARE_MATCH_PRESENCE.get( - presenceKey(accountId), - 'json' - ) + const presence = await getPresence(c.env.DB, accountId) const roomInstanceId = presence?.roomInstance?.roomInstanceId ?? null return c.json(photonAccessToken(accountId, roomInstanceId)) } diff --git a/apps/rooms/src/test/integration/api.test.ts b/apps/rooms/src/test/integration/api.test.ts index 54e1ad3..d98f3aa 100644 --- a/apps/rooms/src/test/integration/api.test.ts +++ b/apps/rooms/src/test/integration/api.test.ts @@ -6,6 +6,7 @@ import '../../rooms.app' import { createRoomInstance, getRoomInstance, + PRESENCE_SCHEMA_DDL, ROOM_INSTANCE_SCHEMA_DDL, ROOM_SCHEMA_DDL, } from '@repo/domain' @@ -50,6 +51,8 @@ beforeAll(async () => { await adminSecretsStore(env.JWT_SECRET).create('test-signing-key') for (const stmt of ROOM_SCHEMA_DDL) await env.DB.prepare(stmt).run() for (const stmt of ROOM_INSTANCE_SCHEMA_DDL) await env.DB.prepare(stmt).run() + // Presence table (read by the photon access-token handler). + for (const stmt of PRESENCE_SCHEMA_DDL) await env.DB.prepare(stmt).run() const insert = env.DB.prepare('INSERT OR IGNORE INTO room (data) VALUES (?1)') await env.DB.batch(importRooms.map((r) => insert.bind(JSON.stringify(r)))) }) @@ -785,10 +788,15 @@ describe('rooms endpoints', () => { it('GET /photon_access_token (bare + /roomserver) returns permissions + presence instance', async () => { // Seed the caller's presence so RoomInstanceId reflects their current instance. - await env.RECFLARE_MATCH_PRESENCE.put( - 'presence:777', - JSON.stringify({ roomInstance: { roomInstanceId: 1000042 } }) - ) + await env.DB.prepare('INSERT OR REPLACE INTO presence (data) VALUES (?1)') + .bind( + JSON.stringify({ + accountId: 777, + roomInstance: { roomInstanceId: 1000042 }, + expiresAt: Math.floor(Date.now() / 1000) + 900, + }) + ) + .run() const headers = await bearer('777') for (const path of ['/photon_access_token', '/roomserver/photon_access_token']) { const res = await SELF.fetch(`${ORIGIN}${path}`, { headers }) diff --git a/apps/rooms/wrangler.jsonc b/apps/rooms/wrangler.jsonc index e4bc784..9b948e0 100644 --- a/apps/rooms/wrangler.jsonc +++ b/apps/rooms/wrangler.jsonc @@ -19,14 +19,6 @@ "migrations_dir": "migrations" } ], - // Shared player-presence KV (owned by the `match` worker). Read-only here to - // resolve the caller's current room instance for the photon access token. - "kv_namespaces": [ - { - "binding": "RECFLARE_MATCH_PRESENCE", - "id": "local" - } - ], // Cross-worker binding to the SignalR notifications hub DO (owned/migrated by // the `notify` worker). We only invoke its RPC methods; no migration here. "durable_objects": { diff --git a/packages/domain/src/index.ts b/packages/domain/src/index.ts index 47097e5..5c27478 100644 --- a/packages/domain/src/index.ts +++ b/packages/domain/src/index.ts @@ -2,3 +2,4 @@ export { RoomInstanceType, Accessibility, Role } from './enums' export * from './accounts-db' export * from './rooms-db' export * from './room-instance-db' +export * from './presence-db' diff --git a/packages/domain/src/presence-db.ts b/packages/domain/src/presence-db.ts new file mode 100644 index 0000000..b480790 --- /dev/null +++ b/packages/domain/src/presence-db.ts @@ -0,0 +1,148 @@ +/** + * Player presence — the room instance a player is currently in, plus the status + * fields the match heartbeat echoes back. Stored on the shared `recflare` D1 with + * the same JSON-blob pattern as the rooms/room_instance tables: the full presence + * is a JSON blob in `data`, and the fields we query on (account_id, + * room_instance_id, room_id, expires_at) are SQLite generated (virtual) columns + * extracted from it. One row per account (unique `account_id`); writes upsert via + * `INSERT OR REPLACE`. + * + * The `match` worker owns presence — written on matchmake/heartbeat, read by the + * heartbeat and the batch `/player` lookup. The `auth` worker seeds it for new + * players (Orientation) and the `rooms` worker reads it (Photon access token). All + * three import these helpers from `@repo/domain`; the `rooms` worker owns the + * schema (migrations/0006_presence.sql). + * + * This replaces the old match-presence KV. D1 gives strong reads (no cross-PoP + * staleness that would read presence as out-of-sync and bounce the player), a + * single-query batch lookup for `/player`, and lets matchmaking count players per + * instance (see {@link countPlayersInInstance}). Rows carry an absolute + * `expiresAt` (epoch seconds); reads filter expired rows out and + * {@link deleteExpiredPresence} purges them. + */ + +/** Presence is kept this long (s) after the last matchmake/heartbeat refresh. */ +export const PRESENCE_TTL_SECONDS = 900 + +/** Schema DDL (mirror of migrations/0006_presence.sql). */ +export const PRESENCE_SCHEMA_DDL: string[] = [ + `CREATE TABLE IF NOT EXISTS presence ( + data TEXT NOT NULL, + account_id INTEGER GENERATED ALWAYS AS (json_extract(data, '$.accountId')) VIRTUAL, + room_instance_id INTEGER GENERATED ALWAYS AS (json_extract(data, '$.roomInstance.roomInstanceId')) VIRTUAL, + room_id INTEGER GENERATED ALWAYS AS (json_extract(data, '$.roomInstance.roomId')) VIRTUAL, + expires_at INTEGER GENERATED ALWAYS AS (json_extract(data, '$.expiresAt')) VIRTUAL + )`, + `CREATE UNIQUE INDEX IF NOT EXISTS idx_presence_account ON presence (account_id)`, + `CREATE INDEX IF NOT EXISTS idx_presence_room_instance ON presence (room_instance_id)`, + `CREATE INDEX IF NOT EXISTS idx_presence_expires ON presence (expires_at)`, +] + +/** + * The presence a caller writes — the room instance the player is in plus the + * status fields the heartbeat echoes. Generic over the room-instance shape so each + * worker keeps its own typing (`match` its full instance, `rooms` just the id). + */ +export interface PresenceInput { + accountId: number + roomInstance: TRoomInstance | null + statusVisibility: number + deviceClass: number + vrMovementMode: number + platform: number + appVersion: string +} + +/** A stored presence row — the input plus its absolute expiry (epoch seconds). */ +export interface StoredPresence extends PresenceInput { + expiresAt: number +} + +const nowSeconds = () => Math.floor(Date.now() / 1000) + +/** + * Upsert a player's presence, stamping a fresh absolute expiry (now + + * PRESENCE_TTL_SECONDS). One row per account: `INSERT OR REPLACE` resolves on the + * unique `account_id` index. Returns the stored row (with its new expiry). + */ +export async function setPresence( + db: D1Database, + input: PresenceInput +): Promise> { + const stored: StoredPresence = { + ...input, + expiresAt: nowSeconds() + PRESENCE_TTL_SECONDS, + } + await db + .prepare('INSERT OR REPLACE INTO presence (data) VALUES (?1)') + .bind(JSON.stringify(stored)) + .run() + return stored +} + +/** Read a player's live presence, or null when they're absent or expired. */ +export async function getPresence( + db: D1Database, + accountId: number, + now = nowSeconds() +): Promise | null> { + const row = await db + .prepare('SELECT data FROM presence WHERE account_id = ?1 AND expires_at > ?2') + .bind(accountId, now) + .first<{ data: string }>() + return row ? (JSON.parse(row.data) as StoredPresence) : null +} + +/** + * Read many players' live presence in one query, keyed by account id (absent or + * expired players are simply missing from the map). Replaces the N point reads the + * batch `/player?id=…` lookup did against KV. + */ +export async function getPresences( + db: D1Database, + accountIds: number[], + now = nowSeconds() +): Promise>> { + const out = new Map>() + if (accountIds.length === 0) return out + const placeholders = accountIds.map((_, i) => `?${i + 1}`).join(', ') + const { results } = await db + .prepare( + `SELECT data FROM presence + WHERE account_id IN (${placeholders}) AND expires_at > ?${accountIds.length + 1}` + ) + .bind(...accountIds, now) + .all<{ data: string }>() + for (const r of results) { + const p = JSON.parse(r.data) as StoredPresence + out.set(p.accountId, p) + } + return out +} + +/** + * How many players are currently in a room instance — the live head-count + * matchmaking can use to spread players and avoid full instances (something KV + * couldn't answer without scanning every key). Counts only unexpired presence. + */ +export async function countPlayersInInstance( + db: D1Database, + roomInstanceId: number, + now = nowSeconds() +): Promise { + const row = await db + .prepare('SELECT COUNT(*) AS n FROM presence WHERE room_instance_id = ?1 AND expires_at > ?2') + .bind(roomInstanceId, now) + .first<{ n: number }>() + return row?.n ?? 0 +} + +/** + * Purge expired presence rows — housekeeping only, since reads already ignore them + * (and `INSERT OR REPLACE` keeps a single row per account, so the table is bounded + * by account count). Returns the number of rows removed. + */ +export async function deleteExpiredPresence(db: D1Database, now = nowSeconds()): Promise { + const res = await db.prepare('DELETE FROM presence WHERE expires_at <= ?1').bind(now).run() + return res.meta.changes ?? 0 +} diff --git a/packages/domain/src/room-instance-db.ts b/packages/domain/src/room-instance-db.ts index ac92bb3..afaa460 100644 --- a/packages/domain/src/room-instance-db.ts +++ b/packages/domain/src/room-instance-db.ts @@ -12,6 +12,8 @@ * the client DTO (`toDto`). */ +import { countPlayersInInstance } from './presence-db' + /** Schema DDL (mirror of migrations/0004_room_instance.sql). */ export const ROOM_INSTANCE_SCHEMA_DDL: string[] = [ `CREATE TABLE IF NOT EXISTS room_instance ( @@ -201,6 +203,37 @@ export async function setRoomInstanceInProgress( return toDto(stored) } +/** + * Recompute an instance's `isFull` flag from live match presence: full once the + * number of players currently present in the instance reaches its `maxCapacity` + * (capacity 0 — unset — is never full). Rewrites the JSON blob (the generated + * `is_full` column follows it) only when the flag actually changes. Returns the + * new fullness, or null when the instance has no row (e.g. the synthetic + * dorm/orientation instances). Matchmaking calls this for the instance a player + * enters and the one they leave, so the flag matchmaking selects on stays accurate. + */ +export async function refreshInstanceFullness( + db: D1Database, + roomInstanceId: number +): Promise { + const row = await db + .prepare('SELECT data FROM room_instance WHERE id = ?1') + .bind(roomInstanceId) + .first<{ data: string }>() + if (!row) return null + const stored = parse(row.data) + const count = await countPlayersInInstance(db, roomInstanceId) + const isFull = stored.maxCapacity > 0 && count >= stored.maxCapacity + if (stored.isFull !== isFull) { + stored.isFull = isFull + await db + .prepare('UPDATE room_instance SET data = ?1 WHERE id = ?2') + .bind(JSON.stringify(stored), roomInstanceId) + .run() + } + return isFull +} + /** * The oldest joinable public instance of a room (not private, not full, joins * enabled, not already in progress), or null when there's none to join. Used by