fix(journal): single-use JWT enforcement for route callback tokens (#2 Phase B)

After Phase A (#442) moved the journal callback token off the browser,
the token was still replayable on the wire until \`exp\` (7 days). This
PR makes each token strictly single-use.

Changes:

- **\`journal.consumed_jwt_jti\` table** — \`jti TEXT PRIMARY KEY,
  consumed_at TIMESTAMPTZ, expires_at TIMESTAMPTZ\`. Picked up by
  drizzle-kit push on deploy.
- **\`createRouteToken\` now sets a \`jti\` claim** (\`randomUUID()\`).
- **\`verifyRouteToken\` atomically consumes the jti** via
  \`INSERT … ON CONFLICT DO NOTHING RETURNING jti\`. Postgres
  serializes the insert, so exactly one concurrent caller wins; the
  rest see an empty result and throw \`TokenAlreadyConsumedError\`.
  Tokens without a \`jti\` claim (i.e. minted before this PR) are
  also rejected — the right call: any in-flight legacy token sitting
  in a planner session is replayable, and we'd rather fail-loud than
  silently grandfather them in.
- **\`consumed-jti-sweep\` job** — daily 03:45 UTC cron that
  \`DELETE WHERE expires_at < now()\`. Keeps the table tiny; offset
  from the other purge jobs to spread load.
- **e2e replay test** — \`integration.test.ts\` now exercises a
  same-token double-submit and asserts the second returns 401 with
  \`/consumed|already/i\`.

UX implication worth flagging: a user who clicks \"Save\" twice (or whose
network retries a failed POST) sees an error on the second attempt.
They go back to the journal for a fresh \"Edit in Planner\" link.

Full repo: pnpm typecheck / lint / test all green (177 + 31 integration).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

apps/journal/app/jobs/consumed-jti-sweep.ts

@@ -0,0 +1,31 @@
+import type { JobDefinition } from "@trails-cool/jobs";
+import { lt } from "drizzle-orm";
+import { consumedJwtJti } from "@trails-cool/db/schema/journal";
+import { getDb } from "../lib/db.ts";
+import { logger } from "../lib/logger.server.ts";
+
+/**
+ * Daily cleanup for the JWT replay-protection table. Each row was
+ * inserted by `verifyRouteToken` to mark a token as consumed; once
+ * the token's `exp` claim has passed, the row is no longer useful
+ * (the JWT itself would fail signature verification before reaching
+ * the consume step). Bound the table to keep it tiny.
+ *
+ * See planner-audit #2 Phase B.
+ */
+export const consumedJtiSweepJob: JobDefinition = {
+  name: "consumed-jti-sweep",
+  cron: "45 3 * * *", // daily at 03:45 UTC (offset from notifications-purge)
+  retryLimit: 1,
+  expireInSeconds: 60,
+  async handler() {
+    const db = getDb();
+    const result = await db
+      .delete(consumedJwtJti)
+      .where(lt(consumedJwtJti.expiresAt, new Date()))
+      .returning({ jti: consumedJwtJti.jti });
+    const purged = result.length;
+    logger.info({ purged }, "consumed-jti-sweep");
+    return { purged };
+  },
+};

apps/journal/app/lib/jwt.server.ts

@@ -1,5 +1,8 @@
 import { SignJWT, jwtVerify } from "jose";
+import { randomUUID } from "node:crypto";
 import { getOrigin, requireSecret } from "./config.server.ts";
+import { getDb } from "./db.ts";
+import { consumedJwtJti } from "@trails-cool/db/schema/journal";
 
 const JWT_SECRET = new TextEncoder().encode(
   requireSecret("JWT_SECRET", "dev-jwt-secret-change-in-production"),
@@ -13,14 +16,57 @@ export async function createRouteToken(routeId: string, permissions: string[] =
     .setIssuer(ISSUER)
     .setExpirationTime("7d")
     .setIssuedAt()
+    .setJti(randomUUID())
     .sign(JWT_SECRET);
 }
 
+export class TokenAlreadyConsumedError extends Error {
+  constructor() {
+    super("Token already consumed");
+    this.name = "TokenAlreadyConsumedError";
+  }
+}
+
+/**
+ * Verify a route token AND atomically consume it. Subsequent calls
+ * with the same token throw `TokenAlreadyConsumedError`.
+ *
+ * The consume step is `INSERT … ON CONFLICT DO NOTHING RETURNING jti`.
+ * Postgres serializes the insert against concurrent attempts, so
+ * exactly one caller observes the returned row — the rest see an
+ * empty result and reject.
+ *
+ * Tokens minted before this PR (no `jti` claim) are rejected outright,
+ * which is the right behavior: pre-existing tokens are already in
+ * planner-session DB rows and would be replayable if accepted. The
+ * user re-saves by going back to the journal for a fresh token.
+ */
 export async function verifyRouteToken(token: string): Promise<{ routeId: string; permissions: string[] }> {
   const { payload } = await jwtVerify(token, JWT_SECRET, {
     issuer: ISSUER,
   });
 
+  if (typeof payload.jti !== "string" || !payload.jti) {
+    throw new TokenAlreadyConsumedError();
+  }
+  if (typeof payload.exp !== "number") {
+    throw new TokenAlreadyConsumedError();
+  }
+
+  const db = getDb();
+  const inserted = await db
+    .insert(consumedJwtJti)
+    .values({
+      jti: payload.jti,
+      expiresAt: new Date(payload.exp * 1000),
+    })
+    .onConflictDoNothing()
+    .returning({ jti: consumedJwtJti.jti });
+
+  if (inserted.length === 0) {
+    throw new TokenAlreadyConsumedError();
+  }
+
   return {
     routeId: payload.route_id as string,
     permissions: payload.permissions as string[],

apps/journal/server.ts

@@ -176,8 +176,9 @@ server.listen(port, async () => {
   const { komootBulkImportJob } = await import("./app/jobs/komoot-bulk-import.ts");
   const { importBatchesSweepJob } = await import("./app/jobs/import-batches-sweep.ts");
   const { sendWelcomeEmailJob } = await import("./app/jobs/send-welcome-email.ts");
+  const { consumedJtiSweepJob } = await import("./app/jobs/consumed-jti-sweep.ts");
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  jobs.push(notificationsFanoutJob, notificationsPurgeJob, komootBulkImportJob as any, importBatchesSweepJob, sendWelcomeEmailJob);
+  jobs.push(notificationsFanoutJob, notificationsPurgeJob, komootBulkImportJob as any, importBatchesSweepJob, sendWelcomeEmailJob, consumedJtiSweepJob);
 
   const boss = createBoss(getDatabaseUrl());
   await startWorker(boss, jobs);

e2e/integration.test.ts

@@ -71,6 +71,28 @@ test.describe("Integration: Planner callback → geometry stored", () => {
     });
     expect(resp.status()).toBe(401);
   });
+
+  test("token is single-use — second submit is rejected (Phase B replay guard)", async ({ request }) => {
+    const seedResp = await request.post(`${JOURNAL}/api/e2e/seed`);
+    const { routeId, token } = await seedResp.json() as { routeId: string; token: string };
+
+    // First save succeeds.
+    const first = await request.post(`${JOURNAL}/api/routes/${routeId}/callback`, {
+      headers: { Authorization: `Bearer ${token}` },
+      data: { gpx: VALID_GPX },
+    });
+    expect(first.status()).toBe(200);
+
+    // Second attempt with the *same* token must fail — the jti has
+    // been consumed.
+    const second = await request.post(`${JOURNAL}/api/routes/${routeId}/callback`, {
+      headers: { Authorization: `Bearer ${token}` },
+      data: { gpx: VALID_GPX },
+    });
+    expect(second.status()).toBe(401);
+    const body = await second.json() as { error: string };
+    expect(body.error).toMatch(/consumed|already/i);
+  });
 });
 
 test.describe("Integration: Journal ↔ Planner handoff", () => {

packages/db/src/schema/journal.ts

@@ -356,3 +356,17 @@ export const notifications = journalSchema.table("notifications", {
     .on(t.recipientUserId, t.type, t.subjectId)
     .where(sql`${t.subjectId} IS NOT NULL`),
 }));
+
+// Single-use JWT enforcement for callback tokens (planner → journal
+// save flow). Each token carries a `jti` claim; the verifier inserts
+// into this table on first use. A second attempt hits the PK conflict
+// and the verifier rejects the token. Rows are swept after `expires_at`.
+// See spec / planner-audit #2 Phase B.
+export const consumedJwtJti = journalSchema.table("consumed_jwt_jti", {
+  jti: text("jti").primaryKey(),
+  consumedAt: timestamp("consumed_at", { withTimezone: true }).notNull().defaultNow(),
+  expiresAt: timestamp("expires_at", { withTimezone: true }).notNull(),
+}, (t) => ({
+  // Sweep runs `DELETE WHERE expires_at < now()` on a daily schedule.
+  expiresAtIdx: index("consumed_jwt_jti_expires_at_idx").on(t.expiresAt),
+}));