diff --git a/DECISIONS.md b/DECISIONS.md index 2ac3a80..f84a980 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -347,3 +347,10 @@ Implementation decisions not spelled out in `memory-bank/implementation-plan.md` - **`WORKERS_ENABLED` (default `true`)** controls whether this process runs workers and cron schedules. Queues are always created and pg-boss always starts, so an API-only process can still enqueue arrival jobs. Tests set it to `false`. - **The queue starts in Fastify's `onReady` hook and stops (gracefully) in `onClose`,** before the database pool closes. - **The server listens on `0.0.0.0` (IPv4).** Clients that resolve `localhost` to IPv6 `::1` first (PowerShell, Node 17+ `fetch`) cannot reach it by that name; use `127.0.0.1`. To revisit when the client dev proxy is set up (Phase 8). + +## Step 7.2 + +- **One job per dispatch,** `resolve-arrivals` with `{ playerId, dispatchId }`, scheduled for the arrival time. The handler resolves everything due for that player under the player lock, the same code as lazy resolution, so it is a no-op when a request got there first. A job that fires early resolves nothing; the next request or the next due job picks the dispatch up. +- **Jobs are sent after the dispatch transaction commits.** A failed send is logged and the dispatch still succeeds, because lazy resolution on read is the safety net. +- **The app clock is `base clock + offset`** (`createOffsetClock`). The offset is always 0 except through `POST /api/test/clock-offset` (`{ offsetMs }`), which is registered only when `NODE_ENV` is not `production` and is not in the route catalog. Job timers use the base (real) clock: `startAfter = arrivalAt - offset`. +- **The job's handler reads time from the same clock as requests,** so tests run handlers directly with a controlled clock (decision 0.5.18). diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index 46b20e0..77844b1 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -1,16 +1,18 @@ import Fastify, { type FastifyInstance } from "fastify"; import { createAuth } from "./auth"; -import { type Clock, systemClock } from "./clock"; +import { type Clock, createOffsetClock, systemClock } from "./clock"; import type { ServerConfig } from "./config"; import { createDatabase, type Database } from "./db/client"; import { registerErrorHandler } from "./http/errors"; import { requirePlayerId } from "./http/session"; +import { appJobs } from "./jobs/definitions"; import { createPgBossQueue, type JobQueue } from "./jobs/queue"; -import type { AppContext } from "./routes/context"; import { registerBoardRoutes } from "./routes/boards"; +import type { AppContext } from "./routes/context"; import { registerDispatchRoutes } from "./routes/dispatches"; import { registerFleetRoutes } from "./routes/fleet"; import { registerPlayerRoutes } from "./routes/player"; +import { registerTestClockRoute } from "./routes/test-clock"; export interface AppDeps { clock?: Clock; @@ -25,13 +27,19 @@ export function buildApp(config: ServerConfig, deps: AppDeps = {}): FastifyInsta const database = deps.database ?? createDatabase(config.databaseUrl); const { db } = database; const auth = createAuth(db, { secret: config.authSecret, baseUrl: config.authUrl }); + const clock = createOffsetClock(deps.clock ?? systemClock); const jobs = deps.jobs ?? - createPgBossQueue(config.databaseUrl, { workers: config.workersEnabled, jobs: [] }); + createPgBossQueue(config.databaseUrl, { + workers: config.workersEnabled, + jobs: appJobs({ db, clock: clock.now }), + }); const ctx: AppContext = { db, auth, - clock: deps.clock ?? systemClock, + clock: clock.now, + // pg-boss timers run on the real clock, not the (possibly offset) game clock. + toBaseTime: clock.toBase, jobs, options: { requirePlayer: (request) => requirePlayerId(request, auth, db), @@ -57,6 +65,7 @@ export function buildApp(config: ServerConfig, deps: AppDeps = {}): FastifyInsta registerBoardRoutes(app, ctx); registerDispatchRoutes(app, ctx); registerFleetRoutes(app, ctx); + if (config.env !== "production") registerTestClockRoute(app, clock); return app; } diff --git a/apps/server/src/clock.ts b/apps/server/src/clock.ts index 40d3df8..1293e95 100644 --- a/apps/server/src/clock.ts +++ b/apps/server/src/clock.ts @@ -4,3 +4,27 @@ import type { RealTime } from "@deadhead/sim"; export type Clock = () => RealTime; export const systemClock: Clock = () => Date.now(); + +/** + * A clock that runs ahead of (or behind) its base clock by an adjustable offset. + * The test-only clock-offset endpoint moves it; production never does (Step 7.2). + */ +export interface OffsetClock { + now: Clock; + offsetMs(): number; + setOffset(ms: number): void; + /** Converts a time on this clock to the base clock, e.g. for pg-boss timers. */ + toBase(time: RealTime): RealTime; +} + +export function createOffsetClock(base: Clock): OffsetClock { + let offset = 0; + return { + now: () => base() + offset, + offsetMs: () => offset, + setOffset: (ms) => { + offset = ms; + }, + toBase: (time) => time - offset, + }; +} diff --git a/apps/server/src/jobs/definitions.ts b/apps/server/src/jobs/definitions.ts new file mode 100644 index 0000000..6e240ea --- /dev/null +++ b/apps/server/src/jobs/definitions.ts @@ -0,0 +1,30 @@ +import type { Clock } from "../clock"; +import type { Db } from "../db/client"; +import { withPlayerLock } from "../game/lock"; +import { resolveArrivedDispatches } from "../game/resolve"; +import type { JobDefinition } from "./queue"; + +export const RESOLVE_ARRIVALS_JOB = "resolve-arrivals"; + +export interface JobContext { + db: Db; + clock: Clock; +} + +/** + * Resolves a player's arrived dispatches, exactly like lazy resolution on read + * (Step 7.2). Scheduled for each dispatch's arrival; does nothing if a request + * already resolved it. + */ +export async function resolveArrivalsJob(ctx: JobContext, data: Record) { + const playerId = data.playerId; + if (typeof playerId !== "string") throw new Error("resolve-arrivals job needs a playerId"); + await withPlayerLock(ctx.db, playerId, (tx) => + resolveArrivedDispatches(tx, playerId, ctx.clock()), + ); +} + +/** Every background job the app runs. */ +export function appJobs(ctx: JobContext): JobDefinition[] { + return [{ name: RESOLVE_ARRIVALS_JOB, handler: (data) => resolveArrivalsJob(ctx, data) }]; +} diff --git a/apps/server/src/routes/context.ts b/apps/server/src/routes/context.ts index 6648c0b..5b5ee86 100644 --- a/apps/server/src/routes/context.ts +++ b/apps/server/src/routes/context.ts @@ -9,6 +9,8 @@ export interface AppContext { db: Db; auth: Auth; clock: Clock; + /** Converts an app-clock time to the real clock pg-boss uses (they differ only in tests). */ + toBaseTime: (time: number) => number; jobs: JobQueue; options: RouteOptions; } diff --git a/apps/server/src/routes/dispatches.ts b/apps/server/src/routes/dispatches.ts index ffe16b3..84a3279 100644 --- a/apps/server/src/routes/dispatches.ts +++ b/apps/server/src/routes/dispatches.ts @@ -2,13 +2,14 @@ import { routes } from "@deadhead/schema"; import type { FastifyInstance } from "fastify"; import { dispatchContract, dispatchReposition, dispatchTutorial } from "../game/dispatch"; import { registerRoute } from "../http/route"; +import { RESOLVE_ARRIVALS_JOB } from "../jobs/definitions"; import type { AppContext } from "./context"; import { withPlayer } from "./with-player"; /** Contract, tutorial and reposition dispatches. */ export function registerDispatchRoutes(app: FastifyInstance, ctx: AppContext): void { - registerRoute(app, routes.createDispatch, ctx.options, async (input, { playerId }) => - withPlayer(ctx, playerId, async (tx, now) => { + registerRoute(app, routes.createDispatch, ctx.options, async (input, { playerId, request }) => { + const result = await withPlayer(ctx, playerId, async (tx, now) => { switch (input.kind) { case "contract": return { dispatch: await dispatchContract(tx, playerId, input, now) }; @@ -17,6 +18,19 @@ export function registerDispatchRoutes(app: FastifyInstance, ctx: AppContext): v case "reposition": return { dispatch: await dispatchReposition(tx, playerId, input, now) }; } - }), - ); + }); + + // Resolve at arrival even if the player never comes back (Step 7.2). Sent after + // the dispatch commits; if sending fails, lazy resolution on read still covers it. + try { + await ctx.jobs.send( + RESOLVE_ARRIVALS_JOB, + { playerId, dispatchId: result.dispatch.id }, + ctx.toBaseTime(result.dispatch.arrivalAt), + ); + } catch (error) { + request.log.error({ err: error }, "Could not schedule the arrival job"); + } + return result; + }); } diff --git a/apps/server/src/routes/test-clock.ts b/apps/server/src/routes/test-clock.ts new file mode 100644 index 0000000..db85727 --- /dev/null +++ b/apps/server/src/routes/test-clock.ts @@ -0,0 +1,20 @@ +import type { FastifyInstance } from "fastify"; +import { z } from "zod"; +import type { OffsetClock } from "../clock"; +import { ApiError } from "../http/errors"; + +const bodySchema = z.object({ offsetMs: z.number().int() }).strict(); + +/** + * Test-only: shifts the server clock so E2E tests can fast-forward to arrivals + * (decision 0.5.18). Never registered in production; not part of the catalog. + */ +export function registerTestClockRoute(app: FastifyInstance, clock: OffsetClock): void { + app.post("/api/test/clock-offset", async (request) => { + const parsed = bodySchema.safeParse(request.body ?? {}); + if (!parsed.success) + throw new ApiError(400, "validation-error", "Expected { offsetMs: integer }."); + clock.setOffset(parsed.data.offsetMs); + return { offsetMs: clock.offsetMs(), realTime: clock.now() }; + }); +} diff --git a/apps/server/test/arrival-job.test.ts b/apps/server/test/arrival-job.test.ts new file mode 100644 index 0000000..1cc7ab3 --- /dev/null +++ b/apps/server/test/arrival-job.test.ts @@ -0,0 +1,120 @@ +import { eq } from "drizzle-orm"; +import { describe, expect, it } from "vitest"; +import { buildApp } from "../src/app"; +import { companies, dispatches, ships } from "../src/db/schema"; +import { RESOLVE_ARRIVALS_JOB, resolveArrivalsJob } from "../src/jobs/definitions"; +import { createRecordingQueue } from "../src/jobs/queue"; +import { makeOwner, setCredits, useTestDatabase } from "./db"; +import { call, getState, putBoard, signUp, testContract, useTestApp } from "./game"; +import { testConfig } from "./helpers"; + +const database = useTestDatabase(); +const testApp = useTestApp(database); + +/** An owner who has just dispatched a 2,000-credit Earth→Moon contract. */ +async function dispatched() { + const { app } = testApp(); + const { db } = database(); + const guest = await signUp(app); + const shipId = await makeOwner(db, guest.playerId, guest.avatarId); + await setCredits(db, guest.playerId, 10_000); + const contract = testContract(); + await putBoard(db, guest.playerId, [contract]); + const response = await call(app, guest, "POST", "/api/dispatches", { + kind: "contract", + contractId: contract.id, + shipId, + pilotId: guest.avatarId, + }); + const dispatch = response.json().dispatch as { id: string; arrivalAt: number; fuelCost: number }; + return { guest, shipId, dispatch }; +} + +const credits = async (playerId: string) => + (await database().db.select().from(companies).where(eq(companies.playerId, playerId)))[0]! + .credits; + +describe("arrival resolution job", () => { + it("is scheduled for the dispatch's arrival time", async () => { + const { jobs } = testApp(); + const { guest, dispatch } = await dispatched(); + expect(jobs.sent).toEqual([ + { + name: RESOLVE_ARRIVALS_JOB, + data: { playerId: guest.playerId, dispatchId: dispatch.id }, + startAfter: dispatch.arrivalAt, + }, + ]); + }); + + it("resolves the dispatch at arrival without any player state fetch", async () => { + const { clock, jobs } = testApp(); + const { db } = database(); + const { guest, shipId, dispatch } = await dispatched(); + const jobCtx = { db, clock: clock.read }; + + // Too early: nothing happens. + clock.now = dispatch.arrivalAt - 1; + await resolveArrivalsJob(jobCtx, jobs.sent[0]!.data); + expect((await db.select().from(dispatches))[0]!.status).toBe("in_flight"); + + clock.now = dispatch.arrivalAt; + await resolveArrivalsJob(jobCtx, jobs.sent[0]!.data); + const [row] = await db.select().from(dispatches).where(eq(dispatches.id, dispatch.id)); + expect(row!.status).toBe("resolved"); + expect(await credits(guest.playerId)).toBe(10_000 - dispatch.fuelCost + 2000); + const [ship] = await db.select().from(ships).where(eq(ships.id, shipId)); + expect(ship).toMatchObject({ status: "idle", locationId: "moon" }); + }); + + it("does nothing if lazy resolution already happened", async () => { + const { app, clock, jobs } = testApp(); + const { db } = database(); + const { guest, dispatch } = await dispatched(); + clock.now = dispatch.arrivalAt + 1000; + + const state = await getState(app, guest); + await resolveArrivalsJob({ db, clock: clock.read }, jobs.sent[0]!.data); + expect(await credits(guest.playerId)).toBe(state.company.credits); + expect((await getState(app, guest)).summaries).toHaveLength(1); + }); +}); + +describe("test-only clock offset endpoint", () => { + it("shifts the server clock outside production", async () => { + const { app, clock } = testApp(); + const response = await app.inject({ + method: "POST", + url: "/api/test/clock-offset", + payload: { offsetMs: 3_600_000 }, + }); + expect(response.statusCode).toBe(200); + const time = (await app.inject({ method: "GET", url: "/api/time" })).json(); + expect(time.realTime).toBe(clock.now + 3_600_000); + }); + + it("schedules arrival jobs on the real clock while an offset is set", async () => { + const { app, jobs } = testApp(); + await app.inject({ + method: "POST", + url: "/api/test/clock-offset", + payload: { offsetMs: 60_000 }, + }); + const { dispatch } = await dispatched(); + expect(jobs.sent[0]!.startAfter).toBe(dispatch.arrivalAt - 60_000); + }); + + it("does not exist in production", async () => { + const app = buildApp(testConfig({ NODE_ENV: "production" }), { + database: database(), + jobs: createRecordingQueue(), + }); + const response = await app.inject({ + method: "POST", + url: "/api/test/clock-offset", + payload: { offsetMs: 1 }, + }); + expect(response.statusCode).toBe(404); + await app.close(); + }); +}); diff --git a/apps/server/test/game.ts b/apps/server/test/game.ts index cfc2e33..a114075 100644 --- a/apps/server/test/game.ts +++ b/apps/server/test/game.ts @@ -29,6 +29,8 @@ export class TestClock { export interface TestApp { app: ReturnType; clock: TestClock; + /** Jobs the app scheduled; tests run their handlers directly. */ + jobs: ReturnType; } /** A fresh app per test on the shared test database, with a controllable clock. */ @@ -36,13 +38,11 @@ export function useTestApp(database: () => Database): () => TestApp { let current: TestApp | undefined; beforeEach(() => { const clock = new TestClock(); + const jobs = createRecordingQueue(); current = { - app: buildApp(testConfig(), { - database: database(), - clock: clock.read, - jobs: createRecordingQueue(), - }), + app: buildApp(testConfig(), { database: database(), clock: clock.read, jobs }), clock, + jobs, }; }); afterEach(() => current?.app.close()); diff --git a/memory-bank/progress.md b/memory-bank/progress.md index 02d876c..44c54e6 100644 --- a/memory-bank/progress.md +++ b/memory-bank/progress.md @@ -72,5 +72,5 @@ ## Phase 7 — Background Workers - [x] **Step 7.1 — pg-boss setup.** `src/jobs/queue.ts`: `JobQueue` interface, pg-boss implementation (queues always created; workers and cron schedules only when `WORKERS_ENABLED=true`), and a recording fake for tests. The app starts the queue on ready and stops it on close. Test runs a no-op job to completion with real pg-boss. -- [ ] Step 7.2 — Arrival resolution job +- [x] **Step 7.2 — Arrival resolution job.** Every dispatch (contract, tutorial, reposition) schedules a `resolve-arrivals` job at its arrival (`src/jobs/definitions.ts`), which runs the same locked resolution as lazy reads and is a no-op if already resolved. Clock is now base clock + offset; test-only `POST /api/test/clock-offset` (not registered in production) moves it. - [ ] Step 7.3 — Expired contract cleanup