Step 7.2: add arrival resolution job and test clock offset

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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).

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

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

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<string, unknown>) {
+  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) }];
+}

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

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 }) =>
+  registerRoute(app, routes.createDispatch, ctx.options, async (input, { playerId, request }) => {
-    withPlayer(ctx, playerId, async (tx, now) => {
+    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;
+  });
 }

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

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

apps/server/test/game.ts

@@ -29,6 +29,8 @@ export class TestClock {
 export interface TestApp {
   app: ReturnType<typeof buildApp>;
   clock: TestClock;
+  /** Jobs the app scheduled; tests run their handlers directly. */
+  jobs: ReturnType<typeof createRecordingQueue>;
 }
 
 /** 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(), {
+      app: buildApp(testConfig(), { database: database(), clock: clock.read, jobs }),
-        database: database(),
-        clock: clock.read,
-        jobs: createRecordingQueue(),
-      }),
       clock,
+      jobs,
     };
   });
   afterEach(() => current?.app.close());

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