perf(sync): walk the mirror domain from the switched tokens

`/api/sync/changes` read the `measurementMirror` domain with one query
that ordered every token-linked reading by the joined token's
`mirror_changed_at`. No index on `measurements` can serve that order, so
every poll read and sorted every row of every token ever switched, even
when nothing had changed past the client's watermark.

The walk now reads the account's tokens with `mirror_changed_at` at or
past the watermark first. Nearly every poll finds none and stops there.
Otherwise it walks each change stamp's readings by
`(ingest_token_id, id)`, the order the index holds, with tokens that
share a stamp walked together so rows interleave by id as before. Rows,
order, page boundaries and the cursor are unchanged; the wire contract
is untouched.

EXPLAIN (ANALYZE, BUFFERS) on a seeded Postgres 16: one account, a
token with 300,000 readings switched at T0, a second token with 50
readings switched at T1, 900,000 untokened readings, page size 200.

  steady-state poll (watermark past every switch)
    before: 53 ms, Merge Join over ~300,000 index rows, 10,827 buffers
    after:  0.07 ms token lookup + 0.02 ms resume probe, 10 buffers
  first page after the switch
    before: 51 ms, top-N heapsort over ~300,000 rows, 10,948 buffers
    after:  0.08 ms + 0.17 ms index scan on (ingest_token_id, id)
  resume in the middle of the big token
    before: 53 ms
    after:  0.02 ms + 0.20 ms, Index Cond on (ingest_token_id, id)

A new integration file pages both forms side by side (tied stamps, a
watermark inside a tie, soft-deleted rows, another account's token,
page sizes from 1 to 100) and asserts identical pages, and that a poll
with nothing switched past its watermark reads no reading at all.

src/lib/sync/changes/read-rows.ts

@@ -28,12 +28,6 @@ const MEASUREMENT_SELECT = {
   ingestToken: APPLE_HEALTH_MIRROR_SELECT.ingestToken,
 } satisfies Prisma.MeasurementSelect;
 
