diff --git a/CHANGELOG.md b/CHANGELOG.md index dd9a927a4..ddee70bfc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,10 @@ server accepts as `acceptedTypes`, taken from the same mapping the batch upload uses. The iOS app can then ask for and send only types the server stores, instead of reading a type the server would skip. (#1340) +- `GET /api/sync/changes?start=now` answers an empty page with a cursor + that starts at the present. The iOS app can page forward from it to + write readings to Apple Health in the background, without first walking + years of history. (#1339) ### Fixed diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 3d075ec93..e0fd36a86 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -4323,6 +4323,19 @@ paths: maxLength: 2048 description: Opaque multi-domain keyset cursor from the previous page. Treat as fully opaque — echo, never parse. Omit for the initial sync. + - in: query + name: start + schema: + description: "`now` answers an empty page (every domain empty, `hasMore: false`) whose `cursor` sits at `serverNow` in + every domain, in the same opaque format as any other page. Paging forward from it yields only what changes + from then on, including `measurementMirror`, without walking the history first. Mutually exclusive with + `cursor` (422). A server without it ignores the parameter and starts from the oldest row." + type: string + const: now + description: "`now` answers an empty page (every domain empty, `hasMore: false`) whose `cursor` sits at `serverNow` in + every domain, in the same opaque format as any other page. Paging forward from it yields only what changes + from then on, including `measurementMirror`, without walking the history first. Mutually exclusive with + `cursor` (422). A server without it ignores the parameter and starts from the oldest row." - in: query name: limit schema: diff --git a/src/app/api/sync/changes/__tests__/start-now.test.ts b/src/app/api/sync/changes/__tests__/start-now.test.ts new file mode 100644 index 000000000..240fd51fa --- /dev/null +++ b/src/app/api/sync/changes/__tests__/start-now.test.ts @@ -0,0 +1,100 @@ +/** + * `GET /api/sync/changes?start=now` (#1339) — a cursor at the present. + * + * The answer is an empty page whose cursor sits at `serverNow` in every + * domain, in the feed's own opaque format, and nothing is read: the point is + * to skip the history walk, so a read here would be the bug. The database is + * deliberately not mocked and not reachable: any read would fail the request, + * so the 200 below is itself the proof that none happened. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { NextRequest } from "next/server"; + +vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() })); +vi.mock("@/lib/rate-limit", () => ({ checkRateLimit: vi.fn() })); +vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() })); +vi.mock("@/lib/modules/gate", () => ({ isModuleEnabled: vi.fn() })); +vi.mock("next/headers", () => ({ + headers: vi.fn(async () => ({ get: () => null })), + cookies: vi.fn(async () => ({ + get: () => undefined, + set: () => {}, + delete: () => {}, + })), +})); + +import { GET } from "../route"; +import { getSession } from "@/lib/auth/session"; +import { checkRateLimit } from "@/lib/rate-limit"; +import { + decodeCursor, + decodeCursorIssuedAt, + SYNC_DOMAINS, +} from "@/lib/sync/cursor"; + +const SESSION_OK = { + session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) }, + user: { id: "user-1", username: "tester", role: "USER" as const }, +}; + +function req(query: string): NextRequest { + return new NextRequest(`http://localhost/api/sync/changes${query}`); +} + +beforeEach(() => { + vi.resetAllMocks(); + vi.mocked(getSession).mockResolvedValue( + SESSION_OK as Awaited>, + ); + vi.mocked(checkRateLimit).mockResolvedValue({ + allowed: true, + remaining: 119, + resetAt: Date.now() + 60_000, + } as Awaited>); +}); + +describe("GET /api/sync/changes?start=now", () => { + it("answers an empty page with a cursor at serverNow in every domain", async () => { + const before = Date.now(); + const res = await GET(req("?start=now")); + const after = Date.now(); + expect(res.status).toBe(200); + const { data } = (await res.json()) as { + data: { + serverNow: string; + cursor: string; + hasMore: boolean; + cursorExpired: boolean; + changes: Record; + }; + }; + expect(data.hasMore).toBe(false); + expect(data.cursorExpired).toBe(false); + expect(Object.keys(data.changes).sort()).toEqual([...SYNC_DOMAINS].sort()); + for (const block of Object.values(data.changes)) { + expect(block).toEqual({ upserts: [], tombstones: [] }); + } + const nowMs = Date.parse(data.serverNow); + expect(nowMs).toBeGreaterThanOrEqual(before); + expect(nowMs).toBeLessThanOrEqual(after); + const cursor = decodeCursor(data.cursor); + expect(cursor).not.toBeNull(); + for (const domain of SYNC_DOMAINS) { + expect(cursor![domain], domain).toEqual({ updatedAtMs: nowMs, id: "0" }); + } + expect(decodeCursorIssuedAt(data.cursor)).toBe(nowMs); + }); + + it("refuses start together with a cursor, and any start value but now", async () => { + const both = await GET(req("?start=now&cursor=abc")); + expect(both.status).toBe(422); + const other = await GET(req("?start=oldest")); + expect(other.status).toBe(422); + }); + + it("still requires a session", async () => { + vi.mocked(getSession).mockResolvedValue(null); + const res = await GET(req("?start=now")); + expect(res.status).toBe(401); + }); +}); diff --git a/src/app/api/sync/changes/route.ts b/src/app/api/sync/changes/route.ts index d714e7fff..76f0fbc8a 100644 --- a/src/app/api/sync/changes/route.ts +++ b/src/app/api/sync/changes/route.ts @@ -43,6 +43,11 @@ * cycle tracking off gets both blocks empty (never a 403 — see the * gate note in the handler) and their cursor watermarks untouched. * + * `start=now` (#1339) answers an empty page whose cursor sits at + * `serverNow` in every domain, in the same opaque format, so a client that + * only wants what changes from now on (the Apple Health write-back) pages + * forward from there instead of draining the whole history first. + * * `apiHandler` + `requireAuth` (cookie OR Bearer; iOS uses Bearer). * Read-only — no idempotency, no write side-effect (unlike the legacy * `/api/sync/state` checkpoint bump). The cursor is owned by the client. @@ -59,15 +64,23 @@ import { sanitiseZodIssues, } from "@/lib/api-response"; import { checkRateLimit } from "@/lib/rate-limit"; -import { pullSyncChanges } from "@/lib/sync/changes/pull"; +import { pullSyncChanges, syncChangesFromNow } from "@/lib/sync/changes/pull"; const DEFAULT_LIMIT = 200; const MAX_LIMIT = 500; -const querySchema = z.object({ - cursor: z.string().min(1).max(2048).optional(), - limit: z.coerce.number().int().min(1).max(MAX_LIMIT).optional(), -}); +// `start=now` (#1339) asks for a cursor at the present instead of a page; +// it contradicts an echoed cursor, so the two together are refused. +const querySchema = z + .object({ + cursor: z.string().min(1).max(2048).optional(), + limit: z.coerce.number().int().min(1).max(MAX_LIMIT).optional(), + start: z.literal("now").optional(), + }) + .refine((q) => !(q.start && q.cursor), { + message: "start and cursor are mutually exclusive", + path: ["start"], + }); export const GET = apiHandler(async (request: NextRequest) => { const { user } = await requireAuth(); @@ -83,6 +96,7 @@ export const GET = apiHandler(async (request: NextRequest) => { const parsed = querySchema.safeParse({ cursor: url.searchParams.get("cursor") ?? undefined, limit: url.searchParams.get("limit") ?? undefined, + start: url.searchParams.get("start") ?? undefined, }); if (!parsed.success) { return apiValidationError( @@ -92,6 +106,10 @@ export const GET = apiHandler(async (request: NextRequest) => { ); } + if (parsed.data.start === "now") { + return apiSuccess(syncChangesFromNow(new Date())); + } + return apiSuccess( await pullSyncChanges({ userId: user.id, diff --git a/src/lib/openapi/routes/sync.ts b/src/lib/openapi/routes/sync.ts index 5444c6aac..4293a2285 100644 --- a/src/lib/openapi/routes/sync.ts +++ b/src/lib/openapi/routes/sync.ts @@ -252,6 +252,12 @@ const syncChangesQuery = z .describe( "Opaque multi-domain keyset cursor from the previous page. Treat as fully opaque — echo, never parse. Omit for the initial sync.", ), + start: z + .literal("now") + .optional() + .describe( + "`now` answers an empty page (every domain empty, `hasMore: false`) whose `cursor` sits at `serverNow` in every domain, in the same opaque format as any other page. Paging forward from it yields only what changes from then on, including `measurementMirror`, without walking the history first. Mutually exclusive with `cursor` (422). A server without it ignores the parameter and starts from the oldest row.", + ), limit: z.coerce .number() .int() diff --git a/src/lib/sync/__tests__/cursor.test.ts b/src/lib/sync/__tests__/cursor.test.ts index 793a9f3c6..80e9bad68 100644 --- a/src/lib/sync/__tests__/cursor.test.ts +++ b/src/lib/sync/__tests__/cursor.test.ts @@ -7,6 +7,8 @@ import { encodeCursor, decodeCursor, decodeCursorIssuedAt, + encodeCursorAtNow, + SYNC_DOMAINS, type SyncCursor, } from "../cursor"; @@ -134,3 +136,19 @@ describe("cursor issue time (sync reset)", () => { expect(decodeCursorIssuedAt("not a token")).toBeNull(); }); }); + +describe("encodeCursorAtNow", () => { + it("places every domain at the instant, before any id in that millisecond", () => { + const nowMs = 1_790_000_000_123; + const token = encodeCursorAtNow(nowMs); + const cursor = decodeCursor(token); + for (const domain of SYNC_DOMAINS) { + expect(cursor?.[domain], domain).toEqual({ updatedAtMs: nowMs, id: "0" }); + } + expect(decodeCursorIssuedAt(token)).toBe(nowMs); + // Every cuid sorts after the tie-breaker, so a row in the same + // millisecond is delivered, not skipped. + expect("0" < "c").toBe(true); + expect("0" < "a").toBe(true); + }); +}); diff --git a/src/lib/sync/changes/pull.ts b/src/lib/sync/changes/pull.ts index cae7aed19..9a81488ef 100644 --- a/src/lib/sync/changes/pull.ts +++ b/src/lib/sync/changes/pull.ts @@ -13,6 +13,7 @@ import { decodeCursor, decodeCursorIssuedAt, encodeCursor, + encodeCursorAtNow, SYNC_DOMAINS, type SyncCursor, type SyncDomain, @@ -108,6 +109,26 @@ function emptyChanges(): SyncChangesBody["changes"] { }; } +/** + * `start=now` (#1339): an empty page with a cursor at `serverNow` in every + * domain, for a client that wants changes from now on (the Apple Health + * write-back) without paging through the history first. `hasMore` is false: + * the client is caught up as of `serverNow` by definition. + */ +export function syncChangesFromNow(serverNow: Date): SyncChangesBody { + annotate({ + action: { name: "sync.changes.start_now" }, + meta: { returned: 0 }, + }); + return { + serverNow: serverNow.toISOString(), + cursor: encodeCursorAtNow(serverNow.getTime()), + hasMore: false, + cursorExpired: false, + changes: emptyChanges(), + }; +} + export async function pullSyncChanges(input: { userId: string; /** The opaque cursor the client echoed, if any. */ diff --git a/src/lib/sync/cursor.ts b/src/lib/sync/cursor.ts index 4d8d1c161..7b3d12d4a 100644 --- a/src/lib/sync/cursor.ts +++ b/src/lib/sync/cursor.ts @@ -102,6 +102,22 @@ export function encodeCursor(cursor: SyncCursor, issuedAtMs?: number): string { return Buffer.from(json, "utf-8").toString("base64url"); } +/** + * A token positioned at `nowMs` in every domain (#1339), the same opaque + * format a drained page hands out: paging forward from it yields only what + * changes from now on, with no walk through the history first. The id `"0"` + * sorts before every cuid, so a row stamped in the very millisecond is + * delivered rather than skipped; the client's upserts are idempotent, so an + * overlap costs nothing and a gap would. + */ +export function encodeCursorAtNow(nowMs: number): string { + const cursor: SyncCursor = {}; + for (const domain of SYNC_DOMAINS) { + cursor[domain] = { updatedAtMs: nowMs, id: "0" }; + } + return encodeCursor(cursor, nowMs); +} + function parseWatermark(raw: unknown): DomainWatermark | null { if (typeof raw !== "object" || raw === null) return null; const u = (raw as { u?: unknown }).u; diff --git a/tests/integration/sync-changes-start-now.test.ts b/tests/integration/sync-changes-start-now.test.ts new file mode 100644 index 000000000..b29f9b9bd --- /dev/null +++ b/tests/integration/sync-changes-start-now.test.ts @@ -0,0 +1,150 @@ +/** + * `GET /api/sync/changes?start=now` (#1339), against Postgres. + * + * A client that only wants what changes from now on (the Apple Health + * write-back) asks for a cursor at the present, then pages forward. The + * history before that instant must not come back on the measurement walk, + * and switching a token's write-back on afterwards must still deliver all of + * that token's readings on `measurementMirror`, because the switch happens + * after the cursor. + */ +import { beforeEach, describe, expect, it } from "vitest"; + +import { GET as syncChanges } from "@/app/api/sync/changes/route"; +import { PATCH as patchToken } from "@/app/api/tokens/[id]/route"; +import { MEASUREMENTS_WRITE_SCOPE } from "@/lib/measurements/scopes"; + +import { callRoute, type RouteAuth } from "../kit/call-route"; +import { + makeMeasurement, + makeSession, + makeToken, + makeUser, + type BearerFixture, +} from "../kit/factories"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +const prisma = getPrismaClient(); + +interface Block { + upserts: Array<{ id: string; appleHealthMirror?: boolean }>; + tombstones: unknown[]; +} +interface Page { + serverNow: string; + cursor: string | null; + hasMore: boolean; + cursorExpired: boolean; + changes: Record; +} + +let userId: string; +let owner: RouteAuth; +let scale: BearerFixture; + +beforeEach(async () => { + await truncateAllTables(prisma); + const user = await makeUser({ timezone: "UTC" }); + userId = user.id; + owner = await makeSession(user); + scale = await makeToken(user, { + permissions: [MEASUREMENTS_WRITE_SCOPE], + name: "Scale", + }); +}); + +async function pull(query: Record) { + const res = await callRoute(syncChanges, { + method: "GET", + route: "/api/sync/changes", + query, + auth: owner, + }); + expect(res.status).toBe(200); + return res.data; +} + +async function pastMillisecond(): Promise { + const start = Date.now(); + while (Date.now() <= start + 1) await new Promise((r) => setTimeout(r, 2)); +} + +describe("a cursor that starts now", () => { + it("skips the history and pages forward to what changes after it", async () => { + const old = []; + for (let i = 0; i < 3; i += 1) { + old.push( + await makeMeasurement( + { id: userId }, + { + source: "EXTERNAL", + ingestTokenId: scale.tokenId, + measuredAt: new Date(Date.UTC(2026, 9, 1 + i, 6)), + externalId: `old-${i}`, + }, + ), + ); + } + await pastMillisecond(); + + const start = await pull({ start: "now" }); + expect(start.hasMore).toBe(false); + expect(start.cursorExpired).toBe(false); + expect(start.cursor).toEqual(expect.any(String)); + for (const block of Object.values(start.changes)) { + expect(block.upserts).toEqual([]); + } + + // Nothing changed yet: the next page is empty in every domain. + const quiet = await pull({ cursor: start.cursor! }); + for (const block of Object.values(quiet.changes)) { + expect(block.upserts).toEqual([]); + expect(block.tombstones).toEqual([]); + } + + await pastMillisecond(); + const fresh = await makeMeasurement( + { id: userId }, + { + source: "EXTERNAL", + ingestTokenId: scale.tokenId, + measuredAt: new Date(Date.UTC(2026, 10, 4, 6)), + externalId: "fresh", + }, + ); + const switched = await callRoute(patchToken, { + method: "PATCH", + route: "/api/tokens/{id}", + params: { id: scale.tokenId }, + body: { mirrorToAppleHealth: true }, + auth: owner, + }); + expect(switched.status).toBe(200); + + const next = await pull({ cursor: start.cursor! }); + expect(next.cursorExpired).toBe(false); + expect(next.changes.measurements.upserts.map((r) => r.id)).toEqual([ + fresh.id, + ]); + const mirrored = next.changes.measurementMirror.upserts; + expect(mirrored.map((r) => r.id).sort()).toEqual( + [...old.map((m) => m.id), fresh.id].sort(), + ); + expect(mirrored.every((r) => r.appleHealthMirror === true)).toBe(true); + + // And the cursor it hands back is a normal one: draining further is quiet. + const after = await pull({ cursor: next.cursor! }); + expect(after.changes.measurements.upserts).toEqual([]); + expect(after.changes.measurementMirror.upserts).toEqual([]); + }); + + it("refuses start=now beside a cursor", async () => { + const res = await callRoute(syncChanges, { + method: "GET", + route: "/api/sync/changes", + query: { start: "now", cursor: "abc" }, + auth: owner, + }); + expect(res.status).toBe(422); + }); +});