feat(sync): let a client start the delta feed at the present

GET /api/sync/changes?start=now answers an empty page whose cursor sits at
serverNow in every domain, in the same opaque format the feed hands out.
A client that only needs what changes from now on, such as the Apple
Health write-back, pages forward from it instead of draining the whole
history first. start and cursor together are refused with 422.

Refs #1339
This commit is contained in:
Marc-André Bombeck committed 2026-11-04 21:24:12 +01:00
1 parent 59434cd463
commit a79fc4168a
9 files changed
+351 -5

No files matched your search

+4
View File
@@ -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
+13
View File
@@ -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:
@@ -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<ReturnType<typeof getSession>>,
);
vi.mocked(checkRateLimit).mockResolvedValue({
allowed: true,
remaining: 119,
resetAt: Date.now() + 60_000,
} as Awaited<ReturnType<typeof checkRateLimit>>);
});
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<string, { upserts: unknown[]; tombstones: unknown[] }>;
};
};
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);
});
});
+23 -5
View File
@@ -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,
+6
View File
@@ -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()
+18
View File
@@ -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);
});
});
+21
View File
@@ -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. */
+16
View File
@@ -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;
@@ -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<string, Block>;
}
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<string, string>) {
const res = await callRoute<Page>(syncChanges, {
method: "GET",
route: "/api/sync/changes",
query,
auth: owner,
});
expect(res.status).toBe(200);
return res.data;
}
async function pastMillisecond(): Promise<void> {
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);
});
});