diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 1ba6ce29e..063be750d 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -40212,7 +40212,8 @@ components: description: False once the client is caught up as of `serverNow`. cursorExpired: type: boolean - description: True when the supplied cursor predates tombstone retention — drop the cursor and do a clean initial sync. + description: True when the supplied cursor predates tombstone retention, or was issued before the account's data was + last restored from a backup or deleted — drop the cursor and do a clean initial sync. changes: type: object properties: diff --git a/prisma/migrations/0363_user_sync_reset_at/migration.sql b/prisma/migrations/0363_user_sync_reset_at/migration.sql new file mode 100644 index 000000000..49ff5f8be --- /dev/null +++ b/prisma/migrations/0363_user_sync_reset_at/migration.sql @@ -0,0 +1,6 @@ +-- When an account's record was last replaced wholesale (a backup restore, or +-- the person deleting all their data). `/api/sync/changes` expires every +-- delta cursor issued before it, so a paired client re-initialises instead +-- of catching up from a position the restore invalidated. Nullable: an +-- account that was never restored or wiped has no reset to honour. +ALTER TABLE "users" ADD COLUMN IF NOT EXISTS "sync_reset_at" TIMESTAMP(3); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 4ff13f258..582310695 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -503,6 +503,14 @@ model User { // instead of running ahead of one. lastSyncedAt DateTime? @map("last_synced_at") + // When the account's record was last replaced wholesale: a backup restore + // committed, or the person deleted all their data. `/api/sync/changes` + // answers `cursorExpired` to any cursor issued before it, because such a + // cursor can no longer be caught up incrementally: a restore writes rows + // back with their original `updatedAt` and removes rows without leaving a + // tombstone, so the delta from an old cursor would miss both. + syncResetAt DateTime? @map("sync_reset_at") + // v0.5.4 ios-coord — daily mood-reminder opt-in. // // When `true` the reminder worker dispatches a `MOOD_REMINDER` diff --git a/src/app/api/admin/data/route.ts b/src/app/api/admin/data/route.ts index 68cdbc007..76e5b2528 100644 --- a/src/app/api/admin/data/route.ts +++ b/src/app/api/admin/data/route.ts @@ -113,7 +113,11 @@ export const DELETE = apiHandler(async (request: NextRequest) => { // Every account row survives; the personal data carried on its own // columns does not. Same classification contract as the model list. - await tx.user.updateMany({ data: USER_RESET }); + // Paired clients drop their delta cursors and start over, instead of + // catching up from a position whose rows are gone (`syncResetAt`). + await tx.user.updateMany({ + data: { ...USER_RESET, syncResetAt: new Date() }, + }); return perModel; }, diff --git a/src/app/api/settings/data/route.ts b/src/app/api/settings/data/route.ts index 2c3187e0f..bc9485205 100644 --- a/src/app/api/settings/data/route.ts +++ b/src/app/api/settings/data/route.ts @@ -23,6 +23,7 @@ import { wipeDelegateKey, WIPE_MODELS, } from "@/lib/data-wipe/wipe-plan"; +import { stampSyncReset } from "@/lib/sync/reset"; export const dynamic = "force-dynamic"; @@ -104,6 +105,9 @@ export const DELETE = apiHandler(async (request: NextRequest) => { // The account row survives; the personal data carried on its own columns // does not. Same classification contract as the model list. await tx.user.update({ where: { id: userId }, data: USER_RESET }); + // Paired clients drop their delta cursors and start over, instead of + // catching up from a position whose rows are gone. + await stampSyncReset(tx, userId); // The nightly off-host copies of the record are part of "all your // data". The request commits with the wipe; the purge job deletes diff --git a/src/app/api/sync/changes/route.ts b/src/app/api/sync/changes/route.ts index 694929cc0..810cf3460 100644 --- a/src/app/api/sync/changes/route.ts +++ b/src/app/api/sync/changes/route.ts @@ -34,7 +34,9 @@ * - `cursorExpired: true` when ANY domain watermark predates the * tombstone-retention horizon — the client drops its cursor and does * a clean initial sync (a deletion older than retention may have - * been pruned, so an incremental delta could silently miss it). + * been pruned, so an incremental delta could silently miss it) — or + * when the cursor was issued before the account's `syncResetAt` (a + * backup restore or a delete-all-data since, see the schema note). * - `syncVersion` is echoed per row so the client can keep its mirror's * version monotonic. * - The `cycleDays` / `cycles` blocks are MODULE-GATED: an account with @@ -64,6 +66,7 @@ import { contextForWire, type MoodContextWire } from "@/lib/mood/context"; import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client"; import { decodeCursor, + decodeCursorIssuedAt, encodeCursor, SYNC_DOMAINS, type DomainWatermark, @@ -271,14 +274,39 @@ export const GET = apiHandler(async (request: NextRequest) => { // A stale cursor (ANY domain watermark older than retention) forces a // clean re-init: a deletion older than retention may already be pruned. - const expired = SYNC_DOMAINS.some((domain) => { + const pastRetention = SYNC_DOMAINS.some((domain) => { const wm = cursor[domain]; return wm !== undefined && wm.updatedAtMs < retentionHorizon.getTime(); }); + // A cursor issued before the account's record was last replaced (a backup + // restore, or the person deleting all their data) cannot be caught up + // either: the restore wrote rows back with their original `updatedAt`, + // behind the cursor's watermarks, and removed rows without tombstones. The + // client re-initialises instead. Compared by the time the cursor was + // issued, not by its watermarks, which a restore leaves meaningless; a + // cursor from before the issue time was recorded counts as issued before + // any reset. + let beforeReset = false; + if (parsed.data.cursor) { + const account = await prisma.user.findUnique({ + where: { id: user.id }, + select: { syncResetAt: true }, + }); + if (account?.syncResetAt) { + const issuedAt = decodeCursorIssuedAt(parsed.data.cursor); + beforeReset = + issuedAt === null || issuedAt < account.syncResetAt.getTime(); + } + } + const expired = pastRetention || beforeReset; if (expired) { annotate({ action: { name: "sync.changes.pull" }, - meta: { cursor_expired: true, returned: 0 }, + meta: { + cursor_expired: true, + cursor_expired_reason: beforeReset ? "sync_reset" : "retention", + returned: 0, + }, }); return apiSuccess({ serverNow: serverNow.toISOString(), @@ -552,7 +580,7 @@ export const GET = apiHandler(async (request: NextRequest) => { return apiSuccess({ serverNow: serverNow.toISOString(), - cursor: encodeCursor(nextCursor), + cursor: encodeCursor(nextCursor, serverNow.getTime()), hasMore, cursorExpired: false, changes: { diff --git a/src/lib/data-wipe/wipe-plan.ts b/src/lib/data-wipe/wipe-plan.ts index 3e91d5379..d6769473b 100644 --- a/src/lib/data-wipe/wipe-plan.ts +++ b/src/lib/data-wipe/wipe-plan.ts @@ -469,6 +469,8 @@ export const USER_KEPT_FIELDS: Readonly> = { "the version of the medical disclaimer the account acknowledged", documentQuotaBytes: "a per-account storage quota set by the operator, not by the person", + syncResetAt: + "sync bookkeeping, not record content: the wipe sets it to its own time rather than clearing it, so a paired client's delta cursor from before the wipe is refused and the client starts over", }; /** diff --git a/src/lib/export/__tests__/restore-backup-round-trip.test.ts b/src/lib/export/__tests__/restore-backup-round-trip.test.ts index 95cb9c53a..991538efe 100644 --- a/src/lib/export/__tests__/restore-backup-round-trip.test.ts +++ b/src/lib/export/__tests__/restore-backup-round-trip.test.ts @@ -759,7 +759,10 @@ describe("backup round trip — export, wire schema, restore", () => { }); const userUpdate = written.find( - (w) => w.model === "user" && w.op === "update", + (w) => + w.model === "user" && + w.op === "update" && + "insightsCachedText" in w.data, ); expect( userUpdate, @@ -798,8 +801,13 @@ describe("backup round trip — export, wire schema, restore", () => { // whether the restore over-invalidates on a genuine no-op. const { written: writtenUnchanged } = await roundTrip(liveSeed, payload); + // The restore also stamps the account's sync reset; only an update + // that touches the cached briefing is what this asks about. const userUpdate = writtenUnchanged.find( - (w) => w.model === "user" && w.op === "update", + (w) => + w.model === "user" && + w.op === "update" && + "insightsCachedText" in w.data, ); expect( userUpdate, diff --git a/src/lib/export/restore-backup.ts b/src/lib/export/restore-backup.ts index 2cf75bdf4..86e6f0a15 100644 --- a/src/lib/export/restore-backup.ts +++ b/src/lib/export/restore-backup.ts @@ -103,6 +103,7 @@ import { restoreEcgData } from "@/lib/export/ecg-backup"; import { restoredMedicationCreatedAt } from "@/lib/export/medication-created-at"; import { invalidateUserData } from "@/lib/cache/invalidate"; import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client"; +import { stampSyncReset } from "@/lib/sync/reset"; import { classifyRestoreFailure, RESTORE_FAILURE_CAUSE_MESSAGES, @@ -2412,6 +2413,11 @@ export async function restoreBackup( foreignBefore, await countForeignReferences(tx, ownerId, tenantEdges), ); + // Paired clients' delta cursors are no longer valid (see + // `User.syncResetAt`). Stamped as the last write, so it commits with + // the data, and stamped again right after the commit below to cover + // a cursor issued while this transaction was still running. + await stampSyncReset(tx, ownerId); progress.sectionsDone = RESTORE_SECTION_STEPS.length; report("sections", true); return { @@ -2479,6 +2485,22 @@ export async function restoreBackup( ); } + // A pull that started before the commit read the pre-restore rows and may + // have been issued a cursor after the stamp inside the transaction. The + // second stamp expires that one too; it can only expire more cursors, so a + // failure here costs a client at most one extra incremental page. + try { + await stampSyncReset(prisma, ownerId); + } catch (err) { + annotate({ + meta: { + restore_sync_reset_failed: true, + restore_sync_reset_error: + err instanceof Error ? err.message : String(err), + }, + }); + } + const { cleared, skipped, expiredTombstonesSkipped } = outcome; const measurementsRestored = streamed.measurementCount - expiredTombstonesSkipped; diff --git a/src/lib/openapi/routes/sync.ts b/src/lib/openapi/routes/sync.ts index 007fd424c..6fdb94d8b 100644 --- a/src/lib/openapi/routes/sync.ts +++ b/src/lib/openapi/routes/sync.ts @@ -257,7 +257,7 @@ export const syncChangesResponse = z cursorExpired: z .boolean() .describe( - "True when the supplied cursor predates tombstone retention — drop the cursor and do a clean initial sync.", + "True when the supplied cursor predates tombstone retention, or was issued before the account's data was last restored from a backup or deleted — drop the cursor and do a clean initial sync.", ), changes: z.object({ measurements: z.object({ diff --git a/src/lib/sync/__tests__/cursor.test.ts b/src/lib/sync/__tests__/cursor.test.ts index 6136b1451..793a9f3c6 100644 --- a/src/lib/sync/__tests__/cursor.test.ts +++ b/src/lib/sync/__tests__/cursor.test.ts @@ -3,7 +3,12 @@ */ import { describe, it, expect } from "vitest"; -import { encodeCursor, decodeCursor, type SyncCursor } from "../cursor"; +import { + encodeCursor, + decodeCursor, + decodeCursorIssuedAt, + type SyncCursor, +} from "../cursor"; describe("sync cursor codec", () => { it("round-trips a per-domain keyset map through encode/decode", () => { @@ -112,3 +117,20 @@ describe("sync cursor codec", () => { }); }); }); + +describe("cursor issue time (sync reset)", () => { + const cursor: SyncCursor = { + measurements: { updatedAtMs: 1_000, id: "clx0m" }, + }; + + it("round-trips the issue time beside the watermarks", () => { + const token = encodeCursor(cursor, 1_790_000_000_000); + expect(decodeCursorIssuedAt(token)).toBe(1_790_000_000_000); + expect(decodeCursor(token)).toEqual(cursor); + }); + + it("reads no issue time from a token that carries none", () => { + expect(decodeCursorIssuedAt(encodeCursor(cursor))).toBeNull(); + expect(decodeCursorIssuedAt("not a token")).toBeNull(); + }); +}); diff --git a/src/lib/sync/cursor.ts b/src/lib/sync/cursor.ts index ebc28d14e..3b207e9d5 100644 --- a/src/lib/sync/cursor.ts +++ b/src/lib/sync/cursor.ts @@ -63,14 +63,27 @@ const CURSOR_VERSION = 2; * `cycleDays`/`cycles` domains; they decode as absent (fresh scan). */ const SUPPORTED_CURSOR_VERSIONS: readonly number[] = [1, 2]; -/** Encode a per-domain keyset map into an opaque token. */ -export function encodeCursor(cursor: SyncCursor): string { +/** + * Encode a per-domain keyset map into an opaque token. + * + * `issuedAtMs` is when the server handed the token out. The watermarks cannot + * say that: a restore writes rows back with their original `updatedAt`, so a + * cursor drained entirely after a restore can still carry watermarks older + * than it. The issue time is what `/api/sync/changes` compares against the + * account's `syncResetAt`. Additive inside the v2 envelope: a token without + * it decodes as before and counts as issued before any reset. + */ +export function encodeCursor(cursor: SyncCursor, issuedAtMs?: number): string { const d: Record = {}; for (const domain of SYNC_DOMAINS) { const wm = cursor[domain]; if (wm) d[domain] = { u: wm.updatedAtMs, i: wm.id }; } - const json = JSON.stringify({ v: CURSOR_VERSION, d }); + const json = JSON.stringify( + issuedAtMs === undefined + ? { v: CURSOR_VERSION, d } + : { v: CURSOR_VERSION, t: issuedAtMs, d }, + ); return Buffer.from(json, "utf-8").toString("base64url"); } @@ -117,3 +130,18 @@ export function decodeCursor(token: string): SyncCursor | null { return null; } } + +/** + * When a token was issued (see {@link encodeCursor}), or null for a token + * that does not say: one from before the field existed, or one that does not + * parse. + */ +export function decodeCursorIssuedAt(token: string): number | null { + try { + const json = Buffer.from(token, "base64url").toString("utf-8"); + const t = (JSON.parse(json) as { t?: unknown } | null)?.t; + return typeof t === "number" && Number.isFinite(t) ? t : null; + } catch { + return null; + } +} diff --git a/src/lib/sync/reset.ts b/src/lib/sync/reset.ts new file mode 100644 index 000000000..06704e52d --- /dev/null +++ b/src/lib/sync/reset.ts @@ -0,0 +1,19 @@ +/** + * Expire every delta cursor an account's paired clients hold. + * + * Called when the account's record is replaced wholesale: a backup restore + * (inside its transaction and again right after the commit) and the + * person's delete-all-data. `/api/sync/changes` answers `cursorExpired` to + * any cursor issued before the stamp, and the client re-initialises. See + * `User.syncResetAt` in the schema for why an incremental catch-up from such + * a cursor would miss rows. + */ +import type { Prisma, PrismaClient } from "@/generated/prisma/client"; + +export async function stampSyncReset( + db: Pick, + userId: string, + at: Date = new Date(), +): Promise { + await db.user.update({ where: { id: userId }, data: { syncResetAt: at } }); +} diff --git a/tests/integration/sync-reset-after-restore.test.ts b/tests/integration/sync-reset-after-restore.test.ts new file mode 100644 index 000000000..d0d8f90d0 --- /dev/null +++ b/tests/integration/sync-reset-after-restore.test.ts @@ -0,0 +1,210 @@ +/** + * A restore or a delete-all-data expires every delta cursor issued before it. + * + * A restore writes rows back with their original `updatedAt` and removes rows + * without leaving tombstones. A paired client that kept its cursor would, on + * its next incremental pull, miss both: the restored rows sit behind its + * watermarks and the removed ones never show up as deletions. So the restore + * stamps `User.syncResetAt`, and `/api/sync/changes` answers `cursorExpired` + * to any cursor issued before it. A cursor the feed hands out after the + * reset keeps working, although it may carry watermarks older than the reset + * (the restored rows'), which is why the comparison is on the cursor's issue + * time. + * + * Mutation checks: drop the `beforeReset` arm in the route and the first two + * tests go red on `cursorExpired`; compare the reset against the cursor's + * watermarks instead of its issue time and the cursor issued after the + * restore expires too. + */ +import { NextRequest } from "next/server"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { cookieJar, headerJar } from "./mock-next-headers"; +import { getPrismaClient, truncateAllTables } from "./setup"; +import { streamFullBackupJson } from "@/lib/export/full-backup-stream"; +import { storeBackupBlob } from "@/lib/export/store-backup-blob"; + +vi.mock("next/headers", async () => { + const { cookieJar, headerJar } = await import("./mock-next-headers"); + return { + headers: vi.fn(async () => ({ + get: (name: string) => headerJar.get(name.toLowerCase()) ?? null, + })), + cookies: vi.fn(async () => ({ + get: (name: string) => { + const value = cookieJar.get(name); + return value ? { name, value } : undefined; + }, + set: (name: string, value: string) => { + cookieJar.set(name, value); + }, + delete: (name: string) => { + cookieJar.delete(name); + }, + })), + }; +}); +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); + cookieJar.clear(); + headerJar.clear(); +}); + +interface Page { + cursor: string | null; + hasMore: boolean; + cursorExpired: boolean; + changes: { + measurements: { + upserts: Array<{ id: string }>; + tombstones: Array<{ id: string }>; + }; + }; +} + +async function pull(cursor?: string): Promise { + const { GET } = await import("@/app/api/sync/changes/route"); + const query = cursor ? `?cursor=${encodeURIComponent(cursor)}` : ""; + const res = await GET( + new NextRequest(`http://localhost/api/sync/changes${query}`), + ); + expect(res.status).toBe(200); + return ((await res.json()) as { data: Page }).data; +} + +/** Drain the feed from `cursor` to the end; the last page's cursor. */ +async function drain( + cursor?: string, +): Promise<{ cursor: string; ids: string[] }> { + const ids: string[] = []; + let next = cursor; + for (;;) { + const page = await pull(next); + expect(page.cursorExpired).toBe(false); + ids.push(...page.changes.measurements.upserts.map((m) => m.id)); + next = page.cursor!; + if (!page.hasMore) return { cursor: next, ids }; + } +} + +async function seedAccount() { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { + username: "sync-reset", + email: "sync-reset@example.test", + role: "ADMIN", + }, + }); + const session = await prisma.session.create({ + data: { userId: user.id, expiresAt: new Date(Date.now() + 600_000) }, + }); + cookieJar.set("healthlog_session", session.id); + const earlier = Date.now() - 3 * 86_400_000; + await prisma.measurement.createMany({ + data: Array.from({ length: 5 }, (_, i) => ({ + id: `m-${i}`, + userId: user.id, + type: "WEIGHT" as const, + value: 80 + i, + unit: "kg", + measuredAt: new Date(earlier + i * 60_000), + // Old row times, as a restored row carries. + createdAt: new Date(earlier), + updatedAt: new Date(earlier + i), + })), + }); + return user; +} + +async function restoreBackup(backupId: string) { + const { POST } = await import("./restore-job-driver"); + const res = await POST( + new Request(`http://localhost/api/admin/backups/${backupId}/restore`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ confirm: "RESTORE" }), + }) as never, + { params: Promise.resolve({ id: backupId }) }, + ); + expect(res.status).toBe(200); +} + +describe("sync reset after a restore or a wipe", () => { + it("expires a cursor issued before a restore, and not one issued after it", async () => { + const user = await seedAccount(); + const prisma = getPrismaClient(); + const { id: backupId } = await storeBackupBlob( + prisma, + { userId: user.id, type: "WEEKLY_AUTO" }, + (write) => + streamFullBackupJson(prisma, user.id, write, { + purpose: "disaster-recovery", + }), + ); + + // A client caught up, then a reading arrived and was synced, and then + // the account was restored to the backup, which removes that reading + // without a tombstone. + const before = await drain(); + await prisma.measurement.create({ + data: { + id: "m-after-backup", + userId: user.id, + type: "WEIGHT", + value: 90, + unit: "kg", + measuredAt: new Date(), + }, + }); + const caughtUp = await drain(before.cursor); + expect(caughtUp.ids).toEqual(["m-after-backup"]); + + await restoreBackup(backupId); + expect( + await prisma.measurement.findUnique({ where: { id: "m-after-backup" } }), + ).toBeNull(); + + // The old cursor would have reported nothing: the removal left no + // tombstone and the restored rows sit behind its watermarks. + const stale = await pull(caughtUp.cursor); + expect(stale.cursorExpired).toBe(true); + expect(stale.changes.measurements.upserts).toEqual([]); + + // Re-initialised: the whole restored record, and the new cursor holds + // although every restored row is older than the reset. + const fresh = await drain(); + expect(fresh.ids.sort()).toEqual(["m-0", "m-1", "m-2", "m-3", "m-4"]); + const next = await pull(fresh.cursor); + expect(next.cursorExpired).toBe(false); + }); + + it("expires a cursor issued before the person deleted all their data", async () => { + await seedAccount(); + const { cursor } = await drain(); + + const { DELETE } = await import("@/app/api/settings/data/route"); + const res = await DELETE( + new Request("http://localhost/api/settings/data", { + method: "DELETE", + body: JSON.stringify({ confirm: "DELETE" }), + }) as never, + ); + expect(res.status).toBe(200); + + expect((await pull(cursor)).cursorExpired).toBe(true); + const fresh = await drain(); + expect(fresh.ids).toEqual([]); + expect((await pull(fresh.cursor)).cursorExpired).toBe(false); + }); + + it("leaves an account that was never restored alone", async () => { + await seedAccount(); + const { cursor } = await drain(); + expect((await pull(cursor)).cursorExpired).toBe(false); + }); +});