-const MEASUREMENT_MIRROR_SELECT = {
-  id: true,
-  source: true,
-  ingestToken: { select: { mirrorToAppleHealth: true, mirrorChangedAt: true } },
-} satisfies Prisma.MeasurementSelect;
-
 const MOOD_SELECT = {
   id: true,
   date: true,
@@ -81,8 +75,15 @@ export type CycleDayRow = Prisma.CycleDayLogGetPayload<{
   include: typeof dayLogSymptomInclude;
 }>;
 export type CycleRow = Prisma.MenstrualCycleGetPayload<object>;
+/** A reading the mirror walk re-delivers, with its token's current setting. */
 export type MeasurementMirrorRow = Prisma.MeasurementGetPayload<{
-  select: typeof MEASUREMENT_MIRROR_SELECT;
+  select: {
+    id: true;
+    source: true;
+    ingestToken: {
+      select: { mirrorToAppleHealth: true; mirrorChangedAt: true };
+    };
+  };
 }>;
 
 /**
@@ -109,41 +110,94 @@ export interface DomainRows {
   measurementMirror: MeasurementMirrorRow[];
 }
 
+const MEASUREMENT_MIRROR_ROW_SELECT = {
+  id: true,
+  source: true,
+} satisfies Prisma.MeasurementSelect;
+
 /**
  * The live readings of every token whose Apple Health setting changed past
  * the watermark, in `(token.mirrorChangedAt, measurement.id)` order. Switching
  * the setting writes the token row only, so the readings keep their
  * `updatedAt` (nothing keyed on it recomputes) and arrive through this walk
  * instead of the measurement one.
+ *
+ * The walk starts from the tokens, not the readings. Ordering readings by a
+ * column of the joined token row cannot use any index on `measurements`, so
+ * the single-query form sorted every row of every token that had ever been
+ * switched, on every poll, even when nothing had changed since the client's
+ * watermark. Reading the switched tokens first is a lookup on a handful of
+ * rows; nearly every poll finds none and ends there. Otherwise the readings
+ * are walked one change stamp at a time, by `(ingest_token_id, id)`, which is
+ * the order the index already holds. Tokens that share a stamp are walked
+ * together so their rows interleave by id exactly as the one-query form
+ * ordered them.
+ *
+ * The token lookup filters on the token's owner. That is exact because a
+ * reading only ever links to a token of its own account: the link is written
+ * for a narrow `measurements:write` Bearer, which never acts on another record
+ * (`requireRecordAuth` refuses the carrier), and a restore keeps the link only
+ * for a token the restoring account holds.
  */
-function readMeasurementMirrorRows(
+async function readMeasurementMirrorRows(
   userId: string,
   wm: DomainWatermark | undefined,
   limit: number,
 ): Promise<MeasurementMirrorRow[]> {
   const at = wm ? new Date(wm.updatedAtMs) : null;
-  return prisma.measurement.findMany({
+  const tokens = await prisma.apiToken.findMany({
     where: {
       userId,
-      deletedAt: null,
-      ingestToken: { mirrorChangedAt: { not: null } },
-      ...(at &&
-        wm && {
-          OR: [
-            { ingestToken: { mirrorChangedAt: { gt: at } } },
-            {
-              AND: [
-                { ingestToken: { mirrorChangedAt: at } },
-                { id: { gt: wm.id } },
-              ],
-            },
-          ],
-        }),
+      mirrorChangedAt: at ? { gte: at } : { not: null },
     },
-    orderBy: [{ ingestToken: { mirrorChangedAt: "asc" } }, { id: "asc" }],
-    take: limit + 1,
-    select: MEASUREMENT_MIRROR_SELECT,
+    orderBy: [{ mirrorChangedAt: "asc" }, { id: "asc" }],
+    select: { id: true, mirrorToAppleHealth: true, mirrorChangedAt: true },
   });
+  if (tokens.length === 0) return [];
+
+  // One group per change stamp, in stamp order.
+  const groups: { stampMs: number; tokens: typeof tokens }[] = [];
+  for (const token of tokens) {
+    const stampMs = token.mirrorChangedAt!.getTime();
+    const last = groups[groups.length - 1];
+    if (last && last.stampMs === stampMs) last.tokens.push(token);
+    else groups.push({ stampMs, tokens: [token] });
+  }
+
+  const out: MeasurementMirrorRow[] = [];
+  for (const group of groups) {
+    const want = limit + 1 - out.length;
+    if (want <= 0) break;
+    // The watermark's own stamp resumes after the last row delivered; every
+    // later stamp starts from its first row.
+    const resume = wm && group.stampMs === wm.updatedAtMs ? wm.id : null;
+    const byId = new Map(group.tokens.map((t) => [t.id, t]));
+    const rows = await prisma.measurement.findMany({
+      where: {
+        ingestTokenId:
+          group.tokens.length === 1
+            ? group.tokens[0].id
+            : { in: group.tokens.map((t) => t.id) },
+        userId,
+        deletedAt: null,
+        ...(resume !== null && { id: { gt: resume } }),
+      },
+      orderBy: { id: "asc" },
+      take: want,
+      select: { ...MEASUREMENT_MIRROR_ROW_SELECT, ingestTokenId: true },
+    });
+    for (const { ingestTokenId, ...row } of rows) {
+      const token = byId.get(ingestTokenId!)!;
+      out.push({
+        ...row,
+        ingestToken: {
+          mirrorToAppleHealth: token.mirrorToAppleHealth,
+          mirrorChangedAt: token.mirrorChangedAt,
+        },
+      });
+    }
+  }
+  return out;
 }
 
 /**

tests/integration/sync-mirror-walk.test.ts

@@ -0,0 +1,215 @@
+/**
+ * The `measurementMirror` walk of `/api/sync/changes`, against Postgres.
+ *
+ * The walk reads the switched tokens first and then each change stamp's
+ * readings by `(ingest_token_id, id)`. The page sequence a client sees must be
+ * exactly the one the earlier single query produced, which ordered every
+ * token-linked reading by `(token.mirrorChangedAt, measurement.id)` and
+ * resumed past `(stamp, id)`. That query is kept below as the reference, and
+ * every page of every walk is compared against it: tied stamps across tokens,
+ * a watermark inside a tie, soft-deleted rows, another account's token, a
+ * token never switched, readings with no token, and page sizes from one row
+ * up to the whole set.
+ */
+import { beforeEach, describe, expect, it } from "vitest";
+
+import { readDomainRows } from "@/lib/sync/changes/read-rows";
+import type { SyncCursor } from "@/lib/sync/cursor";
+
+import { makeMeasurement, makeToken, makeUser } from "../kit/factories";
+import { countStatements } from "./query-counter";
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+const prisma = getPrismaClient();
+
+interface Wm {
+  updatedAtMs: number;
+  id: string;
+}
+
+/** The single-query walk this replaced, verbatim in its semantics. */
+async function referenceWalk(
+  userId: string,
+  wm: Wm | undefined,
+  limit: number,
+) {
+  const at = wm ? new Date(wm.updatedAtMs) : null;
+  return prisma.measurement.findMany({
+    where: {
+      userId,
+      deletedAt: null,
+      ingestToken: { mirrorChangedAt: { not: null } },
+      ...(at &&
+        wm && {
+          OR: [
+            { ingestToken: { mirrorChangedAt: { gt: at } } },
+            {
+              AND: [
+                { ingestToken: { mirrorChangedAt: at } },
+                { id: { gt: wm.id } },
+              ],
+            },
+          ],
+        }),
+    },
+    orderBy: [{ ingestToken: { mirrorChangedAt: "asc" } }, { id: "asc" }],
+    take: limit + 1,
+    select: {
+      id: true,
+      source: true,
+      ingestToken: {
+        select: { mirrorToAppleHealth: true, mirrorChangedAt: true },
+      },
+    },
+  });
+}
+
+async function walk(userId: string, wm: Wm | undefined, limit: number) {
+  const cursor = { v: 2, measurementMirror: wm } as unknown as SyncCursor;
+  const rows = await readDomainRows(userId, cursor, limit, false);
+  return rows.measurementMirror;
+}
+
+/** Page through both walks from `start`, asserting each page is identical. */
+async function pageThrough(userId: string, limit: number, start?: Wm) {
+  let wm = start;
+  const seen: string[] = [];
+  for (let guard = 0; guard < 200; guard += 1) {
+    const [actual, expected] = await Promise.all([
+      walk(userId, wm, limit),
+      referenceWalk(userId, wm, limit),
+    ]);
+    expect(actual, `limit ${limit}, page ${guard}`).toEqual(expected);
+    const page = actual.slice(0, limit);
+    seen.push(...page.map((r) => r.id));
+    if (actual.length <= limit) return { seen, wm };
+    const last = page[page.length - 1];
+    wm = {
+      updatedAtMs: last.ingestToken!.mirrorChangedAt!.getTime(),
+      id: last.id,
+    };
+  }
+  throw new Error("walk did not terminate");
+}
+
+let userId: string;
+let otherId: string;
+const T0 = new Date("2026-10-01T08:00:00.000Z");
+const T1 = new Date("2026-10-02T08:00:00.000Z");
+const T2 = new Date("2026-10-03T08:00:00.000Z");
+
+async function rowsFor(
+  owner: { id: string },
+  tokenId: string | null,
+  n: number,
+  base: number,
+  deleted: number[] = [],
+) {
+  for (let i = 0; i < n; i += 1) {
+    await makeMeasurement(owner, {
+      type: "WEIGHT",
+      unit: "kg",
+      value: 80 + i / 10,
+      source: "EXTERNAL",
+      measuredAt: new Date(Date.UTC(2026, 8, 1, 0, base + i)),
+      ingestTokenId: tokenId,
+      deletedAt: deleted.includes(i) ? new Date() : null,
+    });
+  }
+}
+
+async function stamp(tokenId: string, at: Date, on = true) {
+  await prisma.apiToken.update({
+    where: { id: tokenId },
+    data: { mirrorChangedAt: at, mirrorToAppleHealth: on },
+  });
+}
+
+beforeEach(async () => {
+  await truncateAllTables(prisma);
+  const user = await makeUser({ timezone: "UTC" });
+  const other = await makeUser({ timezone: "UTC" });
+  userId = user.id;
+  otherId = other.id;
+  const [a, b, c, d, never, othersOwn] = await Promise.all([
+    makeToken(user, { name: "A" }),
+    makeToken(user, { name: "B" }),
+    makeToken(user, { name: "C" }),
+    makeToken(user, { name: "D" }),
+    makeToken(user, { name: "never" }),
+    makeToken(other, { name: "others-own" }),
+  ]);
+  await rowsFor(user, a.tokenId, 7, 0, [3]);
+  await rowsFor(user, b.tokenId, 5, 100, [0]);
+  await rowsFor(user, c.tokenId, 3, 200);
+  await rowsFor(user, never.tokenId, 3, 300);
+  await rowsFor(user, d.tokenId, 2, 400);
+  await rowsFor(other, othersOwn.tokenId, 3, 600);
+  await rowsFor(user, null, 4, 700);
+  // A and B share a stamp, so their rows interleave by id.
+  await stamp(a.tokenId, T0);
+  await stamp(b.tokenId, T0, false);
+  await stamp(c.tokenId, T1);
+  await stamp(d.tokenId, T2);
+  await stamp(othersOwn.tokenId, T1);
+});
+
+describe("the measurementMirror walk", () => {
+  it.each([1, 2, 3, 5, 8, 100])(
+    "pages exactly as the single query did, %i rows a page",
+    async (limit) => {
+      const { seen } = await pageThrough(userId, limit);
+      // 6 + 4 live rows on the tied stamp, 3 on C, 2 on D. Nothing from the
+      // never-switched token, nothing of the other account's, nothing
+      // deleted, nothing without a token.
+      expect(seen).toHaveLength(15);
+      expect(new Set(seen).size).toBe(15);
+    },
+  );
+
+  it("resumes inside a tied stamp, and re-delivers a token switched again", async () => {
+    const all = await walk(userId, undefined, 100);
+    const tied = all.filter(
+      (r) => r.ingestToken!.mirrorChangedAt!.getTime() === T0.getTime(),
+    );
+    // Both tied tokens' rows interleave by id, so a watermark three rows into
+    // the tie leaves rows of both still to come.
+    await pageThrough(userId, 2, { updatedAtMs: T0.getTime(), id: tied[2].id });
+
+    const last = all[all.length - 1];
+    const caughtUp: Wm = {
+      updatedAtMs: last.ingestToken!.mirrorChangedAt!.getTime(),
+      id: last.id,
+    };
+    expect(await walk(userId, caughtUp, 100)).toEqual([]);
+
+    const c = await prisma.apiToken.findFirstOrThrow({
+      where: { userId, name: "C" },
+    });
+    await stamp(c.id, new Date("2026-10-04T08:00:00.000Z"), false);
+    const { seen } = await pageThrough(userId, 2, caughtUp);
+    expect(seen).toHaveLength(3);
+  });
+
+  it("answers a poll with nothing switched past its watermark from the token table alone", async () => {
+    const wm = { updatedAtMs: T2.getTime() + 1, id: "" };
+    const measured = await countStatements(() => walk(userId, wm, 100));
+    expect(measured.result).toEqual([]);
+    // The walk's reading-side shape: id, source and the token link.
+    const readingWalks = measured.statements.filter((s) =>
+      s.startsWith(
+        'SELECT "public"."measurements"."id", "public"."measurements"."source"::text, "public"."measurements"."ingest_token_id"',
+      ),
+    );
+    expect(readingWalks).toEqual([]);
+    const tokenLookups = measured.statements.filter((s) =>
+      s.includes('"public"."api_tokens"."mirror_changed_at" FROM'),
+    );
+    expect(tokenLookups).toHaveLength(1);
+  });
+
+  it("leaves the other account's walk to the other account", async () => {
+    const { seen } = await pageThrough(otherId, 2);
+    expect(seen).toHaveLength(3);
+  });
+});